ComfyUI / workflows /scripts /convert_workflow.py

Add remaining API-format workflows; converter fixes (subgraph promoted inputs, rgthree loras, undeclared seed-control widgets)

6d8bd76 verified 4 months ago

Raw History Blame

10.2 kB

#!/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()

	#!/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()