| |
| """ |
| 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 <username>/<space-name> |
| python test_client.py <username>/<space-name> --person me.jpg --garment shirt.jpg |
| python test_client.py <username>/<space-name> --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 |
|
|
| |
| 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 |
|
|
| 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: |
| 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: |
| pass |
|
|
| |
| 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: |
| bad(f"{type(exc).__name__}: {exc}") |
| failures.append("health") |
|
|
| |
| 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: |
| bad(f"{type(exc).__name__}: {exc}") |
| failures.append("validate") |
|
|
| |
| 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"<data URI, {len(res['image'])} chars>") |
| 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: |
| 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()) |
|
|
