Merge pull request #9 from dmandache/fix/resume-when-finished

Fix/resume-when-finished

Author: dmandache

src/imperandi/extract/phase.py

@@ -192,8 +192,6 @@ def extract_phase_volumes(
 
 
 def main(args: argparse.Namespace) -> None:
-    phase_extractor = _load_phase_extractor()
-
     output_path = Path(args.csv_path_out)
     error_path = Path(args.error_csv_path)
     exclude_hash_args = {
@@ -217,8 +215,17 @@ def main(args: argparse.Namespace) -> None:
     paths = resume_ctx["paths"]
     state = resume_ctx["state"]
     can_resume = resume_ctx["can_resume"]
+    already_finished = resume_ctx["already_finished"]
     ckpt = CheckpointManager(paths=paths, config=resume_ctx["config"])
 
+    if already_finished:
+        logger.info(
+            "Resume enabled and matching phase run already finished; skipping execution."
+        )
+        return
+
+    phase_extractor = _load_phase_extractor()
+
     if can_resume and paths.main_checkpoint_path.exists():
         logger.info("Resuming phase from checkpoint: %s", paths.main_checkpoint_path)
         df = pd.read_csv(paths.main_checkpoint_path).copy()

src/imperandi/extract/radiomics.py

@@ -443,9 +443,6 @@ def extract_radiomics_from_dataframe(
 
 
 def main(args: argparse.Namespace) -> None:
-    sitk_module, featureextractor_module = _load_radiomics_dependencies()
-    extractor = _create_radiomics_extractor(featureextractor_module, DEFAULT_SETTINGS)
-
     output_path = Path(args.csv_path_out)
     error_path = Path(args.error_csv_path)
     exclude_hash_args = {
@@ -469,8 +466,18 @@ def main(args: argparse.Namespace) -> None:
     paths = resume_ctx["paths"]
     state = resume_ctx["state"]
     can_resume = resume_ctx["can_resume"]
+    already_finished = resume_ctx["already_finished"]
     ckpt = CheckpointManager(paths=paths, config=resume_ctx["config"])
 
+    if already_finished:
+        logger.info(
+            "Resume enabled and matching radiomics run already finished; skipping execution."
+        )
+        return
+
+    sitk_module, featureextractor_module = _load_radiomics_dependencies()
+    extractor = _create_radiomics_extractor(featureextractor_module, DEFAULT_SETTINGS)
+
     if can_resume and paths.main_checkpoint_path.exists():
         logger.info("Resuming radiomics from checkpoint: %s", paths.main_checkpoint_path)
         df = pd.read_csv(paths.main_checkpoint_path).copy()

src/imperandi/ingest/parse.py

@@ -842,8 +842,21 @@ def process_with_checkpoint(
     paths = resume_ctx["paths"]
     state = resume_ctx["state"]
     can_resume = resume_ctx["can_resume"]
+    already_finished = resume_ctx["already_finished"]
     ckpt = CheckpointManager(paths=paths, config=resume_ctx["config"])
 
+    if already_finished:
+        logger.info(
+            "Resume enabled and matching parse run already finished; skipping execution."
+        )
+        if output_path.exists():
+            return pd.read_csv(output_path)
+        if paths.main_checkpoint_path.exists():
+            return pd.read_csv(paths.main_checkpoint_path).drop(
+                columns=["_source_idx"], errors="ignore"
+            )
+        return df_paths.copy()
+
     if can_resume and paths.main_checkpoint_path.exists():
         logger.info("Resuming parse from checkpoint: %s", paths.main_checkpoint_path)
         df = pd.read_csv(paths.main_checkpoint_path).copy()

src/imperandi/process/convert.py

@@ -536,8 +536,15 @@ def main(args):
     paths = resume_ctx["paths"]
     state = resume_ctx["state"]
     can_resume = resume_ctx["can_resume"]
+    already_finished = resume_ctx["already_finished"]
     ckpt = CheckpointManager(paths=paths, config=resume_ctx["config"])
 
+    if already_finished:
+        logger.info(
+            "Resume enabled and matching convert run already finished; skipping execution."
+        )
+        return
+
     if args.verbose:
         for p in args.csv_path:
             check_file(p)

src/imperandi/process/segment.py

@@ -481,6 +481,7 @@ def segment_volume(
 ) -> List[str]:
     """Run segmentation tasks and optional post‐processing."""
     warnings: List[str] = []
+    ran_any_task = False
     tasks = tasks_config.get("tasks", [])
     if not tasks:
         raise ValueError("No tasks provided in config")
@@ -524,6 +525,7 @@ def segment_volume(
                 task=task_name,
                 **extra,
             )
+            ran_any_task = True
         except Exception as exc:
             logger.error(
                 "Segmentation failed on %s (%s): %s", nifti_path, task_name, exc
@@ -553,6 +555,13 @@ def segment_volume(
     merged_output = str(postprocess.get("output", "merged")).strip() or "merged"
     merged_name = _output_to_filename(merged_output)
     dst = output_dir / merged_name
+    if dst.exists() and not force and not ran_any_task:
+        if verbose:
+            logger.info(
+                "Skip postprocess – output exists and row already has task outputs: %s",
+                dst,
+            )
+        return warnings
     if dst.exists():
         warnings.append(
             f"Postprocess output will overwrite existing file and continue: {dst}"
@@ -800,12 +809,6 @@ def normalize_segment_args(args: argparse.Namespace) -> argparse.Namespace:
 
 def main(args: argparse.Namespace) -> None:
     setup_logging(verbose=getattr(args, "verbose", False))
-    tasks_config = load_segmentation_config(
-        getattr(args, "manifest", None),
-        base_path=Path(__file__).resolve().parents[1],
-    )
-    prefetch_totalsegmentator_models(tasks_config)
-
     output_path = Path(args.csv_path_out)
     error_path = Path(args.error_csv_path)
 
@@ -830,8 +833,21 @@ def main(args: argparse.Namespace) -> None:
     paths = resume_ctx["paths"]
     state = resume_ctx["state"]
     can_resume = resume_ctx["can_resume"]
+    already_finished = resume_ctx["already_finished"]
     ckpt = CheckpointManager(paths=paths, config=resume_ctx["config"])
 
+    if already_finished:
+        logger.info(
+            "Resume enabled and matching segment run already finished; skipping execution."
+        )
+        return
+
+    tasks_config = load_segmentation_config(
+        getattr(args, "manifest", None),
+        base_path=Path(__file__).resolve().parents[1],
+    )
+    prefetch_totalsegmentator_models(tasks_config)
+
     from imperandi.utils.multiprocessing import (
         apply_strategy_env,
         strategy_to_log_dict,

src/imperandi/utils/run_state.py

@@ -229,10 +229,12 @@ def prepare_resume_context(
         args_hash=args_hash,
         input_fingerprint=input_fp,
     )
+    already_finished = can_resume and bool((state or {}).get("finished"))
     return {
         "paths": paths,
         "state": state,
         "can_resume": can_resume,
+        "already_finished": already_finished,
         "config": CheckpointConfig(
             command=command,
             args_hash=args_hash,

tests/test_convert.py

@@ -286,4 +286,4 @@ def fake_convert(work_df, output_dir, verbose, num_workers, on_result):
 
     args.resume = True
     convert_module.main(args)
-    assert work_sizes[-1] == 0
+    assert work_sizes == [1]

tests/test_parse.py

@@ -534,7 +534,7 @@ def update(self, amount):
     assert sum(recorded["updates"]) == 5
 
 
-def test_process_with_checkpoint_counts_resumed_rows_in_progress(
+def test_process_with_checkpoint_skips_when_matching_run_already_finished(
     tmp_path, monkeypatch
 ):
     monkeypatch.setattr(pd.Series, "parallel_apply", pd.Series.apply, raising=False)
@@ -580,8 +580,8 @@ def update(self, amount):
     )
 
     assert calls == []
-    assert recorded["updates"] == [2, 1, 3]
-    assert sum(recorded["updates"]) == 6
+    assert recorded["updates"] == [2, 1]
+    assert sum(recorded["updates"]) == 3
     assert len(out) == 3
 
 

tests/test_phase.py

@@ -115,12 +115,17 @@ def test_main_resume_skips_completed_rows(tmp_path, monkeypatch):
     pd.DataFrame([{"nifti_path": str(nifti)}]).to_csv(csv_path, index=False)
 
     calls = {"count": 0}
+    extractor_loads = {"count": 0}
 
     def fake_process_single_volume(idx, row, *, phase_extractor, verbose=False):
         calls["count"] += 1
         return idx, {"totalseg_phase": "portal"}, None
 
-    monkeypatch.setattr(phase_module, "_load_phase_extractor", lambda: (lambda _: {}))
+    def fake_load_phase_extractor():
+        extractor_loads["count"] += 1
+        return lambda _: {}
+
+    monkeypatch.setattr(phase_module, "_load_phase_extractor", fake_load_phase_extractor)
     monkeypatch.setattr(phase_module, "process_single_volume", fake_process_single_volume)
 
     args = argparse.Namespace(
@@ -140,3 +145,4 @@ def fake_process_single_volume(idx, row, *, phase_extractor, verbose=False):
     args.resume = True
     phase_module.main(args)
     assert calls["count"] == 0
+    assert extractor_loads["count"] == 1

tests/test_radiomics.py

@@ -125,12 +125,13 @@ def test_main_resume_skips_completed_rows(tmp_path, monkeypatch):
     ).to_csv(csv_path, index=False)
 
     calls = {"count": 0}
+    dep_calls = {"count": 0}
 
-    monkeypatch.setattr(
-        radiomics_module,
-        "_load_radiomics_dependencies",
-        lambda: (object(), object()),
-    )
+    def fake_load_deps():
+        dep_calls["count"] += 1
+        return object(), object()
+
+    monkeypatch.setattr(radiomics_module, "_load_radiomics_dependencies", fake_load_deps)
     monkeypatch.setattr(
         radiomics_module,
         "_create_radiomics_extractor",
@@ -162,3 +163,4 @@ def fake_liver(*args, **kwargs):
     args.resume = True
     radiomics_module.main(args)
     assert calls["count"] == 0
+    assert dep_calls["count"] == 1