martin/musicmouse
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add librosa beat/mood analysis and an Ambience background driven by it

Browse Source 
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.
This commit is contained in:
Martin Bauer
2026-09-10 22:42:51 +02:00
parent 8aed3b022b
commit 2e0e6ad199
24 changed files with 2789 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
python-backend/musicmouse/__main__.py
View File

@@ -17,7 +17,7 @@ import asyncio
import contextlib
import logging
import sys
from collections.abc import Coroutine
from collections.abc import Awaitable, Callable, Coroutine
from pathlib import Path
from typing import Any
@@ -31,6 +31,7 @@ from musicmouse.devices.null_transport import NullTransport
from musicmouse.devices.player import Player, VlcPlayer
from musicmouse.devices.serial_link import SerialLink
from musicmouse.library import MusicLibrary
from musicmouse.library.analysis import build_analyzer
from musicmouse.reactions import register_all
from musicmouse.services.base import Service
from musicmouse.services.mqtt import MqttService, build_entities
@@ -43,9 +44,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(
prog="musicmouse", description="Host backend for the MusicMouse RFID music player."
)
parser.add_argument(
"-c", "--config", type=Path, required=True, help="path to config.yml"
)
parser.add_argument("-c", "--config", type=Path, required=True, help="path to config.yml")
parser.add_argument(
"-s",
"--simulate",
@@ -127,9 +126,7 @@ async def run_real(config: Config, config_path: Path, *, hardware: bool = True)
clock=clock,
)
mouse = MusicMouseDevice(bus, link, config.tag_map, port=general.serial_port)
link.attach(
mouse.feed, on_connect=mouse.on_connected, on_disconnect=mouse.on_disconnected
)
link.attach(mouse.feed, on_connect=mouse.on_connected, on_disconnect=mouse.on_disconnected)
else:
mouse = MusicMouseDevice(bus, NullTransport(), config.tag_map, port="none")
@@ -137,9 +134,7 @@ async def run_real(config: Config, config_path: Path, *, hardware: bool = True)
library = await build_library(config)
app = _build_app(config, bus, mouse, player, library, clock=clock)
services = _build_services(
app, mouse, player, clock=clock, config_path=config_path
)
services = _build_services(app, mouse, player, clock=clock, config_path=config_path)
_log.info(
"MusicMouse %s starting: %d figures, %d albums, serial %s, audio %s, mqtt %s",
@@ -157,6 +152,15 @@ async def run_real(config: Config, config_path: Path, *, hardware: bool = True)
[
*([link.run()] if link else []),
player.run(),
# `is_busy` reads `player.is_playing` directly rather than the library
# knowing about playback at all: analysis must never compete with audio
# decoding for CPU, and this device is otherwise idle most of the day, so
# the pass simply resumes once it is. `on_batch` is unset unless the web
# front-end is on - nothing else has a use for the notification.
library.run_analysis(
is_busy=lambda: player.is_playing,
on_batch=_analysis_batch_hook(services),
),
*(service.run() for service in services),
]
)
@@ -184,6 +188,15 @@ async def run_simulated(config: Config, config_path: Path, script: Path | None)
sim.app, sim.app.mouse, sim.player, clock=RealClock(), config_path=config_path
)
tasks = [asyncio.create_task(service.run(), name=service.name) for service in services]
tasks.append(
asyncio.create_task(
sim.app.library.run_analysis(
is_busy=lambda: sim.player.is_playing,
on_batch=_analysis_batch_hook(services),
),
name="library-analysis",
)
)
try:
if script is not None:
await run_script_file(sim, script)
@@ -250,6 +263,7 @@ async def build_library(config: Config) -> MusicLibrary:
library_config.root,
library_config.cache,
frozenset(config.general.audio_extensions),
analyzer=build_analyzer(),
figure_kinds=config.figure_kinds,
)
@@ -279,6 +293,16 @@ def _build_app(
return app
def _analysis_batch_hook(services: list[Service]) -> Callable[[], Awaitable[None]] | None:
"""The web front-end's own hub, if it is running - so open tabs refetch the library
as background analysis lands, instead of only after a manual reload. `None` when
there is no web service, which `MusicLibrary.analyze_pending` treats as "nobody to
tell".
"""
web_service = next((s for s in services if isinstance(s, WebService)), None)
return web_service.hub.broadcast_library if web_service else None
def _build_services(
app: App,
mouse: MusicMouseDevice,

157
python-backend/musicmouse/library/__init__.py
View File

@@ -8,12 +8,22 @@ plain immutable data.
from __future__ import annotations
import asyncio
import contextlib
import logging
from collections.abc import Mapping
import os
from collections.abc import Awaitable, Callable, Collection, Mapping
from dataclasses import replace
from pathlib import Path
from typing import Final
from musicmouse.library.analysis import ANALYZER_VERSION, Analyzer, BeatGrid, NullAnalyzer
from musicmouse.library.analysis import (
ANALYZER_VERSION,
Analyzer,
BeatGrid,
NullAnalyzer,
TrackAnalysis,
TrackCurves,
)
from musicmouse.library.cache import Fingerprint, LibraryCache
from musicmouse.library.models import Album, LibraryTrack, album_id, track_key
from musicmouse.library.scanner import scan_library
@@ -32,10 +42,41 @@ __all__ = [
"LibraryTrack",
"MusicLibrary",
"NullAnalyzer",
"TrackCurves",
"album_id",
"track_key",
]
#: The only kind worth spending DSP on: an audiobook chapter or a podcast episode is
#: tens of minutes of narration with no musical mood to extract, and there are far more
#: of them in a typical library than there are songs.
_ANALYZED_KINDS: Final[tuple[AlbumKind, ...]] = ("music",)
#: How often the background worker checks whether it may resume after `is_busy()` said
#: no. A track's own analysis is a couple of seconds of CPU, so overshooting by this
#: much when playback stops is not worth polling harder for.
_BUSY_POLL_SECONDS: Final = 2.0
#: Analysis is folded back into the live index and persisted this often during a long
#: run, so a first-time pass over an unanalyzed library shows up gradually in open
#: browser tabs rather than only after the whole thing finishes.
_PUBLISH_BATCH_SIZE: Final = 25
def _analyze_one(
analyzer: Analyzer, path: Path
) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]:
"""Runs in a worker thread. Lowers this thread's own scheduling priority first.
On Linux, ``os.nice`` affects only the calling thread, not the whole process - so
this makes idle-time analysis yield CPU to anything else without touching threads
used for other work. Niceness only ever increases and clamps at the OS maximum
(19), so calling this repeatedly on a reused pool thread is harmless.
"""
with contextlib.suppress(OSError):
os.nice(1)
return analyzer.analyze(path)
class MusicLibrary:
"""An immutable index of albums, rebuilt wholesale rather than mutated in place."""
@@ -56,6 +97,10 @@ class MusicLibrary:
#: What each figure holds. The only thing a folder name cannot say.
self.figure_kinds: Mapping[str, AlbumKind] = figure_kinds or {}
self._entries: dict[str, tuple[Album, Fingerprint]] = {}
#: Set by `request_analysis`, consumed by `run_analysis`. An `Event` rather than
#: a queue: a request raised while one is already pending or running just
#: coalesces into it, which is exactly what "look again" should mean here.
self._analysis_requested = asyncio.Event()
# -------------------------------------------------------------------- reading
@@ -85,10 +130,21 @@ class MusicLibrary:
return None
return self.cache.load_beats(track_key(album.tracks[index].path))
def curve(self, identifier: str, index: int) -> TrackCurves | None:
album = self.get(identifier)
if album is None or not 0 <= index < len(album.tracks):
return None
return self.cache.load_curve(track_key(album.tracks[index].path))
# -------------------------------------------------------------------- writing
async def refresh(self) -> None:
"""Rescan from disk. Blocking work happens off the loop; the swap is atomic."""
"""Rescan from disk. Blocking work happens off the loop; the swap is atomic.
Ends by waking the background analyzer: a rescan is exactly when new tracks -
the only ones analysis can be pending for - enter the index, whether that is
the startup scan or a parent tapping "Bibliothek neu einlesen".
"""
known = dict(self._entries)
entries = await asyncio.to_thread(
scan_library,
@@ -100,19 +156,28 @@ class MusicLibrary:
)
self._entries = await asyncio.to_thread(self._with_analysis, entries)
await asyncio.to_thread(self.cache.store_index, self._entries)
self.request_analysis()
def _with_analysis(
self, entries: dict[str, tuple[Album, Fingerprint]]
self,
entries: dict[str, tuple[Album, Fingerprint]],
*,
kinds: Collection[AlbumKind] = _ANALYZED_KINDS,
) -> dict[str, tuple[Album, Fingerprint]]:
"""Fold cached analysis results into the freshly scanned index.
Skipped entirely while ``analysis/`` is empty, which is the normal case until an
analyzer has actually been run - no point stat-ing 900 files that cannot exist.
Restricted to `kinds` (music by default): stat-ing hundreds of book and podcast
tracks whose analysis can never exist would be pure waste. Skipped entirely
while ``analysis/`` is empty, which is the normal case until an analyzer has
actually been run.
"""
if not any(self.cache.analysis.glob("*.json")):
return entries
out: dict[str, tuple[Album, Fingerprint]] = {}
for identifier, (album, fingerprint) in entries.items():
if album.kind not in kinds:
out[identifier] = (album, fingerprint)
continue
tracks = tuple(
replace(track, analysis=self.cache.load_analysis(track_key(track.path)))
for track in album.tracks
@@ -136,12 +201,52 @@ class MusicLibrary:
await library.refresh()
return library
async def analyze_pending(self) -> int:
"""Run the analyzer over tracks that have no current result.
# ------------------------------------------------------------------- analysis
Never called during startup: the index is what playback needs, and analysis is
minutes of DSP per track. Results are keyed by file content, so they can equally
well be produced on a faster machine and the ``analysis/`` folder copied over.
def request_analysis(self) -> None:
"""Wake the background worker to look for tracks with no current analysis.
Idempotent, and safe to call before `run_analysis` has even started a first
time - the request just waits on the event.
"""
self._analysis_requested.set()
async def run_analysis(
self,
*,
is_busy: Callable[[], bool] = lambda: False,
on_batch: Callable[[], Awaitable[None]] | None = None,
) -> None:
"""Analyze pending tracks whenever `request_analysis` wakes this up.
A long-running task, cancelled at shutdown alongside every other one. Loops
forever so a request raised *during* a pass (a rescan mid-analysis) starts
another pass right after, rather than being lost.
"""
while True:
await self._analysis_requested.wait()
self._analysis_requested.clear()
await self.analyze_pending(is_busy=is_busy, on_batch=on_batch)
async def analyze_pending(
self,
*,
kinds: Collection[AlbumKind] = _ANALYZED_KINDS,
is_busy: Callable[[], bool] = lambda: False,
on_batch: Callable[[], Awaitable[None]] | None = None,
batch_size: int = _PUBLISH_BATCH_SIZE,
) -> int:
"""Run the analyzer over tracks of `kinds` that have no current result.
Restricted to music by default - see `_ANALYZED_KINDS`. Checked before every
track, `is_busy()` pauses the whole pass rather than one file: analysis must
never compete with audio decoding for CPU, and a children's player is idle most
of the day, so the pass simply resumes next time it is. A track the analyzer
fails on (corrupt file, DRM, zero length) is still recorded as attempted - with
every scalar left `None` - so it is never retried forever and the frontend falls
back to the un-analyzed baseline for it. Results are folded into the live index
and persisted every `batch_size` tracks, so a long first run is visible in open
browser tabs as it goes rather than only once it finishes.
"""
analyzer = self.analyzer
if analyzer.version < ANALYZER_VERSION:
@@ -149,16 +254,44 @@ class MusicLibrary:
done = 0
for album in self.albums:
if album.kind not in kinds:
continue
for track in album.tracks:
while is_busy():
await asyncio.sleep(_BUSY_POLL_SECONDS)
key = track_key(track.path)
cached = self.cache.load_analysis(key)
if cached is not None and cached.version >= analyzer.version:
continue
analysis, grid = await asyncio.to_thread(analyzer.analyze, track.path)
try:
analysis, grid, curve = await asyncio.to_thread(
_analyze_one, analyzer, track.path
)
except Exception:
_log.warning(
"Analyzer raised on %s; marking it attempted so it is not retried forever",
track.path,
exc_info=True,
)
analysis, grid, curve = TrackAnalysis(version=analyzer.version), None, None
if grid is not None:
self.cache.store_beats(key, grid)
if curve is not None:
self.cache.store_curve(key, curve)
self.cache.store_analysis(key, analysis)
done += 1
if done % batch_size == 0:
await self._publish_analysis(kinds, on_batch)
if done % batch_size:
await self._publish_analysis(kinds, on_batch)
if done:
_log.info("Analyzed %d tracks", done)
return done
async def _publish_analysis(
self, kinds: Collection[AlbumKind], on_batch: Callable[[], Awaitable[None]] | None
) -> None:
self._entries = await asyncio.to_thread(self._with_analysis, self._entries, kinds=kinds)
await asyncio.to_thread(self.cache.store_index, self._entries)
if on_batch is not None:
await on_batch()

103
python-backend/musicmouse/library/analysis.py
View File

@@ -1,16 +1,20 @@
"""Offline audio analysis: the seam, not the implementation.
"""Offline audio analysis, and the seam that lets it be optional.
Nothing here computes anything yet. What it does is fix the shape of the results so
that the analyzer can arrive later without touching the scanner, the cache format, the
API contract or the frontend's data flow.
:class:`LibrosaAnalyzer` (``musicmouse.library.librosa_analyzer``) does the real work,
kept in its own module behind :func:`build_analyzer` so that importing *this* module -
which the scanner, the cache and the web API all do - never pulls in librosa or numpy.
:class:`MusicLibrary` runs whichever analyzer it is given in the background, between
tracks, so the reactive background is a thing a library grows into rather than a
migration.
Two rules hold the design together:
Two rules hold the result shape together:
* **Scalars travel with the index, time series do not.** A 25-minute podcast at 120 BPM
has ~3000 beats; 900 tracks of that inside ``GET /api/library`` would be tens of
megabytes. :class:`TrackAnalysis` is a handful of floats and rides along; the beat
grid lives in its own file and is fetched for the one track that is playing.
* **Every field is optional with a default.** Adding ``danceability`` later needs no
grid and the per-second :class:`TrackCurves` each live in their own file and are
fetched only for the one track that is playing.
* **Every field is optional with a default.** Adding a new scalar later needs no
migration and no cache wipe: unknown keys on disk are dropped on load, missing ones
fall back to the default. Only :data:`ANALYZER_VERSION` moving past what a file
records marks that file stale.
@@ -18,21 +22,28 @@ Two rules hold the design together:
from __future__ import annotations
import logging
from dataclasses import asdict, dataclass, fields
from pathlib import Path
from typing import Any, Final, Protocol
_log = logging.getLogger(__name__)
__all__ = [
"ANALYZER_VERSION",
"Analyzer",
"BeatGrid",
"NullAnalyzer",
"TrackAnalysis",
"TrackCurves",
"build_analyzer",
]
#: Bumped when an analyzer's output changes meaning. Cached results recorded under a
#: lower version are recomputed; results at or above it are left alone.
ANALYZER_VERSION: Final = 1
#: lower version are recomputed; results at or above it are left alone. 2: energy/
#: valence scalars became mean-of-curve instead of whole-track-percentile/heuristic,
#: and every track needs a new TrackCurves artifact producing.
ANALYZER_VERSION: Final = 2
@dataclass(frozen=True, slots=True)
@@ -48,6 +59,10 @@ class TrackAnalysis:
valence: float | None = None
#: 0..1 spectral centroid. Drives the background hue.
brightness: float | None = None
#: 0..1 confidence that :attr:`tempo` is an audible, steady beat rather than an
#: artifact of free-tempo or spoken-word material. Below-threshold tracks should
#: not be pulsed on the beat even though a grid exists for them.
pulse: float | None = None
#: Whether a beat grid for this track exists on disk.
beats: bool = False
@@ -85,22 +100,80 @@ class BeatGrid:
return cls(tuple(flat[0::2]), tuple(flat[1::2]))
@dataclass(frozen=True, slots=True)
class TrackCurves:
"""Per-second samples of the things that vary *within* a track. Regularly
sampled, so just a hop and equal-length arrays, no per-sample timestamps -
contrast :class:`BeatGrid`'s event-based irregular times.
``energy``/``valence`` drive colour; ``drive`` (rhythmic intensity) modulates the
water current's magnitude. Tempo is deliberately absent: measured against a real
library it is flat to within a few percent inside a track, so a per-second tempo
curve would carry estimator noise (including occasional octave errors) and
nothing else. Whole-track tempo keeps setting the current's base magnitude;
``drive`` is the signal that actually varies.
"""
hop_seconds: float
energy: tuple[float, ...]
valence: tuple[float, ...]
drive: tuple[float, ...]
def to_json(self) -> dict[str, Any]:
return {
"hop_seconds": self.hop_seconds,
"energy": list(self.energy),
"valence": list(self.valence),
"drive": list(self.drive),
}
@classmethod
def from_json(cls, data: dict[str, Any]) -> TrackCurves:
return cls(
hop_seconds=float(data["hop_seconds"]),
energy=tuple(data["energy"]),
valence=tuple(data["valence"]),
drive=tuple(data["drive"]),
)
class Analyzer(Protocol):
"""Turns one audio file into cacheable analysis results.
Implementations are CPU-bound and run off the event loop. The real one will live
behind an optional dependency group so the device never installs numpy to play music.
Implementations are CPU-bound and run off the event loop.
"""
version: int
def analyze(self, path: Path) -> tuple[TrackAnalysis, BeatGrid | None]: ...
def analyze(self, path: Path) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]: ...
class NullAnalyzer:
"""Analyzes nothing. Keeps the wiring exercised until a real analyzer lands."""
"""Analyzes nothing. What :func:`build_analyzer` falls back to without librosa."""
version = 0
def analyze(self, path: Path) -> tuple[TrackAnalysis, BeatGrid | None]: # noqa: ARG002
return TrackAnalysis(), None
def analyze(
self,
path: Path, # noqa: ARG002
) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]:
return TrackAnalysis(), None, None
def build_analyzer() -> Analyzer:
"""The real analyzer if its optional dependency group is installed, else a no-op.
``NullAnalyzer.version`` is ``0``, which is below :data:`ANALYZER_VERSION`, so
:meth:`~musicmouse.library.MusicLibrary.analyze_pending` returns immediately and the
background stays at its static baseline - not a crash, not a degraded mode, just the
feature switched off until ``pip install -e '.[analysis]'`` turns it on.
"""
try:
from musicmouse.library.librosa_analyzer import LibrosaAnalyzer
except ImportError:
_log.info(
"librosa is not installed: the reactive background is off. "
"Install the 'analysis' extra to enable it."
)
return NullAnalyzer()
return LibrosaAnalyzer()

13
python-backend/musicmouse/library/cache.py
View File

@@ -8,6 +8,7 @@ different amounts to produce::
├── covers/<album_id>.jpg medium: art pulled out of an ID3 APIC frame
└── analysis/<track_key>.json expensive: minutes of DSP per track
analysis/<track_key>.beats.json
analysis/<track_key>.curve.json
That split is the whole point. A rescan must be free to rebuild ``index.json`` without
destroying analysis, so everything expensive is keyed by a *content* key (see
@@ -24,7 +25,7 @@ from dataclasses import dataclass
from pathlib import Path
from typing import Any, cast
from musicmouse.library.analysis import BeatGrid, TrackAnalysis
from musicmouse.library.analysis import BeatGrid, TrackAnalysis, TrackCurves
from musicmouse.library.models import Album, LibraryTrack
from musicmouse.library.sections import SECTIONS, AlbumKind
@@ -110,6 +111,16 @@ class LibraryCache:
def store_beats(self, key: str, grid: BeatGrid) -> None:
_write_atomic(self.analysis / f"{key}.beats.json", json.dumps(grid.to_json()))
def load_curve(self, key: str) -> TrackCurves | None:
path = self.analysis / f"{key}.curve.json"
try:
return TrackCurves.from_json(json.loads(path.read_text(encoding="utf-8")))
except (OSError, ValueError, KeyError):
return None
def store_curve(self, key: str, curve: TrackCurves) -> None:
_write_atomic(self.analysis / f"{key}.curve.json", json.dumps(curve.to_json()))
# --------------------------------------------------------------------- index
def load_index(self) -> dict[str, tuple[Album, Fingerprint]]:

318
python-backend/musicmouse/library/librosa_analyzer.py Normal file
View File

@@ -0,0 +1,318 @@
"""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]))

