# Fine-tuning guide (model-scaffold — `--task finetune`)

Load-on-demand notes for the target workflow of a clinician-researcher: **fine-tune an
existing pretrained model on collected clinical data**, rather than train from scratch or
design a new architecture. `scaffold.py --task finetune` emits a leakage-safe transfer-learning
repo (frozen→unfrozen schedule, discriminative learning rates, a `PRETRAINED.md` provenance
record); this guide covers the decisions it deliberately leaves to you.

## When to fine-tune (vs train from scratch)
Fine-tuning a pretrained backbone is the right default whenever the labelled clinical set is
small (hundreds–low-thousands of patients) — the regime almost every solo clinical study is
in. From-scratch training needs far more data to beat a fine-tuned ImageNet/medical backbone.
Pick the backbone in `/architecture-zoo`; adapt it here.

## The provenance record is load-bearing — `PRETRAINED.md`
A fine-tune is only reproducible if the **exact** pretrained weights are recorded. `--task
finetune` emits `PRETRAINED.md` with `[VERIFY]` fields; fill them before publishing:
- **Source** — the exact model + weights tag (`timm:resnet50.a1_in1k`, `MedSAM`, a URL/DOI),
  set via `--from-pretrained`.
- **Pretraining data** — the dataset the backbone was pretrained on. **Confirm it does not
  overlap this study's test set.** A backbone pretrained on (or including) your evaluation
  images is *pretraining-set contamination*: leakage that no train/val/test split table can
  see, because it entered through the weights, not the split. This is the fine-tuning analogue
  of benchmark contamination in LLM evaluation (`/mllm-eval`).
- **License**, **checkpoint hash** (sha256), **access date** — for auditability.

`check_training_hygiene.py` fires `PRETRAINED_PROVENANCE_MISSING` (Minor) when a training
script loads pretrained weights (`pretrained=True` / `from_pretrained`) but the repo carries
no `PRETRAINED.md` and no `pretrained:` block in `config.yaml`. The scaffold passes by
construction; a hand-rolled repo (a copied Kaggle notebook that does
`timm.create_model(..., pretrained=True)` and records nothing) fails.

## The frozen→unfrozen schedule
The emitted `train.py` implements the standard two-phase transfer schedule:
1. **Freeze the backbone, warm up the fresh head** for a few epochs (`FREEZE_EPOCHS`). The
   randomly-initialised head would otherwise back-propagate large gradients that damage the
   pretrained features.
2. **Unfreeze with discriminative learning rates** — a small learning rate for the pretrained
   backbone (`BACKBONE_LR`, e.g. 1e-5) and a larger one for the head (`HEAD_LR`, e.g. 1e-3).
   The backbone should *adapt*, not be overwritten.

Tune the two rates and `FREEZE_EPOCHS` per task. For very small datasets, keep more of the
backbone frozen (train only the last block + head) and lean on regularisation (weight decay,
dropout, strong but physiology-preserving augmentation, early stopping on the **val** split).

## BatchNorm and small clinical batches
Fine-tuning with small batches destabilises BatchNorm running statistics. Options: keep the
backbone's BN layers in `eval()` (frozen running stats) while the backbone is frozen; switch
to GroupNorm; or use a larger effective batch (gradient accumulation). Record the choice in
Methods — it materially affects reproducibility.

