Buckets:

AsianPlayer
/

TriDrive

Files

xet

AsianPlayer/TriDrive / code /eval /gaze_off_validation_sample.py

AsianPlayer

16 days ago

download

raw

22.2 kB

"""cwm3/eval/gaze_off_validation_sample.py — human validation sample of the causal face-geometry `gaze_off` state (docs/newtest_protocol_draft.md §4;
pre-registered in docs/newtest_prereg.md; 2026-09-12).  VAL split only (thresholds were fitted on train; val is untouched by the warning head; nothing
here can tune anything on test).
State per 5-Hz grid row: results/targets_face_causal_routes_v2/facehist_causal_{regime}_val.npz — F0_c[:, 7] gaze_off {0,1} (head-yaw hysteresis on
|lm_yaw_dev|, on/off thresholds P90 / P75 = 0.212 / 0.103), F0_c[:, 8] gaze_off_dwell_s, F0_c[:, 4] ear_conf_asym (keypoint-confidence proxy), F0valid;
results/targets_kin_causal_routes_v2/kin_causal_{regime}_val.npz — S0_c[:, 1] head_turned {0,1}, K0valid, dongle (rows aligned with the face file, asserted).
Cells (face x head): on|straight, on|turned, off_brief (dwell <= 2 s)|straight, off_brief|turned, off_sust (dwell > 2 s)|straight, off_sust|turned,
face_invalid|kin_invalid, face_invalid|turned (face_invalid|straight and valid_face|kin_invalid are tiny and not sampled; counts recorded).
Allocation 30 / 25 / 25 / 25 / 25 / 25 / 25 / 10 = 190 + 10 regime top-up (to the regime with fewer windows) = 200; strata = (cell, regime, asym half):
inside every cell with a valid face 50 % of the draws go to |ear_conf_asym| > 0.3 and 50 % to the rest (each half a stratum; face_invalid cells have one
half); allocation over regimes proportional to N; device cap 20 (global); 5 s same-route non-overlap guard; pi_i = n_stratum / N_stratum (> 0 for every
frame row), N_cell / n_cell also stored.  4 calibration windows flagged (cells on|straight, off_brief|turned, off_sust|straight, face_invalid|kin_invalid;
2 / 2 regime) — they join the 8 warning-task calibration clips.  Clip = [t-3, t+1] cabin + road (rendered by newtest_render.py --gaze), frame t highlighted.
Server-only fields (never sent to a rater): cell, gaze_off, dwell_s, head_turned, face_valid, kin_valid, ear_conf_asym, asym_half, pi.
Outputs (append-only): results/gaze_validation/manifest_<v>.csv, frame_counts_<v>.json, run json.
  python eval/gaze_off_validation_sample.py [--seed 0] [--version v1]
Supplementary post-hoc stratum (owner decision 2026-09-13, pre-registered in docs/newtest_prereg_deviations.md BEFORE the draw; never merged into the v1 design):
  --supplement face_invalid_covered --n-supp 20 --version supp_v1 : stratum "gaze_supp" = VAL rows of cell face_invalid|kin_invalid whose cabin video covers t —
  the TIGUAN leaf (cabin video ends at 2284.3 s; all its cell rows lie beyond) and any leaf with no dcamera (local or remote listing) are excluded; on leaves with a
  local dcamera the rows must satisfy t <= duration − 0.25 s (verified coverage); rows on leaves without local video are kept with coverage_verified = False and
  are excluded at render if the fetched cabin video does not cover t (logged, π unchanged).  Strata = regime (device cap 20 within the supplement, 5 s guard);
  π = n / N within the stratum; idx base 91000; own manifest / frame_counts / run json; nothing of v1 is touched.
"""
from __future__ import annotations

import argparse
import hashlib
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]
sys.path.insert(0, str(CWM3 / "common"))
sys.path.insert(0, str(CWM3 / "eval"))
from runmeta import write_run_json                          # noqa: E402
import newtest_sample as NS                                 # noqa: E402  (leaf_of, video_local, allocate_capped, largest_remainder, near_any)

REGIMES = ("manual", "engaged")
FACE = CWM3 / "results" / "targets_face_causal_routes_v2"
KIN = CWM3 / "results" / "targets_kin_causal_routes_v2"
OUT = CWM3 / "results" / "gaze_validation"
CH_GAZE, CH_DWELL, CH_ASYM = 7, 8, 4
CH_TURN = 1
DWELL_BRIEF_S = 2.0
ASYM_TH = 0.3
CELLS = ["on|straight", "on|turned", "off_brief|straight", "off_brief|turned", "off_sust|straight", "off_sust|turned", "face_invalid|kin_invalid", "face_invalid|turned"]
ALLOC = {"on|straight": 30, "on|turned": 25, "off_brief|straight": 25, "off_brief|turned": 25, "off_sust|straight": 25, "off_sust|turned": 25, "face_invalid|kin_invalid": 25, "face_invalid|turned": 10}
TOPUP = 10
DEV_CAP = 20
GUARD_S = 5.0
IDX_BASE = 90000
CALIB_CELLS = {"on|straight": "manual", "off_brief|turned": "engaged", "off_sust|straight": "engaged", "face_invalid|kin_invalid": "manual"}
PRE_S, POST_S = 3.0, 1.0


def opaque_id(uid, t):
    return "g" + hashlib.sha1(f"{uid}|{t:.3f}|cwm3-gaze-v1".encode()).hexdigest()[:12]


