add the option of accessing previous pipelines results in the h5file

Author: MaxBA1602

src/eye_flow.py

@@ -117,6 +117,45 @@ def get(self, key: str, default=None):
         return default
 
 
+class _EyeFlowView:
+    def __init__(self, work_h5: h5py.File) -> None:
+        self.work_h5 = work_h5
+
+    def _pipeline_group_names(self) -> list[str]:
+        eye_flow_group = self.work_h5.get("EyeFlow")
+        if not isinstance(eye_flow_group, h5py.Group):
+            return []
+        return list(eye_flow_group.keys())
+
+    def get(self, key: str, default=None):
+        normalized_key = _normalize_h5_lookup_path(key)
+        if not normalized_key:
+            return default
+
+        explicit = self.work_h5.get(f"EyeFlow/{normalized_key}")
+        if explicit is not None:
+            return explicit
+
+        for pipeline_group_name in reversed(self._pipeline_group_names()):
+            candidate = self.work_h5.get(
+                f"EyeFlow/{pipeline_group_name}/{normalized_key}"
+            )
+            if candidate is not None:
+                return candidate
+        return default
+
+    def __getitem__(self, key: str):
+        found = self.get(key)
+        if found is None:
+            raise KeyError(key)
+        return found
+
+    def __contains__(self, key: object) -> bool:
+        if not isinstance(key, str):
+            return False
+        return self.get(key) is not None
+
+
 class _PipelineInputView:
     def __init__(
         self,
@@ -132,6 +171,7 @@ def __init__(
         self.work = work_h5
         self.hd = holodoppler_h5
         self.dv = doppler_vision_h5
+        self.ef = _EyeFlowView(work_h5)
         self.preferred_input = preferred_input
         self.attrs = _MergedAttrs(
             self.work_h5,

src/pipelines/dual_input_tutorial.py

@@ -15,6 +15,7 @@ class DualInputTutorial(ProcessPipeline):
     - `h5file.hd`: the holodoppler input handle
     - `h5file.dv`: the doppler vision input handle
     - `h5file.work`: the current EyeFlow output/work file
+    - `h5file.ef`: previous EyeFlow pipeline outputs, searchable via `h5file.ef["path"]`
     """
 
     description = "Tutorial: read HD and DV inputs simultaneously in one pipeline."

test/test_dual_input_tutorial.py

@@ -92,6 +92,55 @@ def test_pipeline_reads_hd_and_dv_inputs_together(self) -> None:
                 result.attrs["dv_example_path"],
             )
 
+    def test_eye_flow_proxy_resolves_previous_pipeline_outputs(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            tmp_path = Path(tmp_dir)
+            hd_path = tmp_path / "sample_holodoppler.h5"
+            dv_path = tmp_path / "sample_doppler_vision.h5"
+            work_path = tmp_path / "work.h5"
+
+            with h5py.File(hd_path, "w") as hd_h5:
+                hd_h5.create_dataset("moment0", data=np.array([1.0], dtype=float))
+
+            with h5py.File(dv_path, "w") as dv_h5:
+                dv_h5.create_dataset("Meta/example", data=np.array([2.0], dtype=float))
+
+            with h5py.File(work_path, "w") as work_h5:
+                eye_flow = work_h5.create_group("EyeFlow")
+                previous = eye_flow.create_group("first_pipeline")
+                previous.create_dataset(
+                    "summary/value",
+                    data=np.array([11.0], dtype=float),
+                )
+                latest = eye_flow.create_group("second_pipeline")
+                latest.create_dataset(
+                    "summary/value",
+                    data=np.array([22.0], dtype=float),
+                )
+                latest.create_dataset(
+                    "other/path",
+                    data=np.array([33.0], dtype=float),
+                )
+
+            with (
+                h5py.File(work_path, "r") as work_h5,
+                h5py.File(hd_path, "r") as hd_h5,
+                h5py.File(dv_path, "r") as dv_h5,
+            ):
+                input_view = _PipelineInputView(
+                    work_h5=work_h5,
+                    holodoppler_h5=hd_h5,
+                    doppler_vision_h5=dv_h5,
+                )
+
+                latest_value = np.asarray(input_view.ef["summary/value"])
+                explicit_value = np.asarray(
+                    input_view.ef["second_pipeline/other/path"]
+                )
+
+            np.testing.assert_allclose(latest_value, np.array([22.0], dtype=float))
+            np.testing.assert_allclose(explicit_value, np.array([33.0], dtype=float))
+
 
 if __name__ == "__main__":
     unittest.main()