Commit b06cba91 by PLN (Algolia)

Merge remote-tracking branch 'origin/master'

parents 29ce99d3 328bca59
......@@ -33,6 +33,15 @@ docker build -t fourier:latest armada/api # ~1 GB image, light engine only
(Or build on the dev box and `docker save fourier:latest | ssh erable docker load`.)
> **Base-image drift (bit us 2026-07-23).** A clean rebuild re-pulls
> `python:3.12-slim`, which now tracks Debian **trixie** — and its apt layer
> fails in erable's old Docker (`APT::Update::Post-Invoke … Sub-process returned
> an error`). Two fixes: **pin the base** to a working digest
> (`FROM python:3.12-slim@sha256:…`) so rebuilds are reproducible — the proper
> fix; or, for a **source-only** change (no dep change), skip the base entirely
> and overlay onto the running image:
> `printf 'FROM fourier:latest\nCOPY . /app\n' | docker build -t fourier:latest -f - .`
## 2. State volume
One writable dir at `/data` holds the content-addressed **cache** (transient,
......@@ -67,18 +76,29 @@ MKL_NUM_THREADS=1
## 4. Run
Use the launcher — it bakes in every flag idempotently (rm + run), including the
**memory cap** that a hand-typed `docker run` used to omit:
```bash
docker run -d --name fourier --restart unless-stopped \
--security-opt seccomp=unconfined \
--env-file ~/.config/fourier/fourier.env \
-v /home/pln/srv/fourier/data:/data \
-p 127.0.0.1:9780:9780 \
fourier:latest
deploy/fourier.sh up # (re)launches BOTH the API and the worker container
deploy/fourier.sh ps # status + effective mem caps
```
`-p 127.0.0.1:9780:9780` is the whole security model: loopback-only means the
gateway is the *only* way in, so the injected `X-Tenant`/`X-Scopes` headers can be
trusted. SRE wraps this in a keep-alive systemd unit.
It launches two containers off the same image: **`fourier`** (uvicorn, the HTTP
API) and **`fourier-worker`** (`python -m worker`, the heavy-engine job runner).
Both get `--restart unless-stopped`, `--security-opt seccomp=unconfined`, the
`--env-file`, the `/data` volume, and `--memory` (default `3g` each — override
via `FOURIER_MEM` / `FOURIER_API_MEM` / `FOURIER_WORKER_MEM`).
**Why the cap matters:** erable is a shared ~7.7 GB box (Postgres + the platform
live there too). Uncapped, a runaway separate/sources job could OOM its
neighbours. `docker update --memory` only survives a *restart*, not a rm+run
redeploy — the script is the durable home for the cap, so redeploys keep it.
The API publishes `-p 127.0.0.1:9780:9780`; loopback-only is the whole security
model — the gateway is the *only* way in, so the injected `X-Tenant`/`X-Scopes`
headers can be trusted. uvicorn binds `0.0.0.0` *inside* the container (the
loopback restriction is the host-side `-p`). The worker publishes no ports.
### Two erable gotchas (verified at first deploy, 2026-06-28)
......
......@@ -23,7 +23,7 @@ import cache
import config
import jobs as jobslib
import scopes as scopelib
from engines import ears, feats, signal
from engines import ears, feats, signal, streaming
from engines import loudness as loudness_eng
from engines import grade as grade_eng # VENDORED copy of the Foundry grader (see engines/grade.py)
from engines import naming as naming_eng # VENDORED copy of the Foundry namer (see engines/naming.py)
......@@ -278,57 +278,101 @@ def get_artifact(content_id: str, name: str,
# ── shared analyze flow: upload → content-address → cache → compute (#26) ───────
async def _read_upload(file: UploadFile) -> bytes:
data = await file.read()
# SAFETY (#7): the upload is STREAMED to a temp file in bounded chunks — never
# `await file.read()` into one big `bytes` (a 100 MB body would spike ~100 MB
# resident, and the content-hash + engines add their own copies). Peak memory
# stays O(chunk), so a large upload can't OOM the shared erable box. The size
# ceiling is enforced mid-stream (read at request time, so tests/ops can retune
# config.MAX_UPLOAD_MB live).
async def _spool(file: UploadFile) -> Path:
limit = config.MAX_UPLOAD_MB * 1024 * 1024
if len(data) > limit:
raise HTTPException(413, f"file too large (> {config.MAX_UPLOAD_MB} MB)")
return data
def _cached(kind: str, params: dict, data: bytes, filename: str, compute, response: Response):
"""Content-address the bytes, serve a cache hit instantly (X-Nech-Cache:
hit), else run `compute(path)` and cache it. Returns (content_id, result).
The same engine call is paid ONCE per (content_id, params) — the margin lever."""
suffix = Path(filename or "clip").suffix or ".wav"
with tempfile.NamedTemporaryFile(suffix=suffix, delete=True) as tmp:
tmp.write(data)
suffix = Path(file.filename or "clip").suffix or ".wav"
tmp = tempfile.NamedTemporaryFile(suffix=suffix, delete=False)
path = Path(tmp.name)
size = 0
try:
while chunk := await file.read(1 << 20): # 1 MB at a time
size += len(chunk)
if size > limit:
raise HTTPException(413, {
"error": "payload_too_large",
"detail": f"upload exceeds the {config.MAX_UPLOAD_MB} MB limit for this API",
"docs": "https://api.nech.pl/docs",
})
tmp.write(chunk)
tmp.flush()
cid = cache.content_id(tmp.name)
hit = cache.get(cid, kind, params)
if hit:
response.headers["X-Nech-Cache"] = "hit"
return cid, hit["result"]
try:
result = compute(tmp.name)
except Exception as e:
raise HTTPException(422, f"{kind} analysis failed: {e}")
except BaseException:
tmp.close(); path.unlink(missing_ok=True); raise
else:
tmp.close()
return path
def _source_meta(path: Path) -> dict:
"""Honest provenance for the response: the file's TRUE duration + size, so a
caller always knows the full extent of what they sent — and, paired with an
engine's own `duration_s` / `full_track`, whether it was analyzed whole or
(for the clip-tier engines) only its head."""
meta = {"size_bytes": path.stat().st_size}
try:
p = streaming.probe(str(path))
meta.update(duration_s=round(p["duration_s"], 3), sr=p["sr"], channels=p["channels"])
except Exception:
pass
return meta
def _cached(kind: str, params: dict, path: str, compute, response: Response):
"""Content-address the file, serve a cache hit instantly (X-Nech-Cache: hit),
else run `compute(path)` and cache it. Returns (content_id, result). The same
engine call is paid ONCE per (content_id, params) — the margin lever."""
cid = cache.content_id(path)
hit = cache.get(cid, kind, params)
if hit:
response.headers["X-Nech-Cache"] = "hit"
return cid, hit["result"]
try:
result = compute(path)
except Exception as e:
raise HTTPException(422, f"{kind} analysis failed: {e}")
cache.put(cid, kind, result=result, params=params)
response.headers["X-Nech-Cache"] = "miss"
return cid, result
async def _run(file: UploadFile, kind: str, params: dict, compute, response: Response):
"""Spool the upload (bounded), content-address + cache + compute, always
cleaning up the temp file. Returns (content_id, result, source_meta)."""
path = await _spool(file)
try:
source = _source_meta(path)
cid, result = _cached(kind, params, str(path), compute, response)
return cid, result, source
finally:
path.unlink(missing_ok=True)
@v1.post("/analyze/emotion", operation_id="analyzeEmotion")
async def analyze_emotion(response: Response, file: UploadFile = File(...),
_p: auth.Principal = Depends(require)):
"""Upload a short audio clip → valence/arousal + top emotions. Cached (#26):
the same audio returns instantly on a repeat instead of recomputing."""
data = await _read_upload(file)
params = {"engine": config.EMOTION_ENGINE}
cid, result = _cached("emotion", params, data, file.filename, ears.emotion_read, response)
return {"filename": file.filename, "content_id": cid, "emotion": result}
the same audio returns instantly on a repeat instead of recomputing.
Clip-tier: the emotion engine reads the first ~60 s (`source.duration_s`
reports the full length)."""
cid, result, source = await _run(file, "emotion", {"engine": config.EMOTION_ENGINE},
ears.emotion_read, response)
return {"filename": file.filename, "content_id": cid, "emotion": result, "source": source}
@v1.post("/features", operation_id="features")
async def features(response: Response, file: UploadFile = File(...), rhythm: bool = True,
_p: auth.Principal = Depends(require)):
"""Upload a clip → the ~35-dim audio feature stack (spectral moments, MFCCs,
chroma/key, envelope, + rhythm when `rhythm=true`). Torch-free, cached."""
data = await _read_upload(file)
params = {"engine": "light", "rhythm": rhythm}
cid, result = _cached("features", params, data, file.filename,
lambda p: feats.features(p, rhythm=rhythm), response)
return {"filename": file.filename, "content_id": cid, **result}
chroma/key, envelope, + rhythm when `rhythm=true`). Torch-free, cached.
Clip-tier: analyses the first ~30 s (`source.duration_s` is the full length)."""
cid, result, source = await _run(file, "features", {"engine": "light", "rhythm": rhythm},
lambda p: feats.features(p, rhythm=rhythm), response)
return {"filename": file.filename, "content_id": cid, **result, "source": source}
@v1.post("/analyze/samples", operation_id="analyzeSamples")
......@@ -336,49 +380,47 @@ async def analyze_samples(response: Response, file: UploadFile = File(...), name
_p: auth.Principal = Depends(require)):
"""Upload a one-shot/loop → per-sample EDA + MEASURED role (percs|bass|melodic|
tops|atmos) from the spectrum, never the name. Optional `name` only
disambiguates breaks/drums. Torch-free, cached."""
data = await _read_upload(file)
params = {"engine": "light", "name": name}
cid, result = _cached("samples", params, data, file.filename,
lambda p: feats.sample_profile(p, name=name), response)
return {"filename": file.filename, "content_id": cid, **result}
disambiguates breaks/drums. Torch-free, cached. Clip-tier: first ~30 s."""
cid, result, source = await _run(file, "samples", {"engine": "light", "name": name},
lambda p: feats.sample_profile(p, name=name), response)
return {"filename": file.filename, "content_id": cid, **result, "source": source}
@v1.post("/onsets", operation_id="onsets")
async def onsets(response: Response, file: UploadFile = File(...),
_p: auth.Principal = Depends(require)):
"""Upload audio → onset hit times (s) + tempo + onset rate. The cheapest tier
(no model): rhythmic hits for visual sync / slicing. Torch-free, cached."""
data = await _read_upload(file)
cid, result = _cached("onsets", {"v": 1}, data, file.filename, signal.onsets, response)
return {"filename": file.filename, "content_id": cid, **result}
(no model): rhythmic hits for visual sync / slicing. Torch-free, cached.
Full-track: streams the WHOLE upload (`full_track:true`), not just a head clip."""
cid, result, source = await _run(file, "onsets", {"v": 2}, signal.onsets, response)
return {"filename": file.filename, "content_id": cid, **result, "source": source}
@v1.post("/waveform", operation_id="waveform")
async def waveform(response: Response, file: UploadFile = File(...), bins: int = 800,
_p: auth.Principal = Depends(require)):
"""Upload audio → render-ready waveform: per-bin [min,max] + a 0..1 RMS energy
envelope (`?bins=`, ≤4000). For audio-reactive visuals. Torch-free, cached."""
data = await _read_upload(file)
cid, result = _cached("waveform", {"bins": bins, "v": 1}, data, file.filename,
lambda p: signal.waveform(p, bins=bins), response)
return {"filename": file.filename, "content_id": cid, **result}
envelope (`?bins=`, ≤4000). For audio-reactive visuals. Torch-free, cached.
Full-track: bins span the WHOLE track via an exact streaming reduction."""
cid, result, source = await _run(file, "waveform", {"bins": bins, "v": 2},
lambda p: signal.waveform(p, bins=bins), response)
return {"filename": file.filename, "content_id": cid, **result, "source": source}
@v1.post("/analyze", operation_id="analyzeAll")
async def analyze_all(response: Response, file: UploadFile = File(...),
_p: auth.Principal = Depends(require)):
"""One call → emotion + features + sample role for a clip (fewer round-trips
for hexa). Each is the same engine the dedicated routes use. Cached as a unit."""
data = await _read_upload(file)
for hexa). Each is the same engine the dedicated routes use. Cached as a unit.
Clip-tier composite: emotion ~60 s, features/sample ~30 s of the head."""
params = {"engine": config.EMOTION_ENGINE, "v": 1}
def _compute(p):
return {"emotion": ears.emotion_read(p),
"features": feats.features(p, rhythm=True)["features"],
"sample": feats.sample_profile(p)}
cid, result = _cached("analyze", params, data, file.filename, _compute, response)
return {"filename": file.filename, "content_id": cid, **result}
cid, result, source = await _run(file, "analyze", params, _compute, response)
return {"filename": file.filename, "content_id": cid, **result, "source": source}
@v1.post("/grade", operation_id="grade")
......@@ -388,22 +430,22 @@ async def grade(response: Response, file: UploadFile = File(...), role: str = ""
0..1 + S/A/B/C/D tier, per-rule sub-scores (seam click, zero-crossing
cleanliness, DC, bar self-consistency, level, bass mono-compat) + flags.
`role` ('bass'|'drums'|…) is a hint. WAV/FLAC preferred. Torch-free, cached."""
data = await _read_upload(file)
params = {"role": role or None, "v": 1}
cid, result = _cached("grade", params, data, file.filename,
lambda p: grade_eng.grade(p, role=role or None).model_dump(), response)
return {"filename": file.filename, "content_id": cid, "grade": result}
cid, result, source = await _run(file, "grade", {"role": role or None, "v": 1},
lambda p: grade_eng.grade(p, role=role or None).model_dump(),
response)
return {"filename": file.filename, "content_id": cid, "grade": result, "source": source}
@v1.post("/loudness", operation_id="loudness")
async def loudness(response: Response, file: UploadFile = File(...),
_p: auth.Principal = Depends(require)):
"""Upload audio → BS.1770 integrated LUFS + sample/true peak + crest + the
gain (dB) to hit each delivery target (-14 streaming, -9 club). Torch-free, cached."""
data = await _read_upload(file)
cid, result = _cached("loudness", {"v": 1}, data, file.filename,
loudness_eng.loudness, response)
return {"filename": file.filename, "content_id": cid, **result}
gain (dB) to hit each delivery target (-14 streaming, -9 club). Torch-free, cached.
Full-track: peak/true-peak/RMS are exact; integrated LUFS is chunked-approximate
(`approximate:true`) — pyloudnorm gating can't merge chunks exactly."""
cid, result, source = await _run(file, "loudness", {"v": 2},
loudness_eng.loudness, response)
return {"filename": file.filename, "content_id": cid, **result, "source": source}
@v1.post("/spectrum", operation_id="spectrum")
......@@ -412,22 +454,21 @@ async def spectrum(response: Response, file: UploadFile = File(...),
_p: auth.Principal = Depends(require)):
"""FFT-as-a-service: a downsampled, render-ready spectrogram (`bands`×`frames`,
0..1). `mel=` perceptual vs log-linear; `frames=1` → a single averaged FFT
spectrum. Bounded payload → caches cheaply. Torch-free."""
data = await _read_upload(file)
params = {"bands": bands, "frames": frames, "mel": mel, "v": 1}
cid, result = _cached("spectrum", params, data, file.filename,
lambda p: signal.spectrum(p, bands=bands, frames=frames, mel=mel),
response)
return {"filename": file.filename, "content_id": cid, **result}
spectrum. Bounded payload → caches cheaply. Torch-free. Full-track: frames span
the WHOLE track (streamed windows binned into the global time grid)."""
params = {"bands": bands, "frames": frames, "mel": mel, "v": 2}
cid, result, source = await _run(file, "spectrum", params,
lambda p: signal.spectrum(p, bands=bands, frames=frames, mel=mel),
response)
return {"filename": file.filename, "content_id": cid, **result, "source": source}
@v1.post("/naming", operation_id="naming")
async def naming(response: Response, file: UploadFile = File(...), index: int = 1,
_p: auth.Principal = Depends(require)):
"""Upload a sample → a convention-compliant name (NN_role_character) derived
from the MEASURED role + character, not the file name. Torch-free, cached."""
data = await _read_upload(file)
from the MEASURED role + character, not the file name. Torch-free, cached.
Clip-tier: first ~30 s."""
def _compute(p):
prof = feats.sample_profile(p)
ff = feats.features(p, rhythm=False)["features"]
......@@ -435,8 +476,8 @@ async def naming(response: Response, file: UploadFile = File(...), index: int =
suggested = naming_eng.suggest_name(index, prof["role"], character=character)
return {"suggested_name": suggested, "role": prof["role"], "character": character,
"stem": naming_eng.stem_of(prof["role"]), "lint": naming_eng.lint(suggested)}
cid, result = _cached("naming", {"index": index, "v": 1}, data, file.filename, _compute, response)
return {"filename": file.filename, "content_id": cid, **result}
cid, result, source = await _run(file, "naming", {"index": index, "v": 1}, _compute, response)
return {"filename": file.filename, "content_id": cid, **result, "source": source}
# ── #37 separation seam — no CPU path on erable; env-selected GPU backend later ──
......
......@@ -21,18 +21,27 @@ import db
# ── addressing ──────────────────────────────────────────────────────────────
def content_id(audio_path: str | Path) -> str:
"""sha256 of decoded PCM (+ samplerate). Falls back to raw file bytes if the
format can't be decoded here (still dedupes identical files)."""
format can't be decoded here (still dedupes identical files).
STREAMED (#7): hashes the decoded PCM block-by-block instead of reading the
whole file into RAM — the concatenated float32 bytes are identical to the old
whole-array hash, so existing cache ids stay valid. Bounded memory either way."""
try:
import numpy as np
import soundfile as sf
y, sr = sf.read(str(audio_path), dtype="float32", always_2d=False)
h = hashlib.sha256()
h.update(np.ascontiguousarray(y).tobytes())
with sf.SoundFile(str(audio_path)) as f:
sr = f.samplerate
for block in f.blocks(blocksize=1 << 18, dtype="float32", always_2d=False):
h.update(np.ascontiguousarray(block).tobytes())
h.update(str(sr).encode())
return "cid_" + h.hexdigest()[:32]
except Exception:
raw = Path(audio_path).read_bytes()
return "raw_" + hashlib.sha256(raw).hexdigest()[:32]
h = hashlib.sha256()
with open(audio_path, "rb") as fp:
for chunk in iter(lambda: fp.read(1 << 20), b""):
h.update(chunk)
return "raw_" + h.hexdigest()[:32]
def params_hash(params: dict | None) -> str:
......
......@@ -42,7 +42,11 @@ DB_PATH = Path(os.environ.get("FOURIER_DB", HERE / "_cache" / "fourier.db"))
DEV_TOKEN = os.environ.get("FOURIER_DEV_TOKEN", "")
# ── limits ─────────────────────────────────────────────────────────────────────
MAX_UPLOAD_MB = int(os.environ.get("FOURIER_MAX_UPLOAD_MB", "40"))
# Match the gateway's client_max_body_size (100 MB) so the app never rejects a
# body nginx let through. Safe to accept large uploads now: ingest + hashing +
# the full-track engines all stream in bounded windows (#7), so peak memory is
# O(window), not O(file). The gateway's friendly 413 fires first above this.
MAX_UPLOAD_MB = int(os.environ.get("FOURIER_MAX_UPLOAD_MB", "100"))
# ── artifact store (#27) — local disk on erable; freebox deferred ───────────────
# Big binaries (fetched sources, stems, loops) live content-addressed under here,
......
# Fourier job worker (#25) — systemd --user unit.
#
# SUPERSEDED on erable (2026-07): the worker now runs CONTAINERIZED off the same
# fourier:latest image, launched by `deploy/fourier.sh up` alongside the API (see
# DEPLOY.md §4). This host-venv unit remains as the paved-road alternative for a
# deploy host that runs the worker outside Docker; on erable it is NOT installed.
#
# The worker is a SEPARATE long-lived process from the API container: it claims
# pending jobs and runs the heavy engines (separate/sources/loops). Runs as a
# systemd --user service with lingering on, so it survives logout and starts at
......
#!/usr/bin/env bash
# Idempotent launcher for the Fourier (audio sub-API) containers on erable.
#
# Replaces the hand-typed `docker run` from DEPLOY.md §4 so that redeploys
# reproduce ALL the flags — most importantly the memory cap, which a bare
# `docker run` used to omit (a runaway job could then OOM the DB/platform on
# the shared 7.7 GB box). `docker update` only survives a restart, NOT a
# rm+run redeploy; this script is the durable home for the cap.
#
# Usage:
# deploy/fourier.sh up # (re)launch both containers (default)
# deploy/fourier.sh restart # same as up (rm + run is our restart)
# deploy/fourier.sh down # stop + remove both
# deploy/fourier.sh ps # show status + effective mem caps
#
# Everything is overridable by env; defaults match the live 2026-07 config.
set -euo pipefail
IMAGE="${FOURIER_IMAGE:-fourier:latest}"
DATA_DIR="${FOURIER_DATA_DIR:-$HOME/srv/fourier/data}"
ENV_FILE="${FOURIER_ENV_FILE:-$HOME/.config/fourier/fourier.env}"
BIND="${FOURIER_BIND:-127.0.0.1}" # loopback-only: the gateway is the only way in
PORT="${FOURIER_PORT:-9780}"
# Memory caps — the whole point of this script existing.
# The box has ~7.7 GB total shared with Postgres + the platform; keep the two
# audio containers bounded so a heavy job can't OOM its neighbours. Worker runs
# the heavy engines (separate/sources), API is CPU-light — but we cap both at
# 3g to match the live config. Override per-container if you retune.
MEM="${FOURIER_MEM:-3g}"
API_MEM="${FOURIER_API_MEM:-$MEM}"
WORKER_MEM="${FOURIER_WORKER_MEM:-$MEM}"
# Flags shared by both containers (see DEPLOY.md §4 for the why of each).
common_flags=(
--restart unless-stopped
--security-opt seccomp=unconfined # kernel 4.9 / clone3 — REQUIRED, see DEPLOY.md gotcha #1
--env-file "$ENV_FILE"
-v "$DATA_DIR:/data"
)
preflight() {
[[ -f "$ENV_FILE" ]] || { echo "!! env file missing: $ENV_FILE (see DEPLOY.md §3)" >&2; exit 1; }
mkdir -p "$DATA_DIR"
# Guard the shared box: refuse to launch if the disk is already tight.
local pct; pct=$(df --output=pcent / | tail -1 | tr -dc '0-9')
if (( pct >= 90 )); then
echo "!! root fs at ${pct}% — refusing to launch (free space first; see disk-monitor)" >&2
exit 1
fi
}
down() {
docker rm -f fourier fourier-worker 2>/dev/null || true
}
up() {
preflight
down
# API: serves HTTP on loopback; uvicorn binds 0.0.0.0 INSIDE (loopback is host-side -p).
docker run -d --name fourier "${common_flags[@]}" \
--memory "$API_MEM" \
-p "$BIND:$PORT:$PORT" \
"$IMAGE" \
uvicorn app:app --host 0.0.0.0 --port "$PORT"
# Worker: no ports; claims jobs and runs the heavy engines.
docker run -d --name fourier-worker "${common_flags[@]}" \
--memory "$WORKER_MEM" \
"$IMAGE" \
python -m worker
echo "launched fourier (mem=$API_MEM) + fourier-worker (mem=$WORKER_MEM)"
ps
}
ps() {
docker ps --filter name=fourier --format 'table {{.Names}}\t{{.Status}}'
for c in fourier fourier-worker; do
printf '%s mem-cap: ' "$c"
docker inspect -f '{{.HostConfig.Memory}} bytes' "$c" 2>/dev/null || echo '(not running)'
done
}
case "${1:-up}" in
up|restart) up ;;
down) down; echo "removed fourier + fourier-worker" ;;
ps|status) ps ;;
*) echo "usage: $0 {up|restart|down|ps}" >&2; exit 2 ;;
esac
......@@ -5,11 +5,20 @@ The mastering numbers PLN + hexa actually gate on (reference_postprod_master:
integrated loudness; true-peak is a cheap 4× oversample (catches inter-sample
peaks a raw sample-peak misses). Returns gain-to-target so a caller knows exactly
how much to push for each delivery target.
FULL-TRACK (#7): streams the WHOLE track in bounded windows. Peak / true-peak /
RMS / crest are EXACT (running max / accumulated sum-of-squares). Integrated LUFS
is energy-weighted across per-window meters — pyloudnorm's gating can't be merged
exactly across chunks, so this is a close APPROXIMATION, flagged `approximate:true`
with `lufs_method:"chunked"`. For a steady master the error is well under ~0.5 LU.
"""
import numpy as np
from . import streaming
_EPS = 1e-12
TARGETS = {"streaming": -14.0, "club": -9.0} # LUFS, reference_postprod_master
_METER_MIN_S = 0.4 # pyloudnorm needs ≥ one 400 ms gated block
def _db(x):
......@@ -26,37 +35,65 @@ def _true_peak_dbfs(mono, sr):
return _db(np.max(np.abs(mono)))
def loudness(audio_path, max_seconds=180.0) -> dict:
"""Integrated LUFS, sample/true peak (dBFS), crest, + gain-to-target."""
import soundfile as sf
data, sr = sf.read(str(audio_path), dtype="float64")
if data.shape[0] < sr // 2: # pyloudnorm needs ≥ ~0.4 s for a gated block
def loudness(audio_path, win_s=streaming.DEFAULT_WIN_S) -> dict:
"""Integrated LUFS (chunked-approx), sample/true peak (dBFS, exact), crest,
+ gain-to-target, over the whole track."""
meta = streaming.probe(audio_path)
sr, channels = meta["sr"], meta["channels"]
if meta["frames"] and meta["frames"] < sr // 2:
raise ValueError("clip too short for integrated loudness (need ≥ 0.5 s)")
if max_seconds:
data = data[: int(sr * max_seconds)]
mono = data if data.ndim == 1 else data.mean(axis=1)
rms_db = _db(np.sqrt(np.mean(mono ** 2) + _EPS))
peak_db = _db(np.max(np.abs(mono)))
tp_db = _true_peak_dbfs(mono, sr)
lufs = None
try:
import pyloudnorm as pyln
lufs = float(pyln.Meter(sr).integrated_loudness(data))
meter = pyln.Meter(sr)
except Exception:
pass
meter = None
sumsq = 0.0
n = 0
peak = _EPS
tp = _EPS
energy_sum = 0.0 # Σ dur_i · 10^(LUFS_i/10) — linear loudness, duration-weighted
dur_sum = 0.0
for block, sr, _ in streaming.windows(audio_path, win_s, mono=False, dtype="float64"):
if not len(block):
continue
mono = block if block.ndim == 1 else block.mean(axis=1)
sumsq += float(np.sum(mono ** 2))
n += mono.size
peak = max(peak, float(np.max(np.abs(mono))))
tp = max(tp, 10 ** (_true_peak_dbfs(mono, sr) / 20.0))
if meter is not None and len(block) >= int(_METER_MIN_S * sr):
try:
lw = float(meter.integrated_loudness(block))
except Exception:
lw = float("-inf")
if np.isfinite(lw): # silence-gated windows → skip
w = len(block) / sr
energy_sum += w * (10 ** (lw / 10.0))
dur_sum += w
if n < sr // 2:
raise ValueError("clip too short for integrated loudness (need ≥ 0.5 s)")
gain = ({k: round(t - lufs, 2) for k, t in TARGETS.items()} if lufs is not None
and np.isfinite(lufs) else None)
rms_db = _db(np.sqrt(sumsq / max(n, 1) + _EPS))
peak_db = _db(peak)
tp_db = _db(tp)
lufs = 10.0 * np.log10(energy_sum / dur_sum) if dur_sum > 0 else None
gain = ({k: round(t - lufs, 2) for k, t in TARGETS.items()}
if lufs is not None and np.isfinite(lufs) else None)
return {
"lufs_integrated": round(lufs, 2) if lufs is not None and np.isfinite(lufs) else None,
"lufs_method": "chunked",
"approximate": True, # energy-weighted across windows, not a single gated pass
"peak_dbfs": round(peak_db, 2),
"true_peak_dbfs": round(tp_db, 2),
"rms_dbfs": round(rms_db, 2),
"crest_db": round(peak_db - rms_db, 2),
"gain_to_target_db": gain, # add this many dB to hit each LUFS target
"channels": 1 if data.ndim == 1 else data.shape[1],
"gain_to_target_db": gain, # add this many dB to hit each LUFS target
"channels": int(channels),
"sr": int(sr),
"duration_s": round(meta["duration_s"], 3),
"engine": "light",
"full_track": True,
}
"""Light signal building blocks — onsets + waveform peaks. CPU, NO torch.
"""Light signal building blocks — onsets + waveform + spectrum. CPU, NO torch.
The cheapest analysis tier: thin librosa wrappers that need no model at all.
Especially useful for hexa (a visuals platform): /waveform gives a render-ready
peak/energy envelope, /onsets gives rhythmic hit times for visual sync + slicing.
Self-contained like ears_light/feats (the container has only armada/api/).
FULL-TRACK (#7): these stream the WHOLE upload in bounded windows and aggregate,
instead of only the first 60 s — so a long track is analyzed end-to-end without
buffering it all in RAM. waveform/spectrum aggregate EXACTLY (fixed output grid
filled by streaming reduction); onsets builds one full-track onset envelope (tiny
— hop-downsampled) from per-window pieces. Native sample rate throughout (no
per-block resample, which would seam at window edges). Self-contained like
ears_light/feats (the container has only armada/api/).
"""
import numpy as np
SR = 22050
MAX_S = 60.0
from . import streaming
HOP = 512 # librosa default; fixes frame→time mapping across windows
_MIN_FRAMES_SR = 20 # need ≥ 1/20 s of audio to say anything
_FALLBACK_CAP_S = 600.0 # bounded whole-file decode when a format isn't frame-countable
def _load(audio_path, max_seconds):
def _load_capped(audio_path, sr_target=22050):
"""Bounded whole-file fallback for formats soundfile can't frame-count.
Decodes at most _FALLBACK_CAP_S seconds — never the unbounded file."""
import librosa
y, _ = librosa.load(str(audio_path), sr=SR, mono=True, duration=max_seconds)
if y.size < SR // 20:
y, sr = librosa.load(str(audio_path), sr=sr_target, mono=True, duration=_FALLBACK_CAP_S)
if y.size < sr // _MIN_FRAMES_SR:
raise ValueError("clip too short / undecodable (need ≥ 50 ms of audio)")
return y
return y, sr
def onsets(audio_path, max_seconds=MAX_S) -> dict:
"""Onset hit times (s) + tempo + onset rate. Drift-tolerant tempo via the
onset envelope (librosa 0.11 feature.tempo, not the removed rhythm.tempo)."""
def onsets(audio_path, win_s=streaming.DEFAULT_WIN_S) -> dict:
"""Onset hit times (s) + tempo + onset rate over the WHOLE track. Builds one
full-track onset-strength envelope by concatenating per-window pieces (the
envelope is hop-downsampled, so a 10-min track is only ~tens of KB), then runs
detection + drift-tolerant tempo on it once."""
import librosa
y = _load(audio_path, max_seconds)
onset_env = librosa.onset.onset_strength(y=y, sr=SR)
times = librosa.onset.onset_detect(onset_envelope=onset_env, sr=SR, units="time")
tempo = float(np.atleast_1d(librosa.feature.tempo(onset_envelope=onset_env, sr=SR))[0])
dur = len(y) / SR
meta = streaming.probe(audio_path)
sr = meta["sr"]
if meta["frames"] <= 0: # uncountable format → bounded fallback
y, sr = _load_capped(audio_path)
env = librosa.onset.onset_strength(y=y, sr=sr, hop_length=HOP)
full = False
else:
pieces, full = [], True
for block, sr, _ in streaming.windows(audio_path, win_s, mono=True):
if block.size:
pieces.append(librosa.onset.onset_strength(y=block, sr=sr, hop_length=HOP))
if not pieces:
raise ValueError("clip too short / undecodable (need ≥ 50 ms of audio)")
env = np.concatenate(pieces)
times = librosa.onset.onset_detect(onset_envelope=env, sr=sr, hop_length=HOP, units="time")
tempo = float(np.atleast_1d(librosa.feature.tempo(onset_envelope=env, sr=sr, hop_length=HOP))[0])
dur = meta["duration_s"] or (len(env) * HOP / sr)
return {
"tempo_bpm": round(tempo, 2),
"onsets": [round(float(t), 4) for t in times],
"n_onsets": int(len(times)),
"onset_rate": round(len(times) / dur, 3) if dur else 0.0,
"duration_s": round(dur, 3),
"sr": SR,
"sr": int(sr),
"engine": "light",
"full_track": full,
}
def waveform(audio_path, bins=800, max_seconds=MAX_S) -> dict:
"""Downsampled waveform for rendering: per-bin [min, max] in [-1,1] (draws the
classic waveform) + a 0..1 RMS energy envelope (audio-reactive intensity)."""
y = _load(audio_path, max_seconds)
def waveform(audio_path, bins=800, win_s=streaming.DEFAULT_WIN_S) -> dict:
"""Downsampled waveform for rendering: per-bin [min, max] in [-1,1] + a 0..1
RMS energy envelope, spanning the WHOLE track. Exact: each `bins` bucket is
filled by an unsegmented min/max/mean-square reduction over the streamed
samples, so the result equals a single-pass computation."""
bins = max(1, min(int(bins), 4000))
meta = streaming.probe(audio_path)
total, sr = meta["frames"], meta["sr"]
if total <= 0: # uncountable format → bounded fallback
y, sr = _load_capped(audio_path)
return _waveform_array(y, sr, bins, full=False)
mn = np.full(bins, np.inf, dtype="float64")
mx = np.full(bins, -np.inf, dtype="float64")
sq = np.zeros(bins, dtype="float64")
cnt = np.zeros(bins, dtype="int64")
for block, sr, start in streaming.windows(audio_path, win_s, mono=True):
if not block.size:
continue
idx = ((np.arange(len(block), dtype="int64") + start) * bins // total)
np.clip(idx, 0, bins - 1, out=idx)
np.minimum.at(mn, idx, block)
np.maximum.at(mx, idx, block)
np.add.at(sq, idx, block.astype("float64") ** 2)
np.add.at(cnt, idx, 1)
if cnt.sum() < sr // _MIN_FRAMES_SR:
raise ValueError("clip too short / undecodable (need ≥ 50 ms of audio)")
return _finish_waveform(mn, mx, sq, cnt, bins, total, sr, full=True)
def _waveform_array(y, sr, bins, full):
edges = np.linspace(0, len(y), bins + 1).astype(int)
peaks, rms = [], []
mn = np.zeros(bins); mx = np.zeros(bins); sq = np.zeros(bins); cnt = np.zeros(bins, dtype="int64")
for i in range(bins):
seg = y[edges[i]:edges[i + 1]]
if seg.size == 0:
peaks.append([0.0, 0.0]); rms.append(0.0); continue
peaks.append([round(float(seg.min()), 4), round(float(seg.max()), 4)])
rms.append(float(np.sqrt(np.mean(seg ** 2))))
norm = max(rms) or 1.0
if seg.size:
mn[i], mx[i] = seg.min(), seg.max()
sq[i] = np.sum(seg.astype("float64") ** 2); cnt[i] = seg.size
return _finish_waveform(mn, mx, sq, cnt, bins, len(y), sr, full=full)
def _finish_waveform(mn, mx, sq, cnt, bins, total, sr, full):
empty = cnt == 0
mn = np.where(empty, 0.0, mn)
mx = np.where(empty, 0.0, mx)
rms = np.sqrt(np.where(cnt > 0, sq / np.maximum(cnt, 1), 0.0))
norm = float(rms.max()) or 1.0
return {
"bins": bins,
"peaks": peaks,
"rms": [round(r / norm, 4) for r in rms], # 0..1, peak-normalized
"duration_s": round(len(y) / SR, 3),
"sr": SR,
"peaks": [[round(float(lo), 4), round(float(hi), 4)] for lo, hi in zip(mn, mx)],
"rms": [round(float(r / norm), 4) for r in rms],
"duration_s": round(total / sr, 3) if sr else 0.0,
"sr": int(sr),
"engine": "light",
"full_track": full,
}
def spectrum(audio_path, bands=64, frames=200, mel=True, max_seconds=MAX_S) -> dict:
def spectrum(audio_path, bands=64, frames=200, mel=True, win_s=streaming.DEFAULT_WIN_S) -> dict:
"""FFT-as-a-service: a downsampled, render-ready spectrogram — `bands` ×
`frames`, 0..1 normalized (from dB, ref=peak). mel=True uses perceptual mel
bands (good for visuals); mel=False uses log-spaced linear bins. `frames=1`
collapses to a single time-averaged FFT magnitude spectrum. Bounded payload,
so it caches cheaply. Torch-free."""
`frames`, 0..1 normalized, spanning the WHOLE track. Streams windows, bins each
window's spectrogram columns into the global `frames` grid by absolute time,
accumulates mean POWER per (band, frame), then converts to dB (ref = the
track's global peak power) once at the end. `frames=1` → a single time-averaged
spectrum. Torch-free."""
import librosa
y = _load(audio_path, max_seconds)
bands = max(1, min(int(bands), 256))
frames = max(1, min(int(frames), 1000))
if mel:
S = librosa.feature.melspectrogram(y=y, sr=SR, n_mels=bands)
band_hz = [round(float(f), 1) for f in librosa.mel_frequencies(n_mels=bands, fmax=SR / 2)]
meta = streaming.probe(audio_path)
total_frames, sr = meta["frames"], meta["sr"]
if total_frames <= 0: # uncountable format → bounded fallback
y, sr = _load_capped(audio_path)
total_frames = len(y)
blocks = [(y, sr, 0)]
full = False
else:
mag = np.abs(librosa.stft(y, n_fft=2048, hop_length=512)) ** 2
freqs = librosa.fft_frequencies(sr=SR, n_fft=2048)
edges = np.logspace(np.log10(20), np.log10(SR / 2), bands + 1)
S = np.stack([mag[(freqs >= edges[i]) & (freqs < edges[i + 1])].sum(axis=0)
for i in range(bands)])
band_hz = [round(float(np.sqrt(edges[i] * edges[i + 1])), 1) for i in range(bands)]
Sdb = librosa.power_to_db(S + 1e-12, ref=np.max) # [-80..0] dB
# downsample time → `frames` columns by averaging within each block
cols = Sdb.shape[1]
edges = np.linspace(0, cols, frames + 1).astype(int)
blocks = [Sdb[:, edges[i]:edges[i + 1]].mean(axis=1) if edges[i + 1] > edges[i]
else Sdb[:, min(edges[i], cols - 1)] for i in range(frames)]
spec = (np.clip(np.stack(blocks, axis=1), -80, 0) + 80) / 80.0 # bands×frames, 0..1
blocks = streaming.windows(audio_path, win_s, mono=True)
full = True
acc = np.zeros((bands, frames), dtype="float64") # summed power per (band, global-frame)
cnt = np.zeros(frames, dtype="int64")
gmax = 1e-12 # global peak power for the dB ref
band_hz = None
total_dur = (total_frames / sr) if sr else 0.0
for block, sr, start in blocks:
if block.size < HOP:
continue
if mel:
S = librosa.feature.melspectrogram(y=block, sr=sr, n_mels=bands, hop_length=HOP)
if band_hz is None:
band_hz = [round(float(f), 1) for f in librosa.mel_frequencies(n_mels=bands, fmax=sr / 2)]
else:
mag = np.abs(librosa.stft(block, n_fft=2048, hop_length=HOP)) ** 2
freqs = librosa.fft_frequencies(sr=sr, n_fft=2048)
edges = np.logspace(np.log10(20), np.log10(sr / 2), bands + 1)
S = np.stack([mag[(freqs >= edges[i]) & (freqs < edges[i + 1])].sum(axis=0)
for i in range(bands)])
if band_hz is None:
band_hz = [round(float(np.sqrt(edges[i] * edges[i + 1])), 1) for i in range(bands)]
gmax = max(gmax, float(S.max()))
# map each column to its absolute time → global frame bucket
col_t = (start + np.arange(S.shape[1]) * HOP) / sr
gf = np.zeros(S.shape[1], dtype="int64") if total_dur <= 0 else \
np.clip((col_t / total_dur * frames).astype("int64"), 0, frames - 1)
for b in range(frames):
sel = gf == b
if sel.any():
acc[:, b] += S[:, sel].sum(axis=1)
cnt[b] += int(sel.sum())
if cnt.sum() == 0:
raise ValueError("clip too short / undecodable (need ≥ 50 ms of audio)")
# empty frame buckets (sparse coverage) borrow the nearest filled one's mean
mean_p = np.zeros((bands, frames), dtype="float64")
filled = cnt > 0
mean_p[:, filled] = acc[:, filled] / cnt[filled]
if not filled.all():
idx = np.where(filled)[0]
for b in np.where(~filled)[0]:
mean_p[:, b] = mean_p[:, idx[np.argmin(np.abs(idx - b))]]
Sdb = librosa.power_to_db(mean_p + 1e-12, ref=gmax)
spec = (np.clip(Sdb, -80, 0) + 80) / 80.0
return {
"bands": bands,
"frames": frames,
"mel": bool(mel),
"band_hz": band_hz,
"spec": [[round(float(v), 4) for v in row] for row in spec], # [band][frame]
"duration_s": round(len(y) / SR, 3),
"sr": SR,
"band_hz": band_hz or [],
"spec": [[round(float(v), 4) for v in row] for row in spec],
"duration_s": round(total_dur, 3),
"sr": int(sr),
"engine": "light",
"full_track": full,
}
"""Bounded-memory streaming primitives for full-track analysis. CPU, NO torch.
Producers upload whole tracks, not 60 s clips. Rather than 413-reject a big file
(or buffer the whole thing in RAM), the signal/loudness engines decode it in
fixed time windows and aggregate as they go — so peak memory is O(one window),
not O(file), and a large upload can't OOM the shared erable box.
soundfile (libsndfile 1.2) streams WAV/FLAC/OGG/MP3 with exact frame counts, so
we read at NATIVE sample rate and let each engine pass `sr` to librosa — no
per-block resampling (which would add seam artifacts at window boundaries).
"""
import numpy as np
# 30 s windows: at 96 kHz stereo float64 that is ~44 MB resident — comfortably
# under the container's 3 GB cap, with margin for the engine's own working set.
DEFAULT_WIN_S = 30.0
def probe(path) -> dict:
"""Header-only metadata (no decode): frames, sr, channels, duration_s.
`frames <= 0` means the container can't frame-count this file (rare); callers
fall back to a bounded whole-file decode."""
import soundfile as sf
i = sf.info(str(path))
return {
"frames": int(i.frames),
"sr": int(i.samplerate),
"channels": int(i.channels),
"duration_s": (i.frames / i.samplerate) if i.samplerate else 0.0,
"format": i.format,
}
def windows(path, win_s=DEFAULT_WIN_S, *, mono=True, dtype="float32"):
"""Yield (block, sr, start_frame) over the WHOLE file in `win_s` windows.
`mono=True` downmixes channels (mean); `mono=False` yields 2-D (n, channels).
Non-overlapping and contiguous: start_frame of window k is the running sum of
prior window lengths, so callers can map a block's samples to absolute time.
Bounded memory: only one window is resident at a time."""
import soundfile as sf
with sf.SoundFile(str(path)) as f:
sr = f.samplerate
wf = max(1, int(win_s * sr))
start = 0
while True:
blk = f.read(frames=wf, dtype=dtype, always_2d=not mono)
if len(blk) == 0:
return
block = (blk if blk.ndim == 1 else blk.mean(axis=1)) if mono else blk
yield block, sr, start
start += len(block)
"""Full-track streaming (#7) — the guarantees that make big uploads safe.
Proves the two things unit tests on 2 s clips can't: (1) the signal/loudness
engines now analyse the WHOLE track, past the old 60 s / 180 s head cap, and
(2) streaming didn't change the numbers — content_id stays byte-identical to the
old whole-file hash (so the deployed cache stays warm), and exact reductions
(waveform min/max) equal a single-pass computation.
cd armada/api && python -m pytest tests/test_fulltrack.py -q
"""
import hashlib
import tempfile
from pathlib import Path
import numpy as np
import soundfile as sf
import cache
from engines import loudness, signal, streaming
SR = 22050
def _click_track(bpm=120, secs=70.0, sr=SR):
"""A long pulse train (>60 s) so a head-only analysis would visibly truncate."""
n = int(sr * secs)
y = np.zeros(n, dtype="float32")
step = int(sr * 60 / bpm)
for i in range(0, n, step):
y[i:i + 200] += np.hanning(min(200, n - i)).astype("float32")
return y
def _write(y, sr=SR, suffix=".wav"):
p = Path(tempfile.mktemp(suffix=suffix))
sf.write(str(p), y, sr)
return str(p)
# ── (1) full-track coverage: past the old 60 s / 180 s head cap ─────────────────
def test_onsets_cover_whole_track_not_just_head():
p = _write(_click_track(secs=70.0))
r = signal.onsets(p)
assert r["full_track"] is True
assert r["duration_s"] >= 69.0 # whole track, not capped at 60
assert max(r["onsets"]) > 60.0 # hits found beyond the old head window
assert r["n_onsets"] >= 120 # ~120 bpm × 70 s ≈ 140 pulses
def test_waveform_spans_whole_track():
p = _write(_click_track(secs=70.0))
r = signal.waveform(p, bins=100)
assert r["full_track"] is True and r["bins"] == 100
assert r["duration_s"] >= 69.0
# pulses recur throughout → energy present in the last bins, not only the first
assert r["rms"][-1] > 0.0 and r["rms"][90] > 0.0
def test_loudness_full_track_exact_peak_and_flagged_approx():
y = _click_track(secs=70.0) * 0.5
r = loudness.loudness(_write(y))
assert r["full_track"] is True and r["approximate"] is True
assert r["lufs_method"] == "chunked" and r["lufs_integrated"] is not None
# peak is an EXACT running max over the stream
assert abs(r["peak_dbfs"] - 20 * np.log10(float(np.max(np.abs(y))))) < 0.1
assert r["duration_s"] >= 69.0
# ── (2) streaming didn't change the numbers ─────────────────────────────────────
def test_content_id_streaming_equals_whole_file_hash():
"""The streamed content_id must equal the OLD whole-array hash, or every
deployed cache entry silently invalidates on deploy."""
y = (0.2 * np.sin(2 * np.pi * 220 * np.linspace(0, 5, 5 * SR, endpoint=False))).astype("float32")
p = _write(y)
# reproduce the pre-#7 computation: sha256(whole_pcm.tobytes() + str(sr))
data, sr = sf.read(p, dtype="float32", always_2d=False)
h = hashlib.sha256()
h.update(np.ascontiguousarray(data).tobytes())
h.update(str(sr).encode())
expected = "cid_" + h.hexdigest()[:32]
assert cache.content_id(p) == expected
def test_waveform_reduction_is_exact_vs_single_pass():
"""Streamed per-bin min/max must equal a single-pass numpy computation."""
y = (0.4 * np.sin(2 * np.pi * 110 * np.linspace(0, 8, 8 * SR, endpoint=False))).astype("float32")
p = _write(y)
bins = 64
r = signal.waveform(p, bins=bins)
edges = np.linspace(0, len(y), bins + 1).astype(int)
for i in range(bins):
seg = y[edges[i]:edges[i + 1]]
assert abs(r["peaks"][i][0] - round(float(seg.min()), 4)) < 1e-3
assert abs(r["peaks"][i][1] - round(float(seg.max()), 4)) < 1e-3
def test_streaming_windows_cover_all_samples_once():
"""The window iterator must yield every sample exactly once, contiguously."""
y = np.arange(int(2.5 * SR), dtype="float32")
p = _write(y)
seen = 0
last_end = 0
for block, sr, start in streaming.windows(p, win_s=1.0, mono=True):
assert start == last_end # contiguous, no gap/overlap
seen += len(block)
last_end = start + len(block)
assert seen == len(y)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment