From 3b2c49f89de47c7f7d1110e81291b44ed1aa3aba Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 7 Jun 2026 15:43:52 +0000 Subject: [PATCH 1/2] feat(export): export grani in JSON via --grain-json (closes #73) Aggiunge GrainJsonWriter che serializza i grani di ogni stream in JSON per i client di visualizzazione (PGE-ui#13). Un file per stream: {cache}/{basename}__{stream_id}__grains.json - Schema: stream_id, duration, num_voices, grains[{t,dur,vol,ptr,v}] - t relativo a stream.onset (puo' essere < 0 con onset offset per-voce) - itera stream.voices per preservare l'indice voce - JSON compatto, grani ordinati per t - flag CLI --grain-json, attivo solo con --per-stream https://claude.ai/code/session_015LPLSKXNZW6Qg67gyzVZcy --- src/export/grain_json_writer.py | 102 +++++++++++ src/main.py | 14 +- tests/export/test_grain_json_writer.py | 223 +++++++++++++++++++++++++ 3 files changed, 338 insertions(+), 1 deletion(-) create mode 100644 src/export/grain_json_writer.py create mode 100644 tests/export/test_grain_json_writer.py diff --git a/src/export/grain_json_writer.py b/src/export/grain_json_writer.py new file mode 100644 index 0000000..d4912c4 --- /dev/null +++ b/src/export/grain_json_writer.py @@ -0,0 +1,102 @@ +# src/export/grain_json_writer.py +""" +GrainJsonWriter + +Esporta i grani di uno stream in JSON, consumabile da un client di +visualizzazione (PGE-ui) per disegnare i rettangoli dei grani nella clip +timeline di ogni stream. + +Responsabilita': +- build(): produce la struttura dati (dict) per uno stream +- generate(): serializza la struttura in JSON compatto (stringa) +- write(): scrive il file JSON su disco (un file per stream) + +Schema JSON prodotto (compatto, senza whitespace): + { + "stream_id": "stream1", + "duration": 8.0, + "num_voices": 4, + "grains": [ + {"t": 0.0, "dur": 0.08, "vol": -6.0, "ptr": 0.34, "v": 0}, + ... + ] + } + +Campi grano: + t = grain.onset - stream.onset (onset relativo all'inizio dello stream; + puo' essere < 0 con onset offset per-voce: dato valido) + dur = grain.duration + vol = grain.volume + ptr = grain.pointer_pos (unita' variabile per stream: secondi o frazione) + v = indice voce (da stream.voices) + +I grani sono ordinati per t crescente. num_voices riflette il numero di voci +effettivamente generate (len(stream.voices)). +""" + +import json +import os +from pathlib import Path + + +class GrainJsonWriter: + """ + Esporta i grani di uno stream in un file JSON. + + Itera stream.voices (List[List[Grain]]) per preservare l'indice voce, + che il flat stream.grains perderebbe. + """ + + def build(self, stream) -> dict: + """ + Produce la struttura dati JSON-serializzabile per uno stream. + + Args: + stream: Stream con stream_id, onset, duration, voices + + Returns: + dict con stream_id, duration, num_voices, grains (ordinati per t) + """ + grains = [] + for voice_index, voice in enumerate(stream.voices): + for grain in voice: + grains.append({ + "t": grain.onset - stream.onset, + "dur": grain.duration, + "vol": grain.volume, + "ptr": grain.pointer_pos, + "v": voice_index, + }) + + grains.sort(key=lambda g: g["t"]) + + return { + "stream_id": stream.stream_id, + "duration": stream.duration, + "num_voices": len(stream.voices), + "grains": grains, + } + + def generate(self, stream) -> str: + """Serializza la struttura di build() in JSON compatto (no whitespace).""" + return json.dumps(self.build(stream), separators=(",", ":")) + + def write(self, stream, output_dir: str, yaml_basename: str) -> Path: + """ + Scrive il file JSON dei grani su disco. + + Args: + stream: Stream da esportare + output_dir: directory di output (creata se assente) + yaml_basename: basename del file YAML, usato nel nome file + + Returns: + Path del file JSON scritto: + {output_dir}/{yaml_basename}__{stream_id}__grains.json + """ + os.makedirs(output_dir, exist_ok=True) + filename = f"{yaml_basename}__{stream.stream_id}__grains.json" + path = Path(output_dir) / filename + with open(path, "w") as f: + f.write(self.generate(stream)) + return path diff --git a/src/main.py b/src/main.py index df231ca..37a5ead 100644 --- a/src/main.py +++ b/src/main.py @@ -129,7 +129,8 @@ def main(): "[--log-dir DIR] [--message-level N] " "[--keep-sco] [--sco-dir DIR] " "[--cache] [--cache-dir DIR] " - "[--reaper] [--reaper-path FILE]" + "[--reaper] [--reaper-path FILE] " + "[--grain-json]" ) sys.exit(1) @@ -146,6 +147,7 @@ def main(): per_stream = '--per-stream' in sys.argv or '-p' in sys.argv use_cache = '--cache' in sys.argv reaper_export = '--reaper' in sys.argv + grain_json = '--grain-json' in sys.argv # --reaper-path PATH (default: {yaml_basename}.rpp) reaper_path = None @@ -312,6 +314,16 @@ def main(): ) print(f"Reaper project: {rpp_out}") + if grain_json: + if not per_stream: + print("[grain-json] ignorato: richiede --per-stream") + else: + from export.grain_json_writer import GrainJsonWriter + writer = GrainJsonWriter() + for stream in generator.streams: + json_path = writer.write(stream, cache_dir, yaml_basename) + print(f"Grain JSON: {json_path}") + if do_visualize: print("\nGenerazione partitura grafica...") pdf_file = output_file.rsplit('.', 1)[0] + '.pdf' diff --git a/tests/export/test_grain_json_writer.py b/tests/export/test_grain_json_writer.py new file mode 100644 index 0000000..cec2ea5 --- /dev/null +++ b/tests/export/test_grain_json_writer.py @@ -0,0 +1,223 @@ +# tests/export/test_grain_json_writer.py +""" +TDD suite per GrainJsonWriter. + +Esporta i grani di uno stream in JSON consumabile da un client di +visualizzazione (PGE-ui). Un file per stream. + +Schema JSON prodotto (compatto, senza whitespace): + {"stream_id":"s1","duration":8.0,"num_voices":4, + "grains":[{"t":0.0,"dur":0.08,"vol":-6.0,"ptr":0.34,"v":0}, ...]} + + - t = grain.onset - stream.onset (relativo allo stream; puo' essere < 0) + - dur = grain.duration + - vol = grain.volume + - ptr = grain.pointer_pos + - v = indice voce (da stream.voices) + grains ordinati per t crescente. + +Sezioni: +1. TestTopLevelStructure - chiavi top-level e metadati stream +2. TestGrainMapping - mappatura campi grano e indice voce +3. TestOrdering - grani ordinati per t +4. TestCompactJson - JSON compatto, niente whitespace +5. TestWriteToDisk - write() crea il file, contenuto == generate() +6. TestEdgeCases - stream senza grani, t negativo +""" + +import json +import pytest +from unittest.mock import Mock + +from export.grain_json_writer import GrainJsonWriter + + +# ============================================================================= +# FIXTURES +# ============================================================================= + +def _make_grain(onset, duration=0.08, volume=-6.0, pointer_pos=0.0): + g = Mock() + g.onset = onset + g.duration = duration + g.volume = volume + g.pointer_pos = pointer_pos + return g + + +def _make_stream(stream_id, onset, duration, voices): + """voices: List[List[grain]] - una lista di grani per voce.""" + s = Mock() + s.stream_id = stream_id + s.onset = onset + s.duration = duration + s.voices = voices + return s + + +@pytest.fixture +def writer(): + return GrainJsonWriter() + + +@pytest.fixture +def two_voice_stream(): + # voce 0: due grani; voce 1: un grano + return _make_stream( + "s1", onset=10.0, duration=8.0, + voices=[ + [_make_grain(10.0, pointer_pos=0.34), _make_grain(11.0, pointer_pos=0.5)], + [_make_grain(10.5, pointer_pos=0.1)], + ], + ) + + +# ============================================================================= +# 1. STRUTTURA TOP-LEVEL +# ============================================================================= + +class TestTopLevelStructure: + """Chiavi top-level: stream_id, duration, num_voices, grains.""" + + def test_stream_id_in_output(self, writer, two_voice_stream): + data = writer.build(two_voice_stream) + assert data["stream_id"] == "s1" + + def test_duration_in_output(self, writer, two_voice_stream): + data = writer.build(two_voice_stream) + assert data["duration"] == 8.0 + + def test_num_voices_matches_voices_length(self, writer, two_voice_stream): + """num_voices riflette il numero di voci effettivamente generate.""" + data = writer.build(two_voice_stream) + assert data["num_voices"] == 2 + + def test_grains_is_a_list(self, writer, two_voice_stream): + data = writer.build(two_voice_stream) + assert isinstance(data["grains"], list) + + def test_grain_count_is_total_across_voices(self, writer, two_voice_stream): + data = writer.build(two_voice_stream) + assert len(data["grains"]) == 3 + + +# ============================================================================= +# 2. MAPPATURA CAMPI GRANO +# ============================================================================= + +class TestGrainMapping: + """Ogni grano mappa t/dur/vol/ptr/v correttamente.""" + + def test_t_relative_to_stream_onset(self, writer): + stream = _make_stream("s1", onset=10.0, duration=4.0, + voices=[[_make_grain(12.5)]]) + data = writer.build(stream) + assert data["grains"][0]["t"] == pytest.approx(2.5) + + def test_dur_maps_grain_duration(self, writer): + stream = _make_stream("s1", onset=0.0, duration=4.0, + voices=[[_make_grain(0.0, duration=0.123)]]) + data = writer.build(stream) + assert data["grains"][0]["dur"] == pytest.approx(0.123) + + def test_vol_maps_grain_volume(self, writer): + stream = _make_stream("s1", onset=0.0, duration=4.0, + voices=[[_make_grain(0.0, volume=-12.5)]]) + data = writer.build(stream) + assert data["grains"][0]["vol"] == pytest.approx(-12.5) + + def test_ptr_maps_grain_pointer_pos(self, writer): + stream = _make_stream("s1", onset=0.0, duration=4.0, + voices=[[_make_grain(0.0, pointer_pos=0.42)]]) + data = writer.build(stream) + assert data["grains"][0]["ptr"] == pytest.approx(0.42) + + def test_voice_index_recorded(self, writer): + """Il campo v corrisponde all'indice della voce in stream.voices.""" + stream = _make_stream("s1", onset=0.0, duration=4.0, + voices=[[_make_grain(0.0)], [_make_grain(0.0)]]) + data = writer.build(stream) + voices_seen = {g["v"] for g in data["grains"]} + assert voices_seen == {0, 1} + + +# ============================================================================= +# 3. ORDINAMENTO +# ============================================================================= + +class TestOrdering: + """I grani sono ordinati per t crescente, anche tra voci diverse.""" + + def test_grains_sorted_by_t(self, writer): + stream = _make_stream("s1", onset=0.0, duration=10.0, + voices=[[_make_grain(5.0)], [_make_grain(1.0)], [_make_grain(3.0)]]) + data = writer.build(stream) + ts = [g["t"] for g in data["grains"]] + assert ts == sorted(ts) + + +# ============================================================================= +# 4. JSON COMPATTO +# ============================================================================= + +class TestCompactJson: + """generate() produce JSON valido e compatto (no whitespace).""" + + def test_generate_returns_valid_json(self, writer, two_voice_stream): + text = writer.generate(two_voice_stream) + parsed = json.loads(text) + assert parsed["stream_id"] == "s1" + + def test_generate_has_no_whitespace(self, writer, two_voice_stream): + text = writer.generate(two_voice_stream) + assert ", " not in text + assert ": " not in text + + +# ============================================================================= +# 5. SCRITTURA SU DISCO +# ============================================================================= + +class TestWriteToDisk: + """write() crea il file; il contenuto coincide con generate().""" + + def test_write_creates_file(self, writer, two_voice_stream, tmp_path): + path = writer.write(two_voice_stream, str(tmp_path), "myconfig") + import os + assert os.path.exists(path) + + def test_write_filename_contains_basename_and_stream_id(self, writer, two_voice_stream, tmp_path): + path = writer.write(two_voice_stream, str(tmp_path), "myconfig") + name = str(path) + assert "myconfig" in name and "s1" in name and name.endswith(".json") + + def test_write_content_matches_generate(self, writer, two_voice_stream, tmp_path): + path = writer.write(two_voice_stream, str(tmp_path), "myconfig") + assert open(path).read() == writer.generate(two_voice_stream) + + +# ============================================================================= +# 6. EDGE CASES +# ============================================================================= + +class TestEdgeCases: + """Stream senza grani, t negativo (onset offset per-voce).""" + + def test_stream_without_grains(self, writer): + stream = _make_stream("empty", onset=0.0, duration=4.0, voices=[]) + data = writer.build(stream) + assert data["grains"] == [] + assert data["num_voices"] == 0 + + def test_voice_with_empty_grain_list(self, writer): + stream = _make_stream("s1", onset=0.0, duration=4.0, voices=[[], []]) + data = writer.build(stream) + assert data["grains"] == [] + assert data["num_voices"] == 2 + + def test_negative_t_allowed(self, writer): + """Onset offset per-voce puo' produrre grani con onset < stream.onset.""" + stream = _make_stream("s1", onset=10.0, duration=4.0, + voices=[[_make_grain(9.5)]]) + data = writer.build(stream) + assert data["grains"][0]["t"] == pytest.approx(-0.5) From 254352c0afc31e727fd4724c2a407863ad4516e6 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 7 Jun 2026 20:44:45 +0000 Subject: [PATCH 2/2] fix(export): scrivi grain JSON accanto agli stem .aif invece che in cache_dir I file grain JSON sono sidecar degli stem audio a cui si riferiscono: PGE-ui li trova nella stessa directory degli .aif (output STEMS) invece che in cache_dir, che non ha relazione semantica con l'export e poteva non esistere senza --cache. Riusa il pattern os.path.dirname(abspath) gia' usato per aif_dir nel garbage collect. https://claude.ai/code/session_019N8ZDHmri8tCNCu2WgGRzc --- src/main.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/main.py b/src/main.py index 37a5ead..3cbc9e0 100644 --- a/src/main.py +++ b/src/main.py @@ -319,9 +319,12 @@ def main(): print("[grain-json] ignorato: richiede --per-stream") else: from export.grain_json_writer import GrainJsonWriter + # Sidecar accanto agli stem .aif: PGE-ui trova grain JSON e + # audio nella stessa directory dell'output STEMS. + grain_json_dir = os.path.dirname(os.path.abspath(output_file)) writer = GrainJsonWriter() for stream in generator.streams: - json_path = writer.write(stream, cache_dir, yaml_basename) + json_path = writer.write(stream, grain_json_dir, yaml_basename) print(f"Grain JSON: {json_path}") if do_visualize: