Buckets:
| """cwm3/weak/mine_hard_clips.py — clip-mining strata index v1 and the proposed round-2 human-labelling draw (plan §5.4). | |
| Pool (one row per clip): | |
| TRAIN the 26,578 train-device clips of cwm2/private/clip_map_full.json (15 s windows, anchor t0 = 5 s into the clip) | |
| VAL/TEST the held-out v2 episodes of cwm2/private/heldout_episodes.json (idx -> s15_cockpit/annotate_episodes_v2.json: uid, dongle, | |
| t0, regime, ctrl only — the episodes' `dm` block is a DMS field and is NEVER read here); clip_id = the cwm2 opaque id | |
| (verified against clip_map_heldout.json). Held-out clips are listed for a HUMAN draw only: never sent to any VLM / cloud. | |
| Split = split_of(dongle) (s3 anchors.parquet map, md5 fallback), asserted equal to the pool's own split. | |
| Strata flags per clip, computed on the 10 Hz grid window (prefix [t0-5, t0], horizon (t0, t0+5], response window up to t0+8): | |
| road pressure P_<fam>_h5 (cut_in, lead_braking, lane_excursion, ttc, ttc_drop, badas) from data/build_pressure_targets.route_pressure | |
| old demand D_ego_h5 (steer_rate | decel), D_any_h5 from cwm2/data/build_demand_targets.route_derived (label side, CAN only) | |
| demand/response demand_any_h5 = D_any | P_any in the horizon (or active at t0); response = R >= 1 (build_response_targets, steer | |
| rate lagged 2 samples) within 3 s after the first demand onset -> demand_resp / demand_noresp; hard_response_h5 | |
| driver events hand_leave / hand_return / head_turn in (t0, t0+5]: CAUSAL kinematics where the clip has a row in | |
| results/targets_kin_causal (kin_source = causal_clip), else the old non-causal cwm2 kin at the nearest s12 anchor | |
| within 0.3 s of t0 (kin_source = old_noncausal, flagged), else none | |
| DMS alert TRAIN clips only, from results/targets_dms_teacher/clip_dms_flags_train.csv (derived flags written by the single DMS | |
| reader; this file reads no DMS field). NaN for val/test: held-out draws use data-derived strata only. | |
| night/occlusion route_clock_hour from date-named route ids (85/718 routes; NaN otherwise) -> night_proxy (hour < 6 or >= 19); | |
| occlusion proxies from the causal roi.json mirror (train clips only): co_occupant_present, driver_side_uncertain, | |
| hands_missing_n. Image luminance is NOT available without frames (roi.json / index.json carry no pixel statistics; | |
| frames live in remote shards) -> skipped, stated in the table. | |
| model_disagreement NaN column: needs a train-row inference pass of the cwm3 branches (GPU, later) | |
| active_score TRAIN only, data-derived: sum over binary strata flags of -log(prevalence_in_train_pool), normalised to [0,1] | |
| Round-2 draw (results/mining/round2_draw_v1.csv): train 400 / val 100 / test 100, stratified by primary stratum with fixed quotas, | |
| regime-alternating, device cap (train 12 / held-out 8), same-event guard 10 s (also against every excluded clip), seed 0; test/val | |
| from data-derived strata only (no DMS stratum, no model score). Excluded: s15 annotations_v2.json (all labelled episodes, 4,650), | |
| annotations_v2_cross.json, the 400-clip round-1 H5 queue (cwm2/annotation_human/clips_h5.json), the 187-clip pilot | |
| (cwm2/private/clip_map.json) and the 200 held-out gold clips (cwm2/private/clip_map_heldout.json). | |
| Outputs: results/mining/strata_v1.csv, results/mining/round2_draw_v1.csv, results/tables/mining_v1.md, run json. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import os | |
| import re | |
| import sys | |
| import time | |
| from collections import Counter, defaultdict | |
| from pathlib import Path | |
| os.environ.setdefault("OMP_NUM_THREADS", "8") | |
| import numpy as np # noqa: E402 | |
| import pandas as pd # noqa: E402 | |
| ROOT = Path(__file__).resolve().parents[1] | |
| EXPLORE = ROOT.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| sys.path.insert(0, str(ROOT / "common")) | |
| sys.path.insert(0, str(ROOT / "data")) | |
| sys.path.insert(0, str(CWM2 / "data")) | |
| from runmeta import write_run_json # noqa: E402 | |
| import build_pressure_targets as bp # noqa: E402 | |
| import build_response_targets as br # noqa: E402 | |
| import build_demand_targets as bd # noqa: E402 (cwm2 label-side demand: CAN + R block only, DMS block asserted unread) | |
| OUT = ROOT / "results" / "mining" | |
| TABLES = ROOT / "results" / "tables" | |
| CLIP_MAP = CWM2 / "private" / "clip_map_full.json" | |
| CLIP_MAP_HELD = CWM2 / "private" / "clip_map_heldout.json" | |
| CLIP_MAP_PILOT = CWM2 / "private" / "clip_map.json" | |
| HELD_EPS = CWM2 / "private" / "heldout_episodes.json" | |
| EPS_V2 = EXPLORE / "s15_cockpit" / "annotate_episodes_v2.json" | |
| ANN_V2 = EXPLORE / "s15_cockpit" / "annotations_v2.json" | |
| ANN_CROSS = EXPLORE / "s15_cockpit" / "annotations_v2_cross.json" | |
| H5 = CWM2 / "annotation_human" / "clips_h5.json" | |
| ANCHORS = EXPLORE / "s3_can_baseline" / "results" / "anchors.parquet" | |
| LEAVES = ROOT / "results" / "baton_leaves.json" | |
| ROI = ROOT / "results" / "roi" | |
| KIN_C = ROOT / "results" / "targets_kin_causal" | |
| KIN_OLD = CWM2 / "results" / "targets_kin" | |
| DMS_FLAGS = ROOT / "results" / "targets_dms_teacher" / "clip_dms_flags_train.csv" | |
| S12 = EXPLORE / "s12_wm" / "results" | |
| FS, HOR, RESP_WIN = 10.0, 50, 30 | |
| QUOTA = {"train": {"rare_road": 80, "demand_noresp": 60, "demand_resp": 60, "hand_head": 80, "dms_alert": 60, "occlusion_night": 20, "background": 40}, | |
| "val": {"rare_road": 25, "demand_noresp": 20, "demand_resp": 20, "hand_head": 15, "occlusion_night": 5, "background": 15}, | |
| "test": {"rare_road": 25, "demand_noresp": 20, "demand_resp": 20, "hand_head": 15, "occlusion_night": 5, "background": 15}} | |
| PRIORITY = ["rare_road", "demand_noresp", "demand_resp", "hand_head", "dms_alert", "occlusion_night", "background"] | |
| DEV_CAP = {"train": 12, "val": 8, "test": 8} | |
| MERGE_S = 10.0 | |
| SCORE_FLAGS = ["rare_road_h5", "demand_noresp", "demand_resp", "hand_head_event", "dms_alert_any", "hard_response_h5", "night_proxy", "P_cut_in_h5", "P_lead_braking_h5", "P_lane_excursion_h5"] | |
| DATE_PAT = re.compile(r"__(\d{4})-(\d{2})-(\d{2})--(\d{2})-(\d{2})-(\d{2})__") | |
| def opaque_id(uid, t0, salt="cwm2-full-v1"): | |
| """cwm2 opaque clip id (build_full_manifest.py:19; the held-out manifest uses salt 'cwm2-heldout-v1', build_heldout_manifest.py:30).""" | |
| return "c" + hashlib.sha1(f"{uid}|{t0:.3f}|{salt}".encode()).hexdigest()[:12] | |
| def split_of_fn(): | |
| a = pd.read_parquet(ANCHORS, columns=["dongle", "split"]).drop_duplicates("dongle") | |
| known = dict(zip(a.dongle, a.split)) | |
| def f(d): | |
| if d in known: | |
| return known[d] | |
| h = int(hashlib.md5(d.encode()).hexdigest(), 16) % 100 | |
| return "train" if h < 60 else ("val" if h < 70 else "test") | |
| return f | |
| def leaf_of(uid): | |
| p = uid.split("__") | |
| return "/".join(p[:3]) + "/" + "__".join(p[3:]) | |
| def build_pool(): | |
| split_of = split_of_fn() | |
| cm = json.load(open(CLIP_MAP)) | |
| rows = [] | |
| for cid, c in cm.items(): | |
| sp = split_of(c["dongle"]) | |
| assert sp == "train", (cid, sp) | |
| rows.append(dict(clip_id=cid, idx=c["idx"], uid=c["uid"], dongle=c["dongle"], t0=float(c["t0"]), regime=c["regime"], ctrl=c["ctrl"], | |
| split="train", pool="clip_map_full", pool_family=c.get("family"), pool_stratum=c.get("stratum"), | |
| existing_label=c.get("label"), existing_label_protocol=c.get("label_protocol"))) | |
| eps = {e["idx"]: e for e in json.load(open(EPS_V2))} | |
| held = json.load(open(HELD_EPS)) | |
| hm = json.load(open(CLIP_MAP_HELD)) | |
| for h in held: | |
| e = eps[h["idx"]] | |
| sp = split_of(e["dongle"]) | |
| assert sp == h["split"], (h, sp) | |
| t0 = round(float(e["t0"]), 3) | |
| cid = opaque_id(e["uid"], t0, "cwm2-heldout-v1") | |
| rows.append(dict(clip_id=cid, idx=e["idx"], uid=e["uid"], dongle=e["dongle"], t0=t0, regime=e["regime"], ctrl=(e.get("ctrl") or {}).get("source"), | |
| split=sp, pool="heldout_episodes", pool_family=None, pool_stratum=None, existing_label=None, existing_label_protocol=None)) | |
| df = pd.DataFrame(rows) | |
| # the opaque id rule must reproduce the held-out gold manifest ids | |
| ok = sum(1 for cid, c in hm.items() if opaque_id(c["uid"], round(float(c["t0"]), 3), "cwm2-heldout-v1") == cid) | |
| assert ok == len(hm), (ok, len(hm)) | |
| assert df.clip_id.is_unique | |
| return df | |
| def exclusions(df): | |
| ann = set(int(k) for k in json.load(open(ANN_V2))) | |
| cross = set(int(str(k).split("|")[0]) for k in json.load(open(ANN_CROSS))) # keys are "<idx>|<group>" | |
| h5 = set(c["idx"] for c in json.load(open(H5))["clips"]) | |
| pilot = set(c["idx"] for c in json.load(open(CLIP_MAP_PILOT)).values()) | |
| heldg = set(c["idx"] for c in json.load(open(CLIP_MAP_HELD)).values()) | |
| reason = np.full(len(df), "", object) | |
| for name, s in (("labelled_v2", ann), ("labelled_cross", cross), ("h5_round1_queue", h5), ("pilot_187", pilot), ("heldout_gold_200", heldg)): | |
| m = df.idx.isin(s).to_numpy() & (reason == "") | |
| reason[m] = name | |
| return reason, dict(labelled_v2=len(ann), labelled_cross=len(cross), h5_round1_queue=len(h5), pilot_187=len(pilot), heldout_gold_200=len(heldg)) | |
| def route_cache_entry(uid): | |
| G, tg = bp.load_grid(uid) | |
| rp = bp.route_pressure(G) | |
| rr = br.route_response(G) | |
| rd = bd.route_derived(uid) # cwm2 label-side demand (route quantiles, 6 families incl. steer_rate/decel) | |
| assert rd["T"] == rp["T"] == rr["T"] | |
| fam_d = list(bd.FAMILIES) | |
| ego = rd["fam"][[fam_d.index("steer_rate"), fam_d.index("decel")]].any(0) | |
| d_any = (rd["fam"] & rd["valid"]).any(0) | |
| return dict(t0=tg[0], T=rp["T"], P=rp["fam"], Pv=rp["valid"], mv=rp["mv"], R=rr["R"], Rv=rr["Rvalid"], D_ego=ego, D_any=d_any, | |
| radar=rp["radar_present"]) | |
| def kin_lookup(): | |
| """clip_id -> (E[25,3], EM[25,3], S0[2], Kvalid_frac) from the causal clip-window targets (row with |dt_anchor| minimal); | |
| and per (regime, split): sorted anchor times + old kin arrays for the nearest-anchor fallback.""" | |
| causal = {} | |
| for reg in ("manual", "engaged"): | |
| p = KIN_C / f"kin_causal_{reg}_train.npz" | |
| if not p.exists(): | |
| continue | |
| k = np.load(p, allow_pickle=True) | |
| ok = np.flatnonzero(k["row_ok"]) | |
| cid, dt = k["clip_id"][ok], np.abs(k["dt_anchor"][ok]) | |
| order = np.lexsort((dt, cid)) | |
| E, EM, S0, KV = k["E_c"], k["EM_c"], k["S0_c"], k["Kvalid"] | |
| seen = set() | |
| for j in order: | |
| r = ok[j] | |
| c = cid[j] | |
| if c in seen: | |
| continue | |
| seen.add(c) | |
| causal[c] = (E[r], EM[r], S0[r], float(KV[r].mean()), float(dt[j])) | |
| old = {} | |
| for reg in ("manual", "engaged"): | |
| for sp in ("train", "val", "test"): | |
| p = KIN_OLD / f"kin_{reg}_{sp}.npz" | |
| if not p.exists(): | |
| continue | |
| k = np.load(p, allow_pickle=True) | |
| by = defaultdict(list) | |
| for i, (u, t) in enumerate(zip(k["uid"], k["t"])): | |
| by[u].append((float(t), i)) | |
| old[(reg, sp)] = (dict((u, np.array(sorted(v))) for u, v in by.items()), k["E"], k["EM"], k["S0"], k["Kvalid"]) | |
| return causal, old | |
| def nearest_old(old, regimes, sp, uid, t0): | |
| best = None | |
| for reg in regimes: | |
| ent = old.get((reg, sp)) | |
| if ent is None or uid not in ent[0]: | |
| continue | |
| arr = ent[0][uid] | |
| j = np.searchsorted(arr[:, 0], t0) | |
| for jj in (j - 1, j): | |
| if 0 <= jj < len(arr): | |
| d = abs(arr[jj, 0] - t0) | |
| if d <= 0.3 and (best is None or d < best[0]): | |
| best = (d, reg, int(arr[jj, 1])) | |
| if best is None: | |
| return None | |
| _, reg, i = best | |
| ent = old[(reg, sp)] | |
| return ent[1][i], ent[2][i], ent[3][i], float(ent[4][i].mean()), best[0] | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--seed", type=int, default=0) | |
| ap.add_argument("--force", action="store_true") | |
| a = ap.parse_args() | |
| OUT.mkdir(parents=True, exist_ok=True) | |
| TABLES.mkdir(parents=True, exist_ok=True) | |
| t_start = time.time() | |
| df = build_pool() | |
| reason, excl_sizes = exclusions(df) | |
| df["excluded_reason"] = reason | |
| leaves = json.load(open(LEAVES))["leaves"] | |
| df["video_local"] = [leaves.get(leaf_of(u), {}).get("local_status") == "local_identical_size" for u in df.uid] | |
| hours = [] | |
| for u in df.uid: | |
| m = DATE_PAT.search(u) | |
| hours.append(int(m.group(4)) if m else np.nan) | |
| df["route_clock_hour"] = hours | |
| df["night_proxy"] = np.where(np.isfinite(df.route_clock_hour), (df.route_clock_hour < 6) | (df.route_clock_hour >= 19), np.nan) | |
| # ---- grid-derived flags | |
| cache = {} | |
| fams = bp.FAMILIES | |
| cols = {f"P_{f}_h5": [] for f in fams} | |
| cols.update({f"P0_{f}": [] for f in fams}) | |
| extra = defaultdict(list) | |
| for u, t0 in zip(df.uid, df.t0): | |
| if u not in cache: | |
| cache[u] = route_cache_entry(u) if (bp.GRID / f"{u}.npz").exists() else None | |
| rd = cache[u] | |
| if rd is None: | |
| for f in fams: | |
| cols[f"P_{f}_h5"].append(np.nan) | |
| cols[f"P0_{f}"].append(np.nan) | |
| for k in ("window_in_grid", "moving_at_anchor", "D_ego_h5", "D_any_h5", "demand_any_h5", "demand_at_anchor", "response_after_demand", | |
| "demand_resp", "demand_noresp", "hard_response_h5", "resp_any_h5", "radar_present", "grid_available"): | |
| extra[k].append(np.nan if k != "grid_available" else False) | |
| continue | |
| k0 = int(round((t0 - rd["t0"]) * FS)) | |
| T = rd["T"] | |
| inr = 0 <= k0 < T | |
| hz = slice(k0 + 1, min(k0 + HOR + 1, T)) | |
| win_ok = bool(k0 - 50 >= 0 and k0 + HOR + RESP_WIN < T) | |
| extra["grid_available"].append(True) | |
| extra["window_in_grid"].append(win_ok) | |
| extra["moving_at_anchor"].append(bool(rd["mv"][k0]) if inr else False) | |
| extra["radar_present"].append(bool(rd["radar"])) | |
| P_h = rd["P"][:, hz].any(1) if inr else np.zeros(len(fams), bool) | |
| for f_i, f in enumerate(fams): | |
| cols[f"P_{f}_h5"].append(bool(P_h[f_i])) | |
| cols[f"P0_{f}"].append(bool(rd["P"][f_i, k0]) if inr else False) | |
| p_any = rd["P"].any(0) | |
| dem = (p_any | rd["D_any"]) | |
| extra["D_ego_h5"].append(bool(rd["D_ego"][hz].any()) if inr else False) | |
| extra["D_any_h5"].append(bool(rd["D_any"][hz].any()) if inr else False) | |
| d_h = bool(dem[hz].any()) if inr else False | |
| d_0 = bool(dem[k0]) if inr else False | |
| extra["demand_any_h5"].append(d_h or d_0) | |
| extra["demand_at_anchor"].append(d_0) | |
| R1 = (rd["R"] >= 1) & rd["Rv"] | |
| resp_after = np.nan | |
| if inr and (d_h or d_0): | |
| if d_0: | |
| j = k0 | |
| else: | |
| idx = np.flatnonzero(dem[hz] & ~np.r_[dem[k0], dem[hz][:-1]]) | |
| j = k0 + 1 + int(idx[0]) if len(idx) else k0 + 1 + int(np.flatnonzero(dem[hz])[0]) | |
| resp_after = bool(R1[j + 1:min(j + RESP_WIN + 1, T)].any()) | |
| extra["response_after_demand"].append(resp_after) | |
| extra["demand_resp"].append(bool(resp_after is True)) | |
| extra["demand_noresp"].append(bool(resp_after is False)) | |
| extra["hard_response_h5"].append(bool(((rd["R"] >= 2) & rd["Rv"])[hz].any()) if inr else False) | |
| extra["resp_any_h5"].append(bool(R1[hz].any()) if inr else False) | |
| for k, v in cols.items(): | |
| df[k] = v | |
| for k, v in extra.items(): | |
| df[k] = v | |
| df["rare_road_h5"] = df.P_cut_in_h5.fillna(False).astype(bool) | df.P_lead_braking_h5.fillna(False).astype(bool) | df.P_lane_excursion_h5.fillna(False).astype(bool) | |
| # ---- driver events | |
| causal, old = kin_lookup() | |
| ev = {k: [] for k in ("hand_return_h5", "hand_leave_h5", "head_turn_h5", "hands_off_at_anchor", "head_turned_at_anchor", "kin_source", "kin_valid_frac", "kin_anchor_dt")} | |
| for cid, u, t0, reg, sp in zip(df.clip_id, df.uid, df.t0, df.regime, df.split): | |
| hit = causal.get(cid) | |
| src = "causal_clip" | |
| if hit is None: | |
| regs = ("manual", "engaged") if reg == "mixed" else (reg,) | |
| hit = nearest_old(old, regs, sp, u, t0) | |
| src = "old_noncausal" if hit is not None else "none" | |
| if hit is None: | |
| for k in ev: | |
| ev[k].append(np.nan if k != "kin_source" else "none") | |
| continue | |
| E, EM, S0, kv, dt = hit | |
| on = (E.astype(bool) & EM).any(0) | |
| ev["hand_return_h5"].append(bool(on[0])) | |
| ev["hand_leave_h5"].append(bool(on[1])) | |
| ev["head_turn_h5"].append(bool(on[2])) | |
| ev["hands_off_at_anchor"].append(bool(np.isfinite(S0[0]) and S0[0] < 0.5)) | |
| ev["head_turned_at_anchor"].append(bool(np.isfinite(S0[1]) and S0[1] > 0.5)) | |
| ev["kin_source"].append(src) | |
| ev["kin_valid_frac"].append(kv) | |
| ev["kin_anchor_dt"].append(dt) | |
| for k, v in ev.items(): | |
| df[k] = v | |
| df["hand_head_event"] = df.hand_leave_h5.fillna(False).astype(bool) | df.hand_return_h5.fillna(False).astype(bool) | df.head_turn_h5.fillna(False).astype(bool) | |
| # ---- roi occlusion proxies (train clips only) | |
| roi = {k: [] for k in ("co_occupant_present", "driver_side_uncertain", "hands_missing_n", "roi_selection_rule")} | |
| for cid, sp in zip(df.clip_id, df.split): | |
| p = ROI / f"{cid}.json" | |
| if sp != "train" or not p.exists(): | |
| roi["co_occupant_present"].append(np.nan) | |
| roi["driver_side_uncertain"].append(np.nan) | |
| roi["hands_missing_n"].append(np.nan) | |
| roi["roi_selection_rule"].append(None) | |
| continue | |
| r = json.load(open(p)) | |
| roi["co_occupant_present"].append(bool(r.get("co_occupant_present"))) | |
| roi["driver_side_uncertain"].append(bool(r.get("driver_side_uncertain"))) | |
| roi["hands_missing_n"].append(len(r.get("hands_missing") or [])) | |
| roi["roi_selection_rule"].append(r.get("selection_rule")) | |
| for k, v in roi.items(): | |
| df[k] = v | |
| # ---- DMS flags (train only; derived CSV of the single DMS reader) | |
| fl = pd.read_csv(DMS_FLAGS)[["clip_id", "dms_alert_any", "dms_alert_distract", "dms_alert_gaze", "dms_valid_frac"]] | |
| df = df.merge(fl, on="clip_id", how="left") | |
| assert df.loc[df.split != "train", "dms_alert_any"].isna().all() | |
| df["model_disagreement"] = np.nan # needs a train-row inference pass of the cwm3 branches (GPU) — left NaN in v1 | |
| # ---- primary stratum (flags may be bool, float 0/1 or NaN after merges: _t is the NaN-safe truth test) | |
| def _t(x): | |
| try: | |
| return bool(x == 1) or x is True | |
| except Exception: | |
| return False | |
| def stratum(r): | |
| if _t(r.rare_road_h5): | |
| return "rare_road" | |
| if _t(r.demand_noresp): | |
| return "demand_noresp" | |
| if _t(r.demand_resp): | |
| return "demand_resp" | |
| if _t(r.hand_head_event): | |
| return "hand_head" | |
| if r.split == "train" and _t(r.dms_alert_any): | |
| return "dms_alert" | |
| if _t(r.night_proxy) or (r.split == "train" and (_t(r.co_occupant_present) or (isinstance(r.hands_missing_n, (int, float)) and r.hands_missing_n > 0))): | |
| return "occlusion_night" | |
| return "background" | |
| df["primary_stratum"] = [stratum(r) for r in df.itertuples()] | |
| df.loc[df.grid_available != True, "primary_stratum"] = "no_grid" | |
| # ---- active score (train only, data-derived) | |
| tr = df.split == "train" | |
| prev = {} | |
| score = np.zeros(len(df)) | |
| wsum = 0.0 | |
| for f in SCORE_FLAGS: | |
| col = df[f].fillna(False).astype(bool).to_numpy() | |
| p = float(col[tr].mean()) | |
| prev[f] = p | |
| w = -np.log(max(p, 1e-6)) if 0 < p < 1 else 0.0 | |
| wsum += w | |
| score += w * col | |
| df["active_score"] = np.where(tr, score / max(wsum, 1e-9), np.nan) | |
| df["n_flags"] = sum(df[f].fillna(False).astype(bool).astype(int) for f in SCORE_FLAGS) | |
| # ---- eligibility + same-event guard against excluded clips | |
| excl_by_uid = defaultdict(list) | |
| for u, t0 in zip(df.uid[df.excluded_reason != ""], df.t0[df.excluded_reason != ""]): | |
| excl_by_uid[u].append(t0) | |
| near_excl = np.array([any(abs(t0 - x) < MERGE_S for x in excl_by_uid.get(u, ())) for u, t0 in zip(df.uid, df.t0)]) | |
| df["near_excluded_10s"] = near_excl | |
| df["eligible"] = (df.excluded_reason == "") & ~near_excl & (df.grid_available == True) & (df.window_in_grid == True) & (df.moving_at_anchor == True) & df.regime.isin(["manual", "engaged"]) | |
| # ---- draw | |
| rng = np.random.default_rng(a.seed) | |
| draw_rows, shortfalls = [], {} | |
| for sp in ("train", "val", "test"): | |
| quotas = dict(QUOTA[sp]) | |
| chosen_uid_t = defaultdict(list) | |
| dev_count = Counter() | |
| pool_sp = df[(df.split == sp) & df.eligible] | |
| picked = set() | |
| def take(stratum, n, pool): | |
| got = 0 | |
| cand = pool[(pool.primary_stratum == stratum) & ~pool.clip_id.isin(picked)] | |
| by_reg = {r: cand[cand.regime == r].sample(frac=1.0, random_state=int(rng.integers(0, 2**31 - 1))) for r in ("manual", "engaged")} | |
| iters = {r: iter(by_reg[r].itertuples()) for r in by_reg} | |
| alive = {r: True for r in by_reg} | |
| turn = 0 | |
| while got < n and any(alive.values()): | |
| r = ("manual", "engaged")[turn % 2] | |
| turn += 1 | |
| if not alive[r]: | |
| continue | |
| try: | |
| row = next(iters[r]) | |
| except StopIteration: | |
| alive[r] = False | |
| continue | |
| if dev_count[row.dongle] >= DEV_CAP[sp]: | |
| continue | |
| if any(abs(row.t0 - x) < MERGE_S for x in chosen_uid_t[row.uid]): | |
| continue | |
| picked.add(row.clip_id) | |
| chosen_uid_t[row.uid].append(row.t0) | |
| dev_count[row.dongle] += 1 | |
| got += 1 | |
| draw_rows.append(dict(split=sp, stratum=stratum, quota_stratum=stratum, clip_id=row.clip_id, idx=row.idx, uid=row.uid, dongle=row.dongle, t0=row.t0, | |
| regime=row.regime, ctrl=row.ctrl, active_score=row.active_score, kin_source=row.kin_source, | |
| needs_frames=(sp != "train"), video_local=row.video_local, rare_road_h5=row.rare_road_h5, demand_resp=row.demand_resp, | |
| demand_noresp=row.demand_noresp, hand_head_event=row.hand_head_event, dms_alert_any=row.dms_alert_any, | |
| night_proxy=row.night_proxy, co_occupant_present=row.co_occupant_present, n_flags=row.n_flags)) | |
| return got | |
| for st, n in quotas.items(): | |
| got = take(st, n, pool_sp) | |
| if got < n: | |
| shortfalls[f"{sp}/{st}"] = n - got | |
| short = sum(shortfalls.get(f"{sp}/{st}", 0) for st in quotas) | |
| if short: | |
| extra_got = take("background", short, pool_sp) | |
| shortfalls[f"{sp}/background_refill"] = extra_got | |
| draw = pd.DataFrame(draw_rows) | |
| draw.insert(0, "draw_rank", np.arange(1, len(draw) + 1)) | |
| # ---- write | |
| p_strata, p_draw = OUT / "strata_v1.csv", OUT / "round2_draw_v1.csv" | |
| for p in (p_strata, p_draw): | |
| if p.exists() and not a.force: | |
| raise FileExistsError(p) | |
| df.to_csv(p_strata, index=False) | |
| draw.to_csv(p_draw, index=False) | |
| # ---- table | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| flag_cols = ["rare_road_h5", "P_cut_in_h5", "P_lead_braking_h5", "P_lane_excursion_h5", "P_ttc_h5", "P_ttc_drop_h5", "P_badas_h5", "D_ego_h5", "demand_any_h5", | |
| "demand_resp", "demand_noresp", "hard_response_h5", "hand_head_event", "hand_leave_h5", "hand_return_h5", "head_turn_h5", "dms_alert_any", | |
| "night_proxy", "co_occupant_present", "eligible"] | |
| L = ["# Clip-mining strata index v1 and round-2 draw", "", | |
| f"Generated by `weak/mine_hard_clips.py` (seed {a.seed}, {stamp}); files `results/mining/strata_v1.csv` (one row per clip), `results/mining/round2_draw_v1.csv`. " | |
| "All numbers are script output.", "", | |
| "## Pool", "", "| split | clips | routes | dongles | grid available | window in grid | moving at anchor | eligible for draw |", "|---|---|---|---|---|---|---|---|"] | |
| for sp in ("train", "val", "test"): | |
| d = df[df.split == sp] | |
| L.append(f"| {sp} | {len(d)} | {d.uid.nunique()} | {d.dongle.nunique()} | {int((d.grid_available == True).sum())} | {int((d.window_in_grid == True).sum())} | " | |
| f"{int((d.moving_at_anchor == True).sum())} | {int(d.eligible.sum())} |") | |
| L += ["", "Exclusions (idx-level; the pool row keeps the reason): " + ", ".join(f"{k} {v}" for k, v in excl_sizes.items()) + ".", "", | |
| "| split | " + " | ".join(f"excl:{k}" for k in ["labelled_v2", "labelled_cross", "h5_round1_queue", "pilot_187", "heldout_gold_200"]) + " | near_excluded_10s (not itself excluded) |", | |
| "|---|" + "---|" * 6] | |
| for sp in ("train", "val", "test"): | |
| d = df[df.split == sp] | |
| L.append(f"| {sp} | " + " | ".join(str(int((d.excluded_reason == k).sum())) for k in ["labelled_v2", "labelled_cross", "h5_round1_queue", "pilot_187", "heldout_gold_200"]) + | |
| f" | {int((d.near_excluded_10s & (d.excluded_reason == '')).sum())} |") | |
| L += ["", "## Strata flag prevalence (share of clips with grid; horizon (t0, t0+5]; NaN-safe: missing counted as False)", "", | |
| "| flag | " + " | ".join(("train", "val", "test")) + " |", "|---|---|---|---|"] | |
| for f in flag_cols: | |
| vals = [] | |
| for sp in ("train", "val", "test"): | |
| d = df[(df.split == sp) & (df.grid_available == True)] | |
| vals.append(f"{d[f].fillna(False).astype(bool).mean():.4f}") | |
| L.append(f"| {f} | " + " | ".join(vals) + " |") | |
| L += ["", "kin_source by split (driver events): "] | |
| L += ["| split | causal_clip | old_noncausal | none |", "|---|---|---|---|"] | |
| for sp in ("train", "val", "test"): | |
| c = df[df.split == sp].kin_source.value_counts() | |
| L.append(f"| {sp} | {int(c.get('causal_clip', 0))} | {int(c.get('old_noncausal', 0))} | {int(c.get('none', 0))} |") | |
| L += ["", "## Primary stratum (priority " + " > ".join(PRIORITY) + "; DMS stratum train only)", "", "| stratum | train | val | test | train eligible | val eligible | test eligible |", "|---|---|---|---|---|---|---|"] | |
| for st in PRIORITY + ["no_grid"]: | |
| L.append(f"| {st} | " + " | ".join(str(int(((df.split == sp) & (df.primary_stratum == st)).sum())) for sp in ("train", "val", "test")) + " | " + | |
| " | ".join(str(int(((df.split == sp) & (df.primary_stratum == st) & df.eligible).sum())) for sp in ("train", "val", "test")) + " |") | |
| L += ["", "## Active score (train only, data-derived)", "", "score = sum over flags of -log(prevalence) x flag, normalised by the sum of weights; flags and train prevalences:", ""] | |
| L += ["| flag | train prevalence | weight -log p |", "|---|---|---|"] + [f"| {f} | {prev[f]:.4f} | {(-np.log(max(prev[f], 1e-6)) if 0 < prev[f] < 1 else 0):.3f} |" for f in SCORE_FLAGS] | |
| at = df[tr].active_score | |
| L += ["", f"train active_score: mean {at.mean():.4f}, P50 {at.quantile(0.5):.4f}, P90 {at.quantile(0.9):.4f}, P99 {at.quantile(0.99):.4f}; clips with >= 2 flags: {int((df[tr].n_flags >= 2).sum())}.", "", | |
| "`model_disagreement` is NaN for every row: it needs a train-row inference pass of the cwm3 branches (GPU) and is filled in a later version; " | |
| "no model score of ours touches any val/test row here.", "", | |
| "## Round-2 draw (train 400 / val 100 / test 100; seed 0; device cap train 12 / held-out 8; same-event guard 10 s; regime alternating)", "", | |
| "| split | stratum | n | manual | engaged | devices | routes | mean active_score (train) |", "|---|---|---|---|---|---|---|---|"] | |
| for sp in ("train", "val", "test"): | |
| d = draw[draw.split == sp] | |
| for st in QUOTA[sp]: | |
| e = d[d.stratum == st] | |
| L.append(f"| {sp} | {st} | {len(e)} | {int((e.regime == 'manual').sum())} | {int((e.regime == 'engaged').sum())} | {e.dongle.nunique()} | {e.uid.nunique()} | " | |
| f"{(e.active_score.mean() if sp == 'train' else float('nan')):.4f} |") | |
| L.append(f"| {sp} | **total** | {len(d)} | {int((d.regime == 'manual').sum())} | {int((d.regime == 'engaged').sum())} | {d.dongle.nunique()} | {d.uid.nunique()} | " | |
| f"{(d.active_score.mean() if sp == 'train' else float('nan')):.4f} |") | |
| L += ["", "Quota shortfalls (filled from background): " + (", ".join(f"{k} {v}" for k, v in shortfalls.items()) if shortfalls else "none") + ".", "", | |
| f"kin_source in the draw: " + ", ".join(f"{sp}: {dict(draw[draw.split == sp].kin_source.value_counts())}" for sp in ("train", "val", "test")) + ".", | |
| f"video_local in the draw: " + ", ".join(f"{sp} {int(draw[draw.split == sp].video_local.sum())}/{int((draw.split == sp).sum())}" for sp in ("train", "val", "test")) + | |
| " (held-out clips have no frame package yet: `needs_frames = True`; extraction from the canonical remote copy is required before labelling).", "", | |
| "## Notes / infeasible items", "", | |
| "- Night / occlusion: image mean luminance was NOT computed — roi.json and index.json carry no pixel statistics and the frames live in the remote shards; " | |
| f"`night_proxy` uses the clock hour of date-named route ids only ({int(np.isfinite(df.route_clock_hour).sum())} of {len(df)} clips have one), occlusion proxies come from the causal roi.json mirror (train clips only).", | |
| "- DMS alert stratum: train clips only (`clip_dms_flags_train.csv` from the single DMS reader); val/test rows have NaN and are drawn from data-derived strata only.", | |
| "- Driver events on val/test come from the old non-causal kinematics (`kin_source = old_noncausal`, flagged): no causal pose exists on held-out routes yet.", | |
| "- `demand_any_h5` unions the old D (incl. its ego-motion families) and the new road-pressure P; `demand_noresp` = no R >= 1 within 3 s after the first demand onset.", | |
| "- Held-out clips are listed for the HUMAN draw only and must never be sent to a VLM / cloud service.", ""] | |
| (TABLES / "mining_v1.md").write_text("\n".join(L)) | |
| inputs = [CLIP_MAP, CLIP_MAP_HELD, CLIP_MAP_PILOT, HELD_EPS, EPS_V2, ANN_V2, ANN_CROSS, H5, ANCHORS, LEAVES, DMS_FLAGS, bp.GRID_SRC] | |
| inputs += [KIN_C / f"kin_causal_{r}_train.npz" for r in ("manual", "engaged") if (KIN_C / f"kin_causal_{r}_train.npz").exists()] | |
| inputs += [KIN_OLD / f"kin_{r}_{s}.npz" for r in ("manual", "engaged") for s in ("train", "val", "test") if (KIN_OLD / f"kin_{r}_{s}.npz").exists()] | |
| write_run_json(OUT, "mine_hard_clips", [Path(__file__), Path(bp.__file__), Path(br.__file__), Path(bd.__file__)], inputs, a.seed, t_start, | |
| extra=dict(files_written=[str(p_strata), str(p_draw), str(TABLES / "mining_v1.md")], quotas=QUOTA, priority=PRIORITY, dev_cap=DEV_CAP, | |
| merge_s=MERGE_S, score_flags=SCORE_FLAGS, train_prevalence=prev, exclusions=excl_sizes, shortfalls=shortfalls, | |
| n_pool=int(len(df)), n_draw=int(len(draw)), n_routes_grid=len(cache), dms_fields_used=[], | |
| dms_flags_source=str(DMS_FLAGS), model_disagreement="NaN (needs train-row inference pass, GPU)")) | |
| print("\n".join(L)) | |
| print("MINING_DONE", f"{time.time() - t_start:.0f}s") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 31.6 kB
- Xet hash:
- 8b21ed9b5af39b446ffb63e4cabf8e065d7c9d4637d05a66ca0a4b96445185cc
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.