#!/usr/bin/env python3 """ Anonymous end-to-end check for the virtual try-on Space. Connects with NO token - the same path a browser takes - and exercises /health, /validate and /tryon. pip install "gradio_client==1.13.3" python test_client.py / python test_client.py / --person me.jpg --garment shirt.jpg python test_client.py / --worn --category dress Exit code is 0 only if all three endpoints behave. """ from __future__ import annotations import argparse import base64 import json import os import sys import tempfile import time import urllib.request # --- Prove anonymity: scrub every credential the client might pick up. ------ # for _var in ("HF_TOKEN", "HUGGING_FACE_HUB_TOKEN", "HUGGINGFACE_TOKEN", "HF_API_TOKEN"): if os.environ.pop(_var, None): print(f"[setup] cleared {_var} from env to force an anonymous call") from gradio_client import Client, handle_file # noqa: E402 SAMPLE_PERSON = ("https://huggingface.co/franciszzj/Leffa/resolve/main/" "examples/person1/01350_00.jpg") SAMPLE_GARMENT = ("https://huggingface.co/franciszzj/Leffa/resolve/main/" "examples/garment/01449_00.jpg") GREEN, RED, YELLOW, DIM, RESET = ( "\033[32m", "\033[31m", "\033[33m", "\033[2m", "\033[0m") def ok(msg): print(f"{GREEN} PASS{RESET} {msg}") def bad(msg): print(f"{RED} FAIL{RESET} {msg}") def warn(msg): print(f"{YELLOW} WARN{RESET} {msg}") def head(msg): print(f"\n{'=' * 68}\n{msg}\n{'=' * 68}") def fetch(url: str) -> str: suffix = os.path.splitext(url)[1] or ".jpg" fd, path = tempfile.mkstemp(suffix=suffix) os.close(fd) print(f"{DIM} downloading {url}{RESET}") with urllib.request.urlopen(url, timeout=60) as r, open(path, "wb") as f: f.write(r.read()) return path def show(obj, limit=600): text = json.dumps(obj, indent=2, default=str) if len(text) > limit: text = text[:limit] + f"\n ... [{len(text) - limit} more chars]" print("\n".join(" " + ln for ln in text.splitlines())) def save_data_uri(uri: str, dest: str) -> int: payload = uri.split(",", 1)[1] if "," in uri else uri raw = base64.b64decode(payload) with open(dest, "wb") as f: f.write(raw) return len(raw) def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("space", help="e.g. yourname/vtryon-a (or a full URL)") ap.add_argument("--person") ap.add_argument("--garment") ap.add_argument("--category", default="upper", choices=["upper", "lower", "dress"]) ap.add_argument("--worn", action="store_true", help="garment photo is worn by another person") ap.add_argument("--steps", type=int, default=30) ap.add_argument("--out", default="tryon_result.png") args = ap.parse_args() failures = [] head(f"Connecting ANONYMOUSLY to {args.space}") t0 = time.time() try: client = Client(args.space, hf_token=None, verbose=False) except Exception as exc: # noqa: BLE001 bad(f"could not connect: {type(exc).__name__}: {exc}") print("\n Check the Space exists, is PUBLIC, and is not sleeping/building.") return 1 ok(f"connected in {time.time() - t0:.1f}s with no token") try: names = [e for e in client.view_api(return_format="dict") .get("named_endpoints", {})] print(f"{DIM} endpoints: {names}{RESET}") except Exception: # noqa: BLE001 pass # ---------------------------------------------------------------- health head("1/3 /health (CPU only, zero GPU cost)") try: t0 = time.time() res = client.predict(api_name="/health") dt = time.time() - t0 show(res) if isinstance(res, dict) and "ready" in res and "version" in res: ok(f"returned {{ready, version}} in {dt:.2f}s") if not res.get("ready"): warn("ready=false -> models failed to load. " "Your site should fail over to the mirror.") failures.append("health.ready=false") else: bad(f"unexpected shape: {type(res).__name__}") failures.append("health.shape") except Exception as exc: # noqa: BLE001 bad(f"{type(exc).__name__}: {exc}") failures.append("health") # -------------------------------------------------------------- validate person = args.person or fetch(SAMPLE_PERSON) garment = args.garment or fetch(SAMPLE_GARMENT) head("2/3 /validate (CPU only, zero GPU cost)") try: t0 = time.time() res = client.predict(image=handle_file(person), api_name="/validate") dt = time.time() - t0 show(res) if isinstance(res, dict) and "usable" in res and "reason" in res: ok(f"returned usable={res['usable']} in {dt:.2f}s") if not res["usable"]: warn(f"validator rejected the person photo: {res['reason']}") else: bad(f"unexpected shape: {type(res).__name__}") failures.append("validate.shape") except Exception as exc: # noqa: BLE001 bad(f"{type(exc).__name__}: {exc}") failures.append("validate") # ----------------------------------------------------------------- tryon head(f"3/3 /tryon (GPU; category={args.category} worn={args.worn} " f"steps={args.steps})") try: t0 = time.time() res = client.predict( person_image=handle_file(person), garment_image=handle_file(garment), category=args.category, garment_is_worn=args.worn, request_hash="test-client-001", steps=args.steps, scale=2.5, seed=42, api_name="/tryon", ) dt = time.time() - t0 printable = dict(res) if isinstance(res, dict) else res if isinstance(printable, dict) and printable.get("image"): printable["image"] = ( f"") show(printable) required = {"ok", "image", "error_code", "message", "gpu_seconds"} if not isinstance(res, dict) or not required.issubset(res): bad(f"contract violated; missing {required - set(res or {})}") failures.append("tryon.shape") elif res["ok"]: n = save_data_uri(res["image"], args.out) ok(f"generated in {dt:.1f}s wall / {res['gpu_seconds']:.1f}s GPU") ok(f"image written to {args.out} ({n / 1024:.0f} KB)") if res.get("request_hash") == "test-client-001": ok("request_hash echoed back correctly") else: code = res["error_code"] bad(f"{code}: {res['message']}") if code in ("TIMEOUT", "OUT_OF_MEMORY", "INTERNAL"): warn("this class of error is what the mirror Space is for") elif code in ("NO_PERSON_DETECTED", "NO_GARMENT_FOUND", "BAD_IMAGE"): warn("caller-side error - retrying on the mirror will not help") failures.append(f"tryon.{code}") except Exception as exc: # noqa: BLE001 bad(f"{type(exc).__name__}: {exc}") print("\n A raw exception here means the Space crashed rather than " "returning a structured error, or GPU quota is exhausted.") failures.append("tryon") head("Summary") if failures: bad(f"{len(failures)} problem(s): {', '.join(failures)}") return 1 ok("all three endpoints responded correctly over the anonymous path") return 0 if __name__ == "__main__": sys.exit(main())