Buckets:
| #!/usr/bin/env python3 | |
| """tools/review_tool/apply_review_v3.py — summary of the owner's v3 label validation (relabel_v3.html, items_relabel_v3.json; pre-registered | |
| docs/decisions.md 2026-09-16T16:45:31Z). From the server labels (results/review_tool/labels_latest.json, ids V3_* / V3R_*) it computes | |
| * per cell (engaged hand_return, manual hand_return, engaged hand_leave; plus hand_return pooled): n, labelled, correct (clicked first-frame | |
| within +-TOL = 0.5 s of the automatic stamp dt = 0), timing (clicked frame outside +-TOL), never_* / always_* (no such event in the window), | |
| unsure, unlabelled; PRECISION = correct / (labelled - unsure) with the Wilson 95 % interval; a lenient column counts timing as a real event; | |
| the clicked-dt distribution (stamp bias: v3 stamps the FIRST frame of the new state, so the click is expected at dt ~ -0.1 ... 0.0); | |
| * the pre-registered ACCEPTANCE verdict per cell: precision >= 0.80 AND Wilson lower bound >= 0.65 (PROVISIONAL while items are unlabelled); | |
| * the RECALL estimate from the window block (30 engaged 10 s windows without a v3 hand event): q = yes / (yes + no) with Wilson CI; | |
| recall ~= TP / (TP + FN) with TP = precision(engaged hand_return) x N_v3_returns (engaged test, row_ok & EM, first step) and | |
| FN = q x (population of v3-negative window starts x 0.2 s / 10 s), assuming <= 1 real return per 10 s negative window; the interval | |
| combines the Wilson bounds of precision and q crosswise (rough). | |
| Appends one section to results/tables/review_tool_summary.md (or --out <file> for a dry run) and optionally writes --json-out. | |
| Read-only on everything else; no target file, run dir or manuscript is touched. | |
| python tools/review_tool/apply_review_v3.py [results/review_tool/labels_latest.json] [--items results/review_tool/items_relabel_v3.json] | |
| [--out results/tables/review_tool_summary.md] [--json-out <file>] [--tol 0.5] | |
| """ | |
| from __future__ import annotations | |
| import argparse, json, math, sys, time | |
| from pathlib import Path | |
| CWM3 = Path(__file__).resolve().parents[2] | |
| sys.path.insert(0, str(Path(__file__).resolve().parent)) | |
| import apply_review as AR # noqa: E402 (wilson, load_export) | |
| LABELS_DEFAULT = CWM3 / "results/review_tool/labels_latest.json" | |
| ITEMS_DEFAULT = CWM3 / "results/review_tool/items_relabel_v3.json" | |
| OUT_MD_DEFAULT = CWM3 / "results/tables/review_tool_summary.md" | |
| ACC_PREC, ACC_LB, TOL = 0.80, 0.65, 0.5 | |
| WRONG = {"never_on", "always_on", "never_off", "always_off"} | |
| def cell_counts(items, exp, tol): | |
| c = {"n": len(items), "labelled": 0, "correct": 0, "timing": 0, "never": 0, "always": 0, "unsure": 0, "unlabelled": 0, "other": 0, "dts": []} | |
| for it in items: | |
| l = (exp.get(it["id"]) or {}).get("label", "") | |
| if not l: c["unlabelled"] += 1; continue | |
| c["labelled"] += 1 | |
| if l.startswith("ret_dt=") or l.startswith("off_dt="): | |
| dt = float(l.split("=")[1]); c["dts"].append(dt) | |
| c["correct" if abs(dt) <= tol + 1e-9 else "timing"] += 1 | |
| elif l.startswith("never_"): c["never"] += 1 | |
| elif l.startswith("always_"): c["always"] += 1 | |
| elif l == "unsure": c["unsure"] += 1 | |
| else: c["other"] += 1 | |
| c["n_eff"] = c["labelled"] - c["unsure"] | |
| c["precision"] = c["correct"] / c["n_eff"] if c["n_eff"] else float("nan") | |
| c["ci"] = AR.wilson(c["correct"], c["n_eff"]) | |
| c["precision_lenient"] = (c["correct"] + c["timing"]) / c["n_eff"] if c["n_eff"] else float("nan") | |
| c["ci_lenient"] = AR.wilson(c["correct"] + c["timing"], c["n_eff"]) | |
| return c | |
| def verdict(c): | |
| if c["n_eff"] == 0: return "no labels yet" | |
| ok = c["precision"] >= ACC_PREC and c["ci"][0] >= ACC_LB | |
| v = "ACCEPT" if ok else "REJECT" | |
| return v + (f" (PROVISIONAL: {c['unlabelled']} unlabelled)" if c["unlabelled"] else "") | |
| def fmt_ci(p, ci): | |
| return "--" if not (p == p) else f"{p:.3f} [{ci[0]:.3f}, {ci[1]:.3f}]" | |
| def dt_stats(dts): | |
| if not dts: return "--" | |
| s = sorted(dts); n = len(s); med = s[n // 2] if n % 2 else 0.5 * (s[n // 2 - 1] + s[n // 2]) | |
| return f"n {n}, median {med:+.1f}, mean {sum(s) / n:+.2f}, min {s[0]:+.1f}, max {s[-1]:+.1f}" | |
| def main(argv=None) -> int: | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("labels", nargs="?", default=str(LABELS_DEFAULT)) | |
| ap.add_argument("--items", default=str(ITEMS_DEFAULT)); ap.add_argument("--out", default=str(OUT_MD_DEFAULT)) | |
| ap.add_argument("--json-out", default=None); ap.add_argument("--tol", type=float, default=TOL) | |
| a = ap.parse_args(argv) | |
| exp = AR.load_export(Path(a.labels)); P = json.load(open(a.items, encoding="utf-8")); items = P["items"] | |
| ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) | |
| ev_items = [i for i in items if i["task"] == "V3E"]; w_items = [i for i in items if i["task"] == "V3R"] | |
| cells = {"engaged_hand_return": [i for i in ev_items if i["meta"]["cell"] == "engaged_hand_return"], | |
| "manual_hand_return": [i for i in ev_items if i["meta"]["cell"] == "manual_hand_return"], | |
| "engaged_hand_leave": [i for i in ev_items if i["meta"]["cell"] == "engaged_hand_leave"]} | |
| C = {k: cell_counts(v, exp, a.tol) for k, v in cells.items()} | |
| C["hand_return_pooled"] = cell_counts(cells["engaged_hand_return"] + cells["manual_hand_return"], exp, a.tol) | |
| # ---- recall block | |
| W = {"n": len(w_items), "yes": 0, "no": 0, "unsure": 0, "unlabelled": 0, "other": 0} | |
| for it in w_items: | |
| l = (exp.get(it["id"]) or {}).get("label", "") | |
| W["yes" if l == "yes" else "no" if l == "no" else "unsure" if l == "unsure" else "unlabelled" if not l else "other"] += 1 | |
| W["n_eff"] = W["yes"] + W["no"]; W["q"] = W["yes"] / W["n_eff"] if W["n_eff"] else float("nan"); W["q_ci"] = AR.wilson(W["yes"], W["n_eff"]) | |
| st = P.get("sampling_stats", {}).get("recall_windows", {}) | |
| n_ret = st.get("n_v3_hand_return_engaged_test"); n_starts = st.get("n_valid_window_starts"); step = st.get("start_step_s", 0.2); win_s = st.get("window_s", 10.0) | |
| R = {"n_v3_returns": n_ret, "n_neg_windows": (n_starts * step / win_s) if n_starts else None} | |
| pe = C["engaged_hand_return"] | |
| if R["n_neg_windows"] and pe["n_eff"] and W["n_eff"]: | |
| def rec(p, q): tp = p * n_ret; fn = q * R["n_neg_windows"]; return tp / (tp + fn) if tp + fn > 0 else float("nan") | |
| R.update(tp_est=pe["precision"] * n_ret, fn_est=W["q"] * R["n_neg_windows"], recall=rec(pe["precision"], W["q"]), | |
| recall_lo=rec(pe["ci"][0], W["q_ci"][1]), recall_hi=rec(pe["ci"][1], W["q_ci"][0])) | |
| # ---- section | |
| n_lab = sum(1 for i in items if (exp.get(i["id"]) or {}).get("label")) | |
| lines = [f"\n## Review-tool summary — v3 label validation (fresh events, frame-click) {ts} (labels {Path(a.labels).name}; items {Path(a.items).name} built {P.get('built_utc')}; labelled {n_lab}/{len(items)})\n", | |
| f"Protocol (pre-registered 2026-09-16T16:45:31Z): fresh v3 events on TEST routes with local video, none within +-5 s of the 34 reviewed engaged hand_return events; " | |
| f"21 frames at 0.1 s (-1.0 ... +1.0 s), EVENT = dt 0 = t_anchor + 0.2 s; hand_return: first frame with a hand on the wheel (or never_on / always_on / unsure); " | |
| f"hand_leave: first frame with both hands off (or never_off / always_off / unsure). correct = clicked frame within +-{a.tol:.1f} s of dt 0; timing = real event but " | |
| f"outside +-{a.tol:.1f} s; never/always = no such event in the window. Precision = correct / (labelled - unsure). Acceptance per cell: precision >= {ACC_PREC:.2f} and Wilson " | |
| f"95 % lower bound >= {ACC_LB:.2f}. The hand_return items of both regimes were shown in one shuffled block without the regime.\n", | |
| "| cell | n | labelled | correct (|dt| <= tol) | timing | never | always | unsure | unlabelled | precision [Wilson 95 %] | lenient (correct + timing) | verdict |", | |
| "|---|---|---|---|---|---|---|---|---|---|---|---|"] | |
| for k in ("engaged_hand_return", "manual_hand_return", "engaged_hand_leave", "hand_return_pooled"): | |
| c = C[k] | |
| lines.append(f"| {k} | {c['n']} | {c['labelled']} | {c['correct']} | {c['timing']} | {c['never']} | {c['always']} | {c['unsure']} | {c['unlabelled']} | " | |
| f"{fmt_ci(c['precision'], c['ci'])} | {fmt_ci(c['precision_lenient'], c['ci_lenient'])} | {verdict(c) if k != 'hand_return_pooled' else '(info)'} |") | |
| lines += ["", "Clicked first-frame offsets dt (s) relative to the automatic stamp (v3 stamps the first frame of the new state, so dt ~ -0.1 ... 0.0 is expected):", ""] | |
| lines += [f"- {k}: {dt_stats(C[k]['dts'])}" for k in ("engaged_hand_return", "manual_hand_return", "engaged_hand_leave")] | |
| lines += ["", "### Recall check — 30 random engaged TEST 10 s windows without a v3 hand event (1 Hz frames): 'did a hand return to the wheel and stay >= 1 s?'", "", | |
| "| windows | yes (v3 missed a return) | no | unsure | unlabelled | miss share q = yes / (yes + no) [Wilson 95 %] |", "|---|---|---|---|---|---|", | |
| f"| {W['n']} | {W['yes']} | {W['no']} | {W['unsure']} | {W['unlabelled']} | {fmt_ci(W['q'], W['q_ci'])} |", ""] | |
| if n_starts: | |
| lines.append(f"Window population: {n_starts:,} valid start positions x {step} s = {n_starts * step / 3600:.2f} h of engaged TEST time with no v3 hand_return / hand_leave in " | |
| f"[t0 - 1 s, t0 + 10 s], contiguous anchors, row_ok, wheel reference and hands_on state defined, not overlapping the reviewed +-5 s zones " | |
| f"(= {R['n_neg_windows']:.0f} disjoint 10 s windows); v3 engaged TEST hand_return events (row_ok & EM, first step): {n_ret}.") | |
| if "recall" in R: | |
| lines.append(f"Recall estimate = TP / (TP + FN) with TP = precision(engaged hand_return) x {n_ret} = {R['tp_est']:.1f} and FN = q x {R['n_neg_windows']:.0f} = {R['fn_est']:.1f}: " | |
| f"**{R['recall']:.3f}** (rough interval [{R['recall_lo']:.3f}, {R['recall_hi']:.3f}] from the Wilson bounds of precision and q crosswise; assumes <= 1 real return per " | |
| f"10 s negative window and that the sampled windows represent the v3-negative time).") | |
| else: | |
| lines.append("Recall estimate: not available yet (needs labelled engaged hand_return items AND labelled windows).") | |
| lines += ["", "| item | cell | car | t_event / t0 | label | note |", "|---|---|---|---|---|---|"] | |
| for it in items: | |
| r = exp.get(it["id"]) or {}; m = it["meta"] | |
| tt = m.get("t_event", m.get("t0")); lines.append(f"| {it['id']} | {m['cell']} | {m['car']} | {tt:.1f} | {r.get('label', '') or '--'} | {(r.get('note') or '').replace('|', '/')} |") | |
| text = "\n".join(lines) + "\n" | |
| out = Path(a.out); out.parent.mkdir(parents=True, exist_ok=True) | |
| with open(out, "a", encoding="utf-8") as f: | |
| f.write(text) | |
| print("\n".join(lines[:22])); print(f"\nappended {out} ({len(items)} items, {n_lab} labelled)") | |
| if a.json_out: | |
| summ = {"written_utc": ts, "labels": str(a.labels), "items": str(a.items), "tol_s": a.tol, "acceptance": {"precision_min": ACC_PREC, "wilson_lb_min": ACC_LB}, | |
| "cells": {k: dict({kk: vv for kk, vv in v.items() if kk != "dts"}, dts=v["dts"], verdict=verdict(v)) for k, v in C.items()}, "recall_windows": W, "recall_estimate": R} | |
| Path(a.json_out).write_text(json.dumps(summ, indent=1, default=str)); print(f"wrote {a.json_out}") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 11.5 kB
- Xet hash:
- 6596992275d4cccee3cc6563df64a9a6c6e1c8fda5df78765eb53666b4deeca4
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.