## MedSAM / SAM adaptation
Promptable foundation segmenters (SAM, **MedSAM**) are adapted, not retrained. The common,
solo-doable path is **adapter / decoder fine-tuning**: freeze the heavy image encoder and
fine-tune only the mask decoder (and optionally a lightweight adapter or the prompt encoder)
on your labelled masks. Integrate the upstream implementation
([MedSAM](https://github.com/bowang-lab/MedSAM),
[SAM](https://github.com/facebookresearch/segment-anything)) — do not reimplement it. Keep the
scaffold's split-reading contract (`splits/split_assignment.csv`) so the patient-level split
is preserved, and record the SAM/MedSAM checkpoint in `PRETRAINED.md`. Prompt design (points /
boxes / automatic) is part of the method — report it, and report whether prompts used any
ground-truth information at test time (a leakage trap unique to promptable models).

## Diffusion augmentation — train split only
A diffusion model can synthesise extra training images (an off-the-shelf augmentation, **not**
a novel method here). Two hard rules keep it leakage-safe:
1. **Any generative model used for augmentation must be trained on the TRAIN split only.** If
   the diffusion (or GAN) model saw val/test images, its samples leak that distribution into
   training — the generative analogue of fitting a normaliser on full data
   (`/preprocess-imaging` `NORMALIZATION_LEAKAGE`).
2. **Synthetic images augment TRAIN only, never val/test.** Evaluating on — or augmenting —
   the held-out split with synthetic data invalidates the metric. `check_preprocessing_leakage`
   flags `AUGMENTATION_ON_EVAL`; declare synthetic augmentation in the preprocessing manifest
   as `applies_to: ["train"]`.
Report the synthetic:real ratio and show the result holds without synthetic data (a sensitivity
analysis) — reviewers discount models propped up by synthetic training data.

## Reporting
Fill `methods_stub.md` and `PRETRAINED.md`, then hand off to `/write-paper` +
`/check-reporting`. Fine-tuning specifics reviewers expect: the pretrained source + its
pretraining data (contamination check), the freeze/unfreeze schedule and learning rates,
and — for SAM/MedSAM — the prompt protocol. Report held-out metrics as **mean ± SD over ≥ 3
seeds** (`/model-evaluation` → `/analyze-stats`); fine-tuning is seed-sensitive on small data.

## Hand-offs
- Backbone / architecture choice (incl. SAM/MedSAM, diffusion) → `/architecture-zoo`.
- Data-stage leakage (normalisation fit, augmentation-on-eval, slice crossing) →
  `/preprocess-imaging` (`check_preprocessing_leakage`).
- Split / validation-design audit → `/model-validation` (`check_split_leakage.py`).
- Held-out metrics + CIs → `/model-evaluation` → `/analyze-stats`.
- Provenance/model documentation → `/model-card` (Model Card + Datasheet).

# MLOps wiring guide (model-scaffold)

Load-on-demand notes for taking a scaffolded repo through a **reproducible, reportable**
training run. This is a **wiring and reporting** reference, **not** a training-loop,
hyperparameter-search, or experiment-tracking reimplementation. Training and tracking stay
with the frameworks — MONAI / nnU-Net / timm for the model, Weights & Biases / MLflow /
TensorBoard for tracking, DVC / git-lfs for data — and this guide covers how to wire them so
the run is reproducible and what to record so a reviewer can trust it. Everything here points
to a framework; nothing here replaces one (the [ROADMAP out-of-scope clause](../../../ROADMAP.md)).

## The reproducibility contract (what a run must be able to reproduce)
A training run is reproducible when, from the recorded artifacts alone, someone else can
re-derive the same result. That needs five things pinned, all of which the scaffold already
emits a slot for:
1. **Code** — the git commit of the repo (dirty trees are not reproducible; commit first).
2. **Config** — `config.yaml` as the single source of truth (task, arch, channels, split
   fractions, seed). Do not scatter hyperparameters across the CLI and the code.
3. **Data** — a content hash of the exact dataset (see `/version-dataset`), not just a path.
4. **Seed + determinism** — the scaffold's `seed_everything` seeds Python / NumPy / torch /
   CUDA and sets cuDNN deterministic; record the seed in the tracker and `REPRODUCIBILITY.md`.
5. **Environment** — a pinned `requirements.txt` (exact versions), CUDA / driver, GPU model.

## Experiment tracking (integrate, don't build one)
Log to **W&B**, **MLflow**, or **TensorBoard** — pick one, do not write a tracker. At run
start, log the whole `config.yaml`, the git commit, the dataset hash, and the seed as the run
config; during training log per-epoch train/val loss + the val metric; at the end log the best
checkpoint as an artifact. The tracker's run URL / ID becomes part of the Methods record. Keep
the scaffold's contract intact — the best model is still selected on the **val** split, the
**test** split is still touched once by `evaluate.py`.

## Config management
The emitted `config.yaml` is the SSOT for one run. For sweeps, drive them with **Hydra /
OmegaConf** (or the tracker's sweep feature) that *reads* this config — never hand-edit values
into `train.py`. A hyperparameter that isn't in the config isn't reproducible.

## Data & model versioning
- **Data** — hash the training set with `/version-dataset` and record the manifest hash in the
  run config, so "trained on dataset X" is verifiable, not asserted. For large data use **DVC**
  or **git-lfs**; commit the pointer, not the pixels.
- **Model** — document the trained model with `/model-card` (Model Card + Datasheet); for a
  fine-tune, the pretrained-weight provenance lives in `PRETRAINED.md`
  (see `references/finetuning_guide.md`).

## Environment capture
Pin `requirements.txt` to exact versions before publishing (`pip freeze`, or a `conda env
export`). For a hard reproducibility guarantee, a Dockerfile / container digest removes "works
on my machine". Record the CUDA / driver / GPU — some GPU ops are non-deterministic even with
cuDNN deterministic set, which is why metrics are reported as **mean ± SD over ≥ 3 seeds**, not
a single run.

## Pipeline orchestration — use the framework's, not a new one
For segmentation, **nnU-Net** owns preprocessing → training → inference as a pipeline; build
its `dataset.json` folds from the scaffold's `splits/split_assignment.csv` so the patient-level
split is preserved end to end. **MONAI bundles** package a model + transforms + metadata the
same way. Orchestrate with the framework's own pipeline; do not rebuild one.

## CI for an ML repo (what is worth gating)
CI should gate the **network-free, deterministic** properties — not full training. The scaffold
already ships these: `check_training_hygiene` (seeds, eval-mode inference, train-only loaders,
pretrained provenance) and the split-leakage proof (`/model-validation`). Add a **forward-pass
smoke** (build the model, one batch, check output shape) as a fast job. Do **not** put a real
training run in CI — it is slow, non-deterministic, and not what CI is for.

## What to report (the MLOps reporting checklist)
State, in Methods or a reproducibility appendix: the **compute environment** (framework +
versions, CUDA / driver, GPU), the **seed(s)** and that metrics are mean ± SD over ≥ 3 seeds,
the **config** (as a supplement or a tracker link), the **dataset version** (`/version-dataset`
hash), and the **tracking run** URL / ID. This is the reproducibility half of TRIPOD+AI /
CLAIM 2024; `/check-reporting` covers the items.

## Hand-offs
- Dataset hash / reproducibility-lock → `/version-dataset`.
- Model + dataset documentation → `/model-card`.
- Split / validation-design audit → `/model-validation`.
- Held-out metrics + CIs → `/model-evaluation` → `/analyze-stats`.
- Deployment uncertainty / OOD / monitoring → `/uncertainty-imaging`.
- Reporting fit → `/check-reporting` (TRIPOD+AI / CLAIM / DECIDE-AI).

# Training guide (model-scaffold)

Load-on-demand notes for taking a scaffolded repo to a trained, publishable model. The
scaffold gives you a leakage-safe, reproducible skeleton; this guide covers the
decisions it deliberately leaves to you.

## Swap the model for a production backbone
The emitted `model.py` is a small, CPU-runnable U-Net for the forward-pass smoke test.
For real work, integrate (do not reimplement):
- **MONAI** — `monai.networks.nets.UNet` / `SegResNet`, plus `monai.transforms` and
  `monai.metrics` (DiceMetric, HausdorffDistanceMetric, SurfaceDiceMetric). Keep the
  scaffold's `dataset.py` split-reading contract; replace the model + transforms.
- **nnU-Net (v2)** — for many segmentation tasks the strongest baseline. Use the
  scaffold's `splits/split_assignment.csv` to build the nnU-Net `dataset.json` folds so
  the patient-level split is preserved end to end.
- **3-D** — for volumetric CT/MR, switch to a 3-D U-Net / `SegResNet` and 3-D patches;
  use **TorchIO** for spatial/intensity augmentation.

## Augmentation — train split only
Apply augmentation to the **training** split only, never to val/test. Watch
modality-specific traps: do not horizontal-flip when laterality is a label; window CT
to a clinically sensible HU range before normalising; consider bias-field simulation
for MR. Fit any normalisation statistics on the **training** fold only (fitting on the
whole cohort is preprocessing-before-split leakage — see `/model-validation` MD1).

## Optimisation defaults that travel well
AdamW + a cosine or warmup schedule; gradient clipping for unstable losses; early
stopping / best-checkpoint on the **validation** split (never the test set); automatic
mixed precision (AMP) for speed. Report the metric as **mean ± SD over ≥ 3 seeds**, not
a single run (deep metrics move with seed/init; some GPU ops are non-deterministic even
with cuDNN deterministic set).

## Reproducibility record
Before publishing, complete `REPRODUCIBILITY.md`: pinned `requirements.txt`
(torch/monai/... exact versions), CUDA/driver, GPU model, the git commit of the repo,
and the seed. Pair with `/version-dataset` to hash the exact dataset the model trained
on. For reproducibility-safe wiring of experiment tracking (W&B / MLflow), config / data
/ environment versioning, CI-for-ML, and the MLOps reporting checklist, see
`mlops_guide.md`.

## Hand-offs
- Split / validation-design audit → `/model-validation` (run `check_split_leakage.py`
  on `splits/split_assignment.csv`).
- Held-out metrics (Dice + HD95/NSD, AUROC/AUPRC, calibration, CIs) → `/model-evaluation`
  then `/analyze-stats`.
- Figures (training curve, overlay, confusion) → `/make-figures`.
- Methods + reporting → `/write-paper` (fill the `[VERIFY]` placeholders) and
  `/check-reporting` (CLAIM 2024 / TRIPOD+AI / STARD-AI).

#!/usr/bin/env python3
"""Training-script reproducibility-hygiene linter for a generated model repo (model-scaffold).

A CONSERVATIVE, AST-based linter (it flags, it does not prove) for the network-free
hygiene properties a medical-imaging training repo must have. It is the training-code
analogue of check_generated_code.py: same posture — parse the source, report missing
patterns, never execute torch. It checks the *presence* of the reproducibility
constructs, which is reliably decidable from the AST; it deliberately does NOT claim
to prove semantic correctness of the training loop.

CHECKS (verdicts):
  1. SEED_INCOMPLETE         (Major)  the training script must seed every RNG that
                                      affects a run: random.seed, numpy
                                      (np.random.seed / numpy.random.seed),
                                      torch.manual_seed, torch.cuda.manual_seed_all.
                                      Reports which calls are missing.
  2. MISSING_EVAL_MODE       (Major)  the evaluation/inference script must call
                                      model.eval() AND wrap inference in
                                      torch.no_grad() (or torch.inference_mode());
                                      otherwise dropout/batchnorm stay in train mode
                                      and gradients are tracked.
  3. TRAIN_ON_NONTRAIN_SPLIT (Major)  a training-style DataLoader (shuffle=True) built
                                      from a dataset constructed with split="val" or
                                      split="test" — training on a non-train split.
  4. CUDNN_NONDETERMINISTIC  (Minor)  torch.backends.cudnn.deterministic is not set
                                      True in the training script.
  5. EVAL_SHUFFLE            (Minor)  an evaluation DataLoader uses shuffle=True
                                      (reorders the test set; harmless for metrics but
                                      a smell, and breaks index-aligned outputs).
  6. PRETRAINED_PROVENANCE_MISSING (Minor)  the training script loads PRETRAINED weights
                                      (a `pretrained=True` kwarg or a `from_pretrained`
                                      call — fine-tuning / transfer learning) but the repo
                                      records no pretrained-weight provenance (no non-empty
                                      PRETRAINED.md and no `pretrained:` block in
                                      config.yaml). A fine-tune whose starting checkpoint is
                                      unrecorded is not reproducible. Fires only in --repo
                                      mode (the provenance record is a repo-level artifact).

INPUTS
  --repo   a scaffolded repo directory; train.py and evaluate.py are auto-found.
  --train  explicit path to the training script (overrides --repo discovery).
  --eval   explicit path to the evaluation/inference script.
  (Give --repo, or --train and/or --eval.)

OUTPUT
  A table (stdout) and, with --out, a JSON artifact:
    {train, eval, claims[{verdict, severity, detail, where}], summary}
  SEED_INCOMPLETE / MISSING_EVAL_MODE / TRAIN_ON_NONTRAIN_SPLIT are Major.

Stdlib-only (ast / 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 ast
import json
import sys
from pathlib import Path


def _attr_chain(node: ast.AST) -> str:
    """Dotted name for a Name/Attribute chain ('torch.cuda.manual_seed_all')."""
    parts = []
    while isinstance(node, ast.Attribute):
        parts.append(node.attr)
        node = node.value
    if isinstance(node, ast.Name):
        parts.append(node.id)
    return ".".join(reversed(parts))


def _kw(call: ast.Call, name: str):
    for k in call.keywords:
        if k.arg == name:
            return k.value
    return None


def _is_true(node) -> bool:
    return isinstance(node, ast.Constant) and node.value is True


def _scan(src: str):
    """Extract the hygiene-relevant facts from one script's AST."""
    tree = ast.parse(src)
    facts = {
        "seed_calls": set(),          # normalized RNG seed call chains seen
        "has_eval": False,            # any .eval() call
        "has_no_grad": False,         # torch.no_grad / inference_mode used
        "cudnn_determ": False,        # cudnn.deterministic = True
        "dataset_split": {},          # var name -> split literal it was built with
        "loaders": [],                # (first_arg_name, shuffle_bool)
        "loads_pretrained": False,    # a pretrained=True kwarg or a from_pretrained call
    }
    # dataset var <- Ctor(..., split="X")
    for node in ast.walk(tree):
        if isinstance(node, ast.Assign) and isinstance(node.value, ast.Call):
            sp = _kw(node.value, "split")
            if isinstance(sp, ast.Constant) and isinstance(sp.value, str):
                for tgt in node.targets:
                    if isinstance(tgt, ast.Name):
                        facts["dataset_split"][tgt.id] = sp.value
        if isinstance(node, ast.Assign) and isinstance(node.value, ast.Attribute):
            # torch.backends.cudnn.deterministic = True
            pass

    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            chain = _attr_chain(node.func)
            tail = chain.split(".")[-1]
            if tail in ("seed", "manual_seed", "manual_seed_all"):
                # normalize: random.seed / numpy seed / torch(.cuda).manual_seed*
                if chain.endswith("random.seed") and not chain.startswith(("np", "numpy")):
                    facts["seed_calls"].add("random")
                elif "random.seed" in chain or chain.endswith("np.random.seed"):
                    facts["seed_calls"].add("numpy")
                elif chain.endswith("manual_seed_all"):
                    facts["seed_calls"].add("torch.cuda")
                elif chain.endswith("manual_seed"):
                    facts["seed_calls"].add("torch")
            if tail == "eval" and isinstance(node.func, ast.Attribute):
                facts["has_eval"] = True
            if chain.endswith(("no_grad", "inference_mode")):
                facts["has_no_grad"] = True
            if tail == "DataLoader":
                first = node.args[0] if node.args else None
                name = first.id if isinstance(first, ast.Name) else None
                sh = _kw(node, "shuffle")
                facts["loaders"].append((name, _is_true(sh)))
            # pretrained-weight load: `...(pretrained=True)` or a `...from_pretrained(...)` call
            if _is_true(_kw(node, "pretrained")) or tail == "from_pretrained":
                facts["loads_pretrained"] = True
        # cudnn.deterministic = True (assignment to an Attribute target)
        if isinstance(node, ast.Assign):
            for tgt in node.targets:
                if isinstance(tgt, ast.Attribute) and tgt.attr == "deterministic":
                    if "cudnn" in _attr_chain(tgt) and _is_true(node.value):
                        facts["cudnn_determ"] = True
    return facts


def _has_pretrained_provenance(repo: Path) -> bool:
    """True if the repo records pretrained-weight provenance: a non-empty PRETRAINED.md, or
    a `pretrained:` block in config.yaml."""
    pm = repo / "PRETRAINED.md"
    if pm.is_file() and pm.read_text(encoding="utf-8").strip():
        return True
    cfg = repo / "config.yaml"
    if cfg.is_file():
        for line in cfg.read_text(encoding="utf-8").splitlines():
            if line.strip().startswith("pretrained:"):
                return True
    return False


def analyze(train: str | None, eval_: str | None, repo: str | None = None) -> dict:
    claims = []
    tfacts = _scan(Path(train).read_text(encoding="utf-8")) if train else None
    efacts = _scan(Path(eval_).read_text(encoding="utf-8")) if eval_ else None

    if tfacts is not None:
        need = {"random", "numpy", "torch", "torch.cuda"}
        missing = sorted(need - tfacts["seed_calls"])
        if missing:
            claims.append({
                "verdict": "SEED_INCOMPLETE", "severity": "Major",
                "detail": f"training script does not seed: {', '.join(missing)} "
                          f"(found: {', '.join(sorted(tfacts['seed_calls'])) or 'none'})",
                "where": Path(train).name,
            })
        if not tfacts["cudnn_determ"]:
            claims.append({
                "verdict": "CUDNN_NONDETERMINISTIC", "severity": "Minor",
                "detail": "torch.backends.cudnn.deterministic is not set True in the training script",
                "where": Path(train).name,
            })
        # training on a non-train split (shuffle=True loader from a val/test dataset)
        for name, shuffle in tfacts["loaders"]:
            sp = tfacts["dataset_split"].get(name)
            if shuffle and sp in ("val", "test"):
                claims.append({
                    "verdict": "TRAIN_ON_NONTRAIN_SPLIT", "severity": "Major",
                    "detail": f"a shuffled (training-style) DataLoader is built from dataset "
                              f"'{name}' constructed with split=\"{sp}\" — training on the {sp} split",
                    "where": Path(train).name,
                })
        # fine-tuning: pretrained weights loaded but no provenance recorded (repo mode only)
        if tfacts["loads_pretrained"] and repo and not _has_pretrained_provenance(Path(repo)):
            claims.append({
                "verdict": "PRETRAINED_PROVENANCE_MISSING", "severity": "Minor",
                "detail": "the training script loads pretrained weights (fine-tuning) but the "
                          "repo records no pretrained-weight provenance (a non-empty PRETRAINED.md "
                          "or a 'pretrained:' block in config.yaml). Record the exact "
                          "source / checkpoint / license / hash so the fine-tune is reproducible.",
                "where": Path(train).name,
            })

    if efacts is not None:
        if not (efacts["has_eval"] and efacts["has_no_grad"]):
            miss = []
            if not efacts["has_eval"]:
                miss.append("model.eval()")
            if not efacts["has_no_grad"]:
                miss.append("torch.no_grad()/inference_mode()")
            claims.append({
                "verdict": "MISSING_EVAL_MODE", "severity": "Major",
                "detail": f"evaluation script is missing {', '.join(miss)} before inference "
                          f"(dropout/batchnorm stay in train mode and gradients are tracked)",
                "where": Path(eval_).name,
            })
        for name, shuffle in efacts["loaders"]:
            if shuffle:
                claims.append({
                    "verdict": "EVAL_SHUFFLE", "severity": "Minor",
                    "detail": "an evaluation DataLoader uses shuffle=True (reorders the test set)",
                    "where": Path(eval_).name,
                })

    n_major = sum(1 for c in claims if c["severity"] == "Major")
    return {
        "train": train, "eval": eval_, "claims": claims,
        "summary": {"n_claims": len(claims), "n_major": n_major,
                    "n_minor": 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) | — | seeds all RNGs, cuDNN deterministic, eval() + no_grad() inference |")
    return "\n".join(lines)


def _resolve(repo: str | None, train: str | None, eval_: str | None):
    if repo:
        r = Path(repo)
        if not r.is_dir():
            sys.stderr.write(f"ERROR: --repo not a directory: {repo}\n")
            sys.exit(2)
        train = train or (str(r / "train.py") if (r / "train.py").is_file() else None)
        eval_ = eval_ or (str(r / "evaluate.py") if (r / "evaluate.py").is_file() else None)
    for label, p in (("--train", train), ("--eval", eval_)):
        if p and not Path(p).is_file():
            sys.stderr.write(f"ERROR: {label} not found: {p}\n")
            sys.exit(2)
    if not train and not eval_:
        sys.stderr.write("ERROR: nothing to check; pass --repo or --train/--eval\n")
        sys.exit(2)
    return train, eval_


def main() -> int:
    ap = argparse.ArgumentParser(description="Training-script reproducibility-hygiene linter (model-scaffold).")
    ap.add_argument("--repo", help="scaffolded repo directory (auto-finds train.py / evaluate.py)")
    ap.add_argument("--train", help="explicit training script path")
    ap.add_argument("--eval", dest="eval_", help="explicit evaluation/inference script path")
    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()

    train, eval_ = _resolve(args.repo, args.train, args.eval_)
    result = analyze(train, eval_, repo=args.repo)

    if not args.quiet:
        print("=" * 41)
        print(" Training Hygiene (model-scaffold)")
        print("=" * 41)
        print(render(result))
        print()
        s = result["summary"]
        print(f"MAJOR candidate: {s['n_major']} hygiene issue(s)." if s["n_major"]
              else "OK: training/evaluation reproducibility hygiene present.")

    if args.out:
        Path(args.out).parent.mkdir(parents=True, exist_ok=True)
        Path(args.out).write_text(json.dumps({"detector": "check_training_hygiene", **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 python3
"""Reproducible medical-imaging training-repo scaffold generator (model-scaffold).

Emits a runnable PyTorch training repo for a medical-imaging task with the
reproducibility guarantees baked in *by construction*, so the produced code passes the
lane's integrity gates without hand-editing:

  - a PATIENT-LEVEL, SEED-LOCKED split written as an auditable artifact
    (`splits/split_assignment.csv` + `splits/split_seed.txt`) — disjoint by construction
    (so `check_split_leakage.py` passes), task-independent;
  - a `train.py` that seeds random / numpy / torch (+ cuda) and sets cuDNN deterministic,
    builds the training DataLoader from the TRAIN split only, and an `evaluate.py` that
    runs inference under `model.eval()` + `torch.no_grad()` (so `check_training_hygiene.py`
    passes);
  - `model.py`, `dataset.py`, `losses.py`, `config.yaml`, `requirements.txt`,
    `REPRODUCIBILITY.md`, and a `methods_stub.md` with `[VERIFY: ...]` placeholders.

Tasks (`--task`):
  segmentation   2-D U-Net (Dice + BCE).
  classification small multi-label CNN (BCEWithLogits); swap in a `timm` backbone.
  detection      torchvision Faster R-CNN (ResNet-FPN); FROC / mAP downstream.
  synthesis      Pix2Pix-style U-Net generator + PatchGAN (image-to-image; L1 + GAN).
  ssl            SimCLR encoder + projection head (NT-Xent); pretrain then fine-tune.
  finetune       transfer learning — fine-tune a PRETRAINED backbone (frozen->unfrozen
                 schedule + discriminative LRs) on collected clinical data; the pretrained
                 weight source (`--from-pretrained`) is recorded in PRETRAINED.md so the
                 fine-tune is reproducible. See references/finetuning_guide.md.

It INTEGRATES with the ecosystem (PyTorch; MONAI / nnU-Net / TorchIO / timm / torchvision
referenced in the generated requirements) — it does not reimplement them. The generated
repo is what the user runs on a GPU; the lane's gates verify only the NETWORK-FREE parts
(split disjointness, training hygiene, forward shape).

INPUTS
  --manifest   CSV: one row per image with a patient/subject ID column (+ image/label paths).
  --id-col     ID column (auto-detected: patient_id / subject_id / case_id / id).
  --task       segmentation (default) | classification | detection | synthesis | ssl.
  --out        output repo directory (default: model_repo).
  --seed / --val-frac / --test-frac / --in-channels / --out-channels / --base-channels.

OUTPUT
  A self-contained repo under --out (tree printed to stdout). Deterministic given the
  manifest + seed.

Stdlib + numpy only (numpy is used solely for the deterministic patient-level split; the
GENERATED code uses torch). Exit codes: 0 ok, 2 input/usage error.
"""

from __future__ import annotations

import argparse
import csv
import sys
from pathlib import Path

import numpy as np

ID_HINTS = ("patient_id", "subject_id", "case_id", "patientid", "subjectid", "id", "pid", "mrn")

# Shared determinism preamble, interpolated into every train.py via __SEED_FN__ so each
# emitted training script seeds every RNG and sets cuDNN deterministic (check_training_hygiene).
SEED_FN = '''def seed_everything(seed):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)
    torch.backends.cudnn.deterministic = True
    torch.backends.cudnn.benchmark = False'''

# --------------------------------------------------------------------------- #
# Per-task templates. Placeholders are __TOKENS__ replaced by render().         #
# --------------------------------------------------------------------------- #

# ---- segmentation (2-D U-Net) ----
SEG_MODEL = r'''"""Small configurable 2-D U-Net (generated by model-scaffold).

Plain encoder-decoder with skip connections (Ronneberger 2015); small and CPU-runnable so
a forward pass can be smoke-tested. Swap in MONAI UNet/SegResNet or an nnU-Net plan.
"""
import torch
import torch.nn as nn


def _cbr(cin, cout):
    return nn.Sequential(
        nn.Conv2d(cin, cout, 3, padding=1), nn.BatchNorm2d(cout), nn.ReLU(inplace=True),
        nn.Conv2d(cout, cout, 3, padding=1), nn.BatchNorm2d(cout), nn.ReLU(inplace=True),
    )


class UNet(nn.Module):
    def __init__(self, in_channels=__IN_CH__, out_channels=__OUT_CH__, base=__BASE__):
        super().__init__()
        self.enc1 = _cbr(in_channels, base)
        self.enc2 = _cbr(base, base * 2)
        self.enc3 = _cbr(base * 2, base * 4)
        self.pool = nn.MaxPool2d(2)
        self.bottleneck = _cbr(base * 4, base * 8)
        self.up3 = nn.ConvTranspose2d(base * 8, base * 4, 2, stride=2)
        self.dec3 = _cbr(base * 8, base * 4)
        self.up2 = nn.ConvTranspose2d(base * 4, base * 2, 2, stride=2)
        self.dec2 = _cbr(base * 4, base * 2)
        self.up1 = nn.ConvTranspose2d(base * 2, base, 2, stride=2)
        self.dec1 = _cbr(base * 2, base)
        self.head = nn.Conv2d(base, out_channels, 1)

    def forward(self, x):
        e1 = self.enc1(x)
        e2 = self.enc2(self.pool(e1))
        e3 = self.enc3(self.pool(e2))
        b = self.bottleneck(self.pool(e3))
        d3 = self.dec3(torch.cat([self.up3(b), e3], dim=1))
        d2 = self.dec2(torch.cat([self.up2(d3), e2], dim=1))
        d1 = self.dec1(torch.cat([self.up1(d2), e1], dim=1))
        return self.head(d1)


def build_model():
    return UNet()
'''

SEG_LOSSES = r'''"""Segmentation losses (generated by model-scaffold)."""
import torch
import torch.nn as nn


class DiceBCELoss(nn.Module):
    """BCEWithLogits + soft Dice — a robust default for imbalanced masks."""

    def __init__(self, smooth=1.0):
        super().__init__()
        self.bce = nn.BCEWithLogitsLoss()
        self.smooth = smooth

    def forward(self, logits, target):
        bce = self.bce(logits, target)
        probs = torch.sigmoid(logits)
        num = 2 * (probs * target).sum(dim=(1, 2, 3)) + self.smooth
        den = probs.sum(dim=(1, 2, 3)) + target.sum(dim=(1, 2, 3)) + self.smooth
        dice = 1 - (num / den).mean()
        return bce + dice
'''

# ---- classification (small multi-label CNN) ----
CLS_MODEL = r'''"""Small multi-label classifier (generated by model-scaffold).

A compact CNN for a forward-pass smoke test; for production swap in a pretrained `timm`
backbone (ResNet/DenseNet/EfficientNet/ViT) with a multi-label head.
"""
import torch
import torch.nn as nn


class SmallClassifier(nn.Module):
    def __init__(self, in_channels=__IN_CH__, num_classes=__OUT_CH__, base=__BASE__):
        super().__init__()
        self.features = nn.Sequential(
            nn.Conv2d(in_channels, base, 3, padding=1), nn.BatchNorm2d(base), nn.ReLU(inplace=True),
            nn.MaxPool2d(2),
            nn.Conv2d(base, base * 2, 3, padding=1), nn.BatchNorm2d(base * 2), nn.ReLU(inplace=True),
            nn.MaxPool2d(2),
            nn.Conv2d(base * 2, base * 4, 3, padding=1), nn.BatchNorm2d(base * 4), nn.ReLU(inplace=True),
        )
        self.pool = nn.AdaptiveAvgPool2d(1)
        self.head = nn.Linear(base * 4, num_classes)

    def forward(self, x):
        x = self.pool(self.features(x)).flatten(1)
        return self.head(x)


def build_model():
    return SmallClassifier()
'''

CLS_LOSSES = r'''"""Classification loss (generated by model-scaffold)."""
import torch.nn as nn


def build_loss(pos_weight=None):
    """Multi-label classification: BCEWithLogits (use pos_weight for imbalance).
    For single-label, swap to nn.CrossEntropyLoss()."""
    return nn.BCEWithLogitsLoss(pos_weight=pos_weight)
'''

# ---- detection (torchvision Faster R-CNN) ----
DET_MODEL = r'''"""Detection model (generated by model-scaffold).

A torchvision Faster R-CNN (ResNet-50 FPN). torchvision returns a loss dict in train mode
and detections in eval mode. For instance segmentation use maskrcnn_resnet50_fpn.
"""
import torchvision
from torchvision.models.detection.faster_rcnn import FastRCNNPredictor


def build_model(num_classes=__OUT_CH__ + 1):  # +1 for background
    model = torchvision.models.detection.fasterrcnn_resnet50_fpn(weights=None)
    in_features = model.roi_heads.box_predictor.cls_score.in_features
    model.roi_heads.box_predictor = FastRCNNPredictor(in_features, num_classes)
    return model
'''

DET_LOSSES = r'''"""Detection losses (generated by model-scaffold).

torchvision detection models compute their own multi-task loss dict (classification +
box regression + objectness + rpn box) when called as model(images, targets) in train
mode. This module documents that the training loss is sum(loss_dict.values())."""


def reduce_loss(loss_dict):
    return sum(loss for loss in loss_dict.values())
'''

# ---- synthesis (Pix2Pix-style U-Net generator + PatchGAN) ----
SYN_MODEL = r'''"""Image-to-image synthesis model (generated by model-scaffold).

A small Pix2Pix-style U-Net generator + PatchGAN discriminator (Isola 2017), CPU-runnable
for a forward smoke test. For 3-D or diffusion, see MONAI generative models.
"""
import torch
import torch.nn as nn


def _down(cin, cout):
    return nn.Sequential(nn.Conv2d(cin, cout, 4, 2, 1), nn.BatchNorm2d(cout), nn.LeakyReLU(0.2, True))


def _up(cin, cout):
    return nn.Sequential(nn.ConvTranspose2d(cin, cout, 4, 2, 1), nn.BatchNorm2d(cout), nn.ReLU(True))


class UNetGenerator(nn.Module):
    def __init__(self, in_channels=__IN_CH__, out_channels=__OUT_CH__, base=__BASE__):
        super().__init__()
        self.d1, self.d2, self.d3 = _down(in_channels, base), _down(base, base * 2), _down(base * 2, base * 4)
        self.u3, self.u2 = _up(base * 4, base * 2), _up(base * 4, base)
        self.u1 = nn.ConvTranspose2d(base * 2, out_channels, 4, 2, 1)

    def forward(self, x):
        e1 = self.d1(x)
        e2 = self.d2(e1)
        e3 = self.d3(e2)
        u3 = self.u3(e3)
        u2 = self.u2(torch.cat([u3, e2], dim=1))
        return torch.tanh(self.u1(torch.cat([u2, e1], dim=1)))


class PatchGAN(nn.Module):
    def __init__(self, in_channels=__IN_CH__ + __OUT_CH__, base=__BASE__):
        super().__init__()
        self.net = nn.Sequential(
            nn.Conv2d(in_channels, base, 4, 2, 1), nn.LeakyReLU(0.2, True),
            nn.Conv2d(base, base * 2, 4, 2, 1), nn.BatchNorm2d(base * 2), nn.LeakyReLU(0.2, True),
            nn.Conv2d(base * 2, 1, 4, 1, 1),
        )

    def forward(self, src, tgt):
        return self.net(torch.cat([src, tgt], dim=1))


def build_model():
    return UNetGenerator()


def build_discriminator():
    return PatchGAN()
'''

SYN_LOSSES = r'''"""Synthesis losses (generated by model-scaffold): adversarial (GAN) + L1."""
import torch.nn as nn


def build_losses():
    """Returns (gan_loss, l1_loss). Generator loss = gan + lambda * L1 (lambda ~ 100)."""
    return nn.BCEWithLogitsLoss(), nn.L1Loss()
'''

# ---- ssl (SimCLR encoder + projection head) ----
SSL_MODEL = r'''"""Self-supervised encoder + projection head (generated by model-scaffold).

A SimCLR-style encoder (small CNN) + MLP projection head (Chen 2020). Pretrain with
NT-Xent on two augmented views, then fine-tune the encoder on labels.
"""
import torch
import torch.nn as nn


class Encoder(nn.Module):
    def __init__(self, in_channels=__IN_CH__, base=__BASE__, proj_dim=__BASE__ * 2):
        super().__init__()
        self.backbone = nn.Sequential(
            nn.Conv2d(in_channels, base, 3, padding=1), nn.BatchNorm2d(base), nn.ReLU(True),
            nn.MaxPool2d(2),
            nn.Conv2d(base, base * 2, 3, padding=1), nn.BatchNorm2d(base * 2), nn.ReLU(True),
            nn.AdaptiveAvgPool2d(1),
        )
        self.projector = nn.Sequential(nn.Linear(base * 2, base * 2), nn.ReLU(True),
                                       nn.Linear(base * 2, proj_dim))

    def forward(self, x):
        h = self.backbone(x).flatten(1)
        return h, nn.functional.normalize(self.projector(h), dim=1)


def build_model():
    return Encoder()
'''

SSL_LOSSES = r'''"""NT-Xent contrastive loss (SimCLR; generated by model-scaffold)."""
import torch
import torch.nn as nn


def nt_xent(z1, z2, temperature=0.5):
    n = z1.size(0)
    z = torch.cat([z1, z2], dim=0)
    sim = torch.mm(z, z.t()) / temperature
    sim.fill_diagonal_(-1e9)
    targets = torch.arange(n, device=z.device)
    targets = torch.cat([targets + n, targets], dim=0)
    return nn.functional.cross_entropy(sim, targets)
'''

# ---- per-task dataset templates ----
SEG_DATASET = r'''"""Dataset reading the FROZEN split (segmentation; generated by model-scaffold).
Replace _load_image/_load_label with your DICOM/NIfTI/TIFF reader."""
import csv
from pathlib import Path
import torch
from torch.utils.data import Dataset

ID_COL = "__ID_COL__"


def _read_split(repo_root):
    assign = {}
    with (Path(repo_root) / "splits" / "split_assignment.csv").open(encoding="utf-8") as f:
        for row in csv.DictReader(f):
            assign[row[ID_COL]] = row["split"]
    return assign


class ScaffoldDataset(Dataset):
    def __init__(self, manifest_csv, repo_root, split, transform=None):
        assign = _read_split(repo_root)
        self.rows = [r for r in csv.DictReader(open(manifest_csv, encoding="utf-8"))
                     if assign.get(r[ID_COL]) == split]
        self.transform = transform

    def __len__(self):
        return len(self.rows)

    def _load_image(self, row):
        raise NotImplementedError("plug in your image reader -> CxHxW float tensor")

    def _load_label(self, row):
        raise NotImplementedError("plug in your mask reader -> 1xHxW float tensor in {0,1}")

    def __getitem__(self, i):
        row = self.rows[i]
        x, y = self._load_image(row), self._load_label(row)
        if self.transform is not None:
            x, y = self.transform(x, y)
        return x, y
'''

CLS_DATASET = r'''"""Dataset reading the FROZEN split (classification; generated by model-scaffold).
Replace _load_image/_load_label with your reader."""
import csv
from pathlib import Path
import torch
from torch.utils.data import Dataset

ID_COL = "__ID_COL__"


def _read_split(repo_root):
    assign = {}
    with (Path(repo_root) / "splits" / "split_assignment.csv").open(encoding="utf-8") as f:
        for row in csv.DictReader(f):
            assign[row[ID_COL]] = row["split"]
    return assign


class ScaffoldDataset(Dataset):
    def __init__(self, manifest_csv, repo_root, split, transform=None):
        assign = _read_split(repo_root)
        self.rows = [r for r in csv.DictReader(open(manifest_csv, encoding="utf-8"))
                     if assign.get(r[ID_COL]) == split]
        self.transform = transform

    def __len__(self):
        return len(self.rows)

    def _load_image(self, row):
        raise NotImplementedError("plug in your image reader -> CxHxW float tensor")

    def _load_label(self, row):
        raise NotImplementedError("plug in your label reader -> float tensor [num_classes] (multi-label)")

    def __getitem__(self, i):
        row = self.rows[i]
        x, y = self._load_image(row), self._load_label(row)
        if self.transform is not None:
            x = self.transform(x)
        return x, y
'''

DET_DATASET = r'''"""Detection dataset reading the FROZEN split (generated by model-scaffold).
Returns (image, target) where target = {"boxes": Nx4, "labels": N} per torchvision."""
import csv
from pathlib import Path
import torch
from torch.utils.data import Dataset

ID_COL = "__ID_COL__"


def _read_split(repo_root):
    assign = {}
    with (Path(repo_root) / "splits" / "split_assignment.csv").open(encoding="utf-8") as f:
        for row in csv.DictReader(f):
            assign[row[ID_COL]] = row["split"]
    return assign


class ScaffoldDataset(Dataset):
    def __init__(self, manifest_csv, repo_root, split):
        assign = _read_split(repo_root)
        self.rows = [r for r in csv.DictReader(open(manifest_csv, encoding="utf-8"))
                     if assign.get(r[ID_COL]) == split]

    def __len__(self):
        return len(self.rows)

    def _load_image(self, row):
        raise NotImplementedError("plug in your image reader -> CxHxW float tensor")

    def _load_target(self, row):
        raise NotImplementedError('return {"boxes": FloatTensor[N,4], "labels": Int64Tensor[N]}')

    def __getitem__(self, i):
        row = self.rows[i]
        return self._load_image(row), self._load_target(row)
'''

SYN_DATASET = r'''"""Image-to-image synthesis dataset reading the FROZEN split (generated by model-scaffold).
Returns (source_image, target_image)."""
import csv
from pathlib import Path
import torch
from torch.utils.data import Dataset

ID_COL = "__ID_COL__"


def _read_split(repo_root):
    assign = {}
    with (Path(repo_root) / "splits" / "split_assignment.csv").open(encoding="utf-8") as f:
        for row in csv.DictReader(f):
            assign[row[ID_COL]] = row["split"]
    return assign


class ScaffoldDataset(Dataset):
    def __init__(self, manifest_csv, repo_root, split):
        assign = _read_split(repo_root)
        self.rows = [r for r in csv.DictReader(open(manifest_csv, encoding="utf-8"))
                     if assign.get(r[ID_COL]) == split]

    def __len__(self):
        return len(self.rows)

    def _load_source(self, row):
        raise NotImplementedError("plug in your source-modality reader -> CxHxW float tensor")

    def _load_target(self, row):
        raise NotImplementedError("plug in your target-modality reader -> CxHxW float tensor")

    def __getitem__(self, i):
        row = self.rows[i]
        return self._load_source(row), self._load_target(row)
'''

SSL_DATASET = r'''"""SSL dataset reading the FROZEN split (generated by model-scaffold).
Returns two augmented views of the same image (no labels at pretraining)."""
import csv
from pathlib import Path
import torch
from torch.utils.data import Dataset

ID_COL = "__ID_COL__"


def _read_split(repo_root):
    assign = {}
    with (Path(repo_root) / "splits" / "split_assignment.csv").open(encoding="utf-8") as f:
        for row in csv.DictReader(f):
            assign[row[ID_COL]] = row["split"]
    return assign


class ScaffoldDataset(Dataset):
    def __init__(self, manifest_csv, repo_root, split, augment=None):
        assign = _read_split(repo_root)
        self.rows = [r for r in csv.DictReader(open(manifest_csv, encoding="utf-8"))
                     if assign.get(r[ID_COL]) == split]
        self.augment = augment

    def __len__(self):
        return len(self.rows)

    def _load_image(self, row):
        raise NotImplementedError("plug in your image reader -> CxHxW float tensor")

    def __getitem__(self, i):
        x = self._load_image(self.rows[i])
        if self.augment is None:
            raise NotImplementedError("provide a stochastic augment() for two views")
        return self.augment(x), self.augment(x)
'''

# ---- per-task train templates (all hygienic by construction) ----
SEG_TRAIN = r'''"""Training entry point (segmentation; generated by model-scaffold).
Reproducible: every RNG seeded, cuDNN deterministic, train loader from the TRAIN split,
best model selected on the VAL split. No metric is hard-coded."""
import random
import numpy as np
import torch
from torch.utils.data import DataLoader

from dataset import ScaffoldDataset
from losses import DiceBCELoss
from model import build_model

SEED = __SEED__
REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"


__SEED_FN__


def main():
    seed_everything(SEED)
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    train_ds = ScaffoldDataset(MANIFEST, REPO_ROOT, split="train")
    val_ds = ScaffoldDataset(MANIFEST, REPO_ROOT, split="val")
    train_loader = DataLoader(train_ds, batch_size=4, shuffle=True, num_workers=4)
    val_loader = DataLoader(val_ds, batch_size=4, shuffle=False, num_workers=4)
    model = build_model().to(device)
    criterion = DiceBCELoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
    best_val = float("inf")
    for epoch in range(50):
        model.train()
        for x, y in train_loader:
            x, y = x.to(device), y.to(device)
            optimizer.zero_grad()
            loss = criterion(model(x), y)
            loss.backward()
            optimizer.step()
        model.eval()
        val_loss = 0.0
        with torch.no_grad():
            for x, y in val_loader:
                x, y = x.to(device), y.to(device)
                val_loss += criterion(model(x), y).item()
        val_loss /= max(len(val_loader), 1)
        if val_loss < best_val:
            best_val = val_loss
            torch.save({"epoch": epoch, "model": model.state_dict(), "seed": SEED}, "best.pt")


if __name__ == "__main__":
    main()
'''

CLS_TRAIN = r'''"""Training entry point (classification; generated by model-scaffold).
Reproducible: every RNG seeded, cuDNN deterministic, train loader from the TRAIN split."""
import random
import numpy as np
import torch
from torch.utils.data import DataLoader

from dataset import ScaffoldDataset
from losses import build_loss
from model import build_model

SEED = __SEED__
REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"


__SEED_FN__


def main():
    seed_everything(SEED)
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    train_ds = ScaffoldDataset(MANIFEST, REPO_ROOT, split="train")
    val_ds = ScaffoldDataset(MANIFEST, REPO_ROOT, split="val")
    train_loader = DataLoader(train_ds, batch_size=16, shuffle=True, num_workers=4)
    val_loader = DataLoader(val_ds, batch_size=16, shuffle=False, num_workers=4)
    model = build_model().to(device)
    criterion = build_loss()
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
    best_val = float("inf")
    for epoch in range(50):
        model.train()
        for x, y in train_loader:
            x, y = x.to(device), y.to(device)
            optimizer.zero_grad()
            loss = criterion(model(x), y)
            loss.backward()
            optimizer.step()
        model.eval()
        val_loss = 0.0
        with torch.no_grad():
            for x, y in val_loader:
                x, y = x.to(device), y.to(device)
                val_loss += criterion(model(x), y).item()
        val_loss /= max(len(val_loader), 1)
        if val_loss < best_val:
            best_val = val_loss
            torch.save({"epoch": epoch, "model": model.state_dict(), "seed": SEED}, "best.pt")


if __name__ == "__main__":
    main()
'''

DET_TRAIN = r'''"""Training entry point (detection; generated by model-scaffold).
Reproducible: every RNG seeded, cuDNN deterministic, train loader from the TRAIN split.
torchvision detection models return a loss dict in train mode."""
import random
import numpy as np
import torch
from torch.utils.data import DataLoader

from dataset import ScaffoldDataset
from losses import reduce_loss
from model import build_model

SEED = __SEED__
REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"


__SEED_FN__


def _collate(batch):
    return tuple(zip(*batch))


def main():
    seed_everything(SEED)
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    train_ds = ScaffoldDataset(MANIFEST, REPO_ROOT, split="train")
    train_loader = DataLoader(train_ds, batch_size=2, shuffle=True, num_workers=4, collate_fn=_collate)
    model = build_model().to(device)
    optimizer = torch.optim.SGD(model.parameters(), lr=5e-3, momentum=0.9, weight_decay=5e-4)
    for epoch in range(20):
        model.train()
        for images, targets in train_loader:
            images = [img.to(device) for img in images]
            targets = [{k: v.to(device) for k, v in t.items()} for t in targets]
            optimizer.zero_grad()
            loss = reduce_loss(model(images, targets))
            loss.backward()
            optimizer.step()
        torch.save({"epoch": epoch, "model": model.state_dict(), "seed": SEED}, "best.pt")


if __name__ == "__main__":
    main()
'''

SYN_TRAIN = r'''"""Training entry point (synthesis; generated by model-scaffold).
Reproducible: every RNG seeded, cuDNN deterministic, train loader from the TRAIN split.
Pix2Pix: generator loss = GAN + lambda * L1."""
import random
import numpy as np
import torch
from torch.utils.data import DataLoader

from dataset import ScaffoldDataset
from losses import build_losses
from model import build_model, build_discriminator

SEED = __SEED__
REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"
LAMBDA_L1 = 100.0


__SEED_FN__


def main():
    seed_everything(SEED)
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    train_ds = ScaffoldDataset(MANIFEST, REPO_ROOT, split="train")
    train_loader = DataLoader(train_ds, batch_size=4, shuffle=True, num_workers=4)
    gen, disc = build_model().to(device), build_discriminator().to(device)
    gan_loss, l1_loss = build_losses()
    opt_g = torch.optim.Adam(gen.parameters(), lr=2e-4, betas=(0.5, 0.999))
    opt_d = torch.optim.Adam(disc.parameters(), lr=2e-4, betas=(0.5, 0.999))
    for epoch in range(100):
        gen.train()
        disc.train()
        for src, tgt in train_loader:
            src, tgt = src.to(device), tgt.to(device)
            fake = gen(src)
            opt_d.zero_grad()
            d_real = disc(src, tgt)
            d_fake = disc(src, fake.detach())
            loss_d = 0.5 * (gan_loss(d_real, torch.ones_like(d_real)) +
                            gan_loss(d_fake, torch.zeros_like(d_fake)))
            loss_d.backward()
            opt_d.step()
            opt_g.zero_grad()
            d_fake = disc(src, fake)
            loss_g = gan_loss(d_fake, torch.ones_like(d_fake)) + LAMBDA_L1 * l1_loss(fake, tgt)
            loss_g.backward()
            opt_g.step()
        torch.save({"epoch": epoch, "gen": gen.state_dict(), "seed": SEED}, "best.pt")


if __name__ == "__main__":
    main()
'''

SSL_TRAIN = r'''"""Training entry point (self-supervised pretraining; generated by model-scaffold).
Reproducible: every RNG seeded, cuDNN deterministic, train loader from the TRAIN split.
SimCLR NT-Xent over two augmented views; fine-tune the encoder afterwards."""
import random
import numpy as np
import torch
from torch.utils.data import DataLoader

from dataset import ScaffoldDataset
from losses import nt_xent
from model import build_model

SEED = __SEED__
REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"


__SEED_FN__


def main():
    seed_everything(SEED)
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    train_ds = ScaffoldDataset(MANIFEST, REPO_ROOT, split="train")
    train_loader = DataLoader(train_ds, batch_size=32, shuffle=True, num_workers=4)
    model = build_model().to(device)
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
    for epoch in range(100):
        model.train()
        for v1, v2 in train_loader:
            v1, v2 = v1.to(device), v2.to(device)
            optimizer.zero_grad()
            _, z1 = model(v1)
            _, z2 = model(v2)
            loss = nt_xent(z1, z2)
            loss.backward()
            optimizer.step()
        torch.save({"epoch": epoch, "model": model.state_dict(), "seed": SEED}, "encoder.pt")


if __name__ == "__main__":
    main()
'''

# ---- per-task evaluate templates (all use model.eval() + no_grad()) ----
SEG_EVAL = r'''"""Held-out evaluation (segmentation; generated by model-scaffold).
Inference under model.eval() + torch.no_grad(); writes per-case predictions. Compute
Dice + HD95/NSD with CIs downstream via /model-evaluation + /analyze-stats."""
import csv
import torch
from torch.utils.data import DataLoader
from dataset import ScaffoldDataset
from model import build_model

REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"


def main():
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    test_loader = DataLoader(ScaffoldDataset(MANIFEST, REPO_ROOT, split="test"), batch_size=1, shuffle=False)
    model = build_model().to(device)
    model.load_state_dict(torch.load("best.pt", map_location=device)["model"])
    model.eval()
    rows = []
    with torch.no_grad():
        for i, (x, y) in enumerate(test_loader):
            probs = torch.sigmoid(model(x.to(device)))
            rows.append({"index": i, "pred_positive_voxels": int((probs > 0.5).sum().item())})
    with open("predictions.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["index", "pred_positive_voxels"])
        w.writeheader()
        w.writerows(rows)


if __name__ == "__main__":
    main()
'''

CLS_EVAL = r'''"""Held-out evaluation (classification; generated by model-scaffold).
Inference under model.eval() + torch.no_grad(); writes per-case scores. Compute
AUROC + AUPRC with CIs downstream via /model-evaluation + /analyze-stats."""
import csv
import torch
from torch.utils.data import DataLoader
from dataset import ScaffoldDataset
from model import build_model

REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"


def main():
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    test_loader = DataLoader(ScaffoldDataset(MANIFEST, REPO_ROOT, split="test"), batch_size=1, shuffle=False)
    model = build_model().to(device)
    model.load_state_dict(torch.load("best.pt", map_location=device)["model"])
    model.eval()
    rows = []
    with torch.no_grad():
        for i, (x, y) in enumerate(test_loader):
            scores = torch.sigmoid(model(x.to(device))).squeeze(0).tolist()
            rows.append({"index": i, "scores": scores})
    with open("predictions.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["index", "scores"])
        w.writeheader()
        w.writerows(rows)


if __name__ == "__main__":
    main()
'''

DET_EVAL = r'''"""Held-out evaluation (detection; generated by model-scaffold).
Inference under model.eval() + torch.no_grad(); writes detections. Compute FROC/mAP with a
stated IoU criterion downstream via /model-evaluation + /analyze-stats."""
import json
import torch
from torch.utils.data import DataLoader
from dataset import ScaffoldDataset
from model import build_model

REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"


def _collate(batch):
    return tuple(zip(*batch))


def main():
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    test_loader = DataLoader(ScaffoldDataset(MANIFEST, REPO_ROOT, split="test"), batch_size=1,
                             shuffle=False, collate_fn=_collate)
    model = build_model().to(device)
    model.load_state_dict(torch.load("best.pt", map_location=device)["model"])
    model.eval()
    out = []
    with torch.no_grad():
        for i, (images, _) in enumerate(test_loader):
            preds = model([img.to(device) for img in images])
            out.append({"index": i, "n_boxes": int(preds[0]["boxes"].shape[0])})
    json.dump(out, open("predictions.json", "w"))


if __name__ == "__main__":
    main()
'''

SYN_EVAL = r'''"""Held-out evaluation (synthesis; generated by model-scaffold).
Inference under generator.eval() + torch.no_grad(); writes synthesized images. Compute
SSIM/PSNR + a downstream-task metric via /model-evaluation."""
import torch
from torch.utils.data import DataLoader
from dataset import ScaffoldDataset
from model import build_model

REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"


def main():
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    test_loader = DataLoader(ScaffoldDataset(MANIFEST, REPO_ROOT, split="test"), batch_size=1, shuffle=False)
    gen = build_model().to(device)
    gen.load_state_dict(torch.load("best.pt", map_location=device)["gen"])
    gen.eval()
    with torch.no_grad():
        for i, (src, tgt) in enumerate(test_loader):
            fake = gen(src.to(device))
            torch.save(fake.cpu(), "synth_%d.pt" % i)


if __name__ == "__main__":
    main()
'''

SSL_EVAL = r'''"""Feature extraction (self-supervised; generated by model-scaffold).
Inference under model.eval() + torch.no_grad(); writes encoder embeddings for a
downstream linear-probe / fine-tune evaluation."""
import torch
from torch.utils.data import DataLoader
from dataset import ScaffoldDataset
from model import build_model

REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"


def _identity(x):
    return x


def main():
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    test_ds = ScaffoldDataset(MANIFEST, REPO_ROOT, split="test", augment=_identity)
    test_loader = DataLoader(test_ds, batch_size=1, shuffle=False)
    model = build_model().to(device)
    model.load_state_dict(torch.load("encoder.pt", map_location=device)["model"])
    model.eval()
    feats = []
    with torch.no_grad():
        for v1, _ in test_loader:
            h, _z = model(v1.to(device))
            feats.append(h.cpu())
    torch.save(torch.cat(feats, dim=0) if feats else torch.empty(0), "embeddings.pt")


if __name__ == "__main__":
    main()
'''

# ---- finetune (transfer learning: fine-tune a pretrained backbone) ----
FT_MODEL = r'''"""Fine-tuning backbone (transfer learning; generated by model-scaffold).

Wraps a PRETRAINED classification backbone with a fresh task head. The default is a small
CPU-runnable CNN so the forward pass smoke-tests without a GPU or timm; for real transfer
learning pass pretrained=True to load a `timm` backbone (ResNet/DenseNet/EfficientNet/ViT/
Swin) — whose weight provenance MUST be recorded in PRETRAINED.md. Freeze the backbone,
warm up the head, then unfreeze with discriminative learning rates
(see references/finetuning_guide.md).
"""
import torch.nn as nn

# The exact pretrained source, echoed into PRETRAINED.md (provenance). Override via
# `scaffold.py --from-pretrained ...`; e.g. "timm:resnet50.a1_in1k", "MedSAM", or a URL/DOI.
PRETRAINED_SOURCE = "__PRETRAINED_SOURCE__"


class _SmallBackbone(nn.Module):
    """CPU-runnable stand-in so the forward pass smoke-tests without timm."""

    def __init__(self, in_channels, base):
        super().__init__()
        self.body = nn.Sequential(
            nn.Conv2d(in_channels, base, 3, padding=1), nn.BatchNorm2d(base), nn.ReLU(True),
            nn.MaxPool2d(2),
            nn.Conv2d(base, base * 2, 3, padding=1), nn.BatchNorm2d(base * 2), nn.ReLU(True),
            nn.AdaptiveAvgPool2d(1),
        )
        self.num_features = base * 2

    def forward(self, x):
        return self.body(x).flatten(1)


def _build_backbone(in_channels, base, pretrained):
    if pretrained:
        try:
            import timm
        except ImportError:
            print("NOTE: timm not installed; using a random-init stand-in. Install timm to "
                  "load the pretrained weights recorded in PRETRAINED.md (PRETRAINED_SOURCE).")
        else:
            name = PRETRAINED_SOURCE.split(":", 1)[-1]
            return timm.create_model(name, pretrained=True, in_chans=in_channels, num_classes=0)
    return _SmallBackbone(in_channels, base)


class FineTuneModel(nn.Module):
    def __init__(self, in_channels=__IN_CH__, num_classes=__OUT_CH__, base=__BASE__, pretrained=False):
        super().__init__()
        self.backbone = _build_backbone(in_channels, base, pretrained)
        self.head = nn.Linear(self.backbone.num_features, num_classes)

    def set_backbone_trainable(self, trainable):
        for p in self.backbone.parameters():
            p.requires_grad = trainable

    def forward(self, x):
        return self.head(self.backbone(x))


def build_model(pretrained=False):
    return FineTuneModel(pretrained=pretrained)
'''

FT_LOSSES = r'''"""Fine-tuning loss (generated by model-scaffold)."""
import torch.nn as nn


def build_loss(weight=None):
    """Single-label multi-class transfer target: CrossEntropy (use weight for imbalance).
    For multi-label, swap to nn.BCEWithLogitsLoss()."""
    return nn.CrossEntropyLoss(weight=weight)
'''

FT_DATASET = r'''"""Dataset reading the FROZEN split (fine-tuning / classification; generated by model-scaffold).
Replace _load_image/_load_label with your DICOM/NIfTI/TIFF reader."""
import csv
from pathlib import Path
import torch
from torch.utils.data import Dataset

ID_COL = "__ID_COL__"


def _read_split(repo_root):
    assign = {}
    with (Path(repo_root) / "splits" / "split_assignment.csv").open(encoding="utf-8") as f:
        for row in csv.DictReader(f):
            assign[row[ID_COL]] = row["split"]
    return assign


class ScaffoldDataset(Dataset):
    def __init__(self, manifest_csv, repo_root, split, transform=None):
        assign = _read_split(repo_root)
        self.rows = [r for r in csv.DictReader(open(manifest_csv, encoding="utf-8"))
                     if assign.get(r[ID_COL]) == split]
        self.transform = transform

    def __len__(self):
        return len(self.rows)

    def _load_image(self, row):
        raise NotImplementedError("plug in your image reader -> CxHxW float tensor")

    def _load_label(self, row):
        raise NotImplementedError("plug in your label reader -> int64 class index (single-label)")

    def __getitem__(self, i):
        row = self.rows[i]
        x, y = self._load_image(row), self._load_label(row)
        if self.transform is not None:
            x = self.transform(x)
        return x, y
'''

FT_TRAIN = r'''"""Training entry point (fine-tuning / transfer learning; generated by model-scaffold).
Reproducible: every RNG seeded, cuDNN deterministic, train loader from the TRAIN split.
Transfer learning: load the PRETRAINED backbone (source recorded in PRETRAINED.md), freeze
it and warm up the fresh head, then unfreeze with discriminative learning rates. No metric
is hard-coded."""
import random
import numpy as np
import torch
from torch.utils.data import DataLoader

from dataset import ScaffoldDataset
from losses import build_loss
from model import build_model, PRETRAINED_SOURCE

SEED = __SEED__
REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"
FREEZE_EPOCHS = 5      # head-only warmup while the backbone is frozen
TOTAL_EPOCHS = 50
HEAD_LR = 1e-3         # discriminative LRs: the fresh head learns fast ...
BACKBONE_LR = 1e-5     # ... the pretrained backbone adapts slowly


__SEED_FN__


def main():
    seed_everything(SEED)
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    train_ds = ScaffoldDataset(MANIFEST, REPO_ROOT, split="train")
    val_ds = ScaffoldDataset(MANIFEST, REPO_ROOT, split="val")
    train_loader = DataLoader(train_ds, batch_size=16, shuffle=True, num_workers=4)
    val_loader = DataLoader(val_ds, batch_size=16, shuffle=False, num_workers=4)
    # Load the PRETRAINED backbone (weight provenance is recorded in PRETRAINED.md).
    model = build_model(pretrained=True).to(device)
    criterion = build_loss()
    model.set_backbone_trainable(False)                       # phase 1: head-only warmup
    optimizer = torch.optim.Adam(model.head.parameters(), lr=HEAD_LR)
    best_val = float("inf")
    for epoch in range(TOTAL_EPOCHS):
        if epoch == FREEZE_EPOCHS:                            # phase 2: unfreeze, discriminative LRs
            model.set_backbone_trainable(True)
            optimizer = torch.optim.Adam([
                {"params": model.backbone.parameters(), "lr": BACKBONE_LR},
                {"params": model.head.parameters(), "lr": HEAD_LR},
            ])
        model.train()
        for x, y in train_loader:
            x, y = x.to(device), y.to(device)
            optimizer.zero_grad()
            loss = criterion(model(x), y)
            loss.backward()
            optimizer.step()
        model.eval()
        val_loss = 0.0
        with torch.no_grad():
            for x, y in val_loader:
                x, y = x.to(device), y.to(device)
                val_loss += criterion(model(x), y).item()
        val_loss /= max(len(val_loader), 1)
        if val_loss < best_val:
            best_val = val_loss
            torch.save({"epoch": epoch, "model": model.state_dict(),
                        "seed": SEED, "pretrained_source": PRETRAINED_SOURCE}, "best.pt")


if __name__ == "__main__":
    main()
'''

FT_EVAL = r'''"""Held-out evaluation (fine-tuning / classification; generated by model-scaffold).
Inference under model.eval() + torch.no_grad(); writes per-case scores. Compute AUROC +
AUPRC (+ sensitivity/specificity at the deployment prevalence) with CIs downstream via
/model-evaluation + /analyze-stats."""
import csv
import torch
from torch.utils.data import DataLoader
from dataset import ScaffoldDataset
from model import build_model

REPO_ROOT = "."
MANIFEST = "__MANIFEST_NAME__"


def main():
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    test_loader = DataLoader(ScaffoldDataset(MANIFEST, REPO_ROOT, split="test"), batch_size=1, shuffle=False)
    model = build_model().to(device)   # architecture only; weights come from the fine-tuned checkpoint
    model.load_state_dict(torch.load("best.pt", map_location=device)["model"])
    model.eval()
    rows = []
    with torch.no_grad():
        for i, (x, y) in enumerate(test_loader):
            scores = torch.softmax(model(x.to(device)), dim=1).squeeze(0).tolist()
            rows.append({"index": i, "scores": scores})
    with open("predictions.csv", "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=["index", "scores"])
        w.writeheader()
        w.writerows(rows)


if __name__ == "__main__":
    main()
'''

TASKS = {
    "segmentation": {"model": SEG_MODEL, "dataset": SEG_DATASET, "losses": SEG_LOSSES,
                     "train": SEG_TRAIN, "evaluate": SEG_EVAL, "arch": "unet"},
    "finetune": {"model": FT_MODEL, "dataset": FT_DATASET, "losses": FT_LOSSES,
                 "train": FT_TRAIN, "evaluate": FT_EVAL, "arch": "finetune-backbone"},
    "classification": {"model": CLS_MODEL, "dataset": CLS_DATASET, "losses": CLS_LOSSES,
                       "train": CLS_TRAIN, "evaluate": CLS_EVAL, "arch": "cnn"},
    "detection": {"model": DET_MODEL, "dataset": DET_DATASET, "losses": DET_LOSSES,
                  "train": DET_TRAIN, "evaluate": DET_EVAL, "arch": "fasterrcnn"},
    "synthesis": {"model": SYN_MODEL, "dataset": SYN_DATASET, "losses": SYN_LOSSES,
                  "train": SYN_TRAIN, "evaluate": SYN_EVAL, "arch": "pix2pix"},
    "ssl": {"model": SSL_MODEL, "dataset": SSL_DATASET, "losses": SSL_LOSSES,
            "train": SSL_TRAIN, "evaluate": SSL_EVAL, "arch": "simclr"},
}

CONFIG_YAML = """# Single source of truth for this scaffolded run (generated by model-scaffold).
task: __TASK__
arch: __ARCH__
seed: __SEED__
id_col: __ID_COL__
in_channels: __IN_CH__
out_channels: __OUT_CH__
base_channels: __BASE__
split:
  val_frac: __VAL_FRAC__
  test_frac: __TEST_FRAC__
  assignment: splits/split_assignment.csv
  seed_file: splits/split_seed.txt
manifest: __MANIFEST_NAME__
__PRETRAINED_BLOCK__"""

REQUIREMENTS = """# Generated by model-scaffold. Pin exact versions before publishing.
torch
numpy
# Recommended medical-imaging stack (integrate, do not reimplement):
# monai          # UNet/SegResNet, transforms, metrics (Dice/HD95/NSD)
# torchvision    # detection (Faster / Mask R-CNN, FPN)
# timm           # pretrained classification backbones (ResNet/EfficientNet/ViT/Swin)
# torchio        # 3-D spatial/intensity augmentation
# nibabel pydicom tifffile   # NIfTI / DICOM / TIFF I/O
# tensorboard
"""

REPRO_MD = """# Reproducibility

Generated by `model-scaffold` (task: __TASK__, arch: __ARCH__) with reproducibility baked in.

- **Split**: patient-level, seed-locked (`splits/split_assignment.csv`, seed `__SEED__`);
  disjoint by construction. Verify with `/model-validation`
  (`check_split_leakage.py --splits splits/split_assignment.csv --strict`).
- **Seed**: `__SEED__` applied to random / numpy / torch / torch.cuda; cuDNN deterministic
  (`train.py: seed_everything`). Report metrics as mean +/- SD over >= 3 seeds.
- **Model**: `model.py` (`__ARCH__`). Swap in MONAI / nnU-Net / timm / torchvision for production.
- **Environment** (fill in): python + pip freeze, CUDA / driver, GPU, git commit.
- **Data**: plug your reader into `dataset.py`.

## How to reproduce
`pip install -r requirements.txt` -> implement `dataset.py` I/O -> `python train.py` ->
`python evaluate.py` -> metrics + CIs via `/model-evaluation` -> `/analyze-stats`.
"""

METHODS_MD = """# Methods stub (generated by model-scaffold — fill the [VERIFY] placeholders)

A `__ARCH__` model was trained for __TASK__ (in=`__IN_CH__`, out=`__OUT_CH__`, base
`__BASE__`). Data were split at the **patient level** (val=__VAL_FRAC__, test=__TEST_FRAC__;
[VERIFY: report n patients per split from splits/split_assignment.csv]) with the assignment
frozen and seed-locked (seed __SEED__), so no patient contributed images to more than one
partition. All random number generators (Python, NumPy, PyTorch, CUDA) were seeded and cuDNN
was set deterministic. Held-out performance is reported as
[VERIFY: task-correct metrics with 95% CIs over >= 3 seeds] on the test split.
"""

# Emitted only for --task finetune: the pretrained-weight provenance record. A fine-tune
# whose starting checkpoint is unrecorded is not reproducible or auditable — this is the
# artifact check_training_hygiene looks for (PRETRAINED_PROVENANCE_MISSING).
PRETRAINED_MD = """# Pretrained-weight provenance (fine-tuning; fill the [VERIFY] fields)

This model was fine-tuned from PRETRAINED weights. For the result to be reproducible the
exact source of those weights must be recorded here — a fine-tune whose starting checkpoint
is unknown (ImageNet vs RadImageNet vs a public medical checkpoint) cannot be reproduced or
audited, and a backbone pretrained on data that overlaps this study's test set is a form of
leakage no split table can see.

- **Source**: __PRETRAINED_SOURCE__
  [VERIFY: confirm the exact model + weights tag, e.g. `timm:resnet50.a1_in1k`, `MedSAM`, or a URL/DOI]
- **Pretraining data**: [VERIFY: dataset the backbone was pretrained on — confirm it does NOT
  overlap this study's test set (pretraining-set contamination is unseen leakage)]
- **License**: [VERIFY: license / terms of the pretrained weights]
- **Checkpoint hash**: [VERIFY: sha256 of the downloaded checkpoint file]
- **Access date**: [VERIFY: when the weights were downloaded]

Report the fine-tuning schedule (frozen-vs-unfrozen, discriminative learning rates) in
Methods; see references/finetuning_guide.md.
"""


def _pick_id_col(header, explicit):
    if explicit:
        if explicit in header:
            return explicit
        sys.stderr.write(f"ERROR: --id-col '{explicit}' not in manifest header {header}\n")
        sys.exit(2)
    norm = {h.strip().lower().replace("_", "").replace(" ", ""): h for h in header}
    for hint in ID_HINTS:
        if hint.replace("_", "") in norm:
            return norm[hint.replace("_", "")]
    sys.stderr.write(f"ERROR: no ID column found in {header}; pass --id-col\n")
    sys.exit(2)


def split_patients(ids, seed, val_frac, test_frac):
    uniq = sorted(set(ids))
    rng = np.random.default_rng(seed)
    order = [uniq[i] for i in rng.permutation(len(uniq))]
    n = len(order)
    n_test, n_val = round(n * test_frac), round(n * val_frac)
    assign = {}
    for pid in order[:n_test]:
        assign[pid] = "test"
    for pid in order[n_test:n_test + n_val]:
        assign[pid] = "val"
    for pid in order[n_test + n_val:]:
        assign[pid] = "train"
    return assign


def render(tmpl, repl):
    out = tmpl
    for k, v in repl.items():
        out = out.replace(k, str(v))
    return out


def main() -> int:
    ap = argparse.ArgumentParser(description="Reproducible medical-imaging training-repo scaffold.")
    ap.add_argument("--manifest", required=True, help="CSV: one row per image, with a patient ID column")
    ap.add_argument("--id-col", help="patient/subject ID column (auto-detected if omitted)")
    ap.add_argument("--task", default="segmentation", choices=sorted(TASKS))
    ap.add_argument("--arch", help="architecture label (defaults to the task's default)")
    ap.add_argument("--out", default="model_repo", help="output repo directory")
    ap.add_argument("--seed", type=int, default=42)
    ap.add_argument("--val-frac", type=float, default=0.15)
    ap.add_argument("--test-frac", type=float, default=0.15)
    ap.add_argument("--in-channels", type=int, default=1)
    ap.add_argument("--out-channels", type=int, default=1)
    ap.add_argument("--base-channels", type=int, default=16)
    ap.add_argument("--from-pretrained", default="timm:resnet50.a1_in1k",
                    help="pretrained-weight source recorded as provenance (--task finetune); "
                         "e.g. timm:resnet50.a1_in1k, MedSAM, or a URL/DOI")
    ap.add_argument("--quiet", action="store_true")
    args = ap.parse_args()

    man = Path(args.manifest)
    if not man.is_file():
        sys.stderr.write(f"ERROR: --manifest not found: {args.manifest}\n")
        return 2
    with man.open(encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        header = reader.fieldnames or []
        rows = list(reader)
    if not rows:
        sys.stderr.write("ERROR: manifest has no rows\n")
        return 2
    id_col = _pick_id_col(header, args.id_col)
    ids = [r[id_col] for r in rows if (r.get(id_col) or "").strip()]
    if not ids:
        sys.stderr.write(f"ERROR: no values in ID column '{id_col}'\n")
        return 2

    task = TASKS[args.task]
    arch = args.arch or task["arch"]
    assign = split_patients(ids, args.seed, args.val_frac, args.test_frac)

    out = Path(args.out)
    (out / "splits").mkdir(parents=True, exist_ok=True)
    with (out / "splits" / "split_assignment.csv").open("w", newline="", encoding="utf-8") as f:
        w = csv.writer(f)
        w.writerow([id_col, "split"])
        for pid in sorted(assign):
            w.writerow([pid, assign[pid]])
    (out / "splits" / "split_seed.txt").write_text(f"{args.seed}\n", encoding="utf-8")

    is_finetune = args.task == "finetune"
    pretrained_block = (
        f"pretrained:\n  source: {args.from_pretrained}\n  provenance: PRETRAINED.md\n"
        if is_finetune else ""
    )
    repl = {
        "__SEED__": args.seed, "__IN_CH__": args.in_channels, "__OUT_CH__": args.out_channels,
        "__BASE__": args.base_channels, "__TASK__": args.task, "__ARCH__": arch,
        "__ID_COL__": id_col, "__VAL_FRAC__": args.val_frac, "__TEST_FRAC__": args.test_frac,
        "__MANIFEST_NAME__": man.name, "__SEED_FN__": SEED_FN,
        "__PRETRAINED_SOURCE__": args.from_pretrained, "__PRETRAINED_BLOCK__": pretrained_block,
    }
    files = {
        "model.py": task["model"], "dataset.py": task["dataset"], "losses.py": task["losses"],
        "train.py": task["train"], "evaluate.py": task["evaluate"], "config.yaml": CONFIG_YAML,
        "requirements.txt": REQUIREMENTS, "REPRODUCIBILITY.md": REPRO_MD, "methods_stub.md": METHODS_MD,
    }
    if is_finetune:
        files["PRETRAINED.md"] = PRETRAINED_MD
    for name, tmpl in files.items():
        (out / name).write_text(render(tmpl, repl), encoding="utf-8")

    nt = sum(1 for v in assign.values() if v == "train")
    nv = sum(1 for v in assign.values() if v == "val")
    ns = sum(1 for v in assign.values() if v == "test")
    if not args.quiet:
        print(f"scaffolded {args.task}/{arch} repo -> {out}/")
        print(f"  patients: {len(assign)} (train={nt} val={nv} test={ns}), seed={args.seed}")
        print("  files: " + ", ".join(sorted(files) + ["splits/split_assignment.csv", "splits/split_seed.txt"]))
    return 0


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

#!/usr/bin/env bash
# Regression test for the training-hygiene linter + the scaffold generator
# (model-scaffold). Synthetic, PII-free. Stdlib + numpy only (no torch).
#   (a) a freshly scaffolded repo is clean (all RNGs seeded, cuDNN deterministic,
#       eval()+no_grad(), train-only loader) -> exit 0;
#   (b) bad train/eval fixtures fire SEED_INCOMPLETE, TRAIN_ON_NONTRAIN_SPLIT,
#       MISSING_EVAL_MODE (Major) + CUDNN_NONDETERMINISTIC, EVAL_SHUFFLE (Minor);
#   (c) scaffold.py emits a patient-disjoint, seed-locked split (deterministic).
set -u

HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
SKD="$HERE/../scripts"
HYGIENE="$SKD/check_training_hygiene.py"
SCAFFOLD="$SKD/scaffold.py"
F="$HERE/fixtures"
WORK="$(mktemp -d)"
OUT="$WORK/out.json"
trap 'rm -rf "$WORK"' 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() { python3 -c "
import json
d=json.load(open('$OUT'))
assert any(c['verdict']=='$1' for c in d['claims']), '$1 not found'
"; }
no() { python3 -c "
import json
d=json.load(open('$OUT'))
assert not any(c['verdict']=='$1' for c in d['claims']), '$1 unexpectedly present'
"; }

[[ -f "$HYGIENE" && -f "$SCAFFOLD" ]] || { echo "ENV-ERR: scripts missing" >&2; exit 2; }

# (a) scaffold a clean repo -> hygiene clean, exit 0
printf 'patient_id,image,label\nP1,a,m\nP2,b,n\nP3,c,o\nP4,d,p\n' > "$WORK/m.csv"
python3 "$SCAFFOLD" --manifest "$WORK/m.csv" --out "$WORK/clean" --seed 42 --quiet >/dev/null 2>&1
python3 "$HYGIENE" --repo "$WORK/clean" --out "$OUT" --strict --quiet >/dev/null 2>&1
check "scaffolded repo passes hygiene (exit 0)" test "$?" -eq 0
check "no SEED_INCOMPLETE on clean repo" no SEED_INCOMPLETE
check "no MISSING_EVAL_MODE on clean repo" no MISSING_EVAL_MODE

# (b) bad fixtures -> Major verdicts, exit 1
python3 "$HYGIENE" --train "$F/bad_train.py" --eval "$F/bad_evaluate.py" --out "$OUT" --strict --quiet >/dev/null 2>&1
check "exit 1 on bad train/eval (Major present)" test "$?" -eq 1
check "SEED_INCOMPLETE detected"          has SEED_INCOMPLETE
check "TRAIN_ON_NONTRAIN_SPLIT detected"  has TRAIN_ON_NONTRAIN_SPLIT
check "MISSING_EVAL_MODE detected"        has MISSING_EVAL_MODE
check "CUDNN_NONDETERMINISTIC detected"   has CUDNN_NONDETERMINISTIC
check "EVAL_SHUFFLE detected"             has EVAL_SHUFFLE
check "reports numpy as the only seed found" python3 -c "
import json; d=json.load(open('$OUT'))
c=next(c for c in d['claims'] if c['verdict']=='SEED_INCOMPLETE')
assert 'found: numpy' in c['detail'], c['detail']"

# (c) scaffold emits a patient-disjoint, seed-locked split (deterministic)
check "split_assignment.csv emitted" test -f "$WORK/clean/splits/split_assignment.csv"
check "split seed recorded = 42" bash -c "[ \"\$(cat '$WORK/clean/splits/split_seed.txt')\" = '42' ]"
check "split is patient-disjoint" python3 -c "
import csv
seen={}
for r in csv.DictReader(open('$WORK/clean/splits/split_assignment.csv')):
    seen.setdefault(r['patient_id'],set()).add(r['split'])
assert all(len(s)==1 for s in seen.values()), 'patient crosses splits'"

# (d) breadth: every task scaffolds to valid Python with hygiene-clean train/eval
clean_repo() { python3 "$SCAFFOLD" --manifest "$WORK/m.csv" --task "$1" --out "$WORK/$1" --seed 42 --quiet >/dev/null 2>&1; }
hygiene_ok() { python3 "$HYGIENE" --repo "$WORK/$1" --strict --quiet >/dev/null 2>&1; }
valid_py()  { for f in "$WORK/$1"/*.py; do python3 -c "import ast,sys;ast.parse(open(sys.argv[1]).read())" "$f" || return 1; done; }
for t in classification detection synthesis ssl finetune; do
    if clean_repo "$t" && hygiene_ok "$t" && valid_py "$t"; then
        printf '  PASS  scaffold %s: hygiene-clean + valid Python\n' "$t"
    else
        printf '  FAIL  scaffold %s\n' "$t"; fail=$((fail+1))
    fi
done

# (e) fine-tuning provenance: the scaffold records provenance by construction (no fire),
#     a pretrained-load repo WITHOUT a provenance record fires PRETRAINED_PROVENANCE_MISSING.
check "finetune scaffold emits PRETRAINED.md" test -f "$WORK/finetune/PRETRAINED.md"
check "finetune config.yaml has a pretrained: block" grep -q "^pretrained:" "$WORK/finetune/config.yaml"
python3 "$HYGIENE" --repo "$WORK/finetune" --out "$OUT" --quiet >/dev/null 2>&1
check "no PRETRAINED_PROVENANCE_MISSING on finetune scaffold" no PRETRAINED_PROVENANCE_MISSING
python3 "$HYGIENE" --repo "$F/finetune_no_provenance" --out "$OUT" --quiet >/dev/null 2>&1
check "PRETRAINED_PROVENANCE_MISSING on pretrained-load repo lacking provenance" has PRETRAINED_PROVENANCE_MISSING
check "provenance verdict is Minor" python3 -c "
import json; d=json.load(open('$OUT'))
c=next(c for c in d['claims'] if c['verdict']=='PRETRAINED_PROVENANCE_MISSING')
assert c['severity']=='Minor', c['severity']"

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