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