scripts/validate_gradio.py

"""Exercise actual Gradio session state, uploads, GPU generation and outputs."""
import argparse
import hashlib
import json
import sys
import time
from pathlib import Path

from gradio_client import Client, handle_file

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from grounding_demo.catalog import load_cases
from grounding_demo.console import query_from_prompt
from grounding_demo.protocol import parse_gam


def main():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("url")
    p.add_argument("--output", type=Path, required=True)
    p.add_argument("--limit", type=int, default=0)
    p.add_argument("--route")
    args = p.parse_args()
    if args.output.exists() or args.output.with_suffix(".jsonl").exists():
        raise SystemExit("Refusing to overwrite previous run")
    client = Client(args.url, verbose=False, download_files=False, httpx_kwargs={"trust_env": False})
    health = client.predict(api_name="/service_status")
    if "Inference offline" in health:
        raise SystemExit(health)
    rows = [r for r in load_cases() if not args.route or r["route"] == args.route]
    if args.limit:
        rows = rows[:args.limit]
    results = []
    args.output.parent.mkdir(parents=True, exist_ok=True)
    with args.output.with_suffix(".jsonl").open("x") as stream:
        for row in rows:
            rec = {"id": row["id"], "route": row["route"], "prompt": row["prompt"],
                   "image_sha256": hashlib.sha256((ROOT / row["image"]).read_bytes()).hexdigest()}
            started = time.perf_counter()
            try:
                # Upload current image rather than rely on hidden server default state.
                client.predict(handle_file(str(ROOT / row["image"])), row["route"], row["prompt"], api_name="/choose_upload")
                result = client.predict(query_from_prompt(row["prompt"], row["route"]), row["route"], api_name="/run_console")
                raw = result[4]
                parsed = parse_gam(raw, row["route"])
                rec.update(status="PASS" if parsed.valid and "Complete ·" in result[5] else "FAIL",
                           response=raw, metrics=result[1], normalized=json.loads(result[3]),
                           output_image=result[0], errors=list(parsed.errors), view="direct_prediction_overlay")
            except Exception as exc:
                rec.update(status="ERROR", error=repr(exc))
            rec["elapsed_s"] = time.perf_counter() - started
            results.append(rec)
            stream.write(json.dumps(rec, ensure_ascii=False) + "\n")
            stream.flush()
            print(f"{rec['status']} {row['id']} {rec['elapsed_s']:.2f}s", flush=True)
    failed = sum(r["status"] != "PASS" for r in results)
    report = {"status": "PASS" if not failed else "FAIL", "total": len(results), "failed": failed,
              "url": args.url, "health": health, "results": results,
              "scope": "real_upload_session_generation_protocol_not_independent_accuracy"}
    with args.output.open("x") as output:
        json.dump(report, output, indent=2, ensure_ascii=False)
    raise SystemExit(bool(failed))


if __name__ == "__main__":
    main()