# Medical-imaging preprocessing — modality-aware guidance

Companion to `preprocess-imaging`. This is *produce* knowledge: what preprocessing is standard per
modality, which augmentations preserve versus break physiology, and where leakage hides. It wires
MONAI / TorchIO transforms by name; it does not reimplement them.

## 1. Intensity normalisation by modality

| Modality | Standard intensity handling | Fitted on data? | Leakage risk |
|---|---|---|---|
| **CT** | Fixed HU window/level per task (e.g. lung −600/1500, soft-tissue 40/400), then scale to [0,1] or [−1,1] | **No** — fixed HU bounds are physical, not fitted | None (fixed transform) |
| **MR** | Bias-field correction (N4) → intensity normalisation (z-score, or Nyúl/histogram matching to a reference) | **Yes** — z-score/Nyúl are fitted | Fit the reference/stats on **train only**; per-image z-score is leakage-free |
| **X-ray** | Per-image min–max or z-score; optional CLAHE | Per-image = **no**; dataset z-score = yes | Prefer per-image; if dataset-level, train-only |
| **Ultrasound** | Per-image normalisation; crop the vendor UI/annotation border | Per-image = **no** | Cropping is fixed; watch for burned-in PHI/annotations |
| **Pathology (WSI)** | Stain normalisation (Macenko/Vahadane/Reinhard) to a reference tile | **Yes** — reference is fitted | Choose the reference from the **train** split only |

**Rule of thumb:** a *fixed* transform (HU window, resample to a fixed spacing, fixed crop) never
leaks. A *fitted* transform (z-score with dataset statistics, histogram/stain matching to a reference,
PCA/whitening) leaks unless it is fit on the training split and applied after the split.

## 2. Resampling and geometry

- Resample to a **fixed target spacing** (a task constant, not fitted) — e.g. 1×1×1 mm, or the
  dataset median spacing computed **once on train** and then frozen as a constant.
- Register/reorient to a canonical orientation (e.g. RAS) before patch extraction.
- Foreground cropping by a fixed intensity threshold is fixed (safe); cropping by a *fitted* body-mask
  model is a fitted transform — train-only.

## 3. Augmentation appropriateness (physiology-preserving vs breaking)

Augmentation must keep the image clinically plausible and label-consistent.

| Augmentation | Usually safe | Breaks physiology / label when… |
|---|---|---|
| Flip (L–R) | Most 2-D texture tasks | **Laterality matters** — cardiac silhouette, situs, side-labelled findings; flipping mislabels side |
| Rotation (small) | Most tasks | Large rotations for orientation-dependent tasks (e.g. gravity-dependent effusion/air-fluid levels) |
| Elastic / grid deformation | Soft-tissue segmentation | Rigid structures (bone fracture morphology); can invent/erase small lesions |
| Intensity scale/shift, gamma | CT/MR within-modality | Beyond clinically plausible HU/contrast range; simulates a nonexistent scanner |
| Gaussian noise / blur | Robustness to acquisition | Enough to hide the target finding (micro-nodule, microcalcification) |
| Cutout / random erasing | General | Can erase the sole lesion in a positive case → label noise |
| MixUp / CutMix | Some classification | Segmentation/detection where mixed pixels have no valid mask/box |

**Apply augmentation to the training split only.** Augmenting val/test is test-time augmentation
(TTA): legitimate only if pre-specified and disclosed, never folded silently into the headline metric.

## 4. Where leakage hides (the gate's targets)

1. **Normalisation fit on non-train data** — z-score/histogram/stain statistics computed over
   all/test data. → `NORMALIZATION_LEAKAGE`. Fit on train; apply the frozen statistics to val/test.
2. **Fitted transform before the split** — computing dataset statistics, then splitting. There is no
   train/test distinction yet, so the fit is cross-partition. → `PREPROCESS_BEFORE_SPLIT`.
3. **Patient slices across splits** — splitting *images* instead of *patients*, so a patient's slices
   sit in train and test. → `PATIENT_CROSS_SPLIT`. Split at the patient level, then map slices.
4. **Augmentation on eval** — see §3. → `AUGMENTATION_ON_EVAL`.
5. **Undeclared fit scope** — a fitted transform with no stated scope cannot be cleared.
   → `UNSPECIFIED_FIT_SCOPE`. Declare `fit_scope=train`.

## 5. Library wiring (integrate, don't reimplement)

