Add privacy-conscious waveform fixture extractor

README.md

@@ -48,6 +48,72 @@ eyeflow --data path\to\input.holo
 eyeflow-cli --data path\to\input.holo
 ```
 
+## Extracting Waveform Fixtures
+
+`tools/extract_waveform_fixture.py` creates compact waveform-only H5 files for
+algorithm development and robustness testing. It copies only an explicit
+allow-list of numeric artery/vein waveforms, beat timing, and optional per-beat
+signals. Source filenames, source attributes, images, masks, maps, and
+acquisition metadata are not copied.
+
+### Windows executable
+
+The downloadable `EyeFlowWaveformFixtureExtractor.exe` provides a double-click
+interface:
+
+1. Choose one EyeFlow H5 file or a folder of EyeFlow outputs.
+2. Choose a separate output folder.
+3. Select **Validate inputs** to check compatibility without writing files.
+4. Select **Extract fixtures** to create the compact waveform fixtures.
+
+The executable is published with a `.sha256` checksum file. It is not
+code-signed, so Windows SmartScreen may display an unrecognized-app warning.
+Verify the checksum before running a downloaded copy.
+
+To build the executable from source, install PyInstaller in the active Python
+environment and run:
+
+```powershell
+py -m pip install numpy h5py pyinstaller
+.\tools\build_waveform_fixture_executable.ps1
+```
+
+The executable and checksum are written to
+`dist\waveform-fixture-extractor\`.
+
+### Python command line
+
+Install the two required packages if EyeFlow is not already installed:
+
+```powershell
+py -m pip install numpy h5py
+```
+
+Validate a folder before writing anything:
+
+```powershell
+py tools\extract_waveform_fixture.py "D:\EyeFlowOutputs" `
+  --output-dir "D:\WaveformFixtures" `
+  --recursive `
+  --dry-run
+```
+
+Create the compact fixtures:
+
+```powershell
+py tools\extract_waveform_fixture.py "D:\EyeFlowOutputs" `
+  --output-dir "D:\WaveformFixtures" `
+  --recursive
+```
+
+Output filenames are derived from waveform content rather than source filenames.
+Repeating an extraction reuses an identical fixture. Run the script with
+`--help` for size limits and overwrite behavior.
+
+Metadata removal does not make clinical waveforms non-sensitive. Treat the
+fixtures as patient-derived data, follow the applicable data-governance rules,
+and do not commit them to a public repository without explicit approval.
+
 ## Scope
 
 ### In Scope

test/test_extract_waveform_fixture.py

