diff --git a/apps/backend/phase1_evaluation.py b/apps/backend/phase1_evaluation.py index c1631d75..9b0893a6 100644 --- a/apps/backend/phase1_evaluation.py +++ b/apps/backend/phase1_evaluation.py @@ -18,6 +18,7 @@ DEFAULT_MANIFEST_PATH = REPO_DIR / "tests" / "fixtures" / "phase1_eval_manifest.json" DEFAULT_REPORT_PATH = REPO_DIR / ".runtime" / "reports" / "phase1_eval_report.json" DEFAULT_BENCH_TRACKS_DIR = REPO_DIR / "tests" / "fixtures" / "bench_tracks" +DEFAULT_TRANSCRIPTION_TRACKS_DIR = REPO_DIR / "tests" / "fixtures" / "transcription_tracks" EXPECTED_SPECTRAL_KEYS = { "subBass", "lowBass", @@ -48,6 +49,19 @@ class RealTrackResult: all_passed: bool +@dataclass +class TranscriptionTrackResult: + track_id: str + audio_path: str + category: str + description: str + status: str # "evaluated" | "skipped_audio_missing" | "skipped_analyze_failed" | "skipped_no_transcription" + skip_reason: str | None + checks: list[FixtureCheck] + note_metrics: dict[str, Any] | None + all_passed: bool + + def _write_stereo_wav(path: Path, mono: np.ndarray, sample_rate: int) -> None: mono_arr = np.asarray(mono, dtype=np.float32) stereo = np.stack([mono_arr, mono_arr], axis=1) @@ -137,13 +151,31 @@ def _evaluate_threshold( target = float(config.get("target")) tolerance = float(config.get("tolerance", 0.0)) + direction = str(config.get("direction", "")).strip().lower() if not isinstance(actual, (int, float)): return FixtureCheck( name=f"threshold:{field}", passed=False, message=f"expected numeric target={target}±{tolerance}, actual={actual}", ) - delta = abs(float(actual) - target) + actual_value = float(actual) + if direction == "min": + bound = target - tolerance + passed = actual_value >= bound + return FixtureCheck( + name=f"threshold:{field}", + passed=passed, + message=f"target>={target} tolerance={tolerance} bound={round(bound, 6)} actual={actual}", + ) + if direction == "max": + bound = target + tolerance + passed = actual_value <= bound + return FixtureCheck( + name=f"threshold:{field}", + passed=passed, + message=f"target<={target} tolerance={tolerance} bound={round(bound, 6)} actual={actual}", + ) + delta = abs(actual_value - target) passed = delta <= tolerance return FixtureCheck( name=f"threshold:{field}", @@ -287,6 +319,281 @@ def _evaluate_real_track( ) +def _match_notes( + detected: list[dict[str, Any]], + ground_truth: list[dict[str, Any]], + onset_window_s: float = 0.05, + pitch_tolerance_semitones: int = 1, +) -> list[tuple[int, int]]: + """Greedy onset-sorted matching of detected notes to ground-truth notes. + + For each ground-truth note (in onset order), pair with the earliest unused + detected note whose onset is within ±onset_window_s AND whose pitch differs + by no more than pitch_tolerance_semitones. Returns (gt_index, det_index) + pairs. Mirrors mir_eval.transcription onset/pitch semantics. + """ + gt_indices = sorted( + range(len(ground_truth)), + key=lambda i: float(ground_truth[i].get("onsetSeconds", 0.0)), + ) + det_indices_sorted = sorted( + range(len(detected)), + key=lambda i: float(detected[i].get("onsetSeconds", 0.0)), + ) + used: set[int] = set() + matches: list[tuple[int, int]] = [] + for gt_idx in gt_indices: + gt_onset = float(ground_truth[gt_idx].get("onsetSeconds", 0.0)) + gt_pitch = int(ground_truth[gt_idx].get("pitchMidi", 0)) + for det_idx in det_indices_sorted: + if det_idx in used: + continue + det_onset = float(detected[det_idx].get("onsetSeconds", 0.0)) + if abs(det_onset - gt_onset) > onset_window_s: + continue + det_pitch = int(detected[det_idx].get("pitchMidi", 0)) + if abs(det_pitch - gt_pitch) > pitch_tolerance_semitones: + continue + used.add(det_idx) + matches.append((gt_idx, det_idx)) + break + return matches + + +def _compute_note_metrics( + detected: list[dict[str, Any]], + ground_truth: list[dict[str, Any]], + matches: list[tuple[int, int]], +) -> dict[str, Any]: + """Compute precision/recall/F1 and mean signed pitch error in cents. + + Note: pitchMidi is an integer per the analyze.py contract, so + meanPitchCentsError is always a multiple of 100. This is a deliberate + coarseness limit — switch to a sub-semitone-aware detector if finer + pitch-accuracy reporting is needed. + """ + matched_count = len(matches) + detected_count = len(detected) + gt_count = len(ground_truth) + missed_count = gt_count - matched_count + false_positive_count = detected_count - matched_count + + if detected_count == 0: + precision = 1.0 + else: + precision = matched_count / detected_count + if gt_count == 0: + recall = 1.0 + else: + recall = matched_count / gt_count + if precision + recall == 0.0: + f1 = 0.0 + else: + f1 = (2.0 * precision * recall) / (precision + recall) + + if matched_count == 0: + mean_cents_error = 0.0 + else: + cents_errors = [ + (int(detected[d].get("pitchMidi", 0)) - int(ground_truth[g].get("pitchMidi", 0))) * 100 + for g, d in matches + ] + mean_cents_error = sum(cents_errors) / float(matched_count) + + return { + "precision": round(precision, 4), + "recall": round(recall, 4), + "f1": round(f1, 4), + "meanPitchCentsError": round(mean_cents_error, 4), + "matchedCount": matched_count, + "missedCount": missed_count, + "falsePositiveCount": false_positive_count, + } + + +def _evaluate_transcription_track( + entry: dict[str, Any], + transcription_tracks_dir: Path, + transcribe_runner: Any = None, +) -> TranscriptionTrackResult: + """Evaluate a single transcription-track entry from the manifest. + + Mirrors `_evaluate_real_track` semantics — gracefully skips when the audio + is absent so a missing track never fails the run. `transcribe_runner` is a + callable `(audio_path, extra_flags) -> dict`; defaults to `_run_analyze`. + """ + if transcribe_runner is None: + transcribe_runner = _run_analyze + + track_id = str(entry.get("id") or "unknown") + audio_rel = str(entry.get("audioPath") or "") + category = str(entry.get("category") or "uncategorized") + description = str(entry.get("description") or "") + thresholds = entry.get("thresholds") + analyze_flags = entry.get("analyzeFlags") + ground_truth_notes = entry.get("groundTruthNotes") + + if not isinstance(thresholds, dict): + thresholds = {} + if isinstance(analyze_flags, list): + flags_list = [str(flag) for flag in analyze_flags] + else: + flags_list = ["--transcribe"] + if not isinstance(ground_truth_notes, list): + ground_truth_notes = [] + + if not audio_rel: + return TranscriptionTrackResult( + track_id=track_id, + audio_path="", + category=category, + description=description, + status="skipped_audio_missing", + skip_reason="manifest entry has no audioPath", + checks=[], + note_metrics=None, + all_passed=True, + ) + + audio_path = (transcription_tracks_dir / audio_rel).resolve() + if not audio_path.exists(): + return TranscriptionTrackResult( + track_id=track_id, + audio_path=str(audio_path), + category=category, + description=description, + status="skipped_audio_missing", + skip_reason=( + f"audio not present at {audio_path} — add the file locally to " + "include this track in transcription evaluation" + ), + checks=[], + note_metrics=None, + all_passed=True, + ) + + try: + payload = transcribe_runner(audio_path, flags_list) + except subprocess.CalledProcessError as exc: + stderr_tail = (exc.stderr or "")[-400:] + return TranscriptionTrackResult( + track_id=track_id, + audio_path=str(audio_path), + category=category, + description=description, + status="skipped_analyze_failed", + skip_reason=f"analyze.py failed (exit {exc.returncode}): {stderr_tail}", + checks=[], + note_metrics=None, + all_passed=False, + ) + + transcription_detail = payload.get("transcriptionDetail") if isinstance(payload, dict) else None + if not isinstance(transcription_detail, dict): + return TranscriptionTrackResult( + track_id=track_id, + audio_path=str(audio_path), + category=category, + description=description, + status="skipped_no_transcription", + skip_reason="analyze.py output had no transcriptionDetail block", + checks=[], + note_metrics=None, + all_passed=False, + ) + + detected_notes = transcription_detail.get("notes") or [] + if not isinstance(detected_notes, list): + detected_notes = [] + matches = _match_notes(detected_notes, ground_truth_notes) + note_metrics = _compute_note_metrics(detected_notes, ground_truth_notes, matches) + + eval_payload = dict(payload) + eval_payload["noteMetrics"] = note_metrics + + checks: list[FixtureCheck] = [] + for field, config in thresholds.items(): + if not isinstance(config, dict): + continue + checks.append(_evaluate_threshold(eval_payload, field, config)) + + return TranscriptionTrackResult( + track_id=track_id, + audio_path=str(audio_path), + category=category, + description=description, + status="evaluated", + skip_reason=None, + checks=checks, + note_metrics=note_metrics, + all_passed=all(check.passed for check in checks), + ) + + +def _evaluate_stepped_sine_synthetic( + transcribe_runner: Any = None, +) -> TranscriptionTrackResult: + """Generate a stepped-sine WAV and evaluate it as a harness self-test. + + Four notes at known MIDI pitches with 0.2s gaps. Provides at least one + transcription result even when no real tracks are present. + """ + if transcribe_runner is None: + transcribe_runner = _run_analyze + + sample_rate = 16000 + note_duration_s = 0.6 + gap_s = 0.2 + pitch_midi_values = [60, 64, 67, 72] # C4, E4, G4, C5 + ground_truth: list[dict[str, Any]] = [] + segments: list[np.ndarray] = [] + cursor = 0.0 + for pitch_midi in pitch_midi_values: + freq_hz = 440.0 * (2.0 ** ((pitch_midi - 69) / 12.0)) + n_samples = int(round(note_duration_s * sample_rate)) + t = np.linspace(0.0, note_duration_s, n_samples, endpoint=False, dtype=np.float32) + tone = 0.5 * np.sin(2.0 * np.pi * freq_hz * t) + envelope = np.ones_like(tone) + ramp = max(1, int(round(0.01 * sample_rate))) + envelope[:ramp] = np.linspace(0.0, 1.0, ramp, dtype=np.float32) + envelope[-ramp:] = np.linspace(1.0, 0.0, ramp, dtype=np.float32) + segments.append(tone * envelope) + ground_truth.append( + { + "pitchMidi": pitch_midi, + "onsetSeconds": round(cursor, 4), + "durationSeconds": round(note_duration_s, 4), + } + ) + cursor += note_duration_s + gap = np.zeros(int(round(gap_s * sample_rate)), dtype=np.float32) + segments.append(gap) + cursor += gap_s + + mono = np.concatenate(segments).astype(np.float32) + + with tempfile.TemporaryDirectory(prefix="asa_stepped_sine_") as temp_dir: + wav_path = Path(temp_dir) / "stepped_sine.wav" + _write_stereo_wav(wav_path, mono, sample_rate) + entry = { + "id": "stepped_sine_synthetic", + "audioPath": wav_path.name, + "category": "synthetic_self_test", + "description": "Stepped sine self-test: four monophonic notes (C4, E4, G4, C5).", + "analyzeFlags": ["--transcribe"], + "groundTruthNotes": ground_truth, + "thresholds": { + "noteMetrics.f1": {"target": 0.5, "tolerance": 0.0, "direction": "min"}, + "noteMetrics.meanPitchCentsError": {"target": 0.0, "tolerance": 100.0}, + }, + } + return _evaluate_transcription_track( + entry, + wav_path.parent, + transcribe_runner=transcribe_runner, + ) + + def _evaluate_stability( outputs: list[dict[str, Any]], stability_checks: list[dict[str, Any]], @@ -340,15 +647,23 @@ def run_phase1_evaluation( runs_per_fixture: int = 2, include_real: bool = False, real_tracks_dir: Path = DEFAULT_BENCH_TRACKS_DIR, + include_transcription: bool = False, + transcription_tracks_dir: Path = DEFAULT_TRANSCRIPTION_TRACKS_DIR, + transcribe_runner: Any = None, ) -> dict[str, Any]: manifest = json.loads(manifest_path.read_text(encoding="utf-8")) fixtures = manifest.get("fixtures", []) stability_checks = manifest.get("stabilityChecks", []) real_tracks_manifest = manifest.get("realTracks", []) if include_real else [] + transcription_manifest = ( + manifest.get("transcriptionTracks", []) if include_transcription else [] + ) if not isinstance(fixtures, list) or len(fixtures) == 0: raise ValueError("Manifest must define one or more fixtures.") if not isinstance(real_tracks_manifest, list): raise ValueError("Manifest 'realTracks' must be a list when present.") + if not isinstance(transcription_manifest, list): + raise ValueError("Manifest 'transcriptionTracks' must be a list when present.") fixture_reports: list[dict[str, Any]] = [] passed_checks = 0 @@ -443,11 +758,51 @@ def run_phase1_evaluation( } ) + transcription_reports: list[dict[str, Any]] = [] + transcription_evaluated = 0 + transcription_skipped = 0 + transcription_failed_subprocess = 0 + + if include_transcription: + # Self-test first — guarantees the report has at least one transcription + # row even when no real tracks have been added to the manifest. + self_test = _evaluate_stepped_sine_synthetic(transcribe_runner=transcribe_runner) + if self_test.status == "evaluated": + transcription_evaluated += 1 + passed_checks += sum(1 for check in self_test.checks if check.passed) + failed_checks += sum(1 for check in self_test.checks if not check.passed) + elif self_test.status == "skipped_audio_missing": + transcription_skipped += 1 + else: + transcription_failed_subprocess += 1 + failed_checks += 1 + transcription_reports.append(_transcription_track_to_report(self_test)) + + for raw_entry in transcription_manifest: + if not isinstance(raw_entry, dict): + continue + result = _evaluate_transcription_track( + raw_entry, transcription_tracks_dir, transcribe_runner=transcribe_runner + ) + if result.status == "evaluated": + transcription_evaluated += 1 + passed_checks += sum(1 for check in result.checks if check.passed) + failed_checks += sum(1 for check in result.checks if not check.passed) + elif result.status == "skipped_audio_missing": + transcription_skipped += 1 + else: + transcription_failed_subprocess += 1 + failed_checks += 1 + transcription_reports.append(_transcription_track_to_report(result)) + summary = { "fixtures": len(fixture_reports), "realTracksEvaluated": real_evaluated, "realTracksSkipped": real_skipped, "realTracksAnalyzeFailed": real_failed_subprocess, + "transcriptionTracksEvaluated": transcription_evaluated, + "transcriptionTracksSkipped": transcription_skipped, + "transcriptionTracksAnalyzeFailed": transcription_failed_subprocess, "checksPassed": passed_checks, "checksFailed": failed_checks, "allPassed": failed_checks == 0, @@ -459,13 +814,38 @@ def run_phase1_evaluation( "runsPerFixture": runs_per_fixture, "includeReal": include_real, "realTracksDir": str(real_tracks_dir) if include_real else None, + "includeTranscription": include_transcription, + "transcriptionTracksDir": str(transcription_tracks_dir) if include_transcription else None, "fixtures": fixture_reports, "summary": summary, } if include_real: report["realTracks"] = real_track_reports + if include_transcription: + report["transcriptionTracks"] = transcription_reports report_path.parent.mkdir(parents=True, exist_ok=True) report_path.write_text(json.dumps(report, indent=2), encoding="utf-8") report["reportPath"] = str(report_path) return report + + +def _transcription_track_to_report(result: TranscriptionTrackResult) -> dict[str, Any]: + return { + "id": result.track_id, + "audioPath": result.audio_path, + "category": result.category, + "description": result.description, + "status": result.status, + "skipReason": result.skip_reason, + "noteMetrics": result.note_metrics, + "checks": [ + { + "name": check.name, + "passed": check.passed, + "message": check.message, + } + for check in result.checks + ], + "allPassed": result.all_passed, + } diff --git a/apps/backend/polyphonic_evaluation.py b/apps/backend/polyphonic_evaluation.py index bd94d5a4..78131924 100644 --- a/apps/backend/polyphonic_evaluation.py +++ b/apps/backend/polyphonic_evaluation.py @@ -100,17 +100,27 @@ def summarize_midi_file(midi_path: Path, audio_duration_seconds: float) -> dict[ duration_denominator = max(float(audio_duration_seconds), 1e-9) note_count = len(notes) note_density = note_count / duration_denominator + distinct_pitch_count = len(set(pitch_values)) + mean_active_polyphony = weighted_active_polyphony / max(active_time, 1e-9) flags: list[str] = [] if max_polyphony <= 1: flags.append("monophonic_output") if note_density >= 12.0: flags.append("high_note_density") + if note_density > 15.0: + flags.append("note_clutter") + if distinct_pitch_count > 30 and mean_active_polyphony < 3.0: + flags.append("octave_junk") + if max_polyphony > 8: + flags.append("dense_chords_unusable") + if note_density < 1.0 and float(audio_duration_seconds) >= 10.0: + flags.append("sparse_likely_undertranscribed") min_pitch = min(pitch_values) max_pitch = max(pitch_values) return { "noteCount": note_count, - "distinctPitchCount": len(set(pitch_values)), + "distinctPitchCount": distinct_pitch_count, "pitchRange": { "minMidi": min_pitch, "maxMidi": max_pitch, @@ -119,10 +129,7 @@ def summarize_midi_file(midi_path: Path, audio_duration_seconds: float) -> dict[ }, "maxPolyphony": max_polyphony, "meanTimelinePolyphony": round(weighted_timeline_polyphony / duration_denominator, 4), - "meanActivePolyphony": round( - weighted_active_polyphony / max(active_time, 1e-9), - 4, - ), + "meanActivePolyphony": round(mean_active_polyphony, 4), "averageNoteDurationSeconds": round(total_note_duration / note_count, 4), "noteDensityPerSecond": round(note_density, 4), "flags": flags, diff --git a/apps/backend/scripts/evaluate_phase1.py b/apps/backend/scripts/evaluate_phase1.py index 84359528..0f302b4b 100644 --- a/apps/backend/scripts/evaluate_phase1.py +++ b/apps/backend/scripts/evaluate_phase1.py @@ -17,6 +17,7 @@ DEFAULT_BENCH_TRACKS_DIR, DEFAULT_MANIFEST_PATH, DEFAULT_REPORT_PATH, + DEFAULT_TRANSCRIPTION_TRACKS_DIR, run_phase1_evaluation, ) from phase1_report_html import default_html_report_path, render_html_report @@ -66,6 +67,25 @@ def parse_args() -> argparse.Namespace: f"(default: {DEFAULT_BENCH_TRACKS_DIR})" ), ) + parser.add_argument( + "--include-transcription", + action="store_true", + help=( + "Opt in to evaluating Layer 2 (torchcrepe) transcription against " + "manifest.transcriptionTracks. Always runs the stepped-sine self-" + "test even when the corpus is empty. Missing audio files are " + "skipped with a clear notice rather than failing the run." + ), + ) + parser.add_argument( + "--transcription-tracks-dir", + type=Path, + default=DEFAULT_TRANSCRIPTION_TRACKS_DIR, + help=( + f"Directory holding local transcription reference tracks " + f"(default: {DEFAULT_TRANSCRIPTION_TRACKS_DIR})" + ), + ) parser.add_argument( "--html-report", type=Path, @@ -89,16 +109,30 @@ def main() -> None: runs_per_fixture=max(args.runs, 1), include_real=args.include_real, real_tracks_dir=args.real_tracks_dir, + include_transcription=args.include_transcription, + transcription_tracks_dir=args.transcription_tracks_dir, ) summary_line: dict[str, Any] = { "summary": report["summary"], "reportPath": report["reportPath"], } + notes: list[str] = [] if args.include_real and report["summary"]["realTracksSkipped"] > 0: - summary_line["note"] = ( + notes.append( "Some real tracks were skipped because audio files were not present " f"in {args.real_tracks_dir}. See report for per-track skipReason." ) + if ( + args.include_transcription + and report["summary"]["transcriptionTracksSkipped"] > 0 + ): + notes.append( + "Some transcription tracks were skipped because audio files were " + f"not present in {args.transcription_tracks_dir}. See report for " + "per-track skipReason." + ) + if notes: + summary_line["note"] = " ".join(notes) if args.html_report is not None: if isinstance(args.html_report, Path): diff --git a/apps/backend/scripts/import_midi_to_ground_truth.py b/apps/backend/scripts/import_midi_to_ground_truth.py new file mode 100755 index 00000000..e75a1f59 --- /dev/null +++ b/apps/backend/scripts/import_midi_to_ground_truth.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +"""Convert a reference MIDI file into a `groundTruthNotes` JSON fragment. + +Used to bootstrap entries in `tests/fixtures/phase1_eval_manifest.json` under +`transcriptionTracks[].groundTruthNotes`. Emits a JSON array to stdout so the +output can be pasted directly into the manifest. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +import pretty_midi + + +def _flatten_notes(midi: pretty_midi.PrettyMIDI, offset_seconds: float) -> list[dict]: + notes: list[dict] = [] + for instrument in midi.instruments: + for note in instrument.notes: + start = float(note.start) + offset_seconds + end = float(note.end) + offset_seconds + duration = max(0.0, end - start) + notes.append( + { + "pitchMidi": int(note.pitch), + "onsetSeconds": round(start, 4), + "durationSeconds": round(duration, 4), + } + ) + notes.sort(key=lambda entry: (entry["onsetSeconds"], entry["pitchMidi"])) + return notes + + +def _detect_overlaps(notes: list[dict]) -> list[tuple[int, int]]: + overlaps: list[tuple[int, int]] = [] + sorted_indices = sorted(range(len(notes)), key=lambda i: notes[i]["onsetSeconds"]) + for i_pos in range(len(sorted_indices)): + i = sorted_indices[i_pos] + i_end = notes[i]["onsetSeconds"] + notes[i]["durationSeconds"] + for j_pos in range(i_pos + 1, len(sorted_indices)): + j = sorted_indices[j_pos] + if notes[j]["onsetSeconds"] >= i_end: + break + overlaps.append((i, j)) + return overlaps + + +def _collapse_monophonic_highest(notes: list[dict]) -> list[dict]: + if len(notes) == 0: + return notes + indices_by_onset = sorted(range(len(notes)), key=lambda i: notes[i]["onsetSeconds"]) + keep: set[int] = set(indices_by_onset) + for pos_i in range(len(indices_by_onset)): + i = indices_by_onset[pos_i] + if i not in keep: + continue + i_end = notes[i]["onsetSeconds"] + notes[i]["durationSeconds"] + for pos_j in range(pos_i + 1, len(indices_by_onset)): + j = indices_by_onset[pos_j] + if j not in keep: + continue + if notes[j]["onsetSeconds"] >= i_end: + break + if notes[j]["pitchMidi"] > notes[i]["pitchMidi"]: + keep.discard(i) + break + keep.discard(j) + return [notes[i] for i in sorted(keep, key=lambda i: notes[i]["onsetSeconds"])] + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Convert a MIDI file into the groundTruthNotes JSON fragment for a " + "transcriptionTracks manifest entry." + ) + ) + parser.add_argument("midi_path", type=Path, help="Path to the .mid file to convert.") + parser.add_argument( + "--monophonic-collapse", + choices=("highest", "reject"), + default="reject", + help=( + "Behavior when notes overlap. 'reject' (default) exits non-zero on " + "any overlap. 'highest' keeps the highest pitch in any overlapping " + "group and drops the rest." + ), + ) + parser.add_argument( + "--offset-seconds", + type=float, + default=0.0, + help=( + "Add N seconds to every onset. Use when DAW MIDI export starts at a " + "different point than the audio (e.g. negative track delay)." + ), + ) + return parser.parse_args() + + +def main() -> None: + args = parse_args() + if not args.midi_path.exists(): + print(f"error: MIDI file not found at {args.midi_path}", file=sys.stderr) + raise SystemExit(2) + + try: + midi = pretty_midi.PrettyMIDI(str(args.midi_path)) + except Exception as exc: + print(f"error: failed to parse MIDI: {exc}", file=sys.stderr) + raise SystemExit(2) from exc + + notes = _flatten_notes(midi, args.offset_seconds) + overlaps = _detect_overlaps(notes) + + if len(overlaps) > 0: + if args.monophonic_collapse == "reject": + print( + f"error: {len(overlaps)} overlapping note pair(s) detected. " + "Pass --monophonic-collapse highest to keep the higher pitch in " + "each overlapping group.", + file=sys.stderr, + ) + raise SystemExit(2) + notes = _collapse_monophonic_highest(notes) + + print(json.dumps(notes, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/apps/backend/scripts/score_polyphonic_clip.py b/apps/backend/scripts/score_polyphonic_clip.py new file mode 100755 index 00000000..2a2886f4 --- /dev/null +++ b/apps/backend/scripts/score_polyphonic_clip.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 +"""Interactive scorecard CLI for polyphonic evaluation reports. + +Reads a polyphonic evaluation report produced by `evaluate_polyphonic.py`, +walks the unscored clip/candidate pairs, prompts the reviewer for the five +manual scorecard fields, and writes the updates back into the report in place. + +Replaces hand-editing the report JSON. Idempotent without `--rescore` — already +scored entries are skipped. +""" + +from __future__ import annotations + +import argparse +import json +import shutil +import subprocess +import sys +from pathlib import Path +from typing import Any + +BACKEND_DIR = Path(__file__).resolve().parent.parent +if str(BACKEND_DIR) not in sys.path: + sys.path.insert(0, str(BACKEND_DIR)) + +from polyphonic_evaluation import ( # noqa: E402 - sys.path mutated above + DEFAULT_REPORT_PATH, + build_manual_scorecard, +) + +SCORECARD_FIELDS = ( + "bassRecognizable", + "toplineRecognizable", + "chordsNotObviouslyWrong", + "cleanupMinutes30s", +) + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description=( + "Interactively score polyphonic evaluation clips. Updates the " + "report JSON in place. Pass --no-play to skip audio playback " + "(required for non-interactive / headless usage)." + ) + ) + parser.add_argument( + "--report", + type=Path, + default=DEFAULT_REPORT_PATH, + help=f"Path to the polyphonic evaluation report (default: {DEFAULT_REPORT_PATH})", + ) + parser.add_argument( + "--rescore", + action="store_true", + help="Re-prompt for clips that already have a complete scorecard.", + ) + parser.add_argument( + "--no-play", + action="store_true", + help="Skip audio playback. Required for headless / non-interactive runs.", + ) + parser.add_argument( + "--candidate", + type=str, + default=None, + help="Limit scoring to a single candidate id (e.g. 'basic-pitch').", + ) + return parser.parse_args() + + +def _scorecard_is_complete(scorecard: dict[str, Any] | None) -> bool: + if not isinstance(scorecard, dict): + return False + return all(scorecard.get(field) is not None for field in SCORECARD_FIELDS) + + +def _resolve_player() -> list[str] | None: + """Find a system audio player. Returns the command prefix or None.""" + for binary in ("afplay", "paplay", "aplay"): + path = shutil.which(binary) + if path: + return [path] + return None + + +def _play_audio(player: list[str] | None, audio_path: Path) -> None: + if player is None: + print(" [warn] no audio player found (afplay / paplay / aplay). Skipping playback.") + return + if not audio_path.exists(): + print(f" [warn] audio not found at {audio_path}; skipping playback.") + return + try: + subprocess.run([*player, str(audio_path)], check=False) + except KeyboardInterrupt: + # Allow the reviewer to interrupt playback without exiting the CLI. + print(" [info] playback interrupted; continuing to prompts.") + + +def _prompt_yes_no(label: str, default: str = "n") -> bool: + suffix = "Y/n" if default.lower() == "y" else "y/N" + while True: + raw = input(f" {label} [{suffix}]: ").strip().lower() + if raw == "": + raw = default.lower() + if raw in ("y", "yes"): + return True + if raw in ("n", "no"): + return False + print(" please answer y or n.") + + +def _prompt_float(label: str, *, min_value: float = 0.0) -> float: + while True: + raw = input(f" {label}: ").strip() + try: + value = float(raw) + except ValueError: + print(" please enter a number.") + continue + if value < min_value: + print(f" must be >= {min_value}.") + continue + return value + + +def _prompt_optional_text(label: str) -> str: + return input(f" {label}: ").rstrip("\n") + + +def _score_one( + clip_id: str, + candidate_id: str, + candidate_report: dict[str, Any], + audio_path: Path, + player: list[str] | None, + play: bool, +) -> dict[str, Any]: + metrics = candidate_report.get("metrics") or {} + flags = metrics.get("flags") or [] + print(f"\n=== {clip_id} :: {candidate_id} ===") + print(f" audio: {audio_path}") + print( + " metrics: noteCount={n} maxPolyphony={mp} noteDensityPerSecond={d}".format( + n=metrics.get("noteCount", "?"), + mp=metrics.get("maxPolyphony", "?"), + d=metrics.get("noteDensityPerSecond", "?"), + ) + ) + if flags: + print(f" flags: {', '.join(flags)}") + else: + print(" flags: (none)") + + if play: + _play_audio(player, audio_path) + + existing = candidate_report.get("scorecard") if isinstance(candidate_report.get("scorecard"), dict) else None + bass = _prompt_yes_no("bassRecognizable?") + topline = _prompt_yes_no("toplineRecognizable?") + chords = _prompt_yes_no("chordsNotObviouslyWrong?") + cleanup = _prompt_float("cleanupMinutes30s") + notes_default = (existing or {}).get("notes", "") + notes = _prompt_optional_text(f"notes (current: {notes_default!r})") + if notes == "": + notes = notes_default + + return build_manual_scorecard( + { + "bassRecognizable": bass, + "toplineRecognizable": topline, + "chordsNotObviouslyWrong": chords, + "cleanupMinutes30s": cleanup, + "notes": notes, + } + ) + + +def main() -> None: + args = parse_args() + report_path: Path = args.report + if not report_path.exists(): + print(f"error: report not found at {report_path}", file=sys.stderr) + raise SystemExit(2) + + report = json.loads(report_path.read_text(encoding="utf-8")) + clips = report.get("clips") + if not isinstance(clips, list) or len(clips) == 0: + print("error: report has no clips to score.", file=sys.stderr) + raise SystemExit(2) + + player = _resolve_player() + scored_count = 0 + skipped_count = 0 + + for clip in clips: + if not isinstance(clip, dict): + continue + clip_id = str(clip.get("id") or "unknown") + audio_path_str = clip.get("audioPath") or "" + audio_path = Path(audio_path_str) + candidates = clip.get("candidates") or {} + if not isinstance(candidates, dict): + continue + + for candidate_id, candidate_report in candidates.items(): + if args.candidate and candidate_id != args.candidate: + continue + if not isinstance(candidate_report, dict): + continue + existing = candidate_report.get("scorecard") + if _scorecard_is_complete(existing) and not args.rescore: + skipped_count += 1 + continue + try: + new_scorecard = _score_one( + clip_id=clip_id, + candidate_id=candidate_id, + candidate_report=candidate_report, + audio_path=audio_path, + player=player, + play=not args.no_play, + ) + except EOFError: + print("\n[info] input stream closed; stopping early.", file=sys.stderr) + break + candidate_report["scorecard"] = new_scorecard + scored_count += 1 + + report_path.write_text(json.dumps(report, indent=2), encoding="utf-8") + print( + json.dumps( + { + "reportPath": str(report_path), + "scored": scored_count, + "skippedAlreadyComplete": skipped_count, + }, + indent=2, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/apps/backend/tests/fixtures/phase1_eval_manifest.json b/apps/backend/tests/fixtures/phase1_eval_manifest.json index 41c868f0..adcb7b97 100644 --- a/apps/backend/tests/fixtures/phase1_eval_manifest.json +++ b/apps/backend/tests/fixtures/phase1_eval_manifest.json @@ -44,5 +44,6 @@ { "field": "supersawDetail.isSupersaw", "mode": "exact" }, { "field": "kickDetail.kickCount", "maxDelta": 0.0 } ], - "realTracks": [] + "realTracks": [], + "transcriptionTracks": [] } diff --git a/apps/backend/tests/fixtures/polyphonic_tracks/README.md b/apps/backend/tests/fixtures/polyphonic_tracks/README.md new file mode 100644 index 00000000..45445956 --- /dev/null +++ b/apps/backend/tests/fixtures/polyphonic_tracks/README.md @@ -0,0 +1,64 @@ +# Polyphonic Research Corpus + +This directory holds short polyphonic audio clips used by the **research** harness at `apps/backend/scripts/evaluate_polyphonic.py`. The audio files are **never** committed — they live in your local copy only. + +This corpus is **not** the same thing as `../transcription_tracks/`. Use: + +- **`polyphonic_tracks/`** (this directory) — dense, mixed, mastered material used to evaluate research candidates (`basic-pitch`, `MT3`) against the manual usefulness gates in `docs/POLYPHONIC_TRANSCRIPTION_SPIKE.md`. +- **`transcription_tracks/`** — monophonic-friendly material (vocal leads, isolated bass, simple synth leads) used to evaluate Layer 2 (torchcrepe), which **is** in the product. See `docs/LAYER2_EVALUATION.md`. + +A track that fits one corpus generally does not fit the other. + +## What goes here + +10–20 short clips (20–40s each) drawn from material ASA's polyphonic research is supposed to handle, weighted toward producer-realistic mixes rather than toy or classical proxies: + +1. 3–4 dense electronic mixes (mastered, chord-stack-heavy) +2. 2–3 vocal-heavy material with backing harmony +3. 2–3 piano-only / piano-heavy clips +4. 2–3 busy mastered full-band mixes +5. 1–2 pad + arpeggio / pad + bass slices +6. 1–2 sparse / minimal clips (the corpus needs negative examples too — see the `sparse_likely_undertranscribed` flag in `polyphonic_evaluation.summarize_midi_file`) + +Clips longer than 40 s blow up scoring time; shorter than 20 s rarely give the reviewer enough material to judge fairly. + +Total disk: roughly 50–150 MB depending on encoding. + +## Manifest + +This directory does **not** ship its own checked-in manifest — the polyphonic harness already exercises tempfile-based manifests end-to-end in `tests/test_polyphonic_evaluation.py`. To run the harness on this corpus, point it at a manifest you write locally: + +```bash +cd apps/backend +./venv/bin/python scripts/evaluate_polyphonic.py \ + --manifest /path/to/local_polyphonic_manifest.json +``` + +The manifest format is documented in `docs/POLYPHONIC_TRANSCRIPTION_SPIKE.md`. `audioPath` may be absolute or relative to the manifest file — relative paths typically point into this directory. + +## Sourcing checklist + +1. Use audio you own or license. Nothing copyrighted is committed. +2. Mono or stereo both fine — the harness does not constrain channel count. +3. Format: WAV / FLAC / MP3 / M4A all work via `soundfile` / `librosa`. +4. Pick clips where you, as a reviewer, can confidently answer the five scorecard fields: `bassRecognizable`, `toplineRecognizable`, `chordsNotObviouslyWrong`, `cleanupMinutes30s`, plus optional notes. If you can't answer, the clip isn't useful for the gate. + +## Scoring workflow + +After running the harness: + +```bash +cd apps/backend +./venv/bin/python scripts/score_polyphonic_clip.py \ + --report .runtime/polyphonic_eval/polyphonic_eval_report.json +``` + +The CLI walks unscored clip × candidate pairs, plays the audio (via `afplay` on macOS / `paplay` / `aplay` on Linux), surfaces the diagnostic `flags` and key metrics, and writes the scorecard back into the report JSON in place. Pass `--no-play` to skip playback, `--rescore` to revisit completed entries, `--candidate ` to limit to one candidate. + +Diagnostic flags emitted by `summarize_midi_file` are seeding hints only — they steer your listening attention but do not bypass the manual gates: + +- `note_clutter` — note density > 15 notes/sec; likely junk +- `octave_junk` — >30 distinct pitches with mean active polyphony < 3; suggests scattered octave errors +- `dense_chords_unusable` — max polyphony > 8 simultaneous notes; usually unrecoverable +- `sparse_likely_undertranscribed` — under 1 note/sec across a 10s+ clip; model probably gave up +- `empty_output`, `monophonic_output`, `high_note_density` — pre-existing flags (see `polyphonic_evaluation.py`) diff --git a/apps/backend/tests/fixtures/transcription_tracks/README.md b/apps/backend/tests/fixtures/transcription_tracks/README.md new file mode 100644 index 00000000..5e9a38d0 --- /dev/null +++ b/apps/backend/tests/fixtures/transcription_tracks/README.md @@ -0,0 +1,85 @@ +# Layer 2 (Torchcrepe) Transcription Bench + +This directory holds reference audio used to audit ASA's Layer 2 pitch/note translation. The audio files and their reference MIDI are **never** committed — they live in your local copy only. + +This bench is the counterpart to the polyphonic research corpus at `../polyphonic_tracks/`. Use **this** directory for material that Layer 2 (torchcrepe, monophonic-friendly) is expected to handle well; use the polyphonic directory for dense mixed material that exists only to validate research candidates. + +## What goes here + +8–15 short reference clips (10–40s each) covering the categories Layer 2 is supposed to handle: + +1. 3 monophonic vocal leads — isolated vocal stems or vocal-only renders +2. 3 isolated bass lines — DI bass, bass synth stems, or bass-only renders +3. 2 simple synth leads — saw / square / sine monophonic leads, ideally with light vibrato to exercise the pitch-jump splitter +4. 1 whistled / hummed line — single-source, near-pure-tone material +5. 1–2 acoustic instrument leads (flute, trumpet, sax) — monophonic acoustic timbres +6. 1–2 spare slots for failure modes discovered during audit (rapid runs, glissandi, voice cracks) + +Explicitly **out of scope** for this directory: dense chords, polyphonic piano, full mixes — those belong in `polyphonic_tracks/` and are not Layer 2's job. + +Total disk: roughly 50–100 MB. + +## Registering a track + +Add an entry to `../phase1_eval_manifest.json` under the `transcriptionTracks` array. The minimal shape is: + +```json +{ + "id": "monophonic_synth_lead_01", + "audioPath": "lead_01.flac", + "category": "monophonic_lead", + "description": "Saw lead over silence, monophonic, ~120 BPM", + "analyzeFlags": ["--separate", "--transcribe"], + "groundTruthNotes": [ + { "pitchMidi": 60, "onsetSeconds": 0.50, "durationSeconds": 0.25 } + ], + "thresholds": { + "noteMetrics.f1": { "target": 0.75, "tolerance": 0.0, "direction": "min" }, + "noteMetrics.meanPitchCentsError": { "target": 0.0, "tolerance": 50.0 }, + "transcriptionDetail.averageConfidence": { "target": 0.65, "tolerance": 0.0, "direction": "min" } + } +} +``` + +`audioPath` is relative to this `transcription_tracks/` directory. `groundTruthNotes` uses the same `pitchMidi`/`onsetSeconds`/`durationSeconds` shape as `transcriptionDetail.notes` emitted by analyze.py. Threshold fields under `noteMetrics.*` resolve against the per-track precision/recall/F1 computed by the harness; everything else resolves against the analyzer payload directly. + +### Building ground-truth notes from MIDI + +The harness ships a one-shot converter at `apps/backend/scripts/import_midi_to_ground_truth.py`. The producer workflow: + +1. Export a reference MIDI file from your DAW (Logic / Ableton / Reaper / Cubase / etc.). Quantization is optional — the harness's onset tolerance is ±50 ms. +2. Run the converter: + +```bash +./venv/bin/python scripts/import_midi_to_ground_truth.py reference.mid +``` + +3. Paste the emitted JSON array into your manifest entry's `groundTruthNotes` field. + +Flags: + +- `--monophonic-collapse highest` — when notes overlap, keep the highest pitch and drop the lower ones. Use for chordal MIDI that you want to flatten to a melodic line. +- `--monophonic-collapse reject` (default) — exit non-zero if any notes overlap. Use to catch accidental polyphony. +- `--offset-seconds N` — add `N` seconds to every onset. Use when your DAW reports a negative track delay or you trimmed silence from the head of the audio after exporting MIDI. + +### Ground-truth methodology + +- **Pitch** — DAW MIDI is authoritative. The harness compares semitone-quantized MIDI integers; a ±1 semitone tolerance is baked into the matcher. +- **Onset** — DAW MIDI is authoritative; ±50 ms tolerance. +- **F1 target** — 0.75 is a starting threshold for monophonic material. Tighten per-category once you have a baseline. +- **Confidence** — `transcriptionDetail.averageConfidence` ≥ 0.65 is a sanity check that Layer 2 is not running at the floor; not a substitute for F1. + +## Running the bench + +```bash +# Default — synthetic-only gate, always runnable, runs in CI +./venv/bin/python scripts/evaluate_phase1.py + +# Layer 2 opt-in — runs the stepped-sine self-test plus any registered transcription tracks +./venv/bin/python scripts/evaluate_phase1.py --include-transcription + +# Specify a custom directory for transcription tracks +./venv/bin/python scripts/evaluate_phase1.py --include-transcription --transcription-tracks-dir /path/to/tracks +``` + +The stepped-sine self-test always runs when `--include-transcription` is set, so you get at least one F1 row even before populating the corpus. Missing audio for manifest-registered tracks is reported as skipped (not failed) with a clear notice. diff --git a/apps/backend/tests/test_phase1_evaluation_transcription.py b/apps/backend/tests/test_phase1_evaluation_transcription.py new file mode 100644 index 00000000..a5496d77 --- /dev/null +++ b/apps/backend/tests/test_phase1_evaluation_transcription.py @@ -0,0 +1,333 @@ +"""Pure-function tests for the Layer 2 (transcription) evaluation harness. + +These tests deliberately avoid loading torchcrepe by injecting a fake +transcribe_runner into the harness — they validate the matcher, metrics, +threshold-direction extension, and the MIDI import script without paying the +model-load cost. +""" + +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +import pretty_midi + +from phase1_evaluation import ( + DEFAULT_MANIFEST_PATH, + _compute_note_metrics, + _evaluate_threshold, + _evaluate_transcription_track, + _match_notes, + run_phase1_evaluation, +) + +REPO_DIR = Path(__file__).resolve().parent.parent +IMPORT_SCRIPT = REPO_DIR / "scripts" / "import_midi_to_ground_truth.py" + + +def _note(pitch_midi: int, onset_seconds: float, duration_seconds: float = 0.25) -> dict: + return { + "pitchMidi": pitch_midi, + "onsetSeconds": round(onset_seconds, 4), + "durationSeconds": round(duration_seconds, 4), + } + + +class MatchNotesTests(unittest.TestCase): + def test_perfect_match_pairs_all_notes(self) -> None: + gt = [_note(60, 0.0), _note(62, 0.5), _note(64, 1.0)] + detected = [_note(60, 0.01), _note(62, 0.51), _note(64, 1.02)] + matches = _match_notes(detected, gt) + self.assertEqual(len(matches), 3) + self.assertEqual(sorted(matches), [(0, 0), (1, 1), (2, 2)]) + + def test_missed_note_leaves_gt_unmatched(self) -> None: + gt = [_note(60, 0.0), _note(62, 0.5), _note(64, 1.0)] + detected = [_note(60, 0.0), _note(64, 1.0)] + matches = _match_notes(detected, gt) + gt_matched = {g for g, _d in matches} + self.assertEqual(len(matches), 2) + self.assertNotIn(1, gt_matched) + + def test_spurious_detected_note_is_unmatched(self) -> None: + gt = [_note(60, 0.0)] + detected = [_note(60, 0.0), _note(70, 2.0)] + matches = _match_notes(detected, gt) + self.assertEqual(matches, [(0, 0)]) + + def test_pitch_off_by_two_semitones_does_not_match(self) -> None: + gt = [_note(60, 0.0)] + detected = [_note(62, 0.0)] + matches = _match_notes(detected, gt) + self.assertEqual(matches, []) + + def test_onset_off_by_100ms_does_not_match(self) -> None: + gt = [_note(60, 0.0)] + detected = [_note(60, 0.1)] + matches = _match_notes(detected, gt) + self.assertEqual(matches, []) + + def test_two_detected_within_window_first_wins(self) -> None: + gt = [_note(60, 0.5)] + detected = [_note(60, 0.46), _note(60, 0.52)] + matches = _match_notes(detected, gt) + self.assertEqual(matches, [(0, 0)]) + + +class ComputeNoteMetricsTests(unittest.TestCase): + def test_empty_detected_returns_precision_one(self) -> None: + metrics = _compute_note_metrics([], [_note(60, 0.0)], []) + self.assertEqual(metrics["precision"], 1.0) + self.assertEqual(metrics["recall"], 0.0) + self.assertEqual(metrics["f1"], 0.0) + self.assertEqual(metrics["matchedCount"], 0) + self.assertEqual(metrics["missedCount"], 1) + self.assertEqual(metrics["falsePositiveCount"], 0) + + def test_empty_ground_truth_returns_recall_one(self) -> None: + metrics = _compute_note_metrics([_note(60, 0.0)], [], []) + self.assertEqual(metrics["recall"], 1.0) + self.assertEqual(metrics["precision"], 0.0) + self.assertEqual(metrics["falsePositiveCount"], 1) + + def test_both_empty_returns_f1_one(self) -> None: + metrics = _compute_note_metrics([], [], []) + self.assertEqual(metrics["precision"], 1.0) + self.assertEqual(metrics["recall"], 1.0) + self.assertEqual(metrics["f1"], 1.0) + + def test_perfect_match_metrics(self) -> None: + gt = [_note(60, 0.0), _note(62, 0.5)] + detected = [_note(60, 0.0), _note(62, 0.5)] + matches = _match_notes(detected, gt) + metrics = _compute_note_metrics(detected, gt, matches) + self.assertEqual(metrics["f1"], 1.0) + self.assertEqual(metrics["meanPitchCentsError"], 0.0) + + def test_signed_cents_error_uses_detected_minus_ground_truth(self) -> None: + gt = [_note(60, 0.0)] + detected = [_note(61, 0.0)] + matches = _match_notes(detected, gt) + metrics = _compute_note_metrics(detected, gt, matches) + self.assertEqual(metrics["meanPitchCentsError"], 100.0) + + +class EvaluateThresholdDirectionTests(unittest.TestCase): + def test_min_direction_passes_at_target(self) -> None: + check = _evaluate_threshold( + {"v": 0.75}, "v", {"target": 0.75, "tolerance": 0.0, "direction": "min"} + ) + self.assertTrue(check.passed) + + def test_min_direction_passes_above_target(self) -> None: + check = _evaluate_threshold( + {"v": 0.9}, "v", {"target": 0.75, "tolerance": 0.0, "direction": "min"} + ) + self.assertTrue(check.passed) + + def test_min_direction_fails_below_bound(self) -> None: + check = _evaluate_threshold( + {"v": 0.7}, "v", {"target": 0.75, "tolerance": 0.0, "direction": "min"} + ) + self.assertFalse(check.passed) + + def test_min_direction_tolerance_relaxes_bound(self) -> None: + check = _evaluate_threshold( + {"v": 0.7}, "v", {"target": 0.75, "tolerance": 0.1, "direction": "min"} + ) + self.assertTrue(check.passed) + + def test_max_direction_passes_at_target(self) -> None: + check = _evaluate_threshold( + {"v": 50.0}, "v", {"target": 50.0, "tolerance": 0.0, "direction": "max"} + ) + self.assertTrue(check.passed) + + def test_max_direction_fails_above_bound(self) -> None: + check = _evaluate_threshold( + {"v": 60.0}, "v", {"target": 50.0, "tolerance": 5.0, "direction": "max"} + ) + self.assertFalse(check.passed) + + def test_symmetric_default_still_applies(self) -> None: + check = _evaluate_threshold( + {"v": 1.1}, "v", {"target": 1.0, "tolerance": 0.2} + ) + self.assertTrue(check.passed) + check = _evaluate_threshold( + {"v": 1.5}, "v", {"target": 1.0, "tolerance": 0.2} + ) + self.assertFalse(check.passed) + + +class TranscriptionTrackHarnessTests(unittest.TestCase): + def test_evaluate_transcription_track_with_injected_runner(self) -> None: + """Cover the end-to-end harness path without paying torchcrepe cost. + + The injected runner returns a known transcriptionDetail payload so that + threshold-on-noteMetrics splicing and the all_passed roll-up are + exercised against deterministic inputs. + """ + ground_truth = [_note(60, 0.0), _note(62, 0.5), _note(64, 1.0)] + with tempfile.TemporaryDirectory(prefix="asa_transcription_track_") as temp_dir: + tracks_dir = Path(temp_dir) + audio_path = tracks_dir / "fake_track.wav" + audio_path.write_bytes(b"placeholder") + + def fake_runner(_path: Path, _flags: list[str]) -> dict: + return { + "transcriptionDetail": { + "averageConfidence": 0.8, + "notes": [ + _note(60, 0.0), + _note(62, 0.5), + _note(64, 1.0), + ], + } + } + + entry = { + "id": "perfect_match_synthetic", + "audioPath": "fake_track.wav", + "category": "monophonic_lead", + "description": "Perfect-match fake runner", + "analyzeFlags": ["--transcribe"], + "groundTruthNotes": ground_truth, + "thresholds": { + "noteMetrics.f1": {"target": 0.75, "tolerance": 0.0, "direction": "min"}, + "noteMetrics.meanPitchCentsError": {"target": 0.0, "tolerance": 50.0}, + "transcriptionDetail.averageConfidence": { + "target": 0.65, + "tolerance": 0.0, + "direction": "min", + }, + }, + } + + result = _evaluate_transcription_track(entry, tracks_dir, transcribe_runner=fake_runner) + self.assertEqual(result.status, "evaluated") + self.assertTrue(result.all_passed) + self.assertEqual(result.note_metrics["matchedCount"], 3) + self.assertEqual(result.note_metrics["f1"], 1.0) + + def test_missing_audio_is_skipped_not_failed(self) -> None: + with tempfile.TemporaryDirectory(prefix="asa_transcription_missing_") as temp_dir: + tracks_dir = Path(temp_dir) + entry = { + "id": "absent_audio", + "audioPath": "nope.wav", + "category": "monophonic_lead", + "description": "Audio absent on disk", + "groundTruthNotes": [], + "thresholds": {}, + } + result = _evaluate_transcription_track(entry, tracks_dir) + self.assertEqual(result.status, "skipped_audio_missing") + self.assertTrue(result.all_passed) + self.assertIn("audio not present at", result.skip_reason) + + def test_run_phase1_with_transcription_skips_when_self_test_runner_fakes_success( + self, + ) -> None: + """run_phase1_evaluation with include_transcription should add summary + counters and a transcriptionTracks block — exercised here with an + injected runner so we do not boot the real model. + """ + with tempfile.TemporaryDirectory(prefix="asa_phase1_with_transcription_") as temp_dir: + report_path = Path(temp_dir) / "phase1_eval_report.json" + + def fake_runner(_path: Path, _flags: list[str]) -> dict: + return { + "transcriptionDetail": { + "averageConfidence": 0.9, + "notes": [ + _note(60, 0.0, 0.6), + _note(64, 0.8, 0.6), + _note(67, 1.6, 0.6), + _note(72, 2.4, 0.6), + ], + } + } + + report = run_phase1_evaluation( + manifest_path=DEFAULT_MANIFEST_PATH, + report_path=report_path, + runs_per_fixture=1, + include_transcription=True, + transcription_tracks_dir=Path(temp_dir), + transcribe_runner=fake_runner, + ) + + self.assertTrue(report["includeTranscription"]) + self.assertIn("transcriptionTracks", report) + self.assertGreaterEqual(report["summary"]["transcriptionTracksEvaluated"], 1) + self.assertEqual(report["summary"]["transcriptionTracksAnalyzeFailed"], 0) + self_test_row = report["transcriptionTracks"][0] + self.assertEqual(self_test_row["id"], "stepped_sine_synthetic") + self.assertEqual(self_test_row["status"], "evaluated") + + +class ImportMidiScriptTests(unittest.TestCase): + def _write_midi(self, path: Path, notes: list[tuple[int, float, float]]) -> None: + midi = pretty_midi.PrettyMIDI() + instrument = pretty_midi.Instrument(program=0) + for pitch, start, end in notes: + instrument.notes.append( + pretty_midi.Note(velocity=90, pitch=pitch, start=start, end=end) + ) + midi.instruments.append(instrument) + midi.write(str(path)) + + def _invoke(self, midi_path: Path, *extra_args: str) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, str(IMPORT_SCRIPT), str(midi_path), *extra_args], + capture_output=True, + text=True, + check=False, + ) + + def test_round_trip_emits_expected_notes(self) -> None: + with tempfile.TemporaryDirectory(prefix="asa_import_midi_") as temp_dir: + midi_path = Path(temp_dir) / "ref.mid" + self._write_midi(midi_path, [(60, 0.0, 0.5), (62, 0.5, 1.0), (64, 1.0, 1.5)]) + result = self._invoke(midi_path) + self.assertEqual(result.returncode, 0, msg=result.stderr) + notes = json.loads(result.stdout) + self.assertEqual(len(notes), 3) + self.assertEqual(notes[0]["pitchMidi"], 60) + self.assertEqual(notes[0]["onsetSeconds"], 0.0) + self.assertAlmostEqual(notes[0]["durationSeconds"], 0.5, places=2) + + def test_overlapping_notes_with_default_rejects(self) -> None: + with tempfile.TemporaryDirectory(prefix="asa_import_midi_reject_") as temp_dir: + midi_path = Path(temp_dir) / "chord.mid" + self._write_midi(midi_path, [(60, 0.0, 0.5), (64, 0.1, 0.5)]) + result = self._invoke(midi_path) + self.assertNotEqual(result.returncode, 0) + self.assertIn("overlapping note pair", result.stderr) + + def test_monophonic_collapse_highest_keeps_top_pitch(self) -> None: + with tempfile.TemporaryDirectory(prefix="asa_import_midi_collapse_") as temp_dir: + midi_path = Path(temp_dir) / "chord.mid" + self._write_midi(midi_path, [(60, 0.0, 0.5), (64, 0.1, 0.5), (67, 0.2, 0.5)]) + result = self._invoke(midi_path, "--monophonic-collapse", "highest") + self.assertEqual(result.returncode, 0, msg=result.stderr) + notes = json.loads(result.stdout) + pitches = sorted(note["pitchMidi"] for note in notes) + self.assertEqual(pitches, [67]) + + def test_offset_seconds_shifts_onsets(self) -> None: + with tempfile.TemporaryDirectory(prefix="asa_import_midi_offset_") as temp_dir: + midi_path = Path(temp_dir) / "ref.mid" + self._write_midi(midi_path, [(60, 0.0, 0.5)]) + result = self._invoke(midi_path, "--offset-seconds", "0.25") + self.assertEqual(result.returncode, 0, msg=result.stderr) + notes = json.loads(result.stdout) + self.assertAlmostEqual(notes[0]["onsetSeconds"], 0.25, places=2) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/backend/tests/test_polyphonic_evaluation.py b/apps/backend/tests/test_polyphonic_evaluation.py index a50d336d..c99bf481 100644 --- a/apps/backend/tests/test_polyphonic_evaluation.py +++ b/apps/backend/tests/test_polyphonic_evaluation.py @@ -1,4 +1,6 @@ import json +import subprocess +import sys import tempfile import unittest from pathlib import Path @@ -14,6 +16,9 @@ summarize_midi_file, ) +REPO_DIR = Path(__file__).resolve().parent.parent +SCORECARD_SCRIPT = REPO_DIR / "scripts" / "score_polyphonic_clip.py" + def _write_wav(path: Path, duration_seconds: float = 1.0, sample_rate: int = 22050) -> None: sample_count = int(duration_seconds * sample_rate) @@ -158,6 +163,175 @@ def fake_basic_pitch_runner(clip_id: str, _audio_path: Path, candidate_output_di self.assertEqual(candidate_report["metrics"]["maxPolyphony"], 2) +class PolyphonicFlagRuleTests(unittest.TestCase): + def _summarize(self, note_specs: list[tuple[int, float, float]], duration_s: float) -> dict: + with tempfile.TemporaryDirectory(prefix="asa_polyphonic_flags_") as temp_dir: + midi_path = Path(temp_dir) / "candidate.mid" + _write_midi(midi_path, note_specs) + return summarize_midi_file(midi_path, audio_duration_seconds=duration_s) + + def test_note_clutter_flag_when_density_exceeds_fifteen(self) -> None: + # 32 notes across 2.0s = 16 notes/sec — clears > 15 threshold but + # uses spaced onsets so it does not also trigger dense_chords_unusable. + note_specs = [(60 + (i % 4), i * 0.0625, i * 0.0625 + 0.05) for i in range(32)] + summary = self._summarize(note_specs, duration_s=2.0) + self.assertIn("note_clutter", summary["flags"]) + self.assertNotIn("dense_chords_unusable", summary["flags"]) + self.assertNotIn("octave_junk", summary["flags"]) + self.assertNotIn("sparse_likely_undertranscribed", summary["flags"]) + + def test_octave_junk_flag_when_many_distinct_pitches_low_polyphony(self) -> None: + # 32 distinct pitches strung end-to-end; max polyphony = 1 < 3. + note_specs = [(40 + i, i * 0.1, i * 0.1 + 0.08) for i in range(32)] + summary = self._summarize(note_specs, duration_s=10.0) + self.assertIn("octave_junk", summary["flags"]) + self.assertNotIn("dense_chords_unusable", summary["flags"]) + self.assertNotIn("note_clutter", summary["flags"]) + self.assertNotIn("sparse_likely_undertranscribed", summary["flags"]) + + def test_dense_chords_unusable_flag_when_max_polyphony_above_eight(self) -> None: + # 10 simultaneous notes within first half-second; nothing else fires + # the other new flags (density is 10 notes / 2.0s = 5). + note_specs = [(48 + i, 0.0, 0.5) for i in range(10)] + summary = self._summarize(note_specs, duration_s=2.0) + self.assertIn("dense_chords_unusable", summary["flags"]) + self.assertNotIn("note_clutter", summary["flags"]) + self.assertNotIn("octave_junk", summary["flags"]) + self.assertNotIn("sparse_likely_undertranscribed", summary["flags"]) + + def test_sparse_likely_undertranscribed_flag_when_density_low_over_long_clip(self) -> None: + # 5 notes across 15.0s = 0.33 notes/sec, well below 1.0; clip long enough. + note_specs = [(60, i * 3.0, i * 3.0 + 0.2) for i in range(5)] + summary = self._summarize(note_specs, duration_s=15.0) + self.assertIn("sparse_likely_undertranscribed", summary["flags"]) + self.assertNotIn("note_clutter", summary["flags"]) + self.assertNotIn("dense_chords_unusable", summary["flags"]) + + def test_existing_monophonic_and_high_density_flags_still_emit(self) -> None: + # Regression guard for the pre-existing flags untouched by the new rules. + note_specs = [(60, i * 0.08, i * 0.08 + 0.05) for i in range(20)] + summary = self._summarize(note_specs, duration_s=1.5) + self.assertIn("monophonic_output", summary["flags"]) + self.assertIn("high_note_density", summary["flags"]) + + +class ScorePolyphonicClipScriptTests(unittest.TestCase): + def test_no_play_with_piped_input_writes_scorecard_back(self) -> None: + """Drive score_polyphonic_clip.py via subprocess + stdin. + + Builds a tempdir report JSON with one unscored clip / one candidate, + pipes the prompt answers in, asserts the scorecard fields land in the + written report. + """ + with tempfile.TemporaryDirectory(prefix="asa_scorecard_cli_") as temp_dir: + report_path = Path(temp_dir) / "report.json" + audio_path = Path(temp_dir) / "missing.wav" # ok — playback skipped + report = { + "clips": [ + { + "id": "clip_one", + "audioPath": str(audio_path), + "candidates": { + "basic-pitch": { + "status": "completed", + "metrics": { + "noteCount": 12, + "maxPolyphony": 4, + "noteDensityPerSecond": 6.0, + "flags": ["high_note_density"], + }, + "scorecard": build_manual_scorecard(None), + } + }, + } + ] + } + report_path.write_text(json.dumps(report), encoding="utf-8") + + stdin_lines = [ + "y", # bassRecognizable + "y", # toplineRecognizable + "n", # chordsNotObviouslyWrong + "4.25", # cleanupMinutes30s + "review notes", # notes + "", + ] + result = subprocess.run( + [ + sys.executable, + str(SCORECARD_SCRIPT), + "--report", + str(report_path), + "--no-play", + ], + input="\n".join(stdin_lines), + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(result.returncode, 0, msg=result.stderr) + + updated = json.loads(report_path.read_text(encoding="utf-8")) + scorecard = updated["clips"][0]["candidates"]["basic-pitch"]["scorecard"] + self.assertTrue(scorecard["bassRecognizable"]) + self.assertTrue(scorecard["toplineRecognizable"]) + self.assertFalse(scorecard["chordsNotObviouslyWrong"]) + self.assertEqual(scorecard["cleanupMinutes30s"], 4.25) + self.assertEqual(scorecard["notes"], "review notes") + + def test_already_scored_clip_is_skipped_without_rescore(self) -> None: + with tempfile.TemporaryDirectory(prefix="asa_scorecard_cli_skip_") as temp_dir: + report_path = Path(temp_dir) / "report.json" + complete = build_manual_scorecard( + { + "bassRecognizable": True, + "toplineRecognizable": True, + "chordsNotObviouslyWrong": True, + "cleanupMinutes30s": 2.0, + "notes": "preexisting", + } + ) + report = { + "clips": [ + { + "id": "clip_one", + "audioPath": str(Path(temp_dir) / "missing.wav"), + "candidates": { + "basic-pitch": { + "status": "completed", + "metrics": { + "noteCount": 1, + "maxPolyphony": 1, + "noteDensityPerSecond": 0.1, + "flags": [], + }, + "scorecard": complete, + } + }, + } + ] + } + report_path.write_text(json.dumps(report), encoding="utf-8") + + result = subprocess.run( + [ + sys.executable, + str(SCORECARD_SCRIPT), + "--report", + str(report_path), + "--no-play", + ], + input="", + capture_output=True, + text=True, + check=False, + ) + self.assertEqual(result.returncode, 0, msg=result.stderr) + summary = json.loads(result.stdout) + self.assertEqual(summary["scored"], 0) + self.assertEqual(summary["skippedAlreadyComplete"], 1) + + if __name__ == "__main__": unittest.main() diff --git a/docs/POLYPHONIC_TRANSCRIPTION_SPIKE.md b/docs/POLYPHONIC_TRANSCRIPTION_SPIKE.md index 8599b8d8..f6121163 100644 --- a/docs/POLYPHONIC_TRANSCRIPTION_SPIKE.md +++ b/docs/POLYPHONIC_TRANSCRIPTION_SPIKE.md @@ -165,3 +165,68 @@ Close the question if the outputs show any of these failure patterns: Do not add a polyphonic backend to the product just because a model can emit MIDI. In plain English: "possible" is not the bar. The bar is "good enough that a producer would actually choose to use it." + +## Corpus + Scoring Workflow + +The harness ships with a dedicated fixtures directory and an interactive scorecard CLI to reduce the friction of running this spike on a real corpus. + +This corpus is distinct from the Layer 2 (torchcrepe) bench at `apps/backend/tests/fixtures/transcription_tracks/`. Use: + +- **`tests/fixtures/polyphonic_tracks/`** — dense, mixed, mastered material for *this* research spike. +- **`tests/fixtures/transcription_tracks/`** — monophonic-friendly material for Layer 2, which **is** in the product. See `docs/LAYER2_EVALUATION.md`. + +A clip that fits one corpus generally does not fit the other. + +### Target material distribution + +10–20 short clips, 20–40 seconds each, weighted toward producer-realistic mixes: + +1. 3–4 dense electronic mixes (mastered, chord-stack-heavy) +2. 2–3 vocal-heavy with backing harmony +3. 2–3 piano-only / piano-heavy clips +4. 2–3 busy mastered full-band mixes +5. 1–2 pad + arpeggio / pad + bass slices +6. 1–2 sparse / minimal clips (negative examples for the `sparse_likely_undertranscribed` diagnostic) + +### Sourcing checklist + +1. Use audio you own or license. Nothing copyrighted is committed; the fixtures directory carries only a `README.md`. +2. Mono or stereo both fine. +3. Format: WAV / FLAC / MP3 / M4A all work. +4. Pick clips where you can confidently answer all five scorecard fields. Clips that leave the reviewer unsure are not useful gate material. + +### Where clips live + +`apps/backend/tests/fixtures/polyphonic_tracks/` (never committed). The harness itself does not assume a checked-in manifest — you supply a manifest path via `--manifest`. `audioPath` entries may be absolute or relative to the manifest file. + +### Scoring CLI + +After running `evaluate_polyphonic.py`, score the clips interactively: + +```bash +cd apps/backend +./venv/bin/python scripts/score_polyphonic_clip.py \ + --report .runtime/polyphonic_eval/polyphonic_eval_report.json +``` + +The CLI walks each unscored clip × candidate pair, plays the audio (`afplay` / `paplay` / `aplay`), surfaces the diagnostic `flags` plus key metrics, and writes the scorecard back into the report JSON in place. + +Flags: + +- `--rescore` — re-prompt for clips that already have a complete scorecard. +- `--no-play` — skip audio playback. Required for non-interactive / headless usage. +- `--candidate ` — limit scoring to a single candidate, e.g. `--candidate basic-pitch`. + +After the CLI exits, re-run the per-candidate gate by feeding the updated report through `summarize_candidate_gate` (the harness's main JSON report regenerates the rollups automatically on the next `evaluate_polyphonic.py` run). + +### Diagnostic flags are seeding only + +`summarize_midi_file` now emits a richer `flags` list that steers the reviewer's listening attention. These flags are **diagnostic seeding only** — they do not bypass the five manual usefulness gates, and a clip with no flags can still fail the gates on listen-through: + +- `note_clutter` — > 15 notes / sec; likely junk +- `octave_junk` — > 30 distinct pitches with mean active polyphony < 3; suggests scattered octave errors +- `dense_chords_unusable` — max polyphony > 8 simultaneous notes; usually unrecoverable +- `sparse_likely_undertranscribed` — < 1 note / sec across a 10s+ clip; the model probably gave up +- `empty_output`, `monophonic_output`, `high_note_density` — pre-existing flags + +In plain English: the flags tell you what to listen *for*. The gates tell you whether to ship.