- **MONAI transforms** — `LoadImaged`, `Spacingd`, `ScaleIntensityRanged` (fixed HU window),
  `NormalizeIntensityd` (z-score; set on the train subset), `RandFlipd`/`RandAffined`/`RandGaussianNoised`
  (train pipeline only).
- **TorchIO** — `Resample`, `ZNormalization`, `HistogramStandardization` (fit `landmarks` on train),
  `RandomFlip`/`RandomElasticDeformation` (train only).
- **Stain normalisation (WSI)** — `torchstain` / `staintools` Macenko/Vahadane; fit the reference on a
  train tile.

Record every step in the preprocessing manifest with its `type`, `fit_scope`, and `stage` so the gate
can decide leakage deterministically.

#!/usr/bin/env python3
"""Normaliser-domain gate: is the cohort you are about to predict in the domain the trained
normalisation contract assumes?

`check_preprocessing_leakage.py` asks whether a data-fitted transform was fit on the right SCOPE
(train only, after the split). This asks a different question, at a different moment: at inference,
is the cohort being handed to a trained normaliser in the INTENSITY DOMAIN that normaliser was fit
on? Nothing else in this toolkit asks it, and nothing carries a profiling-stage observation forward
to the moment of use.

**Why this exists.** In demo/05_msd_amos_spleen, a segmentation model trained on CT was applied to
MRI. The training plan carried `CTNormalization` into inference and the inference command had no
argument declaring the incoming modality, so a Hounsfield-unit clip was applied to arbitrary-unit
images. Median Dice was 0.0152. Two independent counterfactual arms — rescale the input, or swap the
normaliser — each recovered it to ~0.29 without retraining, so roughly **0.28 Dice** was the domain
mismatch alone. The run exited 0 and wrote a plausible-looking segmentation for every case.

The profiler had already seen it. `/profile-imaging` flagged `INTENSITY_SCALE_INCONSISTENT` on that
cohort *before training* — as a **Minor**, in a directory no later step reads. The gap this gate
closes is therefore **routing and severity**, not detection: it re-reads an existing profile against
the contract that will actually be applied, at the point where ignoring it costs something.

Reads two JSON files and nothing else — no imaging library, no pixels, no model. stdlib only, so it
travels with the artifacts.

  --profile   a /profile-imaging profile (`profile_imaging_dataset.py` output) with per-case
              `intensity` {min, p01, p50, p99, max} and `split`
  --contract  the normalisation contract that will be applied. Either an nnU-Net-style `plans.json`
              (the scheme and the foreground statistics are read from it) or a small JSON of the
              form {"scheme": "...", "assumes": "hounsfield"|"arbitrary"|"zero_mean",
              "clip": [lo, hi], "mean": m, "std": s}.

Verdicts:

  NORMALIZER_DOMAIN_MISMATCH  (Major)  the contract assumes Hounsfield units and the split contains
                                       no negative voxels — HU is defined by an air floor near
                                       -1000, so a cohort that never goes negative is not in it
  NORMALIZER_SPLIT_DIVERGENCE (Flag)   splits inside one cohort disagree about the intensity domain,
                                       so one contract cannot be right for all of them

**A third check was written and deleted.** `NORMALIZER_CLIP_DESTRUCTIVE` compared each split's 99th
percentile against the contract's clip ceiling. Run against the cohorts this repository already
produces, it fired on **100% of the CT arm and 100% of the MSD training set** — the very data the
plan was fit on. Of course it did: a CT volume's p99 is bone, and clipping bone above a soft-tissue
window is what `CTNormalization` is *for*. A five-number summary cannot distinguish "the window is
doing its job" from "the window is destroying the image", and a check that rejects its own training
data is worse than no check. Deleted rather than tuned.
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

HU_AIR_CEILING = -500.0     # a genuine HU volume has voxels below this (air is near -1000)


def load(p: str) -> dict:
    try:
        return json.loads(Path(p).read_text(encoding="utf-8"))
    except Exception as exc:                                  # noqa: BLE001
        print(f"FATAL: cannot read {p}: {exc}", file=sys.stderr)
        raise SystemExit(2) from exc


def contract_from(obj: dict, channel: str = "0") -> dict:
    """Accept an nnU-Net plans.json or a small hand-written contract."""
    if "configurations" in obj:                                # nnU-Net plans.json
        cfg = obj["configurations"].get("3d_fullres") or next(iter(obj["configurations"].values()))
        schemes = cfg.get("normalization_schemes") or []
        scheme = schemes[0] if schemes else "unknown"
        props = (obj.get("foreground_intensity_properties_per_channel")
                 or obj.get("foreground_intensity_properties_by_modality") or {})
        s = props.get(channel, {})
        return {
            "scheme": scheme,
            "assumes": "hounsfield" if "CT" in scheme else "arbitrary",
            "clip": [s.get("percentile_00_5"), s.get("percentile_99_5")] if s else None,
            "mean": s.get("mean"), "std": s.get("std"),
        }
    c = dict(obj)
    if "scheme" not in c:
        # Fail loudly rather than scoring an unparseable contract as "assumes arbitrary" and
        # returning OK. A gate that passes on input it could not read looks exactly like a gate
        # that passed — this one refuses instead.
        raise SystemExit(
            "FATAL: --contract is neither an nnU-Net plans.json (no 'configurations' key) nor a "
            "contract object (no 'scheme' key). Refusing to evaluate a contract I cannot read; "
            "pass the plans.json the model was trained under, or a JSON with at least "
            '{"scheme": "...", "assumes": "hounsfield"|"arbitrary"}.')
    c.setdefault("assumes", "hounsfield" if "CT" in str(c.get("scheme", "")) else "arbitrary")
    return c


def by_split(profile: dict) -> dict[str, list[dict]]:
    out: dict[str, list[dict]] = {}
    for case in profile.get("cases", []):
        if "intensity" not in case:
            continue
        out.setdefault(case.get("split", "(unsplit)"), []).append(case)
    return out


def main() -> int:
    ap = argparse.ArgumentParser(description="Does the cohort match the normaliser's assumed domain?")
    ap.add_argument("--profile", required=True)
    ap.add_argument("--contract", required=True)
    ap.add_argument("--channel", default="0")
    ap.add_argument("--splits", nargs="*", default=None,
                    help="restrict to these split names (default: every labelled split)")
    ap.add_argument("--out")
    ap.add_argument("--strict", action="store_true", help="exit 1 when a Major is raised")
    a = ap.parse_args()

    profile = load(a.profile)
    contract = contract_from(load(a.contract), a.channel)
    groups = by_split(profile)
    if a.splits:
        groups = {k: v for k, v in groups.items() if k in a.splits}
    if not groups:
        print("FATAL: no per-case intensity found in the profile", file=sys.stderr)
        return 2

    claims, domains = [], {}
    for split, cases in sorted(groups.items()):
        mins = [c["intensity"].get("min") for c in cases if c["intensity"].get("min") is not None]
        if not mins:
            continue
        n_air = sum(1 for m in mins if m <= HU_AIR_CEILING)
        looks_hu = n_air == len(mins)
        domains[split] = "hounsfield" if looks_hu else ("mixed" if n_air else "non-hounsfield")

        if contract.get("assumes") == "hounsfield" and n_air == 0:
            claims.append({
                "verdict": "NORMALIZER_DOMAIN_MISMATCH", "severity": "Major", "split": split,
                "detail": (f"the contract applies {contract.get('scheme')}, which assumes Hounsfield "
                           f"units, but 0 of {len(mins)} case(s) in '{split}' contain a voxel at or "
                           f"below {HU_AIR_CEILING:.0f}. Hounsfield units are defined by an air "
                           f"floor near -1000; a cohort that never goes negative is not in them. "
                           f"Per-case minima range {min(mins):.0f} to {max(mins):.0f}."),
            })
        elif contract.get("assumes") == "hounsfield" and 0 < n_air < len(mins):
            claims.append({
                "verdict": "NORMALIZER_SPLIT_DIVERGENCE", "severity": "Flag", "split": split,
                "detail": (f"{n_air} of {len(mins)} case(s) in '{split}' bottom out near air and the "
                           f"rest do not — mixed modality, or a rescale applied to part of the "
                           f"cohort. One normalisation contract cannot be right for both."),
            })

    if len({d for d in domains.values() if d != "mixed"}) > 1:
        claims.append({
            "verdict": "NORMALIZER_SPLIT_DIVERGENCE", "severity": "Flag", "split": "(cohort)",
            "detail": (f"splits disagree about the intensity domain ({domains}); a single trained "
                       f"contract cannot be correct for all of them."),
        })

    n_major = sum(1 for c in claims if c["severity"] == "Major")
    report = {
        "detector": "check_normalizer_domain",
        "profile": a.profile, "contract": a.contract,
        "contract_read": contract,
        "splits_examined": {k: {"n": len(v), "domain": domains.get(k)} for k, v in groups.items()},
        "claims": claims,
        "summary": {"n_claims": len(claims), "n_major": n_major,
                    "n_flag": len(claims) - n_major,
                    "verdict": "OK" if not claims else ("MAJOR" if n_major else "FLAG")},
    }

    print("=" * 41)
    print(" Normaliser-Domain Gate (preprocess-imaging)")
    print("=" * 41)
    print(f"  contract: {contract.get('scheme')} (assumes {contract.get('assumes')})"
          + (f", clip [{contract['clip'][0]:.0f}, {contract['clip'][1]:.0f}]"
             if contract.get("clip") and None not in contract["clip"] else ""))
    print("| Check | Severity | Detail |")
    print("|---|---|---|")
    if claims:
        for c in claims:
            print(f"| {c['verdict']} | {c['severity']} | {c['detail']} |")
    else:
        print("| (none) | — | every split is in the domain the contract assumes |")
    print()
    print(f"MAJOR candidate: {n_major} domain issue(s)." if n_major
          else ("FLAG: review the divergence above." if claims
                else "OK: every split is in the domain the contract assumes."))

    if a.out:
        Path(a.out).parent.mkdir(parents=True, exist_ok=True)
        Path(a.out).write_text(json.dumps(report, indent=2) + "\n")
        print(f"\nwrote {a.out}")
    return 1 if (a.strict and n_major) else 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
"""Data-stage preprocessing-leakage gate for a medical-imaging pipeline (preprocess-imaging).

