Buckets:
| """cwm3/data/build_pressure_targets.py — ROAD-PRESSURE chains P[N,25,6] on the s12 anchor grid (plan §3 E branch, §5.3). | |
| P is the road-side half of the old demand D (cwm2/data/build_demand_targets.py): it keeps the three D families that are | |
| defined purely by the radar / lane / BADAS channels (ttc, ttc_drop, badas — identical definitions, so the old evaluators | |
| transfer) and adds three road-event families (cut_in, lead_braking, lane_excursion). It contains NO ego-motion family | |
| (no steer_rate, no decel, no pedal): what the ego vehicle does is the RESPONSE R (build_response_targets.py), and the | |
| P-vs-D overlap table (data/report_targets_pr.py) quantifies how often the old demand fired on ego motion alone. | |
| Source: s14_corr/cache/grid10/<uid>.npz {t[N], G[N,32] f32} at 10 Hz. The column layout is recovered by AST-parsing | |
| grid_s14.py (the module is never imported — it imports the DMS reader). Every column read is listed in READ_COLS and | |
| asserted to lie past the DMS block (0–6) and the pose block (7–11): only canspec whitelist CAN columns (vEgo, lane_*, | |
| lead_*) and the R block (badas_risk, ttc, lane_off) are touched. Never reads grid10 columns 0–11. | |
| Whitelist note: lead_dRel / lead_vRel / lead_vLead / lead_valid derive from radar.csv leadOne_*, lane_* from planning.csv | |
| laneLeft_y / laneRight_y (probabilities as masks only), vEgo from vehicle_dynamics.csv — all through canspec.can_features | |
| (the grid10 C block is that function's output). vEgo is used ONLY as the moving gate (> 2 m/s, as every D family) and as | |
| the normaliser of time headway = dRel / max(vEgo, 1); it never defines a family by itself. | |
| Families (order fixed; thresholds documented in docs/decisions.md with the val sensitivity sweep of --sensitivity): | |
| ttc ttc < 4 s (= D ttc; grid ttc = dRel/max(-vRel,0.1), NaN unless lead & approaching) | |
| ttc_drop dttc < -1 s/s & ttc < 8 s (= D ttc_drop; backward 0.2 s difference -> causal at its sample) | |
| cut_in a NEW relevant lead: lead_valid 1 at i after >= 0.5 s without a lead, or the tracked lead's dRel drops by | |
| > 5 m in one 0.1 s sample (a same-target closure of 50 m/s is impossible -> target switch), with time headway | |
| dRel/max(vEgo,1) < 2.5 s; the flag is HELD for 1.0 s (10 samples) so the 0.2 s target grid cannot miss it | |
| lead_braking the SAME tracked lead (lead_valid over the last 1.0 s, no dRel jump > 5 m) decelerates: backward 1.0 s | |
| difference of lead_vLead < -2.0 m/s^2, with time headway < 4 s (val noise floor of that difference on a | |
| steadily tracked lead: P5 -1.25 / P2 -2.0 / P1 -2.8 m/s^2 — a 0.5 s window at -1.5 sat inside the noise) | |
| lane_excursion lane_valid & 2.5 <= lane_width <= 5.0 m & (lane_width/2 - |lane_off|) < 0.9 m (ego edge at/over a line; | |
| 0.9 m = half of a 1.8 m vehicle). lane_valid needs both line probabilities >= 0.5, so crossings during which | |
| openpilot loses a line are invalid, not negative — under-detection is disclosed, not hidden | |
| badas badas_risk > max(0.85, route P95 of moving badas_risk) (= D badas; external forward-risk net) | |
| All families gated by moving = isfinite(vEgo) & vEgo > 2 m/s (not-moving = valid negative, `Moving` stored). | |
| Route quantile (badas only) uses the route's own full signal — a LABEL definition, as in D. | |
| Outputs results/targets_pressure/pressure_{regime}_{split}.npz, row-aligned with s12_wm/results/targets_{regime}_{split}.npz | |
| (uid, t asserted equal; same rows as cwm2/results/targets_demand): | |
| P[N,25,6] bool pressure-high at t+0.2..t+5.0 Pvalid[N,25,6] bool signal present (InRange applied) | |
| P0[N,6] bool pressure-high at the anchor t itself P0valid[N,6] | |
| P_first[N,6] int16 first onset step (0..24) given inactive & valid at t, -1 none, -2 = P0 invalid/active | |
| S0[N,6] f32 continuous margin ratio at t (>1 fired; diagnostic current-signal baseline) | |
| M0[N,10] f32 current margins at t (features): ttc_s (30 = none), headway_s (20 = none), dRel_m (200 = none), | |
| vRel_mps (0 = none), aLead_mps2 (0 = undefined), lat_margin_m (2 = lane invalid), lane_width_m (0), | |
| badas (0 = none), lead_valid, lane_valid M0valid[N,10] bool, margin_names | |
| Moving[N,25], Moving0[N], InRange[N,25], uid, t, dongle, families | |
| Row order = s12 anchor grid; thresholds/constants recorded in run_build_pressure_targets_<utc>.json. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import ast | |
| 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 | |
| ROOT = Path(__file__).resolve().parents[1] | |
| EXPLORE = ROOT.parent | |
| sys.path.insert(0, str(ROOT / "common")) | |
| sys.path.insert(0, str(EXPLORE / "s3_can_baseline")) | |
| from runmeta import write_run_json, sha256_file # noqa: E402 | |
| import canspec # noqa: E402 (whitelist reader; FEATURES names only) | |
| GRID = EXPLORE / "s14_corr" / "cache" / "grid10" | |
| GRID_SRC = EXPLORE / "s14_corr" / "grid_s14.py" | |
| S12 = EXPLORE / "s12_wm" / "results" | |
| OUT = ROOT / "results" / "targets_pressure" | |
| FS = 10.0 | |
| H_STEPS, STEP_GRID = 25, 2 | |
| MOVING_MS = 2.0 | |
| MIN_Q_SAMPLES = 200 | |
| FAMILIES = ["ttc", "ttc_drop", "cut_in", "lead_braking", "lane_excursion", "badas"] | |
| READ_COLS = ["vEgo", "lane_center_offset", "lane_width", "lane_valid", "lead_dRel", "lead_vRel", "lead_vLead", | |
| "lead_valid", "badas_risk", "ttc", "lane_off"] | |
| MARGIN_NAMES = ["ttc_s", "headway_s", "dRel_m", "vRel_mps", "aLead_mps2", "lat_margin_m", "lane_width_m", "badas", | |
| "lead_valid", "lane_valid"] | |
| # thresholds (reused from D where the family exists; new ones swept by --sensitivity on val, see decisions.md) | |
| THR = dict(ttc=4.0, ttc_drop=-1.0, ttc_drop_maxttc=8.0, # D (mine_anomalies.ABS) | |
| cut_in_headway_s=2.5, cut_in_gap_s=0.5, cut_in_jump_m=5.0, cut_in_hold_s=1.0, | |
| lead_brake_a=-2.0, lead_brake_win_s=1.0, lead_brake_headway_s=4.0, same_target_jump_m=5.0, | |
| veh_half_w=0.9, lane_w_min=2.5, lane_w_max=5.0, | |
| badas_abs=0.85, badas_pct=95.0, # D | |
| vego_floor=1.0, ttc_none=30.0, headway_none=20.0, drel_none=200.0, lat_none=2.0) | |
| # ----------------------------------------------------------------------------- shared grid10 helpers | |
| def grid_cols(): | |
| """COLS = DCOLS + PCOLS + canspec.FEATURES + RCOLS, recovered without importing grid_s14.""" | |
| tree = ast.parse(GRID_SRC.read_text()) | |
| lists = {} | |
| for node in tree.body: | |
| if isinstance(node, ast.Assign) and len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): | |
| if node.targets[0].id in ("DCOLS", "PCOLS", "RCOLS"): | |
| lists[node.targets[0].id] = [ast.literal_eval(e) for e in node.value.elts] | |
| return lists["DCOLS"] + lists["PCOLS"] + list(canspec.FEATURES) + lists["RCOLS"], lists | |
| COLS, COL_BLOCKS = grid_cols() | |
| CI = {c: i for i, c in enumerate(COLS)} | |
| N_FORBIDDEN = len(COL_BLOCKS["DCOLS"]) + len(COL_BLOCKS["PCOLS"]) # 12: DMS block + pose block | |
| for _c in READ_COLS: | |
| assert CI[_c] >= N_FORBIDDEN, f"column {_c} lies in the DMS/pose block — refused" | |
| def col(G, name): | |
| assert name in READ_COLS or name in _EXTRA_ALLOWED, name | |
| return G[:, CI[name]].astype(np.float64) | |
| _EXTRA_ALLOWED = set() # build_response_targets registers its own (CAN-only) columns here | |
| def allow_columns(names): | |
| for n in names: | |
| assert CI[n] >= N_FORBIDDEN, f"column {n} lies in the DMS/pose block — refused" | |
| _EXTRA_ALLOWED.add(n) | |
| def load_grid(uid): | |
| z = np.load(GRID / f"{uid}.npz") | |
| G, t = z["G"], z["t"].astype(np.float64) | |
| assert G.shape[1] == len(COLS), (uid, G.shape) | |
| return G, t | |
| def route_quantile(x, mv, p, default): | |
| xs = x[mv] | |
| xs = xs[np.isfinite(xs)] | |
| return float(np.percentile(xs, p)) if len(xs) > MIN_Q_SAMPLES else default | |
| def hold(flag, n): | |
| """flag held True for n samples after each True (inclusive of the trigger sample).""" | |
| out = flag.copy() | |
| for s in range(1, n): | |
| out[s:] |= flag[:-s] | |
| return out | |
| def gather_steps(arr, k, T): | |
| """arr[..., T] indexed at anchors k and offsets 2..50 -> (values[n,25,...], inr[n,25]).""" | |
| offs = (np.arange(1, H_STEPS + 1) * STEP_GRID)[None, :] | |
| ki = k[:, None] + offs | |
| inr = (ki >= 0) & (ki < T) | |
| kic = np.clip(ki, 0, T - 1) | |
| return kic, inr | |
| def anchor_index(t_anchor, t_grid): | |
| k = np.round((t_anchor - t_grid[0]) * FS).astype(np.int64) | |
| worst = float(np.abs(t_grid[0] + k / FS - t_anchor).max()) if len(k) else 0.0 | |
| return k, worst | |
| # ----------------------------------------------------------------------------- pressure per route | |
| def route_pressure(G, thr=None): | |
| """Label-side road-pressure arrays for one route on its 10 Hz grid. `thr`: dict of thresholds (route quantile of | |
| badas computed here when thr is None / lacks 'badas').""" | |
| thr = dict(THR, **(thr or {})) | |
| T = G.shape[0] | |
| v = col(G, "vEgo") | |
| vfin = np.isfinite(v) | |
| mv = vfin & (v > MOVING_MS) | |
| lane_valid = col(G, "lane_valid") > 0.5 | |
| lane_w = col(G, "lane_width") | |
| lane_off = np.abs(col(G, "lane_off")) | |
| lead_valid = col(G, "lead_valid") > 0.5 | |
| dRel = col(G, "lead_dRel") | |
| vRel = col(G, "lead_vRel") | |
| vLead = col(G, "lead_vLead") | |
| badas = col(G, "badas_risk") | |
| ttc = col(G, "ttc") | |
| idx = np.arange(T) | |
| radar_present = bool(lead_valid.any()) | |
| if "badas" not in thr: | |
| thr["badas"] = max(thr["badas_abs"], route_quantile(badas, mv, thr["badas_pct"], thr["badas_abs"])) | |
| # --- ttc / ttc_drop (verbatim D definitions) | |
| dttc = np.full(T, np.nan) | |
| dttc[2:] = (ttc[2:] - ttc[:-2]) * (FS / 2.0) | |
| f_ttc = mv & np.isfinite(ttc) & (ttc < thr["ttc"]) | |
| f_drop = mv & np.isfinite(dttc) & (dttc < thr["ttc_drop"]) & np.isfinite(ttc) & (ttc < thr["ttc_drop_maxttc"]) | |
| # --- radar-derived quantities (same-target bookkeeping) | |
| headway = np.where(lead_valid, dRel / np.maximum(np.nan_to_num(v, nan=0.0), thr["vego_floor"]), np.nan) | |
| drop1 = np.zeros(T, np.float64) # one-sample dRel drop while the lead stays valid | |
| drop1[1:] = np.where(lead_valid[1:] & lead_valid[:-1], dRel[:-1] - dRel[1:], 0.0) | |
| djump = drop1 > thr["cut_in_jump_m"] # target switch that counts as a cut-in (a same-target closure | |
| tjump = drop1 > thr["same_target_jump_m"] # of > 5 m per 0.1 s is impossible); tjump breaks same-target tracking | |
| gap_n = int(round(thr["cut_in_gap_s"] * FS)) | |
| # no lead during the previous gap_n samples (all of i-gap_n .. i-1 invalid) | |
| cs_inv = np.cumsum(np.r_[0, (~lead_valid).astype(np.int64)]) | |
| prev_all_invalid = np.zeros(T, bool) | |
| prev_all_invalid[gap_n:] = (cs_inv[gap_n:gap_n + T - gap_n] - cs_inv[:T - gap_n]) == gap_n | |
| new_lead = lead_valid & prev_all_invalid | |
| switch = djump & (idx >= gap_n) | |
| relevant = np.isfinite(headway) & (headway < thr["cut_in_headway_s"]) | |
| trig = mv & (new_lead | switch) & relevant & (idx >= gap_n) | |
| f_cut = hold(trig, int(round(thr["cut_in_hold_s"] * FS))) | |
| win = int(round(thr["lead_brake_win_s"] * FS)) | |
| same_target = np.zeros(T, bool) # lead valid over i-win..i and no jump inside | |
| cs_val = np.cumsum(np.r_[0, lead_valid.astype(np.int64)]) | |
| cs_jump = np.cumsum(np.r_[0, tjump.astype(np.int64)]) | |
| same_target[win:] = ((cs_val[win + 1:] - cs_val[:T - win]) == win + 1) & ((cs_jump[win + 1:] - cs_jump[1:T - win + 1]) == 0) | |
| aLead = np.full(T, np.nan) | |
| aLead[win:] = (vLead[win:] - vLead[:-win]) / (win / FS) | |
| aLead[~same_target] = np.nan | |
| f_brk = mv & same_target & np.isfinite(aLead) & (aLead < thr["lead_brake_a"]) & np.isfinite(headway) & \ | |
| (headway < thr["lead_brake_headway_s"]) | |
| # --- lane excursion | |
| lat_margin = lane_w / 2.0 - lane_off | |
| lane_ok = lane_valid & np.isfinite(lane_off) & np.isfinite(lane_w) & (lane_w >= thr["lane_w_min"]) & (lane_w <= thr["lane_w_max"]) | |
| f_exc = mv & lane_ok & (lat_margin < thr["veh_half_w"]) | |
| # --- badas | |
| f_bad = mv & np.isfinite(badas) & (badas > thr["badas"]) | |
| fam = np.stack([f_ttc, f_drop, f_cut, f_brk, f_exc, f_bad]) | |
| valid = np.stack([vfin & radar_present, | |
| vfin & radar_present & (idx >= 2), | |
| vfin & radar_present & (idx >= gap_n), | |
| vfin & radar_present & (idx >= win), | |
| vfin & lane_valid & np.isfinite(lane_off) & np.isfinite(lane_w), | |
| vfin & np.isfinite(badas)]) | |
| fam &= valid | |
| S = np.zeros((len(FAMILIES), T), np.float32) | |
| S[0] = np.where(np.isfinite(ttc), thr["ttc"] / np.maximum(ttc, 0.1), 0.0) | |
| S[1] = np.where(np.isfinite(dttc) & np.isfinite(ttc) & (ttc < thr["ttc_drop_maxttc"]), -dttc / abs(thr["ttc_drop"]), 0.0) | |
| S[2] = np.where(np.isfinite(headway), thr["cut_in_headway_s"] / np.maximum(headway, 0.1), 0.0) # closeness proxy | |
| S[3] = np.where(np.isfinite(aLead), aLead / thr["lead_brake_a"], 0.0) | |
| S[4] = np.where(lane_ok, thr["veh_half_w"] / np.maximum(lat_margin, 0.05), 0.0) | |
| S[5] = np.nan_to_num(badas / thr["badas"], nan=0.0) | |
| S[:, ~mv] = 0.0 | |
| Mg = np.zeros((len(MARGIN_NAMES), T), np.float32) | |
| Mg[0] = np.where(np.isfinite(ttc), np.minimum(ttc, thr["ttc_none"]), thr["ttc_none"]) | |
| Mg[1] = np.where(np.isfinite(headway), np.minimum(headway, thr["headway_none"]), thr["headway_none"]) | |
| Mg[2] = np.where(lead_valid, np.minimum(dRel, thr["drel_none"]), thr["drel_none"]) | |
| Mg[3] = np.where(lead_valid, vRel, 0.0) | |
| Mg[4] = np.nan_to_num(aLead, nan=0.0) | |
| Mg[5] = np.where(lane_ok, lat_margin, thr["lat_none"]) | |
| Mg[6] = np.where(lane_ok, lane_w, 0.0) | |
| Mg[7] = np.nan_to_num(badas, nan=0.0) | |
| Mg[8] = lead_valid | |
| Mg[9] = lane_valid | |
| Mgv = np.stack([vfin & radar_present, vfin & radar_present, vfin & radar_present, vfin & radar_present, | |
| vfin & radar_present & (idx >= win), vfin & lane_valid, vfin & lane_valid, vfin & np.isfinite(badas), | |
| vfin & radar_present, vfin]) | |
| return dict(T=T, fam=fam, valid=valid, S=S, M=Mg, Mvalid=Mgv, mv=mv, thr=thr, radar_present=radar_present, | |
| lane_cov=float(lane_valid.mean()), badas_cov=float(np.isfinite(badas).mean()), n_moving=int(mv.sum())) | |
| def first_onset(z0, z0valid, z, zvalid): | |
| """first step with inactive->active transition (z_prev valid & inactive, z valid & active); -1 none; -2 anchor | |
| active/invalid. z[N,25], z0[N].""" | |
| zp = np.concatenate([z0[:, None], z[:, :-1]], 1) | |
| vp = np.concatenate([z0valid[:, None], zvalid[:, :-1]], 1) | |
| onset = ~zp & vp & z & zvalid | |
| first = np.where(onset.any(1), onset.argmax(1), -1).astype(np.int16) | |
| first[~z0valid | z0] = -2 | |
| return first | |
| def gather_route(rd, k): | |
| T = rd["T"] | |
| kic, inr = gather_steps(None, k, T) | |
| k0ok = (k >= 0) & (k < T) | |
| k0c = np.clip(k, 0, T - 1) | |
| P = rd["fam"][:, kic].transpose(1, 2, 0) & inr[:, :, None] | |
| Pv = rd["valid"][:, kic].transpose(1, 2, 0) & inr[:, :, None] | |
| P0 = rd["fam"][:, k0c].T & k0ok[:, None] | |
| P0v = rd["valid"][:, k0c].T & k0ok[:, None] | |
| out = dict(P=P, Pvalid=Pv, P0=P0, P0valid=P0v, | |
| S0=(rd["S"][:, k0c].T * k0ok[:, None]).astype(np.float32), | |
| M0=(rd["M"][:, k0c].T * k0ok[:, None]).astype(np.float32), | |
| M0valid=rd["Mvalid"][:, k0c].T & k0ok[:, None], | |
| Moving=rd["mv"][kic] & inr, Moving0=rd["mv"][k0c] & k0ok, InRange=inr) | |
| out["P_first"] = np.stack([first_onset(P0[:, f], P0v[:, f], P[:, :, f], Pv[:, :, f]) for f in range(len(FAMILIES))], 1) | |
| return out | |
| # ----------------------------------------------------------------------------- split build | |
| def build_split(regime, split, cache, thr_rows, route_fn=route_pressure, gather_fn=gather_route, log=print): | |
| z = np.load(S12 / f"targets_{regime}_{split}.npz", allow_pickle=True) | |
| uid, t, dongle = z["uid"], z["t"].astype(np.float64), z["dongle"] | |
| N = len(t) | |
| out = None | |
| uniq, inv = np.unique(uid, return_inverse=True) | |
| order = np.argsort(inv, kind="stable") | |
| bounds = np.searchsorted(inv[order], np.arange(len(uniq) + 1)) | |
| worst_align = 0.0 | |
| for j, u in enumerate(uniq): | |
| rows = order[bounds[j]:bounds[j + 1]] | |
| if u not in cache: | |
| G, tg = load_grid(u) | |
| rd = route_fn(G) | |
| rd["t0"] = tg[0] | |
| cache[u] = rd | |
| thr_rows.append(dict(uid=u, dongle=str(dongle[rows[0]]), T=rd["T"], **{k: v for k, v in rd.items() | |
| if k in ("radar_present", "lane_cov", "badas_cov", "n_moving")}, | |
| **{f"thr_{k}": v for k, v in rd.get("thr", {}).items()})) | |
| rd = cache[u] | |
| k = np.round((t[rows] - rd["t0"]) * FS).astype(np.int64) | |
| worst_align = max(worst_align, float(np.abs(rd["t0"] + k / FS - t[rows]).max())) | |
| g = gather_fn(rd, k) | |
| if out is None: | |
| out = {key: np.zeros((N,) + arr.shape[1:], arr.dtype) for key, arr in g.items()} | |
| for key, arr in g.items(): | |
| out[key][rows] = arr | |
| assert worst_align < 0.05, f"anchor/grid misalignment {worst_align}" | |
| out.update(uid=uid, t=t, dongle=dongle) | |
| return out, worst_align | |
| def write_npz(path, d, force): | |
| if path.exists() and not force: | |
| raise FileExistsError(f"{path} exists (append-only results; pass --force to overwrite)") | |
| np.savez_compressed(path, **d) | |
| def split_summary(d): | |
| P, Pv, P0, P0v = d["P"], d["Pvalid"], d["P0"], d["P0valid"] | |
| rows = {} | |
| for f, fam in enumerate(FAMILIES): | |
| sel3 = (d["P_first"][:, f] != -2) & Pv[:, :15, f].all(1) | |
| y3 = (d["P_first"][sel3, f] >= 0) & (d["P_first"][sel3, f] < 15) | |
| rows[fam] = dict(valid_frac_steps=float(Pv[:, :, f].mean()), prev_steps=float(P[:, :, f][Pv[:, :, f]].mean()) if Pv[:, :, f].any() else float("nan"), | |
| prev_P0=float(P0[P0v[:, f], f].mean()) if P0v[:, f].any() else float("nan"), | |
| n_inactive_anchors_3s=int(sel3.sum()), onset_anchors_3s=int(y3.sum()), | |
| base_rate_3s=float(y3.mean()) if sel3.any() else float("nan")) | |
| return rows | |
| def sensitivity(regime="manual", split="val"): | |
| """Base-rate sweep of the NEW thresholds on one split (label-definition sensitivity; val by default).""" | |
| variants = {"cut_in_headway_s": [1.5, 2.5, 3.5], "cut_in_jump_m": [3.0, 5.0, 10.0], "cut_in_hold_s": [0.6, 1.0, 2.0], | |
| "lead_brake_a": [-1.5, -2.0, -3.0], "lead_brake_win_s": [0.5, 1.0, 2.0], "lead_brake_headway_s": [3.0, 4.0, 6.0], | |
| "veh_half_w": [0.8, 0.9, 1.0]} | |
| fam_of = {"cut_in": ["cut_in_headway_s", "cut_in_jump_m", "cut_in_hold_s"], | |
| "lead_braking": ["lead_brake_a", "lead_brake_win_s", "lead_brake_headway_s"], "lane_excursion": ["veh_half_w"]} | |
| rows = [] | |
| for fam, keys in fam_of.items(): | |
| for key in keys: | |
| for val in variants[key]: | |
| cache, thr_rows = {}, [] | |
| d, _ = build_split(regime, split, cache, thr_rows, route_fn=lambda G, key=key, val=val: route_pressure(G, {key: val})) | |
| s = split_summary(d)[fam] | |
| rows.append(dict(regime=regime, split=split, family=fam, param=key, value=val, **s)) | |
| print(f"[sens] {fam:15s} {key:22s}={val:6.2f} prev_steps={s['prev_steps']:.5f} base_rate_3s={s['base_rate_3s']:.5f} " | |
| f"onsets_3s={s['onset_anchors_3s']}", flush=True) | |
| return rows | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--regimes", default="manual,engaged") | |
| ap.add_argument("--splits", default="train,val,test") | |
| ap.add_argument("--force", action="store_true") | |
| ap.add_argument("--sensitivity", action="store_true", help="threshold sweep on manual/val (+ engaged/val), writes sensitivity csv") | |
| a = ap.parse_args() | |
| OUT.mkdir(parents=True, exist_ok=True) | |
| t_start = time.time() | |
| if a.sensitivity: | |
| import csv | |
| rows = sensitivity("manual", "val") + sensitivity("engaged", "val") | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| p = OUT / f"sensitivity_val_{stamp}.csv" | |
| with open(p, "w", newline="") as f: | |
| w = csv.DictWriter(f, fieldnames=list(rows[0])) | |
| w.writeheader() | |
| w.writerows(rows) | |
| print("wrote", p) | |
| write_run_json(OUT, "pressure_sensitivity", [Path(__file__)], [S12 / "targets_manual_val.npz", S12 / "targets_engaged_val.npz"], | |
| 0, t_start, extra=dict(csv=str(p), thresholds_default=THR, dms_fields_used=[])) | |
| return | |
| cache, thr_rows, summary, files = {}, [], {}, [] | |
| for regime in a.regimes.split(","): | |
| for split in a.splits.split(","): | |
| d, wa = build_split(regime, split, cache, thr_rows) | |
| src = np.load(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"]), "row alignment" | |
| d.update(families=np.array(FAMILIES), margin_names=np.array(MARGIN_NAMES)) | |
| p = OUT / f"pressure_{regime}_{split}.npz" | |
| write_npz(p, d, a.force) | |
| files.append(str(p)) | |
| summary[f"{regime}_{split}"] = dict(N=int(len(d["t"])), align=wa, **split_summary(d)) | |
| print(f"[{regime}/{split}] N={len(d['t'])} align={wa:.1e} -> {p.name}", flush=True) | |
| for fam, s in summary[f"{regime}_{split}"].items(): | |
| if isinstance(s, dict): | |
| print(f" {fam:15s} prev_steps={s['prev_steps']:.5f} prev_P0={s['prev_P0']:.5f} valid={s['valid_frac_steps']:.3f} " | |
| f"base_rate_3s={s['base_rate_3s']:.5f} ({s['onset_anchors_3s']}/{s['n_inactive_anchors_3s']})", flush=True) | |
| del d | |
| import pandas as pd | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| pd.DataFrame(thr_rows).drop_duplicates("uid").to_csv(OUT / f"route_thresholds_{stamp}.csv", index=False) | |
| (OUT / f"build_stats_{stamp}.json").write_text(json.dumps(summary, indent=1)) | |
| (OUT / "build_stats_latest.json").write_text(json.dumps(summary, indent=1)) | |
| inputs = [S12 / f"targets_{r}_{s}.npz" for r in a.regimes.split(",") for s in a.splits.split(",")] + [GRID_SRC] | |
| grid_files = sorted(GRID / f"{u}.npz" for u in cache) | |
| write_run_json(OUT, "build_pressure_targets", [Path(__file__)], inputs + grid_files, 0, t_start, | |
| extra=dict(files_written=files, families=FAMILIES, margin_names=MARGIN_NAMES, thresholds=THR, | |
| grid_columns_read=[(c, CI[c]) for c in READ_COLS], | |
| grid_columns_never_read=[(COLS[i], i) for i in range(N_FORBIDDEN)], | |
| n_routes=len(cache), dms_fields_used=[], stats=str(OUT / f"build_stats_{stamp}.json"))) | |
| print("PRESSURE_DONE", f"{time.time() - t_start:.0f}s", flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 23.1 kB
- Xet hash:
- f2f258bb598bd33f6073fa7152bd137909471b5ed474165322a52cda0fb1b8d4
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.