feat(parse): one generation-parameter parser in Python; scan links what the resolver cannot decide (#253)
* feat(parse): port the browser generation-parameter parser to Python Adds utils/generation_params.py, a line-for-line port of extractGenerationParams in web/js/comfy-metadata.js, the parser behind the gallery, admin and metadata pages. The scan and live monitor can use it so every surface reads prompts the same way. The JavaScript parser is unchanged. The port reproduces the JavaScript value semantics the result depends on (property order, truthiness, String(), trim(), strict equality, Map keys, JSON.parse number rounding, and the inputs on which it throws), and runs its recursive tracing on an explicit stack so crafted graphs cannot exhaust the Python stack. Tests: - behaviour tests for each tracing rule, with JavaScript results as expected values, including the visit-limit boundary; - golden vectors of the JavaScript parser's results for every named case and every PNG fixture, so the comparison runs without Node; - a live comparison against the JavaScript parser when Node is present; - a randomized differential test over 9,000 generated graphs and workflows, coherent and corrupted, including inputs where the JavaScript throws. Five deliberate porting mistakes (trim, truthiness, prepend handling, visit limit, pair slot) each fail the tests. On a 38,602-file output folder, all 18,491 files with a workflow give identical results. * fix(scan): link the traced prompt when the pair resolver cannot decide The output scan skipped every file the pair resolver called ambiguous or unresolved, so workflows where several samplers share one prompt (a base pass, an upscale pass and a detailer) never reached the Library from a scan. It now links the positive prompt the shared parser traces for those files, the prompt the inspector and the legacy pages show. Files the resolver decides are stored as before, and graphs without a recognized sampler keep the legacy extractor (#252). Read-only scans of a 38,602-file output folder into temporary databases: - links: 1,500 before, 7,954 after; none of the 1,500 dropped or changed; - of the added links that differ from the pre-redesign scan, 1,275 replace a number, a LoRA tag, a fragment or a truncated prompt that scan stored. * refactor(parse): return a uniform pair from the SameValueZero key
This commit is contained in:
Vito
+52
-16
@@ -27,10 +27,22 @@ except ImportError:
|
||||
try:
|
||||
from ...database.operations import PromptDatabase
|
||||
from ...database.pair_ingestion import ingest_resolved_pair
|
||||
from ...utils.generation_params import (
|
||||
PLACEHOLDERS,
|
||||
GenerationParamsError,
|
||||
extract_generation_params,
|
||||
parse_js_json,
|
||||
)
|
||||
from ...utils.prompt_pair_resolver import resolve_prompt_pair
|
||||
except ImportError:
|
||||
from database.operations import PromptDatabase
|
||||
from database.pair_ingestion import ingest_resolved_pair
|
||||
from utils.generation_params import (
|
||||
PLACEHOLDERS,
|
||||
GenerationParamsError,
|
||||
extract_generation_params,
|
||||
parse_js_json,
|
||||
)
|
||||
from utils.prompt_pair_resolver import resolve_prompt_pair
|
||||
|
||||
|
||||
@@ -1511,6 +1523,21 @@ class AdminRoutesMixin:
|
||||
counts["pair_links_added"] += outcome.get("links_added", 0)
|
||||
return counts
|
||||
|
||||
@staticmethod
|
||||
def _graph_prompt_text(metadata):
|
||||
"""The positive prompt the browser parser traces from the graphs, or None."""
|
||||
records = {}
|
||||
for key in ("prompt", "workflow"):
|
||||
value = metadata.get(key)
|
||||
records[key] = parse_js_json(value) if isinstance(value, str) else value
|
||||
try:
|
||||
positive = extract_generation_params(records)["positivePrompt"]
|
||||
except GenerationParamsError:
|
||||
return None
|
||||
if not isinstance(positive, str) or positive == PLACEHOLDERS["positivePrompt"]:
|
||||
return None
|
||||
return positive.strip() or None
|
||||
|
||||
def _ingest_scanned_file(self, media_file, metadata, detailed=False):
|
||||
"""Enrich exact pairs; retain legacy positive-only import for absent pairs."""
|
||||
if not metadata:
|
||||
@@ -1531,23 +1558,32 @@ class AdminRoutesMixin:
|
||||
self.db, str(media_file), canonical, resolution
|
||||
)
|
||||
return result if detailed else result["status"]
|
||||
if resolution["state"] == "ambiguous":
|
||||
return "ambiguous"
|
||||
if resolution["state"] == "unresolved":
|
||||
if canonical.get("prompt") or not isinstance(
|
||||
canonical.get("parameters"), str
|
||||
):
|
||||
if canonical.get("prompt") and resolution["state"] in (
|
||||
"ambiguous",
|
||||
"unresolved",
|
||||
):
|
||||
# The resolver could not decide: link the prompt the inspector and the
|
||||
# legacy pages show for this file (the browser parser, in Python).
|
||||
# Graphs without a recognized sampler keep the legacy extractor below.
|
||||
prompt_text = self._graph_prompt_text(metadata)
|
||||
if not prompt_text:
|
||||
return resolution["state"]
|
||||
parsed = self._parse_comfyui_prompt(metadata)
|
||||
if not (parsed.get("prompt") or parsed.get("parameters")):
|
||||
return None
|
||||
|
||||
prompt_text = self._extract_readable_prompt(parsed)
|
||||
if prompt_text is not None and not isinstance(prompt_text, str):
|
||||
prompt_text = str(prompt_text)
|
||||
if not (prompt_text and prompt_text.strip()):
|
||||
return "found"
|
||||
prompt_text = prompt_text.strip()
|
||||
elif resolution["state"] == "ambiguous":
|
||||
return "ambiguous"
|
||||
elif resolution["state"] == "unresolved" and not isinstance(
|
||||
canonical.get("parameters"), str
|
||||
):
|
||||
return resolution["state"]
|
||||
else:
|
||||
parsed = self._parse_comfyui_prompt(metadata)
|
||||
if not (parsed.get("prompt") or parsed.get("parameters")):
|
||||
return None
|
||||
prompt_text = self._extract_readable_prompt(parsed)
|
||||
if prompt_text is not None and not isinstance(prompt_text, str):
|
||||
prompt_text = str(prompt_text)
|
||||
if not (prompt_text and prompt_text.strip()):
|
||||
return "found"
|
||||
prompt_text = prompt_text.strip()
|
||||
|
||||
prompt_hash = generate_prompt_hash(prompt_text)
|
||||
existing = self.db.get_prompt_by_hash(prompt_hash)
|
||||
|
||||
+809
File diff suppressed because one or more lines are too long
@@ -0,0 +1,444 @@
|
||||
"""Named workflow cases shared by the generation-parameter parser tests.
|
||||
|
||||
Each case is the raw metadata records a file would carry, as text, so the cases
|
||||
exercise ``parse_js_json`` as well as the tracing. The JS parity test runs every
|
||||
case through the legacy browser parser too. Prompt texts and seeds are invented.
|
||||
"""
|
||||
|
||||
import json
|
||||
|
||||
|
||||
def _text(value):
|
||||
return None if value is None else json.dumps(value)
|
||||
|
||||
|
||||
def case(prompt=None, workflow=None):
|
||||
return {"prompt": _text(prompt), "workflow": _text(workflow)}
|
||||
|
||||
|
||||
def encoder(text, clip=("4", 1)):
|
||||
return {
|
||||
"class_type": "CLIPTextEncode",
|
||||
"inputs": {"text": text, "clip": list(clip)},
|
||||
}
|
||||
|
||||
|
||||
def ksampler(positive, negative, model=("4", 0), **settings):
|
||||
inputs = {
|
||||
"seed": 5,
|
||||
"steps": 20,
|
||||
"cfg": 7,
|
||||
"sampler_name": "euler",
|
||||
"scheduler": "normal",
|
||||
"denoise": 1.0,
|
||||
"model": list(model),
|
||||
"positive": list(positive),
|
||||
"negative": list(negative),
|
||||
"latent_image": ["2", 0],
|
||||
}
|
||||
inputs.update(settings)
|
||||
return {"class_type": "KSampler", "inputs": inputs}
|
||||
|
||||
|
||||
CHECKPOINT = {
|
||||
"class_type": "CheckpointLoaderSimple",
|
||||
"inputs": {"ckpt_name": "example.safetensors"},
|
||||
}
|
||||
LATENT = {
|
||||
"class_type": "EmptyLatentImage",
|
||||
"inputs": {"width": 512, "height": 512, "batch_size": 1},
|
||||
}
|
||||
|
||||
|
||||
def _basic():
|
||||
return {
|
||||
"4": CHECKPOINT,
|
||||
"2": LATENT,
|
||||
"6": encoder("a cat on a windowsill"),
|
||||
"7": encoder("blurry, watermark"),
|
||||
"3": ksampler(("6", 0), ("7", 0)),
|
||||
"9": {"class_type": "SaveImage", "inputs": {"images": ["8", 0]}},
|
||||
}
|
||||
|
||||
|
||||
def _prompt_manager(text, prepend="", append=""):
|
||||
return {
|
||||
"class_type": "PromptManager",
|
||||
"inputs": {
|
||||
"text": text,
|
||||
"category": "",
|
||||
"tags": "",
|
||||
"search_text": "",
|
||||
"prepend_text": prepend,
|
||||
"append_text": append,
|
||||
"clip": ["4", 1],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _shared_prompt_pipeline():
|
||||
"""One PromptManager pair feeds a base pass, an upscale pass and a detailer."""
|
||||
return {
|
||||
"1": CHECKPOINT,
|
||||
"5": ksampler(
|
||||
("25", 0),
|
||||
("26", 0),
|
||||
("1", 0),
|
||||
seed=111,
|
||||
steps=4,
|
||||
cfg=1.0,
|
||||
sampler_name="lcm",
|
||||
latent_image=["20", 0],
|
||||
),
|
||||
"6": {
|
||||
"class_type": "VAEDecode",
|
||||
"inputs": {"samples": ["5", 0], "vae": ["1", 2]},
|
||||
},
|
||||
"11": {
|
||||
"class_type": "VAEEncode",
|
||||
"inputs": {"pixels": ["6", 0], "vae": ["1", 2]},
|
||||
},
|
||||
"12": ksampler(
|
||||
("25", 0),
|
||||
("26", 0),
|
||||
("1", 0),
|
||||
seed=222,
|
||||
steps=6,
|
||||
cfg=1.0,
|
||||
sampler_name="lcm",
|
||||
denoise=0.45,
|
||||
latent_image=["11", 0],
|
||||
),
|
||||
"13": {
|
||||
"class_type": "VAEDecode",
|
||||
"inputs": {"samples": ["12", 0], "vae": ["1", 2]},
|
||||
},
|
||||
"17": {
|
||||
"class_type": "FaceDetailer",
|
||||
"inputs": {
|
||||
"seed": 333,
|
||||
"steps": 4,
|
||||
"cfg": 1.0,
|
||||
"sampler_name": "lcm",
|
||||
"image": ["13", 0],
|
||||
"model": ["1", 0],
|
||||
"clip": ["1", 1],
|
||||
"vae": ["1", 2],
|
||||
"positive": ["25", 0],
|
||||
"negative": ["26", 0],
|
||||
},
|
||||
},
|
||||
"20": {
|
||||
"class_type": "EmptyLatentImage",
|
||||
"inputs": {"width": 640, "height": 1536, "batch_size": 1},
|
||||
},
|
||||
"25": _prompt_manager("a lighthouse at dusk, oil painting"),
|
||||
"26": _prompt_manager("blurry, watermark"),
|
||||
"48": {
|
||||
"class_type": "KikoSaveImage",
|
||||
"inputs": {"filename_prefix": ["49", 0], "images": ["17", 0]},
|
||||
},
|
||||
"49": {
|
||||
"class_type": "KikoProjectFilePath",
|
||||
"inputs": {"project_name": "example"},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _custom_sampler(guider):
|
||||
return {
|
||||
"13": CHECKPOINT,
|
||||
"20": encoder("a red fox in snow", clip=("13", 1)),
|
||||
"21": encoder("lowres", clip=("13", 1)),
|
||||
"22": guider,
|
||||
"25": {"class_type": "RandomNoise", "inputs": {"noise_seed": 42}},
|
||||
"16": {"class_type": "KSamplerSelect", "inputs": {"sampler_name": "dpmpp_2m"}},
|
||||
"17": {
|
||||
"class_type": "BasicScheduler",
|
||||
"inputs": {
|
||||
"scheduler": "karras",
|
||||
"steps": 8,
|
||||
"denoise": 1.0,
|
||||
"model": ["13", 0],
|
||||
},
|
||||
},
|
||||
"1300:8": {
|
||||
"class_type": "SamplerCustomAdvanced",
|
||||
"inputs": {
|
||||
"noise": ["25", 0],
|
||||
"guider": ["22", 0],
|
||||
"sampler": ["16", 0],
|
||||
"sigmas": ["17", 0],
|
||||
"latent_image": ["5", 0],
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _workflow_only():
|
||||
return {
|
||||
"nodes": [
|
||||
{
|
||||
"id": 4,
|
||||
"type": "CheckpointLoaderSimple",
|
||||
"mode": 0,
|
||||
"inputs": [],
|
||||
"widgets_values": ["example.safetensors"],
|
||||
},
|
||||
{
|
||||
"id": 6,
|
||||
"type": "CLIPTextEncode",
|
||||
"mode": 0,
|
||||
"inputs": [{"name": "clip", "link": 3}],
|
||||
"widgets_values": ["a tidepool at noon"],
|
||||
},
|
||||
{
|
||||
"id": 7,
|
||||
"type": "CLIPTextEncode",
|
||||
"mode": 0,
|
||||
"inputs": [{"name": "clip", "link": 5}],
|
||||
"widgets_values": ["noisy"],
|
||||
},
|
||||
{
|
||||
"id": 10,
|
||||
"type": "Reroute",
|
||||
"mode": 0,
|
||||
"inputs": [{"name": "", "link": 7}],
|
||||
},
|
||||
{
|
||||
"id": 3,
|
||||
"type": "KSampler",
|
||||
"mode": 0,
|
||||
"inputs": [
|
||||
{"name": "model", "link": 1},
|
||||
{"name": "positive", "link": 8},
|
||||
{"name": "negative", "link": 4},
|
||||
{"name": "latent_image", "link": 2},
|
||||
],
|
||||
"widgets_values": [314, "randomize", 25, 6.5, "dpmpp_2m", "karras", 1],
|
||||
},
|
||||
{
|
||||
"id": 11,
|
||||
"type": "CLIPTextEncode",
|
||||
"mode": 2,
|
||||
"inputs": [],
|
||||
"widgets_values": ["muted node text"],
|
||||
},
|
||||
],
|
||||
"links": [
|
||||
[1, 4, 0, 3, 0, "MODEL"],
|
||||
[3, 4, 1, 6, 0, "CLIP"],
|
||||
[5, 4, 1, 7, 0, "CLIP"],
|
||||
[7, 6, 0, 10, 0, "CONDITIONING"],
|
||||
[8, 10, 0, 3, 1, "CONDITIONING"],
|
||||
[4, 7, 0, 3, 2, "CONDITIONING"],
|
||||
{
|
||||
"id": 2,
|
||||
"origin_id": 99,
|
||||
"origin_slot": 0,
|
||||
"target_id": 3,
|
||||
"target_slot": 3,
|
||||
"type": "LATENT",
|
||||
},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _chain(length):
|
||||
graph = {"4": CHECKPOINT, "1": encoder("deep chain text")}
|
||||
previous = "1"
|
||||
for index in range(length):
|
||||
node_id = str(100 + index)
|
||||
graph[node_id] = {
|
||||
"class_type": "ConditioningPass",
|
||||
"inputs": {"conditioning": [previous, 0]},
|
||||
}
|
||||
previous = node_id
|
||||
graph["3"] = ksampler((previous, 0), ("1", 0))
|
||||
return graph
|
||||
|
||||
|
||||
CASES = {
|
||||
"basic": case(_basic()),
|
||||
"prompt-manager-prepend-append": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"6": _prompt_manager("a cat", prepend="masterpiece", append="4k"),
|
||||
"7": _prompt_manager(
|
||||
"masterpiece, a dog", prepend="masterpiece", append="4k"
|
||||
),
|
||||
"3": ksampler(("6", 0), ("7", 0)),
|
||||
}
|
||||
),
|
||||
"linked-text-concat": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"10": {"class_type": "PrimitiveString", "inputs": {"value": "first part"}},
|
||||
"11": {
|
||||
"class_type": "StringConcatenate",
|
||||
"inputs": {
|
||||
"string_a": ["10", 0],
|
||||
"string_b": "second part",
|
||||
"delimiter": ", ",
|
||||
"seed": 3,
|
||||
"string_c": " ",
|
||||
},
|
||||
},
|
||||
"6": {
|
||||
"class_type": "CLIPTextEncode",
|
||||
"inputs": {"text": ["11", 0], "clip": ["4", 1]},
|
||||
},
|
||||
"7": encoder(""),
|
||||
"3": ksampler(("6", 0), ("7", 0)),
|
||||
}
|
||||
),
|
||||
"prompt-manager-pair": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"8": {
|
||||
"class_type": "PromptManagerPair",
|
||||
"inputs": {
|
||||
"text": "pair positive",
|
||||
"negative_text": "pair negative",
|
||||
"prepend_text": "",
|
||||
"append_text": "",
|
||||
"clip": ["4", 1],
|
||||
},
|
||||
},
|
||||
"3": ksampler(("8", 0), ("8", 1)),
|
||||
}
|
||||
),
|
||||
"conditioning-passthrough": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"6": encoder("through a controlnet"),
|
||||
"7": encoder("negative through it"),
|
||||
"12": {
|
||||
"class_type": "ControlNetApplyAdvanced",
|
||||
"inputs": {
|
||||
"positive": ["6", 0],
|
||||
"negative": ["7", 0],
|
||||
"control_net": ["13", 0],
|
||||
},
|
||||
},
|
||||
"3": ksampler(("12", 0), ("12", 1)),
|
||||
}
|
||||
),
|
||||
"first-sampler-in-js-key-order": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"6": encoder("from sampler twelve"),
|
||||
"7": encoder("from sampler five"),
|
||||
"x": ksampler(("6", 0), ("6", 0)),
|
||||
"12": ksampler(("6", 0), ("6", 0), seed=12),
|
||||
"5": ksampler(("7", 0), ("7", 0), seed=55),
|
||||
}
|
||||
),
|
||||
"custom-sampler-cfg-guider": case(
|
||||
_custom_sampler(
|
||||
{
|
||||
"class_type": "CFGGuider",
|
||||
"inputs": {
|
||||
"model": ["13", 0],
|
||||
"positive": ["20", 0],
|
||||
"negative": ["21", 0],
|
||||
"cfg": 3.5,
|
||||
},
|
||||
}
|
||||
)
|
||||
),
|
||||
"custom-sampler-basic-guider": case(
|
||||
_custom_sampler(
|
||||
{
|
||||
"class_type": "BasicGuider",
|
||||
"inputs": {"model": ["13", 0], "conditioning": ["20", 0]},
|
||||
}
|
||||
)
|
||||
),
|
||||
"no-sampler-loose-settings": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"3": {
|
||||
"class_type": "MyKSamplerVariant",
|
||||
"inputs": {
|
||||
"noise_seed": 9,
|
||||
"steps": 12,
|
||||
"cfg": 2,
|
||||
"sampler_name": "ddim",
|
||||
},
|
||||
},
|
||||
}
|
||||
),
|
||||
"primitive-seed": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"6": encoder("seeded"),
|
||||
"30": {
|
||||
"class_type": "Seed (rgthree)",
|
||||
"inputs": {"other": "x", "seed": 77},
|
||||
},
|
||||
"3": ksampler(("6", 0), ("6", 0), seed=["30", 0]),
|
||||
}
|
||||
),
|
||||
"workflow-only": case(workflow=_workflow_only()),
|
||||
"overwritten-prompt-manager-text": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"25": _prompt_manager("same saved text"),
|
||||
"26": _prompt_manager("same saved text"),
|
||||
"3": ksampler(("25", 0), ("26", 0)),
|
||||
},
|
||||
{
|
||||
"nodes": [
|
||||
{
|
||||
"id": 25,
|
||||
"type": "PromptManager",
|
||||
"widgets_values": ["the real positive"],
|
||||
},
|
||||
{
|
||||
"id": 26,
|
||||
"type": "PromptManager",
|
||||
"widgets_values": ["the real negative"],
|
||||
},
|
||||
],
|
||||
"links": [],
|
||||
},
|
||||
),
|
||||
"conditioning-cycle": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"20": {
|
||||
"class_type": "ConditioningCombine",
|
||||
"inputs": {"conditioning_1": ["21", 0]},
|
||||
},
|
||||
"21": {
|
||||
"class_type": "ConditioningCombine",
|
||||
"inputs": {"conditioning_1": ["20", 0]},
|
||||
},
|
||||
"3": ksampler(("20", 0), ("21", 0)),
|
||||
}
|
||||
),
|
||||
"chain-within-visit-limit": case(_chain(450)),
|
||||
"chain-beyond-visit-limit": case(_chain(600)),
|
||||
"chain-at-last-visit": case(_chain(499)),
|
||||
"chain-one-past-last-visit": case(_chain(500)),
|
||||
"blank-text-and-zero-seed": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"6": encoder(" "),
|
||||
"7": encoder("kept negative"),
|
||||
"3": ksampler(("6", 0), ("7", 0), seed=0),
|
||||
}
|
||||
),
|
||||
"shared-prompt-pipeline": case(_shared_prompt_pipeline()),
|
||||
"null-workflow-node": case(_basic(), {"nodes": [None], "links": []}),
|
||||
"unparsable-records": {"prompt": '{"3": NaN}', "workflow": "not json"},
|
||||
"seed-beyond-double-precision": case(
|
||||
{
|
||||
"4": CHECKPOINT,
|
||||
"6": encoder("x"),
|
||||
"3": ksampler(("6", 0), ("6", 0), cfg=1.0, seed=4974052037430000001),
|
||||
}
|
||||
),
|
||||
"nothing": case(),
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
// Runs the legacy browser parser (comfy-metadata.js extractGenerationParams) on
|
||||
// one JSON case per input line and prints one JSON result per line. The Python
|
||||
// port (utils/generation_params.py) must produce the same results.
|
||||
//
|
||||
// Input line: {"prompt": <text or null>, "workflow": <text or null>}
|
||||
// Each present record is parsed with JSON.parse; a record that fails
|
||||
// to parse is treated as absent, as the browser pages do.
|
||||
// Output line: {"ok": true, "result": {...}} or {"ok": false} when the parser throws.
|
||||
import path from 'node:path';
|
||||
import fs from 'node:fs';
|
||||
import { createRequire } from 'node:module';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
|
||||
const here = path.dirname(fileURLToPath(import.meta.url));
|
||||
const require = createRequire(import.meta.url);
|
||||
const { extractGenerationParams } = require(path.join(here, '..', '..', '..', 'web', 'js', 'comfy-metadata.js'));
|
||||
|
||||
function parsed(text) {
|
||||
if (typeof text !== 'string') return undefined;
|
||||
try { return JSON.parse(text); } catch { return undefined; }
|
||||
}
|
||||
|
||||
for (const line of fs.readFileSync(0, 'utf8').split('\n')) {
|
||||
if (!line.trim()) continue;
|
||||
const item = JSON.parse(line), data = {};
|
||||
const prompt = parsed(item.prompt), workflow = parsed(item.workflow);
|
||||
if (prompt !== undefined) data.prompt = prompt;
|
||||
if (workflow !== undefined) data.workflow = workflow;
|
||||
let out;
|
||||
try { out = { ok: true, result: extractGenerationParams(data) }; } catch { out = { ok: false }; }
|
||||
process.stdout.write(JSON.stringify(out) + '\n');
|
||||
}
|
||||
@@ -0,0 +1,221 @@
|
||||
"""Tests for utils.generation_params, the Python port of the browser parser.
|
||||
|
||||
Expected values are what web/js/comfy-metadata.js extractGenerationParams
|
||||
returns for the same input (tests/test_generation_params_js_parity.py checks
|
||||
that directly when Node is available).
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from utils.generation_params import ( # noqa: E402
|
||||
PLACEHOLDERS,
|
||||
GenerationParamsError,
|
||||
extract_generation_params,
|
||||
extract_from_records,
|
||||
parse_js_json,
|
||||
)
|
||||
|
||||
try:
|
||||
from tests.generation_params_cases import CASES # noqa: E402
|
||||
except ImportError: # discovered with tests/ as the top-level directory
|
||||
from generation_params_cases import CASES # noqa: E402
|
||||
|
||||
|
||||
def run(name):
|
||||
return extract_from_records(CASES[name]["prompt"], CASES[name]["workflow"])
|
||||
|
||||
|
||||
class TracesPrompts(unittest.TestCase):
|
||||
def test_sampler_prompts_and_settings(self):
|
||||
self.assertEqual(
|
||||
run("basic"),
|
||||
{
|
||||
"positivePrompt": "a cat on a windowsill",
|
||||
"negativePrompt": "blurry, watermark",
|
||||
"checkpoint": "example.safetensors",
|
||||
"steps": 20,
|
||||
"cfgScale": 7,
|
||||
"sampler": "euler",
|
||||
"seed": 5,
|
||||
},
|
||||
)
|
||||
|
||||
def test_prompt_manager_joins_prepend_and_append_once(self):
|
||||
result = run("prompt-manager-prepend-append")
|
||||
self.assertEqual(result["positivePrompt"], "masterpiece a cat 4k")
|
||||
# The text already starts with the prepend text, so it is not repeated.
|
||||
self.assertEqual(result["negativePrompt"], "masterpiece, a dog 4k")
|
||||
|
||||
def test_linked_text_joins_text_inputs_with_the_delimiter(self):
|
||||
result = run("linked-text-concat")
|
||||
self.assertEqual(result["positivePrompt"], "first part, second part")
|
||||
self.assertEqual(result["negativePrompt"], PLACEHOLDERS["negativePrompt"])
|
||||
|
||||
def test_prompt_manager_pair_outputs(self):
|
||||
result = run("prompt-manager-pair")
|
||||
self.assertEqual(result["positivePrompt"], "pair positive")
|
||||
self.assertEqual(result["negativePrompt"], "pair negative")
|
||||
|
||||
def test_conditioning_passthrough_follows_the_role(self):
|
||||
result = run("conditioning-passthrough")
|
||||
self.assertEqual(result["positivePrompt"], "through a controlnet")
|
||||
self.assertEqual(result["negativePrompt"], "negative through it")
|
||||
|
||||
def test_several_samplers_sharing_one_prompt_pair(self):
|
||||
result = run("shared-prompt-pipeline")
|
||||
self.assertEqual(result["positivePrompt"], "a lighthouse at dusk, oil painting")
|
||||
self.assertEqual(result["negativePrompt"], "blurry, watermark")
|
||||
self.assertEqual(
|
||||
(result["seed"], result["steps"], result["sampler"]), (111, 4, "lcm")
|
||||
)
|
||||
|
||||
def test_overwritten_prompt_manager_text_is_recovered_from_widgets(self):
|
||||
result = run("overwritten-prompt-manager-text")
|
||||
self.assertEqual(result["positivePrompt"], "the real positive")
|
||||
self.assertEqual(result["negativePrompt"], "the real negative")
|
||||
|
||||
|
||||
class ChoosesTheSampler(unittest.TestCase):
|
||||
def test_first_sampler_follows_javascript_key_order(self):
|
||||
# Integer-like IDs come first in ascending order, then the others.
|
||||
result = run("first-sampler-in-js-key-order")
|
||||
self.assertEqual(result["positivePrompt"], "from sampler five")
|
||||
self.assertEqual(result["seed"], 55)
|
||||
|
||||
def test_custom_sampler_reads_its_guider_and_split_settings(self):
|
||||
result = run("custom-sampler-cfg-guider")
|
||||
self.assertEqual(result["positivePrompt"], "a red fox in snow")
|
||||
self.assertEqual(result["negativePrompt"], "lowres")
|
||||
self.assertEqual(
|
||||
(result["seed"], result["steps"], result["cfgScale"], result["sampler"]),
|
||||
(42, 8, 3.5, "dpmpp_2m"),
|
||||
)
|
||||
|
||||
def test_a_guider_without_prompt_inputs_leaves_prompts_unknown(self):
|
||||
result = run("custom-sampler-basic-guider")
|
||||
self.assertEqual(result["positivePrompt"], PLACEHOLDERS["positivePrompt"])
|
||||
# KSamplerSelect still matches the loose /ksampler/ settings lookup.
|
||||
self.assertEqual(result["sampler"], "dpmpp_2m")
|
||||
self.assertEqual(result["seed"], PLACEHOLDERS["seed"])
|
||||
|
||||
def test_no_sampler_reads_loose_settings(self):
|
||||
result = run("no-sampler-loose-settings")
|
||||
self.assertEqual(result["positivePrompt"], PLACEHOLDERS["positivePrompt"])
|
||||
self.assertEqual(
|
||||
(result["seed"], result["steps"], result["cfgScale"], result["sampler"]),
|
||||
(9, 12, 2, "ddim"),
|
||||
)
|
||||
self.assertEqual(result["checkpoint"], "example.safetensors")
|
||||
|
||||
def test_linked_seed_prefers_the_input_of_the_same_name(self):
|
||||
self.assertEqual(run("primitive-seed")["seed"], 77)
|
||||
|
||||
|
||||
class ReadsTheWorkflow(unittest.TestCase):
|
||||
def test_workflow_only_file(self):
|
||||
self.assertEqual(
|
||||
run("workflow-only"),
|
||||
{
|
||||
"positivePrompt": "a tidepool at noon",
|
||||
"negativePrompt": "noisy",
|
||||
"checkpoint": "example.safetensors",
|
||||
"steps": 25,
|
||||
"cfgScale": 6.5,
|
||||
"sampler": "dpmpp_2m",
|
||||
"seed": 314,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
class StaysBounded(unittest.TestCase):
|
||||
def test_cycles_terminate(self):
|
||||
result = run("conditioning-cycle")
|
||||
self.assertEqual(result["positivePrompt"], PLACEHOLDERS["positivePrompt"])
|
||||
|
||||
def test_long_chains_resolve_within_the_visit_limit(self):
|
||||
self.assertEqual(
|
||||
run("chain-within-visit-limit")["positivePrompt"], "deep chain text"
|
||||
)
|
||||
|
||||
def test_chains_beyond_the_visit_limit_stop_without_recursion_errors(self):
|
||||
result = run("chain-beyond-visit-limit")
|
||||
self.assertEqual(result["positivePrompt"], PLACEHOLDERS["positivePrompt"])
|
||||
self.assertEqual(result["negativePrompt"], "deep chain text")
|
||||
|
||||
def test_the_visit_limit_boundary_matches_the_browser_parser(self):
|
||||
# 499 pass-through nodes and the encoder are exactly 500 visits.
|
||||
self.assertEqual(
|
||||
run("chain-at-last-visit")["positivePrompt"], "deep chain text"
|
||||
)
|
||||
self.assertEqual(
|
||||
run("chain-one-past-last-visit")["positivePrompt"],
|
||||
PLACEHOLDERS["positivePrompt"],
|
||||
)
|
||||
|
||||
def test_very_long_chains_do_not_exhaust_the_python_stack(self):
|
||||
graph = {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "t"}}}
|
||||
for index in range(5000):
|
||||
graph[str(10 + index)] = {
|
||||
"class_type": "P",
|
||||
"inputs": {"conditioning": [str(9 + index if index else 1), 0]},
|
||||
}
|
||||
graph["2"] = {
|
||||
"class_type": "KSampler",
|
||||
"inputs": {
|
||||
"positive": ["5009", 0],
|
||||
"negative": ["1", 0],
|
||||
"model": ["1", 0],
|
||||
},
|
||||
}
|
||||
result = extract_generation_params({"prompt": graph})
|
||||
self.assertEqual(result["positivePrompt"], PLACEHOLDERS["positivePrompt"])
|
||||
self.assertEqual(result["negativePrompt"], "t")
|
||||
|
||||
|
||||
class Placeholders(unittest.TestCase):
|
||||
def test_blank_text_is_a_placeholder_and_zero_seed_is_kept(self):
|
||||
result = run("blank-text-and-zero-seed")
|
||||
self.assertEqual(result["positivePrompt"], PLACEHOLDERS["positivePrompt"])
|
||||
self.assertEqual(result["negativePrompt"], "kept negative")
|
||||
self.assertEqual(result["seed"], 0)
|
||||
|
||||
def test_nothing_recorded(self):
|
||||
self.assertEqual(run("nothing"), dict(PLACEHOLDERS))
|
||||
self.assertEqual(run("unparsable-records"), dict(PLACEHOLDERS))
|
||||
|
||||
|
||||
class MalformedInput(unittest.TestCase):
|
||||
def test_a_null_workflow_node_fails_like_the_browser_parser(self):
|
||||
with self.assertRaises(GenerationParamsError):
|
||||
run("null-workflow-node")
|
||||
|
||||
|
||||
class ParsesLikeJavaScript(unittest.TestCase):
|
||||
def test_rejects_non_standard_numbers(self):
|
||||
for text in ('{"a": NaN}', '{"a": Infinity}', "[-Infinity]", "not json", ""):
|
||||
with self.subTest(text=text):
|
||||
self.assertIsNone(parse_js_json(text))
|
||||
|
||||
def test_large_integers_round_like_doubles(self):
|
||||
self.assertEqual(
|
||||
parse_js_json('{"seed": 9007199254740993}'), {"seed": 9007199254740992.0}
|
||||
)
|
||||
self.assertEqual(
|
||||
parse_js_json('{"seed": 9007199254740991}'), {"seed": 9007199254740991}
|
||||
)
|
||||
|
||||
def test_seeds_beyond_double_precision_round_like_the_browser(self):
|
||||
self.assertEqual(
|
||||
run("seed-beyond-double-precision")["seed"], 4974052037430000001.0
|
||||
)
|
||||
|
||||
def test_duplicate_keys_keep_the_last_value(self):
|
||||
self.assertEqual(parse_js_json('{"a": 1, "a": 2}'), {"a": 2})
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,637 @@
|
||||
"""Randomized differential test: the Python parser equals the browser parser.
|
||||
|
||||
Generates seeded random API graphs and UI workflows, from realistic to
|
||||
malformed, and requires utils.generation_params to return exactly what
|
||||
web/js/comfy-metadata.js extractGenerationParams returns, or to fail where it
|
||||
throws. Runs only where Node is available. Set PM_PARAMS_FUZZ_CASES to run more.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
try:
|
||||
from tests.test_generation_params_js_parity import (
|
||||
NODE,
|
||||
js_equal,
|
||||
run_oracle,
|
||||
run_python,
|
||||
) # noqa: E402
|
||||
except ImportError: # discovered with tests/ as the top-level directory
|
||||
from test_generation_params_js_parity import (
|
||||
NODE,
|
||||
js_equal,
|
||||
run_oracle,
|
||||
run_python,
|
||||
) # noqa: E402
|
||||
|
||||
CASE_COUNT = int(os.environ.get("PM_PARAMS_FUZZ_CASES", "3000"))
|
||||
SEEDS = (20261020, 7, 424242)
|
||||
|
||||
NODE_IDS = [
|
||||
"1",
|
||||
"2",
|
||||
"3",
|
||||
"4",
|
||||
"5",
|
||||
"6",
|
||||
"7",
|
||||
"9",
|
||||
"10",
|
||||
"12",
|
||||
"25",
|
||||
"26",
|
||||
"100",
|
||||
"1300:5",
|
||||
"1300:8",
|
||||
"x",
|
||||
"05",
|
||||
"4294967295",
|
||||
"-1",
|
||||
"constructor",
|
||||
]
|
||||
CLASS_TYPES = [
|
||||
"KSampler",
|
||||
"KSamplerAdvanced",
|
||||
"SamplerCustomAdvanced",
|
||||
"CFGGuider",
|
||||
"BasicGuider",
|
||||
"CLIPTextEncode",
|
||||
"PromptManager",
|
||||
"PromptManagerPair",
|
||||
"PromptManagerText",
|
||||
"StringConcatenate",
|
||||
"PrimitiveString",
|
||||
"ControlNetApplyAdvanced",
|
||||
"ConditioningCombine",
|
||||
"CheckpointLoaderSimple",
|
||||
"UNETLoader",
|
||||
"LoraLoader",
|
||||
"RandomNoise",
|
||||
"KSamplerSelect",
|
||||
"BasicScheduler",
|
||||
"Reroute",
|
||||
"MyKSamplerThing",
|
||||
"SaveImage",
|
||||
"FaceDetailer",
|
||||
"toString",
|
||||
"",
|
||||
]
|
||||
INPUT_KEYS = [
|
||||
"text",
|
||||
"positive",
|
||||
"negative",
|
||||
"model",
|
||||
"conditioning",
|
||||
"conditioning_1",
|
||||
"conditioning_to",
|
||||
"guider",
|
||||
"sigmas",
|
||||
"noise",
|
||||
"sampler",
|
||||
"seed",
|
||||
"noise_seed",
|
||||
"steps",
|
||||
"cfg",
|
||||
"sampler_name",
|
||||
"ckpt_name",
|
||||
"unet_name",
|
||||
"model_name",
|
||||
"prepend_text",
|
||||
"append_text",
|
||||
"negative_text",
|
||||
"delimiter",
|
||||
"string_a",
|
||||
"string_b",
|
||||
"value",
|
||||
"text_g",
|
||||
"prompt",
|
||||
"0",
|
||||
"10",
|
||||
"clip",
|
||||
]
|
||||
TEXTS = [
|
||||
"a cat",
|
||||
" padded text ",
|
||||
"",
|
||||
" ",
|
||||
"bom text ",
|
||||
"line one\nline two",
|
||||
"masterpiece",
|
||||
"masterpiece, a dog",
|
||||
"4k",
|
||||
"a dog 4k",
|
||||
"ünïcödé ✓",
|
||||
", ",
|
||||
]
|
||||
WORKFLOW_TYPES = [
|
||||
"KSampler",
|
||||
"KSamplerAdvanced",
|
||||
"CLIPTextEncode",
|
||||
"PromptManager",
|
||||
"PromptManagerPair",
|
||||
"CheckpointLoaderSimple",
|
||||
"CheckpointLoader",
|
||||
"UNETLoader",
|
||||
"MyUnetLoader",
|
||||
"Reroute",
|
||||
"GetNode",
|
||||
"SetNode",
|
||||
"Note",
|
||||
"toString",
|
||||
"valueOf",
|
||||
]
|
||||
|
||||
|
||||
class Generator:
|
||||
def __init__(self, seed):
|
||||
self.random = random.Random(seed)
|
||||
|
||||
def pick(self, items):
|
||||
return self.random.choice(items)
|
||||
|
||||
def chance(self, probability):
|
||||
return self.random.random() < probability
|
||||
|
||||
def number(self):
|
||||
return self.pick(
|
||||
[
|
||||
0,
|
||||
1,
|
||||
2,
|
||||
3,
|
||||
5,
|
||||
20,
|
||||
7,
|
||||
1.0,
|
||||
1.5,
|
||||
6.5,
|
||||
0.45,
|
||||
1.5e-7,
|
||||
2.5e21,
|
||||
-3,
|
||||
9007199254740991,
|
||||
]
|
||||
)
|
||||
|
||||
def link(self):
|
||||
return [
|
||||
self.pick(NODE_IDS + [3, 12, 5.0]),
|
||||
self.pick([0, 1, 2, 3, 1.0, 0, 0, 0, -1]),
|
||||
]
|
||||
|
||||
def value(self, key):
|
||||
roll = self.random.random()
|
||||
if (
|
||||
key
|
||||
in (
|
||||
"positive",
|
||||
"negative",
|
||||
"model",
|
||||
"conditioning",
|
||||
"conditioning_1",
|
||||
"conditioning_to",
|
||||
"guider",
|
||||
"sigmas",
|
||||
"noise",
|
||||
"sampler",
|
||||
"clip",
|
||||
)
|
||||
and roll < 0.8
|
||||
):
|
||||
return self.link()
|
||||
if roll < 0.3:
|
||||
return self.link()
|
||||
if roll < 0.6:
|
||||
return self.pick(TEXTS)
|
||||
if roll < 0.85:
|
||||
return self.number()
|
||||
return self.pick(
|
||||
[True, False, None, [1, 2, 3], {"nested": "x"}, [], "randomize"]
|
||||
)
|
||||
|
||||
def inputs(self):
|
||||
if self.chance(0.04):
|
||||
return self.pick([[], ["text"], "text", 3, None, True])
|
||||
keys = self.random.sample(INPUT_KEYS, self.random.randint(0, 9))
|
||||
return {key: self.value(key) for key in keys}
|
||||
|
||||
def node(self):
|
||||
if self.chance(0.03):
|
||||
return self.pick([None, "node", 5, [1, 2]])
|
||||
node = {"class_type": self.pick(CLASS_TYPES), "inputs": self.inputs()}
|
||||
if self.chance(0.05):
|
||||
del node["class_type"]
|
||||
return node
|
||||
|
||||
def graph(self):
|
||||
ids = self.random.sample(NODE_IDS, self.random.randint(1, 12))
|
||||
if self.chance(0.03):
|
||||
return [self.node() for _ in ids]
|
||||
return {node_id: self.node() for node_id in ids}
|
||||
|
||||
def workflow_node(self, node_id):
|
||||
if self.chance(0.02):
|
||||
return self.pick([None, 3, "node"])
|
||||
node = {
|
||||
"id": node_id,
|
||||
"type": self.pick(WORKFLOW_TYPES),
|
||||
"mode": self.pick([0, 0, 0, 2, 4]),
|
||||
}
|
||||
inputs = []
|
||||
for _ in range(self.random.randint(0, 4)):
|
||||
item = {
|
||||
"name": self.pick(INPUT_KEYS),
|
||||
"link": self.pick([None, 1, 2, 3, 4, 5, 6, 7, 8, "2"]),
|
||||
"type": self.pick(["CONDITIONING", "MODEL", "CLIP", "*", "STRING"]),
|
||||
}
|
||||
if self.chance(0.2):
|
||||
item["widget"] = {"name": self.pick(INPUT_KEYS)}
|
||||
inputs.append(item)
|
||||
node["inputs"] = (
|
||||
inputs if not self.chance(0.03) else self.pick([{"a": 1}, "abc", None])
|
||||
)
|
||||
node["widgets_values"] = [
|
||||
self.value(self.pick(INPUT_KEYS)) for _ in range(self.random.randint(0, 6))
|
||||
]
|
||||
if self.chance(0.1):
|
||||
node["widgets_values"].insert(1, "randomize")
|
||||
if self.chance(0.05):
|
||||
node["widgets_values_named"] = {
|
||||
"text": self.pick(TEXTS),
|
||||
"seed": self.number(),
|
||||
}
|
||||
return node
|
||||
|
||||
def workflow(self):
|
||||
if self.chance(0.03):
|
||||
return self.pick([[], "workflow", {"nodes": "x"}, {"links": []}])
|
||||
ids = self.random.sample(
|
||||
[1, 2, 3, 4, 5, 6, 7, 8, "9", 10.0], self.random.randint(1, 8)
|
||||
)
|
||||
links = []
|
||||
for link_id in range(1, self.random.randint(1, 9)):
|
||||
origin, slot = self.pick(ids), self.pick([0, 1, 2])
|
||||
if self.chance(0.5):
|
||||
links.append(
|
||||
[
|
||||
link_id,
|
||||
origin,
|
||||
slot,
|
||||
self.pick(ids),
|
||||
0,
|
||||
self.pick(["CONDITIONING", "MODEL", "*", "STRING"]),
|
||||
]
|
||||
)
|
||||
else:
|
||||
links.append(
|
||||
{
|
||||
"id": link_id,
|
||||
"origin_id": origin,
|
||||
"origin_slot": slot,
|
||||
"type": self.pick(["CONDITIONING", "MODEL", "*"]),
|
||||
}
|
||||
)
|
||||
return {
|
||||
"nodes": [self.workflow_node(node_id) for node_id in ids],
|
||||
"links": links,
|
||||
}
|
||||
|
||||
# --- coherent pipelines, then corrupted ------------------------------------
|
||||
|
||||
def text_source(self, graph, ids, depth=0):
|
||||
"""A node id producing text: an encoder, PromptManager, pair or a text chain."""
|
||||
node_id = ids.pop()
|
||||
kind = self.pick(
|
||||
["encoder", "manager", "pair", "concat", "primitive"]
|
||||
if depth < 2
|
||||
else ["encoder", "manager"]
|
||||
)
|
||||
if kind == "encoder":
|
||||
text = (
|
||||
self.pick(TEXTS)
|
||||
if self.chance(0.7)
|
||||
else [self.text_source(graph, ids, depth + 1), 0]
|
||||
)
|
||||
graph[node_id] = {
|
||||
"class_type": "CLIPTextEncode",
|
||||
"inputs": {"text": text, "clip": ["4", 1]},
|
||||
}
|
||||
elif kind == "manager":
|
||||
graph[node_id] = {
|
||||
"class_type": self.pick(["PromptManager", "PromptManagerText"]),
|
||||
"inputs": {
|
||||
"text": self.pick(TEXTS),
|
||||
"prepend_text": self.pick(TEXTS + ["masterpiece"]),
|
||||
"append_text": self.pick(TEXTS + ["4k"]),
|
||||
"clip": ["4", 1],
|
||||
},
|
||||
}
|
||||
elif kind == "pair":
|
||||
graph[node_id] = {
|
||||
"class_type": "PromptManagerPair",
|
||||
"inputs": {
|
||||
"text": self.pick(TEXTS),
|
||||
"negative_text": self.pick(TEXTS),
|
||||
"prepend_text": "",
|
||||
"append_text": "",
|
||||
},
|
||||
}
|
||||
elif kind == "concat":
|
||||
graph[node_id] = {
|
||||
"class_type": "StringConcatenate",
|
||||
"inputs": {
|
||||
"string_a": (
|
||||
[self.text_source(graph, ids, depth + 1), 0] if ids else "x"
|
||||
),
|
||||
"string_b": self.pick(TEXTS),
|
||||
"delimiter": self.pick([", ", " ", 3, "|"]),
|
||||
},
|
||||
}
|
||||
else:
|
||||
graph[node_id] = {
|
||||
"class_type": "PrimitiveString",
|
||||
"inputs": {"value": self.pick(TEXTS)},
|
||||
}
|
||||
return node_id
|
||||
|
||||
def conditioning(self, graph, ids, role):
|
||||
source = self.text_source(graph, ids)
|
||||
slot = (
|
||||
self.pick([0, 1, 2, 3])
|
||||
if graph[source]["class_type"] == "PromptManagerPair"
|
||||
else 0
|
||||
)
|
||||
link = [source, slot]
|
||||
for _ in range(self.pick([0, 0, 1, 2, 3])):
|
||||
if not ids:
|
||||
break
|
||||
node_id = ids.pop()
|
||||
key = self.pick([role, "conditioning", "conditioning_to", "conditioning_1"])
|
||||
graph[node_id] = {
|
||||
"class_type": self.pick(
|
||||
[
|
||||
"ControlNetApplyAdvanced",
|
||||
"ConditioningCombine",
|
||||
"ConditioningSetArea",
|
||||
]
|
||||
),
|
||||
"inputs": {key: link, "strength": 1.0},
|
||||
}
|
||||
link = [node_id, 0]
|
||||
return link
|
||||
|
||||
def pipeline(self):
|
||||
ids = [str(i) for i in self.random.sample(range(1, 400), 120)] + [
|
||||
"1300:7",
|
||||
"x",
|
||||
"constructor",
|
||||
]
|
||||
self.random.shuffle(ids)
|
||||
graph = {
|
||||
"4": {
|
||||
"class_type": self.pick(["CheckpointLoaderSimple", "UNETLoader"]),
|
||||
"inputs": {
|
||||
self.pick(
|
||||
["ckpt_name", "unet_name", "gguf_name"]
|
||||
): "model.safetensors"
|
||||
},
|
||||
}
|
||||
}
|
||||
for _ in range(self.pick([1, 1, 2, 3])):
|
||||
positive, negative = self.conditioning(
|
||||
graph, ids, "positive"
|
||||
), self.conditioning(graph, ids, "negative")
|
||||
settings = {
|
||||
"seed": self.number() if self.chance(0.8) else [ids.pop(), 0],
|
||||
"steps": self.number(),
|
||||
"cfg": self.number(),
|
||||
"sampler_name": self.pick(["euler", "lcm", "dpmpp_2m"]),
|
||||
}
|
||||
if settings["seed"] and isinstance(settings["seed"], list):
|
||||
graph[settings["seed"][0]] = {
|
||||
"class_type": "Seed",
|
||||
"inputs": {"seed": self.number(), "other": "x"},
|
||||
}
|
||||
if self.chance(0.25):
|
||||
guider = ids.pop()
|
||||
graph[guider] = {
|
||||
"class_type": "CFGGuider",
|
||||
"inputs": {
|
||||
"model": ["4", 0],
|
||||
"positive": positive,
|
||||
"negative": negative,
|
||||
"cfg": self.number(),
|
||||
},
|
||||
}
|
||||
noise, chooser, sigmas = ids.pop(), ids.pop(), ids.pop()
|
||||
graph[noise] = {
|
||||
"class_type": "RandomNoise",
|
||||
"inputs": {"noise_seed": self.number()},
|
||||
}
|
||||
graph[chooser] = {
|
||||
"class_type": "KSamplerSelect",
|
||||
"inputs": {"sampler_name": "euler"},
|
||||
}
|
||||
graph[sigmas] = {
|
||||
"class_type": "BasicScheduler",
|
||||
"inputs": {"steps": self.number()},
|
||||
}
|
||||
graph[ids.pop()] = {
|
||||
"class_type": "SamplerCustomAdvanced",
|
||||
"inputs": {
|
||||
"noise": [noise, 0],
|
||||
"guider": [guider, 0],
|
||||
"sampler": [chooser, 0],
|
||||
"sigmas": [sigmas, 0],
|
||||
},
|
||||
}
|
||||
else:
|
||||
graph[ids.pop()] = {
|
||||
"class_type": self.pick(
|
||||
["KSampler", "KSamplerAdvanced", "FaceDetailer"]
|
||||
),
|
||||
"inputs": {
|
||||
"model": ["4", 0],
|
||||
"positive": positive,
|
||||
"negative": negative,
|
||||
**settings,
|
||||
},
|
||||
}
|
||||
for _ in range(self.pick([0, 0, 1, 3])): # corruption
|
||||
node_id = self.pick(list(graph))
|
||||
node = graph[node_id]
|
||||
if (
|
||||
self.chance(0.5)
|
||||
and isinstance(node, dict)
|
||||
and isinstance(node.get("inputs"), dict)
|
||||
and node["inputs"]
|
||||
):
|
||||
key = self.pick(list(graph[node_id]["inputs"]))
|
||||
graph[node_id]["inputs"][key] = self.value(key)
|
||||
else:
|
||||
graph[node_id] = self.node()
|
||||
return graph
|
||||
|
||||
def pipeline_workflow(self):
|
||||
"""A connected workflow: loader, encoders, Reroute or Set/Get hops, sampler."""
|
||||
nodes, links = [], []
|
||||
|
||||
def link(origin, slot, kind):
|
||||
links.append(
|
||||
[len(links) + 1, origin, slot, 0, 0, kind]
|
||||
if self.chance(0.6)
|
||||
else {
|
||||
"id": len(links) + 1,
|
||||
"origin_id": origin,
|
||||
"origin_slot": slot,
|
||||
"type": kind,
|
||||
}
|
||||
)
|
||||
return len(links)
|
||||
|
||||
nodes.append(
|
||||
{
|
||||
"id": 4,
|
||||
"type": self.pick(
|
||||
["CheckpointLoaderSimple", "UNETLoader", "MyUnetLoader"]
|
||||
),
|
||||
"mode": 0,
|
||||
"inputs": [],
|
||||
"widgets_values": ["model.safetensors"],
|
||||
}
|
||||
)
|
||||
refs = []
|
||||
for index, role in enumerate(("positive", "negative")):
|
||||
encoder = 10 + index
|
||||
nodes.append(
|
||||
{
|
||||
"id": encoder,
|
||||
"type": self.pick(
|
||||
["CLIPTextEncode", "PromptManager", "PromptManagerText"]
|
||||
),
|
||||
"mode": self.pick([0, 0, 0, 4]),
|
||||
"inputs": [
|
||||
{"name": "clip", "link": link(4, 1, "CLIP"), "type": "CLIP"}
|
||||
],
|
||||
"widgets_values": [
|
||||
self.pick(TEXTS),
|
||||
"",
|
||||
"",
|
||||
"",
|
||||
self.pick(TEXTS),
|
||||
self.pick(TEXTS),
|
||||
],
|
||||
}
|
||||
)
|
||||
ref = link(encoder, 0, "CONDITIONING")
|
||||
hop = self.pick(["none", "none", "reroute", "setget"])
|
||||
if hop == "reroute":
|
||||
nodes.append(
|
||||
{
|
||||
"id": 20 + index,
|
||||
"type": "Reroute",
|
||||
"mode": 0,
|
||||
"inputs": [{"name": "", "link": ref}],
|
||||
}
|
||||
)
|
||||
ref = link(20 + index, 0, "*")
|
||||
elif hop == "setget":
|
||||
name = self.pick(["cond_" + role, "shared"])
|
||||
nodes.append(
|
||||
{
|
||||
"id": 30 + index,
|
||||
"type": "SetNode",
|
||||
"mode": 0,
|
||||
"inputs": [{"name": "c", "link": ref}],
|
||||
"widgets_values": [name],
|
||||
}
|
||||
)
|
||||
nodes.append(
|
||||
{
|
||||
"id": 40 + index,
|
||||
"type": "GetNode",
|
||||
"mode": 0,
|
||||
"inputs": [],
|
||||
"widgets_values": [name],
|
||||
}
|
||||
)
|
||||
ref = link(40 + index, 0, "CONDITIONING")
|
||||
refs.append(ref)
|
||||
widgets = [
|
||||
self.number(),
|
||||
self.pick(["randomize", "fixed", "x"]),
|
||||
self.number(),
|
||||
self.number(),
|
||||
"euler",
|
||||
"normal",
|
||||
1,
|
||||
]
|
||||
nodes.append(
|
||||
{
|
||||
"id": 3,
|
||||
"type": self.pick(["KSampler", "KSamplerAdvanced"]),
|
||||
"mode": 0,
|
||||
"inputs": [
|
||||
{"name": "model", "link": link(4, 0, "MODEL"), "type": "MODEL"},
|
||||
{"name": "positive", "link": refs[0], "type": "CONDITIONING"},
|
||||
{"name": "negative", "link": refs[1], "type": "CONDITIONING"},
|
||||
],
|
||||
"widgets_values": widgets,
|
||||
}
|
||||
)
|
||||
if self.chance(0.3):
|
||||
nodes[self.random.randrange(len(nodes))] = self.workflow_node(
|
||||
self.pick([1, 2, 99])
|
||||
)
|
||||
return {"nodes": nodes, "links": links}
|
||||
|
||||
def case(self):
|
||||
roll = self.random.random()
|
||||
structured = self.chance(0.7)
|
||||
prompt = (
|
||||
json.dumps(self.pipeline() if structured else self.graph())
|
||||
if roll < 0.75
|
||||
else None
|
||||
)
|
||||
workflow = (
|
||||
json.dumps(self.pipeline_workflow() if structured else self.workflow())
|
||||
if roll > 0.45
|
||||
else None
|
||||
)
|
||||
if self.chance(0.02):
|
||||
prompt = self.pick(['{"1": NaN}', "{", "null", "[]", '"text"', "12"])
|
||||
return {"prompt": prompt, "workflow": workflow}
|
||||
|
||||
|
||||
@unittest.skipUnless(NODE, "Node is required to run the browser parser")
|
||||
class RandomGraphs(unittest.TestCase):
|
||||
def test_random_graphs_match_the_browser_parser(self):
|
||||
for seed in SEEDS:
|
||||
generator = Generator(seed)
|
||||
cases = [generator.case() for _ in range(CASE_COUNT)]
|
||||
expected = run_oracle(cases)
|
||||
self.assertEqual(len(expected), len(cases))
|
||||
mismatches = []
|
||||
for index, (case, result) in enumerate(zip(cases, expected)):
|
||||
actual = run_python(case)
|
||||
if not js_equal(actual, result):
|
||||
mismatches.append((index, case, actual, result))
|
||||
with self.subTest(seed=seed):
|
||||
outcomes = {
|
||||
"ok": sum(r["ok"] for r in expected),
|
||||
"throws": sum(not r["ok"] for r in expected),
|
||||
}
|
||||
self.assertEqual(
|
||||
mismatches[:3],
|
||||
[],
|
||||
f"{len(mismatches)} of {len(cases)} differ; outcomes {outcomes}",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,155 @@
|
||||
"""The Python generation-parameter parser equals the browser parser.
|
||||
|
||||
Golden vectors (tests/fixtures/generation_params/vectors.json) hold the browser
|
||||
parser's results for every named case and every PNG fixture, so this runs on
|
||||
machines without Node. With Node, the live browser parser is compared too and
|
||||
the vectors are checked for staleness. Regenerate them with:
|
||||
|
||||
python tests/test_generation_params_js_parity.py --write-vectors
|
||||
"""
|
||||
|
||||
import json
|
||||
import math
|
||||
import shutil
|
||||
import subprocess # nosec B404 - runs the repository's own Node oracle
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
sys.path.insert(0, str(ROOT))
|
||||
|
||||
from utils.generation_params import ( # noqa: E402
|
||||
GenerationParamsError,
|
||||
extract_from_records,
|
||||
)
|
||||
|
||||
try:
|
||||
from tests.generation_params_cases import CASES # noqa: E402
|
||||
except ImportError: # discovered with tests/ as the top-level directory
|
||||
from generation_params_cases import CASES # noqa: E402
|
||||
|
||||
VECTORS = ROOT / "tests" / "fixtures" / "generation_params" / "vectors.json"
|
||||
ORACLE = ROOT / "tests" / "js" / "support" / "generation-params-oracle.mjs"
|
||||
FIXTURE_DIRS = [ROOT / "tests" / "fixtures"]
|
||||
NODE = shutil.which("node")
|
||||
|
||||
|
||||
def fixture_cases():
|
||||
"""Text records of every PNG fixture, keyed by its repository path."""
|
||||
from PIL import Image
|
||||
|
||||
cases = {}
|
||||
for directory in FIXTURE_DIRS:
|
||||
for path in sorted(directory.rglob("*.png")):
|
||||
with Image.open(path) as image:
|
||||
info = {k: v for k, v in image.info.items() if isinstance(v, str)}
|
||||
if "prompt" in info or "workflow" in info:
|
||||
name = path.relative_to(ROOT).as_posix()
|
||||
cases[name] = {
|
||||
"prompt": info.get("prompt"),
|
||||
"workflow": info.get("workflow"),
|
||||
}
|
||||
return cases
|
||||
|
||||
|
||||
def all_cases():
|
||||
return {**{f"case:{k}": v for k, v in CASES.items()}, **fixture_cases()}
|
||||
|
||||
|
||||
def run_oracle(cases):
|
||||
"""Browser parser results for case records, in order."""
|
||||
lines = "".join(json.dumps(case) + "\n" for case in cases)
|
||||
completed = subprocess.run( # nosec B603 - fixed argv, no shell
|
||||
[NODE, str(ORACLE)],
|
||||
input=lines,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
check=True,
|
||||
timeout=600,
|
||||
)
|
||||
return [json.loads(line) for line in completed.stdout.splitlines()]
|
||||
|
||||
|
||||
def run_python(case):
|
||||
try:
|
||||
return {
|
||||
"ok": True,
|
||||
"result": extract_from_records(case["prompt"], case["workflow"]),
|
||||
}
|
||||
except GenerationParamsError:
|
||||
return {"ok": False}
|
||||
|
||||
|
||||
def js_equal(left, right):
|
||||
"""Equality as JavaScript values: 1 == 1.0, but true is not 1."""
|
||||
if isinstance(left, bool) or isinstance(right, bool):
|
||||
return isinstance(left, bool) and isinstance(right, bool) and left == right
|
||||
if isinstance(left, (int, float)) and isinstance(right, (int, float)):
|
||||
# Every JavaScript number is a double; compare as doubles.
|
||||
left, right = float(left), float(right)
|
||||
return left == right or (math.isnan(left) and math.isnan(right))
|
||||
if isinstance(left, list) and isinstance(right, list):
|
||||
return len(left) == len(right) and all(
|
||||
js_equal(a, b) for a, b in zip(left, right)
|
||||
)
|
||||
if isinstance(left, dict) and isinstance(right, dict):
|
||||
return left.keys() == right.keys() and all(
|
||||
js_equal(left[k], right[k]) for k in left
|
||||
)
|
||||
return type(left) is type(right) and left == right
|
||||
|
||||
|
||||
def write_vectors():
|
||||
cases = all_cases()
|
||||
results = run_oracle(cases.values())
|
||||
vectors = [
|
||||
{"name": name, **case, "expected": result}
|
||||
for (name, case), result in zip(cases.items(), results)
|
||||
]
|
||||
VECTORS.parent.mkdir(parents=True, exist_ok=True)
|
||||
VECTORS.write_text(json.dumps(vectors, indent=1, sort_keys=True) + "\n", "utf-8")
|
||||
print(f"wrote {len(vectors)} vectors to {VECTORS.relative_to(ROOT)}")
|
||||
|
||||
|
||||
class GoldenVectors(unittest.TestCase):
|
||||
def test_every_vector_matches(self):
|
||||
vectors = json.loads(VECTORS.read_text("utf-8"))
|
||||
self.assertGreater(len(vectors), len(CASES))
|
||||
for vector in vectors:
|
||||
with self.subTest(vector=vector["name"]):
|
||||
actual = run_python(vector)
|
||||
self.assertTrue(
|
||||
js_equal(actual, vector["expected"]), (actual, vector["expected"])
|
||||
)
|
||||
|
||||
def test_vectors_cover_every_case_and_fixture(self):
|
||||
names = {vector["name"] for vector in json.loads(VECTORS.read_text("utf-8"))}
|
||||
self.assertEqual(names, set(all_cases()))
|
||||
|
||||
|
||||
@unittest.skipUnless(NODE, "Node is required to run the browser parser")
|
||||
class LiveBrowserParser(unittest.TestCase):
|
||||
def test_live_results_match_python_and_the_vectors(self):
|
||||
cases = all_cases()
|
||||
stored = {
|
||||
vector["name"]: vector["expected"]
|
||||
for vector in json.loads(VECTORS.read_text("utf-8"))
|
||||
}
|
||||
for (name, case), expected in zip(cases.items(), run_oracle(cases.values())):
|
||||
with self.subTest(case=name):
|
||||
self.assertTrue(
|
||||
js_equal(run_python(case), expected), (run_python(case), expected)
|
||||
)
|
||||
self.assertTrue(
|
||||
js_equal(stored[name], expected),
|
||||
"vectors are stale: regenerate them",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if "--write-vectors" in sys.argv:
|
||||
write_vectors()
|
||||
else:
|
||||
unittest.main()
|
||||
@@ -8,6 +8,7 @@ leaves it unresolved, while the inspection profile resolves it.
|
||||
|
||||
import logging
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
from utils import prompt_pair_resolver as resolver
|
||||
@@ -72,12 +73,19 @@ class TestIngestionKeepsDefaultProfile(unittest.TestCase):
|
||||
|
||||
mixin = admin.AdminRoutesMixin.__new__(admin.AdminRoutesMixin)
|
||||
mixin.db = Mock()
|
||||
mixin.db.get_prompt_by_hash.return_value = None
|
||||
mixin.db.save_prompt.return_value = 7
|
||||
spy = _Spy()
|
||||
with patch.object(admin, "resolve_prompt_pair", spy):
|
||||
status = mixin._ingest_scanned_file("/out/f03.png", _f03_metadata())
|
||||
self.assertEqual(status, "unresolved")
|
||||
status = mixin._ingest_scanned_file(Path("/out/f03.png"), _f03_metadata())
|
||||
spy.assert_default_and_gated(self)
|
||||
mixin.db.assert_not_called()
|
||||
# The resolver leaves f03 unresolved; the scan links the prompt the
|
||||
# browser parser traces, as the inspector shows it.
|
||||
self.assertEqual(status, "added")
|
||||
self.assertEqual(
|
||||
mixin.db.save_prompt.call_args.args[0], "a red fox in snow, detailed fur"
|
||||
)
|
||||
mixin.db.link_image_to_prompt.assert_called_once()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+67
-9
@@ -17,6 +17,14 @@ def graph(positive="cat", negative="blur"):
|
||||
}
|
||||
|
||||
|
||||
def sampled_graph(positive="cat", negative="blur"):
|
||||
"""A KSampler with a model link, as real workflows have."""
|
||||
g = graph(positive, negative)
|
||||
g["4"] = {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "m"}}
|
||||
g["3"]["inputs"]["model"] = ["4", 0]
|
||||
return g
|
||||
|
||||
|
||||
class TestPairScan(AdminAPITestCase):
|
||||
def test_pair_scan_repeated_is_unchanged_and_empty_negative_is_supported(self):
|
||||
path = self.output_dir / "pair.png"
|
||||
@@ -28,16 +36,66 @@ class TestPairScan(AdminAPITestCase):
|
||||
self.assertEqual(pairs[0]["negative_text"], "")
|
||||
self.assertEqual(pairs[0]["run_count"], 0)
|
||||
|
||||
def test_ambiguous_sampler_scan_does_not_save_a_guess(self):
|
||||
g = graph()
|
||||
g["4"] = {**g["3"]}
|
||||
meta = {"prompt": json.dumps(g)}
|
||||
self.assertEqual(
|
||||
self.api._ingest_scanned_file(self.output_dir / "multi.png", meta),
|
||||
"ambiguous",
|
||||
)
|
||||
def _scanned(self, name, prompt_graph):
|
||||
meta = {"prompt": json.dumps(prompt_graph)}
|
||||
return self.api._ingest_scanned_file(self.output_dir / name, meta)
|
||||
|
||||
def _texts(self):
|
||||
return [p["text"] for p in self.api.db.get_recent_prompts()["prompts"]]
|
||||
|
||||
def test_samplers_sharing_one_prompt_link_that_prompt(self):
|
||||
g = sampled_graph()
|
||||
g["5"] = {**g["3"]}
|
||||
self.assertEqual(self._scanned("multi.png", g), "added")
|
||||
self.assertEqual(self._texts(), ["cat"])
|
||||
|
||||
def test_samplers_with_different_prompts_link_the_first_sampler_prompt(self):
|
||||
# The prompt the inspector and the legacy panels show for this file.
|
||||
g = sampled_graph()
|
||||
g["6"] = {"class_type": "CLIPTextEncode", "inputs": {"text": "dog"}}
|
||||
g["12"] = {**g["3"], "inputs": {**g["3"]["inputs"], "positive": ["6", 0]}}
|
||||
self.assertEqual(self._scanned("different.png", g), "added")
|
||||
self.assertEqual(self._texts(), ["cat"])
|
||||
|
||||
def test_an_unsupported_pass_through_still_links_the_traced_prompt(self):
|
||||
g = sampled_graph()
|
||||
g["7"] = {
|
||||
"class_type": "MyConditioningTweak",
|
||||
"inputs": {"conditioning": ["1", 0]},
|
||||
}
|
||||
g["3"]["inputs"]["positive"] = ["7", 0]
|
||||
g["5"] = {**g["3"]}
|
||||
self.assertEqual(self._scanned("tweak.png", g), "added")
|
||||
self.assertEqual(self._texts(), ["cat"])
|
||||
|
||||
def test_a_graph_without_a_traceable_prompt_saves_nothing(self):
|
||||
g = sampled_graph()
|
||||
g["5"] = {**g["3"]}
|
||||
for sampler in ("3", "5"):
|
||||
g[sampler]["inputs"]["positive"] = ["9", 0]
|
||||
g["9"] = {
|
||||
"class_type": "BasicGuider",
|
||||
"inputs": {"width": 2048, "height": 2048},
|
||||
}
|
||||
self.assertEqual(self._scanned("untraceable.png", g), "ambiguous")
|
||||
self.assertEqual(self.api.db.list_prompt_pairs()["total"], 0)
|
||||
self.assertEqual(self.api.db.get_recent_prompts()["prompts"], [])
|
||||
self.assertEqual(self._texts(), [])
|
||||
|
||||
def test_numeric_widgets_are_never_stored_as_a_prompt(self):
|
||||
# The pre-redesign scan stored "2048" from a resolution widget here.
|
||||
g = {
|
||||
"1": {"class_type": "CLIPTextEncode", "inputs": {"text": ["8", 0]}},
|
||||
"8": {"class_type": "DisplayText", "inputs": {"text_size": 2048}},
|
||||
"3": {
|
||||
"class_type": "SamplerCustomAdvanced",
|
||||
"inputs": {"guider": ["4", 0], "sigmas": ["5", 0]},
|
||||
},
|
||||
"4": {"class_type": "BasicGuider", "inputs": {"conditioning": ["1", 0]}},
|
||||
}
|
||||
self.assertIn(
|
||||
self._scanned("numbers.png", g), ("ambiguous", "unresolved", "found")
|
||||
)
|
||||
self.assertEqual(self._texts(), [])
|
||||
|
||||
def test_batch_exposes_pair_counts_and_actual_unchanged_status(self):
|
||||
meta = {"parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler"}
|
||||
|
||||
@@ -0,0 +1,943 @@
|
||||
"""Generation parameters traced from ComfyUI metadata: the browser parser in Python.
|
||||
|
||||
A port of ``extractGenerationParams`` in ``web/js/comfy-metadata.js``, the parser
|
||||
behind the gallery, admin and metadata pages. For the same records it returns
|
||||
what that function returns; ``tests/test_generation_params_js_parity.py``
|
||||
enforces it against the JavaScript itself.
|
||||
|
||||
It traces the API ``prompt`` graph from the sampler: the positive and negative
|
||||
inputs back to their text nodes, the model link back to a loader, and the
|
||||
sampler settings, following primitives and pass-through nodes. Fields the API
|
||||
graph cannot give are read from the UI ``workflow`` graph.
|
||||
|
||||
The result depends on JavaScript value semantics, so those are reproduced
|
||||
explicitly: property order, truthiness, ``String()``, ``trim()``, strict
|
||||
equality, ``Map``/``Set`` keys, and the inputs on which the JavaScript throws
|
||||
(raised here as ``GenerationParamsError``). The recursive tracing runs on an
|
||||
explicit stack, so a crafted graph cannot exhaust the Python stack.
|
||||
"""
|
||||
|
||||
import json
|
||||
import math
|
||||
import re
|
||||
from types import MappingProxyType
|
||||
|
||||
__all__ = [
|
||||
"PLACEHOLDERS",
|
||||
"GenerationParamsError",
|
||||
"extract_generation_params",
|
||||
"extract_from_records",
|
||||
"parse_js_json",
|
||||
]
|
||||
|
||||
PLACEHOLDERS = MappingProxyType(
|
||||
{
|
||||
"positivePrompt": "No prompt found",
|
||||
"negativePrompt": "No negative prompt found",
|
||||
"checkpoint": "Unknown",
|
||||
"steps": "Unknown",
|
||||
"cfgScale": "Unknown",
|
||||
"sampler": "Unknown",
|
||||
"seed": "Unknown",
|
||||
}
|
||||
)
|
||||
|
||||
# Upper bound on nodes expanded per lookup; crafted files can contain any graph.
|
||||
MAX_VISITS = 500
|
||||
MODEL_NAME_KEYS = ("ckpt_name", "unet_name", "model_name", "gguf_name")
|
||||
SEED_CONTROL_VALUES = frozenset(("fixed", "increment", "decrement", "randomize"))
|
||||
PASS_THROUGH_TYPES = frozenset(("Reroute", "GetNode", "SetNode"))
|
||||
MODE_MUTED = 2
|
||||
MODE_BYPASSED = 4
|
||||
# Widget order for core nodes, used when a workflow carries no widget names.
|
||||
PROMPT_MANAGER_WIDGETS = (
|
||||
"text",
|
||||
"category",
|
||||
"tags",
|
||||
"search_text",
|
||||
"prepend_text",
|
||||
"append_text",
|
||||
)
|
||||
WIDGET_POSITIONS = {
|
||||
"KSampler": ("seed", "steps", "cfg", "sampler_name", "scheduler", "denoise"),
|
||||
"KSamplerAdvanced": (
|
||||
"add_noise",
|
||||
"noise_seed",
|
||||
"steps",
|
||||
"cfg",
|
||||
"sampler_name",
|
||||
"scheduler",
|
||||
"start_at_step",
|
||||
"end_at_step",
|
||||
"return_with_leftover_noise",
|
||||
),
|
||||
"CLIPTextEncode": ("text",),
|
||||
"PromptManager": PROMPT_MANAGER_WIDGETS,
|
||||
"PromptManagerText": PROMPT_MANAGER_WIDGETS,
|
||||
"PromptManagerFocused": PROMPT_MANAGER_WIDGETS,
|
||||
"PromptManagerSplit": PROMPT_MANAGER_WIDGETS,
|
||||
"PromptManagerPair": (
|
||||
"text",
|
||||
"negative_text",
|
||||
"category",
|
||||
"tags",
|
||||
"search_text",
|
||||
"prepend_text",
|
||||
"append_text",
|
||||
),
|
||||
"CheckpointLoader": ("config_name", "ckpt_name"),
|
||||
"UNETLoader": ("unet_name", "weight_dtype"),
|
||||
}
|
||||
|
||||
# JavaScript regular expressions without the ``u`` flag fold case in ASCII only.
|
||||
_FLAGS = re.IGNORECASE | re.ASCII
|
||||
TEXT_KEYS = re.compile(r"(text|string|prompt|value)(_?[a-z0-9]+)?", _FLAGS)
|
||||
_PROMPT_MANAGER = re.compile(r"promptmanager", _FLAGS)
|
||||
_KSAMPLER = re.compile(r"ksampler", _FLAGS)
|
||||
_CHECKPOINT_LOADER = re.compile(r"checkpointloader", _FLAGS)
|
||||
_UNET = re.compile(r"unet", _FLAGS)
|
||||
_SEED_NAME = re.compile(r"seed\Z")
|
||||
_ARRAY_INDEX = re.compile(r"(?:0|[1-9][0-9]*)\Z")
|
||||
_MAX_ARRAY_INDEX = 2**32 - 2
|
||||
_MAX_SAFE_INTEGER = 2**53
|
||||
# String.prototype.trim() removes WhiteSpace and LineTerminator code points.
|
||||
_JS_WHITESPACE = "\t\n\v\f\r " "
"
|
||||
# Properties every object literal inherits; reading one yields a function or object.
|
||||
_OBJECT_PROTOTYPE_KEYS = frozenset(
|
||||
(
|
||||
"constructor",
|
||||
"hasOwnProperty",
|
||||
"isPrototypeOf",
|
||||
"propertyIsEnumerable",
|
||||
"toLocaleString",
|
||||
"toString",
|
||||
"valueOf",
|
||||
"__proto__",
|
||||
"__defineGetter__",
|
||||
"__defineSetter__",
|
||||
"__lookupGetter__",
|
||||
"__lookupSetter__",
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class GenerationParamsError(ValueError):
|
||||
"""The browser parser throws on this input, so there is no result."""
|
||||
|
||||
|
||||
class _Undefined:
|
||||
"""JavaScript ``undefined``: a missing property, distinct from JSON ``null``."""
|
||||
|
||||
__slots__ = ()
|
||||
|
||||
def __repr__(self):
|
||||
return "undefined"
|
||||
|
||||
|
||||
UNDEFINED = _Undefined()
|
||||
|
||||
|
||||
# --- JavaScript value semantics -------------------------------------------------
|
||||
|
||||
|
||||
def _is_number(value):
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
||||
|
||||
|
||||
def _nullish(value):
|
||||
return value is None or value is UNDEFINED
|
||||
|
||||
|
||||
def _truthy(value):
|
||||
if value is None or value is UNDEFINED or value is False:
|
||||
return False
|
||||
if _is_number(value):
|
||||
return value != 0 and not (isinstance(value, float) and math.isnan(value))
|
||||
if isinstance(value, str):
|
||||
return value != ""
|
||||
return True
|
||||
|
||||
|
||||
def _typeof(value):
|
||||
if value is UNDEFINED:
|
||||
return "undefined"
|
||||
if isinstance(value, bool):
|
||||
return "boolean"
|
||||
if _is_number(value):
|
||||
return "number"
|
||||
if isinstance(value, str):
|
||||
return "string"
|
||||
return "object"
|
||||
|
||||
|
||||
def _number_string(value):
|
||||
"""``String(number)`` for a double, per the ECMAScript Number::toString rules."""
|
||||
if isinstance(value, int):
|
||||
return str(value)
|
||||
if math.isnan(value):
|
||||
return "NaN"
|
||||
if math.isinf(value):
|
||||
return "Infinity" if value > 0 else "-Infinity"
|
||||
if value == 0:
|
||||
return "0"
|
||||
sign = "-" if value < 0 else ""
|
||||
text = repr(abs(value))
|
||||
mantissa, _, exponent = text.partition("e")
|
||||
whole, _, fraction = mantissa.partition(".")
|
||||
raw = whole + fraction
|
||||
leading = len(raw) - len(raw.lstrip("0"))
|
||||
digits = raw.lstrip("0").rstrip("0")
|
||||
point = len(whole) - leading + int(exponent or 0)
|
||||
count = len(digits)
|
||||
if count <= point <= 21:
|
||||
body = digits + "0" * (point - count)
|
||||
elif 0 < point <= 21:
|
||||
body = digits[:point] + "." + digits[point:]
|
||||
elif -6 < point <= 0:
|
||||
body = "0." + "0" * -point + digits
|
||||
else:
|
||||
power = point - 1
|
||||
mark = ("+" if power >= 0 else "-") + str(abs(power))
|
||||
body = (digits if count == 1 else digits[0] + "." + digits[1:]) + "e" + mark
|
||||
return sign + body
|
||||
|
||||
|
||||
def _js_string(value):
|
||||
"""``String(value)`` for any JSON-derived value."""
|
||||
if value is UNDEFINED:
|
||||
return "undefined"
|
||||
if value is None:
|
||||
return "null"
|
||||
if value is True:
|
||||
return "true"
|
||||
if value is False:
|
||||
return "false"
|
||||
if _is_number(value):
|
||||
return _number_string(value)
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
if isinstance(value, list):
|
||||
return ",".join("" if _nullish(item) else _js_string(item) for item in value)
|
||||
return "[object Object]"
|
||||
|
||||
|
||||
def _key(value):
|
||||
return value if isinstance(value, str) else _js_string(value)
|
||||
|
||||
|
||||
def _index(key):
|
||||
"""The array index a property key names, or None."""
|
||||
if (
|
||||
_is_number(key)
|
||||
and not isinstance(key, bool)
|
||||
and float(key).is_integer()
|
||||
and key >= 0
|
||||
):
|
||||
number = int(key)
|
||||
return number if number <= _MAX_ARRAY_INDEX else None
|
||||
text = _key(key)
|
||||
if _ARRAY_INDEX.match(text) and int(text) <= _MAX_ARRAY_INDEX:
|
||||
return int(text)
|
||||
return None
|
||||
|
||||
|
||||
def _get(obj, key):
|
||||
"""Property read ``obj[key]``; reading from null or undefined throws."""
|
||||
if obj is None or obj is UNDEFINED:
|
||||
raise GenerationParamsError(f"cannot read {key!r} of {_js_string(obj)}")
|
||||
if isinstance(obj, dict):
|
||||
return obj.get(_key(key), UNDEFINED)
|
||||
if isinstance(obj, (list, str)):
|
||||
if key == "length":
|
||||
return len(obj)
|
||||
index = _index(key)
|
||||
return obj[index] if index is not None and index < len(obj) else UNDEFINED
|
||||
return UNDEFINED
|
||||
|
||||
|
||||
def _has(obj, key):
|
||||
"""``key in obj``; throws on primitives."""
|
||||
if isinstance(obj, dict):
|
||||
return _key(key) in obj
|
||||
if isinstance(obj, list):
|
||||
index = _index(key)
|
||||
return key == "length" or (index is not None and index < len(obj))
|
||||
raise GenerationParamsError(f"cannot use 'in' on {_typeof(obj)}")
|
||||
|
||||
|
||||
def _own(obj, key):
|
||||
"""``Object.prototype.hasOwnProperty.call(obj, key)``."""
|
||||
if obj is None or obj is UNDEFINED:
|
||||
raise GenerationParamsError("hasOwnProperty on null or undefined")
|
||||
if isinstance(obj, dict):
|
||||
return _key(key) in obj
|
||||
if isinstance(obj, (list, str)):
|
||||
index = _index(key)
|
||||
return key == "length" or (index is not None and index < len(obj))
|
||||
return False
|
||||
|
||||
|
||||
def _ordered_keys(obj):
|
||||
"""Own property order: array indexes ascending, then other keys as inserted."""
|
||||
indexes = sorted(
|
||||
(k for k in obj if _ARRAY_INDEX.match(k) and int(k) <= _MAX_ARRAY_INDEX),
|
||||
key=int,
|
||||
)
|
||||
marked = set(indexes)
|
||||
return indexes + [k for k in obj if k not in marked]
|
||||
|
||||
|
||||
def _entries(obj):
|
||||
"""``Object.entries(obj)``."""
|
||||
if obj is None or obj is UNDEFINED:
|
||||
raise GenerationParamsError("Object.entries of null or undefined")
|
||||
if isinstance(obj, dict):
|
||||
return [(k, obj[k]) for k in _ordered_keys(obj)]
|
||||
if isinstance(obj, (list, str)):
|
||||
return [(str(i), item) for i, item in enumerate(obj)]
|
||||
return []
|
||||
|
||||
|
||||
def _values(obj):
|
||||
"""``Object.values(obj)``."""
|
||||
return [value for _, value in _entries(obj)]
|
||||
|
||||
|
||||
def _strict_equal(left, right):
|
||||
"""``left === right``."""
|
||||
if left is UNDEFINED or right is UNDEFINED or left is None or right is None:
|
||||
return left is right
|
||||
if isinstance(left, bool) or isinstance(right, bool):
|
||||
return isinstance(left, bool) and isinstance(right, bool) and left == right
|
||||
if _is_number(left) and _is_number(right):
|
||||
return left == right
|
||||
if isinstance(left, str) and isinstance(right, str):
|
||||
return left == right
|
||||
return left is right
|
||||
|
||||
|
||||
def _same_value_key(value):
|
||||
"""A hashable key with ``Map``/``Set`` (SameValueZero) equality."""
|
||||
if value is UNDEFINED:
|
||||
return ("undefined", None)
|
||||
if value is None:
|
||||
return ("null", None)
|
||||
if isinstance(value, bool):
|
||||
return ("boolean", value)
|
||||
if _is_number(value):
|
||||
return (
|
||||
("nan", None)
|
||||
if isinstance(value, float) and math.isnan(value)
|
||||
else ("number", value)
|
||||
)
|
||||
if isinstance(value, str):
|
||||
return ("string", value)
|
||||
return ("object", id(value))
|
||||
|
||||
|
||||
def _trim(text):
|
||||
return text.strip(_JS_WHITESPACE)
|
||||
|
||||
|
||||
def _first_truthy(*values):
|
||||
"""``a || b || ...``: the first truthy operand, else the last."""
|
||||
for value in values:
|
||||
if _truthy(value):
|
||||
return value
|
||||
return values[-1]
|
||||
|
||||
|
||||
def _and_chain(obj, *steps):
|
||||
"""``a && a[k1] && a[k1][k2] ...``: stops at the first falsy operand."""
|
||||
value = obj
|
||||
for step in steps:
|
||||
if not _truthy(value):
|
||||
return value
|
||||
value = _get(value, step)
|
||||
return value
|
||||
|
||||
|
||||
def _coalesce(value, fallback):
|
||||
"""``value ?? fallback()``; the fallback runs only when needed."""
|
||||
return fallback() if _nullish(value) else value
|
||||
|
||||
|
||||
def _array(value):
|
||||
"""A value used with an Array method; anything else throws."""
|
||||
if isinstance(value, list):
|
||||
return value
|
||||
raise GenerationParamsError(f"{_typeof(value)} has no array methods")
|
||||
|
||||
|
||||
def _iterate(value):
|
||||
"""``for (const x of value)``: arrays and strings iterate, other values throw."""
|
||||
if isinstance(value, (list, str)):
|
||||
return value
|
||||
raise GenerationParamsError(f"{_typeof(value)} is not iterable")
|
||||
|
||||
|
||||
def _set(obj, key, value):
|
||||
"""``obj[key] = value`` on an object literal (``__proto__`` is not an own key)."""
|
||||
name = _key(key)
|
||||
if name == "__proto__" and name not in obj:
|
||||
return
|
||||
obj[name] = value
|
||||
|
||||
|
||||
def _matches(pattern, value):
|
||||
"""``pattern.test(value || "")``."""
|
||||
return pattern.search(_js_string(value if _truthy(value) else "")) is not None
|
||||
|
||||
|
||||
def _run(generator):
|
||||
"""Drive generator-based recursion on an explicit stack, not the call stack."""
|
||||
stack = [generator]
|
||||
value = None
|
||||
while stack:
|
||||
try:
|
||||
request = stack[-1].send(value)
|
||||
except StopIteration as done:
|
||||
stack.pop()
|
||||
value = done.value
|
||||
continue
|
||||
stack.append(request)
|
||||
value = None
|
||||
return value
|
||||
|
||||
|
||||
# --- Tracing ---------------------------------------------------------------------
|
||||
|
||||
|
||||
def _is_link(value):
|
||||
return (
|
||||
isinstance(value, list)
|
||||
and len(value) == 2
|
||||
and (isinstance(value[0], str) or _is_number(value[0]))
|
||||
and _is_number(value[1])
|
||||
)
|
||||
|
||||
|
||||
def _is_sampler(node):
|
||||
"""A KSampler or a guider (CFGGuider); the model link excludes pass-throughs."""
|
||||
if not _truthy(node):
|
||||
return False
|
||||
inputs = _get(node, "inputs")
|
||||
return (
|
||||
_truthy(inputs)
|
||||
and _is_link(_get(inputs, "positive"))
|
||||
and _is_link(_get(inputs, "negative"))
|
||||
and _is_link(_get(inputs, "model"))
|
||||
)
|
||||
|
||||
|
||||
def _is_custom_sampler(node):
|
||||
"""SamplerCustomAdvanced-style nodes split settings across noise/sampler/sigmas."""
|
||||
if not _truthy(node):
|
||||
return False
|
||||
inputs = _get(node, "inputs")
|
||||
return (
|
||||
_truthy(inputs)
|
||||
and _is_link(_get(inputs, "guider"))
|
||||
and _is_link(_get(inputs, "sigmas"))
|
||||
)
|
||||
|
||||
|
||||
def _workflow_nodes(workflow):
|
||||
if not _truthy(workflow):
|
||||
return []
|
||||
nodes = _get(workflow, "nodes")
|
||||
return nodes if isinstance(nodes, list) else []
|
||||
|
||||
|
||||
def _widgets_of(node):
|
||||
values = _get(node, "widgets_values")
|
||||
return values if isinstance(values, list) else []
|
||||
|
||||
|
||||
class _Context:
|
||||
"""The graph being traced and recovered PromptManager widget text."""
|
||||
|
||||
__slots__ = ("graph", "widget_text")
|
||||
|
||||
def __init__(self, graph, widget_text):
|
||||
self.graph = graph
|
||||
self.widget_text = widget_text
|
||||
|
||||
|
||||
def _create_context(graph, workflow):
|
||||
"""Recover PromptManager text overwritten by the pre-3.2.3 SaveImage patch.
|
||||
|
||||
The signature is two or more PromptManager nodes with identical API text
|
||||
whose workflow widgets differ; only those get their widget text back.
|
||||
Otherwise the API graph is what actually ran and always wins.
|
||||
"""
|
||||
widgets = {}
|
||||
for node in _workflow_nodes(workflow):
|
||||
text = _get(_widgets_of(node), 0)
|
||||
node_type = _get(node, "type")
|
||||
if (
|
||||
not _strict_equal(node_type, "PromptManagerPair")
|
||||
and _matches(_PROMPT_MANAGER, node_type)
|
||||
and isinstance(text, str)
|
||||
):
|
||||
widgets[_js_string(_get(node, "id"))] = text
|
||||
by_api_text = {}
|
||||
for node_id in widgets:
|
||||
api_text = _and_chain(graph, node_id, "inputs", "text")
|
||||
if isinstance(api_text, str):
|
||||
by_api_text.setdefault(api_text, []).append(node_id)
|
||||
widget_text = {}
|
||||
for ids in by_api_text.values():
|
||||
if len(ids) > 1 and len({widgets[node_id] for node_id in ids}) > 1:
|
||||
for node_id in ids:
|
||||
widget_text[node_id] = widgets[node_id]
|
||||
return _Context(graph, widget_text)
|
||||
|
||||
|
||||
def _enter(context, link, seen):
|
||||
"""``(node_id, node)`` for a link; None when unresolved or already expanded."""
|
||||
if not _is_link(link) or len(seen) >= MAX_VISITS:
|
||||
return None
|
||||
node_id = _js_string(link[0])
|
||||
node = _get(context.graph, node_id)
|
||||
key = f"{node_id}:{_js_string(link[1])}"
|
||||
if not _truthy(node) or not _truthy(_get(node, "inputs")) or key in seen:
|
||||
return None
|
||||
seen.add(key)
|
||||
return node_id, node
|
||||
|
||||
|
||||
def _resolve_scalar(context, value, name, seen):
|
||||
"""A scalar input, following a link; an input named like the field wins."""
|
||||
if not _is_link(value):
|
||||
return value
|
||||
entered = _enter(context, value, seen)
|
||||
if entered is None:
|
||||
return UNDEFINED
|
||||
inputs = _get(entered[1], "inputs")
|
||||
if _truthy(name) and _own(inputs, name):
|
||||
return (yield _resolve_scalar(context, _get(inputs, name), name, seen))
|
||||
for item in _values(inputs):
|
||||
resolved = yield _resolve_scalar(context, item, name, seen)
|
||||
if resolved is not UNDEFINED and _typeof(resolved) != "object":
|
||||
return resolved
|
||||
return UNDEFINED
|
||||
|
||||
|
||||
def _resolve_linked_input(context, link, keys):
|
||||
"""The first of ``keys`` on the node a link points to, resolving further links."""
|
||||
seen = set()
|
||||
entered = _enter(context, link, seen)
|
||||
if entered is None:
|
||||
return UNDEFINED
|
||||
inputs = _get(entered[1], "inputs")
|
||||
key = next((k for k in keys if _has(inputs, k)), None)
|
||||
if key is None:
|
||||
return UNDEFINED
|
||||
return (yield _resolve_scalar(context, _get(inputs, key), key, seen))
|
||||
|
||||
|
||||
def _resolve_string(context, value, seen):
|
||||
"""A string input; text-producing nodes may be chained (concat, primitives)."""
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
entered = _enter(context, value, seen)
|
||||
if entered is None:
|
||||
return ""
|
||||
node_id, node = entered
|
||||
inputs = _get(node, "inputs")
|
||||
if _strict_equal(_get(node, "class_type"), "PromptManagerPair"):
|
||||
if _strict_equal(value[1], 3):
|
||||
return (yield _resolve_string(context, _get(inputs, "negative_text"), seen))
|
||||
if _strict_equal(value[1], 2):
|
||||
return (yield _compose_node_text(context, node_id, node, seen))
|
||||
return ""
|
||||
if _has(inputs, "text"):
|
||||
return (yield _compose_node_text(context, node_id, node, seen))
|
||||
delimiter = _get(inputs, "delimiter")
|
||||
if not isinstance(delimiter, str):
|
||||
delimiter = " "
|
||||
parts = []
|
||||
for key, item in _entries(inputs):
|
||||
if TEXT_KEYS.fullmatch(key):
|
||||
parts.append((yield _resolve_string(context, item, seen)))
|
||||
return delimiter.join(part for part in parts if _trim(part))
|
||||
|
||||
|
||||
def _compose_node_text(context, node_id, node, seen):
|
||||
"""An encoder's text, joined like PromptManager: prepend + text + append."""
|
||||
inputs = _get(node, "inputs")
|
||||
if node_id in context.widget_text:
|
||||
raw = context.widget_text[node_id]
|
||||
else:
|
||||
raw = yield _resolve_string(context, _get(inputs, "text"), seen)
|
||||
text = _trim(raw)
|
||||
if not text:
|
||||
return ""
|
||||
# One shared set even across siblings keeps crafted chains linear.
|
||||
prepend = _trim(
|
||||
(yield _resolve_string(context, _get(inputs, "prepend_text"), seen))
|
||||
)
|
||||
append = _trim((yield _resolve_string(context, _get(inputs, "append_text"), seen)))
|
||||
# Older saves stored the already-combined prompt in ``text``.
|
||||
parts = (
|
||||
prepend if prepend and not text.startswith(prepend) else "",
|
||||
text,
|
||||
append if append and not text.endswith(append) else "",
|
||||
)
|
||||
return " ".join(part for part in parts if part)
|
||||
|
||||
|
||||
def _trace_conditioning(context, link, role, seen):
|
||||
"""Follow a conditioning link upstream to the encoder that produced it."""
|
||||
entered = _enter(context, link, seen)
|
||||
if entered is None:
|
||||
return ""
|
||||
node_id, node = entered
|
||||
inputs = _get(node, "inputs")
|
||||
if _strict_equal(_get(node, "class_type"), "PromptManagerPair"):
|
||||
if _strict_equal(link[1], 1):
|
||||
return (yield _resolve_string(context, _get(inputs, "negative_text"), seen))
|
||||
if _strict_equal(link[1], 0):
|
||||
return (yield _compose_node_text(context, node_id, node, seen))
|
||||
return ""
|
||||
if _has(inputs, "text"):
|
||||
return (yield _compose_node_text(context, node_id, node, seen))
|
||||
# Pass-through nodes (ControlNet apply, conditioning combine/set area, ...).
|
||||
following = _first_truthy(
|
||||
_get(inputs, role),
|
||||
_get(inputs, "conditioning"),
|
||||
_get(inputs, "conditioning_to"),
|
||||
_get(inputs, "conditioning_1"),
|
||||
)
|
||||
return (yield _trace_conditioning(context, following, role, seen))
|
||||
|
||||
|
||||
def _trace_model_name(context, link, seen):
|
||||
"""Follow the model link upstream (through LoRA loaders etc.) to a loader."""
|
||||
entered = _enter(context, link, seen)
|
||||
if entered is None:
|
||||
return UNDEFINED
|
||||
inputs = _get(entered[1], "inputs")
|
||||
key = next((k for k in MODEL_NAME_KEYS if isinstance(_get(inputs, k), str)), None)
|
||||
if key is not None:
|
||||
return _get(inputs, key)
|
||||
return (yield _trace_model_name(context, _get(inputs, "model"), seen))
|
||||
|
||||
|
||||
def _find_any_model_name(nodes):
|
||||
for node in nodes:
|
||||
inputs = _and_chain(node, "inputs")
|
||||
if not _truthy(inputs):
|
||||
inputs = {}
|
||||
key = next(
|
||||
(k for k in MODEL_NAME_KEYS if isinstance(_get(inputs, k), str)), None
|
||||
)
|
||||
if key is not None:
|
||||
return _get(inputs, key)
|
||||
return UNDEFINED
|
||||
|
||||
|
||||
def _from_prompt_graph(graph, workflow):
|
||||
context = _create_context(graph, workflow)
|
||||
nodes = _values(graph)
|
||||
custom = next((node for node in nodes if _is_custom_sampler(node)), UNDEFINED)
|
||||
guider = UNDEFINED
|
||||
if custom is not UNDEFINED:
|
||||
entered = _enter(context, _get(_get(custom, "inputs"), "guider"), set())
|
||||
guider = entered[1] if entered is not None else UNDEFINED
|
||||
sampler = (
|
||||
guider
|
||||
if _is_sampler(guider)
|
||||
else next((n for n in nodes if _is_sampler(n)), UNDEFINED)
|
||||
)
|
||||
|
||||
if sampler is UNDEFINED:
|
||||
loose = next(
|
||||
(
|
||||
n
|
||||
for n in nodes
|
||||
if _truthy(n)
|
||||
and _truthy(_get(n, "inputs"))
|
||||
and _matches(_KSAMPLER, _get(n, "class_type"))
|
||||
),
|
||||
UNDEFINED,
|
||||
)
|
||||
inputs = _and_chain(loose, "inputs")
|
||||
if not _truthy(inputs):
|
||||
inputs = {}
|
||||
seed = _coalesce(_get(inputs, "seed"), lambda: _get(inputs, "noise_seed"))
|
||||
return {
|
||||
"checkpoint": _find_any_model_name(nodes),
|
||||
"seed": (yield _resolve_scalar(context, seed, "seed", set())),
|
||||
"steps": (
|
||||
yield _resolve_scalar(context, _get(inputs, "steps"), "steps", set())
|
||||
),
|
||||
"cfgScale": (
|
||||
yield _resolve_scalar(context, _get(inputs, "cfg"), "cfg", set())
|
||||
),
|
||||
"sampler": (
|
||||
yield _resolve_scalar(
|
||||
context, _get(inputs, "sampler_name"), "sampler_name", set()
|
||||
)
|
||||
),
|
||||
}
|
||||
|
||||
inputs = _get(sampler, "inputs")
|
||||
|
||||
def from_custom(field, keys):
|
||||
if custom is UNDEFINED:
|
||||
return UNDEFINED
|
||||
return _run(
|
||||
_resolve_linked_input(context, _get(_get(custom, "inputs"), field), keys)
|
||||
)
|
||||
|
||||
result = {
|
||||
"positivePrompt": (
|
||||
yield _trace_conditioning(
|
||||
context, _get(inputs, "positive"), "positive", set()
|
||||
)
|
||||
),
|
||||
"negativePrompt": (
|
||||
yield _trace_conditioning(
|
||||
context, _get(inputs, "negative"), "negative", set()
|
||||
)
|
||||
),
|
||||
}
|
||||
model = yield _trace_model_name(context, _get(inputs, "model"), set())
|
||||
result["checkpoint"] = model if _truthy(model) else _find_any_model_name(nodes)
|
||||
seed_value = _coalesce(_get(inputs, "seed"), lambda: _get(inputs, "noise_seed"))
|
||||
seed_name = "seed" if _has(inputs, "seed") else "noise_seed"
|
||||
seed = yield _resolve_scalar(context, seed_value, seed_name, set())
|
||||
result["seed"] = _coalesce(
|
||||
seed, lambda: from_custom("noise", ("noise_seed", "seed"))
|
||||
)
|
||||
steps = yield _resolve_scalar(context, _get(inputs, "steps"), "steps", set())
|
||||
result["steps"] = _coalesce(steps, lambda: from_custom("sigmas", ("steps",)))
|
||||
result["cfgScale"] = yield _resolve_scalar(
|
||||
context, _get(inputs, "cfg"), "cfg", set()
|
||||
)
|
||||
sampler_name = yield _resolve_scalar(
|
||||
context, _get(inputs, "sampler_name"), "sampler_name", set()
|
||||
)
|
||||
result["sampler"] = _coalesce(
|
||||
sampler_name, lambda: from_custom("sampler", ("sampler_name",))
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
# --- UI workflow graph -------------------------------------------------------------
|
||||
|
||||
|
||||
def _widget_names(node):
|
||||
inputs = _first_truthy(_get(node, "inputs"), [])
|
||||
from_inputs = [
|
||||
_get(_get(item, "widget"), "name")
|
||||
for item in _array(inputs)
|
||||
if _truthy(item)
|
||||
and _truthy(_get(item, "widget"))
|
||||
and _truthy(_get(_get(item, "widget"), "name"))
|
||||
]
|
||||
if from_inputs:
|
||||
return from_inputs
|
||||
node_type = _get(node, "type")
|
||||
key = _js_string(node_type)
|
||||
if key in _OBJECT_PROTOTYPE_KEYS:
|
||||
raise GenerationParamsError(f"widget positions of {key!r} are not iterable")
|
||||
if key in WIDGET_POSITIONS:
|
||||
return list(WIDGET_POSITIONS[key])
|
||||
if _matches(_CHECKPOINT_LOADER, node_type):
|
||||
return ["ckpt_name"]
|
||||
if _matches(_UNET, node_type):
|
||||
return ["unet_name"]
|
||||
return []
|
||||
|
||||
|
||||
def _named_widgets(node):
|
||||
"""Map widget values to input names, skipping control_after_generate."""
|
||||
named = _get(node, "widgets_values_named")
|
||||
if _truthy(named) and isinstance(named, dict):
|
||||
return dict(named)
|
||||
values = _widgets_of(node)
|
||||
out = {}
|
||||
position = 0
|
||||
for name in _iterate(_widget_names(node)):
|
||||
if position >= len(values):
|
||||
break
|
||||
_set(out, name, values[position])
|
||||
position += 1
|
||||
following = values[position] if position < len(values) else UNDEFINED
|
||||
if (
|
||||
_SEED_NAME.search(_js_string(name))
|
||||
and isinstance(following, str)
|
||||
and following in SEED_CONTROL_VALUES
|
||||
):
|
||||
position += 1
|
||||
return out
|
||||
|
||||
|
||||
def _workflow_to_graph(workflow):
|
||||
"""Convert a UI workflow ({nodes, links}) into the API graph shape.
|
||||
|
||||
Set/Get pairs, Reroutes and bypassed nodes resolve to their real source;
|
||||
muted nodes are dropped.
|
||||
"""
|
||||
nodes = [
|
||||
n
|
||||
for n in _workflow_nodes(workflow)
|
||||
if _truthy(n) and not _nullish(_get(n, "id"))
|
||||
]
|
||||
by_id = {}
|
||||
for node in nodes:
|
||||
by_id[_js_string(_get(node, "id"))] = node
|
||||
links = {}
|
||||
raw_links = _get(workflow, "links")
|
||||
for link in raw_links if isinstance(raw_links, list) else []:
|
||||
if isinstance(link, list):
|
||||
links[_same_value_key(_get(link, 0))] = (
|
||||
_get(link, 1),
|
||||
_get(link, 2),
|
||||
_get(link, 5),
|
||||
)
|
||||
elif _truthy(link) and _typeof(link) == "object":
|
||||
links[_same_value_key(_get(link, "id"))] = (
|
||||
_get(link, "origin_id"),
|
||||
_get(link, "origin_slot"),
|
||||
_get(link, "type"),
|
||||
)
|
||||
setters = {}
|
||||
for node in nodes:
|
||||
if _strict_equal(_get(node, "type"), "SetNode"):
|
||||
setters[_same_value_key(_get(_widgets_of(node), 0))] = _and_chain(
|
||||
node, "inputs", 0, "link"
|
||||
)
|
||||
|
||||
def source(link_id):
|
||||
hops = 0
|
||||
while True:
|
||||
link = links.get(_same_value_key(link_id))
|
||||
node = by_id.get(_js_string(link[0])) if link is not None else None
|
||||
if (
|
||||
node is None
|
||||
or _strict_equal(_get(node, "mode"), MODE_MUTED)
|
||||
or hops > MAX_VISITS
|
||||
):
|
||||
return UNDEFINED
|
||||
hops += 1
|
||||
node_type = _get(node, "type")
|
||||
if _strict_equal(node_type, "Reroute"):
|
||||
link_id = _and_chain(node, "inputs", 0, "link")
|
||||
continue
|
||||
if _strict_equal(node_type, "GetNode"):
|
||||
link_id = setters.get(
|
||||
_same_value_key(_get(_widgets_of(node), 0)), UNDEFINED
|
||||
)
|
||||
continue
|
||||
if _strict_equal(_get(node, "mode"), MODE_BYPASSED):
|
||||
link_type = link[2]
|
||||
same = next(
|
||||
(
|
||||
item
|
||||
for item in _array(_first_truthy(_get(node, "inputs"), []))
|
||||
if _truthy(item)
|
||||
and not _nullish(_get(item, "link"))
|
||||
and (
|
||||
not _truthy(link_type)
|
||||
or _strict_equal(link_type, "*")
|
||||
or _strict_equal(_get(item, "type"), link_type)
|
||||
)
|
||||
),
|
||||
UNDEFINED,
|
||||
)
|
||||
if same is UNDEFINED:
|
||||
return UNDEFINED
|
||||
link_id = _get(same, "link")
|
||||
continue
|
||||
return [_js_string(_get(node, "id")), link[1]]
|
||||
|
||||
graph = {}
|
||||
for node in nodes:
|
||||
mode = _get(node, "mode")
|
||||
node_type = _get(node, "type")
|
||||
if (
|
||||
_strict_equal(mode, MODE_MUTED)
|
||||
or _strict_equal(mode, MODE_BYPASSED)
|
||||
or (isinstance(node_type, str) and node_type in PASS_THROUGH_TYPES)
|
||||
):
|
||||
continue
|
||||
inputs = _named_widgets(node)
|
||||
for item in _iterate(_first_truthy(_get(node, "inputs"), [])):
|
||||
if not _truthy(item) or _nullish(_get(item, "link")):
|
||||
continue
|
||||
found = source(_get(item, "link"))
|
||||
if _truthy(found):
|
||||
_set(inputs, _get(item, "name"), found)
|
||||
_set(graph, _get(node, "id"), {"class_type": node_type, "inputs": inputs})
|
||||
return graph
|
||||
|
||||
|
||||
# --- Public API ----------------------------------------------------------------------
|
||||
|
||||
|
||||
def _has_value(value):
|
||||
return not _nullish(value) and not (isinstance(value, str) and value == "")
|
||||
|
||||
|
||||
def extract_generation_params(data):
|
||||
"""Prompts, model and sampler settings from parsed ``prompt``/``workflow`` records.
|
||||
|
||||
``data`` holds the parsed records (``{"prompt": ..., "workflow": ...}``). Each
|
||||
field falls back to its placeholder when neither graph provides it.
|
||||
Raises ``GenerationParamsError`` where the browser parser throws.
|
||||
"""
|
||||
data = data if _truthy(data) else {}
|
||||
prompt = _get(data, "prompt")
|
||||
workflow = _get(data, "workflow")
|
||||
primary = {}
|
||||
if _truthy(prompt) and _typeof(prompt) == "object":
|
||||
primary = _run(_from_prompt_graph(prompt, workflow))
|
||||
fallback = {}
|
||||
if _truthy(workflow):
|
||||
fallback = _run(_from_prompt_graph(_workflow_to_graph(workflow), None))
|
||||
merged = {}
|
||||
for key, placeholder in PLACEHOLDERS.items():
|
||||
value = primary.get(key, UNDEFINED)
|
||||
if not _has_value(value):
|
||||
value = fallback.get(key, UNDEFINED)
|
||||
merged[key] = value if _has_value(value) else placeholder
|
||||
return merged
|
||||
|
||||
|
||||
def _reject_constant(name):
|
||||
raise ValueError(f"{name} is not JSON")
|
||||
|
||||
|
||||
def _double_integer(text):
|
||||
"""Integers parse as doubles in JavaScript; beyond 2**53 they round."""
|
||||
value = int(text)
|
||||
return value if -_MAX_SAFE_INTEGER <= value <= _MAX_SAFE_INTEGER else float(text)
|
||||
|
||||
|
||||
def parse_js_json(text):
|
||||
"""``JSON.parse(text)``, or None where it throws (or the text is ``null``)."""
|
||||
if not isinstance(text, str):
|
||||
return None
|
||||
try:
|
||||
return json.loads(
|
||||
text, parse_constant=_reject_constant, parse_int=_double_integer
|
||||
)
|
||||
except (ValueError, RecursionError):
|
||||
return None
|
||||
|
||||
|
||||
def extract_from_records(prompt_text, workflow_text):
|
||||
"""``extract_generation_params`` for raw ``prompt``/``workflow`` record text.
|
||||
|
||||
A record that is absent or fails to parse is treated as absent, as the
|
||||
browser pages do.
|
||||
"""
|
||||
data = {}
|
||||
for key, text in (("prompt", prompt_text), ("workflow", workflow_text)):
|
||||
if isinstance(text, str):
|
||||
data[key] = parse_js_json(text)
|
||||
return extract_generation_params(data)
|
||||
Reference in New Issue
Block a user