"""Clip rendering and FFmpeg assembly for the cut-to-music pipeline. Renders each SyncedClip through the existing Reframer + FFmpegExecutor, then concatenates all clips with the chorus audio overlay in a single FFmpeg concat-filter command. Assembly command structure (N clips): ffmpeg -i clip_0_muted.mp4 ... -i clip_N_muted.mp4 -ss {audio_start} -t {total_duration} -i {original_video} -filter_complex "[0:v]...[N-1:v] concat=n=N:v=1:a=0 [vout]; [{N}:a] aformat=... [aout]" -map [vout] -map [aout] -c:v libx264 -preset fast -crf 20 -pix_fmt yuv420p -c:a aac -b:a 192k -ar 44100 -r 30 -movflags +faststart -y {output} Design choices: - Audio comes entirely from the original video's chorus section (clips are muted). - -r 30 forces CFR output (required by TikTok / Instagram). - Speed warp (setpts) is applied AFTER reframing, on the already-cropped 9:16 clip, so the Reframer works with the original video's absolute timestamps. """ from __future__ import annotations import subprocess from pathlib import Path from rich.console import Console from .config import settings from .execution import FFmpegExecutor from .ffmpeg_utils import probe_video from .models import CutToMusicConfig, SyncedClip, VideoSegment from .reframing import Reframer console = Console() class CutToMusicAssembler: """Renders individual clips and assembles the final cut-to-music reel.""" def __init__(self, config: CutToMusicConfig) -> None: self.config = config self.reframer = Reframer( debug=config.debug, output_dir=config.output_dir, min_scene_detection_segment=5.0, ) self.executor = FFmpegExecutor() def render_clip( self, synced_clip: SyncedClip, clip_index: int, work_dir: Path, input_video: Path, ) -> Path: """Render a single SyncedClip to a muted 9:16 intermediate MP4. Steps: 1. Build a synthetic VideoSegment from source_start / source_end. 2. Reframer.generate_crop_path() — uses original video timestamps directly. 3. FFmpegExecutor.generate_clip() — produces reframed 9:16 clip with audio. 4. If speed_factor != 1.0: apply time-warp + strip audio in one FFmpeg pass. 5. Else: strip audio only. Time-warping is applied AFTER reframing so the Reframer's frame-by-frame seek uses unmodified absolute timestamps from the original video. Args: synced_clip: Peak-sync mapping result for this clip. clip_index: 0-based index (used in intermediate filenames). work_dir: Temporary directory for intermediates. input_video: Path to the original source video. Returns: Path to a muted 9:16 intermediate MP4 ready for concatenation. """ synthetic_segment = VideoSegment( start=synced_clip.source_start, end=synced_clip.source_end, score=synced_clip.source_segment.score, description=synced_clip.source_segment.description, segment_type="cut_to_music", ) console.print( f" [dim]Reframing clip {clip_index + 1}: " f"{synced_clip.source_start:.1f}s–{synced_clip.source_end:.1f}s " f"(peak at {synced_clip.source_segment.visual_peak_timestamp:.1f}s)[/dim]" ) crop_path = self.reframer.generate_crop_path( input_video, synthetic_segment, self.config.target_aspect_ratio, ) crop_path.debug = self.config.debug # Build caption renderer if text overlay requested caption_renderer = None if self.config.caption_text and crop_path.keyframes: from .captioning import CaptionConfig, CaptionRenderer caption_renderer = CaptionRenderer( CaptionConfig( text=self.config.caption_text, style=self.config.caption_style.value, position=self.config.caption_position.value, frame_width=crop_path.keyframes[0].width, frame_height=crop_path.keyframes[0].height, ) ) # Use absolute paths to avoid CWD sensitivity in FFmpeg subprocesses reframed_path = (work_dir / f"clip_{clip_index:02d}_reframed.mp4").resolve() self.executor.generate_clip(input_video, reframed_path, crop_path, caption_renderer) if not reframed_path.exists() or reframed_path.stat().st_size == 0: raise RuntimeError( f"Reframed clip is missing or empty: {reframed_path}\n" "This usually means the source segment timestamps are out of bounds." ) # Produce final muted intermediate (with optional speed warp) muted_path = (work_dir / f"clip_{clip_index:02d}_muted.mp4").resolve() sf = synced_clip.speed_factor if abs(sf - 1.0) > 1e-3: console.print( f" [dim]Clip {clip_index + 1}: applying speed factor {sf:.3f}[/dim]" ) try: self._warp_and_strip_audio(reframed_path, muted_path, sf) except subprocess.CalledProcessError as e: console.print( f" [yellow]Warning: speed warp failed for clip {clip_index + 1} " f"(exit {e.returncode}), falling back to direct strip[/yellow]" ) if e.stderr: console.print(f" [dim]FFmpeg: {e.stderr[:400]}[/dim]") # Fallback: strip audio without warping rather than hard-failing self._strip_audio(reframed_path, muted_path) else: self._strip_audio(reframed_path, muted_path) reframed_path.unlink(missing_ok=True) # Loop UGC clips that are shorter than their beat-slot target duration. # This can happen when Gemini selects a short peak segment from an extra video # but the beat slot (derived from onset spacing) is longer. if self.config.extra_videos and synced_clip.slot_duration > 0: is_ugc = any( str(ev.resolve()) == str(input_video.resolve()) for ev in self.config.extra_videos ) if is_ugc: actual_duration = probe_video(muted_path).duration if actual_duration < synced_clip.slot_duration * 0.95: console.print( f" [dim]Clip {clip_index + 1}: looping UGC clip " f"({actual_duration:.2f}s → {synced_clip.slot_duration:.2f}s)[/dim]" ) looped_path = (work_dir / f"clip_{clip_index:02d}_looped.mp4").resolve() self._loop_to_duration(muted_path, looped_path, synced_clip.slot_duration) muted_path.unlink(missing_ok=True) muted_path = looped_path return muted_path def assemble( self, clip_paths: list[Path], synced_clips: list[SyncedClip], source_video: Path, audio_start: float, output_path: Path, ) -> Path: """Concatenate muted clips and overlay chorus audio from the original video. Uses the FFmpeg concat filter for timestamp-precise concatenation. The original source video is the last FFmpeg input, seeked with '-ss' placed *before* '-i' for fast keyframe-level seeking. Audio extraction starts from audio_start (= onset_0.time, the first detected beat). This ensures that clip N, which starts at output_position_N = onset_N - onset_0 in the output video, aligns exactly with beat N in the output audio — because both are measured relative to onset_0. Args: clip_paths: Muted 9:16 intermediate clips (in output order). synced_clips: Corresponding SyncedClip objects (for duration info). source_video: Original source video (chorus audio extracted from here). audio_start: Start of audio extraction in source video (seconds). Should be onsets[0].time so beats align with clip start positions. output_path: Destination MP4 path. Returns: Path to the assembled final video. Raises: ValueError: If clip_paths and synced_clips lengths differ. subprocess.CalledProcessError: If FFmpeg fails. """ if len(clip_paths) != len(synced_clips): raise ValueError( f"clip_paths ({len(clip_paths)}) and synced_clips ({len(synced_clips)}) " "must have the same length" ) n = len(clip_paths) # Probe all clips to get actual rendered durations and the target resolution. # We use actual durations (not slot_duration) for the audio extraction length # so the audio track never exceeds the video length — which would cause black # frames when onsets are unevenly spaced and clips don't fill their beat slots. clip_metas = [probe_video(cp) for cp in clip_paths] total_duration = sum(m.duration for m in clip_metas) # Determine target resolution from the first clip so all clips are # normalised to the same dimensions before concat. When clips come from # different source videos (multi-video mode) the reframer may produce # different 9:16 resolutions depending on each source's native height; # the concat filter requires identical dimensions. first_meta = clip_metas[0] target_w = first_meta.width target_h = first_meta.height # Filter graph: scale every clip to the target resolution, correct SAR, # concatenate video streams, then normalise source audio sample format. sar_parts = "".join( f"[{i}:v] scale={target_w}:{target_h},setsar=1 [v{i}];" for i in range(n) ) concat_inputs = "".join(f"[v{i}]" for i in range(n)) filter_complex = ( f"{sar_parts}" f"{concat_inputs} concat=n={n}:v=1:a=0 [vout];" f" [{n}:a] aformat=sample_fmts=fltp:sample_rates=44100" f":channel_layouts=stereo [aout]" ) cmd: list[str] = [str(settings.ffmpeg_path), "-y"] # N muted clip inputs (use resolved absolute paths) for cp in clip_paths: cmd += ["-i", str(cp.resolve())] # Original video for audio (fast seek before -i). # audio_start = onset_0.time so beat positions in the audio track are measured # from the same reference as output_position (= onset_N - onset_0). cmd += [ "-ss", str(audio_start), "-t", str(total_duration), "-i", str(source_video.resolve()), ] cmd += [ "-filter_complex", filter_complex, "-map", "[vout]", "-map", "[aout]", "-c:v", "libx264", "-preset", "fast", "-crf", "20", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", "-ar", "44100", "-r", "30", # Force CFR — required by TikTok / Instagram "-movflags", "+faststart", str(output_path.resolve()), ] console.print( f"\n[bold cyan]Assembling final reel " f"({n} clips, {total_duration:.1f}s)...[/bold cyan]" ) result = subprocess.run(cmd, capture_output=True) if result.returncode != 0: stderr_text = result.stderr.decode(errors="replace") raise subprocess.CalledProcessError( result.returncode, cmd, stderr=stderr_text, ) return output_path def _warp_and_strip_audio( self, input_path: Path, output_path: Path, speed_factor: float, ) -> None: """Apply time-warp to video stream and remove audio in one pass. setpts=PTS*(1/sf) changes playback speed. Audio is discarded (-an) because the final assembly supplies chorus audio separately. Speed factor is bounded to [0.9, 1.1] by PeakSyncMapper, so a single setpts filter is always sufficient (no atempo chaining needed). Args: input_path: Reframed clip (may have audio). output_path: Warped, muted output clip. speed_factor: Playback speed multiplier (1.0 = no change). """ pts_factor = 1.0 / speed_factor cmd = [ str(settings.ffmpeg_path), "-y", "-i", str(input_path), # fps=30 normalises any VFR timestamps before setpts. # scale=trunc(iw/2)*2:trunc(ih/2)*2 forces even dimensions — required # by libx264 (reframer may produce odd-pixel crops, e.g. 527px wide). "-vf", f"fps=30,scale=trunc(iw/2)*2:trunc(ih/2)*2,setpts=PTS*{pts_factor:.6f}", "-an", "-c:v", "libx264", "-preset", "fast", "-crf", "20", "-pix_fmt", "yuv420p", str(output_path), ] result = subprocess.run(cmd, capture_output=True) if result.returncode != 0: stderr_text = result.stderr.decode(errors="replace") raise subprocess.CalledProcessError( result.returncode, cmd, stderr=stderr_text, ) def _loop_to_duration( self, input_path: Path, output_path: Path, target_duration: float, ) -> None: """Loop a muted clip until it reaches target_duration. Used when a UGC source clip is shorter than the beat-slot it occupies. Input must already be a normalised 30fps H.264/yuv420p clip (muted). Args: input_path: Muted normalised clip to loop. output_path: Output clip of exactly target_duration seconds. target_duration: Desired output duration in seconds. """ cmd = [ str(settings.ffmpeg_path), "-y", "-stream_loop", "-1", "-i", str(input_path), "-t", str(target_duration), "-vf", "fps=30,scale=trunc(iw/2)*2:trunc(ih/2)*2", "-an", "-c:v", "libx264", "-preset", "fast", "-crf", "20", "-pix_fmt", "yuv420p", str(output_path), ] result = subprocess.run(cmd, capture_output=True) if result.returncode != 0: stderr_text = result.stderr.decode(errors="replace") raise subprocess.CalledProcessError(result.returncode, cmd, stderr=stderr_text) def _strip_audio(self, input_path: Path, output_path: Path) -> None: """Re-encode to normalised 30fps H.264/yuv420p and remove audio. We re-encode (rather than stream-copy) so that every intermediate clip has the same frame rate, pixel format, and codec profile. FFmpeg's concat filter requires all inputs to be identical in these properties; if some clips went through _warp_and_strip_audio (which forces fps=30) and others only stream-copied, the concat would silently fail or produce broken output. Args: input_path: Clip with audio. output_path: Normalised, muted clip. """ cmd = [ str(settings.ffmpeg_path), "-y", "-i", str(input_path), # scale=trunc(iw/2)*2:trunc(ih/2)*2 forces even dimensions — required # by libx264 (reframer may produce odd-pixel crops, e.g. 527px wide). "-vf", "fps=30,scale=trunc(iw/2)*2:trunc(ih/2)*2", "-an", "-c:v", "libx264", "-preset", "fast", "-crf", "20", "-pix_fmt", "yuv420p", str(output_path), ] result = subprocess.run(cmd, capture_output=True) if result.returncode != 0: stderr_text = result.stderr.decode(errors="replace") raise subprocess.CalledProcessError( result.returncode, cmd, stderr=stderr_text, )