File size: 10,190 Bytes
cfdb90d 6d8bd76 cfdb90d 6d8bd76 cfdb90d 6d8bd76 cfdb90d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 | #!/usr/bin/env python3
"""Convert a ComfyUI workflow from UI format (Save/Workflow menu) to API format.
Equivalent to the frontend's Workflow > Export (API), but headless. Needs a
running ComfyUI instance because widget-to-input mapping comes from its
GET /object_info (so the instance must have the same custom nodes installed
as the workflow uses).
Handles: subgraphs (flattened, ids become "<instance>:<inner>"), Reroute
splicing, bypassed (mode 4) nodes with type-matched pass-through, muted
(mode 2) nodes, widget-converted-to-input entries, and the phantom
control_after_generate value that follows every seed widget.
Example:
python convert_workflow.py --url http://127.0.0.1:7865 \
../head-swap-flux-klein.json -o ../api/head-swap-flux-klein.json
"""
import argparse
import json
import pathlib
import sys
import requests
VIRTUAL_TYPES = {"Reroute", "Note", "MarkdownNote", "PrimitiveNode"}
WIDGET_SCALAR_TYPES = {"INT", "FLOAT", "STRING", "BOOLEAN", "COMBO"}
SUBGRAPH_INPUT_ID = -10
SUBGRAPH_OUTPUT_ID = -20
class Scope:
"""One graph namespace: either the root workflow or a subgraph instance."""
def __init__(self, graph, prefix="", parent=None, instance_id=None):
self.nodes = {n["id"]: n for n in graph.get("nodes", [])}
self.links = {}
for l in graph.get("links", []):
if isinstance(l, dict): # subgraph defs store links as objects
self.links[l["id"]] = (l["origin_id"], l["origin_slot"], l["target_id"], l["target_slot"])
else: # root stores [id, src, src_slot, dst, dst_slot, type]
self.links[l[0]] = (l[1], l[2], l[3], l[4])
self.prefix = prefix
self.parent = parent # (parent Scope, instance node id) or None
self.instance_id = instance_id
self.children = {} # instance node id -> child Scope
def fid(self, node_id):
return f"{self.prefix}{node_id}"
class Converter:
def __init__(self, workflow, object_info):
self.oi = object_info
self.subgraph_defs = {
sg["id"]: sg for sg in workflow.get("definitions", {}).get("subgraphs", [])
}
self.root = Scope(workflow)
self._build_children(self.root)
def _build_children(self, scope):
for nid, node in scope.nodes.items():
sg = self.subgraph_defs.get(node["type"])
if sg is not None:
child = Scope(sg, prefix=f"{scope.fid(nid)}:", parent=(scope, nid), instance_id=nid)
scope.children[nid] = child
self._build_children(child)
def all_scopes(self):
stack = [self.root]
while stack:
s = stack.pop()
yield s
stack.extend(s.children.values())
def resolve(self, scope, link_id):
"""Follow a link upstream to a real executable (node_fid, output_slot), or None."""
if link_id is None or link_id not in scope.links:
return None
origin_id, origin_slot, _, _ = scope.links[link_id]
if origin_id == SUBGRAPH_INPUT_ID: # crossed into this subgraph from outside
parent_scope, instance_id = scope.parent
instance = parent_scope.nodes[instance_id]
# instances serialize only a subset of the def's inputs (widget-promoted
# ones live in proxyWidgets instead), so match by name, not slot index
name = self.subgraph_defs[instance["type"]]["inputs"][origin_slot]["name"]
entry = next((i for i in instance.get("inputs", []) if i["name"] == name), None)
if entry is None:
return None # unwired promoted widget: inner node keeps its own value
return self.resolve(parent_scope, entry.get("link"))
node = scope.nodes[origin_id]
ntype = node["type"]
if ntype in self.subgraph_defs: # source is a subgraph instance: descend
child = scope.children[origin_id]
for lid, (_, _, tid, tslot) in child.links.items():
if tid == SUBGRAPH_OUTPUT_ID and tslot == origin_slot:
return self.resolve(child, lid)
return None
if ntype == "Reroute":
return self.resolve(scope, node["inputs"][0].get("link"))
mode = node.get("mode", 0)
if mode == 2: # muted: acts as removed
return None
if mode == 4: # bypassed: pass through first linked input of matching type
out_type = node["outputs"][origin_slot].get("type", "*")
for inp in node.get("inputs", []):
if inp.get("link") is None:
continue
in_type = inp.get("type", "*")
if in_type == out_type or "*" in (in_type, out_type):
return self.resolve(scope, inp["link"])
return None
return (scope.fid(origin_id), origin_slot)
def widget_inputs(self, node):
"""Map a node's widgets_values onto named inputs using /object_info order."""
if "Power Lora Loader" in node["type"]:
# rgthree serializes lora rows as dicts among placeholder widgets;
# the API expects them as lora_1..lora_N dict inputs
return {
f"lora_{i}": row
for i, row in enumerate(
(v for v in node.get("widgets_values") or []
if isinstance(v, dict) and "lora" in v),
start=1,
)
}
spec = self.oi[node["type"]].get("input", {})
ordered = list(spec.get("required", {}).items()) + list(spec.get("optional", {}).items())
values = node.get("widgets_values") or []
if isinstance(values, dict): # some packs (rgthree) serialize by name
widget_names = {name for name, s in ordered if self._is_widget(s)}
return {k: v for k, v in values.items() if k in widget_names}
result = {}
self._consume_widgets(ordered, list(values), result)
return result
def _consume_widgets(self, ordered, queue, result, prefix=""):
for name, s in ordered:
if not queue:
return
opts = s[1] if len(s) > 1 and isinstance(s[1], dict) else {}
if s[0] == "COMFY_DYNAMICCOMBO_V3":
# the selection is one widget value; the chosen option pulls in
# nested inputs addressed with dotted ids ("resize_type.megapixels")
selected = queue.pop(0)
result[f"{prefix}{name}"] = selected
option = next((o for o in opts.get("options", []) if o["key"] == selected), None)
if option:
inner = option["inputs"]
inner_ordered = (list(inner.get("required", {}).items())
+ list(inner.get("optional", {}).items()))
self._consume_widgets(inner_ordered, queue, result, prefix=f"{prefix}{name}.")
elif self._is_widget(s):
result[f"{prefix}{name}"] = queue.pop(0)
# discard the UI-only seed-control widget value; some packs
# (e.g. RES4LYF) add it in JS without declaring it in the spec
if queue and (opts.get("control_after_generate")
or (name in ("seed", "noise_seed")
and queue[0] in ("fixed", "increment", "decrement", "randomize"))):
queue.pop(0)
@staticmethod
def _is_widget(s):
kind = s[0]
opts = s[1] if len(s) > 1 and isinstance(s[1], dict) else {}
if opts.get("forceInput"):
return False
return isinstance(kind, list) or kind in WIDGET_SCALAR_TYPES
def convert(self):
prompt = {}
for scope in self.all_scopes():
for nid, node in scope.nodes.items():
ntype = node["type"]
if (ntype in VIRTUAL_TYPES or ntype in self.subgraph_defs
or node.get("mode", 0) in (2, 4)):
continue
if ntype not in self.oi:
raise SystemExit(
f"ERROR: node {scope.fid(nid)} has type {ntype!r} which the ComfyUI "
"instance does not know -- install its custom nodes first")
inputs = self.widget_inputs(node)
for inp in node.get("inputs", []):
source = self.resolve(scope, inp.get("link"))
if source is not None:
inputs[inp["name"]] = list(source)
elif inp.get("link") is not None and not inp.get("widget"):
print(f"warning: {scope.fid(nid)} ({ntype}) input {inp['name']!r} "
"leads to a muted/bypassed dead end; leaving unconnected",
file=sys.stderr)
entry = {"class_type": ntype, "inputs": inputs}
title = node.get("title")
if title:
entry["_meta"] = {"title": title}
prompt[scope.fid(nid)] = entry
return prompt
def main():
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--url", required=True, help="ComfyUI base URL (source of /object_info)")
ap.add_argument("workflow", help="UI-format workflow JSON")
ap.add_argument("-o", "--output", help="where to write the API JSON (default: stdout)")
args = ap.parse_args()
wf = json.loads(pathlib.Path(args.workflow).read_text())
if "nodes" not in wf:
sys.exit("ERROR: input already looks like API format (no 'nodes' array)")
object_info = requests.get(f"{args.url.rstrip('/')}/object_info", timeout=120).json()
prompt = Converter(wf, object_info).convert()
out = json.dumps(prompt, indent=2, sort_keys=True)
if args.output:
pathlib.Path(args.output).write_text(out + "\n")
print(f"wrote {args.output} ({len(prompt)} nodes)", file=sys.stderr)
else:
print(out)
if __name__ == "__main__":
main()
|