@@ -0,0 +1,203 @@
+"""Tests for the standalone waveform-fixture extraction utility."""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+import h5py
+import numpy as np
+
+
+SCRIPT_PATH = Path(__file__).resolve().parents[1] / "tools" / "extract_waveform_fixture.py"
+SPEC = importlib.util.spec_from_file_location("extract_waveform_fixture", SCRIPT_PATH)
+if SPEC is None or SPEC.loader is None:
+    raise RuntimeError(f"Could not load {SCRIPT_PATH}")
+extractor = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = extractor
+SPEC.loader.exec_module(extractor)
+
+
+class WaveformFixtureExtractionTests(unittest.TestCase):
+    def test_extracts_allowlisted_arrays_and_strips_identifying_metadata(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            source_path = root / "patient_identifier_EF.h5"
+            output_dir = root / "fixtures"
+            artery = np.linspace(10.0, 20.0, 201, dtype=np.float32)
+            vein = np.linspace(5.0, 8.0, 201, dtype=np.float32)
+
+            with h5py.File(source_path, "w") as source:
+                source.attrs["patient_name"] = "Must Not Be Copied"
+                analysis = source.create_group("analysis")
+                analysis.create_dataset("retinal_artery_velocity_signal", data=artery)
+                analysis.create_dataset("retinal_vein_velocity_signal", data=vein)
+                analysis.create_dataset(
+                    "beat_indices",
+                    data=np.asarray([0, 100, 200], dtype=np.int32),
+                )
+                analysis.create_dataset(
+                    "time_per_beat",
+                    data=np.asarray([1.0, 1.0], dtype=np.float32),
+                )
+                analysis.create_dataset(
+                    "velocitysignal_filtered",
+                    data=artery - np.float32(0.5),
+                )
+                analysis.create_dataset(
+                    "retinal_velocity_array",
+                    data=np.ones((16, 16, 16), dtype=np.float32),
+                )
+                source.create_dataset(
+                    "patient_name",
+                    data=np.bytes_("Must Not Be Copied"),
+                )
+
+            result = extractor.extract_fixture(source_path, output_dir)
+
+            self.assertTrue(result.created)
+            self.assertNotIn("patient_identifier", result.output_path.name)
+            with h5py.File(result.output_path, "r") as fixture:
+                np.testing.assert_array_equal(
+                    fixture["waveforms/artery/raw"][()],
+                    artery,
+                )
+                np.testing.assert_array_equal(
+                    fixture["waveforms/vein/raw"][()],
+                    vein,
+                )
+                self.assertAlmostEqual(float(fixture["beats/dt_seconds"][()]), 0.01)
+                self.assertNotIn("retinal_velocity_array", fixture)
+                self.assertNotIn("patient_name", fixture)
+                self.assertNotIn("patient_name", fixture.attrs)
+                self.assertTrue(bool(fixture.attrs["metadata_stripped"]))
+                self.assertTrue(bool(fixture.attrs["contains_patient_derived_data"]))
+
+    def test_supports_slim_output_schema_aliases(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            source_path = root / "slim.h5"
+            output_dir = root / "fixtures"
+
+            with h5py.File(source_path, "w") as source:
+                source.create_dataset(
+                    "artery/velocity/signal/value",
+                    data=np.arange(12, dtype=np.float32),
+                )
+                source.create_dataset(
+                    "vein/velocity/signal/value",
+                    data=np.arange(12, dtype=np.float32) + 20,
+                )
+                source.create_dataset(
+                    "perbeat/beat_indices/value",
+                    data=np.asarray([0, 5, 10], dtype=np.int32),
+                )
+                source.create_dataset(
+                    "perbeat/time_per_beat/value",
+                    data=np.asarray([0.5, 0.5], dtype=np.float32),
+                )
+                source.create_dataset(
+                    "artery/velocity/perbeat/signal/value",
+                    data=np.ones((2, 8), dtype=np.float32),
+                )
+
+            result = extractor.extract_fixture(source_path, output_dir)
+
+            with h5py.File(result.output_path, "r") as fixture:
+                self.assertIn("per_beat/artery/raw", fixture)
+                self.assertEqual(
+                    fixture["beats/boundary_indices"].attrs["index_base"],
+                    0,
+                )
+                self.assertAlmostEqual(float(fixture["beats/dt_seconds"][()]), 0.1)
+
+    def test_reuses_identical_fixture_without_overwriting(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            source_path = root / "source.h5"
+            output_dir = root / "fixtures"
+            _write_minimal_source(source_path)
+
+            first = extractor.extract_fixture(source_path, output_dir)
+            second = extractor.extract_fixture(source_path, output_dir)
+
+            self.assertTrue(first.created)
+            self.assertFalse(second.created)
+            self.assertEqual(first.output_path, second.output_path)
+
+    def test_rejects_missing_required_waveform(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            source_path = Path(tmp_dir) / "invalid.h5"
+            with h5py.File(source_path, "w") as source:
+                source.create_dataset(
+                    "analysis/retinal_artery_velocity_signal",
+                    data=np.arange(10, dtype=np.float32),
+                )
+
+            with self.assertRaisesRegex(
+                extractor.FixtureExtractionError,
+                "Missing required EyeFlow waveform datasets",
+            ):
+                extractor.prepare_fixture(source_path, max_output_mb=64)
+
+    def test_recursive_discovery_excludes_output_directory(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            nested = root / "inputs" / "nested"
+            nested.mkdir(parents=True)
+            source_path = nested / "source.h5"
+            _write_minimal_source(source_path)
+            output_dir = root / "inputs" / "fixtures"
+            output_dir.mkdir()
+            _write_minimal_source(output_dir / "waveform_fixture_old.h5")
+
+            discovered = extractor.discover_h5_files(
+                [root / "inputs"],
+                recursive=True,
+                output_dir=output_dir,
+            )
+
+            self.assertEqual(discovered, [source_path.resolve()])
+
+    def test_discovery_allows_source_file_in_output_directory(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            output_dir = Path(tmp_dir)
+            source_path = output_dir / "acquisition.h5"
+            generated_path = output_dir / "waveform_fixture_existing.h5"
+            _write_minimal_source(source_path)
+            _write_minimal_source(generated_path)
+
+            discovered = extractor.discover_h5_files(
+                [output_dir],
+                recursive=False,
+                output_dir=output_dir,
+            )
+
+            self.assertEqual(discovered, [source_path.resolve()])
+
+
+def _write_minimal_source(path: Path) -> None:
+    with h5py.File(path, "w") as source:
+        source.create_dataset(
+            "analysis/retinal_artery_velocity_signal",
+            data=np.arange(12, dtype=np.float32),
+        )
+        source.create_dataset(
+            "analysis/retinal_vein_velocity_signal",
+            data=np.arange(12, dtype=np.float32) + 20,
+        )
+        source.create_dataset(
+            "analysis/beat_indices",
+            data=np.asarray([0, 5, 10], dtype=np.int32),
+        )
+        source.create_dataset(
+            "analysis/time_per_beat",
+            data=np.asarray([0.5, 0.5], dtype=np.float32),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_extract_waveform_fixture_gui.py

@@ -0,0 +1,90 @@
+"""Tests for the executable's non-visual extraction job."""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+import h5py
+import numpy as np
+
+
+TOOLS_DIR = Path(__file__).resolve().parents[1] / "tools"
+if str(TOOLS_DIR) not in sys.path:
+    sys.path.insert(0, str(TOOLS_DIR))
+SCRIPT_PATH = TOOLS_DIR / "extract_waveform_fixture_gui.py"
+SPEC = importlib.util.spec_from_file_location("extract_waveform_fixture_gui", SCRIPT_PATH)
+if SPEC is None or SPEC.loader is None:
+    raise RuntimeError(f"Could not load {SCRIPT_PATH}")
+gui = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = gui
+SPEC.loader.exec_module(gui)
+
+
+class WaveformFixtureGuiJobTests(unittest.TestCase):
+    def test_validation_does_not_write_fixture(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            source_path = root / "source.h5"
+            output_dir = root / "fixtures"
+            _write_minimal_source(source_path)
+            messages: list[str] = []
+
+            summary = gui.run_extraction_job(
+                source_path,
+                output_dir,
+                recursive=False,
+                dry_run=True,
+                log=messages.append,
+            )
+
+            self.assertEqual(summary.succeeded, 1)
+            self.assertEqual(summary.failed, 0)
+            self.assertFalse(output_dir.exists())
+            self.assertTrue(any("compatible" in message for message in messages))
+
+    def test_extraction_job_creates_fixture(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            source_path = root / "source.h5"
+            output_dir = root / "fixtures"
+            _write_minimal_source(source_path)
+
+            summary = gui.run_extraction_job(
+                source_path,
+                output_dir,
+                recursive=False,
+                dry_run=False,
+                log=lambda _message: None,
+            )
+
+            self.assertEqual(summary.created, 1)
+            self.assertEqual(summary.failed, 0)
+            self.assertEqual(len(list(output_dir.glob("waveform_fixture_*.h5"))), 1)
+
+
+def _write_minimal_source(path: Path) -> None:
+    with h5py.File(path, "w") as source:
+        source.create_dataset(
+            "analysis/retinal_artery_velocity_signal",
+            data=np.arange(12, dtype=np.float32),
+        )
+        source.create_dataset(
+            "analysis/retinal_vein_velocity_signal",
+            data=np.arange(12, dtype=np.float32) + 20,
+        )
+        source.create_dataset(
+            "analysis/beat_indices",
+            data=np.asarray([0, 5, 10], dtype=np.int32),
+        )
+        source.create_dataset(
+            "analysis/time_per_beat",
+            data=np.asarray([0.5, 0.5], dtype=np.float32),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()