"""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)