"""Run an evaluation and write per-sample records. Results stream to JSONL as they complete, for two reasons: a crashed or cancelled run keeps whatever it paid for, and a resumed run skips samples already on disk. Given that these runs cost real money, losing a partial run is the failure mode most worth engineering against. Every run also writes a config sidecar, so a number in a table can be traced back to the model, scaffold, prompt, and resolution policy that produced it. """ from __future__ import annotations import json import time from dataclasses import asdict, dataclass from pathlib import Path from typing import Callable, Iterable, Sequence from .dataset import Sample from .geometry import point_in_box, point_in_image from .metrics import SampleResult from .models import GroundingModel from .scaffold import Scaffold @dataclass class RunConfig: """Everything needed to reproduce a run. Written next to the results.""" run_id: str model: dict scaffold: dict n_samples: int dataset_id: str subset_seed: int | None = None notes: str = "" class BudgetExceeded(RuntimeError): """Raised when a run would exceed its spend cap.""" def evaluate( samples: Sequence[Sample], model: GroundingModel, scaffold: Scaffold, out_dir: str | Path, *, run_id: str | None = None, budget_usd: float | None = 5.0, resume: bool = True, progress: bool = True, dataset_id: str = "likaixin/ScreenSpot-Pro", notes: str = "", on_result: Callable[[SampleResult], None] | None = None, ) -> list[SampleResult]: """Evaluate `samples` and return per-sample results. `budget_usd` is a hard stop, defaulting to a low value on purpose: the tooling in this space generally ships without spend caps, and an unattended run is the standard way to get a surprise bill. Pass None to disable. """ out_dir = Path(out_dir) out_dir.mkdir(parents=True, exist_ok=True) run_id = run_id or _default_run_id(model, scaffold) results_path = out_dir / f"{run_id}.jsonl" config_path = out_dir / f"{run_id}.config.json" config = RunConfig( run_id=run_id, model=model.describe(), scaffold=scaffold.describe(), n_samples=len(samples), dataset_id=dataset_id, notes=notes, ) config_path.write_text(json.dumps(asdict(config), indent=2), encoding="utf-8") done: dict[str, SampleResult] = {} if resume and results_path.exists(): done = {r.sample_id: r for r in read_results(results_path)} if done and progress: print(f"[runner] resuming {run_id}: {len(done)} already done") results = list(done.values()) spent = sum(r.cost_usd for r in results) pending = [s for s in samples if s.sample_id not in done] if progress: print( f"[runner] {run_id}: {len(pending)} to run, " f"${spent:.4f} already spent" + (f", cap ${budget_usd:.2f}" if budget_usd else ", no cap") ) with results_path.open("a", encoding="utf-8") as fh: for i, sample in enumerate(pending, start=1): if budget_usd is not None and spent >= budget_usd: print( f"[runner] STOPPING at {i - 1}/{len(pending)}: " f"spent ${spent:.4f} of ${budget_usd:.2f} cap. " f"Re-run with a higher budget_usd to continue." ) break result = run_one(sample, model, scaffold) results.append(result) spent += result.cost_usd # Free the decoded screenshot immediately. Samples cache their image # on first access, and a full sweep holds 1,581 of them — at ~25 MB # per decoded 4K frame that exhausts memory around sample 800. sample.release_image() fh.write(json.dumps(asdict(result), default=_json_default) + "\n") fh.flush() if on_result is not None: on_result(result) if progress and (i % 10 == 0 or i == len(pending)): acc = sum(r.correct for r in results) / len(results) print( f"[runner] {i}/{len(pending)} acc={acc:.3f} " f"spent=${spent:.4f}" ) return results def run_one( sample: Sample, model: GroundingModel, scaffold: Scaffold ) -> SampleResult: """Evaluate a single sample. Never raises; failures become incorrect results.""" outcome = scaffold.locate(model, sample.image, sample.instruction) correct = outcome.point is not None and point_in_box( outcome.point[0], outcome.point[1], sample.gt_box ) off_image = ( outcome.point is not None and not point_in_image( outcome.point[0], outcome.point[1], sample.image_size, tolerance=1.0 ) ) return SampleResult( sample_id=sample.sample_id, correct=correct, pred_point=outcome.point, pred_box=getattr(outcome, "box", None), gt_box=sample.gt_box, area_fraction=sample.area_fraction, size_bucket=sample.size_bucket, data_type=sample.data_type, application=sample.application, parse_failed=outcome.parse_failed, off_image=off_image, raw_response=outcome.last_raw_text[:2000], input_tokens=outcome.total_input_tokens, output_tokens=outcome.total_output_tokens, cost_usd=outcome.total_cost, latency_s=outcome.total_latency, error=outcome.api_error, ) def read_results(path: str | Path) -> list[SampleResult]: """Load a results JSONL back into SampleResult objects. `path` can be a directory (finds the newest `.jsonl`) or a file. """ path = Path(path) if path.is_dir(): candidates = sorted(path.glob("*.jsonl"), key=lambda p: p.stat().st_mtime) if not candidates: raise FileNotFoundError(f"no .jsonl files in {path}") path = candidates[-1] # newest by modification time if not path.exists(): return [] out: list[SampleResult] = [] for line_no, line in enumerate(path.read_text(encoding="utf-8").splitlines(), 1): line = line.strip() if not line: continue try: raw = json.loads(line) except json.JSONDecodeError: # A partial final line is expected if a run was killed mid-write. print(f"[runner] skipping malformed line {line_no} in {path.name}") continue out.append(_result_from_dict(raw)) return out def _result_from_dict(raw: dict) -> SampleResult: def as_tuple(v): return tuple(v) if isinstance(v, list) else v return SampleResult( sample_id=raw["sample_id"], correct=raw["correct"], pred_point=as_tuple(raw.get("pred_point")), pred_box=as_tuple(raw.get("pred_box")), gt_box=as_tuple(raw["gt_box"]), area_fraction=raw["area_fraction"], size_bucket=raw["size_bucket"], data_type=raw["data_type"], application=raw.get("application", ""), parse_failed=raw.get("parse_failed", False), off_image=raw.get("off_image", False), raw_response=raw.get("raw_response", ""), input_tokens=raw.get("input_tokens", 0), output_tokens=raw.get("output_tokens", 0), cost_usd=raw.get("cost_usd", 0.0), latency_s=raw.get("latency_s", 0.0), error=raw.get("error", ""), ) def estimate_cost( samples: Sequence[Sample], model: GroundingModel, scaffold: Scaffold, probe_n: int = 5, ) -> dict: """Measure cost on a few samples, then extrapolate to the full set. Worth running before every full sweep: image tiling makes per-sample cost hard to predict from token prices alone, and a scaffold multiplies it by its pass count. """ probe = list(samples[:probe_n]) if not probe: return {"probe_n": 0} results = [] for s in probe: results.append(run_one(s, model, scaffold)) s.release_image() # same leak as in evaluate(); see the note there total = sum(r.cost_usd for r in results) per_sample = total / len(results) return { "probe_n": len(results), "probe_cost_usd": total, "mean_cost_per_sample": per_sample, "projected_full_cost_usd": per_sample * len(samples), "n_full": len(samples), "mean_input_tokens": sum(r.input_tokens for r in results) / len(results), "mean_latency_s": sum(r.latency_s for r in results) / len(results), "probe_accuracy": sum(r.correct for r in results) / len(results), "probe_parse_failures": sum(r.parse_failed for r in results), } def _default_run_id(model: GroundingModel, scaffold: Scaffold) -> str: safe_model = model.name.replace("/", "_").replace(":", "_") safe_scaffold = scaffold.name.replace("(", "_").replace(")", "").replace( ",", "_" ).replace("=", "").replace(" ", "") stamp = time.strftime("%Y%m%d_%H%M%S") return f"{safe_model}__{safe_scaffold}__{stamp}" def _json_default(obj): if isinstance(obj, tuple): return list(obj) raise TypeError(f"not JSON serializable: {type(obj)}")