mock eyeflow run + structural changes using PyDantic

Author: AdrienGordon

pyproject.toml

@@ -15,6 +15,7 @@ license = { text = "GPL-3.0-only" }
 dependencies = [
     "numpy>=1.24",
     "h5py>=3.9",
+    "pydantic>=2.6",
     "scipy>=1.10",
     "scikit-image>=0.20",
     "joblib>=1.3",

src/cli.py

@@ -23,6 +23,10 @@
 from collections.abc import Callable, Sequence
 from pathlib import Path
 
+from runtime_limits import configure_numeric_threads
+
+configure_numeric_threads()
+
 import h5py
 
 from pipelines import (
@@ -31,7 +35,7 @@
     load_pipeline_catalog,
 )
 from pipelines.core.errors import format_pipeline_exception
-from utils.io import write_combined_results_h5
+from input_output import write_combined_results_h5
 
 
 def _build_pipeline_registry() -> dict[str, PipelineDescriptor]:

src/domain/steps/vessel_velocity_estimator.py

@@ -6,6 +6,8 @@
 
 import numpy as np
 
+from runtime_limits import cap_parallel_jobs
+
 from ._masks import elliptical_mask
 from .base import DomainStep as BaseStep
 
@@ -56,15 +58,15 @@ def run(self, ctx):
         disk, dilation, inpaint = _skimage_dependencies()
         mask = dilation(vessel_mask, disk(local_background_dist)) #TODO add parameter
 
-        n_jobs = _cpu_count() #TODO add parameter for number of parallel jobs
+        n_jobs = cap_parallel_jobs(_cpu_count())
 
         print(f"    - Inpainting fRMS with {n_jobs} parallel jobs")
 
         def _inpaint_frame(frame, mask):
             return inpaint.inpaint_biharmonic(frame, mask)
 
         fRMSbkg = _run_in_parallel(
-            partial(_inpaint_frame, mask=mask), fRMS, n_jobs=-1, chunking=False
+            partial(_inpaint_frame, mask=mask), fRMS, n_jobs=n_jobs, chunking=False
         )
 
         # fRMSbkg = np.stack(np.array([inpaint.inpaint_biharmonic(frame, mask) for frame in fRMS]), axis=0)
@@ -130,6 +132,7 @@ def _run_in_parallel(func, iterable, n_jobs=-1, chunking=False):
         return np.stack([func(item) for item in iterable], axis=0)
     if n_jobs == -1:
         n_jobs = joblib.cpu_count()
+    n_jobs = cap_parallel_jobs(n_jobs)
     results = joblib.Parallel(n_jobs=n_jobs, backend="threading")(
         joblib.delayed(func)(item) for item in iterable
     )

src/eye_flow.py

@@ -14,6 +14,10 @@
 from pathlib import Path
 from tkinter import filedialog, messagebox, ttk
 
+from runtime_limits import configure_numeric_threads
+
+configure_numeric_threads()
+
 from app_settings import (
     LAST_BATCH_LOG_FILENAME,
     AppSettingsStore,
@@ -34,14 +38,13 @@
 
 from pipelines import PipelineDescriptor, load_pipeline_catalog
 from pipelines.core.errors import format_pipeline_exception
-from utils.io import (
+from input_output import (
+    EyeFlowOutputManager,
     HOLO_SUFFIX,
     PipelineInputView,
     ResolvedHoloInput,
-    append_result_group,
     create_zip_from_tree,
     holo_input_status,
-    initialize_output_h5,
     open_h5,
     reset_output_dir,
     resolve_selected_holo_inputs,
@@ -2026,8 +2029,8 @@ def _run_pipelines_to_output(
                 if doppler_vision_h5 is not None
                 else None
             )
-            initialize_output_h5(
-                work_h5,
+            output_manager = EyeFlowOutputManager(work_h5)
+            output_manager.initialize(
                 holodoppler_source_file=(
                     str(holodoppler_h5) if holodoppler_h5 is not None else None
                 ),
@@ -2051,7 +2054,7 @@ def _run_pipelines_to_output(
                     raise RuntimeError(
                         format_pipeline_exception(exc, pipeline)
                     ) from exc
-                append_result_group(work_h5, pipeline.name, result)
+                output_manager.append_pipeline_result(pipeline.name, result)
                 result.output_h5_path = str(output_h5_path)
                 self._log_batch(f"[OK] {pipeline.name}")
                 self._advance_progress()

src/utils/io/__init__.py src/input_output/__init__.pysrc/utils/io/__init__.py renamed to src/input_output/__init__.py

@@ -32,14 +32,15 @@
     write_result_h5,
     write_value_dataset,
 )
-from .output_writer import (
-    is_angioeye_output_key,
-    is_root_mirrored_output_key,
+from .output_manager import (
+    EyeFlowOutputManager,
     pack_dopplerview_analysis_outputs,
     pack_velocity_per_beat_outputs,
     systolic_index_base_for_path,
 )
 from .schema import (
+    DOPPLER_VIEW_ANALYSIS_SCHEMA,
+    DOPPLER_VIEW_SCHEMA,
     DOPPLERVIEW_ANALYSIS_ROOT,
     DOPPLERVIEW_ARTERIAL_VELOCITY_SIGNAL_PATH,
     DOPPLERVIEW_BEAT_INDICES_PATH,
@@ -58,38 +59,30 @@
     DV_H5_LAYOUT,
     DV_RETINAL_ARTERY_MASK_PATH,
     DV_RETINAL_VEIN_MASK_PATH,
-    EYE_FLOW_POSTPROCESS_ROOT,
-    EYE_FLOW_PROCESSING_ROOT,
-    EYE_FLOW_ROOT,
     HD_BATCH_STRIDE_KEY,
     HD_CONFIG_DIR_NAME,
     HD_CONFIG_FILENAME,
     HD_H5_LAYOUT,
     HD_MOMENT0_PATH,
     HD_MOMENT2_PATH,
     HD_SAMPLING_FREQ_KEY,
+    H5DatasetSpec,
+    H5SourceSchema,
     HDF5_SUFFIXES,
+    HOLODOPPLER_SCHEMA,
     HOLO_COMPANION_H5_LAYOUTS,
     HOLO_DATA_DIR_TEMPLATE,
     HOLO_H5_SUBDIR,
     HOLO_SUFFIX,
-    LEGACY_PIPELINES_ROOT,
     HoloCompanionH5Layout,
-    find_pipeline_group,
-    find_postprocess_group,
-    get_postprocess_root,
-    get_processing_root,
+    JsonConfigValueSpec,
     iter_metric_datasets,
-    pipeline_path_candidates,
-    postprocess_path_candidates,
 )
-from .sources import (
+from .inputs import (
     EyeFlowView,
+    HoloInputStatus,
     MergedAttrs,
     PipelineInputView,
-)
-from .resolvers import (
-    HoloInputStatus,
     ResolvedHoloInput,
     holo_input_status,
     resolve_holo_input,
@@ -106,8 +99,11 @@
     "HOLO_SUFFIX",
     "MetricsTree",
     "EyeFlowView",
+    "EyeFlowOutputManager",
     "HoloCompanionH5Layout",
     "HoloInputStatus",
+    "DOPPLER_VIEW_ANALYSIS_SCHEMA",
+    "DOPPLER_VIEW_SCHEMA",
     "DOPPLERVIEW_ANALYSIS_ROOT",
     "DOPPLERVIEW_ARTERIAL_VELOCITY_SIGNAL_PATH",
     "DOPPLERVIEW_BEAT_INDICES_PATH",
@@ -131,6 +127,10 @@
     "HD_MOMENT0_PATH",
     "HD_MOMENT2_PATH",
     "HD_SAMPLING_FREQ_KEY",
+    "H5DatasetSpec",
+    "H5SourceSchema",
+    "HOLODOPPLER_SCHEMA",
+    "JsonConfigValueSpec",
     "MergedAttrs",
     "PipelineInputView",
     "ResolvedHoloInput",
@@ -145,18 +145,12 @@
     "extracted_zip_tree",
     "find_child_group_by_attr",
     "find_first_existing_path",
-    "get_postprocess_root",
-    "get_processing_root",
     "holo_input_status",
     "initialize_output_h5",
     "iter_metric_datasets",
-    "is_angioeye_output_key",
-    "is_root_mirrored_output_key",
     "open_h5",
     "pack_velocity_per_beat_outputs",
     "pack_dopplerview_analysis_outputs",
-    "pipeline_path_candidates",
-    "postprocess_path_candidates",
     "read_array",
     "read_dataset",
     "replace_file_in_zip",
@@ -174,8 +168,4 @@
     "write_metrics_trees_to_h5",
     "write_result_h5",
     "write_value_dataset",
-    "EYE_FLOW_POSTPROCESS_ROOT",
-    "EYE_FLOW_PROCESSING_ROOT",
-    "EYE_FLOW_ROOT",
-    "LEGACY_PIPELINES_ROOT",
 ]

src/utils/io/archives/__init__.py src/input_output/archives/__init__.pysrc/utils/io/archives/__init__.py renamed to src/input_output/archives/__init__.py

@@ -165,9 +165,6 @@ def set_attr_safe(h5obj: h5py.File | h5py.Group | h5py.Dataset, key: str, value)
         h5obj.attrs[key] = str(value)
 
 
-DEFAULT_COMPRESSION_THRESHOLD = 1_000_000
-
-
 def _normalize_dataset_payload(data, ds_attrs):
     original_class = None
     payload = data
@@ -180,11 +177,6 @@ def _normalize_dataset_payload(data, ds_attrs):
     elif isinstance(payload, (list, tuple)):
         payload = np.asarray(payload)
 
-    if isinstance(payload, np.ndarray):
-        if payload.dtype.kind == "f" and payload.dtype.itemsize == 8:
-            payload = payload.astype(np.float32)
-            original_class = original_class or "float64"
-
     if original_class is not None:
         ds_attrs = {} if ds_attrs is None else dict(ds_attrs)
         ds_attrs.setdefault("original_class", original_class)
@@ -193,17 +185,6 @@ def _normalize_dataset_payload(data, ds_attrs):
 
 
 def _get_dataset_creation_kwargs(payload: np.ndarray) -> dict[str, object]:
-    if (
-        isinstance(payload, np.ndarray)
-        and payload.dtype.kind in "bBhHiIlLef"
-        and payload.nbytes >= DEFAULT_COMPRESSION_THRESHOLD
-        and payload.ndim > 0
-    ):
-        return {
-            "compression": "gzip",
-            "compression_opts": 6,
-            "chunks": tuple(min(s, 1024) for s in payload.shape),
-        }
     return {}
 
 
@@ -363,25 +344,18 @@ def append_result_group(
     pipeline_name: str,
     result: "ProcessResult",
 ) -> h5py.Group:
-    from pipelines.core.base import ProcessResult
-    from utils.io.output_writer import is_root_mirrored_output_key
-
-    pipelines_grp = (
-        h5file["EyeFlow"] if "EyeFlow" in h5file else h5file.create_group("EyeFlow")
-    )
-    pipeline_grp = _get_or_replace_group(pipelines_grp, safe_h5_key(pipeline_name))
-    pipeline_grp.attrs["pipeline"] = pipeline_name
+    if "EyeFlow" in h5file:
+        del h5file["EyeFlow"]
+    h5file.attrs["last_pipeline"] = pipeline_name
     if result.attrs:
         for key, value in result.attrs.items():
             if key == "pipeline":
                 continue
-            set_attr_safe(pipeline_grp, key, value)
+            set_attr_safe(h5file, key, value)
     for key, value in result.metrics.items():
-        write_value_dataset(pipeline_grp, key, value)
-        if is_root_mirrored_output_key(key):
-            write_value_dataset(h5file, key, value)
+        write_value_dataset(h5file, key, value)
     h5file.flush()
-    return pipeline_grp
+    return h5file
 
 
 def write_result_h5(