def load_rows(log):
    frames = []
    files = []
    for r in REGIMES:
        fp, kp = FACE / f"facehist_causal_{r}_val.npz", KIN / f"kin_causal_{r}_val.npz"
        files += [fp, kp]
        f = np.load(fp, allow_pickle=True)
        k = np.load(kp, allow_pickle=True)
        assert np.array_equal(f["uid"], k["uid"]) and np.allclose(f["t"], k["t"]), f"{r}: face / kin val rows differ"
        ch = list(map(str, f["channels"]))
        assert ch[CH_GAZE] == "gaze_off" and ch[CH_DWELL] == "gaze_off_dwell_s" and ch[CH_ASYM] == "ear_conf_asym", ch
        assert list(map(str, k["state_names"]))[CH_TURN] == "head_turned"
        F0, S0 = f["F0_c"].astype(float), k["S0_c"].astype(float)
        fv = f["F0valid"].astype(bool)
        kv = k["K0valid"].astype(bool) & np.isfinite(S0[:, CH_TURN])
        gaze_off = fv & (F0[:, CH_GAZE] > 0.5)
        turned = kv & (S0[:, CH_TURN] > 0.5)
        dwell = F0[:, CH_DWELL]
        cell = np.full(len(F0), "", object)
        cell[fv & ~gaze_off & kv & ~turned] = "on|straight"
        cell[fv & ~gaze_off & kv & turned] = "on|turned"
        cell[gaze_off & (dwell <= DWELL_BRIEF_S) & kv & ~turned] = "off_brief|straight"
        cell[gaze_off & (dwell <= DWELL_BRIEF_S) & kv & turned] = "off_brief|turned"
        cell[gaze_off & (dwell > DWELL_BRIEF_S) & kv & ~turned] = "off_sust|straight"
        cell[gaze_off & (dwell > DWELL_BRIEF_S) & kv & turned] = "off_sust|turned"
        cell[~fv & ~kv] = "face_invalid|kin_invalid"
        cell[~fv & kv & turned] = "face_invalid|turned"
        cell[~fv & kv & ~turned] = "face_invalid|straight"
        cell[fv & ~kv] = "valid_face|kin_invalid"
        asym = np.abs(F0[:, CH_ASYM])
        half = np.where(~fv, "na", np.where(np.isfinite(asym) & (asym > ASYM_TH), "hi", "lo"))
        frames.append(pd.DataFrame(dict(uid=f["uid"], t=f["t"].astype(float), dongle=k["dongle"], regime=r, cell=cell, asym_half=half, gaze_off=gaze_off.astype(int),
                                        dwell_s=np.where(fv, dwell, np.nan), head_turned=np.where(kv, turned.astype(float), np.nan), face_valid=fv, kin_valid=kv,
                                        ear_conf_asym=np.where(fv, F0[:, CH_ASYM], np.nan), yaw_dev=np.where(fv, F0[:, 1], np.nan))))
        log(f"{r}: rows {len(F0)} face valid {int(fv.sum())} kin valid {int(kv.sum())} devices {len(np.unique(k['dongle']))}; cells {dict(Counter(cell))}")
    D = pd.concat(frames, ignore_index=True)
    return D, files


TIGUAN_LEAF = "VOLKSWAGEN_TIGUAN_MK2/driver_25/route_2/route_0-87"
SUPP_IDX_BASE = 91000
SUPP_DEV_CAP = 20


def _probe_dur(p):
    import subprocess
    r = subprocess.run(["/usr/bin/ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", str(p)], capture_output=True, text=True)
    try:
        return float(r.stdout.strip())
    except ValueError:
        return float("nan")


