File size: 10,132 Bytes
7c2113f e727aeb 7c2113f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 | """Validate the two shipped workflows against the live node, not against a guess.
Checks the things that actually break on someone else's machine: a node type
that is not registered, a widget count that disagrees with INPUT_TYPES, a link
that points at nothing, and a plan that does not survive the real parsers.
"""
import io
import json
import os
import sys
import types
HERE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# ComfyUI root is two levels above the pack. Derived, not hardcoded, so this
# survives the folder being renamed or the checkout living somewhere else.
sys.path.insert(0, os.path.dirname(os.path.dirname(HERE)))
pkg = types.ModuleType("h3p")
pkg.__path__ = [HERE]
sys.modules["h3p"] = pkg
from h3p import refs as R # noqa: E402
from h3p import plan as PL # noqa: E402
# The widget list straight from INPUT_TYPES, so this cannot drift from the node.
from h3p import h3_ref_chain as H3 # noqa: E402
CORE = {"UNETLoader", "CLIPLoader", "VAELoader", "CreateVideo", "SaveVideo",
"SaveImage", "Note", "MarkdownNote", "PreviewAny", "PreviewImage"}
OURS = {"HandTieClips", "HTCChainPreview", "HTCSeamReport"}
# Declared dependencies, not accidents: the turbo stack this node is actually
# run with. Anything outside these three sets is a pack the reader never asked
# for and must not appear in a shipped example.
DEPS = {"LTX_lora_loader": "ComfyUI-PlagueKind-Nodes",
"H3AdaLNLoRAFix": "ComfyUI-PlagueKind-Nodes",
"H3SLAAttention": "ComfyUI-PlagueKind-Nodes",
"MiniMaxLowVRAMAttention": "ComfyUI-KJNodes (experimental)",
"ModelPreviewOverrideKJ": "ComfyUI-KJNodes"}
# The MODEL wire, in order, from the loader to the chain.
MODEL_PATH = ["UNETLoader", "LTX_lora_loader", "H3AdaLNLoRAFix",
"MiniMaxLowVRAMAttention", "H3SLAAttention",
"ModelPreviewOverrideKJ", "HandTieClips"]
# CLIP must reach the chain THROUGH the LoRA loader, or the text half of every
# LoRA is silently dropped -- a wire that looks fine and costs you the LoRA.
CLIP_PATH = ["CLIPLoader", "LTX_lora_loader", "HandTieClips"]
FAIL = []
def ck(label, ok, detail=""):
print(" %s %s%s" % ("ok " if ok else "FAIL", label,
(" " + detail) if detail else ""))
if not ok:
FAIL.append(label)
SOCKET_TYPES = {"MODEL", "CLIP", "VAE", "IMAGE", "AUDIO", "LATENT",
"CONDITIONING", "VIDEO"}
def widget_names():
"""The flat widget list, in the order widgets_values is indexed by.
A widget is any input whose type is a primitive or a combo AND which is not
marked `forceInput` -- `continuity_state` is a STRING but arrives on a wire.
`seed` occupies TWO entries because the frontend appends
control_after_generate directly after it.
"""
it = H3.HandTieClips.INPUT_TYPES()
names = []
for section in ("required", "optional"):
for name, spec in (it.get(section) or {}).items():
t = spec[0]
cfg = spec[1] if len(spec) > 1 else {}
if isinstance(t, str) and t in SOCKET_TYPES:
continue
if cfg.get("forceInput"):
continue
names.append(name)
if name == "seed":
names.append("control_after_generate")
return names
def main():
expect = widget_names()
print("node declares %d widgets (control_after_generate included)\n"
% len(expect))
for fn in ("HandTieClips_Starter.json", "HandTieClips_Showcase.json"):
p = os.path.join(HERE, "workflows", fn)
print(fn)
wf = json.load(io.open(p, encoding="utf-8"))
nodes = wf["nodes"]
byid = {n["id"]: n for n in nodes}
types_used = {n["type"] for n in nodes}
allowed = CORE | OURS | set(DEPS)
ck("only core, this pack and declared deps", types_used <= allowed,
str(sorted(types_used - allowed)) if types_used - allowed
else "%d types" % len(types_used))
ck("no LoadImage", not any(n["type"] == "LoadImage" for n in nodes))
chain = next(n for n in nodes if n["type"] == "HandTieClips")
# The wires, walked link by link. A patch node that is present but
# bypassed round the side is the failure this catches -- it looks right
# on the canvas and does nothing.
def walk_wire(path, kind):
seen = [path[0]]
cur = next(n for n in nodes if n["type"] == path[0])
for _ in range(len(path)):
nxt = [l for l in wf["links"]
if l[1] == cur["id"] and l[5] == kind]
if not nxt:
break
cur = byid[nxt[0][3]]
seen.append(cur["type"])
return seen
m = walk_wire(MODEL_PATH, "MODEL")
ck("MODEL wire runs the whole turbo stack", m == MODEL_PATH,
" -> ".join(m))
c = walk_wire(CLIP_PATH, "CLIP")
ck("CLIP reaches the chain through the LoRA loader", c == CLIP_PATH,
" -> ".join(c))
ins = [i["name"] for i in chain["inputs"]]
dead = [i for i in ins
if i.startswith("ref_image_") and i != "ref_image_size"
or i in ("reference_video", "voice", "start_image")]
ck("no removed sockets", not dead, str(dead))
ck("widget count matches INPUT_TYPES",
len(chain["widgets_values"]) == len(expect),
"%d vs %d" % (len(chain["widgets_values"]), len(expect)))
wv = dict(zip(expect, chain["widgets_values"]))
ck("control_after_generate=fixed", wv["control_after_generate"] == "fixed")
# Links: every id referenced by a node must exist, and vice versa.
lids = {l[0] for l in wf["links"]}
used = set()
for n in nodes:
for i in n.get("inputs", []):
if i.get("link") is not None:
used.add(i["link"])
for o in n.get("outputs", []):
used.update(o.get("links") or [])
ck("link table consistent", lids == used,
"table=%s used=%s" % (sorted(lids), sorted(used)))
ck("every link endpoint exists",
all(l[1] in byid and l[3] in byid for l in wf["links"]))
ck("all four sockets fed",
all(chain["inputs"][k]["link"] is not None for k in range(4)))
# The plans, through the real parsers.
shots = PL.parse_plan(wv["shot_plan"])
n_shots = len(shots["shots"]) if isinstance(shots, dict) else len(shots)
ck("shot plan parses", n_shots == int(wv["chains"]),
"%d shots, chains=%s" % (n_shots, wv["chains"]))
rp = R.parse_ref_plan(wv["ref_plan"])
refs = rp["refs"]
ck("ref plan parses", True, "%d refs" % len(refs))
# Every @tag in every beat must be declared, or the run dies at hop N.
declared = {r["tag"] for r in refs}
import re
body = [s["beat"] for s in (shots["shots"] if isinstance(shots, dict)
else shots)]
tags = set()
for b in body:
tags |= set(re.findall(r"@([A-Za-z0-9_]+)", b or ""))
ck("every @tag is declared", tags <= declared,
"undeclared=%s" % sorted(tags - declared) if tags - declared
else "tags=%s" % sorted(tags))
# And every declared tag must be active on the hops that use it.
bad = []
for hop, b in enumerate(body):
want = set(re.findall(r"@([A-Za-z0-9_]+)", b or ""))
active = R.active_refs(refs, hop, {r["slot"] for r in refs
if r["file"]})
have = set(R.ordinals(active)) | set(R.subjects(refs))
miss = want - have
if miss:
bad.append("hop %d: %s" % (hop + 1, sorted(miss)))
ck("every tag is scheduled onto the hops that use it", not bad,
"; ".join(bad))
warn = R.check(rp, {r["slot"] for r in refs if r["file"]})
ck("no register warnings", not warn, " | ".join(warn))
# Stale named values. Both shipped workflows once carried a copy
# inherited from the dev workflow they were cloned out of, describing a
# 3x10 s chain with a randomizing seed and references that do not ship.
# It is dormant while Comfy.Workflow.NamedValuesRestore stays off, which
# is exactly why it sat there unnoticed.
named = [n["type"] for n in nodes if "widgets_values_named" in n]
ck("no stale widgets_values_named", not named, str(sorted(set(named))))
# The on-canvas board, where there is one.
cards = [n for n in nodes if n["type"] == "MarkdownNote"]
if cards:
ck("every card is marked and non-empty",
all((n.get("properties") or {}).get("htc_card")
and (n.get("widgets_values") or [""])[0].strip()
for n in cards),
"%d cards" % len(cards))
groups = wf.get("groups") or []
ck("a group wraps the board", len(groups) == 1,
"%d groups" % len(groups))
if groups:
gx, gy, gw, gh = groups[0]["bounding"]
outside = [n["title"] for n in cards
if not (gx <= n["pos"][0]
and gy <= n["pos"][1]
and n["pos"][0] + n["size"][0] <= gx + gw
and n["pos"][1] + n["size"][1] <= gy + gh)]
ck("the group encloses every card", not outside, str(outside))
# A card reaching past x=0 would sit on top of the loaders.
ck("the board stays left of the graph",
all(n["pos"][0] + n["size"][0] <= 0 for n in cards))
print(" %d nodes, %d links, %d KB\n"
% (len(nodes), len(wf["links"]), os.path.getsize(p) // 1024))
if FAIL:
print("%d FAILURE(S): %s" % (len(FAIL), ", ".join(FAIL)))
return 1
print("ALL PASS")
return 0
if __name__ == "__main__":
raise SystemExit(main())
|