Buckets:
| """cwm3/data/fetch_roi.py — pull every clip's roi.json (+ index.json) out of the remote frame shards WITHOUT downloading | |
| the shards. Per shard one remote streaming pass (python3 tarfile 'r|' mode over the 2 GB tar, ~400 clips) emits JSONL | |
| {name, data} for members matching frames/<clip_id>/{roi.json,index.json}; the local side writes | |
| results/roi/<clip_id>.json roi.json content + clip_id + shard + index_meta (index.json without the frame list, | |
| plus n_frames and a check that source_frame_idx == round(source_ts_s*20)) | |
| results/roi_index/shard_NNNN.jsonl.gz the full index.json of every clip in the shard (one line per clip) | |
| results/roi_index/shard_NNNN.done marker (resumable) | |
| results/roi_coverage.json coverage vs the 26,578-clip manifest + histograms (roi_source, co_occupant_present, ...) | |
| Shards: /data/backup/cwm2_derived/frames_full_standard/shard_NNNN.tar (cwm2/ship/dataset_full/shards_ledger.jsonl, remote_ok rows). | |
| Read-only on the remote (never writes there). SSH via the ControlMaster socket /tmp/baton_bak_ctl. | |
| roi.json itself carries no clip_id — the id comes from the tar member path (frames/<clip_id>/roi.json).""" | |
| import argparse, gzip, json, subprocess, sys, time | |
| from collections import Counter | |
| from concurrent.futures import ThreadPoolExecutor, as_completed | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "common")) | |
| from runmeta import write_run_json | |
| REMOTE = "USER@REMOTE_HOST"; RDIR = "/data/backup/cwm2_derived/frames_full_standard" | |
| CTL = "/tmp/baton_bak_ctl"; SSH = ["ssh", "-S", CTL, "-o", "BatchMode=yes", "-o", "ConnectTimeout=20"] | |
| LEDGER = Path("/home/USER/cwm2/ship/dataset_full/shards_ledger.jsonl") | |
| MANIFEST = Path("/home/USER/cwm2/ship/dataset_full/manifest.jsonl") | |
| ROI = ROOT / "results" / "roi"; IDX = ROOT / "results" / "roi_index" | |
| REMOTE_PY = r''' | |
| import sys, tarfile, json | |
| tf = tarfile.open(sys.argv[1], mode="r|") | |
| out = sys.stdout | |
| for m in tf: | |
| if not m.isfile(): continue | |
| n = m.name | |
| if n.endswith("/roi.json") or n.endswith("/index.json"): | |
| data = tf.extractfile(m).read().decode("utf-8") | |
| out.write(json.dumps({"name": n, "data": data}) + "\n") | |
| out.flush() | |
| ''' | |
| def ensure_master(): | |
| if subprocess.run(SSH + ["-O", "check", REMOTE], capture_output=True).returncode == 0: return | |
| subprocess.run(["bash", "/home/USER/cwm2/private/ssh_master.sh"], capture_output=True); time.sleep(2) | |
| if subprocess.run(SSH + ["-O", "check", REMOTE], capture_output=True).returncode != 0: raise RuntimeError("ssh master unavailable") | |
| def fetch_shard(n): | |
| tar = f"{RDIR}/shard_{n:04d}.tar" | |
| t0 = time.time() | |
| r = subprocess.run(SSH + [REMOTE, f"python3 - {tar}"], input=REMOTE_PY, capture_output=True, text=True, timeout=3600) | |
| if r.returncode != 0: | |
| raise RuntimeError(f"shard {n}: rc={r.returncode} {r.stderr[:300]}") | |
| rois, idxs = {}, {} | |
| for line in r.stdout.splitlines(): | |
| rec = json.loads(line); parts = rec["name"].split("/") | |
| cid = parts[-2] | |
| (rois if parts[-1] == "roi.json" else idxs)[cid] = json.loads(rec["data"]) | |
| n_written = 0; bad_tb = 0 | |
| with gzip.open(IDX / f"shard_{n:04d}.jsonl.gz", "wt") as g: | |
| for cid, roi in rois.items(): | |
| ix = idxs.get(cid) | |
| meta = None | |
| if ix is not None: | |
| frames = ix.get("frames", []) | |
| tb_ok = all(int(round(f["source_ts_s"] * 20)) == f["source_frame_idx"] for f in frames) | |
| bad_tb += (not tb_ok) | |
| meta = {k: v for k, v in ix.items() if k != "frames"}; meta.update(n_frames=len(frames), frame_idx_eq_round_ts20=tb_ok) | |
| g.write(json.dumps(dict(clip_id=cid, shard=n, index=ix)) + "\n") | |
| out = dict(roi); out.update(clip_id=cid, shard=n, index_meta=meta) | |
| (ROI / f"{cid}.json").write_text(json.dumps(out, indent=1)); n_written += 1 | |
| (IDX / f"shard_{n:04d}.done").write_text(json.dumps(dict(n_roi=len(rois), n_index=len(idxs), n_bad_timebase=bad_tb, secs=round(time.time() - t0, 1), bytes_stream=len(r.stdout)))) | |
| return n, len(rois), len(idxs), bad_tb, time.time() - t0, len(r.stdout) | |
| def coverage(): | |
| man = [json.loads(l)["clip_id"] for l in open(MANIFEST)] | |
| have = {p.stem for p in ROI.glob("*.json")} | |
| hist = {k: Counter() for k in ("roi_source", "co_occupant_present", "selection_rule", "n_persons_max", "hands_missing", "driver_side_uncertain", "head_from")} | |
| for p in ROI.glob("*.json"): | |
| r = json.load(open(p)) | |
| for k in hist: | |
| v = r.get(k, "<absent>") | |
| if isinstance(v, list): v = ",".join(v) or "none" | |
| hist[k][str(v)] += 1 | |
| missing = sorted(set(man) - have) | |
| rec = dict(utc=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), n_manifest=len(man), n_roi_fetched=len(have), n_in_manifest_fetched=len(set(man) & have), | |
| n_missing=len(missing), n_extra_not_in_manifest=len(have - set(man)), missing_clip_ids=missing[:200], | |
| histograms={k: dict(sorted(v.items(), key=lambda kv: -kv[1])) for k, v in hist.items()}, | |
| shards_done=sorted(int(p.stem.split("_")[1]) for p in IDX.glob("shard_*.done"))) | |
| (ROOT / "results" / "roi_coverage.json").write_text(json.dumps(rec, indent=1)) | |
| return rec | |
| def main(): | |
| ap = argparse.ArgumentParser(); ap.add_argument("--shards", default="", help="e.g. 0,1,2 (default: all remote_ok shards in the ledger)"); ap.add_argument("--workers", type=int, default=3); ap.add_argument("--coverage-only", action="store_true"); a = ap.parse_args() | |
| t_start = time.time(); ROI.mkdir(parents=True, exist_ok=True); IDX.mkdir(parents=True, exist_ok=True) | |
| if not a.coverage_only: | |
| ensure_master() | |
| shards = sorted({json.loads(l)["shard"] for l in open(LEDGER) if json.loads(l).get("remote_ok")}) | |
| if a.shards: shards = [int(s) for s in a.shards.split(",")] | |
| todo = [n for n in shards if not (IDX / f"shard_{n:04d}.done").exists()] | |
| print(f"shards in ledger: {len(shards)} todo: {len(todo)}", flush=True) | |
| tot_b = 0 | |
| with ThreadPoolExecutor(max_workers=a.workers) as ex: | |
| futs = {ex.submit(fetch_shard, n): n for n in todo} | |
| for f in as_completed(futs): | |
| try: | |
| n, nr, ni, bad, secs, nb = f.result(); tot_b += nb | |
| print(f"shard {n:04d} roi={nr} index={ni} bad_timebase={bad} {secs:.0f}s stream={nb/1e6:.1f} MB", flush=True) | |
| except Exception as e: | |
| print(f"FAIL {futs[f]}: {e}", flush=True) | |
| rec = coverage() | |
| summ = {k: rec[k] for k in ("n_manifest", "n_roi_fetched", "n_in_manifest_fetched", "n_missing", "n_extra_not_in_manifest")} | |
| summ["histograms"] = {k: rec["histograms"][k] for k in ("roi_source", "co_occupant_present", "selection_rule")} | |
| summ["n_shards_done"] = len(rec["shards_done"]) | |
| write_run_json(ROOT / "results", "fetch_roi", [__file__], [LEDGER, MANIFEST], None, t_start, extra={"summary": summ}) | |
| print("FETCH_ROI_DONE " + json.dumps(summ), flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 7.17 kB
- Xet hash:
- 6e82dc0a6c8f86471eee8d4313f6bd31dc0de546de40034d6fac654e76b57b0c
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.