def main_supplement(a, log, t_start):
    """post-hoc supplementary stratum gaze_supp (see docstring); frozen draw mechanics (stratified SRS, device cap, guard) reused."""
    assert a.supplement == "face_invalid_covered", a.supplement
    man_p, fc_p = OUT / f"manifest_{a.version}.csv", OUT / f"frame_counts_{a.version}.json"
    for p in (man_p, fc_p):
        if p.exists():
            raise SystemExit(f"append-only: {p} exists — choose a new --version")
    rng = np.random.default_rng(a.seed)
    D, files = load_rows(log)
    C = D[D.cell == "face_invalid|kin_invalid"].copy()
    C["leaf"] = C.uid.map(NS.leaf_of)
    leaves = json.load(open(CWM3 / "results" / "baton_leaves.json"))["leaves"]
    dur = {}
    for leaf in C.leaf.unique():
        p = NS.BATON_LOCAL / leaf / "dcamera.mp4"
        if p.exists():
            dur[leaf] = _probe_dur(p)
    C["local_dcam"] = C.leaf.map(lambda l: l in dur)
    C["remote_dcam"] = C.leaf.map(lambda l: (leaves.get(l, {}).get("dcamera_bytes") or 0) > 0)
    C["coverage_verified"] = C.local_dcam
    C["covered"] = [(t <= dur[l] - 0.25) if l in dur else True for l, t in zip(C.leaf, C.t)]        # unverified leaves: kept, checked at render
    excl_tig = C.leaf == TIGUAN_LEAF
    excl_nodcam = ~(C.local_dcam | C.remote_dcam)
    excl_uncov = C.local_dcam & ~C.covered
    frame = C[~excl_tig & ~excl_nodcam & ~excl_uncov].copy()
    log(f"supplement frame: cell rows {len(C)} -> minus TIGUAN {int(excl_tig.sum())}, minus no-dcamera leaves {int((excl_nodcam & ~excl_tig).sum())}, minus verified-not-covered {int((excl_uncov & ~excl_tig).sum())} "
        f"= N {len(frame)} ({dict(Counter(frame.regime))}; verified coverage {int(frame.coverage_verified.sum())}, unverified {int((~frame.coverage_verified).sum())}; leaves {frame.leaf.nunique()}, devices {frame.dongle.nunique()})")
    if a.expect_n is not None:
        assert len(frame) == a.expect_n, f"frame size {len(frame)} != pre-registered {a.expect_n}"
    # strata = regime, proportional allocation, device cap within the supplement, 5 s guard
    N_r = {r: int((frame.regime == r).sum()) for r in REGIMES}
    n_r = dict(zip(REGIMES, NS.largest_remainder(a.n_supp, [N_r[r] for r in REGIMES])))
    dev_used = Counter()
    drawn, taken = [], defaultdict(list)
    for r in sorted(REGIMES, key=lambda x: N_r[x]):
        pool = frame[frame.regime == r].sort_values(["uid", "t"])
        got = 0
        for j in rng.permutation(len(pool)):
            if got >= n_r[r]:
                break
            row = pool.iloc[j]
            if dev_used[row.dongle] >= SUPP_DEV_CAP or any(abs(row.t - x) < GUARD_S for x in taken[row.uid]):
                continue
            dev_used[row.dongle] += 1
            taken[row.uid].append(float(row.t))
            got += 1
            drawn.append(dict(row.to_dict(), stratum=f"gaze_supp|{r}", N_stratum=N_r[r]))
        for u in drawn:
            if u["stratum"] == f"gaze_supp|{r}":
                u["n_stratum"], u["pi"] = got, got / N_r[r]
        log(f"  gaze_supp {r}: N {N_r[r]} -> n {got} (target {n_r[r]})")
    drawn.sort(key=lambda u: (u["regime"], u["uid"], u["t"]))
    for rank, j in enumerate(rng.permutation(len(drawn))):
        drawn[j]["idx"] = SUPP_IDX_BASE + rank
    for q, u in enumerate(drawn):
        u.update(clip_id=opaque_id(u["uid"], u["t"]), dcam=str(NS.BATON_LOCAL / u["leaf"] / "dcamera.mp4"), qcam=str(NS.BATON_LOCAL / u["leaf"] / "qcamera.mp4"), video_local=NS.video_local(u["uid"]),
                 calibration=False, qpos=q, t_start=round(max(0.0, u["t"] - PRE_S), 3), t_end=round(u["t"] + POST_S, 3), anchor_s=round(min(PRE_S, u["t"]), 3), N_cell=len(frame), n_cell=len(drawn),
                 stratum_group="gaze_supp", design="post_hoc_supplement_2026-09-13")
    cols = ["qpos", "idx", "clip_id", "calibration", "regime", "uid", "dongle", "leaf", "t", "t_start", "t_end", "anchor_s", "cell", "stratum", "stratum_group", "design", "asym_half", "pi", "N_stratum", "n_stratum", "N_cell", "n_cell",
            "gaze_off", "dwell_s", "head_turned", "face_valid", "kin_valid", "ear_conf_asym", "yaw_dev", "coverage_verified", "video_local", "dcam", "qcam"]
    M = pd.DataFrame([{k: u.get(k) for k in cols} for u in drawn], columns=cols)
    assert M.idx.is_unique and M.clip_id.is_unique and ((M.pi > 0) & (M.pi <= 1)).all()
    M.to_csv(man_p, index=False)
    fc = dict(version=a.version, design="post_hoc_supplement_2026-09-13", stratum="gaze_supp", seed=a.seed, generated_utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
              frame=dict(cell_rows=int(len(C)), excluded_tiguan=int(excl_tig.sum()), excluded_no_dcamera=int((excl_nodcam & ~excl_tig).sum()), excluded_verified_not_covered=int((excl_uncov & ~excl_tig).sum()), N=int(len(frame)),
                         N_by_regime=N_r, N_verified=int(frame.coverage_verified.sum()), N_unverified=int((~frame.coverage_verified).sum()), leaves=int(frame.leaf.nunique()), devices=int(frame.dongle.nunique()),
                         leaves_without_local_video=sorted(frame[~frame.coverage_verified].leaf.unique().tolist())),
              drawn=len(M), drawn_by_regime=dict(Counter(M.regime)), devices_drawn=int(M.dongle.nunique()), top_device=int(max(dev_used.values())) if dev_used else 0, video_local=int(M.video_local.sum()),
              coverage_verified_drawn=int(M.coverage_verified.sum()), dev_cap=SUPP_DEV_CAP, guard_s=GUARD_S, clip_window_s=[PRE_S, POST_S], log=[])
    json.dump(fc, open(fc_p, "w"), indent=1, default=lambda o: int(o) if isinstance(o, np.integer) else float(o) if isinstance(o, np.floating) else str(o))
    sha = {p.name: NS.sha256_file(p) for p in (man_p, fc_p)}
    (OUT / f"sha256_{a.version}.txt").write_text("".join(f"{v}  {k}\n" for k, v in sha.items()))
    write_run_json(OUT, "gaze_off_validation_sample_supp", [Path(__file__), Path(NS.__file__)], files + [CWM3 / "results" / "baton_leaves.json"], seed=a.seed, t_start=t_start,
                   extra=dict(version=a.version, supplement=a.supplement, outputs=[str(man_p), str(fc_p)], sha256=sha, n=len(M), frame=fc["frame"], dms_fields_used=[]))
    log(f"DONE in {time.time() - t_start:.0f}s: {len(M)} supplementary windows -> {man_p}; sha256 {sha}")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--seed", type=int, default=0)
    ap.add_argument("--version", default="v1")
    ap.add_argument("--supplement", default="", help="post-hoc supplementary stratum (face_invalid_covered); writes manifest_<version>.csv separately from v1")
    ap.add_argument("--n-supp", type=int, default=20)
    ap.add_argument("--expect-n", type=int, default=None, help="pre-registered frame size; the draw aborts if the computed frame differs")
    a = ap.parse_args()
    t_start = time.time()
    if a.supplement:
        OUT.mkdir(parents=True, exist_ok=True)
        lines = []

        def log(m):
            line = f"[{time.strftime('%H:%M:%S')}] {m}"
            print(line, flush=True)
            lines.append(line)
        return main_supplement(a, log, t_start)
    OUT.mkdir(parents=True, exist_ok=True)
    man_p, fc_p = OUT / f"manifest_{a.version}.csv", OUT / f"frame_counts_{a.version}.json"
    for p in (man_p, fc_p):
        if p.exists():
            raise SystemExit(f"append-only: {p} exists — choose a new --version")
    lines = []

    def log(m):
        line = f"[{time.strftime('%H:%M:%S')}] {m}"
        print(line, flush=True)
        lines.append(line)

    rng = np.random.default_rng(a.seed)
    D, files = load_rows(log)
    cell_counts = D.groupby(["cell", "regime"]).size().unstack(fill_value=0)
    log("cell x regime counts:\n" + cell_counts.to_string())
    # ---- strata = (cell, regime, asym half); allocation: cell -> regimes proportional to N -> halves 50/50 (valid-face cells); top-up to the minority regime
    alloc = {}
    for c in CELLS:
        sub = D[D.cell == c]
        N_r = {r: int((sub.regime == r).sum()) for r in REGIMES}
        n_r = dict(zip(REGIMES, NS.largest_remainder(ALLOC[c], [N_r[r] for r in REGIMES])))
        for r in REGIMES:
            halves = ["na"] if c.startswith("face_invalid") else ["hi", "lo"]
            N_h = {h: int(((sub.regime == r) & (sub.asym_half == h)).sum()) for h in halves}
            if len(halves) == 1:
                n_h = {"na": min(n_r[r], N_h["na"])}
            else:
                base = n_r[r] // 2
                n_h = {"hi": min(base + n_r[r] % 2, N_h["hi"]), "lo": min(base, N_h["lo"])}
                short = n_r[r] - sum(n_h.values())                  # a half too small for its share -> the other half takes the rest
                for h in halves:
                    add = min(short, N_h[h] - n_h[h])
                    n_h[h] += add
                    short -= add
            for h in halves:
                alloc[(c, r, h)] = dict(N=N_h[h], n=n_h[h])
    per_reg = {r: sum(v["n"] for (c, rr, h), v in alloc.items() if rr == r) for r in REGIMES}
    minority = min(per_reg, key=per_reg.get)
    keys_min = [k for k in alloc if k[1] == minority and alloc[k]["N"] > alloc[k]["n"]]
    for k, add in zip(keys_min, NS.largest_remainder(TOPUP, [alloc[k]["N"] for k in keys_min])):
        alloc[k]["n"] += min(add, alloc[k]["N"] - alloc[k]["n"])
    log(f"regime totals before top-up {per_reg}; +{TOPUP} to {minority}")
    # ---- draw: SRS within stratum with the global device cap and the 5 s same-route guard
    dev_used = Counter()
    drawn = []
    taken = defaultdict(list)
    # face-invalid cells first (they are concentrated on very few devices — one engaged device holds 13.9k of the 14.0k face_invalid|kin_invalid rows —
    # and would otherwise find their devices already at the cap), then small strata first so the cap binds on the big valid-face cells
    for key in sorted(alloc, key=lambda k: (not k[0].startswith("face_invalid"), alloc[k]["N"], k)):
        c, r, h = key
        n = alloc[key]["n"]
        pool = D[(D.cell == c) & (D.regime == r) & (D.asym_half == h)].sort_values(["uid", "t"])
        order = rng.permutation(len(pool))
        got = 0
        for j in order:
            if got >= n:
                break
            row = pool.iloc[j]
            if dev_used[row.dongle] >= DEV_CAP or any(abs(row.t - x) < GUARD_S for x in taken[row.uid]):
                continue
            dev_used[row.dongle] += 1
            taken[row.uid].append(float(row.t))
            got += 1
            drawn.append(dict(row.to_dict(), N_stratum=int(alloc[key]["N"]), stratum=f"{c}|{r}|{h}"))
        alloc[key]["got"] = got
        for u in drawn:
            if u["stratum"] == f"{c}|{r}|{h}":
                u["n_stratum"] = got
                u["pi"] = got / alloc[key]["N"]
    log(f"drawn {len(drawn)} windows on {len({u['dongle'] for u in drawn})} devices; per regime {Counter(u['regime'] for u in drawn)}; per cell {Counter(u['cell'] for u in drawn)}; "
        f"top device {max(dev_used.values())}")
    N_cell = D.cell.value_counts().to_dict()
    n_cell = Counter(u["cell"] for u in drawn)
    for u in drawn:
        u["N_cell"], u["n_cell"] = int(N_cell[u["cell"]]), int(n_cell[u["cell"]])
    # ---- ids, calibration, queue
    drawn.sort(key=lambda u: (u["cell"], u["regime"], u["uid"], u["t"]))
    perm = rng.permutation(len(drawn))
    for rank, j in enumerate(perm):
        drawn[j]["idx"] = IDX_BASE + rank
    for u in drawn:
        u["clip_id"] = opaque_id(u["uid"], u["t"])
        u["leaf"] = NS.leaf_of(u["uid"])
        u["dcam"], u["qcam"] = str(NS.BATON_LOCAL / u["leaf"] / "dcamera.mp4"), str(NS.BATON_LOCAL / u["leaf"] / "qcamera.mp4")
        u["video_local"] = NS.video_local(u["uid"])
        u["calibration"] = False
        u["t_start"], u["t_end"], u["anchor_s"] = round(max(0.0, u["t"] - PRE_S), 3), round(u["t"] + POST_S, 3), round(min(PRE_S, u["t"]), 3)
    for c, r in CALIB_CELLS.items():
        cand = [u for u in drawn if u["cell"] == c and u["regime"] == r]
        if not cand:
            cand = [u for u in drawn if u["cell"] == c]
        cand[int(rng.integers(len(cand)))]["calibration"] = True
    calib = [u for u in drawn if u["calibration"]]
    rest = [u for u in drawn if not u["calibration"]]
    queue = [calib[j] for j in rng.permutation(len(calib))] + [rest[j] for j in rng.permutation(len(rest))]
    for q, u in enumerate(queue):
        u["qpos"] = q
    cols = ["qpos", "idx", "clip_id", "calibration", "regime", "uid", "dongle", "leaf", "t", "t_start", "t_end", "anchor_s", "cell", "stratum", "asym_half", "pi", "N_stratum", "n_stratum", "N_cell", "n_cell",
            "gaze_off", "dwell_s", "head_turned", "face_valid", "kin_valid", "ear_conf_asym", "yaw_dev", "video_local", "dcam", "qcam"]
    M = pd.DataFrame([{k: u.get(k) for k in cols} for u in queue], columns=cols)
    assert M.idx.is_unique and M.clip_id.is_unique and ((M.pi > 0) & (M.pi <= 1)).all()
    M.to_csv(man_p, index=False)
    fc = dict(version=a.version, seed=a.seed, generated_utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), n_rows_val={r: int((D.regime == r).sum()) for r in REGIMES},
              n_devices_val=int(D.dongle.nunique()), cells_total=N_cell, cells_by_regime={c: {r: int(cell_counts.loc[c, r]) if c in cell_counts.index else 0 for r in REGIMES} for c in cell_counts.index},
              alloc=ALLOC, topup=TOPUP, topup_regime=minority, dev_cap=DEV_CAP, guard_s=GUARD_S, asym_threshold=ASYM_TH, dwell_brief_s=DWELL_BRIEF_S,
              strata={"|".join(k): v for k, v in alloc.items()}, drawn=len(M), drawn_by_cell=dict(n_cell), drawn_by_regime=dict(Counter(M.regime)), devices_drawn=int(M.dongle.nunique()),
              top_device=int(max(dev_used.values())), video_local=int(M.video_local.sum()), routes=int(M.uid.nunique()), routes_local=int(M[M.video_local].uid.nunique()),
              calibration=[dict(idx=u["idx"], cell=u["cell"], regime=u["regime"]) for u in calib],
              asym_share_among_off={r: float((np.abs(D[(D.regime == r) & (D.gaze_off == 1)].ear_conf_asym) > ASYM_TH).mean()) for r in REGIMES},
              device_concentration={c: dict(Counter(D[D.cell == c].dongle).most_common(3)) for c in CELLS},
              clip_window_s=[PRE_S, POST_S], log=lines)
    json.dump(fc, open(fc_p, "w"), indent=1, default=lambda o: int(o) if isinstance(o, np.integer) else float(o) if isinstance(o, np.floating) else str(o))
    write_run_json(OUT, "gaze_off_validation_sample", [Path(__file__), Path(NS.__file__)], files, seed=a.seed, t_start=t_start,
                   extra=dict(version=a.version, outputs=[str(man_p), str(fc_p)], n=len(M), by_cell=dict(n_cell), dms_fields_used=[]))
    log(f"DONE in {time.time() - t_start:.0f}s: {len(M)} windows -> {man_p}")