14
python-backend/musicmouse/services/web/api.py
View File

@@ -34,7 +34,6 @@ from musicmouse.events import (
from musicmouse.services.web.hub import StateHub
from musicmouse.services.web.schemas import (
AlbumOut,
BeatsOut,
HaConfigOut,
HaDeviceOut,
LibraryOut,
@@ -43,6 +42,8 @@ from musicmouse.services.web.schemas import (
SeekIn,
SettingsIn,
SettingsOut,
TrackCurvesOut,
TrackDetailOut,
VolumeIn,
)
from musicmouse.services.web.settings import (
@@ -86,11 +87,16 @@ def build_router(
)
@router.get("/tracks/{album_id}/{index}/analysis")
def get_track_analysis(album_id: str, index: int) -> BeatsOut:
def get_track_analysis(album_id: str, index: int) -> TrackDetailOut:
grid = app.library.beats(album_id, index)
if grid is None:
curve = app.library.curve(album_id, index)
if grid is None and curve is None:
raise HTTPException(status_code=404, detail="not analyzed")
return BeatsOut(times=list(grid.times), strengths=list(grid.strengths))
return TrackDetailOut(
times=list(grid.times) if grid else [],
strengths=list(grid.strengths) if grid else [],
curve=TrackCurvesOut(**curve.to_json()) if curve else None,
)
@router.post("/library/refresh", status_code=202)
async def refresh_library() -> Response:

22
python-backend/musicmouse/services/web/schemas.py
View File

@@ -15,7 +15,6 @@ from musicmouse.library.analysis import TrackAnalysis
__all__ = [
"AlbumOut",
"BeatsOut",
"HaConfigOut",
"HaDeviceOut",
"LibraryOut",
@@ -24,6 +23,8 @@ __all__ = [
"SeekIn",
"SettingsIn",
"SettingsOut",
"TrackCurvesOut",
"TrackDetailOut",
"TrackOut",
"VolumeIn",
]
@@ -34,6 +35,8 @@ class AnalysisOut(BaseModel):
energy: float | None = None
valence: float | None = None
brightness: float | None = None
#: 0..1 confidence that `tempo` is an audible, steady beat - see `TrackAnalysis`.
pulse: float | None = None
beats: bool = False
@classmethod
@@ -45,6 +48,7 @@ class AnalysisOut(BaseModel):
energy=analysis.energy,
valence=analysis.valence,
brightness=analysis.brightness,
pulse=analysis.pulse,
beats=analysis.beats,
)
@@ -99,9 +103,23 @@ class LibraryOut(BaseModel):
albums: list[AlbumOut]
class BeatsOut(BaseModel):
class TrackCurvesOut(BaseModel):
hop_seconds: float
energy: list[float]
valence: list[float]
drive: list[float]
class TrackDetailOut(BaseModel):
"""Beats + mood/drive curves for the one track currently playing - fetched
together since both are per-track detail the library payload never carries.
``curve`` is ``None`` only when the whole track failed analysis; ``times``/
``strengths`` are empty (not ``None``) for a track with no reliable beat, since a
free-tempo or spoken-word track still has real energy/valence/drive curves."""
times: list[float]
strengths: list[float]
curve: TrackCurvesOut | None = None
class ConnectionOut(BaseModel):

8
python-backend/pyproject.toml
View File

@@ -24,6 +24,10 @@ dependencies = [
[project.optional-dependencies]
dev = ["mypy>=1.10", "pytest-asyncio>=0.23", "pytest>=8.0", "ruff>=0.5"]
# Off by default: the reactive background is nice, not required, and librosa/numpy are
# a real install cost on a Pi. Missing this group means `build_analyzer()` falls back
# to `NullAnalyzer` and the background stays at its static baseline - never a crash.
analysis = ["librosa>=1.0", "numpy>=2.0"]
[project.scripts]
musicmouse = "musicmouse.__main__:main"
@@ -62,5 +66,7 @@ files = ["musicmouse"]
warn_unreachable = true
[[tool.mypy.overrides]]
module = ["vlc", "serial_asyncio", "ruamel.*"]
# numpy ships its own types, but only when the `analysis` extra is installed - a plain
# checkout must still type-check clean, so it needs the same treatment as librosa.
module = ["vlc", "serial_asyncio", "ruamel.*", "librosa.*", "numpy"]
ignore_missing_imports = true

115
python-backend/scripts/fetch_podcast_covers.py Normal file
View File

@@ -0,0 +1,115 @@
#!/usr/bin/env python3
"""One-off: backfill cover art for the podcast shows already in the library.
The scanner now reads a podcast episode's own embedded art first, and a show folder's
``cover.jpg`` second (see ``musicmouse.library.scanner._cover_for_episode``) - but
episodes downloaded before today rarely carry per-episode art, and none of the shows
have a folder-level cover on disk yet. This script fills that gap once, by hand, so it
is not something the running app does on its own.
For every ``Kinderpodcasts/<show>`` folder that has no ``cover.jpg``/``.jpeg``/``.png``
already, it looks the show up on the public iTunes Search API and saves the top match's
artwork as ``cover.jpg`` in that folder. A show it cannot match confidently is left
alone and logged, rather than guessed at - check those by hand afterwards.
This makes real network requests to a third-party service and writes into the real
music library, so it is meant to be run and reviewed by a person, not called from the
app:
python scripts/fetch_podcast_covers.py --config /path/to/config.yml
python scripts/fetch_podcast_covers.py --config /path/to/config.yml --dry-run
"""
from __future__ import annotations
import argparse
import json
import logging
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from musicmouse.config import load_config
_log = logging.getLogger("fetch_podcast_covers")
#: Mirrors scanner.py's `_COVER_NAMES` - what counts as "already has a cover".
_COVER_NAMES = ("cover.jpg", "cover.jpeg", "cover.png", "folder.jpg")
_ITUNES_SEARCH = "https://itunes.apple.com/search"
_TIMEOUT = 10.0
def _has_cover(folder: Path) -> bool:
return any((folder / name).is_file() for name in _COVER_NAMES)
def _find_artwork_url(show_name: str) -> str | None:
query = urllib.parse.urlencode({"media": "podcast", "term": show_name, "limit": 1})
try:
with urllib.request.urlopen(f"{_ITUNES_SEARCH}?{query}", timeout=_TIMEOUT) as response:
body = json.loads(response.read())
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as error:
_log.warning("Lookup for %r failed: %s", show_name, error)
return None
results = body.get("results") or []
if not results:
return None
# iTunes serves a 100x100 thumbnail by default; ask for something worth showing.
artwork = results[0].get("artworkUrl100")
return artwork.replace("100x100", "600x600") if artwork else None
def backfill(root: Path, *, dry_run: bool) -> None:
podcasts_root = root / "Kinderpodcasts"
if not podcasts_root.is_dir():
_log.warning("No Kinderpodcasts folder at %s", podcasts_root)
return
for folder in sorted(podcasts_root.iterdir()):
if not folder.is_dir() or folder.name.startswith("."):
continue
if _has_cover(folder):
_log.info("%-40s already has a cover, skipping", folder.name)
continue
url = _find_artwork_url(folder.name)
if url is None:
_log.warning("%-40s no confident match - fetch this one by hand", folder.name)
continue
_log.info("%-40s -> %s", folder.name, url)
if dry_run:
continue
try:
with urllib.request.urlopen(url, timeout=_TIMEOUT) as response:
data = response.read()
except (urllib.error.URLError, TimeoutError) as error:
_log.warning("%-40s download failed: %s", folder.name, error)
continue
(folder / "cover.jpg").write_bytes(data)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("-c", "--config", type=Path, required=True, help="path to config.yml")
parser.add_argument(
"--dry-run",
action="store_true",
help="print what would be fetched without downloading or writing anything",
)
args = parser.parse_args(argv)
logging.basicConfig(level=logging.INFO, format="%(message)s")
config = load_config(args.config)
backfill(config.general.library.root, dry_run=args.dry_run)
return 0
if __name__ == "__main__":
sys.exit(main())

265
python-backend/tests/test_library.py
View File

@@ -2,32 +2,56 @@
from __future__ import annotations
import asyncio
import contextlib
import json
import threading
from collections.abc import Callable
from pathlib import Path
import pytest
from musicmouse.config import DEFAULT_AUDIO_EXTENSIONS, load_config
from musicmouse.library import Album, MusicLibrary
from musicmouse.library.analysis import ANALYZER_VERSION, BeatGrid, TrackAnalysis
from musicmouse.library import Album, Analyzer, MusicLibrary
from musicmouse.library.analysis import ANALYZER_VERSION, BeatGrid, TrackAnalysis, TrackCurves
from musicmouse.library.cache import LibraryCache
from musicmouse.library.colors import colors_from_id
from musicmouse.library.models import album_id
from musicmouse.library.models import album_id, track_key
from tests.conftest import VALID_CONFIG, write_config, write_track
EXTENSIONS = frozenset(DEFAULT_AUDIO_EXTENSIONS)
async def build(config_dir: Path) -> MusicLibrary:
async def build(config_dir: Path, *, analyzer: Analyzer | None = None) -> MusicLibrary:
config = load_config(write_config(config_dir, VALID_CONFIG))
return await MusicLibrary.build(
config.general.library.root,
config.general.library.cache,
EXTENSIONS,
analyzer=analyzer,
figure_kinds=config.figure_kinds,
)
class FakeAnalyzer:
"""An `Analyzer` that does no DSP, so the worker's own behaviour - which tracks it
touches, how it reacts to `is_busy`, what a raising analyzer does to the batch - can
be tested without librosa."""
def __init__(self, *, version: int = ANALYZER_VERSION, fails: set[str] | None = None) -> None:
self.version = version
self._fails = fails or set()
self.calls: list[Path] = []
def analyze(self, path: Path) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]:
self.calls.append(path)
if path.name in self._fails:
raise RuntimeError(f"boom: {path.name}")
analysis = TrackAnalysis(version=self.version, tempo=100.0, energy=0.5)
curves = TrackCurves(hop_seconds=1.0, energy=(0.5,), valence=(0.5,), drive=(0.5,))
return analysis, BeatGrid((0.1,), (1.0,)), curves
def album_named(library: MusicLibrary, title: str) -> Album:
return next(album for album in library.albums if album.title == title)
@@ -233,25 +257,56 @@ async def test_a_new_episode_only_costs_scanning_that_one_file(config_dir: Path)
async def test_a_rescan_leaves_analysis_alone(config_dir: Path) -> None:
"""The whole reason the cache is a directory rather than one file."""
"""The whole reason the cache is a directory rather than one file.
"Kinderparty Lieder" rather than "Eule": folding analysis back into the index is
restricted to music (see `_ANALYZED_KINDS`), since only music is ever analyzed - a
book's `analysis/` file, if one somehow existed, is not something a rescan need
resurface.
"""
library = await build(config_dir)
album = album_named(library, "Kinderparty Lieder")
from musicmouse.library.models import track_key
key = track_key(album.tracks[0].path)
library.cache.store_analysis(key, TrackAnalysis(version=ANALYZER_VERSION, tempo=128.0))
library.cache.store_beats(key, BeatGrid((0.5, 1.0), (1.0, 0.5)))
curve = TrackCurves(hop_seconds=1.0, energy=(0.4, 0.6), valence=(0.5, 0.5), drive=(0.3, 0.7))
library.cache.store_curve(key, curve)
# Force a full rescan by dropping the cheap part of the cache.
(config_dir / ".cache" / "index.json").unlink()
library = await build(config_dir)
album = album_named(library, "Kinderparty Lieder")
assert album.tracks[0].analysis is not None
assert album.tracks[0].analysis.tempo == 128.0
grid = library.beats(album.id, 0)
assert grid is not None
assert grid.times == (0.5, 1.0)
curve = library.curve(album.id, 0)
assert curve is not None
assert curve.energy == (0.4, 0.6)
async def test_a_rescan_does_not_fold_analysis_into_a_non_music_album(
config_dir: Path,
) -> None:
"""The other half of the restriction above: a book track's `analysis/` file (however
it got there) is not folded into the index, so a rescan never stats hundreds of
book/podcast files that can never have one under normal operation."""
library = await build(config_dir)
album = album_named(library, "Eule")
from musicmouse.library.models import track_key
key = track_key(album.tracks[0].path)
library.cache.store_analysis(key, TrackAnalysis(version=ANALYZER_VERSION, tempo=128.0))
library.cache.store_beats(key, BeatGrid((0.5, 1.0), (1.0, 0.5)))
# Force a full rescan by dropping the cheap part of the cache.
(config_dir / ".cache" / "index.json").unlink()
library = await build(config_dir)
album = album_named(library, "Eule")
assert album.tracks[0].analysis is not None
assert album.tracks[0].analysis.tempo == 128.0
grid = library.beats(album.id, 0)
assert grid is not None
assert grid.times == (0.5, 1.0)
assert album.tracks[0].analysis is None
async def test_analysis_survives_an_analyzer_that_grew_a_field(tmp_path: Path) -> None:
@@ -340,3 +395,189 @@ async def test_a_figures_kind_survives_the_cache(config_dir: Path) -> None:
assert eule.series is not None
assert album_named(second, "Fuchs").kind == "music"
assert album_named(second, "Fuchs").series is None
# -------------------------------------------------------------------- analysis worker
async def _wait_for(predicate: Callable[[], bool], *, timeout: float = 2.0) -> None:
"""Poll until `predicate()` is true, rather than guessing a sleep duration."""
async def poll() -> None:
while not predicate():
await asyncio.sleep(0.01)
await asyncio.wait_for(poll(), timeout=timeout)
async def _cancel(task: asyncio.Task[object]) -> None:
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
async def test_analyze_pending_only_touches_music_albums(config_dir: Path) -> None:
"""Books and podcasts are the majority of a real library and none of them get a
background - see `_ANALYZED_KINDS`."""
analyzer = FakeAnalyzer()
library = await build(config_dir, analyzer=analyzer)
music_track_count = sum(len(a.tracks) for a in library.albums if a.kind == "music")
assert music_track_count == 5 # fuchs (3) + Kinderparty Lieder (2)
done = await library.analyze_pending()
assert done == 5
assert len(analyzer.calls) == 5
touched = {
next(a for a in library.albums if p in {t.path for t in a.tracks}).kind
for p in analyzer.calls
}
assert touched == {"music"}
async def test_analyze_pending_persists_curves_alongside_beats(config_dir: Path) -> None:
"""`MusicLibrary.curve()` mirrors `.beats()` - both are per-track detail fetched
for the one track currently playing, written together every pass."""
analyzer = FakeAnalyzer()
library = await build(config_dir, analyzer=analyzer)
album = next(a for a in library.albums if a.kind == "music")
await library.analyze_pending()
curve = library.curve(album.id, 0)
assert curve is not None
assert curve.hop_seconds == 1.0
assert curve.energy == (0.5,)
assert library.beats(album.id, 0) is not None
async def test_analyze_pending_does_not_repeat_work_already_cached(config_dir: Path) -> None:
analyzer = FakeAnalyzer()
library = await build(config_dir, analyzer=analyzer)
await library.analyze_pending()
again = await library.analyze_pending()
assert again == 0
assert len(analyzer.calls) == 5
async def test_a_failing_track_is_marked_attempted_and_not_retried(config_dir: Path) -> None:
"""A corrupt file, a DRM'd one, or one the analyzer just chokes on must not abort
the batch, and must not be retried on every single future pass either."""
analyzer = FakeAnalyzer(fails={"00 - lied.mp3"})
library = await build(config_dir, analyzer=analyzer)
done = await library.analyze_pending()
assert done == 5 # the failure still counts as "handled"
kinderparty = album_named(library, "Kinderparty Lieder")
failed = next(t for t in kinderparty.tracks if t.path.name == "00 - lied.mp3")
cached = library.cache.load_analysis(track_key(failed.path))
assert cached is not None
assert cached.version == ANALYZER_VERSION
assert cached.tempo is None # attempted, not computed
again = await library.analyze_pending()
assert again == 0
assert len(analyzer.calls) == 5
async def test_is_busy_pauses_the_pass_until_it_clears(
config_dir: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
import musicmouse.library as library_module
monkeypatch.setattr(library_module, "_BUSY_POLL_SECONDS", 0.01)
analyzer = FakeAnalyzer()
library = await build(config_dir, analyzer=analyzer)
busy = True
task = asyncio.create_task(library.analyze_pending(is_busy=lambda: busy))
try:
await asyncio.sleep(0.05)
assert analyzer.calls == [] # never got past the busy check
busy = False
await asyncio.wait_for(task, timeout=2)
finally:
if not task.done():
await _cancel(task)
assert len(analyzer.calls) == 5
async def test_run_analysis_does_the_work_refresh_left_pending(config_dir: Path) -> None:
"""The trigger this feature actually depends on: `refresh()` - at startup, or from
a parent's "Bibliothek neu einlesen" - ends by requesting analysis, and
`run_analysis` is what turns that request into finished work, without a second
`refresh()` needed to see it."""
analyzer = FakeAnalyzer()
library = await build(config_dir, analyzer=analyzer) # refresh() already requested
batches: list[int] = []
async def on_batch() -> None:
batches.append(len(analyzer.calls))
worker = asyncio.create_task(library.run_analysis(on_batch=on_batch))
try:
await _wait_for(lambda: bool(batches))
finally:
await _cancel(worker)
assert batches[-1] == 5
kinderparty = album_named(library, "Kinderparty Lieder")
assert kinderparty.tracks[0].analysis is not None
assert kinderparty.tracks[0].analysis.tempo == 100.0
async def test_requests_raised_during_a_pass_coalesce_into_one_more_pass(
config_dir: Path,
) -> None:
"""Calling `request_analysis` three times while a pass is already running must not
queue three more passes - just one, right after the current one finishes."""
release = threading.Event()
class BlockingOnceAnalyzer(FakeAnalyzer):
def __init__(self) -> None:
super().__init__()
self._blocked = False
def analyze(self, path: Path) -> tuple[TrackAnalysis, BeatGrid | None, TrackCurves | None]:
if not self._blocked:
self._blocked = True
release.wait(timeout=2)
return super().analyze(path)
analyzer = BlockingOnceAnalyzer()
library = await build(config_dir, analyzer=analyzer)
pass_count = 0
original = library.analyze_pending
async def counting(**kwargs: object) -> int:
nonlocal pass_count
pass_count += 1
return await original(**kwargs) # type: ignore[arg-type]
library.analyze_pending = counting # type: ignore[method-assign]
worker = asyncio.create_task(library.run_analysis())
try:
await _wait_for(lambda: pass_count >= 1) # the first pass has started
await asyncio.sleep(0.05) # ... and is now blocked inside the analyzer
library.request_analysis()
library.request_analysis()
library.request_analysis()
release.set()
await _wait_for(lambda: pass_count >= 2)
# Give the coalesced pass a moment to actually run - everything is already
# cached, so it finds nothing pending and returns without incrementing further.
await asyncio.sleep(0.1)
finally:
await _cancel(worker)
assert pass_count == 2
assert len(analyzer.calls) == 5

199
python-backend/tests/test_librosa_analyzer.py Normal file
View File

@@ -0,0 +1,199 @@
"""`LibrosaAnalyzer` against synthesised signals, not shipped audio fixtures.
Skipped wholesale when the ``analysis`` extra is not installed - exactly the state a
checkout of this repo is in until someone runs ``pip install -e '.[analysis]'``, and
what :func:`musicmouse.library.analysis.build_analyzer` falls back to `NullAnalyzer` for.
Warnings become errors project-wide (see ``pytest.ini_options.filterwarnings`` in
``pyproject.toml``), and numba emits a benign one on its first JIT compile per process -
so this module, alone, turns that off rather than loosening the project-wide setting.
"""
from __future__ import annotations
from pathlib import Path
import pytest
librosa = pytest.importorskip("librosa")
np = pytest.importorskip("numpy")
sf = pytest.importorskip("soundfile")
from musicmouse.library.librosa_analyzer import LibrosaAnalyzer # noqa: E402
pytestmark = pytest.mark.filterwarnings("ignore")
_SR = 22050
def _click_train(bpm: float, seconds: float = 20.0, sr: int = _SR) -> np.ndarray:
"""A metronome: short decaying clicks exactly `bpm` apart, easy for a beat tracker
to lock onto - real music is messier, but this makes a known-answer test possible.
"""
y = np.zeros(int(seconds * sr), dtype=np.float32)
period = 60.0 / bpm
click = np.hanning(200).astype(np.float32)
t = 0.0
while t < seconds:
i = int(t * sr)
n = min(len(click), len(y) - i)
if n > 0:
y[i : i + n] += click[:n]
t += period
return y
def _sine(freq: float, seconds: float = 12.0, sr: int = _SR, amp: float = 0.5) -> np.ndarray:
t = np.linspace(0, seconds, int(seconds * sr), endpoint=False)
return (amp * np.sin(2 * np.pi * freq * t)).astype(np.float32)
def _write(tmp_path: Path, name: str, y: np.ndarray, sr: int = _SR) -> Path:
path = tmp_path / name
sf.write(path, y, sr)
return path
def _close_to_bpm(tempo: float, target: float, tolerance: float = 8.0) -> bool:
"""Beat trackers routinely report a tempo at half or double the "true" one - both
are the same beat grid, just every-other-click or twice-per-click. Any of the three
counts as a correct detection."""
candidates = (target, target / 2, target * 2)
return any(abs(tempo - candidate) <= tolerance for candidate in candidates)
def test_tempo_from_a_click_train(tmp_path: Path) -> None:
path = _write(tmp_path, "clicks.wav", _click_train(120.0))
analysis, grid, curves = LibrosaAnalyzer().analyze(path)
assert analysis.tempo is not None
assert _close_to_bpm(analysis.tempo, 120.0)
assert grid is not None
assert len(grid.times) > 10
assert curves is not None
def test_brightness_orders_a_high_tone_above_a_low_one(tmp_path: Path) -> None:
low = _write(tmp_path, "low.wav", _sine(300.0))
high = _write(tmp_path, "high.wav", _sine(3000.0))
analyzer = LibrosaAnalyzer()
low_analysis, _, _ = analyzer.analyze(low)
high_analysis, _, _ = analyzer.analyze(high)
assert low_analysis.brightness is not None
assert high_analysis.brightness is not None
assert high_analysis.brightness > low_analysis.brightness
def test_energy_orders_a_loud_signal_above_a_quiet_one(tmp_path: Path) -> None:
loud = _write(tmp_path, "loud.wav", _sine(440.0, amp=0.9))
quiet = _write(tmp_path, "quiet.wav", _sine(440.0, amp=0.09))
analyzer = LibrosaAnalyzer()
loud_analysis, _, _ = analyzer.analyze(loud)
quiet_analysis, _, _ = analyzer.analyze(quiet)
assert loud_analysis.energy is not None
assert quiet_analysis.energy is not None
assert loud_analysis.energy > quiet_analysis.energy
def test_pulse_is_higher_for_a_steady_beat_than_a_plain_tone(tmp_path: Path) -> None:
"""What `pulse` is for: telling a track that actually pulses apart from one that
merely has *a* tempo number attached to it, like a sustained tone or narration."""
clicks = _write(tmp_path, "clicks.wav", _click_train(120.0))
tone = _write(tmp_path, "tone.wav", _sine(440.0))
analyzer = LibrosaAnalyzer()
click_analysis, _, _ = analyzer.analyze(clicks)
tone_analysis, _, _ = analyzer.analyze(tone)
assert click_analysis.pulse is not None
assert tone_analysis.pulse is not None
assert click_analysis.pulse > tone_analysis.pulse
def test_a_corrupt_file_is_analyzed_as_nothing_rather_than_raising(tmp_path: Path) -> None:
path = tmp_path / "corrupt.mp3"
path.write_bytes(b"not an audio file")
analysis, grid, curves = LibrosaAnalyzer().analyze(path)
assert analysis.version == LibrosaAnalyzer.version
assert analysis.tempo is None
assert analysis.energy is None
assert analysis.valence is None
assert analysis.brightness is None
assert analysis.pulse is None
assert analysis.beats is False
assert grid is None
assert curves is None
def test_a_zero_length_file_is_analyzed_as_nothing_rather_than_raising(tmp_path: Path) -> None:
path = _write(tmp_path, "empty.wav", np.zeros(0, dtype=np.float32))
analysis, grid, curves = LibrosaAnalyzer().analyze(path)
assert analysis.version == LibrosaAnalyzer.version
assert analysis.tempo is None
assert grid is None
assert curves is None
def test_curves_are_produced_alongside_the_scalars(tmp_path: Path) -> None:
path = _write(tmp_path, "clicks.wav", _click_train(120.0))
_, _, curves = LibrosaAnalyzer().analyze(path)
assert curves is not None
assert curves.hop_seconds == 1.0
assert len(curves.energy) == len(curves.valence) == len(curves.drive) >= 1
def test_whole_track_scalars_are_the_mean_of_their_curves(tmp_path: Path) -> None:
path = _write(tmp_path, "clicks.wav", _click_train(120.0))
analysis, _, curves = LibrosaAnalyzer().analyze(path)
assert curves is not None
assert analysis.energy == pytest.approx(sum(curves.energy) / len(curves.energy))
assert analysis.valence == pytest.approx(sum(curves.valence) / len(curves.valence))
def test_energy_curve_tracks_a_loud_then_quiet_signal(tmp_path: Path) -> None:
loud = _sine(440.0, seconds=10.0, amp=0.9)
quiet = _sine(440.0, seconds=10.0, amp=0.09)
path = _write(tmp_path, "loud_then_quiet.wav", np.concatenate([loud, quiet]))
_, _, curves = LibrosaAnalyzer().analyze(path)
assert curves is not None
half = len(curves.energy) // 2
first_half_mean = sum(curves.energy[:half]) / half
second_half_mean = sum(curves.energy[half:]) / (len(curves.energy) - half)
assert first_half_mean > second_half_mean
def test_drive_is_higher_for_a_steady_beat_than_a_plain_tone(tmp_path: Path) -> None:
"""What `drive` is for: a track's rhythmic intensity, the signal that actually
varies within a track (unlike tempo, which is flat to within a few percent inside
a real recording) - see `LibrosaAnalyzer._drive_curve`."""
clicks = _write(tmp_path, "clicks.wav", _click_train(120.0))
tone = _write(tmp_path, "tone.wav", _sine(440.0))
analyzer = LibrosaAnalyzer()
_, _, click_curves = analyzer.analyze(clicks)
_, _, tone_curves = analyzer.analyze(tone)
assert click_curves is not None
assert tone_curves is not None
click_mean = sum(click_curves.drive) / len(click_curves.drive)
tone_mean = sum(tone_curves.drive) / len(tone_curves.drive)
assert click_mean > tone_mean
def test_a_clip_shorter_than_one_hop_still_produces_a_one_sample_curve(tmp_path: Path) -> None:
path = _write(tmp_path, "short.wav", _sine(440.0, seconds=0.3))
_, _, curves = LibrosaAnalyzer().analyze(path)
assert curves is not None
assert len(curves.energy) == 1

46
python-backend/tests/test_web.py
View File

@@ -17,6 +17,8 @@ import pytest
from fastapi import FastAPI
from musicmouse.config import WebConfig, load_config
from musicmouse.library.analysis import BeatGrid, TrackCurves
from musicmouse.library.models import track_key
from musicmouse.services.web.service import build_app
from musicmouse.simulator.harness import Simulation, build_simulation
from tests.conftest import VALID_CONFIG, write_config
@@ -96,6 +98,50 @@ async def test_unanalyzed_tracks_report_no_analysis(client: Client) -> None:
assert (await client.get(f"/api/tracks/{album['id']}/0/analysis")).status_code == 404
async def test_an_analyzed_track_reports_its_beats_and_curves(
client: Client, sim: Simulation
) -> None:
album = await album_by_title(client, "Eule")
library_album = sim.app.library.get(album["id"])
assert library_album is not None
key = track_key(library_album.tracks[0].path)
sim.app.library.cache.store_beats(key, BeatGrid((0.5, 1.0), (1.0, 0.5)))
sim.app.library.cache.store_curve(
key, TrackCurves(hop_seconds=1.0, energy=(0.4, 0.6), valence=(0.5, 0.5), drive=(0.3, 0.7))
)
response = await client.get(f"/api/tracks/{album['id']}/0/analysis")
assert response.status_code == 200
body = response.json()
assert body["times"] == [0.5, 1.0]
assert body["curve"]["energy"] == [0.4, 0.6]
assert body["curve"]["valence"] == [0.5, 0.5]
assert body["curve"]["drive"] == [0.3, 0.7]
async def test_a_track_with_a_curve_but_no_reliable_beat_still_reports_200(
client: Client, sim: Simulation
) -> None:
"""A free-tempo/spoken-word track has no usable beat grid, but energy/valence/
drive are computed regardless - the endpoint must not 404 just because `beats`
came back empty."""
album = await album_by_title(client, "Eule")
library_album = sim.app.library.get(album["id"])
assert library_album is not None
key = track_key(library_album.tracks[0].path)
sim.app.library.cache.store_curve(
key, TrackCurves(hop_seconds=1.0, energy=(0.5,), valence=(0.5,), drive=(0.5,))
)
response = await client.get(f"/api/tracks/{album['id']}/0/analysis")
assert response.status_code == 200
body = response.json()
assert body["times"] == []
assert body["curve"]["energy"] == [0.5]
# -------------------------------------------------------------------------- state