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
This commit is contained in:
58
lightsync/audio/beats.py
Normal file
58
lightsync/audio/beats.py
Normal file
@@ -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)
|
||||
Reference in New Issue
Block a user