231 lines
12 KiB
Python
231 lines
12 KiB
Python
"""Inspect a ComfyUI schema and prepare small workflows without executing them."""
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from urllib.parse import urlsplit
|
|
from urllib.request import urlopen
|
|
|
|
|
|
NODE_TYPES = ("EmptyImage", "ChickHideNode", "ChickSaveImage", "NotifyMeNode", "PrimitiveStringMultiline")
|
|
|
|
|
|
def inputs_of(node):
|
|
spec = node.get("input", {})
|
|
return {**spec.get("required", {}), **spec.get("optional", {})}
|
|
|
|
|
|
def output_slots(node):
|
|
types = node.get("output", [])
|
|
names = node.get("output_name", [])
|
|
lists = node.get("output_is_list", [])
|
|
return [{"slot": i, "type": kind, "name": names[i] if i < len(names) else kind,
|
|
"is_list": bool(lists[i]) if i < len(lists) else False}
|
|
for i, kind in enumerate(types)]
|
|
|
|
|
|
def image_slot(node):
|
|
slots = [out["slot"] for out in output_slots(node) if out["type"] == "IMAGE" and not out["is_list"]]
|
|
return slots[0] if len(slots) == 1 else None
|
|
|
|
|
|
def analyze(info):
|
|
if not isinstance(info, dict):
|
|
raise ValueError("object_info must be a JSON object")
|
|
for name in NODE_TYPES:
|
|
if name in info and not isinstance(info[name], dict):
|
|
raise ValueError("node schemas must be JSON objects")
|
|
nodes = {name: info.get(name, {}) for name in NODE_TYPES}
|
|
blockers = []
|
|
expected = {
|
|
"EmptyImage": {"width": "INT", "height": "INT", "batch_size": "INT", "color": "INT"},
|
|
"ChickHideNode": {"password": "STRING", "title": "STRING", "fps": "INT",
|
|
"compress": [2, 6, 8], "combine_video": "BOOLEAN", "images": "IMAGE"},
|
|
"ChickSaveImage": {"images": "IMAGE", "filename_prefix": "STRING"},
|
|
}
|
|
for name, required in expected.items():
|
|
if name not in info:
|
|
blockers.append(f"missing node: {name}")
|
|
continue
|
|
spec = nodes[name]
|
|
if spec.get("is_input_list", False):
|
|
blockers.append(f"list input contract requires review: {name}")
|
|
declared = inputs_of(spec)
|
|
for key, kind in required.items():
|
|
value = declared.get(key)
|
|
if not isinstance(value, (list, tuple)) or not value or value[0] != kind:
|
|
blockers.append(f"unexpected input contract: {name}.{key}")
|
|
supplied = set(required)
|
|
extra = set(spec.get("input", {}).get("required", {})) - supplied
|
|
if extra:
|
|
blockers.append(f"additional required inputs: {name}: {', '.join(sorted(extra))}")
|
|
for name in ("EmptyImage", "ChickHideNode"):
|
|
if name in info and image_slot(nodes[name]) is None:
|
|
blockers.append(f"expected one scalar IMAGE output: {name}")
|
|
if "ChickSaveImage" in info and not nodes["ChickSaveImage"].get("output_node", False):
|
|
blockers.append("ChickSaveImage must be an output node")
|
|
callback_blockers = []
|
|
notify = nodes["NotifyMeNode"]
|
|
if "NotifyMeNode" not in info:
|
|
callback_blockers.append("missing node: NotifyMeNode")
|
|
else:
|
|
spec = notify.get("input", {})
|
|
for group, key, kind in (("required", "callback_url", "STRING"),
|
|
("optional", "id", "STRING"),
|
|
("optional", "msg", "STRING"),
|
|
("optional", "passthrough", "*")):
|
|
field = spec.get(group, {}).get(key)
|
|
if not isinstance(field, (list, tuple)) or not field or field[0] != kind:
|
|
callback_blockers.append(f"unexpected input contract: NotifyMeNode.{key}")
|
|
if set(spec.get("required", {})) != {"callback_url"}:
|
|
callback_blockers.append("NotifyMeNode must require only callback_url")
|
|
if notify.get("is_input_list", False):
|
|
callback_blockers.append("NotifyMeNode must accept scalar inputs")
|
|
if (notify.get("output") != ["*"] or notify.get("output_name") != ["passthrough"]):
|
|
callback_blockers.append("NotifyMeNode must expose one wildcard passthrough output")
|
|
if not notify.get("output_node", False):
|
|
callback_blockers.append("NotifyMeNode must remain an output node")
|
|
text_blockers = list(callback_blockers)
|
|
source = nodes["PrimitiveStringMultiline"]
|
|
text_slot = None
|
|
if "PrimitiveStringMultiline" not in info:
|
|
text_blockers.append("missing example text source: PrimitiveStringMultiline")
|
|
else:
|
|
value = inputs_of(source).get("value")
|
|
if not isinstance(value, (list, tuple)) or not value or value[0] != "STRING":
|
|
text_blockers.append("unexpected input contract: PrimitiveStringMultiline.value")
|
|
if set(source.get("input", {}).get("required", {})) - {"value"} or source.get("is_input_list", False):
|
|
text_blockers.append("unsupported example text source inputs")
|
|
slots = [out["slot"] for out in output_slots(source) if out["type"] == "STRING" and not out["is_list"]]
|
|
if len(slots) == 1:
|
|
text_slot = slots[0]
|
|
else:
|
|
text_blockers.append("example text source must expose one scalar STRING output")
|
|
return {
|
|
"nodes": {name: {"registered": name in info, "outputs": output_slots(nodes[name])} for name in NODE_TYPES},
|
|
"carrier_workflow_ready": not blockers,
|
|
"carrier_blockers": blockers,
|
|
"callback_wiring_ready": not callback_blockers,
|
|
"callback_blockers": callback_blockers,
|
|
"text_callback_workflow_ready": not text_blockers,
|
|
"text_callback_blockers": text_blockers,
|
|
"text_output_slot": text_slot,
|
|
"execution_submitted": False,
|
|
"next_checks": ["Fill callback_url and set your task id before running a notification workflow.",
|
|
"Check that the receiver gets exactly the connected text in msg, or an empty string without a connection.",
|
|
"Download the saved carrier PNG and verify its payload and metadata."]}
|
|
|
|
|
|
def build_prompts(info, report):
|
|
results = {}
|
|
if report["carrier_workflow_ready"]:
|
|
results["carrier-smoke"] = {
|
|
"1": {"class_type": "EmptyImage", "inputs": {"width": 16, "height": 16, "batch_size": 1, "color": 255}},
|
|
"2": {"class_type": "ChickHideNode", "inputs": {
|
|
"images": ["1", image_slot(info["EmptyImage"])], "password": "", "title": "Smoke",
|
|
"fps": 16, "compress": 2, "combine_video": True}},
|
|
"3": {"class_type": "ChickSaveImage", "inputs": {
|
|
"images": ["2", image_slot(info["ChickHideNode"])], "filename_prefix": "ChickTools/smoke"}},
|
|
}
|
|
if report["callback_wiring_ready"]:
|
|
results["notify-empty-smoke"] = {
|
|
"1": {"class_type": "NotifyMeNode", "inputs": {"callback_url": "", "id": "demo-task-001"}},
|
|
}
|
|
if report["text_callback_workflow_ready"]:
|
|
results["notify-text-smoke"] = {
|
|
"1": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "Sample message"}},
|
|
"2": {"class_type": "NotifyMeNode", "inputs": {
|
|
"callback_url": "", "id": "demo-task-001", "msg": ["1", report["text_output_slot"]],
|
|
"passthrough": ["1", report["text_output_slot"]]}},
|
|
}
|
|
return results
|
|
|
|
|
|
def build_canvas(prompt, info):
|
|
"""Render the minimal prompt as a canvas without copying schema defaults or secrets."""
|
|
widget_order = {"EmptyImage": ["width", "height", "batch_size", "color"],
|
|
"LoadImage": ["image"], "ChickRestoreImage": ["password"],
|
|
"ImageBatchTestPattern": ["batch_size", "start_from", "text_x", "text_y",
|
|
"width", "height", "font", "font_size"],
|
|
"ChickHideNode": ["password", "title", "fps", "compress", "combine_video"],
|
|
"ChickSaveImage": ["filename_prefix"], "NotifyMeNode": ["callback_url", "id"],
|
|
"PrimitiveStringMultiline": ["value"]}
|
|
nodes, links = {}, []
|
|
for order, (id, data) in enumerate(prompt.items()):
|
|
kind, inputs = data["class_type"], data["inputs"]
|
|
names = [name for name in widget_order[kind] if name in inputs_of(info[kind])]
|
|
ports = []
|
|
for key, field in inputs_of(info[kind]).items():
|
|
if key not in names or isinstance(inputs.get(key), list):
|
|
port = {"name": key, "type": field[0], "link": None}
|
|
if key in names:
|
|
port["widget"] = {"name": key}
|
|
ports.append(port)
|
|
nodes[id] = {"id": int(id), "type": kind, "pos": [60 + order * 330, 140], "size": [300, 240],
|
|
"flags": {}, "order": order, "mode": 0, "inputs": ports,
|
|
"outputs": [{"name": out["name"], "type": out["type"], "slot_index": out["slot"], "links": []}
|
|
for out in output_slots(info[kind])],
|
|
"properties": {"Node name for S&R": kind},
|
|
"widgets_values": [inputs.get(key, "")
|
|
if not isinstance(inputs.get(key), list) else "" for key in names]}
|
|
for id, data in prompt.items():
|
|
for input_index, port in enumerate(nodes[id]["inputs"]):
|
|
value = data["inputs"].get(port["name"])
|
|
if not isinstance(value, list):
|
|
continue
|
|
source, slot = value
|
|
link_id = len(links) + 1
|
|
source_type = nodes[source]["outputs"][slot]["type"]
|
|
links.append([link_id, int(source), slot, int(id), input_index, source_type])
|
|
port["link"] = link_id
|
|
nodes[source]["outputs"][slot]["links"].append(link_id)
|
|
return {"last_node_id": max(node["id"] for node in nodes.values()), "last_link_id": len(links),
|
|
"nodes": list(nodes.values()), "links": links, "groups": [], "config": {},
|
|
"extra": {"ds": {"scale": 0.7, "offset": [0, 0]}}, "version": 0.4}
|
|
|
|
|
|
def read_info(base_url):
|
|
parsed = urlsplit(base_url)
|
|
if parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.query or parsed.fragment or parsed.username:
|
|
raise ValueError("base URL must be HTTP(S) without credentials, query or fragment")
|
|
# Read-only: this tool never posts to /prompt or calls a callback endpoint.
|
|
with urlopen(base_url.rstrip("/") + "/object_info", timeout=15) as response:
|
|
return json.load(response)
|
|
|
|
|
|
def write_preparation(info, output):
|
|
report = analyze(info)
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
prompts = build_prompts(info, report)
|
|
for name, prompt in prompts.items():
|
|
for suffix, value in (("api", prompt), ("canvas", build_canvas(prompt, info))):
|
|
with (output / f"{name}.{suffix}.json").open("w", encoding="utf-8", newline="\n") as handle:
|
|
json.dump(value, handle, ensure_ascii=False, indent=2)
|
|
handle.write("\n")
|
|
with (output / "report.json").open("w", encoding="utf-8", newline="\n") as handle:
|
|
json.dump(report, handle, ensure_ascii=False, indent=2)
|
|
handle.write("\n")
|
|
return report
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
source = parser.add_mutually_exclusive_group(required=True)
|
|
source.add_argument("--base-url", help="ComfyUI API base URL")
|
|
source.add_argument("--object-info", type=Path, help="Saved /object_info response")
|
|
parser.add_argument("--out-dir", type=Path, required=True, help="New output directory for this environment")
|
|
args = parser.parse_args()
|
|
if args.out_dir.exists():
|
|
parser.error("choose a new output directory so old workflows cannot be mistaken for fresh results")
|
|
try:
|
|
info = json.loads(args.object_info.read_text(encoding="utf-8")) if args.object_info else read_info(args.base_url)
|
|
report = write_preparation(info, args.out_dir)
|
|
except (OSError, ValueError, TypeError) as exc:
|
|
parser.exit(1, f"Preparation failed ({type(exc).__name__}); check the schema or connection.\n")
|
|
print(json.dumps(report, ensure_ascii=False, indent=2))
|
|
return 0 if report["carrier_workflow_ready"] and report["callback_wiring_ready"] else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|