780 lines
32 KiB
Python
780 lines
32 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import copy
|
|
import itertools
|
|
import folder_paths
|
|
import comfy.samplers
|
|
|
|
|
|
|
|
def parse_prompt_input_nested(prompt_input: str):
|
|
"""
|
|
Parse prompt input that supports plain text and arbitrarily deep nested arrays.
|
|
Recursively creates Cartesian products from nested lists.
|
|
|
|
Rules:
|
|
- A plain string is a single prompt option
|
|
- A flat list of strings ["a", "b"] = multiple OPTIONS (OR logic)
|
|
- A list containing sub-lists = SEQUENCE (AND logic, Cartesian product)
|
|
- Nesting can be arbitrarily deep
|
|
|
|
Args:
|
|
prompt_input: Plain text string OR JSON string containing prompts
|
|
|
|
Returns:
|
|
List of final prompt strings
|
|
|
|
Examples:
|
|
Plain text: "my prompt" -> ["my prompt"]
|
|
Simple options: '["a", "b", "c"]' -> ["a", "b", "c"]
|
|
Cartesian: '[["a", "b"], ["1", "2"]]' -> ["a, 1", "a, 2", "b, 1", "b, 2"]
|
|
Recursive: '["Photo of", ["a cat", "a dog"], ["in space", ["eating pizza", "sleeping"]]]'
|
|
-> ["Photo of, a cat, in space, eating pizza",
|
|
"Photo of, a cat, in space, sleeping",
|
|
"Photo of, a dog, in space, eating pizza",
|
|
"Photo of, a dog, in space, sleeping"]
|
|
"""
|
|
|
|
def recursive_cartesian(item):
|
|
"""Recursively expand nested prompt structures into flat string options."""
|
|
# Base Case: string/primitive -> single option
|
|
if not isinstance(item, list):
|
|
return [str(item)]
|
|
|
|
# Check if this list contains any sub-lists
|
|
has_nested = any(isinstance(sub, list) for sub in item)
|
|
|
|
# Flat list of strings -> OPTIONS (OR logic)
|
|
if not has_nested:
|
|
return [str(x) for x in item]
|
|
|
|
# List containing sub-lists -> SEQUENCE (AND logic, Cartesian product)
|
|
# Each element is recursively resolved into its options, then we take the product
|
|
normalized_groups = []
|
|
for sub in item:
|
|
normalized_groups.append(recursive_cartesian(sub))
|
|
|
|
# Generate Cartesian product across the resolved groups
|
|
combinations = itertools.product(*normalized_groups)
|
|
return [", ".join(combo) for combo in combinations]
|
|
|
|
# Handle None or empty string
|
|
if not prompt_input or prompt_input.strip() == "":
|
|
return [""]
|
|
|
|
prompt_input = prompt_input.strip()
|
|
|
|
# Try parsing as JSON
|
|
try:
|
|
parsed = json.loads(prompt_input)
|
|
except (json.JSONDecodeError, ValueError):
|
|
# Not valid JSON - treat as plain text
|
|
if (prompt_input.startswith('"') and prompt_input.endswith('"')) or \
|
|
(prompt_input.startswith("'") and prompt_input.endswith("'")):
|
|
prompt_input = prompt_input[1:-1]
|
|
return [prompt_input]
|
|
|
|
# Handle empty parsed result
|
|
if not parsed:
|
|
return [""]
|
|
|
|
# Start the recursion
|
|
return recursive_cartesian(parsed)
|
|
|
|
|
|
def normalize_str(s):
|
|
"""Normalize string paths by replacing backslashes with forward slashes"""
|
|
if isinstance(s, str):
|
|
return s.replace("\\", "/").strip()
|
|
return s
|
|
|
|
|
|
def parse_json_with_error(json_str, name):
|
|
"""Parse JSON string with helpful error messages"""
|
|
try:
|
|
return json.loads(json_str.strip())
|
|
except json.JSONDecodeError as e:
|
|
raise ValueError(f"JSON Error in {name}: {e}")
|
|
|
|
|
|
def parse_float_input(input_str):
|
|
"""Parse float input that could be JSON array, comma-separated, or single value"""
|
|
try:
|
|
val = json.loads(input_str)
|
|
if isinstance(val, list):
|
|
return [float(x) for x in val]
|
|
return [float(val)]
|
|
except:
|
|
try:
|
|
if "," in input_str:
|
|
return [float(x.strip()) for x in input_str.split(",")]
|
|
return [float(input_str)]
|
|
except:
|
|
return [1.0]
|
|
|
|
|
|
def parse_string_input(input_str):
|
|
"""Parse string input that could be JSON array or single value"""
|
|
try:
|
|
val = json.loads(input_str.strip())
|
|
if isinstance(val, list):
|
|
return [str(x) for x in val]
|
|
return [str(val)]
|
|
except:
|
|
return [input_str]
|
|
|
|
|
|
def get_files_from_folder(input_string, type_key):
|
|
"""
|
|
Get files from a folder path, or return the input if it's a single file.
|
|
|
|
Args:
|
|
input_string: Path to file or folder (folder ends with /)
|
|
type_key: Type of files to search for (e.g., "checkpoints", "loras")
|
|
|
|
Returns:
|
|
List of file paths
|
|
"""
|
|
input_norm = input_string.replace("\\", "/")
|
|
if ":" in input_norm:
|
|
input_norm = input_norm.split(":")[0]
|
|
if not input_norm.endswith("/"):
|
|
return [input_string]
|
|
|
|
target_folder = input_norm.rstrip("/")
|
|
all_files = folder_paths.get_filename_list(type_key)
|
|
found = []
|
|
for f in all_files:
|
|
f_norm = f.replace("\\", "/")
|
|
if f_norm.startswith(target_folder + "/"):
|
|
found.append(f)
|
|
if not found:
|
|
print(f"[GridTester] Warning: No files found in folder '{input_string}' for type '{type_key}'")
|
|
return []
|
|
return found
|
|
|
|
|
|
def parse_lora_definition(lora_string):
|
|
"""
|
|
Parse LoRA definition string into list of (name, model_str, clip_str) tuples.
|
|
|
|
Format: "lora1.safetensors:1.0:0.8 + lora2.safetensors:0.5:0.5"
|
|
|
|
If no strength is specified, defaults to 1.0 for both model and CLIP.
|
|
|
|
Args:
|
|
lora_string: String defining LoRAs with optional strengths
|
|
|
|
Returns:
|
|
List of (name, model_strength, clip_strength) tuples
|
|
"""
|
|
if lora_string == "None":
|
|
return []
|
|
definitions = []
|
|
parts = lora_string.split(" + ")
|
|
for part in parts:
|
|
part = part.strip()
|
|
if ":" in part:
|
|
segments = part.split(":")
|
|
name = segments[0].strip()
|
|
m_str = float(segments[1]) if len(segments) > 1 else 1.0
|
|
c_str = float(segments[2]) if len(segments) > 2 else m_str # Default CLIP to same as model
|
|
definitions.append((name, m_str, c_str))
|
|
else:
|
|
# No strength specified, default to 1.0 for both
|
|
definitions.append((part, 1.0, 1.0))
|
|
return definitions
|
|
|
|
|
|
def _expand_lora_weight_arrays(lora_string):
|
|
"""
|
|
Expand LoRA weight arrays into individual LoRA strings.
|
|
|
|
Supports array notation in strength values for grid searching:
|
|
- "lora.safetensors:[0.5, 0.8]:1.0" -> ["lora.safetensors:0.5:1.0", "lora.safetensors:0.8:1.0"]
|
|
- "lora.safetensors:1.0:[0.5, 0.8]" -> ["lora.safetensors:1.0:0.5", "lora.safetensors:1.0:0.8"]
|
|
- "lora.safetensors:[0.5, 0.8]:[0.6, 1.0]" -> 4 combinations via Cartesian product
|
|
- "lora1:[0.5, 0.8]:1.0 + lora2:1.0:1.0" -> expands lora1 weights, keeps lora2 fixed in each
|
|
|
|
Args:
|
|
lora_string: Single LoRA definition (may contain " + " for stacked LoRAs)
|
|
|
|
Returns:
|
|
List of expanded LoRA strings with concrete weight values
|
|
"""
|
|
if lora_string == "None":
|
|
return ["None"]
|
|
|
|
# Split stacked LoRAs
|
|
stack_parts = lora_string.split(" + ")
|
|
|
|
# For each part, check if it has array weights
|
|
parts_expansions = []
|
|
has_arrays = False
|
|
|
|
for part in stack_parts:
|
|
part = part.strip()
|
|
|
|
# Check for bracket notation in strength values
|
|
# Match pattern: name:[array_or_value]:[array_or_value] or name:[array_or_value]
|
|
bracket_match = re.search(r'\[[\d.,\s-]+\]', part)
|
|
if not bracket_match:
|
|
# No arrays in this part, keep as-is
|
|
parts_expansions.append([part])
|
|
continue
|
|
|
|
has_arrays = True
|
|
|
|
# Parse the part: split on ":" but respect brackets
|
|
# Strategy: find the name (everything before first ":"), then parse strength fields
|
|
segments = []
|
|
current = ""
|
|
depth = 0
|
|
for ch in part:
|
|
if ch == '[':
|
|
depth += 1
|
|
current += ch
|
|
elif ch == ']':
|
|
depth -= 1
|
|
current += ch
|
|
elif ch == ':' and depth == 0:
|
|
segments.append(current)
|
|
current = ""
|
|
else:
|
|
current += ch
|
|
if current:
|
|
segments.append(current)
|
|
|
|
name = segments[0]
|
|
|
|
# Parse each strength segment - could be "[0.5, 0.8]" or just "1.0"
|
|
strength_lists = []
|
|
for i in range(1, len(segments)):
|
|
seg = segments[i].strip()
|
|
if seg.startswith('[') and seg.endswith(']'):
|
|
# Array notation - parse values
|
|
inner = seg[1:-1]
|
|
values = [v.strip() for v in inner.split(',') if v.strip()]
|
|
strength_lists.append(values)
|
|
else:
|
|
strength_lists.append([seg])
|
|
|
|
# Generate all combinations of strengths for this LoRA
|
|
if strength_lists:
|
|
combos = list(itertools.product(*strength_lists))
|
|
expanded = [f"{name}:{':'.join(combo)}" for combo in combos]
|
|
parts_expansions.append(expanded)
|
|
else:
|
|
parts_expansions.append([part])
|
|
|
|
if not has_arrays:
|
|
return [lora_string]
|
|
|
|
# Cartesian product across stacked LoRAs
|
|
result = []
|
|
for combo in itertools.product(*parts_expansions):
|
|
result.append(" + ".join(combo))
|
|
|
|
if len(result) > 1:
|
|
print(f"[GridTester] 🎛️ LoRA weight arrays expanded to {len(result)} combinations")
|
|
|
|
return result
|
|
|
|
|
|
def expand_lora_stack(lora_input):
|
|
"""
|
|
Expand LoRA stacks with folder support and weight array expansion.
|
|
|
|
Handles formats like:
|
|
- "lora.safetensors"
|
|
- "lora.safetensors:0.8:0.6"
|
|
- "lora.safetensors:[0.5, 0.8]:1.0" (weight array - grid searches strengths)
|
|
- "lora1:0.8:0.6 + lora2:1.0:1.0"
|
|
- "folder/ + lora.safetensors" (expands each LoRA individually)
|
|
- "folder/* + lora.safetensors" (stacks ALL LoRAs in folder together)
|
|
- "folder/:0.8:0.6 + lora.safetensors"
|
|
- "folder/*:0.8:0.6" (stacks ALL LoRAs with same strength)
|
|
|
|
Args:
|
|
lora_input: LoRA specification string or list
|
|
|
|
Returns:
|
|
List of expanded LoRA stack strings
|
|
"""
|
|
def to_list(x):
|
|
return x if isinstance(x, list) else [x]
|
|
|
|
raw_loras = to_list(lora_input)
|
|
expanded_loras = []
|
|
|
|
for l in raw_loras:
|
|
if l == "None":
|
|
expanded_loras.append("None")
|
|
continue
|
|
|
|
stack_parts = l.split(" + ")
|
|
expanded_parts = []
|
|
|
|
for part in stack_parts:
|
|
if ":" in part:
|
|
p_split = part.split(":", 1)
|
|
base_path = p_split[0].strip()
|
|
args = ":" + p_split[1].strip()
|
|
else:
|
|
base_path = part.strip()
|
|
args = ""
|
|
|
|
norm_path = base_path.replace("\\", "/")
|
|
|
|
# Check for folder/* syntax (stack all together)
|
|
if norm_path.endswith("/*") or (norm_path.endswith("*") and "/" in norm_path):
|
|
# Remove the /* or * suffix
|
|
folder_path = norm_path.rstrip("/*").rstrip("*").rstrip("/")
|
|
found_files = get_files_from_folder(folder_path + "/", "loras")
|
|
|
|
if found_files:
|
|
# Stack ALL files together as a single entry (not separate combinations)
|
|
stacked = " + ".join([f"{f}{args}" for f in found_files])
|
|
expanded_parts.append([stacked])
|
|
print(f"[GridTester] 📚 Folder/* syntax: Stacking {len(found_files)} LoRAs from '{folder_path}' together")
|
|
else:
|
|
print(f"[GridTester] ⚠️ No LoRAs found in folder: {folder_path}")
|
|
expanded_parts.append([])
|
|
|
|
# Check for regular folder/ syntax (expand individually)
|
|
elif norm_path.endswith("/"):
|
|
found_files = get_files_from_folder(base_path, "loras")
|
|
expanded_parts.append([f"{f}{args}" for f in found_files])
|
|
else:
|
|
expanded_parts.append([part])
|
|
|
|
# Skip if any part expanded to empty
|
|
if not all(expanded_parts):
|
|
continue
|
|
|
|
for combo in itertools.product(*expanded_parts):
|
|
joined = " + ".join(combo)
|
|
# Now expand any weight arrays in the joined result
|
|
weight_expanded = _expand_lora_weight_arrays(joined)
|
|
expanded_loras.extend(weight_expanded)
|
|
|
|
return expanded_loras
|
|
|
|
|
|
def _normalize_ltx_audio_mode(entry):
|
|
"""Convert audio_mode='both' to ['on', 'off'] for cartesian expansion."""
|
|
if entry.get("audio_mode") == "both":
|
|
e = dict(entry)
|
|
e["audio_mode"] = ["on", "off"]
|
|
return e
|
|
return entry
|
|
|
|
|
|
def _expand_ltx_entry(entry):
|
|
"""Expand a single ltx_video entry into a list of fully-resolved LTX configs.
|
|
|
|
Each LTX field that is a list (except clip_models which has special semantics)
|
|
contributes a Cartesian axis. Returns a list of dicts ready for the orchestrator.
|
|
"""
|
|
e = _normalize_ltx_audio_mode(entry)
|
|
|
|
def to_list(x):
|
|
return x if isinstance(x, list) else [x]
|
|
|
|
# clip_models special case:
|
|
# - [str, str] (2 strings) -> single value (a CLIP pair)
|
|
# - [[str, str], [str, str], ...] -> sweep
|
|
raw_clip = e.get("clip_models", ["", ""])
|
|
if isinstance(raw_clip, list) and len(raw_clip) == 2 and all(isinstance(c, str) for c in raw_clip):
|
|
clip_options = [raw_clip]
|
|
elif isinstance(raw_clip, list) and all(isinstance(c, list) for c in raw_clip):
|
|
clip_options = raw_clip
|
|
else:
|
|
clip_options = [raw_clip]
|
|
|
|
# input_image: special-case folder expansion before to_list
|
|
raw_input_image = e.get("input_image")
|
|
if isinstance(raw_input_image, str) and raw_input_image and (raw_input_image.endswith("/") or raw_input_image.endswith("\\")):
|
|
folder_path = raw_input_image.rstrip("/\\")
|
|
if os.path.isdir(folder_path):
|
|
exts = (".png", ".jpg", ".jpeg", ".webp")
|
|
input_image_axis = sorted(
|
|
os.path.join(folder_path, f)
|
|
for f in os.listdir(folder_path)
|
|
if f.lower().endswith(exts)
|
|
)
|
|
if not input_image_axis:
|
|
input_image_axis = [None] # empty folder → text-to-video
|
|
else:
|
|
# Folder doesn't exist on disk — leave as-is, preflight will surface the error
|
|
input_image_axis = [raw_input_image]
|
|
else:
|
|
input_image_axis = to_list(raw_input_image) if raw_input_image is not None else [None]
|
|
|
|
# All other fields: simple to_list
|
|
field_axes = {
|
|
"model": to_list(e.get("model", "")),
|
|
"clip_models": clip_options,
|
|
"vae_video": to_list(e.get("vae_video", "")),
|
|
"vae_audio": to_list(e.get("vae_audio", "")),
|
|
"latent_upscaler": to_list(e.get("latent_upscaler", "")),
|
|
"duration_seconds": to_list(e.get("duration_seconds", 5)),
|
|
"frame_rate": to_list(e.get("frame_rate", 25)),
|
|
"sampler_stage1": to_list(e.get("sampler_stage1", "euler_ancestral_cfg_pp")),
|
|
"sampler_stage2": to_list(e.get("sampler_stage2", "euler_cfg_pp")),
|
|
"sigmas_stage1": to_list(e.get("sigmas_stage1", "")),
|
|
"sigmas_stage2": to_list(e.get("sigmas_stage2", "")),
|
|
"cfg": to_list(e.get("cfg", 1.0)),
|
|
"seed": to_list(e.get("seed", 0)),
|
|
"input_image": input_image_axis,
|
|
"image_strength_stage1": to_list(e.get("image_strength_stage1", 0.8)),
|
|
"image_strength_stage2": to_list(e.get("image_strength_stage2", 1.0)),
|
|
"img_compression": to_list(e.get("img_compression", 18)),
|
|
"audio_mode": to_list(e.get("audio_mode", "on")),
|
|
"lora": to_list(e.get("lora", "None")),
|
|
"positive": to_list(e.get("positive", "")),
|
|
"negative": to_list(e.get("negative", "")),
|
|
"width": to_list(e.get("width", 446)),
|
|
"height": to_list(e.get("height", 576)),
|
|
}
|
|
|
|
keys = list(field_axes.keys())
|
|
values = [field_axes[k] for k in keys]
|
|
|
|
expanded = []
|
|
for combo in itertools.product(*values):
|
|
cfg = dict(zip(keys, combo))
|
|
cfg["model_type"] = "ltx_video"
|
|
expanded.append(cfg)
|
|
return expanded
|
|
|
|
|
|
def expand_configs(raw_configs, pos_prompts, neg_prompts, denoise_values, seed, extra_seeds, ckpt_name=None):
|
|
"""
|
|
Expand raw config entries into full configuration combinations.
|
|
|
|
Supports per-config prompts: if a config entry has "positive" and/or "negative" keys,
|
|
those override the external pos_prompts/neg_prompts for that entry. This allows each
|
|
config to define its own nested array prompts for Cartesian product expansion.
|
|
|
|
Args:
|
|
raw_configs: List of config dictionaries with potentially wildcarded values
|
|
pos_prompts: List of positive prompts (fallback when config doesn't define prompts)
|
|
neg_prompts: List of negative prompts (fallback when config doesn't define prompts)
|
|
denoise_values: List of denoise values
|
|
seed: Base seed value
|
|
extra_seeds: List of additional random seeds
|
|
ckpt_name: Checkpoint name from node input (used as default when model is "Default")
|
|
|
|
Returns:
|
|
List of fully expanded config dictionaries
|
|
"""
|
|
ALL_SCHEDULERS = comfy.samplers.KSampler.SCHEDULERS
|
|
ALL_SAMPLERS = comfy.samplers.KSampler.SAMPLERS
|
|
expanded = []
|
|
|
|
# Default prompt pairing strategy (used when config doesn't define its own prompts)
|
|
if len(pos_prompts) > 1 and len(neg_prompts) > 1 and len(pos_prompts) == len(neg_prompts):
|
|
print("[GridTester] Detected matching prompt lists. Using 1-to-1 Pairing.")
|
|
default_prompt_pairs = list(zip(pos_prompts, neg_prompts))
|
|
else:
|
|
default_prompt_pairs = list(itertools.product(pos_prompts, neg_prompts))
|
|
|
|
def to_list(x):
|
|
return x if isinstance(x, list) else [x]
|
|
|
|
for entry in raw_configs:
|
|
# Branch early for LTX video — different field shape, separate cartesian
|
|
if entry.get("model_type") == "ltx_video":
|
|
expanded.extend(_expand_ltx_entry(entry))
|
|
continue
|
|
|
|
# ==== PER-CONFIG PROMPT HANDLING ====
|
|
# If this config defines its own prompts, parse them and use instead of defaults
|
|
if "positive" in entry or "negative" in entry:
|
|
raw_positive = entry.get("positive", "")
|
|
raw_negative = entry.get("negative", "")
|
|
|
|
# Handle both string and already-parsed list/nested-list formats
|
|
if isinstance(raw_positive, (list, tuple)):
|
|
# Already a list/nested structure - pass as JSON string for parse_prompt_input_nested
|
|
config_pos = parse_prompt_input_nested(json.dumps(raw_positive))
|
|
elif isinstance(raw_positive, str) and raw_positive.strip():
|
|
config_pos = parse_prompt_input_nested(raw_positive)
|
|
else:
|
|
config_pos = pos_prompts # Fallback to external
|
|
|
|
if isinstance(raw_negative, (list, tuple)):
|
|
config_neg = parse_prompt_input_nested(json.dumps(raw_negative))
|
|
elif isinstance(raw_negative, str) and raw_negative.strip():
|
|
config_neg = parse_prompt_input_nested(raw_negative)
|
|
else:
|
|
config_neg = neg_prompts # Fallback to external
|
|
|
|
# Build prompt pairs for this config entry
|
|
if len(config_pos) > 1 and len(config_neg) > 1 and len(config_pos) == len(config_neg):
|
|
entry_prompt_pairs = list(zip(config_pos, config_neg))
|
|
else:
|
|
entry_prompt_pairs = list(itertools.product(config_pos, config_neg))
|
|
|
|
if entry_prompt_pairs:
|
|
print(f"[GridTester] Config defines {len(entry_prompt_pairs)} prompt combinations (overriding node inputs)")
|
|
else:
|
|
entry_prompt_pairs = default_prompt_pairs
|
|
|
|
# Expand wildcards
|
|
samplers = ALL_SAMPLERS if entry.get("sampler") == "*" else to_list(entry.get("sampler", "euler"))
|
|
schedulers = ALL_SCHEDULERS if entry.get("scheduler") == "*" else to_list(entry.get("scheduler", "normal"))
|
|
steps_l = to_list(entry.get("steps", 20))
|
|
cfgs = to_list(entry.get("cfg", 7.0))
|
|
clip_skips = to_list(entry.get("clip_skip", 0))
|
|
|
|
# Extract model type and related fields for non-checkpoint models
|
|
model_type = entry.get("model_type", "checkpoint")
|
|
clip_type_str = entry.get("clip_type", "stable_diffusion")
|
|
text_encoders = entry.get("text_encoders", [])
|
|
gguf_options = entry.get("gguf_options", {})
|
|
|
|
# Expand model folders (using correct folder type based on model_type)
|
|
raw_models = to_list(entry.get("model", "Default"))
|
|
expanded_models = []
|
|
if model_type == "gguf":
|
|
model_folder = "unet_gguf"
|
|
elif model_type == "diffusion_model":
|
|
model_folder = "diffusion_models"
|
|
else:
|
|
model_folder = "checkpoints"
|
|
|
|
for m in raw_models:
|
|
if m == "Default":
|
|
# Use the checkpoint name from node input instead of "Default"
|
|
if ckpt_name:
|
|
expanded_models.append(ckpt_name)
|
|
else:
|
|
expanded_models.append("Default")
|
|
else:
|
|
expanded_models.extend(get_files_from_folder(m, model_folder))
|
|
|
|
# Expand VAE list
|
|
raw_vaes = to_list(entry.get("vae", "Default"))
|
|
|
|
# Expand LoRA stacks (strengths are now in the lora string itself)
|
|
expanded_loras = expand_lora_stack(entry.get("lora", "None"))
|
|
|
|
# Get LoRA trigger word omissions (if specified)
|
|
lora_omit_triggers = entry.get("lora_omit_triggers", [])
|
|
if not isinstance(lora_omit_triggers, list):
|
|
lora_omit_triggers = [lora_omit_triggers]
|
|
|
|
# Get LoRA trigger word append settings (if specified)
|
|
lora_triggerwords_append_settings = entry.get("lora_triggerwords_append_settings", {})
|
|
if not isinstance(lora_triggerwords_append_settings, dict):
|
|
lora_triggerwords_append_settings = {}
|
|
|
|
# Get model-specific prompt prefix/suffix (quality tags for specific model families)
|
|
model_prompt_prefix = entry.get("model_prompt_prefix", "")
|
|
model_prompt_suffix = entry.get("model_prompt_suffix", "")
|
|
if not isinstance(model_prompt_prefix, str):
|
|
model_prompt_prefix = str(model_prompt_prefix)
|
|
if not isinstance(model_prompt_suffix, str):
|
|
model_prompt_suffix = str(model_prompt_suffix)
|
|
|
|
# Get attention mode(s) for testing different attention implementations
|
|
# Valid: "default", "xformers", "pytorch", "flash", "sage", "sage3", "sub_quad", "split", "*"
|
|
VALID_ATTENTION_MODES = ["default", "xformers", "pytorch", "flash", "sage", "sage3", "sub_quad", "split"]
|
|
raw_attention = to_list(entry.get("attention_mode", "default"))
|
|
if raw_attention == ["*"]:
|
|
attention_modes = VALID_ATTENTION_MODES
|
|
else:
|
|
attention_modes = [a for a in raw_attention if a in VALID_ATTENTION_MODES] or ["default"]
|
|
|
|
# Per-config resolutions (override sampler's resolutions_json when present)
|
|
raw_resolutions = entry.get("resolutions", None)
|
|
if raw_resolutions and isinstance(raw_resolutions, list) and len(raw_resolutions) > 0:
|
|
config_resolutions = [tuple(r) for r in raw_resolutions] # [(w, h), ...]
|
|
else:
|
|
config_resolutions = [None] # Single None = no override, use global resolutions
|
|
|
|
# Extra Model & Sampling Options
|
|
# Model sampling override: can be "none" or a specific type
|
|
raw_model_sampling = entry.get("model_sampling_override", "none")
|
|
if isinstance(raw_model_sampling, list):
|
|
model_sampling_overrides = raw_model_sampling
|
|
else:
|
|
model_sampling_overrides = [raw_model_sampling]
|
|
|
|
# Model sampling shift values (comma-separated string -> list of floats)
|
|
raw_shift = entry.get("model_sampling_shift", "1.73")
|
|
model_sampling_shifts = [float(s.strip()) for s in str(raw_shift).split(",") if s.strip()]
|
|
if not model_sampling_shifts:
|
|
model_sampling_shifts = [1.73]
|
|
|
|
# Flux-specific shift values
|
|
raw_flux_max = entry.get("model_sampling_flux_max_shift", "1.15")
|
|
flux_max_shifts = [float(s.strip()) for s in str(raw_flux_max).split(",") if s.strip()]
|
|
if not flux_max_shifts:
|
|
flux_max_shifts = [1.15]
|
|
|
|
raw_flux_base = entry.get("model_sampling_flux_base_shift", "0.5")
|
|
flux_base_shifts = [float(s.strip()) for s in str(raw_flux_base).split(",") if s.strip()]
|
|
if not flux_base_shifts:
|
|
flux_base_shifts = [0.5]
|
|
|
|
# Advanced sampling pipeline
|
|
use_advanced = entry.get("use_advanced_sampling", False)
|
|
if use_advanced:
|
|
advanced_sampling_values = [True, False] # Grid: test with and without
|
|
else:
|
|
advanced_sampling_values = [False]
|
|
|
|
raw_guider = entry.get("advanced_guider", "cfg_guider")
|
|
advanced_guiders = to_list(raw_guider) if use_advanced else ["cfg_guider"]
|
|
|
|
raw_scheduler_adv = entry.get("advanced_scheduler", "basic")
|
|
advanced_schedulers = to_list(raw_scheduler_adv) if use_advanced else ["basic"]
|
|
|
|
# Flux guidance
|
|
use_flux_guid = entry.get("use_flux_guidance", False)
|
|
if use_flux_guid:
|
|
raw_guid_val = entry.get("flux_guidance_value", "3.5")
|
|
flux_guidance_values = [float(v.strip()) for v in str(raw_guid_val).split(",") if v.strip()]
|
|
if not flux_guidance_values:
|
|
flux_guidance_values = [3.5]
|
|
else:
|
|
flux_guidance_values = [0.0] # 0.0 = disabled sentinel
|
|
|
|
# Build all combinations
|
|
base_combos = []
|
|
for combo in itertools.product(samplers, schedulers, steps_l, cfgs, clip_skips, expanded_loras,
|
|
denoise_values, entry_prompt_pairs, expanded_models, raw_vaes,
|
|
attention_modes, model_sampling_overrides, model_sampling_shifts,
|
|
flux_max_shifts, flux_base_shifts,
|
|
advanced_sampling_values, advanced_guiders, advanced_schedulers,
|
|
flux_guidance_values, config_resolutions):
|
|
# Skip redundant model sampling parameter combinations
|
|
override = combo[11]
|
|
if override == "none":
|
|
# Skip all but first shift values when override is disabled
|
|
if combo[12] != model_sampling_shifts[0] or combo[13] != flux_max_shifts[0] or combo[14] != flux_base_shifts[0]:
|
|
continue
|
|
elif override == "flux":
|
|
# Skip non-flux shift values
|
|
if combo[12] != model_sampling_shifts[0]:
|
|
continue
|
|
else: # aura_flow or sd3
|
|
# Skip flux shift values
|
|
if combo[13] != flux_max_shifts[0] or combo[14] != flux_base_shifts[0]:
|
|
continue
|
|
|
|
# Skip advanced sampling sub-options when advanced sampling is off
|
|
if not combo[15]: # use_advanced_sampling == False
|
|
if combo[16] != advanced_guiders[0] or combo[17] != advanced_schedulers[0]:
|
|
continue
|
|
|
|
base_combos.append({
|
|
"sampler": combo[0],
|
|
"scheduler": combo[1],
|
|
"steps": combo[2],
|
|
"cfg": combo[3],
|
|
"clip_skip": combo[4],
|
|
"lora": combo[5],
|
|
"denoise": combo[6],
|
|
"positive": combo[7][0],
|
|
"negative": combo[7][1],
|
|
"model": combo[8],
|
|
"vae": combo[9],
|
|
"attention_mode": combo[10],
|
|
"model_sampling_override": combo[11],
|
|
"model_sampling_shift": combo[12],
|
|
"model_sampling_flux_max_shift": combo[13],
|
|
"model_sampling_flux_base_shift": combo[14],
|
|
"use_advanced_sampling": combo[15],
|
|
"advanced_guider": combo[16],
|
|
"advanced_scheduler": combo[17],
|
|
"flux_guidance_value": combo[18],
|
|
"use_deep_shrink": entry.get("use_deep_shrink", False),
|
|
"deep_shrink_block_number": entry.get("deep_shrink_block_number", 3),
|
|
"deep_shrink_downscale_factor": entry.get("deep_shrink_downscale_factor", 2.0),
|
|
"deep_shrink_start_percent": entry.get("deep_shrink_start_percent", 0.0),
|
|
"deep_shrink_end_percent": entry.get("deep_shrink_end_percent", 0.35),
|
|
"deep_shrink_downscale_after_skip": entry.get("deep_shrink_downscale_after_skip", True),
|
|
"deep_shrink_downscale_method": entry.get("deep_shrink_downscale_method", "bicubic"),
|
|
"deep_shrink_upscale_method": entry.get("deep_shrink_upscale_method", "bicubic"),
|
|
"resolution": combo[19], # (w, h) tuple or None
|
|
# Per-entry seed override (Revise modal sets this explicitly).
|
|
# Builder UI does NOT emit entry["seed"] — it uses the node-level
|
|
# seed widget instead — so this fall-through is safe for that flow.
|
|
"seed": int(entry["seed"]) if entry.get("seed") not in (None, "") else seed,
|
|
"seed_behavior": entry.get("seed_behavior", "fixed"),
|
|
"full_run_seed_behavior": entry.get("full_run_seed_behavior", "fixed"),
|
|
"model_type": model_type,
|
|
"clip_type": clip_type_str,
|
|
"text_encoders": list(text_encoders),
|
|
"gguf_options": dict(gguf_options) if gguf_options else {},
|
|
"lora_omit_triggers": list(lora_omit_triggers),
|
|
"lora_triggerwords_append_settings": dict(lora_triggerwords_append_settings),
|
|
"model_prompt_prefix": model_prompt_prefix.strip(),
|
|
"model_prompt_suffix": model_prompt_suffix.strip()
|
|
})
|
|
|
|
# Apply full_run_seed override (per-config seed that overrides node seed)
|
|
full_run_seed = entry.get("full_run_seed", 0)
|
|
if full_run_seed and int(full_run_seed) > 0:
|
|
for c in base_combos:
|
|
c["seed"] = int(full_run_seed)
|
|
|
|
# Apply base seed and extra seeds
|
|
for c in base_combos:
|
|
expanded.append(c)
|
|
for extra_seed in extra_seeds:
|
|
new_c = copy.deepcopy(c)
|
|
new_c["seed"] = extra_seed
|
|
expanded.append(new_c)
|
|
|
|
return expanded
|
|
|
|
|
|
def prepare_input_jobs(optional_latent, resolutions):
|
|
"""
|
|
Prepare input jobs from either optional latent or resolution list.
|
|
|
|
Args:
|
|
optional_latent: Optional latent tensor input
|
|
resolutions: List of [width, height] pairs
|
|
|
|
Returns:
|
|
List of job dictionaries with label, width, height, latent, batch_idx
|
|
"""
|
|
input_jobs = []
|
|
|
|
if optional_latent is not None:
|
|
batch_count = optional_latent["samples"].shape[0]
|
|
for i in range(batch_count):
|
|
single_sample = optional_latent["samples"][i].unsqueeze(0)
|
|
input_jobs.append({
|
|
"label": f"Input {i+1}",
|
|
"width": single_sample.shape[3] * 8,
|
|
"height": single_sample.shape[2] * 8,
|
|
"latent": {"samples": single_sample},
|
|
"batch_idx": i
|
|
})
|
|
else:
|
|
for res in resolutions:
|
|
input_jobs.append({
|
|
"label": f"{res[0]}x{res[1]}",
|
|
"width": res[0],
|
|
"height": res[1],
|
|
"latent": None,
|
|
"batch_idx": 0
|
|
})
|
|
|
|
return input_jobs
|
|
|
|
|
|
def sanitize_session_name(session_name):
|
|
"""Sanitize session name to be filesystem-safe"""
|
|
session_name = re.sub(r'[^\w\-]', '', session_name)
|
|
if not session_name:
|
|
session_name = "default_session"
|
|
return session_name |