Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Vito Sansevero
|
c9678245f8 | ||
|
Vito Sansevero
|
5f21894196 | ||
|
Vito Sansevero
|
02e339ca1f | ||
|
Vito Sansevero
|
91bd060344 | ||
|
Vito Sansevero
|
5b3ddb887d |
+16
-52
@@ -27,22 +27,10 @@ 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
|
||||
|
||||
|
||||
@@ -1523,21 +1511,6 @@ 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:
|
||||
@@ -1558,32 +1531,23 @@ class AdminRoutesMixin:
|
||||
self.db, str(media_file), canonical, resolution
|
||||
)
|
||||
return result if detailed else result["status"]
|
||||
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"]
|
||||
elif resolution["state"] == "ambiguous":
|
||||
if 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()
|
||||
if resolution["state"] == "unresolved":
|
||||
if canonical.get("prompt") or not isinstance(
|
||||
canonical.get("parameters"), str
|
||||
):
|
||||
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()
|
||||
|
||||
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
@@ -1,444 +0,0 @@
|
||||
"""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(),
|
||||
}
|
||||
@@ -1,32 +0,0 @@
|
||||
// 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');
|
||||
}
|
||||
@@ -1,221 +0,0 @@
|
||||
"""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()
|
||||
@@ -1,637 +0,0 @@
|
||||
"""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()
|
||||
@@ -1,155 +0,0 @@
|
||||
"""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,7 +8,6 @@ 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
|
||||
@@ -73,19 +72,12 @@ 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(Path("/out/f03.png"), _f03_metadata())
|
||||
status = mixin._ingest_scanned_file("/out/f03.png", _f03_metadata())
|
||||
self.assertEqual(status, "unresolved")
|
||||
spy.assert_default_and_gated(self)
|
||||
# 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()
|
||||
mixin.db.assert_not_called()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+9
-67
@@ -17,14 +17,6 @@ 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"
|
||||
@@ -36,66 +28,16 @@ class TestPairScan(AdminAPITestCase):
|
||||
self.assertEqual(pairs[0]["negative_text"], "")
|
||||
self.assertEqual(pairs[0]["run_count"], 0)
|
||||
|
||||
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._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")
|
||||
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",
|
||||
)
|
||||
self.assertEqual(self._texts(), [])
|
||||
self.assertEqual(self.api.db.list_prompt_pairs()["total"], 0)
|
||||
self.assertEqual(self.api.db.get_recent_prompts()["prompts"], [])
|
||||
|
||||
def test_batch_exposes_pair_counts_and_actual_unchanged_status(self):
|
||||
meta = {"parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler"}
|
||||
|
||||
@@ -1,943 +0,0 @@
|
||||
"""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