The background worker now runs a real librosa analyzer (tempo, beat grid, per-second energy/valence curves) instead of only the null baseline, kept behind build_analyzer() so a plain checkout without the analysis extra still runs fine. The web player reads that per-track analysis and drives a new animated "Ambience" background (bubbles, colour, current) that reacts to the beat and the mood curve as the track plays, plus a debug overlay for tuning it. Also adds a one-off script to backfill podcast cover art from iTunes.
319 lines
15 KiB
Python
319 lines
15 KiB
Python
"""A concrete :class:`~musicmouse.library.analysis.Analyzer` built on librosa.
|
|
|
|
Only reached through :func:`musicmouse.library.analysis.build_analyzer`, so nothing
|
|
else in the app ever imports librosa or numpy: a device without the ``analysis`` extra
|
|
installed never executes this module at all.
|
|
|
|
Every scalar here is a signal-processing proxy, not a measurement of how a track
|
|
actually feels - ``valence`` most of all, see its section below. Good enough to drive
|
|
an ambient background; not a music information retrieval research result.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import math
|
|
import warnings
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
|
|
import librosa
|
|
import numpy as np
|
|
|
|
from musicmouse.library.analysis import ANALYZER_VERSION, BeatGrid, TrackAnalysis, TrackCurves
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
__all__ = ["LibrosaAnalyzer"]
|
|
|
|
#: librosa's own default. Ample for everything below - the highest band that matters,
|
|
#: the brightness ceiling, sits well under this rate's 11025 Hz Nyquist frequency.
|
|
_SAMPLE_RATE = 22050
|
|
_HOP_LENGTH = 512
|
|
|
|
#: `energy`: the 80th-percentile RMS frame, in dB, mapped floor..ceil to 0..1. -32 dB is
|
|
#: a quiet passage, -8 dB is a hot, compressed master. A percentile rather than the mean
|
|
#: or max so a quiet intro or a gap between phrases doesn't drag a loud track down, and
|
|
#: one clipped peak doesn't blow it out.
|
|
_ENERGY_DB_FLOOR = -32.0
|
|
_ENERGY_DB_CEIL = -8.0
|
|
|
|
#: `brightness`: the median spectral centroid (robust to a single loud transient), in
|
|
#: Hz, log-mapped floor..ceil to 0..1. 300 Hz is a dark, bass/vocal-heavy mix; 4000 Hz is
|
|
#: bright, sparkly production.
|
|
_BRIGHTNESS_HZ_FLOOR = 300.0
|
|
_BRIGHTNESS_HZ_CEIL = 4000.0
|
|
|
|
#: `valence`'s tempo term: 60 BPM reads as a lullaby, 150 BPM as a romp.
|
|
_TEMPO_BPM_FLOOR = 60.0
|
|
_TEMPO_BPM_CEIL = 150.0
|
|
|
|
#: Key is a global property of a track; the middle minute is its most representative
|
|
#: one and this keeps the most expensive feature (chroma) off the tail of a long track.
|
|
_CHROMA_EXCERPT_SECONDS = 120.0
|
|
|
|
#: Seconds per curve sample. Fixed at analysis time, not user-facing - changing this
|
|
#: needs a re-analysis and an ANALYZER_VERSION bump, unlike the frontend's own
|
|
#: unrelated "curve sample interval" debug slider, which just smooths already-fetched
|
|
#: samples for live preview.
|
|
_ANALYSIS_HOP_SECONDS = 1.0
|
|
|
|
#: `drive`'s blend of onset activity (steadier, measured 33-55% relative spread across
|
|
#: a track) and local pulse strength (more dynamic but spikier, 59-189%) - weighted
|
|
#: toward activity so a busy chorus reads clearly without the current twitching on
|
|
#: every transient.
|
|
_DRIVE_ACTIVITY_WEIGHT = 0.6
|
|
_DRIVE_PLP_WEIGHT = 0.4
|
|
|
|
#: Krumhansl-Kessler key profiles (Krumhansl & Kessler 1982), starting from C. Every
|
|
#: other key is scored by rotating these twelve weights, not by transposing the audio.
|
|
_MAJOR_PROFILE = np.array([6.35, 2.23, 3.48, 2.33, 4.38, 4.09, 2.52, 5.19, 2.39, 3.66, 2.29, 2.88])
|
|
_MINOR_PROFILE = np.array([6.33, 2.68, 3.52, 5.38, 2.60, 3.53, 2.54, 4.75, 3.98, 2.69, 3.34, 3.17])
|
|
|
|
|
|
def _clip01(value: float) -> float:
|
|
return max(0.0, min(1.0, value))
|
|
|
|
|
|
def _normalize(value: float, floor: float, ceil: float) -> float:
|
|
"""Linear map ``floor..ceil`` to ``0..1``, clipped at both ends."""
|
|
return _clip01((value - floor) / (ceil - floor))
|
|
|
|
|
|
def _center_excerpt(y: np.ndarray, sr: float, seconds: float) -> np.ndarray:
|
|
max_samples = int(seconds * sr)
|
|
if y.size <= max_samples:
|
|
return y
|
|
start = (y.size - max_samples) // 2
|
|
return y[start : start + max_samples]
|
|
|
|
|
|
def _frames_per_window(sr: float, hop_length: int, window_seconds: float) -> int:
|
|
return max(1, round(window_seconds * sr / hop_length))
|
|
|
|
|
|
def _windowed(
|
|
values: np.ndarray, frames_per_window: int, reduce: Callable[[np.ndarray], float]
|
|
) -> np.ndarray:
|
|
"""Buckets an already-computed per-frame array (RMS, spectral centroid, onset
|
|
envelope, PLP - anything on the STFT hop grid) into `frames_per_window`-wide
|
|
windows, applying `reduce` to each. The last window is short rather than dropped,
|
|
so a track's tail is never silently excluded from its own curve."""
|
|
n = max(1, math.ceil(values.size / frames_per_window))
|
|
out = np.empty(n)
|
|
for i in range(n):
|
|
w = values[i * frames_per_window : (i + 1) * frames_per_window]
|
|
out[i] = reduce(w if w.size else values[-1:])
|
|
return out
|
|
|
|
|
|
def _energy_window(w: np.ndarray) -> float:
|
|
"""The same 80th-percentile+dB statistic as the whole-track `energy` scalar,
|
|
applied to one window."""
|
|
db = librosa.amplitude_to_db(np.array([np.percentile(w, 80)]), ref=1.0)[0]
|
|
return _normalize(float(db), _ENERGY_DB_FLOOR, _ENERGY_DB_CEIL)
|
|
|
|
|
|
def _brightness_window(w: np.ndarray) -> float:
|
|
"""The same median+log statistic as the whole-track `brightness` scalar, applied
|
|
to one window."""
|
|
hz = max(float(np.median(w)), 1.0) # guard log2(0)
|
|
floor, ceil = math.log2(_BRIGHTNESS_HZ_FLOOR), math.log2(_BRIGHTNESS_HZ_CEIL)
|
|
return _normalize(math.log2(hz), floor, ceil)
|
|
|
|
|
|
def _norm95(values: np.ndarray) -> np.ndarray:
|
|
"""Scale so the array's 95th percentile maps to 1.0 - a robust max that one loud
|
|
transient can't blow out, the same reasoning as the beat grid's strengths."""
|
|
ceiling = float(np.percentile(values, 95))
|
|
return values / ceiling if ceiling > 0 else np.zeros_like(values)
|
|
|
|
|
|
def _majorness(chroma_mean: np.ndarray) -> float:
|
|
"""Best major key-profile correlation minus best minor one, over all 12 rotations.
|
|
|
|
Positive means the track's pitch-class distribution fits a major key better than
|
|
any minor one; negative the other way round. `np.corrcoef` is undefined for a
|
|
perfectly flat chroma vector (silence, pure noise) - `nan_to_num` turns that into
|
|
"no signal either way" rather than raising.
|
|
"""
|
|
|
|
def best_fit(profile: np.ndarray) -> float:
|
|
return max(
|
|
float(np.nan_to_num(np.corrcoef(chroma_mean, np.roll(profile, i))[0, 1]))
|
|
for i in range(12)
|
|
)
|
|
|
|
return best_fit(_MAJOR_PROFILE) - best_fit(_MINOR_PROFILE)
|
|
|
|
|
|
class LibrosaAnalyzer:
|
|
"""Turns one music file into :class:`TrackAnalysis` plus an optional beat grid
|
|
and :class:`TrackCurves`."""
|
|
|
|
version = ANALYZER_VERSION
|
|
|
|
def analyze(self, path: Path) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]:
|
|
"""Never raises: a file this can't make sense of is analyzed as "nothing".
|
|
|
|
A corrupt file, a DRM'd one, or a zero-length one must not abort a batch of
|
|
hundreds - the fallback records `version` so :meth:`MusicLibrary.analyze_pending`
|
|
does not retry it forever, while every scalar stays `None` so the frontend falls
|
|
back to the un-analyzed baseline look rather than something half-computed.
|
|
"""
|
|
try:
|
|
y, sr = librosa.load(path, sr=_SAMPLE_RATE, mono=True)
|
|
except Exception:
|
|
_log.warning("Could not decode %s; leaving it unanalyzed", path, exc_info=True)
|
|
return TrackAnalysis(version=self.version), None, None
|
|
|
|
if y.size == 0:
|
|
_log.warning("%s decoded to no audio; leaving it unanalyzed", path)
|
|
return TrackAnalysis(version=self.version), None, None
|
|
|
|
try:
|
|
return self._analyze(y, sr)
|
|
except Exception:
|
|
_log.warning("Analysis failed for %s; leaving it unanalyzed", path, exc_info=True)
|
|
return TrackAnalysis(version=self.version), None, None
|
|
|
|
def _analyze(
|
|
self, y: np.ndarray, sr: float
|
|
) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]:
|
|
# One onset envelope feeds tempo, the beat grid's strengths, `pulse` and
|
|
# `drive` - the single most expensive shared computation, so it is done once.
|
|
onset_env = librosa.onset.onset_strength(y=y, sr=sr, hop_length=_HOP_LENGTH)
|
|
tempo_raw, beat_frames = librosa.beat.beat_track(
|
|
onset_envelope=onset_env, sr=sr, hop_length=_HOP_LENGTH
|
|
)
|
|
tempo_bpm = float(np.atleast_1d(tempo_raw)[0])
|
|
grid = self._beat_grid(onset_env, beat_frames, sr)
|
|
|
|
frames_per_window = _frames_per_window(sr, _HOP_LENGTH, _ANALYSIS_HOP_SECONDS)
|
|
|
|
rms = librosa.feature.rms(y=y, hop_length=_HOP_LENGTH)[0]
|
|
energy_curve = _windowed(rms, frames_per_window, _energy_window)
|
|
energy = float(np.mean(energy_curve))
|
|
|
|
centroid = librosa.feature.spectral_centroid(y=y, sr=sr, hop_length=_HOP_LENGTH)[0]
|
|
# Scalar `brightness` stays the whole-track median exactly as before this
|
|
# refactor - it no longer drives anything on the frontend (only its curve
|
|
# feeds `valence` below), so its own meaning is deliberately left unchanged.
|
|
brightness = _brightness_window(centroid)
|
|
brightness_curve = _windowed(centroid, frames_per_window, _brightness_window)
|
|
|
|
majorness_norm, tempo_norm = self._valence_terms(y, sr, tempo_bpm)
|
|
valence_curve = np.clip(
|
|
0.5 * majorness_norm + 0.3 * brightness_curve + 0.2 * tempo_norm, 0.0, 1.0
|
|
)
|
|
valence = float(np.mean(valence_curve))
|
|
|
|
pulse = self._pulse(onset_env, sr, tempo_bpm)
|
|
drive_curve = self._drive_curve(onset_env, sr, frames_per_window)
|
|
|
|
curves = TrackCurves(
|
|
hop_seconds=_ANALYSIS_HOP_SECONDS,
|
|
energy=tuple(float(v) for v in energy_curve),
|
|
valence=tuple(float(v) for v in valence_curve),
|
|
drive=tuple(float(v) for v in drive_curve),
|
|
)
|
|
|
|
analysis = TrackAnalysis(
|
|
version=self.version,
|
|
tempo=tempo_bpm,
|
|
energy=energy,
|
|
valence=valence,
|
|
brightness=brightness,
|
|
pulse=pulse,
|
|
beats=grid is not None,
|
|
)
|
|
return analysis, grid, curves
|
|
|
|
def _beat_grid(
|
|
self, onset_env: np.ndarray, beat_frames: np.ndarray, sr: float
|
|
) -> BeatGrid | None:
|
|
if beat_frames.size == 0:
|
|
return None
|
|
times = librosa.frames_to_time(beat_frames, sr=sr, hop_length=_HOP_LENGTH)
|
|
raw_strengths = onset_env[np.clip(beat_frames, 0, onset_env.size - 1)]
|
|
# The 95th percentile rather than the max, so one loud crash does not flatten
|
|
# every other beat's strength toward zero.
|
|
scale = float(np.percentile(raw_strengths, 95))
|
|
strengths = raw_strengths / scale if scale > 0 else np.zeros_like(raw_strengths)
|
|
return BeatGrid(
|
|
tuple(float(t) for t in times),
|
|
tuple(_clip01(float(s)) for s in strengths),
|
|
)
|
|
|
|
def _valence_terms(self, y: np.ndarray, sr: float, tempo_bpm: float) -> tuple[float, float]:
|
|
"""majorness_norm, tempo_norm - the two whole-track-constant terms of the
|
|
valence formula. See the module docstring for why valence is a heuristic, not
|
|
a measurement.
|
|
|
|
Weighted for a *children's* library specifically: mode (major/minor) is the
|
|
strongest and most legible cue in this repertoire - a minor-key children's song
|
|
is almost always deliberately sad or spooky, unlike in pop where mode is a much
|
|
weaker signal. Tempo adds romp-vs-lullaby. Both stay whole-track constants:
|
|
key is a global property of a track, and (unlike brightness) chroma-based key
|
|
detection is too expensive and too noisy over a short window to be worth
|
|
computing per second for what is now only a secondary, "nudge" contribution to
|
|
color - see `TrackCurves.valence`.
|
|
"""
|
|
excerpt = _center_excerpt(y, sr, _CHROMA_EXCERPT_SECONDS)
|
|
chroma = librosa.feature.chroma_cqt(y=excerpt, sr=sr, hop_length=_HOP_LENGTH)
|
|
majorness = _majorness(chroma.mean(axis=1))
|
|
majorness_norm = _clip01((majorness + 1.0) / 2.0)
|
|
tempo_norm = _normalize(tempo_bpm, _TEMPO_BPM_FLOOR, _TEMPO_BPM_CEIL)
|
|
return majorness_norm, tempo_norm
|
|
|
|
def _drive_curve(self, onset_env: np.ndarray, sr: float, frames_per_window: int) -> np.ndarray:
|
|
"""Rhythmic intensity per window, 0..1, normalised *within the track*.
|
|
|
|
This - not tempo - is what the frontend's water current breathes with while a
|
|
track plays. Measured against a real library, local tempo is flat to within a
|
|
few percent inside a track (recorded children's music is played to a click);
|
|
the residual "variation" a per-second tempo curve would show is mostly
|
|
estimator noise, including occasional octave errors. Onset activity and local
|
|
pulse strength (PLP) both genuinely vary within a track (33-55% and 59-189%
|
|
relative spread respectively) and don't carry that failure mode.
|
|
|
|
Per-track normalisation (each component scaled by its own 95th percentile) is
|
|
deliberate: absolute "how energetic is this song" is already carried by
|
|
whole-track `tempo` (the current's base magnitude) and by `energy` (colour).
|
|
This curve is for relative shape within the track - the intro is calmer than
|
|
the chorus - so every track uses its own full 0..1 range rather than a
|
|
uniformly quiet song sitting flat near zero throughout.
|
|
"""
|
|
activity = _windowed(onset_env, frames_per_window, lambda w: float(np.mean(w)))
|
|
plp = librosa.beat.plp(onset_envelope=onset_env, sr=sr, hop_length=_HOP_LENGTH)
|
|
pulse_curve = _windowed(plp, frames_per_window, lambda w: float(np.mean(w)))
|
|
activity_term = _DRIVE_ACTIVITY_WEIGHT * _norm95(activity)
|
|
pulse_term = _DRIVE_PLP_WEIGHT * _norm95(pulse_curve)
|
|
return np.clip(activity_term + pulse_term, 0.0, 1.0)
|
|
|
|
def _pulse(self, onset_env: np.ndarray, sr: float, tempo_bpm: float) -> float:
|
|
"""0..1 confidence that `tempo` is an audible, steady beat.
|
|
|
|
The onset envelope's autocorrelation at the beat period, relative to its value
|
|
at lag zero: a track that truly pulses at `tempo` has a strong echo of itself
|
|
one beat later; free-tempo or spoken-word material does not, even though
|
|
`beat_track` always returns *some* grid for it.
|
|
"""
|
|
if tempo_bpm <= 0 or onset_env.size < 2:
|
|
return 0.0
|
|
period_frames = round((60.0 / tempo_bpm) * sr / _HOP_LENGTH)
|
|
if not 0 < period_frames < onset_env.size:
|
|
return 0.0
|
|
with warnings.catch_warnings():
|
|
# A known-spurious warning from numba's complex-magnitude dufunc under
|
|
# this numba/numpy/librosa combination (confirmed: the input here has no
|
|
# NaN/Inf, and the result is a normal finite float) - narrowly silenced
|
|
# rather than left to spam the log once per track analyzed.
|
|
warnings.filterwarnings(
|
|
"ignore", message="invalid value encountered in cast", category=RuntimeWarning
|
|
)
|
|
ac = librosa.autocorrelate(onset_env)
|
|
if ac[0] <= 0:
|
|
return 0.0
|
|
return _clip01(float(ac[period_frames] / ac[0]))
|