Buckets:
| """cwm3/weak/gpt_calibration.py — GPT weak-label CALIBRATION batch: sampling, spend watch, and the pre-registered gate. | |
| python gpt_calibration.py sample --n 100 --seed 0 --stamp <UTC> # writes cwm2/results/W1d/gpt_calibration_clips_<stamp>.{csv,txt}; no API | |
| python gpt_calibration.py watch --clips <txt> --n-total 100 [--abort-usd 12] # ledger-only projection; exit 3 = abort condition met | |
| python gpt_calibration.py gate --clips <csv> --stamp <UTC> # kappa (+bootstrap CI), per-field tables, spend, cache hit rate -> W1d md/json | |
| Pool (pre-registered): the 187-clip pilot manifest of cwm2/annotation_package (frames on disk, anchor = episode t0 at 5.000 s of the | |
| 15 s window, TRAIN devices only) intersected by episode idx with cwm3/data/labels.train_label_table(("consensus",)) — the 979 clean TRAIN | |
| consensus rows used by the warning head (protocol consensus, split_of(dongle) == train, group not in {G11, [annotator]}, label != unsure, | |
| anchored on the s12 TRAIN packs). Sampling: proportional stratification by the human binary label y, within stratum a seed-0 | |
| permutation (numpy default_rng) of the SORTED clip ids. Every sampled clip is re-asserted to be on a TRAIN-split device via | |
| run_gpt._split_of_clips before anything is written. | |
| Gate: GPT `prompt_warranted_now` binarised (warranted_attention / warranted_hands -> 1, not_warranted -> 0, unsure EXCLUDED and its rate | |
| reported) vs human y; Cohen's kappa >= 0.6 (point estimate) passes; 95 % CI by clip bootstrap (B = 2000, seed 0). Per-field agreement | |
| for behavior_task_compatibility / task family+demand / concern_persists / confidence vs the human label is DESCRIPTIVE only. | |
| No openpilot driver-monitoring value is read anywhere here (labels.py reads none; returns are the package's own GPT outputs). | |
| """ | |
| from __future__ import annotations | |
| import argparse, hashlib, json, sqlite3, sys | |
| from collections import Counter | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| EXPLORE = CWM3.parent | |
| PKG = EXPLORE / "cwm2" / "annotation_package" | |
| W1D = EXPLORE / "cwm2" / "results" / "W1d" | |
| LEDGER = CWM3 / "weak" / "budget_ledger.sqlite" | |
| CLIP_MAP = EXPLORE / "cwm2" / "private" / "clip_map.json" | |
| PASSES = ("observe_prefix", "ground_relation") | |
| DEMAND_ORDER = {"low": 0, "moderate": 1, "high": 2} | |
| BIN = {"warranted_attention": 1, "warranted_hands": 1, "not_warranted": 0} | |
| def sha256_lines(ids) -> str: | |
| return hashlib.sha256(("\n".join(sorted(ids)) + "\n").encode()).hexdigest() | |
| def kappa(y, g) -> float: | |
| y = np.asarray(y, int); g = np.asarray(g, int); n = len(y) | |
| if n == 0: | |
| return float("nan") | |
| po = float((y == g).mean()) | |
| pe = float(((y == 1).mean() * (g == 1).mean()) + ((y == 0).mean() * (g == 0).mean())) | |
| return float("nan") if pe >= 1.0 else (po - pe) / (1.0 - pe) | |
| def boot_ci(y, g, B=2000, seed=0): | |
| rng = np.random.default_rng(seed); y = np.asarray(y, int); g = np.asarray(g, int); n = len(y); ks = [] | |
| for _ in range(B): | |
| i = rng.integers(0, n, n); k = kappa(y[i], g[i]) | |
| if np.isfinite(k): | |
| ks.append(k) | |
| ks = np.asarray(ks) | |
| return (float(np.percentile(ks, 2.5)), float(np.percentile(ks, 97.5)), int(len(ks))) | |
| # ----------------------------------------------------------------------------- sample | |
| def cmd_sample(a): | |
| sys.path.insert(0, str(CWM3 / "data")); sys.path.insert(0, str(CWM3 / "weak")); sys.path.insert(0, str(PKG)) | |
| from labels import train_label_table | |
| import run_gpt # noqa: F401 (patches pkg_common; no API, no argv side effects outside __main__ except --ledger append) | |
| import pkg_common as pc | |
| tab = train_label_table(("consensus",)) | |
| assert len(tab) == 979, len(tab) | |
| cm = json.load(open(CLIP_MAP)) | |
| man = pc.manifest_by_id() | |
| idx2clip = {v["idx"]: k for k, v in cm.items() if k in man and pc.clip_has_frames(k)} | |
| tab = tab.assign(clip_id=tab.idx.map(idx2clip)) | |
| pool = tab[tab.clip_id.notna()].copy() | |
| pool["human_label"] = pool.clip_id.map(lambda c: cm[c]["label"]) | |
| assert ((pool.human_label == "warranted") == (pool.y == 1)).all(), "pilot label / table y mismatch" | |
| pool = pool.sort_values("clip_id").reset_index(drop=True) | |
| n1 = int(round(a.n * (pool.y == 1).mean())); n0 = a.n - n1 | |
| rng = np.random.default_rng(a.seed) | |
| parts = [] | |
| for yv, k in ((1, n1), (0, n0)): | |
| ids = sorted(pool[pool.y == yv].clip_id) | |
| perm = rng.permutation(len(ids)) | |
| parts.append(pool[pool.clip_id.isin([ids[i] for i in perm[:k]])]) | |
| smp = pd.concat(parts).sort_values("clip_id").reset_index(drop=True) | |
| sp = run_gpt._split_of_clips(list(smp.clip_id)) | |
| bad = {c: s for c, s in sp.items() if s != "train"} | |
| assert not bad, f"non-train clips in sample: {bad}" | |
| smp["device_split"] = smp.clip_id.map(sp) | |
| cols = ["clip_id", "idx", "y", "human_label", "regime", "ep_regime", "dongle", "device_split", "t0", "group", "protocol", "row", "t_anchor"] | |
| W1D.mkdir(parents=True, exist_ok=True) | |
| csv = W1D / f"gpt_calibration_clips_{a.stamp}.csv"; txt = W1D / f"gpt_calibration_clips_{a.stamp}.txt" | |
| assert not csv.exists() and not txt.exists(), "append-only: refuse to overwrite an existing sample file" | |
| smp[[c for c in cols if c in smp]].to_csv(csv, index=False) | |
| txt.write_text("".join(f"{c}\n" for c in smp.clip_id)) | |
| h = sha256_lines(smp.clip_id) | |
| print(f"pool: {len(pool)} pilot clips in the 979 clean TRAIN consensus rows (y=1 {int((pool.y == 1).sum())}, y=0 {int((pool.y == 0).sum())}, " | |
| f"devices {pool.dongle.nunique()})") | |
| print(f"sample: n={len(smp)} (seed {a.seed}; stratified y=1 {n1} / y=0 {n0}); regime {smp.regime.value_counts().to_dict()}; " | |
| f"devices {smp.dongle.nunique()}; device_split {smp.device_split.value_counts().to_dict()}") | |
| print(f"wrote {csv}\nwrote {txt}\nclip_list_sha256 {h}") | |
| # ----------------------------------------------------------------------------- ledger helpers | |
| def batch_rows(clips, model, preset, passes=PASSES): | |
| con = sqlite3.connect(str(LEDGER)); con.row_factory = sqlite3.Row | |
| q = f"SELECT * FROM requests WHERE model=? AND preset=? AND pass IN ({','.join('?' * len(passes))})" | |
| rows = [dict(r) for r in con.execute(q, (model, preset, *passes)) if r["clip_id"] in set(clips)] | |
| total_spent = con.execute("SELECT COALESCE(SUM(actual_cost_usd),0) FROM requests WHERE status IN ('completed','incomplete','refused')").fetchone()[0] | |
| cap = con.execute("SELECT value FROM meta WHERE key='cap_usd'").fetchone()[0] | |
| con.close() | |
| return rows, float(total_spent), float(cap) | |
| def spend_summary(rows): | |
| out = {} | |
| for p in PASSES: | |
| pr = [r for r in rows if r["pass"] == p] | |
| done = [r for r in pr if r["status"] in ("completed", "incomplete", "refused")] # billed rows (incomplete after budget.py reconcile) | |
| out[p] = dict(n_rows=len(pr), n_completed=sum(r["status"] == "completed" for r in pr), | |
| n_incomplete_billed=sum(r["status"] in ("incomplete", "refused") for r in pr), | |
| n_needs_reconcile=sum(r["status"] == "needs_reconcile" for r in pr), | |
| n_failed=sum(r["status"] == "failed" for r in pr), | |
| n_other=sum(r["status"] not in ("completed", "incomplete", "refused", "needs_reconcile", "failed") for r in pr), | |
| spent_usd=float(sum(r["actual_cost_usd"] or 0 for r in done)), | |
| mean_cost_usd=float(np.mean([r["actual_cost_usd"] or 0 for r in done])) if done else float("nan"), | |
| input_tokens=int(sum(r["actual_in"] or 0 for r in done)), | |
| cached_tokens=int(sum(r["actual_cached"] or 0 for r in done)), | |
| output_tokens=int(sum(r["actual_out"] or 0 for r in done)), | |
| reasoning_tokens=int(sum(r["actual_reasoning"] or 0 for r in done)), | |
| retries=int(sum(r["retries"] or 0 for r in pr))) | |
| out[p]["cache_hit_rate"] = out[p]["cached_tokens"] / out[p]["input_tokens"] if out[p]["input_tokens"] else float("nan") | |
| return out | |
| def cmd_watch(a): | |
| clips = [l.strip() for l in Path(a.clips).read_text().splitlines() if l.strip()] | |
| rows, total_spent, cap = batch_rows(clips, a.model, a.preset) | |
| s = spend_summary(rows) | |
| done_all = [r for r in rows if r["status"] in ("completed", "incomplete", "refused")] | |
| n_done = len(done_all); spent = sum(r["actual_cost_usd"] or 0 for r in done_all) | |
| # projection rule (pre-stated): mean cost per completed request in this batch x (2 passes x n_total clips) | |
| mean_req = spent / n_done if n_done else float("nan") | |
| proj = mean_req * 2 * a.n_total if n_done else float("nan") | |
| n_rec = sum(v["n_needs_reconcile"] for v in s.values()); n_rows = sum(v["n_rows"] for v in s.values()) | |
| for p, v in s.items(): | |
| print(f"{p:<16} completed {v['n_completed']:>3} incomplete(billed) {v['n_incomplete_billed']} needs_reconcile {v['n_needs_reconcile']} failed {v['n_failed']} other {v['n_other']} " | |
| f"spent ${v['spent_usd']:.4f} mean/req ${v['mean_cost_usd']:.4f} cache_hit {v['cache_hit_rate']:.3f} retries {v['retries']}") | |
| print(f"batch: {n_done} completed requests, spent ${spent:.4f}, mean/req ${mean_req:.4f}, PROJECTED batch total ${proj:.2f} " | |
| f"(abort > ${a.abort_usd:.2f}); ledger total spent ${total_spent:.4f} / cap ${cap:.2f}") | |
| rc = 0 | |
| if n_done and proj > a.abort_usd: | |
| print(f"ABORT: projected ${proj:.2f} > ${a.abort_usd:.2f}"); rc = 3 | |
| if n_rows and n_rec / n_rows > 0.05: | |
| print(f"ABORT: needs_reconcile rate {n_rec}/{n_rows} > 5 %"); rc = 3 | |
| if total_spent >= a.stop_total_usd: | |
| print(f"ABORT: cumulative ledger spend ${total_spent:.2f} >= ${a.stop_total_usd:.2f}"); rc = 3 | |
| return rc | |
| # ----------------------------------------------------------------------------- gate | |
| def read_relation(model, clip): | |
| p = PKG / "returns" / model / "ground_relation" / f"{clip}.json" | |
| if not p.is_file(): | |
| return None | |
| return json.load(open(p)) | |
| def cmd_gate(a): | |
| smp = pd.read_csv(a.clips, dtype={"clip_id": str}) | |
| clips = list(smp.clip_id); h = sha256_lines(clips) | |
| recs = [] | |
| for r in smp.itertuples(): | |
| env = read_relation(a.model, r.clip_id) | |
| rec = dict(clip_id=r.clip_id, idx=int(r.idx), y=int(r.y), regime=r.regime, dongle=r.dongle, group=r.group, | |
| status="missing", preset=None, prompt=None, gpt_bin=None, compatibility=None, families=None, max_demand=None, | |
| n_tasks=None, vehicle_response=None, concern=None, confidence=None, reason_codes=None, cost_usd=None) | |
| if env is None: | |
| pp = PKG / "returns" / a.model / "observe_prefix" / f"{r.clip_id}.json" # why is the relation missing? (prefix incomplete -> pass skipped) | |
| rec["status"] = "missing_relation:prefix_" + (json.load(open(pp)).get("status", "?") if pp.is_file() else "absent") | |
| if env is not None: | |
| rec["status"] = env.get("status"); rec["preset"] = env.get("preset"); rec["cost_usd"] = env.get("actual_cost_usd") | |
| ann = env.get("annotation") or {} | |
| if env.get("preset") != a.preset: | |
| rec["status"] = f"preset_mismatch:{env.get('preset')}" | |
| elif ann: | |
| rec["prompt"] = ann.get("prompt_warranted_now"); rec["gpt_bin"] = BIN.get(rec["prompt"]) | |
| rec["compatibility"] = ann.get("behavior_task_compatibility") | |
| tasks = ann.get("tasks") or [] | |
| rec["families"] = "|".join(sorted({t.get("family", "?") for t in tasks})) or "none_listed" | |
| dl = [DEMAND_ORDER.get(t.get("demand_level"), -1) for t in tasks] | |
| rec["max_demand"] = ({0: "low", 1: "moderate", 2: "high"}.get(max(dl), "unknown") if dl else "none_listed") | |
| rec["n_tasks"] = len(tasks); rec["vehicle_response"] = ann.get("observed_vehicle_response") | |
| rec["concern"] = ann.get("concern_persists"); rec["confidence"] = ann.get("confidence") | |
| rec["reason_codes"] = "|".join(ann.get("reason_codes") or []) | |
| recs.append(rec) | |
| df = pd.DataFrame(recs) | |
| n = len(df); answered = df[df.prompt.notna()]; unsure = answered[answered.prompt == "unsure"] | |
| use = answered[answered.gpt_bin.notna()] | |
| k = kappa(use.y, use.gpt_bin); lo, hi, nb = boot_ci(use.y, use.gpt_bin, B=a.boot, seed=0) | |
| cm = pd.crosstab(use.y, use.gpt_bin.astype(int), rownames=["human_y"], colnames=["gpt_bin"], dropna=False).reindex(index=[0, 1], columns=[0, 1], fill_value=0) | |
| tp, fn, fp, tn = int(cm.loc[1, 1]), int(cm.loc[1, 0]), int(cm.loc[0, 1]), int(cm.loc[0, 0]) | |
| acc = (tp + tn) / max(len(use), 1); sens = tp / max(tp + fn, 1); spec = tn / max(tn + fp, 1) | |
| by_regime = {} | |
| for rg, g in use.groupby("regime"): | |
| by_regime[rg] = dict(n=len(g), kappa=kappa(g.y, g.gpt_bin), human_pos=int(g.y.sum()), gpt_pos=int(g.gpt_bin.sum())) | |
| # descriptive per-field tables vs human label (all answered clips incl. unsure) | |
| fields = {} | |
| for f in ("prompt", "compatibility", "max_demand", "families", "concern", "vehicle_response", "confidence"): | |
| ct = pd.crosstab(answered[f].fillna("NA"), answered.y, dropna=False).reindex(columns=[0, 1], fill_value=0) | |
| fields[f] = {str(kx): {"human_0": int(v[0]), "human_1": int(v[1])} for kx, v in ct.iterrows()} | |
| comp_bin = answered.compatibility.map({"sustained_incompatible": 1, "compatible": 0, "briefly_incompatible": 0}) | |
| cb = answered[comp_bin.notna()] | |
| k_comp = kappa(cb.y, comp_bin[comp_bin.notna()].astype(int)) if len(cb) else float("nan") | |
| rows, total_spent, cap = batch_rows(clips, a.model, a.preset) | |
| sp = spend_summary(rows) | |
| batch_spent = sum(v["spent_usd"] for v in sp.values()) | |
| tot_in = sum(v["input_tokens"] for v in sp.values()); tot_cached = sum(v["cached_tokens"] for v in sp.values()) | |
| passed = bool(np.isfinite(k) and k >= a.gate) | |
| res = dict(stamp=a.stamp, model=a.model, preset=a.preset, clip_list_csv=str(a.clips), clip_list_sha256=h, n_clips=n, | |
| n_relation_returns=int(df.prompt.notna().sum()), n_missing=int(df.status.astype(str).str.startswith("missing").sum()), | |
| status_counts=Counter(df.status.astype(str)).most_common(), | |
| unsure_n=int(len(unsure)), unsure_rate=float(len(unsure) / max(len(answered), 1)), | |
| n_used_for_kappa=int(len(use)), human_pos_used=int(use.y.sum()), gpt_pos_used=int(use.gpt_bin.sum()), | |
| kappa=k, kappa_ci95=[lo, hi], bootstrap_B=a.boot, bootstrap_valid=nb, gate_threshold=a.gate, gate_passed=passed, | |
| confusion=dict(tp=tp, fn=fn, fp=fp, tn=tn), accuracy=acc, sensitivity=sens, specificity=spec, | |
| prompt_value_counts=Counter(answered.prompt).most_common(), kappa_by_regime=by_regime, | |
| kappa_compatibility_sustained_vs_human=k_comp, n_compatibility_binarised=int(len(cb)), | |
| fields_vs_human=fields, spend=dict(batch_usd=batch_spent, per_pass=sp, ledger_total_usd=total_spent, cap_usd=cap, | |
| cache_hit_rate=(tot_cached / tot_in if tot_in else float("nan")), | |
| input_tokens=tot_in, cached_tokens=tot_cached), | |
| per_clip=df.to_dict(orient="records")) | |
| out_dir = Path(a.out_dir) if a.out_dir else W1D; out_dir.mkdir(parents=True, exist_ok=True) | |
| jp = out_dir / f"gpt_calibration_{a.stamp}.json"; mp = out_dir / f"gpt_calibration_{a.stamp}.md" | |
| assert not jp.exists() and not mp.exists(), "append-only: refuse to overwrite an existing report" | |
| json.dump(res, open(jp, "w"), indent=1, default=float) | |
| L = [f"# GPT weak-label calibration batch — {a.stamp}", "", | |
| f"Model `{a.model}`, preset `{a.preset}`, detail high, passes observe_prefix + ground_relation (sync). Clip list `{Path(a.clips).name}` " | |
| f"(n = {n}, sha256 `{h}`), human labels = the 979 clean TRAIN consensus rows (`cwm3/data/labels.py`); TRAIN devices only.", "", | |
| "## Gate (pre-registered: Cohen's kappa of binarised `prompt_warranted_now` vs human label >= 0.6)", "", | |
| f"* ground_relation returns: {res['n_relation_returns']} / {n} (missing {res['n_missing']}; status counts {res['status_counts']})", | |
| f"* unsure: {res['unsure_n']} / {len(answered)} answered = {res['unsure_rate']:.3f} (excluded from kappa)", | |
| f"* used for kappa: n = {len(use)} (human warranted {int(use.y.sum())}, GPT warranted {int(use.gpt_bin.sum())})", | |
| f"* **kappa = {k:.3f}**, 95 % clip-bootstrap CI [{lo:.3f}, {hi:.3f}] (B = {a.boot}, {nb} valid) -> **{'PASS' if passed else 'FAIL'}** (threshold {a.gate})", | |
| f"* confusion (human rows x GPT cols): tn {tn} fp {fp} / fn {fn} tp {tp}; accuracy {acc:.3f}, sensitivity {sens:.3f}, specificity {spec:.3f}", | |
| f"* prompt_warranted_now counts: {res['prompt_value_counts']}", | |
| f"* by regime (descriptive): " + "; ".join(f"{r}: n {v['n']} kappa {v['kappa']:.3f} (human+ {v['human_pos']}, GPT+ {v['gpt_pos']})" for r, v in by_regime.items()), | |
| f"* compatibility binarised (sustained_incompatible -> 1) vs human: kappa {k_comp:.3f} on n = {len(cb)} (descriptive)", "", | |
| "## Per-field distributions vs the human label (descriptive; counts human_0 / human_1)", ""] | |
| for f, tab in fields.items(): | |
| L += [f"### {f}", "", "| value | human not_warranted | human warranted |", "|---|---|---|"] | |
| L += [f"| {v} | {c['human_0']} | {c['human_1']} |" for v, c in tab.items()] | |
| L.append("") | |
| L += ["## Spend (ledger `cwm3/weak/budget_ledger.sqlite`, rows of this batch)", "", | |
| "| pass | completed | incomplete (billed) | needs_reconcile | failed | spent USD | mean USD/req | input tok | cached tok | cache hit | output tok | reasoning tok | retries |", "|---|---|---|---|---|---|---|---|---|---|---|---|---|"] | |
| for p, v in sp.items(): | |
| L.append(f"| {p} | {v['n_completed']} | {v['n_incomplete_billed']} | {v['n_needs_reconcile']} | {v['n_failed']} | {v['spent_usd']:.4f} | {v['mean_cost_usd']:.4f} | {v['input_tokens']} | " | |
| f"{v['cached_tokens']} | {v['cache_hit_rate']:.3f} | {v['output_tokens']} | {v['reasoning_tokens']} | {v['retries']} |") | |
| L += ["", f"Batch spend **${batch_spent:.4f}**; ledger total spent ${total_spent:.4f} of cap ${cap:.2f}; cache hit rate (cached / input tokens) " | |
| f"{res['spend']['cache_hit_rate']:.3f}.", "", f"JSON: `{jp.name}` (per-clip rows included)."] | |
| mp.write_text("\n".join(L) + "\n") | |
| print("\n".join(L[:14])) | |
| print(f"wrote {mp}\nwrote {jp}") | |
| return 0 if passed else 4 | |
| def main(): | |
| ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) | |
| sub = ap.add_subparsers(dest="cmd", required=True) | |
| p = sub.add_parser("sample"); p.add_argument("--n", type=int, default=100); p.add_argument("--seed", type=int, default=0); p.add_argument("--stamp", required=True) | |
| p = sub.add_parser("watch"); p.add_argument("--clips", required=True); p.add_argument("--n-total", type=int, default=100) | |
| p.add_argument("--model", default="gpt-5"); p.add_argument("--preset", default="economy"); p.add_argument("--abort-usd", type=float, default=12.0) | |
| p.add_argument("--stop-total-usd", type=float, default=90.0) | |
| p = sub.add_parser("gate"); p.add_argument("--clips", required=True); p.add_argument("--stamp", required=True); p.add_argument("--model", default="gpt-5") | |
| p.add_argument("--preset", default="economy"); p.add_argument("--gate", type=float, default=0.6); p.add_argument("--boot", type=int, default=2000); p.add_argument("--out-dir", default="", help="default cwm2/results/W1d; use a scratch dir to test on partial returns") | |
| a = ap.parse_args() | |
| return {"sample": cmd_sample, "watch": cmd_watch, "gate": cmd_gate}[a.cmd](a) | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 19.9 kB
- Xet hash:
- ba63f66e6f824bc6cceddc89cb43a501c1419e8bac016e9e9117ea7d4b59a243
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.