uj-mastering-master/webapp.py

# SPDX-License-Identifier: Apache-2.0
"""
Web front end for the mastering toolkit.

Same backend as the desktop app — `master_core.AudioFile` for decode,
`metrics.METRICS[...].compute` for the numbers — with the PyQt5 layer replaced by
a JSON API and a browser canvas. `metrics.Metric` was already split into a
backend-neutral `compute` and a view-aware `build_spec`, so this costs no changes
to the analysis code; `metrics_json` is the new renderer seam.

Run it:

    uv run ujm-web                       # http://127.0.0.1:8000
    uv run ujm-web --host 0.0.0.0 --port 8080 --open

Design notes:

- **The browser holds the audio; the server holds only computed data.** The
  client keeps every uploaded track in OPFS and PUTs the bytes for each analysis
  request. The server spools the body to a temp file, decodes, computes, and
  deletes it inside a `finally` — there is no upload directory and no audio file
  survives a request. `web_session`'s docstring explains why the bytes still have
  to reach this process (librosa decodes by path) and what that does and doesn't
  guarantee. Concretely: hosting this does not accumulate the user's masters on
  the host.
- **Nothing is keyed on a client-supplied path.** `file_id` is the SHA-256 of the
  audio bytes, so there is no traversal surface and no filename collision; the
  only place a client string reaches the spool is its extension, which is matched
  against an allow-list.
- **Analysis runs in the event loop's threadpool, not in the loop.** Metric
  computes are CPU-bound numpy/scipy calls that do release the GIL; a plain `def`
  endpoint is dispatched to Starlette's threadpool, so a long spectrogram doesn't
  block status requests.
- **Bodies are read as a stream.** `await request.stream()` is written straight to
  the spool, so a 200 MB master never sits in RAM as one `bytes` object.
- **Nothing is served by a CDN.** The front end (uPlot for curves, a plain canvas
  for the spectrogram) is vendored under `static/` and the server binds loopback
  by default: mastering work-in-progress shouldn't leak out of the box.
- **`/api/plot` is a pure view render.** Lin/log, time mode and the reference-line
  set are query parameters, so a view change costs a JSON re-serialisation from
  the cache — never a recompute. That is the property that made the desktop app
  feel fast; the web UI keeps it.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import logging
import tempfile
from pathlib import Path
from typing import Any, Optional

from fastapi import Body, FastAPI, HTTPException, Query, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles

from logger_setup import setup_logging
from metrics import METRICS, DEFAULT_METRIC_ID
from metrics_json import serialize_ref_lines, serialize_spec, serialize_view
from plotspec import RefLineProps, ViewState, X_ABSOLUTE, X_RELATIVE, apply_x_mode
from web_session import AUDIO_SUFFIXES, SESSION, DEFAULT_METRIC, file_id_for

STATIC_DIR = Path(__file__).parent / "static"

logger = logging.getLogger(__name__)

app = FastAPI(
  title="uj-mastering-master (web)",
  description="Audio mastering analysis metrics over HTTP.",
  version="0.1.0",
)


# ---- request helpers -------------------------------------------------------

def _query(request: Request, key: str) -> Optional[str]:
  """Read a query parameter.

  FastAPI's `Query` only populates declared signature parameters, and these
  endpoints take a raw body rather than a parameter list, so the two values the
  client sends alongside the audio are pulled off the URL directly.
  """
  value = request.query_params.get(key)
  return value or None


def _suffix_or_400(name: str) -> str:
  suffix = Path(name or "").suffix.lower()
  if suffix not in AUDIO_SUFFIXES:
    raise HTTPException(
      status_code=400,
      detail=f"Unsupported format '{suffix or '?'}' "
             f"(expected one of {', '.join(sorted(AUDIO_SUFFIXES))})",
    )
  return suffix


def _require_file(file_id: str):
  entry = SESSION.entry(file_id)
  if entry is None:
    raise HTTPException(status_code=404, detail="Track not analysed in this session")
  return entry


def _metric_or_400(metric: Optional[str]) -> str:
  uid = metric or DEFAULT_METRIC_ID
  if uid not in METRICS:
    raise HTTPException(status_code=400, detail=f"Unknown metric: {uid}")
  return uid


# ---- catalogue & session ---------------------------------------------------

def _metric_catalogue() -> list[dict]:
  return [
    {
      "id": m.id,
      "display_name": m.display_name,
      "default": m.id == DEFAULT_METRIC_ID,
    }
    for m in METRICS.values()
  ]


@app.get("/api/metrics")
def list_metrics():
  return {"metrics": _metric_catalogue(), "default": DEFAULT_METRIC_ID}


@app.get("/api/session")
def session_state():
  return {
    "files": [f.metadata() for f in SESSION.files.values()],
    "overlay": list(SESSION.overlay),
    "default_metric": DEFAULT_METRIC_ID,
    # Explicit, so the front end can state it in the UI and a reader of the API
    # can confirm there is no audio store behind this server.
    "audio_storage": "client (browser OPFS) — nothing persisted server-side",
  }


# ---- analysis --------------------------------------------------------------

@app.post("/api/analyse")
async def analyse(
  request: Request,
  file_id: Optional[str] = Query(None, description="SHA-256 of the body; computed if absent"),
  name: str = Query("upload", description="Original filename, for display only"),
  metric: Optional[str] = Query(None),
):
  """Decode and measure one track, sent as a raw body.

  Returns the snapshot plus the serialised metric payload in a single round trip:
  the client has to send the bytes either way, so making it then ask again for the
  plot would upload the same file twice.

  Declared `async` so the body can be streamed to the spool with `await`; the
  decode/compute that follows is a blocking call, so it is handed to the
  threadpool explicitly to keep the event loop free.
  """
  uid = _metric_or_400(metric)
  suffix = _suffix_or_400(name)
  display_name = Path(name).name

  spool = tempfile.NamedTemporaryFile(prefix="ujm-", suffix=suffix, delete=False)
  path = Path(spool.name)
  try:
    size = 0
    with spool:
      async for chunk in request.stream():
        spool.write(chunk)
        size += len(chunk)
    if size == 0:
      raise HTTPException(status_code=400, detail="Empty request body")

    fid = file_id or await run_in_threadpool(_hash_file, path)

    def _work() -> tuple[Any, Any]:
      entry = SESSION.load(path, fid, display_name, size)
      data = SESSION.compute_metric(path, fid, uid, display_name, size)
      spec = METRICS[uid].build_spec(data, ViewState())
      return entry, spec

    entry, spec = await run_in_threadpool(_work)
  finally:
    # The guarantee this design makes: no audio outlives the request.
    path.unlink(missing_ok=True)

  payload = serialize_spec(spec)
  return {
    "file": entry.metadata(),
    "metric": uid,
    "metric_name": METRICS[uid].display_name,
    "payload": payload,
    "axes": payload["axes"],
    "overlay": list(SESSION.overlay),
  }


def _hash_file(path: Path) -> str:
  """SHA-256 of a spooled body, read in chunks."""
  digest = hashlib.sha256()
  with path.open("rb") as handle:
    for block in iter(lambda: handle.read(1024 * 1024), b""):
      digest.update(block)
  return digest.hexdigest()[:32]


# ---- plotting --------------------------------------------------------------

@app.get("/api/metadata")
def metadata(file_id: str):
  return _require_file(file_id).metadata()


@app.post("/api/file")
async def analyse_extra(request: Request, file_id: Optional[str] = Query(None),
                        name: str = Query("upload")):
  """Register a track without plotting it.

  Used when the client wants the file on the list (duration, amplitudes, tags)
  for a metric that is already drawn, so the plot endpoint doesn't need to
  re-upload it.
  """
  suffix = _suffix_or_400(name)
  display_name = Path(name).name

  spool = tempfile.NamedTemporaryFile(prefix="ujm-", suffix=suffix, delete=False)
  path = Path(spool.name)
  try:
    size = 0
    with spool:
      async for chunk in request.stream():
        spool.write(chunk)
        size += len(chunk)
    if size == 0:
      raise HTTPException(status_code=400, detail="Empty request body")

    fid = file_id or await run_in_threadpool(_hash_file, path)
    entry = await run_in_threadpool(SESSION.observe, path, fid, display_name, size)
  finally:
    path.unlink(missing_ok=True)

  return {"file": entry.metadata(), "overlay": list(SESSION.overlay)}


@app.post("/api/overlay")
def set_overlay(payload: dict = Body(...)):
  ids = payload.get("file_ids")
  if not isinstance(ids, list):
    raise HTTPException(status_code=400, detail="'file_ids' must be a list")
  return {"overlay": SESSION.set_overlay([str(i) for i in ids])}


@app.delete("/api/file")
def delete_file(file_id: str):
  if not SESSION.drop(file_id):
    raise HTTPException(status_code=404, detail="Track not analysed in this session")
  return {"overlay": list(SESSION.overlay),
          "note": "server forgot the analysis; the browser still holds the audio"}


@app.get("/api/plot")
def plot(
  metric: str = Query(DEFAULT_METRIC),
  y_log: Optional[bool] = Query(None),
  x_log: Optional[bool] = Query(None),
  x_mode: str = Query(X_ABSOLUTE),
  refs: str = Query("", description="JSON list of reference lines"),
):
  """Draw the current overlay of `metric`, honouring view state.

  Pure view render: every dataset here was already computed by `/api/analyse`, so
  a lin/log flip or a time-mode change never touches numpy. Returns one
  serialised dataset per drawn file plus the shared axis description, mirroring
  the desktop `show_specs([(label, PlotSpec, colour), ...])`.
  """
  if metric not in METRICS:
    raise HTTPException(status_code=400, detail=f"Unknown metric: {metric}")
  if x_mode not in (X_ABSOLUTE, X_RELATIVE):
    raise HTTPException(status_code=400, detail=f"Unknown x_mode: {x_mode}")

  view = ViewState(y_log=y_log, x_log=x_log, x_mode=x_mode)
  ref_lines = _parse_refs(refs)

  entries = SESSION.overlay_entries()
  if not entries:
    return {
      "metric": metric, "metric_name": METRICS[metric].display_name,
      "view": serialize_view(view), "refs": serialize_ref_lines(ref_lines),
      "datasets": [], "axes": None, "warnings": [],
    }

  datasets, warnings = [], []
  for index, entry in enumerate(entries):
    data = entry.metric_data.get(metric)
    if data is None:
      # Never analysed this metric for this track: the client has the bytes and
      # knows to send them. Say so rather than recomputing from nothing.
      warnings.append(f"{entry.snapshot.song_name}: {METRICS[metric].display_name} "
                      f"not computed yet")
      continue
    try:
      spec = METRICS[metric].build_spec(data, view)
    except Exception as exc:
      logger.exception("build_spec failed for %s", metric)
      warnings.append(f"{entry.snapshot.song_name}: {exc}")
      continue

    if view.x_mode == X_RELATIVE:
      spec = apply_x_mode(spec, view.x_mode)

    datasets.append({
      "file_id": entry.snapshot.file_id,
      "label": entry.snapshot.song_name,
      "index": index,
      "payload": serialize_spec(spec),
      # The client assigns colours; it keeps a song's colour stable across
      # metric switches the same way the desktop palette-by-row does.
      "palette_index": _palette_index(entry.snapshot.file_id),
    })

  # Heatmaps don't overlay: only the first drawn dataset is rendered, as on the
  # desktop, so the browser isn't sent z-grids it will throw away.
  if datasets and datasets[0]["payload"]["heatmap"] is not None and len(datasets) > 1:
    warnings.append(
      f"{METRICS[metric].display_name}: one track at a time — "
      f"showing {datasets[0]['label']}"
    )
    datasets = datasets[:1]

  return {
    "metric": metric,
    "metric_name": METRICS[metric].display_name,
    "view": serialize_view(view),
    "refs": serialize_ref_lines(ref_lines),
    "datasets": datasets,
    "axes": datasets[0]["payload"]["axes"] if datasets else None,
    "warnings": warnings,
    "recompute_free": True,
  }


def _palette_index(file_id: str) -> int:
  """Stable palette slot for a track, by first-seen order in the session."""
  order = list(SESSION.files.keys())
  try:
    return order.index(file_id)
  except ValueError:
    return 0


def _parse_refs(raw: str) -> list[RefLineProps]:
  if not raw:
    return []
  try:
    items = json.loads(raw)
  except ValueError:
    raise HTTPException(status_code=400, detail="refs must be JSON")
  if not isinstance(items, list):
    raise HTTPException(status_code=400, detail="refs must be a JSON list")
  lines = []
  for item in items:
    if not isinstance(item, dict) or "value" not in item:
      continue
    try:
      value = float(item["value"])
    except (TypeError, ValueError):
      continue
    lines.append(RefLineProps(
      value=value,
      color=str(item.get("color", "#444444")),
      style=str(item.get("style", "dash")),
      label=str(item.get("label", "")),
    ))
  return lines


# ---- front end -------------------------------------------------------------

@app.get("/")
def index():
  index_file = STATIC_DIR / "index.html"
  if not index_file.exists():
    raise HTTPException(status_code=500, detail="Front end assets missing")
  return FileResponse(index_file)


if STATIC_DIR.exists():
  app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")


@app.exception_handler(Exception)
async def unhandled(request, exc):  # pragma: no cover - safety net
  logger.exception("Unhandled error on %s", request.url.path)
  return JSONResponse(status_code=500, content={"detail": str(exc)})


# ---- entry point -----------------------------------------------------------

# Our CLI speaks `logger_setup`'s vocabulary (ERROR/WARN/INFO/DEBUG/TRACE);
# uvicorn only accepts its own names. Translate rather than passing through —
# `--log-level WARN` used to die with KeyError: 'warn'.
_UVICORN_LEVEL = {
  "ERROR": "error",
  "WARN": "warning",
  "INFO": "info",
  "DEBUG": "debug",
  "TRACE": "debug",
}


def main() -> None:
  import uvicorn

  parser = argparse.ArgumentParser(
    prog="ujm-web",
    description="Mastering analysis toolkit — web front end",
  )
  parser.add_argument("--host", default="127.0.0.1",
                      help="interface to bind (default: %(default)s — loopback)")
  parser.add_argument("--port", type=int, default=8000)
  parser.add_argument("--open", action="store_true",
                      help="open the UI in a browser on startup")
  parser.add_argument("--reload", action="store_true", help="auto-reload on edits")
  parser.add_argument("--log-level", "-l", default="INFO",
                      choices=["ERROR", "WARN", "INFO", "DEBUG", "TRACE"])
  parser.add_argument("--log-file", action="store_true",
                      help="also write audio_analysis.log")
  args = parser.parse_args()

  setup_logging(level=args.log_level, log_to_file=args.log_file)
  url = f"http://{args.host if args.host != '0.0.0.0' else '127.0.0.1'}:{args.port}"
  logger.info("Web UI at %s", url)
  logger.info("Audio is stored in the browser (OPFS); "
              "uploads are spooled to temp and deleted after each request")

  if args.open:
    import threading
    import webbrowser
    threading.Timer(1.0, lambda: webbrowser.open(url)).start()

  uvicorn.run("webapp:app", host=args.host, port=args.port, reload=args.reload,
              log_level=_UVICORN_LEVEL[args.log_level])


if __name__ == "__main__":
  main()