Buckets:
| """cwm3/eval/gaze_off_validation_analysis.py — pre-registered analysis of the gaze_off human validation (docs/newtest_prereg.md §4; 2026-09-12). | |
| Inputs: results/gaze_validation/manifest_<v>.csv (design: cell, pi, N_stratum, gaze_off, dwell_s, head_turned, face_valid, kin_valid, ear_conf_asym, dongle) | |
| and the gaze records (stage `gaze`) of cwm2/annotation_human/labels_newtest.jsonl (two blind raters per window; ADJ* records resolve disagreements). | |
| Per window: human off-road = view_region != road_ahead (unknown excluded); sustained off-road = check_kind == sustained_off_road; mirror check = check_kind == | |
| mirror_check; eyes-only glance = eyes_vs_head == head_on_eyes_off; occluded = occlusion != none. Consensus = both raters agree; disagreement -> ADJ record if | |
| present, else excluded from the precision / recall estimates (counted). Weights w = 1 / pi (HT to the val prevalence; N_stratum stored). | |
| Metrics (HT-weighted unless stated): rater kappa (unweighted; view_region collapsed on/off-road, check_kind, eyes_vs_head); confusion gaze_off state | |
| {on, off_brief, off_sust, face_invalid} x human {on, off}; precision / recall of gaze_off = 1 vs human off-road and vs sustained off-road; share of gaze_off = 1 | |
| that are mirror checks; share of gaze_off = 1 and of face_invalid with occlusion; recall of eyes-only glances by gaze_off (pre-stated expectation ~ 0: gaze_off is a | |
| head-yaw hysteresis on |lm_yaw_dev|, P90 / P75 = 0.212 / 0.103); device-cluster bootstrap CIs (2,000). Output: a section appended to | |
| results/tables/gaze_off_validation.md + CSVs in results/gaze_validation/analysis_<stamp>/ + run json. --self-test: synthetic labels with a known generative | |
| model -> HT precision / recall recover the population values; writes nothing. | |
| python eval/gaze_off_validation_analysis.py [--version v1] [--n-boot 2000] [--panel G1] [--self-test] | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| import time | |
| from collections import Counter, defaultdict | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| CWM2 = CWM3.parent / "cwm2" | |
| for p in (CWM3 / "eval", CWM3 / "common"): | |
| sys.path.insert(0, str(p)) | |
| from runmeta import write_run_json # noqa: E402 | |
| import newtest_analysis as NA # noqa: E402 (DeviceBoot, pct, wording, f3, sha256_file) | |
| GAZEV = CWM3 / "results" / "gaze_validation" | |
| STORE = CWM2 / "annotation_human" / "labels_newtest.jsonl" | |
| TABLE = CWM3 / "results" / "tables" / "gaze_off_validation.md" | |
| DEVIATIONS = GAZEV / "deviations_v1.json" | |
| STATES = ("on", "off_brief", "off_sust", "face_invalid") | |
| def kappa(a, b, labels): | |
| a, b = np.asarray(a), np.asarray(b) | |
| if len(a) == 0: | |
| return np.nan | |
| po = float(np.mean(a == b)) | |
| pe = float(sum(np.mean(a == l) * np.mean(b == l) for l in labels)) | |
| return (po - pe) / (1 - pe) if pe < 1 else np.nan | |
| def state_of(row): | |
| if not bool(row.face_valid): | |
| return "face_invalid" | |
| if int(row.gaze_off) == 0: | |
| return "on" | |
| return "off_brief" if (np.isfinite(row.dwell_s) and row.dwell_s <= 2.0) else "off_sust" | |
| def load_gaze_records(store, aliases=()): | |
| by = defaultdict(dict) | |
| if not Path(store).exists(): | |
| return by | |
| amap = dict(x.split("=", 1) for x in aliases if "=" in x) | |
| for line in open(store): | |
| try: | |
| r = json.loads(line) | |
| except json.JSONDecodeError: | |
| continue | |
| if r.get("stage") == "gaze" and r.get("kind") != "skip": | |
| by[int(r["idx"])].setdefault(amap.get(str(r["annotator"]), str(r["annotator"])), r.get("fields", {})) | |
| return by | |
| def consensus_frame(M, by): | |
| """per window: consensus human labels (agree / adjudicated) + per-rater pairs for kappa.""" | |
| rows, pairs = [], defaultdict(list) | |
| for r in M.itertuples(): | |
| recs = by.get(int(r.idx), {}) | |
| blind = {a_: f for a_, f in recs.items() if not a_.upper().startswith("ADJ")} | |
| adj = {a_: f for a_, f in recs.items() if a_.upper().startswith("ADJ")} | |
| fs = list(blind.values()) | |
| rec = dict(idx=r.idx, dongle=r.dongle, regime=r.regime, cell=r.cell, state=state_of(r), pi=r.pi, w=1.0 / r.pi, n_raters=len(blind), gaze_off=int(r.gaze_off), face_valid=bool(r.face_valid)) | |
| def lab(name, collapse=None): | |
| vals = [f.get(name) for f in fs] | |
| if collapse: | |
| vals = [collapse(v) for v in vals] | |
| vals = [v for v in vals if v not in (None, "unknown")] | |
| if len(fs) >= 2 and len(vals) == 2: | |
| pairs[name].append(tuple(vals)) | |
| if len(vals) >= 2 and len(set(vals)) == 1: | |
| return vals[0], "consensus" | |
| if adj: | |
| v = list(adj.values())[0].get(name) | |
| v = collapse(v) if collapse else v | |
| return (v, "adjudicated") if v not in (None, "unknown") else (None, "unknown") | |
| if len(vals) == 1 and len(fs) == 1: | |
| return vals[0], "single" | |
| return None, ("disagree" if len(vals) >= 2 else "unknown") | |
| offroad = lambda v: None if v in (None, "unknown") else ("on" if v == "road_ahead" else "off") # noqa: E731 | |
| rec["human_offroad"], rec["human_offroad_status"] = lab("view_region", offroad) | |
| rec["check_kind"], rec["check_kind_status"] = lab("check_kind") | |
| rec["eyes_vs_head"], rec["eyes_vs_head_status"] = lab("eyes_vs_head") | |
| rec["occlusion"], rec["occlusion_status"] = lab("occlusion") | |
| rec["head_motion"], rec["head_motion_status"] = lab("head_motion") | |
| rows.append(rec) | |
| return pd.DataFrame(rows), pairs | |
| def estimates(C, n_boot, seed=0): | |
| B = NA.DeviceBoot(C.dongle.to_numpy(), n_boot, seed) | |
| w = C.w.to_numpy() | |
| g = (C.gaze_off == 1).to_numpy() & C.face_valid.to_numpy() | |
| fi = ~C.face_valid.to_numpy() | |
| off = (C.human_offroad == "off").to_numpy() | |
| on = (C.human_offroad == "on").to_numpy() | |
| lab_off = off | on | |
| sust = (C.check_kind == "sustained_off_road").to_numpy() | |
| lab_ck = C.check_kind.notna().to_numpy() | |
| mirror = (C.check_kind == "mirror_check").to_numpy() | |
| eyes_only = (C.eyes_vs_head == "head_on_eyes_off").to_numpy() | |
| lab_eh = C.eyes_vs_head.notna().to_numpy() | |
| occl = (C.occlusion.notna() & (C.occlusion != "none")).to_numpy() | |
| lab_oc = C.occlusion.notna().to_numpy() | |
| stats = { | |
| "precision_gaze_off_vs_offroad": lambda pk: NA.ht_ratio(w[pk][g[pk] & lab_off[pk]], off[pk][g[pk] & lab_off[pk]], np.ones(int((g[pk] & lab_off[pk]).sum()))), | |
| "recall_gaze_off_vs_offroad": lambda pk: NA.ht_ratio(w[pk][off[pk] & C.face_valid.to_numpy()[pk]], g[pk][off[pk] & C.face_valid.to_numpy()[pk]], np.ones(int((off[pk] & C.face_valid.to_numpy()[pk]).sum()))), | |
| "precision_gaze_off_vs_sustained": lambda pk: NA.ht_ratio(w[pk][g[pk] & lab_ck[pk]], sust[pk][g[pk] & lab_ck[pk]], np.ones(int((g[pk] & lab_ck[pk]).sum()))), | |
| "recall_gaze_off_vs_sustained": lambda pk: NA.ht_ratio(w[pk][sust[pk] & C.face_valid.to_numpy()[pk]], g[pk][sust[pk] & C.face_valid.to_numpy()[pk]], np.ones(int((sust[pk] & C.face_valid.to_numpy()[pk]).sum()))), | |
| "share_gaze_off_mirror_check": lambda pk: NA.ht_ratio(w[pk][g[pk] & lab_ck[pk]], mirror[pk][g[pk] & lab_ck[pk]], np.ones(int((g[pk] & lab_ck[pk]).sum()))), | |
| "share_gaze_off_occluded": lambda pk: NA.ht_ratio(w[pk][g[pk] & lab_oc[pk]], occl[pk][g[pk] & lab_oc[pk]], np.ones(int((g[pk] & lab_oc[pk]).sum()))), | |
| "share_face_invalid_occluded": lambda pk: NA.ht_ratio(w[pk][fi[pk] & lab_oc[pk]], occl[pk][fi[pk] & lab_oc[pk]], np.ones(int((fi[pk] & lab_oc[pk]).sum()))), | |
| "recall_eyes_only_glances": lambda pk: NA.ht_ratio(w[pk][eyes_only[pk] & C.face_valid.to_numpy()[pk]], g[pk][eyes_only[pk] & C.face_valid.to_numpy()[pk]], np.ones(int((eyes_only[pk] & C.face_valid.to_numpy()[pk]).sum()))), | |
| "prevalence_human_offroad_val": lambda pk: NA.ht_ratio(w[pk][lab_off[pk]], off[pk][lab_off[pk]], np.ones(int(lab_off[pk].sum()))), | |
| "prevalence_gaze_off_val": lambda pk: NA.ht_ratio(w[pk], g[pk].astype(float), np.ones(len(pk))), | |
| } | |
| rows = [] | |
| full = np.arange(len(C)) | |
| for k, fn in stats.items(): | |
| v = fn(full) | |
| b = B.run(fn) | |
| lo, hi = NA.pct(b) | |
| n = {"precision_gaze_off_vs_offroad": int((g & lab_off).sum()), "recall_gaze_off_vs_offroad": int((off & C.face_valid).sum()), "precision_gaze_off_vs_sustained": int((g & lab_ck).sum()), | |
| "recall_gaze_off_vs_sustained": int((sust & C.face_valid).sum()), "share_gaze_off_mirror_check": int((g & lab_ck).sum()), "share_gaze_off_occluded": int((g & lab_oc).sum()), | |
| "share_face_invalid_occluded": int((fi & lab_oc).sum()), "recall_eyes_only_glances": int((eyes_only & C.face_valid).sum()), "prevalence_human_offroad_val": int(lab_off.sum()), | |
| "prevalence_gaze_off_val": len(C)}[k] | |
| rows.append(dict(metric=k, value=v, lo=lo, hi=hi, n=n)) | |
| conf = pd.crosstab(C.state, C.human_offroad.fillna("unlabelled"), values=C.w, aggfunc="sum").fillna(0) | |
| conf_n = pd.crosstab(C.state, C.human_offroad.fillna("unlabelled")) | |
| return pd.DataFrame(rows), conf, conf_n | |
| def self_test(): | |
| rng = np.random.default_rng(3) | |
| n = 4000 | |
| devs = rng.choice([f"D{i}" for i in range(10)], n) | |
| face_valid = rng.random(n) > 0.1 | |
| off_true = rng.random(n) < 0.15 | |
| gaze = (face_valid & off_true & (rng.random(n) < 0.7)) | (face_valid & ~off_true & (rng.random(n) < 0.05)) | |
| pop = pd.DataFrame(dict(idx=np.arange(n), dongle=devs, regime="manual", face_valid=face_valid, gaze_off=gaze.astype(int), dwell_s=np.where(gaze, rng.uniform(0.5, 4, n), np.nan), head_turned=0.0, | |
| off=off_true)) | |
| pop["cell"] = np.where(~pop.face_valid, "face_invalid|kin_invalid", np.where(pop.gaze_off == 1, "off_brief|straight", "on|straight")) | |
| truth_prec = float((pop.gaze_off.astype(bool) & pop.off).sum() / pop.gaze_off.sum()) | |
| truth_rec = float((pop.gaze_off.astype(bool) & pop.off & pop.face_valid).sum() / (pop.off & pop.face_valid).sum()) | |
| est_p, est_r = [], [] | |
| for rep in range(40): | |
| rr = np.random.default_rng(rep) | |
| parts = [] | |
| for cell, n_c in (("on|straight", 60), ("off_brief|straight", 60), ("face_invalid|kin_invalid", 30)): | |
| sub = pop[pop.cell == cell] | |
| pick = sub.iloc[rr.choice(len(sub), min(n_c, len(sub)), replace=False)].copy() | |
| pick["pi"] = len(pick) / len(sub) | |
| parts.append(pick) | |
| S = pd.concat(parts) | |
| by = {int(i): {"R1": dict(view_region="road_ahead" if not o else "center_console", check_kind="none" if not o else "instrument_or_console_check", eyes_vs_head="head_and_eyes_on_road", occlusion="none"), | |
| "R2": dict(view_region="road_ahead" if not o else "center_console", check_kind="none" if not o else "instrument_or_console_check", eyes_vs_head="head_and_eyes_on_road", occlusion="none")} | |
| for i, o in zip(S.idx, S.off)} | |
| C, _ = consensus_frame(S, by) | |
| E, _, _ = estimates(C, 20, rep) | |
| est_p.append(float(E[E.metric == "precision_gaze_off_vs_offroad"].value.iloc[0])) | |
| est_r.append(float(E[E.metric == "recall_gaze_off_vs_offroad"].value.iloc[0])) | |
| print(f"self-test precision truth {truth_prec:.3f} HT mean {np.mean(est_p):.3f} (sd {np.std(est_p):.3f}); recall truth {truth_rec:.3f} HT mean {np.mean(est_r):.3f} (sd {np.std(est_r):.3f})") | |
| ok = abs(np.mean(est_p) - truth_prec) < 0.04 and abs(np.mean(est_r) - truth_rec) < 0.06 | |
| assert abs(kappa(["a", "b", "a", "b"], ["a", "b", "a", "b"], ["a", "b"]) - 1.0) < 1e-9 and kappa(["a", "a", "b", "b"], ["a", "b", "a", "b"], ["a", "b"]) == 0.0 | |
| print("SELF-TEST", "PASS" if ok else "FAIL") | |
| return ok | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--version", default="v1") | |
| ap.add_argument("--n-boot", type=int, default=2000) | |
| ap.add_argument("--store", default=str(STORE)) | |
| ap.add_argument("--deviations", default=str(DEVIATIONS)) | |
| ap.add_argument("--panel", default="", help="committee-consensus mode: the panel id's single record per window is the label; kappa reported as n/a") | |
| ap.add_argument("--supplement", default="supp_v1", help="post-hoc supplementary gaze stratum manifest (results/gaze_validation/manifest_<supp>.csv); '' = none") | |
| ap.add_argument("--alias", nargs="*", default=[], help="OLD=NEW annotator id renames applied in memory") | |
| ap.add_argument("--self-test", action="store_true") | |
| a = ap.parse_args() | |
| if a.self_test: | |
| sys.exit(0 if self_test() else 1) | |
| t0 = time.time() | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| fc = json.load(open(GAZEV / f"frame_counts_{a.version}.json")) | |
| # pre-analysis exclusions (renderer deviations): windows with no cabin video at t are NOT missing at random — they are a frame-level category of the | |
| # automatic state ("face_invalid because the cabin video ended before t"). They are reported as their own category with their HT weight (1 / pi of | |
| # the original design) and removed from the human-label estimates; their strata mates keep the original pi (the category is estimated, not dropped). | |
| M["no_cabin_video"] = False | |
| dev_rows = [] | |
| if Path(a.deviations).exists(): | |
| for e in json.load(open(a.deviations))["exclusions"]: | |
| i = int(e["idx"]) | |
| if i in set(M.idx): | |
| M.loc[M.idx == i, "no_cabin_video"] = e.get("reason") == "no_cabin_video_at_t" | |
| dev_rows.append(dict(idx=i, cell=e.get("cell"), regime=e.get("regime"), reason=e.get("reason"), t=e.get("t"), leaf=e.get("leaf"))) | |
| other = [x for x in dev_rows if x["reason"] != "no_cabin_video_at_t"] | |
| if other: # genuine render failures: unit non-response, re-weight the stratum | |
| for x in other: | |
| r = M[M.idx == x["idx"]].iloc[0] | |
| mates = M[(M.stratum == r.stratum) & (M.idx != x["idx"])] | |
| if len(mates) and r.n_stratum > 1: | |
| M.loc[mates.index, "pi"] = M.loc[mates.index, "pi"] * (r.n_stratum - 1) / r.n_stratum | |
| M = M[~M.idx.isin([x["idx"] for x in other])].reset_index(drop=True) | |
| ncv = M[M.no_cabin_video] | |
| w_ncv_share = {} | |
| for c in sorted(M.cell.unique()): | |
| sub = M[M.cell == c] | |
| w_ncv_share[c] = float(np.sum(1 / sub[sub.no_cabin_video].pi) / np.sum(1 / sub.pi)) if len(sub) else np.nan | |
| print(f"pre-analysis exclusions: {len(dev_rows)} ({dict(Counter(x['reason'] for x in dev_rows))}); no-cabin-video windows {len(ncv)} -> HT share of their cells: " | |
| + ", ".join(f"{c} {v:.3f}" for c, v in w_ncv_share.items() if v > 0)) | |
| M = M[~M.no_cabin_video].reset_index(drop=True) | |
| by = load_gaze_records(a.store, a.alias) | |
| C, pairs = consensus_frame(M, by) | |
| if a.panel: | |
| pairs = {} # no independent raters: agreement is not estimable | |
| n_lab = int(C.n_raters.gt(0).sum()) | |
| print(f"windows {len(M)}; with gaze records {n_lab}; double-rated {int(C.n_raters.ge(2).sum())}; consensus off-road labels {int(C.human_offroad.notna().sum())}") | |
| if n_lab == 0: | |
| print("no gaze records yet — nothing to estimate") | |
| return | |
| out = GAZEV / f"analysis_{stamp}" | |
| out.mkdir(parents=True, exist_ok=True) | |
| E, conf, conf_n = estimates(C, a.n_boot) | |
| # ---- post-hoc supplementary stratum gaze_supp (deviation 2026-09-13): reported alone, next to the original cell, and in one labelled pooled row; never merged into the v1 weights | |
| supp_rows = [] | |
| sp = GAZEV / f"manifest_{a.supplement}.csv" if a.supplement else None | |
| if sp is not None and sp.exists(): | |
| Ms = pd.read_csv(sp) | |
| dev_s = GAZEV / f"deviations_{a.supplement}.json" | |
| if dev_s.exists(): | |
| ex = {int(e["idx"]) for e in json.load(open(dev_s))["exclusions"]} | |
| Ms = Ms[~Ms.idx.isin(ex)].reset_index(drop=True) | |
| Cs, _ = consensus_frame(Ms, by) | |
| Cs = Cs[Cs.n_raters > 0] | |
| Co = C[(C.cell == "face_invalid|kin_invalid") & (C.n_raters > 0)] | |
| def occ_share(X): | |
| m = X.occlusion.notna() | |
| return NA.ht_ratio(X.w.to_numpy()[m], (X.occlusion[m] != "none").to_numpy().astype(float), np.ones(int(m.sum()))) if m.any() else np.nan | |
| def occ_boot(X, seed): | |
| if not len(X): | |
| return (np.nan, np.nan) | |
| B = NA.DeviceBoot(X.dongle.to_numpy(), a.n_boot, seed) | |
| m = X.occlusion.notna().to_numpy() | |
| w, o = X.w.to_numpy(), (X.occlusion.fillna("").to_numpy() != "none").astype(float) | |
| return NA.pct(B.run(lambda pk: NA.ht_ratio(w[pk][m[pk]], o[pk][m[pk]], np.ones(int(m[pk].sum()))))) | |
| for label, X in (("original cell (5 ratable windows; design weights; 0.803 of the cell's weight is 'no cabin video')", Co), (f"supplementary stratum gaze_supp ({a.supplement}; own weights within its frame)", Cs), | |
| ("POST HOC POOLED — cell rows with cabin video (union of both samples, each unit with its own design weight; not pre-registered)", pd.concat([Co, Cs], ignore_index=True))): | |
| lo, hi = occ_boot(X, 11) | |
| supp_rows.append(dict(estimate=label, n_rated=int(len(X)), n_windows_manifest=(int(len(Ms)) if "supplementary" in label else (int((C.cell == "face_invalid|kin_invalid").sum()) if "original" in label else int(len(Ms) + (C.cell == "face_invalid|kin_invalid").sum()))), | |
| share_face_invalid_occluded=occ_share(X), lo=lo, hi=hi, occlusion_kinds=dict(Counter(X.occlusion.dropna())) if len(X) else {})) | |
| print("supplementary stratum rows:", [(r["estimate"][:40], r["n_rated"], round(r["share_face_invalid_occluded"], 3) if np.isfinite(r["share_face_invalid_occluded"]) else None) for r in supp_rows]) | |
| kap = {} | |
| if pairs.get("view_region"): | |
| P = np.array(pairs["view_region"]) | |
| kap["view_region_on_off"] = kappa(P[:, 0], P[:, 1], ["on", "off"]) | |
| for k in ("check_kind", "eyes_vs_head", "occlusion", "head_motion"): | |
| if pairs.get(k): | |
| P = np.array(pairs[k]) | |
| kap[k] = kappa(P[:, 0], P[:, 1], sorted(set(P.ravel()))) | |
| prereg_now = NA.sha256_file(NA.PREREG) if NA.PREREG.exists() else "missing" | |
| prereg_reg = json.load(open(NA.PREREG_HASH)).get("sha256") if NA.PREREG_HASH.exists() else None | |
| L = [f"## gaze_off human validation — manifest {a.version}, run {stamp} (eval/gaze_off_validation_analysis.py)", "", | |
| f"Pre-registration sha256 now `{prereg_now}` vs registered `{prereg_reg}` -> {'MATCH' if prereg_now == prereg_reg else 'MISMATCH'}. Windows {len(M)} (VAL split, {fc['n_devices_val']} devices); " | |
| f"rated {n_lab}, double-rated {int(C.n_raters.ge(2).sum())}; HT weights 1 / pi to the val prevalence (strata = cell x regime x |ear_conf_asym| half); {a.n_boot} device-cluster resamples.", "", | |
| "### Rater agreement (unweighted Cohen's kappa over double-rated windows; pre-specified target >= 0.6 for view_region on/off)", ""] | |
| if a.panel: | |
| L += [f"**n/a — committee consensus (single operator `{a.panel}`; deviation 2026-09-13, `docs/newtest_prereg_deviations.md`): the κ ≥ 0.6 gate is not applicable and not passed; " | |
| "inter-rater reliability is not estimable (effective n_raters = 1).**"] | |
| else: | |
| L += ["| field | kappa | n pairs |", "|---|---|---|"] | |
| for k, v in kap.items(): | |
| L.append(f"| {k} | {NA.f3(v)} | {len(pairs['view_region' if k == 'view_region_on_off' else k])} |") | |
| L += ["", "### HT-weighted metrics (val prevalence) [device-cluster CI]", "", "| metric | value | CI | n windows |", "|---|---|---|---|"] | |
| for r in E.itertuples(): | |
| L.append(f"| {r.metric} | {NA.f3(r.value)} | [{NA.f3(r.lo, 2)}, {NA.f3(r.hi, 2)}] | {r.n} |") | |
| if dev_rows: | |
| L += ["", f"### Pre-analysis exclusions ({len(dev_rows)}; deviations log)", "", "| idx | cell | regime | t | reason | leaf |", "|---|---|---|---|---|---|"] | |
| L += [f"| {x['idx']} | {x['cell']} | {x['regime']} | {x['t']} | {x['reason']} | {x['leaf']} |" for x in dev_rows] | |
| L += ["", "HT share of each cell that is 'no cabin video at t' (design weights 1 / pi; a frame-level category of the automatic state, NOT a human label): " | |
| + ", ".join(f"{c} {v:.3f}" for c, v in w_ncv_share.items() if v > 0) + ". The occlusion share of face_invalid is estimated on the rated windows only (cabin video present)."] | |
| if supp_rows: | |
| L += ["", f"### face_invalid|kin_invalid — original cell, supplementary stratum `gaze_supp` and a labelled post-hoc pooled row (deviation 2026-09-13; `share_face_invalid_occluded`)", "", | |
| "| estimate | windows (manifest) | rated | share occluded [CI] | occlusion kinds |", "|---|---|---|---|---|"] | |
| for r in supp_rows: | |
| L.append(f"| {r['estimate']} | {r['n_windows_manifest']} | {r['n_rated']} | {NA.f3(r['share_face_invalid_occluded'])} [{NA.f3(r['lo'], 2)}, {NA.f3(r['hi'], 2)}] | {r['occlusion_kinds']} |") | |
| L += ["", "The supplementary stratum has its own frame (827 VAL rows of the cell whose cabin video covers t) and inclusion probabilities; it is never merged into the pre-registered cell weight. " | |
| "The pooled row is a ratio estimator over the union of the two samples (each unit with its own design weight) and is labelled post hoc wherever it appears."] | |
| L += ["", "Pre-stated expectation: `recall_eyes_only_glances` ≈ 0 — gaze_off is a head-yaw hysteresis on |lm_yaw_dev| (P90 / P75 = 0.212 / 0.103) and cannot see eyes-only glances; " | |
| "the statement in the paper is therefore 'the state detects head-away episodes, not eye-only glances', with the measured recall.", "", | |
| "### Confusion: automatic state x human view_region (HT-weighted sums; counts in parentheses)", "", "| state | " + " | ".join(conf.columns) + " |", "|---|" + "---|" * len(conf.columns)] | |
| for st in conf.index: | |
| L.append(f"| {st} | " + " | ".join(f"{conf.loc[st, c]:.0f} ({int(conf_n.loc[st, c]) if c in conf_n.columns else 0})" for c in conf.columns) + " |") | |
| L.append("") | |
| md = "\n".join(L) + "\n" | |
| if not TABLE.exists(): | |
| TABLE.write_text("# gaze_off human validation (eval/gaze_off_validation_analysis.py; sections appended per run)\n\n") | |
| with open(TABLE, "a") as f: | |
| f.write(md) | |
| print(md) | |
| C.to_csv(out / "windows.csv", index=False) | |
| E.to_csv(out / "metrics.csv", index=False) | |
| if supp_rows: | |
| pd.DataFrame(supp_rows).to_csv(out / "face_invalid_supplement.csv", index=False) | |
| write_run_json(out, "gaze_off_validation_analysis", [Path(__file__), Path(NA.__file__)], [GAZEV / f"manifest_{a.version}.csv"] + ([Path(a.store)] if Path(a.store).exists() else []), seed=0, t_start=t0, | |
| input_listing=False, extra=dict(version=a.version, n_rated=n_lab, kappa=kap, metrics=E.to_dict("records"), dms_fields_used=[])) | |
| print(f"DONE in {time.time() - t0:.0f}s -> {out}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 23.1 kB
- Xet hash:
- 8f6d05b495701ad6da772c349784a37268315a9a76ef423f6e4d07a404a71d17
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.