Merge branch 'feat/enhance-radiomic-exctractor'

Author: dmandache

src/imperandi/extract/phase.py

@@ -22,7 +22,7 @@
 
 logger = logging.getLogger(__name__)
 DEFAULT_CHECKPOINT_EVERY_ROWS = 50
-DEFAULT_CHECKPOINT_EVERY_SEC = 350
+DEFAULT_CHECKPOINT_EVERY_SEC = 5 * 60
 
 
 def _load_phase_extractor() -> Callable[[Any], Dict[str, Any]]:
@@ -72,6 +72,12 @@ def add_phase_arguments(
         default=None,
         help="CSV path for failed rows only (default: <csv_dir>/phase_errors.csv).",
     )
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        default=False,
+        help="Recompute phase even when `totalseg_phase` is already populated.",
+    )
     parser.add_argument("--verbose", "-v", action="store_true", help="Verbose logging")
     add_checkpoint_arguments(
         parser,
@@ -135,6 +141,14 @@ def parse_arguments() -> argparse.Namespace:
     return args
 
 
+def _has_populated_value(value: Any) -> bool:
+    if pd.isna(value):
+        return False
+    if isinstance(value, str):
+        return bool(value.strip())
+    return True
+
+
 def process_single_volume(
     idx: int,
     row: Mapping[str, Any],
@@ -212,6 +226,7 @@ def main(args: argparse.Namespace) -> None:
 
     errors_by_idx: Dict[int, Dict[str, Any]] = {}
     completed_indices: set[int] = set()
+    force = bool(getattr(args, "force", False))
     if can_resume:
         completed_indices = {
             int(i)
@@ -227,6 +242,22 @@ def main(args: argparse.Namespace) -> None:
                     except Exception:
                         pass
 
+    if not force and "totalseg_phase" in df.columns:
+        prefilled_indices = {
+            int(df.at[idx, "_source_idx"])
+            for idx in df.index
+            if _has_populated_value(df.at[idx, "totalseg_phase"])
+        }
+        newly_completed = prefilled_indices - completed_indices
+        if newly_completed:
+            completed_indices.update(prefilled_indices)
+            for source_idx in newly_completed:
+                errors_by_idx.pop(source_idx, None)
+            logger.info(
+                "Skipping %d row(s) with existing totalseg_phase (use --force to recompute)",
+                len(newly_completed),
+            )
+
     def _checkpoint_write(*, force: bool = False) -> None:
         err_df = (
             pd.DataFrame(list(errors_by_idx.values())) if errors_by_idx else pd.DataFrame()

src/imperandi/extract/radiomics.py

@@ -21,7 +21,7 @@
 
 logger = logging.getLogger(__name__)
 DEFAULT_CHECKPOINT_EVERY_ROWS = 50
-DEFAULT_CHECKPOINT_EVERY_SEC = 350
+DEFAULT_CHECKPOINT_EVERY_SEC = 5 * 60
 
 DEFAULT_SETTINGS = {
     "binWidth": 25,

src/imperandi/ingest/parse.py

@@ -43,9 +43,7 @@
 warnings.filterwarnings("ignore")
 logger = logging.getLogger(__name__)
 DEFAULT_CHECKPOINT_EVERY_ROWS = 10_000
-DEFAULT_CHECKPOINT_EVERY_SEC = 350
-PARSE_WORKER_BATCH_SIZE = 16384
-PARSE_CHECKPOINT_SCHEMA_VERSION = 1
+DEFAULT_CHECKPOINT_EVERY_SEC = 5 * 60  # 5 minutes
 
 # Make reading tolerant of non-conformant values
 config.settings.reading_validation_mode = config.IGNORE  # or config.WARN

src/imperandi/process/convert.py

@@ -31,7 +31,25 @@
 
 logger = logging.getLogger(__name__)
 DEFAULT_CHECKPOINT_EVERY_ROWS = 50
-DEFAULT_CHECKPOINT_EVERY_SEC = 350
+DEFAULT_CHECKPOINT_EVERY_SEC = 5 * 60 # 5 minutes
+
+
+def _lower_log_level_one_step(level: int) -> int:
+    if level >= logging.CRITICAL:
+        return logging.ERROR
+    if level >= logging.ERROR:
+        return logging.WARNING
+    if level >= logging.WARNING:
+        return logging.INFO
+    if level >= logging.INFO:
+        return logging.DEBUG
+    return logging.DEBUG
+
+
+def _configure_dicom2nifti_convert_logger(base_logger: logging.Logger) -> None:
+    base_level = base_logger.getEffectiveLevel()
+    dicom2nifti_level = _lower_log_level_one_step(base_level)
+    logging.getLogger("dicom2nifti.convert_dicom").setLevel(dicom2nifti_level)
 
 
 # Function to parse command-line arguments
@@ -331,6 +349,7 @@ def process_single_volume(k, row, output_dir, verbose, return_status=False):
               "converted", "skipped", or "failed".
     """
     setup_logging(verbose=verbose)
+    _configure_dicom2nifti_convert_logger(logger)
 
     def _result(export_path, error_row, status):
         if return_status:

tests/test_convert.py

@@ -3,6 +3,7 @@
 import tarfile
 import zipfile
 import argparse
+import logging
 from pathlib import Path
 
 # Ensure src/ is on sys.path for imports
@@ -29,6 +30,40 @@ def test_convert_list_str_to_list_invalid():
     assert out == s
 
 
+@pytest.mark.parametrize(
+    ("base_level", "expected_level"),
+    [
+        (logging.CRITICAL, logging.ERROR),
+        (logging.ERROR, logging.WARNING),
+        (logging.WARNING, logging.INFO),
+        (logging.INFO, logging.DEBUG),
+        (logging.DEBUG, logging.DEBUG),
+        (logging.NOTSET, logging.DEBUG),
+    ],
+)
+def test_lower_log_level_one_step(base_level, expected_level):
+    assert convert_module._lower_log_level_one_step(base_level) == expected_level
+
+
+def test_configure_dicom2nifti_convert_logger_one_level_lower():
+    base_logger = logging.getLogger("imperandi.process.convert.test_base")
+    target_logger = logging.getLogger("dicom2nifti.convert_dicom")
+    old_base_level = base_logger.level
+    old_target_level = target_logger.level
+
+    try:
+        base_logger.setLevel(logging.INFO)
+        convert_module._configure_dicom2nifti_convert_logger(base_logger)
+        assert target_logger.level == logging.DEBUG
+
+        base_logger.setLevel(logging.WARNING)
+        convert_module._configure_dicom2nifti_convert_logger(base_logger)
+        assert target_logger.level == logging.INFO
+    finally:
+        base_logger.setLevel(old_base_level)
+        target_logger.setLevel(old_target_level)
+
+
 def make_series(tmp_path, output_dir, create_nifti=False):
     # create a dummy series_dir with files
     series_dir = tmp_path / "series"

tests/test_phase.py

@@ -194,6 +194,88 @@ def fake_load_phase_extractor():
     assert extractor_loads["count"] == 1
 
 
+def test_main_skips_rows_with_existing_totalseg_phase_when_not_forced(
+    tmp_path, monkeypatch
+):
+    nifti = tmp_path / "valid.nii.gz"
+    nifti.write_text("nifti")
+    csv_path = tmp_path / "nifti_index.csv"
+    pd.DataFrame(
+        [{"nifti_path": str(nifti), "totalseg_phase": "portal"}]
+    ).to_csv(csv_path, index=False)
+
+    calls = {"count": 0}
+    monkeypatch.setattr(phase_module.nib, "load", lambda _: object())
+    monkeypatch.setattr(
+        phase_module,
+        "_load_phase_extractor",
+        lambda: (lambda _, quiet=True: {"phase": "arterial"}),
+    )
+
+    def fake_process_single_volume(idx, row, *, phase_extractor, verbose=False):
+        calls["count"] += 1
+        return idx, {"totalseg_phase": "arterial"}, None
+
+    monkeypatch.setattr(phase_module, "process_single_volume", fake_process_single_volume)
+
+    args = argparse.Namespace(
+        csv_path=str(csv_path),
+        csv_path_out=str(tmp_path / "out.csv"),
+        error_csv_path=str(tmp_path / "errors.csv"),
+        verbose=False,
+        force=False,
+        checkpoint_every_rows=1,
+        checkpoint_every_sec=3600,
+        resume=False,
+        strict_resume=False,
+    )
+    phase_module.main(args)
+
+    assert calls["count"] == 0
+    out_df = pd.read_csv(args.csv_path_out)
+    assert out_df.loc[0, "totalseg_phase"] == "portal"
+
+
+def test_main_force_recomputes_existing_totalseg_phase(tmp_path, monkeypatch):
+    nifti = tmp_path / "valid.nii.gz"
+    nifti.write_text("nifti")
+    csv_path = tmp_path / "nifti_index.csv"
+    pd.DataFrame(
+        [{"nifti_path": str(nifti), "totalseg_phase": "portal"}]
+    ).to_csv(csv_path, index=False)
+
+    calls = {"count": 0}
+    monkeypatch.setattr(phase_module.nib, "load", lambda _: object())
+    monkeypatch.setattr(
+        phase_module,
+        "_load_phase_extractor",
+        lambda: (lambda _, quiet=True: {"phase": "arterial"}),
+    )
+
+    def fake_process_single_volume(idx, row, *, phase_extractor, verbose=False):
+        calls["count"] += 1
+        return idx, {"totalseg_phase": "arterial"}, None
+
+    monkeypatch.setattr(phase_module, "process_single_volume", fake_process_single_volume)
+
+    args = argparse.Namespace(
+        csv_path=str(csv_path),
+        csv_path_out=str(tmp_path / "out.csv"),
+        error_csv_path=str(tmp_path / "errors.csv"),
+        verbose=False,
+        force=True,
+        checkpoint_every_rows=1,
+        checkpoint_every_sec=3600,
+        resume=False,
+        strict_resume=False,
+    )
+    phase_module.main(args)
+
+    assert calls["count"] == 1
+    out_df = pd.read_csv(args.csv_path_out)
+    assert out_df.loc[0, "totalseg_phase"] == "arterial"
+
+
 def test_main_preserves_foreign_columns_from_existing_output(tmp_path, monkeypatch):
     nifti = tmp_path / "valid.nii.gz"
     nifti.write_text("nifti")