Merge pull request #8 from jimpi-dev/feature-addworkfowimport

Feature workflow import + custom node for workflowui app import/config
This commit is contained in:
jimpi-dev
2026-03-08 00:04:55 +01:00
committed by GitHub
7 changed files with 1478 additions and 72 deletions
+4 -15
View File
@@ -1,21 +1,10 @@
"""
ComfyUI custom node: WorkflowUI plugin.
Registers API routes so WorkflowUI (and other clients) can:
- Delete files from ComfyUI output/input/temp (POST /delete, POST /workflowui/media/delete)
- List and browse media (GET /workflowui/media/list, GET /workflowui/media/tree)
- Detect plugin (GET /workflowui/media/capabilities)
- Installed modules/versions for run metadata (GET /workflowui/version_info)
No graph nodes are added; this is a route-only plugin.
"""
from .version_info import __version__
from .routes import register_routes
from .workflow_ui_link import NODE_CLASS_MAPPINGS as LINK_MAPPINGS
from .workflow_ui_link import NODE_DISPLAY_NAME_MAPPINGS as LINK_DISPLAY_NAMES
register_routes()
# Startup banner
_banner_text = f" WorkflowUI Plugin v{__version__} "
_width = max(len(_banner_text) + 2, 24)
_border = "+" + "-" * (_width - 2) + "+"
@@ -25,7 +14,7 @@ print("|" + _banner_text.center(_width - 2) + "|")
print(_border)
print()
NODE_CLASS_MAPPINGS = {}
NODE_DISPLAY_NAME_MAPPINGS = {}
NODE_CLASS_MAPPINGS = dict(LINK_MAPPINGS)
NODE_DISPLAY_NAME_MAPPINGS = dict(LINK_DISPLAY_NAMES)
__all__ = ["NODE_CLASS_MAPPINGS", "NODE_DISPLAY_NAME_MAPPINGS"]
+1 -1
View File
@@ -1,7 +1,7 @@
[project]
name = "WorkflowUIPlugin"
description = "This custom node is a plugin for: WorkflowUI Application, found here: https://github.com/jimpi-dev/WorkflowUI"
version = "1.0.4"
version = "1.0.6"
license = { file = "LICENSE" }
[project.urls]
+235 -33
View File
@@ -1,36 +1,64 @@
"""
WorkflowUI plugin: ComfyUI API routes for media access and deletion.
- POST /delete and POST /workflowui/media/delete — delete a file by filename/subfolder/type
(WorkflowUI calls POST /delete with JSON body).
- GET /workflowui/media/view — serve a file from output/input/temp; ?preview=webp|jpeg|SIZE for compressed image previews.
- GET /workflowui/media/list — list files in output/input/temp (optional subfolder).
- GET /workflowui/media/tree — tree of subfolders and files for browsing.
- GET /workflowui/media/capabilities — report supported features (delete, list, view).
- GET /workflowui/version_info — installed custom node modules and versions (for run metadata).
"""
from __future__ import annotations
import hashlib
import importlib.util
import json
import io
import logging
import mimetypes
import os
import sys
from pathlib import Path
from aiohttp import web
logger = logging.getLogger(__name__)
def _load_workflow_converter():
try:
from .workflow_converter import WorkflowConverter
return WorkflowConverter
except SyntaxError as e:
if "null bytes" not in str(e).lower():
raise
logger.warning(
"WorkflowUIPlugin: workflow_converter.py contains null bytes in source; "
"loading with null bytes stripped (check deployment/sync)."
)
pkg_dir = os.path.dirname(os.path.abspath(__file__))
path = os.path.join(pkg_dir, "workflow_converter.py")
with open(path, "rb") as f:
source = f.read().replace(b"\x00", b"").decode("utf-8", errors="replace")
fullname = (__package__ or "WorkflowUIPlugin") + ".workflow_converter"
class _Loader:
def __init__(self, src, pathname):
self.src = src
self.pathname = pathname
def create_module(self, spec):
return None
def exec_module(self, module):
code = compile(self.src, self.pathname, "exec")
exec(code, module.__dict__)
spec = importlib.util.spec_from_loader(fullname, _Loader(source, path), origin=path)
module = importlib.util.module_from_spec(spec)
sys.modules[fullname] = module
spec.loader.exec_module(module)
return module.WorkflowConverter
MEDIA_TYPES = ("output", "input", "temp")
PREVIEW_MAX_DIM_DEFAULT = 512
PREVIEW_QUALITY_DEFAULT = 85
IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tiff", ".tif"}
UPLOAD_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".gif"}
UPLOAD_VIDEO_EXTENSIONS = {".mp4", ".webm", ".mkv", ".mov"}
UPLOAD_AUDIO_EXTENSIONS = {".mp3", ".wav", ".ogg", ".flac", ".m4a"}
def _get_media_root(folder_type: str):
"""Resolve ComfyUI media root for output/input/temp. Returns None if invalid."""
try:
import folder_paths
out = folder_paths.get_directory_by_type(folder_type)
@@ -40,20 +68,14 @@ def _get_media_root(folder_type: str):
def _resolve_media_path(folder_type: str, subfolder: str, filename: str) -> tuple[str | None, str | None]:
"""
Resolve full_path for a media file. Ensures path is under base (no traversal).
Returns (full_path, None) on success; (None, error_message) on validation failure.
"""
if folder_type not in MEDIA_TYPES:
return None, "Invalid type; use output, input, or temp"
base = _get_media_root(folder_type)
if not base:
return None, "Could not resolve media directory"
# Normalize subfolder: no leading/trailing slashes, no '..'
sub = (subfolder or "").strip().replace("\\", "/").strip("/")
if ".." in sub or sub.startswith("/"):
return None, "Invalid subfolder"
# Filename must be a single segment (no path)
name = (filename or "").strip()
if not name or "/" in name or "\\" in name or ".." in name:
return None, "Invalid filename"
@@ -65,7 +87,6 @@ def _resolve_media_path(folder_type: str, subfolder: str, filename: str) -> tupl
def _delete_file(folder_type: str, subfolder: str, filename: str) -> tuple[bool, str, int]:
"""Delete one file. Returns (success, message, http_status)."""
full, err = _resolve_media_path(folder_type, subfolder, filename)
if err:
return False, err, 400
@@ -92,7 +113,6 @@ def _make_image_preview(
max_dim: int,
quality: int,
) -> tuple[bytes | None, str | None]:
"""Return (body, content_type) for a resized/compressed preview, or (None, None) on failure."""
try:
from PIL import Image
except ImportError:
@@ -124,7 +144,6 @@ def _make_image_preview(
async def _handle_view(request: web.Request) -> web.Response:
"""GET ?filename=...&subfolder=...&type=output&preview=webp|jpeg|SIZE — serve file; optional compressed image preview."""
filename = request.query.get("filename", "").strip()
subfolder = request.query.get("subfolder", "")
folder_type = (request.query.get("type") or "output").strip().lower()
@@ -142,7 +161,6 @@ async def _handle_view(request: web.Request) -> web.Response:
if not content_type:
content_type = "application/octet-stream"
# Optional preview: only for images; format webp (default) or jpeg, optional max dimension
if preview and _is_image_file(filename, content_type):
max_dim = PREVIEW_MAX_DIM_DEFAULT
quality = PREVIEW_QUALITY_DEFAULT
@@ -168,7 +186,6 @@ async def _handle_view(request: web.Request) -> web.Response:
async def _handle_delete(request: web.Request) -> web.Response:
"""POST body: JSON { filename, subfolder?, type? }. type defaults to 'output'."""
try:
body = await request.json()
except Exception:
@@ -186,8 +203,74 @@ async def _handle_delete(request: web.Request) -> web.Response:
return web.json_response({"ok": False, "error": msg}, status=status)
def _content_hash_name(content: bytes, ext: str) -> str:
h = hashlib.sha256(content).hexdigest()[:16]
return f"{h}{ext}"
async def _handle_upload(request: web.Request) -> web.Response:
try:
reader = await request.multipart()
except Exception as e:
logger.debug("WorkflowUIPlugin: upload multipart error: %s", e)
return web.json_response({"error": "Invalid multipart body"}, status=400)
media_type = (request.query.get("type") or "image").strip().lower()
if media_type not in ("image", "video", "audio"):
return web.json_response({"error": "type must be image, video, or audio"}, status=400)
if media_type == "image":
allowed_ext = UPLOAD_IMAGE_EXTENSIONS
elif media_type == "video":
allowed_ext = UPLOAD_VIDEO_EXTENSIONS
else:
allowed_ext = UPLOAD_AUDIO_EXTENSIONS
part = None
while True:
try:
p = await reader.next()
except Exception:
p = None
if p is None:
break
name = getattr(p, "name", None) or ""
if name in ("file", "image"):
part = p
break
if part is None:
return web.json_response({"error": "Missing form field 'file' or 'image'"}, status=400)
filename = getattr(part, "filename", None) or "upload"
ext = Path(filename).suffix.lower()
if ext not in allowed_ext:
return web.json_response(
{"error": f"Unsupported {media_type} format. Allowed: {', '.join(sorted(allowed_ext))}"},
status=400,
)
try:
content = await part.read()
except Exception as e:
logger.warning("WorkflowUIPlugin: upload read failed: %s", e)
return web.json_response({"error": "Failed to read uploaded file"}, status=400)
if not content:
return web.json_response({"error": "Empty file"}, status=400)
base = _get_media_root("input")
if not base:
return web.json_response({"error": "Could not resolve input directory"}, status=500)
safe_name = _content_hash_name(content, ext)
full = os.path.join(base, safe_name)
try:
with open(full, "wb") as f:
f.write(content)
except OSError as e:
logger.warning("WorkflowUIPlugin: upload write failed for %s: %s", full, e)
return web.json_response({"error": f"Failed to save file: {e}"}, status=500)
return web.json_response({
"filename": safe_name,
"name": safe_name,
"subfolder": "",
"type": "input",
})
def _list_dir(folder_type: str, subfolder: str) -> tuple[list | None, str | None]:
"""List files in type/subfolder. Returns (list of {name, size, mtime}, error)."""
if folder_type not in MEDIA_TYPES:
return None, "Invalid type"
base = _get_media_root(folder_type)
@@ -223,7 +306,6 @@ def _list_dir(folder_type: str, subfolder: str) -> tuple[list | None, str | None
async def _handle_list(request: web.Request) -> web.Response:
"""GET ?type=output&subfolder= (subfolder optional)."""
folder_type = (request.query.get("type") or "output").strip().lower()
subfolder = request.query.get("subfolder", "")
data, err = _list_dir(folder_type, subfolder)
@@ -233,7 +315,6 @@ async def _handle_list(request: web.Request) -> web.Response:
def _tree_dir(folder_type: str, subfolder: str) -> tuple[dict | None, str | None]:
"""Return tree { folders: [name], files: [{filename, size, mtime}] } for one level."""
if folder_type not in MEDIA_TYPES:
return None, "Invalid type"
base = _get_media_root(folder_type)
@@ -268,7 +349,6 @@ def _tree_dir(folder_type: str, subfolder: str) -> tuple[dict | None, str | None
async def _handle_tree(request: web.Request) -> web.Response:
"""GET ?type=output&subfolder= (one level: folders + files)."""
folder_type = (request.query.get("type") or "output").strip().lower()
subfolder = request.query.get("subfolder", "")
data, err = _tree_dir(folder_type, subfolder)
@@ -283,7 +363,6 @@ async def _handle_tree(request: web.Request) -> web.Response:
async def _handle_capabilities(request: web.Request) -> web.Response:
"""GET /workflowui/media/capabilities — report what this plugin supports."""
return web.json_response({
"workflowui_plugin": True,
"delete": True,
@@ -291,29 +370,152 @@ async def _handle_capabilities(request: web.Request) -> web.Response:
"tree": True,
"view": True,
"view_preview": True,
"upload": True,
})
async def _handle_version_info(request: web.Request) -> web.Response:
"""GET /workflowui/version_info — installed custom node modules and versions for run metadata."""
from .version_info import get_version_info_payload
return web.json_response(get_version_info_payload())
async def _handle_workflows_list(request: web.Request) -> web.Response:
from .workflow_export import list_workflows
try:
items = list_workflows()
return web.json_response(items)
except Exception as e:
logger.warning("WorkflowUIPlugin: workflows list failed: %s", e)
return web.json_response([])
async def _handle_workflow_by_id(request: web.Request) -> web.Response:
from .workflow_export import load_workflow
workflow_id = request.match_info.get("id", "").strip()
if not workflow_id:
return web.json_response({"error": "Missing workflow id"}, status=400)
try:
name, graph = load_workflow(workflow_id)
return web.json_response({"name": name, "graph": graph})
except FileNotFoundError as e:
return web.json_response({"error": str(e)}, status=404)
except ValueError as e:
return web.json_response({"error": str(e)}, status=400)
except Exception as e:
logger.exception("WorkflowUIPlugin: load workflow %s failed", workflow_id)
return web.json_response({"error": str(e)}, status=500)
WORKFLOWCONVERTER_MAX_CONTENT_LENGTH = 1 * 1024 * 1024
def _strip_null_bytes(obj):
if isinstance(obj, dict):
out = {}
for k, v in obj.items():
key = k
if isinstance(k, str) and ("\x00" in k or "\u0000" in k):
key = k.replace("\x00", "").replace("\u0000", "")
out[key] = _strip_null_bytes(v)
return out
if isinstance(obj, list):
return [_strip_null_bytes(v) for v in obj]
if isinstance(obj, str):
if "\x00" in obj or "\u0000" in obj:
return obj.replace("\x00", "").replace("\u0000", "")
return obj
return obj
async def _handle_workflowconverter_convert_post(request: web.Request) -> web.Response:
logger.info("WorkflowUIPlugin: workflowconverter/convert POST received")
if request.content_length is not None and request.content_length > WORKFLOWCONVERTER_MAX_CONTENT_LENGTH:
return web.json_response({
"success": False,
"error": f"Request too large. Maximum size is {WORKFLOWCONVERTER_MAX_CONTENT_LENGTH // (1024 * 1024)} MB",
}, status=413)
try:
raw = await request.read()
raw = raw.replace(b"\x00", b"")
body = json.loads(raw.decode("utf-8", errors="replace"))
except Exception as e:
logger.debug("WorkflowUIPlugin: workflowconverter/convert invalid JSON: %s", e)
return web.json_response({"success": False, "error": f"Invalid JSON: {e!s}"}, status=400)
if not isinstance(body, dict):
return web.json_response({"success": False, "error": "Body must be a JSON object"}, status=400)
body = _strip_null_bytes(body)
try:
WorkflowConverter = _load_workflow_converter()
if WorkflowConverter.is_api_format(body):
return web.json_response(body)
if "nodes" not in body or "links" not in body:
return web.json_response({
"success": False,
"error": "Invalid workflow format - missing nodes or links",
}, status=400)
if not isinstance(body.get("nodes"), list):
return web.json_response({
"success": False,
"error": "Invalid workflow format - 'nodes' must be a list",
}, status=400)
if not isinstance(body.get("links"), list):
return web.json_response({
"success": False,
"error": "Invalid workflow format - 'links' must be a list",
}, status=400)
api_prompt = WorkflowConverter.convert_to_api(body)
num_nodes = len(body.get("nodes", []))
num_links = len(body.get("links", []))
num_converted = len(api_prompt)
logger.info(
"WorkflowUIPlugin: workflowconverter converted %s nodes, %s links -> %s API nodes",
num_nodes, num_links, num_converted,
)
return web.json_response(api_prompt)
except Exception as e:
import traceback
tb = traceback.format_exc()
logger.error("WorkflowUIPlugin: workflowconverter/convert failed: %s", e)
logger.error("Traceback: %s", tb)
err_msg = str(e).strip() or "Internal server error during conversion"
if len(err_msg) > 500:
err_msg = err_msg[:497] + "..."
return web.json_response({
"success": False,
"error": err_msg,
"traceback": tb,
}, status=500)
async def _handle_workflowconverter_convert_get(request: web.Request) -> web.Response:
return web.json_response({
"name": "WorkflowUI Workflow Converter",
"version": "2.1.0",
"description": "Converts non-API workflow format to API format for execution (subgraphs, bypassed nodes, GetNode/SetNode). Part of WorkflowUIPlugin.",
"usage": "POST a workflow JSON to this endpoint to convert it to API format",
"endpoint": "/workflowui/workflowconverter/convert",
"source": "Based on Seth A. Robinson's comfyui-workflow-to-api-converter-endpoint",
"repository": "https://github.com/SethRobinson/comfyui-workflow-to-api-converter-endpoint",
})
def register_routes():
"""Register routes with ComfyUI PromptServer. Call from __init__.py."""
try:
from server import PromptServer
routes = PromptServer.instance.routes
except Exception as e:
logger.warning("WorkflowUIPlugin: could not get PromptServer routes: %s", e)
return
# WorkflowUI backend calls POST /delete with JSON { filename, subfolder, type }
routes.post("/delete")(_handle_delete)
routes.post("/workflowui/media/delete")(_handle_delete)
routes.post("/workflowui/media/upload")(_handle_upload)
routes.get("/workflowui/media/view")(_handle_view)
routes.get("/workflowui/media/list")(_handle_list)
routes.get("/workflowui/media/tree")(_handle_tree)
routes.get("/workflowui/media/capabilities")(_handle_capabilities)
routes.get("/workflowui/version_info")(_handle_version_info)
logger.info("WorkflowUIPlugin: registered media routes (delete, view, list, tree, capabilities, version_info)")
routes.get("/workflowui/workflows")(_handle_workflows_list)
routes.get("/workflowui/workflows/{id}")(_handle_workflow_by_id)
routes.post("/workflowui/workflowconverter/convert")(_handle_workflowconverter_convert_post)
routes.get("/workflowui/workflowconverter/convert")(_handle_workflowconverter_convert_get)
logger.info("WorkflowUIPlugin: registered media + workflow export + workflowconverter routes")
+1 -23
View File
@@ -4,7 +4,7 @@ import json
import re
from pathlib import Path
__version__ = "1.0.4"
__version__ = "1.0.10"
_PLUGIN_DIR = Path(__file__).resolve().parent
_CUSTOM_NODES_DIR = _PLUGIN_DIR.parent
@@ -14,7 +14,6 @@ _PYPROJECT_VERSION_RE = re.compile(r'^\s*version\s*=\s*["\']([^"\']+)["\']', re.
def _version_from_init_py(init_path: Path) -> str | None:
"""Try to read __version__ from a package __init__.py without importing."""
if not init_path.is_file():
return None
try:
@@ -26,7 +25,6 @@ def _version_from_init_py(init_path: Path) -> str | None:
def _version_from_metadata(package_name: str) -> str | None:
"""Try to get version from importlib.metadata (pip-installed packages)."""
try:
from importlib.metadata import version
for name in (package_name, package_name.replace("_", "-"), package_name.replace("-", "_")):
@@ -40,7 +38,6 @@ def _version_from_metadata(package_name: str) -> str | None:
def _version_from_pyproject_toml(package_dir: Path) -> str | None:
"""Try to read version from pyproject.toml ([project] or [tool.poetry] version)."""
path = package_dir / "pyproject.toml"
if not path.is_file():
return None
@@ -54,7 +51,6 @@ def _version_from_pyproject_toml(package_dir: Path) -> str | None:
def _version_from_package_json(package_dir: Path) -> str | None:
"""Try to read version from package.json."""
path = package_dir / "package.json"
if not path.is_file():
return None
@@ -68,7 +64,6 @@ def _version_from_package_json(package_dir: Path) -> str | None:
def _version_from_version_file(package_dir: Path) -> str | None:
"""Try __version__ in version.py (or similar) in the package dir."""
for name in ("version.py", "version.json"):
path = package_dir / name
if not path.is_file():
@@ -89,20 +84,10 @@ def _version_from_version_file(package_dir: Path) -> str | None:
def _is_git_repo(package_dir: Path) -> bool:
"""True if package_dir is the root of a git repo (has .git)."""
return (package_dir / ".git").exists()
def get_installed_modules() -> list[dict]:
"""
List installed custom node modules (directories under custom_nodes with __init__.py)
and their version if discoverable.
Returns a list of {"name": str, "version": str | null} sorted by name.
Version is resolved from: __init__.py, importlib.metadata, pyproject.toml,
package.json, version.py; if still unknown and the dir is a git repo, "nightly".
Does not import custom node code.
"""
result: list[dict] = []
if not _CUSTOM_NODES_DIR.is_dir():
return result
@@ -129,13 +114,6 @@ def get_installed_modules() -> list[dict]:
def get_version_info_payload() -> dict:
"""
Build the JSON payload for GET /workflowui/version_info.
Includes:
- workflowui_plugin_version: this plugin's version
- installed_custom_nodes: list of { name, version } for each custom node package
"""
return {
"workflowui_plugin_version": __version__,
"installed_custom_nodes": get_installed_modules(),
+800
View File
@@ -0,0 +1,800 @@
import json
import logging
from typing import Dict, Any, List, Tuple, Optional, Union
logger = logging.getLogger(__name__)
try:
import nodes
except ImportError as e:
raise ImportError('Cannot import ComfyUI nodes module. This converter must be run within the ComfyUI environment. Make sure ComfyUI is properly initialized before using the converter.') from e
_node_info_cache = {}
def _strip_nulls(s: str) -> str:
if not isinstance(s, str):
return s
return s.replace('\x00', '').replace('\x00', '')
def get_node_info_for_type(node_type: str) -> Dict[str, Any]:
global _node_info_cache
node_type = _strip_nulls(node_type)
if node_type not in _node_info_cache:
if node_type in nodes.NODE_CLASS_MAPPINGS:
try:
obj_class = nodes.NODE_CLASS_MAPPINGS[node_type]
info = {}
info['input'] = obj_class.INPUT_TYPES()
info['input_order'] = {key: list(value.keys()) for key, value in obj_class.INPUT_TYPES().items()}
_node_info_cache[node_type] = info
except ValueError as e:
if 'null bytes' in str(e).lower():
logger.error("WorkflowConverter: node type '%s' triggered 'source code string cannot contain null bytes'. Treating as unknown (using fallback input order). Check this custom node's source or ComfyUI.", node_type, exc_info=True)
_node_info_cache[node_type] = None
return None
raise
except Exception as e:
logger.debug('Could not get node info for %s: %s', node_type, e)
_node_info_cache[node_type] = None
else:
_node_info_cache[node_type] = None
return _node_info_cache.get(node_type)
class WorkflowConverter:
@staticmethod
def is_subgraph_uuid(node_type: str) -> bool:
if not node_type or not isinstance(node_type, str):
return False
if len(node_type) == 36 and node_type.count('-') == 4:
parts = node_type.split('-')
if len(parts) == 5 and all((len(p) in [8, 4, 4, 4, 12] for i, p in enumerate(parts) if i == 0 or i == 4 or len(p) == 4)):
return True
return False
@staticmethod
def expand_subgraph(subgraph_node_id: int, subgraph_def: Dict[str, Any], workflow_links: List) -> Tuple[List[Dict], List]:
expanded_nodes = []
expanded_links = []
internal_nodes = subgraph_def.get('nodes', [])
internal_links = subgraph_def.get('links', [])
max_link_id = 0
for link in workflow_links:
if isinstance(link, (list, tuple)) and len(link) > 0:
link_id = link[0]
if isinstance(link_id, int) and link_id > max_link_id:
max_link_id = link_id
link_id_remap = {}
next_link_id = max_link_id + 1
for link in internal_links:
if isinstance(link, dict):
old_id = link.get('id')
if old_id is not None:
link_id_remap[old_id] = next_link_id
next_link_id += 1
internal_link_map = {}
for link in internal_links:
if isinstance(link, dict):
link_id = link.get('id')
internal_link_map[link_id] = link
subgraph_inputs = subgraph_def.get('inputs', [])
subgraph_outputs = subgraph_def.get('outputs', [])
input_slot_to_internal = {}
internal_to_output_slot = {}
for idx, input_def in enumerate(subgraph_inputs):
input_link_ids = input_def.get('linkIds', [])
targets = []
for link_id in input_link_ids:
if link_id in internal_link_map:
link = internal_link_map[link_id]
target_id = link.get('target_id')
target_slot = link.get('target_slot')
targets.append((target_id, target_slot))
if targets:
input_slot_to_internal[idx] = targets
for idx, output_def in enumerate(subgraph_outputs):
output_link_ids = output_def.get('linkIds', [])
for link_id in output_link_ids:
if link_id in internal_link_map:
link = internal_link_map[link_id]
origin_id = link.get('origin_id')
origin_slot = link.get('origin_slot')
internal_to_output_slot[origin_id, origin_slot] = idx
for node in internal_nodes:
internal_id = node.get('id')
expanded_node = node.copy()
expanded_node['id'] = f'{subgraph_node_id}:{internal_id}'
if 'inputs' in expanded_node:
updated_inputs = []
for input_info in expanded_node['inputs']:
input_link = input_info.get('link')
if input_link in internal_link_map:
link_data = internal_link_map[input_link]
if link_data.get('origin_id') == -10:
input_copy = input_info.copy()
input_copy['link'] = None
updated_inputs.append(input_copy)
else:
input_copy = input_info.copy()
if input_link in link_id_remap:
input_copy['link'] = link_id_remap[input_link]
updated_inputs.append(input_copy)
else:
updated_inputs.append(input_info)
expanded_node['inputs'] = updated_inputs
expanded_nodes.append(expanded_node)
for link in internal_links:
if isinstance(link, dict):
origin_id = link.get('origin_id')
target_id = link.get('target_id')
if origin_id in [-10, -20] or target_id in [-10, -20]:
continue
old_link_id = link.get('id')
new_link_id = link_id_remap.get(old_link_id, old_link_id)
expanded_link = [new_link_id, f'{subgraph_node_id}:{origin_id}', link.get('origin_slot'), f'{subgraph_node_id}:{target_id}', link.get('target_slot'), link.get('type')]
expanded_links.append(expanded_link)
return (expanded_nodes, expanded_links, input_slot_to_internal, internal_to_output_slot)
@staticmethod
def is_api_format(workflow: Dict[str, Any]) -> bool:
if 'nodes' in workflow and 'links' in workflow:
return False
for key, value in workflow.items():
if key in ['prompt', 'extra_data', 'client_id']:
continue
if isinstance(value, dict) and 'class_type' in value:
return True
return False
@staticmethod
def _strip_null_bytes(obj: Any) -> Any:
if isinstance(obj, dict):
out = {}
for k, v in obj.items():
key = k.replace('\x00', '').replace('\x00', '') if isinstance(k, str) else k
out[key] = WorkflowConverter._strip_null_bytes(v)
return out
if isinstance(obj, list):
return [WorkflowConverter._strip_null_bytes(v) for v in obj]
if isinstance(obj, str):
return obj.replace('\x00', '').replace('\x00', '')
return obj
@staticmethod
def convert_to_api(workflow: Dict[str, Any]) -> Dict[str, Any]:
workflow = WorkflowConverter._strip_null_bytes(workflow)
if WorkflowConverter.is_api_format(workflow):
return workflow
workflow_nodes = workflow.get('nodes', [])
links = workflow.get('links', [])
node_types_in_workflow = sorted(set((n.get('type') or n.get('class_type') for n in workflow_nodes if isinstance(n, dict) and (n.get('type') or n.get('class_type')))))
logger.info('WorkflowConverter: starting conversion, %s nodes, %s links. Node types: %s', len(workflow_nodes), len(links), node_types_in_workflow[:40])
subgraph_defs = {}
definitions = workflow.get('definitions', {})
for subgraph in definitions.get('subgraphs', []):
subgraph_id = subgraph.get('id')
if subgraph_id:
subgraph_defs[subgraph_id] = subgraph
subgraph_input_mappings = {}
subgraph_output_mappings = {}
subgraph_slot_to_input_idx = {}
max_iterations = 10
iteration = 0
while iteration < max_iterations:
iteration += 1
expanded_nodes = []
found_subgraph = False
for node in workflow_nodes:
node_type = _strip_nulls(node.get('type') or '')
node_id = node.get('id')
if WorkflowConverter.is_subgraph_uuid(node_type) and node_type in subgraph_defs:
found_subgraph = True
logger.debug(f'Expanding subgraph node {node_id} (type: {node_type}) - iteration {iteration}')
subgraph_def = subgraph_defs[node_type]
sg_nodes, sg_links, input_map, output_map = WorkflowConverter.expand_subgraph(node_id, subgraph_def, links)
expanded_nodes.extend(sg_nodes)
links.extend(sg_links)
subgraph_input_mappings[str(node_id)] = input_map
subgraph_output_mappings[str(node_id)] = output_map
outer_inputs = node.get('inputs', [])
subgraph_inputs = subgraph_def.get('inputs', [])
subgraph_input_name_to_idx = {}
for idx, sg_input in enumerate(subgraph_inputs):
input_name = sg_input.get('name')
if input_name:
subgraph_input_name_to_idx[input_name] = idx
slot_mapping = {}
for outer_slot, outer_input in enumerate(outer_inputs):
outer_name = outer_input.get('name')
if outer_name and outer_name in subgraph_input_name_to_idx:
slot_mapping[outer_slot] = subgraph_input_name_to_idx[outer_name]
logger.debug(f"Subgraph {node_id}: outer slot {outer_slot} ('{outer_name}') -> subgraph input idx {subgraph_input_name_to_idx[outer_name]}")
subgraph_slot_to_input_idx[str(node_id)] = slot_mapping
else:
expanded_nodes.append(node)
workflow_nodes = expanded_nodes
if not found_subgraph:
logger.debug(f'Subgraph expansion complete after {iteration} iteration(s)')
break
if iteration >= max_iterations:
logger.warning(f'Reached maximum subgraph expansion iterations ({max_iterations}). There may be circular subgraph references.')
def resolve_subgraph_output(node_id_str, slot, depth=0):
if depth > 100:
logger.warning(f'Max recursion depth reached resolving subgraph output for node {node_id_str}')
return (node_id_str, slot)
if node_id_str in subgraph_output_mappings:
output_map = subgraph_output_mappings[node_id_str]
for (internal_node, internal_slot), out_slot in output_map.items():
if out_slot == slot:
new_node_id = f'{node_id_str}:{internal_node}'
return resolve_subgraph_output(new_node_id, internal_slot, depth + 1)
return (node_id_str, slot)
def resolve_subgraph_input(node_id_str, slot, depth=0):
if depth > 100:
logger.warning(f'Max recursion depth reached resolving subgraph input for node {node_id_str}')
return (node_id_str, slot)
if node_id_str in subgraph_input_mappings:
input_map = subgraph_input_mappings[node_id_str]
subgraph_input_idx = slot
if node_id_str in subgraph_slot_to_input_idx:
slot_mapping = subgraph_slot_to_input_idx[node_id_str]
if slot in slot_mapping:
subgraph_input_idx = slot_mapping[slot]
if subgraph_input_idx in input_map:
targets = input_map[subgraph_input_idx]
if targets:
internal_node, internal_slot = targets[0]
new_node_id = f'{node_id_str}:{internal_node}'
return resolve_subgraph_input(new_node_id, internal_slot, depth + 1)
return (node_id_str, slot)
def resolve_subgraph_input_all(node_id_str, slot, depth=0):
if depth > 100:
logger.warning(f'Max recursion depth reached resolving subgraph input for node {node_id_str}')
return [(node_id_str, slot)]
if node_id_str in subgraph_input_mappings:
input_map = subgraph_input_mappings[node_id_str]
subgraph_input_idx = slot
if node_id_str in subgraph_slot_to_input_idx:
slot_mapping = subgraph_slot_to_input_idx[node_id_str]
if slot in slot_mapping:
subgraph_input_idx = slot_mapping[slot]
if subgraph_input_idx in input_map:
targets = input_map[subgraph_input_idx]
results = []
for internal_node, internal_slot in targets:
new_node_id = f'{node_id_str}:{internal_node}'
results.extend(resolve_subgraph_input_all(new_node_id, internal_slot, depth + 1))
return results if results else [(node_id_str, slot)]
return [(node_id_str, slot)]
node_input_updates = {}
updated_links = []
for link in links:
if len(link) >= 6:
link_id = link[0]
source_id = link[1]
source_slot = link[2]
target_id = link[3]
target_slot = link[4]
link_type = link[5] if len(link) > 5 else None
source_id_str = str(source_id)
source_id, source_slot = resolve_subgraph_output(source_id_str, source_slot)
target_id_str = str(target_id)
all_targets = resolve_subgraph_input_all(target_id_str, target_slot)
for resolved_target_id, resolved_target_slot in all_targets:
if resolved_target_id != target_id_str:
if resolved_target_id not in node_input_updates:
node_input_updates[resolved_target_id] = {}
node_input_updates[resolved_target_id][resolved_target_slot] = link_id
resolved_target_id, resolved_target_slot = all_targets[0]
target_id = resolved_target_id
target_slot = resolved_target_slot
updated_links.append([link_id, source_id, source_slot, target_id, target_slot, link_type])
links = updated_links
for node in workflow_nodes:
node_id_str = str(node.get('id'))
if node_id_str in node_input_updates and 'inputs' in node:
slot_to_link = node_input_updates[node_id_str]
inputs = node.get('inputs', [])
for i, input_info in enumerate(inputs):
if i in slot_to_link:
input_info['link'] = slot_to_link[i]
link_map = {}
nodes_with_connected_outputs = set()
for link in links:
if len(link) >= 6:
link_id = link[0]
source_id = link[1]
source_slot = link[2]
target_id = link[3]
target_slot = link[4]
link_type = link[5] if len(link) > 5 else None
link_map[link_id] = {'source_id': source_id, 'source_slot': source_slot, 'target_id': target_id, 'target_slot': target_slot, 'type': link_type}
nodes_with_connected_outputs.add(source_id)
primitive_values = {}
nodes_to_exclude = set()
bypassed_nodes = set()
set_node_sources = {}
get_node_vars = {}
node_by_id = {}
for node in workflow_nodes:
node_by_id[str(node.get('id'))] = node
for node in workflow_nodes:
node_id = node.get('id')
node_type = _strip_nulls(node.get('type') or '')
node_mode = node.get('mode', 0)
if node_mode == 4:
bypassed_nodes.add(str(node_id))
logger.debug(f'Tracking bypassed node {node_id} ({node_type})')
if node_type == 'PrimitiveNode':
node_id_str = str(node_id)
widget_values = node.get('widgets_values')
if widget_values and len(widget_values) > 0:
primitive_values[node_id_str] = widget_values[0]
elif node_type == 'SetNode':
widget_values = node.get('widgets_values')
if widget_values and len(widget_values) > 0:
var_name = widget_values[0]
node_inputs = node.get('inputs', [])
if node_inputs:
for input_info in node_inputs:
input_link = input_info.get('link')
if input_link is not None and input_link in link_map:
link_data = link_map[input_link]
set_node_sources[var_name] = (link_data['source_id'], link_data['source_slot'])
logger.debug(f"SetNode '{var_name}' receives from node {link_data['source_id']} slot {link_data['source_slot']}")
break
elif node_type == 'GetNode':
widget_values = node.get('widgets_values')
if widget_values and len(widget_values) > 0:
var_name = widget_values[0]
get_node_vars[str(node_id)] = var_name
logger.debug(f"GetNode {node_id} provides variable '{var_name}'")
outputs = node.get('outputs', [])
has_connected_output = False
for output in outputs:
output_links = output.get('links', [])
if output_links and len(output_links) > 0:
has_connected_output = True
break
if node_type == 'LoadImageOutput':
nodes_to_exclude.add(str(node_id))
logger.debug(f'Marking node {node_id} ({node_type}) for exclusion - UI-only node type')
elif not outputs or not has_connected_output:
node_class = nodes.NODE_CLASS_MAPPINGS.get(node_type) if hasattr(nodes, 'NODE_CLASS_MAPPINGS') else None
is_output_node = node_class and hasattr(node_class, 'OUTPUT_NODE') and node_class.OUTPUT_NODE
has_connected_input = False
if not is_output_node and (not node_class):
node_inputs = node.get('inputs', [])
for input_info in node_inputs:
if input_info.get('link') is not None:
has_connected_input = True
break
if not is_output_node and (not has_connected_input):
nodes_to_exclude.add(str(node_id))
logger.debug(f'Marking node {node_id} ({node_type}) for exclusion - no connected outputs')
else:
logger.debug(f'Keeping node {node_id} ({node_type}) - OUTPUT_NODE=True or has connected inputs despite no connected outputs')
def trace_through_get_set_nodes(source_node_id, source_slot, visited=None):
if visited is None:
visited = set()
source_node_id_str = str(source_node_id)
if source_node_id_str in visited:
return (source_node_id, source_slot)
visited.add(source_node_id_str)
if source_node_id_str in get_node_vars:
var_name = get_node_vars[source_node_id_str]
if var_name in set_node_sources:
actual_source_id, actual_source_slot = set_node_sources[var_name]
logger.debug(f"Tracing GetNode {source_node_id} (var '{var_name}') -> actual source {actual_source_id} slot {actual_source_slot}")
return trace_through_get_set_nodes(actual_source_id, actual_source_slot, visited)
else:
logger.warning(f"GetNode {source_node_id} references variable '{var_name}' but no SetNode found for it")
return (source_node_id, source_slot)
def trace_through_bypassed(source_node_id, source_slot, visited=None):
if visited is None:
visited = set()
if source_node_id in visited:
return (source_node_id, source_slot)
visited.add(source_node_id)
if source_node_id not in bypassed_nodes:
return (source_node_id, source_slot)
for node in workflow_nodes:
if str(node.get('id')) == str(source_node_id):
node_type = node.get('type', 'unknown')
node_inputs = node.get('inputs', [])
node_outputs = node.get('outputs', [])
output_type = None
if node_outputs and source_slot < len(node_outputs):
output_type = node_outputs[source_slot].get('type')
if node_inputs:
linked_input = None
fallback_linked_input = None
for idx, input_info in enumerate(node_inputs):
input_link = input_info.get('link')
input_type = input_info.get('type')
if input_link is not None and input_link in link_map:
if fallback_linked_input is None:
fallback_linked_input = input_link
if output_type and input_type == output_type:
linked_input = input_link
break
if linked_input is None:
linked_input = fallback_linked_input
if linked_input is not None:
link_data = link_map[linked_input]
upstream_source_id = link_data['source_id']
upstream_source_slot = link_data['source_slot']
upstream_source_id, upstream_source_slot = trace_through_get_set_nodes(upstream_source_id, upstream_source_slot)
return trace_through_bypassed(upstream_source_id, upstream_source_slot, visited)
break
return (source_node_id, source_slot)
api_prompt = {}
for node in workflow_nodes:
node_id = str(node.get('id'))
node_type = _strip_nulls(node.get('type') or '')
if not node_type:
continue
node_mode = node.get('mode', 0)
if node_mode == 2:
logger.debug(f'Skipping muted node {node_id} ({node_type})')
continue
elif node_mode == 4:
logger.debug(f'Skipping bypassed/disabled node {node_id} ({node_type})')
continue
if node_type in ['Note', 'PrimitiveNode', 'GetNode', 'SetNode']:
logger.debug(f'Skipping {node_type} node {node_id}')
continue
if node_id in nodes_to_exclude:
logger.debug(f'Skipping {node_type} node {node_id} - no connected outputs')
continue
api_node = {'inputs': {}, 'class_type': node_type}
if 'title' in node:
api_node['_meta'] = {'title': node['title']}
elif hasattr(nodes, 'NODE_DISPLAY_NAME_MAPPINGS') and node_type in nodes.NODE_DISPLAY_NAME_MAPPINGS:
api_node['_meta'] = {'title': nodes.NODE_DISPLAY_NAME_MAPPINGS[node_type]}
else:
api_node['_meta'] = {'title': node_type}
link_inputs = {}
primitive_inputs = {}
node_inputs = node.get('inputs', [])
if node_inputs:
for input_info in node_inputs:
input_name = input_info.get('name')
input_link = input_info.get('link')
if input_link is not None and input_link in link_map:
link_data = link_map[input_link]
source_node_id = link_data['source_id']
source_slot = link_data['source_slot']
actual_source_id, actual_source_slot = trace_through_get_set_nodes(source_node_id, source_slot)
if str(actual_source_id) in bypassed_nodes:
actual_source_id, actual_source_slot = trace_through_bypassed(actual_source_id, actual_source_slot)
if str(actual_source_id) in bypassed_nodes:
logger.debug(f'Could not trace through bypassed node {source_node_id} for input {input_name}, falling back to widget value')
continue
actual_source_id, actual_source_slot = trace_through_get_set_nodes(actual_source_id, actual_source_slot)
actual_source_id, actual_source_slot = resolve_subgraph_output(str(actual_source_id), actual_source_slot)
source_node_id_str = str(actual_source_id)
if source_node_id_str in primitive_values:
primitive_inputs[input_name] = primitive_values[source_node_id_str]
elif actual_source_id in nodes_to_exclude:
logger.debug(f'Skipping input {input_name} from excluded node {source_node_id_str}')
elif actual_source_id in bypassed_nodes:
logger.warning(f'Could not resolve bypassed node {source_node_id} for input {input_name}, skipping connection')
else:
if actual_source_id != source_node_id:
logger.info(f'Bypassed disabled node {source_node_id}, connecting {input_name} to {actual_source_id} (slot {actual_source_slot}) instead')
link_inputs[input_name] = [str(actual_source_id), actual_source_slot]
logger.info('WorkflowConverter: processing node id=%s type=%s', node_id, node_type)
ordered_inputs = WorkflowConverter._get_ordered_inputs(node_type, node)
widget_values = node.get('widgets_values')
widget_inputs = {}
if widget_values is not None:
if isinstance(widget_values, dict):
for key, value in widget_values.items():
if key in ['videopreview', 'preview']:
continue
if key not in link_inputs:
widget_inputs[key] = value
elif isinstance(widget_values, list):
has_dict_widgets = any((isinstance(v, dict) for v in widget_values))
if has_dict_widgets:
WorkflowConverter._process_dict_widget_values(widget_values, widget_inputs, link_inputs)
else:
widget_mappings = WorkflowConverter._get_widget_mappings(node_type, node)
filtered_values = WorkflowConverter._filter_control_values(widget_values)
if widget_mappings:
for i, value in enumerate(filtered_values):
if i < len(widget_mappings):
widget_name = widget_mappings[i]
if widget_name and widget_name not in link_inputs:
widget_inputs[widget_name] = value
elif filtered_values:
logger.warning(f"Could not map widget values for unknown node type '{node_type}' (node {node_id})")
default_inputs = WorkflowConverter._get_default_inputs(node_type, widget_inputs, primitive_inputs, link_inputs)
if ordered_inputs:
for input_name in ordered_inputs:
if input_name in widget_inputs:
api_node['inputs'][input_name] = widget_inputs[input_name]
elif input_name in primitive_inputs:
api_node['inputs'][input_name] = primitive_inputs[input_name]
elif input_name in default_inputs:
api_node['inputs'][input_name] = default_inputs[input_name]
for input_name in ordered_inputs:
if input_name in link_inputs and input_name not in api_node['inputs']:
api_node['inputs'][input_name] = link_inputs[input_name]
for key, value in widget_inputs.items():
if key not in api_node['inputs']:
api_node['inputs'][key] = value
for key, value in primitive_inputs.items():
if key not in api_node['inputs']:
api_node['inputs'][key] = value
for key, value in default_inputs.items():
if key not in api_node['inputs']:
api_node['inputs'][key] = value
for key, value in link_inputs.items():
if key not in api_node['inputs']:
api_node['inputs'][key] = value
else:
for key, value in widget_inputs.items():
api_node['inputs'][key] = value
for key, value in primitive_inputs.items():
if key not in api_node['inputs']:
api_node['inputs'][key] = value
for key, value in default_inputs.items():
if key not in api_node['inputs']:
api_node['inputs'][key] = value
for key, value in link_inputs.items():
if key not in api_node['inputs']:
api_node['inputs'][key] = value
api_prompt[node_id] = api_node
return api_prompt
@staticmethod
def _process_dict_widget_values(widget_values: List[Any], widget_inputs: Dict[str, Any], link_inputs: Dict[str, Any]) -> None:
lora_counter = 0
for value in widget_values:
if isinstance(value, dict):
if not value:
continue
elif 'type' in value:
widget_name = value.get('type')
if widget_name and widget_name not in link_inputs:
widget_inputs[widget_name] = value
elif 'lora' in value:
lora_counter += 1
widget_name = f'lora_{lora_counter}'
if widget_name not in link_inputs:
clean_value = {k: v for k, v in value.items() if k != 'strengthTwo' or v is not None}
widget_inputs[widget_name] = clean_value
else:
logger.debug(f'Unknown dict widget value structure: {value}')
elif isinstance(value, str):
if value == '':
widget_inputs['➕ Add Lora'] = value
@staticmethod
def _filter_control_values(widget_values: List[Any]) -> List[Any]:
filtered = []
skip_next = False
for i, value in enumerate(widget_values):
if skip_next:
skip_next = False
continue
if value in ['fixed', 'increment', 'decrement', 'randomize']:
continue
if i + 1 < len(widget_values):
next_val = widget_values[i + 1]
if next_val in ['fixed', 'increment', 'decrement', 'randomize']:
filtered.append(value)
continue
filtered.append(value)
return filtered
@staticmethod
def _get_ordered_inputs(node_type: str, node: Dict[str, Any]) -> List[str]:
node_type = _strip_nulls(node_type)
properties = node.get('properties', {})
if 'Node name for S&R' in properties:
node_type = _strip_nulls(properties['Node name for S&R'])
node_info = get_node_info_for_type(node_type)
if node_info and 'input_order' in node_info:
input_order = node_info['input_order']
input_names = []
for section in ['required', 'optional']:
if section in input_order:
input_names.extend(input_order[section])
if input_names:
return input_names
if hasattr(nodes, 'NODE_CLASS_MAPPINGS') and node_type in nodes.NODE_CLASS_MAPPINGS:
try:
node_class = nodes.NODE_CLASS_MAPPINGS[node_type]
input_types = node_class.INPUT_TYPES()
input_names = []
for section in ['required', 'optional']:
if section in input_types:
for input_name in input_types[section].keys():
input_names.append(input_name)
if input_names:
return input_names
except ValueError as e:
if 'null bytes' in str(e).lower():
logger.warning("WorkflowConverter: node type '%s' raised null bytes in INPUT_TYPES(); using fallback order", node_type)
return []
except Exception as e:
logger.debug('Could not get input order from node class for %s: %s', node_type, e)
return []
@staticmethod
def _get_dynamic_combo_sub_inputs(input_name: str, input_spec: Any, widget_values: List[Any], current_idx: int) -> List[str]:
if not isinstance(input_spec, (list, tuple)) or len(input_spec) < 2:
return []
input_type = input_spec[0]
if not isinstance(input_type, str) or not input_type.startswith('COMFY_') or 'COMBO' not in input_type:
return []
spec_options = input_spec[1] if len(input_spec) > 1 else {}
if not isinstance(spec_options, dict):
return []
options = spec_options.get('options', [])
if not options or current_idx >= len(widget_values):
return []
selected_value = widget_values[current_idx]
for option in options:
if isinstance(option, dict) and option.get('key') == selected_value:
sub_inputs = option.get('inputs', {})
sub_input_names = []
for section in ['required', 'optional']:
if section in sub_inputs:
for sub_name in sub_inputs[section].keys():
sub_input_names.append(f'{input_name}.{sub_name}')
return sub_input_names
return []
@staticmethod
def _get_widget_mappings(node_type: str, node: Dict[str, Any]) -> List[Optional[str]]:
node_type = _strip_nulls(node_type)
properties = node.get('properties', {})
if 'Node name for S&R' in properties:
node_type = _strip_nulls(properties['Node name for S&R'])
widget_values = node.get('widgets_values', [])
if isinstance(widget_values, dict):
widget_values = []
node_info = get_node_info_for_type(node_type)
if node_info and 'input' in node_info:
try:
input_def = node_info['input']
widget_names = []
widget_idx = 0
for section in ['required', 'optional']:
if section in input_def:
for input_name, input_spec in input_def[section].items():
if isinstance(input_spec, (list, tuple)) and len(input_spec) >= 1:
input_type = input_spec[0]
is_widget = False
is_dynamic_combo = False
if isinstance(input_type, list):
is_widget = True
elif input_type in ['INT', 'FLOAT', 'STRING', 'BOOLEAN', 'COMBO']:
is_widget = True
elif isinstance(input_type, str) and input_type.startswith('COMFY_') and ('COMBO' in input_type):
is_widget = True
is_dynamic_combo = True
elif isinstance(input_type, str) and (not input_type.isupper()):
is_widget = True
if is_widget:
widget_names.append(input_name)
if is_dynamic_combo and widget_values:
sub_inputs = WorkflowConverter._get_dynamic_combo_sub_inputs(input_name, input_spec, widget_values, widget_idx)
widget_names.extend(sub_inputs)
widget_idx += 1 + len(sub_inputs)
else:
widget_idx += 1
if widget_names:
return widget_names
except Exception as e:
logger.debug(f'Could not get widget mappings from node info for {node_type}: {e}')
if hasattr(nodes, 'NODE_CLASS_MAPPINGS') and node_type in nodes.NODE_CLASS_MAPPINGS:
try:
node_class = nodes.NODE_CLASS_MAPPINGS[node_type]
input_types = node_class.INPUT_TYPES()
widget_names = []
widget_idx = 0
for section in ['required', 'optional']:
if section in input_types:
for input_name, input_spec in input_types[section].items():
if isinstance(input_spec, tuple) and len(input_spec) >= 1:
input_type = input_spec[0]
is_widget = False
is_dynamic_combo = False
if isinstance(input_type, list):
is_widget = True
elif input_type in ['INT', 'FLOAT', 'STRING', 'BOOLEAN', 'COMBO']:
is_widget = True
elif isinstance(input_type, str) and input_type.startswith('COMFY_') and ('COMBO' in input_type):
is_widget = True
is_dynamic_combo = True
elif isinstance(input_type, str) and (not input_type.isupper()):
is_widget = True
if is_widget:
widget_names.append(input_name)
if is_dynamic_combo and widget_values:
sub_inputs = WorkflowConverter._get_dynamic_combo_sub_inputs(input_name, input_spec, widget_values, widget_idx)
widget_names.extend(sub_inputs)
widget_idx += 1 + len(sub_inputs)
else:
widget_idx += 1
if widget_names:
return widget_names
except ValueError as e:
if 'null bytes' in str(e).lower():
logger.warning("WorkflowConverter: node type '%s' raised null bytes in INPUT_TYPES(); using fallback widget order", node_type)
return []
except Exception as e:
logger.debug('Could not get widget mappings from node class for %s: %s', node_type, e)
widget_values = node.get('widgets_values', [])
if not isinstance(widget_values, list) or len(widget_values) == 0:
return []
properties = node.get('properties', {})
ue_properties = properties.get('ue_properties', {})
widget_ue_connectable = ue_properties.get('widget_ue_connectable', {})
if widget_ue_connectable and isinstance(widget_ue_connectable, dict):
widget_names = list(widget_ue_connectable.keys())
if widget_names and len(widget_names) >= len(widget_values):
return widget_names[:len(widget_values)]
all_inputs = []
connected_inputs = set()
widget_flagged_inputs = []
for input_info in node.get('inputs', []):
input_name = input_info.get('name')
if input_name:
all_inputs.append(input_name)
if input_info.get('link') is not None:
connected_inputs.add(input_name)
if input_info.get('widget'):
widget_flagged_inputs.append(input_name)
if widget_flagged_inputs:
if len(widget_values) > len(widget_flagged_inputs):
potential_widgets = [inp for inp in all_inputs if inp not in connected_inputs and inp not in widget_flagged_inputs]
return widget_flagged_inputs + potential_widgets[:len(widget_values) - len(widget_flagged_inputs)]
return widget_flagged_inputs
unconnected = [inp for inp in all_inputs if inp not in connected_inputs]
if unconnected and len(unconnected) >= len(widget_values):
return unconnected[:len(widget_values)]
return []
@staticmethod
def _get_default_inputs(node_type: str, widget_inputs: Dict[str, Any], primitive_inputs: Dict[str, Any], link_inputs: Dict[str, Any]) -> Dict[str, Any]:
default_inputs = {}
node_type = _strip_nulls(node_type)
if not hasattr(nodes, 'NODE_CLASS_MAPPINGS') or node_type not in nodes.NODE_CLASS_MAPPINGS:
return default_inputs
try:
node_class = nodes.NODE_CLASS_MAPPINGS[node_type]
input_types = node_class.INPUT_TYPES()
for section in ['required', 'optional']:
if section not in input_types:
continue
for input_name, input_spec in input_types[section].items():
if input_name in widget_inputs or input_name in primitive_inputs or input_name in link_inputs:
continue
if isinstance(input_spec, (list, tuple)) and len(input_spec) >= 1:
input_type = input_spec[0]
spec_options = input_spec[1] if len(input_spec) >= 2 else {}
if isinstance(spec_options, dict) and 'default' in spec_options:
default_inputs[input_name] = spec_options['default']
logger.debug(f"Adding default value for {node_type}.{input_name}: {spec_options['default']}")
elif isinstance(input_type, list) and len(input_type) > 0:
default_inputs[input_name] = input_type[0]
logger.debug(f'Adding implicit combo default for {node_type}.{input_name}: {input_type[0]}')
elif input_type == 'COMBO' and isinstance(spec_options, dict) and ('options' in spec_options):
options = spec_options['options']
if isinstance(options, list) and len(options) > 0:
default_inputs[input_name] = options[0]
logger.debug(f'Adding implicit COMBO default for {node_type}.{input_name}: {options[0]}')
except ValueError as e:
if 'null bytes' in str(e).lower():
logger.warning("WorkflowConverter: node type '%s' raised null bytes in _get_default_inputs; skipping defaults", node_type)
except Exception as e:
logger.debug('Could not get default inputs for %s: %s', node_type, e)
return default_inputs
+248
View File
@@ -0,0 +1,248 @@
from __future__ import annotations
import json
import logging
import os
from pathlib import Path
from typing import Any
logger = logging.getLogger(__name__)
_PLUGIN_DIR = Path(__file__).resolve().parent
_COMFYUI_ROOT = _PLUGIN_DIR.parent.parent
def _workflow_ui_link_widget_order() -> list[str]:
order = ["form_label"]
for i in range(8):
order.append(f"type_{i}")
order.append(f"name_{i}")
for i in range(8):
order.append(f"input_text_{i}")
for i in range(8):
order.append(f"input_number_{i}")
for i in range(8):
order.append(f"input_image_{i}")
for i in range(8):
order.append(f"input_video_{i}")
for i in range(8):
order.append(f"input_audio_{i}")
for i in range(8):
order.append(f"input_boolean_{i}")
return order
_WIDGET_ORDER: dict[str, list[str]] = {
"WorkflowUILink": _workflow_ui_link_widget_order(),
"WorkflowUI Link": _workflow_ui_link_widget_order(),
"CLIPTextEncode": ["text"],
"Text Multiline": ["text"],
"PrimitiveStringMultiline": ["value"],
"EmptyLatentImage": ["width", "height", "batch_size"],
"EmptySD3LatentImage": ["width", "height", "batch_size"],
"SDXLEmptyLatentSizePicker+": ["resolution", "batch_size", "width_override", "height_override"],
"KSampler": ["seed", "steps", "cfg", "sampler_name", "scheduler", "denoise"],
"BasicScheduler": ["scheduler", "steps", "denoise"],
"UNETLoader": ["unet_name", "weight_dtype"],
"DiffusionModelLoader": ["unet_name", "weight_dtype"],
"LoadDiffusionModel": ["unet_name", "weight_dtype"],
"StableCascadeCheckpointLoader": ["key_opt_b", "key_opt_c", "cache_mode"],
"StableCascade_CheckpointLoader": ["key_opt_b", "key_opt_c", "cache_mode"],
"SD3CheckpointLoader": ["ckpt_name", "shift"],
"SD3LoadCheckpoint": ["ckpt_name", "shift"],
"SaveImage": ["filename_prefix"],
"Save Image": ["filename_prefix"],
"Save Image (api)": ["filename_prefix"],
"SaveImageNode": ["filename_prefix"],
}
def get_workflows_directory() -> Path | None:
env_path = os.environ.get("WORKFLOWUI_WORKFLOWS_DIR", "").strip()
if env_path:
p = Path(env_path).resolve()
if p.is_dir():
return p
return None
default = _COMFYUI_ROOT / "user" / "default" / "workflows"
if default.is_dir():
return default
return None
def list_workflows() -> list[dict[str, str]]:
wf_dir = get_workflows_directory()
if not wf_dir:
return []
result: list[dict[str, str]] = []
try:
for path in sorted(wf_dir.glob("*.json")):
if path.is_file():
stem = path.stem
label = stem.replace("_", " ").strip() or stem
result.append({"id": stem, "label": label})
except OSError as e:
logger.warning("WorkflowUIPlugin: list_workflows failed: %s", e)
return result
def _is_api_format(data: dict[str, Any]) -> bool:
if not data or not isinstance(data, dict):
return False
for v in data.values():
if isinstance(v, dict) and ("class_type" in v or "inputs" in v):
return True
return False
def _extract_graph(workflow: dict[str, Any]) -> dict[str, Any]:
if not isinstance(workflow, dict):
return workflow
if workflow.get("nodes") is not None:
return workflow
if _is_api_format(workflow):
return workflow
for key in ("graph", "workflow", "prompt"):
inner = workflow.get(key)
if isinstance(inner, dict):
if inner.get("nodes") is not None or _is_api_format(inner):
return inner
deeper = inner.get("graph") or inner.get("workflow")
if isinstance(deeper, dict):
return deeper
return workflow
def _build_link_map(links: list[Any]) -> dict[int | str, tuple[str, int]]:
result: dict[int | str, tuple[str, int]] = {}
if not isinstance(links, list):
return result
for link in links:
if not isinstance(link, (list, tuple)) or len(link) < 6:
continue
link_id, origin_id, origin_slot = link[0], link[1], link[2]
try:
origin_slot_int = int(origin_slot)
except (TypeError, ValueError):
continue
result[link_id] = (str(origin_id), origin_slot_int)
return result
def _inputs_from_ui_node(
node: dict[str, Any],
class_type: str,
link_map: dict[int | str, tuple[str, int]],
) -> dict[str, Any]:
inputs: dict[str, Any] = {}
node_inputs = node.get("inputs")
widgets_values = (
node.get("widgets_values")
or node.get("widgetsValues")
or node.get("widget_values")
)
widget_order = _WIDGET_ORDER.get(class_type)
if isinstance(widgets_values, dict):
inputs = dict(widgets_values)
elif isinstance(widgets_values, list) and widget_order:
for idx, field in enumerate(widget_order):
if idx < len(widgets_values):
inputs[field] = widgets_values[idx]
if isinstance(node_inputs, list):
widget_only_names: list[str] = []
for inp in node_inputs:
if not isinstance(inp, dict):
continue
name = inp.get("name")
if not name or not isinstance(name, str):
continue
link = inp.get("link")
if link is not None:
conn = link_map.get(link)
if conn is not None:
inputs[name] = [conn[0], conn[1]]
else:
widget_only_names.append(name)
for idx, name in enumerate(widget_only_names):
if name not in inputs and isinstance(widgets_values, list) and idx < len(widgets_values):
inputs[name] = widgets_values[idx]
elif isinstance(node_inputs, dict) and not inputs:
inputs = dict(node_inputs)
return inputs
def _ensure_saveimage_defaults(api_graph: dict[str, Any]) -> None:
save_types = ("SaveImage", "Save Image", "Save Image (api)", "SaveImageNode")
for node in api_graph.values():
if not isinstance(node, dict) or node.get("class_type") not in save_types:
continue
inputs = node.get("inputs")
if isinstance(inputs, dict) and not inputs.get("filename_prefix"):
inputs["filename_prefix"] = "ComfyUI"
_NODE_MODE_DISABLED = 4
def ui_to_api_format(workflow: dict[str, Any]) -> dict[str, Any]:
workflow = _extract_graph(workflow)
nodes = workflow.get("nodes")
if not isinstance(nodes, list):
if _is_api_format(workflow):
return workflow
return {}
links = workflow.get("links")
link_map = _build_link_map(links) if isinstance(links, list) else {}
out: dict[str, Any] = {}
for node in nodes:
if not isinstance(node, dict):
continue
if node.get("mode") == _NODE_MODE_DISABLED:
continue
nid = node.get("id")
if nid is None:
continue
node_id = str(nid)
class_type = node.get("type") or node.get("class_type")
if not class_type:
continue
inputs = _inputs_from_ui_node(node, class_type, link_map)
meta = node.get("_meta")
if not meta and isinstance(node.get("properties"), dict):
s_r_name = (node.get("properties") or {}).get("Node name for S&R")
if isinstance(s_r_name, str) and s_r_name.strip():
meta = {"title": s_r_name.strip()}
normalized: dict[str, Any] = {"class_type": class_type, "inputs": inputs}
if meta:
normalized["_meta"] = meta
out[node_id] = normalized
_ensure_saveimage_defaults(out)
return out
def load_workflow(workflow_id: str) -> tuple[str, dict[str, Any]]:
wf_dir = get_workflows_directory()
if not wf_dir:
raise FileNotFoundError("Workflow directory not configured")
safe_id = (workflow_id or "").strip()
if not safe_id or "/" in safe_id or "\\" in safe_id or ".." in safe_id:
raise ValueError("Invalid workflow id")
path = (wf_dir / safe_id).with_suffix(".json")
if not path.is_file():
raise FileNotFoundError(f"Workflow not found: {safe_id}")
try:
raw = path.read_text(encoding="utf-8", errors="replace")
data = json.loads(raw)
except json.JSONDecodeError as e:
raise ValueError(f"Invalid workflow JSON: {e}") from e
if not isinstance(data, dict):
raise ValueError("Workflow must be a JSON object")
name = data.get("name")
if not isinstance(name, str) and isinstance(data.get("workflow"), dict):
name = data["workflow"].get("name")
if not isinstance(name, str) or not name.strip():
name = safe_id.replace("_", " ").strip() or safe_id
graph = _extract_graph(data)
if not graph or (isinstance(graph, dict) and not graph.get("nodes") and not _is_api_format(graph)):
raise ValueError("Workflow has no executable nodes")
return name, graph
+189
View File
@@ -0,0 +1,189 @@
from __future__ import annotations
import logging
import os
from typing import Any
logger = logging.getLogger(__name__)
try:
import torch
import numpy as np
from PIL import Image, ImageOps, ImageSequence
_HAS_IMAGE_DEPS = True
except ImportError:
_HAS_IMAGE_DEPS = False
MAX_INPUT_SLOTS = 8
TYPE_OPTIONS = ["", "text", "number", "seed", "image", "video", "audio", "boolean", "select"]
def _input_field_for_type(slot: int, typ: str) -> str:
if typ == "image":
return f"input_image_{slot}"
if typ == "video":
return f"input_video_{slot}"
if typ == "audio":
return f"input_audio_{slot}"
if typ == "boolean":
return f"input_boolean_{slot}"
if typ in ("number", "seed"):
return f"input_number_{slot}"
return f"input_text_{slot}" # text, select, or unknown
def _get_media_files(content_types: list[str]) -> list[str]:
try:
import folder_paths
input_dir = folder_paths.get_input_directory()
if not os.path.isdir(input_dir):
return []
files = [f for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f))]
return sorted(folder_paths.filter_files_content_types(files, content_types))
except Exception as e:
logger.warning("WorkflowUILink: could not list media files: %s", e)
return []
def _build_input_types() -> dict:
try:
import folder_paths
input_dir = folder_paths.get_input_directory()
all_files = [f for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f))]
image_files = sorted(folder_paths.filter_files_content_types(all_files, ["image"]))
video_files = sorted(folder_paths.filter_files_content_types(all_files, ["video"]))
audio_files = sorted(folder_paths.filter_files_content_types(all_files, ["audio"]))
except Exception:
image_files = video_files = audio_files = []
required = {
"form_label": ("STRING", {
"default": "Generation parameters",
"placeholder": "Header shown above inputs in WorkflowUI app (e.g. 'Generation parameters'). This text appears as the section title in the app.",
}),
}
for i in range(MAX_INPUT_SLOTS):
required[f"type_{i}"] = (TYPE_OPTIONS, {"default": ""})
required[f"name_{i}"] = ("STRING", {
"default": "",
"placeholder": f"Name and label in app (e.g. 'Positive prompt'; use 'seed' for master seed)",
})
image_opts = [""] + (image_files or [])
video_opts = [""] + (video_files or [])
audio_opts = [""] + (audio_files or [])
optional = {}
for i in range(MAX_INPUT_SLOTS):
optional[f"input_text_{i}"] = ("STRING", {"default": "", "multiline": True})
optional[f"input_number_{i}"] = ("INT", {"default": 0, "min": 0, "max": 2**31 - 1})
optional[f"input_image_{i}"] = (image_opts, {"image_upload": True})
optional[f"input_video_{i}"] = (video_opts,)
optional[f"input_audio_{i}"] = (audio_opts,)
optional[f"input_boolean_{i}"] = ("BOOLEAN", {"default": False})
return {"required": required, "optional": optional}
def _build_return_types() -> tuple[str, ...]:
return (
("STRING",) * MAX_INPUT_SLOTS
+ ("INT",) * MAX_INPUT_SLOTS
+ ("IMAGE",) * MAX_INPUT_SLOTS
+ ("STRING",) * MAX_INPUT_SLOTS # video filenames
+ ("STRING",) * MAX_INPUT_SLOTS # audio filenames
+ ("BOOLEAN",) * MAX_INPUT_SLOTS # boolean values
)
def _build_return_names() -> tuple[str, ...]:
names = []
for i in range(MAX_INPUT_SLOTS):
names.append(f"text_{i}")
for i in range(MAX_INPUT_SLOTS):
names.append(f"number_{i}")
for i in range(MAX_INPUT_SLOTS):
names.append(f"image_{i}")
for i in range(MAX_INPUT_SLOTS):
names.append(f"video_{i}")
for i in range(MAX_INPUT_SLOTS):
names.append(f"audio_{i}")
for i in range(MAX_INPUT_SLOTS):
names.append(f"boolean_{i}")
return tuple(names)
def _load_image_from_path(image: str) -> Any:
if not _HAS_IMAGE_DEPS or not image or not isinstance(image, str):
return None
try:
import folder_paths
import node_helpers
image_path = folder_paths.get_annotated_filepath(image)
img = node_helpers.pillow(Image.open, image_path)
output_images = []
for i in ImageSequence.Iterator(img):
i = node_helpers.pillow(ImageOps.exif_transpose, i)
if i.mode in ("RGBA", "P"):
i = i.convert("RGB")
arr = np.array(i).astype(np.float32) / 255.0
output_images.append(torch.from_numpy(arr)[None,])
if not output_images:
return None
return torch.cat(output_images, dim=0) if len(output_images) > 1 else output_images[0]
except Exception as e:
logger.warning("WorkflowUILink: failed to load image %s: %s", image, e)
return None
def _empty_image_placeholder() -> Any:
if not _HAS_IMAGE_DEPS:
return None
return torch.zeros(1, 64, 64, 3)
class WorkflowUILink:
@classmethod
def INPUT_TYPES(cls) -> dict:
return _build_input_types()
RETURN_TYPES = _build_return_types()
RETURN_NAMES = _build_return_names()
FUNCTION = "run"
CATEGORY = "WorkflowUI"
def run(self, form_label: str, **kwargs: Any) -> tuple:
_empty_img = _empty_image_placeholder() if _HAS_IMAGE_DEPS else None
results: list[Any] = []
for i in range(MAX_INPUT_SLOTS):
results.append(kwargs.get(f"input_text_{i}", ""))
for i in range(MAX_INPUT_SLOTS):
results.append(kwargs.get(f"input_number_{i}", 0))
for i in range(MAX_INPUT_SLOTS):
val = kwargs.get(f"input_image_{i}")
if isinstance(val, str) and val.strip():
loaded = _load_image_from_path(val)
results.append(loaded if loaded is not None else _empty_img)
else:
results.append(_empty_img if val is None else val)
for i in range(MAX_INPUT_SLOTS):
val = kwargs.get(f"input_video_{i}")
results.append(val if isinstance(val, str) else "")
for i in range(MAX_INPUT_SLOTS):
val = kwargs.get(f"input_audio_{i}")
results.append(val if isinstance(val, str) else "")
for i in range(MAX_INPUT_SLOTS):
val = kwargs.get(f"input_boolean_{i}", False)
results.append(bool(val))
return tuple(results)
NODE_CLASS_MAPPINGS = {
"WorkflowUILink": WorkflowUILink,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"WorkflowUILink": "WorkflowUI Link",
}