diff --git a/CLAUDE.md b/CLAUDE.md index 2a7aa062..97ac113d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -174,6 +174,7 @@ Artifact access goes through `artifact_storage.py` rather than direct disk paths 15. **`dsp_bandbank.py` + `dsp_utils.py`**: Shared DSP primitives — `BatchedBandpass` (4th-order Butterworth bandpass bank) and cross-module utility functions. 16. **`phase1_evaluation.py` + `phase1_report_html.py`, `polyphonic_evaluation.py`, `beat_evaluation.py` + `beat_report_html.py`, `loudness_rec_evaluation.py`, `recommendation_evaluation.py`**: Offline evaluation harnesses (deterministic-metric / detector-stability reporting, research-only polyphonic transcription, the beat/downbeat measurement gate that benchmarks CPJKU/beat_this against the shipping kick-accent heuristic, loudness-recommendation reachability, and the recommendation-quality scorer that grades Phase 2 recommendations against known-settings fixtures — see [`apps/backend/NEEDS.md`](apps/backend/NEEDS.md)). Not on the product path; driven by `scripts/evaluate_*.py`. Deleting them restores the product exactly. 17. **`utils/cleanup.py`**: Periodic artifact-cleanup helpers used by the server background-task loop. +18. **`transcription_pianoroll.py`**: Renders the pitch-note translation stage's `transcriptionDetail` as a velocity-encoded `(pitch, time)` uint8 matrix via [`symusic`](https://github.com/Yikai-Liao/symusic). Backs `GET /api/analysis-runs/{run_id}/transcription/pianoroll`. Derived view, never overrides Phase 1; the response cites Phase 1's `bpm` + `timeSignature` so chain of custody is preserved. Note: `transcriptionDetail` is *stripped* from `measurement.result` (see `analysis_runtime.py:685`) and lives in the `pitchNoteTranslation` stage instead — the route reads from there. The subprocess isolation means `analyze.py` works as a standalone CLI. Check `apps/backend/JSON_SCHEMA.md` before adding new analyzer output fields. Check `apps/backend/ARCHITECTURE.md` for the full HTTP flow and contract details. diff --git a/apps/backend/ARCHITECTURE.md b/apps/backend/ARCHITECTURE.md index ef245606..2683d965 100644 --- a/apps/backend/ARCHITECTURE.md +++ b/apps/backend/ARCHITECTURE.md @@ -16,6 +16,7 @@ | `url_ingest.py` | SSRF-guarded URL-mode ingestion for `POST /api/analysis-runs`. Fetches a public `http`/`https` audio file and streams the bytes through the same downstream pipeline as a multipart upload, enforcing the shared 100 MiB cap. | | `audio_mime.py` | Canonical, host-independent filename→MIME resolution for ingested audio (`canonical_audio_mime`). Mirrors the frontend map in `apps/ui/src/services/audioFile.ts` so a `.flac` resolves to `audio/flac` on every OS — stdlib `mimetypes` is host-dependent (`audio/x-flac` on macOS vs `audio/flac` on Linux), which failed the test gate and could mislabel a FLAC handed to Gemini. Imported directly by `server_phase2.py` and `url_ingest.py`, and reached by `server.py` via `server_phase2._get_audio_mime_type`. | | `csv_export.py` | CSV exporters for Phase 1 time-series fields, keyed by dotted JSON path. Backs `GET /api/analysis-runs/{run_id}/export/csv/{field_path}` and keeps the route handler a thin lookup-and-serve. | +| `transcription_pianoroll.py` | Renders the pitch-note translation stage's `transcriptionDetail` as a velocity-encoded `(pitch, time)` uint8 matrix via `symusic`. Backs `GET /api/analysis-runs/{run_id}/transcription/pianoroll`. Derived view — Phase 1 stays authoritative; the response cites the Phase 1 `bpm` and `timeSignature` so chain of custody (PURPOSE invariant #2) is preserved. | | `stage_status.py` | Collapses the eight internal stage statuses into the additive client-facing `publicStatus` field carried on every stage in the run snapshot. | | `server_samples.py` + `sample_generation.py` / `sample_theory.py` / `sample_synthesis.py` / `sample_drums.py` | Phase 3 audition-sample generation. PyTheory musical plan, FluidSynth (sine-additive fallback) audio render, NumPy drum one-shots, and a citation manifest tying every clip back to a Phase 1 field. On-demand only — not part of the staged-execution queue. See [`docs/SAMPLE_GENERATION.md`](../../docs/SAMPLE_GENERATION.md). | | `dsp_bandbank.py` + `dsp_utils.py` | Shared DSP primitives: `BatchedBandpass` (4th-order Butterworth bandpass bank with zero-phase `filtfilt`) and cross-module utility functions. | @@ -88,6 +89,7 @@ Custom routes: - `GET /api/analysis-runs/{run_id}/artifacts` and `…/artifacts/{artifact_id}` - `GET /api/analysis-runs/{run_id}/source-audio` — re-serves the original ingested audio for the run. Owner-only (no admin bypass). Saves a round-trip vs looking up the source artifact id first. - `GET /api/analysis-runs/{run_id}/export/csv/{field_path}` — CSV export of a Phase 1 time-series field. See [`docs/adr/0001-phase1-json-schema-v1.md`](../../docs/adr/0001-phase1-json-schema-v1.md) and the registry in [`csv_export.py`](csv_export.py). +- `GET /api/analysis-runs/{run_id}/transcription/pianoroll` — velocity-encoded pianoroll matrix derived from the pitch-note translation stage's `transcriptionDetail`. Query params: `mode` (`frame`|`onset`, default `frame`), `pitchLow` (default 21), `pitchHigh` (default 109, exclusive), `tpq` (default 4). Response cites the Phase 1 `bpm` and `timeSignature` so every cell traces back to a measurement. Status codes: 200 (payload), 400 (`INVALID_MODE`/`INVALID_PITCH_RANGE`/`INVALID_TPQ`), 404 (`RUN_NOT_FOUND`/`TRANSCRIPTION_NOT_REQUESTED`/`TRANSCRIPTION_NOT_AVAILABLE`), 409 (`MEASUREMENT_NOT_COMPLETED`/`TRANSCRIPTION_NOT_COMPLETED`). Implementation in [`transcription_pianoroll.py`](transcription_pianoroll.py). - `POST /api/analysis-runs/{run_id}/spectral-enhancements/{kind}` — on-demand spectral artifacts. `kind` is one of `cqt`, `hpss`, `onset`, `chroma_interactive`, or `reassigned` (sharper transient/frequency localization via `librosa.reassigned_spectrogram`). - `POST /api/analysis-runs/{run_id}/pitch-note-translations` - `POST /api/analysis-runs/{run_id}/interpretations` diff --git a/apps/backend/requirements.txt b/apps/backend/requirements.txt index 35b298bf..bb6278cd 100644 --- a/apps/backend/requirements.txt +++ b/apps/backend/requirements.txt @@ -58,6 +58,11 @@ soundfile==0.13.1 soxr==1.0.0 starlette==0.52.1 submitit==1.5.4 +# Symbolic music library — fast MIDI/Score primitives with a C++ core. Used by +# transcription_pianoroll.py to convert transcriptionDetail.notes into a +# velocity-encoded matrix for the UI heatmap. Ships cp311 wheels for darwin-arm64 +# and manylinux_x86_64 — no compile step. +symusic==0.6.0 sympy==1.14.0 threadpoolctl==3.6.0 torch==2.10.0 diff --git a/apps/backend/server.py b/apps/backend/server.py index 3346d929..6ee33f55 100644 --- a/apps/backend/server.py +++ b/apps/backend/server.py @@ -48,6 +48,7 @@ ) from utils.cleanup import cleanup_artifacts import csv_export +import transcription_pianoroll import upload_limits import url_ingest from server_upload import ( # noqa: F401 — re-exported for test backward compat @@ -2544,6 +2545,200 @@ async def export_run_field_as_csv( ) +@app.get("/api/analysis-runs/{run_id}/transcription/pianoroll") +async def get_transcription_pianoroll( + run_id: str, + mode: str = Query( + "frame", + description="'frame' (sustained notes painted across their duration) " + "or 'onset' (note starts only — useful for rhythmic visualization).", + ), + pitch_low: int = Query( + transcription_pianoroll.DEFAULT_PITCH_LOW, + alias="pitchLow", + description="Lower MIDI pitch bound (inclusive). Default: 21 (A0).", + ), + pitch_high: int = Query( + transcription_pianoroll.DEFAULT_PITCH_HIGH, + alias="pitchHigh", + description="Upper MIDI pitch bound (exclusive). Default: 109 " + "(one above C8 — the 88-key piano range).", + ), + tpq: int = Query( + transcription_pianoroll.DEFAULT_TPQ, + description="Time resolution in ticks per quarter note. Default: 4 " + "(one row per 16th note at typical tempos).", + ), + x_asa_user_id: str | None = Header(None), + x_asa_user_email: str | None = Header(None), +) -> JSONResponse: + """Render the run's transcriptionDetail as a velocity-encoded pianoroll. + + Derived view — Phase 1's ``transcriptionDetail`` (sourced from the + pitch-note translation stage) stays authoritative. The response cites + Phase 1's ``bpm`` + ``timeSignature`` so the UI surface can attribute + every cell back to a measurement (invariant #2 in ``PURPOSE.md``). + + Status mapping is deliberately split between 404 (won't appear without + user action) and 409 (wait or finish a prerequisite): + + * 404 ``RUN_NOT_FOUND`` — run doesn't exist or is not owned by the caller. + * 409 ``MEASUREMENT_NOT_COMPLETED`` — measurement stage hasn't finished. + * 404 ``TRANSCRIPTION_NOT_REQUESTED`` — run was created with pitch-note + translation disabled. The caller should enable transcription on a new run. + * 409 ``TRANSCRIPTION_NOT_COMPLETED`` — pitch-note stage is queued/running. + * 404 ``TRANSCRIPTION_NOT_AVAILABLE`` — pitch-note stage completed (or + failed / was interrupted) without producing a ``transcriptionDetail`` + payload. Caller's affordance is "re-run the transcription stage." + * 400 ``INVALID_MODE`` / ``INVALID_PITCH_RANGE`` / ``INVALID_TPQ`` — + query parameter validation. Returned as structured codes rather than + FastAPI's default 422 so clients can branch on them. + """ + # Up-front query validation so the route returns structured error codes + # rather than letting ``render_pianoroll`` raise generic ``ValueError`` + # — clients should be able to switch on the specific failure. + if mode not in ("frame", "onset"): + return JSONResponse( + status_code=400, + content={ + "error": { + "code": "INVALID_MODE", + "message": f"mode must be 'frame' or 'onset'; got {mode!r}.", + } + }, + ) + if pitch_low < 0 or pitch_high > 128 or pitch_low >= pitch_high: + return JSONResponse( + status_code=400, + content={ + "error": { + "code": "INVALID_PITCH_RANGE", + "message": ( + f"pitch range [{pitch_low}, {pitch_high}) is invalid; " + "must satisfy 0 <= pitchLow < pitchHigh <= 128." + ), + } + }, + ) + if tpq < 1: + return JSONResponse( + status_code=400, + content={ + "error": { + "code": "INVALID_TPQ", + "message": f"tpq must be >= 1; got {tpq}.", + } + }, + ) + + user_context = _resolve_route_user_context(x_asa_user_id, x_asa_user_email) + if isinstance(user_context, JSONResponse): + return user_context + + runtime = get_analysis_runtime() + try: + snapshot = runtime.get_run(run_id, owner_user_id=user_context.user_id) + except (KeyError, PermissionError): + return _run_not_found_response(run_id) + + stages = snapshot.get("stages", {}) if isinstance(snapshot, dict) else {} + measurement = stages.get("measurement", {}) or {} + if measurement.get("status") != "completed": + return JSONResponse( + status_code=409, + content={ + "error": { + "code": "MEASUREMENT_NOT_COMPLETED", + "message": ( + "Measurement stage must complete before the " + "transcription pianoroll can be rendered." + ), + } + }, + ) + + pn_stage = stages.get("pitchNoteTranslation", {}) or {} + pn_status = pn_stage.get("status") + if pn_status == "not_requested": + return JSONResponse( + status_code=404, + content={ + "error": { + "code": "TRANSCRIPTION_NOT_REQUESTED", + "message": ( + "This run was created with pitch-note translation " + "disabled. Enable transcription on a new run to " + "render the pianoroll." + ), + } + }, + ) + if pn_status in ("ready", "queued", "running"): + return JSONResponse( + status_code=409, + content={ + "error": { + "code": "TRANSCRIPTION_NOT_COMPLETED", + "message": ( + "Pitch-note translation has not finished. Wait for " + "the stage to complete and try again." + ), + } + }, + ) + + pn_result = pn_stage.get("result") + transcription_detail = ( + pn_result.get("transcriptionDetail") + if isinstance(pn_result, dict) + else None + ) + if not isinstance(transcription_detail, dict): + # Covers: completed-but-null result, failed attempt, interrupted + # attempt — the user's recourse is identical (re-run transcription). + return JSONResponse( + status_code=404, + content={ + "error": { + "code": "TRANSCRIPTION_NOT_AVAILABLE", + "message": ( + "Pitch-note translation did not produce a " + "transcriptionDetail payload for this run." + ), + } + }, + ) + + measurement_result = ( + measurement.get("result") + if isinstance(measurement.get("result"), dict) + else {} + ) + bpm_value = measurement_result.get("bpm") + time_signature_value = measurement_result.get("timeSignature") + + payload = transcription_pianoroll.render_pianoroll( + transcription_detail, + bpm=( + float(bpm_value) + if isinstance(bpm_value, (int, float)) and bpm_value > 0 + else None + ), + time_signature=( + time_signature_value + if isinstance(time_signature_value, str) + else None + ), + mode=mode, # type: ignore[arg-type] + pitch_low=pitch_low, + pitch_high=pitch_high, + tpq=tpq, + ) + return JSONResponse( + content=transcription_pianoroll.payload_to_json_dict(payload) + ) + + _ENHANCEMENT_GENERATORS = { "cqt": ("generate_cqt_spectrogram", ["spectrogram_cqt"], True), "hpss": ("generate_hpss_spectrograms", ["spectrogram_harmonic", "spectrogram_percussive"], True), diff --git a/apps/backend/tests/test_server.py b/apps/backend/tests/test_server.py index ce8ddb88..655435ec 100644 --- a/apps/backend/tests/test_server.py +++ b/apps/backend/tests/test_server.py @@ -4984,6 +4984,301 @@ def test_unknown_run_returns_404(self) -> None: self.assertEqual(payload["error"]["code"], "RUN_NOT_FOUND") +class TranscriptionPianorollRouteTests(unittest.TestCase): + """Route tests for GET /api/analysis-runs/{run_id}/transcription/pianoroll. + + The pure pianoroll rasterization is covered by + tests/test_transcription_pianoroll.py. These tests verify only the HTTP + shell — status codes, error envelopes, and that the route correctly + threads transcriptionDetail out of the pitch-note translation stage + (not measurement, which strips it at analysis_runtime.py:685). + """ + + def _decode_json_response(self, response) -> dict: + return json.loads(response.body.decode("utf-8")) + + def _call_route( + self, + run_id: str, + *, + mode: str = "frame", + pitch_low: int = 21, + pitch_high: int = 109, + tpq: int = 4, + ): + # FastAPI's ``Query(...)`` defaults are sentinel objects, not the + # resolved values, so a direct unit-test call has to pass explicit + # ints/strings or the validation block sees the sentinel and bails + # with ``INVALID_MODE``. Tests opt-in via this wrapper. + return asyncio.run( + server.get_transcription_pianoroll( + run_id, + mode=mode, + pitch_low=pitch_low, + pitch_high=pitch_high, + tpq=tpq, + ) + ) + + def _make_run( + self, + runtime, + *, + measurement_payload: dict | None = None, + complete_measurement: bool = True, + pitch_note_mode: str = "auto", + translation_status: str | None = None, + translation_result: dict | None = None, + ) -> str: + created = runtime.create_run( + filename="track.mp3", + content=b"fake-audio", + mime_type="audio/mpeg", + pitch_note_mode=pitch_note_mode, + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + ) + run_id = created["runId"] + if complete_measurement: + runtime.complete_measurement( + run_id, + payload=( + measurement_payload + if measurement_payload is not None + else {"bpm": 128, "timeSignature": "4/4"} + ), + provenance={ + "schemaVersion": "measurement.v1", + "engineVersion": "analyze.py", + }, + diagnostics={"backendDurationMs": 1000}, + ) + if translation_status is not None: + runtime.create_pitch_note_attempt( + run_id, + backend_id="torchcrepe-viterbi", + mode="auto", + status=translation_status, + result=translation_result, + provenance={"engineVersion": "torchcrepe-viterbi"}, + ) + return run_id + + @staticmethod + def _transcription_result(notes: list[dict]) -> dict: + return { + "transcriptionDetail": { + "transcriptionMethod": "torchcrepe-viterbi", + "noteCount": len(notes), + "averageConfidence": 0.85, + "dominantPitches": [], + "pitchRange": { + "minMidi": 60, + "maxMidi": 67, + "minName": "C4", + "maxName": "G4", + }, + "stemSeparationUsed": False, + "fullMixFallback": True, + "stemsTranscribed": ["full_mix"], + "perStemAverageConfidence": {}, + "notes": notes, + } + } + + @staticmethod + def _note( + pitch: int = 60, + onset: float = 0.0, + duration: float = 0.5, + confidence: float = 0.9, + ) -> dict: + return { + "pitchMidi": pitch, + "pitchName": "C4", + "onsetSeconds": onset, + "durationSeconds": duration, + "confidence": confidence, + "stemSource": "full_mix", + } + + def test_returns_payload_on_happy_path(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_pianoroll_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run( + runtime, + translation_status="completed", + translation_result=self._transcription_result([self._note()]), + ) + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = self._call_route(run_id) + + self.assertEqual(response.status_code, 200) + payload = self._decode_json_response(response) + self.assertEqual(payload["mode"], "frame") + self.assertEqual(payload["pitchLow"], 21) + self.assertEqual(payload["pitchHigh"], 109) + self.assertEqual(payload["ticksPerQuarter"], 4) + self.assertEqual(payload["quartersPerMinute"], 128.0) + self.assertEqual(payload["timeSignature"], "4/4") + self.assertEqual(payload["noteCount"], 1) + # 88 pitch rows (21..109 exclusive); pitch 60 lands at index 60-21=39. + self.assertEqual(len(payload["frames"]), 88) + target_row = payload["frames"][60 - 21] + self.assertGreater(sum(target_row), 0) + + def test_unknown_run_returns_404(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_pianoroll_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = self._call_route("does-not-exist") + + self.assertEqual(response.status_code, 404) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "RUN_NOT_FOUND") + + def test_measurement_not_completed_returns_409(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_pianoroll_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run(runtime, complete_measurement=False) + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = self._call_route(run_id) + + self.assertEqual(response.status_code, 409) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "MEASUREMENT_NOT_COMPLETED") + + def test_transcription_not_requested_returns_404(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_pianoroll_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run(runtime, pitch_note_mode="off") + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = self._call_route(run_id) + + self.assertEqual(response.status_code, 404) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "TRANSCRIPTION_NOT_REQUESTED") + + def test_transcription_running_returns_409(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_pianoroll_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run(runtime, translation_status="running") + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = self._call_route(run_id) + + self.assertEqual(response.status_code, 409) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "TRANSCRIPTION_NOT_COMPLETED") + + def test_transcription_failed_returns_404_not_available(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_pianoroll_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run( + runtime, + translation_status="failed", + translation_result=None, + ) + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = self._call_route(run_id) + + self.assertEqual(response.status_code, 404) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "TRANSCRIPTION_NOT_AVAILABLE") + + def test_transcription_completed_with_null_detail_returns_404(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_pianoroll_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run( + runtime, + translation_status="completed", + translation_result={"transcriptionDetail": None}, + ) + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = self._call_route(run_id) + + self.assertEqual(response.status_code, 404) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "TRANSCRIPTION_NOT_AVAILABLE") + + def test_invalid_mode_returns_400(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_pianoroll_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run( + runtime, + translation_status="completed", + translation_result=self._transcription_result([]), + ) + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = self._call_route(run_id, mode="garbage") + + self.assertEqual(response.status_code, 400) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "INVALID_MODE") + + def test_invalid_pitch_range_returns_400(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_pianoroll_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run( + runtime, + translation_status="completed", + translation_result=self._transcription_result([]), + ) + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = self._call_route( + run_id, pitch_low=100, pitch_high=50 + ) + + self.assertEqual(response.status_code, 400) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "INVALID_PITCH_RANGE") + + def test_invalid_tpq_returns_400(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_pianoroll_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run( + runtime, + translation_status="completed", + translation_result=self._transcription_result([]), + ) + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = self._call_route(run_id, tpq=0) + + self.assertEqual(response.status_code, 400) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "INVALID_TPQ") + + class AudioMimeTypeTests(unittest.TestCase): """The Gemini-upload labeling path must resolve canonical, host-stable types. diff --git a/apps/backend/tests/test_transcription_pianoroll.py b/apps/backend/tests/test_transcription_pianoroll.py new file mode 100644 index 00000000..2f5f85c4 --- /dev/null +++ b/apps/backend/tests/test_transcription_pianoroll.py @@ -0,0 +1,336 @@ +"""Unit tests for transcription_pianoroll. + +Stdlib unittest to match the rest of apps/backend/tests/. Pure-Python — no +audio fixtures, no network, no Phase 1 run. All inputs are minimal dicts +shaped like a real transcriptionDetail. +""" + +from __future__ import annotations + +import unittest + +import numpy as np + +from transcription_pianoroll import ( + DEFAULT_PITCH_LOW, + VELOCITY_FLOOR, + _parse_time_signature, + _velocity_from_confidence, + build_score, + payload_to_json_dict, + render_pianoroll, +) + + +def _mock_transcription(notes: list[dict]) -> dict: + return { + "transcriptionMethod": "torchcrepe-viterbi", + "noteCount": len(notes), + "averageConfidence": 0.9, + "notes": notes, + } + + +def _make_note( + pitch: int, + onset: float, + duration: float, + *, + confidence: float = 0.9, + stem: str = "full_mix", +) -> dict: + return { + "pitchMidi": pitch, + "pitchName": "C4", # irrelevant for the matrix + "onsetSeconds": onset, + "durationSeconds": duration, + "confidence": confidence, + "stemSource": stem, + } + + +class VelocityMappingTests(unittest.TestCase): + def test_zero_confidence_maps_to_floor(self): + self.assertEqual(_velocity_from_confidence(0.0), VELOCITY_FLOOR) + + def test_full_confidence_maps_to_127(self): + self.assertEqual(_velocity_from_confidence(1.0), 127) + + def test_half_confidence_midrange(self): + # 64 + 63 * 0.5 = 95.5; Python's banker's rounding takes 95.5 -> 96. + self.assertEqual(_velocity_from_confidence(0.5), 96) + + def test_below_zero_clipped_to_floor(self): + self.assertEqual(_velocity_from_confidence(-2.0), VELOCITY_FLOOR) + + def test_above_one_clipped_to_max(self): + self.assertEqual(_velocity_from_confidence(5.0), 127) + + def test_non_numeric_input_clamps_to_floor(self): + self.assertEqual(_velocity_from_confidence("not-a-number"), VELOCITY_FLOOR) + self.assertEqual(_velocity_from_confidence(None), VELOCITY_FLOOR) + + +class TimeSignatureParseTests(unittest.TestCase): + def test_valid_4_4(self): + self.assertEqual(_parse_time_signature("4/4"), (4, 4)) + + def test_valid_3_4(self): + self.assertEqual(_parse_time_signature("3/4"), (3, 4)) + + def test_valid_7_8(self): + self.assertEqual(_parse_time_signature("7/8"), (7, 8)) + + def test_none_input(self): + self.assertIsNone(_parse_time_signature(None)) + + def test_malformed_returns_none(self): + self.assertIsNone(_parse_time_signature("not-a-sig")) + + def test_non_integer_returns_none(self): + self.assertIsNone(_parse_time_signature("4/four")) + + def test_too_many_parts(self): + self.assertIsNone(_parse_time_signature("4/4/4")) + + +class BuildScoreTests(unittest.TestCase): + def test_empty_notes_produces_empty_track(self): + score = build_score( + _mock_transcription([]), bpm=120.0, time_signature="4/4" + ) + self.assertEqual(len(score.tracks), 1) + self.assertEqual(len(score.tracks[0].notes), 0) + + def test_tempo_emitted_when_bpm_positive(self): + score = build_score( + _mock_transcription([]), bpm=128.0, time_signature="4/4" + ) + self.assertEqual(len(score.tempos), 1) + self.assertAlmostEqual(score.tempos[0].qpm, 128.0) + + def test_no_tempo_when_bpm_missing(self): + score = build_score( + _mock_transcription([]), bpm=None, time_signature="4/4" + ) + self.assertEqual(len(score.tempos), 0) + + def test_no_tempo_when_bpm_zero(self): + score = build_score( + _mock_transcription([]), bpm=0.0, time_signature="4/4" + ) + self.assertEqual(len(score.tempos), 0) + + def test_time_signature_emitted_when_present(self): + score = build_score( + _mock_transcription([]), bpm=120.0, time_signature="3/4" + ) + self.assertEqual(len(score.time_signatures), 1) + self.assertEqual(score.time_signatures[0].numerator, 3) + self.assertEqual(score.time_signatures[0].denominator, 4) + + def test_no_time_signature_when_malformed(self): + score = build_score( + _mock_transcription([]), bpm=120.0, time_signature="garbage" + ) + self.assertEqual(len(score.time_signatures), 0) + + def test_notes_carried_through(self): + notes_in = [ + _make_note(60, 0.0, 0.5), + _make_note(64, 0.5, 0.5, confidence=1.0), + ] + score = build_score( + _mock_transcription(notes_in), bpm=120.0, time_signature="4/4" + ) + track_notes = sorted(score.tracks[0].notes, key=lambda n: n.time) + self.assertEqual(len(track_notes), 2) + self.assertEqual(track_notes[0].pitch, 60) + self.assertEqual(track_notes[1].pitch, 64) + self.assertEqual(track_notes[1].velocity, 127) + + def test_malformed_note_skipped(self): + notes_in = [ + _make_note(60, 0.0, 0.5), + { + "pitchMidi": "bad", + "onsetSeconds": 1.0, + "durationSeconds": 0.5, + }, + _make_note(64, 1.5, 0.5), + ] + score = build_score( + _mock_transcription(notes_in), bpm=120.0, time_signature="4/4" + ) + self.assertEqual(len(score.tracks[0].notes), 2) + + def test_zero_or_negative_duration_skipped(self): + notes_in = [ + _make_note(60, 0.0, 0.0), + _make_note(62, 0.5, -0.1), + _make_note(64, 1.0, 0.5), + ] + score = build_score( + _mock_transcription(notes_in), bpm=120.0, time_signature="4/4" + ) + self.assertEqual(len(score.tracks[0].notes), 1) + self.assertEqual(score.tracks[0].notes[0].pitch, 64) + + def test_out_of_range_pitch_skipped(self): + notes_in = [ + _make_note(60, 0.0, 0.5), + _make_note(128, 1.0, 0.5), # 128 is invalid + _make_note(-1, 2.0, 0.5), + ] + score = build_score( + _mock_transcription(notes_in), bpm=120.0, time_signature="4/4" + ) + self.assertEqual(len(score.tracks[0].notes), 1) + + def test_non_dict_transcription_is_safe(self): + score = build_score(None, bpm=120.0, time_signature="4/4") # type: ignore[arg-type] + self.assertEqual(len(score.tracks), 1) + self.assertEqual(len(score.tracks[0].notes), 0) + + +class RenderPianorollTests(unittest.TestCase): + def test_empty_transcription_returns_zero_matrix(self): + payload = render_pianoroll(_mock_transcription([]), bpm=120.0) + self.assertEqual(payload.note_count, 0) + self.assertEqual(payload.frames.dtype, np.uint8) + self.assertEqual(int(payload.frames.sum()), 0) + + def test_single_note_lands_in_correct_pitch_row(self): + notes_in = [_make_note(60, 0.0, 0.5, confidence=1.0)] + payload = render_pianoroll(_mock_transcription(notes_in), bpm=120.0) + target_row_index = 60 - DEFAULT_PITCH_LOW + target_row = payload.frames[target_row_index] + self.assertGreater(int(target_row.sum()), 0) + # No spill into adjacent pitch rows. + non_target = np.delete(payload.frames, target_row_index, axis=0) + self.assertEqual(int(non_target.sum()), 0) + + def test_pitch_range_validation_rejects_inverted(self): + with self.assertRaises(ValueError): + render_pianoroll( + _mock_transcription([]), + bpm=120.0, + pitch_low=50, + pitch_high=50, + ) + + def test_pitch_range_validation_rejects_negative(self): + with self.assertRaises(ValueError): + render_pianoroll( + _mock_transcription([]), + bpm=120.0, + pitch_low=-1, + pitch_high=10, + ) + + def test_pitch_range_validation_rejects_above_128(self): + with self.assertRaises(ValueError): + render_pianoroll( + _mock_transcription([]), + bpm=120.0, + pitch_low=0, + pitch_high=200, + ) + + def test_tpq_validation(self): + with self.assertRaises(ValueError): + render_pianoroll(_mock_transcription([]), bpm=120.0, tpq=0) + + def test_mode_validation(self): + with self.assertRaises(ValueError): + render_pianoroll( + _mock_transcription([]), + bpm=120.0, + mode="bogus", # type: ignore[arg-type] + ) + + def test_onset_mode_is_sparser_than_frame_mode(self): + notes_in = [_make_note(60, 0.0, 1.0, confidence=1.0)] + frame_payload = render_pianoroll( + _mock_transcription(notes_in), bpm=120.0, mode="frame" + ) + onset_payload = render_pianoroll( + _mock_transcription(notes_in), bpm=120.0, mode="onset" + ) + row = 60 - DEFAULT_PITCH_LOW + # Frame mode paints the sustain; onset mode marks only the note start. + self.assertGreater( + int(frame_payload.frames[row].sum()), + int(onset_payload.frames[row].sum()), + ) + + def test_velocity_floor_visible_for_zero_confidence(self): + notes_in = [_make_note(60, 0.0, 0.5, confidence=0.0)] + payload = render_pianoroll(_mock_transcription(notes_in), bpm=120.0) + row = payload.frames[60 - DEFAULT_PITCH_LOW] + non_zero = row[row > 0] + self.assertGreater(len(non_zero), 0) + self.assertEqual(int(non_zero.max()), VELOCITY_FLOOR) + + def test_metadata_passes_through(self): + payload = render_pianoroll( + _mock_transcription([]), + bpm=128.0, + time_signature="3/4", + ) + self.assertEqual(payload.quarters_per_minute, 128.0) + self.assertEqual(payload.time_signature, (3, 4)) + self.assertEqual(payload.mode, "frame") + + def test_works_without_bpm(self): + notes_in = [_make_note(60, 0.0, 0.5)] + payload = render_pianoroll(_mock_transcription(notes_in), bpm=None) + self.assertIsNone(payload.quarters_per_minute) + # Pianoroll still renders — symusic falls back to qpm=120 internally. + self.assertGreater(int(payload.frames.sum()), 0) + + +class PayloadJsonDictTests(unittest.TestCase): + def test_round_trip_has_expected_keys(self): + notes_in = [_make_note(60, 0.0, 0.5)] + payload = render_pianoroll( + _mock_transcription(notes_in), bpm=128.0, time_signature="4/4" + ) + out = payload_to_json_dict(payload) + self.assertEqual( + set(out.keys()), + { + "mode", + "pitchLow", + "pitchHigh", + "ticksPerQuarter", + "quartersPerMinute", + "timeSignature", + "noteCount", + "frames", + }, + ) + self.assertEqual(out["mode"], "frame") + self.assertEqual(out["timeSignature"], "4/4") + self.assertAlmostEqual(out["quartersPerMinute"], 128.0) + self.assertEqual(out["noteCount"], 1) + self.assertIsInstance(out["frames"], list) + self.assertEqual(len(out["frames"]), payload.frames.shape[0]) + # Each row is a list of ints (uint8 -> Python int via tolist()). + self.assertTrue( + all(isinstance(cell, int) for cell in out["frames"][0]) + ) + + def test_missing_bpm_serializes_none(self): + payload = render_pianoroll(_mock_transcription([]), bpm=None) + out = payload_to_json_dict(payload) + self.assertIsNone(out["quartersPerMinute"]) + + def test_missing_time_signature_serializes_none(self): + payload = render_pianoroll(_mock_transcription([]), bpm=120.0) + out = payload_to_json_dict(payload) + self.assertIsNone(out["timeSignature"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/backend/transcription_pianoroll.py b/apps/backend/transcription_pianoroll.py new file mode 100644 index 00000000..ecf83655 --- /dev/null +++ b/apps/backend/transcription_pianoroll.py @@ -0,0 +1,254 @@ +"""Convert ``transcriptionDetail.notes`` to a velocity-encoded pianoroll matrix. + +Derived view — Phase 1's ``transcriptionDetail`` is the authority and is never +overridden here (invariant #1 from ``PURPOSE.md``). This module re-renders that +measurement as a 2D heatmap-ready ``uint8`` matrix using ``symusic`` for the +rasterization. The pianoroll cites the same notes and Phase 1 ``bpm`` / +``timeSignature`` the HTTP envelope already carries; chain of custody is +preserved (invariant #2). + +Public surface: + +1. :func:`render_pianoroll` — top-level entry point. Takes a + ``transcriptionDetail`` dict + Phase 1 tempo/meter, returns a + :class:`PianorollPayload` (mode, pitch axis bounds, frames matrix, metadata). +2. :func:`payload_to_json_dict` — JSON-friendly projection of the payload for the + HTTP envelope. ``frames`` becomes a nested list. At ``tpq=4`` typical clips + stay well under a megabyte, but larger matrices should be persisted through + ``artifact_storage`` instead of inlining. +3. :func:`build_score` — exposed for tests and any future Score-level consumer + (MIDI download, etc.). Tempo / time-signature events are emitted at ``t=0`` + to match Phase 1's static (per-track) analysis. + +Shape contract (per the symusic tutorial): + +- ``Score.pianoroll(modes=[m], pitch_range=[low, high], encode_velocity=True)`` + returns ``(num_modes, num_tracks, high - low, time_steps)`` uint8. +- We render a single mode + single track, so the result is sliced to + ``(pitch_count, time_steps)`` before being handed back. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal + +import numpy as np +from symusic import Note, Score, Tempo, TimeSignature, Track + +PianorollMode = Literal["frame", "onset"] + +# Default to the 88-key piano range (A0 = 21 inclusive, C8 = 108 inclusive ⇒ +# upper bound 109 exclusive). Phase 1 transcription rarely emits notes outside +# this range; clamping keeps the matrix compact for transport. +DEFAULT_PITCH_LOW = 21 +DEFAULT_PITCH_HIGH = 109 + +# tpq=4 is one row per 16th note at typical tempos. A 30 s clip at 120 BPM at +# tpq=4 fits in ~88 × 240 cells ≈ 21 KB uncompressed — small enough to inline. +# Lower values risk losing onsets close together; higher values blow up the +# matrix for marginal perceptual gain. +DEFAULT_TPQ = 4 + +# Internal score resolution before the resample. 480 tpq is the General MIDI +# convention and is more than enough precision for note times that arrive in +# seconds from the transcription backend. +SCORE_TPQ = 480 + +# Confidence ∈ [0, 1] is mapped onto MIDI velocity ∈ [VELOCITY_FLOOR, 127] so a +# zero-confidence note is still visible in the heatmap (otherwise it would be +# indistinguishable from a silent cell), and a maximally confident note maxes +# out at 127. The floor is also what callers can assert against in tests. +VELOCITY_FLOOR = 64 +VELOCITY_RANGE = 127 - VELOCITY_FLOOR + + +@dataclass(frozen=True) +class PianorollPayload: + """A flat 2D ``(pitch, time)`` velocity matrix + the axis metadata.""" + + mode: PianorollMode + pitch_low: int + pitch_high: int + ticks_per_quarter: int + quarters_per_minute: float | None + time_signature: tuple[int, int] | None + note_count: int + frames: np.ndarray # shape: (pitch_high - pitch_low, time_steps), dtype=uint8 + + +def _parse_time_signature(value: str | None) -> tuple[int, int] | None: + """Parse a ``"4/4"`` / ``"3/4"`` string. Returns ``None`` on anything weird. + + Phase 1 emits ``timeSignature`` as a string like ``"4/4"`` (currently the + assumed-4/4 fallback per ``JSON_SCHEMA.md``); we tolerate ``None`` and + malformed inputs so this module remains useful when meter is uncertain. + """ + if not isinstance(value, str): + return None + parts = value.strip().split("/") + if len(parts) != 2: + return None + try: + return int(parts[0]), int(parts[1]) + except ValueError: + return None + + +def _velocity_from_confidence(confidence: object) -> int: + """Map ``[0, 1]`` confidence onto velocity ``[VELOCITY_FLOOR, 127]``. + + Out-of-range / non-numeric values clamp to the endpoints rather than raise + — Phase 1 confidence floors at ``TRANSCRIPTION_CONFIDENCE_FLOOR = 0.05`` so + the typical input is well-behaved, but a defensive clamp here means a + surprise upstream emits a still-visible heatmap row. + """ + try: + value = float(confidence) # type: ignore[arg-type] + except (TypeError, ValueError): + value = 0.0 + clipped = max(0.0, min(1.0, value)) + return int(round(VELOCITY_FLOOR + VELOCITY_RANGE * clipped)) + + +def build_score( + transcription_detail: dict, + *, + bpm: float | None, + time_signature: str | None, +) -> Score: + """Build a ``symusic.Score`` (seconds time-unit) from a transcription dict. + + The Score carries Phase 1 tempo and meter at ``t=0`` so downstream consumers + (e.g. a ``.mid`` download) inherit the citation. Notes that fail validation + (missing keys, non-numeric times, out-of-range pitches, zero/negative + duration) are silently skipped — the matrix is a derived view, so emitting + an obviously corrupt cell is worse than dropping the note. + """ + score = Score(SCORE_TPQ, ttype="Second") + + if bpm is not None and float(bpm) > 0: + score.tempos.append(Tempo(time=0.0, qpm=float(bpm), ttype="Second")) + + parsed_signature = _parse_time_signature(time_signature) + if parsed_signature is not None: + numerator, denominator = parsed_signature + score.time_signatures.append( + TimeSignature( + time=0.0, + numerator=numerator, + denominator=denominator, + ttype="Second", + ) + ) + + track = Track(name="transcription", program=0, ttype="Second") + notes = ( + transcription_detail.get("notes") + if isinstance(transcription_detail, dict) + else None + ) + if isinstance(notes, list): + for note in notes: + if not isinstance(note, dict): + continue + try: + pitch = int(note["pitchMidi"]) + onset = float(note["onsetSeconds"]) + duration = float(note["durationSeconds"]) + except (KeyError, TypeError, ValueError): + continue + if duration <= 0 or pitch < 0 or pitch > 127 or onset < 0: + continue + velocity = _velocity_from_confidence(note.get("confidence", 0.0)) + track.notes.append( + Note( + time=onset, + duration=duration, + pitch=pitch, + velocity=velocity, + ttype="Second", + ) + ) + score.tracks.append(track) + return score + + +def render_pianoroll( + transcription_detail: dict, + *, + bpm: float | None, + time_signature: str | None = None, + mode: PianorollMode = "frame", + pitch_low: int = DEFAULT_PITCH_LOW, + pitch_high: int = DEFAULT_PITCH_HIGH, + tpq: int = DEFAULT_TPQ, +) -> PianorollPayload: + """Render transcription notes as a ``(pitch, time)`` uint8 matrix. + + ``mode`` is one of ``"frame"`` (sustains visible) or ``"onset"`` (note + starts only — useful for rhythmic visualization). ``pitch_low`` is + inclusive, ``pitch_high`` is exclusive. + + Raises ``ValueError`` on out-of-band axis arguments rather than silently + clamping — the route layer should validate query params before reaching + here, and an invalid value is almost always a caller bug. + """ + if pitch_low < 0 or pitch_high > 128 or pitch_low >= pitch_high: + raise ValueError( + f"invalid pitch range [{pitch_low}, {pitch_high}); " + "must satisfy 0 <= pitch_low < pitch_high <= 128" + ) + if tpq < 1: + raise ValueError(f"tpq must be >= 1; got {tpq}") + if mode not in ("frame", "onset"): + raise ValueError(f"mode must be 'frame' or 'onset'; got {mode!r}") + + score = build_score( + transcription_detail, bpm=bpm, time_signature=time_signature + ) + resampled = score.resample(tpq=tpq, min_dur=1) + matrix = resampled.pianoroll( + modes=[mode], + pitch_range=[pitch_low, pitch_high], + encode_velocity=True, + ) + # Slice to (pitch, time): single mode + single track in this module. + frames = np.ascontiguousarray(matrix[0, 0], dtype=np.uint8) + + return PianorollPayload( + mode=mode, + pitch_low=pitch_low, + pitch_high=pitch_high, + ticks_per_quarter=tpq, + quarters_per_minute=( + float(bpm) if bpm is not None and float(bpm) > 0 else None + ), + time_signature=_parse_time_signature(time_signature), + note_count=len(score.tracks[0].notes) if score.tracks else 0, + frames=frames, + ) + + +def payload_to_json_dict(payload: PianorollPayload) -> dict: + """Project the payload into a JSON-serializable dict for the HTTP envelope. + + ``frames`` becomes a nested list of ints (rows are pitches, columns are + time steps). Keep the field names camelCase to match the rest of the + ``phase1`` envelope — there is no conversion layer (see ``CLAUDE.md`` + tripwire #3). + """ + return { + "mode": payload.mode, + "pitchLow": payload.pitch_low, + "pitchHigh": payload.pitch_high, + "ticksPerQuarter": payload.ticks_per_quarter, + "quartersPerMinute": payload.quarters_per_minute, + "timeSignature": ( + None + if payload.time_signature is None + else f"{payload.time_signature[0]}/{payload.time_signature[1]}" + ), + "noteCount": payload.note_count, + "frames": payload.frames.tolist(), + }