trav
10ced778af
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.
192 lines
7.8 KiB
Python
192 lines
7.8 KiB
Python
"""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)
|