"""Top-level orchestrator for the cut-to-music pipeline. Coordinates the 4-phase Semantic Peak-Sync algorithm: Phase 1: Dual Gemini queries (chorus + visual peaks) — run concurrently Phase 2: Local onset detection (librosa) Phase 3: Peak-sync mapping — visual peaks → audio onsets Phase 4: Clip rendering (reframe each clip) + FFmpeg assembly """ from __future__ import annotations import shutil from pathlib import Path from rich.console import Console from rich.table import Table from .cut_to_music_assembler import CutToMusicAssembler from .cut_to_music_intelligence import CutToMusicIntelligence from .models import ChorusWindow, CutToMusicConfig, SyncedClip, VisualPeakSegment from .onset_detection import AudioOnset, InsufficientOnsetError, OnsetDetector from .peak_sync import PeakSyncMapper console = Console() class CutToMusicProcessor: """Runs the complete cut-to-music pipeline end-to-end.""" def __init__(self, config: CutToMusicConfig) -> None: self.config = config self.intelligence = CutToMusicIntelligence() self.onset_detector = OnsetDetector(min_spacing=config.min_onset_spacing) self.mapper = PeakSyncMapper(config) self.assembler = CutToMusicAssembler(config) def process(self, output_dir: Path | None = None) -> Path: """Execute the full cut-to-music pipeline. Args: output_dir: Where to write the assembled reel. Overrides ``config.output_dir`` for this call only; the config object is not mutated. Defaults to ``config.output_dir`` (``./output``). Returns: Path to the assembled output MP4. Raises: InsufficientOnsetError: If fewer than 2 strong onsets are detected. RuntimeError: On Gemini API or FFmpeg failure. """ # Phases 1+2 chorus, visual_peaks, onsets = self.analyze_only() console.print( f"[green]✓[/green] Chorus: {chorus.chorus_start:.1f}s–{chorus.chorus_end:.1f}s" f" ({chorus.chorus_end - chorus.chorus_start:.1f}s)" ) if chorus.description: console.print(f" [dim]{chorus.description}[/dim]") console.print( f"[green]✓[/green] Visual peaks: {len(visual_peaks)} segments identified " f"(using best {self.config.num_segments})" ) if not visual_peaks: raise RuntimeError( "Gemini returned no usable visual peak segments. " "Try a different video or check the Gemini API response." ) self._print_visual_peaks(visual_peaks) if len(onsets) < len(visual_peaks): console.print( f"[yellow]Warning: Only {len(onsets)} strong onset(s) found " f"(requested {self.config.num_segments}). " f"Reducing to {len(onsets)} clip(s).[/yellow]" ) console.print( f"[green]✓[/green] {len(onsets)} onset(s) detected: " + ", ".join(f"{o.time:.2f}s" for o in onsets) ) # Phases 3+3.5+4 synced_clips = self.sync(visual_peaks, onsets, chorus) return self.render(synced_clips, output_dir=output_dir) def sync( self, visual_peaks: list[VisualPeakSegment], onsets: list[AudioOnset], chorus: ChorusWindow, ) -> list[SyncedClip]: """Phases 3+3.5: map visual peaks to audio onsets and snap clip starts to scene cuts. Takes the output of *analyze_only()* — optionally after user-adjustment in an external UI — and returns fully-specified SyncedClip objects ready for rendering. Args: visual_peaks: Scored visual peak segments from Gemini (or user-adjusted). onsets: Beat onset timestamps from the chorus audio. chorus: Chorus window used to interpret the onset times. Returns: SyncedClip list sorted ascending by output position. """ console.print("\n[bold cyan]Phase 3: Mapping visual peaks to audio onsets...[/bold cyan]") synced_clips = self.mapper.map(visual_peaks, onsets, chorus.chorus_start) self._print_sync_table(synced_clips) console.print(" Snapping clip starts to scene boundaries...") synced_clips = self._snap_clip_starts(synced_clips) total_duration = sum(sc.slot_duration for sc in synced_clips) console.print(f"[green]✓[/green] Total output duration: {total_duration:.1f}s") return synced_clips def render( self, synced_clips: list[SyncedClip], output_dir: Path | None = None, ) -> Path: """Phase 4: render clips and assemble the final beat-synced output video. Takes the output of *sync()* — optionally after further user inspection — and produces the final ``cut_to_music.mp4``. The audio start position is derived from ``synced_clips[0].audio_onset_time`` so no additional parameters are needed. Args: synced_clips: Fully-specified clip mappings from *sync()*. output_dir: Where to write the assembled reel. Overrides ``config.output_dir`` for this call only; the config object is not mutated. Defaults to ``config.output_dir`` (``./output``). Returns: Path to the assembled output MP4. """ video_path = self.config.input_video out_dir = output_dir or self.config.output_dir console.print("\n[bold cyan]Phase 4: Rendering clips...[/bold cyan]") out_dir.mkdir(parents=True, exist_ok=True) work_dir = out_dir / "work" work_dir.mkdir(exist_ok=True) try: clip_paths: list[Path] = [] for i, sc in enumerate(synced_clips): clip_source = sc.source_segment.source_video or video_path muted = self.assembler.render_clip( synced_clip=sc, clip_index=i, work_dir=work_dir, input_video=clip_source, ) clip_paths.append(muted) console.print( f"[green]✓[/green] Clip {i + 1}/{len(synced_clips)} rendered" ) # audio_start = synced_clips[0].audio_onset_time (= onsets[0].time): # audio and video are both measured from onset_0, so clip N at # output_position_N = onset_N - onset_0 aligns exactly with beat N. output_path = out_dir / "cut_to_music.mp4" self.assembler.assemble( clip_paths=clip_paths, synced_clips=synced_clips, source_video=video_path, audio_start=synced_clips[0].audio_onset_time, output_path=output_path, ) finally: if not self.config.debug and work_dir.exists(): shutil.rmtree(work_dir, ignore_errors=True) console.print(f"\n[bold green]Done![/bold green] Output: {output_path}") return output_path def analyze_only(self) -> tuple[ChorusWindow, list[VisualPeakSegment], list]: """Run Phases 1 + 2 only (for --dry-run mode). Returns: Tuple of (chorus, visual_peaks, onsets) for display. """ video_path = self.config.input_video console.print("\n[bold cyan]Phase 1: Analyzing video with Gemini...[/bold cyan]") chorus, visual_peaks = self.intelligence.analyze( video_path, num_segments=self.config.num_segments + 1, segment_duration=self.config.segment_duration, extra_videos=self.config.extra_videos, video_source=self.config.video_source, ) console.print("\n[bold cyan]Phase 2: Detecting beat onsets in chorus...[/bold cyan]") try: onsets = self.onset_detector.extract_onsets( video_path, chorus_start=chorus.chorus_start, chorus_end=chorus.chorus_end, num_onsets=self.config.num_segments, ) except InsufficientOnsetError as e: console.print(f"[yellow]Warning:[/yellow] {e}") onsets = [] return chorus, visual_peaks, onsets def _snap_clip_starts(self, synced_clips: list[SyncedClip]) -> list[SyncedClip]: """Snap each clip's source_start to the nearest PySceneDetect scene cut. Gemini timestamps have ±0.5–1s precision. For each SyncedClip we probe a ±2s window around source_start and, if a frame-accurate cut is found there, snap to it — provided: - The clip stays at least 1s long after the adjustment. - The visual peak remains inside the new clip window. After snapping source_start, source_end is shifted to source_start + slot_duration (clamped to the segment's end) so the clip duration stays equal to the slot. This keeps sum(video durations) == sum(slot_durations) == the audio extraction length, preventing silent frames at the end of the assembled reel. """ result = [] for sc in synced_clips: source_video = sc.source_segment.source_video or self.config.input_video peak_ts = sc.source_segment.visual_peak_timestamp # Search window must reach at least 0.5s past the visual peak so # the scene boundary *starting* the peak's scene is always found. window_after = max(2.0, peak_ts - sc.source_start + 0.5) boundary = self.assembler.reframer.find_scene_boundary_near( source_video, sc.source_start, window_after=window_after, before_time=peak_ts, ) if ( boundary is not None and abs(boundary - sc.source_start) > 0.05 # Ensure the full slot fits in the source segment after the snap. # sc.source_end only reaches source_start + slot_duration, so for # forward snaps it can be less than boundary + slot_duration; check # against source_segment.end (the Gemini-identified window) instead. and sc.source_segment.end - boundary >= sc.slot_duration * 0.9 and sc.source_segment.visual_peak_timestamp > boundary ): delta = boundary - sc.source_start new_start = round(boundary, 3) # Move source_end so the clip stays exactly slot_duration long. # Clamp to the segment's end boundary in case the shift overshoots. new_end = min( round(new_start + sc.slot_duration, 3), sc.source_segment.end, ) # End-snap: if new_end falls within 0.3s past a scene cut in the # source, trim it back to that cut. This prevents end-of-clip # stray frames when a forward start-snap moves source_end into # the opening frames of a new scene. end_boundary = self.assembler.reframer.find_scene_boundary_near( source_video, new_end, window_before=0.3, window_after=0.1, before_time=new_end, ) if end_boundary is not None and (new_end - end_boundary) < 0.3: new_end = round(end_boundary, 3) new_duration = round(new_end - new_start, 3) # Recompute speed_factor against slot_duration now that source_end moved. new_sf = new_duration / sc.slot_duration if sc.slot_duration > 0 else 1.0 if abs(new_sf - 1.0) > self.config.speed_factor_tolerance: new_sf = 1.0 console.print( f" [dim]Snap start: {sc.source_start:.2f}s → {boundary:.2f}s " f"({delta:+.2f}s)[/dim]" ) sc = sc.model_copy(update={ "source_start": new_start, "source_end": new_end, "duration": new_duration, "speed_factor": new_sf, }) result.append(sc) return result def _print_visual_peaks(self, peaks: list[VisualPeakSegment]) -> None: multi_source = len({p.source_video for p in peaks if p.source_video}) > 1 table = Table(show_header=True, header_style="bold magenta") table.add_column("#", style="cyan", width=4) table.add_column("Segment", style="white", width=18) table.add_column("Visual Peak", style="yellow", width=14) table.add_column("Score", style="green", width=8) if multi_source: table.add_column("Source", style="blue", width=20) table.add_column("Description", style="white") for i, seg in enumerate(sorted(peaks, key=lambda s: s.score, reverse=True), 1): row = [ str(i), f"{seg.start:.1f}s – {seg.end:.1f}s", f"{seg.visual_peak_timestamp:.2f}s", f"{seg.score:.2f}", ] if multi_source: row.append(seg.source_video.name if seg.source_video else "–") row.append(seg.description[:60] + ("…" if len(seg.description) > 60 else "")) table.add_row(*row) console.print(table) def _print_sync_table(self, synced_clips: list[SyncedClip]) -> None: table = Table(show_header=True, header_style="bold magenta") table.add_column("#", style="cyan", width=4) table.add_column("Source clip", style="white", width=18) table.add_column("Output pos", style="yellow", width=12) table.add_column("Onset", style="green", width=10) table.add_column("Speed", style="white", width=8) table.add_column("Dur", style="white", width=7) for i, sc in enumerate(synced_clips, 1): sf_str = f"{sc.speed_factor:.2f}x" if abs(sc.speed_factor - 1.0) > 1e-3 else "1.00×" table.add_row( str(i), f"{sc.source_start:.1f}s – {sc.source_end:.1f}s", f"{sc.output_position:.2f}s", f"{sc.audio_onset_time:.2f}s", sf_str, f"{sc.duration:.1f}s", ) console.print(table)