Buckets:
| """cwm3/data/ship_watch_once.py — ONE poll of the prediction-pack shipping queue (run by the low-frequency watcher every 30 min). | |
| Reads results/runs_preds_ready_to_ship.txt (run-dir names appended by the model-stage agent as evaluations close), ships every | |
| name not yet in results/runs_preds_shipped_runs.txt with ship_runs_preds.py --runs <names> (--include '^$' --exclude : ONLY the | |
| listed dirs are touched; keep-list unchanged), appends one decisions.md row per batch with GB freed, records the names, and prints | |
| exactly one line when a batch was shipped or a failure occurred (silent otherwise).""" | |
| import json, re, subprocess, sys, time | |
| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parents[1]; PY = sys.executable | |
| READY = ROOT / "results" / "runs_preds_ready_to_ship.txt"; STATE = ROOT / "results" / "runs_preds_shipped_runs.txt"; RUNS = ROOT / "results" / "runs" | |
| LOG = ROOT / "results" / "logs" / "ship_watch.log"; DEC = ROOT / "docs" / "decisions.md" | |
| def expand(tok): | |
| """bash-like brace expansion ({a,b,c} and {0..4}) followed by glob, relative to results/runs.""" | |
| import glob, itertools | |
| m = re.search(r"\{([^{}]*)\}", tok) | |
| if not m: return sorted(str(Path(p).relative_to(RUNS)) for p in glob.glob(str(RUNS / tok))) or [tok] | |
| body = m.group(1); rng = re.fullmatch(r"(\d+)\.\.(\d+)", body) | |
| alts = [str(i) for i in range(int(rng.group(1)), int(rng.group(2)) + 1)] if rng else body.split(",") | |
| out = [] | |
| for alt in alts: out += expand(tok[:m.start()] + alt + tok[m.end():]) | |
| return list(dict.fromkeys(out)) | |
| def main(): | |
| if not READY.exists(): return | |
| raw = [l.split()[0].strip().rstrip(",;:").removeprefix("results/runs/").rstrip("/") for l in READY.read_text().splitlines() if l.strip() and not l.lstrip().startswith("#")] # first token = run dir (optionally prefixed results/runs/); trailing punctuation tolerated | |
| done = set(l.strip() for l in STATE.read_text().splitlines()) if STATE.exists() else set() | |
| SKIP = STATE.with_name("runs_preds_skipped_entries.txt"); skipped = set(SKIP.read_text().splitlines()) if SKIP.exists() else set() | |
| dirs, files = [], [] | |
| for tok in raw: | |
| for cand in expand(tok): # brace {a,b} / {0..4} + glob, relative to results/runs | |
| cp = RUNS / cand | |
| if cp.is_file() and cand not in files: files.append(cand) # file-level entry: ship exactly this file | |
| elif cp.is_dir() and cand not in dirs: dirs.append(cand) # dir-level entry: every pred_*.npz (except pred_layout) of the run | |
| new_dirs = [n for n in dirs if n not in done and re.fullmatch(r"[A-Za-z0-9_.\-/]+", n) and ".." not in n] | |
| new_files = [f_ for f_ in files if f_ not in done and ".." not in f_] | |
| valid = new_dirs + new_files | |
| bad = [tok for tok in dict.fromkeys(raw) if tok not in skipped and not any((RUNS / c).exists() or c in done for c in expand(tok))] # already-shipped (deleted) files count as resolved | |
| if bad: | |
| with open(SKIP, "a") as f: f.write("\n".join(bad) + "\n") | |
| if not valid and not bad: return | |
| stamp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) | |
| if bad: print(f"{stamp} WARN listed but not a run dir (skipped): {bad}", flush=True) | |
| if not valid: return | |
| cmd = [PY, str(ROOT / "data" / "ship_runs_preds.py"), "--include", "^$", "--exclude"] + (["--runs", *new_dirs] if new_dirs else []) + (["--files", *new_files] if new_files else []) + (["--dry-run"] if "--dry-run" in sys.argv else []) | |
| r = subprocess.run(cmd, capture_output=True, text=True) | |
| if "--dry-run" in sys.argv: print(r.stdout[-3000:]); return | |
| with open(LOG, "a") as f: f.write(f"=== {stamp} batch {valid}\n{r.stdout}\n{r.stderr}\n") | |
| m = re.search(r"SHIP_RUNS_PREDS_DONE (\{.*\})", r.stdout); s = json.loads(m.group(1)) if m else None | |
| if "SHIP_RUNS_PREDS_LOCKED" in r.stdout: return # another shipper holds the lock; retry next poll | |
| if r.returncode != 0 or s is None or s.get("n_files_failed", 1) > 0: | |
| print(f"{stamp} SHIP_FAIL rc={r.returncode} runs={valid} summary={s} stderr={r.stderr[-300:]}", flush=True); return | |
| with open(STATE, "a") as f: f.write("\n".join(valid) + "\n") | |
| with open(DEC, "a") as f: | |
| f.write(f"| {stamp[:10]} | Prediction packs of {len(new_dirs)} newly closed run dir(s) + {len(new_files)} file-level entries archived by the shipping watcher (`data/ship_watch_once.py` → `ship_runs_preds.py --runs/--files`): {', '.join(new_dirs + new_files)}. | `results/runs_preds_ledger.jsonl`, `run_ship_runs_preds_*.json`: {s['n_files_ok']} files, {s['GB_archived']} GB archived and deleted after remote sha256 match, {s['n_files_failed']} failed; df free {s['disk_free_before_GB']} → {s['disk_free_after_GB']} GB (+{s['disk_freed_GB']} GB). | Raw predictions of these runs live only at `{s['remote_dir']}/<run>/`; kept locally: BEST + ckpts, config, history, metrics, pred_layout_*, run.json, logs. |\n") | |
| print(f"{stamp} SHIPPED runs={valid} files={s['n_files_ok']} GB={s['GB_archived']} freed={s['disk_freed_GB']} GB", flush=True) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 5.13 kB
- Xet hash:
- bb415ad200af198bb0b5ce3eee744f2a21a4b19d9026fdeb6fef7ab372b5024c
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.