Buckets:
| """Sweep driver for Flint x ProgramBench (solo PIV vs team PIV) against ONE serving job. | |
| Workflow, end to end (every step resumable; the manifest is the single source of truth): | |
| 1. `sweep.py plan` read tasks/tasks-balanced-30.txt and specs/<condition>.yaml; write one spec | |
| per (condition, task) to sweep/specs/<condition>__<task>.yaml (only the | |
| `name:` line and `world.params.task` differ from the condition's spec) | |
| and sweep/manifest.json with one entry per (task, condition, seed), | |
| status `planned`. Re-running keeps the status of existing entries. | |
| 2. `sweep.py serve` print `swarmlab job serve RecursiveSI/flint ...`; with `--launch`, submit | |
| it and record the serve job id. `sweep.py endpoint` checks readiness | |
| (`swarmlab job endpoint --json`) and records the URL, stage and cost. | |
| 3. `sweep.py run` refuse unless the endpoint is ready; launch `swarmlab job run <spec> | |
| --endpoint <serve> --bucket <bucket> --launch --json` for planned entries, | |
| oldest first, keeping at most --max-concurrent task jobs running; poll | |
| `swarmlab job status --json` and mark entries done/failed. `--dry-run` | |
| only prints the commands. Re-running continues from the manifest. | |
| `--stop-at N` bounds the TOTAL number of manifest entries ever launched | |
| (status other than `planned`), across invocations: once N entries are | |
| launched, no invocation launches more (`--max-concurrent` stays the cap | |
| on jobs running at once). `--retry-failed` first resets entries that are | |
| `failed` with no graded score back to `planned` (the failed attempt's | |
| fields move to the entry's `attempts` list), so they count against | |
| --stop-at again only when relaunched. | |
| 4. `sweep.py fetch` `swarmlab job fetch <tag>/<run_id>` every done entry into runs/, then | |
| `swarmlab programbench grade runs/<run_id> --launch --wait --json`, and | |
| record the grade job and the score (`--dry-run` prints only). | |
| 5. `sweep.py status` the budget line first, then the launched count, per-condition counts, | |
| the serve job's state and cost, task CPU hours. | |
| 6. `analyze.py` results/summary.{md,json} from the fetched, graded runs. | |
| Budget (hard rule: the whole sweep stays under --budget-usd, default $100). Every job this sweep | |
| launches is a row of the launch ledger runs/jobs.jsonl (serve, task and grade jobs; each row has | |
| `timeout_s` and `rate_usd_per_hour`, the flavor's list price: h200 $5/h, cpu-upgrade $0.03/h). | |
| For each row, `swarmlab job status --json` (`hf jobs inspect`) gives the stage and the billed | |
| running seconds (finished jobs are cached in the manifest): | |
| spent = sum over jobs of running_h x rate (running jobs: up to now) | |
| committed = sum over running/pending jobs of (timeout_h - running_h) x rate | |
| next = the job about to launch: its planned timeout_h x rate (the same command | |
| without --launch, --json: plan.timeout_s, plan.rate_usd_per_hour) | |
| `run`, `fetch` (grade jobs) and `serve --launch` refuse a launch when spent + committed + next | |
| > budget, so even if every job runs to its timeout the bill stays under the budget. The serve | |
| job's default timeout is 12h: $60 of the $100 at most; `sweep.py serve` prints that ceiling. | |
| A job with an unknown flavor rate, or whose status cannot be read, makes the check refuse. | |
| Every external command goes to sweep/commands.log (UTC timestamp, exit code, argv; never a | |
| secret: commands carry job ids and paths only). The swarmlab CLI is the `swarmlab` script next | |
| to this interpreter, or `$SWARMLAB_CMD` (e.g. "uv run --project <checkout> swarmlab"). Commands | |
| run with the experiment dir as cwd, so the launch ledger is runs/jobs.jsonl here. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| import re | |
| import shlex | |
| import subprocess | |
| import sys | |
| import time | |
| from collections import Counter | |
| from pathlib import Path | |
| from typing import Any | |
| EXP = Path(__file__).resolve().parent.parent | |
| BUCKET = "hf://buckets/cmpatino/flint-programbench" | |
| MODEL = "RecursiveSI/flint" | |
| PLACEHOLDER = "abishekvashok__cmatrix.5c082c6" | |
| FINISHED_OK = {"COMPLETED"} | |
| FINISHED_BAD = {"ERROR", "CANCELED", "CANCELLED", "DELETED", "TIMEOUT"} | |
| DONE = ("done", "fetched", "graded") | |
| def now() -> str: | |
| return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) | |
| def swarmlab_cmd() -> list[str]: | |
| if os.environ.get("SWARMLAB_CMD"): | |
| return shlex.split(os.environ["SWARMLAB_CMD"]) | |
| local = Path(sys.executable).parent / "swarmlab" | |
| return [str(local)] if local.exists() else ["swarmlab"] | |
| class Sweep: | |
| def __init__(self, exp: Path) -> None: | |
| self.exp = exp | |
| self.dir = exp / "sweep" | |
| self.path = self.dir / "manifest.json" | |
| self.m: dict[str, Any] = (json.loads(self.path.read_text()) if self.path.exists() | |
| else {"bucket": BUCKET, "serve": {}, "runs": []}) | |
| def save(self) -> None: | |
| self.dir.mkdir(parents=True, exist_ok=True) | |
| self.m["updated_at"] = now() | |
| tmp = self.path.with_suffix(".json.tmp") | |
| tmp.write_text(json.dumps(self.m, indent=1, sort_keys=True) + "\n") | |
| os.replace(tmp, self.path) # atomic: a crash leaves the old or the new manifest | |
| def log(self, argv: list[str], code: int | str) -> None: | |
| self.dir.mkdir(parents=True, exist_ok=True) | |
| with (self.dir / "commands.log").open("a") as fh: | |
| fh.write(f"{now()}\texit={code}\t{shlex.join(argv)}\n") | |
| def call(self, args: list[str], *, dry: bool = False) -> dict[str, Any] | None: | |
| """Run `swarmlab <args>` (always with --json), log it, return its JSON object. Dry: print.""" | |
| argv = [*swarmlab_cmd(), *args] | |
| if dry: | |
| print(shlex.join(["swarmlab", *args])) | |
| self.log(argv, "dry-run") | |
| return None | |
| p = subprocess.run(argv, cwd=self.exp, capture_output=True, text=True, | |
| check=False) | |
| self.log(argv, p.returncode) | |
| try: | |
| out = json.loads(p.stdout.strip().splitlines()[-1]) if p.stdout.strip() else {} | |
| except json.JSONDecodeError: | |
| out = {"ok": False, "error": p.stdout[-400:]} | |
| if p.returncode != 0 or out.get("ok") is False: | |
| out.setdefault("ok", False) | |
| out.setdefault("error", (p.stderr or p.stdout)[-400:]) | |
| return out | |
| def runs(self) -> list[dict[str, Any]]: | |
| return self.m["runs"] | |
| def bucket(self) -> str: | |
| return self.m.get("bucket", BUCKET) | |
| # ---- budget --------------------------------------------------------------------------------- | |
| TERMINAL = FINISHED_OK | FINISHED_BAD | |
| def ledger_rows(sw: Sweep) -> list[dict[str, Any]]: | |
| path = sw.exp / "runs" / "jobs.jsonl" | |
| rows = [json.loads(ln) for ln in path.read_text().splitlines() if ln.strip()] \ | |
| if path.exists() else [] | |
| return [r for r in rows if r.get("job_id")] | |
| def spend(sw: Sweep) -> dict[str, Any]: | |
| """{spent, committed, jobs, unknown} in USD over every ledger job (module docstring).""" | |
| cache = sw.m.setdefault("jobs", {}) | |
| spent = committed = 0.0 | |
| unknown: list[str] = [] | |
| for row in ledger_rows(sw): | |
| jid, rate, timeout_s = row["job_id"], row.get("rate_usd_per_hour"), row.get("timeout_s") | |
| info = cache.get(jid) | |
| if not info or info.get("stage") not in TERMINAL: | |
| st = sw.call(["job", "status", jid, "--json"]) or {} | |
| if not st.get("ok"): | |
| unknown.append(jid) | |
| continue | |
| info = {"stage": (st.get("stage") or "").upper(), "secs": st.get("running_secs") or 0, | |
| "kind": row.get("kind"), "rate": rate, "timeout_s": timeout_s} | |
| cache[jid] = info | |
| if rate is None or timeout_s is None: | |
| unknown.append(jid) | |
| continue | |
| spent += info["secs"] / 3600 * rate | |
| if info["stage"] not in TERMINAL: | |
| committed += max(0, timeout_s - info["secs"]) / 3600 * rate | |
| sw.save() | |
| return {"spent": spent, "committed": committed, "jobs": len(cache), "unknown": unknown} | |
| def check_budget(sw: Sweep, budget: float, plan: dict[str, Any] | None, what: str) -> float: | |
| """Refuse (SystemExit) unless spent + committed + this job's worst case <= budget.""" | |
| if not plan or plan.get("rate_usd_per_hour") is None or not plan.get("timeout_s"): | |
| raise SystemExit(f"budget: cannot price {what} (no plan timeout/rate); not launching") | |
| nxt = plan["timeout_s"] / 3600 * plan["rate_usd_per_hour"] | |
| sp = spend(sw) | |
| if sp["unknown"]: | |
| raise SystemExit(f"budget: cannot read jobs {sp['unknown']}; not launching {what}") | |
| total = sp["spent"] + sp["committed"] + nxt | |
| if total > budget: | |
| raise SystemExit(f"budget: spent ${sp['spent']:.2f} + committed ${sp['committed']:.2f} " | |
| f"+ {what} worst ${nxt:.2f} = ${total:.2f} > ${budget:.2f}; stopping") | |
| return nxt | |
| def priced(sw: Sweep, args: list[str], budget: float, what: str) -> None: | |
| """Plan `args` without --launch/--wait and check the budget for it.""" | |
| dry = [x for x in args if x not in ("--launch", "--wait")] | |
| out = sw.call(dry) or {} | |
| if not out.get("ok"): | |
| raise SystemExit(f"planning {what} failed: {out.get('error')}") | |
| check_budget(sw, budget, out.get("plan"), what) | |
| # ---- plan ------------------------------------------------------------------------------------ | |
| def read_tasks(path: Path) -> list[str]: | |
| return [ln.strip() for ln in path.read_text().splitlines() | |
| if ln.strip() and not ln.lstrip().startswith("#")] | |
| def spec_for(text: str, condition: str, task: str) -> str: | |
| """The condition's spec with only the `name:` line and the world's task replaced.""" | |
| name = f"pb-{condition}-{task.replace('__', '--')}" # run ids split on "__" | |
| out, n_name = re.subn(r"(?m)^name:.*$", f"name: {name}", text, count=1) | |
| out, n_task = re.subn(r"(task:\s*)" + re.escape(PLACEHOLDER), r"\g<1>" + task, out) | |
| if n_name != 1 or n_task != 1: | |
| raise SystemExit(f"{condition}: expected one name line and one task {PLACEHOLDER} " | |
| f"(found {n_name}, {n_task})") | |
| return out | |
| def cmd_plan(sw: Sweep, a: argparse.Namespace) -> None: | |
| import yaml # the swarmlab env has PyYAML | |
| tasks = list(dict.fromkeys(read_tasks(Path(a.tasks)))) # duplicates once | |
| conditions = a.conditions.split(",") | |
| seeds = [int(s) for s in a.seeds.split(",")] | |
| have = {r["key"]: r for r in sw.runs} | |
| (sw.dir / "specs").mkdir(parents=True, exist_ok=True) | |
| runs: list[dict[str, Any]] = [] | |
| for task in tasks: # task-major, so a partial sweep still holds whole pairs | |
| for cond in conditions: | |
| src = (sw.exp / "specs" / f"{cond}.yaml").read_text() | |
| text = spec_for(src, cond, task) | |
| doc = yaml.safe_load(text) | |
| (arm,) = doc["arms"] | |
| assert doc["arms"][arm]["world"]["params"]["task"] == task | |
| dest = sw.dir / "specs" / f"{cond}__{task}.yaml" | |
| dest.write_text(text) | |
| for seed in seeds: | |
| key = f"{cond}__{task}__s{seed}" | |
| runs.append(have.pop(key, None) or { | |
| "key": key, "task": task, "condition": cond, "seed": seed, "arm": arm, | |
| "spec": str(dest.relative_to(sw.exp)), "status": "planned", | |
| "run_id": f"{doc['name']}__{arm}__s{seed}"}) | |
| runs.extend(have.values()) # entries planned earlier from another task list are kept | |
| sw.m["runs"] = runs | |
| sw.m.setdefault("created_at", now()) | |
| sw.save() | |
| print(f"manifest {sw.path}: {len(runs)} runs ({len(tasks)} tasks x {conditions} x seeds " | |
| f"{seeds}); specs in {sw.dir / 'specs'}") | |
| # ---- serve / endpoint ------------------------------------------------------------------------ | |
| def cmd_serve(sw: Sweep, a: argparse.Namespace) -> None: | |
| args = ["job", "serve", MODEL, "--engine", "sglang", "--profile", "flint", "--flavor", | |
| "h200", "--timeout", a.timeout, "--bucket", sw.bucket] | |
| from swarmlab.jobs.launch import FLAVOR_USD_PER_HOUR, parse_duration | |
| ceiling = parse_duration(a.timeout) / 3600 * FLAVOR_USD_PER_HOUR["h200"] | |
| print(f"# GPU ceiling: h200 ${FLAVOR_USD_PER_HOUR['h200']:.2f}/h x {a.timeout} = " | |
| f"${ceiling:.2f} (budget ${a.budget_usd:.2f})") | |
| if not a.launch: | |
| sw.call(args, dry=True) | |
| print("not launched (pass --launch to submit the serve job)") | |
| return | |
| if sw.m["serve"].get("job_id"): | |
| raise SystemExit(f"manifest already has serve job {sw.m['serve']['job_id']}") | |
| priced(sw, [*args, "--json"], a.budget_usd, "the serve job") | |
| out = sw.call([*args, "--launch", "--json"]) or {} | |
| if not out.get("ok"): | |
| raise SystemExit(f"serve failed: {out.get('error')}") | |
| sw.m["serve"] = {"job_id": out["job"]["job_id"], "url": out["job"].get("url"), | |
| "launched_at": now(), "timeout": a.timeout} | |
| sw.save() | |
| print(f"serve job {sw.m['serve']['job_id']} recorded") | |
| def refresh_endpoint(sw: Sweep) -> dict[str, Any]: | |
| job = sw.m["serve"].get("job_id") | |
| if not job: | |
| return {"ready": False, "detail": "no serve job in the manifest (sweep.py serve --launch)"} | |
| ep = sw.call(["job", "endpoint", job, "--json"]) or {} | |
| if ep.get("ok"): | |
| sw.m["serve"].update({k: ep.get(k) for k in ("url", "base_url", "served_model_name", | |
| "ready", "stage", "running_secs", | |
| "cost_usd", "remaining_s")}, | |
| checked_at=now()) | |
| sw.save() | |
| return ep | |
| def cmd_endpoint(sw: Sweep, a: argparse.Namespace) -> None: | |
| ep = refresh_endpoint(sw) | |
| keys = ("job_id", "stage", "ready", "url", "served_model_name", "cost_usd", "remaining_s", | |
| "detail", "error") | |
| print(json.dumps({k: ep.get(k) for k in keys if ep.get(k) is not None}, indent=1)) | |
| # ---- run ------------------------------------------------------------------------------------- | |
| def poll(sw: Sweep) -> None: | |
| for r in sw.runs: | |
| if r["status"] != "running": | |
| continue | |
| st = sw.call(["job", "status", r["job_id"], "--json"]) or {} | |
| if not st.get("ok"): | |
| continue | |
| stage = (st.get("stage") or "").upper() | |
| r.update(job_stage=stage, running_secs=st.get("running_secs"), cost_usd=st.get("cost_usd")) | |
| if stage in FINISHED_OK | FINISHED_BAD: | |
| r["status"] = "done" if stage in FINISHED_OK else "failed" | |
| r["finished_at"] = now() | |
| if st.get("message"): | |
| r["message"] = st["message"] | |
| sw.save() | |
| ATTEMPT_KEYS = ("job_id", "tag", "job_url", "launched_at", "finished_at", "job_stage", | |
| "running_secs", "cost_usd", "message", "error") | |
| def launched_entries(sw: Sweep) -> int: | |
| """Manifest entries ever launched: every status but `planned` (the --stop-at bound).""" | |
| return sum(r["status"] != "planned" for r in sw.runs) | |
| def retryable(sw: Sweep) -> list[dict[str, Any]]: | |
| """Entries `--retry-failed` resets: status `failed` and no graded score.""" | |
| return [r for r in sw.runs if r["status"] == "failed" and r.get("score") is None] | |
| def reset_failed(sw: Sweep) -> int: | |
| """`--retry-failed`: back to `planned`; the failed attempt's fields go to `attempts`.""" | |
| rows = retryable(sw) | |
| for r in rows: | |
| r.setdefault("attempts", []).append( | |
| {k: r.pop(k) for k in ATTEMPT_KEYS if k in r} | {"reset_at": now()}) | |
| r["status"] = "planned" | |
| if rows: | |
| sw.save() | |
| return len(rows) | |
| def run_args(sw: Sweep, r: dict[str, Any], serve: str) -> list[str]: | |
| return ["job", "run", str(sw.exp / r["spec"]), "--endpoint", serve, "--bucket", sw.bucket, | |
| "--arm", r["arm"], "--seeds", str(r["seed"]), "--launch", "--json"] | |
| def cmd_run(sw: Sweep, a: argparse.Namespace) -> None: | |
| serve = sw.m["serve"].get("job_id") or "<serve-job-id>" | |
| if a.dry_run: | |
| retry = {id(r) for r in retryable(sw)} if a.retry_failed else set() | |
| planned = [r for r in sw.runs if r["status"] == "planned" or id(r) in retry] | |
| before = launched_entries(sw) - len(retry) | |
| allowed = max(0, a.stop_at - before) if a.stop_at else len(planned) | |
| print(f"# would check: swarmlab job endpoint {serve} --json (ready required); before each " | |
| f"launch, the same command without --launch prices it against ${a.budget_usd:.0f}") | |
| if a.retry_failed: | |
| print(f"# --retry-failed would reset {len(retry)} failed entries to planned") | |
| print(f"# --stop-at {a.stop_at or 'off'}: {before} entries already launched (not " | |
| f"planned); {allowed if a.stop_at else 'no'} more allowed") | |
| for r in planned[:allowed]: | |
| sw.call(run_args(sw, r, serve), dry=True) | |
| print(f"# {min(len(planned), allowed)} of {len(planned)} planned would launch; " | |
| f"at most {a.max_concurrent} at once") | |
| return | |
| if not sw.m["serve"].get("job_id"): | |
| raise SystemExit("no serve job in the manifest: run `sweep.py serve --launch` first") | |
| if a.retry_failed: | |
| print(f"--retry-failed: reset {reset_failed(sw)} failed entries to planned") | |
| launched = 0 | |
| while True: | |
| poll(sw) | |
| running = sum(r["status"] == "running" for r in sw.runs) | |
| queue = [r for r in sw.runs if r["status"] == "planned"] | |
| allowed = max(0, a.stop_at - launched_entries(sw)) if a.stop_at else len(queue) | |
| slots = min(a.max_concurrent - running, allowed, len(queue)) | |
| if slots > 0: | |
| ep = refresh_endpoint(sw) | |
| if not ep.get("ready"): | |
| raise SystemExit(f"endpoint not ready: {ep.get('detail') or ep.get('error')}") | |
| for r in queue[:max(slots, 0)]: | |
| priced(sw, run_args(sw, r, serve), a.budget_usd, f"task job {r['key']}") | |
| out = sw.call(run_args(sw, r, serve)) or {} | |
| if not out.get("ok"): | |
| r.update(status="failed", error=str(out.get("error"))[-400:], finished_at=now()) | |
| else: | |
| job = out["job"] | |
| r.update(status="running", job_id=job["job_id"], tag=job["tag"], | |
| job_url=job.get("url"), launched_at=now()) | |
| launched += 1 | |
| sw.save() | |
| running = sum(r["status"] == "running" for r in sw.runs) | |
| left = [r for r in sw.runs if r["status"] == "planned"] | |
| capped = bool(a.stop_at) and launched_entries(sw) >= a.stop_at | |
| if a.once or not running and (not left or capped): | |
| break | |
| print(f"{now()} running={running} planned={len(left)} launched={launched} " | |
| f"(manifest total {launched_entries(sw)}/{a.stop_at or 'all'})", flush=True) | |
| time.sleep(a.poll) | |
| print(f"launched {launched}; " + ", ".join(f"{k}={v}" for k, v in | |
| Counter(r["status"] for r in sw.runs).items())) | |
| # ---- fetch + grade --------------------------------------------------------------------------- | |
| def cmd_fetch(sw: Sweep, a: argparse.Namespace) -> None: | |
| out_dir = sw.exp / "runs" | |
| for r in sw.runs: | |
| if r["status"] == "done": | |
| args = ["job", "fetch", f"{r['tag']}/{r['run_id']}", "--out", str(out_dir), | |
| "--bucket", sw.bucket, "--json"] | |
| if a.dry_run: | |
| sw.call(args, dry=True) | |
| else: | |
| got = sw.call(args) or {} | |
| if not got.get("ok", True) or "fetched" not in got: | |
| r["fetch_error"] = str(got.get("error"))[-400:] | |
| else: | |
| r.update(status="fetched", run_dir=str(Path(got["fetched"]).relative_to(sw.exp) | |
| if Path(got["fetched"]).is_absolute() | |
| else got["fetched"])) | |
| sw.save() | |
| if r["status"] == "fetched" or (a.dry_run and r["status"] == "done"): | |
| run_dir = r.get("run_dir") or f"runs/{r['run_id']}" | |
| args = ["programbench", "grade", run_dir, "--bucket", sw.bucket, "--launch", "--wait", | |
| "--json"] | |
| if a.dry_run: | |
| sw.call(args, dry=True) | |
| continue | |
| priced(sw, args, a.budget_usd, f"grade job {r['key']}") | |
| got = sw.call(args) or {} | |
| r["grade_job"] = (got.get("job") or {}).get("job_id") | |
| if got.get("ok") and got.get("score"): | |
| s = got["score"] | |
| r.update(status="graded", score=s.get("score"), passed=s.get("passed"), | |
| total=s.get("total"), graded_at=now()) | |
| else: | |
| r["grade_error"] = str(got.get("error"))[-400:] | |
| sw.save() | |
| # ---- status ---------------------------------------------------------------------------------- | |
| def cmd_status(sw: Sweep, a: argparse.Namespace) -> None: | |
| if a.refresh: | |
| poll(sw) | |
| refresh_endpoint(sw) | |
| if a.refresh: | |
| sp = spend(sw) | |
| left = a.budget_usd - sp["spent"] - sp["committed"] | |
| print(f"budget ${a.budget_usd:.2f}: spent ${sp['spent']:.2f}, committed worst case " | |
| f"${sp['committed']:.2f} (running jobs to their timeouts), remaining " | |
| f"${left:.2f} over {sp['jobs']} ledger jobs" | |
| + (f"; UNREADABLE {sp['unknown']}" if sp["unknown"] else "")) | |
| else: | |
| print(f"budget ${a.budget_usd:.2f}: spend not computed (--refresh polls every ledger job)") | |
| print(f"launched entries (not planned): {launched_entries(sw)} of {len(sw.runs)} " | |
| f"(the run --stop-at bound counts these)") | |
| cols = ["planned", "running", "done", "fetched", "failed", "graded"] | |
| by: dict[str, Counter] = {} | |
| for r in sw.runs: | |
| by.setdefault(r["condition"], Counter())[r["status"]] += 1 | |
| print(f"{'condition':<12}" + "".join(f"{c:>9}" for c in cols) + f"{'total':>8}" | |
| + f"{'mean score':>12}") | |
| for cond, c in by.items(): | |
| scores = [r["score"] for r in sw.runs if r["condition"] == cond | |
| and r.get("score") is not None] | |
| mean = f"{sum(scores) / len(scores):.3f}" if scores else "-" | |
| print(f"{cond:<12}" + "".join(f"{c[k]:>9}" for k in cols) + f"{sum(c.values()):>8}" | |
| + f"{mean:>12}") | |
| s = sw.m["serve"] | |
| hrs = (s.get("running_secs") or 0) / 3600 | |
| cost = f"${s['cost_usd']:.2f}" if s.get("cost_usd") is not None else "-" | |
| print(f"serve job {s.get('job_id') or '-'}: stage {s.get('stage') or '-'}, ready " | |
| f"{s.get('ready', '-')}, {hrs:.2f} GPU h, {cost} (as of {s.get('checked_at') or '-'})") | |
| tries = [*sw.runs, *(t for r in sw.runs for t in r.get("attempts", []))] # retried too | |
| secs = [t.get("running_secs") or 0 for t in tries] | |
| usd = sum(t.get("cost_usd") or 0 for t in tries) | |
| print(f"task jobs: {sum(1 for x in secs if x)} with time, {sum(secs) / 3600:.2f} CPU-job h, " | |
| f"${usd:.2f}" + ("" if a.refresh else " (manifest values; --refresh to poll)")) | |
| def main(argv: list[str] | None = None) -> None: | |
| p = argparse.ArgumentParser(description=__doc__.split("\n")[0]) | |
| p.add_argument("--exp-dir", type=Path, default=EXP, help="experiment dir (default: %(default)s)") | |
| sub = p.add_subparsers(dest="cmd", required=True) | |
| s = sub.add_parser("plan", help="write sweep/manifest.json and sweep/specs/") | |
| s.add_argument("--tasks", default=str(EXP / "tasks" / "tasks-balanced-30.txt")) | |
| s.add_argument("--conditions", default="solo-piv,team-piv") | |
| s.add_argument("--seeds", default="0") | |
| s = sub.add_parser("serve", help="print (with --launch submit) the serve job") | |
| s.add_argument("--timeout", default="12h") | |
| s.add_argument("--launch", action="store_true") | |
| s.add_argument("--budget-usd", type=float, default=100.0) | |
| sub.add_parser("endpoint", help="the serve job's readiness, URL and cost") | |
| s = sub.add_parser("run", help="launch task jobs from the manifest (resumable)") | |
| s.add_argument("--max-concurrent", type=int, default=6) | |
| s.add_argument("--stop-at", type=int, default=0, | |
| help="at most N manifest entries launched in total, across invocations " | |
| "(entries not `planned`; 0: no bound)") | |
| s.add_argument("--retry-failed", action="store_true", | |
| help="first reset `failed` entries with no graded score to `planned`") | |
| s.add_argument("--poll", type=float, default=120.0, help="seconds between status polls") | |
| s.add_argument("--once", action="store_true", help="one poll + launch pass, then exit") | |
| s.add_argument("--dry-run", action="store_true") | |
| s.add_argument("--budget-usd", type=float, default=100.0) | |
| s = sub.add_parser("fetch", help="fetch done runs into runs/ and grade them") | |
| s.add_argument("--dry-run", action="store_true") | |
| s.add_argument("--budget-usd", type=float, default=100.0) | |
| s = sub.add_parser("status", help="per-condition counts, serve cost, CPU hours, budget") | |
| s.add_argument("--refresh", action="store_true", help="poll jobs and the endpoint first") | |
| s.add_argument("--budget-usd", type=float, default=100.0) | |
| a = p.parse_args(argv) | |
| sw = Sweep(a.exp_dir.resolve()) | |
| if a.cmd != "plan" and not sw.path.exists(): | |
| raise SystemExit(f"no manifest at {sw.path}: run `sweep.py plan` first") | |
| {"plan": cmd_plan, "serve": cmd_serve, "endpoint": cmd_endpoint, "run": cmd_run, | |
| "fetch": cmd_fetch, "status": cmd_status}[a.cmd](sw, a) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 26 kB
- Xet hash:
- 5dba9c8d3abe89605576392b8483bb3ead649b8e3e9e263a288f66a6a2960644
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.