"""Percussive onset detection for cut-to-music beat alignment. Uses librosa to find the N most prominent, evenly-spaced percussive transients within a chorus window extracted from the source video. """ from __future__ import annotations import subprocess import tempfile from dataclasses import dataclass from pathlib import Path import numpy as np from .config import settings class InsufficientOnsetError(ValueError): """Raised when no strong percussive onsets are found in the chorus window.""" @dataclass class AudioOnset: """A detected percussive onset.""" time: float """Absolute time in source audio (seconds), within the original video timeline.""" prominence: float """Onset strength at this point (0-1 normalized relative to the max in the window).""" class OnsetDetector: """Detects percussive transients in a chorus window using librosa.""" def __init__(self, min_spacing: float = 2.0) -> None: """ Args: min_spacing: Minimum seconds between selected onsets to prevent clumping. """ self.min_spacing = min_spacing def extract_onsets( self, video_path: Path, chorus_start: float, chorus_end: float, num_onsets: int, ) -> list[AudioOnset]: """Extract the N most prominent, evenly-spaced onsets in the chorus window. Algorithm: 1. Extract WAV from [chorus_start, chorus_end] via FFmpeg (mono, 22050 Hz). 2. Load into librosa. 3. Compute onset_strength envelope (aggregate='median' for robustness). 4. Run onset_detect with backtrack=True so each onset timestamp is shifted to the preceding local energy minimum — the visual cut arrives *on* the transient rather than just after it. 5. Filter: keep only onsets with prominence >= 50th percentile. 6. Apply min_spacing greedy filter (earliest first). 7. If still > num_onsets after spacing filter, take top-N by prominence. 8. Sort ascending by time, convert to absolute timestamps. Args: video_path: Source video path (audio will be extracted). chorus_start: Start of chorus window in source video (seconds). chorus_end: End of chorus window in source video (seconds). num_onsets: Target number of onsets to return. Returns: List of AudioOnset sorted ascending by time. May contain fewer than num_onsets if insufficient strong onsets exist. Raises: InsufficientOnsetError: If no strong onsets are detected after all filtering. RuntimeError: If FFmpeg audio extraction fails. """ import librosa with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp: wav_path = Path(tmp.name) try: self._extract_audio_wav(video_path, chorus_start, chorus_end, wav_path) y, sr = librosa.load(str(wav_path), sr=22050, mono=True) finally: wav_path.unlink(missing_ok=True) # Onset strength envelope (median aggregation is more robust than mean for music) oenv = librosa.onset.onset_strength(y=y, sr=sr, aggregate=np.median) # Detect onsets; backtrack=True shifts each onset to the preceding energy trough onset_frames = librosa.onset.onset_detect( onset_envelope=oenv, sr=sr, backtrack=True, units="frames", ) if len(onset_frames) == 0: raise InsufficientOnsetError( "No percussive onsets found in chorus window. " "The audio may be too sparse or ambient — try a different video section." ) # Convert frames to relative times (seconds within the extracted WAV) frame_times = librosa.frames_to_time(onset_frames, sr=sr) onset_strengths = oenv[onset_frames] # Normalize strength to [0, 1] max_strength = float(onset_strengths.max()) if onset_strengths.max() > 0 else 1.0 normalized = onset_strengths / max_strength # Filter: keep only those at or above the 50th percentile of strength threshold = float(np.percentile(normalized, 50)) mask = normalized >= threshold frame_times = frame_times[mask] normalized = normalized[mask] if len(frame_times) == 0: raise InsufficientOnsetError( "No strong percussive onsets found after filtering. " "Try adjusting the chorus window or check that the video has a clear beat." ) # Build onset objects (still relative to wav start, i.e. chorus_start) candidates = [ AudioOnset(time=float(t), prominence=float(p)) for t, p in zip(frame_times, normalized, strict=True) ] # Apply minimum spacing greedy filter spaced = self._apply_spacing_filter(candidates, self.min_spacing) if not spaced: raise InsufficientOnsetError( f"No onsets remain after applying {self.min_spacing}s minimum spacing. " "Try reducing --min-onset-spacing or choosing a more rhythmic chorus section." ) # If more than needed, keep top-N by prominence if len(spaced) > num_onsets: spaced.sort(key=lambda o: o.prominence, reverse=True) spaced = spaced[:num_onsets] # Sort by time (ascending) and convert to absolute timestamps spaced.sort(key=lambda o: o.time) return [ AudioOnset(time=onset.time + chorus_start, prominence=onset.prominence) for onset in spaced ] def _extract_audio_wav( self, video_path: Path, start: float, end: float, output_path: Path, ) -> None: """Extract an audio segment to WAV using FFmpeg. Uses -ss before -i for fast (keyframe-level) seeking — sufficient for audio. Output: mono, 22050 Hz WAV. Args: video_path: Source video. start: Start time in source (seconds). end: End time in source (seconds). output_path: Destination WAV path. Raises: RuntimeError: If FFmpeg returns a non-zero exit code. """ duration = end - start cmd = [ str(settings.ffmpeg_path), "-y", "-ss", str(start), "-t", str(duration), "-i", str(video_path), "-ac", "1", "-ar", "22050", "-vn", "-f", "wav", str(output_path), ] result = subprocess.run(cmd, capture_output=True) if result.returncode != 0: raise RuntimeError( f"FFmpeg audio extraction failed:\n{result.stderr.decode(errors='replace')}" ) def _apply_spacing_filter( self, onsets: list[AudioOnset], min_spacing: float, ) -> list[AudioOnset]: """Greedy forward filter: keep earliest onset, then skip within min_spacing. Args: onsets: Onsets sorted ascending by time. min_spacing: Minimum seconds between kept onsets. Returns: Filtered list, still sorted ascending by time. """ if not onsets: return [] kept: list[AudioOnset] = [onsets[0]] for onset in onsets[1:]: if onset.time - kept[-1].time >= min_spacing: kept.append(onset) return kept