added data saving feature

python_app/storage/npz/store.py

@@ -36,6 +36,50 @@ def _compose_snapshot_stem(name: str, name_prefix: str) -> str:
     return sanitize_path_component(f"{name_prefix}_{base}" if name_prefix else base)
 
 
+class SnapshotStreamWriter:
+    """Append aligned raw/preprocessed/result collections to a snapshot dir in chunks.
+
+    A streaming alternative to :meth:`NpzStore.save_runtime_snapshot_numpy` for long
+    recordings: the caller flushes small chunks as measurements arrive and frees them,
+    so memory stays flat instead of buffering every measurement. The on-disk layout is
+    identical (``raw``/``preprocessed``/``results`` subtrees), and per-collection
+    directory indices continue across chunks via a running offset per stage.
+    """
+
+    __slots__ = ("_dir", "_raw_written", "_preprocessed_written", "_result_written")
+
+    def __init__(self, snapshot_dir: Path) -> None:
+        self._dir = snapshot_dir
+        self._raw_written = 0
+        self._preprocessed_written = 0
+        self._result_written = 0
+
+    @property
+    def directory(self) -> Path:
+        return self._dir
+
+    @property
+    def result_count(self) -> int:
+        """Number of result collections written to disk so far."""
+        return self._result_written
+
+    def append(
+        self,
+        raw: list[SweepCollection],
+        preprocessed: list[SweepCollection],
+        results: list[ResultCollection],
+    ) -> None:
+        """Write one chunk of (already aligned) collections, continuing each stage's index."""
+        save_trace_history_numpy(self._dir / "raw", raw, index_offset=self._raw_written)
+        save_trace_history_numpy(
+            self._dir / "preprocessed", preprocessed, index_offset=self._preprocessed_written
+        )
+        save_result_history_numpy(self._dir / "results", results, index_offset=self._result_written)
+        self._raw_written += len(raw)
+        self._preprocessed_written += len(preprocessed)
+        self._result_written += len(results)
+
+
 class NpzStore(StoreApi):
     """Persist preprocess sets and runtime snapshots using NumPy files."""
 
@@ -194,6 +238,35 @@ class NpzStore(StoreApi):
         logger.info("Saved binary runtime snapshot (last_n=%d) to %s", last_n, snapshot_dir)
         return snapshot_dir
 
+    def snapshot_directory(
+        self, output_root_dir: Path, snapshot_name: str, *, name_prefix: str = ""
+    ) -> Path:
+        """Return the directory a snapshot save would create for these inputs.
+
+        Lets callers check ``.exists()`` *before* acquiring data (e.g. the disk
+        recorder arms a run only once the destination is free), so a name clash is
+        reported up front instead of after the measurements are collected. Note a
+        blank ``snapshot_name`` resolves to a fresh timestamp each call, so this is
+        meaningful only for explicit names — exactly the case that can collide.
+        """
+        return output_root_dir / _compose_snapshot_stem(snapshot_name, name_prefix)
+
+    def create_snapshot_stream(
+        self, output_root_dir: Path, snapshot_name: str, *, name_prefix: str = ""
+    ) -> SnapshotStreamWriter:
+        """Create an empty snapshot directory and return a chunked stream writer for it.
+
+        Raises ``FileExistsError`` if the directory already exists (same guard as
+        :meth:`save_runtime_snapshot_numpy`), so a name clash is reported at arm time.
+        """
+        snapshot_dir = self.snapshot_directory(output_root_dir, snapshot_name, name_prefix=name_prefix)
+        output_root_dir.mkdir(parents=True, exist_ok=True)
+        if snapshot_dir.exists():
+            raise FileExistsError(f"Snapshot directory already exists: {snapshot_dir}")
+        snapshot_dir.mkdir(parents=True, exist_ok=False)
+        logger.info("Opened streaming snapshot directory %s", snapshot_dir)
+        return SnapshotStreamWriter(snapshot_dir)
+
     def save_runtime_snapshot_numpy(
         self,
         output_root_dir: Path,