From 92cc09e908da6631ccfe2880250b0bf0fee75567 Mon Sep 17 00:00:00 2001
From: Claude <claude@groll.cloud>
Date: Mon, 6 Apr 2026 15:35:58 +0000
Subject: [PATCH] feat(02-03): add librosa beat detection module
 (lightsync/audio/beats.py)

- analyze(path) -> {tempo: float, beats: [float, ...]} using librosa.beat.beat_track
- Returns beat timestamps in seconds converted from librosa frame indices
- analyze_async() wraps in asyncio.to_thread for non-blocking use from FastAPI
- Supports MP3/WAV/FLAC/OGG via librosa's ffmpeg/soundfile fallback chain
---
 lightsync/audio/beats.py | 58 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 58 insertions(+)
 create mode 100644 lightsync/audio/beats.py

diff --git a/lightsync/audio/beats.py b/lightsync/audio/beats.py
new file mode 100644
index 0000000..cff64d8
--- /dev/null
+++ b/lightsync/audio/beats.py
@@ -0,0 +1,58 @@
+"""Beat detection module using librosa (Phase 02-03 architecture redesign).
+
+analyze(path) -> {"tempo": float, "beats": [float, ...]}
+
+Uses librosa.beat.beat_track to detect tempo and beat timestamps.
+Runs in asyncio.to_thread — can take several seconds for long files.
+"""
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+
+
+def analyze(path: str) -> dict[str, Any]:
+    """Analyze beat structure of an audio file.
+
+    Args:
+        path: Absolute path to audio file (MP3, WAV, FLAC, OGG supported via librosa/ffmpeg).
+
+    Returns:
+        {"tempo": float, "beats": [float, ...]}
+        where beats is a list of timestamps in seconds.
+
+    Raises:
+        FileNotFoundError: if path does not exist
+        RuntimeError: if librosa analysis fails
+    """
+    import librosa  # noqa: PLC0415 — deferred import, heavy package
+
+    p = Path(path)
+    if not p.exists():
+        raise FileNotFoundError(f"Audio file not found: {path}")
+
+    # Load mono audio at native sample rate
+    # librosa handles MP3 via audioread/soundfile fallbacks automatically
+    y, sr = librosa.load(path, sr=None, mono=True)
+
+    # beat_track returns (tempo, beat_frames) — frames are sample indices
+    tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr)
+
+    # Convert beat frame indices to timestamps in seconds
+    beat_times: list[float] = librosa.frames_to_time(beat_frames, sr=sr).tolist()
+
+    # tempo may be a numpy scalar — cast to Python float
+    tempo_value = float(np.asarray(tempo).flat[0])
+
+    return {
+        "tempo": tempo_value,
+        "beats": beat_times,
+    }
+
+
+async def analyze_async(path: str) -> dict[str, Any]:
+    """Async wrapper — runs analyze() in a thread pool to avoid blocking the event loop."""
+    return await asyncio.to_thread(analyze, path)