Buckets:
| """cwm3/data/ship_grid_preds.py — archive the raw cwm2 grid predictions (pred_test.npz / pred_val.npz per cell) to the | |
| remote derived store, verify by remote sha256, and delete the local copy ONLY after the remote digest matches. | |
| Remote: USER@REMOTE_HOST:/data/backup/cwm2_derived/grid_preds/<cell>/pred_{test,val}.npz (per-file rsync, no tar). | |
| Kept locally untouched: pred_layout_*.npz, BEST + ckpt_*.pt, logs, run_*.json, metrics, history. | |
| Ledger (append-only): cwm3/results/grid_preds_ledger.jsonl one row per file per attempt | |
| {cell, file, bytes, sha256, remote_sha256, remote_ok, local_deleted, secs_transfer, ts} | |
| Resumable: a file whose ledger row says remote_ok and whose local copy is gone is skipped; if the local copy still exists | |
| the remote digest is re-checked before deleting. Never deletes anything on the remote; never uses --delete. | |
| SSH via the ControlMaster socket /tmp/baton_bak_ctl (cwm2/private/ssh_master.sh re-establishes it; no password here).""" | |
| import argparse, json, os, 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 | |
| GRID = Path("/home/USER/cwm2/results/grid") | |
| REMOTE = "USER@REMOTE_HOST"; RDIR = "/data/backup/cwm2_derived/grid_preds" | |
| CTL = "/tmp/baton_bak_ctl"; SSH = ["ssh", "-S", CTL, "-o", "BatchMode=yes", "-o", "ConnectTimeout=20"] | |
| LEDGER = ROOT / "results" / "grid_preds_ledger.jsonl" | |
| FILES = ("pred_test.npz", "pred_val.npz") | |
| 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 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 else None | |
| def ledger_rows(): | |
| rows = [] | |
| if LEDGER.exists(): | |
| for l in open(LEDGER): | |
| l = l.strip() | |
| if l: rows.append(json.loads(l)) | |
| return rows | |
| def main(): | |
| ap = argparse.ArgumentParser(); ap.add_argument("--dry-run", action="store_true"); ap.add_argument("--limit", type=int, default=0); a = ap.parse_args() | |
| t_start = time.time(); ensure_master() | |
| free0 = shutil.disk_usage("/home/USER").free | |
| prev_ok = {(r["cell"], r["file"]) for r in ledger_rows() if r.get("remote_ok")} | |
| cells = sorted(d for d in GRID.iterdir() if d.is_dir() and (d / "PRED_DONE").exists()) | |
| todo = [(c.name, c / f) for c in cells for f in FILES if (c / f).exists()] | |
| skipped_no_pred_done = sorted(d.name for d in GRID.iterdir() if d.is_dir() and not (d / "PRED_DONE").exists() and any((d / f).exists() for f in FILES)) | |
| print(f"cells with PRED_DONE: {len(cells)} files to ship: {len(todo)} cells skipped (no PRED_DONE): {skipped_no_pred_done}", flush=True) | |
| if a.limit: todo = todo[:a.limit] | |
| subprocess.run(SSH + [REMOTE, f"mkdir -p {RDIR}"], check=True) | |
| moved_bytes = 0; n_ok = n_fail = 0; xfer_s = 0.0; sha_s = 0.0 | |
| for cell, p in todo: | |
| size = p.stat().st_size | |
| t0 = time.time(); dig = sha256_file(p); sha_s += time.time() - t0 | |
| rpath = f"{RDIR}/{cell}/{p.name}" | |
| if (cell, p.name) in prev_ok: | |
| rd = remote_sha(rpath) # local still exists: re-verify before deleting | |
| t_x = 0.0 | |
| else: | |
| if a.dry_run: | |
| print(f"DRY {cell}/{p.name} {size/1e6:.0f} MB {dig[:12]}"); continue | |
| subprocess.run(SSH + [REMOTE, f"mkdir -p {RDIR}/{cell}"], check=True) | |
| t1 = time.time() | |
| subprocess.run(["rsync", "-a", "--partial", "--inplace", "-e", " ".join(SSH), str(p), f"{REMOTE}:{RDIR}/{cell}/"], check=True) | |
| t_x = time.time() - t1; xfer_s += t_x | |
| rd = remote_sha(rpath) | |
| ok = (rd == dig) | |
| deleted = False | |
| if ok and not a.dry_run: | |
| p.unlink(); deleted = True; moved_bytes += size; n_ok += 1 | |
| else: | |
| n_fail += 1 | |
| with open(LEDGER, "a") as f: | |
| f.write(json.dumps(dict(cell=cell, file=p.name, 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"{cell}/{p.name} {size/1e6:.0f} MB ok={ok} deleted={deleted} xfer={t_x:.0f}s ({(size/1e6/max(t_x,1e-6)) if t_x else 0:.0f} MB/s)", flush=True) | |
| free1 = shutil.disk_usage("/home/USER").free | |
| summ = dict(n_files_ok=n_ok, n_files_failed=n_fail, bytes_archived_and_deleted=moved_bytes, GB_archived=round(moved_bytes / 1e9, 2), | |
| transfer_secs=round(xfer_s, 1), mean_MBps=round(moved_bytes / 1e6 / max(xfer_s, 1e-6), 1), sha_secs=round(sha_s, 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}", cells_skipped_no_pred_done=skipped_no_pred_done) | |
| if not a.dry_run: | |
| write_run_json(ROOT / "results", "ship_grid_preds", [__file__], [LEDGER] if LEDGER.exists() else [], None, t_start, extra={"summary": summ}) | |
| print("SHIP_GRID_PREDS_DONE " + json.dumps(summ), flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 5.56 kB
- Xet hash:
- a371b8dafa6162357829291b76156cfe4f733de38c5dbef187f45c19ff86a0d3
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.