Jason HokuandClaude Opus 4.7
7a2c3c905e
Phase 1 of consolidating run-level config into the Builder UI. The Builder UI's "Run Settings" section is now rendered at the TOP of the UI (was previously below config arrays) and gains six new fields with help tooltips: - overwrite_existing (bool, default false) - flush_batch_every (int, default 4) - lora_triggerwords_mode (enum: None / Append To End / Append To Start / Read From Config, default None) - save_conditioning_cache_to_file (bool, default false) - enable_model_cache (bool, default false) - vae_batch_size (int, default 4, range -1..64) State plumbs through state_to_configs_json -> _session_settings -> sampler_node.run_tests -> run_generation_loop. Builder UI values override the corresponding widgets on UltimateSamplerGrid when set (non-default). Widget values remain the fallback so existing workflows keep working without changes. Logs an "⚙️ <name> overridden by Builder UI: <value>" line whenever an override fires, so it's obvious in the console. Next: once the user has tested this thoroughly, the corresponding widgets on the SamplerGridTester node can be removed/hidden to simplify the generator UX (deferred — coordinate with user). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1409 lines
62 KiB
Python
1409 lines
62 KiB
Python
"""
|
|
Ultimate Config Builder - Complete HTML UI Version
|
|
ALL data stored in single widget (lora_config)
|
|
Python reads everything from that widget
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import builtins
|
|
import folder_paths
|
|
from typing import List, Dict, Any
|
|
import server
|
|
from aiohttp import web
|
|
import hashlib
|
|
from .network_utils import civitai_fetch_by_hash
|
|
|
|
|
|
def safe_print(*args, **kwargs):
|
|
"""
|
|
Windows-safe print that survives colorama/stdout corruption.
|
|
Falls back to raw sys.__stdout__ if the wrapped stdout is broken,
|
|
and silently drops the message if even that fails. This prevents
|
|
OSError [Errno 22] from Windows console bugs crashing node execution.
|
|
"""
|
|
try:
|
|
builtins.print(*args, **kwargs)
|
|
except (OSError, ValueError):
|
|
try:
|
|
msg = " ".join(str(a) for a in args) + kwargs.get("end", "\n")
|
|
sys.__stdout__.write(msg)
|
|
sys.__stdout__.flush()
|
|
except Exception:
|
|
pass # Drop silently — don't crash the node over a log line
|
|
|
|
|
|
# Shadow the module-level `print` so every print() call in this file routes
|
|
# through safe_print. Without this, sibling methods (process_lora_array,
|
|
# lookup_triggers_endpoint, etc.) still hit the broken stdout and crash with
|
|
# OSError [Errno 22] on Windows — the local `print = safe_print` in
|
|
# generate_config only covers that one function.
|
|
print = safe_print
|
|
|
|
|
|
|
|
|
|
# =============================================================================
|
|
# INLINE LORA_UTILS FUNCTIONS (for compatibility)
|
|
# =============================================================================
|
|
|
|
def load_json_from_file(file_path):
|
|
"""Load JSON data from file"""
|
|
try:
|
|
with open(file_path, 'r') as json_file:
|
|
data = json.load(json_file)
|
|
return data
|
|
except FileNotFoundError:
|
|
return None
|
|
except json.JSONDecodeError:
|
|
print(f"[ConfigBuilder] Error decoding JSON in file: {file_path}")
|
|
return None
|
|
|
|
|
|
def save_dict_to_json(data_dict, file_path):
|
|
"""Save dictionary to JSON file"""
|
|
try:
|
|
# Ensure directory exists
|
|
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
|
with open(file_path, 'w') as json_file:
|
|
json.dump(data_dict, json_file, indent=4)
|
|
print(f"[ConfigBuilder] Data saved to {file_path}")
|
|
except Exception as e:
|
|
print(f"[ConfigBuilder] Error saving JSON to file: {e}")
|
|
|
|
|
|
def calculate_sha256(file_path):
|
|
"""Calculate SHA256 hash of a file"""
|
|
sha256_hash = hashlib.sha256()
|
|
with open(file_path, "rb") as f:
|
|
for chunk in iter(lambda: f.read(4096), b""):
|
|
sha256_hash.update(chunk)
|
|
return sha256_hash.hexdigest()
|
|
|
|
|
|
def get_model_version_info(hash_value):
|
|
"""Fetch model version info from Civitai API using hash"""
|
|
return civitai_fetch_by_hash(hash_value)
|
|
|
|
|
|
def load_and_save_tags(lora_name, force_fetch=False, auto_fetch=True):
|
|
"""
|
|
Load trigger tags for a LoRA, fetching from Civitai API if necessary.
|
|
Caches results to loras_tags.json.
|
|
|
|
Args:
|
|
lora_name: Name of the LoRA file (may include path with / or \\)
|
|
force_fetch: Force fetch from API even if cached
|
|
auto_fetch: If False, don't hash uncached LoRAs (just save empty entry)
|
|
|
|
Returns:
|
|
List of trigger words/tags
|
|
"""
|
|
output_dir = folder_paths.get_output_directory()
|
|
json_tags_path = os.path.join(output_dir, "benchmarks", "loras_tags.json")
|
|
|
|
lora_tags = load_json_from_file(json_tags_path)
|
|
|
|
# Normalize the lora_name to use forward slashes for cache lookup
|
|
normalized_name = lora_name.replace("\\", "/")
|
|
|
|
# Try to find in cache
|
|
output_tags = None
|
|
found_in_cache = False
|
|
|
|
if lora_tags is not None:
|
|
# Try original name first
|
|
if lora_name in lora_tags:
|
|
output_tags = lora_tags[lora_name]
|
|
found_in_cache = True
|
|
# If not found, try normalized name
|
|
elif normalized_name != lora_name and normalized_name in lora_tags:
|
|
output_tags = lora_tags[normalized_name]
|
|
found_in_cache = True
|
|
# If still not found, try backslash version
|
|
else:
|
|
backslash_name = normalized_name.replace("/", "\\")
|
|
if backslash_name in lora_tags:
|
|
output_tags = lora_tags[backslash_name]
|
|
found_in_cache = True
|
|
|
|
# If found in cache, return immediately
|
|
if found_in_cache:
|
|
return output_tags if output_tags is not None else []
|
|
|
|
# NOT IN CACHE - Need to fetch from API
|
|
lora_path = folder_paths.get_full_path("loras", lora_name)
|
|
|
|
# Check if lora_path is valid before attempting to hash
|
|
if lora_path is None:
|
|
print(f"[ConfigBuilder] ⚠️ LoRA file not found: {lora_name}")
|
|
return []
|
|
|
|
# Only hash if force_fetch is True OR (not in cache AND auto_fetch is True)
|
|
if force_fetch or auto_fetch:
|
|
print(f"[ConfigBuilder] 🔄 Fetching tags for {lora_name}")
|
|
try:
|
|
LORAsha256 = calculate_sha256(lora_path)
|
|
model_info = get_model_version_info(LORAsha256)
|
|
|
|
if model_info is not None and "trainedWords" in model_info:
|
|
print(f"[ConfigBuilder] ✅ Tags found for {lora_name}")
|
|
if lora_tags is None:
|
|
lora_tags = {}
|
|
lora_tags[normalized_name] = model_info["trainedWords"]
|
|
save_dict_to_json(lora_tags, json_tags_path)
|
|
return model_info["trainedWords"]
|
|
else:
|
|
print(f"[ConfigBuilder] No tags found for {lora_name}")
|
|
if lora_tags is None:
|
|
lora_tags = {}
|
|
lora_tags[normalized_name] = []
|
|
save_dict_to_json(lora_tags, json_tags_path)
|
|
return []
|
|
except Exception as e:
|
|
print(f"[ConfigBuilder] Error processing {lora_name}: {e}")
|
|
return []
|
|
else:
|
|
# Not in cache and auto_fetch=False
|
|
if lora_tags is None:
|
|
lora_tags = {}
|
|
lora_tags[normalized_name] = []
|
|
save_dict_to_json(lora_tags, json_tags_path)
|
|
return []
|
|
|
|
|
|
class UltimateConfigBuilder:
|
|
"""
|
|
Config builder with complete HTML UI.
|
|
All data is stored in the lora_config widget as a single JSON object.
|
|
"""
|
|
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
sessions = cls.get_available_sessions()
|
|
|
|
return {
|
|
"required": {
|
|
# Session Management (hidden, controlled by HTML)
|
|
"session_name": ("STRING", {
|
|
"default": "my_test_session",
|
|
"multiline": False
|
|
}),
|
|
"load_session": (sessions, {
|
|
"default": sessions[0] if sessions else "None"
|
|
}),
|
|
|
|
# Sampler Settings (hidden, controlled by HTML)
|
|
"samplers": ("STRING", {
|
|
"default": "euler, dpmpp_2m",
|
|
"multiline": False
|
|
}),
|
|
"schedulers": ("STRING", {
|
|
"default": "normal, karras",
|
|
"multiline": False
|
|
}),
|
|
"steps": ("STRING", {
|
|
"default": "20, 30",
|
|
"multiline": False
|
|
}),
|
|
"cfg": ("STRING", {
|
|
"default": "7.0",
|
|
"multiline": False
|
|
}),
|
|
|
|
# LoRA Configuration (ACTUAL DATA STORAGE - contains EVERYTHING)
|
|
"lora_config": ("STRING", {
|
|
"default": cls.get_default_config(),
|
|
"multiline": True
|
|
}),
|
|
|
|
# Options (hidden, controlled by HTML)
|
|
"include_none": ("BOOLEAN", {
|
|
"default": False
|
|
}),
|
|
},
|
|
"optional": {
|
|
"model": ("STRING", {
|
|
"default": "",
|
|
"multiline": False
|
|
}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("STRING", "STRING")
|
|
RETURN_NAMES = ("configs_json", "session_name")
|
|
FUNCTION = "generate_config"
|
|
CATEGORY = "sampling/testing"
|
|
OUTPUT_NODE = True
|
|
|
|
@staticmethod
|
|
def get_default_config():
|
|
"""Return default complete configuration"""
|
|
config = {
|
|
"session_name": "my_test_session",
|
|
"include_none": False,
|
|
"global_positive_groups": [],
|
|
"global_negative": "",
|
|
"config_arrays": [
|
|
{
|
|
"name": "Config 1",
|
|
"samplers": "euler, dpmpp_2m",
|
|
"schedulers": "normal, karras",
|
|
"steps": "20, 30",
|
|
"cfg": "7.0",
|
|
"model": "",
|
|
"loras": ["None"],
|
|
"lora_omit_triggers": [],
|
|
"lora_triggerwords_append_settings": {},
|
|
"combine": True,
|
|
"positive_prompt_groups": [],
|
|
"negative_prompt": "",
|
|
"use_custom_prompts": False
|
|
}
|
|
]
|
|
}
|
|
return json.dumps(config, indent=2, ensure_ascii=False)
|
|
|
|
@staticmethod
|
|
def get_available_sessions() -> List[str]:
|
|
"""Scan benchmarks folder for available sessions, sorted by newest first"""
|
|
sessions = ["None"]
|
|
try:
|
|
output_dir = folder_paths.get_output_directory()
|
|
benchmarks_dir = os.path.join(output_dir, "benchmarks")
|
|
|
|
if os.path.exists(benchmarks_dir):
|
|
session_items = []
|
|
for item in os.listdir(benchmarks_dir):
|
|
item_path = os.path.join(benchmarks_dir, item)
|
|
manifest_path = os.path.join(item_path, "manifest.json")
|
|
|
|
if os.path.isdir(item_path) and os.path.exists(manifest_path):
|
|
# Get modification time for sorting
|
|
mtime = os.path.getmtime(item_path)
|
|
session_items.append((item, mtime))
|
|
|
|
# Sort by modification time (newest first)
|
|
session_items.sort(key=lambda x: x[1], reverse=True)
|
|
sessions.extend([item[0] for item in session_items])
|
|
except Exception as e:
|
|
print(f"[ConfigBuilder] Warning: Could not scan sessions: {e}")
|
|
|
|
return sessions
|
|
|
|
@staticmethod
|
|
def expand_lora_folders(lora_list: List[str]) -> List[str]:
|
|
"""
|
|
Expand folder references to individual LoRA files.
|
|
|
|
Args:
|
|
lora_list: List of LoRA strings (may include folders)
|
|
|
|
Returns:
|
|
List with folders expanded to individual files
|
|
"""
|
|
expanded = []
|
|
available_loras = folder_paths.get_filename_list("loras")
|
|
|
|
for lora_str in lora_list:
|
|
if not lora_str or lora_str == "None":
|
|
continue
|
|
|
|
# Parse out the name (before any : strength modifiers)
|
|
lora_name = lora_str.split(":")[0]
|
|
|
|
# Check if it's a folder reference
|
|
is_folder = lora_name.endswith("/") or lora_name.endswith("/*")
|
|
|
|
if is_folder:
|
|
# Remove trailing / or /*
|
|
folder_name = lora_name.rstrip("/*").rstrip("/")
|
|
|
|
# Normalize to forward slashes for comparison
|
|
folder_prefix = folder_name.replace("\\", "/") + "/"
|
|
|
|
# Find all LoRAs in this folder
|
|
for lora_file in available_loras:
|
|
normalized_file = lora_file.replace("\\", "/")
|
|
if normalized_file.startswith(folder_prefix):
|
|
expanded.append(lora_file)
|
|
|
|
lora_count = len([l for l in available_loras if l.replace('\\', '/').startswith(folder_prefix)])
|
|
print(f"[ConfigBuilder] Expanded folder '{lora_name}' to {lora_count} LoRAs")
|
|
else:
|
|
# Regular LoRA file
|
|
expanded.append(lora_str)
|
|
|
|
return expanded
|
|
|
|
@staticmethod
|
|
def lookup_lora_triggers(lora_list: List[str]) -> Dict[str, List[str]]:
|
|
"""
|
|
Lookup trigger words for a list of LoRAs.
|
|
|
|
Args:
|
|
lora_list: List of LoRA strings (may include strengths and folders)
|
|
|
|
Returns:
|
|
Dict mapping lora name to list of trigger words
|
|
"""
|
|
# First, expand any folder references
|
|
expanded_list = UltimateConfigBuilder.expand_lora_folders(lora_list)
|
|
|
|
trigger_map = {}
|
|
|
|
for lora_str in expanded_list:
|
|
if not lora_str or lora_str == "None":
|
|
continue
|
|
|
|
# Handle combined LoRAs (e.g., "lora1 + lora2")
|
|
if " + " in lora_str:
|
|
parts = lora_str.split(" + ")
|
|
for part in parts:
|
|
part = part.strip()
|
|
if part and part != "None":
|
|
lora_name = part.split(":")[0]
|
|
if lora_name not in trigger_map:
|
|
try:
|
|
triggers = load_and_save_tags(
|
|
lora_name,
|
|
force_fetch=False,
|
|
auto_fetch=True
|
|
)
|
|
trigger_map[lora_name] = triggers if triggers else []
|
|
except Exception as e:
|
|
print(f"[ConfigBuilder] Error fetching triggers for {lora_name}: {e}")
|
|
trigger_map[lora_name] = []
|
|
else:
|
|
# Single LoRA
|
|
lora_name = lora_str.split(":")[0]
|
|
if lora_name and lora_name != "None" and lora_name not in trigger_map:
|
|
try:
|
|
triggers = load_and_save_tags(
|
|
lora_name,
|
|
force_fetch=False,
|
|
auto_fetch=True
|
|
)
|
|
trigger_map[lora_name] = triggers if triggers else []
|
|
except Exception as e:
|
|
print(f"[ConfigBuilder] Error fetching triggers for {lora_name}: {e}")
|
|
trigger_map[lora_name] = []
|
|
|
|
return trigger_map
|
|
|
|
@staticmethod
|
|
def parse_int_list(value: str) -> List[int]:
|
|
"""Parse comma-separated integers"""
|
|
items = UltimateConfigBuilder.parse_comma_list(value)
|
|
result = []
|
|
for item in items:
|
|
try:
|
|
# Cast to float first to handle strings like "20.0", then to int
|
|
result.append(int(float(item)))
|
|
except ValueError:
|
|
print(f"[ConfigBuilder] Warning: Could not parse integer '{item}'")
|
|
return result
|
|
|
|
@staticmethod
|
|
def parse_comma_list(value) -> List[str]:
|
|
"""Parse comma-separated string or pass through list"""
|
|
if isinstance(value, list):
|
|
return [str(item).strip() for item in value if str(item).strip()]
|
|
if not value or str(value).strip() == "":
|
|
return []
|
|
return [item.strip() for item in str(value).split(",") if item.strip()]
|
|
|
|
@staticmethod
|
|
def parse_number_list(value: str) -> List[float]:
|
|
"""Parse comma-separated numbers"""
|
|
items = UltimateConfigBuilder.parse_comma_list(value)
|
|
result = []
|
|
for item in items:
|
|
try:
|
|
result.append(float(item))
|
|
except ValueError:
|
|
print(f"[ConfigBuilder] Warning: Could not parse '{item}'")
|
|
return result
|
|
|
|
@staticmethod
|
|
def process_lora_array(config_array: Dict, include_none: bool) -> List[str]:
|
|
"""
|
|
Process a SINGLE config array and return its lora strings.
|
|
|
|
Args:
|
|
config_array: Single config array dict from config_arrays
|
|
include_none: Whether to include "None" in results
|
|
|
|
Returns:
|
|
List of lora strings for this config array
|
|
"""
|
|
array_name = config_array.get("name", "Unnamed Config")
|
|
|
|
# FIX: Force combine to True.
|
|
# The UI defaults new configs to 'combine: false' but the Preview
|
|
# treats them as combined. We enforce True here to match the Preview/Stack behavior.
|
|
combine = True
|
|
|
|
loras = config_array.get("loras", [])
|
|
lora_bypass_states = config_array.get("lora_bypass_states", {})
|
|
|
|
# Filter out bypassed loras, then convert to strings
|
|
lora_strings = []
|
|
for lora in loras:
|
|
if not lora or lora == "None":
|
|
continue
|
|
lora_str = str(lora)
|
|
# Extract lora name (path before first colon) to check bypass state
|
|
lora_name = lora_str.split(":")[0] if ":" in lora_str else lora_str
|
|
if lora_bypass_states.get(lora_name, False):
|
|
continue # Skip bypassed
|
|
lora_strings.append(lora_str)
|
|
|
|
# Apply weight arrays (bracket notation) for "Compare Strengths" entries.
|
|
# The Builder UI stores per-LoRA strength arrays in a side-channel
|
|
# (config_array.lora_weight_arrays[name + "_model" / "_clip"]). Rewrite
|
|
# each lora string from "name:m:c" -> "name:[m_arr]:[c_arr|c]" so the
|
|
# orchestrator's _expand_lora_weight_arrays can fan them out at runtime.
|
|
# SYNC WARNING: must match convertStateToConfigs() in conf-builder-utilities.js.
|
|
weight_arrays = config_array.get("lora_weight_arrays", {}) or {}
|
|
|
|
def _apply_weight_array(lora_str: str) -> str:
|
|
name, _sep, rest = lora_str.partition(":")
|
|
parts = rest.split(":") if rest else []
|
|
model_str = parts[0] if parts else "1.00"
|
|
clip_str = parts[1] if len(parts) > 1 else model_str
|
|
model_arr = weight_arrays.get(name + "_model")
|
|
clip_arr = weight_arrays.get(name + "_clip")
|
|
if model_arr and len(model_arr) > 1:
|
|
model_part = "[" + ", ".join(str(v) for v in model_arr) + "]"
|
|
if clip_arr and len(clip_arr) > 1:
|
|
# Legacy/edge case: separate clip array → preserves Cartesian behavior
|
|
# for backward-compat with workflows saved before the locked-semantics
|
|
# simplification (2026-04-28).
|
|
clip_part = "[" + ", ".join(str(v) for v in clip_arr) + "]"
|
|
return f"{name}:{model_part}:{clip_part}"
|
|
# Locked semantics: omit the clip part entirely.
|
|
# parse_lora_definition (config_utils.py) defaults clip = model when no
|
|
# third segment is present, so "name:[a,b,c]" expands to N configs each
|
|
# with model=clip=value. This avoids the Cartesian explosion that
|
|
# "name:[m]:[c]" would trigger in expand_lora_stack.
|
|
return f"{name}:{model_part}"
|
|
if clip_arr and len(clip_arr) > 1:
|
|
# Edge: clip-only array (no model array) — emit as before.
|
|
return f"{name}:{model_str}:[" + ", ".join(str(v) for v in clip_arr) + "]"
|
|
return lora_str
|
|
|
|
lora_strings = [_apply_weight_array(s) for s in lora_strings]
|
|
|
|
# Add combined version if requested
|
|
if combine and len(lora_strings) > 1:
|
|
# Folder refs ("name/") are excluded from stacking — they expand at
|
|
# runtime into multiple loras. Detect by the NAME portion (before the
|
|
# first colon) ending in "/", not the whole string, so that
|
|
# "folder/:[a,b]:[c,d]" (folder ref + strength array) is also excluded.
|
|
stackable = [s for s in lora_strings if not s.split(":", 1)[0].endswith("/")]
|
|
if len(stackable) > 1:
|
|
# When combine is true, ONLY return the combined version
|
|
combined = " + ".join(stackable)
|
|
lora_strings = [combined]
|
|
print(f"[ConfigBuilder] {array_name}: Combined {len(stackable)} LoRAs into stack")
|
|
|
|
# Remove duplicates while preserving order
|
|
seen = set()
|
|
unique_strings = []
|
|
for item in lora_strings:
|
|
if item not in seen:
|
|
seen.add(item)
|
|
unique_strings.append(item)
|
|
|
|
print(f"[ConfigBuilder] {array_name}: Processed {len(unique_strings)} LoRA configs")
|
|
return unique_strings
|
|
|
|
@staticmethod
|
|
def state_to_configs_json(state: dict) -> str:
|
|
"""
|
|
Pure transformer: builder UI state -> configs_json string.
|
|
|
|
Single source of truth. Called by:
|
|
- generate_config() (run-time, via lora_config widget)
|
|
- /configbuilder/preview (preview endpoint, via POST body)
|
|
|
|
Both code paths share this function. By construction, the preview
|
|
and the actual node output cannot disagree.
|
|
|
|
Args:
|
|
state: parsed builder UI state (the JSON object stored in the
|
|
lora_config widget — config_arrays, prompts, etc.)
|
|
Returns:
|
|
The configs_json string (top-level: {"configs": [...], "_distribution": ..., "_session_settings": ...}).
|
|
"""
|
|
# SINGLE SOURCE OF TRUTH for builder UI state -> configs_json.
|
|
|
|
actual_include_none = state.get("include_none", False)
|
|
config_arrays = state.get("config_arrays", [])
|
|
|
|
# Global prompts (used when per-config prompts are not defined)
|
|
global_positive_groups = state.get("global_positive_groups", [])
|
|
global_negative = state.get("global_negative", "")
|
|
|
|
if not config_arrays:
|
|
config_arrays = [{
|
|
"name": "Config 1",
|
|
"samplers": ["euler"],
|
|
"schedulers": ["normal"],
|
|
"steps": "20",
|
|
"cfg": "7.0",
|
|
"model": "",
|
|
"loras": ["None"],
|
|
"lora_omit_triggers": [],
|
|
"lora_triggerwords_append_settings": {},
|
|
"combine": True,
|
|
"positive_prompt_groups": [],
|
|
"negative_prompt": "",
|
|
"use_custom_prompts": False
|
|
}]
|
|
|
|
configs_output = []
|
|
total_lora_configs = 0
|
|
|
|
for config_array in config_arrays:
|
|
# Parse values from this config array
|
|
sampler_list = UltimateConfigBuilder.parse_comma_list(config_array.get("samplers", "euler"))
|
|
scheduler_list = UltimateConfigBuilder.parse_comma_list(config_array.get("schedulers", "normal"))
|
|
steps_list = UltimateConfigBuilder.parse_int_list(config_array.get("steps", "20"))
|
|
cfg_list = UltimateConfigBuilder.parse_number_list(config_array.get("cfg", "7.0"))
|
|
models_raw = config_array.get("models", ["None"])
|
|
omit_triggers = config_array.get("lora_omit_triggers", [])
|
|
lora_triggerwords_append_settings = config_array.get("lora_triggerwords_append_settings", {})
|
|
|
|
# Extra Model & Sampling Options
|
|
model_sampling_override = config_array.get("model_sampling_override", "none")
|
|
model_sampling_shift = config_array.get("model_sampling_shift", "1.73")
|
|
model_sampling_flux_max_shift = config_array.get("model_sampling_flux_max_shift", "1.15")
|
|
model_sampling_flux_base_shift = config_array.get("model_sampling_flux_base_shift", "0.5")
|
|
use_advanced_sampling = config_array.get("use_advanced_sampling", False)
|
|
advanced_guider = config_array.get("advanced_guider", "cfg_guider")
|
|
advanced_scheduler = config_array.get("advanced_scheduler", "basic")
|
|
use_flux_guidance = config_array.get("use_flux_guidance", False)
|
|
flux_guidance_value = config_array.get("flux_guidance_value", "3.5")
|
|
|
|
# Process models - handle both object format {path, type} and legacy string format
|
|
model_strings = []
|
|
model_type = "checkpoint" # default
|
|
model_bypass_states = config_array.get("model_bypass_states", {})
|
|
for m in models_raw:
|
|
if isinstance(m, dict):
|
|
# Trust the user's type selection even if no file is picked yet
|
|
# (otherwise switching to LTX/diffusion_model type without yet
|
|
# selecting a file silently falls back to checkpoint).
|
|
if m.get("type"):
|
|
model_type = m.get("type", "checkpoint")
|
|
path = m.get("path", "")
|
|
if path and path != "None" and not model_bypass_states.get(path, False):
|
|
model_strings.append(str(path))
|
|
elif isinstance(m, str) and m and m != "None" and not model_bypass_states.get(m, False):
|
|
model_strings.append(str(m))
|
|
|
|
# Process loras for this config
|
|
lora_strings = UltimateConfigBuilder.process_lora_array(config_array, actual_include_none)
|
|
total_lora_configs += len(lora_strings)
|
|
|
|
# Create ONE config for this array
|
|
config = {
|
|
"sampler": sampler_list if len(sampler_list) > 1 else sampler_list[0] if sampler_list else "euler",
|
|
"scheduler": scheduler_list if len(scheduler_list) > 1 else scheduler_list[0] if scheduler_list else "normal",
|
|
"steps": steps_list if len(steps_list) > 1 else steps_list[0] if steps_list else 20,
|
|
"cfg": cfg_list if len(cfg_list) > 1 else cfg_list[0] if cfg_list else 7.0,
|
|
"lora": lora_strings if len(lora_strings) > 1 else lora_strings[0] if lora_strings else "None",
|
|
"model": model_strings if len(model_strings) > 1 else model_strings[0] if model_strings else "None"
|
|
}
|
|
|
|
# Always include all fields with defaults to prevent manifest data loss
|
|
# when fields are toggled off in the UI and the session is reloaded
|
|
seed_behavior = config_array.get("seed_behavior", "fixed")
|
|
config["seed_behavior"] = seed_behavior
|
|
|
|
# Full run seed behavior (applied before/after entire grid test session)
|
|
full_run_seed_behavior = config_array.get("full_run_seed_behavior", "fixed")
|
|
config["full_run_seed_behavior"] = full_run_seed_behavior
|
|
|
|
# Full run seed (overrides node seed when > 0)
|
|
full_run_seed = config_array.get("full_run_seed", 0)
|
|
config["full_run_seed"] = int(full_run_seed) if full_run_seed else 0
|
|
|
|
# Process VAEs — always include (default "None" if unset)
|
|
vaes_raw = config_array.get("vaes", ["None"])
|
|
vae_strings = [str(v) for v in vaes_raw if v and v != "None"]
|
|
if vae_strings:
|
|
config["vae"] = vae_strings if len(vae_strings) > 1 else vae_strings[0]
|
|
else:
|
|
config["vae"] = "None"
|
|
|
|
# Always include model_type and related fields
|
|
config["model_type"] = model_type
|
|
if model_type != "checkpoint":
|
|
text_encoders = config_array.get("text_encoders", [])
|
|
if text_encoders:
|
|
config["text_encoders"] = [te for te in text_encoders if te and te != "None"]
|
|
clip_type = config_array.get("clip_type", "")
|
|
if clip_type:
|
|
config["clip_type"] = clip_type
|
|
if model_type == "gguf":
|
|
gguf_options = config_array.get("gguf_options", {})
|
|
if gguf_options:
|
|
config["gguf_options"] = gguf_options
|
|
if model_type == "ltx_video":
|
|
# Flatten configArray.ltx_video into top-level fields for the orchestrator.
|
|
# Mirrors the JS-side flattening in conf-builder-utilities.convertStateToConfigs.
|
|
ltx = config_array.get("ltx_video") or {}
|
|
if ltx.get("clip_models"):
|
|
config["clip_models"] = ltx["clip_models"]
|
|
if ltx.get("vae_video"):
|
|
config["vae_video"] = ltx["vae_video"]
|
|
if ltx.get("vae_audio"):
|
|
config["vae_audio"] = ltx["vae_audio"]
|
|
if ltx.get("latent_upscaler"):
|
|
config["latent_upscaler"] = ltx["latent_upscaler"]
|
|
if ltx.get("duration_seconds") is not None:
|
|
config["duration_seconds"] = ltx["duration_seconds"]
|
|
if ltx.get("frame_rate") is not None:
|
|
config["frame_rate"] = ltx["frame_rate"]
|
|
if ltx.get("sampler_stage1"):
|
|
config["sampler_stage1"] = ltx["sampler_stage1"]
|
|
if ltx.get("sampler_stage2"):
|
|
config["sampler_stage2"] = ltx["sampler_stage2"]
|
|
if ltx.get("sigmas_stage1"):
|
|
config["sigmas_stage1"] = ltx["sigmas_stage1"]
|
|
if ltx.get("sigmas_stage2"):
|
|
config["sigmas_stage2"] = ltx["sigmas_stage2"]
|
|
if ltx.get("input_image") is not None:
|
|
config["input_image"] = ltx["input_image"]
|
|
if ltx.get("image_strength_stage1") is not None:
|
|
config["image_strength_stage1"] = ltx["image_strength_stage1"]
|
|
if ltx.get("image_strength_stage2") is not None:
|
|
config["image_strength_stage2"] = ltx["image_strength_stage2"]
|
|
if ltx.get("img_compression") is not None:
|
|
config["img_compression"] = ltx["img_compression"]
|
|
if ltx.get("audio_mode"):
|
|
config["audio_mode"] = ltx["audio_mode"]
|
|
|
|
# Always include omit triggers (empty list if none)
|
|
config["lora_omit_triggers"] = omit_triggers if omit_triggers else []
|
|
|
|
# Always include trigger append settings (empty dict if none)
|
|
config["lora_triggerwords_append_settings"] = lora_triggerwords_append_settings if lora_triggerwords_append_settings else {}
|
|
|
|
# Per-config resolutions (override sampler's resolutions_json)
|
|
raw_resolutions = config_array.get("resolutions", [])
|
|
if raw_resolutions and len(raw_resolutions) > 0:
|
|
# Convert "WxH" strings to [W, H] arrays for config_utils.expand_configs()
|
|
parsed_res = []
|
|
for r in raw_resolutions:
|
|
if isinstance(r, str) and "x" in r:
|
|
parts = r.split("x")
|
|
parsed_res.append([int(parts[0]), int(parts[1])])
|
|
elif isinstance(r, (list, tuple)) and len(r) == 2:
|
|
parsed_res.append([int(r[0]), int(r[1])])
|
|
if parsed_res:
|
|
config["resolutions"] = parsed_res
|
|
|
|
# Attention mode(s) for testing different attention implementations
|
|
attention_modes = config_array.get("attention_modes", ["default"])
|
|
if isinstance(attention_modes, list):
|
|
filtered = [a for a in attention_modes if a and a != "default"]
|
|
if filtered:
|
|
config["attention_mode"] = filtered if len(filtered) > 1 else filtered[0]
|
|
else:
|
|
config["attention_mode"] = "default"
|
|
elif isinstance(attention_modes, str):
|
|
config["attention_mode"] = attention_modes
|
|
else:
|
|
config["attention_mode"] = "default"
|
|
|
|
# Model prompt prefix/suffix (quality tags prepended/appended to prompts)
|
|
model_prompt_prefix = config_array.get("model_prompt_prefix", "")
|
|
config["model_prompt_prefix"] = model_prompt_prefix.strip() if model_prompt_prefix else ""
|
|
model_prompt_suffix = config_array.get("model_prompt_suffix", "")
|
|
config["model_prompt_suffix"] = model_prompt_suffix.strip() if model_prompt_suffix else ""
|
|
|
|
# Always include model sampling options with defaults
|
|
config["model_sampling_override"] = model_sampling_override if model_sampling_override else "none"
|
|
if model_sampling_override and model_sampling_override != "none":
|
|
if model_sampling_override == "flux":
|
|
config["model_sampling_flux_max_shift"] = model_sampling_flux_max_shift
|
|
config["model_sampling_flux_base_shift"] = model_sampling_flux_base_shift
|
|
elif model_sampling_override == "flux2":
|
|
config["model_sampling_shift"] = model_sampling_shift if model_sampling_shift else "2.02"
|
|
else:
|
|
config["model_sampling_shift"] = model_sampling_shift
|
|
config["use_advanced_sampling"] = use_advanced_sampling or False
|
|
if use_advanced_sampling:
|
|
config["advanced_guider"] = advanced_guider
|
|
config["advanced_scheduler"] = advanced_scheduler
|
|
config["use_flux_guidance"] = use_flux_guidance or False
|
|
if use_flux_guidance:
|
|
config["flux_guidance_value"] = flux_guidance_value
|
|
|
|
# ==== PROMPT HANDLING ====
|
|
# Priority: per-config > global > node inputs (omitted = use node inputs)
|
|
use_custom = config_array.get("use_custom_prompts", False)
|
|
per_config_positive_groups = config_array.get("positive_prompt_groups", [])
|
|
per_config_negative = config_array.get("negative_prompt", "")
|
|
|
|
if use_custom and per_config_positive_groups:
|
|
# Per-config prompts override everything
|
|
# Store as nested array format for parse_prompt_input_nested() compatibility
|
|
config["positive"] = per_config_positive_groups
|
|
if per_config_negative:
|
|
config["negative"] = per_config_negative
|
|
config["_prompt_source"] = "custom"
|
|
elif global_positive_groups:
|
|
# Global prompts override node inputs
|
|
config["positive"] = global_positive_groups
|
|
if global_negative:
|
|
config["negative"] = global_negative
|
|
config["_prompt_source"] = "global"
|
|
# If neither, omit "positive"/"negative" keys - node inputs will be used as fallback
|
|
|
|
# --- CONFIG SCHEMA VALIDATION ---
|
|
# Sanity check for typos / dropped fields. Only includes keys this
|
|
# function ALWAYS sets unconditionally — fields that are filled in
|
|
# later by the orchestrator (seed, denoise) or only set for
|
|
# specific model types (clip_type, text_encoders, gguf_options,
|
|
# ltx_*) are intentionally excluded so they don't fire false alarms.
|
|
_EXPECTED_CONFIG_KEYS = {
|
|
"sampler", "scheduler", "steps", "cfg", "seed_behavior",
|
|
"model", "model_type", "lora", "vae",
|
|
"model_sampling_override", "use_advanced_sampling", "use_flux_guidance",
|
|
"model_prompt_prefix", "model_prompt_suffix", "attention_mode",
|
|
}
|
|
missing = _EXPECTED_CONFIG_KEYS - set(config.keys())
|
|
if missing:
|
|
print(f"[ConfigBuilder] ⚠️ Config schema drift: missing keys {missing} in config for '{config.get('model', '?')}'")
|
|
|
|
configs_output.append(config)
|
|
|
|
# Build the output object with configs and optional distribution settings
|
|
output_obj = {"configs": configs_output}
|
|
|
|
# Embed distribution config if enabled
|
|
if state.get("distribution_enabled") and state.get("worker_urls"):
|
|
output_obj["_distribution"] = {
|
|
"enabled": True,
|
|
"worker_urls": [u for u in state["worker_urls"] if u and u.strip()],
|
|
"claim_timeout": state.get("claim_timeout", 600),
|
|
"use_master_encoding": state.get("use_master_encoding", False),
|
|
"sync_models_to_workers": state.get("sync_models_to_workers", False)
|
|
}
|
|
|
|
# Embed session-level settings (upscaling, cooldown) if enabled
|
|
session_settings = {}
|
|
upscaling_data = state.get("upscaling", {})
|
|
if upscaling_data and upscaling_data.get("enabled", False):
|
|
# Filter out inactive pipelines and inactive steps within pipelines
|
|
pipelines = upscaling_data.get("pipelines", [])
|
|
active_pipelines = []
|
|
for p in pipelines:
|
|
if p.get("active", True) is False:
|
|
continue
|
|
active_steps = [s for s in p.get("steps", []) if s.get("active", True) is not False]
|
|
if active_steps:
|
|
active_pipelines.append({**p, "steps": active_steps})
|
|
if active_pipelines:
|
|
session_settings["upscaling"] = {
|
|
"enabled": True,
|
|
"save_pre_upscale": upscaling_data.get("save_pre_upscale", False),
|
|
"run_upscales_at_end": upscaling_data.get("run_upscales_at_end", False),
|
|
"hires_prompt_adjust": upscaling_data.get("hires_prompt_adjust", False),
|
|
"hires_prompt_behavior": upscaling_data.get("hires_prompt_behavior", "append_end"),
|
|
"hires_prompt_text": upscaling_data.get("hires_prompt_text", ""),
|
|
"pipelines": active_pipelines
|
|
}
|
|
cooldown_data = state.get("cooldown", {})
|
|
if cooldown_data and cooldown_data.get("enabled", False):
|
|
session_settings["cooldown"] = cooldown_data
|
|
# Start At Job # (skip to a specific job number)
|
|
start_at_job = state.get("start_at_job", 0)
|
|
if start_at_job and int(start_at_job) > 0:
|
|
session_settings["start_at_job"] = int(start_at_job)
|
|
# Image save format (only emit when non-default to keep configs_json clean)
|
|
image_format = state.get("image_format", "webp")
|
|
if image_format and image_format != "webp":
|
|
session_settings["image_format"] = image_format
|
|
# Generator-node override settings — emit only when non-default so the
|
|
# widget value on the SamplerGridTester node wins for users who haven't
|
|
# explicitly set them in the Builder UI. When the Generator widgets are
|
|
# eventually removed, these can be made unconditional.
|
|
ov_existing = state.get("overwrite_existing", False)
|
|
if ov_existing:
|
|
session_settings["overwrite_existing"] = True
|
|
|
|
flush_be = state.get("flush_batch_every", 4)
|
|
try:
|
|
flush_be = int(flush_be)
|
|
except (TypeError, ValueError):
|
|
flush_be = 4
|
|
if flush_be != 4:
|
|
session_settings["flush_batch_every"] = flush_be
|
|
|
|
lora_tw = state.get("lora_triggerwords_mode", "None")
|
|
if lora_tw and lora_tw != "None":
|
|
session_settings["lora_triggerwords_mode"] = lora_tw
|
|
|
|
save_cc = state.get("save_conditioning_cache_to_file", False)
|
|
if save_cc:
|
|
session_settings["save_conditioning_cache_to_file"] = True
|
|
|
|
en_mc = state.get("enable_model_cache", False)
|
|
if en_mc:
|
|
session_settings["enable_model_cache"] = True
|
|
|
|
vae_bs = state.get("vae_batch_size", 4)
|
|
try:
|
|
vae_bs = int(vae_bs)
|
|
except (TypeError, ValueError):
|
|
vae_bs = 4
|
|
if vae_bs != 4:
|
|
session_settings["vae_batch_size"] = vae_bs
|
|
|
|
if session_settings:
|
|
output_obj["_session_settings"] = session_settings
|
|
|
|
json_output = json.dumps(output_obj, indent=2, ensure_ascii=False)
|
|
return json_output
|
|
|
|
def generate_config(
|
|
self,
|
|
session_name,
|
|
load_session,
|
|
samplers,
|
|
schedulers,
|
|
steps,
|
|
cfg,
|
|
lora_config,
|
|
include_none,
|
|
model=""
|
|
):
|
|
"""
|
|
Generate configuration.
|
|
|
|
NOTE: All widget parameters are IGNORED!
|
|
The actual data comes from the lora_config widget which contains everything.
|
|
"""
|
|
# Use Windows-safe print for all logging in this function. Prevents
|
|
# OSError [Errno 22] from colorama/stdout corruption crashing the node.
|
|
print = safe_print
|
|
|
|
print(f"\n{'='*80}")
|
|
print(f"[ConfigBuilder] 🎯 Generating Configuration")
|
|
print(f"{'='*80}")
|
|
|
|
# Parse the COMPLETE state from lora_config widget
|
|
try:
|
|
state = json.loads(lora_config)
|
|
except json.JSONDecodeError as e:
|
|
print(f"[ConfigBuilder] ⚠️ Error parsing lora_config: {e}")
|
|
print(f"[ConfigBuilder] Using default config")
|
|
state = json.loads(self.get_default_config())
|
|
|
|
actual_session_name = state.get("session_name", session_name)
|
|
|
|
# ============================================================================
|
|
# SINGLE SOURCE OF TRUTH: state_to_configs_json is also called by
|
|
# POST /configbuilder/preview, so the preview cannot disagree with this.
|
|
# ============================================================================
|
|
json_output = UltimateConfigBuilder.state_to_configs_json(state)
|
|
|
|
# Brief summary log (computed from the helper's output)
|
|
try:
|
|
parsed = json.loads(json_output)
|
|
n_configs = len(parsed.get("configs", []))
|
|
except Exception:
|
|
n_configs = 0
|
|
print(f"[ConfigBuilder] 📊 Session: {actual_session_name}")
|
|
print(f"[ConfigBuilder] 📊 Configs: {n_configs}")
|
|
print(f"{'='*80}\n")
|
|
|
|
return (json_output, actual_session_name)
|
|
|
|
|
|
# API endpoint for trigger word lookup
|
|
@server.PromptServer.instance.routes.post("/configbuilder/lookup_triggers")
|
|
async def lookup_triggers_endpoint(request):
|
|
"""API endpoint to lookup trigger words for LoRAs"""
|
|
try:
|
|
data = await request.json()
|
|
lora_list = data.get("loras", [])
|
|
|
|
print(f"[ConfigBuilder] 🔍 Lookup request for {len(lora_list)} LoRAs")
|
|
|
|
trigger_map = UltimateConfigBuilder.lookup_lora_triggers(lora_list)
|
|
|
|
print(f"[ConfigBuilder] ✅ Found triggers for {len(trigger_map)} LoRAs")
|
|
|
|
return web.json_response({
|
|
"triggers": trigger_map
|
|
})
|
|
except Exception as e:
|
|
print(f"[ConfigBuilder] ❌ Error in lookup_triggers endpoint: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return web.json_response({
|
|
"error": str(e)
|
|
}, status=500)
|
|
|
|
|
|
# API endpoint for detailed LoRA metadata lookup
|
|
@server.PromptServer.instance.routes.post("/configbuilder/lookup_lora_metadata")
|
|
async def lookup_lora_metadata_endpoint(request):
|
|
"""API endpoint to lookup full metadata for a specific LoRA from CivitAI"""
|
|
try:
|
|
data = await request.json()
|
|
lora_name = data.get("lora_name", "")
|
|
force_refresh = data.get("force_refresh", False)
|
|
|
|
if not lora_name:
|
|
return web.json_response({
|
|
"error": "No LoRA name provided"
|
|
}, status=400)
|
|
|
|
print(f"[ConfigBuilder] 🔍 Full metadata lookup request for: {lora_name}")
|
|
|
|
# Check disk cache first to avoid expensive SHA256 hashing
|
|
output_dir = folder_paths.get_output_directory()
|
|
model_data_dir = os.path.join(output_dir, "benchmarks", "model-data", lora_name.replace("/", "_").replace("\\", "_").replace(".safetensors", ""))
|
|
metadata_file = os.path.join(model_data_dir, "metadata.json")
|
|
|
|
if os.path.exists(metadata_file) and not force_refresh:
|
|
try:
|
|
cached = load_json_from_file(metadata_file)
|
|
if cached and cached.get("name"):
|
|
# Get file modification time for cache date display
|
|
cache_mtime = os.path.getmtime(metadata_file)
|
|
cache_date = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(cache_mtime))
|
|
print(f"[ConfigBuilder] ✅ Using cached metadata for: {lora_name} (cached on {cache_date})")
|
|
return web.json_response({
|
|
"metadata": cached,
|
|
"saved_to": metadata_file,
|
|
"cached": True,
|
|
"cache_date": cache_date
|
|
})
|
|
except Exception:
|
|
pass # Cache miss or corrupt file - fall through to fresh lookup
|
|
|
|
# Get the full path to the LoRA file
|
|
lora_path = folder_paths.get_full_path("loras", lora_name)
|
|
|
|
if lora_path is None:
|
|
return web.json_response({
|
|
"error": f"LoRA file not found: {lora_name}"
|
|
}, status=404)
|
|
|
|
# Calculate the hash (expensive - only when not cached)
|
|
lora_hash = calculate_sha256(lora_path)
|
|
short_hash = lora_hash[:10] # First 10 characters for short hash
|
|
|
|
print(f"[ConfigBuilder] 📊 Hash calculated: {short_hash}")
|
|
|
|
# Fetch metadata from CivitAI
|
|
model_info = get_model_version_info(lora_hash)
|
|
|
|
if model_info is None:
|
|
return web.json_response({
|
|
"error": "No metadata found on CivitAI",
|
|
"hash": lora_hash,
|
|
"short_hash": short_hash
|
|
}, status=404)
|
|
|
|
# Extract relevant information
|
|
metadata = {
|
|
"name": model_info.get("name", "Unknown"),
|
|
"model_name": model_info.get("model", {}).get("name", "Unknown") if isinstance(model_info.get("model"), dict) else "Unknown",
|
|
"trained_words": model_info.get("trainedWords", []),
|
|
"base_model": model_info.get("baseModel", "Unknown"),
|
|
"description": model_info.get("description", ""),
|
|
"tags": model_info.get("model", {}).get("tags", []) if isinstance(model_info.get("model"), dict) else [],
|
|
"images": [],
|
|
"url": f"https://civitai.com/models/{model_info.get('modelId', '')}" if model_info.get("modelId") else "",
|
|
"hash": lora_hash,
|
|
"short_hash": short_hash,
|
|
"file_path": lora_path,
|
|
"stats": model_info.get("stats", {}),
|
|
"creator": model_info.get("creator", {}).get("username", "Unknown") if isinstance(model_info.get("creator"), dict) else "Unknown"
|
|
}
|
|
|
|
# Extract images
|
|
if "images" in model_info and isinstance(model_info["images"], list):
|
|
for img in model_info["images"][:5]: # Limit to first 5 images
|
|
if isinstance(img, dict) and "url" in img:
|
|
metadata["images"].append({
|
|
"url": img["url"],
|
|
"nsfw": img.get("nsfw", "None"),
|
|
"width": img.get("width", 0),
|
|
"height": img.get("height", 0)
|
|
})
|
|
|
|
# Save metadata to file
|
|
output_dir = folder_paths.get_output_directory()
|
|
model_data_dir = os.path.join(output_dir, "benchmarks", "model-data", lora_name.replace("/", "_").replace("\\", "_").replace(".safetensors", ""))
|
|
os.makedirs(model_data_dir, exist_ok=True)
|
|
|
|
metadata_file = os.path.join(model_data_dir, "metadata.json")
|
|
save_dict_to_json(metadata, metadata_file)
|
|
|
|
print(f"[ConfigBuilder] ✅ Metadata saved to: {metadata_file}")
|
|
|
|
# Compare fresh tags with stored loras_tags.json (only on force_refresh)
|
|
tags_changed = False
|
|
old_tags = []
|
|
new_tags = metadata.get("trained_words", [])
|
|
if force_refresh and new_tags:
|
|
json_tags_path = os.path.join(output_dir, "benchmarks/loras_tags.json")
|
|
if os.path.exists(json_tags_path):
|
|
lora_tags = load_json_from_file(json_tags_path) or {}
|
|
normalized = lora_name.replace("\\", "/")
|
|
backslash = lora_name.replace("/", "\\")
|
|
old_tags = lora_tags.get(lora_name, lora_tags.get(normalized, lora_tags.get(backslash, [])))
|
|
if old_tags is None:
|
|
old_tags = []
|
|
# Compare sorted lists to detect any difference
|
|
if sorted(old_tags) != sorted(new_tags):
|
|
tags_changed = True
|
|
print(f"[ConfigBuilder] ⚠️ Tags changed for {lora_name}: {old_tags} -> {new_tags}")
|
|
|
|
return web.json_response({
|
|
"metadata": metadata,
|
|
"saved_to": metadata_file,
|
|
"tags_changed": tags_changed,
|
|
"old_tags": old_tags,
|
|
"new_tags": new_tags
|
|
})
|
|
|
|
except Exception as e:
|
|
print(f"[ConfigBuilder] ❌ Error in lookup_lora_metadata endpoint: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return web.json_response({
|
|
"error": str(e)
|
|
}, status=500)
|
|
|
|
|
|
# API endpoint for detailed Model/Checkpoint metadata lookup
|
|
@server.PromptServer.instance.routes.post("/configbuilder/lookup_model_metadata")
|
|
async def lookup_model_metadata_endpoint(request):
|
|
"""API endpoint to lookup full metadata for a model/checkpoint from CivitAI"""
|
|
try:
|
|
data = await request.json()
|
|
model_name = data.get("model_name", "")
|
|
model_type = data.get("model_type", "checkpoint")
|
|
force_refresh = data.get("force_refresh", False)
|
|
|
|
if not model_name:
|
|
return web.json_response({
|
|
"error": "No model name provided"
|
|
}, status=400)
|
|
|
|
print(f"[ConfigBuilder] 🔍 Full metadata lookup request for model: {model_name} (type: {model_type})")
|
|
|
|
# Check disk cache first to avoid expensive SHA256 hashing
|
|
output_dir = folder_paths.get_output_directory()
|
|
model_data_dir = os.path.join(output_dir, "benchmarks", "model-data", model_name.replace("/", "_").replace("\\", "_").replace(".safetensors", "").replace(".ckpt", "").replace(".gguf", ""))
|
|
metadata_file = os.path.join(model_data_dir, "metadata.json")
|
|
|
|
if os.path.exists(metadata_file) and not force_refresh:
|
|
try:
|
|
cached = load_json_from_file(metadata_file)
|
|
if cached and cached.get("name"):
|
|
# Get file modification time for cache date display
|
|
cache_mtime = os.path.getmtime(metadata_file)
|
|
cache_date = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(cache_mtime))
|
|
print(f"[ConfigBuilder] ✅ Using cached metadata for model: {model_name} (cached on {cache_date})")
|
|
return web.json_response({
|
|
"metadata": cached,
|
|
"saved_to": metadata_file,
|
|
"cached": True,
|
|
"cache_date": cache_date
|
|
})
|
|
except Exception:
|
|
pass # Cache miss or corrupt file - fall through to fresh lookup
|
|
|
|
# Resolve the full path based on model type
|
|
model_path = None
|
|
if model_type == "gguf":
|
|
try:
|
|
model_path = folder_paths.get_full_path("unet_gguf", model_name)
|
|
except (KeyError, Exception):
|
|
pass
|
|
if model_path is None:
|
|
model_path = folder_paths.get_full_path("diffusion_models", model_name)
|
|
elif model_type == "diffusion_model":
|
|
model_path = folder_paths.get_full_path("diffusion_models", model_name)
|
|
else:
|
|
model_path = folder_paths.get_full_path("checkpoints", model_name)
|
|
|
|
if model_path is None:
|
|
return web.json_response({
|
|
"error": f"Model file not found: {model_name}"
|
|
}, status=404)
|
|
|
|
# Calculate the hash (expensive - only when not cached)
|
|
model_hash = calculate_sha256(model_path)
|
|
short_hash = model_hash[:10]
|
|
|
|
print(f"[ConfigBuilder] 📊 Hash calculated: {short_hash}")
|
|
|
|
# Fetch metadata from CivitAI
|
|
model_info = get_model_version_info(model_hash)
|
|
|
|
if model_info is None:
|
|
return web.json_response({
|
|
"error": "No metadata found on CivitAI",
|
|
"hash": model_hash,
|
|
"short_hash": short_hash
|
|
}, status=404)
|
|
|
|
# Extract relevant information
|
|
metadata = {
|
|
"name": model_info.get("name", "Unknown"),
|
|
"model_name": model_info.get("model", {}).get("name", "Unknown") if isinstance(model_info.get("model"), dict) else "Unknown",
|
|
"trained_words": model_info.get("trainedWords", []),
|
|
"base_model": model_info.get("baseModel", "Unknown"),
|
|
"description": model_info.get("description", ""),
|
|
"tags": model_info.get("model", {}).get("tags", []) if isinstance(model_info.get("model"), dict) else [],
|
|
"images": [],
|
|
"url": f"https://civitai.com/models/{model_info.get('modelId', '')}" if model_info.get("modelId") else "",
|
|
"hash": model_hash,
|
|
"short_hash": short_hash,
|
|
"file_path": model_path,
|
|
"stats": model_info.get("stats", {}),
|
|
"creator": model_info.get("creator", {}).get("username", "Unknown") if isinstance(model_info.get("creator"), dict) else "Unknown"
|
|
}
|
|
|
|
# Extract images
|
|
if "images" in model_info and isinstance(model_info["images"], list):
|
|
for img in model_info["images"][:5]:
|
|
if isinstance(img, dict) and "url" in img:
|
|
metadata["images"].append({
|
|
"url": img["url"],
|
|
"nsfw": img.get("nsfw", "None"),
|
|
"width": img.get("width", 0),
|
|
"height": img.get("height", 0)
|
|
})
|
|
|
|
# Save metadata to file
|
|
output_dir = folder_paths.get_output_directory()
|
|
model_data_dir = os.path.join(output_dir, "benchmarks", "model-data", model_name.replace("/", "_").replace("\\", "_").replace(".safetensors", "").replace(".ckpt", "").replace(".gguf", ""))
|
|
os.makedirs(model_data_dir, exist_ok=True)
|
|
|
|
metadata_file = os.path.join(model_data_dir, "metadata.json")
|
|
save_dict_to_json(metadata, metadata_file)
|
|
|
|
print(f"[ConfigBuilder] ✅ Model metadata saved to: {metadata_file}")
|
|
|
|
return web.json_response({
|
|
"metadata": metadata,
|
|
"saved_to": metadata_file
|
|
})
|
|
|
|
except Exception as e:
|
|
print(f"[ConfigBuilder] ❌ Error in lookup_model_metadata endpoint: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return web.json_response({
|
|
"error": str(e)
|
|
}, status=500)
|
|
|
|
|
|
# Fast count endpoint — returns just file counts per category for change detection
|
|
@server.PromptServer.instance.routes.get("/configbuilder/model_counts")
|
|
async def get_model_counts_endpoint(request):
|
|
"""Return file counts per model category. Very fast — uses folder_paths internal cache."""
|
|
try:
|
|
counts = {}
|
|
for cat in ["checkpoints", "diffusion_models", "text_encoders", "vae", "loras"]:
|
|
try:
|
|
counts[cat] = len(folder_paths.get_filename_list(cat))
|
|
except (KeyError, Exception):
|
|
counts[cat] = 0
|
|
for cat in ["unet_gguf", "clip_gguf", "upscale_models"]:
|
|
try:
|
|
counts[cat] = len(folder_paths.get_filename_list(cat))
|
|
except (KeyError, Exception):
|
|
counts[cat] = 0
|
|
return web.json_response(counts)
|
|
except Exception as e:
|
|
return web.json_response({"error": str(e)}, status=500)
|
|
|
|
# API endpoint to get all model lists for unified model selector
|
|
@server.PromptServer.instance.routes.get("/configbuilder/model_lists")
|
|
async def get_model_lists_endpoint(request):
|
|
"""Return all model lists for the config builder's unified model selector."""
|
|
try:
|
|
checkpoints = folder_paths.get_filename_list("checkpoints")
|
|
diffusion_models = folder_paths.get_filename_list("diffusion_models")
|
|
text_encoders = folder_paths.get_filename_list("text_encoders")
|
|
|
|
# GGUF lists - may not exist if ComfyUI-GGUF is not installed
|
|
try:
|
|
unet_gguf = folder_paths.get_filename_list("unet_gguf")
|
|
except (KeyError, Exception):
|
|
unet_gguf = []
|
|
try:
|
|
clip_gguf = folder_paths.get_filename_list("clip_gguf")
|
|
except (KeyError, Exception):
|
|
clip_gguf = []
|
|
|
|
# CLIPType options (matching ComfyUI's CLIPLoader and DualCLIPLoader)
|
|
clip_types = [
|
|
"stable_diffusion", "stable_cascade", "sd3", "stable_audio",
|
|
"mochi", "ltxv", "pixart", "cosmos", "lumina2", "wan",
|
|
"hidream", "chroma", "ace", "flux", "flux2"
|
|
]
|
|
dual_clip_types = [
|
|
"sdxl", "sd3", "flux", "flux2", "hunyuan_video", "hidream",
|
|
"hunyuan_image", "hunyuan_video_15"
|
|
]
|
|
|
|
# Upscale model list
|
|
try:
|
|
upscale_models = folder_paths.get_filename_list("upscale_models")
|
|
except (KeyError, Exception):
|
|
upscale_models = []
|
|
|
|
# Latent upscale model list (separate folder; used by LatentUpscaleModelLoader,
|
|
# e.g. LTX 2.3 spatial upscaler)
|
|
try:
|
|
latent_upscale_models = folder_paths.get_filename_list("latent_upscale_models")
|
|
except (KeyError, Exception):
|
|
latent_upscale_models = []
|
|
|
|
# VAE list
|
|
vae_list = folder_paths.get_filename_list("vae")
|
|
|
|
# Sampler and scheduler lists from ComfyUI core
|
|
import comfy.samplers
|
|
sampler_names = list(comfy.samplers.KSampler.SAMPLERS)
|
|
scheduler_names = list(comfy.samplers.KSampler.SCHEDULERS)
|
|
|
|
return web.json_response({
|
|
"checkpoints": checkpoints,
|
|
"diffusion_models": diffusion_models,
|
|
"unet_gguf": unet_gguf,
|
|
"text_encoders": text_encoders,
|
|
"clip_gguf": clip_gguf,
|
|
"clip_types": clip_types,
|
|
"dual_clip_types": dual_clip_types,
|
|
"vae": vae_list,
|
|
"upscale_models": upscale_models,
|
|
"latent_upscale_models": latent_upscale_models,
|
|
"samplers": sampler_names,
|
|
"schedulers": scheduler_names
|
|
})
|
|
except Exception as e:
|
|
print(f"[ConfigBuilder] Error in model_lists endpoint: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return web.json_response({"error": str(e)}, status=500)
|
|
|
|
|
|
# API endpoint to refresh model/lora lists (triggered by "Update Node Definitions")
|
|
@server.PromptServer.instance.routes.post("/configbuilder/refresh_models")
|
|
async def refresh_models_endpoint(request):
|
|
"""
|
|
API endpoint to signal frontend to clear its caches.
|
|
Called when ComfyUI updates node definitions.
|
|
"""
|
|
try:
|
|
print(f"[ConfigBuilder] 🔄 Refresh signal received - clearing frontend caches")
|
|
|
|
return web.json_response({
|
|
"status": "ok",
|
|
"message": "Frontend caches should be cleared"
|
|
})
|
|
except Exception as e:
|
|
print(f"[ConfigBuilder] ❌ Error in refresh_models endpoint: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return web.json_response({
|
|
"error": str(e)
|
|
}, status=500)
|
|
|
|
|
|
@server.PromptServer.instance.routes.get("/configbuilder/get_lora_triggers")
|
|
async def get_lora_triggers_endpoint(request):
|
|
"""Get trigger words for a specific LoRA from loras_tags.json"""
|
|
try:
|
|
lora_name = request.query.get("lora_name", "")
|
|
if not lora_name:
|
|
return web.json_response({"error": "Missing lora_name"}, status=400)
|
|
|
|
json_tags_path = os.path.join(folder_paths.get_output_directory(), "benchmarks/loras_tags.json")
|
|
triggers = []
|
|
if os.path.exists(json_tags_path):
|
|
from .lora_utils import load_json_from_file
|
|
lora_tags = load_json_from_file(json_tags_path) or {}
|
|
# Try exact match, normalized, and backslash variants
|
|
normalized = lora_name.replace("\\", "/")
|
|
backslash = lora_name.replace("/", "\\")
|
|
triggers = lora_tags.get(lora_name, lora_tags.get(normalized, lora_tags.get(backslash, [])))
|
|
|
|
return web.json_response({"lora_name": lora_name, "triggers": triggers})
|
|
except Exception as e:
|
|
return web.json_response({"error": str(e)}, status=500)
|
|
|
|
|
|
@server.PromptServer.instance.routes.post("/configbuilder/save_lora_triggers")
|
|
async def save_lora_triggers_endpoint(request):
|
|
"""Save edited trigger words for a LoRA to loras_tags.json"""
|
|
try:
|
|
data = await request.json()
|
|
lora_name = data.get("lora_name", "")
|
|
triggers = data.get("triggers", [])
|
|
|
|
if not lora_name:
|
|
return web.json_response({"error": "Missing lora_name"}, status=400)
|
|
|
|
json_tags_path = os.path.join(folder_paths.get_output_directory(), "benchmarks/loras_tags.json")
|
|
from .lora_utils import load_json_from_file, save_dict_to_json
|
|
lora_tags = {}
|
|
if os.path.exists(json_tags_path):
|
|
lora_tags = load_json_from_file(json_tags_path) or {}
|
|
|
|
# Normalize the name for consistent storage
|
|
normalized = lora_name.replace("\\", "/")
|
|
lora_tags[normalized] = triggers
|
|
|
|
save_dict_to_json(lora_tags, json_tags_path)
|
|
|
|
# Clear the trigger word LRU cache so changes take effect immediately
|
|
from .trigger_words import clear_trigger_caches
|
|
clear_trigger_caches()
|
|
|
|
print(f"[ConfigBuilder] ✏️ Saved {len(triggers)} trigger words for: {normalized}")
|
|
return web.json_response({"status": "saved", "lora_name": normalized, "triggers": triggers})
|
|
except Exception as e:
|
|
return web.json_response({"error": str(e)}, status=500)
|
|
|
|
|
|
@server.PromptServer.instance.routes.post("/configbuilder/preview")
|
|
async def preview_endpoint(request):
|
|
"""
|
|
Preview endpoint — runs state_to_configs_json on the POSTed builder UI
|
|
state and returns the resulting configs_json string. The Builder UI's
|
|
JSON Preview panel calls this. Single source of truth: this endpoint and
|
|
generate_config() share state_to_configs_json, so they cannot disagree.
|
|
"""
|
|
try:
|
|
body = await request.json()
|
|
state = body.get("state")
|
|
if not isinstance(state, dict):
|
|
return web.json_response(
|
|
{"error": "Missing or invalid 'state' object in request body"},
|
|
status=400,
|
|
)
|
|
configs_json = UltimateConfigBuilder.state_to_configs_json(state)
|
|
return web.json_response({"configs_json": configs_json})
|
|
except Exception as e:
|
|
return web.json_response({"error": str(e)}, status=500) |