Buckets:

AsianPlayer
/

TriDrive

Files

xet

AsianPlayer/TriDrive / code /data /fetch_baton_leaves.py

AsianPlayer

16 days ago

download

raw

6.83 kB

"""cwm3/data/fetch_baton_leaves.py — fetch BATON leaves back from the canonical remote (USER@REMOTE_HOST:/data/backup/BATON/<leaf>)
into the same local path (/home/USER/BATON/<leaf>): dcamera.mp4 + qcamera.mp4 (+ the leaf's CSV/metadata files if missing
locally). Per leaf: rsync -a --partial --inplace (never --delete), then verification of every fetched file — remote `stat -c %s` +
`sha256sum` (one ssh per leaf, ControlMaster /tmp/baton_bak_ctl) vs local size + sha256. Ledger (append-only)
results/baton_local_fetch_ledger.jsonl {leaf, file, bytes, local_sha, remote_sha, remote_bytes, verified, ts}. Leaves are resolved from
a round-2 draw CSV (uid -> <CAR>/<dongle>/<routeid>/<route_leaf>) for the requested splits. Owner request 2026-09-12."""
import argparse, json, os, subprocess, sys, time
from pathlib import Path
import pandas as pd
ROOT = Path(__file__).resolve().parents[1]; sys.path.insert(0, str(ROOT / "common"))
from runmeta import write_run_json, sha256_file
LROOT = Path("/home/USER/BATON"); REMOTE = "USER@REMOTE_HOST"; RROOT = "/data/backup/BATON"
SSH = ["ssh", "-S", "/tmp/baton_bak_ctl", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20"]; LEDGER = ROOT / "results" / "baton_local_fetch_ledger.jsonl"
VIDEOS = ("dcamera.mp4", "qcamera.mp4")

def leaf_of_uid(u): p = u.split("__"); return "/".join(p[:3]) + "/" + "__".join(p[3:])

def ensure_master():
    if subprocess.run(SSH + ["-O", "check", REMOTE], capture_output=True).returncode == 0: return
    subprocess.run(["bash", str(ROOT.parent / "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 remote_listing(leaf):
    r = subprocess.run(SSH + [REMOTE, f"cd '{RROOT}/{leaf}' 2>/dev/null && for f in *; do [ -f \"$f\" ] && printf '%s\\t%s\\n' \"$f\" \"$(stat -c %s \"$f\")\"; done"], capture_output=True, text=True, timeout=300)
    out = {}
    for line in r.stdout.splitlines():
        n, s = line.split("\t"); out[n] = int(s)
    return out

def remote_sha(leaf, names):
    cmd = "; ".join(f"printf '%s\\t' '{n}'; sha256sum '{RROOT}/{leaf}/{n}' | cut -d' ' -f1" for n in names)
    r = subprocess.run(SSH + [REMOTE, cmd], capture_output=True, text=True, timeout=7200); out = {}
    for line in r.stdout.splitlines():
        p = line.split("\t")
        if len(p) == 2: out[p[0]] = p[1]
    return out

def main():
    ap = argparse.ArgumentParser(); ap.add_argument("--csv", default=str(ROOT / "results" / "mining" / "round2_draw_v1.csv")); ap.add_argument("--splits", default="test"); ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--leaves-file", default="", help="text file with one leaf path per line (first whitespace field; 'BATON/' prefix tolerated; # comments) — overrides --csv/--splits"); ap.add_argument("--tag", default="")
    ap.add_argument("--leaves-full-line", action="store_true", help="2026-09-16 WP5: the whole stripped line of --leaves-file is the leaf (car dirs with spaces, e.g. 'AUDI A3 3RD GEN/...'); default keeps the first-whitespace-field rule")
    ap.add_argument("--csv-only", action="store_true", help="2026-09-16 WP5: fetch only the leaf's non-video files that are missing locally (CSV / metadata.json); videos untouched")
    a = ap.parse_args(); t_start = time.time()
    if a.leaves_file:
        leaves = sorted({(l.strip() if a.leaves_full_line else l.split()[0]).removeprefix("BATON/").removeprefix(str(LROOT) + "/").rstrip("/") for l in open(a.leaves_file) if l.strip() and not l.lstrip().startswith("#")}); sub = []
        print(f"leaves file {a.leaves_file}: distinct leaves {len(leaves)}", flush=True)
    else:
        df = pd.read_csv(a.csv); sub = df[df.split.isin(a.splits.split(","))]; leaves = sorted({leaf_of_uid(u) for u in sub.uid})
        print(f"splits {a.splits}: clips {len(sub)}, distinct leaves {len(leaves)}", flush=True)
    ensure_master()
    done = set()
    if LEDGER.exists():
        for l in open(LEDGER):
            if l.strip():
                r = json.loads(l)
                if r.get("verified"): done.add((r["leaf"], r["file"]))
    tot_b = 0; n_ok = n_fail = 0; missing_remote = []; xfer_s = 0.0; rows_all = []
    for i, leaf in enumerate(leaves):
        rl = remote_listing(leaf)
        if not rl: missing_remote.append(leaf); print(f"MISSING_ON_REMOTE {leaf}", flush=True); continue
        want = [n for n in VIDEOS if n in rl] + [n for n in rl if not n.endswith((".mp4", ".hevc")) and not (LROOT / leaf / n).exists()]
        want = [n for n in want if (leaf, n) not in done and not (n in VIDEOS and (LROOT / leaf / n).exists() and (LROOT / leaf / n).stat().st_size == rl[n] and (leaf, n) in done)]
        if a.csv_only: want = [n for n in want if n not in VIDEOS]
        if a.dry_run: print(f"DRY {leaf}: {[(n, rl[n]) for n in want]}"); continue
        (LROOT / leaf).mkdir(parents=True, exist_ok=True); t1 = time.time()
        for n in want:
            subprocess.run(["rsync", "-a", "--partial", "--inplace", "-s", "-e", " ".join(SSH), f"{REMOTE}:{RROOT}/{leaf}/{n}", str(LROOT / leaf / n)], check=True)   # -s: remote args not shell-split (car dirs with spaces; 2026-09-16)
        xfer_s += time.time() - t1
        shas = remote_sha(leaf, want)
        for n in want:
            p = LROOT / leaf / n; lb = p.stat().st_size if p.exists() else -1; ls = sha256_file(p) if p.exists() else None; ok = (lb == rl[n]) and (ls == shas.get(n))
            row = dict(leaf=leaf, file=n, bytes=lb, local_sha=ls, remote_sha=shas.get(n), remote_bytes=rl[n], verified=bool(ok), ts=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))
            with open(LEDGER, "a") as f: f.write(json.dumps(row) + "\n")
            rows_all.append(row); n_ok += ok; n_fail += (not ok); tot_b += lb if ok else 0
        print(f"[{i+1}/{len(leaves)}] {leaf[:60]} files {len(want)} ok {sum(1 for n in want if (leaf, n) and rows_all and True)} total {tot_b/1e9:.1f} GB {time.time()-t_start:.0f}s", flush=True)
    summ = dict(splits=(a.tag or a.splits), source=(a.leaves_file or a.csv), clips=int(len(sub)), leaves=len(leaves), files_verified=n_ok, files_failed=n_fail, GB_transferred_verified=round(tot_b / 1e9, 2), transfer_secs=round(xfer_s, 1), MBps=round(tot_b / 1e6 / max(xfer_s, 1e-6), 1), missing_on_remote=missing_remote, wall_s=round(time.time() - t_start))
    if not a.dry_run: write_run_json(ROOT / "results", "fetch_baton_leaves" + (f"_{a.tag}" if a.tag else ""), [Path(__file__)], [Path(a.leaves_file or a.csv)], None, t_start, extra={"summary": summ}, input_listing=False)
    print(("FETCH_FAILED " if (n_fail or missing_remote) else "FETCH_DONE ") + json.dumps(summ), flush=True); sys.exit(6 if (n_fail or missing_remote) else 0)

if __name__ == "__main__":
    main()

Xet Storage Details

Size:: 6.83 kB
Xet hash:: 04bd98e2f28cd3f5384cddcbaa535c1d3c5f69c5c296ba473a1fb2c326fd225d

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""cwm3/data/fetch_baton_leaves.py — fetch BATON leaves back from the canonical remote (USER@REMOTE_HOST:/data/backup/BATON/<leaf>)
	into the same local path (/home/USER/BATON/<leaf>): dcamera.mp4 + qcamera.mp4 (+ the leaf's CSV/metadata files if missing
	locally). Per leaf: rsync -a --partial --inplace (never --delete), then verification of every fetched file — remote `stat -c %s` +
	`sha256sum` (one ssh per leaf, ControlMaster /tmp/baton_bak_ctl) vs local size + sha256. Ledger (append-only)
	results/baton_local_fetch_ledger.jsonl {leaf, file, bytes, local_sha, remote_sha, remote_bytes, verified, ts}. Leaves are resolved from
	a round-2 draw CSV (uid -> <CAR>/<dongle>/<routeid>/<route_leaf>) for the requested splits. Owner request 2026-09-12."""
	import argparse, json, os, subprocess, sys, time
	from pathlib import Path
	import pandas as pd
	ROOT = Path(__file__).resolve().parents[1]; sys.path.insert(0, str(ROOT / "common"))
	from runmeta import write_run_json, sha256_file
	LROOT = Path("/home/USER/BATON"); REMOTE = "USER@REMOTE_HOST"; RROOT = "/data/backup/BATON"
	SSH = ["ssh", "-S", "/tmp/baton_bak_ctl", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20"]; LEDGER = ROOT / "results" / "baton_local_fetch_ledger.jsonl"
	VIDEOS = ("dcamera.mp4", "qcamera.mp4")

	def leaf_of_uid(u): p = u.split("__"); return "/".join(p[:3]) + "/" + "__".join(p[3:])

	def ensure_master():
	if subprocess.run(SSH + ["-O", "check", REMOTE], capture_output=True).returncode == 0: return
	subprocess.run(["bash", str(ROOT.parent / "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 remote_listing(leaf):
	r = subprocess.run(SSH + [REMOTE, f"cd '{RROOT}/{leaf}' 2>/dev/null && for f in *; do [ -f \"$f\" ] && printf '%s\\t%s\\n' \"$f\" \"$(stat -c %s \"$f\")\"; done"], capture_output=True, text=True, timeout=300)
	out = {}
	for line in r.stdout.splitlines():
	n, s = line.split("\t"); out[n] = int(s)
	return out

	def remote_sha(leaf, names):
	cmd = "; ".join(f"printf '%s\\t' '{n}'; sha256sum '{RROOT}/{leaf}/{n}' \| cut -d' ' -f1" for n in names)
	r = subprocess.run(SSH + [REMOTE, cmd], capture_output=True, text=True, timeout=7200); out = {}
	for line in r.stdout.splitlines():
	p = line.split("\t")
	if len(p) == 2: out[p[0]] = p[1]
	return out

	def main():
	ap = argparse.ArgumentParser(); ap.add_argument("--csv", default=str(ROOT / "results" / "mining" / "round2_draw_v1.csv")); ap.add_argument("--splits", default="test"); ap.add_argument("--dry-run", action="store_true")
	ap.add_argument("--leaves-file", default="", help="text file with one leaf path per line (first whitespace field; 'BATON/' prefix tolerated; # comments) — overrides --csv/--splits"); ap.add_argument("--tag", default="")
	ap.add_argument("--leaves-full-line", action="store_true", help="2026-09-16 WP5: the whole stripped line of --leaves-file is the leaf (car dirs with spaces, e.g. 'AUDI A3 3RD GEN/...'); default keeps the first-whitespace-field rule")
	ap.add_argument("--csv-only", action="store_true", help="2026-09-16 WP5: fetch only the leaf's non-video files that are missing locally (CSV / metadata.json); videos untouched")
	a = ap.parse_args(); t_start = time.time()
	if a.leaves_file:
	leaves = sorted({(l.strip() if a.leaves_full_line else l.split()[0]).removeprefix("BATON/").removeprefix(str(LROOT) + "/").rstrip("/") for l in open(a.leaves_file) if l.strip() and not l.lstrip().startswith("#")}); sub = []
	print(f"leaves file {a.leaves_file}: distinct leaves {len(leaves)}", flush=True)
	else:
	df = pd.read_csv(a.csv); sub = df[df.split.isin(a.splits.split(","))]; leaves = sorted({leaf_of_uid(u) for u in sub.uid})
	print(f"splits {a.splits}: clips {len(sub)}, distinct leaves {len(leaves)}", flush=True)
	ensure_master()
	done = set()
	if LEDGER.exists():
	for l in open(LEDGER):
	if l.strip():
	r = json.loads(l)
	if r.get("verified"): done.add((r["leaf"], r["file"]))
	tot_b = 0; n_ok = n_fail = 0; missing_remote = []; xfer_s = 0.0; rows_all = []
	for i, leaf in enumerate(leaves):
	rl = remote_listing(leaf)
	if not rl: missing_remote.append(leaf); print(f"MISSING_ON_REMOTE {leaf}", flush=True); continue
	want = [n for n in VIDEOS if n in rl] + [n for n in rl if not n.endswith((".mp4", ".hevc")) and not (LROOT / leaf / n).exists()]
	want = [n for n in want if (leaf, n) not in done and not (n in VIDEOS and (LROOT / leaf / n).exists() and (LROOT / leaf / n).stat().st_size == rl[n] and (leaf, n) in done)]
	if a.csv_only: want = [n for n in want if n not in VIDEOS]
	if a.dry_run: print(f"DRY {leaf}: {[(n, rl[n]) for n in want]}"); continue
	(LROOT / leaf).mkdir(parents=True, exist_ok=True); t1 = time.time()
	for n in want:
	subprocess.run(["rsync", "-a", "--partial", "--inplace", "-s", "-e", " ".join(SSH), f"{REMOTE}:{RROOT}/{leaf}/{n}", str(LROOT / leaf / n)], check=True) # -s: remote args not shell-split (car dirs with spaces; 2026-09-16)
	xfer_s += time.time() - t1
	shas = remote_sha(leaf, want)
	for n in want:
	p = LROOT / leaf / n; lb = p.stat().st_size if p.exists() else -1; ls = sha256_file(p) if p.exists() else None; ok = (lb == rl[n]) and (ls == shas.get(n))
	row = dict(leaf=leaf, file=n, bytes=lb, local_sha=ls, remote_sha=shas.get(n), remote_bytes=rl[n], verified=bool(ok), ts=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))
	with open(LEDGER, "a") as f: f.write(json.dumps(row) + "\n")
	rows_all.append(row); n_ok += ok; n_fail += (not ok); tot_b += lb if ok else 0
	print(f"[{i+1}/{len(leaves)}] {leaf[:60]} files {len(want)} ok {sum(1 for n in want if (leaf, n) and rows_all and True)} total {tot_b/1e9:.1f} GB {time.time()-t_start:.0f}s", flush=True)
	summ = dict(splits=(a.tag or a.splits), source=(a.leaves_file or a.csv), clips=int(len(sub)), leaves=len(leaves), files_verified=n_ok, files_failed=n_fail, GB_transferred_verified=round(tot_b / 1e9, 2), transfer_secs=round(xfer_s, 1), MBps=round(tot_b / 1e6 / max(xfer_s, 1e-6), 1), missing_on_remote=missing_remote, wall_s=round(time.time() - t_start))
	if not a.dry_run: write_run_json(ROOT / "results", "fetch_baton_leaves" + (f"_{a.tag}" if a.tag else ""), [Path(__file__)], [Path(a.leaves_file or a.csv)], None, t_start, extra={"summary": summ}, input_listing=False)
	print(("FETCH_FAILED " if (n_fail or missing_remote) else "FETCH_DONE ") + json.dumps(summ), flush=True); sys.exit(6 if (n_fail or missing_remote) else 0)

	if __name__ == "__main__":
	main()