From 10a6cf59abbb210a386112449e951ea75e1951cf Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 02:48:24 +0000 Subject: [PATCH 01/11] test: add failing regression for silent 120-BPM tempo fallback analysis._bpm_from_beats returned 120.0 on every low-evidence path (<3 beats, non-positive regression slope, non-positive median interval), so an un-estimable tempo was silently reported as 120 instead of unknown. This regression asserts those paths yield 0.0 (unknown); it fails on the current tree (120.0 == 0.0), pinning the silent fallback ahead of the fix. Receipt: analysis.py:96,113,119,123 (verified at main). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- tests/test_tempo_evidence.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) create mode 100644 tests/test_tempo_evidence.py diff --git a/tests/test_tempo_evidence.py b/tests/test_tempo_evidence.py new file mode 100644 index 0000000..5b027c6 --- /dev/null +++ b/tests/test_tempo_evidence.py @@ -0,0 +1,20 @@ +"""Tempo/beat evidence — A-series PR-A2. + +Receipts (verified at main): analysis.py silently returned 120 BPM on several +failure paths (`:96,:113,:119,:123`), `:98` assumed 4/4 (`downbeats=beats[::4]`), +and BeatGrid carried no confidence. This regression pins the first fix: an +un-estimable tempo must be UNKNOWN (0.0), never a silent 120. +""" + +from __future__ import annotations + +from stemforge.analysis import _bpm_from_beats + + +# --------------------------------------------------------------------------- # +# RED: an un-estimable tempo is UNKNOWN (0.0), never a silent 120 +# --------------------------------------------------------------------------- # +def test_bpm_unknown_is_zero_not_silent_120(): + assert _bpm_from_beats([1.0, 2.0], regression=True) == 0.0 # <3 beats + assert _bpm_from_beats([3.0, 2.0, 1.0], regression=True) == 0.0 # non-positive slope + assert _bpm_from_beats([1.0, 1.0, 1.0], regression=False) == 0.0 # non-positive median From 3a3276a13527fd5dad32ff22693d11e18e991afd Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 02:48:40 +0000 Subject: [PATCH 02/11] =?UTF-8?q?feat:=20tempo/beat=20evidence=20=E2=80=94?= =?UTF-8?q?=20honest=20unknowns=20+=20shared=20engine=20fallback=20chain?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stop passing off assumed tempo as measured, and record why each engine fallback happened instead of swallowing it. analysis: - _bpm_from_beats returns 0.0 (UNKNOWN) — never a silent 120 — for <3 beats, a non-positive slope, or a non-positive median interval. - BeatGrid gains bpm_confidence, bpm_candidates (always incl. half/double), meter_confidence, tempo_assumed, and fallback_chain; source_bpm==0.0 means unknown (no Optional spread). New helpers _bpm_confidence / _bpm_candidates / _meter_confidence back them. - The librosa downbeats are a 4/4 assumption, so that path is flagged tempo_assumed with low meter_confidence. - analyze() records the beat_this->librosa hop as EngineAttempt evidence. io_utils: shared EngineAttempt {engine,status,reason} schema + sanitize_reason (drops absolute paths, collapses subprocess dumps, deterministic). drum_midi: when the grid tempo is unknown the .mid still needs a tempo to serialize, so 120 is written but flagged tempo_source=midi_serialization_default with tempo_assumed=true; a measured tempo is tempo_source=beat_grid. stretch: the rubberband->signalsmith->librosa chain records a sanitized EngineAttempt per hop (the reasons the bare excepts dropped); match_bpm_file and stretch_stems manifests carry attempts[]. Turns the A2 regression green. Receipt: analysis.py:96,98,113,119,123; drum_midi.py:56; stretch.py hops. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- src/stemforge/analysis.py | 136 ++++++++++++++++++++++++++++++------- src/stemforge/drum_midi.py | 15 +++- src/stemforge/io_utils.py | 37 ++++++++++ src/stemforge/stretch.py | 59 +++++++++++++--- 4 files changed, 212 insertions(+), 35 deletions(-) diff --git a/src/stemforge/analysis.py b/src/stemforge/analysis.py index d11a89b..0ea2473 100644 --- a/src/stemforge/analysis.py +++ b/src/stemforge/analysis.py @@ -14,16 +14,23 @@ import numpy as np -from .io_utils import AudioTensor, save_audio +from .io_utils import AudioTensor, EngineAttempt, sanitize_reason, save_audio @dataclass class BeatGrid: - source_bpm: float + source_bpm: float # 0.0 == UNKNOWN (never a silent 120) beats: list[float] = field(default_factory=list) # beat times, seconds downbeats: list[float] = field(default_factory=list) # downbeat times, seconds engine: str = "librosa" beats_per_bar: int = 4 + # Tempo/meter evidence — so downstream stages (and the manifest) can tell a + # measured tempo from an assumed one instead of trusting a bare number. + bpm_confidence: float = 0.0 # 0..1 (0.0 when unknown) + bpm_candidates: list[float] = field(default_factory=list) # always incl. half/double + meter_confidence: float = 0.0 # 0..1 confidence in beats_per_bar + tempo_assumed: bool = False # True => bpm/meter is a fallback assumption + fallback_chain: list[dict] = field(default_factory=list) # EngineAttempt dicts @property def seconds_per_beat(self) -> float: @@ -39,7 +46,12 @@ def to_dict(self) -> dict[str, Any]: return { "engine": self.engine, "source_bpm": round(self.source_bpm, 3), + "bpm_confidence": round(self.bpm_confidence, 3), + "bpm_candidates": [round(c, 3) for c in self.bpm_candidates], + "tempo_assumed": self.tempo_assumed, "beats_per_bar": self.beats_per_bar, + "meter_confidence": round(self.meter_confidence, 3), + "fallback_chain": self.fallback_chain, "num_beats": len(self.beats), "num_downbeats": len(self.downbeats), "beats": [round(b, 4) for b in self.beats], @@ -53,13 +65,25 @@ def analyze( device: str = "cuda", bpm_from_regression: bool = True, ) -> BeatGrid: + """Analyze tempo/beats, recording the engine fallback chain as evidence. + + Never silently substitutes 120 BPM: when the tempo can't be measured the grid + reports ``source_bpm == 0.0`` (unknown) with ``tempo_assumed`` set. + """ + chain: list[dict] = [] if engine == "beat_this": try: - return _analyze_beat_this(audio, device, bpm_from_regression) - except Exception: - # fall through to librosa rather than fail the whole run + grid = _analyze_beat_this(audio, device, bpm_from_regression) + chain.append(EngineAttempt("beat_this", "used").to_dict()) + grid.fallback_chain = chain + return grid + except Exception as e: # noqa: BLE001 - fall through to librosa, but record why + chain.append(EngineAttempt("beat_this", "fell_through", sanitize_reason(e)).to_dict()) engine = "librosa" - return _analyze_librosa(audio, bpm_from_regression) + grid = _analyze_librosa(audio, bpm_from_regression) + chain.append(EngineAttempt("librosa", "used").to_dict()) + grid.fallback_chain = chain + return grid # --------------------------------------------------------------------------- # @@ -78,8 +102,16 @@ def _analyze_beat_this(audio: AudioTensor, device: str, regression: bool) -> Bea downbeats = [float(b) for b in np.asarray(downbeats).ravel()] bpm = _bpm_from_beats(beats, regression) bpb = _beats_per_bar(beats, downbeats) - return BeatGrid(source_bpm=bpm, beats=beats, downbeats=downbeats, - engine="beat_this", beats_per_bar=bpb) + # beat_this returns real downbeats: the meter is measured (not assumed), and + # tempo is assumed only when we couldn't derive a BPM at all. + return BeatGrid( + source_bpm=bpm, beats=beats, downbeats=downbeats, + engine="beat_this", beats_per_bar=bpb, + bpm_confidence=_bpm_confidence(beats, regression), + bpm_candidates=_bpm_candidates(bpm), + meter_confidence=_meter_confidence(beats, downbeats), + tempo_assumed=(bpm <= 0.0), + ) # --------------------------------------------------------------------------- # @@ -90,50 +122,108 @@ def _analyze_librosa(audio: AudioTensor, regression: bool) -> BeatGrid: y = audio.to_mono().samples[0] sr = audio.sample_rate - tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr, units="frames") + _tempo, beat_frames = librosa.beat.beat_track(y=y, sr=sr, units="frames") beat_times = librosa.frames_to_time(beat_frames, sr=sr) beats = [float(t) for t in beat_times] - tempo_val = float(np.asarray(tempo).ravel()[0]) if np.size(tempo) else 120.0 - bpm = _bpm_from_beats(beats, regression) if len(beats) >= 3 else tempo_val + # UNKNOWN (0.0) — never a silent 120 — when the beats can't support a BPM. + bpm = _bpm_from_beats(beats, regression) downbeats = beats[::4] # heuristic: assume 4/4, downbeat every 4 beats - return BeatGrid(source_bpm=bpm, beats=beats, downbeats=downbeats, - engine="librosa", beats_per_bar=4) + # The librosa downbeats are a 4/4 ASSUMPTION, so the meter is always assumed + # here (low meter confidence), independent of the BPM estimate quality. + return BeatGrid( + source_bpm=bpm, beats=beats, downbeats=downbeats, + engine="librosa", beats_per_bar=4, + bpm_confidence=_bpm_confidence(beats, regression), + bpm_candidates=_bpm_candidates(bpm), + meter_confidence=min(_meter_confidence(beats, downbeats), 0.2), + tempo_assumed=True, + ) # --------------------------------------------------------------------------- # -# BPM estimation +# BPM estimation + evidence # --------------------------------------------------------------------------- # def _bpm_from_beats(beats: list[float], regression: bool) -> float: """Linear regression of beat index -> time gives seconds/beat robustly. (Averaging raw inter-beat intervals is jitter-sensitive; the slope of a line fit through (index, time) is not — see madmom issue #416.) + + Returns ``0.0`` (UNKNOWN) — never a silent 120 — when there is too little + evidence (<3 beats), a non-positive regression slope, or a non-positive median + inter-beat interval. """ if len(beats) < 3: - return 120.0 + return 0.0 arr = np.asarray(beats, dtype=np.float64) if regression: idx = np.arange(len(arr)) slope, _ = np.polyfit(idx, arr, 1) # seconds per beat if slope <= 0: - return 120.0 + return 0.0 return 60.0 / slope intervals = np.diff(arr) med = float(np.median(intervals)) - return 60.0 / med if med > 0 else 120.0 + return 60.0 / med if med > 0 else 0.0 + + +def _bpm_confidence(beats: list[float], regression: bool) -> float: + """0..1 confidence in the BPM: R² of the index→time fit (regression) or + ``1 - CV`` of the inter-beat intervals. 0.0 whenever the BPM is unknown.""" + if len(beats) < 3: + return 0.0 + arr = np.asarray(beats, dtype=np.float64) + if regression: + idx = np.arange(len(arr)) + slope, intercept = np.polyfit(idx, arr, 1) + if slope <= 0: + return 0.0 + pred = slope * idx + intercept + ss_res = float(np.sum((arr - pred) ** 2)) + ss_tot = float(np.sum((arr - arr.mean()) ** 2)) + return float(np.clip(1.0 - ss_res / ss_tot, 0.0, 1.0)) if ss_tot > 0 else 0.0 + intervals = np.diff(arr) + med = float(np.median(intervals)) + if med <= 0: + return 0.0 + cv = float(np.std(intervals) / med) + return float(np.clip(1.0 - cv, 0.0, 1.0)) + + +def _bpm_candidates(bpm: float) -> list[float]: + """Always surface the half/double alternates alongside the estimate so a + half/double detection error is visible, not silently locked in. Empty when + the BPM is unknown.""" + if bpm <= 0: + return [] + return sorted({round(bpm, 3), round(bpm / 2, 3), round(bpm * 2, 3)}) def _beats_per_bar(beats: list[float], downbeats: list[float]) -> int: """Estimate meter from spacing of downbeats within the beat sequence.""" if len(downbeats) < 2 or len(beats) < 2: return 4 - spb = 60.0 # placeholder, use beat spacing b = np.asarray(beats) d = np.asarray(downbeats) - # median beats between consecutive downbeats - counts = [] - for i in range(len(d) - 1): - counts.append(int(np.sum((b >= d[i]) & (b < d[i + 1])))) + counts = [int(np.sum((b >= d[i]) & (b < d[i + 1]))) for i in range(len(d) - 1)] counts = [c for c in counts if c > 0] - _ = spb return int(np.median(counts)) if counts else 4 + + +def _meter_confidence(beats: list[float], downbeats: list[float]) -> float: + """0..1 confidence in the meter: how consistent the beats-per-bar counts are + across the piece (all bars equal → 1.0). 0.0 when there is too little to tell.""" + if len(downbeats) < 3 or len(beats) < 2: + return 0.0 + b = np.asarray(beats) + d = np.asarray(downbeats) + counts = [int(np.sum((b >= d[i]) & (b < d[i + 1]))) for i in range(len(d) - 1)] + counts = [c for c in counts if c > 0] + if len(counts) < 2: + return 0.0 + arr = np.asarray(counts, dtype=np.float64) + med = float(np.median(arr)) + if med <= 0: + return 0.0 + frac_agree = float(np.mean(np.abs(arr - med) < 0.5)) # share of bars at the modal count + return float(np.clip(frac_agree, 0.0, 1.0)) diff --git a/src/stemforge/drum_midi.py b/src/stemforge/drum_midi.py index 8b0b925..5cb99a0 100644 --- a/src/stemforge/drum_midi.py +++ b/src/stemforge/drum_midi.py @@ -63,7 +63,13 @@ def transcribe( beat_grid: Any = None, ) -> dict[str, Any]: out_dir = Path(out_dir) - bpm = float(getattr(beat_grid, "source_bpm", 0) or 120.0) + # A real, measured tempo drives the .mid; when it's UNKNOWN (0.0) we still must + # write *some* tempo to serialize the file, so we use 120 but flag it as a + # serialization default (tempo_assumed) rather than passing it off as detected. + src_bpm = float(getattr(beat_grid, "source_bpm", 0) or 0.0) + tempo_assumed = src_bpm <= 0.0 + bpm = src_bpm if src_bpm > 0 else 120.0 + tempo_source = "beat_grid" if src_bpm > 0 else "midi_serialization_default" # 1) SOTA path: let ADTOF (or another ADT) write the MIDI, we adopt it. if cfg.external_cmd: @@ -71,7 +77,12 @@ def transcribe( # 2) Parts-based path: build the MIDI from separated parts. if drum_parts: - return _from_parts(drum_parts, cfg, out_dir, bpm) + res = _from_parts(drum_parts, cfg, out_dir, bpm) + if "file" in res: # annotate a produced MIDI with its tempo provenance + res["tempo_bpm"] = round(bpm, 3) + res["tempo_source"] = tempo_source + res["tempo_assumed"] = tempo_assumed + return res return { "skipped": ( diff --git a/src/stemforge/io_utils.py b/src/stemforge/io_utils.py index 8d69c2b..2be9581 100644 --- a/src/stemforge/io_utils.py +++ b/src/stemforge/io_utils.py @@ -9,6 +9,7 @@ import hashlib import json +import re import subprocess from dataclasses import dataclass from pathlib import Path @@ -20,6 +21,42 @@ PathLike = str | Path +# --------------------------------------------------------------------------- # +# Engine fallback evidence (shared by analysis + stretch) +# --------------------------------------------------------------------------- # +@dataclass +class EngineAttempt: + """One hop in an engine fallback chain — the analysis beat_this→librosa hop or + the stretch rubberband→signalsmith→librosa chain. + + ``status`` is ``"used"`` (this engine produced the output), ``"fell_through"`` + (it failed and the chain continued), or ``"failed"``. ``reason`` is a + sanitized, single-line message (no absolute paths or subprocess dumps), empty + when the engine was used. + """ + + engine: str + status: str + reason: str = "" + + def to_dict(self) -> dict[str, str]: + return {"engine": self.engine, "status": self.status, "reason": self.reason} + + +_PATH_RE = re.compile(r"(?:[A-Za-z]:\\|/)[\w./\\-]+") + + +def sanitize_reason(err: Any, limit: int = 200) -> str: + """A deterministic, log-safe reason string for a fallback record: drop + absolute paths, collapse newlines/whitespace (kills multi-line subprocess + dumps), and truncate — so manifests stay reproducible and leak no host paths. + """ + s = err if isinstance(err, str) else f"{type(err).__name__}: {err}" + s = _PATH_RE.sub("", s) + s = " ".join(s.split()) + return s[:limit] + + # --------------------------------------------------------------------------- # # Audio container # --------------------------------------------------------------------------- # diff --git a/src/stemforge/stretch.py b/src/stemforge/stretch.py index 0a6353c..470e085 100644 --- a/src/stemforge/stretch.py +++ b/src/stemforge/stretch.py @@ -16,7 +16,15 @@ import numpy as np -from .io_utils import AudioTensor, PathLike, load_audio, save_audio, slugify +from .io_utils import ( + AudioTensor, + EngineAttempt, + PathLike, + load_audio, + sanitize_reason, + save_audio, + slugify, +) try: # pragma: no cover from .orchestrator import StretchCfg @@ -41,6 +49,7 @@ def stretch_with_engine( engine: str = "rubberband", crisp: int = 5, preserve_formant: bool = False, + attempts: list[dict] | None = None, ) -> tuple[AudioTensor, str]: """Stretch and report the engine ACTUALLY used (after any fallback). @@ -48,20 +57,36 @@ def stretch_with_engine( the engine that actually produced the audio — so callers record the truth, not the request (the old code fell through to librosa without ever updating the label). Identity ratio does no work and reports ``"none"``. + + When an ``attempts`` list is supplied, each hop appends a sanitized + :class:`EngineAttempt` record — so the reason a higher-quality engine fell + through (previously swallowed by a bare ``except``) is captured as evidence. """ + def _rec(eng: str, status: str, reason: str = "") -> None: + if attempts is not None: + attempts.append(EngineAttempt(eng, status, reason).to_dict()) + if abs(ratio - 1.0) < 1e-6: + _rec("none", "used") return audio, "none" if engine == "rubberband": try: - return _rubberband(audio, ratio, crisp, preserve_formant), "rubberband" - except Exception: + out = _rubberband(audio, ratio, crisp, preserve_formant) + _rec("rubberband", "used") + return out, "rubberband" + except Exception as e: # noqa: BLE001 - degrade, but record why + _rec("rubberband", "fell_through", sanitize_reason(e)) engine = "signalsmith" if engine == "signalsmith": try: - return _signalsmith(audio, ratio), "signalsmith" - except Exception: - pass - return _librosa(audio, ratio), "librosa" + out = _signalsmith(audio, ratio) + _rec("signalsmith", "used") + return out, "signalsmith" + except Exception as e: # noqa: BLE001 - degrade, but record why + _rec("signalsmith", "fell_through", sanitize_reason(e)) + out = _librosa(audio, ratio) + _rec("librosa", "used") + return out, "librosa" def time_stretch( @@ -168,15 +193,18 @@ def match_bpm_file( return {"skipped": "no detectable BPM; pass source_bpm to override"} ratio = float(target_bpm) / src - out, used_engine = stretch_with_engine(audio, ratio, engine=engine, crisp=crisp) + attempts: list[dict] = [] + out, used_engine = stretch_with_engine(audio, ratio, engine=engine, crisp=crisp, + attempts=attempts) stem = out_name or f"{slugify(Path(input_path).stem)}_{int(round(float(target_bpm)))}bpm.wav" outp = save_audio(Path(out_dir) / stem, out) except Exception as e: # noqa: BLE001 - never raise; surface as a soft error - return {"error": f"match-bpm failed: {e}"} + return {"error": f"match-bpm failed: {sanitize_reason(e)}"} return { "engine": used_engine, # actual engine after any fallback "engine_requested": engine, + "attempts": attempts, # per-hop fallback evidence (sanitized) "detect_engine": detect_engine, "source_bpm": round(src, 3), "source_bpm_detected": round(detected, 3), @@ -213,6 +241,7 @@ def stretch_stems( "files": {}, } engines_used: list[str] = [] + attempts_all: list[dict] = [] for name, path in stems.items(): target = cfg.per_stem_target_bpm.get(name, cfg.target_bpm) if not target: @@ -221,7 +250,7 @@ def stretch_stems( audio = load_audio(path) formant = cfg.preserve_formant and name == "vocals" out, used = stretch_with_engine(audio, ratio, engine=cfg.engine, crisp=cfg.crisp, - preserve_formant=formant) + preserve_formant=formant, attempts=attempts_all) engines_used.append(used) outp = save_audio(out_dir / f"{name}_{int(round(float(target)))}bpm.wav", out) result["ratios"][name] = round(ratio, 4) @@ -229,4 +258,14 @@ def stretch_stems( if engines_used: uniq = sorted(set(engines_used)) result["engine"] = uniq[0] if len(uniq) == 1 else uniq + # Dedupe the per-stem hops (the chain is the same for every stem) but keep the + # sanitized fall-through reasons as manifest evidence. + seen: set[tuple[str, str, str]] = set() + deduped: list[dict] = [] + for a in attempts_all: + key = (a["engine"], a["status"], a["reason"]) + if key not in seen: + seen.add(key) + deduped.append(a) + result["attempts"] = deduped return result From 97443ad71e38d1f671a041c98de22dbbd9de010b Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 02:48:59 +0000 Subject: [PATCH 03/11] test: cover BPM confidence/candidates, meter, chains, and stretch/drum wiring Round out PR-A2 acceptance beyond the silent-120 regression: - known beat sequence still yields ~120 with high confidence; candidates always include half/double; unknown BPM has zero confidence and no candidates. - meter_confidence is high for a clean 4/4 grid, lower for ragged bars. - BeatGrid.to_dict carries the evidence fields. - analyze records the beat_this->librosa fallback chain with a sanitized reason; an un-estimable tempo is source_bpm 0.0 + tempo_assumed; the librosa meter is flagged assumed with low meter_confidence. - EngineAttempt.to_dict + sanitize_reason (path scrub, newline collapse, deterministic). - match_bpm_file manifest carries attempts[]; stretch_with_engine records fell_through hops with sanitized reasons. - drum_midi flags the 120 serialization default vs a measured beat_grid tempo. - source guard: analysis has no silent-120 fallback. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- tests/test_tempo_evidence.py | 195 ++++++++++++++++++++++++++++++++++- 1 file changed, 191 insertions(+), 4 deletions(-) diff --git a/tests/test_tempo_evidence.py b/tests/test_tempo_evidence.py index 5b027c6..1de45f5 100644 --- a/tests/test_tempo_evidence.py +++ b/tests/test_tempo_evidence.py @@ -2,19 +2,206 @@ Receipts (verified at main): analysis.py silently returned 120 BPM on several failure paths (`:96,:113,:119,:123`), `:98` assumed 4/4 (`downbeats=beats[::4]`), -and BeatGrid carried no confidence. This regression pins the first fix: an -un-estimable tempo must be UNKNOWN (0.0), never a silent 120. +and BeatGrid carried no confidence. drum_midi baked 120 into the .mid when the +grid was absent. stretch swallowed fallback reasons at both hops. This pins the +fixes: unknown tempo is 0.0 (never a silent 120) with confidence/candidates, +the librosa meter is flagged assumed, and both analysis and stretch record a +shared, sanitized EngineAttempt chain. + +GPU-free and deterministic (beat_this is absent, so analyze uses librosa). """ from __future__ import annotations -from stemforge.analysis import _bpm_from_beats +import inspect +import types +from pathlib import Path + +import numpy as np +import soundfile as sf + +from stemforge import analysis, drum_midi +from stemforge.analysis import ( + BeatGrid, + _bpm_candidates, + _bpm_confidence, + _bpm_from_beats, + _meter_confidence, + analyze, +) +from stemforge.io_utils import AudioTensor, EngineAttempt, ensure_dir, sanitize_reason +from stemforge.orchestrator import load_config +from stemforge.stretch import match_bpm_file + +SR = 44100 + + +def _burst_track(times: list[float], dur: float = 2.5, tau: float = 0.05, seed: int = 0) -> np.ndarray: + """Broadband decaying-noise hits (librosa onset_detect finds these).""" + rng = np.random.default_rng(seed) + y = np.zeros(int(dur * SR), dtype=np.float32) + for t0 in times: + i = int(t0 * SR) + n = int(0.3 * SR) + tt = np.arange(n) / SR + seg = (rng.standard_normal(n) * np.exp(-tt / tau)).astype(np.float32) + end = min(y.size, i + n) + y[i:end] += seg[: end - i] + return y + + +def _write(path: Path, y: np.ndarray) -> str: + sf.write(str(path), y, SR, subtype="FLOAT") + return str(path) + + +def _two_clicks() -> AudioTensor: + y = np.zeros(SR, dtype=np.float32) + y[100:130] = 0.8 + y[SR // 2:SR // 2 + 30] = 0.8 + return AudioTensor(y, SR) # --------------------------------------------------------------------------- # -# RED: an un-estimable tempo is UNKNOWN (0.0), never a silent 120 +# commit 1 — RED: an un-estimable tempo is UNKNOWN (0.0), never a silent 120 # --------------------------------------------------------------------------- # def test_bpm_unknown_is_zero_not_silent_120(): assert _bpm_from_beats([1.0, 2.0], regression=True) == 0.0 # <3 beats assert _bpm_from_beats([3.0, 2.0, 1.0], regression=True) == 0.0 # non-positive slope assert _bpm_from_beats([1.0, 1.0, 1.0], regression=False) == 0.0 # non-positive median + + +# --------------------------------------------------------------------------- # +# commit 3 — evidence, candidates, meter confidence, chains, drum/stretch wiring +# --------------------------------------------------------------------------- # +def test_bpm_known_sequence_and_confidence(): + beats = [i * 0.5 for i in range(9)] # exactly 120 BPM + assert abs(_bpm_from_beats(beats, regression=True) - 120.0) < 0.5 # contract preserved + assert _bpm_confidence(beats, regression=True) > 0.99 # near-perfect line + assert _bpm_confidence([1.0, 2.0], regression=True) == 0.0 # unknown -> 0 conf + + +def test_bpm_candidates_include_half_and_double(): + assert _bpm_candidates(120.0) == [60.0, 120.0, 240.0] + assert _bpm_candidates(0.0) == [] # unknown -> no candidates + + +def test_meter_confidence_consistent_vs_irregular(): + beats = [i * 0.5 for i in range(17)] + downbeats = beats[::4] # a clean 4/4 grid + assert _meter_confidence(beats, downbeats) > 0.9 + irregular = [beats[0], beats[3], beats[9], beats[11]] # ragged bar spacing + assert _meter_confidence(beats, irregular) < _meter_confidence(beats, downbeats) + + +def test_beatgrid_to_dict_carries_evidence(): + g = BeatGrid(source_bpm=120.0, beats=[0.0, 0.5, 1.0, 1.5], downbeats=[0.0, 1.0], + bpm_confidence=0.9, bpm_candidates=[60.0, 120.0, 240.0], + meter_confidence=0.8, tempo_assumed=False, + fallback_chain=[{"engine": "beat_this", "status": "used", "reason": ""}]) + d = g.to_dict() + assert d["source_bpm"] == 120.0 + for k in ("bpm_confidence", "bpm_candidates", "meter_confidence", "tempo_assumed", + "fallback_chain"): + assert k in d + assert d["tempo_assumed"] is False + assert d["bpm_candidates"] == [60.0, 120.0, 240.0] + + +def test_analyze_records_fallback_chain(): + """beat_this is absent -> analyze records a fell_through hop with a sanitized + reason, then librosa used.""" + g = analyze(_two_clicks()) # default engine="beat_this" -> falls to librosa + assert g.engine == "librosa" + chain = g.fallback_chain + assert [c["engine"] for c in chain] == ["beat_this", "librosa"] + assert chain[0]["status"] == "fell_through" and chain[0]["reason"] + assert chain[1]["status"] == "used" + assert "/" not in chain[0]["reason"] or "" in chain[0]["reason"] + + +def test_analyze_unknown_source_bpm_is_flagged(): + g = analyze(_two_clicks(), engine="librosa") + assert g.source_bpm == 0.0 # unknown, not 120 + assert g.tempo_assumed is True + assert g.bpm_confidence == 0.0 + assert g.bpm_candidates == [] + + +def test_librosa_path_marks_meter_assumed(): + g = analyze(_two_clicks(), engine="librosa") + assert g.tempo_assumed is True # librosa downbeats are a 4/4 assumption + assert g.meter_confidence <= 0.2 + + +def test_engine_attempt_and_sanitize_reason(): + assert EngineAttempt("rubberband", "fell_through", "boom").to_dict() == { + "engine": "rubberband", "status": "fell_through", "reason": "boom"} + r = sanitize_reason("failed at /usr/local/bin/rubberband\nstack line 2\nline 3") + assert "/usr/local" not in r and "" in r + assert "\n" not in r # subprocess dump collapsed + assert sanitize_reason(RuntimeError("x")) == "RuntimeError: x" # deterministic + + +def test_match_bpm_file_manifest_carries_attempts(tmp_path, sine): + samples, sr = sine + wav = _write(tmp_path / "loop.wav", samples.T if samples.ndim == 2 else samples) + res = match_bpm_file(Path(wav), target_bpm=140, out_dir=tmp_path / "m", + source_bpm=120, engine="librosa") + assert "attempts" in res and isinstance(res["attempts"], list) + assert res["attempts"], "stretch manifest must carry a non-empty attempts[]" + last = res["attempts"][-1] + assert last["engine"] == "librosa" and last["status"] == "used" + + +def test_stretch_records_fallthrough_reasons(monkeypatch, sine): + from stemforge import stretch as st + + def raise_rb(*a, **k): + raise RuntimeError("rubberband missing at /usr/bin/rubberband") + + def raise_ss(*a, **k): + raise RuntimeError("signalsmith boom") + + monkeypatch.setattr(st, "_rubberband", raise_rb) + monkeypatch.setattr(st, "_signalsmith", raise_ss) + attempts: list[dict] = [] + _out, used = st.stretch_with_engine(AudioTensor(*sine), 1.5, engine="rubberband", + attempts=attempts) + assert used == "librosa" + kinds = {(a["engine"], a["status"]) for a in attempts} + assert ("rubberband", "fell_through") in kinds + assert ("signalsmith", "fell_through") in kinds + assert ("librosa", "used") in kinds + rb = next(a for a in attempts if a["engine"] == "rubberband") + assert "" in rb["reason"] and "/usr/bin" not in rb["reason"] + + +def _drum_parts(tmp_path) -> dict[str, str]: + return {"kick": _write(tmp_path / "kick.wav", _burst_track([0.1, 0.6, 1.1, 1.6]))} + + +def test_drum_midi_flags_serialization_default_tempo(tmp_path): + res = drum_midi.transcribe(None, _drum_parts(tmp_path), load_config().drums.midi, + ensure_dir(tmp_path / "out"), beat_grid=None) + assert "file" in res + assert res["tempo_source"] == "midi_serialization_default" + assert res["tempo_assumed"] is True + assert res["tempo_bpm"] == 120.0 + + +def test_drum_midi_uses_measured_tempo(tmp_path): + grid = types.SimpleNamespace(source_bpm=128.0) + res = drum_midi.transcribe(None, _drum_parts(tmp_path), load_config().drums.midi, + ensure_dir(tmp_path / "out"), beat_grid=grid) + assert res["tempo_source"] == "beat_grid" + assert res["tempo_assumed"] is False + assert res["tempo_bpm"] == 128.0 + + +def test_analysis_source_has_no_silent_120_fallback(): + """Guard against reintroducing a silent 120: the BPM helpers must not + manufacture a tempo on failure.""" + src = inspect.getsource(analysis) + assert "return 120" not in src + assert "else 120" not in src From 042c80c156af412b0d730ddd6a7f6d34962f89b7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 03:10:20 +0000 Subject: [PATCH 04/11] test: add failing regression for cross-part drum bleed duplicates MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit drum_midi had no cross-part de-duplication, so separation bleed — a weak phantom onset in one part coincident with a loud hit in another — survived as a duplicate coincident note. This regression drops a weak phantom into the toms part on top of each loud kick and asserts no coincident cross-pitch notes survive within 18 ms. It fails on the current tree (3 bleed duplicates survive), pinning the loss. Receipt: drum_midi.py parts-based path (no cross-part de-dup) at main. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- tests/test_drum_calibration.py | 79 ++++++++++++++++++++++++++++++++++ 1 file changed, 79 insertions(+) create mode 100644 tests/test_drum_calibration.py diff --git a/tests/test_drum_calibration.py b/tests/test_drum_calibration.py new file mode 100644 index 0000000..565b95b --- /dev/null +++ b/tests/test_drum_calibration.py @@ -0,0 +1,79 @@ +"""Drum calibration — A-series PR-A4. + +Receipt (verified at main): drum_midi had NO cross-part de-dup, so separation +bleed (a weak phantom in one part coincident with a loud hit in another) survived +as a duplicate coincident note. This regression pins that: coincident cross-part +bleed must not produce duplicate notes. + +GPU-free and deterministic — synthetic hits, seeded RNG. +""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import pytest +import soundfile as sf + +pretty_midi = pytest.importorskip("pretty_midi") + +from stemforge import drum_midi # noqa: E402 +from stemforge.io_utils import ensure_dir # noqa: E402 +from stemforge.orchestrator import load_config # noqa: E402 + +SR = 44100 + + +def _noise(dur: float, tau: float, seed: int, amp: float = 1.0) -> np.ndarray: + rng = np.random.default_rng(seed) + t = np.arange(int(dur * SR)) / SR + return (amp * rng.standard_normal(t.size) * np.exp(-t / tau)).astype(np.float32) + + +def _sine(dur: float, hz: float, tau: float, amp: float = 0.9) -> np.ndarray: + t = np.arange(int(dur * SR)) / SR + return (amp * np.sin(2 * np.pi * hz * t) * np.exp(-t / tau)).astype(np.float32) + + +def _place(hits: list[tuple[float, np.ndarray]], total_s: float = 2.0) -> np.ndarray: + y = np.zeros(int(total_s * 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 + + +def _write(path: Path, y: np.ndarray) -> str: + sf.write(str(path), y, SR, subtype="FLOAT") + return str(path) + + +def _run(parts, tmp_path, **cfg_over): + cfg = load_config().drums.midi + for k, v in cfg_over.items(): + setattr(cfg, k, v) + return drum_midi._from_parts(parts, cfg, ensure_dir(tmp_path / "out"), bpm=120.0) + + +# --------------------------------------------------------------------------- # +# RED — separation bleed must not produce coincident cross-part duplicates +# --------------------------------------------------------------------------- # +def test_cross_part_bleed_is_deduped(tmp_path): + """A weak phantom in the toms part, coincident with each loud kick (bleed), + must not survive as a duplicate note. Fails pre-fix (no cross-part de-dup).""" + kick = _place([(0.3, _noise(0.3, 0.05, 1, amp=1.0)), + (0.9, _noise(0.3, 0.05, 2, amp=1.0)), + (1.5, _noise(0.3, 0.05, 3, amp=1.0))]) + toms = _place([(0.304, _noise(0.3, 0.05, 4, amp=0.12)), # phantom bleed x3 + (0.904, _noise(0.3, 0.05, 5, amp=0.12)), + (1.504, _noise(0.3, 0.05, 6, amp=0.12)), + (0.6, _sine(0.4, 90.0, 0.12, amp=0.9))]) # one real (low) tom + res = _run({"kick": _write(tmp_path / "kick.wav", kick), + "toms": _write(tmp_path / "toms.wav", toms)}, tmp_path) + mid = pretty_midi.PrettyMIDI(res["file"]) + times = sorted((round(n.start, 4), n.pitch) for inst in mid.instruments for n in inst.notes) + coincident = sum(1 for a, b in zip(times, times[1:]) + if b[0] - a[0] <= 0.018 and a[1] != b[1]) + assert coincident == 0, f"cross-part bleed duplicates survived: {coincident}" From d73f261f39002ad63d96eacd7dad9d04c50632a3 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 03:10:54 +0000 Subject: [PATCH 05/11] =?UTF-8?q?feat:=20calibrate=20drum=20MIDI=20?= =?UTF-8?q?=E2=80=94=20per-part=20velocity,=20bleed=20de-dup,=20tom=20spli?= =?UTF-8?q?t,=20config?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rework the parts-based drum transcription around calibrated, configurable stages, and record the per-event evidence. - Per-part velocity normalization against each part's ~98th-percentile strength (was a single global peak), then a kit-level balance gain — a quiet hi-hat and a loud kick both use the full velocity range and ghost notes stay audible. - Cross-part de-dup: cluster onsets within dedupe_window_ms and drop a coincident cross-part hit clearly weaker than the loudest (separation bleed), keeping the most probable. Comparable simultaneous hits (real kick+snare) and same-part ghosts survive. Manifest reports duplicates_removed. - Toms split low/mid/high by dominant pitch (GM 45/47/50); off -> single mid tom. - Hi-hat open/closed decay windows + ratio, and the PR-A1 cymbal thresholds, are moved into DrumMidiCfg (config-ified) with a validate() that fails fast on an out-of-range value; load_config runs it. - Manifest gains a per-event record (time, part, note, raw_strength, normalized_strength, velocity). Turns the bleed regression green. Inherited tests adapted to the new velocity function and the now-config-ified cymbal knobs. Receipt: drum_midi.py velocity/global-peak, no cross-part de-dup, single tom note, hard-coded hat test (verified at main). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- configs/default.yaml | 20 ++++ src/stemforge/drum_midi.py | 178 +++++++++++++++++++++++++------- src/stemforge/orchestrator.py | 43 +++++++- tests/test_drum_midi_cymbals.py | 12 +-- tests/test_dsp.py | 11 +- 5 files changed, 212 insertions(+), 52 deletions(-) diff --git a/configs/default.yaml b/configs/default.yaml index cb3a6bd..98850ca 100644 --- a/configs/default.yaml +++ b/configs/default.yaml @@ -96,6 +96,26 @@ drums: velocity_window_ms: 50 velocity_hop_ms: 10 external_cmd: null + # --- Calibration (PR-A4) --- + velocity_percentile: 98 # per-part reference level (robust to a stray peak) + dedupe_window_ms: 18 # cross-part coincidence window + dedupe_weak_ratio: 0.5 # drop a coincident cross-part hit weaker than this x the loudest (bleed) + hat_open_short_ms: 25 # hi-hat open/closed decay test + hat_open_long_ms: 130 + hat_open_ratio: 0.35 + tom_split: true # split toms low/mid/high by dominant pitch + tom_low_hz: 100 + tom_mid_hz: 220 + cymbal_hf_cut_hz: 6000 # cymbal onset classifier (config-ified from PR-A1) + cymbal_min_hf_ratio: 0.30 + cymbal_min_peak: 0.001 + cymbal_attack_ms: 30 + cymbal_sustain_ms: 180 + cymbal_closed_decay_max: 0.30 + cymbal_crash_decay_min: 0.55 + cymbal_bright_centroid_hz: 7000 + cymbal_ambiguous: generic # generic | drop + cymbal_generic_note: 42 output: root: out # per-song bundle: out//{stems,midi,drums,stretched} diff --git a/src/stemforge/drum_midi.py b/src/stemforge/drum_midi.py index 5cb99a0..4c3c4d2 100644 --- a/src/stemforge/drum_midi.py +++ b/src/stemforge/drum_midi.py @@ -29,7 +29,7 @@ import shutil import subprocess from pathlib import Path -from typing import Any, Callable +from typing import Any import numpy as np @@ -47,7 +47,10 @@ "hihat": 42, # closed by default "hihat_closed": 42, "hihat_open": 46, - "toms": 47, # mid tom + "toms": 47, # mid tom (default when tom-split is off) + "toms_low": 45, # low/floor tom + "toms_mid": 47, + "toms_high": 50, # high tom "ride": 51, "crash": 49, } @@ -156,16 +159,15 @@ def canonical_drum_part(part: str) -> str | None: # --------------------------------------------------------------------------- # -# Parts-based transcription (fully implemented) +# Parts-based transcription (calibrated — PR-A4) # --------------------------------------------------------------------------- # def _from_parts(drum_parts: dict[str, str], cfg: "DrumMidiCfg", out_dir: Path, bpm: float) -> dict[str, Any]: import pretty_midi - events: list[tuple[float, int, float]] = [] # (start, note, raw_peak) - global_peak = 1e-9 + raw_events: list[dict[str, Any]] = [] # {time, part, note, raw_strength} counts: dict[str, int] = {} - cymbal_classes: dict[str, int] = {} # GM-class -> count (default-path evidence) - cymbal_rejected = 0 # onsets in a cymbal bucket that were gated/dropped + cymbal_classes: dict[str, int] = {} # GM-class -> count (default-path evidence) + cymbal_rejected = 0 # onsets in a cymbal bucket that were gated/dropped for part, file in drum_parts.items(): canon = canonical_drum_part(part) @@ -185,35 +187,43 @@ def _peak(t: float) -> float: return _peak_in_window(env, env_t, t, cfg.velocity_window_ms / 1000.0) \ if cfg.velocity_from_stems else 1.0 - if canon == "cymbals": - for t in onsets: + for t in onsets: + if canon == "cymbals": cls = _classify_cymbal(_cymbal_features(y, sr, float(t), cfg), cfg) note = _cymbal_note(cls, cfg) - if note is None: # gated (not a cymbal) or ambiguous-dropped + if note is None: # gated (not a cymbal) or ambiguous-dropped cymbal_rejected += 1 continue - peak = _peak(t) - events.append((float(t), note, float(peak))) - global_peak = max(global_peak, peak) label = cls if cls in GM else "generic" cymbal_classes[label] = cymbal_classes.get(label, 0) + 1 - else: - note_fn = _note_selector(canon, cfg, y, sr) - for t in onsets: - peak = _peak(t) - events.append((float(t), note_fn(t), float(peak))) - global_peak = max(global_peak, peak) - - if not events: + elif canon == "toms": + note = _tom_note(y, sr, float(t), cfg) + elif canon == "hihat": + note = _hihat_note(y, sr, float(t), cfg) + else: + note = GM.get(canon, GM["snare"]) + raw_events.append({"time": float(t), "part": canon, "note": int(note), + "raw_strength": float(_peak(t))}) + + if not raw_events: return {"skipped": "no onsets detected in drum parts"} + _normalize_velocities(raw_events, cfg) # per-part, then kit balance + kept, duplicates_removed = _dedupe_cross_part(raw_events, cfg) + pm = pretty_midi.PrettyMIDI(initial_tempo=bpm) inst = pretty_midi.Instrument(program=0, is_drum=True, name="drums") - for start, note, peak in events: - vel = _scale_velocity(peak, global_peak) if cfg.velocity_from_stems else 100 - inst.notes.append(pretty_midi.Note(velocity=vel, pitch=note, - start=start, end=start + NOTE_LEN_S)) - inst.notes.sort(key=lambda n: n.start) + records: list[dict[str, Any]] = [] + for e in sorted(kept, key=lambda x: x["time"]): + vel = _velocity_from_norm(e["normalized_strength"]) if cfg.velocity_from_stems else 100 + inst.notes.append(pretty_midi.Note(velocity=vel, pitch=e["note"], + start=e["time"], end=e["time"] + NOTE_LEN_S)) + records.append({ + "time": round(e["time"], 4), "part": e["part"], "note": e["note"], + "raw_strength": round(e["raw_strength"], 6), + "normalized_strength": round(e["normalized_strength"], 4), + "velocity": vel, + }) pm.instruments.append(inst) target = out_dir / "drums.mid" @@ -221,19 +231,107 @@ def _peak(t: float) -> float: return { "source": "parts", "file": str(target), - "note_count": len(events), + "note_count": len(kept), "onsets_per_part": counts, "expanded_7class": bool(cfg.expand_to_7class), "cymbal_classes": cymbal_classes, "cymbal_rejected": cymbal_rejected, + "duplicates_removed": duplicates_removed, + "events": records, } -def _note_selector(part: str, cfg: "DrumMidiCfg", y: np.ndarray, sr: int) -> Callable[[float], int]: - """Return a function onset_time -> GM note, applying 5->7 hi-hat expansion.""" - if part == "hihat" and cfg.expand_to_7class: - return lambda t: GM["hihat_open"] if _hat_is_open(y, sr, t) else GM["hihat_closed"] - return lambda t, n=GM.get(part, GM["snare"]): n +# --------------------------------------------------------------------------- # +# Calibration: per-part velocity normalization, cross-part de-dup, part voicing +# --------------------------------------------------------------------------- # +def _normalize_velocities(events: list[dict[str, Any]], cfg: "DrumMidiCfg") -> None: + """Normalize each part against its own ~98th-percentile strength (so a quiet + hi-hat and a loud kick both use the full velocity range and ghost notes stay + audible), then apply the kit-level balance gain. Adds ``normalized_strength``.""" + pct = float(getattr(cfg, "velocity_percentile", 98.0)) + balance = getattr(cfg, "kit_balance", {}) or {} + by_part: dict[str, list[dict[str, Any]]] = {} + for e in events: + by_part.setdefault(e["part"], []).append(e) + for part, evs in by_part.items(): + strengths = np.asarray([e["raw_strength"] for e in evs], dtype=np.float64) + ref = float(np.percentile(strengths, pct)) if strengths.size else 1.0 + if ref <= 1e-9: + ref = float(strengths.max()) if strengths.size else 1.0 + ref = ref if ref > 1e-9 else 1.0 + gain = float(balance.get(part, 1.0)) + for e in evs: + e["normalized_strength"] = float(np.clip(e["raw_strength"] / ref * gain, 0.0, 1.0)) + + +def _dedupe_cross_part(events: list[dict[str, Any]], cfg: "DrumMidiCfg", + ) -> tuple[list[dict[str, Any]], int]: + """Cluster onsets within ``dedupe_window_ms`` and drop a CROSS-part hit that is + clearly weaker than the loudest in the cluster (separation bleed) — keeping the + most probable. Comparable simultaneous hits (a real kick+snare) and same-part + ghost notes survive, since only clearly-weaker cross-part coincidences go.""" + if not events: + return [], 0 + win = float(getattr(cfg, "dedupe_window_ms", 18.0)) / 1000.0 + ratio = float(getattr(cfg, "dedupe_weak_ratio", 0.5)) + order = sorted(events, key=lambda e: e["time"]) + kept: list[dict[str, Any]] = [] + removed = 0 + i = 0 + while i < len(order): + j = i + while j + 1 < len(order) and order[j + 1]["time"] - order[i]["time"] <= win: + j += 1 + cluster = order[i:j + 1] + if len(cluster) == 1: + kept.append(cluster[0]) + else: + strongest = max(cluster, key=lambda e: e["raw_strength"]) + for e in cluster: + if e is strongest: + kept.append(e) + elif e["part"] != strongest["part"] and \ + e["raw_strength"] < ratio * strongest["raw_strength"]: + removed += 1 # separation bleed → drop + else: + kept.append(e) # same-part or comparable → keep + i = j + 1 + return kept, removed + + +def _velocity_from_norm(norm: float) -> int: + """Normalized strength (0..1) -> MIDI velocity with a floor so ghosts stay audible.""" + ratio = float(np.clip(norm, 0.0, 1.0)) ** 0.5 # perceptual-ish compression + return int(np.clip(round(10 + 117 * ratio), 1, 127)) + + +def _hihat_note(y: np.ndarray, sr: int, t: float, cfg: "DrumMidiCfg") -> int: + if cfg.expand_to_7class and _hat_is_open(y, sr, t, cfg): + return GM["hihat_open"] + return GM["hihat_closed"] + + +def _tom_note(y: np.ndarray, sr: int, t: float, cfg: "DrumMidiCfg") -> int: + """Split toms into low/mid/high by the onset's dominant pitch.""" + if not getattr(cfg, "tom_split", True): + return GM["toms"] + hz = _dominant_hz(y, sr, t) + if hz < float(getattr(cfg, "tom_low_hz", 100.0)): + return GM["toms_low"] + if hz < float(getattr(cfg, "tom_mid_hz", 220.0)): + return GM["toms_mid"] + return GM["toms_high"] + + +def _dominant_hz(y: np.ndarray, sr: int, t: float, win_ms: float = 60.0) -> float: + """Dominant (peak-magnitude) frequency in a short window after the onset.""" + i0 = max(0, int(t * sr)) + seg = y[i0:i0 + max(1, int(win_ms / 1000.0 * sr))].astype(np.float64) + if seg.size < 4: + return 0.0 + spec = np.abs(np.fft.rfft(seg * np.hanning(seg.size))) + freqs = np.fft.rfftfreq(seg.size, d=1.0 / sr) + return float(freqs[int(np.argmax(spec))]) # --------------------------------------------------------------------------- # @@ -357,20 +455,20 @@ def _peak_in_window(env: np.ndarray, env_t: np.ndarray, t: float, window_s: floa return float(np.max(env[mask])) -def _scale_velocity(peak: float, global_peak: float) -> int: - """Perceptual-ish scaling: sqrt compression, floor at 10 so hits stay audible.""" - ratio = (peak / global_peak) ** 0.5 if global_peak > 0 else 0.0 - return int(np.clip(round(10 + 117 * ratio), 1, 127)) - +def _hat_is_open(y: np.ndarray, sr: int, t: float, cfg: "DrumMidiCfg" = None) -> bool: + """Open vs closed hi-hat via decay: open hats sustain, closed die fast. -def _hat_is_open(y: np.ndarray, sr: int, t: float, short_ms: float = 25.0, long_ms: float = 130.0) -> bool: - """Open vs closed hi-hat via decay: open hats sustain, closed die fast.""" + Windows and the decay ratio are configurable (PR-A4); the defaults reproduce + the previous hard-coded behavior.""" + short_ms = _cfg_num(cfg, "hat_open_short_ms", 25.0) if cfg is not None else 25.0 + long_ms = _cfg_num(cfg, "hat_open_long_ms", 130.0) if cfg is not None else 130.0 + ratio = _cfg_num(cfg, "hat_open_ratio", 0.35) if cfg is not None else 0.35 i0 = int(t * sr) a = _rms_at(y, i0, int(short_ms / 1000.0 * sr)) b = _rms_at(y, i0 + int(long_ms / 1000.0 * sr), int(short_ms / 1000.0 * sr)) if a <= 1e-9: return False - return (b / a) > 0.35 # still ringing well after the hit => open + return (b / a) > ratio # still ringing well after the hit => open def _rms_at(y: np.ndarray, start: int, length: int) -> float: diff --git a/src/stemforge/orchestrator.py b/src/stemforge/orchestrator.py index 90134ee..ef5c86e 100644 --- a/src/stemforge/orchestrator.py +++ b/src/stemforge/orchestrator.py @@ -201,6 +201,45 @@ class DrumMidiCfg: velocity_window_ms: float = 50.0 velocity_hop_ms: float = 10.0 external_cmd: str | None = None + # --- Calibration (PR-A4) --------------------------------------------- # + velocity_percentile: float = 98.0 # per-part reference level (robust to a stray peak) + dedupe_window_ms: float = 18.0 # cross-part coincidence window + dedupe_weak_ratio: float = 0.5 # drop a coincident cross-part hit weaker than this × the loudest + # Hi-hat open/closed decay test (was hard-coded). + hat_open_short_ms: float = 25.0 + hat_open_long_ms: float = 130.0 + hat_open_ratio: float = 0.35 + # Toms split low/mid/high by dominant pitch. + tom_split: bool = True + tom_low_hz: float = 100.0 # < low_hz => low tom + tom_mid_hz: float = 220.0 # < mid_hz => mid tom, else high tom + # Cymbal onset classifier (config-ified from PR-A1 module defaults). + cymbal_hf_cut_hz: float = 6000.0 + cymbal_min_hf_ratio: float = 0.30 + cymbal_min_peak: float = 1e-3 + cymbal_attack_ms: float = 30.0 + cymbal_sustain_ms: float = 180.0 + cymbal_closed_decay_max: float = 0.30 + cymbal_crash_decay_min: float = 0.55 + cymbal_bright_centroid_hz: float = 7000.0 + cymbal_ambiguous: str = "generic" # "generic" | "drop" + cymbal_generic_note: int = 42 + + def validate(self) -> "DrumMidiCfg": + """Fail fast on an out-of-range calibration value (config validated, A4).""" + if not 0.0 < self.velocity_percentile <= 100.0: + raise ValueError(f"velocity_percentile must be in (0,100], got {self.velocity_percentile}") + if self.dedupe_window_ms < 0: + raise ValueError(f"dedupe_window_ms must be >= 0, got {self.dedupe_window_ms}") + if not 0.0 <= self.dedupe_weak_ratio <= 1.0: + raise ValueError(f"dedupe_weak_ratio must be in [0,1], got {self.dedupe_weak_ratio}") + if self.cymbal_ambiguous not in {"generic", "drop"}: + raise ValueError(f"cymbal_ambiguous must be 'generic' or 'drop', got {self.cymbal_ambiguous!r}") + if not 0.0 <= self.cymbal_min_hf_ratio <= 1.0: + raise ValueError(f"cymbal_min_hf_ratio must be in [0,1], got {self.cymbal_min_hf_ratio}") + if self.tom_low_hz > self.tom_mid_hz: + raise ValueError(f"tom_low_hz ({self.tom_low_hz}) must be <= tom_mid_hz ({self.tom_mid_hz})") + return self @dataclass @@ -264,7 +303,9 @@ def load_config(path: PathLike | None = None, overrides: dict[str, Any] | None = overrides["separation.preset"] = key # normalized for the manifest for dotted, value in overrides.items(): _set_dotted(data, dotted, value) - return _build(RunConfig, data) + cfg = _build(RunConfig, data) + cfg.drums.midi.validate() # fail fast on an out-of-range drum-calibration value (A4) + return cfg def _set_dotted(d: dict[str, Any], dotted: str, value: Any) -> None: diff --git a/tests/test_drum_midi_cymbals.py b/tests/test_drum_midi_cymbals.py index c388928..2865ca1 100644 --- a/tests/test_drum_midi_cymbals.py +++ b/tests/test_drum_midi_cymbals.py @@ -217,12 +217,12 @@ def test_manifest_reports_cymbal_counts(tmp_path): for k in res["cymbal_classes"]) -def test_stock_config_needs_no_new_fields(tmp_path): - """Prior configs parse unchanged: the stock DrumMidiCfg has no cymbal_* fields, - yet the cymbal path runs with module defaults (getattr fallbacks).""" +def test_cymbal_config_defaults_and_still_voices(tmp_path): + """The cymbal knobs are config-ified (PR-A4) with sane defaults; a stock config + carries them and the cymbal path still voices via the generic policy.""" cfg = _midi_cfg() - assert not hasattr(cfg, "cymbal_ambiguous") # not a dataclass field - assert not hasattr(cfg, "cymbal_generic_note") + assert cfg.cymbal_ambiguous == "generic" + assert cfg.cymbal_generic_note == 42 parts = {"other": _write(tmp_path / "other.wav", _cymbal_bucket())} res = drum_midi._from_parts(parts, cfg, ensure_dir(tmp_path / "out"), bpm=120.0) - assert sum(res["cymbal_classes"].values()) >= 1 # defaults -> generic policy, still voices + assert sum(res["cymbal_classes"].values()) >= 1 diff --git a/tests/test_dsp.py b/tests/test_dsp.py index f28037b..3e334ca 100644 --- a/tests/test_dsp.py +++ b/tests/test_dsp.py @@ -1,6 +1,6 @@ import numpy as np -from stemforge.drum_midi import GM, _hat_is_open, _peak_in_window, _rms_at, _scale_velocity +from stemforge.drum_midi import GM, _hat_is_open, _peak_in_window, _rms_at, _velocity_from_norm def test_gm_mapping(): @@ -8,12 +8,13 @@ def test_gm_mapping(): assert GM["hihat_open"] == 46 and GM["hihat_closed"] == 42 -def test_scale_velocity_monotonic_and_bounded(): - vals = [_scale_velocity(p, 1.0) for p in (0.0, 0.25, 0.5, 1.0)] +def test_velocity_from_norm_monotonic_and_bounded(): + # PR-A4: velocity now maps normalized (0..1) per-part strength (was global-peak) + vals = [_velocity_from_norm(p) for p in (0.0, 0.25, 0.5, 1.0)] assert vals == sorted(vals) assert all(1 <= v <= 127 for v in vals) - assert vals[-1] == 127 # peak == global => full velocity - assert _scale_velocity(0.0, 1.0) >= 1 # floor keeps quiet hits audible + assert vals[-1] == 127 # full normalized strength => full velocity + assert _velocity_from_norm(0.0) >= 1 # floor keeps quiet/ghost hits audible def test_peak_in_window(): From 1acb2d60dbd7403064d73da6c2ccb9f222f88662 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 03:11:08 +0000 Subject: [PATCH 06/11] test: cover per-part velocity, ghost survival, tom split, and config validation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round out PR-A4 acceptance beyond the bleed regression: - per-part velocity normalization: a quiet part still reaches a strong velocity for its loudest hit, and every hit (incl. a ghost) stays above the audible floor — ghost notes survive. - toms split low/mid/high by dominant pitch; tom_split off collapses to one note. - a real, comparable kick+snare on the same beat is NOT treated as bleed (both survive; duplicates_removed == 0). - manifest carries per-event raw/normalized strength + velocity. - config validation rejects out-of-range values (percentile, weak-ratio, cymbal_ambiguous, tom bounds) directly and via load_config. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- tests/test_drum_calibration.py | 103 ++++++++++++++++++++++++++++++--- 1 file changed, 95 insertions(+), 8 deletions(-) diff --git a/tests/test_drum_calibration.py b/tests/test_drum_calibration.py index 565b95b..0889673 100644 --- a/tests/test_drum_calibration.py +++ b/tests/test_drum_calibration.py @@ -1,9 +1,11 @@ """Drum calibration — A-series PR-A4. -Receipt (verified at main): drum_midi had NO cross-part de-dup, so separation -bleed (a weak phantom in one part coincident with a loud hit in another) survived -as a duplicate coincident note. This regression pins that: coincident cross-part -bleed must not produce duplicate notes. +Receipts (verified at main): drum_midi scaled every hit against ONE global peak +(no per-part normalization), had NO cross-part de-dup (separation bleed produced +phantom coincident notes), voiced all toms as one GM note, hard-coded the hi-hat +decay test, and recorded no per-event strength. This pins the fixes: per-part +velocity normalization, cross-part bleed de-dup (keep-most-probable), toms split +by pitch, config-driven thresholds (validated), and a per-event manifest. GPU-free and deterministic — synthetic hits, seeded RNG. """ @@ -20,7 +22,7 @@ from stemforge import drum_midi # noqa: E402 from stemforge.io_utils import ensure_dir # noqa: E402 -from stemforge.orchestrator import load_config # noqa: E402 +from stemforge.orchestrator import DrumMidiCfg, load_config # noqa: E402 SR = 44100 @@ -50,11 +52,15 @@ def _write(path: Path, y: np.ndarray) -> str: return str(path) -def _run(parts, tmp_path, **cfg_over): +def _cfg(**over): cfg = load_config().drums.midi - for k, v in cfg_over.items(): + for k, v in over.items(): setattr(cfg, k, v) - return drum_midi._from_parts(parts, cfg, ensure_dir(tmp_path / "out"), bpm=120.0) + return cfg + + +def _run(parts, tmp_path, **cfg_over): + return drum_midi._from_parts(parts, _cfg(**cfg_over), ensure_dir(tmp_path / "out"), bpm=120.0) # --------------------------------------------------------------------------- # @@ -77,3 +83,84 @@ def test_cross_part_bleed_is_deduped(tmp_path): coincident = sum(1 for a, b in zip(times, times[1:]) if b[0] - a[0] <= 0.018 and a[1] != b[1]) assert coincident == 0, f"cross-part bleed duplicates survived: {coincident}" + + +# --------------------------------------------------------------------------- # +# Per-part velocity normalization + ghost survival +# --------------------------------------------------------------------------- # +def test_velocity_normalized_per_part_and_ghosts_survive(tmp_path): + # a quiet part (hi-hat-ish) and a loud part (kick): both reach a strong velocity + # for their own loudest hit, and a quiet ghost stays clearly audible. + kick = _place([(0.3, _noise(0.3, 0.05, 1, amp=1.0)), (0.9, _noise(0.3, 0.05, 2, amp=1.0))]) + snare = _place([(0.6, _noise(0.3, 0.05, 3, amp=0.15)), # whole part is quiet + (1.2, _noise(0.3, 0.05, 4, amp=0.15)), + (1.5, _noise(0.3, 0.05, 5, amp=0.04))]) # a ghost within the quiet part + res = _run({"kick": _write(tmp_path / "k.wav", kick), + "snare": _write(tmp_path / "s.wav", snare)}, tmp_path) + by_part: dict[str, list[int]] = {} + for e in res["events"]: + by_part.setdefault(e["part"], []).append(e["velocity"]) + # per-part normalization: the quiet snare part still reaches a strong velocity + assert max(by_part["snare"]) >= 90 + assert max(by_part["kick"]) >= 90 + # every hit (incl. the ghost) stays audible (velocity floor) + assert all(v >= 10 for vs in by_part.values() for v in vs) + assert min(by_part["snare"]) >= 10 + + +# --------------------------------------------------------------------------- # +# Toms split by pitch +# --------------------------------------------------------------------------- # +def test_toms_split_low_mid_high_by_pitch(tmp_path): + toms = _place([(0.2, _sine(0.4, 80.0, 0.12)), # low + (0.8, _sine(0.4, 150.0, 0.12)), # mid + (1.4, _sine(0.4, 300.0, 0.12))]) # high + res = _run({"toms": _write(tmp_path / "toms.wav", toms)}, tmp_path) + pitches = {n.pitch for inst in pretty_midi.PrettyMIDI(res["file"]).instruments for n in inst.notes} + assert drum_midi.GM["toms_low"] in pitches + assert drum_midi.GM["toms_high"] in pitches + # tom_split off -> everything collapses to the single mid-tom note + res2 = _run({"toms": _write(tmp_path / "t2.wav", toms)}, tmp_path, tom_split=False) + p2 = {n.pitch for inst in pretty_midi.PrettyMIDI(res2["file"]).instruments for n in inst.notes} + assert p2 == {drum_midi.GM["toms"]} + + +def test_comparable_simultaneous_hits_survive(tmp_path): + # a real, comparable kick+snare on the same beat is NOT bleed — both survive + kick = _place([(0.5, _noise(0.3, 0.05, 1, amp=1.0))]) + snare = _place([(0.502, _noise(0.3, 0.05, 2, amp=0.9))]) + res = _run({"kick": _write(tmp_path / "k.wav", kick), + "snare": _write(tmp_path / "s.wav", snare)}, tmp_path) + pitches = [n.pitch for inst in pretty_midi.PrettyMIDI(res["file"]).instruments for n in inst.notes] + assert drum_midi.GM["kick"] in pitches and drum_midi.GM["snare"] in pitches + assert res["duplicates_removed"] == 0 + + +# --------------------------------------------------------------------------- # +# Per-event manifest + config validation +# --------------------------------------------------------------------------- # +def test_manifest_records_per_event_strengths(tmp_path): + kick = _place([(0.3, _noise(0.3, 0.05, 1)), (0.9, _noise(0.3, 0.05, 2))]) + res = _run({"kick": _write(tmp_path / "k.wav", kick)}, tmp_path) + assert "events" in res and res["events"] + ev = res["events"][0] + for key in ("time", "part", "note", "raw_strength", "normalized_strength", "velocity"): + assert key in ev + assert 1 <= ev["velocity"] <= 127 + assert 0.0 <= ev["normalized_strength"] <= 1.0 + assert "duplicates_removed" in res + + +def test_config_validation_rejects_bad_values(): + assert DrumMidiCfg().validate() is not None # defaults are valid + with pytest.raises(ValueError): + DrumMidiCfg(velocity_percentile=0).validate() + with pytest.raises(ValueError): + DrumMidiCfg(dedupe_weak_ratio=2.0).validate() + with pytest.raises(ValueError): + DrumMidiCfg(cymbal_ambiguous="nonsense").validate() + with pytest.raises(ValueError): + DrumMidiCfg(tom_low_hz=500, tom_mid_hz=100).validate() + # load_config runs validation -> a bad override is rejected at load time + with pytest.raises(ValueError): + load_config(overrides={"drums.midi.cymbal_ambiguous": "bogus"}) From 76ddc0f4c2a28eb7967c7f5c1a1f06484013d4bd Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 03:18:34 +0000 Subject: [PATCH 07/11] feat: accuracy benchmark loop with a synthetic corpus and CI threshold gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A render-roundtrip benchmark that measures the A1..A4 accuracy work and fails the build on a regression. - benchmarks/corpus.py: deterministic synthetic fixtures with a known ground truth — constant 90/120/140 BPM tempo trains, and a backbeat drum groove with per-part hits, ghost notes, a low-tom fill, and a cross-part bleed variant. - benchmarks/metrics.py: pure metrics — tempo abs error, half/double (octave) correctness, drum onset F1, cross-part duplicate rate, velocity rank correlation. - benchmarks/run_benchmarks.py: runs analysis + drum_midi over the corpus, computes metrics, compares to thresholds.yaml, writes a JSON report, and exits non-zero on any violation. Metrics needing an absent model (melodic F1, etc.) are reported as skipped with a reason — never silently dropped. - Every report embeds provenance: fixture SHA-256, drum-midi config hash, and library/model versions captured at run time. - thresholds.yaml: bounds with margin over a clean baseline (tempo err 2.3, drum F1 1.0, duplicate rate 0.0, velocity corr 0.64). - .github/workflows/benchmarks.yml: unit tests + the benchmark threshold gate on every PR/push, uploading the report artifact. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- .github/workflows/benchmarks.yml | 30 ++++ benchmarks/README.md | 50 ++++++ benchmarks/__init__.py | 9 ++ benchmarks/corpus.py | 151 +++++++++++++++++++ benchmarks/metrics.py | 86 +++++++++++ benchmarks/run_benchmarks.py | 251 +++++++++++++++++++++++++++++++ benchmarks/thresholds.yaml | 9 ++ 7 files changed, 586 insertions(+) create mode 100644 .github/workflows/benchmarks.yml create mode 100644 benchmarks/README.md create mode 100644 benchmarks/__init__.py create mode 100644 benchmarks/corpus.py create mode 100644 benchmarks/metrics.py create mode 100644 benchmarks/run_benchmarks.py create mode 100644 benchmarks/thresholds.yaml diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml new file mode 100644 index 0000000..b1f5fca --- /dev/null +++ b/.github/workflows/benchmarks.yml @@ -0,0 +1,30 @@ +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 + pip install numpy soundfile librosa pyyaml pretty_midi + pip install -e . --no-deps + - 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 diff --git a/benchmarks/README.md b/benchmarks/README.md new file mode 100644 index 0000000..ce69aed --- /dev/null +++ b/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. diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py new file mode 100644 index 0000000..0f81aa5 --- /dev/null +++ b/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 diff --git a/benchmarks/corpus.py b/benchmarks/corpus.py new file mode 100644 index 0000000..d0e28d0 --- /dev/null +++ b/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 diff --git a/benchmarks/metrics.py b/benchmarks/metrics.py new file mode 100644 index 0000000..031c1d3 --- /dev/null +++ b/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 diff --git a/benchmarks/run_benchmarks.py b/benchmarks/run_benchmarks.py new file mode 100644 index 0000000..284c60f --- /dev/null +++ b/benchmarks/run_benchmarks.py @@ -0,0 +1,251 @@ +"""Run the StemForge accuracy benchmarks and enforce thresholds. + +Usage: + python -m benchmarks.run_benchmarks [--report out.json] [--corpus-dir dir] + +Exits non-zero if any metric violates ``benchmarks/thresholds.yaml`` — wired into +CI so an accuracy regression fails the build. Every report embeds provenance +(fixture SHA-256, config hash, library/model versions) captured at run time. + +Metrics implemented here run fully offline: tempo abs error + half/double +(analysis), drum per-class F1, cross-part duplicate rate, and velocity rank +correlation (drum_midi). Metrics that need an unavailable model are reported as +``skipped`` with a reason — never silently dropped. +""" + +from __future__ import annotations + +import argparse +import json +import platform +import sys +import tempfile +from pathlib import Path +from typing import Any + +import numpy as np +import soundfile as sf +import yaml + +from stemforge import __version__ as sf_version +from stemforge import analysis, drum_midi +from stemforge.io_utils import sha256_bytes +from stemforge.orchestrator import load_config + +from . import corpus, metrics + +_THRESHOLDS = Path(__file__).resolve().parent / "thresholds.yaml" + + +# --------------------------------------------------------------------------- # +# Provenance +# --------------------------------------------------------------------------- # +def _versions() -> dict[str, str]: + vers = {"stemforge": sf_version, "python": platform.python_version(), + "numpy": np.__version__, "soundfile": sf.__version__} + try: + import librosa + vers["librosa"] = librosa.__version__ + except Exception: # noqa: BLE001 + vers["librosa"] = "absent" + try: + import pretty_midi + vers["pretty_midi"] = getattr(pretty_midi, "__version__", "unknown") + except Exception: # noqa: BLE001 + vers["pretty_midi"] = "absent" + return vers + + +def _config_hash() -> str: + import dataclasses + return sha256_bytes(json.dumps( + dataclasses.asdict(load_config().drums.midi), sort_keys=True, default=str).encode()) + + +# --------------------------------------------------------------------------- # +# Evaluators +# --------------------------------------------------------------------------- # +def _eval_tempo() -> dict[str, Any]: + errors, half_double_hits, n = [], 0, 0 + per_fixture = {} + for tf in corpus.tempo_fixtures(): + grid = analysis.analyze(analysis.AudioTensor(tf.audio, tf.sr), engine="librosa") + err = metrics.tempo_abs_error(tf.true_bpm, grid.source_bpm) + hd = metrics.half_double_ok(tf.true_bpm, grid.bpm_candidates or [grid.source_bpm]) + errors.append(err) + half_double_hits += int(hd) + n += 1 + per_fixture[tf.name] = {"true_bpm": tf.true_bpm, "est_bpm": round(grid.source_bpm, 2), + "abs_error": round(err, 3), "half_double_ok": hd} + return { + "tempo_abs_error_bpm_mean": float(np.mean(errors)) if errors else 0.0, + "tempo_abs_error_bpm_max": float(np.max(errors)) if errors else 0.0, + "half_double_correct": half_double_hits / n if n else 0.0, + "per_fixture": per_fixture, + } + + +def _run_drum(df: corpus.DrumFixture, work: Path) -> dict[str, Any]: + parts: dict[str, str] = {} + for part, y in df.parts.items(): + p = work / f"{df.name}_{part}.wav" + sf.write(str(p), y, df.sr, subtype="FLOAT") + parts[part] = str(p) + cfg = load_config().drums.midi + res = drum_midi._from_parts(parts, cfg, work, bpm=120.0) + return res + + +def _eval_drums() -> dict[str, Any]: + _CLASS_NOTES = {"kick": {drum_midi.GM["kick"]}, "snare": {drum_midi.GM["snare"]}, + "toms": {drum_midi.GM["toms_low"], drum_midi.GM["toms_mid"], + drum_midi.GM["toms_high"], drum_midi.GM["toms"]}} + out: dict[str, Any] = {"per_fixture": {}} + f1s: dict[str, list[float]] = {"kick": [], "snare": [], "toms": []} + dup_rates, vel_corrs = [], [] + with tempfile.TemporaryDirectory() as td: + work = Path(td) + for df in corpus.drum_fixtures(): + res = _run_drum(df, work) + events = res.get("events", []) + fx: dict[str, Any] = {"duplicates_removed": res.get("duplicates_removed", 0)} + for cls, notes in _CLASS_NOTES.items(): + pred = [e["time"] for e in events if e["note"] in notes] + tp, fp, fn = metrics.match_events(df.truth.get(cls, []), pred) + score = metrics.f1(tp, fp, fn) + f1s[cls].append(score) + fx[f"{cls}_f1"] = round(score, 3) + # duplicate rate from the written MIDI (time, pitch) + import pretty_midi + mid = pretty_midi.PrettyMIDI(res["file"]) + notes_tp = [(n.start, n.pitch) for inst in mid.instruments for n in inst.notes] + dr = metrics.duplicate_rate(notes_tp) + fx["duplicate_rate"] = round(dr, 4) + dup_rates.append(dr) + # velocity rank correlation on the snare part (varied amplitudes) + vc = _velocity_corr(df, events, "snare") + if vc is not None: + fx["snare_velocity_rank_corr"] = round(vc, 3) + vel_corrs.append(vc) + out["per_fixture"][df.name] = fx + # Gate on the reliable backbone classes (kick+snare); toms F1 is reported per + # fixture but not gated — a two-hit tom fill makes its F1 statistically noisy. + backbone = f1s["kick"] + f1s["snare"] + out["drum_f1_min"] = float(min(backbone)) if backbone else 0.0 + out["drum_toms_f1_mean"] = float(np.mean(f1s["toms"])) if f1s["toms"] else 0.0 + out["drum_kick_f1_mean"] = float(np.mean(f1s["kick"])) if f1s["kick"] else 0.0 + out["duplicate_rate_max"] = float(max(dup_rates)) if dup_rates else 0.0 + out["velocity_rank_corr_min"] = float(min(vel_corrs)) if vel_corrs else 1.0 + return out + + +def _velocity_corr(df: corpus.DrumFixture, events: list[dict[str, Any]], part: str): + truth = df.truth.get(part, []) + amps = df.amps.get(part, []) + if len(truth) != len(amps) or len(truth) < 2: + return None + pred = sorted((e["time"], e["velocity"]) for e in events if e["part"] == part) + paired_amp, paired_vel = [], [] + for t, a in zip(truth, amps): + near = min(pred, key=lambda tv: abs(tv[0] - t), default=None) if pred else None + if near is not None and abs(near[0] - t) <= 0.05: + paired_amp.append(a) + paired_vel.append(near[1]) + if len(paired_amp) < 2: + return None + return metrics.rank_corr(paired_amp, paired_vel) + + +# --------------------------------------------------------------------------- # +# Orchestration +# --------------------------------------------------------------------------- # +def run() -> dict[str, Any]: + import time + t0 = time.perf_counter() + tempo = _eval_tempo() + drums = _eval_drums() + metrics_flat = { + "tempo_abs_error_bpm_max": tempo["tempo_abs_error_bpm_max"], + "half_double_correct": tempo["half_double_correct"], + "drum_f1_min": drums["drum_f1_min"], + "duplicate_rate_max": drums["duplicate_rate_max"], + "velocity_rank_corr_min": drums["velocity_rank_corr_min"], + } + return { + "provenance": { + "fixture_sha256": corpus.fixture_sha256(), + "config_hash": _config_hash(), + "versions": _versions(), + }, + "metrics": metrics_flat, + "detail": {"tempo": tempo, "drums": drums}, + "skipped": { + "melodic_onset_pitch_f1": "requires basic-pitch (not installed in the offline harness)", + "beat_alignment": "deferred — needs a beat-aligned reference render", + "note_fragmentation": "deferred to the melodic path (basic-pitch)", + "quantization_displacement": "covered by the A3 ledger (max_displacement_s); " + "not re-measured here without basic-pitch", + }, + "runtime_s": round(time.perf_counter() - t0, 3), + } + + +def load_thresholds(path: Path = _THRESHOLDS) -> dict[str, Any]: + return yaml.safe_load(path.read_text(encoding="utf-8")) or {} + + +def evaluate(report: dict[str, Any], thresholds: dict[str, Any]) -> list[str]: + """Return a list of human-readable threshold violations (empty == pass).""" + m = report["metrics"] + violations: list[str] = [] + + def _max(name: str) -> None: + lim = thresholds.get(name) + if lim is not None and m.get(name, 0.0) > lim: + violations.append(f"{name}={m[name]:.3f} exceeds max {lim}") + + def _min(name: str) -> None: + lim = thresholds.get(name) + if lim is not None and m.get(name, 0.0) < lim: + violations.append(f"{name}={m[name]:.3f} below min {lim}") + + _max("tempo_abs_error_bpm_max") + _min("half_double_correct") + _min("drum_f1_min") + _max("duplicate_rate_max") + _min("velocity_rank_corr_min") + return violations + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description="Run StemForge accuracy benchmarks.") + ap.add_argument("--report", type=Path, default=None, help="write the JSON report here") + ap.add_argument("--corpus-dir", type=Path, default=None, help="also materialize the corpus .wavs") + args = ap.parse_args(argv) + + if args.corpus_dir: + corpus.write_corpus(args.corpus_dir) + report = run() + thresholds = load_thresholds() + violations = evaluate(report, thresholds) + report["thresholds"] = thresholds + report["violations"] = violations + report["passed"] = not violations + + text = json.dumps(report, indent=2) + if args.report: + args.report.write_text(text, encoding="utf-8") + print(text) + for skip, why in report["skipped"].items(): + print(f"[skipped] {skip}: {why}", file=sys.stderr) + if violations: + print(f"\nBENCHMARK FAILED — {len(violations)} threshold violation(s):", file=sys.stderr) + for v in violations: + print(f" - {v}", file=sys.stderr) + return 1 + print("\nBENCHMARK PASSED", file=sys.stderr) + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/benchmarks/thresholds.yaml b/benchmarks/thresholds.yaml new file mode 100644 index 0000000..a2fd488 --- /dev/null +++ b/benchmarks/thresholds.yaml @@ -0,0 +1,9 @@ +# StemForge accuracy thresholds (PR-A5). CI fails the build if a metric regresses +# past one of these bounds. Values are set with margin over a clean baseline run +# (see benchmarks/README.md); tighten them as accuracy improves. + +tempo_abs_error_bpm_max: 6.0 # max |true - estimated| BPM over the tempo corpus (baseline 2.3) +half_double_correct: 1.0 # fraction of fixtures whose true octave is in the candidates +drum_f1_min: 0.9 # worst per-fixture kick/snare onset F1 (baseline 1.0) +duplicate_rate_max: 0.02 # cross-part coincident-duplicate rate — bleed de-dup keeps it ~0 +velocity_rank_corr_min: 0.4 # velocity vs render-amplitude rank correlation (baseline 0.64) From 9b860fc4b1836943927aea2309ef1226b87923f4 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 03:18:34 +0000 Subject: [PATCH 08/11] test: cover the benchmark harness, threshold enforcement, and provenance - pure metric functions (match/F1, cross-pitch duplicate rate, half/double, rank correlation, unknown-tempo full miss). - corpus determinism (stable fixture SHA-256, expected fixture set). - end-to-end harness: all gated metrics present, the shipped thresholds pass on the current tree, provenance (fixture sha256 + config hash + versions) present, and skipped metrics reported with a reason. - threshold enforcement actually gates (an unreachable bound trips a violation). - the CLI returns 0 on the passing baseline and writes the JSON report. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- tests/test_benchmarks.py | 89 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) create mode 100644 tests/test_benchmarks.py diff --git a/tests/test_benchmarks.py b/tests/test_benchmarks.py new file mode 100644 index 0000000..5241bc0 --- /dev/null +++ b/tests/test_benchmarks.py @@ -0,0 +1,89 @@ +"""Benchmark harness — A-series PR-A5. + +Verifies the accuracy harness runs offline, computes the gated metrics, enforces +thresholds (a strict bound must trip a violation), carries provenance, and reports +— never silently drops — the metrics that need an unavailable model. +""" + +from __future__ import annotations + +import pytest + +pytest.importorskip("pretty_midi") +pytest.importorskip("librosa") + +from benchmarks import corpus, metrics # noqa: E402 +from benchmarks import run_benchmarks as rb # noqa: E402 + + +# --------------------------------------------------------------------------- # +# Pure metric functions +# --------------------------------------------------------------------------- # +def test_match_events_and_f1(): + tp, fp, fn = metrics.match_events([1.0, 2.0, 3.0], [1.01, 2.0, 5.0], tol_s=0.05) + assert (tp, fp, fn) == (2, 1, 1) + assert metrics.f1(2, 1, 1) == pytest.approx(2 / 3, rel=1e-6) + assert metrics.f1(0, 3, 3) == 0.0 + + +def test_duplicate_rate_counts_only_cross_pitch_coincidence(): + # two different-pitch notes within 18 ms -> a duplicate; same pitch -> not + assert metrics.duplicate_rate([(0.0, 36), (0.005, 45)]) == 1.0 + assert metrics.duplicate_rate([(0.0, 36), (0.005, 36)]) == 0.0 + assert metrics.duplicate_rate([(0.0, 36), (0.5, 45)]) == 0.0 + + +def test_half_double_and_rank_corr(): + assert metrics.half_double_ok(120.0, [60.0, 120.0, 240.0]) is True + assert metrics.half_double_ok(140.0, [142.3, 71.1, 284.6]) is True # right octave, small drift + assert metrics.half_double_ok(100.0, [150.0, 75.0]) is False + assert metrics.rank_corr([1, 2, 3], [10, 20, 30]) == pytest.approx(1.0) + assert metrics.rank_corr([1, 2, 3], [30, 20, 10]) == pytest.approx(-1.0) + assert metrics.tempo_abs_error(120.0, 0.0) == 120.0 # unknown -> full miss + + +# --------------------------------------------------------------------------- # +# Corpus determinism +# --------------------------------------------------------------------------- # +def test_fixture_sha256_is_stable(): + assert corpus.fixture_sha256() == corpus.fixture_sha256() + assert len(corpus.tempo_fixtures()) == 3 + assert {df.name for df in corpus.drum_fixtures()} == {"groove_clean", "groove_bleed"} + + +# --------------------------------------------------------------------------- # +# End-to-end harness + threshold enforcement + provenance +# --------------------------------------------------------------------------- # +def test_harness_runs_and_passes_thresholds(): + report = rb.run() + m = report["metrics"] + for key in ("tempo_abs_error_bpm_max", "half_double_correct", "drum_f1_min", + "duplicate_rate_max", "velocity_rank_corr_min"): + assert key in m + # the shipped thresholds pass on the current tree + assert rb.evaluate(report, rb.load_thresholds()) == [] + # provenance present + prov = report["provenance"] + assert len(prov["fixture_sha256"]) == 64 + assert len(prov["config_hash"]) == 64 + assert prov["versions"]["stemforge"] + # skipped metrics are reported with a reason, not dropped + assert "melodic_onset_pitch_f1" in report["skipped"] + + +def test_thresholds_actually_gate(): + report = rb.run() + # an impossibly strict duplicate-rate bound must trip a violation + strict = dict(rb.load_thresholds()) + strict["drum_f1_min"] = 1.01 # unreachable (F1 <= 1) + violations = rb.evaluate(report, strict) + assert any("drum_f1_min" in v for v in violations) + + +def test_main_exit_code_and_report(tmp_path): + rc = rb.main(["--report", str(tmp_path / "r.json")]) + assert rc == 0 # baseline passes + assert (tmp_path / "r.json").is_file() + import json + saved = json.loads((tmp_path / "r.json").read_text()) + assert saved["passed"] is True and saved["violations"] == [] From e80521dd7db34bd5b7495c18a857938063519cf7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 03:24:56 +0000 Subject: [PATCH 09/11] ci: install test + web extras so the benchmark workflow can run pytest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The benchmarks workflow installed only a handful of runtime deps and then ran `pytest -q`, which failed with "pytest: command not found" (exit 127) — pytest (and the webapp test deps fastapi/httpx/uvicorn/python-multipart) were never installed. Install the project with its dev+ui extras instead; torch stays out (heavy libs load lazily and the GPU paths are mocked, so the suite is CPU-only). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- .github/workflows/benchmarks.yml | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/.github/workflows/benchmarks.yml b/.github/workflows/benchmarks.yml index b1f5fca..73d5a36 100644 --- a/.github/workflows/benchmarks.yml +++ b/.github/workflows/benchmarks.yml @@ -16,8 +16,10 @@ jobs: - name: Install run: | python -m pip install --upgrade pip - pip install numpy soundfile librosa pyyaml pretty_midi - pip install -e . --no-deps + # 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) From 0ce0fcc62ea3dabca59193f2b25889361bc35e55 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 03:27:04 +0000 Subject: [PATCH 10/11] ci: put repo root on sys.path (pytest pythonpath) so `benchmarks` imports Under a bare `pytest` invocation (CI) the repo root is not on sys.path, so `tests/test_benchmarks.py`'s `import benchmarks` failed with ModuleNotFoundError (it worked locally only because `python -m pytest` adds the CWD). Set `pythonpath = ["."]` in the pytest config so the top-level benchmarks package resolves under both invocations. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- pyproject.toml | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 95a21fc..d6eb63d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -90,3 +90,6 @@ target-version = "py311" [tool.pytest.ini_options] testpaths = ["tests"] addopts = "-q" +# Put the repo root on sys.path so tests can import the top-level `benchmarks` +# package under a bare `pytest` invocation (CI), matching `python -m pytest`. +pythonpath = ["."] From 46a18168490df0c243b9f23caa9a7ab2914db2cd Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 10 Jul 2026 03:34:09 +0000 Subject: [PATCH 11/11] test: strip ANSI in the CLI help assertion (CI forces color) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GitHub Actions sets FORCE_COLOR, so Typer/Rich colorizes `--help` and splits the `--preset` token with escape codes (`-\x1b[0m\x1b[1;36m-preset`), breaking the raw substring check — the test passed locally (no color) but failed in CI. Compare against the ANSI-stripped output instead. Full suite is green under FORCE_COLOR=1. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Dn8XfTBWPfKA9poDbkYQU6 --- tests/test_cli.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/tests/test_cli.py b/tests/test_cli.py index 94c2584..e8fa703 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from pathlib import Path from typer.testing import CliRunner @@ -10,6 +11,15 @@ runner = CliRunner() +_ANSI = re.compile(r"\x1b\[[0-9;]*m") + + +def _plain(text: str) -> str: + """Strip ANSI color codes. CI sets FORCE_COLOR, so Rich colorizes --help and + splits option tokens (e.g. `--preset`) with escape codes; comparisons must run + against the plain text, not the styled bytes.""" + return _ANSI.sub("", text) + class _FakePipeline: """Captures the config the CLI built; returns a minimal manifest.""" @@ -93,7 +103,7 @@ def test_help_lists_preset(): for cmd in ("run", "separate"): result = runner.invoke(cli.app, [cmd, "--help"]) assert result.exit_code == 0 - assert "--preset" in result.output + assert "--preset" in _plain(result.output) def test_setup_sota_invokes_provisioner(monkeypatch, tmp_path):