| """ |
| evaluate.py -- Part B, Section 3.4 (evaluation and error analysis). |
| |
| Reports on the frozen test split: |
| mAP@0.50, mAP@0.75, mAP@0.95, mAP@[.50:.05:.95], |
| precision / recall / F1 at a tuned confidence threshold, |
| mean IoU of matched detections, inference time per image, |
| recall stratified by polyp size (small / medium / large), |
| a confidence-threshold sweep and a precision-recall curve, |
| >= 8 seed-selected qualitative panels covering TP, FP and FN, |
| and a failure-pattern summary linked back to the EDA. |
| |
| Usage |
| ----- |
| python evaluate.py --ckpt ../Results/model_weights/det_fasterrcnn_resnet50_fpn_ce_seed16.pt |
| python evaluate.py --compare det_fasterrcnn_resnet50_fpn_ce_seed16 \ |
| det_fasterrcnn_resnet50_fpn_focal-a0.25g2.0_seed16 |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import time |
| from pathlib import Path |
|
|
| import numpy as np |
| import pandas as pd |
| import torch |
|
|
| from common import (DET_SPLIT_CSV, METRICS_DIR, OVERLAY_DIR, PLOTS_DIR, SEED, |
| SEED_TAG, KvasirDetectionDataset, _mpl, average_precision, |
| box_iou_np, coco_map, get_device, make_loader, |
| prf_at_threshold, progress, save_json, seeded_rng, set_seed, |
| size_stratified_recall) |
| from model import build_detector, set_roi_loss |
|
|
|
|
| |
| def _torch_load(path): |
| """torch>=2.6 defaults weights_only=True, which cannot restore the config |
| dict we store alongside the weights; older torch has no such argument.""" |
| try: |
| return torch.load(path, map_location="cpu", weights_only=False) |
| except TypeError: |
| return torch.load(path, map_location="cpu") |
|
|
|
|
| def load_detector(ckpt_path: Path): |
| ck = _torch_load(ckpt_path) |
| cfg = ck["args"] |
| set_roi_loss(cfg.get("loss", "ce"), alpha=cfg.get("focal_alpha", 0.25), |
| gamma=cfg.get("focal_gamma", 2.0), fg_weight=cfg.get("fg_weight", 4.0)) |
| model = build_detector(cfg.get("arch", "fasterrcnn_resnet50_fpn"), |
| pretrained_backbone=False, |
| img_size=cfg.get("img_size", 512), |
| trainable_backbone_layers=cfg.get("trainable_layers", 3)) |
| model.load_state_dict(ck["model_state"]) |
| model.eval() |
| return model, ck, cfg |
|
|
|
|
| @torch.no_grad() |
| def infer_split(model, cfg, split: str, bs: int, workers: int): |
| device = get_device() |
| model.to(device).eval() |
| ds = KvasirDetectionDataset(DET_SPLIT_CSV, split, augment=False, |
| img_size=cfg.get("img_size", 512)) |
| dl = make_loader(ds, bs, shuffle=False, num_workers=workers) |
|
|
| preds, gts, stems, areas = [], [], [], [] |
| |
| for images, _ in dl: |
| model([images[0].to(device)]) |
| break |
| if device.type == "cuda": |
| torch.cuda.synchronize() |
|
|
| t0 = time.time() |
| for images, targets in progress(dl, desc=f"infer[{split}]"): |
| outs = model([im.to(device) for im in images]) |
| for im, o, t in zip(images, outs, targets): |
| preds.append({"boxes": o["boxes"].cpu().numpy(), |
| "scores": o["scores"].cpu().numpy()}) |
| gts.append({"boxes": t["boxes"].numpy()}) |
| stems.append(t["stem"]) |
| areas.append(float(im.shape[-1] * im.shape[-2])) |
| if device.type == "cuda": |
| torch.cuda.synchronize() |
| dt = time.time() - t0 |
| timing = {"total_seconds": dt, "n_images": len(stems), |
| "ms_per_image": 1000 * dt / max(1, len(stems)), |
| "fps": len(stems) / max(1e-9, dt), "device": str(device), |
| "batch_size": bs} |
| return preds, gts, stems, areas, timing |
|
|
|
|
| |
| def threshold_sweep(preds, gts, iou_thr: float = 0.5, |
| thresholds=np.arange(0.05, 0.96, 0.05)) -> pd.DataFrame: |
| rows = [prf_at_threshold(preds, gts, iou_thr, float(t)) for t in thresholds] |
| return pd.DataFrame(rows) |
|
|
|
|
| def plot_pr_and_sweep(preds, gts, sweep: pd.DataFrame, tag: str): |
| plt = _mpl() |
| ap = average_precision(preds, gts, 0.5) |
| fig, axes = plt.subplots(1, 2, figsize=(11.5, 4.4)) |
| if "recall_curve" in ap: |
| axes[0].plot(ap["recall_curve"], ap["precision_curve"], lw=2, color="#4C72B0") |
| axes[0].set_xlabel("recall"); axes[0].set_ylabel("precision") |
| axes[0].set_title(f"PR curve @IoU 0.50 (AP={ap['ap']:.3f})") |
| axes[0].set_xlim(0, 1); axes[0].set_ylim(0, 1.02); axes[0].grid(alpha=0.3) |
|
|
| axes[1].plot(sweep["score_threshold"], sweep["precision"], marker="o", ms=3, label="precision") |
| axes[1].plot(sweep["score_threshold"], sweep["recall"], marker="s", ms=3, label="recall") |
| axes[1].plot(sweep["score_threshold"], sweep["f1"], marker="^", ms=3, label="F1") |
| best = sweep.loc[sweep["f1"].idxmax()] |
| axes[1].axvline(best["score_threshold"], ls="--", c="k", lw=1) |
| axes[1].set_xlabel("confidence threshold") |
| axes[1].set_title(f"Operating point sweep (best F1={best['f1']:.3f} " |
| f"@ {best['score_threshold']:.2f})") |
| axes[1].legend(); axes[1].grid(alpha=0.3) |
| fig.suptitle(tag, fontsize=11) |
| fig.tight_layout() |
| out = PLOTS_DIR / f"pr_sweep_{tag}.png" |
| fig.savefig(out, dpi=160); plt.close(fig) |
| print(f"[saved] {out}") |
|
|
|
|
| def qualitative(preds, gts, stems, cfg, tag: str, score_thr: float, |
| iou_thr: float = 0.5, n: int = 8): |
| """Seed-selected panels guaranteed to contain true positives, false |
| positives and false negatives.""" |
| plt = _mpl() |
| from PIL import Image |
|
|
| df = pd.read_csv(DET_SPLIT_CSV).set_index("image_id") |
| per = [] |
| for i, (p, g, s) in enumerate(zip(preds, gts, stems)): |
| keep = p["scores"] >= score_thr |
| pb, ps = p["boxes"][keep], p["scores"][keep] |
| gb = g["boxes"] |
| ious = box_iou_np(pb, gb) if len(pb) and len(gb) else np.zeros((len(pb), len(gb))) |
| tp = int((ious.max(axis=1) >= iou_thr).sum()) if ious.size else 0 |
| fp = len(pb) - tp |
| fn = len(gb) - (int((ious.max(axis=0) >= iou_thr).sum()) if ious.size else 0) |
| per.append({"i": i, "stem": s, "tp": tp, "fp": fp, "fn": fn, |
| "n_pred": len(pb), "n_gt": len(gb), |
| "best_iou": float(ious.max()) if ious.size else 0.0}) |
| pdf = pd.DataFrame(per) |
|
|
| rng = seeded_rng(offset=9) |
| picks: list[int] = [] |
|
|
| def take(sub, k): |
| sub = sub[~sub["i"].isin(picks)] |
| if len(sub) == 0: |
| return |
| idx = rng.choice(len(sub), size=min(k, len(sub)), replace=False) |
| picks.extend(sub.iloc[idx]["i"].tolist()) |
|
|
| take(pdf[(pdf.tp > 0) & (pdf.fp == 0) & (pdf.fn == 0)], 3) |
| take(pdf[pdf.fp > 0], 3) |
| take(pdf[pdf.fn > 0], 3) |
| take(pdf, n) |
| picks = picks[:max(n, 8)] |
|
|
| ncol = 4 |
| nrow = int(np.ceil(len(picks) / ncol)) |
| fig, axes = plt.subplots(nrow, ncol, figsize=(4.2 * ncol, 4.5 * nrow)) |
| axes = np.atleast_1d(axes).ravel() |
| recs = [] |
| for k, i in enumerate(picks): |
| stem = stems[i] |
| r = df.loc[stem] |
| img = Image.open(r["image_path"]).convert("RGB") |
| size = cfg.get("img_size", 512) |
| img = img.resize((size, size), Image.BILINEAR) |
| axes[k].imshow(np.asarray(img)) |
| for b in gts[i]["boxes"]: |
| axes[k].add_patch(plt.Rectangle((b[0], b[1]), b[2] - b[0], b[3] - b[1], |
| fill=False, edgecolor="lime", lw=2.2)) |
| keep = preds[i]["scores"] >= score_thr |
| pbs, pss = preds[i]["boxes"][keep], preds[i]["scores"][keep] |
| ious = box_iou_np(pbs, gts[i]["boxes"]) if len(pbs) and len(gts[i]["boxes"]) \ |
| else np.zeros((len(pbs), max(1, len(gts[i]["boxes"])))) |
| for j, (b, sc) in enumerate(zip(pbs, pss)): |
| best = float(ious[j].max()) if ious.size else 0.0 |
| col = "red" if best >= iou_thr else "orange" |
| axes[k].add_patch(plt.Rectangle((b[0], b[1]), b[2] - b[0], b[3] - b[1], |
| fill=False, edgecolor=col, lw=1.8, ls="-")) |
| axes[k].text(b[0], max(10, b[1] - 4), f"{sc:.2f} / IoU {best:.2f}", |
| color=col, fontsize=7, |
| bbox=dict(facecolor="black", alpha=0.45, pad=0.8, edgecolor="none")) |
| row = pdf.iloc[i] |
| axes[k].set_title(f"{stem[:18]}\nTP={row.tp} FP={row.fp} FN={row.fn}", fontsize=8) |
| axes[k].axis("off") |
| recs.append({"image_id": stem, "tp": int(row.tp), "fp": int(row.fp), |
| "fn": int(row.fn), "n_pred": int(row.n_pred), "n_gt": int(row.n_gt), |
| "best_iou": round(float(row.best_iou), 3), |
| "max_confidence": round(float(pss.max()), 3) if len(pss) else 0.0}) |
| for k in range(len(picks), len(axes)): |
| axes[k].axis("off") |
| fig.suptitle(f"Kvasir-SEG test predictions -- {tag} " |
| f"(green = GT, red = matched detection, orange = false positive; " |
| f"conf>={score_thr:.2f}, seed={SEED})", fontsize=11) |
| fig.tight_layout() |
| out = OVERLAY_DIR / f"det_examples_{tag}.png" |
| fig.savefig(out, dpi=145); plt.close(fig) |
| print(f"[saved] {out}") |
|
|
| pd.DataFrame(recs).to_csv(METRICS_DIR / f"det_examples_{tag}.csv", index=False) |
| return pdf |
|
|
|
|
| def failure_patterns(pdf: pd.DataFrame, preds, gts, areas, score_thr: float) -> dict: |
| """Quantified failure taxonomy, each row linkable to an EDA finding.""" |
| n_img = len(pdf) |
| small_missed = medium_missed = large_missed = 0 |
| for p, g, a in zip(preds, gts, areas): |
| keep = p["scores"] >= score_thr |
| pb = p["boxes"][keep] |
| gb = g["boxes"] |
| if len(gb) == 0: |
| continue |
| ious = box_iou_np(pb, gb) if len(pb) else np.zeros((0, len(gb))) |
| best = ious.max(axis=0) if len(pb) else np.zeros(len(gb)) |
| for j in range(len(gb)): |
| if best[j] >= 0.5: |
| continue |
| frac = (gb[j, 2] - gb[j, 0]) * (gb[j, 3] - gb[j, 1]) / max(1.0, a) |
| if frac < 0.01: |
| small_missed += 1 |
| elif frac < 0.05: |
| medium_missed += 1 |
| else: |
| large_missed += 1 |
| return { |
| "images_evaluated": int(n_img), |
| "images_fully_correct": int(((pdf.fp == 0) & (pdf.fn == 0) & (pdf.tp > 0)).sum()), |
| "images_with_false_positive": int((pdf.fp > 0).sum()), |
| "images_with_missed_polyp": int((pdf.fn > 0).sum()), |
| "images_with_no_detection_at_all": int((pdf.n_pred == 0).sum()), |
| "images_with_duplicate_detections": int((pdf.n_pred > pdf.n_gt).sum()), |
| "missed_by_size": {"small_lt_1pct": small_missed, |
| "medium_1_to_5pct": medium_missed, |
| "large_ge_5pct": large_missed}, |
| "localisation_near_misses_iou_0.3_to_0.5": |
| int(((pdf.best_iou >= 0.3) & (pdf.best_iou < 0.5)).sum()), |
| } |
|
|
|
|
| |
| def evaluate_one(args): |
| model, ck, cfg = load_detector(Path(args.ckpt)) |
| tag = ck.get("tag", Path(args.ckpt).stem) |
| preds, gts, stems, areas, timing = infer_split(model, cfg, args.split, args.bs, args.workers) |
|
|
| sweep = threshold_sweep(preds, gts, args.iou_thr) |
| best_row = sweep.loc[sweep["f1"].idxmax()] |
| score_thr = args.score_thr if args.score_thr is not None else float(best_row["score_threshold"]) |
|
|
| maps = coco_map(preds, gts) |
| op = prf_at_threshold(preds, gts, args.iou_thr, score_thr) |
| sizes = size_stratified_recall(preds, gts, areas, args.iou_thr, score_thr) |
|
|
| out = {"tag": tag, "split": args.split, "seed": SEED, |
| "loss_description": ck.get("loss_description", cfg.get("loss")), |
| "confidence_threshold_used": score_thr, |
| "confidence_threshold_selection": "argmax F1 on the sweep" |
| if args.score_thr is None else "user supplied", |
| "mAP@0.50": maps["AP@0.50"], "mAP@0.75": maps["AP@0.75"], |
| "mAP@0.95": maps["AP@0.95"], |
| "mAP@[.50:.05:.95]": maps["mAP@[.50:.05:.95]"], |
| "AP_per_iou_threshold": maps, |
| "operating_point": op, |
| "recall_by_polyp_size": sizes, |
| "timing": timing, "checkpoint": str(args.ckpt), "config": cfg} |
|
|
| pdf = qualitative(preds, gts, stems, cfg, tag, score_thr, args.iou_thr, args.n_examples) |
| out["failure_patterns"] = failure_patterns(pdf, preds, gts, areas, score_thr) |
|
|
| save_json(out, METRICS_DIR / f"eval_{tag}_{args.split}.json") |
| sweep.to_csv(METRICS_DIR / f"sweep_{tag}_{args.split}.csv", index=False) |
| pdf.to_csv(METRICS_DIR / f"per_image_{tag}_{args.split}.csv", index=False) |
| plot_pr_and_sweep(preds, gts, sweep, tag) |
|
|
| print(f"\n=== {tag} / {args.split} ===") |
| print(f" mAP@0.50 {out['mAP@0.50']:.4f}") |
| print(f" mAP@0.75 {out['mAP@0.75']:.4f}") |
| print(f" mAP@0.95 {out['mAP@0.95']:.4f}") |
| print(f" mAP@[.50:.05:.95] {out['mAP@[.50:.05:.95]']:.4f}") |
| print(f" @conf={score_thr:.2f}: P={op['precision']:.4f} R={op['recall']:.4f} " |
| f"F1={op['f1']:.4f}") |
| print(f" TP={op['true_positives']} FP={op['false_positives']} FN={op['false_negatives']}") |
| print(f" mean IoU (matched) {op['mean_iou_matched']:.4f}") |
| print(f" inference {timing['ms_per_image']:.1f} ms/image " |
| f"({timing['fps']:.1f} FPS on {timing['device']})") |
| print(" recall by polyp size:") |
| for k, v in sizes.items(): |
| print(f" {k:8s} n={v['n_gt']:4d} recall={v['recall']:.4f}") |
| print(" failure patterns:", json.dumps(out["failure_patterns"], indent=4)) |
|
|
|
|
| def compare(tags, split: str = "test"): |
| """Section 3.3.4: how the loss-function ablation moves each metric.""" |
| rows = [] |
| for t in tags: |
| p = METRICS_DIR / f"eval_{t}_{split}.json" |
| if not p.exists(): |
| print(f"[warn] missing {p}") |
| continue |
| m = json.loads(p.read_text()) |
| op, sz = m["operating_point"], m["recall_by_polyp_size"] |
| rows.append({ |
| "run": t, "loss": m.get("loss_description", "")[:46], |
| "mAP@0.50": m["mAP@0.50"], "mAP@0.75": m["mAP@0.75"], |
| "mAP@[.50:.95]": m["mAP@[.50:.05:.95]"], |
| "precision": op["precision"], "recall": op["recall"], "F1": op["f1"], |
| "FP": op["false_positives"], "FN": op["false_negatives"], |
| "mean_IoU": op["mean_iou_matched"], |
| "recall_small": sz["small"]["recall"], "recall_medium": sz["medium"]["recall"], |
| "recall_large": sz["large"]["recall"], |
| "ms_per_image": m["timing"]["ms_per_image"]}) |
| if not rows: |
| return |
| df = pd.DataFrame(rows) |
| out = METRICS_DIR / f"ablation_comparison_{split}_{SEED_TAG}.csv" |
| df.round(4).to_csv(out, index=False) |
| print(f"\n=== Loss-function ablation ({split}) ===") |
| print(df.round(4).to_string(index=False)) |
| if len(df) >= 2: |
| base, alt = df.iloc[0], df.iloc[1] |
| print(f"\nDelta ({alt['run']} - {base['run']}):") |
| for k in ["mAP@0.50", "mAP@[.50:.95]", "precision", "recall", "F1", |
| "recall_small", "recall_medium", "recall_large"]: |
| print(f" {k:16s} {alt[k] - base[k]:+.4f}") |
| print(f" {'FP':16s} {alt['FP'] - base['FP']:+d}") |
| print(f" {'FN':16s} {alt['FN'] - base['FN']:+d}") |
| print(f"[saved] {out}") |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser(description="Kvasir-SEG detector evaluation (seed 16)") |
| ap.add_argument("--ckpt") |
| ap.add_argument("--split", default="test", choices=["train", "val", "test"]) |
| ap.add_argument("--bs", type=int, default=4) |
| ap.add_argument("--workers", type=int, default=2) |
| ap.add_argument("--iou-thr", type=float, default=0.5) |
| ap.add_argument("--score-thr", type=float, default=None, |
| help="fixed confidence threshold; default picks argmax-F1 from the sweep") |
| ap.add_argument("--n-examples", type=int, default=8) |
| ap.add_argument("--compare", nargs="*", default=None) |
| args = ap.parse_args() |
|
|
| set_seed(SEED) |
| if args.compare: |
| compare(args.compare, args.split) |
| return |
| if not args.ckpt: |
| ap.error("--ckpt is required (or use --compare)") |
| evaluate_one(args) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
