Buckets:
| """cwm3/data/report_targets_pr.py — summary tables for the disentangled target families -> results/tables/targets_pr.md. | |
| Reads results/targets_pressure, results/targets_response, results/targets_behaviour and cwm2/results/targets_demand (row-aligned; | |
| uid/t asserted). Every number in the markdown is computed here; nothing is typed by hand. | |
| §1 P chains: per regime x split x family: valid fraction, prevalence (steps / anchor), onsets within 3 s and base rate | |
| (population = inactive & valid at t, first 15 steps valid — the transition_metrics contract). | |
| §2 R grades: per regime x split: grade shares, R0 prevalence, first-passage onsets within 3 s (any / hard), component bits. | |
| §3 B events: coverage by source and base rates within 3 s per event. | |
| §4 P vs D overlap (the "demand/response disentangling" claim): over steps where both are valid — how often the old demand | |
| fires with NO road pressure (D_any & ~P_any), split into ego-only families, and how often road pressure fires with no old | |
| demand; anchor-level: among anchors with a D onset within 3 s, share with a P onset within 3 s, and vice versa; and how | |
| often an ego-motion demand onset (steer_rate / decel) is a RESPONSE (R >= 1 at the same step) rather than a road event. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "common")) | |
| from runmeta import write_run_json # noqa: E402 | |
| PRESS, RESP, BEH = ROOT / "results" / "targets_pressure", ROOT / "results" / "targets_response", ROOT / "results" / "targets_behaviour" | |
| DEM = ROOT.parent / "cwm2" / "results" / "targets_demand" | |
| TABLES = ROOT / "results" / "tables" | |
| REGIMES, SPLITS = ("manual", "engaged"), ("train", "val", "test") | |
| def onset_3s(z0, z0v, z, zv): | |
| zp = np.concatenate([z0[:, None], z[:, :-1]], 1) | |
| vp = np.concatenate([z0v[:, None], zv[:, :-1]], 1) | |
| onset = ~zp & vp & z & zv | |
| first = np.where(onset.any(1), onset.argmax(1), -1) | |
| sel = z0v & ~z0 & zv[:, :15].all(1) | |
| y = (first >= 0) & (first < 15) | |
| return sel, y & sel | |
| def fmt(x, nd=4): | |
| if isinstance(x, (float, np.floating)): | |
| return "nan" if not np.isfinite(x) else f"{x:.{nd}f}" | |
| return str(x) | |
| def table(rows, cols, nd=4): | |
| out = ["| " + " | ".join(cols) + " |", "|" + "---|" * len(cols)] | |
| for r in rows: | |
| out.append("| " + " | ".join(fmt(r.get(c, ""), nd) for c in cols) + " |") | |
| return out | |
| def main(): | |
| t0 = time.time() | |
| TABLES.mkdir(parents=True, exist_ok=True) | |
| p_rows, r_rows, b_rows, ov_rows, files, missing = [], [], [], [], [], [] | |
| for reg in REGIMES: | |
| for sp in SPLITS: | |
| fp, fr, fb, fd = PRESS / f"pressure_{reg}_{sp}.npz", RESP / f"response_{reg}_{sp}.npz", BEH / f"behaviour_{reg}_{sp}.npz", DEM / f"demand_{reg}_{sp}.npz" | |
| if not (fp.exists() and fr.exists()): | |
| missing.append(f"{reg}/{sp}") | |
| continue | |
| P, R = np.load(fp, allow_pickle=True), np.load(fr, allow_pickle=True) | |
| files += [fp, fr] | |
| fams = list(map(str, P["families"])) | |
| N = len(P["t"]) | |
| for f, fam in enumerate(fams): | |
| z, zv, z0, z0v = P["P"][:, :, f], P["Pvalid"][:, :, f], P["P0"][:, f], P["P0valid"][:, f] | |
| sel, y = onset_3s(z0, z0v, z, zv) | |
| p_rows.append(dict(regime=reg, split=sp, family=fam, n=N, valid_frac_steps=zv.mean(), prev_steps=z[zv].mean() if zv.any() else np.nan, | |
| prev_P0=z0[z0v].mean() if z0v.any() else np.nan, n_inactive_anchors_3s=int(sel.sum()), onset_anchors_3s=int(y.sum()), | |
| base_rate_3s=y.sum() / max(sel.sum(), 1), moving_frac_P0=P["Moving0"].mean())) | |
| Rg, Rv, R0, R0v = R["R"], R["Rvalid"], R["R0"], R["R0valid"] | |
| sel_a, y_a = onset_3s(R0 >= 1, R0v, Rg >= 1, Rv) | |
| sel_h, y_h = onset_3s(R0 >= 2, R0v, Rg >= 2, Rv) | |
| bits = R["Bits"][Rv] | |
| r_rows.append(dict(regime=reg, split=sp, n=N, valid_frac_steps=Rv.mean(), none=(Rg[Rv] == 0).mean(), mild=(Rg[Rv] == 1).mean(), hard=(Rg[Rv] == 2).mean(), | |
| prev_R0_any=(R0[R0v] >= 1).mean(), prev_R0_hard=(R0[R0v] == 2).mean(), | |
| n_inactive_3s_any=int(sel_a.sum()), onsets_3s_any=int(y_a.sum()), base_rate_3s_any=y_a.sum() / max(sel_a.sum(), 1), | |
| n_inactive_3s_hard=int(sel_h.sum()), onsets_3s_hard=int(y_h.sum()), base_rate_3s_hard=y_h.sum() / max(sel_h.sum(), 1), | |
| bit_brake=((bits & 1) > 0).mean(), bit_aego_mild=((bits & 2) > 0).mean(), bit_sr_mild=((bits & 4) > 0).mean(), | |
| bit_aego_hard=((bits & 8) > 0).mean(), bit_sr_hard=((bits & 16) > 0).mean())) | |
| if fb.exists(): | |
| B = np.load(fb, allow_pickle=True) | |
| files.append(fb) | |
| src = B["B_source"] | |
| names = {0: "none", 1: "causal_clip", 2: "old_noncausal", 3: "causal_routes"} | |
| for code, nm in names.items(): | |
| rows_c = src == code | |
| if not rows_c.any(): | |
| continue | |
| rec = dict(regime=reg, split=sp, source=nm, n_rows=int(rows_c.sum()), frac_rows=rows_c.mean()) | |
| if code: | |
| inact0 = np.isfinite(B["B0"]) & (B["B0"] < 0.5) | |
| for e, ev in enumerate(map(str, B["events"])): | |
| Bf = B["B_first"][:, e] | |
| sel = rows_c & (Bf != -2) & B["Bvalid"][:, :15, e].all(1) & inact0[:, e] | |
| y = sel & (Bf >= 0) & (Bf < 15) | |
| rec[f"{ev}_n3s"] = int(sel.sum()) | |
| rec[f"{ev}_onsets3s"] = int(y.sum()) | |
| rec[f"{ev}_base3s"] = y.sum() / max(sel.sum(), 1) | |
| b_rows.append(rec) | |
| if fd.exists(): | |
| D = np.load(fd, allow_pickle=True) | |
| files.append(fd) | |
| assert np.array_equal(D["uid"], P["uid"]) and np.array_equal(D["t"], P["t"]) | |
| dfams = list(map(str, D["families"])) | |
| ego = [dfams.index("steer_rate"), dfams.index("decel")] | |
| road_d = [dfams.index(f) for f in ("ttc", "ttc_drop", "lane", "badas")] | |
| Dv_any = D["Dvalid"].any(2) | |
| D_any = (D["D"] & D["Dvalid"]).any(2) | |
| D_ego = (D["D"][:, :, ego] & D["Dvalid"][:, :, ego]).any(2) | |
| D_road = (D["D"][:, :, road_d] & D["Dvalid"][:, :, road_d]).any(2) | |
| Pv_any = P["Pvalid"].any(2) | |
| P_any = (P["P"] & P["Pvalid"]).any(2) | |
| both = Dv_any & Pv_any & D["Moving"] & P["Moving"] | |
| n_both = int(both.sum()) | |
| Ract = (Rg >= 1) & Rv | |
| # anchor-level onsets within 3 s (any family) | |
| def any_first(z, zv, z0, z0v): | |
| sel, y = onset_3s(z0, z0v, z, zv) | |
| return sel, y | |
| D0_any = (D["D0"] & D["D0valid"]).any(1) | |
| P0_any = (P["P0"] & P["P0valid"]).any(1) | |
| selD, yD = any_first(D_any, Dv_any, D0_any, D["D0valid"].any(1)) | |
| selP, yP = any_first(P_any, Pv_any, P0_any, P["P0valid"].any(1)) | |
| D0_ego = (D["D0"][:, ego] & D["D0valid"][:, ego]).any(1) | |
| selE, yE = any_first(D_ego, D["Dvalid"][:, :, ego].any(2), D0_ego, D["D0valid"][:, ego].any(1)) | |
| # ego demand onsets: preceded by road pressure within the previous 3 s (incl. the anchor state P0) or not | |
| zpE = np.concatenate([D0_ego[:, None], D_ego[:, :-1]], 1) | |
| onE = D_ego & ~zpE | |
| Pcum = np.concatenate([P0_any[:, None], P_any], 1).astype(np.int32) # [N,26]: P0 then steps 0..24 | |
| cs = np.cumsum(Pcum, 1) | |
| j = np.arange(25) | |
| lo = np.maximum(j - 15, 0) # window: steps j-15..j (+ P0 when lo == 0) | |
| p_prior = (cs[:, j + 1] - np.where(lo > 0, cs[:, lo], 0)) > 0 # any P in the 3 s up to and incl. step j | |
| # NOTE: the ego families (|sr| > 100 deg/s, aEgo < -2) are strict subsets of the R >= 1 rule (|sr| > 30, aEgo < -0.8), | |
| # so "ego demand onset coincides with a response" is 1.0 by construction and is not reported as a finding. | |
| ov_rows.append(dict(regime=reg, split=sp, n_steps_both_valid_moving=n_both, | |
| D_any_rate=D_any[both].mean(), P_any_rate=P_any[both].mean(), | |
| D_without_P=(D_any & ~P_any)[both].mean(), P_without_D=(P_any & ~D_any)[both].mean(), | |
| D_and_P=(D_any & P_any)[both].mean(), | |
| frac_of_D_steps_without_P=(D_any & ~P_any)[both].sum() / max(D_any[both].sum(), 1), | |
| frac_of_D_steps_ego_only=(D_ego & ~D_road & ~P_any)[both].sum() / max(D_any[both].sum(), 1), | |
| frac_of_P_steps_without_D=(P_any & ~D_any)[both].sum() / max(P_any[both].sum(), 1), | |
| ego_onsets_with_P_same_step=(onE & P_any)[both].sum() / max(onE[both].sum(), 1), | |
| ego_onsets_with_P_prior_3s=(onE & p_prior)[both].sum() / max(onE[both].sum(), 1), | |
| ego_onsets_with_hard_R_same_step=(onE & (Rg >= 2) & Rv)[both].sum() / max(onE[both].sum(), 1), | |
| anchors_D_onset3s=int(yD.sum()), anchors_P_onset3s=int(yP.sum()), | |
| D_onset3s_with_P_onset3s=(yD & yP).sum() / max(yD.sum(), 1), | |
| P_onset3s_with_D_onset3s=(yD & yP).sum() / max(yP.sum(), 1), | |
| ego_onset3s_without_P_onset3s=(yE & ~yP).sum() / max(yE.sum(), 1))) | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| L = ["# Disentangled target families P / R / B — base rates, onsets and the P-vs-D overlap", "", | |
| f"Generated by `data/report_targets_pr.py` at {stamp} from `results/targets_pressure`, `results/targets_response`, `results/targets_behaviour` and " | |
| "`cwm2/results/targets_demand` (row-aligned with the s12 anchor grid; uid/t asserted). Every number is script output.", | |
| "Population for `base_rate_3s` = anchors inactive & valid at t with the first 15 future steps valid (the `eval_s13.transition_metrics` contract); " | |
| "`prev_steps` = mean over valid future steps; `prev_P0` = at the anchor.", ""] | |
| if missing: | |
| L += [f"Missing (not built): {', '.join(missing)}", ""] | |
| L += ["## 1. Road-pressure chains P (radar / lane / BADAS only — no ego-motion family)", ""] | |
| L += table(p_rows, ["regime", "split", "family", "n", "valid_frac_steps", "prev_steps", "prev_P0", "n_inactive_anchors_3s", "onset_anchors_3s", "base_rate_3s"]) | |
| L += ["", "Definitions / thresholds: `data/build_pressure_targets.py` header and `results/targets_pressure/run_build_pressure_targets_latest.json` " | |
| "(`thresholds`); sensitivity sweep on val: `results/targets_pressure/sensitivity_val_*.csv`.", "", | |
| "## 2. Graded vehicle response R (none / mild / hard; `Resp = R >= 1` is bit-identical to cwm2's Resp chain)", ""] | |
| L += table(r_rows, ["regime", "split", "n", "valid_frac_steps", "none", "mild", "hard", "prev_R0_any", "prev_R0_hard", "n_inactive_3s_any", "onsets_3s_any", | |
| "base_rate_3s_any", "n_inactive_3s_hard", "onsets_3s_hard", "base_rate_3s_hard"]) | |
| L += ["", "Component bits (share of valid steps): "] | |
| L += table(r_rows, ["regime", "split", "bit_brake", "bit_aego_mild", "bit_sr_mild", "bit_aego_hard", "bit_sr_hard"]) | |
| L += ["", "## 3. Driver-behaviour events B (source per row; causal where available)", ""] | |
| bcols = ["regime", "split", "source", "n_rows", "frac_rows"] + [f"{ev}_{k}" for ev in ("hand_return", "hand_leave", "head_turn_onset") for k in ("n3s", "onsets3s", "base3s")] | |
| L += table(b_rows, bcols) | |
| L += ["", "`causal_clip` rows exist on train routes only (clip windows; 0 on s12 val/test routes — `results/targets_kin_causal/build_stats_latest.json`); " | |
| "`old_noncausal` = cwm2 targets_kin fallback (route-median ROI incl. the future; hand-event onsets agree with the causal ones only at F1 ~ 0.3), disclosed via `B_source`.", "", | |
| "## 4. P vs D overlap — how much of the old demand was ego motion", "", | |
| "Step level (steps where D and P are both valid and moving): `D_without_P` = old demand fires with no road pressure at that step; " | |
| "`frac_of_D_steps_ego_only` = share of D-active steps carried by steer_rate/decel alone with no road family (D or P) active; " | |
| "`ego_onsets_with_P_prior_3s` = share of ego-family (steer_rate / decel) onset steps with road pressure active at that step or in the 3 s before it " | |
| "(a road cause is visible); `ego_onsets_with_P_same_step` = same, at the step only; `ego_onsets_with_hard_R_same_step` = share graded hard by R. " | |
| "By construction every ego-family onset is a response (R >= 1): |steer_rate| > 100 deg/s and aEgo < -2 m/s^2 are subsets of the R rule, " | |
| "so the old D mixed responses into 'demand' — the number is 1.0 and is not tabulated. Anchor level: `D_onset3s_with_P_onset3s` = among anchors with a D onset within 3 s, share that also has a P onset within 3 s.", ""] | |
| L += table(ov_rows, ["regime", "split", "n_steps_both_valid_moving", "D_any_rate", "P_any_rate", "D_without_P", "P_without_D", "D_and_P", | |
| "frac_of_D_steps_without_P", "frac_of_D_steps_ego_only", "frac_of_P_steps_without_D", "ego_onsets_with_P_same_step", | |
| "ego_onsets_with_P_prior_3s", "ego_onsets_with_hard_R_same_step"]) | |
| L += [""] | |
| L += table(ov_rows, ["regime", "split", "anchors_D_onset3s", "anchors_P_onset3s", "D_onset3s_with_P_onset3s", "P_onset3s_with_D_onset3s", "ego_onset3s_without_P_onset3s"]) | |
| L += ["", f"Run record: `results/tables/run_report_targets_pr_{stamp}.json`.", ""] | |
| (TABLES / "targets_pr.md").write_text("\n".join(L)) | |
| (TABLES / f"targets_pr_{stamp}.json").write_text(json.dumps(dict(P=p_rows, R=r_rows, B=b_rows, overlap=ov_rows), indent=1, default=float)) | |
| write_run_json(TABLES, "report_targets_pr", [Path(__file__)], files, 0, t0, extra=dict(table=str(TABLES / "targets_pr.md"), missing=missing, dms_fields_used=[])) | |
| print("\n".join(L)) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 14.7 kB
- Xet hash:
- 1fad738f4b6f2c362e1b1ddb35c1c3ff2cba9ebf647c230da2a04ae4a336edfa
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.