| """Run the loop demo against a local ComfyUI server. |
| |
| This is a repo-local validation harness for |
| `docs/automations/loop-demo/LOOP_demo_pipeline.json`. It submits an API-format version |
| of the queue-controller demo graph to a running ComfyUI server and verifies |
| that four EXR files are written across four prompt executions. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| from pathlib import Path |
| import time |
| import urllib.error |
| import urllib.parse |
| import urllib.request |
|
|
|
|
| def _validate_http_url(url: str) -> None: |
| parsed = urllib.parse.urlsplit(str(url or "")) |
| if parsed.scheme not in {"http", "https"} or not parsed.netloc: |
| raise RuntimeError(f"Only http(s) ComfyUI server URLs are allowed: {url!r}") |
|
|
|
|
| def _post_json(url: str, payload: dict) -> dict: |
| _validate_http_url(url) |
| request = urllib.request.Request( |
| url, |
| data=json.dumps(payload).encode("utf-8"), |
| headers={"Content-Type": "application/json"}, |
| ) |
| with urllib.request.urlopen(request, timeout=30) as response: |
| return json.load(response) |
|
|
|
|
| def _get_json(url: str, timeout: float = 15) -> dict: |
| _validate_http_url(url) |
| with urllib.request.urlopen(url, timeout=timeout) as response: |
| return json.load(response) |
|
|
|
|
| def _assert_inside(parent: Path, child: Path) -> None: |
| parent = parent.resolve() |
| child = child.resolve() |
| if parent != child and parent not in child.parents: |
| raise RuntimeError(f"Refusing to touch path outside repo: {child}") |
|
|
|
|
| def _dump_server_debug(server: str) -> None: |
| for endpoint in ("queue", "history?max_items=20"): |
| try: |
| payload = _get_json(f"{server}/{endpoint}", timeout=10) |
| except Exception as exc: |
| print(f"Could not read /{endpoint}: {exc}") |
| continue |
| print(f"/{endpoint}:") |
| print(json.dumps(payload, indent=2)[:8000]) |
|
|
|
|
| def build_prompt(output_base: Path) -> dict: |
| """Build the API prompt equivalent of the current loop demo. |
| |
| The canvas workflow contains a ComfyUI subgraph node. The API prompt below |
| flattens that demo subgraph to its current single inner node, |
| `EasyResize_Koolook`, so the prompt can be submitted directly to `/prompt`. |
| """ |
| return { |
| "12": {"class_type": "easy int", "inputs": {"value": 4}}, |
| "10": { |
| "class_type": "Text Multiline", |
| "inputs": {"text": str(output_base) + "\\"}, |
| }, |
| "9": { |
| "class_type": "Easy_Pattern", |
| "inputs": { |
| "batch_size": ["12", 0], |
| "width": 512, |
| "height": 512, |
| "bg_color_mode": "Custom", |
| "show_text": True, |
| "text_color_mode": "White", |
| "start_from": 1, |
| "step": 1, |
| "font_size": 256, |
| "position": "center", |
| "zero_pad": 0, |
| "bg_color": "#C71585", |
| "text_color": "#FFFFFF", |
| "prefix": "", |
| "suffix": "", |
| }, |
| }, |
| "8": { |
| "class_type": "EasyAIPipeline", |
| "inputs": { |
| "shot_duration": 81, |
| "seed_value": 453453453, |
| "instruction": "Place your base folder path in the FIELD below", |
| "base_directory_path": ["10", 0], |
| "extension": ".%04d.exr", |
| "shot_name": "faceLoop", |
| "ai_method": "", |
| "version": "1", |
| "disable_versioning": False, |
| "enable_overwrite": False, |
| "no_subfolders": True, |
| }, |
| }, |
| "22": {"class_type": "easy int", "inputs": {"value": 0}}, |
| "3": { |
| "class_type": "ImageFromBatch", |
| "inputs": {"image": ["9", 0], "batch_index": ["22", 0], "length": 1}, |
| }, |
| "19": { |
| "class_type": "EasyResize_Koolook", |
| "inputs": { |
| "image": ["3", 0], |
| "base_on": "Width", |
| "base_size": 512, |
| "aspect_ratio": "16:9", |
| "divisible_by": 32, |
| "upscale_method": "nearest-exact", |
| "keep_proportion": "stretch", |
| "crop_position": "center", |
| "pad_color_mode": "Black", |
| "panel_color_mode": "Black", |
| "device": "cpu", |
| "pad_color": "0, 0, 0", |
| "panel_color": "0, 0, 0", |
| "invert_composed_MASK": False, |
| }, |
| }, |
| "21": { |
| "class_type": "Koolook_LoopStatus", |
| "inputs": { |
| "value": ["19", 0], |
| "index": ["22", 0], |
| "total": ["12", 0], |
| "filepath": ["8", 0], |
| "label": "EXR_SAFE", |
| "auto_queue_next": True, |
| "index_node_id": "22", |
| "server_url": "http://127.0.0.1:8188", |
| "max_auto_queue_depth": 100, |
| "remaining_auto_queue_depth": -1, |
| }, |
| }, |
| "4": { |
| "class_type": "SaveEXRFrames", |
| "inputs": { |
| "images": ["21", 0], |
| "filepath": ["8", 0], |
| "tonemap": "linear", |
| "start_frame": ["22", 0], |
| "overwrite": True, |
| "save_workflow": "none", |
| }, |
| }, |
| } |
|
|
|
|
| def run(server: str, output_base: Path, repo_root: Path) -> int: |
| _assert_inside(repo_root, output_base) |
| run_output = output_base / f"run-{int(time.time() * 1000)}" |
| _assert_inside(repo_root, run_output) |
| run_output.mkdir(parents=True, exist_ok=True) |
|
|
| system = _get_json(f"{server}/system_stats") |
| print(f"ComfyUI {system['system']['comfyui_version']} at {server}") |
|
|
| payload = { |
| "prompt": build_prompt(run_output.resolve()), |
| "client_id": "koolook-loop-test", |
| } |
| try: |
| queued = _post_json(f"{server}/prompt", payload) |
| except urllib.error.HTTPError as exc: |
| print(exc.read().decode("utf-8", errors="replace")) |
| return 1 |
|
|
| prompt_id = queued["prompt_id"] |
| print(f"queued {prompt_id}") |
| for _ in range(240): |
| time.sleep(0.5) |
| history = _get_json(f"{server}/history/{prompt_id}") |
| if prompt_id in history: |
| status = history[prompt_id].get("status", {}) |
| print(json.dumps(status, indent=2)) |
| if status.get("status_str") != "success": |
| return 1 |
| break |
| else: |
| print("Timed out waiting for ComfyUI history.") |
| return 1 |
|
|
| files = [] |
| for _ in range(240): |
| files = sorted(run_output.rglob("*.exr")) |
| if len(files) >= 4: |
| break |
| time.sleep(0.5) |
| for file in files: |
| print(file) |
| if len(files) != 4: |
| print(f"Expected 4 EXRs, found {len(files)}.") |
| _dump_server_debug(server) |
| return 1 |
| print("OK: wrote 4 EXR frames.") |
| return 0 |
|
|
|
|
| def main() -> int: |
| repo_root = Path(__file__).resolve().parents[1] |
| parser = argparse.ArgumentParser() |
| parser.add_argument("--server", default="http://127.0.0.1:8188") |
| parser.add_argument( |
| "--output-base", |
| type=Path, |
| default=repo_root / ".tmp" / "comfy-loop-test-output", |
| ) |
| args = parser.parse_args() |
| return run(args.server.rstrip("/"), args.output_base, repo_root) |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(main()) |
|
|
