Merge pull request #14 from DigitalHolography/improve-imports-global

feature/Improve_imports_global

Author: MaxBA1602

README.md

@@ -106,12 +106,16 @@ To see more complete examples, check out `src/pipelines/basic_stats.py` and `src
 ### Simple Pipeline Structure
 
 ```python
-from pipelines import ProcessPipeline, ProcessResult
+from pipelines import ProcessPipeline, ProcessResult, registerPipeline
 
+@registerPipeline(
+    name="My Analysis",
+    description="Calculates a custom clinical metric.",
+    required_deps=["torch>=2.2"],
+)
 class MyAnalysis(ProcessPipeline):
-    description = "Calculates a custom clinical metric."
-
     def run(self, h5file):
+        import torch
         # 1. Read data using h5py
         # 2. Perform calculations
         # 3. Return metrics and artifacts

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."
     )