| |
| """Call the qwen-edit-turbo RunPod serverless endpoint ASYNCHRONOUSLY (/run). |
| |
| Standalone: python 3.8+, stdlib only. Submits the job, then either polls until |
| done (default) or exits immediately with the job id (--no-wait) so you can |
| collect later with --job-id. |
| |
| # submit and wait |
| python qwen_edit_async.py --api-key $RUNPOD_API_KEY \ |
| --image person.jpg --ref-image outfit.png \ |
| --prompt "put the outfit from the second image on the person" |
| |
| # fire-and-forget, collect later |
| python qwen_edit_async.py ... --no-wait # prints JOB_ID |
| python qwen_edit_async.py --api-key $RUNPOD_API_KEY --job-id JOB_ID |
| |
| Arguments are identical to qwen_edit_sync.py, plus: |
| --no-wait submit only; print the job id and exit |
| --job-id ID skip submission; poll/collect an existing job |
| --poll SECONDS poll interval (default 3) |
| --cancel ID cancel a queued/running job and exit |
| """ |
| import argparse |
| import base64 |
| import json |
| import os |
| import pathlib |
| import sys |
| import time |
| import urllib.error |
| import urllib.request |
|
|
| DEFAULT_ENDPOINT = "dom5lwr0o5wq6u" |
| TERMINAL = ("COMPLETED", "FAILED", "CANCELLED", "TIMED_OUT") |
|
|
|
|
| def api(args, method, path, payload=None, timeout=90): |
| req = urllib.request.Request( |
| f"https://api.runpod.ai/v2/{args.endpoint_id}/{path}", method=method, |
| data=json.dumps(payload).encode() if payload is not None else None, |
| headers={"Content-Type": "application/json", |
| "Authorization": f"Bearer {args.api_key}"}) |
| try: |
| with urllib.request.urlopen(req, timeout=timeout) as r: |
| return json.load(r) |
| except urllib.error.HTTPError as e: |
| sys.exit(f"ERROR: HTTP {e.code} on /{path}: {e.read().decode(errors='replace')[:2000]}") |
|
|
|
|
| def build_input(args): |
| images, params = [], {"prompt": args.prompt, "mode": args.mode} |
| images.append({"name": os.path.basename(args.image), |
| "image": base64.b64encode(open(args.image, "rb").read()).decode()}) |
| if args.ref_image: |
| images.append({"name": os.path.basename(args.ref_image), |
| "image": base64.b64encode(open(args.ref_image, "rb").read()).decode()}) |
| if args.seed is not None: |
| params["seed"] = args.seed |
| for name in ("input_max_dim", "ref_max_dim", "output_max_dim"): |
| v = getattr(args, name) |
| if v is not None: |
| params[name] = v |
| if args.lora_skin_fix: |
| params["lora_skin_fix"] = True |
| params["lora_skin_fix_strength"] = args.lora_skin_fix_strength |
| if args.lora_qwen4play: |
| params["lora_qwen4play"] = True |
| params["lora_qwen4play_strength"] = args.lora_qwen4play_strength |
|
|
| payload = {"images": images, "params": params} |
| if args.workflow: |
| payload["workflow"] = args.workflow |
| if args.workflow_json: |
| payload["workflow_json"] = json.load(open(args.workflow_json)) |
| overrides = {} |
| for s in args.set or []: |
| k, v = s.split("=", 1) |
| try: |
| overrides[k] = json.loads(v) |
| except json.JSONDecodeError: |
| overrides[k] = v |
| if overrides: |
| payload["set"] = overrides |
| return payload |
|
|
|
|
| def save_outputs(output, out_dir): |
| out_dir = pathlib.Path(out_dir) |
| out_dir.mkdir(parents=True, exist_ok=True) |
| saved = [] |
| for i, img in enumerate(output.get("images", [])): |
| p = out_dir / f"{int(time.time())}-{i}-{img['filename']}" |
| p.write_bytes(base64.b64decode(img["data"])) |
| saved.append(str(p)) |
| return saved |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser(description=__doc__, |
| formatter_class=argparse.RawDescriptionHelpFormatter) |
| ap.add_argument("--api-key", default=os.environ.get("RUNPOD_API_KEY")) |
| ap.add_argument("--endpoint-id", default=DEFAULT_ENDPOINT) |
| ap.add_argument("--image") |
| ap.add_argument("--ref-image") |
| ap.add_argument("--prompt") |
| ap.add_argument("--mode", default="turbo-8", choices=["turbo-4", "turbo-8", "quality"]) |
| ap.add_argument("--seed", type=int) |
| ap.add_argument("--input-max-dim", type=int) |
| ap.add_argument("--ref-max-dim", type=int) |
| ap.add_argument("--output-max-dim", type=int) |
| ap.add_argument("--lora-skin-fix", action="store_true") |
| ap.add_argument("--lora-skin-fix-strength", type=float, default=1.0) |
| ap.add_argument("--lora-qwen4play", action="store_true") |
| ap.add_argument("--lora-qwen4play-strength", type=float, default=1.0) |
| ap.add_argument("--workflow") |
| ap.add_argument("--set", action="append", metavar="NODE.INPUT=VALUE") |
| ap.add_argument("--workflow-json") |
| ap.add_argument("--out", default=".") |
| ap.add_argument("--timeout", type=float, default=1800) |
| ap.add_argument("--poll", type=float, default=3.0) |
| ap.add_argument("--no-wait", action="store_true") |
| ap.add_argument("--job-id") |
| ap.add_argument("--cancel", metavar="JOB_ID") |
| args = ap.parse_args() |
| if not args.api_key: |
| sys.exit("ERROR: pass --api-key or set RUNPOD_API_KEY") |
|
|
| if args.cancel: |
| print(json.dumps(api(args, "POST", f"cancel/{args.cancel}"), indent=2)) |
| return |
|
|
| if args.job_id: |
| job_id = args.job_id |
| else: |
| if not args.image or not args.prompt: |
| sys.exit("ERROR: --image and --prompt are required to submit a job") |
| job = api(args, "POST", "run", {"input": build_input(args)}) |
| job_id = job["id"] |
| print(f"# submitted job {job_id}", file=sys.stderr) |
| if args.no_wait: |
| print(job_id) |
| return |
|
|
| t0 = time.monotonic() |
| while True: |
| result = api(args, "GET", f"status/{job_id}") |
| status = result.get("status") |
| if status in TERMINAL: |
| break |
| if time.monotonic() - t0 > args.timeout: |
| sys.exit(f"ERROR: timed out after {args.timeout}s (last status {status}); " |
| f"job {job_id} is still yours to collect with --job-id") |
| print(f"# {status} ... {time.monotonic()-t0:.0f}s", file=sys.stderr) |
| time.sleep(args.poll) |
|
|
| if status != "COMPLETED": |
| sys.exit(f"ERROR: {json.dumps(result, indent=2)[:3000]}") |
| output = result["output"] |
| if "error" in output: |
| sys.exit(f"ERROR from handler: {output['error']}") |
| for p in save_outputs(output, args.out): |
| print(p) |
| print(f"# seed={output.get('seed')} delay={result.get('delayTime')}ms " |
| f"exec={result.get('executionTime')}ms", file=sys.stderr) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