`model-validation`'s split-leakage gate proves the train/val/test *split* is
patient-disjoint. But leakage also enters one stage earlier — in **preprocessing** —
and the split table cannot see it. The three classic data-stage leaks
(Kapoor & Narayanan, Patterns 2023; Varoquaux & Cheplygina, npj Digit Med 2022;
CLAIM 2024 data-partition/preprocessing items) are:

  1. fitting a dataset-level normalisation / scaler on data that is NOT the training
     split (the test intensity distribution leaks into training);
  2. running any data-fitted transform BEFORE the split exists (there is no train/test
     distinction yet, so the fit is inherently cross-partition);
  3. the same patient's slices landing in more than one split (slice-level overlap that
     a per-image manifest hides).

This gate reads a declarative **preprocessing manifest** (JSON — the artifact this
skill emits, or one the researcher writes) and decides each of these by rule and by
set arithmetic on the patient IDs, not from prose. A *per-image* / *per-sample*
transform (each image normalised by its own statistics) is leakage-free and never
fires; only a *dataset-fitted* transform can leak.

CHECKS (verdicts):
  1. PREPROCESS_BEFORE_SPLIT  (Major)  a data-fitted transform runs before the split
                                       (stage=before_split) — the fit spans partitions.
  2. NORMALIZATION_LEAKAGE    (Major)  a data-fitted transform is fit on a non-train
                                       scope (all/full/dataset/test/both) after the split.
  3. PATIENT_CROSS_SPLIT      (Major)  a patient_id whose units appear in >= 2 splits.
  4. AUGMENTATION_ON_EVAL     (Minor)  an augmentation is applied to val/test (train-time
                                       augmentation folded into evaluation / undisclosed TTA).
  5. UNSPECIFIED_FIT_SCOPE    (Minor)  a data-fitted transform declares no fit_scope — the
                                       leak cannot be ruled out; declare train-only.
  6. MISSING_SEED             (Minor)  no split_seed — the split cannot be regenerated.

