feat(export): export grani in JSON via --grain-json (#73)

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

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,19 @@ 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
+                # 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, grain_json_dir, yaml_basename)
+                    print(f"Grain JSON: {json_path}")
+
         if do_visualize:
             print("\nGenerazione partitura grafica...")
             pdf_file = output_file.rsplit('.', 1)[0] + '.pdf'

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)