Reuse cleaning filter logic for radiomics

docs/source/manifests.md

@@ -173,7 +173,9 @@ radiomics:
         imageType:
           Original: {}
   filters:
-    phase: [ARTERIAL, PORTAL_VENOUS]
+    - kind: keep
+      rules:
+        - {column: phase, op: in, value: [ARTERIAL, PORTAL_VENOUS]}
 ```
 
 Copy a built-in YAML manifest as the starting point because the full built-in
@@ -945,8 +947,29 @@ For a shared configuration, the existing `radiomics.pyradiomics` mapping and
 `modalities` or the shared `pyradiomics` mapping; configurations are not merged.
 Manifest settings take precedence over the CLI settings file.
 
-`radiomics.filters` maps an existing cohort column to its accepted values. Use
-the canonical `phase` output for phase-based filtering.
+`radiomics.filters` accepts a list of filters using the same `kind`, `logic`,
+`rules`, and `keep_null` options as [cleaning filters](#cleaning-filters).
+Filters run in order on cohort rows; `type` and `scope` are unnecessary.
+All cleaning operators and null/list matching semantics are shared. Unlike
+cleaning, radiomics reports an error if a referenced column is absent.
+
+```yaml
+radiomics:
+  filters:
+    - kind: keep
+      logic: and
+      rules:
+        - {column: phase, op: in, value: [PORTAL_VENOUS, ARTERIAL]}
+        - {column: followup_months, op: lte, value: 6}
+    - kind: discard
+      rules:
+        - {column: SeriesDescription, op: icontains, value: localizer}
+```
+
+The existing column-to-accepted-values mapping remains supported, as does
+`--filter column=value1,value2`. Manifest rules override CLI filters for every
+column they reference; other CLI filters are combined with the manifest filters.
+`--skip_filter` disables both. Use the canonical `phase` output for phase filters.
 
 ## Validation checklist
 

src/imperandi/builtin_datasets_config/manifests/generic.yaml

@@ -331,6 +331,6 @@ radiomics:
         imageType:
           Original: {}
   filters:
-    # Each key is a cohort-table column; a row is retained when its value is in
-    # the configured list. Multiple filter keys are combined with AND.
-    phase: [PORTAL_VENOUS, ARTERIAL]
+    - kind: keep
+      rules:
+        - {column: phase, op: in, value: [PORTAL_VENOUS, ARTERIAL]}

src/imperandi/extract/radiomics.py

@@ -14,6 +14,7 @@
     setup_logging,
 )
 from imperandi.utils.columns import MASK_PATH_COLUMN_PATTERN, is_mask_path_column
+from imperandi.utils.conditions import filter_mask, validate_filter_group
 from imperandi.utils.misc import print_args
 from imperandi.utils.manifest import load_manifest
 from imperandi.utils.checkpoint_cli import add_checkpoint_arguments
@@ -272,48 +273,58 @@ def _normalize_cli_filters(filter_args: Optional[list[str]]) -> dict[str, list[s
     return normalized
 
 
-def _load_manifest_radiomics_filters(
-    manifest_arg: Optional[str],
-) -> dict[str, list[Any]]:
+def _normalize_filter_groups(raw_filters: Any) -> list[dict]:
+    """Convert legacy column/value filters to the cleaning filter syntax."""
+    if raw_filters is None:
+        return []
+    if isinstance(raw_filters, dict):
+        groups = []
+        for column, values in raw_filters.items():
+            column_name = str(column).strip()
+            if not column_name:
+                raise ValueError("Radiomics filters contain an empty column name.")
+            if not isinstance(values, list) or not values:
+                raise ValueError(
+                    f"Radiomics filters[{column_name!r}] must be a non-empty list."
+                )
+            groups.append(
+                {
+                    "kind": "keep",
+                    "rules": [{"column": column_name, "op": "in", "value": values}],
+                }
+            )
+    elif isinstance(raw_filters, list):
+        groups = raw_filters
+    else:
+        raise ValueError(
+            "Radiomics filters must be an object or a list of filter groups."
+        )
+    return [
+        validate_filter_group(group, f"radiomics.filters[{index}]")
+        for index, group in enumerate(groups)
+    ]
+
+
+def _load_manifest_radiomics_filters(manifest_arg: Optional[str]) -> list[dict]:
     if not manifest_arg:
-        return {}
+        return []
 
     manifest = load_manifest(
         manifest_arg,
         base_path=Path(__file__).resolve().parents[1],
     )
     radiomics_settings = manifest.get("radiomics")
     if radiomics_settings is None:
-        return {}
+        return []
     if not isinstance(radiomics_settings, dict):
         raise ValueError(
             "Manifest radiomics settings must be an object under key 'radiomics'."
         )
+    return _normalize_filter_groups(radiomics_settings.get("filters"))
 
-    raw_filters = radiomics_settings.get("filters")
-    if raw_filters is None:
-        return {}
-    if not isinstance(raw_filters, dict):
-        raise ValueError("Manifest radiomics.filters must be an object.")
-
-    normalized: dict[str, list[Any]] = {}
-    for column, values in raw_filters.items():
-        column_name = str(column).strip()
-        if not column_name:
-            raise ValueError(
-                "Manifest radiomics.filters contains an empty column name."
-            )
-        if not isinstance(values, list):
-            raise ValueError(
-                f"Manifest radiomics.filters[{column_name!r}] must be a list."
-            )
-        if not values:
-            raise ValueError(
-                f"Manifest radiomics.filters[{column_name!r}] must not be empty."
-            )
-        normalized[column_name] = values
 
-    return normalized
+def _filter_columns(filters: list[dict]) -> set[str]:
+    return {rule["column"] for group in filters for rule in group["rules"]}
 
 
 def _resolve_pyradiomics_settings_source(
@@ -405,55 +416,56 @@ def _create_extractors_for_rows(
     return row_modalities, extractors
 
 
-def _resolve_radiomics_filters(args: argparse.Namespace) -> dict[str, list[Any]]:
+def _resolve_radiomics_filters(args: argparse.Namespace) -> list[dict]:
     if getattr(args, "skip_filter", False):
         logger.info("Radiomics row filters skipped via --skip_filter")
-        return {}
+        return []
 
     cli_filters = getattr(args, "filters", {}) or {}
     manifest_filters = _load_manifest_radiomics_filters(getattr(args, "manifest", None))
-    effective_filters = dict(cli_filters)
-
-    if manifest_filters:
-        overlapping_columns = sorted(set(cli_filters) & set(manifest_filters))
-        for column in overlapping_columns:
-            logger.info(
-                "Manifest radiomics filter overrides CLI filter for column '%s'",
-                column,
-            )
-        effective_filters.update(manifest_filters)
-
-    if effective_filters:
-        logger.info("Radiomics row filters resolved: %s", effective_filters)
-    else:
-        logger.info("Radiomics row filters resolved: none")
+    manifest_columns = _filter_columns(manifest_filters)
+    for column in sorted(set(cli_filters) & manifest_columns):
+        logger.info(
+            "Manifest radiomics filter overrides CLI filter for column '%s'",
+            column,
+        )
+    effective_filters = (
+        _normalize_filter_groups(
+            {
+                column: values
+                for column, values in cli_filters.items()
+                if column not in manifest_columns
+            }
+        )
+        + manifest_filters
+    )
+    logger.info("Radiomics row filters resolved: %s", effective_filters or "none")
     return effective_filters
 
 
 def _apply_explicit_filters(
     df: pd.DataFrame,
-    filters: dict[str, list[Any]],
+    filters: list[dict],
 ) -> pd.DataFrame:
     if not filters:
         logger.info("No radiomics row filters applied; keeping %d rows", len(df))
         return df
 
-    missing_columns = [column for column in filters if column not in df.columns]
+    missing_columns = _filter_columns(filters) - set(df.columns)
     if missing_columns:
         raise ValueError(
             "Radiomics filter column(s) missing from input CSV: "
             + ", ".join(sorted(missing_columns))
         )
 
-    filtered = df.copy()
+    filtered = df
     logger.info("Applying radiomics row filters to %d rows", len(filtered))
-    for column, allowed_values in filters.items():
+    for group in filters:
         before_count = len(filtered)
-        filtered = filtered[filtered[column].isin(allowed_values)]
+        filtered = filtered[filter_mask(filtered, group)].copy()
         logger.info(
-            "Radiomics filter applied | column=%s | values=%s | rows=%d -> %d",
-            column,
-            allowed_values,
+            "Radiomics filter applied | filter=%s | rows=%d -> %d",
+            group,
             before_count,
             len(filtered),
         )
@@ -961,7 +973,7 @@ def main(args: argparse.Namespace) -> None:
         "volume_id",
         "nifti_path",
         "Modality",
-        *sorted(effective_filters),
+        *sorted(_filter_columns(effective_filters)),
     ]
     semantic_fp = fingerprint_csv_semantic(
         args.csv_path,

src/imperandi/ingest/clean.py

@@ -45,9 +45,9 @@
 from imperandi.utils.dicom_timing import DICOM_CLOCK_DTYPES
 from imperandi.utils.conditions import (
     SUPPORTED_OPERATORS,
+    filter_mask,
     rule_mask,
-    rules_mask,
-    validate_rule_group,
+    validate_filter_group,
 )
 from imperandi.ingest.defaults import (
     DEFAULT_DICOM_TAGS,
@@ -2148,13 +2148,9 @@ def validate_cleaning_manifest(manifest: dict) -> list[dict]:
             _get_hook_outputs(step)
 
         if step_type == "filter":
-            if step.get("kind") not in {"keep", "discard"}:
-                raise ValueError("Filter steps must define kind='keep' or 'discard'.")
+            step.update(validate_filter_group(step, f"cleaning.steps[{index - 1}]"))
             if step.get("scope") not in {"row", "volume"}:
                 raise ValueError("Filter steps must define scope='row' or 'volume'.")
-            step.update(validate_rule_group(step, f"cleaning.steps[{index - 1}]"))
-            if "keep_null" in step and not isinstance(step["keep_null"], bool):
-                raise ValueError("Filter steps must define keep_null as a boolean.")
 
         _validate_step_config(step)
 
@@ -2362,15 +2358,7 @@ def _run_filter_step(df: pd.DataFrame, step: dict) -> pd.DataFrame:
         )
         return df
 
-    mask = rules_mask(df, step["rules"], step.get("logic", "and"))
-    if step.get("keep_null"):
-        null_mask = rules_mask(
-            df,
-            [{"column": column, "op": "is_null"} for column in required_columns],
-            "or",
-        )
-        mask = mask | null_mask if step["kind"] == "keep" else mask & ~null_mask
-    return df[mask if step["kind"] == "keep" else ~mask].copy()
+    return df[filter_mask(df, step)].copy()
 
 
 def _safe_uid_keyword(value):

src/imperandi/utils/conditions.py

@@ -1,4 +1,4 @@
-"""Shared manifest predicates for cleaning filters and phase assignments."""
+"""Shared manifest predicates for cleaning, radiomics, and phase assignments."""
 
 from __future__ import annotations
 
@@ -165,3 +165,30 @@ def rules_mask(df: pd.DataFrame, rules: list[Mapping], logic: str) -> pd.Series:
         current = rule_mask(df, rule)
         mask = mask & current if logic == "and" else mask | current
     return mask
+
+
+def validate_filter_group(group: Mapping, field: str) -> dict:
+    """Validate a keep/discard filter using the shared rule syntax."""
+    if not isinstance(group, Mapping):
+        raise ValueError(f"{field} must be a mapping.")
+    if group.get("kind") not in {"keep", "discard"}:
+        raise ValueError("Filter steps must define kind='keep' or 'discard'.")
+    normalized = {**group, **validate_rule_group(group, field)}
+    if "keep_null" in group and not isinstance(group["keep_null"], bool):
+        raise ValueError("Filter steps must define keep_null as a boolean.")
+    return normalized
+
+
+def filter_mask(df: pd.DataFrame, group: Mapping) -> pd.Series:
+    """Return rows to retain for a validated keep/discard filter."""
+    mask = rules_mask(df, group["rules"], group.get("logic", "and"))
+    if group["kind"] == "discard":
+        mask = ~mask
+    if group.get("keep_null"):
+        columns = {rule["column"] for rule in group["rules"]}
+        mask |= rules_mask(
+            df,
+            [{"column": column, "op": "is_null"} for column in columns],
+            "or",
+        )
+    return mask

tests/unit/test_cli.py

@@ -66,7 +66,15 @@ def test_startup_logs_resolved_manifest_on_dry_run(command, tmp_path, caplog):
         elif name == "segment.py":
             assert logged.segmentation["modalities"]["CT"]["tasks"]
         elif name == "radiomics.py":
-            assert logged.filters == {"phase": ["MANIFEST_PHASE"]}
+            assert logged.filters == [
+                {
+                    "kind": "keep",
+                    "logic": "and",
+                    "rules": [
+                        {"column": "phase", "op": "in", "value": ["MANIFEST_PHASE"]}
+                    ],
+                }
+            ]
             assert (
                 logged.pyradiomics_settings["modalities"]["CT"]["setting"]["binWidth"]
                 == 17
@@ -407,8 +415,12 @@ def test_cli_radiomics_accepts_repeatable_filter_flags(tmp_path, capsys):
         line for line in output.splitlines() if line.strip().startswith("filters")
     )
     assert exit_code == 0
-    assert "'phase': ['portal', 'arteriel']" in filters_line
-    assert "'followup_months': ['0', '3']" in filters_line
+    assert (
+        "'column': 'phase', 'op': 'in', 'value': ['portal', 'arteriel']" in filters_line
+    )
+    assert (
+        "'column': 'followup_months', 'op': 'in', 'value': ['0', '3']" in filters_line
+    )
 
 
 def test_cli_parse_accepts_snapshot_flags(tmp_path):