Buckets:
| """cwm3/eval/newtest_sample.py — freeze the three warning systems, their fixed-load thresholds and the NEW human-evaluation test set (S1 / S2 / S3) | |
| with an inclusion probability per clip (docs/newtest_protocol_draft.md §1; pre-registration docs/newtest_prereg.md; 2026-09-12). | |
| Frozen systems (scores read from published layouts only; nothing is retrained): | |
| A full predictive model + warning head `W_learned_bothhgb_t15_w10_n979`, MEAN over the 3 frozen seeds of results/runs/W_I1_all_minus_road2drv_s{0,1,2}_final_merged/ | |
| B strongest current-state-only head `W_learned_curhgb_t15_w10_n979dwfc` in results/runs/W_I1_all_minus_road2drv_s0_final_dwfc/ (seed-invariant by construction) | |
| C openpilot DMS chain att@3s compose.legacy_compose(pA, None, 15)["att_only"] on cwm2/baselines/legacy_cockpitwm/results/marginals_{regime}_s0.npz (TEST) | |
| and marginals_{regime}_val_s0.npz (VAL; the frozen s0 chain checkpoints scored on the val pack by | |
| score_val_marginals.py, verified against the published test marginals). NOTE the protocol draft named | |
| att_{regime}_val.npz["A"] for val — that array is the chain's binary future-state LABEL pack, not a prediction. | |
| The DMS chain enters ONLY as the labelled comparator through the existing compose path (exactly as eval/operating_points.py); no DMS field is an input anywhere. | |
| Thresholds (frozen before any clip is served): per system x regime, bisection on alerts / hour with the 60 s per-route cooldown | |
| (operating_points.thr_for_rate -> train_warning_head.alerts_per_hour) on the VAL layouts to 10 / h and 20 / h (primary, deployable) and on the TEST | |
| layouts (secondary, "equal achieved load"); the achieved load of every threshold on both splits is recorded (the val -> test transfer facts). A target | |
| that cannot be reached because the score saturates (the DMS att@3s chain at 10 / h) is recorded with reachable = false and its floor (alerts / h at | |
| threshold = max score); the stored threshold is then the max score. | |
| Sampling frame = the TEST-split exposure of the two regime layouts (5-Hz s12 grid), partitioned into | |
| S1 alert union: alerts of A, B, C at the frame threshold min(thr_val20, thr_test20) per system x regime, merged per route with a 10 s same-event | |
| guard (a new event starts when an alert is > 10 s after the current event's first alert; anchor t = the first alert); fire pattern = the systems with | |
| a frame-threshold alert in [t-10, t+10] (the same window rule as fired_*; the cluster's own systems are kept as pattern_cluster); | |
| cell = regime x pattern {A, B, C, A+B, A+C, B+C, A+B+C}; | |
| 160 clips = per regime A 14, B 14, C 14, A+B 12, A+C 10, B+C 10, A+B+C 6; device cap 10 over S1; stratified SRS within (cell, device-group): | |
| a device whose remaining cap budget is below its count in the cell is its own stratum (n <= budget, pi exact); among the never-cappable devices | |
| those earning >= 1 draw are their own stratum and the rest are pooled into one stratum (an exact SRS); allocation proportional to N with the cap, | |
| SRS within -> pi_i = n_stratum / N_stratum. Devices whose TOTAL count exceeds the cap have their draws PRE-ALLOCATED across their (cell, device) | |
| strata greedily by N (>= 1 in each of their largest cells) so the cap costs as little frame coverage as possible. A stratum whose device is | |
| already at the cap receives no draw: its events are UNCOVERED (pi = 0); | |
| N_covered / coverage per cell are stored so the analysis post-stratifies the HT weights to the full cell (w_i = (N_cell / N_covered) / pi_i) and | |
| reports the coverage. Cells are allocated in increasing N_cell so the cap binds on the big single-system cells. | |
| S2 silent moving background: grid rows with speed > 2 m/s at t, not within 12.5 s of any S1 union event or excluded anchor; 40 clips (20 / regime), | |
| strata (regime, device-group) as above with device cap 6, SRS of rows with a 15 s non-overlap guard; pi_i = n_stratum / N_stratum rows. | |
| S3 model-independent event candidates: the 100 round-2 TEST clips (results/mining/round2_draw_v1.csv == cwm2/annotation_human/clips_r2.json; their | |
| prefix cuts exist and are reused), pi_i = quota_stratum / N_eligible_stratum (strata_v1.csv, test, eligible); optional +N hard_response_h5 clips drawn | |
| here with the mine_hard_clips rule (seed 1, device cap 8, 10 s guard, regime alternating; --s3-hard N). S3 is used for paired event-level analyses | |
| and AUROC, never for HT natural rates. | |
| Exclusion (S1 / S2 only): anything within 12.5 s (same route) of a clean-test gold anchor (score_clean568.episode_table(), clean_test), of a round-1 H5 | |
| TEST clip (clips_h5.json, split test) or of a round-2 TEST clip. | |
| Per clip (server-side only, never sent to a rater): fired_{A,B,C}_{val10,val20,test10,test20} (an alert of that system at that threshold in [t-10, t+10]), | |
| t_alert_{A,B,C}_{frame,val10,val20,test10,test20} (first such alert), score_{A,B,C}_anchor, stratum, cell, pattern, pi, N / n of its stratum, phase | |
| (1 = S3 + the first 60 S1 clips; 2 = the rest), calibration flag (8 warning clips: S1 3, S2 3, S3 2, 4 / 4 regime; + 4 gaze windows flagged by | |
| gaze_off_validation_sample.py = the 12-clip calibration set), video_local (both cameras on disk NOW — | |
| results/baton_leaves.json is stale after the 2026-09-12 prune), needs_frames. | |
| Outputs (append-only; refuses to overwrite an existing version): results/newtest/manifest_<v>.csv, frame_counts_<v>.json, thresholds_frozen_<v>.json | |
| (+ sha256 of every layout / marginals file), run_newtest_sample_<stamp>.json (+ inputs listing). | |
| python eval/newtest_sample.py [--seed 0] [--version v1] [--s3-hard 20] | |
| Unit tests: tests/test_newtest_sample.py (synthetic frames: alert semantics == train_warning_head.alerts_per_hour, merge guard, exclusion, allocation / caps, | |
| pi sanity, seed reproducibility). | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import os | |
| import sys | |
| import time | |
| from collections import Counter, defaultdict | |
| from pathlib import Path | |
| os.environ.setdefault("OMP_NUM_THREADS", "4") | |
| import numpy as np # noqa: E402 | |
| import pandas as pd # noqa: E402 | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| for p in (CWM3 / "eval", CWM3 / "common", CWM2 / "eval", CWM2, CWM2 / "baselines" / "legacy_cockpitwm"): | |
| sys.path.insert(0, str(p)) | |
| REGIMES = ("manual", "engaged") | |
| SYSTEMS = ("A", "B", "C") | |
| SEEDS_A = (0, 1, 2) | |
| RUN_A = "W_I1_all_minus_road2drv_s{seed}_final_merged" | |
| COL_A = "W_learned_bothhgb_t15_w10_n979" | |
| RUN_B = "W_I1_all_minus_road2drv_s0_final_dwfc" | |
| COL_B = "W_learned_curhgb_t15_w10_n979dwfc" | |
| DMS_DIR = CWM2 / "baselines" / "legacy_cockpitwm" / "results" | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| TAU = 15 | |
| COOLDOWN = 60.0 | |
| GRID_DT = 0.2 | |
| POINTS = {"val10": ("val", 10.0), "val20": ("val", 20.0), "test10": ("test", 10.0), "test20": ("test", 20.0)} | |
| PATTERNS = ("A", "B", "C", "A+B", "A+C", "B+C", "A+B+C") | |
| S1_ALLOC = {"A": 14, "B": 14, "C": 14, "A+B": 12, "A+C": 10, "B+C": 10, "A+B+C": 6} # per regime -> 80 / regime, 160 | |
| S1_DEV_CAP = 10 | |
| S2_PER_REGIME = 20 | |
| S2_DEV_CAP = 6 | |
| S2_GUARD_S = 15.0 | |
| MERGE_S = 10.0 | |
| EXCL_S = 12.5 | |
| ALERT_WIN_S = 10.0 | |
| MOVING_MPS = 2.0 | |
| VEGO_CH = 3 # s12 canspec_v2.TARGETS = [steeringAngleDeg, gasPressed, brakePressed, vEgo, aEgo] | |
| S3_QUOTA = {"rare_road": 25, "demand_noresp": 20, "demand_resp": 20, "hand_head": 15, "occlusion_night": 5, "background": 15} # mine_hard_clips.QUOTA["test"] | |
| S3_HARD_DEV_CAP = 8 | |
| S3_HARD_SEED = 1 | |
| IDX_BASE = {"S1": 70000, "S2": 80000, "S3_hard": 85000} # s15 episode idx max = 42049 -> no collision; 5 digits for the server's file names | |
| N_PHASE1_S1 = 60 | |
| CALIB = {"S1": {"manual": 2, "engaged": 1}, "S2": {"manual": 1, "engaged": 2}, "S3": {"manual": 1, "engaged": 1}} # 8 warning clips (+ 4 gaze windows) | |
| BATON_LOCAL = Path("/home/USER/BATON") | |
| OUT_DIR = CWM3 / "results" / "newtest" | |
| MINING = CWM3 / "results" / "mining" | |
| ANN = CWM2 / "annotation_human" | |
| # ----------------------------------------------------------------------------------------------------------------------- small helpers | |
| def leaf_of(uid): | |
| p = uid.split("__") | |
| return "/".join(p[:3]) + "/" + "__".join(p[3:]) | |
| def opaque_id(uid, t): | |
| return "n" + hashlib.sha1(f"{uid}|{t:.3f}|cwm3-newtest-v1".encode()).hexdigest()[:12] | |
| def sha256_file(p, chunk=1 << 22): | |
| h = hashlib.sha256() | |
| with open(p, "rb") as f: | |
| for b in iter(lambda: f.read(chunk), b""): | |
| h.update(b) | |
| return h.hexdigest() | |
| def video_local(uid): | |
| d = BATON_LOCAL / leaf_of(uid) | |
| return bool((d / "dcamera.mp4").exists() and (d / "qcamera.mp4").exists()) | |
| def pattern_of(systems): | |
| return "+".join(s for s in SYSTEMS if s in systems) | |
| # ----------------------------------------------------------------------------------------------------------------------- alert semantics | |
| def alert_rows(uid, t, s, thr, cooldown=COOLDOWN): | |
| """Row indices (into the given arrays) of the alerts of one score at threshold thr with the per-route cooldown — the SAME semantics as | |
| train_warning_head.alerts_per_hour (rows visited in (uid, t) order; an alert fires when s >= thr and >= cooldown since the route's last alert).""" | |
| order = np.lexsort((t, uid)) | |
| u, tt, ss = uid[order], t[order], s[order] | |
| fire = np.isfinite(ss) & (ss >= thr) | |
| out = [] | |
| last_uid, last_t = None, -1e9 | |
| for j in range(len(u)): | |
| if u[j] != last_uid: | |
| last_uid, last_t = u[j], -1e9 | |
| if fire[j] and tt[j] - last_t >= cooldown: | |
| out.append(order[j]) | |
| last_t = tt[j] | |
| return np.asarray(out, dtype=int) | |
| def alerts_per_hour(uid, t, s, thr, cooldown=COOLDOWN): | |
| n = len(alert_rows(uid, t, s, thr, cooldown)) | |
| hours = len(uid) * GRID_DT / 3600.0 | |
| return n, hours, n / hours if hours > 0 else np.nan | |
| def by_route(uid, t): | |
| """uid -> sorted array of t (for [t-10, t+10] queries).""" | |
| d = defaultdict(list) | |
| for u, x in zip(uid, t): | |
| d[u].append(float(x)) | |
| return {u: np.sort(np.asarray(v)) for u, v in d.items()} | |
| def first_in_window(times_by_uid, uid, t, lo=ALERT_WIN_S, hi=ALERT_WIN_S): | |
| arr = times_by_uid.get(uid) | |
| if arr is None or not len(arr): | |
| return np.nan | |
| i = int(np.searchsorted(arr, t - lo - 1e-9)) | |
| if i < len(arr) and arr[i] <= t + hi + 1e-9: | |
| return float(arr[i]) | |
| return np.nan | |
| class RouteRows: | |
| """(uid, t) -> row index of the nearest grid row within tol.""" | |
| def __init__(self, uid, t, tol=0.11): | |
| self.tol = tol | |
| order = np.lexsort((t, uid)) | |
| self.u, self.t, self.rows = uid[order], t[order].astype(float), order | |
| self.bounds = {} | |
| starts = np.flatnonzero(np.r_[True, self.u[1:] != self.u[:-1]]) | |
| ends = np.r_[starts[1:], len(self.u)] | |
| for a, b in zip(starts, ends): | |
| self.bounds[self.u[a]] = (a, b) | |
| def row(self, uid, t): | |
| ab = self.bounds.get(uid) | |
| if ab is None: | |
| return None | |
| a, b = ab | |
| tt = self.t[a:b] | |
| j = int(np.searchsorted(tt, t)) | |
| best = None | |
| for k in (j - 1, j): | |
| if 0 <= k < len(tt) and abs(tt[k] - t) <= self.tol and (best is None or abs(tt[k] - t) < abs(tt[best] - t)): | |
| best = k | |
| return None if best is None else int(self.rows[a + best]) | |
| # ----------------------------------------------------------------------------------------------------------------------- merge / exclusion / allocation (pure, testable) | |
| def merge_events(alerts, merge_s=MERGE_S): | |
| """alerts: list of (uid, t, system, regime) -> list of events dict(uid, t, regime, systems:set, alerts:list) with the anchor-based guard: | |
| per route (both regimes together), sorted by t; a new event starts when an alert is > merge_s after the current event's FIRST alert.""" | |
| per = defaultdict(list) | |
| for u, t, x, r in alerts: | |
| per[u].append((float(t), x, r)) | |
| events = [] | |
| for u in sorted(per): | |
| cur = None | |
| for t, x, r in sorted(per[u], key=lambda z: (z[0], z[1])): | |
| if cur is None or t - cur["t"] > merge_s: | |
| cur = dict(uid=u, t=t, regime=r, systems={x}, alerts=[(t, x, r)]) | |
| events.append(cur) | |
| else: | |
| cur["systems"].add(x) | |
| cur["alerts"].append((t, x, r)) | |
| return events | |
| def near_any(anchors_by_uid, uid, t, tol): | |
| arr = anchors_by_uid.get(uid) | |
| if arr is None or not len(arr): | |
| return False | |
| i = int(np.searchsorted(arr, t)) | |
| for k in (i - 1, i): | |
| if 0 <= k < len(arr) and abs(arr[k] - t) <= tol + 1e-9: | |
| return True | |
| return False | |
| def largest_remainder(n, weights): | |
| w = np.asarray(weights, float) | |
| if n <= 0 or w.sum() <= 0: | |
| return [0] * len(w) | |
| raw = n * w / w.sum() | |
| base = np.floor(raw).astype(int) | |
| for i in np.argsort(-(raw - base), kind="mergesort")[: n - int(base.sum())]: | |
| base[i] += 1 | |
| return base.tolist() | |
| def allocate_capped(N_by, n, avail_by): | |
| """Allocate n draws over strata proportional to N_by[k] (largest remainder), never exceeding avail_by[k]; leftover is redistributed to strata with slack. | |
| -> {k: n_k}.""" | |
| keys = list(N_by) | |
| alloc = {k: 0 for k in keys} | |
| remaining = min(int(n), int(sum(min(N_by[k], avail_by.get(k, N_by[k])) for k in keys))) | |
| for _ in range(len(keys) + 2): | |
| if remaining <= 0: | |
| break | |
| active = [k for k in keys if alloc[k] < min(N_by[k], avail_by.get(k, N_by[k]))] | |
| if not active: | |
| break | |
| add = largest_remainder(remaining, [N_by[k] for k in active]) | |
| moved = 0 | |
| for k, a in zip(active, add): | |
| a = min(a, min(N_by[k], avail_by.get(k, N_by[k])) - alloc[k]) | |
| alloc[k] += a | |
| moved += a | |
| remaining -= moved | |
| if moved == 0: # every active stratum rounded to 0: give 1 to the largest active stratum | |
| k = max(active, key=lambda z: N_by[z]) | |
| alloc[k] += 1 | |
| remaining -= 1 | |
| return alloc | |
| def device_groups(N_by_dev, n, remaining=None): | |
| """Strata over devices for one cell: a device whose remaining cap budget is smaller than its count is ALWAYS its own stratum (its draws are limited, | |
| pi = n_d / N_d stays exact); among the never-cappable devices those whose proportional share earns >= 1 draw are their own stratum and the rest are | |
| pooled into '_pooled' (an SRS over the pool is exact because none of its devices can hit the cap inside the cell). -> {dev: group}.""" | |
| tot = sum(N_by_dev.values()) | |
| groups = {} | |
| for d, N in N_by_dev.items(): | |
| limited = remaining is not None and remaining.get(d, N) < N | |
| groups[d] = d if (limited or (tot > 0 and n * N / tot >= 1.0)) else "_pooled" | |
| if list(groups.values()).count("_pooled") == 1: # a pooled group of one device is just that device | |
| for d, g in groups.items(): | |
| if g == "_pooled": | |
| groups[d] = d | |
| return groups | |
| def preallocate_capped(units, cell_of, n_by_cell, cap, extra="fixed"): | |
| """Coverage-maximising placement of the draws of devices whose TOTAL count exceeds the cap: each such device's `cap` draws are spread over its | |
| (cell, device) strata greedily by N_{cell,d} descending (1 draw to each of its largest cells, the rest proportional), so that one draw covers as many | |
| frame events as possible; per cell the fixed draws leave >= 1 draw for the other devices. Devices with N_d <= cap never bind and get no fixed draws. | |
| extra = "fixed": the cap - k spare draws are fixed proportional to N over the planned cells; "free": only the 1-per-cell floor is fixed and the spare | |
| draws are left to the proportional allocation inside the planned cells (S2: two cells, many devices). Planned devices receive NO draw outside their | |
| plan (stratified_draw enforces it), so the budget reserved for the large cells cannot be spent in small ones. | |
| -> ({(cell, dongle): n_fixed}, planned devices).""" | |
| N = Counter((cell_of(u), u["dongle"]) for u in units) | |
| N_d = Counter(u["dongle"] for u in units) | |
| fixed, planned = {}, set() | |
| for d, tot in N_d.items(): | |
| if tot <= cap: | |
| continue | |
| planned.add(d) | |
| cells = sorted([c for (c, dd) in N if dd == d], key=lambda c: (-N[(c, d)], c)) | |
| k = min(cap, len(cells)) | |
| base = {c: 1 for c in cells[:k]} | |
| if extra == "fixed": | |
| for c, add in zip(cells[:k], largest_remainder(cap - k, [N[(c, d)] for c in cells[:k]])): | |
| base[c] += add | |
| for c, n in base.items(): | |
| fixed[(c, d)] = min(n, N[(c, d)]) | |
| for c, n_cell in n_by_cell.items(): | |
| items = sorted([(d, fixed[(cc, d)]) for (cc, d) in fixed if cc == c], key=lambda x: N[(c, x[0])]) | |
| while items and sum(n for _, n in items) > max(n_cell - 1, 0): | |
| d, n = items[0] | |
| if n > 1: | |
| fixed[(c, d)] -= 1 | |
| items[0] = (d, n - 1) | |
| else: | |
| del fixed[(c, d)] | |
| items.pop(0) | |
| items.sort(key=lambda x: N[(c, x[0])]) | |
| return fixed, planned | |
| def stratified_draw(units, n, dev_cap, dev_used, rng, guard=None, unit_key=lambda u: (u["uid"], u["t"]), fixed=None, planned=None, extra_free=False): | |
| """Stratified SRS of `units` (dicts with 'dongle', 'uid', 't') into device-group strata (device_groups): allocation proportional to N with the global | |
| device cap, SRS within each stratum (a stratum whose devices are already at the cap gets no draw -> its units are uncovered: pi = 0, reported as | |
| N_uncovered / coverage so the analysis can post-stratify to the full cell). With `guard`, a unit within `guard` s of an already drawn unit of the same | |
| route is rejected and redrawn (a negligible perturbation of pi for dense row frames; documented). | |
| -> (drawn units with pi / stratum fields added, strata table, coverage dict). Mutates dev_used (Counter of draws per device).""" | |
| by_dev = Counter(u["dongle"] for u in units) | |
| remaining = {d: max(0, dev_cap - dev_used[d]) for d in by_dev} | |
| fixed = {d: min(v, by_dev[d], remaining[d]) for d, v in (fixed or {}).items() if d in by_dev} | |
| planned = set(planned or ()) | |
| for d in by_dev: | |
| if d in planned and d not in fixed: | |
| remaining[d] = 0 # planned elsewhere: no draw in this cell (its budget is reserved for its planned cells) | |
| grp = device_groups(by_dev, n, remaining) | |
| for d in by_dev: # planned devices are always their own stratum | |
| if d in planned: | |
| grp[d] = d | |
| if sum(1 for d, g in grp.items() if g == "_pooled") == 1: | |
| for d, g in grp.items(): | |
| if g == "_pooled": | |
| grp[d] = d | |
| N_g = Counter(grp[d] for d in by_dev.elements()) | |
| avail_g = {g: sum(min(by_dev[d], remaining[d]) for d in by_dev if grp[d] == g) for g in N_g} | |
| n_fixed = sum(fixed.values()) | |
| alloc = dict(fixed) | |
| if extra_free: # spare draws may also go to the planned devices of this cell (above their floor) | |
| free = {g: N_g[g] for g in N_g if avail_g[g] - alloc.get(g, 0) > 0} | |
| add = allocate_capped(free, max(0, n - n_fixed), {g: avail_g[g] - alloc.get(g, 0) for g in free}) | |
| else: | |
| free = {g: N_g[g] for g in N_g if g not in fixed} | |
| add = allocate_capped(free, max(0, n - n_fixed), {g: avail_g[g] for g in free}) | |
| for g, v in add.items(): | |
| alloc[g] = alloc.get(g, 0) + v | |
| for g in N_g: | |
| alloc.setdefault(g, 0) | |
| short = n - sum(alloc.values()) | |
| if short > 0: # fallback: the free strata cannot absorb the target -> planned devices with slack in THIS cell | |
| slack = {g: avail_g[g] - alloc[g] for g in N_g if avail_g[g] - alloc[g] > 0} | |
| for g, v in allocate_capped({g: N_g[g] for g in slack}, short, slack).items(): | |
| alloc[g] += v | |
| drawn, table = [], [] | |
| for g in sorted(N_g, key=lambda z: (-N_g[z], str(z))): | |
| pool = [u for u in units if grp[u["dongle"]] == g] | |
| pool.sort(key=unit_key) | |
| n_g = alloc[g] | |
| order = rng.permutation(len(pool)) | |
| got, chosen = 0, [] | |
| taken_by_uid = defaultdict(list) | |
| for j in order: | |
| if got >= n_g: | |
| break | |
| u = pool[j] | |
| assert dev_used[u["dongle"]] < dev_cap # by construction (avail_g) — never a cap hit inside the SRS | |
| if guard is not None and any(abs(u["t"] - x) < guard for x in taken_by_uid[u["uid"]]): | |
| continue | |
| chosen.append(u) | |
| taken_by_uid[u["uid"]].append(u["t"]) | |
| dev_used[u["dongle"]] += 1 | |
| got += 1 | |
| for u in chosen: | |
| v = dict(u) | |
| v.update(dev_group=str(g), N_stratum=int(N_g[g]), n_stratum=int(got), pi=got / N_g[g]) | |
| drawn.append(v) | |
| table.append(dict(dev_group=str(g), N=int(N_g[g]), n_alloc=int(n_g), n=int(got), n_devices=int(sum(1 for d in by_dev if grp[d] == g)), avail=int(avail_g[g]))) | |
| N_cov = sum(r["N"] for r in table if r["n"] > 0) | |
| cov = dict(N=len(units), N_covered=int(N_cov), N_uncovered=int(len(units) - N_cov), coverage=(N_cov / len(units)) if units else float("nan"), | |
| uncovered_groups=[r["dev_group"] for r in table if r["n"] == 0]) | |
| for u in drawn: | |
| u["N_covered_cell"], u["coverage_cell"] = cov["N_covered"], cov["coverage"] | |
| return drawn, table, cov | |
| def exact_device_coverage(units, n): | |
| """Coverage of the literal per-device rule (no pooling): share of units on devices that would receive >= 1 draw under proportional allocation.""" | |
| by_dev = Counter(u["dongle"] for u in units) | |
| alloc = largest_remainder(n, [by_dev[d] for d in by_dev]) | |
| covered = sum(by_dev[d] for d, a in zip(by_dev, alloc) if a > 0) | |
| return covered / max(len(units), 1) | |
| # ----------------------------------------------------------------------------------------------------------------------- scores | |
| def load_scores(log): | |
| """-> scores[(X, regime, split)] = dict(uid, t, dongle, s), packs, hours, files (for hashing).""" | |
| import warning_gold as WG # unchanged cwm2 code: load_pack | |
| from compose import legacy_compose | |
| files = [] | |
| packs = {(r, sp): WG.load_pack(r, sp) for r in REGIMES for sp in ("test", "val")} | |
| for r in REGIMES: | |
| for sp in ("test", "val"): | |
| files.append(S12R / f"targets_{r}_{sp}.npz") | |
| S = {} | |
| for r in REGIMES: | |
| for sp in ("test", "val"): | |
| pk = packs[(r, sp)] | |
| acc = None | |
| for seed in SEEDS_A: | |
| p = CWM3 / "results" / "runs" / RUN_A.format(seed=seed) / f"pred_layout_{r}_{sp}.npz" | |
| files.append(p) | |
| z = np.load(p, allow_pickle=True) | |
| names = list(map(str, z["w_names"])) | |
| assert COL_A in names, (p, COL_A) | |
| assert np.array_equal(z["uid"], pk["uid"]) and np.allclose(z["t"].astype(float), pk["t"].astype(float)), f"{p}: rows != s12 pack" | |
| v = z["w"][:, names.index(COL_A)].astype(np.float64) | |
| if "w_mask" in z.files: | |
| v = np.where(z["w_mask"], v, np.nan) | |
| acc = v if acc is None else acc + v | |
| S[("A", r, sp)] = dict(uid=pk["uid"], t=pk["t"].astype(float), dongle=pk["dongle"], s=acc / len(SEEDS_A)) | |
| p = CWM3 / "results" / "runs" / RUN_B / f"pred_layout_{r}_{sp}.npz" | |
| files.append(p) | |
| z = np.load(p, allow_pickle=True) | |
| names = list(map(str, z["w_names"])) | |
| assert COL_B in names, (p, COL_B) | |
| assert np.array_equal(z["uid"], pk["uid"]), f"{p}: rows != s12 pack" | |
| v = z["w"][:, names.index(COL_B)].astype(np.float64) | |
| if "w_mask" in z.files: | |
| v = np.where(z["w_mask"], v, np.nan) | |
| S[("B", r, sp)] = dict(uid=pk["uid"], t=pk["t"].astype(float), dongle=pk["dongle"], s=v) | |
| # C: test = legacy marginals (own rows), val = att_<regime>_val.npz["A"] aligned with the s12 val pack | |
| p = DMS_DIR / f"marginals_{r}_s0.npz" | |
| files.append(p) | |
| z = np.load(p, allow_pickle=True) | |
| S[("C", r, "test")] = dict(uid=z["uid"], t=z["t"].astype(float), dongle=z["dongle"], s=legacy_compose(z["pA"].astype(np.float32), None, TAU)["att_only"].astype(np.float64)) | |
| p = DMS_DIR / f"marginals_{r}_val_s0.npz" | |
| assert p.exists(), f"{p} missing: run cwm2/baselines/legacy_cockpitwm/score_val_marginals.py --regime {r} --seed 0 --verify-test first" | |
| files.append(p) | |
| z = np.load(p, allow_pickle=True) | |
| pk = packs[(r, "val")] | |
| assert set(z["uid"]) <= set(pk["uid"]), f"{p}: routes not in the s12 val pack" | |
| S[("C", r, "val")] = dict(uid=z["uid"], t=z["t"].astype(float), dongle=z["dongle"], s=legacy_compose(z["pA"].astype(np.float32), None, TAU)["att_only"].astype(np.float64)) | |
| hours = {r: len(packs[(r, "test")]["uid"]) * GRID_DT / 3600.0 for r in REGIMES} | |
| for (X, r, sp), d in sorted(S.items()): | |
| fin = np.isfinite(d["s"]) | |
| log(f"scores {X} {r:8s} {sp:4s}: rows {len(d['s'])} routes {len(np.unique(d['uid']))} devices {len(np.unique(d['dongle']))} finite {fin.mean():.4f} " | |
| f"min {np.nanmin(d['s']):.4f} max {np.nanmax(d['s']):.4f} share at max {(d['s'] >= np.nanmax(d['s']) - 1e-12).mean():.5f}") | |
| return S, packs, hours, files | |
| def freeze_thresholds(S, log): | |
| """-> thr[X][r][point] = dict(thr, target, fit_split, reachable, floor_alerts_per_hour_fit, achieved_fit, achieved_test, achieved_val, n_alerts_test).""" | |
| import operating_points as OP # thr_for_rate (bisection, 24 it) -> train_warning_head.alerts_per_hour (unchanged) | |
| thr = {X: {r: {} for r in REGIMES} for X in SYSTEMS} | |
| for X in SYSTEMS: | |
| for r in REGIMES: | |
| for point, (sp, rate) in POINTS.items(): | |
| d = S[(X, r, sp)] | |
| smax = float(np.nanmax(d["s"])) | |
| floor = OP.TW.alerts_per_hour(d["uid"], d["t"], d["s"], smax)[2] | |
| reachable = bool(floor <= rate + 1e-9) | |
| th = OP.thr_for_rate(d["uid"], d["t"], d["s"], rate) if reachable else smax | |
| ach = {s2: OP.TW.alerts_per_hour(S[(X, r, s2)]["uid"], S[(X, r, s2)]["t"], S[(X, r, s2)]["s"], th) for s2 in ("test", "val")} | |
| thr[X][r][point] = dict(thr=float(th), target_alerts_per_hour=rate, fit_split=sp, reachable=reachable, floor_alerts_per_hour_fit=float(floor), | |
| achieved_fit=float(ach[sp][2]), achieved_test=float(ach["test"][2]), achieved_val=float(ach["val"][2]), | |
| n_alerts_test=int(ach["test"][0]), n_alerts_val=int(ach["val"][0]), hours_test=float(ach["test"][1]), hours_val=float(ach["val"][1])) | |
| log(f"thr {X} {r:8s} {point:6s}: thr {th:.5f} target {rate:4.0f}/h on {sp}; achieved fit {ach[sp][2]:.2f}/h, test {ach['test'][2]:.2f}/h, val {ach['val'][2]:.2f}/h" | |
| + ("" if reachable else f" ** NOT REACHABLE: floor {floor:.2f}/h (score saturates at {smax:.4f}); threshold = max score")) | |
| f = min(thr[X][r]["val20"]["thr"], thr[X][r]["test20"]["thr"]) | |
| d = S[(X, r, "test")] | |
| n_f, h_f, r_f = OP.TW.alerts_per_hour(d["uid"], d["t"], d["s"], f) | |
| thr[X][r]["frame"] = dict(thr=float(f), rule="min(thr_val20, thr_test20)", n_alerts_test=int(n_f), achieved_test=float(r_f), hours_test=float(h_f)) | |
| log(f"thr {X} {r:8s} frame : thr {f:.5f} -> {n_f} alerts, {r_f:.2f}/h on test") | |
| return thr | |
| # ----------------------------------------------------------------------------------------------------------------------- exclusion anchors | |
| def exclusion_anchors(log): | |
| """uid -> sorted t of: clean-test gold anchors (568-episode clean_test), round-1 H5 TEST clips, round-2 TEST clips. -> (dict, counts).""" | |
| from score_clean568 import episode_table # the clean-568 episode table (the same function the gold uses) | |
| EP = episode_table() | |
| gold = EP[EP.clean_test] | |
| h5 = json.load(open(ANN / "clips_h5.json"))["clips"] | |
| h5t = [c for c in h5 if c["split"] == "test"] | |
| r2 = pd.read_csv(MINING / "round2_draw_v1.csv") | |
| r2 = r2[r2.split == "test"] | |
| src = {"clean_test_gold": list(zip(gold.uid, gold.t0.astype(float))), "round1_h5_test": [(c["uid"], float(c["t0"])) for c in h5t], | |
| "round2_test": list(zip(r2.uid, r2.t0.astype(float)))} | |
| d = defaultdict(list) | |
| for name, lst in src.items(): | |
| for u, t in lst: | |
| d[u].append(t) | |
| counts = {k: len(v) for k, v in src.items()} | |
| log(f"exclusion anchors: {counts} -> {sum(counts.values())} anchors on {len(d)} routes") | |
| return {u: np.sort(np.asarray(v)) for u, v in d.items()}, counts, src | |
| # ----------------------------------------------------------------------------------------------------------------------- main | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--seed", type=int, default=0) | |
| ap.add_argument("--version", default="v1") | |
| ap.add_argument("--s3-hard", type=int, default=20, help="optional hard_response_h5 clips added to S3 (flagged optional, phase 2); 0 = none") | |
| ap.add_argument("--out", default=str(OUT_DIR)) | |
| a = ap.parse_args() | |
| t_start = time.time() | |
| out = Path(a.out) | |
| out.mkdir(parents=True, exist_ok=True) | |
| man_p, fc_p, thr_p = out / f"manifest_{a.version}.csv", out / f"frame_counts_{a.version}.json", out / f"thresholds_frozen_{a.version}.json" | |
| for p in (man_p, fc_p, thr_p): | |
| if p.exists(): | |
| raise SystemExit(f"append-only: {p} exists — choose a new --version") | |
| log_p = out / f"newtest_sample_{a.version}.log" | |
| lines = [] | |
| def log(m): | |
| line = f"[{time.strftime('%H:%M:%S')}] {m}" | |
| print(line, flush=True) | |
| lines.append(line) | |
| with open(log_p, "a") as f: | |
| f.write(line + "\n") | |
| rng = np.random.default_rng(a.seed) | |
| from runmeta import write_run_json | |
| import operating_points as OP | |
| S, packs, hours, files = load_scores(log) | |
| dongle_of = {} | |
| for (X, r, sp), d in S.items(): | |
| for u, dg in zip(d["uid"], d["dongle"]): | |
| dongle_of.setdefault(u, dg) | |
| # sanity: the local alert semantics equal train_warning_head.alerts_per_hour on a real layout | |
| d = S[("A", "manual", "test")] | |
| assert len(alert_rows(d["uid"], d["t"], d["s"], 0.5)) == OP.TW.alerts_per_hour(d["uid"], d["t"], d["s"], 0.5)[0] | |
| thr = freeze_thresholds(S, log) | |
| # ---- alert lists per (X, regime, point) on the TEST layouts | |
| alerts = {} | |
| for X in SYSTEMS: | |
| for r in REGIMES: | |
| d = S[(X, r, "test")] | |
| for point in list(POINTS) + ["frame"]: | |
| rows = alert_rows(d["uid"], d["t"], d["s"], thr[X][r][point]["thr"]) | |
| alerts[(X, r, point)] = by_route(d["uid"][rows], d["t"][rows]) | |
| # ---- S1 union frame | |
| raw = [] | |
| for X in SYSTEMS: | |
| for r in REGIMES: | |
| for u, arr in alerts[(X, r, "frame")].items(): | |
| raw += [(u, t, X, r) for t in arr] | |
| events = merge_events(raw, MERGE_S) | |
| n_union_raw = len(events) | |
| excl, excl_counts, excl_src = exclusion_anchors(log) | |
| kept, n_excl_by = [], Counter() | |
| for e in events: | |
| if near_any(excl, e["uid"], e["t"], EXCL_S): | |
| hit = next(name for name, lst in excl_src.items() if any(u == e["uid"] and abs(t - e["t"]) <= EXCL_S for u, t in lst)) | |
| n_excl_by[hit] += 1 | |
| continue | |
| e["dongle"] = dongle_of[e["uid"]] | |
| e["pattern_cluster"] = pattern_of(e["systems"]) | |
| e["pattern"] = pattern_of({X for X in SYSTEMS if np.isfinite(first_in_window(alerts[(X, e["regime"], "frame")], e["uid"], e["t"]))}) | |
| kept.append(e) | |
| events = kept | |
| log(f"S1 union: {len(raw)} frame alerts -> {n_union_raw} events (10 s guard) -> {len(events)} after the 12.5 s exclusion ({dict(n_excl_by)} removed); " | |
| f"window-rule pattern == cluster pattern for {np.mean([e['pattern'] == e['pattern_cluster'] for e in events]):.3f} of the events") | |
| cell_counts = {r: {p: sum(1 for e in events if e["regime"] == r and e["pattern"] == p) for p in PATTERNS} for r in REGIMES} | |
| for r in REGIMES: | |
| log(f" S1 cells {r}: " + ", ".join(f"{p} {cell_counts[r][p]}" for p in PATTERNS) + f"; devices {len({e['dongle'] for e in events if e['regime'] == r})}; " | |
| f"top device {Counter(e['dongle'] for e in events if e['regime'] == r).most_common(1)}") | |
| # ---- S1 draw: cells in increasing N, strata (cell, device-group), global device cap | |
| dev_used = Counter() | |
| s1, s1_table = [], [] | |
| order = sorted([(r, p) for r in REGIMES for p in PATTERNS], key=lambda rp: (cell_counts[rp[0]][rp[1]], rp[0], rp[1])) | |
| fixed_all, planned_s1 = preallocate_capped(events, lambda e: f"{e['regime']}|{e['pattern']}", {f"{r}|{p}": min(S1_ALLOC[p], cell_counts[r][p]) for r in REGIMES for p in PATTERNS}, S1_DEV_CAP) | |
| log(f"S1 pre-allocation of over-cap devices: {len(planned_s1)} devices, {sum(fixed_all.values())} fixed draws over {len(fixed_all)} (cell, device) strata") | |
| for r, p in order: | |
| units = [e for e in events if e["regime"] == r and e["pattern"] == p] | |
| n_target = min(S1_ALLOC[p], len(units)) | |
| fixed_c = {d: v for (c, d), v in fixed_all.items() if c == f"{r}|{p}"} | |
| drawn, table, cov = stratified_draw(units, n_target, S1_DEV_CAP, dev_used, rng, fixed=fixed_c, planned=planned_s1) | |
| for u in drawn: | |
| u.update(stratum="S1", cell=f"{r}|{p}", N_cell=len(units), n_cell=len(drawn)) | |
| s1 += drawn | |
| s1_table.append(dict(regime=r, pattern=p, N_cell=len(units), n_target=S1_ALLOC[p], n=len(drawn), n_devices_cell=len({e["dongle"] for e in units}), strata=table, **cov, | |
| coverage_literal_per_device_rule=exact_device_coverage(units, n_target) if units else np.nan)) | |
| log(f" S1 draw {r:8s} {p:6s}: N {len(units):4d} -> n {len(drawn):2d} (target {S1_ALLOC[p]}); strata {len(table)}; covered {cov['N_covered']}/{len(units)} ({cov['coverage']:.2f}; uncovered groups {cov['uncovered_groups']}); " | |
| f"literal per-device rule would cover {s1_table[-1]['coverage_literal_per_device_rule']:.2f}") | |
| log(f"S1 drawn {len(s1)} clips; per regime {Counter(u['regime'] for u in s1)}; devices {len({u['dongle'] for u in s1})}; top device share {max(dev_used.values()) / max(len(s1), 1):.3f}") | |
| # ---- S2 background frame: moving rows, >= 12.5 s from every union event (pre-exclusion, i.e. every alert cluster) and every excluded anchor | |
| union_by_uid = defaultdict(list) | |
| for e in merge_events(raw, MERGE_S): | |
| union_by_uid[e["uid"]].append(e["t"]) | |
| union_by_uid = {u: np.sort(np.asarray(v)) for u, v in union_by_uid.items()} | |
| s2, s2_table, s2_counts = [], [], {} | |
| dev_used2 = Counter() | |
| s2_units = {} | |
| speed_all = {} | |
| for r in REGIMES: | |
| Xr = np.load(S12R / f"targets_{r}_test.npz", allow_pickle=True)["X"] | |
| speed_all[r] = Xr[:, -1, VEGO_CH].astype(float) # vEgo at t (context X = t-3.9 .. t at 10 Hz, last step = t) | |
| del Xr | |
| for r in REGIMES: | |
| pk = packs[(r, "test")] | |
| speed = speed_all[r] | |
| moving = np.isfinite(speed) & (speed > MOVING_MPS) | |
| ok = np.zeros(len(speed), bool) | |
| idx = np.flatnonzero(moving) | |
| for i in idx: | |
| u, t = pk["uid"][i], float(pk["t"][i]) | |
| ok[i] = not near_any(union_by_uid, u, t, EXCL_S) and not near_any(excl, u, t, EXCL_S) | |
| units = [dict(uid=pk["uid"][i], t=float(pk["t"][i]), regime=r, dongle=pk["dongle"][i], speed_anchor_mps=float(speed[i])) for i in np.flatnonzero(ok)] | |
| s2_counts[r] = dict(N_rows=int(len(speed)), N_rows_moving=int(moving.sum()), N_rows_frame=int(ok.sum()), hours=hours[r], rows_per_hour=int(round(3600 / GRID_DT))) | |
| s2_units[r] = units | |
| fixed_s2, planned_s2 = preallocate_capped([u for r in REGIMES for u in s2_units[r]], lambda u: u["regime"], {r: S2_PER_REGIME for r in REGIMES}, S2_DEV_CAP, extra="free") | |
| for r in REGIMES: | |
| units = s2_units[r] | |
| drawn, table, cov = stratified_draw(units, S2_PER_REGIME, S2_DEV_CAP, dev_used2, rng, guard=S2_GUARD_S, fixed={d: v for (c, d), v in fixed_s2.items() if c == r}, planned=planned_s2, extra_free=True) | |
| for u in drawn: | |
| u.update(stratum="S2", cell=f"{r}|silent", pattern="", N_cell=len(units), n_cell=len(drawn)) | |
| s2 += drawn | |
| s2_table.append(dict(regime=r, N_rows_frame=len(units), n=len(drawn), strata=table, n_devices_frame=len({u["dongle"] for u in units}), **cov)) | |
| log(f"S2 {r:8s}: rows {s2_counts[r]['N_rows']} moving {s2_counts[r]['N_rows_moving']} frame {s2_counts[r]['N_rows_frame']} -> drawn {len(drawn)} on {len({u['dongle'] for u in drawn})} devices; " | |
| f"covered {cov['N_covered']}/{len(units)} ({cov['coverage']:.3f})") | |
| # ---- S3: the 100 round-2 TEST clips (+ optional hard_response draw) | |
| r2 = json.load(open(ANN / "clips_r2.json")) | |
| draw = pd.read_csv(MINING / "round2_draw_v1.csv") | |
| draw_t = draw[draw.split == "test"].set_index("idx") | |
| strata = pd.read_csv(MINING / "strata_v1.csv", low_memory=False) | |
| T = strata[(strata.split == "test") & (strata.eligible == True)] # noqa: E712 | |
| n_elig = T.primary_stratum.value_counts().to_dict() | |
| mh = json.load(open(MINING / "run_mine_hard_clips_latest.json")) | |
| assert mh["quotas"]["test"] == S3_QUOTA, (mh["quotas"]["test"], S3_QUOTA) | |
| s3 = [] | |
| for c in r2["clips"]: | |
| dr = draw_t.loc[int(c["idx"])] | |
| st = str(dr.stratum) | |
| s3.append(dict(uid=c["uid"], t=float(c["t0"]), regime=str(c["regime"]), dongle=c["dongle"], stratum="S3", cell=f"{c['regime']}|{st}", pattern="", | |
| s3_stratum=st, s3_draw_rank=int(dr.draw_rank), s15_idx=int(c["idx"]), clip_id_r2=c["clip_id"], N_cell=int(n_elig[st]), n_cell=int(S3_QUOTA[st]), | |
| N_stratum=int(n_elig[st]), n_stratum=int(S3_QUOTA[st]), pi=S3_QUOTA[st] / n_elig[st], dev_group="", N_covered_cell=int(n_elig[st]), coverage_cell=1.0, dcam=c["dcam"], qcam=c["qcam"], | |
| prefix_cab=str(ANN / c["clips"]["prefix_cab"]), prefix_road=str(ANN / c["clips"]["prefix_road"]), full_cab=c["clips"]["full_cab"], full_road=c["clips"]["full_road"])) | |
| log(f"S3: {len(s3)} round-2 test clips; pi by stratum " + ", ".join(f"{k} {S3_QUOTA[k]}/{n_elig[k]}={S3_QUOTA[k] / n_elig[k]:.4f}" for k in S3_QUOTA)) | |
| s3h, s3h_info = [], {} | |
| if a.s3_hard > 0: | |
| rng_h = np.random.default_rng(S3_HARD_SEED) | |
| H = T[(T.hard_response_h5 == True) & ~T.clip_id.isin(draw.clip_id) & T.regime.isin(REGIMES)] # noqa: E712 | |
| excl_r2 = defaultdict(list) | |
| for u, t in list(zip(draw.uid, draw.t0)) + [(u, t) for lst in excl_src.values() for u, t in lst]: | |
| excl_r2[u].append(float(t)) | |
| excl_r2 = {u: np.sort(np.asarray(v)) for u, v in excl_r2.items()} | |
| H = H[[not near_any(excl_r2, u, float(t), MERGE_S) for u, t in zip(H.uid, H.t0)]] | |
| s3h_info = dict(N_eligible_hard_test=int(len(H)), by_regime=H.regime.value_counts().to_dict(), seed=S3_HARD_SEED, dev_cap=S3_HARD_DEV_CAP, quota=a.s3_hard) | |
| by_reg = {r: H[H.regime == r].sample(frac=1.0, random_state=int(rng_h.integers(0, 2**31 - 1))) for r in REGIMES} | |
| iters = {r: iter(by_reg[r].itertuples()) for r in REGIMES} | |
| alive = {r: True for r in REGIMES} | |
| devc, chosen_t, turn = Counter(), defaultdict(list), 0 | |
| while len(s3h) < a.s3_hard and any(alive.values()): | |
| r = REGIMES[turn % 2] | |
| turn += 1 | |
| if not alive[r]: | |
| continue | |
| try: | |
| row = next(iters[r]) | |
| except StopIteration: | |
| alive[r] = False | |
| continue | |
| if devc[row.dongle] >= S3_HARD_DEV_CAP or any(abs(row.t0 - x) < MERGE_S for x in chosen_t[row.uid]): | |
| continue | |
| devc[row.dongle] += 1 | |
| chosen_t[row.uid].append(float(row.t0)) | |
| s3h.append(dict(uid=row.uid, t=float(row.t0), regime=str(row.regime), dongle=row.dongle, stratum="S3_hard", cell=f"{row.regime}|hard_response", pattern="", | |
| s3_stratum="hard_response", s3_draw_rank=len(s3h), s15_idx=int(row.idx), clip_id_r2=row.clip_id, N_cell=int(len(H)), n_cell=a.s3_hard, | |
| N_stratum=int(len(H)), n_stratum=a.s3_hard, pi=a.s3_hard / len(H), dev_group="", N_covered_cell=int(len(H)), coverage_cell=1.0)) | |
| for u in s3h: | |
| u["n_cell"] = u["n_stratum"] = len(s3h) | |
| u["pi"] = len(s3h) / len(H) | |
| u["dcam"] = str(BATON_LOCAL / leaf_of(u["uid"]) / "dcamera.mp4") | |
| u["qcam"] = str(BATON_LOCAL / leaf_of(u["uid"]) / "qcamera.mp4") | |
| u["full_cab"] = str(EXPLORE / "s15_cockpit" / "annotate_clips_v2" / f"{u['s15_idx']:05d}_cab.mp4") | |
| u["full_road"] = str(EXPLORE / "s15_cockpit" / "annotate_clips_v2" / f"{u['s15_idx']:05d}_road.mp4") | |
| log(f"S3_hard (optional): eligible {len(H)} ({s3h_info['by_regime']}) -> drawn {len(s3h)}; regimes {Counter(u['regime'] for u in s3h)}") | |
| # ---- per-clip system fields (server-side only) | |
| rowsA = {r: RouteRows(S[("A", r, "test")]["uid"], S[("A", r, "test")]["t"]) for r in REGIMES} | |
| rowsC = {r: RouteRows(S[("C", r, "test")]["uid"], S[("C", r, "test")]["t"]) for r in REGIMES} | |
| clips = s1 + s2 + s3 + s3h | |
| for u in clips: | |
| r = u["regime"] | |
| reg_for_rows = r if rowsA[r].row(u["uid"], u["t"]) is not None else next((q for q in REGIMES if rowsA[q].row(u["uid"], u["t"]) is not None), r) | |
| u["layout_regime"] = reg_for_rows | |
| i = rowsA[reg_for_rows].row(u["uid"], u["t"]) | |
| j = rowsC[reg_for_rows].row(u["uid"], u["t"]) | |
| u["score_A_anchor"] = float(S[("A", reg_for_rows, "test")]["s"][i]) if i is not None else np.nan | |
| u["score_B_anchor"] = float(S[("B", reg_for_rows, "test")]["s"][i]) if i is not None else np.nan | |
| u["score_C_anchor"] = float(S[("C", reg_for_rows, "test")]["s"][j]) if j is not None else np.nan | |
| if "speed_anchor_mps" not in u: | |
| u["speed_anchor_mps"] = float(speed_all[reg_for_rows][i]) if i is not None else np.nan | |
| for X in SYSTEMS: | |
| for point in list(POINTS) + ["frame"]: | |
| ta = first_in_window(alerts[(X, reg_for_rows, point)], u["uid"], u["t"]) | |
| u[f"t_alert_{X}_{point}"] = ta | |
| if point != "frame": | |
| u[f"fired_{X}_{point}"] = bool(np.isfinite(ta)) | |
| u[f"thr_{X}_frame"] = thr[X][reg_for_rows]["frame"]["thr"] | |
| u["video_local"] = video_local(u["uid"]) | |
| u["leaf"] = leaf_of(u["uid"]) | |
| u["dongle"] = str(u["dongle"]) | |
| u.setdefault("dcam", str(BATON_LOCAL / u["leaf"] / "dcamera.mp4")) | |
| u.setdefault("qcam", str(BATON_LOCAL / u["leaf"] / "qcamera.mp4")) | |
| for u in s1: # the window-rule pattern equals the systems with a frame alert in [t-10, t+10] (same regime layout) | |
| assert u["layout_regime"] == u["regime"] and u["pattern"] == pattern_of({X for X in SYSTEMS if np.isfinite(u[f"t_alert_{X}_frame"])}), (u["uid"], u["t"], u["pattern"]) | |
| assert all(np.isfinite(u[f"t_alert_{X}_frame"]) and u[f"t_alert_{X}_frame"] - u["t"] <= ALERT_WIN_S + 1e-6 for X in u["systems"]) | |
| # ---- ids, phases, calibration, queue order | |
| for k, (stratum, lst) in enumerate((("S1", s1), ("S2", s2))): | |
| lst.sort(key=lambda u: (u["regime"], u["cell"], u["uid"], u["t"])) | |
| perm = rng.permutation(len(lst)) | |
| for rank, j in enumerate(perm): | |
| lst[j]["idx"] = IDX_BASE[stratum] + rank | |
| for u in s3 + s3h: | |
| u["idx"] = u["s15_idx"] | |
| for u in clips: | |
| u["clip_id"] = opaque_id(u["uid"], u["t"]) if u["stratum"] in ("S1", "S2") else u["clip_id_r2"] | |
| u["anchor_s"] = float(min(5.0, u["t"])) | |
| u["needs_frames"] = u["stratum"] in ("S1", "S2", "S3_hard") or not Path(u.get("prefix_cab", "/nonexistent")).exists() | |
| u["optional"] = u["stratum"] == "S3_hard" | |
| # phase 1 = S3 + the first 60 S1 clips (regime-interleaved random order); phase 2 = remaining S1, S2, optional S3_hard | |
| s1_by_r = {r: [u for u in s1 if u["regime"] == r] for r in REGIMES} | |
| for r in REGIMES: | |
| s1_by_r[r] = [s1_by_r[r][j] for j in rng.permutation(len(s1_by_r[r]))] | |
| inter = [u for pair in zip(*[s1_by_r[r] for r in REGIMES]) for u in pair] | |
| inter += [u for r in REGIMES for u in s1_by_r[r] if u not in inter] | |
| for k, u in enumerate(inter): | |
| u["s1_order"] = k | |
| u["phase"] = 1 if k < N_PHASE1_S1 else 2 | |
| for u in s3: | |
| u["phase"] = 1 | |
| for u in s2 + s3h: | |
| u["phase"] = 2 | |
| # calibration: 4 per stratum (S1 phase-1, S2, S3), 2 per regime each -> 12, flagged (rated by every rater + owner before Phase 1) | |
| for u in clips: | |
| u["calibration"] = False | |
| for st, lst in (("S1", inter[:N_PHASE1_S1]), ("S2", s2), ("S3", s3)): | |
| for r in REGIMES: | |
| cand = [u for u in lst if u["regime"] == r and not u.get("optional")] | |
| for j in rng.permutation(len(cand))[: CALIB[st][r]]: | |
| cand[j]["calibration"] = True | |
| # queue: calibration first, then phase 1 (shuffled), then phase 2 (shuffled) | |
| calib = [u for u in clips if u["calibration"]] | |
| p1 = [u for u in clips if u["phase"] == 1 and not u["calibration"]] | |
| p2 = [u for u in clips if u["phase"] == 2 and not u["calibration"]] | |
| queue = [calib[j] for j in rng.permutation(len(calib))] + [p1[j] for j in rng.permutation(len(p1))] + [p2[j] for j in rng.permutation(len(p2))] | |
| for q, u in enumerate(queue): | |
| u["qpos"] = q | |
| # ---- manifest | |
| cols = ["qpos", "idx", "clip_id", "stratum", "phase", "calibration", "optional", "regime", "layout_regime", "uid", "dongle", "leaf", "t_anchor", "anchor_s", "cell", "pattern", "pattern_cluster", "dev_group", | |
| "pi", "N_cell", "n_cell", "N_covered_cell", "coverage_cell", "N_stratum", "n_stratum", "s3_stratum", "s3_draw_rank", "s15_idx", "speed_anchor_mps", "video_local", "needs_frames"] | |
| cols += [f"fired_{X}_{p}" for X in SYSTEMS for p in POINTS] + [f"t_alert_{X}_{p}" for X in SYSTEMS for p in list(POINTS) + ["frame"]] | |
| cols += [f"score_{X}_anchor" for X in SYSTEMS] + [f"thr_{X}_frame" for X in SYSTEMS] + ["dcam", "qcam", "prefix_cab", "prefix_road", "full_cab", "full_road"] | |
| rows = [] | |
| for u in queue: | |
| rec = {c: u.get(c, np.nan) for c in cols} | |
| rec["t_anchor"] = u["t"] | |
| rows.append(rec) | |
| M = pd.DataFrame(rows, columns=cols) | |
| assert M.idx.is_unique and M.clip_id.is_unique | |
| assert ((M.pi > 0) & (M.pi <= 1)).all() | |
| M.to_csv(man_p, index=False) | |
| # ---- frame counts + thresholds json | |
| fc = dict(version=a.version, seed=a.seed, generated_utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), hours_test=hours, | |
| n_rows_test={r: int(len(packs[(r, 'test')]['uid'])) for r in REGIMES}, n_routes_test={r: int(len(np.unique(packs[(r, 'test')]['uid']))) for r in REGIMES}, | |
| n_devices_test={r: int(len(np.unique(packs[(r, 'test')]['dongle']))) for r in REGIMES}, | |
| S1=dict(frame_alerts=len(raw), union_events_before_exclusion=n_union_raw, excluded=dict(n_excl_by), union_events=len(events), exclusion_anchor_counts=excl_counts, | |
| exclusion_window_s=EXCL_S, merge_s=MERGE_S, alloc_per_regime=S1_ALLOC, dev_cap=S1_DEV_CAP, cells=cell_counts, drawn=len(s1), | |
| drawn_per_regime={r: sum(1 for u in s1 if u["regime"] == r) for r in REGIMES}, devices_frame=len({e["dongle"] for e in events}), devices_drawn=len({u["dongle"] for u in s1}), | |
| top_device_frame=Counter(e["dongle"] for e in events).most_common(3), top_device_drawn_share=max(dev_used.values()) / max(len(s1), 1), table=s1_table, | |
| frame_alerts_by_system={X: {r: int(sum(len(v) for v in alerts[(X, r, 'frame')].values())) for r in REGIMES} for X in SYSTEMS}, | |
| video_local_frame_share=float(np.mean([video_local(e["uid"]) for e in events])) if events else np.nan), | |
| S2=dict(counts=s2_counts, per_regime=S2_PER_REGIME, dev_cap=S2_DEV_CAP, guard_s=S2_GUARD_S, moving_mps=MOVING_MPS, drawn=len(s2), table=s2_table), | |
| S3=dict(n=len(s3), quota=S3_QUOTA, N_eligible=n_elig, by_regime=Counter(u["regime"] for u in s3), by_stratum=Counter(u["s3_stratum"] for u in s3), devices=len({u["dongle"] for u in s3}), | |
| video_local=int(sum(u["video_local"] for u in s3)), prefix_cuts_exist=int(sum(Path(u["prefix_cab"]).exists() for u in s3))), | |
| S3_hard=dict(s3h_info, drawn=len(s3h)), | |
| phases={f"phase{k}": dict(Counter(u["stratum"] for u in clips if u["phase"] == k)) for k in (1, 2)}, calibration=[dict(idx=u["idx"], stratum=u["stratum"], regime=u["regime"]) for u in calib], | |
| video_local_by_stratum={s: dict(n=int(sum(1 for u in clips if u["stratum"] == s)), local=int(sum(u["video_local"] for u in clips if u["stratum"] == s)), | |
| routes=len({u["uid"] for u in clips if u["stratum"] == s}), routes_local=len({u["uid"] for u in clips if u["stratum"] == s and u["video_local"]})) for s in ("S1", "S2", "S3", "S3_hard")}, | |
| fired_share_drawn={f"{X}_{p}": {s: float(np.mean([u[f'fired_{X}_{p}'] for u in clips if u['stratum'] == s])) if any(u['stratum'] == s for u in clips) else np.nan for s in ("S1", "S2", "S3")} for X in SYSTEMS for p in POINTS}) | |
| 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)) | |
| layout_sha = {str(p): sha256_file(p) for p in files} | |
| tj = dict(version=a.version, generated_utc=fc["generated_utc"], systems=dict(A=dict(runs=[RUN_A.format(seed=s) for s in SEEDS_A], col=COL_A, rule="mean over seeds"), B=dict(run=RUN_B, col=COL_B), | |
| C=dict(test=f"{DMS_DIR}/marginals_<regime>_s0.npz -> compose.legacy_compose(pA, None, {TAU})['att_only']", val=f"{DMS_DIR}/marginals_<regime>_val_s0.npz (score_val_marginals.py: frozen s0 chain checkpoints on the val pack) -> same compose")), | |
| cooldown_s=COOLDOWN, bisection="operating_points.thr_for_rate (24 iterations) on alerts/h of train_warning_head.alerts_per_hour", points=POINTS, hours_test=hours, | |
| hours_val={r: float(len(packs[(r, 'val')]['uid']) * GRID_DT / 3600.0) for r in REGIMES}, thresholds=thr, frame_rule="min(thr_val20, thr_test20)", | |
| not_reachable=[f"{X}/{r}/{p}" for X in SYSTEMS for r in REGIMES for p in POINTS if not thr[X][r][p]["reachable"]], layout_sha256=layout_sha, dms_fields_used=[]) | |
| json.dump(tj, open(thr_p, "w"), indent=1) | |
| write_run_json(out, "newtest_sample", [Path(__file__), Path(OP.__file__), Path(OP.TW.__file__), Path(sys.modules["compose"].__file__)], | |
| files + [MINING / "round2_draw_v1.csv", MINING / "strata_v1.csv", ANN / "clips_h5.json", ANN / "clips_r2.json"], seed=a.seed, t_start=t_start, | |
| extra=dict(version=a.version, outputs=[str(man_p), str(fc_p), str(thr_p)], n_clips=len(M), by_stratum=dict(Counter(M.stratum)), not_reachable=tj["not_reachable"], | |
| log=lines, dms_fields_used=[])) | |
| log(f"DONE in {time.time() - t_start:.0f}s: {len(M)} clips ({dict(Counter(M.stratum))}) -> {man_p}, {fc_p}, {thr_p}") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 51.9 kB
- Xet hash:
- 64a20c6ab9affb8e55511c32edbfd3a32d306452feafdaa15d5e966c184ddbce
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.