diff --git a/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py b/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py index 28e245b3..b5f1571b 100644 --- a/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py +++ b/services/analysis-engine/src/bandscope_analysis/temporal/analyzer.py @@ -24,6 +24,36 @@ (DeprecationWarning, r".*pkg_resources is deprecated.*", r".*librosa.*"), (FutureWarning, r".*Numba.*", r".*numba.*"), ) +# ponytail: assumes 4/4; upgrade to meter estimation or a madmom DBN if other meters matter. +BEATS_PER_BAR = 4 + + +def _estimate_downbeats( + onset_env: NDArray[np.floating[Any]], + beat_frames: NDArray[np.integer[Any]], + beat_times: NDArray[np.floating[Any]], + beats_per_bar: int = BEATS_PER_BAR, +) -> list[float]: + """Pick the bar phase whose beats carry the most onset energy as the downbeats. + + Downbeats are typically the strongest onset in a bar, so instead of blindly + treating beat 0 as the downbeat we sample the onset-strength envelope at each + beat and choose the phase (0..beats_per_bar-1) with the highest mean strength. + This looks at the actual audio rather than assuming beat 0 starts the bar. + """ + if len(beat_times) == 0: + return [] + if len(beat_times) < beats_per_bar or len(onset_env) == 0: + return [float(beat_times[0])] + idx = np.clip(beat_frames, 0, len(onset_env) - 1) + beat_strength = onset_env[idx] + best_phase, best_score = 0, -np.inf + for phase in range(beats_per_bar): + window = beat_strength[phase::beats_per_bar] + score = float(np.mean(window)) if len(window) else -np.inf + if score > best_score: + best_score, best_phase = score, phase + return [float(bt) for i, bt in enumerate(beat_times) if (i - best_phase) % beats_per_bar == 0] class TemporalAnalyzer: @@ -95,9 +125,10 @@ def analyze(self, audio_path: str | Path) -> TemporalFeatures: # Convert frame indices to time (seconds) beat_times: NDArray[np.floating[Any]] = librosa.frames_to_time(beat_frames, sr=sr) - # Extract downbeats (simple approximation: every 4th beat) - # A real model might use madmom or complex DBNs for precise downbeats - downbeat_times = [float(bt) for i, bt in enumerate(beat_times) if i % 4 == 0] + # Place downbeats on the strongest-onset bar phase (looks at the audio, + # not a blind "every 4th beat from index 0"). + onset_env = librosa.onset.onset_strength(y=y_array, sr=sr) + downbeat_times = _estimate_downbeats(onset_env, beat_frames, beat_times) bpm_val = float(tempo[0]) if isinstance(tempo, np.ndarray) else float(tempo) diff --git a/services/analysis-engine/tests/test_temporal.py b/services/analysis-engine/tests/test_temporal.py index c3a673c2..6ce90ae1 100644 --- a/services/analysis-engine/tests/test_temporal.py +++ b/services/analysis-engine/tests/test_temporal.py @@ -9,6 +9,7 @@ import soundfile as sf # type: ignore from bandscope_analysis.temporal import TemporalAnalyzer +from bandscope_analysis.temporal.analyzer import _estimate_downbeats @pytest.fixture @@ -200,3 +201,27 @@ def fake_load(*args: object, **kwargs: object) -> tuple[np.ndarray, int]: features = TemporalAnalyzer().analyze(test_wav) assert features["bpm"] == 120.0 + + +def test_estimate_downbeats_picks_strongest_onset_phase() -> None: + """Downbeats land on the bar phase with the most onset energy, not index 0.""" + onset = np.full(200, 1.0) + beat_frames = np.arange(16) * 10 + beat_times = beat_frames * 0.1 + # Accent phase 2 (beats 2, 6, 10, 14) — the old "every 4th from 0" would miss this. + for i in range(2, 16, 4): + onset[beat_frames[i]] = 10.0 + downbeats = _estimate_downbeats(onset, beat_frames, beat_times) + assert downbeats == [float(beat_times[i]) for i in range(2, 16, 4)] + + +def test_estimate_downbeats_empty() -> None: + """No beats yields no downbeats.""" + empty = np.array([]) + assert _estimate_downbeats(empty, empty, empty) == [] + + +def test_estimate_downbeats_too_few_beats_returns_first() -> None: + """Fewer beats than a bar falls back to the first beat as the downbeat.""" + onset = np.ones(50) + assert _estimate_downbeats(onset, np.array([0, 10]), np.array([0.0, 0.5])) == [0.0]