uj-mastering-master/web_session.py

# 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