A data-fitted transform is one whose `type` is a fitted operation (normalization,
standardize, scaler, min-max, clip_percentile, histogram_match, pca, whitening,
feature_selection, resample, …) AND whose `fit_scope` is not per-sample/none/fixed.
A genuinely fixed transform (a fixed HU window, a resample to a spacing you chose
in advance) is not data-fitted and never leaks — declare `fit_scope: fixed` and it
stays silent.

Resampling is on that list because the *target* is so often derived rather than
chosen: nnU-Net sets its target spacing from a percentile of the dataset
fingerprint, so a resample fitted over every case carries held-out geometry into
the training grid just as an intensity statistic would. "Resample to a fixed
spacing never leaks" is true only when the spacing really is fixed.

MANIFEST (JSON)
  {
    "split_seed": 42,
    "transforms": [
      {"name": "...", "type": "standardize", "fit_scope": "train", "stage": "after_split"},
      {"name": "...", "type": "augmentation", "stage": "after_split", "applies_to": ["train"]}
    ],
    "split_assignment": [
      {"patient_id": "P001", "unit_id": "P001_s1", "split": "train"}, ...
    ]
  }
  fit_scope   : train / training / train_val / dev  -> OK; all / full / dataset / test /
                both / combined -> leak; sample / per_image / instance / none -> not data-fitted.
  stage       : before_split / after_split (synonyms: pre-split / post-split).
  split       : train/val/test synonyms collapse (training/validation/holdout/…).

