Buckets:
| """cwm3/data/baton_hours.py — the BATON hours the paper may cite, computed from the sources the CWM3 pipeline actually uses. | |
| Benchmark subset = the s12 anchor grid (s12_wm/results/targets_{manual,engaged}_{train,val,test}.npz: uid, dongle, t; anchors on a 0.2 s | |
| stride). Per route the VIDEO duration is the cabin video's frame count / 20 (frame/20 axis, the pipeline's time base; frame counts from | |
| results/pack_v2/container_fps.json = cv2/ffprobe headers of the canonical files) and is partitioned by the canonical regime flag of the | |
| 10 Hz CAN grid (s14_corr/cache/grid10/<uid>.npz `engaged` = canspec_v2 cs_enabled OR cruiseState_enabled with 1 s debounce): engaged hours, | |
| manual hours, unknown = video time not covered by the CAN grid. Anchor hours = n_rows × 0.2 s. Devices = distinct dongles. | |
| Full corpus = the 1,489 canonical leaves of results/baton_leaves.json; their durations are probed once on the remote (ffprobe nb_frames / | |
| duration of dcamera.mp4, one batched ssh over the ControlMaster; cached in results/baton_full_corpus_durations.json). | |
| Writes results/tables/baton_hours.md (+ baton_hours.json, run json).""" | |
| import json, subprocess, sys, time | |
| from pathlib import Path | |
| import numpy as np | |
| ROOT = Path(__file__).resolve().parents[1]; EXPLORE = ROOT.parent; sys.path.insert(0, str(ROOT / "common")) | |
| from runmeta import write_run_json | |
| S12R = EXPLORE / "s12_wm" / "results"; GRID10 = EXPLORE / "s14_corr" / "cache" / "grid10"; FPS = ROOT / "results" / "pack_v2" / "container_fps.json" | |
| LEAVES = ROOT / "results" / "baton_leaves.json"; DUR_CACHE = ROOT / "results" / "baton_full_corpus_durations.json" | |
| REMOTE = "USER@REMOTE_HOST"; RROOT = "/data/backup/BATON"; SSH = ["ssh", "-S", "/tmp/baton_bak_ctl", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20"] | |
| REGIMES, SPLITS = ("manual", "engaged"), ("train", "val", "test") | |
| def leaf_of_uid(u): p = u.split("__"); return "/".join(p[:3]) + "/" + "__".join(p[3:]) | |
| def full_corpus_durations(leaves): | |
| if DUR_CACHE.exists(): return json.load(open(DUR_CACHE)) | |
| out = {} | |
| for i in range(0, len(leaves), 150): | |
| chunk = leaves[i:i + 150] | |
| cmd = "; ".join(f"printf '%s\\t' '{l}'; ffprobe -v error -select_streams v:0 -show_entries stream=nb_frames,duration,avg_frame_rate -of csv=p=0 '{RROOT}/{l}/dcamera.mp4' 2>/dev/null || echo ERR" for l in chunk) | |
| r = subprocess.run(SSH + [REMOTE, cmd], capture_output=True, text=True, timeout=3600) | |
| for line in r.stdout.splitlines(): | |
| parts = line.split("\t") | |
| if len(parts) != 2: continue | |
| l, v = parts; f = v.split(",") | |
| # ffprobe csv field order is avg_frame_rate, duration, nb_frames (ffprobe orders by its own stream-entry order, not by the request) | |
| out[l] = None if v.strip() == "ERR" or len(f) < 3 else dict(raw=v.strip(), avg_frame_rate=f[0], duration=float(f[1]) if f[1] not in ("", "N/A") else None, nb_frames=int(f[2]) if f[2].isdigit() else None) | |
| print(f" probed {min(i + 150, len(leaves))}/{len(leaves)}", flush=True) | |
| DUR_CACHE.write_text(json.dumps(out, indent=1)); return out | |
| def main(): | |
| t_start = time.time(); fps = json.load(open(FPS)); lv = json.load(open(LEAVES)) | |
| # ---------------- benchmark subset (s12 grid) | |
| rows = {}; route_split = {}; route_regimes = {} | |
| for r in REGIMES: | |
| for s in SPLITS: | |
| z = np.load(S12R / f"targets_{r}_{s}.npz", allow_pickle=True); u, c = np.unique(z["uid"], return_counts=True) | |
| rows[(r, s)] = dict(n_rows=int(len(z["t"])), routes=set(u.tolist()), dongles=set(np.unique(z["dongle"]).tolist())) | |
| for uu in u: route_split[uu] = s; route_regimes.setdefault(uu, set()).add(r) | |
| routes = sorted(route_split) | |
| video = {} | |
| for u in routes: | |
| nfr = (fps.get(u, {}).get("cabin") or {}).get("n_frames") or (fps.get(u, {}).get("front") or {}).get("n_frames") | |
| vid_h = nfr / 20.0 / 3600 if nfr else float("nan") | |
| g = GRID10 / f"{u}.npz"; eng_h = man_h = grid_h = 0.0 | |
| if g.exists(): | |
| zz = np.load(g); e = zz["engaged"].astype(bool); grid_h = len(e) / 36000; eng_h = e.sum() / 36000; man_h = (~e).sum() / 36000 | |
| video[u] = dict(video_h=vid_h, grid_h=grid_h, engaged_h=eng_h, manual_h=man_h, unknown_h=max(0.0, (vid_h if vid_h == vid_h else grid_h) - grid_h), split=route_split[u], dongle=u.split("__")[1], has_fps=bool(nfr), has_grid=g.exists()) | |
| table = {} | |
| for s in SPLITS: | |
| rs = [u for u in routes if route_split[u] == s] | |
| table[s] = dict(routes=len(rs), devices=len({video[u]["dongle"] for u in rs}), video_h=float(np.nansum([video[u]["video_h"] for u in rs])), routes_without_fps=sum(not video[u]["has_fps"] for u in rs), routes_without_grid=sum(not video[u]["has_grid"] for u in rs), | |
| video_manual_h=sum(video[u]["manual_h"] for u in rs), video_engaged_h=sum(video[u]["engaged_h"] for u in rs), video_unknown_h=sum(video[u]["unknown_h"] for u in rs), grid_h=sum(video[u]["grid_h"] for u in rs), | |
| anchors={r: dict(n_rows=rows[(r, s)]["n_rows"], hours=rows[(r, s)]["n_rows"] * 0.2 / 3600, routes=len(rows[(r, s)]["routes"]), devices=len(rows[(r, s)]["dongles"])) for r in REGIMES}) | |
| tot = dict(routes=len(routes), devices=len({video[u]["dongle"] for u in routes}), video_h=float(np.nansum([v["video_h"] for v in video.values()])), video_manual_h=sum(v["manual_h"] for v in video.values()), video_engaged_h=sum(v["engaged_h"] for v in video.values()), | |
| video_unknown_h=sum(v["unknown_h"] for v in video.values()), grid_h=sum(v["grid_h"] for v in video.values()), anchor_h={r: sum(rows[(r, s)]["n_rows"] for s in SPLITS) * 0.2 / 3600 for r in REGIMES}, anchor_rows={r: sum(rows[(r, s)]["n_rows"] for s in SPLITS) for r in REGIMES}, | |
| routes_without_fps=sum(not v["has_fps"] for v in video.values()), routes_without_grid=sum(not v["has_grid"] for v in video.values())) | |
| tot["anchor_h_total"] = sum(tot["anchor_h"].values()); tot["anchor_rows_total"] = sum(tot["anchor_rows"].values()) | |
| # ---------------- full corpus (1,489 canonical leaves) | |
| leaves = sorted(lv["leaves"]); print(f"full corpus: {len(leaves)} leaves; probing durations on the remote (cached after first run)", flush=True) | |
| dur = full_corpus_durations(leaves); ok = {l: d for l, d in dur.items() if d} | |
| full = dict(leaves=len(leaves), leaves_probed=len(ok), leaves_failed=len(leaves) - len(ok), devices=len({l.split("/")[1] for l in leaves}), vehicles=len({l.split("/")[0] for l in leaves}), | |
| video_h_frame20=sum((d["nb_frames"] or 0) for d in ok.values()) / 20 / 3600, video_h_container=sum((d["duration"] or 0) for d in ok.values()) / 3600, dcamera_TB=sum((v["dcamera_bytes"] or 0) for v in lv["leaves"].values()) / 1e12, | |
| s12_leaves_in_corpus=sum(leaf_of_uid(u) in lv["leaves"] for u in routes)) | |
| rec = dict(utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), benchmark_by_split=table, benchmark_total=tot, full_corpus=full, | |
| definitions=dict(video_h="cabin video frame count / 20 (frame/20 axis; results/pack_v2/container_fps.json)", regime_partition="grid10 `engaged` flag per 0.1 s (canspec_v2 canonical, 1 s debounce); unknown = video time without CAN grid", | |
| anchor_h="n_rows x 0.2 s (s12 anchor grid; manual anchors = human-control rows, engaged = clean-cruise rows)", devices="distinct dongle ids", full_corpus="1,489 canonical remote leaves; ffprobe nb_frames/20 (frame/20) and container duration")) | |
| (ROOT / "results" / "tables" / "baton_hours.json").write_text(json.dumps(rec, indent=1)) | |
| L = ["# BATON hours — from the CWM3 pipeline's own sources", "", f"Generated {rec['utc']} by cwm3/data/baton_hours.py (numbers only from this script; raw values in results/tables/baton_hours.json).", "", | |
| "Definitions: video hours = cabin video frame count / 20 (frame/20 axis, the pipeline time base; frame counts from `results/pack_v2/container_fps.json`); manual / engaged hours = CAN-grid time (10 Hz grid10 rows / 36,000) partitioned by the canonical `engaged` flag (canspec_v2, 1 s debounce) — the CAN grid can extend slightly past the video on a few routes, so manual + engaged = grid hours ≈ video hours; unknown = video time with no CAN grid; anchor hours = s12 rows × 0.2 s; devices = distinct dongles.", "", | |
| "## CWM3 benchmark subset (the s12 anchor grid: 644 routes)", "", "| split | routes | devices | video h | of which manual h | engaged h | unknown h | anchor rows manual / engaged | anchor h manual / engaged | anchor devices manual / engaged |", "|---|---|---|---|---|---|---|---|---|---|"] | |
| for s in SPLITS: | |
| t = table[s]; a = t["anchors"] | |
| L.append(f"| {s} | {t['routes']} | {t['devices']} | {t['video_h']:.1f} | {t['video_manual_h']:.1f} | {t['video_engaged_h']:.1f} | {t['video_unknown_h']:.1f} | {a['manual']['n_rows']:,} / {a['engaged']['n_rows']:,} | {a['manual']['hours']:.1f} / {a['engaged']['hours']:.1f} | {a['manual']['devices']} / {a['engaged']['devices']} |") | |
| L.append(f"| **total** | {tot['routes']} | {tot['devices']} | **{tot['video_h']:.1f}** | {tot['video_manual_h']:.1f} | {tot['video_engaged_h']:.1f} | {tot['video_unknown_h']:.1f} | {tot['anchor_rows']['manual']:,} / {tot['anchor_rows']['engaged']:,} | {tot['anchor_h']['manual']:.1f} / {tot['anchor_h']['engaged']:.1f} (**{tot['anchor_h_total']:.1f}**) | — |") | |
| L += ["", f"Routes without a frame count: {tot['routes_without_fps']}; without a CAN grid: {tot['routes_without_grid']}. CAN-grid hours (10 Hz rows / 36,000): {tot['grid_h']:.1f}.", "", | |
| "## Full canonical corpus (remote /data/backup/BATON)", "", f"- leaves {full['leaves']} ({full['leaves_probed']} probed, {full['leaves_failed']} failed), vehicles {full['vehicles']}, devices {full['devices']}, dcamera {full['dcamera_TB']:.3f} TB", | |
| f"- video hours: **{full['video_h_frame20']:.1f} h** on the frame/20 axis (ffprobe nb_frames / 20); {full['video_h_container']:.1f} h by container duration", f"- all {full['s12_leaves_in_corpus']} / {tot['routes']} benchmark routes are canonical leaves", "", | |
| "## What the paper should cite", "", f"Recommended: \"the CWM3 benchmark subset of BATON: **{tot['video_h']:.1f} h of cabin+road video from {tot['routes']} routes on {tot['devices']} devices** ({tot['video_manual_h']:.1f} h under manual control, {tot['video_engaged_h']:.1f} h engaged, {tot['video_unknown_h']:.1f} h without CAN), of which **{tot['anchor_h_total']:.1f} h are valid 5-Hz anchors** ({tot['anchor_h']['manual']:.1f} h manual, {tot['anchor_h']['engaged']:.1f} h engaged); the full canonical corpus holds {full['video_h_frame20']:.1f} h across {full['leaves']} route leaves and {full['devices']} devices.\"", | |
| "", "Do not cite the hand-written '165 h': it matches neither the benchmark subset's video hours nor the anchor hours computed here."] | |
| (ROOT / "results" / "tables" / "baton_hours.md").write_text("\n".join(L) + "\n") | |
| write_run_json(ROOT / "results" / "tables", "baton_hours", [Path(__file__)], [FPS, LEAVES] + [S12R / f"targets_{r}_{s}.npz" for r in REGIMES for s in SPLITS], None, t_start, extra=dict(summary=dict(benchmark_total=tot, full_corpus=full)), input_listing=False) | |
| print("BATON_HOURS_DONE", json.dumps(dict(benchmark_total={k: (round(v, 2) if isinstance(v, float) else v) for k, v in tot.items() if k not in ("anchor_h", "anchor_rows")}, anchor_h=tot["anchor_h"], full_corpus={k: (round(v, 2) if isinstance(v, float) else v) for k, v in full.items()}))) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 11.6 kB
- Xet hash:
- 7534ecc56ed7f9c6f2679c608f11640ff03e5cbc47ca9170b5c7cca16318509f
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.