scripts/compare_cv.py

#!/usr/bin/env python3
"""Compare two cross-validation runs (cv_results.json files) on the measures that
decide whether a new training recipe replaces the shipped weights.

    python scripts/compare_cv.py reports/cv/cv_results.json runs/cv_v2/cv_results.json [--out reports/cv_v2/comparison.md]

Nothing here looks at the images or the predictions; it only reads the two
aggregates, so it cannot tune anything on the held-out data.
"""
from __future__ import annotations

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from cellquant.labels import ALL_LABELS  # noqa: E402


def pct(x: float) -> str:
    return f"{100 * x:.1f}%"


def delta(a: float, b: float) -> str:
    d = 100 * (b - a)
    return f"{d:+.1f}"


def headline_rows(agg: dict) -> dict[str, float]:
    det, det8 = agg["detection_base"], agg["detection_tta8"]
    rows = {
        "Nucleus F1, IoU ≥ 0.5": det["cells"]["f1"],
        "Nucleus precision, IoU ≥ 0.5": det["cells"]["precision"],
        "Nucleus recall, IoU ≥ 0.5": det["cells"]["recall"],
        "Nucleus F1, IoU ≥ 0.3": det["cells_iou30"]["f1"],
        "Nucleus F1, centroid rule": det["cells_centroid"]["f1"],
        "Nucleus F1, IoU ≥ 0.5, 8× TTA": det8["cells"]["f1"],
        "Puncta F1": det["nostrils"]["f1"],
        "Puncta F1, 8× TTA": det8["nostrils"]["f1"],
    }
    # attribute accuracies pooled over matched nuclei (weights = matched count per image)
    rule = agg["rule_variants"]["rule_nos=head_cell=contains"]
    rows["Attribute accuracy (both, matched nuclei)"] = rule["pooled_cell_attr_both"]
    rows["Punctum type accuracy (matched puncta)"] = rule["pooled_nos_type_acc"]
    for lab in ALL_LABELS:
        rows[f"F1 [{lab}]"] = agg["micro_base"][lab]["f1"]
    for lab in ("Cell, Antibody -",):
        r = agg["micro_base"][lab]
        rows[f"Precision [{lab}]"] = r["precision"]
        rows[f"Recall [{lab}]"] = r["recall"]
    return rows


def main(argv=None):
    ap = argparse.ArgumentParser()
    ap.add_argument("old")
    ap.add_argument("new")
    ap.add_argument("--names", default="v1,v2")
    ap.add_argument("--out", default="")
    a = ap.parse_args(argv)
    A = json.load(open(a.old))["aggregate"]
    B = json.load(open(a.new))["aggregate"]
    na, nb = a.names.split(",")
    L = [f"# Cross-validation comparison: {na} vs {nb}", "",
         f"`{a.old}` vs `{a.new}`; micro-averaged over the held-out images (base run, no TTA unless stated).", "",
         f"| Measure | {na} | {nb} | Δ (points) |", "|---|---:|---:|---:|"]
    ra, rb = headline_rows(A), headline_rows(B)
    for k in ra:
        L.append(f"| {k} | {pct(ra[k])} | {pct(rb[k])} | {delta(ra[k], rb[k])} |")
    L += ["", "## Per held-out image", "",
          f"| Image | Nuclei F1 {na} | Nuclei F1 {nb} | Δ | Centroid F1 {na} | Centroid F1 {nb} | Antibody acc {na} | Antibody acc {nb} | Puncta F1 {na} | Puncta F1 {nb} |",
          "|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|"]
    for k in sorted(set(A["per_image"]) | set(B["per_image"])):
        sa, sb = A["per_image"].get(k, {}).get("base"), B["per_image"].get(k, {}).get("base")
        if not sa or not sb:
            continue
        L.append(f"| {k} | {pct(sa['cell_det_f1'])} | {pct(sb['cell_det_f1'])} | {delta(sa['cell_det_f1'], sb['cell_det_f1'])} | "
                 f"{pct(sa['cell_det_f1_centroid'])} | {pct(sb['cell_det_f1_centroid'])} | {pct(sa['cell_attr_antibody'])} | {pct(sb['cell_attr_antibody'])} | "
                 f"{pct(sa['nos_det_f1'])} | {pct(sb['nos_det_f1'])} |")
    L += ["", "## Robustness (nuclei F1 over the 14 rotated / perturbed runs per image)", "",
          f"| Image | {na} mean ± sd | {nb} mean ± sd |", "|---|---|---|"]
    for k in sorted(set(A["per_image"]) & set(B["per_image"])):
        ra_, rb_ = A["per_image"][k]["robustness"]["cell_det_f1"], B["per_image"][k]["robustness"]["cell_det_f1"]
        L.append(f"| {k} | {100*ra_['mean']:.1f} ± {100*ra_['std']:.1f} | {100*rb_['mean']:.1f} ± {100*rb_['std']:.1f} |")
    L += ["", "## Antibody-negative class", "",
          f"| | {na} | {nb} |", "|---|---:|---:|"]
    for lab in ("Cell, Antibody -", "Cell, Nostril, Antibody -"):
        x, y = A["micro_base"][lab], B["micro_base"][lab]
        L.append(f"| {lab}: GT / predicted / TP | {x['gt_count']} / {x['pred_count']} / {x['tp']} | {y['gt_count']} / {y['pred_count']} / {y['tp']} |")
    sa, sb = A.get("antibody_threshold_sweep"), B.get("antibody_threshold_sweep")
    if sa and sb:
        ba = max(sa["sweep"], key=lambda r: r["neg_f1"])
        bb = max(sb["sweep"], key=lambda r: r["neg_f1"])
        at = [r for r in sa["sweep"] if abs(r["thr"] - 0.5) < 1e-6][0]
        bt = [r for r in sb["sweep"] if abs(r["thr"] - 0.5) < 1e-6][0]
        L.append(f"| Antibody − F1 among matched nuclei at the default threshold 0.5 | {pct(at['neg_f1'])} (P {pct(at['neg_precision'])}, R {pct(at['neg_recall'])}) | {pct(bt['neg_f1'])} (P {pct(bt['neg_precision'])}, R {pct(bt['neg_recall'])}) |")
        L.append(f"| Best threshold in the sweep (not used) | {pct(ba['neg_f1'])} at {ba['thr']:.2f} | {pct(bb['neg_f1'])} at {bb['thr']:.2f} |")
    L.append("")
    text = "\n".join(L)
    print(text)
    if a.out:
        os.makedirs(os.path.dirname(a.out), exist_ok=True)
        with open(a.out, "w") as f:
            f.write(text)
        print("wrote", a.out)


if __name__ == "__main__":
    main()