Spaces:
Running on Zero
Running on Zero
File size: 3,207 Bytes
28dbedc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | """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()
|