diff --git a/docs/source/manifests.md b/docs/source/manifests.md index ccd8535..dd17d08 100644 --- a/docs/source/manifests.md +++ b/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 diff --git a/src/imperandi/builtin_datasets_config/manifests/generic.yaml b/src/imperandi/builtin_datasets_config/manifests/generic.yaml index 388b500..28e9e84 100644 --- a/src/imperandi/builtin_datasets_config/manifests/generic.yaml +++ b/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]} diff --git a/src/imperandi/extract/radiomics.py b/src/imperandi/extract/radiomics.py index 818f4a7..aba2608 100644 --- a/src/imperandi/extract/radiomics.py +++ b/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,11 +273,41 @@ 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, @@ -284,36 +315,16 @@ def _load_manifest_radiomics_filters( ) 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, diff --git a/src/imperandi/ingest/clean.py b/src/imperandi/ingest/clean.py index ae14c37..427c62e 100644 --- a/src/imperandi/ingest/clean.py +++ b/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): diff --git a/src/imperandi/utils/conditions.py b/src/imperandi/utils/conditions.py index 2dea378..0740cdc 100644 --- a/src/imperandi/utils/conditions.py +++ b/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 diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index ea2fde3..5e58e87 100644 --- a/tests/unit/test_cli.py +++ b/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): diff --git a/tests/unit/test_radiomics.py b/tests/unit/test_radiomics.py index 456437c..d5709d5 100644 --- a/tests/unit/test_radiomics.py +++ b/tests/unit/test_radiomics.py @@ -459,11 +459,13 @@ def test_resolve_radiomics_filters_prefers_manifest_for_same_column( filters = radiomics_module._resolve_radiomics_filters(args) - assert filters == { - "phase": ["portal"], - "followup_months": ["0", "3"], - "site": ["A"], - } + assert filters == radiomics_module._normalize_filter_groups( + { + "followup_months": ["0", "3"], + "phase": ["portal"], + "site": ["A"], + } + ) def test_resolve_radiomics_filters_skip_filter_bypasses_manifest_and_cli(tmp_path): @@ -475,20 +477,22 @@ def test_resolve_radiomics_filters_skip_filter_bypasses_manifest_and_cli(tmp_pat manifest=str(manifest_path), ) - assert radiomics_module._resolve_radiomics_filters(args) == {} + assert radiomics_module._resolve_radiomics_filters(args) == [] def test_apply_explicit_filters_requires_existing_columns(): df = pd.DataFrame([{"phase": "portal"}]) with pytest.raises(ValueError, match="missing from input CSV"): - radiomics_module._apply_explicit_filters(df, {"site": ["A"]}) + radiomics_module._apply_explicit_filters( + df, radiomics_module._normalize_filter_groups({"site": ["A"]}) + ) def test_apply_explicit_filters_keeps_rows_when_no_filters(): df = pd.DataFrame([{"phase": "mixte"}, {"phase": "portal"}]) - out = radiomics_module._apply_explicit_filters(df, {}) + out = radiomics_module._apply_explicit_filters(df, []) pd.testing.assert_frame_equal(out, df) @@ -504,12 +508,103 @@ def test_apply_explicit_filters_uses_and_semantics_and_preserves_order(): out = radiomics_module._apply_explicit_filters( df, - {"phase": ["portal"], "followup_months": [0, 3]}, + radiomics_module._normalize_filter_groups( + {"phase": ["portal"], "followup_months": [0, 3]} + ), ) assert out["row_id"].tolist() == ["keep-1", "keep-2"] +@pytest.mark.parametrize("kind, expected", [("keep", [4, 1, 3]), ("discard", [1, 2])]) +def test_radiomics_filter_groups_share_cleaning_semantics(kind, expected): + df = pd.DataFrame( + { + "phase": [["portal", "arterial"], [], ["other"], ["other"]], + "months": [0, None, 2, 6], + "description": ["diagnostic", "diagnostic", "LOCALIZER", "diagnostic"], + }, + index=[4, 1, 2, 3], + ) + groups = radiomics_module._normalize_filter_groups( + [ + { + "kind": kind, + "logic": "or", + "keep_null": True, + "rules": [ + {"column": "phase", "op": "in", "value": ["portal"]}, + {"column": "months", "op": "gte", "value": 6}, + ], + }, + ] + ) + out = radiomics_module._apply_explicit_filters(df, groups) + assert out.index.tolist() == expected + + groups.append( + radiomics_module._normalize_filter_groups( + [ + { + "kind": "discard", + "rules": [ + { + "column": "description", + "op": "icontains", + "value": "localizer", + } + ], + } + ] + )[0] + ) + out = radiomics_module._apply_explicit_filters(df, groups) + assert out.index.tolist() == [index for index in expected if index != 2] + + +@pytest.mark.parametrize( + "group, message", + [ + (None, "must be a mapping"), + ({"kind": "invalid", "rules": []}, "kind="), + ({"kind": "keep", "rules": []}, "non-empty 'rules'"), + ( + {"kind": "keep", "rules": [{"column": "phase", "op": "invalid"}]}, + "Unsupported filter operator", + ), + ( + {"kind": "keep", "rules": [{"column": "phase", "op": "eq"}]}, + "requires a 'value'", + ), + ( + { + "kind": "keep", + "logic": "xor", + "rules": [{"column": "phase", "op": "is_null"}], + }, + "logic must be", + ), + ( + { + "kind": "keep", + "keep_null": "true", + "rules": [{"column": "phase", "op": "is_null"}], + }, + "keep_null as a boolean", + ), + ( + {"kind": "keep", "rules": [{"column": "phase", "op": "is_null"}] * 2}, + "for multiple rules", + ), + ], +) +def test_manifest_radiomics_validates_filter_groups(tmp_path, group, message): + manifest_path = tmp_path / "manifest.yaml" + manifest_path.write_text(yaml.safe_dump({"radiomics": {"filters": [group]}})) + with pytest.raises(ValueError, match=message): + radiomics_module._load_manifest_radiomics_filters(str(manifest_path)) + + def test_configure_pyradiomics_output_toggles_logger_state(): pyradiomics_logger = logging.getLogger("radiomics") old_disabled = pyradiomics_logger.disabled @@ -765,7 +860,10 @@ def fake_liver(*args, **kwargs): assert dep_calls["count"] == 1 -def test_main_resume_only_tracks_radiomics_manifest_section(tmp_path, monkeypatch): +@pytest.mark.parametrize("declarative", [False, True]) +def test_main_resume_only_tracks_radiomics_manifest_section( + tmp_path, monkeypatch, declarative +): portal_nifti = tmp_path / "portal.nii.gz" arterial_nifti = tmp_path / "arterial.nii.gz" portal_mask = tmp_path / "portal_mask.nii.gz" @@ -789,9 +887,25 @@ def test_main_resume_only_tracks_radiomics_manifest_section(tmp_path, monkeypatc ] ).to_csv(csv_path, index=False) manifest_path = tmp_path / "manifest.yaml" - manifest_path.write_text( - '{"dataset_name": "v1", "radiomics": {"filters": {"phase": ["portal"]}}}' - ) + + def write_manifest(dataset_name, phase): + filters = ( + [ + { + "kind": "keep", + "rules": [{"column": "phase", "op": "eq", "value": phase}], + } + ] + if declarative + else {"phase": [phase]} + ) + manifest_path.write_text( + yaml.safe_dump( + {"dataset_name": dataset_name, "radiomics": {"filters": filters}} + ) + ) + + write_manifest("v1", "portal") monkeypatch.setattr( radiomics_module, @@ -841,20 +955,23 @@ def fake_liver( ) radiomics_module.main(args) - manifest_path.write_text( - '{"dataset_name": "v1", "radiomics": {"filters": {"phase": ["arterial"]}}}' - ) + write_manifest("v1", "arterial") args.resume = True radiomics_module.main(args) # Changes outside the command-specific section must not invalidate resume. - manifest_path.write_text( - '{"dataset_name": "v2", "radiomics": {"filters": {"phase": ["arterial"]}}}' - ) + write_manifest("v2", "arterial") radiomics_module.main(args) assert processed == ["portal.nii.gz", "arterial.nii.gz"] + # Referenced rule columns must participate in semantic resume fingerprints. + updated = pd.read_csv(csv_path) + updated["phase"] = ["arterial", "portal"] + updated.to_csv(csv_path, index=False) + radiomics_module.main(args) + assert processed == ["portal.nii.gz", "arterial.nii.gz", "portal.nii.gz"] + def test_main_preserves_foreign_columns_from_existing_output(tmp_path, monkeypatch): nifti = tmp_path / "good.nii.gz" @@ -968,7 +1085,10 @@ def test_main_does_not_filter_phase_by_default(tmp_path, monkeypatch): assert sorted(out_df["phase"].tolist()) == ["mixte", "portal"] -def test_main_applies_explicit_filters_from_manifest_and_cli(tmp_path, monkeypatch): +@pytest.mark.parametrize("declarative", [False, True]) +def test_main_applies_explicit_filters_from_manifest_and_cli( + tmp_path, monkeypatch, declarative +): nifti_a = tmp_path / "a.nii.gz" nifti_b = tmp_path / "b.nii.gz" nifti_c = tmp_path / "c.nii.gz" @@ -1002,7 +1122,17 @@ def test_main_applies_explicit_filters_from_manifest_and_cli(tmp_path, monkeypat ] ).to_csv(csv_path, index=False) manifest_path = tmp_path / "manifest.yaml" - manifest_path.write_text('{"radiomics": {"filters": {"phase": ["portal"]}}}') + filters = ( + [ + { + "kind": "keep", + "rules": [{"column": "phase", "op": "eq", "value": "portal"}], + } + ] + if declarative + else {"phase": ["portal"]} + ) + manifest_path.write_text(yaml.safe_dump({"radiomics": {"filters": filters}})) monkeypatch.setattr( radiomics_module,