Buckets:
| """cwm3/data/labels.py — human warning labels for the learned warning head W (plan §3 W; cwm2/docs/data_and_labels.md). | |
| Sources (column `source`, kept separate everywhere): | |
| consensus clean TRAIN consensus rows: protocol == consensus AND split_of(dongle) == train AND group ∉ {G11, [annotator]} AND label != unsure | |
| (1,148 rows) anchored on the s12 TRAIN packs of either regime with the gold_eval rule (t in [t0-2, t0+0.2], latest anchor, | |
| BAD routes skipped; an episode anchored in both regimes keeps the latest anchor, ties -> its own regime, as warning_gold.build_gold) | |
| -> 979 rows / 182 warranted / 91 devices (measured 2026-09-10). | |
| legacy_v1 legacy single-annotator TRAIN rows (protocol legacy_v1_single_annotator, label != unsure; κ ≈ 0.15 in the audit) anchored the same | |
| way -> 1,250 rows / 412 warranted. Weak; used only as an explicitly labelled extension of the label curve, never in the headline. | |
| h5 cwm2/annotation_human/labels_h5.jsonl `prompt_warranted_now` (five-question session) when the file exists; ABSENT on 2026-09-10. | |
| The clean gold (test 568 / val 115) is NEVER returned by train_label_table (asserted: no test-device row); gold_val_table returns the clean-VAL | |
| rows anchored on the val packs (s12-latest anchor) for the warning head's temperature fit. No openpilot DMS field is read. | |
| python labels.py --check | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| CWM2 = EXPLORE / "cwm2" | |
| S12R = EXPLORE / "s12_wm" / "results" | |
| H5_STORE = CWM2 / "annotation_human" / "labels_h5.jsonl" | |
| for p in (CWM2 / "eval", CWM2 / "data", CWM2, CWM2 / "baselines" / "legacy_cockpitwm"): | |
| sys.path.insert(0, str(p)) | |
| LENIENT = {"G11", "[annotator]"} | |
| REGIMES = ("manual", "engaged") | |
| def episode_labels(): | |
| """score_clean568.episode_table(): one row per annotated episode (idx, label, protocol, group, uid, dongle, t0, ep_regime, s12_split, y, clean*).""" | |
| from score_clean568 import episode_table | |
| return episode_table() | |
| def load_pack(regime, split): | |
| d = np.load(S12R / f"targets_{regime}_{split}.npz", allow_pickle=True) | |
| return {"uid": d["uid"], "dongle": d["dongle"], "t": d["t"]} | |
| def _anchor(lab, split): | |
| """latest s12 anchor of each episode over both regimes' packs of `split` -> DataFrame idx, regime, row, t_anchor.""" | |
| from score_clean568 import anchor_hits | |
| hits = {r: anchor_hits({split: load_pack(r, split)}, lab) for r in REGIMES} | |
| rows = [] | |
| for e in lab.itertuples(): | |
| cands = {r: hits[r][e.idx] for r in REGIMES if e.idx in hits[r]} | |
| if not cands: | |
| continue | |
| best_r = max(cands, key=lambda r: (cands[r][2], r == e.ep_regime)) | |
| rows.append(dict(idx=e.idx, regime=best_r, row=int(cands[best_r][1]), t_anchor=float(cands[best_r][2]), n_regimes_anchored=len(cands))) | |
| return pd.DataFrame(rows) | |
| def h5_labels(): | |
| """prompt_warranted_now records (source h5) or an empty frame with the reason.""" | |
| if not H5_STORE.exists(): | |
| return pd.DataFrame(columns=["idx", "y", "source", "uid", "dongle", "t0"]), f"{H5_STORE} absent (no five-question records yet)" | |
| recs = [json.loads(l) for l in H5_STORE.read_text().splitlines() if l.strip()] | |
| rows = [] | |
| for r in recs: | |
| a = r.get("answers", r) | |
| if "prompt_warranted_now" not in a: | |
| continue | |
| v = str(a["prompt_warranted_now"]).lower() | |
| y = 1 if v in ("yes", "warranted", "true", "1") else 0 if v in ("no", "not_warranted", "false", "0") else None | |
| if y is None: | |
| continue | |
| rows.append(dict(idx=r.get("idx", r.get("clip_id")), y=y, source="h5", uid=r.get("uid", ""), dongle=r.get("dongle", ""), t0=r.get("t0", np.nan))) | |
| return pd.DataFrame(rows), f"{len(rows)} usable records" | |
| def train_label_table(sources=("consensus",)): | |
| """TRAIN-device labelled episodes anchored on the TRAIN packs. Columns: idx y source protocol group uid dongle t0 ep_regime regime row t_anchor.""" | |
| EP = episode_labels() | |
| lab = EP[(EP.label != "unsure") & (EP.s12_split == "train")] | |
| parts = [] | |
| if "consensus" in sources: | |
| c = lab[(lab.protocol == "consensus") & (~lab.group.isin(LENIENT))].copy() | |
| c["source"] = "consensus" | |
| parts.append(c) | |
| if "legacy_v1" in sources: | |
| l = lab[lab.protocol == "legacy_v1_single_annotator"].copy() | |
| l["source"] = "legacy_v1" | |
| parts.append(l) | |
| tab = pd.concat(parts, ignore_index=True) if parts else lab.iloc[:0].assign(source="") | |
| anc = _anchor(tab, "train") | |
| out = tab.merge(anc, on="idx", how="inner") | |
| assert (out.s12_split == "train").all(), "a non-train device reached the training label table" | |
| if "h5" in sources: | |
| h5, note = h5_labels() | |
| out.attrs["h5_note"] = note | |
| if len(h5): | |
| out = pd.concat([out, h5], ignore_index=True) | |
| cols = ["idx", "y", "source", "protocol", "group", "uid", "dongle", "t0", "ep_regime", "regime", "row", "t_anchor", "n_regimes_anchored"] | |
| return out[[c for c in cols if c in out]].reset_index(drop=True) | |
| def gold_val_table(): | |
| """clean-VAL rows (consensus, val devices, not lenient, not unsure) anchored on the VAL packs (s12-latest anchor).""" | |
| EP = episode_labels() | |
| lab = EP[EP.clean_val & (EP.label != "unsure")].copy() | |
| anc = _anchor(lab, "val") | |
| out = lab.merge(anc, on="idx", how="inner") | |
| out["source"] = "clean_val" | |
| return out[["idx", "y", "source", "uid", "dongle", "t0", "ep_regime", "regime", "row", "t_anchor"]].reset_index(drop=True) | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--check", action="store_true") | |
| ap.add_argument("--dump-rows", default="", help="write {regime: sorted train rows} of the anchored label table to this npz and exit") | |
| ap.add_argument("--sources", default="consensus,legacy_v1") | |
| a = ap.parse_args() | |
| if a.dump_rows: | |
| tab = train_label_table(sources=tuple(a.sources.split(","))) | |
| np.savez(a.dump_rows, **{r: np.sort(tab[tab.regime == r].row.unique()).astype(np.int64) for r in REGIMES}, n_rows=np.array(len(tab))) | |
| print(f"wrote {a.dump_rows}: " + ", ".join(f"{r} {int((tab.regime == r).sum())}" for r in REGIMES)) | |
| return | |
| for src in (("consensus",), ("consensus", "legacy_v1")): | |
| t = train_label_table(src) | |
| print(f"train sources={src}: rows {len(t)} pos {int(t.y.sum())} devices {t.dongle.nunique()} regimes {t.regime.value_counts().to_dict()} " | |
| f"by source {t.groupby('source').y.agg(['count', 'sum']).to_dict()}") | |
| v = gold_val_table() | |
| print(f"clean_val anchored: rows {len(v)} pos {int(v.y.sum())} devices {v.dongle.nunique()}") | |
| h5, note = h5_labels() | |
| print(f"h5: {note}") | |
| if a.check: | |
| assert len(train_label_table(("consensus",))) == 979 | |
| print("check OK") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 7.03 kB
- Xet hash:
- 3895b61fb4629d7c5d69485faa6506d67049a23ebfe34e687cd05d66033e80fd
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.