INPUTS
  --manifest  preprocessing manifest JSON (required).

OUTPUT
  A reconciliation table (stdout) and, with --out, a JSON artifact:
    {manifest, n_transforms, n_units, n_patients, partitions{name:count}, seed,
     claims[{verdict, severity, detail, where}], summary}
  PREPROCESS_BEFORE_SPLIT / NORMALIZATION_LEAKAGE / PATIENT_CROSS_SPLIT are Major.

Stdlib-only (json / argparse / pathlib). Exit codes: 0 clean (or report-only),
1 Major claim(s) found (with --strict), 2 input/usage error.
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

# Transform types whose parameters are FIT from data (so their fit scope matters).
FIT_BASED_TYPES = {
    "normalization", "normalize", "normalisation", "standardize", "standardise",
    "standardization", "zscore", "z-score", "z_score", "scaler", "scaling",
    "minmax", "min-max", "min_max", "clip_percentile", "percentile_clip",
    "percentile_clipping", "histogram_match", "histogram_matching",
    "histogram_equalization", "histogram_equalisation", "pca", "whitening",
    "feature_selection", "intensity_normalization", "intensity_normalisation",
    "nyul", "zca",
    # Resampling belongs here whenever the *target* is derived from the data. nnU-Net's
    # target spacing is a percentile of the dataset fingerprint, so a resample fitted over
    # all cases carries held-out geometry into the training grid exactly as an intensity
    # statistic would. A target that is genuinely fixed declares fit_scope=fixed and drops
    # out in _is_fit_based -- that is the case the docstring used to describe as if it
    # were the only one.
    "resample", "resampling", "respacing", "resample_spacing", "target_spacing",
    "spacing_normalization", "spacing_normalisation",
}
# fit_scope values that make a transform NOT data-fitted (per-sample or fixed).
SAMPLE_SCOPES = {
    "sample", "per_sample", "per-sample", "per_image", "per-image", "perimage",
    "instance", "per_instance", "none", "fixed", "self",
}
# fit_scope values that leak (fit touches non-training data).
NON_TRAIN_SCOPES = {
    "all", "full", "dataset", "entire", "everything", "combined", "both",
    "test", "testing", "holdout", "hold-out", "trainvaltest", "train_val_test",
    "train+test", "all_data", "whole",
}
TRAIN_OK_SCOPES = {
    "train", "training", "train_val", "trainval", "train+val", "development",
    "dev", "fold_train", "train_only",
}
BEFORE_STAGES = {"before_split", "before-split", "pre_split", "pre-split", "before", "pre"}
SPLIT_SYNONYM = {
    "train": "train", "training": "train",
    "val": "val", "validation": "val", "valid": "val", "dev": "val",
    "test": "test", "testing": "test", "holdout": "test", "hold-out": "test",
    "eval": "test", "evaluation": "test",
}
EVAL_SPLITS = {"val", "test"}


def _norm(s) -> str:
    return str(s).strip().lower() if s is not None else ""


def _is_fit_based(t: dict) -> bool:
    typ = _norm(t.get("type"))
    scope = _norm(t.get("fit_scope"))
    if typ not in FIT_BASED_TYPES:
        return False
    if scope in SAMPLE_SCOPES:
        return False
    return True


