DemonGatanjieuandClaude Opus 5.5
c84b5960fd
list_workflows gives each card a route read from its saved graph: its passes in run order, the model each draws with, a split's steps, an enlargement, control models such as a Tile LLLite and what comes after. A Flash run picked a one-pass card for a two-model job; it saw names only. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
184 lines
8.6 KiB
Python
184 lines
8.6 KiB
Python
"""A saved workflow's route in one line for the AI (list_workflows): its passes in the order they run,
|
|
each with the model file it draws with, the part of a split it runs, its redraw strength, an
|
|
enlargement before it and the control models patched in, then the picture nodes after the samplers.
|
|
So an AI picking a card sees "one pass" from "base, then a Tile redraw on a second model" without
|
|
opening it. Read from the serialized graph alone, as the workflow details' flow is on the page
|
|
(web/modules/recipe_flow.js); widget names come from ComfyUI's node types (`names` in tests)."""
|
|
|
|
import re
|
|
|
|
CONTROL = re.compile(r"lllite|controlnet|t2iadapter|ipadapter", re.I)
|
|
UPSCALE = re.compile(r"upscale|scaleby|imagescale", re.I)
|
|
SAVE = re.compile(r"^(SaveImage|PreviewImage)|save", re.I)
|
|
PLUMBING = re.compile(r"^(VAEDecode|VAEEncode|Reroute|Primitive|Note|MarkdownNote)")
|
|
LOADER_VALUE = re.compile(r"\.(safetensors|ckpt|pt|pth|bin|gguf|sft)$", re.I)
|
|
WIDGET_KINDS = ("INT", "FLOAT", "STRING", "BOOLEAN", "COMBO")
|
|
SEEDS = ("seed", "noise_seed")
|
|
|
|
|
|
def widget_names(node_type):
|
|
"""A node type's widget names in the order its values are saved, or [] for a type not loaded."""
|
|
import nodes # ComfyUI's registry, filled once ComfyUI has loaded the packs
|
|
|
|
cls = nodes.NODE_CLASS_MAPPINGS.get(node_type)
|
|
try:
|
|
spec = cls.INPUT_TYPES() if cls else {}
|
|
except Exception: # noqa: BLE001 - a pack's own code; its values stay unnamed
|
|
return []
|
|
names = []
|
|
for group in ("required", "optional"):
|
|
for name, item in (spec.get(group) or {}).items():
|
|
kind = item[0] if isinstance(item, (list, tuple)) and item else item
|
|
options = item[1] if isinstance(item, (list, tuple)) and len(item) > 1 and isinstance(item[1], dict) else {}
|
|
if options.get("forceInput") or not (isinstance(kind, list) or kind in WIDGET_KINDS):
|
|
continue
|
|
names.append(name)
|
|
if kind == "INT" and (options.get("control_after_generate") or name in SEEDS):
|
|
names.append("control_after_generate")
|
|
return names
|
|
|
|
|
|
def _values(node):
|
|
values = node.get("widgets_values")
|
|
return values if isinstance(values, list) else list((values or {}).values())
|
|
|
|
|
|
def _off(node):
|
|
return node.get("mode") in (2, 4)
|
|
|
|
|
|
def _file(node):
|
|
return next((value for value in _values(node) if isinstance(value, str) and LOADER_VALUE.search(value)), "")
|
|
|
|
|
|
def _short(file):
|
|
return LOADER_VALUE.sub("", re.split(r"[\\/]", str(file))[-1])
|
|
|
|
|
|
def _sampler(node):
|
|
names = {item.get("name") for item in node.get("inputs") or []}
|
|
return {"model", "positive"} <= names and any("latent" in str(name) for name in names)
|
|
|
|
|
|
def route(workflow, names=widget_names):
|
|
"""[{model, loras, controls, steps, window, denoise, upscale, kind, state}] in the order the
|
|
samplers run, and the picture nodes after them: (stages, after)."""
|
|
nodes = {str(node.get("id")): node for node in (workflow or {}).get("nodes") or [] if isinstance(node, dict)}
|
|
links = {}
|
|
for link in (workflow or {}).get("links") or []:
|
|
if isinstance(link, list) and len(link) >= 2:
|
|
links[str(link[0])] = str(link[1])
|
|
elif isinstance(link, dict):
|
|
links[str(link.get("id"))] = str(link.get("origin_id"))
|
|
source = lambda item: nodes.get(links.get(str(item.get("link")))) if item and item.get("link") is not None else None # noqa: E731
|
|
named = {}
|
|
|
|
def value(node, name):
|
|
if node["type"] not in named:
|
|
named[node["type"]] = names(node["type"])
|
|
at = named[node["type"]].index(name) if name in named[node["type"]] else -1
|
|
values = _values(node)
|
|
return values[at] if 0 <= at < len(values) else None
|
|
|
|
def feeders(sampler):
|
|
seen, before, todo = set(), set(), [sampler]
|
|
while todo:
|
|
current = todo.pop()
|
|
for item in current.get("inputs") or []:
|
|
if current is sampler and item.get("name") in ("positive", "negative"):
|
|
continue
|
|
found = source(item)
|
|
key = found and str(found["id"])
|
|
if not found or key in seen or key in before:
|
|
continue
|
|
if _sampler(found):
|
|
before.add(key)
|
|
else:
|
|
seen.add(key)
|
|
todo.append(found)
|
|
return seen, before
|
|
|
|
def model_of(sampler):
|
|
out = {"model": "", "loras": [], "controls": []}
|
|
node = source(next((item for item in sampler.get("inputs") or [] if item.get("name") == "model"), None))
|
|
for _hop in range(16):
|
|
if not node:
|
|
break
|
|
if not _off(node):
|
|
if re.search("lora", node["type"], re.I):
|
|
out["loras"].append(_short(_file(node)) or node["type"])
|
|
elif CONTROL.search(node["type"]):
|
|
out["controls"].append(_short(_file(node)) or node.get("title") or node["type"])
|
|
elif _file(node) and not any(item.get("name") == "model" for item in node.get("inputs") or []):
|
|
out["model"] = _short(_file(node))
|
|
break
|
|
node = source(next((item for item in node.get("inputs") or [] if item.get("name") == "model"), None))
|
|
return out
|
|
|
|
samplers = [node for node in nodes.values() if _sampler(node)]
|
|
fed = {str(node["id"]): feeders(node) for node in samplers}
|
|
depth = {}
|
|
|
|
def level(key, guard=0):
|
|
if key not in depth:
|
|
depth[key] = 0 if guard > 16 else 1 + max([level(item, guard + 1) for item in fed[key][1]] or [0])
|
|
return depth[key]
|
|
|
|
stages = []
|
|
for node in sorted(samplers, key=lambda item: (level(str(item["id"])), int(item["id"]) if str(item["id"]).isdigit() else 0)):
|
|
key = str(node["id"])
|
|
steps, start, end = value(node, "steps"), value(node, "start_at_step"), value(node, "end_at_step")
|
|
window = None if start is None or not isinstance(steps, int) else [start, min(end if isinstance(end, int) else steps, steps)]
|
|
enlarge = next((nodes[item] for item in fed[key][0] if UPSCALE.search(nodes[item]["type"])), None)
|
|
stages.append({**model_of(node), "steps": steps, "window": window, "denoise": value(node, "denoise"),
|
|
"upscale": (value(enlarge, "scale_by") or True) if enlarge else None,
|
|
"kind": "continue" if fed[key][1] else "base",
|
|
"state": "off" if _off(node) else "skipped" if window and window[0] >= window[1] else "run"})
|
|
staged = set().union(*[fed[str(node["id"])][0] for node in samplers]) if samplers else set()
|
|
after, todo = [], list(samplers)
|
|
users = {}
|
|
for node in nodes.values():
|
|
for item in node.get("inputs") or []:
|
|
found = source(item)
|
|
if found:
|
|
users.setdefault(str(found["id"]), []).append(node)
|
|
seen = set()
|
|
while todo:
|
|
for user in users.get(str(todo.pop()["id"]), []):
|
|
key = str(user["id"])
|
|
if key in seen or key in staged or _sampler(user):
|
|
continue
|
|
seen.add(key)
|
|
if not PLUMBING.search(user["type"]) and not SAVE.search(user["type"]) and not _off(user):
|
|
after.append(user.get("title") or user["type"])
|
|
todo.append(user)
|
|
return stages, after
|
|
|
|
|
|
def line(workflow, names=widget_names):
|
|
"""The route as one line for the AI, or "" when it has no sampler."""
|
|
stages, after = route(workflow, names)
|
|
parts = []
|
|
for stage in stages:
|
|
if stage["state"] != "run":
|
|
continue
|
|
words = ["base" if stage["kind"] == "base" else "redraw" if stage["upscale"] else "continue"]
|
|
if stage["upscale"]:
|
|
words.insert(0, f"enlarge {stage['upscale']}x," if stage["upscale"] is not True else "enlarge,")
|
|
if stage["window"]:
|
|
words.append(f"steps {stage['window'][0]}-{stage['window'][1]}")
|
|
elif isinstance(stage["steps"], int):
|
|
words.append(f"{stage['steps']} steps")
|
|
if isinstance(stage["denoise"], (int, float)) and stage["denoise"] < 1:
|
|
words.append(f"denoise {stage['denoise']}")
|
|
words.append(f"on {stage['model'] or 'its model'}")
|
|
if stage["loras"]:
|
|
words.append("+ LoRA " + ", ".join(dict.fromkeys(stage["loras"])))
|
|
if stage["controls"]:
|
|
words.append("with " + ", ".join(dict.fromkeys(stage["controls"])))
|
|
parts.append(" ".join(words))
|
|
if not parts:
|
|
return ""
|
|
passes = f"{len(parts)} pass{'es' if len(parts) > 1 else ''}: "
|
|
return passes + " -> ".join(parts) + (f" -> then {', '.join(dict.fromkeys(after))}" if after else "")
|