cellquant / scripts /compare_cv.py
stevafernandes's picture
Deploy CellQuant
812b6ee verified
Raw History Blame Contribute Delete
5.47 kB
#!/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()