Buckets:
| """cwm3/data/ship_runs_preds.py — archive prediction packs of CLOSED cwm3 runs (results/runs/<run>/) to the remote derived store, | |
| verify by remote sha256, and delete the local copy ONLY after the remote digest matches (same protocol as ship_grid_preds.py). | |
| Shipped per run: every pred_*.npz except pred_layout_* (any size) + any other file > 100 MB that is not on the KEEP list. | |
| KEPT locally (never shipped/deleted): BEST + every ckpt_*.pt (the BEST checkpoint is among them), config.json, history*.jsonl, | |
| metrics_*.json, pred_layout_*.npz / pred_layout/*, run_*.json, *.log, inputs_sha256_*, stats.npz, memory.json, *_DONE, ckpt_pruned.jsonl. | |
| Scope (regex, --include / --exclude): default = week-1 base runs {U0_kin_can(_noK|_vc|_crop)?, V0_can4s, V0_can8s, E0_pooled, | |
| E0_tok4s, E0_tok8s}_s{0,1,2} plus *_smoke* / *synthetic* / *_ep8 / *_rpe400k diagnostics; EXCLUDED (still evaluated): | |
| I1_*, F1_*, W*, OOF*/oof, U0_dmspre*, U0_kin_can_noK_dmspre* (exclusion wins over any include pattern). | |
| Remote: USER@REMOTE_HOST:/data/backup/cwm3_derived/runs_preds/<run>/<file> (per-file rsync -a --partial --inplace; never --delete) | |
| Ledger (append-only): results/runs_preds_ledger.jsonl {run, file, bytes, sha256, remote_sha256, remote_ok, local_deleted, remote_path, secs_transfer, ts} | |
| Resumable: ledger rows with remote_ok and no local file are skipped; a still-present local file is re-verified remotely before deletion.""" | |
| import argparse, json, re, shutil, subprocess, sys, time | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parents[1] | |
| sys.path.insert(0, str(ROOT / "common")) | |
| from runmeta import write_run_json, sha256_file | |
| RUNS = ROOT / "results" / "runs"; LEDGER = ROOT / "results" / "runs_preds_ledger.jsonl" | |
| REMOTE = "USER@REMOTE_HOST"; RDIR = "/data/backup/cwm3_derived/runs_preds" | |
| SSH = ["ssh", "-S", "/tmp/baton_bak_ctl", "-o", "BatchMode=yes", "-o", "ConnectTimeout=20"] | |
| INCLUDE = [r"^(U0_kin_can(_noK|_vc|_crop)?|V0_can4s|V0_can8s|E0_pooled|E0_tok4s|E0_tok8s)_s[0-2]$", r"_smoke", r"synthetic", r"_ep8$", r"_rpe400k$"] | |
| EXCLUDE = [r"^I1_", r"^F1_", r"^W", r"^OOF", r"^oof$", r"^U0_dmspre", r"^U0_kin_can_noK_dmspre"] | |
| SHIP_RE = re.compile(r"^pred_(?!layout).*\.npz$") # every prediction pack (pred_{regime}_{split}.npz, pred_*_train_labelrows.npz, ...) except pred_layout_* | |
| KEEP_RE = re.compile(r"^(BEST|ckpt_.*\.pt|config\.json|history.*\.jsonl|metrics_.*\.json|pred_layout.*|run_.*\.json|.*\.log|inputs_sha256_.*|stats\.npz|memory\.json|.*_DONE|ckpt_pruned\.jsonl)$") | |
| BIG = 100e6 | |
| 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_sha(path): | |
| r = subprocess.run(SSH + [REMOTE, f"test -f '{path}' && sha256sum '{path}' | cut -d' ' -f1"], capture_output=True, text=True, timeout=1800) | |
| return r.stdout.strip() if r.returncode == 0 and r.stdout.strip() else None | |
| def in_scope(name, inc, exc): | |
| return not any(re.search(p, name) for p in exc) and any(re.search(p, name) for p in inc) | |
| def ship_files(run_dir): | |
| out = [] | |
| for p in sorted(run_dir.rglob("*")): | |
| if not p.is_file(): continue | |
| rel = str(p.relative_to(run_dir)); base = p.name | |
| if KEEP_RE.match(base) or rel.startswith("pred_layout"): continue | |
| if SHIP_RE.match(base) or p.stat().st_size > BIG: out.append(p) | |
| return out | |
| def main(): | |
| ap = argparse.ArgumentParser(); ap.add_argument("--dry-run", action="store_true"); ap.add_argument("--include", nargs="*", default=INCLUDE); ap.add_argument("--exclude", nargs="*", default=EXCLUDE); ap.add_argument("--runs", nargs="*", default=[], help="explicit run dir names (still subject to --exclude)"); ap.add_argument("--files", nargs="*", default=[], help="explicit files relative to results/runs (shipped as-is unless on the KEEP list)") | |
| a = ap.parse_args(); t_start = time.time() | |
| import fcntl | |
| lockf = open(ROOT / 'results' / '.ship_runs_preds.lock', 'w') | |
| try: fcntl.flock(lockf, fcntl.LOCK_EX | fcntl.LOCK_NB) | |
| except BlockingIOError: print('SHIP_RUNS_PREDS_LOCKED another shipper is running; nothing done', flush=True); return | |
| runs = sorted(d.name for d in RUNS.iterdir() if d.is_dir() and in_scope(d.name, a.include, a.exclude)) | |
| runs += sorted(r.strip('/') for r in a.runs if (RUNS / r).is_dir() and r.strip('/') not in runs and not any(re.search(p, Path(r).name) for p in a.exclude)) # explicit names may be nested (e.g. oof/<run>_f0) | |
| prev_ok = set() | |
| if LEDGER.exists(): | |
| for l in open(LEDGER): | |
| if l.strip(): | |
| r = json.loads(l) | |
| if r.get("remote_ok"): prev_ok.add((r["run"], r["file"])) | |
| jobs = [(rn, p) for rn in runs for p in ship_files(RUNS / rn)] | |
| for f_ in a.files: # file-level entries: exactly these files, run = their parent dir | |
| p = RUNS / f_ | |
| if p.is_file() and not KEEP_RE.match(p.name) and not p.name.startswith('pred_layout') and (str(p.parent.relative_to(RUNS)), p) not in jobs: jobs.append((str(p.parent.relative_to(RUNS)), p)) | |
| runs = sorted(set(runs) | {rn for rn, _ in jobs}) | |
| tot = sum(p.stat().st_size for _, p in jobs) | |
| print(f"runs in scope: {len(runs)} files to ship: {len(jobs)} bytes: {tot/1e9:.2f} GB", flush=True) | |
| if a.dry_run: | |
| for rn, p in jobs: print(f" DRY {rn}/{p.relative_to(RUNS / rn)} {p.stat().st_size/1e6:.0f} MB") | |
| print("kept example:", sorted(q.name for q in (RUNS / runs[0]).iterdir() if q.is_file() and (RUNS / runs[0] / q.name) not in [p for r_, p in jobs]) if runs else None); return | |
| ensure_master(); subprocess.run(SSH + [REMOTE, f"mkdir -p {RDIR}"], check=True) | |
| free0 = shutil.disk_usage(str(ROOT)).free; moved = 0; n_ok = n_fail = 0; xfer_s = 0.0 | |
| for rn, p in jobs: | |
| rel = str(p.relative_to(RUNS / rn)); rpath = f"{RDIR}/{rn}/{rel}" | |
| if not p.exists(): continue # vanished since listing (verified+deleted by another pass) | |
| size = p.stat().st_size; dig = sha256_file(p) | |
| t_x = 0.0 | |
| if (rn, rel) in prev_ok: rd = remote_sha(rpath) | |
| else: | |
| subprocess.run(SSH + [REMOTE, f"mkdir -p '{RDIR}/{rn}/{Path(rel).parent}'"], check=True) | |
| t1 = time.time(); subprocess.run(["rsync", "-a", "--partial", "--inplace", "-e", " ".join(SSH), str(p), f"{REMOTE}:{RDIR}/{rn}/{rel}"], check=True); t_x = time.time() - t1; xfer_s += t_x | |
| rd = remote_sha(rpath) | |
| ok = (rd == dig); deleted = False | |
| if ok: | |
| try: p.unlink(); deleted = True; moved += size | |
| except FileNotFoundError: deleted = True # already removed by a verified earlier pass | |
| n_ok += 1 | |
| else: n_fail += 1 | |
| with open(LEDGER, "a") as f: | |
| f.write(json.dumps(dict(run=rn, file=rel, bytes=size, sha256=dig, remote_sha256=rd, remote_ok=ok, local_deleted=deleted, remote_path=rpath, secs_transfer=round(t_x, 1), ts=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))) + "\n") | |
| print(f"{rn}/{rel} {size/1e6:.0f} MB ok={ok} deleted={deleted} xfer={t_x:.0f}s", flush=True) | |
| free1 = shutil.disk_usage(str(ROOT)).free | |
| summ = dict(n_runs=len(runs), runs=runs, n_files_ok=n_ok, n_files_failed=n_fail, bytes_archived_and_deleted=moved, GB_archived=round(moved / 1e9, 2), transfer_secs=round(xfer_s, 1), | |
| mean_MBps=round(moved / 1e6 / max(xfer_s, 1e-6), 1), disk_free_before_GB=round(free0 / 1e9, 1), disk_free_after_GB=round(free1 / 1e9, 1), disk_freed_GB=round((free1 - free0) / 1e9, 2), remote_dir=f"{REMOTE}:{RDIR}") | |
| write_run_json(ROOT / "results", "ship_runs_preds", [__file__], [LEDGER] if LEDGER.exists() else [], None, t_start, extra={"summary": summ}, input_listing=False) | |
| print("SHIP_RUNS_PREDS_DONE " + json.dumps(summ), flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 8.1 kB
- Xet hash:
- 3f6e08e8a59359feb8d5dd93b3d6e79d2a676d1dd219d7d080a7f1f7f5f04cd0
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.