v0.34.0: Calculate BPM from the audio, on the right-click menu

Spectral-flux onsets + autocorrelation in numpy, decoded by the ffmpeg CLI; no new dependency. Runs on a worker with the status-bar progress bar, for one track or many, and the batch is one Ctrl+Z. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-02 14:55:35 -07:00
parent 4c0aaae463
commit b3d6c01774
7 changed files with 504 additions and 4 deletions
@@ -0,0 +1,197 @@
+"""Round 73: Calculate BPM.
+
+Right-click → Calculate BPM measures the tempo from the audio itself
+(``bpm_detect``: spectral-flux onsets, autocorrelation, a 120 BPM prior),
+with the status-bar progress bar, for one track or many. Results overwrite
+whatever BPM a track had, as one Ctrl+Z.
+
+The beats here are synthetic — a kick on every beat, optionally hats on the
+off-beats and noise — so the right answer is known exactly.
+"""
+import shutil
+import time
+import wave
+
+import numpy as np
+import pytest
+from PyQt6.QtCore import QCoreApplication
+
+from lintunes import bpm_detect
+from lintunes.bpm_detect import (
+    SAMPLE_RATE, BpmWorker, decode_mono, detect_file, estimate_bpm, fold_bpm,
+)
+from lintunes.library_manager import LibraryManager
+from lintunes.models import Library, Track
+
+
+def _beat(bpm, seconds=30, hats=False, noise=0.0, sr=SAMPLE_RATE):
+    rng = np.random.default_rng(7)
+    n = int(seconds * sr)
+    out = np.zeros(n, np.float32)
+    t = np.arange(int(0.08 * sr)) / sr
+    kick = (np.sin(2 * np.pi * 60 * t) * np.exp(-t * 30)).astype(np.float32)
+    hat_len = int(0.03 * sr)
+    hat = (rng.standard_normal(hat_len)
+           * np.exp(-np.arange(hat_len) / sr * 150) * 0.3).astype(np.float32)
+    period = 60.0 / bpm
+    beat = 0
+    while (beat + 1) * period < seconds:
+        i = int(beat * period * sr)
+        out[i:i + len(kick)] += kick
+        if hats:
+            j = int((beat + 0.5) * period * sr)
+            out[j:j + hat_len] += hat
+        beat += 1
+    return out + noise * rng.standard_normal(n).astype(np.float32)
+
+
+def _pump(until, timeout=5.0):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        QCoreApplication.processEvents()
+        if until():
+            return True
+        time.sleep(0.01)
+    return False
+
+
+class TestEstimate:
+    @pytest.mark.parametrize("bpm", [72, 90, 100, 120, 128, 140, 155, 174])
+    def test_kick_on_every_beat(self, bpm):
+        assert abs(fold_bpm(estimate_bpm(_beat(bpm))) - bpm) <= 1
+
+    @pytest.mark.parametrize("bpm", [85, 120, 174])
+    def test_offbeat_hats_and_noise_dont_double_it(self, bpm):
+        found = estimate_bpm(_beat(bpm, hats=True, noise=0.05))
+        assert abs(fold_bpm(found) - bpm) <= 1
+
+    def test_silence_and_noise_have_no_beat(self):
+        assert estimate_bpm(np.zeros(SAMPLE_RATE * 10, np.float32)) is None
+        noise = np.random.default_rng(1).standard_normal(SAMPLE_RATE * 10)
+        assert estimate_bpm(noise.astype(np.float32)) is None
+
+    def test_too_short_for_four_beats(self):
+        assert estimate_bpm(_beat(120, seconds=1)) is None
+        assert estimate_bpm(np.zeros(10, np.float32)) is None
+
+    def test_fold(self):
+        assert fold_bpm(60) == 120
+        assert fold_bpm(240) == 120
+        assert fold_bpm(127.6) == 128
+        assert fold_bpm(174) == 174
+
+
+@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="needs ffmpeg")
+class TestDecode:
+    def test_wav_round_trip(self, tmp_path):
+        path = tmp_path / "beat.wav"
+        samples = _beat(128, seconds=20, sr=44100)
+        pcm = (np.clip(samples, -1, 1) * 32000).astype("<i2")
+        with wave.open(str(path), "wb") as f:
+            f.setnchannels(1)
+            f.setsampwidth(2)
+            f.setframerate(44100)
+            f.writeframes(pcm.tobytes())
+        decoded = decode_mono(str(path))
+        assert abs(len(decoded) - 20 * SAMPLE_RATE) < SAMPLE_RATE // 10
+        assert detect_file(str(path)) == 128
+
+    def test_unreadable_file_raises_oserror(self, tmp_path):
+        bad = tmp_path / "bad.mp3"
+        bad.write_bytes(b"not audio at all")
+        with pytest.raises(OSError):
+            decode_mono(str(bad))
+
+
+class TestWorker:
+    def test_reports_each_result_and_a_summary(self, qapp):
+        answers = {"a": 120, "b": None}
+
+        def detect(location):
+            if location == "c":
+                raise OSError("nope")
+            return answers[location]
+
+        worker = BpmWorker([(1, "a", "A"), (2, "b", "B"), (3, "c", "C")],
+                           detect=detect)
+        found, summaries = [], []
+        worker.detected.connect(lambda tid, bpm: found.append((tid, bpm)))
+        worker.finished.connect(summaries.append)
+        worker.start()
+        assert _pump(lambda: summaries)
+        assert found == [(1, 120)]
+        assert summaries[0] == {"done": 1, "failed": 1, "no_beat": 1,
+                                "cancelled": False}
+        assert not worker.busy()
+
+    def test_cancel_keeps_what_was_measured(self, qapp):
+        worker = BpmWorker([(1, "a", "A"), (2, "b", "B")])
+
+        def detect(location):
+            worker.cancel()
+            return 100
+
+        worker._detect = detect
+        found, summaries = [], []
+        worker.detected.connect(lambda tid, bpm: found.append(tid))
+        worker.finished.connect(summaries.append)
+        worker.start()
+        assert _pump(lambda: summaries)
+        assert found == [1]
+        assert summaries[0]["cancelled"]
+
+
+def _manager(tmp_path, bpms):
+    library = Library()
+    for tid, bpm in bpms.items():
+        library.tracks[tid] = Track(track_id=tid, name=f"Song {tid}", bpm=bpm)
+    return LibraryManager(library, tmp_path / "data")
+
+
+class TestSetTracksBpm:
+    def test_per_track_values_overwrite_and_undo_as_one(self, qapp, tmp_path):
+        manager = _manager(tmp_path, {1: 0, 2: 95, 3: 140})
+        manager.set_tracks_bpm({1: 120, 2: 128, 3: 140})
+        tracks = manager.library.tracks
+        assert [tracks[i].bpm for i in (1, 2, 3)] == [120, 128, 140]
+        manager.undo_stack.undo()
+        assert [tracks[i].bpm for i in (1, 2, 3)] == [0, 95, 140]
+        manager.undo_stack.redo()
+        assert [tracks[i].bpm for i in (1, 2, 3)] == [120, 128, 140]
+
+    def test_nothing_changed_records_nothing(self, qapp, tmp_path):
+        manager = _manager(tmp_path, {1: 120})
+        manager.set_tracks_bpm({1: 120, 99: 100})
+        assert not manager.undo_stack.can_undo()
+
+
+class TestWindow:
+    def test_menu_signal_runs_the_batch_through_the_status_bar(
+            self, qapp, tmp_path, monkeypatch):
+        from lintunes.gui import sounds
+        from lintunes.gui.main_window import MainWindow
+        from lintunes.preferences import Preferences
+
+        monkeypatch.setattr(bpm_detect, "detect_file",
+                            lambda location: {"/a.mp3": 90}.get(location))
+        monkeypatch.setattr(
+            "lintunes.export.web_support.ffmpeg_available", lambda: True)
+        monkeypatch.setattr(sounds, "play_done", lambda prefs: None)
+        manager = _manager(tmp_path, {1: 0, 2: 0, 3: 0})
+        manager.library.tracks[1].location = "/a.mp3"
+        manager.library.tracks[2].location = "/b.mp3"
+        # Nothing on disk to tag; the library edit is what's under test.
+        monkeypatch.setattr(manager, "_write_track_tags",
+                            lambda track, fields: True)
+        window = MainWindow(manager, Preferences(tmp_path / "data"))
+        try:
+            window._library_view.table.bpm_requested.emit([1, 2, 3])
+            assert window._bpm_worker is not None
+            assert window._sync_progress.maximum() == 2   # 3 has no file
+            assert _pump(lambda: not window._bpm_worker.busy()
+                         and window._sync_progress.isHidden())
+            tracks = manager.library.tracks
+            assert (tracks[1].bpm, tracks[2].bpm, tracks[3].bpm) == (90, 0, 0)
+            assert "no steady beat in 1" in window.statusBar().currentMessage()
+        finally:
+            window.close()