gig-line/Tools/comfyui_generate.py

#!/usr/bin/env python3
"""Generate image assets from the local ComfyUI instance.

The project needs textures it cannot photograph: skin detail, hair strands, fabric
and surface grime. ComfyUI runs locally at `127.0.0.1:8188`, so this is a
**human-triggered local tool**, not a network service — nothing in CI calls it, and
no credential is involved.

    # what is installed
    python3 Tools/comfyui_generate.py models

    # one image
    python3 Tools/comfyui_generate.py txt2img \
        --prompt "seamless photoreal human skin texture, fine pores, neutral light" \
        --negative "text, watermark, hair, clothing" \
        --model juggernautXL_ragnarok.safetensors \
        --size 1024x1024 --seed 7 \
        --out Art/textures/generated/T_Character_SkinDetail.png

Outputs land in the given path and a JSON sidecar records the exact prompt, model,
seed and workflow, so an asset can be regenerated or rejected later without
guessing how it was made.

Licence: the output is a generated image from a locally hosted model. `THIRD_PARTY
NOTICES.md` must record the model and its licence before the texture ships; the
sidecar carries what is needed to do that.
"""

from __future__ import annotations

import argparse
import json
import os
import random
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

HOST = os.environ.get("COMFYUI_HOST", "http://127.0.0.1:8188")
DEFAULT_MODEL = "juggernautXL_ragnarok.safetensors"


def api(path: str) -> dict:
    with urllib.request.urlopen(f"{HOST}{path}", timeout=30) as response:
        return json.loads(response.read().decode())


def post(path: str, body: dict) -> dict:
    request = urllib.request.Request(
        f"{HOST}{path}", data=json.dumps(body).encode(), method="POST",
        headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(request, timeout=60) as response:
        raw = response.read().decode()
        return json.loads(raw) if raw.strip() else {}


def checkpoints() -> list[str]:
    info = api("/object_info/CheckpointLoaderSimple")
    return info["CheckpointLoaderSimple"]["input"]["required"]["ckpt_name"][0]


def workflow(prompt: str, negative: str, model: str, width: int, height: int,
             seed: int, steps: int, cfg: float, sampler: str, scheduler: str) -> dict:
    """A minimal SDXL/SD1.5 txt2img graph in ComfyUI's API format.

    Node ids are strings and references are `[node_id, output_index]`, which is the
    format `/prompt` expects — the UI's workflow JSON is a different shape and cannot
    be posted directly.
    """
    return {
        "1": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": model}},
        "2": {"class_type": "CLIPTextEncode",
              "inputs": {"text": prompt, "clip": ["1", 1]}},
        "3": {"class_type": "CLIPTextEncode",
              "inputs": {"text": negative, "clip": ["1", 1]}},
        "4": {"class_type": "EmptyLatentImage",
              "inputs": {"width": width, "height": height, "batch_size": 1}},
        "5": {"class_type": "KSampler",
              "inputs": {"seed": seed, "steps": steps, "cfg": cfg,
                         "sampler_name": sampler, "scheduler": scheduler,
                         "denoise": 1.0, "model": ["1", 0],
                         "positive": ["2", 0], "negative": ["3", 0],
                         "latent_image": ["4", 0]}},
        "6": {"class_type": "VAEDecode", "inputs": {"samples": ["5", 0], "vae": ["1", 2]}},
        "7": {"class_type": "SaveImage",
              "inputs": {"images": ["6", 0], "filename_prefix": "gigline"}},
    }


def generate(args) -> int:
    width, height = (int(v) for v in args.size.lower().split("x"))
    seed = args.seed if args.seed is not None else random.randrange(0, 2**31)
    graph = workflow(args.prompt, args.negative, args.model, width, height, seed,
                     args.steps, args.cfg, args.sampler, args.scheduler)
    queued = post("/prompt", {"prompt": graph})
    prompt_id = queued.get("prompt_id")
    if not prompt_id:
        print(f"error: ComfyUI rejected the prompt: {queued}", file=sys.stderr)
        return 1
    print(f"queued {prompt_id} ({args.model}, {width}x{height}, seed {seed})")

    deadline = time.time() + args.timeout
    while time.time() < deadline:
        history = api(f"/history/{prompt_id}")
        entry = history.get(prompt_id)
        if entry and entry.get("outputs"):
            images = []
            for node in entry["outputs"].values():
                images.extend(node.get("images", []))
            if not images:
                print("error: job finished with no image", file=sys.stderr)
                return 1
            image = images[0]
            query = urllib.parse.urlencode({
                "filename": image["filename"],
                "subfolder": image.get("subfolder", ""),
                "type": image.get("type", "output")})
            with urllib.request.urlopen(f"{HOST}/view?{query}", timeout=60) as response:
                data = response.read()
            os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
            with open(args.out, "wb") as handle:
                handle.write(data)
            sidecar = os.path.splitext(args.out)[0] + ".json"
            with open(sidecar, "w") as handle:
                json.dump({
                    "tool": "Tools/comfyui_generate.py",
                    "generated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                    "prompt": args.prompt,
                    "negative": args.negative,
                    "model": args.model,
                    "size": [width, height],
                    "seed": seed,
                    "steps": args.steps,
                    "cfg": args.cfg,
                    "sampler": args.sampler,
                    "scheduler": args.scheduler,
                    "workflow": graph,
                    "comfyui_prompt_id": prompt_id,
                    "note": "generated locally; record the model licence in "
                            "THIRD_PARTY_NOTICES.md before shipping",
                }, handle, indent=2)
            print(f"wrote {args.out} (+ sidecar)")
            return 0
        time.sleep(args.poll)
    print("error: timed out waiting for the image", file=sys.stderr)
    return 1


def main(argv=None) -> int:
    argv = sys.argv[1:] if argv is None else argv
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd", required=True)
    sub.add_parser("models", help="list installed checkpoints")
    t = sub.add_parser("txt2img", help="generate one image")
    t.add_argument("--prompt", required=True)
    t.add_argument("--negative", default="text, watermark, signature, blurry, low quality")
    t.add_argument("--model", default=DEFAULT_MODEL)
    t.add_argument("--size", default="1024x1024")
    t.add_argument("--seed", type=int, default=None)
    t.add_argument("--steps", type=int, default=28)
    t.add_argument("--cfg", type=float, default=5.5)
    t.add_argument("--sampler", default="dpmpp_2m")
    t.add_argument("--scheduler", default="karras")
    t.add_argument("--out", required=True)
    t.add_argument("--timeout", type=float, default=300.0)
    t.add_argument("--poll", type=float, default=1.5)
    args = ap.parse_args(argv)

    if args.cmd == "models":
        for name in checkpoints():
            print(name)
        return 0
    return generate(args)


if __name__ == "__main__":
    raise SystemExit(main())