PR-A5: accuracy benchmark loop — synthetic corpus + CI threshold gate

.github/workflows/benchmarks.yml

@@ -0,0 +1,32 @@
+name: benchmarks
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+
+jobs:
+  accuracy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+      - name: Install
+        run: |
+          python -m pip install --upgrade pip
+          # core deps + test (pytest, httpx) + web (fastapi, uvicorn, multipart)
+          # extras. torch is intentionally NOT installed — heavy libs load lazily
+          # and the GPU-dependent paths are mocked, so the suite is CPU-only.
+          pip install -e ".[dev,ui]"
+      - name: Unit tests
+        run: pytest -q
+      - name: Accuracy benchmarks (threshold gate)
+        run: python -m benchmarks.run_benchmarks --report benchmark-report.json
+      - name: Upload report
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: benchmark-report
+          path: benchmark-report.json

benchmarks/README.md

@@ -0,0 +1,50 @@
+# StemForge accuracy benchmarks (A-series PR-A5)
+
+A synthetic **render-roundtrip** corpus with a known ground truth, a set of
+accuracy metrics, and a threshold gate wired into CI. It measures the accuracy
+work from PR-A1..A4 (cymbal MIDI, tempo evidence, drum calibration) and fails the
+build on a regression.
+
+## Run
+
+```bash
+python -m benchmarks.run_benchmarks                     # print a JSON report, exit 1 on a violation
+python -m benchmarks.run_benchmarks --report report.json --corpus-dir corpus/
+```
+
+## Corpus (`corpus.py`)
+
+Everything is rendered deterministically (seeded `numpy.default_rng`) so runs
+reproduce and the fixture SHA-256 is stable:
+
+- **Tempo** — noise-pulse trains at constant **90 / 120 / 140 BPM**.
+- **Drums** — a backbeat groove with per-part ground truth (kick / snare / toms),
+  quiet **ghost** snares, and a short low-tom fill; plus a **bleed** variant with
+  weak phantom onsets in the toms part coincident with each kick.
+
+## Metrics (`metrics.py`)
+
+| Metric | Source | Gated |
+|--------|--------|-------|
+| `tempo_abs_error_bpm_max` | analysis (A2) | ✅ |
+| `half_double_correct` | analysis candidates (A2) | ✅ |
+| `drum_f1_min` (kick+snare onset F1) | drum_midi (A1/A4) | ✅ |
+| `duplicate_rate_max` (cross-part) | drum_midi de-dup (A4) | ✅ |
+| `velocity_rank_corr_min` | drum_midi velocity (A4) | ✅ |
+| toms F1, per-fixture detail | drum_midi | reported, not gated (sparse) |
+
+**Skipped (reported, never silently dropped):** melodic onset/pitch F1,
+note fragmentation (need `basic-pitch`, absent in the offline harness);
+beat alignment (needs a beat-aligned reference); quantization displacement
+(covered by the A3 ledger's `max_displacement_s`).
+
+## Thresholds (`thresholds.yaml`)
+
+Bounds with margin over a clean baseline. CI (`.github/workflows/benchmarks.yml`)
+runs `run_benchmarks` and fails on any violation. Tighten as accuracy improves.
+
+## Provenance
+
+Every report embeds the **fixture SHA-256**, the drum-midi **config hash**, and
+**library/model versions** captured at run time — so a report is traceable to the
+exact corpus, config, and dependency set that produced it.

benchmarks/__init__.py

@@ -0,0 +1,9 @@
+"""StemForge accuracy benchmarks (A-series PR-A5).
+
+A synthetic render-roundtrip corpus with a known ground truth, a set of accuracy
+metrics, and a threshold gate wired into CI. The corpus is generated
+deterministically (seeded RNG) so runs are reproducible; every report embeds the
+fixture SHA-256, the config hash, and library/model versions as provenance.
+"""
+
+from . import corpus, metrics  # noqa: F401

benchmarks/corpus.py

@@ -0,0 +1,151 @@
+"""Deterministic synthetic corpus with a known ground truth.
+
+Every fixture is rendered from parameters the metrics can check against — beat
+grids at a known BPM, drum grooves with a known hit list per part, and a bleed
+variant with deliberate cross-part phantoms. No external audio, no GPU, no
+network. Determinism (seeded ``default_rng``) makes the fixture SHA-256 stable.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import numpy as np
+import soundfile as sf
+
+from stemforge.io_utils import sha256_bytes
+
+SR = 22050
+
+
+# --------------------------------------------------------------------------- #
+# Renderers
+# --------------------------------------------------------------------------- #
+def _noise_hit(n: int, seed: int, amp: float, tau: float) -> np.ndarray:
+    rng = np.random.default_rng(seed)
+    t = np.arange(n) / SR
+    return (amp * rng.standard_normal(n) * np.exp(-t / tau)).astype(np.float32)
+
+
+def _tone_hit(n: int, hz: float, amp: float, tau: float) -> np.ndarray:
+    t = np.arange(n) / SR
+    return (amp * np.sin(2 * np.pi * hz * t) * np.exp(-t / tau)).astype(np.float32)
+
+
+def _render(hits: list[tuple[float, np.ndarray]], dur: float) -> np.ndarray:
+    y = np.zeros(int(dur * SR), dtype=np.float32)
+    for start, sig in hits:
+        i = int(start * SR)
+        end = min(y.size, i + sig.size)
+        y[i:end] += sig[: end - i]
+    return y
+
+
+# --------------------------------------------------------------------------- #
+# Fixtures
+# --------------------------------------------------------------------------- #
+@dataclass
+class TempoFixture:
+    name: str
+    true_bpm: float
+    audio: np.ndarray
+    sr: int = SR
+
+
+@dataclass
+class DrumFixture:
+    name: str
+    parts: dict[str, np.ndarray]                 # part -> mono audio
+    truth: dict[str, list[float]]                # part -> hit times (seconds)
+    amps: dict[str, list[float]]                 # part -> render amplitude per hit (for velocity corr)
+    bleed_pairs: list[tuple[str, float]] = field(default_factory=list)  # (part, time) phantom hits
+    sr: int = SR
+
+
+def tempo_fixtures() -> list[TempoFixture]:
+    """Constant-tempo noise-pulse trains at 90 / 120 / 140 BPM."""
+    out: list[TempoFixture] = []
+    for k, bpm in enumerate((90.0, 120.0, 140.0)):
+        dur = 6.0
+        period = 60.0 / bpm
+        n = int(0.04 * SR)
+        hits = [(t, _noise_hit(n, seed=100 + k, amp=0.9, tau=0.02))
+                for t in np.arange(0.2, dur, period)]
+        out.append(TempoFixture(name=f"tempo_{int(bpm)}", true_bpm=bpm, audio=_render(hits, dur)))
+    return out
+
+
+def drum_fixtures() -> list[DrumFixture]:
+    """A backbeat groove with per-part ground truth, plus a bleed variant."""
+    dur = 4.0
+    period = 0.5  # 120 BPM quarter notes
+    beats = list(np.arange(0.2, dur, period))
+    n = int(0.28 * SR)
+
+    kick_times = [b for i, b in enumerate(beats) if i % 2 == 0]
+    snare_times = [b for i, b in enumerate(beats) if i % 2 == 1]
+    tom_times = [3.45, 3.6]                             # a short low-tom fill, off the kick/snare grid
+    ghost_times = [t + 0.25 for t in snare_times[:-1]]  # quiet ghost snares off the beat
+
+    kick = _render([(t, _noise_hit(n, 10 + i, amp=1.0, tau=0.05)) for i, t in enumerate(kick_times)], dur)
+    snare_hits = [(t, _noise_hit(n, 20 + i, amp=0.9, tau=0.05)) for i, t in enumerate(snare_times)]
+    ghost_hits = [(t, _noise_hit(n, 40 + i, amp=0.12, tau=0.04)) for i, t in enumerate(ghost_times)]
+    snare = _render(snare_hits + ghost_hits, dur)
+    toms = _render([(t, _tone_hit(n, 90.0, amp=0.9, tau=0.12)) for t in tom_times], dur)
+
+    clean = DrumFixture(
+        name="groove_clean",
+        parts={"kick": kick, "snare": snare, "toms": toms},
+        truth={"kick": kick_times, "snare": snare_times + ghost_times, "toms": tom_times},
+        amps={"kick": [1.0] * len(kick_times),
+              "snare": [0.9] * len(snare_times) + [0.12] * len(ghost_times),
+              "toms": [0.9] * len(tom_times)},
+    )
+
+    # bleed variant: weak phantom kick coincident in the toms part (separation spill)
+    phantom_times = kick_times
+    toms_bleed = _render(
+        [(t, _tone_hit(n, 90.0, amp=0.9, tau=0.12)) for t in tom_times]
+        + [(t + 0.004, _noise_hit(n, 60 + i, amp=0.12, tau=0.05)) for i, t in enumerate(phantom_times)],
+        dur)
+    bleed = DrumFixture(
+        name="groove_bleed",
+        parts={"kick": kick, "snare": snare, "toms": toms_bleed},
+        truth={"kick": kick_times, "snare": snare_times + ghost_times, "toms": tom_times},
+        amps={"kick": [1.0] * len(kick_times),
+              "snare": [0.9] * len(snare_times) + [0.12] * len(ghost_times),
+              "toms": [0.9] * len(tom_times)},
+        bleed_pairs=[("toms", t + 0.004) for t in phantom_times],
+    )
+    return [clean, bleed]
+
+
+# --------------------------------------------------------------------------- #
+# Provenance
+# --------------------------------------------------------------------------- #
+def fixture_sha256() -> str:
+    """A stable SHA-256 over the whole rendered corpus (provenance for reports)."""
+    h_parts: list[bytes] = []
+    for tf in tempo_fixtures():
+        h_parts.append(tf.audio.tobytes())
+    for df in drum_fixtures():
+        for name in sorted(df.parts):
+            h_parts.append(df.parts[name].tobytes())
+    return sha256_bytes(b"".join(h_parts))
+
+
+def write_corpus(dest: Path) -> dict[str, Any]:
+    """Materialize the corpus as .wav files under ``dest`` (for inspection/CI)."""
+    dest.mkdir(parents=True, exist_ok=True)
+    manifest: dict[str, Any] = {"tempo": {}, "drums": {}, "sha256": fixture_sha256()}
+    for tf in tempo_fixtures():
+        p = dest / f"{tf.name}.wav"
+        sf.write(str(p), tf.audio, tf.sr, subtype="FLOAT")
+        manifest["tempo"][tf.name] = {"path": str(p), "true_bpm": tf.true_bpm}
+    for df in drum_fixtures():
+        for part, y in df.parts.items():
+            sf.write(str(dest / f"{df.name}_{part}.wav"), y, df.sr, subtype="FLOAT")
+        manifest["drums"][df.name] = {"truth": df.truth}
+    return manifest

benchmarks/metrics.py

@@ -0,0 +1,86 @@
+"""Accuracy metrics for the benchmark corpus.
+
+Pure functions over ground truth + StemForge output — no I/O, deterministic.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+
+
+def tempo_abs_error(true_bpm: float, est_bpm: float) -> float:
+    """Absolute BPM error. An unknown estimate (0.0) is a full miss (= true_bpm)."""
+    if est_bpm <= 0:
+        return float(true_bpm)
+    return float(abs(true_bpm - est_bpm))
+
+
+def half_double_ok(true_bpm: float, candidates: list[float], rel_tol: float = 0.05) -> bool:
+    """True if the true BPM matches the estimate OR one of its half/double
+    candidates within an octave-relative tolerance (a half/double detection error
+    is recoverable — the correct octave is surfaced in the candidate list)."""
+    return any(abs(true_bpm - c) <= rel_tol * true_bpm for c in (candidates or []))
+
+
+def match_events(truth: list[float], pred: list[float], tol_s: float = 0.05) -> tuple[int, int, int]:
+    """Greedy one-to-one match of predicted onset times to truth within tol.
+    Returns (true_positives, false_positives, false_negatives)."""
+    truth_sorted = sorted(truth)
+    used = [False] * len(truth_sorted)
+    tp = 0
+    for p in sorted(pred):
+        best, best_d = -1, tol_s
+        for i, t in enumerate(truth_sorted):
+            if used[i]:
+                continue
+            d = abs(p - t)
+            if d <= best_d:
+                best, best_d = i, d
+        if best >= 0:
+            used[best] = True
+            tp += 1
+    fp = len(pred) - tp
+    fn = len(truth_sorted) - tp
+    return tp, fp, fn
+
+
+def f1(tp: int, fp: int, fn: int) -> float:
+    if tp == 0:
+        return 0.0
+    prec = tp / (tp + fp) if (tp + fp) else 0.0
+    rec = tp / (tp + fn) if (tp + fn) else 0.0
+    return float(2 * prec * rec / (prec + rec)) if (prec + rec) else 0.0
+
+
+def duplicate_rate(note_times_pitches: list[tuple[float, int]], window_s: float = 0.018) -> float:
+    """Fraction of adjacent note pairs that are coincident (<= window) but a
+    DIFFERENT pitch — i.e. cross-part duplicates. 0.0 is clean."""
+    seq = sorted(note_times_pitches)
+    if len(seq) < 2:
+        return 0.0
+    dup = sum(1 for a, b in zip(seq, seq[1:])
+              if b[0] - a[0] <= window_s and a[1] != b[1])
+    return float(dup) / (len(seq) - 1)
+
+
+def rank_corr(a: list[float], b: list[float]) -> float:
+    """Spearman rank correlation (velocity-vs-render-amplitude ordering).
+    Returns 0.0 when undefined (constant input or < 2 points)."""
+    if len(a) != len(b) or len(a) < 2:
+        return 0.0
+    ra = _rankdata(a)
+    rb = _rankdata(b)
+    ra -= ra.mean()
+    rb -= rb.mean()
+    denom = float(np.sqrt((ra ** 2).sum() * (rb ** 2).sum()))
+    if denom <= 1e-12:
+        return 0.0
+    return float((ra * rb).sum() / denom)
+
+
+def _rankdata(x: list[float]) -> np.ndarray:
+    arr = np.asarray(x, dtype=np.float64)
+    order = arr.argsort()
+    ranks = np.empty(len(arr), dtype=np.float64)
+    ranks[order] = np.arange(len(arr), dtype=np.float64)
+    return ranks