nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
feat/247-python-params-parser
ComfyAssets-ComfyUI_PromptM…/tests/inspection_parity.py
T
Vito 6d2213ead5 feat(inspect): declarative node support matrix (#250) 
* feat(inspect): add a declarative inspection node semantics table

* fix(inspect): bound and isolate the semantics table

Bound the semantics loader: at most 1 MiB is read, containers may nest
at most eight levels, and a RecursionError from the JSON parser becomes
a SemanticsError instead of escaping.

The table is no longer loaded at import. get_semantics() reads it on
first use and caches only a valid result, and the resolver imports it
lazily, only for a profile with semantics on. The default ingestion
profile never imports the module, so a missing or corrupt table cannot
break scan, monitor or admin import. Inspection of a prompt graph with
an unusable table is unresolved with the reason "Inspection node
semantics are unavailable" and an error log, rather than a crash.

INSPECTION_PROFILE now has semantics on; no strategy consumes the table
yet, so resolution is unchanged.

* feat(inspect): resolve guider-based samplers

Under INSPECTION_PROFILE the resolver discovers samplers from the
semantics table and follows SamplerCustomAdvanced through its guider
link (Reroutes allowed) to a listed guider:

- CFGGuider: positive and negative, so the pair resolves.
- BasicGuider: positive only. The state is "absent" with the reason
  "Guider has no negative input", the existing missing-negative path.
- Any other guider, including DualCFGGuider (D17), is unsupported-node
  and names that guider.

Guider hops draw visits, scan and depth from the existing budgets. A
SamplerCustomAdvanced upstream in latent ancestry is an earlier stage,
so the result stays multistage-ambiguous. Positive and negative
provenance record guider_id. FaceDetailer (detailer-stage) is not a
candidate yet: deciding it needs output ancestry. DEFAULT_PROFILE never
loads the table and is unchanged.

Vector changes (inspection profile only, default vectors unchanged):
- f07b: no-sampler -> resolved (CFGGuider).
- f17: no-sampler -> unsupported-node (FluxGuidance, Task 2.3).
- f01b: no-sampler -> unsupported-node (EmptyFlux2LatentImage latent,
  Task 2.5).
- f01: no-sampler -> ambiguous (24 SamplerCustomAdvanced branches).

The no-sampler tests (resolver and inspect route) now use a sampler
class the inspector does not know, because f17 has a supported one.

* feat(inspect): trace conditioning through supported pass-through and selector nodes

* feat(inspect): treat detailers as sampler stages

* fix(inspect): choose selectors that share one condition together

* fix(inspect): bound detailer stage chains by the depth limit

* feat(inspect): support common text encoders and string sources

* fix(inspect): keep pair execution evidence behind unreadable string sources

* feat(inspect): accept any non-sampler latent origin

* fix(inspect): queue each node once in breadth-first ancestry walks

The latent and detailer image walks queued every link before checking
whether its node was already seen, and popped from the list head. A dense
shared fan-in grew the queue far past the visit budget before any visit
was charged. Both walks now use a deque and enqueue each node at most
once, charging every examined edge as scanned work; outcomes are
unchanged.

* fix(inspect): treat unlisted pipe and embed samplers as possible stages

The latent and detailer ancestry walks only flagged unlisted nodes with a
linked latent_image, guider, positive or negative input, so easy-use pipe
samplers, Impact pipe detailers and embed-driven video samplers were walked
through and the final sampler's prompt resolved silently. A separate
stage-shaped input set (adding pipes, embeds and custom-sampler parts)
now makes such nodes unsupported-node, named.

* fix(inspect): walk each sampler's latent ancestry once per resolution

Every selector branch run and candidate re-walked the same latent
ancestry, so conditioning-carrying origins such as WanImageToVideo turned
a three-selector graph from ambiguous into work-limit. The walk outcome is
now cached per sampler; branch runs of one candidate pay for it once and
other candidates are charged its visit cost, as the text memo does. Walks
that ran out of work are not cached. Every link input is still walked.

* fix(inspect): treat unlisted model-driven samplers as possible stages

Unlisted samplers that link a model with conditioning or a latent under
other input names (XlabsSampler, generic video samplers) were still walked
through, so the final sampler's prompt resolved silently. An unlisted node
with a linked model plus conditioning, neg_conditioning, samples or latent
is now unsupported-node, named; model patches such as LoraLoader and
ModelSampling nodes are not.

* feat(inspect): compute facts from the resolved sampler

* fix(inspect): merge facts across sampler branches when ambiguous

* fix(inspect): merge facts over unresolved stages and share stage inputs via the table

* feat(inspect): explain why a prompt could not be resolved

* feat(inspect): port legacy parser rules the new resolver missed (PromptManager)

R1: before 3.2.3 a save patch overwrote every PromptManager node's saved
text with the last-run node's prompt. The legacy viewer recognizes the
signature (comfy-metadata.js createContext: two or more PromptManager
workflow nodes with one API text and different widget values). Inspection
now refuses those nodes as malformed instead of resolving to another
node's prompt; the inspector hands the parsed workflow to the resolver for
this check only. Lands before the new pass-throughs (R6) that would
otherwise make these graphs resolve to the wrong text.

R7: this repository's node classes up to 07a5af3 (PromptManagerV2,
Positive, Negative, Detailer, V2Text, PositiveText, NegativeText) resolve
with their own outputs and their exact strip and ", " join.

The default (ingestion) profile ignores both.

* feat(inspect): port legacy parser rules the new resolver missed (A1111)