def check(manifest: dict) -> list[dict]:
    claims: list[dict] = []
    transforms = manifest.get("transforms") or []

    for t in transforms:
        name = t.get("name") or t.get("type") or "(unnamed)"
        typ = _norm(t.get("type"))
        scope = _norm(t.get("fit_scope"))
        stage = _norm(t.get("stage"))

        if _is_fit_based(t):
            if stage in BEFORE_STAGES:
                claims.append({
                    "verdict": "PREPROCESS_BEFORE_SPLIT", "severity": "Major",
                    "detail": (f"data-fitted transform '{name}' ({typ}) runs before the split "
                               f"(stage=before_split); the fit spans train and test"),
                    "where": name,
                })
            elif scope in NON_TRAIN_SCOPES:
                claims.append({
                    "verdict": "NORMALIZATION_LEAKAGE", "severity": "Major",
                    "detail": (f"data-fitted transform '{name}' ({typ}) is fit on a non-train "
                               f"scope ('{scope}'); test-set statistics leak into training"),
                    "where": name,
                })
            elif scope not in TRAIN_OK_SCOPES:
                # data-fitted, after split, but fit_scope undeclared/unknown -> ambiguous
                claims.append({
                    "verdict": "UNSPECIFIED_FIT_SCOPE", "severity": "Minor",
                    "detail": (f"data-fitted transform '{name}' ({typ}) declares no train-only "
                               f"fit_scope ('{scope or 'missing'}'); declare fit_scope=train so "
                               f"leakage can be ruled out"),
                    "where": name,
                })

        if typ in ("augmentation", "augment", "aug"):
            applies = [_norm(x) for x in (t.get("applies_to") or [])]
            applies = {SPLIT_SYNONYM.get(a, a) for a in applies}
            leaked = sorted(applies & EVAL_SPLITS)
            if leaked:
                claims.append({
                    "verdict": "AUGMENTATION_ON_EVAL", "severity": "Minor",
                    "detail": (f"augmentation '{name}' is applied to {'/'.join(leaked)}; "
                               f"train-time augmentation on an evaluation split folds "
                               f"undisclosed test-time augmentation into the reported metric"),
                    "where": name,
                })

    # Patient-level cross-split (set arithmetic).
    rows = manifest.get("split_assignment") or []
    pat_to_splits: dict[str, set] = {}
    for r in rows:
        pid = r.get("patient_id") or r.get("subject_id") or r.get("patient") or r.get("id")
        sp = SPLIT_SYNONYM.get(_norm(r.get("split")), _norm(r.get("split")))
        if pid is None or not sp:
            continue
        pat_to_splits.setdefault(str(pid), set()).add(sp)
    offenders = sorted(p for p, s in pat_to_splits.items() if len(s) >= 2)
    if offenders:
        ex = offenders[0]
        claims.append({
            "verdict": "PATIENT_CROSS_SPLIT", "severity": "Major",
            "detail": (f"{len(offenders)} of {len(pat_to_splits)} patients have units in "
                       f">= 2 splits (e.g. '{ex}' in {'/'.join(sorted(pat_to_splits[ex]))}); "
                       f"the same patient in train and test inflates every metric. "
                       f"Offenders: {', '.join(offenders)}"),
            "where": ex,
        })

    # Reproducibility.
    if manifest.get("split_seed") is None and rows:
        claims.append({
            "verdict": "MISSING_SEED", "severity": "Minor",
            "detail": "no split_seed recorded; the split cannot be regenerated or re-verified",
            "where": "split_seed",
        })

    return claims


def analyze(manifest_path: str) -> dict:
    p = Path(manifest_path)
    if not p.is_file():
        sys.stderr.write(f"ERROR: manifest not found: {manifest_path}\n")
        sys.exit(2)
    try:
        manifest = json.loads(p.read_text(encoding="utf-8"))
    except (json.JSONDecodeError, ValueError) as e:
        sys.stderr.write(f"ERROR: manifest is not valid JSON: {e}\n")
        sys.exit(2)
    if not isinstance(manifest, dict):
        sys.stderr.write("ERROR: manifest JSON must be an object\n")
        sys.exit(2)

    claims = check(manifest)
    rows = manifest.get("split_assignment") or []
    partitions: dict[str, int] = {}
    patients: set = set()
    for r in rows:
        sp = SPLIT_SYNONYM.get(_norm(r.get("split")), _norm(r.get("split")))
        if sp:
            partitions[sp] = partitions.get(sp, 0) + 1
        pid = r.get("patient_id") or r.get("subject_id") or r.get("patient") or r.get("id")
        if pid is not None:
            patients.add(str(pid))
    n_major = sum(1 for c in claims if c["severity"] == "Major")
    return {
        "manifest": str(p),
        "n_transforms": len(manifest.get("transforms") or []),
        "n_units": len(rows),
        "n_patients": len(patients),
        "partitions": dict(sorted(partitions.items())),
        "seed": manifest.get("split_seed"),
        "claims": claims,
        "summary": {
            "n_claims": len(claims),
            "n_major": n_major,
            "n_flag": len(claims) - n_major,
            "verdict": "MAJOR_CANDIDATE" if n_major else "OK",
        },
    }


