Merge pull request #13 from DigitalHolography/12-pipeline-possible-danger-of-contamination

feature/Added_the_PipelineDescriptor_dataclass

Author: Drdaag

src/angio_eye.py

@@ -15,6 +15,7 @@
     sv_ttk = None
 
 from pipelines import (
+    PipelineDescriptor,
     ProcessPipeline,
     ProcessResult,
     load_pipeline_catalog,
@@ -70,7 +71,7 @@ def __init__(self) -> None:
         self.title("HDF5 Process")
         self.geometry("800x600")
         self.h5_file: h5py.File | None = None
-        self.pipeline_registry: dict[str, ProcessPipeline] = {}
+        self.pipeline_registry: dict[str, PipelineDescriptor] = {}
         self.pipeline_check_vars: dict[str, tk.BooleanVar] = {}
         self.last_process_result: ProcessResult | None = None
         self.last_process_pipeline: ProcessPipeline | None = None
@@ -329,12 +330,12 @@ def _register_pipelines(self) -> None:
         self._populate_pipeline_checks(available, missing)
 
     def _populate_pipeline_checks(
-        self, available: list[ProcessPipeline], missing: list[ProcessPipeline]
+        self, available: list[PipelineDescriptor], missing: list[PipelineDescriptor]
     ) -> None:
         for child in self.pipeline_checks_inner.winfo_children():
             child.destroy()
         self.pipeline_check_vars = {}
-        rows: list[ProcessPipeline] = [*available, *missing]
+        rows: list[PipelineDescriptor] = [*available, *missing]
         for idx, pipeline in enumerate(rows):
             is_available = getattr(pipeline, "available", True)
             var = tk.BooleanVar(value=is_available)
@@ -422,8 +423,8 @@ def run_selected_pipeline(self) -> None:
                 "Missing pipeline", "Select a pipeline before running."
             )
             return
-        pipeline = self.pipeline_registry.get(name)
-        if pipeline is None:
+        pipeline_desc = self.pipeline_registry.get(name)
+        if pipeline_desc is None:
             messagebox.showerror(
                 "Pipeline missing", f"Pipeline '{name}' is not registered."
             )
@@ -432,6 +433,7 @@ def run_selected_pipeline(self) -> None:
             messagebox.showwarning("Missing file", "Load a .h5 file first.")
             return
         try:
+            pipeline = pipeline_desc.instantiate()
             result = pipeline.run(self.h5_file)
         except Exception as exc:  # noqa: BLE001
             messagebox.showerror("Pipeline error", f"Pipeline failed: {exc}")
@@ -522,7 +524,7 @@ def run_batch(self) -> None:
             )
             return
 
-        pipelines: list[ProcessPipeline] = []
+        pipelines: list[PipelineDescriptor] = []
         missing: list[str] = []
         for name in selected_names:
             pipeline = self.pipeline_registry.get(name)
@@ -646,15 +648,16 @@ def _safe_pipeline_suffix(self, name: str) -> str:
     def _run_pipelines_on_file(
         self,
         h5_path: Path,
-        pipelines: Sequence[ProcessPipeline],
+        pipelines: Sequence[PipelineDescriptor],
         output_root: Path,
     ) -> None:
         data_dir = output_root / h5_path.stem
         data_dir.mkdir(parents=True, exist_ok=True)
         combined_h5_out = data_dir / f"{h5_path.stem}_pipelines_result.h5"
         pipeline_results: list[tuple[str, ProcessResult]] = []
         with h5py.File(h5_path, "r") as h5file:
-            for pipeline in pipelines:
+            for pipeline_desc in pipelines:
+                pipeline = pipeline_desc.instantiate()
                 result = pipeline.run(h5file)
                 pipeline_results.append((pipeline.name, result))
                 self._log_batch(f"[OK] {h5_path.name} -> {pipeline.name}")

src/cli.py

@@ -19,25 +19,30 @@
 import sys
 import tempfile
 import zipfile
-from collections.abc import Iterable, Sequence
+from collections.abc import Sequence
 from pathlib import Path
 
 import h5py
 
