"""Compact 4-metric paper table on benchmark/v1/val. User-requested columns (and ONLY these): AUROC (binary, tick-level) AP_v (per-video AP, max-pool ALERT score per clip) F1* (oracle F1 — best F1 over all thresholds, fair-per-method) DAUS (Driver-Alert Utility Score, hit-rate 0.30, config B') Layout: one row per method. - VLAlert: honest pick = highest mean rank across (AUROC, AP_v, F1*, DAUS). Ranking uses all 21 VLAlert variants in per_tick/. - Baselines: ResNet50-LSTM, R3D-18, MViT-V2-S, Open-BADAS, Gemini-2.5-Flash-Lite (zero-shot). Each at its OWN best F1* threshold. Outputs: eval_results/benchmark_v1_val/paper_4metric_table.md eval_results/benchmark_v1_val/paper_4metric_sweep.md (all 21 VLAlert variants) Run: python tools/build_paper_4metric_table.py """ from __future__ import annotations import json from collections import defaultdict from pathlib import Path import numpy as np import torch from sklearn.metrics import (average_precision_score, precision_recall_curve, roc_auc_score) ROOT = Path("PROJECT_ROOT") PT_DIR = ROOT / "eval_results/benchmark_v1_val/per_tick" OUT_DIR = ROOT / "eval_results/benchmark_v1_val" DAUS_JSON = OUT_DIR / "daus_v1_val.json" BASELINES = [ ("resnet50_lstm", "ResNet50-LSTM"), ("r3d18", "R3D-18"), ("mvit_v2_s", "MViT-V2-S"), ("badas", "Open-BADAS"), ("gemini_zeroshot", "Gemini-2.5-Flash-Lite (zero-shot)"), ] def _safe(fn, *a, **kw): try: v = fn(*a, **kw) return float(v) if np.isfinite(v) else float("nan") except Exception: return float("nan") def metrics_one(pt_path: Path) -> dict | None: """Return {AUROC, AP_v, F1*, thr*, n_ticks, n_video, slug}.""" d = torch.load(pt_path, weights_only=False, map_location="cpu") if "scores_binary" not in d or "tick_label" not in d: return None ids = list(d.get("ids", [])) y3 = d["tick_label"].numpy().astype(np.int64) scores = d["scores_binary"].numpy().astype(np.float64) y_alert = (y3 == 2).astype(np.int64) mask = np.isfinite(scores) & (y3 >= 0) # AUROC binary auc = _safe(roc_auc_score, y_alert[mask], scores[mask]) # F1* try: prec, rec, thrs = precision_recall_curve(y_alert[mask], scores[mask]) f1s = (2 * prec * rec / np.where(prec + rec > 0, prec + rec, 1.0)) i_star = int(np.argmax(f1s[:-1])) f1_star = float(f1s[i_star]) thr_star = float(thrs[i_star]) except Exception: f1_star = thr_star = float("nan") # AP_v (per-video max-pool) per_vid_s = defaultdict(float) per_vid_l = defaultdict(int) for vid, lab, sc in zip(ids, y3, scores): if not np.isfinite(sc): continue per_vid_s[vid] = max(per_vid_s[vid], float(sc)) per_vid_l[vid] = max(per_vid_l[vid], int(lab == 2)) if per_vid_s: v_s = np.array(list(per_vid_s.values())) v_l = np.array(list(per_vid_l.values())) AP_v = _safe(average_precision_score, v_l, v_s) if 0 < v_l.sum() < len(v_l) else float("nan") else: AP_v = float("nan") return { "slug": pt_path.stem, "n_ticks": int(mask.sum()), "n_video": len(per_vid_s), "AUROC": auc, "AP_v": AP_v, "F1_star": f1_star, "thr_star": thr_star, } def fmt(v, p=3, dash="—"): return dash if v is None or not np.isfinite(v) else f"{v:.{p}f}" def main(): # ── DAUS lookup (from prior compute_daus_v1_val.py run) ── daus_map = {} if DAUS_JSON.exists(): d = json.loads(DAUS_JSON.read_text()) for slug, r in d.get("results", {}).items(): v = r.get("DAUS") daus_map[slug] = (float(v) if v is not None and (isinstance(v, (int, float)) and np.isfinite(v)) else float("nan")) # ── Per-method metrics ── rows = {} for p in sorted(PT_DIR.glob("*.pt")): m = metrics_one(p) if m is None: continue m["DAUS"] = daus_map.get(m["slug"], float("nan")) rows[m["slug"]] = m print(f" {m['slug']:35s} AUROC={fmt(m['AUROC'])} " f"AP_v={fmt(m['AP_v'])} F1*={fmt(m['F1_star'])} DAUS={fmt(m['DAUS'])}") # ── Honest VLAlert pick: mean-rank over 4 metrics ── vl = [r for r in rows.values() if r["slug"].startswith("vlalert_")] for metric in ("AUROC", "AP_v", "F1_star", "DAUS"): ranked = sorted(vl, key=lambda r: -(r[metric] if np.isfinite(r[metric]) else -1)) for i, r in enumerate(ranked): r.setdefault("ranks", {})[metric] = i + 1 for r in vl: r["rank_mean"] = float(np.mean(list(r["ranks"].values()))) vl.sort(key=lambda r: r["rank_mean"]) winner = vl[0] print(f"\n[honest pick] VLAlert winner = {winner['slug']} " f"(mean rank across 4 metrics = {winner['rank_mean']:.2f})") # ── Build compact paper table ── paper_rows = [winner] for slug, _name in BASELINES: if slug in rows: paper_rows.append(rows[slug]) else: print(f" [warn] missing {slug}") def pretty_name(r): if r["slug"] == winner["slug"]: return f"**VLAlert** _(={r['slug']})_" for slug, name in BASELINES: if r["slug"] == slug: return name return r["slug"] lines = ["# Final paper table — benchmark/v1/val (4 metrics)", "", f"Honest VLAlert winner (mean rank across AUROC, AP_v, F1, DAUS): " f"`{winner['slug']}` (mean rank {winner['rank_mean']:.2f}).", "", "Baselines: each at its own F1* oracle threshold (fair comparison).", "", "| Method | AUROC↑ | AP_v↑ | F1↑ | DAUS↑ |", "| :--- | ---: | ---: | ---: | ---: |"] for r in paper_rows: lines.append("| " + " | ".join([ pretty_name(r), fmt(r["AUROC"]), fmt(r["AP_v"]), fmt(r["F1_star"]), fmt(r["DAUS"], 4), ]) + " |") out_main = OUT_DIR / "paper_4metric_table.md" out_main.write_text("\n".join(lines) + "\n") print(f"\n[save] {out_main}") # ── Appendix: all 21 VLAlert variants ── vl_sorted = sorted(vl, key=lambda r: r["rank_mean"]) lines = ["# VLAlert variant sweep — benchmark/v1/val (4 metrics)", "", "Sorted by mean rank across AUROC, AP_v, F1, DAUS. Honest pick = top row.", "", "| # | Variant | AUROC↑ | AP_v↑ | F1↑ | DAUS↑ | mean_rank |", "| ---: | :--- | ---: | ---: | ---: | ---: | ---: |"] for i, r in enumerate(vl_sorted, 1): tag = "🏆 " if i == 1 else "" lines.append("| " + " | ".join([ str(i), tag + r["slug"], fmt(r["AUROC"]), fmt(r["AP_v"]), fmt(r["F1_star"]), fmt(r["DAUS"], 4), f"{r['rank_mean']:.2f}", ]) + " |") out_sweep = OUT_DIR / "paper_4metric_sweep.md" out_sweep.write_text("\n".join(lines) + "\n") print(f"[save] {out_sweep}") if __name__ == "__main__": main()