DumiFlex
e41ae4ded0
Wildcard steps list their options with weights and why each was cut, the pool chain, locked seed, separator and matched value; fixed values, combines and node templates show what they read; constraint steps link the steps they re-weighted; Model Info, Toggles, Injector and Context Loop rows carry their own facts.
243 lines
9.9 KiB
Python
243 lines
9.9 KiB
Python
"""WP_ModelInfo — writes $model_family / $model_variant / $model_name into the
|
|
context so derivations (and, through a wildcard, constraints) can branch on
|
|
the loaded checkpoint.
|
|
|
|
Detection logic lives in ``engine/model_info.py``; this node only reads the
|
|
two facts ComfyUI hands it — the MODEL's config class and the checkpoint file
|
|
name behind the MODEL wire — and writes the result like a Context Injector.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from comfy_api.latest import io # pyright: ignore[reportMissingImports]
|
|
|
|
from engine import model_info, negatives
|
|
from wp_nodes.types import ModelInfoWidgetInput, PipelineContext
|
|
|
|
|
|
def _config_class_name(model: Any) -> str:
|
|
"""``model.model.model_config``'s class name (``SDXL``, ``Flux``, …).
|
|
|
|
``model`` is a ComfyUI ModelPatcher. Read defensively: anything that is
|
|
not shaped like one yields ``""`` and the family is left empty.
|
|
"""
|
|
inner = getattr(model, "model", None)
|
|
config = getattr(inner, "model_config", None)
|
|
return type(config).__name__ if config is not None else ""
|
|
|
|
|
|
def _pin_source(key: str, wired: dict[str, str], pinned: str, detected: str) -> str:
|
|
if key in wired:
|
|
return "wired"
|
|
return "pinned" if pinned else detected
|
|
|
|
|
|
class WPModelInfo(io.ComfyNode):
|
|
"""Chain source that names the loaded model."""
|
|
|
|
@classmethod
|
|
def define_schema(cls):
|
|
return io.Schema(
|
|
node_id="WP_ModelInfo",
|
|
display_name="WP Model Info",
|
|
category="wildcard-pipeline",
|
|
description=(
|
|
"Writes $model_family, $model_variant and $model_name so rules "
|
|
"can branch on the checkpoint. Put it first in the chain."
|
|
),
|
|
inputs=[
|
|
PipelineContext.Input(
|
|
"upstream",
|
|
optional=True,
|
|
tooltip=(
|
|
"Optional upstream Context to extend. Only nodes AFTER "
|
|
"this one see the model variables, so it usually goes "
|
|
"first."
|
|
),
|
|
),
|
|
io.Model.Input(
|
|
"model",
|
|
optional=True,
|
|
tooltip=(
|
|
"The model from your checkpoint loader. Gives "
|
|
"$model_family (sdxl, flux, …) and, through the "
|
|
"loader, the file name. If a node that loads LoRAs "
|
|
"from the prompt sits on the model, take MODEL from "
|
|
"BEFORE it, or the graph loops."
|
|
),
|
|
),
|
|
*(
|
|
io.String.Input(
|
|
key,
|
|
optional=True,
|
|
force_input=True,
|
|
tooltip=tip,
|
|
)
|
|
for key, tip in (
|
|
("family", "Text from another node for $model_family. "
|
|
"Beats detection and the pin."),
|
|
("variant", "Text from another node for $model_variant. "
|
|
"Beats the rules and the pin."),
|
|
("name", "A checkpoint file name from another node. The "
|
|
"variant rules read it, as they read the loader's."),
|
|
)
|
|
),
|
|
# Detected values, pins and the variant rules all live in one
|
|
# DOM widget (src/components/model-info/ModelInfoWidget.vue).
|
|
ModelInfoWidgetInput.Input(
|
|
"wp_model_info", socketless=True, default="", optional=True,
|
|
),
|
|
],
|
|
outputs=[PipelineContext.Output("context")],
|
|
hidden=[io.Hidden.unique_id, io.Hidden.prompt],
|
|
not_idempotent=True,
|
|
)
|
|
|
|
@classmethod
|
|
def fingerprint_inputs(cls, **kwargs):
|
|
"""Always run. The node is cheap, and a cached run sends no UI payload,
|
|
so the widget's family would read "after a run" after every run whose
|
|
inputs matched the last one. NaN never equals itself, so the cache
|
|
never matches."""
|
|
return float("nan")
|
|
|
|
@classmethod
|
|
def execute(
|
|
cls,
|
|
upstream: PipelineContext | None = None,
|
|
model: Any = None,
|
|
family: str | None = None,
|
|
variant: str | None = None,
|
|
name: str | None = None,
|
|
wp_model_info: str = "",
|
|
):
|
|
cfg = model_info.parse_config(wp_model_info)
|
|
# A wired text input beats the panel's pin (an empty wire leaves it).
|
|
wired = {
|
|
k: v.strip()
|
|
for k, v in (("family", family), ("variant", variant), ("name", name))
|
|
if isinstance(v, str) and v.strip()
|
|
}
|
|
pins = {k: wired.get(k) or getattr(cfg, k) for k in ("family", "variant", "name")}
|
|
upstream_ctx: dict = upstream.context if upstream is not None else {}
|
|
upstream_debug: dict = upstream.debug if upstream is not None else {}
|
|
upstream_internals: dict = upstream.internals if upstream is not None else {}
|
|
|
|
hidden = getattr(cls, "hidden", None)
|
|
node_id = getattr(hidden, "unique_id", None) if hidden is not None else None
|
|
prompt = getattr(hidden, "prompt", None) if hidden is not None else None
|
|
|
|
warnings: list[dict] = []
|
|
|
|
file_name = pins["name"]
|
|
name_source = ("wired" if "name" in wired else "pinned") if file_name else "loader"
|
|
if not file_name and model is not None:
|
|
file_name = model_info.find_checkpoint_name(prompt, node_id)
|
|
name_source = "loader"
|
|
|
|
rules, problems = model_info.compile_variant_rules(cfg.rules)
|
|
for problem in problems:
|
|
warnings.append({
|
|
"type": "model_info_bad_rule",
|
|
"severity": "warn",
|
|
"message": f"WP Model Info skipped a variant rule: {problem}",
|
|
})
|
|
|
|
config_class = _config_class_name(model)
|
|
family = pins["family"] or model_info.family_from_config_name(config_class)
|
|
stem = model_info.model_stem(file_name)
|
|
variant = pins["variant"] or model_info.detect_variant(stem, rules)
|
|
|
|
if model is None and not file_name and not family and not variant:
|
|
warnings.append({
|
|
"type": "model_info_nothing_detected",
|
|
"severity": "warn",
|
|
"message": (
|
|
"WP Model Info has no model wired and nothing pinned, so "
|
|
"the model variables are empty."
|
|
),
|
|
})
|
|
|
|
values = {
|
|
model_info.FAMILY_VAR: family,
|
|
model_info.VARIANT_VAR: variant,
|
|
model_info.NAME_VAR: stem,
|
|
}
|
|
|
|
ctx: dict = dict(upstream_ctx)
|
|
ctx.update(values)
|
|
|
|
out_internals = dict(upstream_internals)
|
|
# A write replaces a variable's negatives (send-to-negative rule).
|
|
neg_table = out_internals.get(negatives.NEG_KEY)
|
|
if isinstance(neg_table, dict) and any(k in neg_table for k in values):
|
|
out_internals[negatives.NEG_KEY] = {
|
|
k: v for k, v in neg_table.items() if k not in values
|
|
}
|
|
# Internal-ness is last-write-wins like the value: this write decides
|
|
# it for all three names, flagging the ones the widget marks internal
|
|
# and clearing any flag an earlier writer left on the others.
|
|
internal_vars = {f"model_{k}" for k in cfg.internal}
|
|
flags = {
|
|
k: v for k, v in (out_internals.get("__wp_internal_flags__") or {}).items()
|
|
if k not in values
|
|
}
|
|
flags.update({k: True for k in values if k in internal_vars})
|
|
if flags or "__wp_internal_flags__" in out_internals:
|
|
out_internals["__wp_internal_flags__"] = flags
|
|
|
|
sources = {
|
|
model_info.FAMILY_VAR: _pin_source("family", wired, cfg.family, "model"),
|
|
model_info.VARIANT_VAR: _pin_source("variant", wired, cfg.variant, "rules"),
|
|
model_info.NAME_VAR: name_source,
|
|
}
|
|
# WP Debug: what each value was read from.
|
|
matched = next(
|
|
(r for r in rules if stem and not pins["variant"] and r.regex.search(stem)),
|
|
None,
|
|
)
|
|
details: dict[str, dict[str, Any]] = {
|
|
model_info.FAMILY_VAR: (
|
|
{"config_class": config_class} if config_class and not pins["family"] else {}
|
|
),
|
|
model_info.VARIANT_VAR: {
|
|
**({"rule": matched.pattern, "rule_variant": matched.variant} if matched else {}),
|
|
**({"rules_count": len(rules)} if not pins["variant"] else {}),
|
|
**({"stem": stem} if stem else {}),
|
|
},
|
|
model_info.NAME_VAR: {"file": file_name} if file_name else {},
|
|
}
|
|
traces = [
|
|
{
|
|
"node": "WP_ModelInfo",
|
|
"binding": name,
|
|
"internal": name in internal_vars,
|
|
"type": sources[name],
|
|
"value": value,
|
|
**({"detail": details[name]} if details[name] else {}),
|
|
}
|
|
for name, value in values.items()
|
|
]
|
|
|
|
debug = dict(upstream_debug)
|
|
if node_id is not None:
|
|
for row in traces + warnings:
|
|
row["node_id"] = str(node_id)
|
|
debug["__wp_trace__"] = list(debug.get("__wp_trace__", [])) + traces
|
|
if warnings:
|
|
debug["__wp_warnings__"] = list(debug.get("__wp_warnings__", [])) + warnings
|
|
|
|
# What the widget shows under DETECTED until the next run: the family
|
|
# is only knowable here, and the rest confirms the canvas preview.
|
|
detected = {
|
|
"family": family,
|
|
"variant": variant,
|
|
"name": stem,
|
|
"sources": {k.removeprefix("model_"): v for k, v in sources.items()},
|
|
}
|
|
return io.NodeOutput(
|
|
PipelineContext.Type(context=ctx, debug=debug, internals=out_internals),
|
|
ui={"wp_model_info": [detected]},
|
|
)
|