-from pipelines import ProcessPipeline, ProcessResult, load_all_pipelines
+from pipelines import (
+    PipelineDescriptor,
+    ProcessResult,
+    load_pipeline_catalog,
+)
 from pipelines.core.utils import write_combined_results_h5
 
 
-def _build_pipeline_registry() -> dict[str, ProcessPipeline]:
-    pipelines = load_all_pipelines()
-    return {p.name: p for p in pipelines}
+def _build_pipeline_registry() -> dict[str, PipelineDescriptor]:
+    available, _ = load_pipeline_catalog()
+    # pipelines = load_all_pipelines()
+    return {p.name: p for p in available}
 
 
 def _load_pipeline_list(
-    path: Path, registry: dict[str, ProcessPipeline]
-) -> list[ProcessPipeline]:
+    path: Path, registry: dict[str, PipelineDescriptor]
+) -> list[PipelineDescriptor]:
     raw_lines = path.read_text(encoding="utf-8").splitlines()
-    selected: list[ProcessPipeline] = []
+    selected: list[PipelineDescriptor] = []
     missing: list[str] = []
     for line in raw_lines:
         name = line.strip()
@@ -92,7 +97,7 @@ def _prepare_data_root(
 
 def _run_pipelines_on_file(
     h5_path: Path,
-    pipelines: Sequence[ProcessPipeline],
+    pipelines: Sequence[PipelineDescriptor],
     output_root: Path,
 ) -> list[Path]:
     outputs: list[Path] = []
@@ -101,7 +106,8 @@ def _run_pipelines_on_file(
     combined_h5_out = data_dir / f"{h5_path.stem}_pipelines_result.h5"
     pipeline_results: list[tuple[str, ProcessResult]] = []
     with h5py.File(h5_path, "r") as h5file:
-        for pipeline in pipelines:
+        for pipeline_desc in pipelines:
+            pipeline = pipeline_desc.instantiate()
             result = pipeline.run(h5file)
             pipeline_results.append((pipeline.name, result))
             print(f"[OK] {h5_path.name} -> {pipeline.name}")
@@ -200,7 +206,7 @@ def run_cli(
             shutil.rmtree(work_tempdir_path, ignore_errors=True)
 
 
-def main(argv: Iterable[str] | None = None) -> int:
+def main(argv: Sequence[str] | None = None) -> int:
     parser = argparse.ArgumentParser(
         description="Run AngioEye pipelines over a folder of HDF5 files."
     )

src/pipelines/__init__.py

@@ -4,35 +4,16 @@
 import pkgutil
 
 # import inspect
-from .core.base import PIPELINE_REGISTRY, ProcessPipeline, ProcessResult
+from .core.base import (
+    PIPELINE_REGISTRY,
+    MissingPipeline,
+    PipelineDescriptor,
+    ProcessPipeline,
+    ProcessResult,
+)
 from .core.utils import write_combined_results_h5, write_result_h5
 
 
-class MissingPipeline(ProcessPipeline):
-    """Placeholder for pipelines whose dependencies are missing."""
-
-    available = False
-    missing_deps: list[str]
-    requires: list[str]
-
-    def __init__(
-        self, name: str, description: str, missing_deps: list[str], requires: list[str]
-    ) -> None:
-        super().__init__()
-        self.name = name
-        self.description = description or "Pipeline unavailable (missing dependencies)."
-        self.missing_deps = missing_deps
-        self.requires = requires
-
-    def run(self, h5file):
-        missing = ", ".join(
-            self.missing_deps or self.requires or ["unknown dependency"]
-        )
-        raise ImportError(
-            f"Pipeline '{self.name}' unavailable. Missing dependencies: {missing}"
-        )
-
-
 def _module_docstring(module_name: str) -> str:
     spec = importlib.util.find_spec(module_name)
     if not spec or not spec.origin:
@@ -96,9 +77,9 @@ def _missing_requirements(requires: list[str]) -> list[str]:
     return missing
 
 
-def _discover_pipelines() -> tuple[list[ProcessPipeline], list[MissingPipeline]]:
-    available: list[ProcessPipeline] = []
-    missing: list[MissingPipeline] = []
+def _discover_pipelines() -> tuple[list[PipelineDescriptor], list[PipelineDescriptor]]:
+    available: list[PipelineDescriptor] = []
+    missing: list[PipelineDescriptor] = []
     # seen_classes = set()
 
     for module_info in pkgutil.iter_modules(__path__):
@@ -124,28 +105,32 @@ def _discover_pipelines() -> tuple[list[ProcessPipeline], list[MissingPipeline]]
         except Exception as e:
             # Fallback for unknown failures (SyntaxError, etc.)
             missing.append(
-                MissingPipeline(
-                    module_info.name, f"Error: {e}", ["Unknown"], ["Unknown"]
+                PipelineDescriptor(
+                    name=module_info.name,
+                    description=f"Import Error: {e}",
+                    available=False,
+                    error_msg=str(e),
                 )
             )
 
     for _name, cls in PIPELINE_REGISTRY.items():
+        desc = PipelineDescriptor(
+            name=cls.name,
+            description=cls.description,
+            available=cls.available,
+            requires=cls.requires,
+            missing_deps=cls.missing_deps,
+            pipeline_cls=cls,
+        )
         if getattr(cls, "is_available", True):
-            inst = cls()
+            # inst = cls()
             # The GUI needs thoses values
-            inst.name = cls.name
-            inst.available = True
-            inst.requires = cls.required_deps
-            available.append(inst)
+            # inst.name = cls.name
+            # inst.available = True
+            # inst.requires = cls.required_deps
+            available.append(desc)
         else:
-            missing.append(
-                MissingPipeline(
-                    name=getattr(cls, "name", cls.__name__),
-                    description=getattr(cls, "description", ""),
-                    missing_deps=getattr(cls, "missing_deps", []),
-                    requires=getattr(cls, "required_deps", []),
-                )
-            )
+            missing.append(desc)
 
         # except ImportError as exc:
         #     # Capture missing dependency if ModuleNotFoundError has a name.
@@ -192,15 +177,23 @@ def _discover_pipelines() -> tuple[list[ProcessPipeline], list[MissingPipeline]]
     return available, missing
 
 
-def load_all_pipelines(include_missing: bool = False) -> list[ProcessPipeline]:
-    """
-    Discover and instantiate pipelines. Optionally include placeholders for missing deps.
-    """
-    available, missing = _discover_pipelines()
-    return available + missing if include_missing else available
+# def load_all_pipelines(
+#     include_missing: bool = False,
+# ) -> list[type[ProcessPipeline] | MissingPipeline]:
+#     """
+#     Discover  pipelines. Optionally include placeholders for missing deps.
+#     """
+#     available, missing = _discover_pipelines()
+#     # Cast to a common list type for the type checker
+#     combined: list[type[ProcessPipeline] | MissingPipeline] = list(available)
+#     if include_missing:
+#         combined.extend(missing)
+#     return combined
 
 
-def load_pipeline_catalog() -> tuple[list[ProcessPipeline], list[MissingPipeline]]:
+def load_pipeline_catalog() -> tuple[
+    list[PipelineDescriptor], list[PipelineDescriptor]
+]:
     """Return (available, missing) pipelines for UI/CLI surfaces."""
     return _discover_pipelines()
 
@@ -216,7 +209,7 @@ def load_pipeline_catalog() -> tuple[list[ProcessPipeline], list[MissingPipeline
     "ProcessResult",
     "write_result_h5",
     "write_combined_results_h5",
-    "load_all_pipelines",
+    # "load_all_pipelines",
     "load_pipeline_catalog",
     "MissingPipeline",
     *[_cls.__name__ for _cls in (p.__class__ for p in _AVAILABLE)],

src/pipelines/core/__init__.py

@@ -1,4 +1,4 @@
-from .base import ProcessPipeline, ProcessResult
+from .base import MissingPipeline, ProcessPipeline, ProcessResult
 from .utils import (
     safe_h5_key,
     write_combined_results_h5,
@@ -7,6 +7,7 @@
 
 __all__ = [
     "ProcessPipeline",
+    "MissingPipeline",
     "ProcessResult",
     "safe_h5_key",
     "write_result_h5",