def render(result: dict) -> str:
    lines = ["| Check | Severity | Detail |", "|---|---|---|"]
    for c in result["claims"]:
        lines.append(f"| {c['verdict']} | {c['severity']} | {c['detail']} |")
    if len(lines) == 2:
        lines.append("| (none) | — | preprocessing manifest is leakage-safe |")
    return "\n".join(lines)


def main() -> int:
    ap = argparse.ArgumentParser(description="Data-stage preprocessing-leakage gate.")
    ap.add_argument("--manifest", required=True, help="preprocessing manifest JSON")
    ap.add_argument("--out", help="write JSON artifact to this path")
    ap.add_argument("--strict", action="store_true", help="exit 1 if any Major claim exists")
    ap.add_argument("--quiet", action="store_true", help="suppress stdout table")
    args = ap.parse_args()

    result = analyze(args.manifest)

    if not args.quiet:
        print("=" * 41)
        print(" Preprocessing-Leakage Gate (preprocess-imaging)")
        print("=" * 41)
        p = result
        print(f"  transforms={p['n_transforms']}  units={p['n_units']}  "
              f"patients={p['n_patients']}  partitions={p['partitions']}  seed={p['seed']}")
        print(render(result))
        print()
        s = result["summary"]
        if s["n_major"]:
            print(f"MAJOR candidate: {s['n_major']} preprocessing-leakage issue(s).")
        elif s["n_flag"]:
            print(f"MINOR flag: {s['n_flag']} preprocessing hygiene issue(s) (see table).")
        else:
            print("OK: preprocessing manifest is leakage-safe.")

    if args.out:
        Path(args.out).parent.mkdir(parents=True, exist_ok=True)
        Path(args.out).write_text(json.dumps({"detector": "check_preprocessing_leakage", **result}, indent=2), encoding="utf-8")
        if not args.quiet:
            print(f"\nwrote {args.out}")

    return 1 if (args.strict and result["summary"]["n_major"]) else 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env bash
# Regression test for the preprocessing-leakage gate (preprocess-imaging).
# Synthetic, PII-free JSON manifests reproduce each verdict class. Stdlib-only (python3).
set -u

HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SCRIPT="$HERE/../scripts/check_preprocessing_leakage.py"
CH="$HERE/../scripts/check_preprocessing_leakage_challenge"
TMP="$(mktemp -d -t preproc_XXXX)"
OUT="$TMP/out.json"
trap 'rm -rf "$TMP"' EXIT

