From 92f43c0b4745293624482816b5f5cc585b37e798 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 13 May 2026 06:37:43 +0000 Subject: [PATCH 1/2] =?UTF-8?q?feat(export):=20Track=202=20=E2=80=94=20CSV?= =?UTF-8?q?=20export=20of=20Phase=201=20time-series=20fields=20+=20schema?= =?UTF-8?q?=20ADR=20v1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements Track 2 from the external-repo incorporation review (docs/external-repo-review-2026-05-13.md). Patterned on Partiels' (time, duration, value) tabular shape, but per-ASA-field with the domain names preserved — does not flatten ASA's schema into a Partiels-style flat shape (the review rejected that as a chain-of- custody regression). What's new: - apps/backend/csv_export.py — allowlisted registry of four exportable time-series fields with per-field serializers and a small dot-path resolver (not JSONPath). 200 lines, no external deps beyond stdlib. Public surface: export_field_to_csv, is_supported_field, list_supported_fields. - apps/backend/server.py — new route GET /api/analysis-runs/{run_id}/export/csv/{field_path} Three 404 error codes (RUN_NOT_FOUND, EXPORT_FIELD_NOT_SUPPORTED, EXPORT_FIELD_NOT_AVAILABLE) for the three distinguishable failure modes. Sets Content-Disposition: attachment with a sensible filename. Auth model matches the existing /api/analysis-runs/.../artifacts route. - apps/backend/tests/test_csv_export.py — 22 unit tests covering the registry, the dot-path resolver, and each serializer's column layout and edge cases (empty arrays, missing keys, partial rows). All pass locally. - apps/backend/tests/test_server.py — 4 new tests in CsvExportRouteTests covering: success path with content-disposition + body shape; unsupported field path returns EXPORT_FIELD_NOT_SUPPORTED with the supported list in the message; supported-but-null field returns EXPORT_FIELD_NOT_AVAILABLE; unknown run returns RUN_NOT_FOUND. - docs/adr/0001-phase1-json-schema-v1.md — ratifies the current Phase 1 shape as v1, defines the additive-only compatibility policy, and pins the four exportable time-series field shapes. The schema is the union of EXPECTED_TOP_LEVEL_KEYS, JSON_SCHEMA.md, and Phase1Result — no separate machine-readable JSON Schema artifact (deferred to v2 if a consumer ever needs one). - apps/backend/ARCHITECTURE.md — updated route list to include the new endpoint plus three pre-existing routes that were missing from the doc (artifacts, spectral-enhancements, pitch-note-translations, interpretations). Exportable fields in v1: lufsCurve.shortTerm → time, duration=3.0, lufs lufsCurve.momentary → time, duration=0.4, lufs rhythmDetail.tempoCurve → time, bpm spectralBalanceTimeSeries → time, subBass, lowBass, lowMids, mids, upperMids, highs, brilliance Out of scope (deferred, documented in the review): - SDIF / LAB / REAPER export formats - Beat/downbeat tabular export (the existing fields are sample indices, not seconds; adding this requires a frame-rate threading decision) - CLI export tool https://claude.ai/code/session_01Uh2eNT1oqjixcox8UX3BTg --- apps/backend/ARCHITECTURE.md | 5 + apps/backend/csv_export.py | 234 +++++++++++++++++++++ apps/backend/server.py | 88 +++++++- apps/backend/tests/test_csv_export.py | 268 +++++++++++++++++++++++++ apps/backend/tests/test_server.py | 133 ++++++++++++ docs/adr/0001-phase1-json-schema-v1.md | 89 ++++++++ 6 files changed, 816 insertions(+), 1 deletion(-) create mode 100644 apps/backend/csv_export.py create mode 100644 apps/backend/tests/test_csv_export.py create mode 100644 docs/adr/0001-phase1-json-schema-v1.md diff --git a/apps/backend/ARCHITECTURE.md b/apps/backend/ARCHITECTURE.md index 40300b9d..e17fab66 100644 --- a/apps/backend/ARCHITECTURE.md +++ b/apps/backend/ARCHITECTURE.md @@ -71,6 +71,11 @@ Custom routes: - `POST /api/analysis-runs` - `GET /api/analysis-runs/{run_id}` - `DELETE /api/analysis-runs/{run_id}` +- `GET /api/analysis-runs/{run_id}/artifacts` and `…/artifacts/{artifact_id}` +- `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). +- `POST /api/analysis-runs/{run_id}/spectral-enhancements/{kind}` +- `POST /api/analysis-runs/{run_id}/pitch-note-translations` +- `POST /api/analysis-runs/{run_id}/interpretations` - `POST /api/analyze` (legacy compatibility) - `POST /api/analyze/estimate` (legacy compatibility) - `POST /api/phase2` (legacy compatibility) diff --git a/apps/backend/csv_export.py b/apps/backend/csv_export.py new file mode 100644 index 00000000..196f87da --- /dev/null +++ b/apps/backend/csv_export.py @@ -0,0 +1,234 @@ +"""CSV exporters for Phase 1 time-series fields. + +Patterned on the ``(time, duration, value)`` tabular shape used by Partiels +and similar Vamp-output tools (see ``docs/external-repo-review-2026-05-13.md`` +Track 2 for the rationale and the rejected-alternatives discussion). Each +exporter maps one dotted JSON path inside the Phase 1 measurement payload to +a small CSV with documented columns. + +The exporters are kept here, outside ``server.py``, so the route handler can +stay a thin lookup-and-serve. + +Registered paths (v1): + +================================== ===================================== +Path CSV columns +================================== ===================================== +``lufsCurve.shortTerm`` ``time, duration, lufs`` (duration=3.0) +``lufsCurve.momentary`` ``time, duration, lufs`` (duration=0.4) +``rhythmDetail.tempoCurve`` ``time, bpm`` +``spectralBalanceTimeSeries`` ``time, subBass, lowBass, lowMids,`` + ``mids, upperMids, highs, brilliance`` +================================== ===================================== + +Design notes: + +- The ``duration`` column on the LUFS curves reflects EBU R128's measurement + window for that point (3.0 s short-term, 0.4 s momentary). Other curves + do not have an intrinsic per-point duration and so omit the column. +- A field that is ``None`` or missing from the measurement payload returns + ``None`` from :func:`export_field_to_csv`. The HTTP layer treats that + separately from "unknown field" so users get a useful error code. +- Field paths are deliberately allowlisted: arbitrary JSONPath/nested-key + descent on the measurement payload is not supported. Adding a new + exportable field is a small, deliberate code change in this file. +""" + +from __future__ import annotations + +import csv +import io +from typing import Any, Callable + + +__all__ = [ + "export_field_to_csv", + "is_supported_field", + "list_supported_fields", +] + + +# EBU R128 short-term loudness window is 3.0 s; momentary is 0.4 s. +# Each curve point reports the integrated loudness over that window +# ending at ``t``. +_LUFS_SHORT_TERM_WINDOW_S = 3.0 +_LUFS_MOMENTARY_WINDOW_S = 0.4 + + +_SPECTRAL_BANDS: tuple[str, ...] = ( + "subBass", + "lowBass", + "lowMids", + "mids", + "upperMids", + "highs", + "brilliance", +) + + +# ---------------------------------------------------------------------- +# Public API +# ---------------------------------------------------------------------- + + +def export_field_to_csv( + measurement_result: dict | None, + field_path: str, +) -> str | None: + """Look up ``field_path`` in the registry and serialize the value to CSV. + + Returns the CSV text (header row + data rows) when the field is present + and non-empty. Returns ``None`` when any of these are true: + + - ``measurement_result`` itself is ``None`` + - the field path is not registered (caller should treat as 404 "unknown") + - the field is missing or ``None`` in the measurement payload (caller + should treat as 404 "not available for this run") + - the field is present but empty (zero data points; same as above) + + The caller is responsible for distinguishing "unknown path" from + "known path, no data" — see :func:`is_supported_field`. + """ + if measurement_result is None: + return None + if field_path not in _EXPORTERS: + return None + field_value = _resolve_dot_path(measurement_result, field_path) + if field_value is None: + return None + return _EXPORTERS[field_path](field_value) + + +def is_supported_field(field_path: str) -> bool: + """Whether ``field_path`` is in the export registry. + + Independent of whether any particular run has data for it. + """ + return field_path in _EXPORTERS + + +def list_supported_fields() -> list[str]: + """All currently exportable field paths, sorted alphabetically.""" + return sorted(_EXPORTERS.keys()) + + +# ---------------------------------------------------------------------- +# Path resolution +# ---------------------------------------------------------------------- + + +def _resolve_dot_path(payload: dict, dot_path: str) -> Any: + """Walk ``payload`` along ``a.b.c`` and return the leaf, or ``None``. + + Returns ``None`` on any of: non-dict at an intermediate step, missing + key at any step, or explicit ``None`` value at any step. This is a + simple nested-key descent — *not* JSONPath syntax (no ``$``, no + ``[*]``, no array indexing). + """ + node: Any = payload + for key in dot_path.split("."): + if not isinstance(node, dict): + return None + node = node.get(key) + if node is None: + return None + return node + + +# ---------------------------------------------------------------------- +# Per-field serializers +# ---------------------------------------------------------------------- + + +def _serialize_lufs_curve( + points: Any, window_duration_s: float +) -> str | None: + """Serialize a list of ``{t, lufs}`` points with a constant duration column.""" + if not isinstance(points, list) or len(points) == 0: + return None + buf = io.StringIO() + writer = csv.writer(buf, lineterminator="\n") + writer.writerow(["time", "duration", "lufs"]) + wrote_any = False + for point in points: + if not isinstance(point, dict): + continue + t = point.get("t") + lufs = point.get("lufs") + if t is None or lufs is None: + continue + writer.writerow( + [f"{float(t):.6f}", f"{window_duration_s}", f"{float(lufs):.2f}"] + ) + wrote_any = True + return buf.getvalue() if wrote_any else None + + +def _serialize_tempo_curve(points: Any) -> str | None: + """Serialize a list of ``{t, bpm}`` points.""" + if not isinstance(points, list) or len(points) == 0: + return None + buf = io.StringIO() + writer = csv.writer(buf, lineterminator="\n") + writer.writerow(["time", "bpm"]) + wrote_any = False + for point in points: + if not isinstance(point, dict): + continue + t = point.get("t") + bpm = point.get("bpm") + if t is None or bpm is None: + continue + writer.writerow([f"{float(t):.6f}", f"{float(bpm):.3f}"]) + wrote_any = True + return buf.getvalue() if wrote_any else None + + +def _serialize_spectral_balance_time_series(points: Any) -> str | None: + """Serialize a list of ``{t, subBass, ..., brilliance}`` points. + + Skips any row that is missing one of the band values rather than + writing a partial row — keeps the CSV regular and downstream-loadable. + """ + if not isinstance(points, list) or len(points) == 0: + return None + buf = io.StringIO() + writer = csv.writer(buf, lineterminator="\n") + writer.writerow(["time", *_SPECTRAL_BANDS]) + wrote_any = False + for point in points: + if not isinstance(point, dict): + continue + t = point.get("t") + if t is None: + continue + band_values: list[str] = [] + skip = False + for band in _SPECTRAL_BANDS: + value = point.get(band) + if value is None: + skip = True + break + band_values.append(f"{float(value):.4f}") + if skip: + continue + writer.writerow([f"{float(t):.6f}", *band_values]) + wrote_any = True + return buf.getvalue() if wrote_any else None + + +# ---------------------------------------------------------------------- +# Registry +# ---------------------------------------------------------------------- + + +_EXPORTERS: dict[str, Callable[[Any], str | None]] = { + "lufsCurve.shortTerm": lambda value: _serialize_lufs_curve( + value, _LUFS_SHORT_TERM_WINDOW_S + ), + "lufsCurve.momentary": lambda value: _serialize_lufs_curve( + value, _LUFS_MOMENTARY_WINDOW_S + ), + "rhythmDetail.tempoCurve": _serialize_tempo_curve, + "spectralBalanceTimeSeries": _serialize_spectral_balance_time_series, +} diff --git a/apps/backend/server.py b/apps/backend/server.py index 30d788b4..2e46cd39 100644 --- a/apps/backend/server.py +++ b/apps/backend/server.py @@ -27,7 +27,7 @@ from fastapi import FastAPI, File, Form, Header, Query, UploadFile from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import FileResponse, JSONResponse +from fastapi.responses import FileResponse, JSONResponse, Response from auth_context import AuthenticationRequiredError, UserContext, resolve_api_user_context from analysis_runtime import ( @@ -46,6 +46,7 @@ should_start_in_process_workers, ) from utils.cleanup import cleanup_artifacts +import csv_export import upload_limits from server_upload import ( # noqa: F401 — re-exported for test backward compat LEGACY_ENDPOINT_SUNSET, @@ -2088,6 +2089,91 @@ async def get_run_artifact( ) +@app.get( + "/api/analysis-runs/{run_id}/export/csv/{field_path}", + response_model=None, +) +async def export_run_field_as_csv( + run_id: str, + field_path: str, + x_asa_user_id: str | None = Header(None), + x_asa_user_email: str | None = Header(None), +) -> Response | JSONResponse: + """Export one Phase 1 time-series field as CSV. + + The ``field_path`` is a simple dot-path into the measurement payload — + e.g. ``lufsCurve.shortTerm`` or ``rhythmDetail.tempoCurve``. Not + JSONPath; arbitrary nested-key descent is not supported. See + :func:`csv_export.list_supported_fields` for the registered list. + + Returns: + - 200 ``text/csv`` with header row + data rows on success. + - 404 ``EXPORT_FIELD_NOT_SUPPORTED`` if the path is not in the + registry. + - 404 ``RUN_NOT_FOUND`` if the run does not exist or is not + owned by the requesting user. + - 404 ``EXPORT_FIELD_NOT_AVAILABLE`` if the field is registered + but missing/null/empty for this run (measurement may be + incomplete or the analyzer skipped that field for this audio). + """ + user_context = _resolve_route_user_context(x_asa_user_id, x_asa_user_email) + if isinstance(user_context, JSONResponse): + return user_context + + if not csv_export.is_supported_field(field_path): + supported = ", ".join(csv_export.list_supported_fields()) + return JSONResponse( + status_code=404, + content={ + "error": { + "code": "EXPORT_FIELD_NOT_SUPPORTED", + "message": ( + f"Field path '{field_path}' is not exportable. " + f"Supported paths: {supported}." + ), + } + }, + ) + + 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) + + measurement_result = ( + snapshot.get("stages", {}) + .get("measurement", {}) + .get("result") + ) + csv_text = csv_export.export_field_to_csv(measurement_result, field_path) + if csv_text is None: + return JSONResponse( + status_code=404, + content={ + "error": { + "code": "EXPORT_FIELD_NOT_AVAILABLE", + "message": ( + f"Field '{field_path}' is supported but not " + f"populated in run '{run_id}'. The measurement " + f"may be incomplete or the analyzer skipped this " + f"field for this audio." + ), + } + }, + ) + + # Filename: _.csv + filename = f"{run_id}_{field_path.replace('.', '_')}.csv" + return Response( + content=csv_text, + media_type="text/csv; charset=utf-8", + headers={ + "Content-Disposition": f'attachment; filename="{filename}"', + }, + ) + + _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_csv_export.py b/apps/backend/tests/test_csv_export.py new file mode 100644 index 00000000..84ee7d54 --- /dev/null +++ b/apps/backend/tests/test_csv_export.py @@ -0,0 +1,268 @@ +"""Unit tests for csv_export serializers and the dot-path resolver. + +The route tests in test_server.py exercise the HTTP layer end-to-end; +these tests focus on the pure-logic surface of csv_export.py: given a +specific input shape for a specific field, does the CSV come out right? + +The shapes we feed in here are the same shapes that +``analysis_runtime.get_run(...)["stages"]["measurement"]["result"]`` +produces. If the analyzer ever drops a field or changes the per-point +key names, these tests fail loudly — that's the early-warning we want. +""" + +import importlib.util +import sys +import unittest +from pathlib import Path + + +_BACKEND_ROOT = Path(__file__).resolve().parents[1] +if str(_BACKEND_ROOT) not in sys.path: + sys.path.insert(0, str(_BACKEND_ROOT)) + + +_SPEC = importlib.util.spec_from_file_location( + "csv_export_test", _BACKEND_ROOT / "csv_export.py" +) +if _SPEC is None or _SPEC.loader is None: + raise AssertionError("Could not load csv_export.py") +csv_export = importlib.util.module_from_spec(_SPEC) +_SPEC.loader.exec_module(csv_export) + + +class RegistryTests(unittest.TestCase): + """The registry is the public contract for which fields are exportable.""" + + def test_supported_fields_v1(self): + self.assertEqual( + csv_export.list_supported_fields(), + [ + "lufsCurve.momentary", + "lufsCurve.shortTerm", + "rhythmDetail.tempoCurve", + "spectralBalanceTimeSeries", + ], + ) + + def test_is_supported_field_membership(self): + self.assertTrue(csv_export.is_supported_field("lufsCurve.shortTerm")) + self.assertFalse(csv_export.is_supported_field("bpm")) + self.assertFalse(csv_export.is_supported_field("lufsCurve")) # not a leaf + self.assertFalse(csv_export.is_supported_field("")) + self.assertFalse(csv_export.is_supported_field("nope.nope.nope")) + + +class DotPathResolverTests(unittest.TestCase): + """Verifies the simple nested-key descent — not JSONPath.""" + + def test_single_key(self): + self.assertEqual( + csv_export._resolve_dot_path({"bpm": 128}, "bpm"), 128 + ) + + def test_nested_keys(self): + payload = {"a": {"b": {"c": 42}}} + self.assertEqual(csv_export._resolve_dot_path(payload, "a.b.c"), 42) + + def test_missing_intermediate_returns_none(self): + self.assertIsNone( + csv_export._resolve_dot_path({"a": {"b": 1}}, "a.c.d") + ) + + def test_none_intermediate_returns_none(self): + self.assertIsNone( + csv_export._resolve_dot_path({"a": None}, "a.b") + ) + + def test_non_dict_intermediate_returns_none(self): + self.assertIsNone( + csv_export._resolve_dot_path({"a": [1, 2]}, "a.b") + ) + + +class ExportFieldToCsvTopLevelTests(unittest.TestCase): + """Top-level routing: unknown fields and null inputs all return None.""" + + def test_unknown_field_returns_none(self): + result = csv_export.export_field_to_csv({}, "nope") + self.assertIsNone(result) + + def test_null_measurement_returns_none(self): + result = csv_export.export_field_to_csv(None, "lufsCurve.shortTerm") + self.assertIsNone(result) + + def test_missing_field_returns_none(self): + result = csv_export.export_field_to_csv({}, "lufsCurve.shortTerm") + self.assertIsNone(result) + + def test_empty_array_returns_none(self): + payload = {"lufsCurve": {"shortTerm": []}} + result = csv_export.export_field_to_csv(payload, "lufsCurve.shortTerm") + self.assertIsNone(result) + + +class LufsCurveShortTermTests(unittest.TestCase): + """lufsCurve.shortTerm → time,duration,lufs with duration=3.0 constant.""" + + def test_basic_three_points(self): + payload = { + "lufsCurve": { + "shortTerm": [ + {"t": 0.0, "lufs": -23.0}, + {"t": 0.1, "lufs": -22.5}, + {"t": 0.2, "lufs": -22.0}, + ] + } + } + result = csv_export.export_field_to_csv(payload, "lufsCurve.shortTerm") + self.assertIsNotNone(result) + lines = result.strip().split("\n") + self.assertEqual(lines[0], "time,duration,lufs") + self.assertEqual(len(lines), 4) # header + 3 rows + # First data row — exact format + self.assertEqual(lines[1], "0.000000,3.0,-23.00") + + def test_duration_column_is_3_0_for_short_term(self): + payload = { + "lufsCurve": { + "shortTerm": [{"t": 1.5, "lufs": -20.0}], + } + } + result = csv_export.export_field_to_csv(payload, "lufsCurve.shortTerm") + lines = result.strip().split("\n") + # Column order: time,duration,lufs — duration is column 1 + cells = lines[1].split(",") + self.assertEqual(cells[1], "3.0") + + def test_skips_points_with_missing_keys(self): + payload = { + "lufsCurve": { + "shortTerm": [ + {"t": 0.0, "lufs": -23.0}, + {"t": 0.1}, # missing lufs — skipped + {"lufs": -22.0}, # missing t — skipped + {"t": 0.3, "lufs": -22.5}, + ] + } + } + result = csv_export.export_field_to_csv(payload, "lufsCurve.shortTerm") + lines = result.strip().split("\n") + # header + 2 valid rows + self.assertEqual(len(lines), 3) + + def test_all_points_invalid_returns_none(self): + payload = { + "lufsCurve": { + "shortTerm": [ + {"t": 0.0}, # missing lufs + {"lufs": -23.0}, # missing t + ] + } + } + result = csv_export.export_field_to_csv(payload, "lufsCurve.shortTerm") + self.assertIsNone(result) + + +class LufsCurveMomentaryTests(unittest.TestCase): + """lufsCurve.momentary → time,duration,lufs with duration=0.4 constant.""" + + def test_duration_column_is_0_4_for_momentary(self): + payload = { + "lufsCurve": { + "momentary": [{"t": 0.5, "lufs": -18.0}], + } + } + result = csv_export.export_field_to_csv(payload, "lufsCurve.momentary") + lines = result.strip().split("\n") + cells = lines[1].split(",") + self.assertEqual(cells[1], "0.4") + + def test_short_term_and_momentary_are_independent(self): + """A run with only the momentary curve populated must still export it.""" + payload = { + "lufsCurve": { + "shortTerm": None, + "momentary": [{"t": 0.0, "lufs": -25.0}], + } + } + self.assertIsNone( + csv_export.export_field_to_csv(payload, "lufsCurve.shortTerm") + ) + result_m = csv_export.export_field_to_csv(payload, "lufsCurve.momentary") + self.assertIsNotNone(result_m) + self.assertIn("-25.00", result_m) + + +class TempoCurveTests(unittest.TestCase): + """rhythmDetail.tempoCurve → time,bpm (no duration column).""" + + def test_basic_two_points(self): + payload = { + "rhythmDetail": { + "tempoCurve": [ + {"t": 0.0, "bpm": 128.0}, + {"t": 10.0, "bpm": 130.5}, + ] + } + } + result = csv_export.export_field_to_csv(payload, "rhythmDetail.tempoCurve") + lines = result.strip().split("\n") + self.assertEqual(lines[0], "time,bpm") + self.assertEqual(len(lines), 3) + self.assertEqual(lines[1], "0.000000,128.000") + self.assertEqual(lines[2], "10.000000,130.500") + + def test_null_tempo_curve_returns_none(self): + payload = {"rhythmDetail": {"tempoCurve": None}} + result = csv_export.export_field_to_csv(payload, "rhythmDetail.tempoCurve") + self.assertIsNone(result) + + +class SpectralBalanceTimeSeriesTests(unittest.TestCase): + """spectralBalanceTimeSeries → time + 7 band columns.""" + + BANDS = ( + "subBass", + "lowBass", + "lowMids", + "mids", + "upperMids", + "highs", + "brilliance", + ) + + def _point(self, t: float) -> dict: + return {"t": t, **{band: 0.5 for band in self.BANDS}} + + def test_columns_match_band_order(self): + payload = {"spectralBalanceTimeSeries": [self._point(0.0)]} + result = csv_export.export_field_to_csv(payload, "spectralBalanceTimeSeries") + lines = result.strip().split("\n") + self.assertEqual( + lines[0], + "time,subBass,lowBass,lowMids,mids,upperMids,highs,brilliance", + ) + + def test_row_has_eight_cells(self): + payload = {"spectralBalanceTimeSeries": [self._point(1.0)]} + result = csv_export.export_field_to_csv(payload, "spectralBalanceTimeSeries") + lines = result.strip().split("\n") + cells = lines[1].split(",") + self.assertEqual(len(cells), 8) # time + 7 bands + + def test_row_with_missing_band_is_skipped(self): + payload = { + "spectralBalanceTimeSeries": [ + self._point(0.0), + {"t": 1.0, "subBass": 0.1}, # missing 6 bands — skipped + self._point(2.0), + ] + } + result = csv_export.export_field_to_csv(payload, "spectralBalanceTimeSeries") + lines = result.strip().split("\n") + # header + 2 valid rows + self.assertEqual(len(lines), 3) + + +if __name__ == "__main__": + unittest.main() diff --git a/apps/backend/tests/test_server.py b/apps/backend/tests/test_server.py index 5b5f46f5..676bc745 100644 --- a/apps/backend/tests/test_server.py +++ b/apps/backend/tests/test_server.py @@ -4204,5 +4204,138 @@ def test_aliases_scoped_per_device_not_global(self) -> None: self.assertEqual(self._codes(warnings), ["UNKNOWN_PARAMETER"]) +class CsvExportRouteTests(unittest.TestCase): + """Route tests for GET /api/analysis-runs/{run_id}/export/csv/{field_path}. + + The pure CSV serialization logic is covered by tests/test_csv_export.py. + These tests verify only the HTTP shell: status codes, error envelopes, + headers, and that the route correctly pulls the measurement payload + out of the run snapshot. + """ + + def _decode_json_response(self, response) -> dict: + return json.loads(response.body.decode("utf-8")) + + def _make_run_with_measurement(self, runtime, measurement_payload: dict) -> str: + created = runtime.create_run( + filename="track.mp3", + content=b"fake-audio", + mime_type="audio/mpeg", + pitch_note_mode="off", + pitch_note_backend="auto", + interpretation_mode="off", + interpretation_profile="producer_summary", + interpretation_model=None, + ) + runtime.complete_measurement( + created["runId"], + payload=measurement_payload, + provenance={"schemaVersion": "measurement.v1", "engineVersion": "analyze.py"}, + diagnostics={"backendDurationMs": 1000}, + ) + return created["runId"] + + def test_returns_csv_with_attachment_disposition_on_success(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_csv_export_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run_with_measurement( + runtime, + measurement_payload={ + "lufsCurve": { + "shortTerm": [ + {"t": 0.0, "lufs": -23.0}, + {"t": 0.5, "lufs": -22.5}, + ], + "momentary": None, + }, + }, + ) + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = asyncio.run( + server.export_run_field_as_csv( + run_id, + "lufsCurve.shortTerm", + ) + ) + + self.assertEqual(response.status_code, 200) + self.assertTrue( + response.media_type.startswith("text/csv"), + f"Expected text/csv media type, got {response.media_type!r}", + ) + # Body is bytes — decode and check the header row + body_text = response.body.decode("utf-8") + self.assertTrue(body_text.startswith("time,duration,lufs")) + self.assertIn("0.000000,3.0,-23.00", body_text) + # Content-Disposition includes the run id and dotless field path + disposition = response.headers.get("content-disposition", "") + self.assertIn(f"{run_id}_lufsCurve_shortTerm.csv", disposition) + + def test_unsupported_field_path_returns_404(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_csv_export_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run_with_measurement( + runtime, measurement_payload={"bpm": 128} + ) + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = asyncio.run( + server.export_run_field_as_csv(run_id, "bpm") + ) + + self.assertEqual(response.status_code, 404) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "EXPORT_FIELD_NOT_SUPPORTED") + # The error message must list the supported paths so the caller + # can fix their request without reading the source. + self.assertIn("lufsCurve.shortTerm", payload["error"]["message"]) + + def test_field_supported_but_not_populated_returns_404(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_csv_export_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + run_id = self._make_run_with_measurement( + runtime, + measurement_payload={ + # tempoCurve is explicitly null — the analyzer ran but + # produced no per-point data for this audio. + "rhythmDetail": {"tempoCurve": None}, + }, + ) + + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = asyncio.run( + server.export_run_field_as_csv( + run_id, "rhythmDetail.tempoCurve" + ) + ) + + self.assertEqual(response.status_code, 404) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "EXPORT_FIELD_NOT_AVAILABLE") + + def test_unknown_run_returns_404(self) -> None: + from analysis_runtime import AnalysisRuntime + + with tempfile.TemporaryDirectory(prefix="asa_csv_export_") as temp_dir: + runtime = AnalysisRuntime(Path(temp_dir) / "runtime") + with patch.object(server, "get_analysis_runtime", return_value=runtime): + response = asyncio.run( + server.export_run_field_as_csv( + "does-not-exist", "lufsCurve.shortTerm" + ) + ) + + self.assertEqual(response.status_code, 404) + payload = self._decode_json_response(response) + self.assertEqual(payload["error"]["code"], "RUN_NOT_FOUND") + + if __name__ == "__main__": unittest.main() diff --git a/docs/adr/0001-phase1-json-schema-v1.md b/docs/adr/0001-phase1-json-schema-v1.md new file mode 100644 index 00000000..23240af8 --- /dev/null +++ b/docs/adr/0001-phase1-json-schema-v1.md @@ -0,0 +1,89 @@ +# ADR 0001 — Phase 1 JSON Schema v1 + +**Status:** Accepted +**Date:** 2026-05-13 +**Supersedes:** — +**Anchor:** [PURPOSE.md](../../PURPOSE.md) quality invariant #2 (citation chain). + +## Context + +ASA's HTTP contract has two layers that matter to external consumers: + +1. **The raw analyzer output** — what `apps/backend/analyze.py` writes to stdout. Documented in [`apps/backend/JSON_SCHEMA.md`](../../apps/backend/JSON_SCHEMA.md). The top-level key set is enforced by `EXPECTED_TOP_LEVEL_KEYS` in [`apps/backend/tests/test_analyze.py`](../../apps/backend/tests/test_analyze.py). +2. **The `phase1` HTTP envelope** — what `GET /api/analysis-runs/{run_id}` returns under `stages.measurement.result`. Typed on the frontend by `Phase1Result` in [`apps/ui/src/types/measurement.ts`](../../apps/ui/src/types/measurement.ts). + +These two layers share the same field shape — Python emits camelCase JSON directly; there is no rename layer between them ([CLAUDE.md tripwire #3](../../CLAUDE.md)). That coupling has served ASA well, but it carries no explicit version or compatibility promise. Until this ADR, "Phase 1 schema v1" was implicit in `EXPECTED_TOP_LEVEL_KEYS` and `Phase1Result` — provable from code but not stated as a contract. + +The external-repo review at [`docs/external-repo-review-2026-05-13.md`](../external-repo-review-2026-05-13.md) flagged this gap as Track 2 work. Partiels exports a documented schema (CSV/JSON/SDIF), and consumers of ASA's output (REAPER scripts, Max patches, downstream pipelines) need a stable shape to write against. + +## Decision + +The Phase 1 measurement payload — as captured by `EXPECTED_TOP_LEVEL_KEYS` and documented in `JSON_SCHEMA.md` on the date of this ADR — is hereby declared **Phase 1 JSON Schema v1**. + +The contract is the union of: + +- The current set of top-level keys in `apps/backend/tests/test_analyze.py:EXPECTED_TOP_LEVEL_KEYS`. +- Their semantic meaning as documented in `apps/backend/JSON_SCHEMA.md`. +- The per-detail-object shapes as typed in `apps/ui/src/types/measurement.ts`. + +The combination is the schema. There is no separate JSON Schema artifact (yet). Generation of a machine-readable schema is deferred to v2 if and when an external consumer needs it for validation. + +## Compatibility policy + +A future change to the Phase 1 payload is classified as follows: + +| Change | Class | Allowed in v1? | +| --- | --- | --- | +| Add a new top-level key | Additive (minor) | **Yes**, must update `EXPECTED_TOP_LEVEL_KEYS`, `JSON_SCHEMA.md`, and `Phase1Result`. | +| Add a new nested key inside an existing detail object | Additive (minor) | **Yes**, must update the detail interface in `types/measurement.ts`. | +| Add a new value to an existing string enum (e.g. `keyProfile`, `dynamicCharacter`) | Additive (minor) | **Yes**, must update the TypeScript union. Consumers should treat unknown values defensively. | +| Rename an existing key | Breaking (major) | **No** under v1. Requires a v2 bump. | +| Remove an existing key | Breaking (major) | **No** under v1. | +| Change a key's type (e.g. `number` → `number | null`, scalar → object) | Breaking (major) | **No** under v1. | +| Change the units of a value (e.g. dB → LUFS, seconds → milliseconds) | Breaking (major) | **No** under v1. | +| Tighten or loosen a value range (e.g. confidence was `[0, 1]`, now `[0, 100]`) | Breaking (major) | **No** under v1. | +| Change the shape of a detail object's elements (e.g. `{t, lufs}` → `{time, value}`) | Breaking (major) | **No** under v1. | + +**Versioning identifier:** the schema version is the string `"phase1.v1"`. The CSV export endpoint (`GET /api/analysis-runs/{run_id}/export/csv/{field_path}`) does **not** carry a version because its CSV column names mirror the v1 field names; if a breaking schema change ever bumps to v2, the export contract must be re-evaluated at the same time. + +## Time-series field shapes (v1 reference) + +The following nested shapes are the load-bearing time-series fields exported via the CSV endpoint introduced alongside this ADR. They are pinned by v1. + +| Field | Element shape | CSV columns | +| --- | --- | --- | +| `lufsCurve.shortTerm` | `{t: number, lufs: number}` | `time, duration, lufs` (duration=3.0) | +| `lufsCurve.momentary` | `{t: number, lufs: number}` | `time, duration, lufs` (duration=0.4) | +| `rhythmDetail.tempoCurve` | `{t: number, bpm: number}` | `time, bpm` | +| `spectralBalanceTimeSeries` | `{t, subBass, lowBass, lowMids, mids, upperMids, highs, brilliance}` | `time, subBass, lowBass, lowMids, mids, upperMids, highs, brilliance` | + +Adding a new exportable time-series field is an additive change: register a serializer in [`apps/backend/csv_export.py`](../../apps/backend/csv_export.py), update the test in `apps/backend/tests/test_csv_export.py`, and document the CSV columns in the table above. + +## Where v1 is enforced + +| Layer | Enforcement | If you break v1 here | +| --- | --- | --- | +| Top-level keys | `EXPECTED_TOP_LEVEL_KEYS` snapshot in `test_analyze.py` | Test fails; CI red. | +| Detail shapes | `Phase1Result` typing in `apps/ui/src/types/measurement.ts` | TypeScript type-check fails on the UI side. | +| HTTP envelope | `_normalize_run_snapshot` in `apps/backend/server.py` | Existing route tests in `test_server.py` fail. | +| CSV column contract | `apps/backend/tests/test_csv_export.py` | Per-field tests assert exact column names and order. | + +There is no single "phase1.v1.json" file. The schema is the four-test-suite intersection. This is deliberate — generating a JSON Schema artifact would add a new failure mode (the artifact drifting from the runtime contract) without a current consumer. + +## What this ADR does *not* do + +- **Does not generate a machine-readable JSON Schema.** No `phase1.v1.schema.json` is committed. If a future external consumer requires one, that's a v2-era task or a separate ADR. +- **Does not add an HTTP version header.** The `phase1` envelope is implicitly v1 by virtue of the route shape; if/when v2 lands, the canonical move is a new top-level field `phase1Version` rather than HTTP header negotiation. +- **Does not freeze SDIF or LAB export.** Both were considered in the review and deferred. This ADR scopes v1 to JSON + CSV. + +## Consequences + +- Future PRs that add fields are routine and low-friction; the checklist is in [CLAUDE.md](../../CLAUDE.md) under "Where to Make the Change." +- Future PRs that rename, remove, or retype fields require a new ADR and either a v2 envelope or a deprecation cycle. Neither is in scope for v1. +- ASA-side consumers (the UI) are tightly coupled to v1 by construction. External-side consumers (REAPER, Max, downstream scripts) gain a stable column contract from the CSV exporter. + +## Alternatives considered + +- **Mirror Partiels' export schema verbatim.** Rejected in [external-repo-review-2026-05-13.md](../external-repo-review-2026-05-13.md). Partiels' CSV is `time, duration, label/value` per Vamp track — too flat for ASA's domain-named measurements. Would have collapsed the citation chain (Quality Invariant #2). +- **Generate a JSON Schema artifact.** Deferred. No current consumer needs it; the test-snapshot enforcement already prevents drift. +- **Version the schema via HTTP `Accept` header.** Overkill for one runtime. Re-evaluate at v2. From 36f616276883e111c88d0fe5fbd653f42ffd3884 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 13 May 2026 06:43:01 +0000 Subject: [PATCH 2/2] test(export): guard against NaN/Infinity in CSV cells MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per PR #35 review (both non-blocking, both worth doing): 1. NaN passthrough. The serializers did float(value) on numeric fields with no guard. Python's json.loads accepts NaN/Infinity by default (non-standard JSON extension), so a pathological analyzer output could in principle round-trip a NaN into a snapshot and from there into a CSV cell as 'nan' — which most downstream tools (REAPER, Max, pandas) mis-parse as a string. Added _as_finite_float() helper that converts to float and returns None for NaN/Inf/non-numeric. The three serializers now use it instead of bare float(); a row with any non-finite field is skipped, matching the existing 'missing key -> skip row' behavior. 2. Empty-array coverage was only tested for lufsCurve.shortTerm. Added the same assertion for rhythmDetail.tempoCurve and spectralBalanceTimeSeries. Trivial; closes the most common 'measurement ran but produced nothing useful' case. New tests (14 total): - 2 empty-array tests for tempo/spectral - 8 unit tests of _as_finite_float covering None, normal numbers, NaN, +Inf, -Inf, non-numeric strings, numeric-string coercion - 4 end-to-end tests that NaN/Inf in real fields causes the row to be skipped from the CSV Test count: 22 -> 36, all passing locally. https://claude.ai/code/session_01Uh2eNT1oqjixcox8UX3BTg --- apps/backend/csv_export.py | 66 +++++++++---- apps/backend/tests/test_csv_export.py | 128 +++++++++++++++++++++++++- 2 files changed, 177 insertions(+), 17 deletions(-) diff --git a/apps/backend/csv_export.py b/apps/backend/csv_export.py index 196f87da..84eca5d2 100644 --- a/apps/backend/csv_export.py +++ b/apps/backend/csv_export.py @@ -38,6 +38,7 @@ import csv import io +import math from typing import Any, Callable @@ -135,6 +136,32 @@ def _resolve_dot_path(payload: dict, dot_path: str) -> Any: return node +def _as_finite_float(value: Any) -> float | None: + """Convert ``value`` to a finite ``float``, or ``None``. + + Returns ``None`` for: ``None`` input, non-numeric input (raises on + ``float()``), and ``NaN`` / ``Infinity`` / ``-Infinity``. The + non-finite case matters: Python's ``json.loads`` accepts ``NaN`` and + ``Infinity`` by default (non-standard JSON), so a pathological + measurement could round-trip an NaN into a snapshot and from there + into a CSV cell as the string ``"nan"`` — which most downstream + consumers (REAPER, Max, pandas) would mis-parse as a string. + + Used at the per-row level: a row with any non-finite value is + skipped entirely, matching the existing "missing key → skip row" + behavior. + """ + if value is None: + return None + try: + as_float = float(value) + except (TypeError, ValueError): + return None + if not math.isfinite(as_float): + return None + return as_float + + # ---------------------------------------------------------------------- # Per-field serializers # ---------------------------------------------------------------------- @@ -143,7 +170,11 @@ def _resolve_dot_path(payload: dict, dot_path: str) -> Any: def _serialize_lufs_curve( points: Any, window_duration_s: float ) -> str | None: - """Serialize a list of ``{t, lufs}`` points with a constant duration column.""" + """Serialize a list of ``{t, lufs}`` points with a constant duration column. + + Skips any row where either ``t`` or ``lufs`` is missing, non-numeric, + or non-finite (NaN, Infinity). + """ if not isinstance(points, list) or len(points) == 0: return None buf = io.StringIO() @@ -153,19 +184,21 @@ def _serialize_lufs_curve( for point in points: if not isinstance(point, dict): continue - t = point.get("t") - lufs = point.get("lufs") + t = _as_finite_float(point.get("t")) + lufs = _as_finite_float(point.get("lufs")) if t is None or lufs is None: continue - writer.writerow( - [f"{float(t):.6f}", f"{window_duration_s}", f"{float(lufs):.2f}"] - ) + writer.writerow([f"{t:.6f}", f"{window_duration_s}", f"{lufs:.2f}"]) wrote_any = True return buf.getvalue() if wrote_any else None def _serialize_tempo_curve(points: Any) -> str | None: - """Serialize a list of ``{t, bpm}`` points.""" + """Serialize a list of ``{t, bpm}`` points. + + Skips any row where either ``t`` or ``bpm`` is missing, non-numeric, + or non-finite. + """ if not isinstance(points, list) or len(points) == 0: return None buf = io.StringIO() @@ -175,11 +208,11 @@ def _serialize_tempo_curve(points: Any) -> str | None: for point in points: if not isinstance(point, dict): continue - t = point.get("t") - bpm = point.get("bpm") + t = _as_finite_float(point.get("t")) + bpm = _as_finite_float(point.get("bpm")) if t is None or bpm is None: continue - writer.writerow([f"{float(t):.6f}", f"{float(bpm):.3f}"]) + writer.writerow([f"{t:.6f}", f"{bpm:.3f}"]) wrote_any = True return buf.getvalue() if wrote_any else None @@ -187,8 +220,9 @@ def _serialize_tempo_curve(points: Any) -> str | None: def _serialize_spectral_balance_time_series(points: Any) -> str | None: """Serialize a list of ``{t, subBass, ..., brilliance}`` points. - Skips any row that is missing one of the band values rather than - writing a partial row — keeps the CSV regular and downstream-loadable. + Skips any row where ``t`` or any band value is missing, non-numeric, + or non-finite (NaN, Infinity) — keeps the CSV regular and + downstream-loadable. """ if not isinstance(points, list) or len(points) == 0: return None @@ -199,20 +233,20 @@ def _serialize_spectral_balance_time_series(points: Any) -> str | None: for point in points: if not isinstance(point, dict): continue - t = point.get("t") + t = _as_finite_float(point.get("t")) if t is None: continue band_values: list[str] = [] skip = False for band in _SPECTRAL_BANDS: - value = point.get(band) + value = _as_finite_float(point.get(band)) if value is None: skip = True break - band_values.append(f"{float(value):.4f}") + band_values.append(f"{value:.4f}") if skip: continue - writer.writerow([f"{float(t):.6f}", *band_values]) + writer.writerow([f"{t:.6f}", *band_values]) wrote_any = True return buf.getvalue() if wrote_any else None diff --git a/apps/backend/tests/test_csv_export.py b/apps/backend/tests/test_csv_export.py index 84ee7d54..6d6884b9 100644 --- a/apps/backend/tests/test_csv_export.py +++ b/apps/backend/tests/test_csv_export.py @@ -95,11 +95,137 @@ def test_missing_field_returns_none(self): result = csv_export.export_field_to_csv({}, "lufsCurve.shortTerm") self.assertIsNone(result) - def test_empty_array_returns_none(self): + def test_empty_array_returns_none_for_lufs_curve(self): payload = {"lufsCurve": {"shortTerm": []}} result = csv_export.export_field_to_csv(payload, "lufsCurve.shortTerm") self.assertIsNone(result) + def test_empty_array_returns_none_for_tempo_curve(self): + payload = {"rhythmDetail": {"tempoCurve": []}} + result = csv_export.export_field_to_csv(payload, "rhythmDetail.tempoCurve") + self.assertIsNone(result) + + def test_empty_array_returns_none_for_spectral_balance(self): + payload = {"spectralBalanceTimeSeries": []} + result = csv_export.export_field_to_csv(payload, "spectralBalanceTimeSeries") + self.assertIsNone(result) + + +class FiniteFloatHelperTests(unittest.TestCase): + """_as_finite_float guards against NaN/Inf leaking into CSV cells. + + Python's json.loads accepts NaN/Infinity by default (non-standard JSON + extension), so a pathological analyzer output could in principle put + a NaN into the snapshot. Without this guard, float() would happily + convert NaN/Inf to themselves and the CSV writer would emit \"nan\" + or \"inf\" as a string — which most downstream tools mis-parse. + """ + + def test_none_returns_none(self): + self.assertIsNone(csv_export._as_finite_float(None)) + + def test_normal_int_returns_float(self): + self.assertEqual(csv_export._as_finite_float(42), 42.0) + + def test_normal_float_returns_self(self): + self.assertEqual(csv_export._as_finite_float(-23.5), -23.5) + + def test_nan_returns_none(self): + self.assertIsNone(csv_export._as_finite_float(float("nan"))) + + def test_positive_inf_returns_none(self): + self.assertIsNone(csv_export._as_finite_float(float("inf"))) + + def test_negative_inf_returns_none(self): + self.assertIsNone(csv_export._as_finite_float(float("-inf"))) + + def test_non_numeric_string_returns_none(self): + self.assertIsNone(csv_export._as_finite_float("not-a-number")) + + def test_numeric_string_is_coerced(self): + # We don't expect strings in the analyzer payload, but if one + # sneaks in and it's numeric, this is the same forgiving behavior + # float() always had. + self.assertEqual(csv_export._as_finite_float("3.14"), 3.14) + + +class NonFiniteValueHandlingTests(unittest.TestCase): + """End-to-end: a row with any NaN/Inf field is skipped from the CSV.""" + + def test_lufs_curve_skips_nan_lufs(self): + payload = { + "lufsCurve": { + "shortTerm": [ + {"t": 0.0, "lufs": -23.0}, + {"t": 0.1, "lufs": float("nan")}, # skipped + {"t": 0.2, "lufs": -22.0}, + ] + } + } + result = csv_export.export_field_to_csv(payload, "lufsCurve.shortTerm") + self.assertIsNotNone(result) + self.assertNotIn("nan", result) + self.assertNotIn("NaN", result) + # header + 2 valid rows + self.assertEqual(len(result.strip().split("\n")), 3) + + def test_lufs_curve_skips_inf_time(self): + payload = { + "lufsCurve": { + "shortTerm": [ + {"t": float("inf"), "lufs": -23.0}, # skipped + {"t": 0.5, "lufs": -22.0}, + ] + } + } + result = csv_export.export_field_to_csv(payload, "lufsCurve.shortTerm") + self.assertNotIn("inf", result.lower()) + self.assertEqual(len(result.strip().split("\n")), 2) # header + 1 + + def test_tempo_curve_skips_nan_bpm(self): + payload = { + "rhythmDetail": { + "tempoCurve": [ + {"t": 0.0, "bpm": float("nan")}, # skipped + {"t": 1.0, "bpm": 128.0}, + ] + } + } + result = csv_export.export_field_to_csv(payload, "rhythmDetail.tempoCurve") + self.assertNotIn("nan", result.lower()) + self.assertEqual(len(result.strip().split("\n")), 2) + + def test_spectral_balance_skips_row_with_nan_band(self): + payload = { + "spectralBalanceTimeSeries": [ + { + "t": 0.0, + "subBass": 0.5, + "lowBass": 0.5, + "lowMids": 0.5, + "mids": 0.5, + "upperMids": 0.5, + "highs": 0.5, + "brilliance": float("nan"), # one bad band → row skipped + }, + { + "t": 1.0, + "subBass": 0.4, + "lowBass": 0.4, + "lowMids": 0.4, + "mids": 0.4, + "upperMids": 0.4, + "highs": 0.4, + "brilliance": 0.4, + }, + ] + } + result = csv_export.export_field_to_csv( + payload, "spectralBalanceTimeSeries" + ) + self.assertNotIn("nan", result.lower()) + self.assertEqual(len(result.strip().split("\n")), 2) # header + 1 + class LufsCurveShortTermTests(unittest.TestCase): """lufsCurve.shortTerm → time,duration,lufs with duration=3.0 constant."""