98 lines
4.7 KiB
Python
98 lines
4.7 KiB
Python
"""Offline checks for the saved example graphs (no ComfyUI dependency)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
def workflow_problems(data: dict) -> list[str]:
|
|
problems = []
|
|
nodes = data.get("nodes", [])
|
|
if not isinstance(nodes, list) or not nodes:
|
|
return ["expected a nonempty UI workflow nodes list"]
|
|
by_id = {n["id"]: n for n in nodes}
|
|
if len(by_id) != len(nodes):
|
|
problems.append("duplicate node ids")
|
|
links = data.get("links", [])
|
|
by_link = {link[0]: link for link in links if isinstance(link, list) and len(link) == 6}
|
|
if len(by_link) != len(links):
|
|
problems.append("duplicate or malformed links (expected six fields)")
|
|
for link in by_link.values():
|
|
ident, source_id, source_slot, target_id, target_slot, _kind = link
|
|
source, target = by_id.get(source_id), by_id.get(target_id)
|
|
if source is None or target is None:
|
|
problems.append(f"link {ident} refers to a missing node")
|
|
continue
|
|
outputs, inputs = source.get("outputs", []), target.get("inputs", [])
|
|
if not isinstance(source_slot, int) or not 0 <= source_slot < len(outputs):
|
|
problems.append(f"link {ident} has an invalid source slot")
|
|
elif ident not in (outputs[source_slot].get("links") or []):
|
|
problems.append(f"link {ident} is missing from its source output")
|
|
if not isinstance(target_slot, int) or not 0 <= target_slot < len(inputs):
|
|
problems.append(f"link {ident} has an invalid target slot")
|
|
elif inputs[target_slot].get("link") != ident:
|
|
problems.append(f"link {ident} is missing from its target input")
|
|
for n in nodes:
|
|
for slot, inp in enumerate(n.get("inputs", [])):
|
|
ident = inp.get("link")
|
|
if ident is not None and (ident not in by_link or by_link[ident][3:5] != [n["id"], slot]):
|
|
problems.append(f"node {n['id']} input {slot} has a stale link")
|
|
for slot, output in enumerate(n.get("outputs", [])):
|
|
for ident in output.get("links") or []:
|
|
if ident not in by_link or by_link[ident][1:3] != [n["id"], slot]:
|
|
problems.append(f"node {n['id']} output {slot} has a stale link")
|
|
if data.get("last_node_id", -1) < max(by_id):
|
|
problems.append("last_node_id is behind the graph")
|
|
if by_link and data.get("last_link_id", -1) < max(by_link):
|
|
problems.append("last_link_id is behind the graph")
|
|
|
|
groups = data.get("groups", [])
|
|
if not groups:
|
|
problems.append("no stage groups")
|
|
notes = [n for n in nodes if n["type"] == "AUSBOSS_NODES_WorkflowNote"]
|
|
if not notes:
|
|
problems.append("no Workflow Note with setup instructions")
|
|
for n in notes:
|
|
try:
|
|
note = json.loads(n["widgets_values"][0])
|
|
if not note.get("title") or not note.get("body"):
|
|
problems.append(f"note {n['id']} needs a title and instructions")
|
|
if "\\n" in note.get("body", ""):
|
|
problems.append(f"note {n['id']} contains escaped line breaks instead of Markdown lines")
|
|
except (ValueError, KeyError, IndexError, TypeError, AttributeError):
|
|
problems.append(f"note {n['id']} is not valid note JSON")
|
|
|
|
rectangles = []
|
|
for n in nodes:
|
|
size = n.get("size", [0, 0])
|
|
if isinstance(size, dict):
|
|
size = [size.get("0", 0), size.get("1", 0)]
|
|
x, y = n.get("pos", [0, 0])
|
|
width, height = size
|
|
if width <= 0 or height <= 0:
|
|
problems.append(f"node {n['id']} has no usable saved size")
|
|
rectangles.append((n["id"], x, y - 30, x + width, y + height))
|
|
if n not in notes and not any(
|
|
gx <= x and gy <= y - 30 and x + width <= gx + gw + 1 and y + height <= gy + gh + 1
|
|
for gx, gy, gw, gh in (g["bounding"] for g in groups)
|
|
):
|
|
problems.append(f"node {n['id']} is outside its stage group")
|
|
for i, (ident, x0, y0, x1, y1) in enumerate(rectangles):
|
|
for other, ox0, oy0, ox1, oy1 in rectangles[i + 1:]:
|
|
if min(x1, ox1) > max(x0, ox0) + 1 and min(y1, oy1) > max(y0, oy0) + 1:
|
|
problems.append(f"nodes {ident} and {other} overlap (including title bars)")
|
|
return problems
|
|
|
|
|
|
def example_problems(directory: Path) -> list[str]:
|
|
problems = []
|
|
for path in sorted(directory.glob("*.json")):
|
|
try:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
issues = workflow_problems(data)
|
|
except (OSError, ValueError, KeyError, TypeError, AttributeError) as exc:
|
|
issues = [f"malformed saved workflow: {exc}"]
|
|
problems.extend(f"{path.name}: {issue}" for issue in issues)
|
|
return problems
|