fail=0
check() { local label="$1"; shift
    if "$@" >/dev/null 2>&1; then printf '  PASS  %s\n' "$label"
    else printf '  FAIL  %s\n' "$label"; fail=$((fail+1)); fi
}
has_verdict() { python3 -c "
import json,sys
d=json.load(open('$OUT'))
assert any(c['verdict']=='$1' for c in d['claims']), '$1 not found'
"; }
no_verdict() { python3 -c "
import json,sys
d=json.load(open('$OUT'))
assert not any(c['verdict']=='$1' for c in d['claims']), '$1 unexpectedly present'
"; }

[[ -f "$SCRIPT" ]] || { echo "ENV-ERR: script missing" >&2; exit 2; }

# (1) leak fixture -> 3 Major verdicts + exit 1
python3 "$SCRIPT" --manifest "$CH/fixture/manifest_leak.json" --out "$OUT" --strict --quiet >/dev/null 2>&1
check "exit 1 (leak manifest)" test "$?" -eq 1
check "PREPROCESS_BEFORE_SPLIT detected" has_verdict PREPROCESS_BEFORE_SPLIT
check "NORMALIZATION_LEAKAGE detected" has_verdict NORMALIZATION_LEAKAGE
check "PATIENT_CROSS_SPLIT detected" has_verdict PATIENT_CROSS_SPLIT
check "AUGMENTATION_ON_EVAL detected" has_verdict AUGMENTATION_ON_EVAL

# (2) clean fixture -> exit 0, no Major
python3 "$SCRIPT" --manifest "$CH/fixture/manifest_clean.json" --strict --quiet >/dev/null 2>&1
check "exit 0 (clean manifest)" test "$?" -eq 0

# (3) data-fitted transform after split with NO fit_scope -> Minor UNSPECIFIED_FIT_SCOPE, exit 0
cat > "$TMP/unspec.json" <<'EOF'
{"split_seed": 1,
 "transforms": [{"name": "scaler", "type": "minmax", "stage": "after_split"}],
 "split_assignment": [{"patient_id": "A", "split": "train"}, {"patient_id": "B", "split": "test"}]}
EOF
python3 "$SCRIPT" --manifest "$TMP/unspec.json" --out "$OUT" --quiet >/dev/null 2>&1
check "UNSPECIFIED_FIT_SCOPE detected" has_verdict UNSPECIFIED_FIT_SCOPE
python3 "$SCRIPT" --manifest "$TMP/unspec.json" --strict --quiet >/dev/null 2>&1
check "unspecified-scope is Minor (exit 0 under --strict)" test "$?" -eq 0

# (4) split present but no split_seed -> Minor MISSING_SEED
cat > "$TMP/noseed.json" <<'EOF'
{"transforms": [{"name": "z", "type": "standardize", "fit_scope": "train", "stage": "after_split"}],
 "split_assignment": [{"patient_id": "A", "split": "train"}, {"patient_id": "B", "split": "test"}]}
EOF
python3 "$SCRIPT" --manifest "$TMP/noseed.json" --out "$OUT" --quiet >/dev/null 2>&1
check "MISSING_SEED detected" has_verdict MISSING_SEED

# (5) per-image normalisation BEFORE split -> NOT a leak (sample scope is leakage-free)
cat > "$TMP/persample.json" <<'EOF'
{"split_seed": 3,
 "transforms": [{"name": "perimg", "type": "normalization", "fit_scope": "sample", "stage": "before_split"}],
 "split_assignment": [{"patient_id": "A", "split": "train"}, {"patient_id": "B", "split": "test"}]}
EOF
python3 "$SCRIPT" --manifest "$TMP/persample.json" --out "$OUT" --quiet >/dev/null 2>&1
check "per-image transform before split does NOT fire PREPROCESS_BEFORE_SPLIT" no_verdict PREPROCESS_BEFORE_SPLIT
python3 "$SCRIPT" --manifest "$TMP/persample.json" --strict --quiet >/dev/null 2>&1
check "exit 0 on per-image-safe manifest" test "$?" -eq 0

# (6) resampling. A target spacing chosen in advance is fixed and never leaks; a target
# derived from the cohort (nnU-Net takes a percentile of the dataset fingerprint) is a
# fitted parameter, and fitting it over every case carries held-out geometry into the
# training grid exactly as an intensity statistic would.
cat > "$TMP/resample_fitted.json" <<'EOF'
{"split_seed": 42,
 "transforms": [{"name": "resample_to_fingerprint_median", "type": "resample",
                 "fit_scope": "all", "stage": "before_split"}],
 "split_assignment": [{"patient_id": "A", "split": "train"}, {"patient_id": "B", "split": "test"}]}
EOF
python3 "$SCRIPT" --manifest "$TMP/resample_fitted.json" --out "$OUT" --quiet >/dev/null 2>&1
check "data-derived resample target before split fires PREPROCESS_BEFORE_SPLIT" \
      has_verdict PREPROCESS_BEFORE_SPLIT

cat > "$TMP/resample_fixed.json" <<'EOF'
{"split_seed": 42,
 "transforms": [{"name": "resample_to_1mm_iso", "type": "resample",
                 "fit_scope": "fixed", "stage": "before_split"}],
 "split_assignment": [{"patient_id": "A", "split": "train"}, {"patient_id": "B", "split": "test"}]}
EOF
python3 "$SCRIPT" --manifest "$TMP/resample_fixed.json" --out "$OUT" --quiet >/dev/null 2>&1
check "resample to a declared fixed spacing does NOT fire PREPROCESS_BEFORE_SPLIT" \
      no_verdict PREPROCESS_BEFORE_SPLIT
python3 "$SCRIPT" --manifest "$TMP/resample_fixed.json" --strict --quiet >/dev/null 2>&1
check "exit 0 on fixed-spacing resample" test "$?" -eq 0

# (7) the shipped challenge card passes
check "challenge verify.sh passes" bash "$CH/verify.sh"

echo "fail=$fail"; [[ "$fail" -eq 0 ]] && echo "ALL PASS" || echo "FAILURES: $fail"
exit "$fail"