#!/usr/bin/env python3 """Check a running ComfyUI that has the pack: every node is offered, and every example workflow still fits the nodes this ComfyUI has. python scripts/compat_boot.py --url http://127.0.0.1:8188 [--log comfyui.log] It asks the server what nodes it offers (GET /object_info) and compares: 1. Every public node of the pack is there (the list in validate_nodes.py). 2. For every node in every example workflow, subgraphs included: this ComfyUI offers a node of that type, and the example gives a value or a socket for each input the node requires today. A core node that was renamed, dropped, or given a new required input shows up here by name. 3. With --log: the pack's banner in ComfyUI's log names no module that failed to load. Nothing is queued and no model is loaded. Standard library only; it talks to the address it is given and nothing else. Exit code 1 lists the problems. """ from __future__ import annotations import argparse import ast import json from pathlib import Path import re import sys import urllib.error import urllib.request ROOT = Path(__file__).resolve().parent.parent # Node types the page draws by itself; the server does not list them. FRONTEND_ONLY = {"Note", "MarkdownNote", "PrimitiveNode", "Reroute"} SUBGRAPH = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$") def public_node_ids() -> set[str]: """PUBLIC_NODE_IDS, read out of validate_nodes.py without running it.""" tree = ast.parse((ROOT / "scripts" / "validate_nodes.py").read_text(encoding="utf-8")) for node in tree.body: if isinstance(node, ast.Assign) and any(isinstance(target, ast.Name) and target.id == "PUBLIC_NODE_IDS" for target in node.targets): return set(ast.literal_eval(node.value)) raise ValueError("validate_nodes.py has no PUBLIC_NODE_IDS") def example_nodes(data: dict): """(where, node) for every node of a workflow, subgraphs included.""" for node in data.get("nodes") or []: yield "", node for graph in (data.get("definitions") or {}).get("subgraphs") or []: for node in graph.get("nodes") or []: yield f" in subgraph '{graph.get('name') or graph.get('id')}'", node def _widget(spec) -> bool: if not isinstance(spec, (list, tuple)) or not spec: return False options = spec[1] if len(spec) > 1 and isinstance(spec[1], dict) else {} return not options.get("forceInput") and (isinstance(spec[0], list) or spec[0] in ("INT", "FLOAT", "BOOLEAN", "STRING", "COMBO")) def given_names(node: dict, info: dict) -> set[str]: """Connected sockets and usable serialized widget values.""" names = {entry.get("name") for entry in node.get("inputs") or [] if isinstance(entry, dict) and entry.get("link") is not None} for saved in (node.get("widgets_values"), node.get("widgets_values_named")): if isinstance(saved, dict): names.update(name for name, value in saved.items() if value is not None and _widget(required_of(info, name))) return {name for name in names if name} def required_of(info: dict, name: str): return ((info.get("input") or {}).get("required") or {}).get(name) def _grows(spec) -> bool: """True for an input whose sockets are made as links are added.""" kind = spec[0] if isinstance(spec, (list, tuple)) and spec else spec return isinstance(kind, str) and ("AUTOGROW" in kind or "DYNAMIC" in kind) def example_problems(directory: Path, offered: dict) -> list[str]: problems = [] for path in sorted(directory.glob("*.json")): try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError) as exc: problems.append(f"{path.name}: cannot be read ({exc})") continue if not isinstance(data, dict) or not isinstance(data.get("nodes"), list): continue subgraphs = {graph.get("id") for graph in (data.get("definitions") or {}).get("subgraphs") or []} for where, node in example_nodes(data): kind = str(node.get("type")) if kind in FRONTEND_ONLY or kind in subgraphs: continue if SUBGRAPH.fullmatch(kind): problems.append(f"{path.name}: unresolved subgraph '{kind}' (#{node.get('id')}{where})") continue info = offered.get(kind) if info is None: problems.append(f"{path.name}: this ComfyUI has no node '{kind}' (#{node.get('id')}{where})") continue required = set((info.get("input") or {}).get("required") or {}) # Older positional saves cannot identify widgets by name, but # required sockets still need a connection. positional = not isinstance(node.get("widgets_values_named"), dict) and isinstance(node.get("widgets_values"), list) and bool(node["widgets_values"]) given = given_names(node, info) # An input that grows (values.a, values.b, ...) is saved under # its parts' names, and may have none of them filled. roots = given | {name.split(".", 1)[0] for name in given} for name in sorted(required - roots): spec = required_of(info, name) if _grows(spec) or (positional and _widget(spec)): continue problems.append(f"{path.name}: '{kind}' (#{node.get('id')}{where}) now requires '{name}', and the example has no value for it") return problems def log_problems(text: str) -> list[str]: problems = [] for line in text.splitlines(): plain = re.sub(r"\x1b\[[0-9;]*m", "", line) if "Failed to load:" in plain: problems.append("the pack's banner says: " + plain.strip()) elif "IMPORT FAILED" in plain and "ausboss" in plain.lower(): problems.append("ComfyUI's log says: " + plain.strip()) return problems def fetch(url: str) -> dict: with urllib.request.urlopen(url, timeout=60) as response: return json.loads(response.read().decode("utf-8")) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.split("\n", 1)[0]) parser.add_argument("--url", default="http://127.0.0.1:8188", help="the ComfyUI to ask") parser.add_argument("--log", help="ComfyUI's log file, to read the pack's banner from") args = parser.parse_args(argv) base = args.url.rstrip("/") try: stats = fetch(base + "/system_stats") offered = fetch(base + "/object_info") except (OSError, ValueError, urllib.error.URLError) as exc: print(f"ERROR could not ask {base}: {exc}") return 1 system = stats.get("system") or {} print(f"ComfyUI {system.get('comfyui_version', '?')}, frontend {system.get('required_frontend_version', '?')}, {len(offered)} nodes offered.") problems = [f"the pack's node '{key}' is not offered" for key in sorted(public_node_ids() - set(offered))] problems += example_problems(ROOT / "example_workflows", offered) if args.log: try: problems += log_problems(Path(args.log).read_text(encoding="utf-8", errors="replace")) except OSError as exc: problems.append(f"could not read the log {args.log}: {exc}") for problem in problems: print("ERROR " + problem) if problems: print(f"\n{len(problems)} problem(s) found.") return 1 print("OK - every public node is offered and every example fits.") return 0 if __name__ == "__main__": sys.exit(main())