File size: 5,838 Bytes
41a10e6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
#!/usr/bin/env python3
"""Call the faceswap-lissie RunPod serverless endpoint ASYNCHRONOUSLY (/run).

Standalone: python 3.8+, stdlib only. Same arguments as
faceswap_lissie_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 JOB_ID  cancel a queued/running job and exit

    # fire-and-forget, collect later
    python faceswap_lissie_async.py --body-image scene.jpg --face-image face.jpg --no-wait
    python faceswap_lissie_async.py --job-id JOB_ID --out ./results
"""
import argparse
import base64
import json
import os
import pathlib
import sys
import time
import urllib.error
import urllib.request

DEFAULT_ENDPOINT = "qkeibkeoefcrk3"
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 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("--body-image")
    ap.add_argument("--face-image")
    ap.add_argument("--instruction")
    ap.add_argument("--expression-prompt")
    ap.add_argument("--negative-prompt")
    ap.add_argument("--steps", type=int)
    ap.add_argument("--cfg", type=float)
    ap.add_argument("--seed", type=int)
    ap.add_argument("--rmbg-sensitivity", type=float)
    ap.add_argument("--rmbg-process-res", type=int)
    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.body_image or not args.face_image:
            sys.exit("ERROR: --body-image and --face-image are required to submit a job")
        params = {"body_image": os.path.basename(args.body_image),
                  "face_image": os.path.basename(args.face_image)}
        for cli, p in [("instruction", "instruction"), ("expression_prompt", "expression_prompt"),
                       ("negative_prompt", "negative_prompt"), ("steps", "steps"),
                       ("cfg", "cfg"), ("seed", "seed"),
                       ("rmbg_sensitivity", "rmbg_sensitivity"),
                       ("rmbg_process_res", "rmbg_process_res")]:
            v = getattr(args, cli)
            if v is not None:
                params[p] = v
        payload = {
            "images": [
                {"name": os.path.basename(args.body_image),
                 "image": base64.b64encode(open(args.body_image, "rb").read()).decode()},
                {"name": os.path.basename(args.face_image),
                 "image": base64.b64encode(open(args.face_image, "rb").read()).decode()},
            ],
            "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
        job = api(args, "POST", "run", {"input": payload})
        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 (job {job_id} still collectable 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']}")
    out_dir = pathlib.Path(args.out)
    out_dir.mkdir(parents=True, exist_ok=True)
    for i, img in enumerate(output["images"]):
        p = out_dir / f"{int(time.time())}-{i}-{img['filename']}"
        p.write_bytes(base64.b64decode(img["data"]))
        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()