92cc09e908
- 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
59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
"""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)
|