diff --git a/CLAUDE.md b/CLAUDE.md index 7cf3323..252cbd4 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -256,6 +256,18 @@ persistence) → GUI (Qt widgets that read the manager and connect to its signal it never belongs in git). Audio under ~3 s has no fingerprint at all ("Empty fingerprint"), which is a reported failure, not a crash. +- **`lintunes/bpm_detect.py`** — right-click → Calculate BPM (Round 73). + Spectral-flux onset envelope → autocorrelation → every tempo 60–200 scored + over four beat multiples, times a log-Gaussian prior at 120 BPM (the + half/double-tempo tiebreak), then `fold_bpm` into 70–180. Plain numpy plus + the `ffmpeg` CLI (decodes the first 120 s to mono 11 kHz) — **no librosa**, + which would drag numba/scipy/scikit-learn in for one function. `BpmWorker` + is ExportWorker's shape on the shared status-bar widgets (`_busy_worker`), + reports each result as it lands so a cancel keeps them, and the batch is + applied by `LibraryManager.set_tracks_bpm` as **one** undoable command, + overwriting existing BPMs (trav's call). The transport's tap button stays + for the songs it gets wrong. + - **`lintunes/filename_tags.py`** — the offline half of Identify Track, and the answer to its biggest limitation: **AcoustID only knows music somebody submitted**, so an underground/SoundCloud rip fingerprints perfectly and diff --git a/lintunes/__init__.py b/lintunes/__init__.py index fc877be..5b09605 100644 --- a/lintunes/__init__.py +++ b/lintunes/__init__.py @@ -1,3 +1,3 @@ """LinTunes — iTunes-style music library manager and player for Linux.""" -__version__ = "0.33.0" +__version__ = "0.34.0" diff --git a/lintunes/bpm_detect.py b/lintunes/bpm_detect.py new file mode 100644 index 0000000..17fdc3b --- /dev/null +++ b/lintunes/bpm_detect.py @@ -0,0 +1,191 @@ +"""Calculate BPM: tempo detection from the audio itself. + +The textbook method, in numpy: decode to mono, build an onset-strength +envelope (spectral flux — how much louder each frequency got since the last +frame), autocorrelate it, and score every tempo from 60 to 200 BPM by how +strongly the envelope repeats at one, two, three and four beats. A +log-Gaussian prior centred on 120 BPM breaks the tie between a tempo and its +half (librosa's trick). No librosa/aubio: numpy is already a dependency, and +librosa drags numba/scipy/scikit-learn behind it for this one function. + +Shaped like ``fingerprint.py``: pure numpy first (tested offline against +synthetic beats), then the ``ffmpeg`` subprocess — the CLI binary, detected +at runtime like ``fpcalc``, never a pip dependency — then ``BpmWorker`` on a +daemon thread. Only the first ``MAX_SECONDS`` are analysed: a song's tempo +is settled long before then, and a batch of a few hundred stays quick. + +Like any detector it can land on half or double the tempo a person would +tap; ``fold_bpm`` keeps the answer in a sensible range, and the transport's +tap button is still there for the ones it gets wrong. +""" +import subprocess +import threading + +import numpy as np +from PyQt6.QtCore import QObject, pyqtSignal + +SAMPLE_RATE = 11025 +MAX_SECONDS = 120 +WINDOW = 1024 +HOP = 128 +MIN_BPM, MAX_BPM = 60.0, 200.0 +PRIOR_BPM = 120.0 +PRIOR_OCTAVES = 1.0 +HARMONICS = 4 + + +def onset_envelope(samples: np.ndarray, sr: int = SAMPLE_RATE): + """Spectral-flux onset strength, one value per hop. Returns + ``(envelope, frames_per_second)``; the envelope is empty for audio + shorter than one window.""" + samples = np.asarray(samples, dtype=np.float32) + count = 1 + (len(samples) - WINDOW) // HOP if len(samples) >= WINDOW else 0 + if count < 2: + return np.zeros(0, dtype=np.float32), sr / HOP + window = np.hanning(WINDOW).astype(np.float32) + frames = np.lib.stride_tricks.sliding_window_view(samples, WINDOW)[::HOP] + parts = [] + previous = None + chunk = 2048 # frames at a time: ~10k frames x 513 bins at once is 40 MB + for start in range(0, count, chunk): + spectrum = np.abs(np.fft.rfft(frames[start:start + chunk] * window, + axis=1)) + level = np.log1p(100.0 * spectrum).astype(np.float32) + joined = level if previous is None else np.vstack([previous, level]) + rise = np.maximum(np.diff(joined, axis=0), 0.0).sum(axis=1) + parts.append(rise if previous is not None else np.r_[0.0, rise]) + previous = level[-1:] + return np.concatenate(parts).astype(np.float32), sr / HOP + + +def _autocorrelation(envelope: np.ndarray) -> np.ndarray: + """Biased autocorrelation (via FFT) of the mean-removed envelope, + normalised so lag 0 is 1. Biased on purpose: it decays with lag, which + leans a tempo-vs-half-tempo tie toward the faster, beat-level one.""" + env = envelope.astype(np.float64) + # Subtract a ~1 s moving average so slow loudness swells aren't "beats". + width = max(int(SAMPLE_RATE / HOP), 1) + env = env - np.convolve(env, np.ones(width) / width, mode="same") + env = np.maximum(env, 0.0) + env -= env.mean() + size = 1 << int(np.ceil(np.log2(2 * len(env)))) + spectrum = np.fft.rfft(env, size) + acf = np.fft.irfft(spectrum * np.conj(spectrum), size)[:len(env)] + return acf / acf[0] if acf[0] > 0 else np.zeros_like(acf) + + +def estimate_bpm(samples: np.ndarray, sr: int = SAMPLE_RATE) -> float | None: + """The tempo of `samples` in BPM, or None when nothing repeats (silence, + a drone, audio too short to hold four beats).""" + envelope, fps = onset_envelope(samples, sr) + # Four beats at the slowest tempo must fit, or there's nothing to score. + if (len(envelope) <= 60.0 * fps / MIN_BPM * HARMONICS + or not np.any(envelope > 0)): + return None + acf = _autocorrelation(envelope) + lags = np.arange(len(acf)) + candidates = np.arange(MIN_BPM, MAX_BPM + 0.05, 0.1) + beat_lags = 60.0 * fps / candidates + score = np.zeros_like(candidates) + for k in range(1, HARMONICS + 1): + score += np.interp(beat_lags * k, lags, acf) + score /= HARMONICS + if score.max() <= 0.05: + return None + prior = np.exp(-0.5 * (np.log2(candidates / PRIOR_BPM) + / PRIOR_OCTAVES) ** 2) + return float(candidates[np.argmax(score * prior)]) + + +def fold_bpm(bpm: float, low: float = 70.0, high: float = 180.0) -> int: + """Fold a half/double-tempo answer into [low, high] and round — Track.bpm + is a whole number, as in iTunes.""" + while bpm < low: + bpm *= 2 + while bpm > high: + bpm /= 2 + return int(round(bpm)) + + +def decode_mono(path: str, sr: int = SAMPLE_RATE, + max_seconds: int = MAX_SECONDS) -> np.ndarray: + """The first `max_seconds` of `path` as mono float32 at `sr`, decoded by + the ffmpeg CLI. Raises OSError when ffmpeg fails or yields nothing.""" + cmd = ["ffmpeg", "-nostdin", "-v", "error", "-i", str(path), + "-t", str(max_seconds), "-vn", "-ac", "1", "-ar", str(sr), + "-f", "f32le", "-"] + try: + result = subprocess.run(cmd, capture_output=True, timeout=120) + except (OSError, subprocess.TimeoutExpired) as e: + raise OSError(f"ffmpeg could not run: {e}") from e + if result.returncode != 0: + detail = result.stderr.decode(errors="replace").strip().splitlines() + raise OSError(detail[-1] if detail else "ffmpeg failed") + samples = np.frombuffer(result.stdout, dtype=np.float32) + if len(samples) == 0: + raise OSError("no audio") + return samples + + +def detect_file(path: str) -> int | None: + """The folded BPM of the file at `path`, or None if it has no beat. + Raises OSError when the file can't be decoded.""" + bpm = estimate_bpm(decode_mono(path)) + return fold_bpm(bpm) if bpm is not None else None + + +class BpmWorker(QObject): + """Runs Calculate BPM over a batch on a daemon thread (ExportWorker's + shape). Handed `(track_id, location, label)` snapshots so it never reads + the library off the GUI thread; reports each result as it lands, so a + cancel keeps everything already measured.""" + + progress = pyqtSignal(int, int, str) # done, total, label of the next + detected = pyqtSignal(int, int) # track_id, bpm + finished = pyqtSignal(dict) # {"done", "failed", "no_beat", + # "cancelled"} + + def __init__(self, items: list[tuple[int, str, str]], parent=None, + detect=None): + super().__init__(parent) + self._items = list(items) + self._detect = detect # None: detect_file, looked up when run + self._busy = False + self._cancel = threading.Event() + + def busy(self) -> bool: + return self._busy + + def cancel(self): + self._cancel.set() + + def start(self): + if self._busy: + return + self._busy = True + threading.Thread(target=self._run, daemon=True).start() + + def _run(self): + summary = {"done": 0, "failed": 0, "no_beat": 0, "cancelled": False} + total = len(self._items) + try: + for i, (track_id, location, label) in enumerate(self._items): + if self._cancel.is_set(): + summary["cancelled"] = True + break + self.progress.emit(i, total, label) + try: + bpm = (self._detect or detect_file)(location) + except Exception: + summary["failed"] += 1 + continue + if bpm is None: + summary["no_beat"] += 1 + continue + summary["done"] += 1 + self.detected.emit(track_id, bpm) + else: + self.progress.emit(total, total, "") + finally: + self._busy = False + self.finished.emit(summary) diff --git a/lintunes/gui/main_window.py b/lintunes/gui/main_window.py index 9441e68..87b0e62 100644 --- a/lintunes/gui/main_window.py +++ b/lintunes/gui/main_window.py @@ -10,7 +10,7 @@ from PyQt6.QtWidgets import ( from PyQt6.QtCore import Qt, QEvent, QObject, QTimer, pyqtSignal from PyQt6.QtGui import QAction, QKeySequence -from lintunes import device_sync, music_folder, theme, url_import +from lintunes import bpm_detect, device_sync, music_folder, theme, url_import from lintunes.andtunes import layout as andtunes_layout from lintunes.andtunes import install as andtunes_install from lintunes.andtunes import sync as andtunes_sync @@ -223,6 +223,9 @@ class MainWindow(QMainWindow): self._identify_running = False # Import from URL: also shares the status-bar progress widgets. self._url_worker = None + # Calculate BPM: also shares the status-bar progress widgets. + self._bpm_worker = None + self._bpm_results: dict[int, int] = {} self._url_target = None # url_import.ImportTarget, or None self._url_inserted = 0 # songs added to the target so far self._url_imported = 0 @@ -256,6 +259,7 @@ class MainWindow(QMainWindow): view.table.tracks_changed.connect(self._update_totals) view.table.download_art_requested.connect(self._download_album_art) view.table.identify_requested.connect(self._identify_tracks) + view.table.bpm_requested.connect(self._calculate_bpm) view.table.remove_from_library_requested.connect( lambda ids: self._delete_tracks(ids, delete_files=False)) view.table.delete_from_library_requested.connect( @@ -550,12 +554,17 @@ class MainWindow(QMainWindow): def _busy_worker(self): """The transfer currently owning the status-bar progress widgets.""" for worker in (self._andtunes_planner, self._andtunes_worker, - self._export_worker, self._url_worker): + self._export_worker, self._url_worker, + self._bpm_worker): if worker is not None and worker.busy(): return worker return None def _confirm_cancel_sync(self): + if self._bpm_worker is not None and self._bpm_worker.busy(): + # Nothing to lose: what's been measured is kept. + self._bpm_worker.cancel() + return if self._url_worker is not None and self._url_worker.busy(): self._confirm_cancel_url_import() return @@ -1586,6 +1595,70 @@ class MainWindow(QMainWindow): return self._enqueue_identify(tracks) + # ---- Calculate BPM ---- + + def _calculate_bpm(self, track_ids: list[int]): + """Measure the tempo of the selection on a worker thread, with the + status-bar progress bar; results land as one undoable edit.""" + if not exporter.web_support.ffmpeg_available(): + QMessageBox.warning( + self, "Calculate BPM", + "Calculating BPM needs ffmpeg, which is not installed.\n\n" + "Install your distro's “ffmpeg” package and try again.") + return + if self._busy_worker() is not None: + self.statusBar().showMessage( + "Wait for the current transfer to finish first", 6000) + return + items = [] + for tid in track_ids: + track = self._manager.library.tracks.get(tid) + if track is not None and track.location: + label = (f"{track.artist} — {track.name}" if track.artist + else track.name) or Path(track.location).name + items.append((tid, track.location, label)) + if not items: + return + self._bpm_results = {} + worker = bpm_detect.BpmWorker(items, self) + worker.progress.connect(self._on_bpm_progress) + worker.detected.connect(self._bpm_results.__setitem__) + worker.finished.connect(self._on_bpm_finished) + self._bpm_worker = worker + self._sync_label.setText("Calculating BPM…") + self._sync_progress.setRange(0, len(items)) + self._sync_progress.setValue(0) + self._sync_cancel.show() + self._sync_label.show() + self._sync_progress.show() + worker.start() + + def _on_bpm_progress(self, done: int, total: int, label: str): + self._sync_progress.setRange(0, total) + self._sync_progress.setValue(done) + if label: + short = label if len(label) <= 40 else label[:39] + "…" + where = f"{done + 1} of {total} · " if total > 1 else "" + self._sync_label.setText(f"BPM {where}{short}") + + def _on_bpm_finished(self, summary: dict): + self._hide_sync_widgets() + results, self._bpm_results = self._bpm_results, {} + self._manager.set_tracks_bpm(results) + if len(results) == 1 and not summary["cancelled"]: + msg = f"BPM: {next(iter(results.values()))}" + else: + msg = f"BPM set for {len(results)} track(s)" + if summary["cancelled"]: + msg = "Calculate BPM stopped — " + msg[0].lower() + msg[1:] + if summary["no_beat"]: + msg += f" · no steady beat in {summary['no_beat']}" + if summary["failed"]: + msg += f" · {summary['failed']} couldn't be read" + if not summary["cancelled"]: + sounds.play_done(self._prefs) + self.statusBar().showMessage(msg, 10000) + def _identify_setup_problem(self) -> str: """Why an AcoustID lookup can't run on this machine, or ''.""" if not fpcalc_available(): diff --git a/lintunes/gui/track_table.py b/lintunes/gui/track_table.py index 61e60ab..c9410d8 100644 --- a/lintunes/gui/track_table.py +++ b/lintunes/gui/track_table.py @@ -553,6 +553,7 @@ class TrackTableView(QTableView): rating_edited = pyqtSignal(int, int) # track_id, new rating 0-100 download_art_requested = pyqtSignal(list) # selected track ids identify_requested = pyqtSignal(list) # selected track ids + bpm_requested = pyqtSignal(list) # selected track ids remove_from_library_requested = pyqtSignal(list) # track ids, file kept delete_from_library_requested = pyqtSignal(list) # track ids, file trashed @@ -860,7 +861,7 @@ class TrackTableView(QTableView): locations = self.selected_locations() reveal_action = copy_path_action = download_art_action = None - identify_action = None + identify_action = bpm_action = None if locations: menu.addSeparator() reveal_action = menu.addAction("Reveal in File Browser") @@ -868,6 +869,7 @@ class TrackTableView(QTableView): download_art_action = menu.addAction("Download Album Art…") # Fingerprinting reads the audio, so this needs a file on disk. identify_action = menu.addAction("Identify Track…") + bpm_action = menu.addAction("Calculate BPM") # Single track only: one click shouldn't open a browser tab per song. # Needs no file, so it's offered for a dangling location too. @@ -921,6 +923,8 @@ class TrackTableView(QTableView): self.download_art_requested.emit(self.selected_track_ids()) elif identify_action is not None and chosen is identify_action: self.identify_requested.emit(self.selected_track_ids()) + elif bpm_action is not None and chosen is bpm_action: + self.bpm_requested.emit(self.selected_track_ids()) elif youtube_action is not None and chosen is youtube_action: track = self.model_.track_at(self.selected_source_rows()[0]) QDesktopServices.openUrl(youtube_search_url(track)) diff --git a/lintunes/library_manager.py b/lintunes/library_manager.py index 3a78d10..b793eab 100644 --- a/lintunes/library_manager.py +++ b/lintunes/library_manager.py @@ -627,6 +627,29 @@ class LibraryManager(QObject): redo=lambda: [self._revert_track_fields(tid, new) for tid, new, _old in changes])) + def set_tracks_bpm(self, values: dict[int, int]): + """Calculate BPM's results: a different bpm per track, tag written to + each file, recorded as ONE undoable command.""" + changes = [] # (track_id, new_fields, old_fields) + with timed("set_tracks_bpm (%d tracks)", len(values)): + for track_id, bpm in values.items(): + track = self.library.tracks.get(track_id) + if not track or track.bpm == bpm: + continue + changed = {"bpm": bpm} + if not self._write_track_tags(track, changed): + continue + changes.append((track_id, changed, {"bpm": track.bpm})) + self._apply_track_fields(track_id, changed) + if not changes: + return + self.undo_stack.push(Command( + "Calculate BPM", + undo=lambda: [self._revert_track_fields(tid, old) + for tid, _new, old in changes], + redo=lambda: [self._revert_track_fields(tid, new) + for tid, new, _old in changes])) + def _write_track_tags(self, track, field_map: dict) -> bool: # Library-only fields (rating, size, start/stop times) have no tag # representation; don't rewrite the audio file for them. diff --git a/tests/test_round73.py b/tests/test_round73.py new file mode 100644 index 0000000..e03fe55 --- /dev/null +++ b/tests/test_round73.py @@ -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("