R3: an accumulating batch saver repeats the A1111 parameters record once
per batch item, identical except "Batch index: N". The bounded PNG reader
now keeps the last copy (Pillow's rule, which #165 relied on) instead of
calling the file unreadable. Any other difference, including a missing or
second batch index, is still a conflict, and decoding the copies draws from
the duplicate role budget.

R4: parameters text that starts with "Negative prompt:" has an empty
positive (#165), instead of showing the delimiter line as the positive.

R5: the settings footer may be followed by sd-dynamic-prompts' one-line
"Template:" and "Negative Template:" records, and may be "Steps: N" alone
when there is no full footer.

R10: commas inside A1111's quoted settings values no longer split them, so
an "ADetailer prompt" with a comma keeps the other facts known.

The facts' A1111 reader mirrors R4, R5 and R10; the default (ingestion)
profile keeps the historical split. Also from the Task 2.7 review: a guard
test fails on any uncoded unresolved reason without a user line (M1), and
the encoder bounds role lines to the client's 512 code units (M4).

* feat(inspect): port legacy parser rules the new resolver missed (installed nodes)

R6 and R9 for nodes whose sources are installed beside ComfyUI, each
entry's input names and output slots read from the node's code (the task
report lists file and line):

- detailer stages UltimateSDUpscale (image), UltimateSDUpscaleNoUpscale
  (upscaled_image) and DetailerForEach (image), judged with their upstream
  samplers like FaceDetailer;
- pass-throughs ControlNetInpaintingAliMamaApply and LTXVConditioning
  (positive 0, negative 1) and FluxKontextMultiReferenceLatentMethod;
- encoder Text to Conditioning, strings TextInput and DisplayText;
- facts model pass-throughs Lora Loader (LoraManager), Power Lora Loader
  (rgthree), ModelPatchTorchSettings, CFGNorm, PerturbedAttentionGuidance.

Impact Pack encodes a detailer's non-empty wildcard instead of its positive
input, so FaceDetailer and DetailerForEach now list it as a prompt
override: a literal wildcard is decided at run time and a linked one names
its source, instead of resolving to the positive input. The overwritten
PromptManager refusal (R1) still holds behind the new pass-throughs. The
default profile does not read the table.

* feat(inspect): port legacy parser rules the new resolver missed (upstream nodes)

R6 and R9 for nodes not installed beside ComfyUI here. Their input names
and output slots come from each pack's source at the commit the saved
workflows record in the node's cnr_id and ver properties (the task report
lists repository, commit, file and line):

- samplers ClownsharKSampler_Beta (RES4LYF), KSampler (Efficient) and
  ttN KSampler_v2 (latent input "latent");
- pass-throughs SeedVarianceEnhancer and ApplyInstantIDAdvanced (positive
  1, negative 2; output 0 is the model);
- selector Switch conditioning [Crystools];
- encoder smZ CLIPTextEncode: its text when with_SDXL is false; when true
  the connected CLIP decides at run time which inputs are read, so the
  prompt is known only when text, text_g and text_l agree;
- strings ttN text, CR Text, CR Prompt Text and TextBoxMira;
- facts model pass-through CR Apply LoRA Stack.

Inputs that replace what a sampler reads are not followed: RES4LYF options
(may carry a guider, and with sigmas replace the settings), an Efficiency
script and a ttN XY plot leave the prompt unsupported and the facts
unresolved. The overwritten PromptManager refusal (R1) still holds behind
SeedVarianceEnhancer. The default profile does not read the table.

* test(inspect): cover batch copies past the role budget and orphan widgets

* fix(inspect): match the batch index only in the settings footer

Review I-1: duplicate A1111 parameters copies were compared after blanking
"Batch index: N" anywhere in the record, so copies whose prompts differ
only by that phrase were accepted. The field is now blanked only inside
the record's one "Steps: N, ..." footer line, which must hold it exactly
once; any other difference stays a conflict.

Review M-4: the inspector parses the workflow record only when the API
graph can show the overwritten-PromptManager signature (a PromptManager
class, or a node with PromptManager's search_text input).

* fix(inspect): refuse overwritten PromptManager text inside subgraphs

Controller ruling on review M-1: PromptManager nodes inside subgraphs run
as "<instance>:<inner>" API IDs, nested instances chaining their IDs. The
overwritten-text check now walks subgraph instances (bounded in depth and
nodes) and reads a text widget stored by name, so these nodes are refused
like top-level ones, anchored or not. A definition without an instance
runs nothing and is ignored.

Review M-5: the historical adapters skip an explicit null prepend or append,
as the 07a5af3 code's "if prepend_text:" did.

* test(inspect): import unittest one way in the inspection suites

Code scanning flagged the mixed 'import unittest' and 'from unittest import
mock' forms. Use 'import unittest' plus 'import unittest.mock' everywhere.

* fix(history): name the recorded run's current-file prompts apart from its frozen fields

Now that the inspector resolves more saved files, the Recorded run dialog can
hold two regions named "Positive prompt" (and two of each Copy/Send action):
the frozen run field and the inline current file's prompt. Assistive
technology could not tell them apart, and the real-host chronology test read
two <pre> elements.

Prompt blocks take an optional context label. The inline current file inside
a recorded run names its regions "Current file positive prompt" and its
actions "Copy/Send current file positive prompt"; the visible headings stay
as in the concept. The real-host test now also scopes its frozen-field reads
to the "Frozen run prompt texts" section, and a disposable-server workflow
pins the distinct names with a resolved current file.

* fix(inspect): drop preview-only stages from unanchored candidates

A stage whose outputs reach only non-file outputs (PreviewImage and the other
non_file_outputs) cannot have written a saved file, yet it stayed an
unanchored candidate, so a preview-only FaceDetailer made the saved branch
ambiguous (f01b).

The inspection profile now drops such stages before detailer absorption,
using two linear backward passes over the graph (charged like the consumer
map; a failed pass keeps every stage). A stage without consumers, or one that
also reaches any other terminal, is kept, and nothing is dropped when every
stage is preview-only. Raw plus detailed double saves both write files and
stay ambiguous. A file in the host temp folder may itself be a preview, so it
is inspected with PREVIEW_INSPECTION_PROFILE, which keeps those stages.

Workspace facts follow the same choice: an absent or unresolved result that
names a sampler merges facts across that sampler's stages only, so the
dropped detailer's settings no longer blur the saved file's facts.

f01b now resolves (absent negative, guider has none); its vector is updated
and the expected failure removed. The default profile is unchanged:
differential against 39dcc40 identical over 160,256 cases. Over 60,054
inspection cases no previously resolved or absent result changed.

* fix(history): name the current file's prompt pair button apart from the frozen run's

The recorded run dialog nests the current-file inspector, whose footer repeated the "Send prompt pair" name. Name it from the prompt context, scope the real-host test to the frozen region, and add a dialog-wide duplicate-name check.
2026-10-19 17:26:37 -07:00

1036 lines
32 KiB
Python
Raw Permalink Blame History

"""Deterministic synthetic ComfyUI-shaped fixtures for inspection parity (#240).
Every graph here is original synthetic content (CC0): neutral texts, generic model
file names and invented filename prefixes. Nothing is copied from user outputs.
The shapes reproduce the parity audit's fixture numbering (``f01``..``f24``).
``build(name)`` returns ``{"prompt", "workflow", "extra_chunks"}``. ``write_png``
writes the two records the way core ``SaveImage`` does (Pillow ``PngInfo.add_text``,
``compress_level=4``, a 64x64 constant colour image), then splices any extra raw
``tEXt`` chunks, with real CRC32, before ``IEND``.
Regenerate the committed files with::
python3 -m tests.inspection_parity --write tests/fixtures/inspection_parity
"""
import argparse
import io
import json
import struct
import tempfile
import zlib
from pathlib import Path
from PIL import Image
from PIL.PngImagePlugin import PngInfo
POS = "a red fox in snow, detailed fur"
NEG = "blurry, lowres"
FIXTURE_DIR = Path(__file__).resolve().parent / "fixtures" / "inspection_parity"
VECTORS_PATH = FIXTURE_DIR / "vectors.json"
MAX_FIXTURE_BYTES = 32 * 1024
_SIGNATURE = b"\x89PNG\r\n\x1a\n"
ABSENT = "<absent>"
FACT_FIELDS = ("seed", "steps", "cfg", "sampler", "scheduler", "model")
EXPECT_KEYS = (
"metadata_status",
"state",
"reason",
"reason_code",
"selection",
"positive",
"negative",
"candidate_count",
"facts",
)
# Small enough to commit (each file stays under MAX_FIXTURE_BYTES).
COMMITTED = (
"f01b",
"f02",
"f03",
"f06a",
"f06b",
"f06c",
"f06d",
"f06e",
"f07a",
"f07b",
"f08",
"f09",
"f10",
"f11",
"f12",
"f14",
"f15",
"f16",
"f17",
"f19",
"f20",
"f21",
"f22",
"f23",
"f24",
)
# Built in a temporary directory at test time (too large to commit).
TEST_TIME = ("f01", "f04", "f05")
def _ui(prompt, extra_nodes=0, subgraphs=0, big_widget=None):
"""A UI workflow: one node per prompt node plus inert reroutes/notes/primitives."""
nodes = [
{"id": index + 1, "type": node["class_type"], "mode": 0}
for index, node in enumerate(prompt.values())
]
base = len(nodes)
for offset in range(extra_nodes):
kind = ("Reroute", "Note", "PrimitiveNode")[offset % 3]
nodes.append({"id": base + offset + 1, "type": kind, "mode": 0})
if big_widget is not None:
nodes.append(
{"id": 999999, "type": "Note", "mode": 0, "widgets_values": [big_widget]}
)
workflow = {"last_node_id": len(nodes), "nodes": nodes, "links": [], "version": 0.4}
if subgraphs:
workflow["definitions"] = {
"subgraphs": [
{"id": f"sg-{index}", "nodes": [], "links": []}
for index in range(subgraphs)
]
}
return workflow
def _enc(text):
return {"class_type": "CLIPTextEncode", "inputs": {"text": text, "clip": ["4", 1]}}
def _linked_enc(link):
return {"class_type": "CLIPTextEncode", "inputs": {"text": link, "clip": ["4", 1]}}
def _ksampler_graph(positive, negative, sampler="KSampler", latent=None, extra=None):
graph = {
"4": {
"class_type": "CheckpointLoaderSimple",
"inputs": {"ckpt_name": "sdxl_base.safetensors"},
},
"5": latent
or {
"class_type": "EmptyLatentImage",
"inputs": {"width": 1024, "height": 1024, "batch_size": 1},
},
"6": positive,
"7": negative,
"3": {
"class_type": sampler,
"inputs": {
"seed": 42,
"steps": 25,
"cfg": 6.5,
"sampler_name": "euler",
"scheduler": "normal",
"denoise": 1.0,
"model": ["4", 0],
"positive": ["6", 0],
"negative": ["7", 0],
"latent_image": ["5", 0],
},
},
"8": {
"class_type": "VAEDecode",
"inputs": {"samples": ["3", 0], "vae": ["4", 2]},
},
"9": {
"class_type": "SaveImage",
"inputs": {"filename_prefix": "ComfyUI", "images": ["8", 0]},
},
}
if sampler == "KSamplerAdvanced":
inputs = graph["3"]["inputs"]
inputs["noise_seed"] = inputs.pop("seed")
inputs.pop("denoise")
inputs.update(
add_noise="enable",
start_at_step=0,
end_at_step=10000,
return_with_leftover_noise="disable",
)
graph.update(extra or {})
return graph
def _basic(prompt, workflow=None, extra_chunks=()):
return {
"prompt": prompt,
"workflow": workflow,
"extra_chunks": [tuple(chunk) for chunk in extra_chunks],
}
def _flux2_branches(branches, gates=4):
"""Subgraph-expanded flux2 shape: N SaveImage branches plus a preview-only detailer.
Each branch is SamplerCustomAdvanced <- BasicGuider <- (easy ifElse /
ReferenceLatent) x gates <- FluxGuidance <- CLIPTextEncode <- JoinStringMulti.
Every SaveImage shares one JoinStringMulti ``filename_prefix``. A FaceDetailer
with a different prompt feeds only a PreviewImage.
"""
graph = {
"11": {
"class_type": "VAELoader",
"inputs": {"vae_name": "vae.safetensors"},
},
"12": {
"class_type": "CLIPLoader",
"inputs": {"clip_name": "text_encoder.safetensors", "type": "flux2"},
},
"13": {
"class_type": "UNETLoader",
"inputs": {
"unet_name": "flux-2-klein-9b-fp8.safetensors",
"weight_dtype": "default",
},
},
"14": {"class_type": "LoadImage", "inputs": {"image": "reference.png"}},
"21": {
"class_type": "StringConstantMultiline",
"inputs": {"string": POS + ",", "strip_newlines": False},
},
"22": {
"class_type": "StringConstantMultiline",
"inputs": {"string": "soft winter light,", "strip_newlines": False},
},
"31": {"class_type": "PrimitiveString", "inputs": {"value": "Project"}},
"32": {"class_type": "PrimitiveString", "inputs": {"value": "fixture"}},
"33": {"class_type": "PrimitiveString", "inputs": {"value": "IMG"}},
"34": {
"class_type": "JoinStringMulti",
"inputs": {
"inputcount": 2,
"delimiter": "_",
"return_list": False,
"string_1": ["32", 0],
"string_2": ["33", 0],
},
},
"35": {
"class_type": "JoinStringMulti",
"inputs": {
"inputcount": 4,
"delimiter": "/",
"return_list": False,
"string_1": ["31", 0],
"string_2": ["32", 0],
"string_3": ["34", 0],
"string_4": ["34", 0],
},
},
"41": {
"class_type": "PrimitiveStringMultiline",
"inputs": {"value": "Detailer pass: refine the face"},
},
"42": {
"class_type": "CLIPTextEncode",
"inputs": {"text": ["41", 0], "clip": ["12", 0]},
},
"43": {
"class_type": "ConditioningZeroOut",
"inputs": {"conditioning": ["42", 0]},
},
"44": {
"class_type": "FaceDetailer",
"inputs": {
"image": ["14", 0],
"model": ["13", 0],
"clip": ["12", 0],
"vae": ["11", 0],
"positive": ["42", 0],
"negative": ["43", 0],
"seed": 271828182845,
"steps": 20,
"cfg": 1,
"sampler_name": "euler",
"scheduler": "simple",
},
},
"45": {"class_type": "PreviewImage", "inputs": {"images": ["44", 0]}},
}
for branch in range(branches):
scope = f"{1300 + branch}"
def sub(local, scope=scope):
return f"{scope}:{local}"
graph[f"{1100 + branch}"] = {
"class_type": "StringConstantMultiline",
"inputs": {
"string": f"Scene {branch}: unique shot description",
"strip_newlines": False,
},
}
graph[f"{1200 + branch}"] = {
"class_type": "JoinStringMulti",
"inputs": {
"inputcount": 3,
"delimiter": " ",
"return_list": False,
"string_1": ["21", 0],
"string_2": [f"{1100 + branch}", 0],
"string_3": ["22", 0],
},
}
graph[sub(1)] = {
"class_type": "CLIPTextEncode",
"inputs": {"text": [f"{1200 + branch}", 0], "clip": ["12", 0]},
}
graph[sub(2)] = {
"class_type": "FluxGuidance",
"inputs": {"guidance": 3.5, "conditioning": [sub(1), 0]},
}
previous = sub(2)
for gate in range(gates):
graph[sub(10 + gate)] = {
"class_type": "ImageExists",
"inputs": {"image": ["14", 0]},
}
graph[sub(20 + gate)] = {
"class_type": "VAEEncode",
"inputs": {"pixels": [sub(10 + gate), 0], "vae": ["11", 0]},
}
graph[sub(30 + gate)] = {
"class_type": "ReferenceLatent",
"inputs": {
"conditioning": [previous, 0],
"latent": [sub(20 + gate), 0],
},
}
graph[sub(40 + gate)] = {
"class_type": "easy ifElse",
"inputs": {
"boolean": [sub(10 + gate), 1],
"on_true": [sub(30 + gate), 0],
"on_false": [previous, 0],
},
}
previous = sub(40 + gate)
graph[sub(3)] = {
"class_type": "BasicGuider",
"inputs": {"model": ["13", 0], "conditioning": [previous, 0]},
}
graph[sub(4)] = {
"class_type": "RandomNoise",
"inputs": {"noise_seed": 1000 + branch},
}
graph[sub(5)] = {
"class_type": "KSamplerSelect",
"inputs": {"sampler_name": "euler"},
}
graph[sub(6)] = {
"class_type": "Flux2Scheduler",
"inputs": {"steps": 20, "width": 1024, "height": 1024},
}
graph[sub(7)] = {
"class_type": "EmptyFlux2LatentImage",
"inputs": {"width": 1024, "height": 1024, "batch_size": 1},
}
graph[sub(8)] = {
"class_type": "SamplerCustomAdvanced",
"inputs": {
"noise": [sub(4), 0],
"guider": [sub(3), 0],
"sampler": [sub(5), 0],
"sigmas": [sub(6), 0],
"latent_image": [sub(7), 0],
},
}
graph[sub(9)] = {
"class_type": "VAEDecode",
"inputs": {"samples": [sub(8), 0], "vae": ["11", 0]},
}
graph[f"{1400 + branch}"] = {
"class_type": "SaveImage",
"inputs": {"filename_prefix": ["35", 0], "images": [sub(9), 0]},
}
return graph
def _top_level(graph):
return {key: node for key, node in graph.items() if ":" not in key}
def _f01():
graph = _flux2_branches(24)
return _basic(graph, _ui(_top_level(graph), 450, subgraphs=24))
def _f01b():
graph = _flux2_branches(1)
return _basic(graph, _ui(_top_level(graph), subgraphs=1))
def _f02():
graph = _ksampler_graph(_enc(POS), _enc(NEG))
return _basic(graph, _ui(graph, 600))
def _f03():
side = {
f"{100 + index}": {"class_type": "PrimitiveInt", "inputs": {"value": 1}}
for index in range(499)
}
return _basic(_ksampler_graph(_enc(POS), _enc(NEG), extra=side))
def _f04():
graph = _ksampler_graph(_enc(POS), _enc(NEG))
return _basic(graph, _ui(graph, big_widget="A" * 2_500_000))
def _f05():
graph = _ksampler_graph(_enc(POS), _enc(NEG))
return _basic(graph, _ui(graph, big_widget='"' * 900_000))
def _f06a():
sdxl = {
"class_type": "CLIPTextEncodeSDXL",
"inputs": {
"text_g": POS,
"text_l": POS,
"width": 1024,
"height": 1024,
"crop_w": 0,
"crop_h": 0,
"target_width": 1024,
"target_height": 1024,
"clip": ["4", 1],
},
}
return _basic(_ksampler_graph(sdxl, _enc(NEG)))
def _f06b():
flux = {
"class_type": "CLIPTextEncodeFlux",
"inputs": {"clip_l": POS, "t5xxl": POS, "guidance": 3.5, "clip": ["4", 1]},
}
return _basic(_ksampler_graph(flux, _enc(NEG)))
def _f06c():
guidance = {
"class_type": "FluxGuidance",
"inputs": {"guidance": 3.5, "conditioning": ["10", 0]},
}
return _basic(_ksampler_graph(guidance, _enc(NEG), extra={"10": _enc(POS)}))
def _f06d():
combine = {
"class_type": "ConditioningCombine",
"inputs": {"conditioning_1": ["10", 0], "conditioning_2": ["11", 0]},
}
extra = {"10": _enc(POS), "11": _enc("a second subject")}
return _basic(_ksampler_graph(combine, _enc(NEG), extra=extra))
def _f06e():
def apply():
return {
"class_type": "ControlNetApplyAdvanced",
"inputs": {
"positive": ["10", 0],
"negative": ["11", 0],
"control_net": ["12", 0],
"image": ["13", 0],
"strength": 1.0,
"start_percent": 0,
"end_percent": 1,
},
}
extra = {
"10": _enc(POS),
"11": _enc(NEG),
"12": {
"class_type": "ControlNetLoader",
"inputs": {"control_net_name": "control.safetensors"},
},
"13": {"class_type": "LoadImage", "inputs": {"image": "control.png"}},
}
graph = _ksampler_graph(apply(), apply(), extra=extra)
graph["3"]["inputs"]["negative"] = ["7", 1]
return _basic(graph)
def _f07a():
return _basic(_ksampler_graph(_enc(POS), _enc(NEG), "KSamplerAdvanced"))
def _f07b():
graph = _ksampler_graph(_enc(POS), _enc(NEG))
graph["3"] = {
"class_type": "SamplerCustomAdvanced",
"inputs": {
"noise": ["20", 0],
"guider": ["21", 0],
"sampler": ["22", 0],
"sigmas": ["23", 0],
"latent_image": ["5", 0],
},
}
graph.update(
{
"20": {"class_type": "RandomNoise", "inputs": {"noise_seed": 42}},
"21": {
"class_type": "CFGGuider",
"inputs": {
"model": ["4", 0],
"positive": ["6", 0],
"negative": ["7", 0],
"cfg": 6.5,
},
},
"22": {"class_type": "KSamplerSelect", "inputs": {"sampler_name": "euler"}},
"23": {
"class_type": "BasicScheduler",
"inputs": {
"model": ["4", 0],
"scheduler": "normal",
"steps": 25,
"denoise": 1.0,
},
},
}
)
return _basic(graph)
def _f08():
extra = {
"30": {
"class_type": "StringConcatenate",
"inputs": {"string_a": ["31", 0], "string_b": ", winter", "delimiter": ""},
},
"31": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": POS}},
}
return _basic(_ksampler_graph(_enc(["30", 0]), _enc(NEG), extra=extra))
def _f09():
graph = _ksampler_graph(_enc(POS), _enc(NEG))
graph["40"] = {
"class_type": "LatentUpscaleBy",
"inputs": {
"upscale_method": "nearest-exact",
"scale_by": 1.5,
"samples": ["3", 0],
},
}
graph["41"] = {
"class_type": "KSampler",
"inputs": {**graph["3"]["inputs"], "latent_image": ["40", 0], "denoise": 0.5},
}
graph["8"]["inputs"]["samples"] = ["41", 0]
return _basic(graph)
def _f10():
latent = {
"class_type": "VAEEncode",
"inputs": {"pixels": ["50", 0], "vae": ["4", 2]},
}
extra = {"50": {"class_type": "LoadImage", "inputs": {"image": "input.png"}}}
return _basic(_ksampler_graph(_enc(POS), _enc(NEG), latent=latent, extra=extra))
def _f11():
latent = {
"class_type": "EmptyHunyuanLatentVideo",
"inputs": {"width": 640, "height": 384, "length": 17, "batch_size": 1},
}
return _basic(_ksampler_graph(_enc(POS), _enc(NEG), latent=latent))
def _f12():
def link(name, link_id):
return {"name": name, "link": link_id}
workflow = {
"nodes": [
{
"id": 3,
"type": "KSampler",
"mode": 0,
"inputs": [
link("model", 1),
link("positive", 2),
link("negative", 3),
link("latent_image", 4),
],
"widgets_values": [42, "fixed", 25, 6.5, "euler", "normal", 1.0],
},
{
"id": 4,
"type": "CheckpointLoaderSimple",
"mode": 0,
"inputs": [],
"widgets_values": ["sdxl_base.safetensors"],
},
{
"id": 6,
"type": "CLIPTextEncode",
"mode": 0,
"inputs": [],
"widgets_values": [POS],
},
{
"id": 7,
"type": "CLIPTextEncode",
"mode": 0,
"inputs": [],
"widgets_values": [NEG],
},
{
"id": 5,
"type": "EmptyLatentImage",
"mode": 0,
"inputs": [],
"widgets_values": [1024, 1024, 1],
},
],
"links": [
[1, 4, 0, 3, 0, "MODEL"],
[2, 6, 0, 3, 1, "CONDITIONING"],
[3, 7, 0, 3, 2, "CONDITIONING"],
[4, 5, 0, 3, 3, "LATENT"],
],
}
return _basic(None, workflow)
def _f14():
"""ComfyUI writes ``_meta`` titles and ``"is_changed": [NaN]`` outside inputs."""
graph = _ksampler_graph(_enc(POS), _enc(NEG))
for node in graph.values():
node["_meta"] = {"title": node["class_type"]}
graph["4"]["is_changed"] = [float("nan")]
return _basic(graph, _ui(graph))
def _f15():
"""Byte-identical duplicate ``prompt``/``workflow`` chunks (four copies each)."""
graph = _ksampler_graph(_enc(POS), _enc(NEG))
workflow = _ui(graph)
prompt_text, workflow_text = json.dumps(graph), json.dumps(workflow)
extra = [("prompt", prompt_text), ("workflow", workflow_text)] * 3
return _basic(graph, workflow, extra)
def _f16():
"""A second ``prompt`` chunk that differs from the first."""
graph = _ksampler_graph(_enc(POS), _enc(NEG))
other = _ksampler_graph(_enc("a grey wolf in fog"), _enc(NEG))
return _basic(graph, _ui(graph), [("prompt", json.dumps(other))])
def _f17():
"""SamplerCustomAdvanced + BasicGuider only: no positive/negative inputs."""
graph = {
"10": {
"class_type": "UNETLoader",
"inputs": {"unet_name": "flux1-dev.safetensors", "weight_dtype": "default"},
},
"11": {
"class_type": "DualCLIPLoader",
"inputs": {
"clip_name1": "clip_l.safetensors",
"clip_name2": "t5xxl.safetensors",
"type": "flux",
},
},
"12": {"class_type": "VAELoader", "inputs": {"vae_name": "ae.safetensors"}},
"6": {
"class_type": "CLIPTextEncode",
"inputs": {"text": POS, "clip": ["11", 0]},
},
"26": {
"class_type": "FluxGuidance",
"inputs": {"guidance": 3.5, "conditioning": ["6", 0]},
},
"22": {
"class_type": "BasicGuider",
"inputs": {"model": ["10", 0], "conditioning": ["26", 0]},
},
"25": {"class_type": "RandomNoise", "inputs": {"noise_seed": 42}},
"16": {"class_type": "KSamplerSelect", "inputs": {"sampler_name": "euler"}},
"17": {
"class_type": "BasicScheduler",
"inputs": {
"model": ["10", 0],
"scheduler": "simple",
"steps": 20,
"denoise": 1.0,
},
},
"27": {
"class_type": "EmptySD3LatentImage",
"inputs": {"width": 1024, "height": 1024, "batch_size": 1},
},
"13": {
"class_type": "SamplerCustomAdvanced",
"inputs": {
"noise": ["25", 0],
"guider": ["22", 0],
"sampler": ["16", 0],
"sigmas": ["17", 0],
"latent_image": ["27", 0],
},
},
"8": {
"class_type": "VAEDecode",
"inputs": {"samples": ["13", 0], "vae": ["12", 0]},
},
"9": {
"class_type": "SaveImage",
"inputs": {"filename_prefix": "ComfyUI", "images": ["8", 0]},
},
}
return _basic(graph, _ui(graph))
DETAILER_POS = "a red fox in snow, sharp detailed face"
def _f19():
"""KSampler -> VAEDecode -> FaceDetailer -> SaveImage; the detailer re-prompts."""
graph = _ksampler_graph(_enc(POS), _enc(NEG))
graph["10"] = _enc(DETAILER_POS)
graph["50"] = {
"class_type": "FaceDetailer",
"inputs": {
"image": ["8", 0],
"model": ["4", 0],
"clip": ["4", 1],
"vae": ["4", 2],
"positive": ["10", 0],
"negative": ["7", 0],
"seed": 7,
"steps": 12,
"cfg": 4.5,
"sampler_name": "euler",
"scheduler": "normal",
"denoise": 0.4,
},
}
graph["9"]["inputs"]["images"] = ["50", 0]
return _basic(graph)
def _f20():
"""``easy ifElse`` with a literal boolean picks the positive branch."""
select = {
"class_type": "easy ifElse",
"inputs": {"boolean": True, "on_true": ["10", 0], "on_false": ["11", 0]},
}
extra = {"10": _enc(POS), "11": _enc("a grey wolf in fog")}
return _basic(_ksampler_graph(select, _enc(NEG), extra=extra))
def _f21():
"""``JoinStringMulti`` with ``return_list: true`` feeds the positive encoder."""
join = {
"class_type": "JoinStringMulti",
"inputs": {
"inputcount": 2,
"delimiter": ", ",
"return_list": True,
"string_1": ["11", 0],
"string_2": ["12", 0],
},
}
extra = {
"10": join,
"11": {"class_type": "PrimitiveString", "inputs": {"value": "a red fox"}},
"12": {"class_type": "PrimitiveString", "inputs": {"value": "in snow"}},
}
positive = _linked_enc(["10", 0])
return _basic(_ksampler_graph(positive, _enc(NEG), extra=extra))
def _f22():
"""``PromptSearchList`` (database search at run time) feeds the positive."""
search = {
"class_type": "PromptSearchList",
"inputs": {
"text": "fox",
"category": "",
"tags": "",
"min_rating": 0,
"limit": 50,
},
}
positive = _linked_enc(["10", 0])
return _basic(_ksampler_graph(positive, _enc(NEG), extra={"10": search}))
def _f24():
"""``TextEncodeQwenImage21``: slot 0 is the positive, slot 1 the negative."""
qwen = {
"class_type": "TextEncodeQwenImage21",
"inputs": {
"clip": ["4", 1],
"prompt": POS,
"negative_prompt": NEG,
"vae": ["4", 2],
"resolution": 1024,
},
}
graph = _ksampler_graph(qwen, None)
del graph["7"]
graph["3"]["inputs"]["negative"] = ["6", 1]
return _basic(graph)
def _f23():
"""The negative is ``ConditioningZeroOut`` of the positive encoder."""
zero = {"class_type": "ConditioningZeroOut", "inputs": {"conditioning": ["6", 0]}}
return _basic(_ksampler_graph(_enc(POS), zero))
_BUILDERS = {
"f01": _f01,
"f01b": _f01b,
"f02": _f02,
"f03": _f03,
"f04": _f04,
"f05": _f05,
"f06a": _f06a,
"f06b": _f06b,
"f06c": _f06c,
"f06d": _f06d,
"f06e": _f06e,
"f07a": _f07a,
"f07b": _f07b,
"f08": _f08,
"f09": _f09,
"f10": _f10,
"f11": _f11,
"f12": _f12,
"f14": _f14,
"f15": _f15,
"f16": _f16,
"f17": _f17,
"f19": _f19,
"f20": _f20,
"f21": _f21,
"f22": _f22,
"f23": _f23,
"f24": _f24,
}
def build(name):
"""Return a fresh ``{"prompt", "workflow", "extra_chunks"}`` for one fixture."""
try:
builder = _BUILDERS[name]
except KeyError:
raise KeyError(f"Unknown inspection parity fixture: {name}") from None
return builder()
def fixture_path(name, directory=FIXTURE_DIR):
return Path(directory) / f"{name}.png"
def expected_text_chunks(built):
"""The decoded ``(keyword, text)`` chunks ``write_png`` emits, in file order."""
chunks = []
if built["prompt"] is not None:
chunks.append(("prompt", json.dumps(built["prompt"])))
if built["workflow"] is not None:
chunks.append(("workflow", json.dumps(built["workflow"])))
return chunks + [tuple(chunk) for chunk in built["extra_chunks"]]
def _raw_chunk(kind, data):
return (
struct.pack(">I", len(data))
+ kind
+ data
+ struct.pack(">I", zlib.crc32(kind + data) & 0xFFFFFFFF)
)
def write_png(path, built):
"""Write ``built`` as core SaveImage does, then splice extra raw text chunks."""
info = PngInfo()
if built["prompt"] is not None:
info.add_text("prompt", json.dumps(built["prompt"]))
if built["workflow"] is not None:
info.add_text("workflow", json.dumps(built["workflow"]))
buffer = io.BytesIO()
with Image.new("RGB", (64, 64), (40, 80, 120)) as image:
image.save(buffer, format="PNG", pnginfo=info, compress_level=4)
data = buffer.getvalue()
iend = data.rindex(b"IEND") - 4
extra = b"".join(
_raw_chunk(b"tEXt", key.encode("latin-1") + b"\0" + text.encode("latin-1"))
for key, text in built["extra_chunks"]
)
with open(path, "wb") as stream:
stream.write(data[:iend] + extra + data[iend:])
def write_all(directory, names=COMMITTED):
directory = Path(directory)
directory.mkdir(parents=True, exist_ok=True)
for name in names:
write_png(fixture_path(name, directory), build(name))
return [fixture_path(name, directory) for name in names]
def read_text_chunks(path):
"""Decode every tEXt/zTXt/iTXt chunk as ``(keyword, text)``, duplicates kept."""
with open(path, "rb") as stream:
data = stream.read()
if not data.startswith(_SIGNATURE):
raise ValueError(f"Not a PNG: {path}")
chunks, offset = [], len(_SIGNATURE)
while offset < len(data):
length, kind = struct.unpack(">I4s", data[offset : offset + 8])
body = data[offset + 8 : offset + 8 + length]
offset += 12 + length
if kind == b"tEXt":
key, _, text = body.partition(b"\0")
chunks.append((key.decode("latin-1"), text.decode("latin-1")))
elif kind == b"zTXt":
key, _, rest = body.partition(b"\0")
chunks.append(
(key.decode("latin-1"), zlib.decompress(rest[1:]).decode("latin-1"))
)
elif kind == b"iTXt":
key, _, rest = body.partition(b"\0")
flag, fields = rest[0], rest[2:].split(b"\0", 2)
text = zlib.decompress(fields[2]) if flag else fields[2]
chunks.append((key.decode("latin-1"), text.decode("utf-8")))
elif kind == b"IEND":
break
return chunks
def load_vectors(path=VECTORS_PATH):
with open(path, encoding="utf-8") as stream:
return json.load(stream)
def _role_text(status, text):
if status == "resolved":
return text
return ABSENT if status == "absent" else None
def _observe_inspection(read):
from utils.prompt_inspector import inspect_prompt_evidence
from utils.prompt_pair_resolver import INSPECTION_PROFILE
if read is None: # inspect_current_file's unreadable-media evidence
return "unresolved", "Media metadata could not be read", None, None, {}
evidence = inspect_prompt_evidence(read.metadata, profile=INSPECTION_PROFILE)
resolution, prompts = evidence["resolution"], evidence["prompts"]
roles = [
_role_text(prompts[role]["status"], prompts[role]["text"])
for role in ("positive", "negative")
]
positive, negative = roles
return resolution["state"], resolution["reason"], positive, negative, resolution
def _observe_default(path):
from utils.prompt_pair_resolver import DEFAULT_PROFILE, resolve_prompt_pair
# Last chunk wins, as Pillow's ``image.text`` (the metadata extractor) does.
records = dict(read_text_chunks(path))
canonical = {
key: value for key, value in records.items() if key in ("prompt", "parameters")
}
if isinstance(canonical.get("prompt"), str):
canonical["prompt"] = json.loads(canonical["prompt"])
resolved = resolve_prompt_pair(canonical, profile=DEFAULT_PROFILE)
state = resolved["state"]
roles = [
(
resolved[role]["text"]
if state == "resolved" and isinstance(resolved[role], dict)
else (ABSENT if state == "absent" else None)
)
for role in ("positive", "negative")
]
resolved = {**resolved, "candidate_count": len(resolved.get("candidates") or [])}
positive, negative = roles
return "available", state, resolved.get("reason", ""), positive, negative, resolved
def _observe_path(vector, path):
from utils.prompt_inspector import MetadataReadError, read_inspection_file
if vector["profile"] != "inspection":
return _observe_default(path)
try:
read, status = read_inspection_file(path), "available"
except MetadataReadError:
read, status = None, "unreadable"
state, reason, positive, negative, resolution = _observe_inspection(read)
return status, state, reason, positive, negative, resolution
def observe(vector, directory=FIXTURE_DIR):
"""Today's outcome for one vector, shaped like ``vector["expect"]`` minus facts.
``inspection`` is the workspace inspector path (``read_inspection_file`` then
``inspect_prompt_evidence``) with ``INSPECTION_PROFILE``. ``default`` pins
the resolver's ``DEFAULT_PROFILE`` only: ``resolve_prompt_pair`` on the role
records decoded from the PNG text chunks. Ingestion's own readers are
covered by their own suites.
Committed fixtures are read from ``directory``, which must contain them.
Test-time fixtures are built into a temporary directory; the file is
written and closed before it is read, and closed again before cleanup.
"""
name = vector["fixture"]
if name in TEST_TIME:
with tempfile.TemporaryDirectory() as temporary:
path = fixture_path(name, temporary)
write_png(path, build(name))
observed = _observe_path(vector, path)
else:
path = fixture_path(name, directory)
if not path.is_file():
raise FileNotFoundError(path)
observed = _observe_path(vector, path)
status, state, reason, positive, negative, resolution = observed
return {
"metadata_status": status,
"state": state,
"reason": reason,
"reason_code": resolution.get("reason_code"),
"selection": resolution.get("selection"),
"positive": positive,
"negative": negative,
"candidate_count": resolution.get("candidate_count", 0),
}
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("--write", required=True, help="output directory")
args = parser.parse_args(argv)
for written in write_all(args.write):
print(written)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink