egutierrez
f12272d002
- docs/capabilities/INDEX.md - docs/capabilities/comfyui.md - python/functions/browser/comfyui_export_workflow_ui.md - python/functions/browser/comfyui_export_workflow_ui.py - python/functions/browser/comfyui_load_workflow_ui.md - python/functions/browser/comfyui_load_workflow_ui.py - python/functions/browser/comfyui_queue_prompt_ui.md - python/functions/browser/comfyui_queue_prompt_ui.py - python/functions/browser/comfyui_refresh_nodes_ui.md - python/functions/browser/comfyui_refresh_nodes_ui.py - ... Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
111 lines
3.7 KiB
Python
111 lines
3.7 KiB
Python
"""Valida un workflow ComfyUI contra el catalogo /object_info del servidor.
|
|
|
|
Cruza los class_type del workflow contra los nodos disponibles y los nombres de
|
|
modelos (checkpoints, loras, vae, controlnet, upscale) contra los combos
|
|
enumerados de cada nodo. Asi se detectan nodos o modelos faltantes ANTES de
|
|
encolar (POST /prompt), evitando un HTTP 400 del servidor.
|
|
|
|
Compone comfyui_object_info (no reimplementa la consulta HTTP).
|
|
|
|
Impura: red (HTTP GET via comfyui_object_info). Solo stdlib.
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
if _THIS_DIR not in sys.path:
|
|
sys.path.insert(0, _THIS_DIR)
|
|
|
|
from comfyui_object_info import comfyui_object_info # noqa: E402
|
|
|
|
# inputs cuyo valor es el nombre (string) de un asset/modelo en disco
|
|
_MODEL_INPUTS = {
|
|
"ckpt_name",
|
|
"lora_name",
|
|
"vae_name",
|
|
"control_net_name",
|
|
"model_name",
|
|
"unet_name",
|
|
"clip_name",
|
|
"style_model_name",
|
|
"gligen_name",
|
|
}
|
|
|
|
|
|
def comfyui_validate_workflow(
|
|
workflow: dict,
|
|
server: str = "127.0.0.1:8188",
|
|
timeout: float = 30.0,
|
|
) -> dict:
|
|
"""Valida un workflow API format contra un servidor ComfyUI vivo.
|
|
|
|
Args:
|
|
workflow: dict en API format ({node_id: {class_type, inputs}}).
|
|
server: host:port del servidor ComfyUI (sin esquema).
|
|
timeout: timeout de la consulta HTTP en segundos.
|
|
|
|
Returns:
|
|
dict {ok, valid, missing_nodes, missing_models, error}:
|
|
- ok: la validacion se pudo ejecutar (el servidor respondio).
|
|
- valid: el workflow no tiene nodos ni modelos faltantes.
|
|
- missing_nodes: lista de class_type ausentes en el servidor.
|
|
- missing_models: lista de {node, input, value} con valores de modelo
|
|
no presentes en el combo enumerado correspondiente.
|
|
- error: mensaje si no se pudo consultar el servidor (ok=False).
|
|
"""
|
|
try:
|
|
obj_info = comfyui_object_info(server=server, timeout=timeout)
|
|
except RuntimeError as exc:
|
|
return {
|
|
"ok": False,
|
|
"valid": False,
|
|
"missing_nodes": [],
|
|
"missing_models": [],
|
|
"error": str(exc),
|
|
}
|
|
|
|
missing_nodes = []
|
|
missing_models = []
|
|
for nid, node in workflow.items():
|
|
if not isinstance(node, dict):
|
|
continue
|
|
ctype = node.get("class_type")
|
|
if ctype not in obj_info:
|
|
missing_nodes.append(ctype)
|
|
continue
|
|
spec = obj_info[ctype].get("input", {})
|
|
allowed = {}
|
|
for section in ("required", "optional"):
|
|
for name, decl in (spec.get(section) or {}).items():
|
|
if isinstance(decl, list) and decl and isinstance(decl[0], list):
|
|
allowed[name] = set(decl[0])
|
|
for in_name, val in node.get("inputs", {}).items():
|
|
if in_name in _MODEL_INPUTS and isinstance(val, str):
|
|
opts = allowed.get(in_name)
|
|
if opts is not None and val not in opts:
|
|
missing_models.append(
|
|
{"node": nid, "input": in_name, "value": val}
|
|
)
|
|
|
|
# dedup missing_nodes preservando orden
|
|
seen = set()
|
|
missing_nodes = [c for c in missing_nodes if not (c in seen or seen.add(c))]
|
|
valid = not missing_nodes and not missing_models
|
|
return {
|
|
"ok": True,
|
|
"valid": valid,
|
|
"missing_nodes": missing_nodes,
|
|
"missing_models": missing_models,
|
|
"error": "",
|
|
}
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import json
|
|
|
|
sys.path.insert(0, _THIS_DIR)
|
|
from comfyui_build_txt2img_workflow import comfyui_build_txt2img_workflow
|
|
|
|
wf = comfyui_build_txt2img_workflow("dreamshaper_8.safetensors", "a cat")
|
|
print(json.dumps(comfyui_validate_workflow(wf), indent=2))
|