Buckets:
| """cwm3/weak/dms_teacher_targets.py — DMS weak-teacher SOFT TARGETS on the s12 anchor grid (plan §5.2), TRAIN and VAL rows only. | |
| *** THIS FILE (and, later, weak/dms_teacher_pretrain.py) IS THE ONLY cwm3 CODE ALLOWED TO READ openpilot DMS FIELDS. *** | |
| It reads grid10 columns 0-6 (face_yaw, face_pitch, face_roll, blink, eye, isDistracted, faceProb; written by | |
| s14_corr/dstate.route_dstate from driver_state.csv, already faceProb<0.5 -> NaN and >1 s joint-constant holds -> NaN at | |
| 20 Hz) plus the whitelisted vEgo column for the moving gate. Everything it writes is a TARGET / SELECTION artefact: | |
| `dms_lineage: target_only` in feature_provenance.yaml; models/ train/ eval/ data/ never import weak/ (tests/test_no_dms_inputs.py). | |
| Test rows are never written (asserted). The DMS is openpilot's own driver-monitoring net: using it as a teacher for the | |
| driver branch is a disclosed weak-supervision step, never a test-time input (plan §0). | |
| Per route (10 Hz): | |
| valid = finite(face_yaw) & faceProb >= 0.5 & not stale10 | |
| stale10 = (face_yaw, face_pitch) identical to the previous sample for >= STALE_N = 3 consecutive samples (>= 0.3 s): | |
| repeated identical angles are export/hold artefacts, not measurements (dstate catches only >1 s joint holds) | |
| channels K = 6: face_yaw [rad], face_pitch [rad], eye_open (= min(L,R) eye-open prob), blink (= max(L,R) blink prob), | |
| distracted_soft = mean of isDistracted over the centred 1.1 s window of valid samples (>= 6 of 11 valid), | |
| gaze_off_soft = mean of (|face_yaw| > 0.4 rad) over the same window (mine_anomalies dms_gaze rule) | |
| w_route = valid fraction over MOVING samples x (1 - stale fraction) (route reliability weight; 0 when no DMS file) | |
| Anchors: T0[N,K] f32 at t (NaN invalid), T0valid[N]; T[N,25,K] f16 at t+0.2..t+5.0, Tvalid[N,25]; Th[N,20,K] f16 history | |
| t-3.8..t @ 5 Hz (incl. t), Thvalid[N,20]; w_route[N] f32; norm_mean/norm_std[K] from the TRAIN valid samples of both | |
| regimes (for the pretraining loss); uid, t, dongle, channels. -> results/targets_dms_teacher/dms_{regime}_{train,val}.npz | |
| Clip flags (train-device clips of cwm2/private/clip_map_full.json, for clip MINING only — selection, never input): | |
| window [t0-5, t0+10] on the grid: dms_alert_distract = sustained(isDistracted >= 0.5 & moving, 1.5 s), | |
| dms_alert_gaze = sustained(|face_yaw| > 0.4 & moving, 1.5 s) (mine_anomalies SUSTAIN 15 samples), dms_alert_any, | |
| dms_valid_frac, dms_present -> results/targets_dms_teacher/clip_dms_flags_train.csv | |
| run.json records dms_fields_used and every grid10 column index read. Table: results/tables/dms_teacher.md. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import sys | |
| import time | |
| 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 | |
| sys.path.insert(0, str(ROOT / "common")) | |
| sys.path.insert(0, str(ROOT / "data")) | |
| from runmeta import write_run_json # noqa: E402 | |
| import build_pressure_targets as bp # noqa: E402 (grid10 layout recovery, anchor grid, gather helpers; CAN-only there) | |
| OUT = ROOT / "results" / "targets_dms_teacher" | |
| TABLES = ROOT / "results" / "tables" | |
| CLIP_MAP = EXPLORE / "cwm2" / "private" / "clip_map_full.json" | |
| ANCHORS = EXPLORE / "s3_can_baseline" / "results" / "anchors.parquet" | |
| DMS_FIELDS = ["face_yaw", "face_pitch", "face_roll", "blink", "eye", "isDistracted", "faceProb"] | |
| DMS_IDX = [bp.CI[c] for c in DMS_FIELDS] | |
| assert DMS_IDX == list(range(7)), DMS_IDX # the DMS block is grid10 columns 0-6 | |
| VEGO = bp.CI["vEgo"] | |
| CHANNELS = ["face_yaw", "face_pitch", "eye_open", "blink", "distracted_soft", "gaze_off_soft"] | |
| K = len(CHANNELS) | |
| FACEPROB_MIN, STALE_N, GAZE_YAW, SOFT_WIN, SOFT_MIN_VALID = 0.5, 3, 0.4, 11, 6 | |
| SUSTAIN_N = 15 # 1.5 s (s15_cockpit/mine_anomalies.SUSTAIN distract/gaze) | |
| FS, H, HIST = 10.0, 25, 20 | |
| SPLITS_ALLOWED = ("train", "val") | |
| def runs_ge(b, n): | |
| """True on every sample of a run of True of length >= n (vectorised analyze_safety.sustained).""" | |
| b = np.asarray(b, bool) | |
| if not b.any(): | |
| return np.zeros_like(b) | |
| d = np.diff(np.r_[0, b.astype(np.int8), 0]) | |
| starts, ends = np.flatnonzero(d == 1), np.flatnonzero(d == -1) | |
| out = np.zeros_like(b) | |
| for s, e in zip(starts, ends): | |
| if e - s >= n: | |
| out[s:e] = True | |
| return out | |
| def moving_mean(x, valid, win, min_valid): | |
| """centred window mean over valid samples; NaN where fewer than min_valid valid.""" | |
| xv = np.where(valid, x, 0.0) | |
| k = np.ones(win) | |
| num = np.convolve(xv, k, mode="same") | |
| den = np.convolve(valid.astype(float), k, mode="same") | |
| out = np.full(len(x), np.nan) | |
| ok = den >= min_valid | |
| out[ok] = num[ok] / den[ok] | |
| return out | |
| def route_dms(G): | |
| T = G.shape[0] | |
| D = G[:, :7].astype(np.float64) | |
| yaw, pitch, blink, eye, dis, fp = D[:, 0], D[:, 1], D[:, 3], D[:, 4], D[:, 5], np.nan_to_num(D[:, 6], nan=0.0) | |
| v = G[:, VEGO].astype(np.float64) | |
| mv = np.isfinite(v) & (v > bp.MOVING_MS) | |
| fin = np.isfinite(yaw) & np.isfinite(pitch) | |
| same = np.zeros(T, bool) | |
| same[1:] = fin[1:] & fin[:-1] & (yaw[1:] == yaw[:-1]) & (pitch[1:] == pitch[:-1]) | |
| r = runs_ge(same, STALE_N - 1) # STALE_N identical samples = STALE_N-1 consecutive "same" flags | |
| stale = r.copy() | |
| starts = r & ~np.r_[False, r[:-1]] | |
| stale[:-1] |= starts[1:] # the sample before the run is the first of the identical values | |
| valid = fin & (fp >= FACEPROB_MIN) & ~stale | |
| present = bool(np.isfinite(D[:, 6]).any() and fin.any()) | |
| dsoft = moving_mean(np.nan_to_num(dis, nan=0.0), valid & np.isfinite(dis), SOFT_WIN, SOFT_MIN_VALID) | |
| gsoft = moving_mean((np.abs(np.nan_to_num(yaw, nan=0.0)) > GAZE_YAW).astype(float), valid, SOFT_WIN, SOFT_MIN_VALID) | |
| X = np.stack([yaw, pitch, eye, blink, dsoft, gsoft], 1).astype(np.float32) | |
| X[~valid] = np.nan | |
| n_mv = int(mv.sum()) | |
| valid_frac_mv = float(valid[mv].mean()) if n_mv else 0.0 | |
| stale_frac = float(stale[fin].mean()) if fin.any() else 0.0 | |
| w = valid_frac_mv * (1.0 - stale_frac) if present else 0.0 | |
| alert_dis = runs_ge((np.nan_to_num(dis, nan=0.0) >= 0.5) & valid & mv, SUSTAIN_N) | |
| alert_gaze = runs_ge((np.abs(np.nan_to_num(yaw, nan=0.0)) > GAZE_YAW) & valid & mv, SUSTAIN_N) | |
| return dict(T=T, X=X, valid=valid, mv=mv, w=float(w), present=present, valid_frac_mv=valid_frac_mv, stale_frac=stale_frac, | |
| alert_dis=alert_dis, alert_gaze=alert_gaze, dis_rate=float(np.nanmean(dis[valid])) if valid.any() else float("nan")) | |
| def gather_dms(rd, k): | |
| T = rd["T"] | |
| kic, inr = bp.gather_steps(None, k, T) | |
| k0ok = (k >= 0) & (k < T) | |
| k0c = np.clip(k, 0, T - 1) | |
| hk = k[:, None] - (np.arange(HIST)[::-1] * 2)[None, :] # t-3.8 .. t @ 5 Hz | |
| hin = (hk >= 0) & (hk < T) | |
| hkc = np.clip(hk, 0, T - 1) | |
| Tf = rd["X"][kic] | |
| Tf[~inr] = np.nan | |
| Th = rd["X"][hkc] | |
| Th[~hin] = np.nan | |
| T0 = rd["X"][k0c] | |
| T0[~k0ok] = np.nan | |
| return dict(T0=T0.astype(np.float32), T0valid=rd["valid"][k0c] & k0ok, T=Tf.astype(np.float16), Tvalid=rd["valid"][kic] & inr, | |
| Th=Th.astype(np.float16), Thvalid=rd["valid"][hkc] & hin, w_route=np.full(len(k), rd["w"], np.float32), | |
| Moving0=rd["mv"][k0c] & k0ok) | |
| def split_of_map(): | |
| a = pd.read_parquet(ANCHORS, columns=["dongle", "split"]).drop_duplicates("dongle") | |
| return dict(zip(a.dongle, a.split)) | |
| def clip_flags(cache, log): | |
| import hashlib | |
| cm = json.load(open(CLIP_MAP)) | |
| known = split_of_map() | |
| rows = [] | |
| for cid, c in cm.items(): | |
| d = c["dongle"] | |
| sp = known.get(d) | |
| if sp is None: | |
| h = int(hashlib.md5(d.encode()).hexdigest(), 16) % 100 | |
| sp = "train" if h < 60 else ("val" if h < 70 else "test") | |
| assert sp == "train", (cid, d, sp) # DMS flags are computed for TRAIN-device clips only | |
| u = c["uid"] | |
| if u not in cache: | |
| G, tg = bp.load_grid(u) | |
| rd = route_dms(G) | |
| rd["t0"] = tg[0] | |
| cache[u] = rd | |
| rd = cache[u] | |
| k0 = int(round((c["t0"] - rd["t0"]) * FS)) | |
| lo, hi = max(0, k0 - 50), min(rd["T"], k0 + 101) | |
| w = slice(lo, hi) | |
| n = hi - lo | |
| rows.append(dict(clip_id=cid, idx=c["idx"], uid=u, dongle=d, t0=c["t0"], dms_present=rd["present"], | |
| dms_valid_frac=float(rd["valid"][w].mean()) if n else float("nan"), | |
| dms_alert_distract=bool(rd["alert_dis"][w].any()), dms_alert_gaze=bool(rd["alert_gaze"][w].any()), | |
| dms_distracted_frac=float(np.nanmean(np.where(rd["valid"][w], rd["X"][w, 4], np.nan))) if rd["valid"][w].any() else float("nan"), | |
| window_in_grid=bool(k0 - 50 >= 0 and k0 + 100 < rd["T"]), w_route=rd["w"])) | |
| df = pd.DataFrame(rows) | |
| df["dms_alert_any"] = df.dms_alert_distract | df.dms_alert_gaze | |
| return df | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regimes", default="manual,engaged") | |
| ap.add_argument("--splits", default="train,val") | |
| ap.add_argument("--force", action="store_true") | |
| ap.add_argument("--no-clip-flags", action="store_true") | |
| a = ap.parse_args() | |
| for s in a.splits.split(","): | |
| assert s in SPLITS_ALLOWED, f"split {s} refused: DMS teacher targets are written for train/val rows only" | |
| OUT.mkdir(parents=True, exist_ok=True) | |
| TABLES.mkdir(parents=True, exist_ok=True) | |
| t_start = time.time() | |
| cache, stats, files = {}, {}, [] | |
| # TRAIN normalisation stats (both regimes) are computed from the train files' valid anchors after the first pass | |
| built = {} | |
| for regime in a.regimes.split(","): | |
| for split in a.splits.split(","): | |
| d, wa = bp.build_split(regime, split, cache, [], route_fn=route_dms, gather_fn=gather_dms) | |
| src = np.load(bp.S12 / f"targets_{regime}_{split}.npz", allow_pickle=True) | |
| assert np.array_equal(src["uid"], d["uid"]) and np.array_equal(src["t"], d["t"]) | |
| built[(regime, split)] = d | |
| uids = np.unique(d["uid"]) | |
| pres = np.array([cache[u]["present"] for u in uids]) | |
| T0v, Tv, T0 = d["T0valid"], d["Tvalid"], d["T0"] | |
| X = T0[T0v] | |
| stats[f"{regime}_{split}"] = dict(N=int(len(d["t"])), align=wa, n_routes=int(len(uids)), n_routes_with_dms=int(pres.sum()), | |
| T0valid_frac=float(T0v.mean()), Tvalid_frac_steps=float(Tv.mean()), Tvalid_all25_frac=float(Tv.all(1).mean()), | |
| Thvalid_all20_frac=float(d["Thvalid"].all(1).mean()), | |
| w_route_mean=float(d["w_route"].mean()), w_route_pct=[float(x) for x in np.percentile(d["w_route"], [10, 50, 90])], | |
| stale_frac_routes_mean=float(np.mean([cache[u]["stale_frac"] for u in uids])), | |
| valid_frac_moving_routes_mean=float(np.mean([cache[u]["valid_frac_mv"] for u in uids])), | |
| yaw_abs_mean=float(np.nanmean(np.abs(X[:, 0]))) if len(X) else float("nan"), | |
| pitch_mean=float(np.nanmean(X[:, 1])) if len(X) else float("nan"), | |
| eye_open_lt05_rate=float(np.nanmean(X[:, 2] < 0.5)) if len(X) else float("nan"), | |
| blink_gt05_rate=float(np.nanmean(X[:, 3] > 0.5)) if len(X) else float("nan"), | |
| distracted_soft_mean=float(np.nanmean(X[:, 4])) if len(X) else float("nan"), | |
| distracted_soft_gt05_rate=float(np.nanmean(X[:, 4] > 0.5)) if len(X) else float("nan"), | |
| gaze_off_soft_mean=float(np.nanmean(X[:, 5])) if len(X) else float("nan"), | |
| gaze_off_soft_gt05_rate=float(np.nanmean(X[:, 5] > 0.5)) if len(X) else float("nan")) | |
| print(f"[{regime}/{split}] N={len(d['t'])} routes_with_dms={pres.sum()}/{len(uids)} T0valid={T0v.mean():.3f} Tvalid={Tv.mean():.3f} " | |
| f"w_route mean={d['w_route'].mean():.3f} distracted_soft>0.5={stats[f'{regime}_{split}']['distracted_soft_gt05_rate']:.4f} " | |
| f"gaze_off>0.5={stats[f'{regime}_{split}']['gaze_off_soft_gt05_rate']:.4f}", flush=True) | |
| tr = [built[k]["T0"][built[k]["T0valid"]] for k in built if k[1] == "train"] | |
| if tr: | |
| Xtr = np.concatenate(tr) | |
| norm_mean, norm_std = np.nanmean(Xtr, 0).astype(np.float32), np.nanstd(Xtr, 0).astype(np.float32) | |
| else: | |
| norm_mean, norm_std = np.zeros(K, np.float32), np.ones(K, np.float32) | |
| for (regime, split), d in built.items(): | |
| d.update(channels=np.array(CHANNELS), norm_mean=norm_mean, norm_std=norm_std, norm_source=np.array("train anchors, both regimes, valid T0")) | |
| p = OUT / f"dms_{regime}_{split}.npz" | |
| bp.write_npz(p, d, a.force) | |
| files.append(str(p)) | |
| stats["norm"] = dict(channels=CHANNELS, mean=norm_mean.tolist(), std=norm_std.tolist()) | |
| if not a.no_clip_flags: | |
| df = clip_flags(cache, print) | |
| pf = OUT / "clip_dms_flags_train.csv" | |
| if pf.exists() and not a.force: | |
| raise FileExistsError(pf) | |
| df.to_csv(pf, index=False) | |
| files.append(str(pf)) | |
| stats["clip_flags_train"] = dict(n_clips=int(len(df)), dms_present_frac=float(df.dms_present.mean()), dms_valid_frac_mean=float(df.dms_valid_frac.mean()), | |
| alert_distract_frac=float(df.dms_alert_distract.mean()), alert_gaze_frac=float(df.dms_alert_gaze.mean()), | |
| alert_any_frac=float(df.dms_alert_any.mean()), window_in_grid_frac=float(df.window_in_grid.mean())) | |
| print("[clip flags]", stats["clip_flags_train"], flush=True) | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| (OUT / f"build_stats_{stamp}.json").write_text(json.dumps(stats, indent=1)) | |
| (OUT / "build_stats_latest.json").write_text(json.dumps(stats, indent=1)) | |
| # ---- table | |
| keys = [k for k in stats if "_" in k and k not in ("clip_flags_train",) and isinstance(stats[k], dict) and "N" in stats[k]] | |
| cols = ["N", "n_routes", "n_routes_with_dms", "T0valid_frac", "Tvalid_frac_steps", "Tvalid_all25_frac", "Thvalid_all20_frac", "w_route_mean", | |
| "stale_frac_routes_mean", "yaw_abs_mean", "pitch_mean", "eye_open_lt05_rate", "blink_gt05_rate", "distracted_soft_mean", | |
| "distracted_soft_gt05_rate", "gaze_off_soft_mean", "gaze_off_soft_gt05_rate"] | |
| lines = ["# DMS weak-teacher targets — coverage and base rates", "", | |
| f"Generated by `weak/dms_teacher_targets.py` ({stamp}); every number is script output (`results/targets_dms_teacher/build_stats_{stamp}.json`).", | |
| "Rows = s12 anchors of TRAIN / VAL only (test rows are never written). `T0valid` = DMS valid at the anchor (faceProb >= 0.5, finite, not stale);", | |
| f"`stale` = identical (yaw, pitch) for >= {STALE_N} consecutive 10 Hz samples; soft channels = centred {SOFT_WIN / FS:.1f} s means over valid samples.", | |
| "DMS fields read (grid10 columns 0-6): " + ", ".join(DMS_FIELDS) + "; plus vEgo (moving gate). This file is the ONLY DMS reader in cwm3.", "", | |
| "| regime_split | " + " | ".join(cols) + " |", "|---|" + "---|" * len(cols)] | |
| for k in keys: | |
| s = stats[k] | |
| lines.append(f"| {k} | " + " | ".join(f"{s[c]:.4f}" if isinstance(s[c], float) else str(s[c]) for c in cols) + " |") | |
| lines += ["", f"Normalisation (train anchors, both regimes): mean {np.round(norm_mean, 4).tolist()}, std {np.round(norm_std, 4).tolist()} for {CHANNELS}.", ""] | |
| if "clip_flags_train" in stats: | |
| c = stats["clip_flags_train"] | |
| lines += ["## Clip-level DMS flags (26,578 TRAIN-device clips; clip mining only)", "", | |
| f"| n_clips | dms_present | dms_valid_frac (mean) | alert_distract (sustained >= 1.5 s) | alert_gaze (|yaw| > 0.4 rad >= 1.5 s) | alert_any | window_in_grid |", | |
| "|---|---|---|---|---|---|---|", | |
| f"| {c['n_clips']} | {c['dms_present_frac']:.4f} | {c['dms_valid_frac_mean']:.4f} | {c['alert_distract_frac']:.4f} | {c['alert_gaze_frac']:.4f} | {c['alert_any_frac']:.4f} | {c['window_in_grid_frac']:.4f} |", ""] | |
| lines += ["Firewall: `run_dms_teacher_targets_<utc>.json` lists `dms_fields_used`; tests/test_no_dms_inputs.py asserts models/ train/ eval/ data/ never import weak/ and that the DMS tokens appear in no other cwm3 file.", ""] | |
| (TABLES / "dms_teacher.md").write_text("\n".join(lines)) | |
| inputs = [bp.S12 / f"targets_{r}_{s}.npz" for r in a.regimes.split(",") for s in a.splits.split(",")] + [bp.GRID_SRC, CLIP_MAP] | |
| inputs += sorted(bp.GRID / f"{u}.npz" for u in cache) | |
| write_run_json(OUT, "dms_teacher_targets", [Path(__file__)], inputs, 0, t_start, | |
| extra=dict(files_written=files, dms_fields_used=DMS_FIELDS, grid_columns_read=[(c, bp.CI[c]) for c in DMS_FIELDS + ["vEgo"]], | |
| splits_written=a.splits.split(","), test_rows_written=False, channels=CHANNELS, | |
| constants=dict(FACEPROB_MIN=FACEPROB_MIN, STALE_N=STALE_N, GAZE_YAW=GAZE_YAW, SOFT_WIN=SOFT_WIN, SOFT_MIN_VALID=SOFT_MIN_VALID, | |
| SUSTAIN_N=SUSTAIN_N), n_routes=len(cache), table=str(TABLES / "dms_teacher.md"), | |
| this_file_is_the_only_dms_reader=True)) | |
| print("DMS_TEACHER_DONE", f"{time.time() - t_start:.0f}s", flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 17.9 kB
- Xet hash:
- 0c4e3b16eb930682a982132cafecd59f21084eec4c5c30cc1e5d02a81e098689
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.