310 lines
12 KiB
Python
310 lines
12 KiB
Python
|
# SPDX-License-Identifier: Apache-2.0
|
||
|
|
"""
|
||
|
|
Headless analysis service for the web UI.
|
||
|
|
|
||
|
|
This is the server-side equivalent of `AnalysisResultsManager` + its worker
|
||
|
|
threads: it owns the decode-and-compute lifecycle and the per-file metric cache,
|
||
|
|
but it knows nothing about a GUI toolkit. No PyQt import appears in this file or
|
||
|
|
anything it imports — that property is what lets `webapp.py` run under a plain
|
||
|
|
ASGI server on a box with no display.
|
||
|
|
|
||
|
|
## Where the audio lives
|
||
|
|
|
||
|
|
The browser is the store of record. The client keeps every uploaded file in OPFS
|
||
|
|
(origin-private file system) and re-sends the bytes for each analysis request; the
|
||
|
|
server spools them to a temporary file, analyses it, and deletes it in a
|
||
|
|
`finally` block. Nothing this module or the API layer does leaves an audio file
|
||
|
|
behind on the host.
|
||
|
|
|
||
|
|
That is a deliberate trade, made because the analysis backend is
|
||
|
|
`master_core.AudioFile` -> `librosa.load(path)` -> numpy/scipy. Decoding needs a
|
||
|
|
real path, and the metrics need the samples in *this* process, so the bytes must
|
||
|
|
transit the host. What the design guarantees is that they never *persist* there:
|
||
|
|
no upload directory, no files surviving a request, and no server state keyed on a
|
||
|
|
path the client chose. A crash mid-analysis can leave a file in `tempfile`'s
|
||
|
|
directory, which is the OS's to clean; there is no application-owned location
|
||
|
|
that accumulates audio.
|
||
|
|
|
||
|
|
The cost is bandwidth: switching metrics re-uploads the track. For a local
|
||
|
|
workstation deployment that is a loopback copy of a few tens of megabytes, which
|
||
|
|
is cheaper than the alternative (an LRU cache of decoded signals) — and the
|
||
|
|
client only re-sends when the metric isn't already computed, because the computed
|
||
|
|
*data* is cached here, keyed by content hash.
|
||
|
|
|
||
|
|
## Content addressing
|
||
|
|
|
||
|
|
`file_id` is the SHA-256 of the audio bytes, computed by the client. Using the
|
||
|
|
hash rather than the filename means:
|
||
|
|
|
||
|
|
- two takes with the same name never collide, and renaming a file doesn't orphan
|
||
|
|
its cached metrics;
|
||
|
|
- the server can key its cache on the bytes actually analysed, so a client that
|
||
|
|
re-sends the same file gets the cached result without re-decoding;
|
||
|
|
- nothing the client sends is used to construct a path, so there is no traversal
|
||
|
|
surface to defend.
|
||
|
|
|
||
|
|
## Differences from the desktop manager, and why
|
||
|
|
|
||
|
|
- **No prefetch.** The desktop app warms every metric for the loaded file so a
|
||
|
|
dropdown switch is instant. Analysing in a browser is request-driven, and
|
||
|
|
prefetching seven metrics on every upload would occupy a core for minutes on a
|
||
|
|
long track while the user looks at one plot. So `compute()` runs on demand and
|
||
|
|
the result is cached.
|
||
|
|
- **One lock, not a thread-per-job.** Long analysis holds the CPU — numpy and
|
||
|
|
scipy release the GIL, but concurrent librosa loads on the same box just thrash
|
||
|
|
the cache and make every request slower. Serialising here also keeps the
|
||
|
|
`metric_data` cache simple to reason about.
|
||
|
|
- **The decoded signal is not kept.** Once a metric is computed, the `AudioFile`
|
||
|
|
is dropped: the samples are the expensive part (tens of MB of float32 for a
|
||
|
|
4-minute track), and every metric we ship is computed from one pass. Holding
|
||
|
|
them to save a decode on a *second* metric costs more memory than the re-upload
|
||
|
|
costs time.
|
||
|
|
|
||
|
|
`AudioFile` itself memoises the K-weighted signal so the three metrics that share
|
||
|
|
it (LUFS and PSR via `_short_term_lufs`, Crest Factor's windowing) don't each
|
||
|
|
re-filter the audio within a single request.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import hashlib
|
||
|
|
import logging
|
||
|
|
import threading
|
||
|
|
import time
|
||
|
|
from dataclasses import dataclass, field
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any, Optional
|
||
|
|
|
||
|
|
from master_core import AudioFile
|
||
|
|
from metrics import METRICS, DEFAULT_METRIC_ID
|
||
|
|
|
||
|
|
logger = logging.getLogger(__name__)
|
||
|
|
|
||
|
|
# Suffixes the decoder is expected to handle. The client filters on these too,
|
||
|
|
# but the server is the authority — a client is not a trusted input.
|
||
|
|
AUDIO_SUFFIXES = frozenset({
|
||
|
|
".wav", ".mp3", ".flac", ".ogg", ".oga", ".opus", ".m4a", ".aac",
|
||
|
|
".aiff", ".aif", ".aifc", ".wma", ".mp4", ".webm",
|
||
|
|
})
|
||
|
|
|
||
|
|
# Spool chunk while receiving the body: 1 MiB keeps peak memory flat regardless
|
||
|
|
# of file size while not hammering the syscall boundary.
|
||
|
|
_SPOOL_CHUNK = 1024 * 1024
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class AudioFileSnapshot:
|
||
|
|
"""The bits of an `AudioFile` the API needs, without the decoded signal."""
|
||
|
|
file_id: str
|
||
|
|
name: str
|
||
|
|
song_name: str
|
||
|
|
sample_rate: int
|
||
|
|
duration: float
|
||
|
|
max_amplitude: float
|
||
|
|
avg_amplitude: float
|
||
|
|
size_bytes: int = 0
|
||
|
|
|
||
|
|
@property
|
||
|
|
def channels(self) -> int:
|
||
|
|
"""Always 1: analysis is mono.
|
||
|
|
|
||
|
|
`librosa.load` downmixes to mono by default and `AudioFile` stores the mono
|
||
|
|
signal, so every metric here measures the mono sum — same as the desktop app.
|
||
|
|
The field is kept explicit (rather than omitted) because a "0.0 dBTP on a
|
||
|
|
stereo master" reading is only meaningful once you know that.
|
||
|
|
"""
|
||
|
|
return 1
|
||
|
|
|
||
|
|
def as_dict(self) -> dict[str, Any]:
|
||
|
|
return {
|
||
|
|
"file_id": self.file_id,
|
||
|
|
"name": self.name,
|
||
|
|
"song_name": self.song_name,
|
||
|
|
"sample_rate": self.sample_rate,
|
||
|
|
"duration": self.duration,
|
||
|
|
"max_amplitude": self.max_amplitude,
|
||
|
|
"avg_amplitude": self.avg_amplitude,
|
||
|
|
"channels": self.channels,
|
||
|
|
"size_bytes": self.size_bytes,
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class SessionFile:
|
||
|
|
"""One analysed track: its snapshot plus every metric computed for it so far."""
|
||
|
|
snapshot: AudioFileSnapshot
|
||
|
|
metric_data: dict[str, Any] = field(default_factory=dict)
|
||
|
|
|
||
|
|
def metadata(self) -> dict[str, Any]:
|
||
|
|
payload = self.snapshot.as_dict()
|
||
|
|
payload["metrics_cached"] = sorted(self.metric_data.keys())
|
||
|
|
return payload
|
||
|
|
|
||
|
|
|
||
|
|
def file_id_for(data: bytes) -> str:
|
||
|
|
"""Content hash used as the session key (see the module docstring)."""
|
||
|
|
return hashlib.sha256(data).hexdigest()[:32]
|
||
|
|
|
||
|
|
|
||
|
|
# Spool chunk while receiving a body: 1 MiB keeps peak memory flat regardless of
|
||
|
|
# file size while not hammering the syscall boundary.
|
||
|
|
SPOOL_CHUNK = 1024 * 1024
|
||
|
|
|
||
|
|
|
||
|
|
class AnalysisSession:
|
||
|
|
"""Server-side state: analysed tracks keyed by content hash, and the view set."""
|
||
|
|
|
||
|
|
def __init__(self, max_files: int = 64) -> None:
|
||
|
|
self.files: dict[str, SessionFile] = {}
|
||
|
|
# File ids currently drawn (the desktop app's file-list checkboxes).
|
||
|
|
self.overlay: list[str] = []
|
||
|
|
self.max_files = max_files
|
||
|
|
self._lock = threading.RLock()
|
||
|
|
|
||
|
|
# ---- loading ------------------------------------------------------------
|
||
|
|
|
||
|
|
def load(self, path: Path, file_id: str, name: str, size_bytes: int = 0) -> SessionFile:
|
||
|
|
"""Decode `path` (a spool that the caller owns) into the session.
|
||
|
|
|
||
|
|
Idempotent by `file_id`: a client re-sending the same bytes for a metric that
|
||
|
|
isn't cached yet re-uses the existing snapshot rather than re-decoding.
|
||
|
|
"""
|
||
|
|
with self._lock:
|
||
|
|
existing = self.files.get(file_id)
|
||
|
|
if existing is not None:
|
||
|
|
return existing
|
||
|
|
|
||
|
|
t0 = time.perf_counter()
|
||
|
|
audio = AudioFile(str(path))
|
||
|
|
snapshot = AudioFileSnapshot(
|
||
|
|
file_id=file_id,
|
||
|
|
name=name,
|
||
|
|
song_name=audio.song_name,
|
||
|
|
sample_rate=int(audio.sr),
|
||
|
|
duration=float(len(audio.y_mono) / audio.sr) if audio.sr else 0.0,
|
||
|
|
max_amplitude=float(audio.max_amplitude),
|
||
|
|
avg_amplitude=float(audio.avg_amplitude),
|
||
|
|
size_bytes=size_bytes,
|
||
|
|
)
|
||
|
|
entry = SessionFile(snapshot=snapshot)
|
||
|
|
logger.info(
|
||
|
|
"Loaded %s in %.1fs (%.0f s @ %d Hz, %s)",
|
||
|
|
snapshot.song_name, time.perf_counter() - t0,
|
||
|
|
snapshot.duration, snapshot.sample_rate, file_id[:8],
|
||
|
|
)
|
||
|
|
with self._lock:
|
||
|
|
self.files[file_id] = entry
|
||
|
|
self._evict_locked()
|
||
|
|
if file_id not in self.overlay:
|
||
|
|
self.overlay.append(file_id)
|
||
|
|
return entry
|
||
|
|
|
||
|
|
# ---- metrics ------------------------------------------------------------
|
||
|
|
|
||
|
|
def compute_metric(self, path: Path, file_id: str, metric_id: str,
|
||
|
|
name: str = "upload", size_bytes: int = 0) -> Any:
|
||
|
|
"""Metric data for a track, computed (and cached) if needed.
|
||
|
|
|
||
|
|
`path` is a spool the caller owns and will delete; this method only reads it.
|
||
|
|
"""
|
||
|
|
metric = METRICS.get(metric_id)
|
||
|
|
if metric is None:
|
||
|
|
raise KeyError(f"Unknown metric: {metric_id}")
|
||
|
|
|
||
|
|
with self._lock:
|
||
|
|
entry = self.files.get(file_id)
|
||
|
|
if entry is not None:
|
||
|
|
cached = entry.metric_data.get(metric_id)
|
||
|
|
if cached is not None:
|
||
|
|
return cached
|
||
|
|
|
||
|
|
t0 = time.perf_counter()
|
||
|
|
audio = AudioFile(str(path))
|
||
|
|
if entry is None:
|
||
|
|
# First contact with these bytes: snapshot it while we have them decoded.
|
||
|
|
entry = self.load(path, file_id, name, size_bytes)
|
||
|
|
data = metric.compute(audio)
|
||
|
|
elapsed = time.perf_counter() - t0
|
||
|
|
logger.info("%s on %s in %.2fs", metric.display_name,
|
||
|
|
entry.snapshot.song_name, elapsed)
|
||
|
|
|
||
|
|
with self._lock:
|
||
|
|
# Keep the first result if a concurrent request won the race — identical,
|
||
|
|
# because both hashed the same bytes.
|
||
|
|
entry.metric_data.setdefault(metric_id, data)
|
||
|
|
return entry.metric_data[metric_id]
|
||
|
|
|
||
|
|
def observe(self, path: Path, file_id: str, name: str, size_bytes: int = 0) -> SessionFile:
|
||
|
|
"""Decode just enough to register a track (used for metadata-only requests)."""
|
||
|
|
return self.load(path, file_id, name, size_bytes)
|
||
|
|
|
||
|
|
# ---- lifecycle ----------------------------------------------------------
|
||
|
|
|
||
|
|
def drop(self, file_id: str) -> bool:
|
||
|
|
"""Forget a track and everything computed for it.
|
||
|
|
|
||
|
|
Nothing to delete on disk: the audio was never persisted. The browser still
|
||
|
|
holds its copy, which is why this is the client's decision to make.
|
||
|
|
"""
|
||
|
|
with self._lock:
|
||
|
|
entry = self.files.pop(file_id, None)
|
||
|
|
if file_id in self.overlay:
|
||
|
|
self.overlay.remove(file_id)
|
||
|
|
return entry is not None
|
||
|
|
|
||
|
|
def clear(self) -> None:
|
||
|
|
with self._lock:
|
||
|
|
self.files.clear()
|
||
|
|
self.overlay.clear()
|
||
|
|
|
||
|
|
def _evict_locked(self) -> None:
|
||
|
|
"""Bound the cache so a long-lived server can't grow without limit.
|
||
|
|
|
||
|
|
Evicts whole tracks, oldest first. The client can always re-upload, so the
|
||
|
|
worst case is a re-decode rather than an error.
|
||
|
|
"""
|
||
|
|
while len(self.files) > self.max_files:
|
||
|
|
oldest = next(iter(self.files))
|
||
|
|
self.files.pop(oldest, None)
|
||
|
|
if oldest in self.overlay:
|
||
|
|
self.overlay.remove(oldest)
|
||
|
|
logger.info("Evicted %s from the analysis cache", oldest[:8])
|
||
|
|
|
||
|
|
# ---- view set -----------------------------------------------------------
|
||
|
|
|
||
|
|
def set_overlay(self, file_ids: list[str]) -> list[str]:
|
||
|
|
"""Replace the drawn set, keeping only ids this session has analysed."""
|
||
|
|
with self._lock:
|
||
|
|
self.overlay = [f for f in file_ids if f in self.files]
|
||
|
|
return list(self.overlay)
|
||
|
|
|
||
|
|
def toggle(self, file_id: str, checked: bool) -> list[str]:
|
||
|
|
with self._lock:
|
||
|
|
if file_id not in self.files:
|
||
|
|
return list(self.overlay)
|
||
|
|
if checked and file_id not in self.overlay:
|
||
|
|
self.overlay.append(file_id)
|
||
|
|
elif not checked and file_id in self.overlay:
|
||
|
|
self.overlay.remove(file_id)
|
||
|
|
return list(self.overlay)
|
||
|
|
|
||
|
|
def overlay_entries(self) -> list[SessionFile]:
|
||
|
|
with self._lock:
|
||
|
|
return [self.files[f] for f in self.overlay if f in self.files]
|
||
|
|
|
||
|
|
def entry(self, file_id: str) -> Optional[SessionFile]:
|
||
|
|
with self._lock:
|
||
|
|
return self.files.get(file_id)
|
||
|
|
|
||
|
|
def snapshot_size(self) -> int:
|
||
|
|
"""Number of tracks held in the analysis cache — surfaced in /api/session."""
|
||
|
|
with self._lock:
|
||
|
|
return len(self.files)
|
||
|
|
|
||
|
|
|
||
|
|
# One process-wide session. The server is single-user by design (it runs on a
|
||
|
|
# workstation and analyses files the user picked), so per-connection sessions
|
||
|
|
# would only duplicate an expensive decode of the same audio.
|
||
|
|
SESSION = AnalysisSession()
|
||
|
|
DEFAULT_METRIC = DEFAULT_METRIC_ID
|