"""Peak-sync mapping engine: align visual peaks to audio onsets. Maps N video segments to N audio onsets by aligning each segment's visual peak timestamp to the corresponding onset, then computes the source slice [source_start, source_end] and output timeline position. Worked example -------------- Chorus: 90.0s–108.0s in source. Onsets (absolute): [90.4, 93.1, 96.0, 99.2, 102.7] Visual peaks (sorted by time): [14.2, 38.7, 61.4, 82.1, 105.3] Forward slots: [2.7, 2.9, 3.2, 3.5, 3.5(mirrored)] ← one beat interval each Clip 0: V=14.2, A=90.4, P=0.0s, slot=2.7s, gap_before=1.35, gap_after=1.35 source=[12.85, 15.55] Clip 1: V=38.7, A=93.1, P=2.7s, slot=2.9s, gap_before=1.45, gap_after=1.45 source=[37.25, 40.15] Clip 4: V=105.3, A=102.7, P=12.3s, slot=3.5s(mirrored), gap_before=1.75, gap_after=1.75 source=[103.55, 107.05] → may clamp to segment bounds Total output duration: 2.7+2.9+3.2+3.5+3.5 = 15.8s (within 12–17s target). Cuts land at output positions 0.0, 2.7, 5.6, 8.8, 12.3s — each aligned to a beat. Visual peak of each clip is centred in its beat slot (build-up → peak → aftermath). """ from __future__ import annotations from .models import CutToMusicConfig, SyncedClip, VisualPeakSegment from .onset_detection import AudioOnset, InsufficientOnsetError class PeakSyncMapper: """Computes SyncedClip objects from visual peaks and audio onsets.""" def __init__(self, config: CutToMusicConfig) -> None: self.config = config def map( self, segments: list[VisualPeakSegment], onsets: list[AudioOnset], chorus_start: float, ) -> list[SyncedClip]: """Pair each segment to an onset and compute source/output coordinates. Pairing strategy: - Sort segments ascending by visual_peak_timestamp (temporal order). - Sort onsets ascending by time (already guaranteed by OnsetDetector). - Pair index-by-index (positional 1:1 match). - If len(onsets) < len(segments): drop lowest-score segments first to match the available onset count. For each pair (segment_n, onset_n): output_position_n = onset_n.time - onset_0.time (output starts at t=0 when the first beat hits) gap_before_n = onset_n.time - onset_{n-1}.time (use first inter-onset gap as mirror for clip 0) gap_after_n = onset_{n+1}.time - onset_n.time (use last inter-onset gap as mirror for clip N-1) source_start_n = V_peak_n - gap_before_n source_end_n = V_peak_n + gap_after_n speed_factor = clip_duration / (gap_before + gap_after) (only applied when within ±speed_factor_tolerance of 1.0) Args: segments: VisualPeakSegments from Gemini, sorted descending by score. onsets: AudioOnsets from librosa, sorted ascending by time. chorus_start: Absolute time of chorus start in source (for reference only). Returns: List of SyncedClip sorted ascending by output_position. Raises: InsufficientOnsetError: If onsets list is empty. """ if not onsets: raise InsufficientOnsetError( "Cannot map peaks to beats: no audio onsets provided." ) selected = self._select_segments(segments, onsets) gaps = self._calculate_gaps(onsets[: len(selected)]) synced: list[SyncedClip] = [] for i, (seg, onset) in enumerate(zip(selected, onsets[: len(selected)], strict=True)): gap_before = gaps[i][0] gap_after = gaps[i][1] slot_duration = gap_before + gap_after output_position = onset.time - onsets[0].time source_start = seg.visual_peak_timestamp - gap_before source_end = seg.visual_peak_timestamp + gap_after # Clamp to segment bounds source_start = max(source_start, seg.start) source_end = min(source_end, seg.end) # If clamping shortened the clip, re-center on visual peak available = source_end - source_start if available < slot_duration * 0.5: # Very short segment: center on peak, use whatever is available half = available / 2.0 source_start = max(seg.start, seg.visual_peak_timestamp - half) source_end = min(seg.end, seg.visual_peak_timestamp + half) clip_duration = source_end - source_start speed_factor = ( clip_duration / slot_duration if slot_duration > 0 else 1.0 ) # Outside tolerance → crop-only mode (no time warp) if abs(speed_factor - 1.0) > self.config.speed_factor_tolerance: speed_factor = 1.0 synced.append( SyncedClip( source_segment=seg, audio_onset_time=onset.time, source_start=source_start, source_end=source_end, output_position=output_position, speed_factor=speed_factor, duration=clip_duration, slot_duration=slot_duration, ) ) return synced def _select_segments( self, segments: list[VisualPeakSegment], onsets: list[AudioOnset], ) -> list[VisualPeakSegment]: """Trim segment list to match onset count, then sort by visual_peak_timestamp. Segments arrive sorted descending by score (highest first from Gemini). If we need to drop some, we pop from the end (lowest score). We then re-sort ascending by visual_peak_timestamp for positional matching with time-sorted onsets so the output flows chronologically. Args: segments: Score-sorted (descending) list of segments. onsets: Time-sorted (ascending) list of onsets (determines target count). Returns: Up to len(onsets) segments, sorted ascending by visual_peak_timestamp. """ target = min(len(segments), len(onsets)) # Sort descending by score to ensure we keep the best ones best = sorted(segments, key=lambda s: s.score, reverse=True)[:target] # Re-sort ascending by visual_peak_timestamp for chronological output return sorted(best, key=lambda s: s.visual_peak_timestamp) def _calculate_gaps(self, onsets: list[AudioOnset]) -> list[tuple[float, float]]: """Compute (gap_before, gap_after) in seconds for each onset. Each clip spans exactly ONE inter-onset gap, split evenly around the visual peak. This guarantees that when clips are sequentially concatenated, each clip starts at output_position = onset_N - onset_0, and the assembly audio (starting from onset_0) places beats exactly at clip transition points. Slot for clip N = onset_{N+1} - onset_N (one beat gap). For the last clip, the slot mirrors the preceding inter-onset gap. gap_before = gap_after = slot / 2 (visual peak centered in clip). Args: onsets: Time-sorted list of onsets (at most N elements). Returns: List of (gap_before, gap_after) tuples, one per onset. """ n = len(onsets) if n == 1: return [(1.5, 1.5)] # Fallback: 1.5s each side for a single onset times = [o.time for o in onsets] gaps: list[tuple[float, float]] = [] for i in range(n): if i < n - 1: slot = times[i + 1] - times[i] else: slot = times[i] - times[i - 1] # Mirror the last inter-onset gap half = slot / 2.0 gaps.append((half, half)) return gaps