Buckets:
| #!/usr/bin/env python3 | |
| """tools/vlm_handson/build_validation_frames.py — frame-level hands-on validation set for the VLM hands-on gold check (pre-registration | |
| docs/decisions.md 2026-09-16T23:26Z "vision-based hands-on gold"). NEW files only; read-only on every input. | |
| Sources (the owner's clicks in results/review_tool/labels_latest.json): | |
| (a) pass-4 items results/review_tool/items_relabel_hr_engaged_r4.json (ids <orig>__r4, 21 frames dt = -1.0 … +1.0 s): | |
| never_on -> all frames 0 | always_on -> all frames 1 | ret_dt=<dt> -> 0 for dt < click, 1 for dt >= click | unsure / unlabelled -> item skipped | |
| (b) v3 page items results/review_tool/items_relabel_v3.json (ids V3_*, 31 frames dt = -2.0 … +1.0 s): same rules. | |
| Frames are RE-EXTRACTED CLEAN (no red box, no text) from the local videos with exactly the review tool's convention | |
| (tools/review_tool/build_review_tool.py: VIO.video_path, grab = plain cv2 seek, FPS = 20, frame index round(20 t)), so that the VLM sees the | |
| same frame the owner labelled; each clean frame is checked against the owner's overlay jpg (mean |diff| outside the box / text band after the | |
| tool's 640-px resize) and the check is stored per frame. The causal driver box (results/pose_causal_routes[_v2]/<uid>.npz, native px) at the | |
| frame is stored with the frame; the driver side is the route-median box centre relative to the frame centre (dside_prior is NaN on most routes). | |
| Output: results/vlm_handson/validation_frames.json + results/vlm_handson/frames_clean/<item>_<k>.jpg (quality 95) + build log. | |
| python tools/vlm_handson/build_validation_frames.py [--out results/vlm_handson] | |
| """ | |
| from __future__ import annotations | |
| import argparse, json, sys, time | |
| from pathlib import Path | |
| import numpy as np | |
| CWM3 = Path(__file__).resolve().parents[2] | |
| sys.path.insert(0, str(CWM3 / "tools/review_tool")); sys.path.insert(0, str(CWM3 / "common")) | |
| import build_review_tool as B # noqa: E402 (VIO, grab, nearest_pose_row, FPS) | |
| from runmeta import write_run_json # noqa: E402 | |
| RT = CWM3 / "results/review_tool" | |
| ITEMS_R4 = RT / "items_relabel_hr_engaged_r4.json" | |
| ITEMS_V3 = RT / "items_relabel_v3.json" | |
| LABELS = RT / "labels_latest.json" | |
| SKIP = ("", "unsure") | |
| TEXT_BAND_PX = 32 # overlay text band at the top of the 640-px review image | |
| BOX_DILATE_PX = 8 # red rectangle is 4 px thick in native px -> ~2 px at 640; dilate generously | |
| def frame_labels(label: str, dts): | |
| """owner item label -> per-frame 0/1 list (None -> skip item) + click dt (None if not a clicked return).""" | |
| if label in SKIP: | |
| return None, None | |
| if label == "never_on": | |
| return [0] * len(dts), None | |
| if label == "always_on": | |
| return [1] * len(dts), None | |
| if label.startswith("ret_dt="): | |
| c = float(label.split("=", 1)[1]) | |
| return [int(dt >= c - 1e-6) for dt in dts], c | |
| raise ValueError(f"unexpected label {label!r}") | |
| def overlay_check(clean_bgr, overlay_path: Path, box_native): | |
| """mean |clean(resized like the tool) - overlay| outside the (dilated) box and the text band; also the same statistic for a +-2-frame | |
| neighbour is NOT computed here (one seek per frame keeps the tool convention) — a small value (JPEG noise, ~2-4) confirms the same frame.""" | |
| import cv2 | |
| ov = cv2.imread(str(overlay_path)) | |
| if ov is None: | |
| return None | |
| H, W = clean_bgr.shape[:2] | |
| s = ov.shape[1] / W | |
| cl = cv2.resize(clean_bgr, (ov.shape[1], int(round(H * s))), interpolation=cv2.INTER_AREA) | |
| if cl.shape != ov.shape: | |
| return None | |
| m = np.ones(ov.shape[:2], bool) | |
| m[:TEXT_BAND_PX, :] = False | |
| if box_native is not None and np.all(np.isfinite(box_native)): | |
| x0, y0, x1, y1 = [int(round(float(v) * s)) for v in box_native] | |
| d = BOX_DILATE_PX | |
| # mask the rectangle outline only (the interior is compared too — it is the driver) | |
| for (a0, b0, a1, b1) in ((x0 - d, y0 - d, x1 + d, y0 + d), (x0 - d, y1 - d, x1 + d, y1 + d), (x0 - d, y0 - d, x0 + d, y1 + d), (x1 - d, y0 - d, x1 + d, y1 + d)): | |
| m[max(0, b0):max(0, b1), max(0, a0):max(0, a1)] = False | |
| diff = np.abs(cl.astype(np.int16) - ov.astype(np.int16)).mean(axis=2) | |
| return float(diff[m].mean()) if m.any() else None | |
| def main(argv=None) -> int: | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--out", default=str(CWM3 / "results/vlm_handson")) | |
| a = ap.parse_args(argv) | |
| import cv2 | |
| cv2.setNumThreads(2) | |
| t0 = time.time() | |
| out = Path(a.out); img_dir = out / "frames_clean"; img_dir.mkdir(parents=True, exist_ok=True) | |
| logf = open(out / "build_validation_frames.log", "a") | |
| def log(s): | |
| print(s, flush=True); logf.write(s + "\n"); logf.flush() | |
| labels = json.load(open(LABELS, encoding="utf-8")) | |
| r4 = json.load(open(ITEMS_R4, encoding="utf-8")) | |
| v3 = json.load(open(ITEMS_V3, encoding="utf-8")) | |
| src_items = [("pass4", it) for it in r4["items"]] + [("v3", it) for it in v3["items"] if it["id"].startswith("V3_")] | |
| items, skipped = [], [] | |
| for src, it in src_items: | |
| lab = labels.get(it["id"], {}).get("label", "") | |
| dts = [float(f["dt"]) for f in it["frames"]] | |
| y, click = frame_labels(lab, dts) | |
| if y is None: | |
| skipped.append({"id": it["id"], "source": src, "label": lab or "<unlabelled>"}); continue | |
| items.append({"id": it["id"], "source": src, "orig_id": it["meta"].get("orig_id", it["id"]), "uid": it["meta"]["uid"], "car": it["meta"]["car"], | |
| "regime": it["meta"]["regime"], "event": it["meta"]["event"], "t_event": float(it["meta"]["t_event"]), "owner_label": lab, | |
| "click_dt": click, "rater": labels[it["id"]].get("rater"), "label_ts": labels[it["id"]].get("server_ts"), | |
| "frames": [{"k": k, "t": float(f["t"]), "dt": dts[k], "frame_idx": int(round(float(f["t"]) * B.FPS)), "label": int(y[k]), | |
| "file_overlay": f"results/review_tool/{f['file']}"} for k, f in enumerate(it["frames"])]}) | |
| log(f"labelled items: {len(items)} (pass4 {sum(i['source']=='pass4' for i in items)}, v3 {sum(i['source']=='v3' for i in items)}); skipped {len(skipped)}: " | |
| + ", ".join(f"{s['id']}={s['label']}" for s in skipped)) | |
| for it in items: | |
| if it["source"] == "v3": | |
| for f, g in zip(it["frames"], next(x for x in v3["items"] if x["id"] == it["id"])["frames"]): | |
| assert f["frame_idx"] == int(g["frame_idx"]), (it["id"], f, g) # v3 items carry the frame index used by the page | |
| # ---- extraction, one video at a time | |
| by_uid = {} | |
| for it in items: | |
| by_uid.setdefault(it["uid"], []).append(it) | |
| dside = {} | |
| n_fr = n_nan = 0; checks = [] | |
| for uid, its in by_uid.items(): | |
| vp = B.VIO.video_path(uid) | |
| boxes = B.VIO.load_driver_boxes(uid) | |
| W, H = boxes["native_wh"] | |
| finite = boxes["box"][np.all(np.isfinite(boxes["box"]), 1)] | |
| cx_med = float(np.median((finite[:, 0] + finite[:, 2]) / 2) / W) if len(finite) else float("nan") | |
| with np.load(B.VIO.pose_path(uid), allow_pickle=True) as z: | |
| dp, ds = float(z["dside_prior"]), str(z["dside_source"]) | |
| side = "right" if cx_med >= 0.5 else "left" | |
| dside[uid] = {"box_cx_median_frac": round(cx_med, 4), "driver_side": side, "dside_prior": None if np.isnan(dp) else dp, "dside_source": ds, | |
| "pose_source": boxes["source"], "native_wh": [W, H], "header_fps": round(B.VIO.video_header_fps(vp), 4)} | |
| cap = cv2.VideoCapture(str(vp)) | |
| for it in its: | |
| it["driver_side"] = side; it["native_wh"] = [W, H] | |
| fis = np.array([f["frame_idx"] for f in it["frames"]], np.int64) | |
| bx, _ = B.VIO.boxes_at(boxes, fis) | |
| for f, b in zip(it["frames"], bx): | |
| src_b = "pose_exact" | |
| if not np.all(np.isfinite(b)): | |
| j = B.nearest_pose_row(boxes, f["t"]) | |
| if j is not None and np.all(np.isfinite(boxes["box"][j])): | |
| b = boxes["box"][j]; src_b = "pose_nearest<=4fr" | |
| else: | |
| b = None; src_b = "none"; n_nan += 1 | |
| fr = B.grab(cap, f["frame_idx"]) | |
| if fr is None: | |
| raise RuntimeError(f"{it['id']} frame {f['frame_idx']} not decodable from {vp}") | |
| assert fr.shape[1] == W and fr.shape[0] == H, (fr.shape, W, H) | |
| fn = f"{it['id']}_{f['k']:02d}.jpg" | |
| cv2.imwrite(str(img_dir / fn), fr, [cv2.IMWRITE_JPEG_QUALITY, 95]) | |
| chk = overlay_check(fr, CWM3 / f["file_overlay"], b) | |
| f.update({"file_clean": f"results/vlm_handson/frames_clean/{fn}", "box_native": None if b is None else [round(float(v), 1) for v in b], | |
| "box_source": src_b, "overlay_check_mad": None if chk is None else round(chk, 3)}) | |
| if chk is not None: | |
| checks.append(chk) | |
| n_fr += 1 | |
| cap.release() | |
| log(f" {uid[:60]:60s} items {len(its):2d} side={side} cx={cx_med:.2f} fps_hdr={dside[uid]['header_fps']} frames so far {n_fr} {time.time()-t0:6.1f}s") | |
| checks = np.array(checks) | |
| worst = sorted(((f["overlay_check_mad"], it["id"], f["k"]) for it in items for f in it["frames"] if f["overlay_check_mad"] is not None), reverse=True)[:8] | |
| log(f"frames {n_fr}; box undefined {n_nan}; overlay check MAD: median {np.median(checks):.2f} p95 {np.percentile(checks, 95):.2f} max {checks.max():.2f}; worst {worst}") | |
| n_pos = sum(f["label"] for it in items for f in it["frames"]) | |
| summary = {"n_items": len(items), "n_frames": n_fr, "n_pos": n_pos, "n_neg": n_fr - n_pos, | |
| "by_source": {s: {"items": sum(i["source"] == s for i in items), "frames": sum(len(i["frames"]) for i in items if i["source"] == s)} for s in ("pass4", "v3")}, | |
| "by_regime": {r: {"items": sum(i["regime"] == r for i in items), "frames": sum(len(i["frames"]) for i in items if i["regime"] == r)} for r in ("engaged", "manual")}, | |
| "by_owner_label": {}, "clicked_returns": [{"id": i["id"], "regime": i["regime"], "click_dt": i["click_dt"], "t_click": round(i["t_event"] + i["click_dt"], 2)} for i in items if i["click_dt"] is not None], | |
| "overlay_check_mad": {"median": round(float(np.median(checks)), 3), "p95": round(float(np.percentile(checks, 95)), 3), "max": round(float(checks.max()), 3), | |
| "note": "mean |clean frame resized as the tool - owner overlay jpg| outside the box outline and text band; JPEG-noise-level values (< ~6) = same frame"}, | |
| "n_box_undefined": n_nan} | |
| for i in items: | |
| k = i["owner_label"].split("=")[0]; summary["by_owner_label"][k] = summary["by_owner_label"].get(k, 0) + 1 | |
| doc = {"built_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "purpose": "frame-level hands-on validation set for the VLM hands-on gold pre-registration (decisions 2026-09-16T23:26Z)", | |
| "provenance": {"labels": str(LABELS.relative_to(CWM3)), "items_pass4": str(ITEMS_R4.relative_to(CWM3)), "items_v3": str(ITEMS_V3.relative_to(CWM3)), | |
| "label_rules": {"never_on": "all frames 0", "always_on": "all frames 1", "ret_dt=<c>": "0 for dt < c, 1 for dt >= c", "unsure/unlabelled": "item skipped"}, | |
| "frame_convention": "tools/review_tool/build_review_tool.py: cv2 plain seek (grab) at frame_idx = round(20 t), FPS = 20 — identical to the frames the owner labelled", | |
| "box": "results/pose_causal_routes_v2 preferred else results/pose_causal_routes, key box (native px) at the exact frame_idx (else nearest row within 4 frames)", | |
| "driver_side": "route-median box centre x vs frame centre (dside_prior is NaN / population prior on these routes)", | |
| "clean_frames": "JPEG quality 95 of the raw decoded frame, native resolution, no overlay"}, | |
| "skipped_items": skipped, "driver_side_by_uid": dside, "summary": summary, "items": items} | |
| (out / "validation_frames.json").write_text(json.dumps(doc, indent=1, ensure_ascii=False)) | |
| write_run_json(out, "build_validation_frames", [Path(__file__), CWM3 / "tools/review_tool/build_review_tool.py"], [LABELS, ITEMS_R4, ITEMS_V3], seed=None, t_start=t0, | |
| extra={"summary": summary}, input_listing=False) | |
| log(f"summary {json.dumps(summary)}") | |
| log(f"VALIDATION_FRAMES_DONE {time.time()-t0:.1f}s") | |
| return 0 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |
Xet Storage Details
- Size:
- 12.6 kB
- Xet hash:
- fada8bf8b49a3f6091c133f04ef6550df3397caeaf238ee99a5d9341888ecbb2
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.