x-stride
fa673ce0d4
SDXL checkpoint merging with DARE/TIES deltas, bezier UNet profiles, stack visualizer, example workflow PNG, and API smoke tests.
139 lines
4.5 KiB
Python
139 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Smoke-test ComfyUI XS DARE TIES via the running ComfyUI HTTP API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
BASE = "http://127.0.0.1:8188"
|
|
POLL_SEC = 0.5
|
|
TIMEOUT_SEC = 120
|
|
|
|
|
|
def req(method: str, path: str, body: dict | None = None) -> tuple[int, dict | list | str]:
|
|
data = None if body is None else json.dumps(body).encode()
|
|
r = urllib.request.Request(
|
|
BASE + path,
|
|
data=data,
|
|
headers={"Content-Type": "application/json"} if body is not None else {},
|
|
method=method,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(r, timeout=30) as resp:
|
|
raw = resp.read()
|
|
if not raw:
|
|
return resp.status, ""
|
|
return resp.status, json.loads(raw)
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read().decode()
|
|
try:
|
|
return e.code, json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return e.code, raw
|
|
|
|
|
|
def pick_checkpoints() -> tuple[str, str]:
|
|
code, data = req("GET", "/models/checkpoints")
|
|
if code != 200 or not isinstance(data, list) or len(data) < 2:
|
|
raise RuntimeError(f"need >=2 checkpoints, got {code}: {data!r}")
|
|
return data[0], data[1]
|
|
|
|
|
|
def wait_prompt(prompt_id: str) -> dict:
|
|
deadline = time.time() + TIMEOUT_SEC
|
|
while time.time() < deadline:
|
|
code, hist = req("GET", f"/history/{prompt_id}")
|
|
if code == 200 and isinstance(hist, dict) and prompt_id in hist:
|
|
return hist[prompt_id]
|
|
time.sleep(POLL_SEC)
|
|
raise TimeoutError(f"prompt {prompt_id} did not finish within {TIMEOUT_SEC}s")
|
|
|
|
|
|
def test_profiles() -> None:
|
|
name = "_dare_ties_smoke_test"
|
|
bezier = [{"x": 0.0, "y": 1.0}, {"x": 0.5, "y": 0.5}, {"x": 1.0, "y": 0.0}]
|
|
code, body = req(
|
|
"POST",
|
|
"/dare_ties_profiles/save",
|
|
{"name": name, "bezier": bezier, "display_name": "smoke", "description": "api test"},
|
|
)
|
|
assert code == 200 and body == {"ok": True}, f"profile save failed: {code} {body!r}"
|
|
|
|
code, got = req("GET", f"/dare_ties_profiles/get/{name}")
|
|
assert code == 200 and got.get("name") == name and got.get("bezier") == bezier, got
|
|
|
|
code, body = req("DELETE", f"/dare_ties_profiles/{name}")
|
|
assert code == 200 and body == {"ok": True}, f"profile delete failed: {code} {body!r}"
|
|
print(" profiles REST OK")
|
|
|
|
|
|
def test_visualizer_prompt(ckpt_a: str, ckpt_b: str) -> str:
|
|
prompt = {
|
|
"1": {
|
|
"class_type": "DARETIESModelStack",
|
|
"inputs": {
|
|
"ckpt_name_1": ckpt_a,
|
|
"model_weight_1": 1.0,
|
|
"clip_weight_1": 1.0,
|
|
"enabled_1": True,
|
|
"ckpt_name_2": ckpt_b,
|
|
"model_weight_2": 0.5,
|
|
"clip_weight_2": 0.5,
|
|
"enabled_2": True,
|
|
"ckpt_name_3": "None",
|
|
"model_weight_3": 1.0,
|
|
"clip_weight_3": 1.0,
|
|
"enabled_3": False,
|
|
},
|
|
},
|
|
"2": {
|
|
"class_type": "DARETIESStackVisualizer",
|
|
"inputs": {
|
|
"model_stack": ["1", 0],
|
|
"normalize_like_merge": True,
|
|
"column_scale": "stacked_share",
|
|
"checkpoint_color_mode": "hue_per_child",
|
|
"brightness_from_unet": True,
|
|
"width": 640,
|
|
"height": 240,
|
|
},
|
|
},
|
|
"3": {
|
|
"class_type": "PreviewImage",
|
|
"inputs": {"images": ["2", 0]},
|
|
},
|
|
}
|
|
code, body = req("POST", "/prompt", {"prompt": prompt})
|
|
assert code == 200 and isinstance(body, dict) and "prompt_id" in body, f"queue failed: {code} {body!r}"
|
|
prompt_id = body["prompt_id"]
|
|
entry = wait_prompt(prompt_id)
|
|
status = entry.get("status", {}).get("status_str")
|
|
assert status == "success", entry
|
|
assert "3" in entry.get("outputs", {}), entry
|
|
print(f" visualizer OK prompt_id={prompt_id}")
|
|
return prompt_id
|
|
|
|
|
|
def main() -> int:
|
|
print(f"ComfyUI API smoke test @ {BASE}")
|
|
code, info = req("GET", "/object_info/DARETIESStackVisualizer")
|
|
if code != 200:
|
|
print(f"FAIL: ComfyUI not reachable or node missing ({code})", file=sys.stderr)
|
|
return 1
|
|
|
|
ckpt_a, ckpt_b = pick_checkpoints()
|
|
print(f" checkpoints {ckpt_a!r}, {ckpt_b!r}")
|
|
|
|
test_profiles()
|
|
test_visualizer_prompt(ckpt_a, ckpt_b)
|
|
print("PASS")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|