Buckets:
| """cwm3/weak/gpt_main_batch.py — MAIN GPT weak-label batch (only after the calibration gate PASSED): clip selection and local frame extraction. | |
| python gpt_main_batch.py select --n 600 --seed 0 --stamp <UTC> # no API, no GPU | |
| python gpt_main_batch.py extract --clips <txt> [--limit N] # ONE CUDA context; run under `flock /tmp/s13_gpu.lock`; resumable; never ships/evicts | |
| Selection (pre-registered 2026-09-13): (A) the pilot clips of cwm2/annotation_package that are NOT in the 979 clean TRAIN consensus rows | |
| (frames already on disk; TRAIN devices); (B) TRAIN clips of cwm2/private/clip_map_full.json WITHOUT a human label (label None), regime in | |
| {manual, engaged} stratified 50/50, within regime a seed-0 numpy.default_rng permutation of the SORTED clip ids, keeping clips whose | |
| BATON videos and CAN evidence exist locally, until n - |A| clips. Every clip is re-asserted TRAIN via run_gpt._split_of_clips. | |
| Extraction = extract_full.process (same layout / ROI rule / density as the shipped full set: wide 2 Hz, crops 5 Hz, causal DMS-free ROI with the | |
| dongle driver-side prior of private/person_diag.json) into cwm2/ship/dataset_full/frames/<clip_id>/; the shards ledger is untouched. | |
| Frames for (B) are read by the runner with CWM2_DATASET_ROOT=cwm2/ship/dataset_full; (A) uses the package's own root. | |
| """ | |
| from __future__ import annotations | |
| import argparse, collections, hashlib, json, os, sys, time | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1]; EXPLORE = CWM3.parent; CWM2 = EXPLORE / "cwm2" | |
| PKG = CWM2 / "annotation_package"; W1D = CWM2 / "results" / "W1d"; DS_FULL = CWM2 / "ship" / "dataset_full" | |
| CLIP_MAP_FULL = CWM2 / "private" / "clip_map_full.json"; CLIP_MAP_PILOT = CWM2 / "private" / "clip_map.json" | |
| def sha256_lines(ids): | |
| return hashlib.sha256(("\n".join(sorted(ids)) + "\n").encode()).hexdigest() | |
| def cmd_select(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 | |
| tab = train_label_table(("consensus",)); assert len(tab) == 979 | |
| pil = json.load(open(CLIP_MAP_PILOT)); full = json.load(open(CLIP_MAP_FULL)) | |
| idx979 = set(tab.idx) | |
| A = sorted(c for c, v in pil.items() if v["idx"] not in idx979 and (PKG / "frames" / c / "index.json").is_file()) | |
| rows = [dict(clip_id=c, source="pilot_leftover", idx=pil[c]["idx"], uid=pil[c]["uid"], dongle=pil[c]["dongle"], t0=pil[c]["t0"], regime=pil[c]["regime"], | |
| ctrl=pil[c]["ctrl"], human_label=pil[c].get("label"), group=pil[c].get("group"), dataset_root=str(PKG)) for c in A] | |
| nB = a.n - len(A); per = {"manual": nB // 2, "engaged": nB - nB // 2} | |
| rng = np.random.default_rng(a.seed); ev = DS_FULL / "evidence"; dropped = collections.Counter() | |
| for rg in ("manual", "engaged"): | |
| ids = sorted(c for c, v in full.items() if v.get("label") is None and v["regime"] == rg) | |
| perm = rng.permutation(len(ids)); taken = 0 | |
| for i in perm: | |
| if taken >= per[rg]: | |
| break | |
| c = ids[i]; v = full[c] | |
| if not (os.path.exists(v["dcam"]) and os.path.exists(v["qcam"])): | |
| dropped["video_missing"] += 1; continue | |
| if not ((ev / f"{c}.prefix.json").is_file() and (ev / f"{c}.outcome.json").is_file()): | |
| dropped["evidence_missing"] += 1; continue | |
| rows.append(dict(clip_id=c, source="full_unlabelled", idx=v["idx"], uid=v["uid"], dongle=v["dongle"], t0=v["t0"], regime=rg, ctrl=v["ctrl"], | |
| human_label=None, group=None, dataset_root=str(DS_FULL))); taken += 1 | |
| df = pd.DataFrame(rows) | |
| sp = run_gpt._split_of_clips(list(df.clip_id)); bad = {c: s for c, s in sp.items() if s != "train"} | |
| assert not bad, f"non-train clips selected: {bad}" | |
| df["device_split"] = df.clip_id.map(sp) | |
| W1D.mkdir(parents=True, exist_ok=True) | |
| csv = W1D / f"gpt_main_clips_{a.stamp}.csv"; txtA = W1D / f"gpt_main_clips_{a.stamp}_pilot.txt"; txtB = W1D / f"gpt_main_clips_{a.stamp}_full.txt" | |
| for p in (csv, txtA, txtB): | |
| assert not p.exists(), f"append-only: {p} exists" | |
| df.to_csv(csv, index=False) | |
| txtA.write_text("".join(f"{c}\n" for c in df[df.source == "pilot_leftover"].clip_id)) | |
| txtB.write_text("".join(f"{c}\n" for c in df[df.source == "full_unlabelled"].clip_id)) | |
| print(f"selected {len(df)}: pilot_leftover {len(A)} (labels {collections.Counter(df[df.source == 'pilot_leftover'].human_label.fillna('None')).most_common()}), " | |
| f"full_unlabelled {int((df.source == 'full_unlabelled').sum())} (manual {per['manual']} / engaged {per['engaged']}); dropped {dict(dropped)}; " | |
| f"devices {df.dongle.nunique()}; ctrl {collections.Counter(df.ctrl).most_common()}; device_split {collections.Counter(df.device_split)}") | |
| print(f"wrote {csv}\nwrote {txtA}\nwrote {txtB}\nclip_list_sha256 {sha256_lines(df.clip_id)}") | |
| def cmd_extract(a): | |
| sys.path.insert(0, str(CWM2 / "data")) | |
| import extract_frames as EF | |
| import extract_full as XF | |
| t_start = time.strftime("%Y-%m-%dT%H:%M:%S"); t = time.time() | |
| from ultralytics import YOLO | |
| model = YOLO(str(EF.YOLO_W)) | |
| priv = json.load(open(CLIP_MAP_FULL)) | |
| ids = [l.strip() for l in Path(a.clips).read_text().splitlines() if l.strip()] | |
| if a.limit: | |
| ids = ids[:a.limit] | |
| diag = json.load(open(CWM2 / "private" / "person_diag.json")); dside = dict(diag.get("dside", {})) # read-only prior | |
| print(f"clips={len(ids)} -> {XF.FR} dongle side priors={len(dside)}", flush=True) | |
| readers = XF.Readers(); n_ok = n_skip = 0; fails = []; rules = collections.Counter() | |
| for i, cid in enumerate(ids): | |
| try: | |
| r, _ = XF.process(cid, priv[cid], model, readers, dside) | |
| n_ok += r == "ok"; n_skip += r == "skip" | |
| if r == "ok": | |
| rules[json.load(open(XF.FR / cid / "roi.json")).get("selection_rule")] += 1 | |
| except Exception as e: # noqa: BLE001 | |
| fails.append((cid, repr(e)[:200])) | |
| if (i + 1) % 20 == 0: | |
| print(f"{i + 1}/{len(ids)} ok={n_ok} skip={n_skip} fail={len(fails)} {time.time() - t:.0f}s", flush=True) | |
| W1D.mkdir(parents=True, exist_ok=True) | |
| run = dict(item="W1d_gpt_main", step="extract", command=" ".join(sys.argv), started=t_start, finished=time.strftime("%Y-%m-%dT%H:%M:%S"), | |
| code_sha256={p.name: XF.sha256(p) for p in (Path(__file__), CWM2 / "data" / "extract_full.py", CWM2 / "data" / "extract_frames.py")}, | |
| inputs_sha256={str(p): XF.sha256(p) for p in (Path(a.clips), CLIP_MAP_FULL, CWM2 / "private" / "person_diag.json", EF.YOLO_W)}, | |
| params=dict(wide_hz=XF.WIDE_HZ, crop_hz=XF.CROP_HZ, wide_w=XF.WIDE_W, head_sz=XF.HEAD_SZ, hand_sz=XF.HAND_SZ, jpg_q=XF.JPG_Q, detector="yolo11x-pose"), | |
| n_clips=len(ids), ok=n_ok, skip=n_skip, fail=len(fails), fails=fails, selection_rule_this_run=dict(rules), elapsed_s=round(time.time() - t, 1)) | |
| with open(W1D / "gpt_main_extract_run.json", "a") as f: | |
| f.write(json.dumps(run) + "\n") | |
| print(f"EXTRACT_GPT_MAIN_DONE ok={n_ok} skip={n_skip} fail={len(fails)} rules={dict(rules)} in {time.time() - t:.0f}s", flush=True) | |
| def main(): | |
| ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) | |
| sub = ap.add_subparsers(dest="cmd", required=True) | |
| p = sub.add_parser("select"); p.add_argument("--n", type=int, default=600); p.add_argument("--seed", type=int, default=0); p.add_argument("--stamp", required=True) | |
| p = sub.add_parser("extract"); p.add_argument("--clips", required=True); p.add_argument("--limit", type=int, default=0) | |
| a = ap.parse_args() | |
| return {"select": cmd_select, "extract": cmd_extract}[a.cmd](a) | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 7.91 kB
- Xet hash:
- 7183536d995fc36593ea735d3c907964a8dee7e31318bc948cf61977b1b918fa
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.