if __name__ == "__main__":
    main()

Xet Storage Details

Size:: 22.2 kB
Xet hash:: 5c0e529e301f292da0754c411fd6719d1e77524928b4f51915ef0868fbfc484a

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""cwm3/eval/gaze_off_validation_sample.py — human validation sample of the causal face-geometry `gaze_off` state (docs/newtest_protocol_draft.md §4;
	pre-registered in docs/newtest_prereg.md; 2026-09-12). VAL split only (thresholds were fitted on train; val is untouched by the warning head; nothing
	here can tune anything on test).

	State per 5-Hz grid row: results/targets_face_causal_routes_v2/facehist_causal_{regime}_val.npz — F0_c[:, 7] gaze_off {0,1} (head-yaw hysteresis on
	\|lm_yaw_dev\|, on/off thresholds P90 / P75 = 0.212 / 0.103), F0_c[:, 8] gaze_off_dwell_s, F0_c[:, 4] ear_conf_asym (keypoint-confidence proxy), F0valid;
	results/targets_kin_causal_routes_v2/kin_causal_{regime}_val.npz — S0_c[:, 1] head_turned {0,1}, K0valid, dongle (rows aligned with the face file, asserted).
	Cells (face x head): on\|straight, on\|turned, off_brief (dwell <= 2 s)\|straight, off_brief\|turned, off_sust (dwell > 2 s)\|straight, off_sust\|turned,
	face_invalid\|kin_invalid, face_invalid\|turned (face_invalid\|straight and valid_face\|kin_invalid are tiny and not sampled; counts recorded).
	Allocation 30 / 25 / 25 / 25 / 25 / 25 / 25 / 10 = 190 + 10 regime top-up (to the regime with fewer windows) = 200; strata = (cell, regime, asym half):
	inside every cell with a valid face 50 % of the draws go to \|ear_conf_asym\| > 0.3 and 50 % to the rest (each half a stratum; face_invalid cells have one
	half); allocation over regimes proportional to N; device cap 20 (global); 5 s same-route non-overlap guard; pi_i = n_stratum / N_stratum (> 0 for every
	frame row), N_cell / n_cell also stored. 4 calibration windows flagged (cells on\|straight, off_brief\|turned, off_sust\|straight, face_invalid\|kin_invalid;
	2 / 2 regime) — they join the 8 warning-task calibration clips. Clip = [t-3, t+1] cabin + road (rendered by newtest_render.py --gaze), frame t highlighted.
	Server-only fields (never sent to a rater): cell, gaze_off, dwell_s, head_turned, face_valid, kin_valid, ear_conf_asym, asym_half, pi.
	Outputs (append-only): results/gaze_validation/manifest_<v>.csv, frame_counts_<v>.json, run json.
	python eval/gaze_off_validation_sample.py [--seed 0] [--version v1]
	Supplementary post-hoc stratum (owner decision 2026-09-13, pre-registered in docs/newtest_prereg_deviations.md BEFORE the draw; never merged into the v1 design):
	--supplement face_invalid_covered --n-supp 20 --version supp_v1 : stratum "gaze_supp" = VAL rows of cell face_invalid\|kin_invalid whose cabin video covers t —
	the TIGUAN leaf (cabin video ends at 2284.3 s; all its cell rows lie beyond) and any leaf with no dcamera (local or remote listing) are excluded; on leaves with a
	local dcamera the rows must satisfy t <= duration − 0.25 s (verified coverage); rows on leaves without local video are kept with coverage_verified = False and
	are excluded at render if the fetched cabin video does not cover t (logged, π unchanged). Strata = regime (device cap 20 within the supplement, 5 s guard);
	π = n / N within the stratum; idx base 91000; own manifest / frame_counts / run json; nothing of v1 is touched.
	"""
	from __future__ import annotations

	import argparse
	import hashlib
	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]
	sys.path.insert(0, str(CWM3 / "common"))
	sys.path.insert(0, str(CWM3 / "eval"))
	from runmeta import write_run_json # noqa: E402
	import newtest_sample as NS # noqa: E402 (leaf_of, video_local, allocate_capped, largest_remainder, near_any)

	REGIMES = ("manual", "engaged")
	FACE = CWM3 / "results" / "targets_face_causal_routes_v2"
	KIN = CWM3 / "results" / "targets_kin_causal_routes_v2"
	OUT = CWM3 / "results" / "gaze_validation"
	CH_GAZE, CH_DWELL, CH_ASYM = 7, 8, 4
	CH_TURN = 1
	DWELL_BRIEF_S = 2.0
	ASYM_TH = 0.3
	CELLS = ["on\|straight", "on\|turned", "off_brief\|straight", "off_brief\|turned", "off_sust\|straight", "off_sust\|turned", "face_invalid\|kin_invalid", "face_invalid\|turned"]
	ALLOC = {"on\|straight": 30, "on\|turned": 25, "off_brief\|straight": 25, "off_brief\|turned": 25, "off_sust\|straight": 25, "off_sust\|turned": 25, "face_invalid\|kin_invalid": 25, "face_invalid\|turned": 10}
	TOPUP = 10
	DEV_CAP = 20
	GUARD_S = 5.0
	IDX_BASE = 90000
	CALIB_CELLS = {"on\|straight": "manual", "off_brief\|turned": "engaged", "off_sust\|straight": "engaged", "face_invalid\|kin_invalid": "manual"}
	PRE_S, POST_S = 3.0, 1.0


	def opaque_id(uid, t):
	return "g" + hashlib.sha1(f"{uid}\|{t:.3f}\|cwm3-gaze-v1".encode()).hexdigest()[:12]


	def load_rows(log):
	frames = []
	files = []
	for r in REGIMES:
	fp, kp = FACE / f"facehist_causal_{r}_val.npz", KIN / f"kin_causal_{r}_val.npz"
	files += [fp, kp]
	f = np.load(fp, allow_pickle=True)
	k = np.load(kp, allow_pickle=True)
	assert np.array_equal(f["uid"], k["uid"]) and np.allclose(f["t"], k["t"]), f"{r}: face / kin val rows differ"
	ch = list(map(str, f["channels"]))
	assert ch[CH_GAZE] == "gaze_off" and ch[CH_DWELL] == "gaze_off_dwell_s" and ch[CH_ASYM] == "ear_conf_asym", ch
	assert list(map(str, k["state_names"]))[CH_TURN] == "head_turned"
	F0, S0 = f["F0_c"].astype(float), k["S0_c"].astype(float)
	fv = f["F0valid"].astype(bool)
	kv = k["K0valid"].astype(bool) & np.isfinite(S0[:, CH_TURN])
	gaze_off = fv & (F0[:, CH_GAZE] > 0.5)
	turned = kv & (S0[:, CH_TURN] > 0.5)
	dwell = F0[:, CH_DWELL]
	cell = np.full(len(F0), "", object)
	cell[fv & ~gaze_off & kv & ~turned] = "on\|straight"
	cell[fv & ~gaze_off & kv & turned] = "on\|turned"
	cell[gaze_off & (dwell <= DWELL_BRIEF_S) & kv & ~turned] = "off_brief\|straight"
	cell[gaze_off & (dwell <= DWELL_BRIEF_S) & kv & turned] = "off_brief\|turned"
	cell[gaze_off & (dwell > DWELL_BRIEF_S) & kv & ~turned] = "off_sust\|straight"
	cell[gaze_off & (dwell > DWELL_BRIEF_S) & kv & turned] = "off_sust\|turned"
	cell[~fv & ~kv] = "face_invalid\|kin_invalid"
	cell[~fv & kv & turned] = "face_invalid\|turned"
	cell[~fv & kv & ~turned] = "face_invalid\|straight"
	cell[fv & ~kv] = "valid_face\|kin_invalid"
	asym = np.abs(F0[:, CH_ASYM])
	half = np.where(~fv, "na", np.where(np.isfinite(asym) & (asym > ASYM_TH), "hi", "lo"))
	frames.append(pd.DataFrame(dict(uid=f["uid"], t=f["t"].astype(float), dongle=k["dongle"], regime=r, cell=cell, asym_half=half, gaze_off=gaze_off.astype(int),
	dwell_s=np.where(fv, dwell, np.nan), head_turned=np.where(kv, turned.astype(float), np.nan), face_valid=fv, kin_valid=kv,
	ear_conf_asym=np.where(fv, F0[:, CH_ASYM], np.nan), yaw_dev=np.where(fv, F0[:, 1], np.nan))))
	log(f"{r}: rows {len(F0)} face valid {int(fv.sum())} kin valid {int(kv.sum())} devices {len(np.unique(k['dongle']))}; cells {dict(Counter(cell))}")
	D = pd.concat(frames, ignore_index=True)
	return D, files


	TIGUAN_LEAF = "VOLKSWAGEN_TIGUAN_MK2/driver_25/route_2/route_0-87"
	SUPP_IDX_BASE = 91000
	SUPP_DEV_CAP = 20


	def _probe_dur(p):
	import subprocess
	r = subprocess.run(["/usr/bin/ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "csv=p=0", str(p)], capture_output=True, text=True)
	try:
	return float(r.stdout.strip())
	except ValueError:
	return float("nan")


	def main_supplement(a, log, t_start):
	"""post-hoc supplementary stratum gaze_supp (see docstring); frozen draw mechanics (stratified SRS, device cap, guard) reused."""
	assert a.supplement == "face_invalid_covered", a.supplement
	man_p, fc_p = OUT / f"manifest_{a.version}.csv", OUT / f"frame_counts_{a.version}.json"
	for p in (man_p, fc_p):
	if p.exists():
	raise SystemExit(f"append-only: {p} exists — choose a new --version")
	rng = np.random.default_rng(a.seed)
	D, files = load_rows(log)
	C = D[D.cell == "face_invalid\|kin_invalid"].copy()
	C["leaf"] = C.uid.map(NS.leaf_of)
	leaves = json.load(open(CWM3 / "results" / "baton_leaves.json"))["leaves"]
	dur = {}
	for leaf in C.leaf.unique():
	p = NS.BATON_LOCAL / leaf / "dcamera.mp4"
	if p.exists():
	dur[leaf] = _probe_dur(p)
	C["local_dcam"] = C.leaf.map(lambda l: l in dur)
	C["remote_dcam"] = C.leaf.map(lambda l: (leaves.get(l, {}).get("dcamera_bytes") or 0) > 0)
	C["coverage_verified"] = C.local_dcam
	C["covered"] = [(t <= dur[l] - 0.25) if l in dur else True for l, t in zip(C.leaf, C.t)] # unverified leaves: kept, checked at render
	excl_tig = C.leaf == TIGUAN_LEAF
	excl_nodcam = ~(C.local_dcam \| C.remote_dcam)
	excl_uncov = C.local_dcam & ~C.covered
	frame = C[~excl_tig & ~excl_nodcam & ~excl_uncov].copy()
	log(f"supplement frame: cell rows {len(C)} -> minus TIGUAN {int(excl_tig.sum())}, minus no-dcamera leaves {int((excl_nodcam & ~excl_tig).sum())}, minus verified-not-covered {int((excl_uncov & ~excl_tig).sum())} "
	f"= N {len(frame)} ({dict(Counter(frame.regime))}; verified coverage {int(frame.coverage_verified.sum())}, unverified {int((~frame.coverage_verified).sum())}; leaves {frame.leaf.nunique()}, devices {frame.dongle.nunique()})")
	if a.expect_n is not None:
	assert len(frame) == a.expect_n, f"frame size {len(frame)} != pre-registered {a.expect_n}"
	# strata = regime, proportional allocation, device cap within the supplement, 5 s guard
	N_r = {r: int((frame.regime == r).sum()) for r in REGIMES}
	n_r = dict(zip(REGIMES, NS.largest_remainder(a.n_supp, [N_r[r] for r in REGIMES])))
	dev_used = Counter()
	drawn, taken = [], defaultdict(list)
	for r in sorted(REGIMES, key=lambda x: N_r[x]):
	pool = frame[frame.regime == r].sort_values(["uid", "t"])
	got = 0
	for j in rng.permutation(len(pool)):
	if got >= n_r[r]:
	break
	row = pool.iloc[j]
	if dev_used[row.dongle] >= SUPP_DEV_CAP or any(abs(row.t - x) < GUARD_S for x in taken[row.uid]):
	continue
	dev_used[row.dongle] += 1
	taken[row.uid].append(float(row.t))
	got += 1
	drawn.append(dict(row.to_dict(), stratum=f"gaze_supp\|{r}", N_stratum=N_r[r]))
	for u in drawn:
	if u["stratum"] == f"gaze_supp\|{r}":
	u["n_stratum"], u["pi"] = got, got / N_r[r]
	log(f" gaze_supp {r}: N {N_r[r]} -> n {got} (target {n_r[r]})")
	drawn.sort(key=lambda u: (u["regime"], u["uid"], u["t"]))
	for rank, j in enumerate(rng.permutation(len(drawn))):
	drawn[j]["idx"] = SUPP_IDX_BASE + rank
	for q, u in enumerate(drawn):
	u.update(clip_id=opaque_id(u["uid"], u["t"]), dcam=str(NS.BATON_LOCAL / u["leaf"] / "dcamera.mp4"), qcam=str(NS.BATON_LOCAL / u["leaf"] / "qcamera.mp4"), video_local=NS.video_local(u["uid"]),
	calibration=False, qpos=q, t_start=round(max(0.0, u["t"] - PRE_S), 3), t_end=round(u["t"] + POST_S, 3), anchor_s=round(min(PRE_S, u["t"]), 3), N_cell=len(frame), n_cell=len(drawn),
	stratum_group="gaze_supp", design="post_hoc_supplement_2026-09-13")
	cols = ["qpos", "idx", "clip_id", "calibration", "regime", "uid", "dongle", "leaf", "t", "t_start", "t_end", "anchor_s", "cell", "stratum", "stratum_group", "design", "asym_half", "pi", "N_stratum", "n_stratum", "N_cell", "n_cell",
	"gaze_off", "dwell_s", "head_turned", "face_valid", "kin_valid", "ear_conf_asym", "yaw_dev", "coverage_verified", "video_local", "dcam", "qcam"]
	M = pd.DataFrame([{k: u.get(k) for k in cols} for u in drawn], columns=cols)
	assert M.idx.is_unique and M.clip_id.is_unique and ((M.pi > 0) & (M.pi <= 1)).all()
	M.to_csv(man_p, index=False)
	fc = dict(version=a.version, design="post_hoc_supplement_2026-09-13", stratum="gaze_supp", seed=a.seed, generated_utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
	frame=dict(cell_rows=int(len(C)), excluded_tiguan=int(excl_tig.sum()), excluded_no_dcamera=int((excl_nodcam & ~excl_tig).sum()), excluded_verified_not_covered=int((excl_uncov & ~excl_tig).sum()), N=int(len(frame)),
	N_by_regime=N_r, N_verified=int(frame.coverage_verified.sum()), N_unverified=int((~frame.coverage_verified).sum()), leaves=int(frame.leaf.nunique()), devices=int(frame.dongle.nunique()),
	leaves_without_local_video=sorted(frame[~frame.coverage_verified].leaf.unique().tolist())),
	drawn=len(M), drawn_by_regime=dict(Counter(M.regime)), devices_drawn=int(M.dongle.nunique()), top_device=int(max(dev_used.values())) if dev_used else 0, video_local=int(M.video_local.sum()),
	coverage_verified_drawn=int(M.coverage_verified.sum()), dev_cap=SUPP_DEV_CAP, guard_s=GUARD_S, clip_window_s=[PRE_S, POST_S], log=[])
	json.dump(fc, open(fc_p, "w"), indent=1, default=lambda o: int(o) if isinstance(o, np.integer) else float(o) if isinstance(o, np.floating) else str(o))
	sha = {p.name: NS.sha256_file(p) for p in (man_p, fc_p)}
	(OUT / f"sha256_{a.version}.txt").write_text("".join(f"{v} {k}\n" for k, v in sha.items()))
	write_run_json(OUT, "gaze_off_validation_sample_supp", [Path(__file__), Path(NS.__file__)], files + [CWM3 / "results" / "baton_leaves.json"], seed=a.seed, t_start=t_start,
	extra=dict(version=a.version, supplement=a.supplement, outputs=[str(man_p), str(fc_p)], sha256=sha, n=len(M), frame=fc["frame"], dms_fields_used=[]))
	log(f"DONE in {time.time() - t_start:.0f}s: {len(M)} supplementary windows -> {man_p}; sha256 {sha}")


	def main():
	ap = argparse.ArgumentParser()
	ap.add_argument("--seed", type=int, default=0)
	ap.add_argument("--version", default="v1")
	ap.add_argument("--supplement", default="", help="post-hoc supplementary stratum (face_invalid_covered); writes manifest_<version>.csv separately from v1")
	ap.add_argument("--n-supp", type=int, default=20)
	ap.add_argument("--expect-n", type=int, default=None, help="pre-registered frame size; the draw aborts if the computed frame differs")
	a = ap.parse_args()
	t_start = time.time()
	if a.supplement:
	OUT.mkdir(parents=True, exist_ok=True)
	lines = []

	def log(m):
	line = f"[{time.strftime('%H:%M:%S')}] {m}"
	print(line, flush=True)
	lines.append(line)
	return main_supplement(a, log, t_start)
	OUT.mkdir(parents=True, exist_ok=True)
	man_p, fc_p = OUT / f"manifest_{a.version}.csv", OUT / f"frame_counts_{a.version}.json"
	for p in (man_p, fc_p):
	if p.exists():
	raise SystemExit(f"append-only: {p} exists — choose a new --version")
	lines = []

	def log(m):
	line = f"[{time.strftime('%H:%M:%S')}] {m}"
	print(line, flush=True)
	lines.append(line)

	rng = np.random.default_rng(a.seed)
	D, files = load_rows(log)
	cell_counts = D.groupby(["cell", "regime"]).size().unstack(fill_value=0)
	log("cell x regime counts:\n" + cell_counts.to_string())
	# ---- strata = (cell, regime, asym half); allocation: cell -> regimes proportional to N -> halves 50/50 (valid-face cells); top-up to the minority regime
	alloc = {}
	for c in CELLS:
	sub = D[D.cell == c]
	N_r = {r: int((sub.regime == r).sum()) for r in REGIMES}
	n_r = dict(zip(REGIMES, NS.largest_remainder(ALLOC[c], [N_r[r] for r in REGIMES])))
	for r in REGIMES:
	halves = ["na"] if c.startswith("face_invalid") else ["hi", "lo"]
	N_h = {h: int(((sub.regime == r) & (sub.asym_half == h)).sum()) for h in halves}
	if len(halves) == 1:
	n_h = {"na": min(n_r[r], N_h["na"])}
	else:
	base = n_r[r] // 2
	n_h = {"hi": min(base + n_r[r] % 2, N_h["hi"]), "lo": min(base, N_h["lo"])}
	short = n_r[r] - sum(n_h.values()) # a half too small for its share -> the other half takes the rest
	for h in halves:
	add = min(short, N_h[h] - n_h[h])
	n_h[h] += add
	short -= add
	for h in halves:
	alloc[(c, r, h)] = dict(N=N_h[h], n=n_h[h])
	per_reg = {r: sum(v["n"] for (c, rr, h), v in alloc.items() if rr == r) for r in REGIMES}
	minority = min(per_reg, key=per_reg.get)
	keys_min = [k for k in alloc if k[1] == minority and alloc[k]["N"] > alloc[k]["n"]]
	for k, add in zip(keys_min, NS.largest_remainder(TOPUP, [alloc[k]["N"] for k in keys_min])):
	alloc[k]["n"] += min(add, alloc[k]["N"] - alloc[k]["n"])
	log(f"regime totals before top-up {per_reg}; +{TOPUP} to {minority}")
	# ---- draw: SRS within stratum with the global device cap and the 5 s same-route guard
	dev_used = Counter()
	drawn = []
	taken = defaultdict(list)
	# face-invalid cells first (they are concentrated on very few devices — one engaged device holds 13.9k of the 14.0k face_invalid\|kin_invalid rows —
	# and would otherwise find their devices already at the cap), then small strata first so the cap binds on the big valid-face cells
	for key in sorted(alloc, key=lambda k: (not k[0].startswith("face_invalid"), alloc[k]["N"], k)):
	c, r, h = key
	n = alloc[key]["n"]
	pool = D[(D.cell == c) & (D.regime == r) & (D.asym_half == h)].sort_values(["uid", "t"])
	order = rng.permutation(len(pool))
	got = 0
	for j in order:
	if got >= n:
	break
	row = pool.iloc[j]
	if dev_used[row.dongle] >= DEV_CAP or any(abs(row.t - x) < GUARD_S for x in taken[row.uid]):
	continue
	dev_used[row.dongle] += 1
	taken[row.uid].append(float(row.t))
	got += 1
	drawn.append(dict(row.to_dict(), N_stratum=int(alloc[key]["N"]), stratum=f"{c}\|{r}\|{h}"))
	alloc[key]["got"] = got
	for u in drawn:
	if u["stratum"] == f"{c}\|{r}\|{h}":
	u["n_stratum"] = got
	u["pi"] = got / alloc[key]["N"]
	log(f"drawn {len(drawn)} windows on {len({u['dongle'] for u in drawn})} devices; per regime {Counter(u['regime'] for u in drawn)}; per cell {Counter(u['cell'] for u in drawn)}; "
	f"top device {max(dev_used.values())}")
	N_cell = D.cell.value_counts().to_dict()
	n_cell = Counter(u["cell"] for u in drawn)
	for u in drawn:
	u["N_cell"], u["n_cell"] = int(N_cell[u["cell"]]), int(n_cell[u["cell"]])
	# ---- ids, calibration, queue
	drawn.sort(key=lambda u: (u["cell"], u["regime"], u["uid"], u["t"]))
	perm = rng.permutation(len(drawn))
	for rank, j in enumerate(perm):
	drawn[j]["idx"] = IDX_BASE + rank
	for u in drawn:
	u["clip_id"] = opaque_id(u["uid"], u["t"])
	u["leaf"] = NS.leaf_of(u["uid"])
	u["dcam"], u["qcam"] = str(NS.BATON_LOCAL / u["leaf"] / "dcamera.mp4"), str(NS.BATON_LOCAL / u["leaf"] / "qcamera.mp4")
	u["video_local"] = NS.video_local(u["uid"])
	u["calibration"] = False
	u["t_start"], u["t_end"], u["anchor_s"] = round(max(0.0, u["t"] - PRE_S), 3), round(u["t"] + POST_S, 3), round(min(PRE_S, u["t"]), 3)
	for c, r in CALIB_CELLS.items():
	cand = [u for u in drawn if u["cell"] == c and u["regime"] == r]
	if not cand:
	cand = [u for u in drawn if u["cell"] == c]
	cand[int(rng.integers(len(cand)))]["calibration"] = True
	calib = [u for u in drawn if u["calibration"]]
	rest = [u for u in drawn if not u["calibration"]]
	queue = [calib[j] for j in rng.permutation(len(calib))] + [rest[j] for j in rng.permutation(len(rest))]
	for q, u in enumerate(queue):
	u["qpos"] = q
	cols = ["qpos", "idx", "clip_id", "calibration", "regime", "uid", "dongle", "leaf", "t", "t_start", "t_end", "anchor_s", "cell", "stratum", "asym_half", "pi", "N_stratum", "n_stratum", "N_cell", "n_cell",
	"gaze_off", "dwell_s", "head_turned", "face_valid", "kin_valid", "ear_conf_asym", "yaw_dev", "video_local", "dcam", "qcam"]
	M = pd.DataFrame([{k: u.get(k) for k in cols} for u in queue], columns=cols)
	assert M.idx.is_unique and M.clip_id.is_unique and ((M.pi > 0) & (M.pi <= 1)).all()
	M.to_csv(man_p, index=False)
	fc = dict(version=a.version, seed=a.seed, generated_utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), n_rows_val={r: int((D.regime == r).sum()) for r in REGIMES},
	n_devices_val=int(D.dongle.nunique()), cells_total=N_cell, cells_by_regime={c: {r: int(cell_counts.loc[c, r]) if c in cell_counts.index else 0 for r in REGIMES} for c in cell_counts.index},
	alloc=ALLOC, topup=TOPUP, topup_regime=minority, dev_cap=DEV_CAP, guard_s=GUARD_S, asym_threshold=ASYM_TH, dwell_brief_s=DWELL_BRIEF_S,
	strata={"\|".join(k): v for k, v in alloc.items()}, drawn=len(M), drawn_by_cell=dict(n_cell), drawn_by_regime=dict(Counter(M.regime)), devices_drawn=int(M.dongle.nunique()),
	top_device=int(max(dev_used.values())), video_local=int(M.video_local.sum()), routes=int(M.uid.nunique()), routes_local=int(M[M.video_local].uid.nunique()),
	calibration=[dict(idx=u["idx"], cell=u["cell"], regime=u["regime"]) for u in calib],
	asym_share_among_off={r: float((np.abs(D[(D.regime == r) & (D.gaze_off == 1)].ear_conf_asym) > ASYM_TH).mean()) for r in REGIMES},
	device_concentration={c: dict(Counter(D[D.cell == c].dongle).most_common(3)) for c in CELLS},
	clip_window_s=[PRE_S, POST_S], log=lines)
	json.dump(fc, open(fc_p, "w"), indent=1, default=lambda o: int(o) if isinstance(o, np.integer) else float(o) if isinstance(o, np.floating) else str(o))
	write_run_json(OUT, "gaze_off_validation_sample", [Path(__file__), Path(NS.__file__)], files, seed=a.seed, t_start=t_start,
	extra=dict(version=a.version, outputs=[str(man_p), str(fc_p)], n=len(M), by_cell=dict(n_cell), dms_fields_used=[]))
	log(f"DONE in {time.time() - t_start:.0f}s: {len(M)} windows -> {man_p}")


	if __name__ == "__main__":
	main()