# ANCHOR: Imports & Setup from __future__ import annotations import os import csv import logging import re import math import torch import torch.nn as nn from pathlib import Path from types import MethodType from collections import OrderedDict import gradio as gr import modules.scripts as scripts from modules import shared, devices, script_callbacks logger = logging.getLogger("AnimaArtistForge") logger.setLevel(logging.INFO) # Internal defaults COMBINE_OUTPUT_AVG = "output_avg" FUSION_INTERPOLATE = "interpolate" _artist_database = set() _db_initialized = False _cached_db_regex = None # Global Text Conditioning Cache _COND_CACHE = OrderedDict() _COND_CACHE_LIMIT = 64 # ANCHOR: Mathematical Helpers def _lerp(a: float, b: float, t: float) -> float: return float(a + (b - a) * t) def _adain(target: torch.Tensor, style: torch.Tensor, eps: float = 1e-6) -> torch.Tensor: t_mean = target.mean(dim=1, keepdim=True) s_mean = style.mean(dim=1, keepdim=True) t_std = target.float().std(dim=1, keepdim=True).add(eps).to(target.dtype) s_std = style.float().std(dim=1, keepdim=True).add(eps).to(style.dtype) return (target - t_mean) / t_std * s_std + s_mean def _spectral_hybrid(target: torch.Tensor, artist: torch.Tensor, alpha: float, beta: float = 2.0) -> torch.Tensor: try: seq_dim = -2 L = target.shape[seq_dim] orig_dtype = target.dtype t_32 = target.to(dtype=torch.float32) a_32 = artist.to(dtype=torch.float32) f_target = torch.fft.rfft(t_32, n=L, dim=seq_dim) f_artist = torch.fft.rfft(a_32, n=L, dim=seq_dim) num_bins = f_target.shape[seq_dim] bins = torch.linspace(0.0, 1.0, num_bins, device=target.device, dtype=torch.float32) W = torch.pow(bins, beta).view(*([1] * (target.dim() - 2)), num_bins, 1) W_eff = (1.0 - alpha) + alpha * W f_hybrid = (1.0 - W_eff) * f_artist + W_eff * f_target res = torch.fft.irfft(f_hybrid, n=L, dim=seq_dim) return res.to(dtype=orig_dtype) except Exception as e: logger.error(f"[AnimaSpectral] FFT fallback to LERP: {e}") return (1.0 - alpha) * target + alpha * artist def _cross_token_adain(target: torch.Tensor, artist: torch.Tensor, alpha: float, eps: float = 1e-6) -> torch.Tensor: try: seq_dim = -2 mean_t = target.mean(dim=seq_dim, keepdim=True) std_t = target.std(dim=seq_dim, keepdim=True).clamp_min(eps) mean_a = artist.mean(dim=seq_dim, keepdim=True) std_a = artist.std(dim=seq_dim, keepdim=True).clamp_min(eps) normalized = (target - mean_t) / std_t adain_target = normalized * std_a + mean_a return (1.0 - alpha) * target + alpha * adain_target except Exception as e: logger.error(f"[AnimaAdaIN] AdaIN fallback to LERP: {e}") return (1.0 - alpha) * target + alpha * artist def _textural_contrast_recovery(base_context: torch.Tensor, artist_a: torch.Tensor, artist_b: torch.Tensor, alpha: float, delta_scale: float = 1.0, beta: float = 2.0) -> torch.Tensor: try: seq_dim = -2 L = base_context.shape[seq_dim] orig_dtype = base_context.dtype a_aligned = _align_sequence_length(artist_a, base_context) b_aligned = _align_sequence_length(artist_b, base_context) delta = (b_aligned - a_aligned).to(dtype=torch.float32) f_delta = torch.fft.rfft(delta, n=L, dim=seq_dim) num_bins = f_delta.shape[seq_dim] bins = torch.linspace(0.0, 1.0, num_bins, device=base_context.device, dtype=torch.float32) W = torch.pow(bins, beta).view(*([1] * (delta.dim() - 2)), num_bins, 1) f_delta_high = f_delta * W * float(delta_scale) delta_high = torch.fft.irfft(f_delta_high, n=L, dim=seq_dim).to(orig_dtype) return (1.0 - alpha) * a_aligned + alpha * b_aligned + delta_high except Exception as e: logger.error(f"[TexturalContrastRecovery] Math failure, falling back to LERP: {e}") return (1.0 - alpha) * artist_a + alpha * artist_b def _axes_dims_from_head_dim(head_dim: int) -> list[int]: hd = max(0, int(head_dim)) dim_h = (hd // 6) * 2 dim_w = dim_h dim_t = hd - 2 * dim_h axes = [dim_t, dim_h, dim_w] if sum(axes) != hd or any(v <= 0 for v in axes): return [hd] return axes def _build_rope_scale_vector(head_dim: int, axes_dims: list[int], high_scale: float, low_scale: float, beta: float, device: torch.device, dtype: torch.dtype) -> torch.Tensor: if not axes_dims or sum(axes_dims) != head_dim: axes_dims = [head_dim] is_3axis = len(axes_dims) == 3 pieces: list[torch.Tensor] = [] for axis_idx, axis_dim in enumerate(axes_dims): n_pairs = axis_dim // 2 if n_pairs <= 0: pieces.append(torch.ones(axis_dim, device=device, dtype=dtype)) continue if is_3axis and axis_idx == 0: scales = torch.full((n_pairs,), float(low_scale), device=device, dtype=torch.float32) else: d = torch.zeros(1, device=device, dtype=torch.float32) if n_pairs == 1 else torch.linspace(0.0, 1.0, n_pairs, device=device, dtype=torch.float32) scales = high_scale + (low_scale - high_scale) * d.pow(float(beta)) pieces.append(scales.to(dtype).repeat_interleave(2)) if axis_dim % 2: pieces.append(torch.ones(1, device=device, dtype=dtype)) out = torch.cat(pieces, dim=0) if out.numel() < head_dim: out = torch.nn.functional.pad(out, (0, head_dim - out.numel()), value=1.0) return out[:head_dim] def _apply_rope_scaling(rope: any, scale: torch.Tensor) -> any: if rope is None: return None if isinstance(rope, torch.Tensor): try: scale_len = scale.numel() rope_shape = list(rope.shape) target_dim = -1 for idx, dim_sz in enumerate(rope_shape): if dim_sz == scale_len: target_dim = idx break if target_dim == -1: for idx, dim_sz in enumerate(rope_shape): if dim_sz == (scale_len // 2): target_dim = idx break if target_dim == -1: return rope sh = [1] * len(rope_shape) sh[target_dim] = -1 if rope_shape[target_dim] == (scale_len // 2): scale_proj = scale[:(scale_len // 2)].view(*sh).to(device=rope.device, dtype=rope.dtype) else: scale_proj = scale.view(*sh).to(device=rope.device, dtype=rope.dtype) return rope * scale_proj except Exception as e: logger.error(f"[AnimaArtistMixer] RoPE scaling bypassed: {e}") return rope elif isinstance(rope, (tuple, list)): return type(rope)(_apply_rope_scaling(r, scale) for r in rope) return rope def _broadcast_batch(t: torch.Tensor, batch_size: int, target_shape: tuple[int, ...] | None = None) -> torch.Tensor: if t.shape[0] != batch_size: if t.shape[0] == 1: t = t.expand(batch_size, *([-1] * (t.dim() - 1))) elif batch_size % t.shape[0] == 0: t = t.repeat(batch_size // t.shape[0], *([1] * (t.dim() - 1))) else: t = t[:1].expand(batch_size, *([-1] * (t.dim() - 1))) if target_shape is not None and t.dim() < len(target_shape): diff = len(target_shape) - t.dim() for _ in range(diff): t = t.unsqueeze(1) expand_sizes = list(t.shape) for i in range(1, 1 + diff): expand_sizes[i] = target_shape[i] t = t.expand(*expand_sizes) return t def _align_sequence_length(src: torch.Tensor, tgt: torch.Tensor) -> torch.Tensor: if src.shape[-2] != tgt.shape[-2]: if src.shape[-2] < tgt.shape[-2]: pad_shape = list(tgt.shape) pad_shape[-2] = tgt.shape[-2] - src.shape[-2] padding = torch.zeros(*pad_shape, device=src.device, dtype=src.dtype) return torch.cat([src, padding], dim=-2) else: slices = [slice(None)] * src.dim() slices[-2] = slice(0, tgt.shape[-2]) return src[tuple(slices)] return src def _resolve_mask(cou: list[int] | None, batch_size: int) -> list[bool]: if cou is None or len(cou) != batch_size: return [True] * batch_size return [c == 0 for c in cou] def calculate_curve_factor(progress: float, start: float, end: float, peak: float, curve: str) -> float: start = max(0.0, min(1.0, start)) end = max(0.0, min(1.0, end)) if end < start: start, end = end, start if progress < start or progress > end: return 0.0 if end - start <= 1e-6: return 1.0 local = (progress - start) / max(end - start, 1e-6) if curve == "Hold": return 1.0 if curve == "Front loaded": return 1.0 - 0.75 * local if curve == "Back loaded": return 0.25 + 0.75 * local peak = max(start, min(end, peak)) peak_local = (peak - start) / max(end - start, 1e-6) if curve == "Triangle": if local <= peak_local: return local / max(peak_local, 1e-6) return (1.0 - local) / max(1.0 - peak_local, 1e-6) if local <= peak_local: raw = local / max(peak_local, 1e-6) else: raw = (1.0 - local) / max(1.0 - peak_local, 1e-6) raw = max(0.0, min(1.0, raw)) return raw * raw * (3.0 - 2.0 * raw) def apply_norm_limiting(base_idx: torch.Tensor, target_idx: torch.Tensor, active_strength: float, limit_ratio: float) -> torch.Tensor: delta = target_idx - base_idx if delta.numel() == 0: return base_idx with torch.no_grad(): flat_delta = delta.detach().float().flatten() flat_base = base_idx.detach().float().flatten() delta_norm = flat_delta.norm().clamp_min(1e-6) base_norm = flat_base.norm().clamp_min(1e-6) limit = base_norm * float(limit_ratio) scale = torch.minimum(torch.ones_like(delta_norm), limit / delta_norm) return base_idx + delta * scale.to(device=delta.device, dtype=delta.dtype) * float(active_strength) def _batched_artists_forward(cross_attn_module, x, context, rope_emb, transformer_options, individuals, weights, bsz, *args, **kwargs): n = len(individuals) if n < 2: return None kv_list = [] for artist_i in individuals: artist_b = _broadcast_batch(artist_i, bsz, target_shape=context.shape).to(device=context.device, dtype=context.dtype) kv_list.append(artist_b) kv_lens = {kv.shape[-2] for kv in kv_list} if len(kv_lens) > 1: return None x_rep = x.repeat(n, *([1] * (x.dim() - 1))) kv_stacked = torch.cat(kv_list, dim=0) rope_rep = rope_emb if rope_rep is not None and torch.is_tensor(rope_rep): if rope_rep.dim() > 0 and rope_rep.shape[0] == bsz: rope_rep = rope_rep.repeat(n, *([1] * (rope_rep.dim() - 1))) new_opts = dict(transformer_options) if isinstance(transformer_options, dict) else {} cou = new_opts.get("cond_or_uncond") if cou is not None: new_opts["cond_or_uncond"] = list(cou) * n out = cross_attn_module.original_forward(x_rep, kv_stacked, rope_emb=rope_rep, transformer_options=new_opts, *args, **kwargs) out = out.view(n, bsz, *out.shape[1:]) w_t = torch.tensor(weights, device=out.device, dtype=out.dtype).view(n, *([1] * (out.dim() - 1))) return (out * w_t).sum(dim=0) # INFO: Presets Configuration Dictionary (Pristine & Decoupled with clean names) PRESETS = { "Default": { "replication_method": "None (Bypass to Classic Parallel Attention)", "comp_strength": 1.0, "style_strength": 1.0, "color_strength": 1.0, "style_isolation_strength": 0.0, "curve_type": "Hold", "isolation_start_step": 0.0, "isolation_end_step": 1.0, "auto_distribute": False, "enable_manual": False }, "Custom": {}, "First Shapes // Second Linework": { "replication_method": "Spectral (LERoPE)", "comp_strength": 1.5, "style_strength": 1.0, "color_strength": 0.8, "style_isolation_strength": 0.3, "curve_type": "Hold", "isolation_start_step": 0.30, "isolation_end_step": 0.80, "auto_distribute": True, "enable_manual": False }, "First Layout // Second Colors": { "replication_method": "AdaIN", "comp_strength": 0.7, "style_strength": 1.3, "color_strength": 1.6, "style_isolation_strength": 0.5, "curve_type": "Hold", "isolation_start_step": 0.20, "isolation_end_step": 0.90, "auto_distribute": True, "enable_manual": False }, "First Structure // Second Texture": { "replication_method": "Textural Recovery (Anti-Median)", "comp_strength": 1.0, "style_strength": 1.5, "color_strength": 1.5, "style_isolation_strength": 0.4, "curve_type": "Hold", "isolation_start_step": 0.10, "isolation_end_step": 0.90, "auto_distribute": False, "enable_manual": False }, "Causal Delta Splicing": { "replication_method": "Decoupled Causal Delta Splicing", "comp_strength": 1.1, "style_strength": 1.2, "color_strength": 1.4, "style_isolation_strength": 0.0, "curve_type": "Hold", "isolation_start_step": 0.0, "isolation_end_step": 1.0, "auto_distribute": False, "enable_manual": False } } # ANCHOR: File Access & CSV Tag DB Parsing def get_tac_tags_path() -> Path | None: try: from modules.paths import extensions_dir ext_path = Path(extensions_dir).absolute() except ImportError: ext_path = Path().absolute().joinpath("extensions").absolute() possible_tac_dirs = list(ext_path.glob("*tag-autocomplete*")) + list(ext_path.glob("*tagcomplete*")) for folder in possible_tac_dirs: tags_folder = folder.joinpath("tags") if tags_folder.exists(): return tags_folder sibling_tac = Path(scripts.basedir()).parent.joinpath("stable-diffusion-webui-tag-autocomplete", "tags") if sibling_tac.exists(): return sibling_tac.absolute() return None def load_artists(): global _artist_database, _db_initialized, _cached_db_regex if _db_initialized: return _artist_database _artist_database.clear() extension_dir = Path(scripts.basedir()) local_txt = extension_dir.joinpath("artists.txt") if local_txt.exists(): try: with open(local_txt, "r", encoding="utf-8") as f: for line in f: name = line.strip().lower() if name and not name.startswith("#"): _artist_database.add(name.replace("_", " ")) logger.info(f"[AnimaArtistMixer] Loaded overrides from {local_txt}") except Exception as e: logger.error(f"Error reading local override file: {e}") tac_tags_path = get_tac_tags_path() if tac_tags_path: csv_files = [tac_tags_path.joinpath("danbooru.csv"), tac_tags_path.joinpath("novelai.csv")] for csv_path in csv_files: if csv_path.exists(): try: loaded_count = 0 with open(csv_path, "r", encoding="utf-8") as f: reader = csv.reader(f) for row in reader: if len(row) >= 2: name = row[0].strip().replace("_", " ").lower() category = row[1].strip() if category == "1": _artist_database.add(name) loaded_count += 1 logger.info(f"[AnimaArtistMixer] Loaded {loaded_count} artists from {csv_path}") break except Exception as e: logger.error(f"Error parsing CSV {csv_path}: {e}") else: logger.warning("[AnimaArtistMixer] Tag Autocomplete database not found. Relying on syntax prefix (@) and local list.") if _artist_database: sorted_db = sorted(list(_artist_database), key=len, reverse=True) db_escaped = '|'.join(re.escape(name) for name in sorted_db) _cached_db_regex = ( re.compile(r'\(\s*\b(' + db_escaped + r')\b\s*:\s*([\d.]+)\s*\)', re.IGNORECASE), re.compile(r'\b(' + db_escaped + r')\b', re.IGNORECASE) ) _db_initialized = True return _artist_database # ANCHOR: Prompt Cleansing & Parsing def clean_empty_groups(text: str) -> str: prev = None while prev != text: prev = text text = re.sub(r'\([\s,|,:.\d-]*\)', '', text) text = re.sub(r'\[[\s,|,:.\d-]*\]', '', text) text = re.sub(r'\{[\s,|,:.\d-]*\}', '', text) text = re.sub(r'[:|]\s*(?=[,()\[\]]|$)', '', text) text = re.sub(r',\s*,', ',', text) text = re.sub(r'\s+', ' ', text) return text.strip() def isolate_extra_networks(text: str) -> tuple[str, list[str]]: extra_networks = [] def repl(m): extra_networks.append(m.group(0)) return f" __EXT_NET_{len(extra_networks)-1}__ " safe_text = re.sub(r'<[^>]+>', repl, text) return safe_text, extra_networks def restore_extra_networks(text: str, extra_networks: list[str]) -> str: for idx, tag in enumerate(extra_networks): text = text.replace(f"__EXT_NET_{idx}__", tag) return text def extract_artist_tags_only(text: str) -> list[str]: temp_text, extra_networks = isolate_extra_networks(text) artists = [] artist_db = load_artists() pat_paren_at_weight = re.compile( r'\(\s*(@(?:\\.|[^\s,()\[\]:|])+(?:\s+(?:\\.|[^\s,()\[\]:|])+)*)\s*:\s*([\d.]+)\s*\)' ) for m in pat_paren_at_weight.finditer(temp_text): name = m.group(1).strip().replace("\\(", "(").replace("\\)", ")").replace("_", " ") artists.append(name.lower()) pat_at_weight = re.compile( r'(@(?:\\.|[^\s,()\[\]:|])+(?:\s+(?:\\.|[^\s,()\[\]:|])+)*)\s*:\s*([\d.]+)\b' ) for m in pat_at_weight.finditer(temp_text): name = m.group(1).strip().replace("\\(", "(").replace("\\)", ")").replace("_", " ") if name.lower() not in artists: artists.append(name.lower()) pat_at_raw = re.compile( r'(@(?:\\.|[^\s,()\[\]:|])+(?:\s+(?:\\.|[^\s,()\[\]:|])+)*)' ) for m in pat_at_raw.finditer(temp_text): name = m.group(1).strip().replace("\\(", "(").replace("\\)", ")").replace("_", " ") if name.lower() not in artists: artists.append(name.lower()) if artist_db and _cached_db_regex: pat_db_paren_weight, pat_db_raw = _cached_db_regex for m in pat_db_paren_weight.finditer(temp_text): name = m.group(1).replace("_", " ").lower() if name not in artists: artists.append(name) for m in pat_db_raw.finditer(temp_text): name = m.group(1).replace("_", " ").lower() if name not in artists: artists.append(name) return artists def parse_scheduled_artists(prompt_str: str, total_steps: int = 20) -> dict[str, list[tuple[float, float]]]: artist_ranges = {} bracket_pattern = re.compile(r'\[([^\[\]]+)\]') for m in bracket_pattern.finditer(prompt_str): content = m.group(1) parts = re.split(r'(? 3 parts (A until step, B after step) if len(parts) == 3: val_str = parts[2] try: val = float(val_str) if val < 0: step_pct = (total_steps + val) / total_steps else: step_pct = val if val < 1.0 else (val / total_steps) step_pct = max(0.0, min(1.0, step_pct)) except ValueError: step_pct = 0.5 part_a, part_b = parts[0], parts[1] artists_a = extract_artist_tags_only(part_a) for a in artists_a: artist_ranges.setdefault(a, []).append((0.0, step_pct)) artists_b = extract_artist_tags_only(part_b) for b in artists_b: artist_ranges.setdefault(b, []).append((step_pct, 1.0)) # Case 2: [ A : step ] or [ : B : step ] (which registers as ['', B, step]) elif len(parts) == 2: part_a, val_str = parts[0], parts[1] try: val = float(val_str) if val < 0: step_pct = (total_steps + val) / total_steps else: step_pct = val if val < 1.0 else (val / total_steps) step_pct = max(0.0, min(1.0, step_pct)) artists_a = extract_artist_tags_only(part_a) for a in artists_a: artist_ranges.setdefault(a, []).append((0.0, step_pct)) except ValueError: pass return artist_ranges def extract_artists_and_base(prompt: str) -> tuple[list[tuple[str, float]], str]: temp_prompt, extra_networks = isolate_extra_networks(prompt) artists_with_weights: list[tuple[str, float]] = [] artist_db = load_artists() pat_paren_at_weight = re.compile( r'\(\s*(@(?:\\.|[^\s,()\[\]:|])+(?:\s+(?:\\.|[^\s,()\[\]:|])+)*)\s*:\s*([\d.]+)\s*\)' ) def repl_paren_at_weight(m): name = m.group(1).strip().replace("\\(", "(").replace("\\)", ")").replace("_", " ") weight = float(m.group(2)) artists_with_weights.append((name, weight)) return "" temp_prompt = pat_paren_at_weight.sub(repl_paren_at_weight, temp_prompt) pat_at_weight = re.compile( r'(@(?:\\.|[^\s,()\[\]:|])+(?:\s+(?:\\.|[^\s,()\[\]:|])+)*)\s*:\s*([\d.]+)\b' ) def repl_at_weight(m): name = m.group(1).strip().replace("\\(", "(").replace("\\)", ")").replace("_", " ") weight = float(m.group(2)) artists_with_weights.append((name, weight)) return "" temp_prompt = pat_at_weight.sub(repl_at_weight, temp_prompt) pat_at_raw = re.compile( r'(@(?:\\.|[^\s,()\[\]:|])+(?:\s+(?:\\.|[^\s,()\[\]:|])+)*)' ) def repl_at_raw(m): name = m.group(1).strip().replace("\\(", "(").replace("\\)", ")").replace("_", " ") artists_with_weights.append((name, 1.0)) return "" temp_prompt = pat_at_raw.sub(repl_at_raw, temp_prompt) if artist_db and _cached_db_regex: pat_db_paren_weight, pat_db_raw = _cached_db_regex def repl_db_paren_weight(m): name = m.group(1).replace("_", " ").lower() weight = float(m.group(2)) artists_with_weights.append((name, weight)) return "" temp_prompt = pat_db_paren_weight.sub(repl_db_paren_weight, temp_prompt) def repl_db_raw(m): name = m.group(1).replace("_", " ").lower() artists_with_weights.append((name, 1.0)) return "" temp_prompt = pat_db_raw.sub(repl_db_raw, temp_prompt) clean_prompt = clean_empty_groups(temp_prompt) clean_prompt = re.sub(r'^\s*,\s*', '', clean_prompt) clean_prompt = re.sub(r'\s*,\s*$', '', clean_prompt) clean_prompt = re.sub(r'\s*,\s*$', '', clean_prompt) clean_prompt = re.sub(r'\s*,\s*,+', ', ', clean_prompt) clean_prompt = re.sub(r'\s+', ' ', clean_prompt).strip() clean_prompt = restore_extra_networks(clean_prompt, extra_networks) return artists_with_weights, clean_prompt # ANCHOR: Conditioning Caching & Cleanups def get_conditioning_cached(sd_model, packaged_prompt: str, use_cache: bool = True) -> torch.Tensor: global _COND_CACHE if not use_cache: return _encode_conditioning_raw(sd_model, packaged_prompt) model_id = id(getattr(sd_model, "cond_stage_model", None)) or id(sd_model) cache_key = (model_id, packaged_prompt) if cache_key in _COND_CACHE: _COND_CACHE.move_to_end(cache_key) return _COND_CACHE[cache_key] cond = _encode_conditioning_raw(sd_model, packaged_prompt) _COND_CACHE[cache_key] = cond while len(_COND_CACHE) > _COND_CACHE_LIMIT: _COND_CACHE.popitem(last=False) return cond def _encode_conditioning_raw(sd_model, packaged_prompt: str) -> torch.Tensor: if hasattr(sd_model, "get_learned_conditioning"): cond = sd_model.get_learned_conditioning([packaged_prompt])[0] elif hasattr(sd_model, "cond_stage_model"): cond = sd_model.cond_stage_model([packaged_prompt]) else: raise RuntimeError("No compatible Text Encoder pipeline found.") if isinstance(cond, tuple): cond = cond[0] if cond.ndim == 3: cond = cond[0] return cond # ANCHOR: Attention Forward Patched Mechanism def anima_artist_cross_attn_forward(cross_attn_module, x, context=None, rope_emb=None, transformer_options=None, *args, **kwargs): if context is None or getattr(cross_attn_module, '_disabled', False): return cross_attn_module.original_forward(x, context, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) strength = getattr(cross_attn_module, 'strength', 1.0) normalize_weights = getattr(cross_attn_module, 'normalize_weights', True) sdxl_replication_mode = getattr(cross_attn_module, 'sdxl_replication_mode', False) replication_method = getattr(cross_attn_module, 'replication_method', 'spectral') block_index = getattr(cross_attn_module, 'block_index', 0) total_blocks = getattr(cross_attn_module, 'total_blocks', 28) comp_strength = getattr(cross_attn_module, 'comp_strength', 1.0) style_strength = getattr(cross_attn_module, 'style_strength', 1.0) color_strength = getattr(cross_attn_module, 'color_strength', 1.0) start_step_pct = getattr(cross_attn_module, 'start_step_pct', 0.0) end_step_pct = getattr(cross_attn_module, 'end_step_pct', 1.0) curve_type = getattr(cross_attn_module, 'curve_type', 'Hold') curve_peak = getattr(cross_attn_module, 'curve_peak', 0.5) turbo_mode = getattr(cross_attn_module, 'turbo_mode', False) use_norm_limiting = getattr(cross_attn_module, 'use_norm_limiting', False) norm_limit_ratio = getattr(cross_attn_module, 'norm_limit_ratio', 0.6) style_isolation_strength = getattr(cross_attn_module, 'style_isolation_strength', 0.0) isolation_start_step = getattr(cross_attn_module, 'isolation_start_step', 0.30) isolation_end_step = getattr(cross_attn_module, 'isolation_end_step', 0.80) artists_list = getattr(cross_attn_module, 'artists_list', []) artist_schedules = getattr(cross_attn_module, 'artist_schedules', {}) xyz_isolation_mode = getattr(cross_attn_module, 'xyz_isolation_mode', "Merged Blend") total_steps = 1 current_step = 0 if isinstance(transformer_options, dict): total_steps = max(1, transformer_options.get("total_steps", shared.state.sampling_steps)) current_step = transformer_options.get("step", shared.state.sampling_step) else: total_steps = max(1, shared.state.sampling_steps) current_step = shared.state.sampling_step progress = current_step / total_steps # ANCHOR: Low-Step Turbo Controller if turbo_mode: if progress < 0.85: curve_mult = 1.35 else: decay_progress = (progress - 0.85) / 0.15 curve_mult = 1.35 * (math.cos(decay_progress * math.pi * 0.5) ** 2) else: curve_mult = calculate_curve_factor(progress, start_step_pct, end_step_pct, curve_peak, curve_type) if curve_mult <= 1e-6: return cross_attn_module.original_forward(x, context, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) current_bsz = x.shape[0] cou = (transformer_options or {}).get("cond_or_uncond") if isinstance(transformer_options, dict) else None mask = _resolve_mask(cou, current_bsz) # ANCHOR: Pure Causal Delta Splicing Path if replication_method == "decoupled_delta": delta_tensors = getattr(cross_attn_module, 'delta_tensors', []) if not delta_tensors or not any(mask): return cross_attn_module.original_forward(x, context, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) modified_context = context.clone() t = block_index / max(1, total_blocks - 1) w_layout = float(comp_strength * ((1.0 - t) ** 1.5) * curve_mult) w_texture = float(color_strength * (t ** 1.5) * curve_mult) for batch_idx in range(current_bsz): if mask[batch_idx]: slice_ctx = modified_context[batch_idx] if xyz_isolation_mode == "First Artist Only" and len(delta_tensors) >= 1: delta_1 = _align_sequence_length(delta_tensors[0], slice_ctx) slice_ctx = slice_ctx + w_layout * delta_1 elif xyz_isolation_mode == "Second Artist Only" and len(delta_tensors) >= 2: delta_2 = _align_sequence_length(delta_tensors[1], slice_ctx) slice_ctx = slice_ctx + w_texture * delta_2 else: if len(delta_tensors) >= 1: delta_1 = _align_sequence_length(delta_tensors[0], slice_ctx) slice_ctx = slice_ctx + w_layout * delta_1 if len(delta_tensors) >= 2: delta_2 = _align_sequence_length(delta_tensors[1], slice_ctx) slice_ctx = slice_ctx + w_texture * delta_2 modified_context[batch_idx] = slice_ctx return cross_attn_module.original_forward(x, modified_context, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) # ANCHOR: Original Blending Pathways (FFT / AdaIN / Spectral Bypass) auto_distribute = getattr(cross_attn_module, 'auto_distribute', False) enable_manual = getattr(cross_attn_module, 'enable_manual', False) layout_conds = getattr(cross_attn_module, 'layout_pure_conds', None) style_conds = getattr(cross_attn_module, 'style_pure_conds', None) detail_conds = getattr(cross_attn_module, 'detail_pure_conds', None) if (auto_distribute or enable_manual) and (layout_conds is not None or style_conds is not None or detail_conds is not None): b_early = total_blocks // 3 b_mid = 2 * total_blocks // 3 current_isolation = 0.0 if style_isolation_strength > 1e-4: if isolation_start_step <= progress <= isolation_end_step: range_len = isolation_end_step - isolation_start_step if range_len <= 1e-6: current_isolation = style_isolation_strength else: local_p = (progress - isolation_start_step) / range_len mult = local_p / 0.15 if local_p < 0.15 else ((1.0 - local_p) / 0.15 if local_p > 0.85 else 1.0) current_isolation = style_isolation_strength * mult def blend_cond_lists(pure_list, mixed_list): if pure_list is None or mixed_list is None: return None blended_list = [] for p_c, m_c in zip(pure_list, mixed_list): p_c_aligned = _align_sequence_length(p_c.to(device=m_c.device, dtype=m_c.dtype), m_c) blended = (1.0 - current_isolation) * m_c + current_isolation * p_c_aligned blended_list.append(blended) return blended_list layout_conds_blended = blend_cond_lists(layout_conds, getattr(cross_attn_module, 'layout_mixed_conds', None)) style_conds_blended = blend_cond_lists(style_conds, getattr(cross_attn_module, 'style_mixed_conds', None)) detail_conds_blended = blend_cond_lists(detail_conds, getattr(cross_attn_module, 'detail_mixed_conds', None)) t = block_index / max(1, total_blocks - 1) w_layout = math.cos(math.pi * 0.5 * t) ** 2 w_style = math.sin(math.pi * t) ** 2 w_detail = math.sin(math.pi * 0.5 * t) ** 2 active_conditionings = [] user_weights = [] def filter_active_components(conds_list, weights_list, sub_artists): if conds_list is None: return None, [] act_c, act_w = [], [] for c, w, name in zip(conds_list, weights_list, sub_artists): is_active = any(s <= progress <= e for s, e in artist_schedules[name]) if (artist_schedules and name in artist_schedules) else (start_step_pct <= progress <= end_step_pct) if is_active: act_c.append(c) act_w.append(w) return act_c, act_w def inject_partition_role(p_conds, p_weights, sub_artists, curve_weight, band_mult): filtered_c, filtered_w = filter_active_components(p_conds, p_weights, sub_artists) if filtered_c and curve_weight > 1e-4: for c, w in zip(filtered_c, filtered_w): active_conditionings.append(c) user_weights.append(w * curve_weight * band_mult) inject_partition_role(layout_conds_blended, getattr(cross_attn_module, 'layout_weights', []), getattr(cross_attn_module, 'layout_artists_list', []), w_layout, comp_strength) inject_partition_role(style_conds_blended, getattr(cross_attn_module, 'style_weights', []), getattr(cross_attn_module, 'style_artists_list', []), w_style, style_strength) inject_partition_role(detail_conds_blended, getattr(cross_attn_module, 'detail_weights', []), getattr(cross_attn_module, 'detail_artists_list', []), w_detail, color_strength) active_strength = strength * curve_mult conditionings = active_conditionings else: conditionings = getattr(cross_attn_module, 'mixed_conds', []) user_weights = getattr(cross_attn_module, 'user_weights', []) active_strength = strength * curve_mult if not conditionings: return cross_attn_module.original_forward(x, context, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) try: if not any(mask): return cross_attn_module.original_forward(x, context, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) individuals = [] for raw_cond in conditionings: artist_tensor = raw_cond.to(device=context.device, dtype=context.dtype) if artist_tensor.dim() == 2: artist_tensor = artist_tensor.unsqueeze(0) individuals.append(artist_tensor) if not (auto_distribute or enable_manual): active_indices = [] for idx, artist_name in enumerate(artists_list): is_active = any(s <= progress <= e for s, e in artist_schedules[artist_name]) if (artist_schedules and artist_name in artist_schedules) else (start_step_pct <= progress <= end_step_pct) if is_active: active_indices.append(idx) if not active_indices: return cross_attn_module.original_forward(x, context, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) individuals = [individuals[i] for i in active_indices if i < len(individuals)] user_weights = [user_weights[i] for i in active_indices if i < len(user_weights)] # ANCHOR: XYZ Isolated Artist Splicing Router for legacy paths if xyz_isolation_mode == "First Artist Only" and len(individuals) >= 1: individuals = [individuals[0]] user_weights = [1.0] elif xyz_isolation_mode == "Second Artist Only" and len(individuals) >= 2: individuals = [individuals[1]] user_weights = [1.0] if normalize_weights: total_w = sum(user_weights) if user_weights else 1.0 ws = [w / total_w for w in user_weights] if total_w > 0 else [1.0 / len(individuals)] * len(individuals) else: ws = list(user_weights) # Rotary Positional Scaling Logic if rope_emb is not None and len(individuals) >= 2: try: num_heads = getattr(cross_attn_module, "num_heads", None) or (transformer_options.get("n_heads", 8) if isinstance(transformer_options, dict) else 8) c_dim = x.shape[-1] head_dim = c_dim // num_heads if num_heads > 0 else 64 axes = _axes_dims_from_head_dim(head_dim) scale_vec = _build_rope_scale_vector( head_dim, axes, _lerp(0.1, comp_strength, progress), _lerp(style_strength, 2.0, progress), 25.0, context.device, context.dtype ) rope_emb = _apply_rope_scaling(rope_emb, scale_vec) except Exception as e: logger.error(f"[AnimaArtistMixer] RoPE scale bypass: {e}") # ANCHOR: Context Aligned vs Parallel Attention Routing if sdxl_replication_mode and replication_method in ["spectral", "adain", "contrast_recovery"]: active_context = context[:current_bsz] blended_artists = None current_w_sum = 0.0 for artist_tensor, w in zip(individuals, ws): artist_b = _broadcast_batch(artist_tensor, current_bsz, target_shape=active_context.shape).to(device=context.device, dtype=context.dtype) artist_b = _align_sequence_length(artist_b, active_context) if blended_artists is None: blended_artists = artist_b current_w_sum = float(w) else: new_w_sum = current_w_sum + float(w) alpha = float(w) / new_w_sum if new_w_sum > 0 else 0.5 if replication_method == "spectral": blended_artists = _spectral_hybrid(blended_artists, artist_b, alpha, beta=max(0.1, comp_strength)) elif replication_method == "adain": blended_artists = _cross_token_adain(blended_artists, artist_b, alpha) elif replication_method == "contrast_recovery": blended_artists = _textural_contrast_recovery( active_context, blended_artists, artist_b, alpha=alpha, delta_scale=style_isolation_strength * 2.0, beta=max(0.5, comp_strength) ) current_w_sum = new_w_sum if blended_artists is None: blended_artists = active_context target_val = cross_attn_module.original_forward(x, blended_artists, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) base_out = cross_attn_module.original_forward(x, context, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) out = base_out.clone() for idx, active in enumerate(mask): if active: target_idx = target_val[idx] base_idx = base_out[idx] interp_val = apply_norm_limiting(base_idx, target_idx, active_strength, norm_limit_ratio) if use_norm_limiting else (base_idx * (1.0 - active_strength) + target_idx * active_strength) base_mean, base_std = base_idx.mean(), base_idx.std() target_mean, target_std = interp_val.mean(), interp_val.std() if target_std > 1e-5: gamma = 1.0 - (active_strength * 0.4) out[idx] = (interp_val - target_mean) * ((target_std * (1.0 - gamma) + base_std * gamma) / target_std) + (target_mean * (1.0 - gamma) + base_mean * gamma) else: out[idx] = interp_val return out # ANCHOR: Parallel Attention Blending (Pristine Legacy LERP Math) artist_total = _batched_artists_forward( cross_attn_module, x, context, rope_emb, transformer_options, individuals, ws, current_bsz, *args, **kwargs ) if artist_total is None or artist_total.shape != x.shape: artist_total = torch.zeros_like(x) for artist_tensor, w in zip(individuals, ws): artist_b = _broadcast_batch(artist_tensor, current_bsz, target_shape=context.shape) artist_total += cross_attn_module.original_forward(x, artist_b, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) * w base_out = cross_attn_module.original_forward(x, context, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) out = base_out.clone() for idx, active in enumerate(mask): if active: target_idx = artist_total[idx] base_idx = base_out[idx] target_val = apply_norm_limiting(base_idx, target_idx, active_strength, norm_limit_ratio) if use_norm_limiting else (base_idx * (1.0 - active_strength) + target_idx * active_strength) base_mean, base_std = base_idx.mean(), base_idx.std() target_mean, target_std = target_val.mean(), target_val.std() if target_std > 1e-5: gamma = 1.0 - (active_strength * 0.4) out[idx] = (target_val - target_mean) * ((target_val.std() * (1.0 - gamma) + base_std * gamma) / target_std) + (target_mean * (1.0 - gamma) + base_mean * gamma) else: out[idx] = target_val return out except Exception as e: logger.error(f"[AnimaCrossAttn] Fallback triggered due to forward exception: {e}") cross_attn_module._disabled = True return cross_attn_module.original_forward(x, context, rope_emb=rope_emb, transformer_options=transformer_options, *args, **kwargs) # ANCHOR: WebUI Monkeypatch Controller class AnimaArtistMixerSwitch: def __init__(self, model, strength, comp_strength, style_strength, color_strength, start_step_pct, end_step_pct, pure_conds, mixed_conds, user_weights, start_block, end_block, layout_pure_conds=None, layout_mixed_conds=None, layout_weights=None, layout_artists_list=None, style_pure_conds=None, style_mixed_conds=None, style_weights=None, style_artists_list=None, detail_pure_conds=None, detail_mixed_conds=None, detail_weights=None, detail_artists_list=None, sdxl_replication_mode=False, replication_method="spectral", curve_type="Hold", curve_peak=0.5, use_norm_limiting=False, norm_limit_ratio=0.6, style_isolation_strength=0.0, isolation_start_step=0.30, isolation_end_step=0.80, artists_list=None, artist_schedules=None, turbo_mode=False, xyz_isolation_mode="Merged Blend", delta_tensors=None, auto_distribute=False, enable_manual=False): self.model = model self.strength = strength self.comp_strength = comp_strength self.style_strength = style_strength self.color_strength = color_strength self.start_step_pct = start_step_pct self.end_step_pct = end_step_pct self.pure_conds = pure_conds self.mixed_conds = mixed_conds self.user_weights = user_weights self.start_block = start_block self.end_block = end_block self.sdxl_replication_mode = sdxl_replication_mode self.replication_method = replication_method self.curve_type = curve_type self.curve_peak = curve_peak self.turbo_mode = turbo_mode self.use_norm_limiting = use_norm_limiting self.norm_limit_ratio = norm_limit_ratio self.layout_pure_conds = layout_pure_conds self.layout_mixed_conds = layout_mixed_conds self.layout_weights = layout_weights self.layout_artists_list = layout_artists_list self.style_pure_conds = style_pure_conds self.style_mixed_conds = style_mixed_conds self.style_weights = style_weights self.style_artists_list = style_artists_list self.detail_pure_conds = detail_pure_conds self.detail_mixed_conds = detail_mixed_conds self.detail_weights = detail_weights self.detail_artists_list = detail_artists_list self.style_isolation_strength = style_isolation_strength self.isolation_start_step = isolation_start_step self.isolation_end_step = isolation_end_step self.artists_list = artists_list or [] self.artist_schedules = artist_schedules or {} self.xyz_isolation_mode = xyz_isolation_mode self.delta_tensors = delta_tensors self.auto_distribute = auto_distribute self.enable_manual = enable_manual def set_patches(self): if not hasattr(self.model, 'blocks'): return num_blocks = len(self.model.blocks) sb = max(0, int(self.start_block)) eb = num_blocks - 1 if int(self.end_block) < 0 else min(num_blocks - 1, int(self.end_block)) for i in range(num_blocks): block = self.model.blocks[i] if hasattr(block, 'cross_attn'): if sb <= i <= eb: block.cross_attn.strength = self.strength block.cross_attn.normalize_weights = True block.cross_attn.comp_strength = self.comp_strength block.cross_attn.style_strength = self.style_strength block.cross_attn.color_strength = self.color_strength block.cross_attn.start_step_pct = self.start_step_pct block.cross_attn.end_step_pct = self.end_step_pct block.cross_attn.apply_to_uncond = False block.cross_attn.sdxl_replication_mode = self.sdxl_replication_mode block.cross_attn.replication_method = self.replication_method block.cross_attn.curve_type = self.curve_type block.cross_attn.curve_peak = self.curve_peak block.cross_attn.turbo_mode = self.turbo_mode block.cross_attn.use_norm_limiting = self.use_norm_limiting block.cross_attn.norm_limit_ratio = self.norm_limit_ratio block.cross_attn.pure_conds = self.pure_conds block.cross_attn.mixed_conds = self.mixed_conds block.cross_attn.user_weights = self.user_weights block.cross_attn.layout_pure_conds = self.layout_pure_conds block.cross_attn.layout_mixed_conds = self.layout_mixed_conds block.cross_attn.layout_weights = self.layout_weights block.cross_attn.layout_artists_list = self.layout_artists_list block.cross_attn.style_pure_conds = self.style_pure_conds block.cross_attn.style_mixed_conds = self.style_mixed_conds block.cross_attn.style_weights = self.style_weights block.cross_attn.style_artists_list = self.style_artists_list block.cross_attn.detail_pure_conds = self.detail_pure_conds block.cross_attn.detail_mixed_conds = self.detail_mixed_conds block.cross_attn.detail_weights = self.detail_weights block.cross_attn.detail_artists_list = self.detail_artists_list block.cross_attn.style_isolation_strength = self.style_isolation_strength block.cross_attn.isolation_start_step = self.isolation_start_step block.cross_attn.isolation_end_step = self.isolation_end_step block.cross_attn.artists_list = self.artists_list block.cross_attn.artist_schedules = self.artist_schedules block.cross_attn.xyz_isolation_mode = self.xyz_isolation_mode block.cross_attn.delta_tensors = self.delta_tensors block.cross_attn.auto_distribute = self.auto_distribute block.cross_attn.enable_manual = self.enable_manual block.cross_attn.block_index = i block.cross_attn.total_blocks = num_blocks block.cross_attn._disabled = False if not hasattr(block.cross_attn, 'is_anima_artist_wrapper'): block.cross_attn.original_forward = block.cross_attn.forward block.cross_attn.forward = MethodType(anima_artist_cross_attn_forward, block.cross_attn) block.cross_attn.is_anima_artist_wrapper = True else: self._remove_patch_from_block(block) def set_origin(self): if hasattr(self.model, 'blocks'): for block in self.model.blocks: self._remove_patch_from_block(block) def _remove_patch_from_block(self, block): if hasattr(block, 'cross_attn'): if hasattr(block.cross_attn, 'is_anima_artist_wrapper'): block.cross_attn.forward = block.cross_attn.original_forward delattr(block.cross_attn, 'original_forward') delattr(block.cross_attn, 'is_anima_artist_wrapper') attrs_to_remove = [ 'strength', 'normalize_weights', 'comp_strength', 'style_strength', 'color_strength', 'start_step_pct', 'end_step_pct', 'apply_to_uncond', 'sdxl_replication_mode', 'replication_method', 'curve_type', 'curve_peak', 'use_norm_limiting', 'norm_limit_ratio', 'pure_conds', 'mixed_conds', 'user_weights', 'layout_pure_conds', 'layout_mixed_conds', 'layout_weights', 'layout_artists_list', 'style_pure_conds', 'style_mixed_conds', 'style_weights', 'style_artists_list', 'detail_pure_conds', 'detail_mixed_conds', 'detail_weights', 'detail_artists_list', 'block_index', 'total_blocks', '_disabled', 'turbo_mode', 'style_isolation_strength', 'isolation_start_step', 'isolation_end_step', 'artists_list', 'artist_schedules', 'xyz_isolation_mode', 'delta_tensors', 'auto_distribute', 'enable_manual' ] for attr in attrs_to_remove: if hasattr(block.cross_attn, attr): delattr(block.cross_attn, attr) # ANCHOR: Main WebUI Script Hook class Script(scripts.Script): def __init__(self): self.mixer_switch = None self.active = False def title(self): return "Anima Artist Mixer" def show(self, is_img2img): return scripts.AlwaysVisible def ui(self, is_img2img): with gr.Accordion("Anima Artist Mixer", open=False): with gr.Row(): enabled = gr.Checkbox(label="Enable Artist Mixer", value=False) turbo_mode = gr.Checkbox(label="Enable Low-Step Mode (Tapered early-fry)", value=False) preset_dropdown = gr.Dropdown( label="Presets", choices=list(PRESETS.keys()), value="Default" ) with gr.Row(): style_isolation_strength = gr.Slider( label="Style Isolation Strength", minimum=0.0, maximum=1.0, value=0.0, step=0.05 ) replication_method = gr.Dropdown( label="Artist Blending Mathematics", choices=[ "Spectral (LERoPE)", "Classic LERP", "AdaIN", "Textural Recovery (Anti-Median)", "Decoupled Causal Delta Splicing", "None (Bypass to Classic Parallel Attention)" ], value="None (Bypass to Classic Parallel Attention)" ) with gr.Row(): isolation_start_step = gr.Slider( label="Isolation Start Step %", minimum=0.0, maximum=1.0, value=0.0, step=0.05 ) isolation_end_step = gr.Slider( label="Isolation End Step %", minimum=0.0, maximum=1.0, value=1.0, step=0.05 ) with gr.Accordion("Assign Specific Artists to Layer Roles & Strengths", open=False): with gr.Row(): auto_distribute = gr.Checkbox( label="Auto-Distribute Artists by Prompt Order", value=False ) enable_manual = gr.Checkbox( label="Override with Manual Layer Roles", value=False ) with gr.Column(visible=False) as manual_input_container: gr.Markdown("Type specific artist name(s) (comma-separated) to assign them to specific roles.") layout_artists = gr.Textbox( label="Layout & Composition Roles (Early Layers)", placeholder="e.g. wlop, greg rutkowski", value="" ) style_artists = gr.Textbox( label="Painting Style & Shape Roles (Middle Layers)", placeholder="e.g. sakimichan, artstation", value="" ) detail_artists = gr.Textbox( label="Color, Contrast & Texture Roles (Late Layers)", placeholder="e.g. mucha, daito", value="" ) gr.Markdown("---") keep_baseline = gr.Checkbox( label="Keep a Soft Mix for Unassigned Artists", value=True, visible=False ) baseline_strength = gr.Slider( label="Soft Mix Strength", minimum=0.0, maximum=0.8, value=0.25, step=0.05, visible=False ) gr.Markdown("### **Fine-Tune Layer Strengths**") comp_strength = gr.Slider( label="Composition Influence (Early Layers)", minimum=0.1, maximum=2.0, value=1.0, step=0.1 ) style_strength = gr.Slider( label="Painting Influence (Middle Layers)", minimum=0.0, maximum=2.0, value=1.0, step=0.1 ) color_strength = gr.Slider( label="Color & Texture Influence (Late Layers)", minimum=0.0, maximum=2.0, value=1.0, step=0.1 ) def sync_ui_visibilities(auto, manual, keep): roles_active = auto or manual return { manual_input_container: gr.update(visible=manual), keep_baseline: gr.update(visible=roles_active), baseline_strength: gr.update(visible=roles_active and keep) } auto_distribute.change( fn=lambda active: gr.update(value=False) if active else gr.update(), inputs=[auto_distribute], outputs=[enable_manual] ) enable_manual.change( fn=lambda active: gr.update(value=False) if active else gr.update(), inputs=[enable_manual], outputs=[auto_distribute] ) for trigger in [auto_distribute, enable_manual, keep_baseline]: trigger.change( fn=sync_ui_visibilities, inputs=[auto_distribute, enable_manual, keep_baseline], outputs=[manual_input_container, keep_baseline, baseline_strength] ) with gr.Accordion("Advanced: Step Ranges & Fine-Tuning", open=False): enable_cache = gr.Checkbox( label="Enable Text Encoding Cache", value=True ) gr.Markdown("### **Step Blending & Multi-Curve Gates**") curve_type = gr.Dropdown( label="Step Influence Curve Shape", choices=["Hold", "Smooth", "Triangle", "Front loaded", "Back loaded"], value="Hold" ) curve_peak = gr.Slider( label="Influence Curve Peak Position", minimum=0.0, maximum=1.0, value=0.5, step=0.05, visible=False ) with gr.Row(): start_step_pct = gr.Slider(label="Start Style at Step %", minimum=0.0, maximum=1.0, value=0.0, step=0.05) end_step_pct = gr.Slider(label="Stop Style at Step %", minimum=0.0, maximum=1.0, value=1.0, step=0.05) with gr.Row(): start_block = gr.Number(label="First Active Model Layer", value=0, precision=0) end_block = gr.Number(label="Last Active Model Layer", value=-1, precision=0) gr.Markdown("### **Safety & Blending Constraints**") with gr.Row(): use_norm_limiting = gr.Checkbox( label="Enable Quality-Safe Norm Limiting", value=False ) norm_limit_ratio = gr.Slider( label="Norm Limiting Ratio Limit", minimum=0.1, maximum=2.0, value=0.6, step=0.05, visible=False ) curve_type.change( fn=lambda val: gr.update(visible=val in ["Smooth", "Triangle"] and not turbo_mode), inputs=[curve_type], outputs=[curve_peak] ) use_norm_limiting.change( fn=lambda active: gr.update(visible=active), inputs=[use_norm_limiting], outputs=[norm_limit_ratio] ) preset_targets = [ replication_method, comp_strength, style_strength, color_strength, style_isolation_strength, curve_type, isolation_start_step, isolation_end_step, auto_distribute, enable_manual ] def _apply_preset(choice): p = PRESETS.get(choice, {}) if not p: return tuple(gr.update() for _ in range(10)) return ( p["replication_method"], p["comp_strength"], p["style_strength"], p["color_strength"], p["style_isolation_strength"], p["curve_type"], p["isolation_start_step"], p["isolation_end_step"], p.get("auto_distribute", gr.update()), p.get("enable_manual", gr.update()) ) preset_dropdown.change( fn=_apply_preset, inputs=[preset_dropdown], outputs=preset_targets ) return [ enabled, style_isolation_strength, replication_method, isolation_start_step, isolation_end_step, auto_distribute, enable_manual, layout_artists, style_artists, detail_artists, keep_baseline, baseline_strength, comp_strength, style_strength, color_strength, enable_cache, curve_type, curve_peak, start_step_pct, end_step_pct, start_block, end_block, use_norm_limiting, norm_limit_ratio, turbo_mode ] def process_batch(self, p, enabled, style_isolation_strength, replication_method, isolation_start_step, isolation_end_step, auto_distribute, enable_manual, layout_artists_str, style_artists_str, detail_artists_str, keep_baseline, baseline_strength, comp_strength, style_strength, color_strength, enable_cache, curve_type, curve_peak, start_step_pct, end_step_pct, start_block, end_block, use_norm_limiting, norm_limit_ratio, turbo_mode, **kwargs): self.__init__() xyz_overrides = getattr(p, "anima_xyz_overrides", {}) if "enabled" in xyz_overrides: enabled = xyz_overrides["enabled"] if "replication_method" in xyz_overrides: replication_method = xyz_overrides["replication_method"] if "preset" in xyz_overrides: preset_choice = xyz_overrides["preset"] if preset_choice in PRESETS and preset_choice != "Custom": p_dict = PRESETS[preset_choice] replication_method = p_dict["replication_method"] comp_strength = p_dict["comp_strength"] style_strength = p_dict["style_strength"] color_strength = p_dict["color_strength"] style_isolation_strength = p_dict["style_isolation_strength"] curve_type = p_dict["curve_type"] isolation_start_step = p_dict["isolation_start_step"] isolation_end_step = p_dict["isolation_end_step"] auto_distribute = p_dict["auto_distribute"] enable_manual = p_dict["enable_manual"] xyz_isolation_mode = xyz_overrides.get("xyz_isolation_mode", "Merged Blend") if not enabled: return method_str = str(replication_method).lower() if "spectral" in method_str: sanitized_method = "spectral" elif "adain" in method_str: sanitized_method = "adain" elif "contrast" in method_str: sanitized_method = "contrast_recovery" elif "decoupled" in method_str: sanitized_method = "decoupled_delta" elif "lerp" in method_str: sanitized_method = "lerp" else: sanitized_method = "none" sdxl_replication_mode = (sanitized_method != "none" and sanitized_method != "lerp") raw_prompt = p.prompt total_steps = getattr(p, "steps", 20) or 20 # Parse artist-specific schedules before extracting tags artist_schedules = parse_scheduled_artists(raw_prompt, total_steps) artists_with_weights, base_prompt = extract_artists_and_base(raw_prompt) if not artists_with_weights: return artists = [aw[0] for aw in artists_with_weights] user_weights = [aw[1] for aw in artists_with_weights] logger.info(f"[AnimaArtistMixer] Found Prompt Artists: {list(zip(artists, user_weights))} | Clean Base: {base_prompt}") p.prompt = base_prompt self.active = True sd_model = p.sd_model # ANCHOR: Execution Mode Branching Selection delta_tensors = None pure_conds = [] mixed_conds = [] with torch.no_grad(), devices.autocast(): if sanitized_method == "decoupled_delta": try: base_cond = get_conditioning_cached(sd_model, base_prompt, use_cache=enable_cache) delta_tensors = [] for artist_name in artists: isolated_prompt = f"{base_prompt}, {artist_name}" artist_cond = get_conditioning_cached(sd_model, isolated_prompt, use_cache=enable_cache) aligned_base = _align_sequence_length(base_cond, artist_cond) delta_tensors.append(artist_cond - aligned_base) except Exception as e: logger.error(f"Isolated decoupled delta text encoding execution failed: {e}") self.active = False return else: for artist in artists: pure_prompt = artist mixed_prompt = f"{artist}\n{base_prompt}" if base_prompt else artist try: p_cond = get_conditioning_cached(sd_model, pure_prompt, use_cache=enable_cache) m_cond = get_conditioning_cached(sd_model, mixed_prompt, use_cache=enable_cache) except Exception as e: logger.error(f"No compatible Text Encoder pipeline found: {e}") self.active = False return pure_conds.append(p_cond) mixed_conds.append(m_cond) # ANCHOR: Segment-allocated partition mapper (tracks artist names) def partition_conds(cond_list): l_conds, s_conds, d_conds = [], [], [] l_weights, s_weights, d_weights = [], [], [] l_artists, s_artists, d_artists = [], [], [] if not cond_list: return None, None, None, None, None, None, None, None, None if auto_distribute: n = len(artists_with_weights) if n == 1: for idx, (artist_name, w) in enumerate(artists_with_weights): c = cond_list[idx] l_conds.append(c); l_weights.append(w); l_artists.append(artist_name) s_conds.append(c); s_weights.append(w); s_artists.append(artist_name) d_conds.append(c); d_weights.append(w); d_artists.append(artist_name) elif n == 2: for idx, (artist_name, w) in enumerate(artists_with_weights): c = cond_list[idx] if idx == 0: l_conds.append(c); l_weights.append(w); l_artists.append(artist_name) s_conds.append(c); s_weights.append(w); s_artists.append(artist_name) if keep_baseline: d_conds.append(c); d_weights.append(w * baseline_strength); d_artists.append(artist_name) else: d_conds.append(c); d_weights.append(w); d_artists.append(artist_name) if keep_baseline: l_conds.append(c); l_weights.append(w * baseline_strength); l_artists.append(artist_name) s_conds.append(c); s_weights.append(w * baseline_strength); s_artists.append(artist_name) else: s1 = max(1, n // 3) s2 = max(s1 + 1, 2 * n // 3) for idx, (artist_name, w) in enumerate(artists_with_weights): c = cond_list[idx] is_layout = idx < s1 is_style = s1 <= idx < s2 is_detail = idx >= s2 if is_layout: l_conds.append(c); l_weights.append(w); l_artists.append(artist_name) elif keep_baseline: l_conds.append(c); l_weights.append(w * baseline_strength); l_artists.append(artist_name) if is_style: s_conds.append(c); s_weights.append(w); s_artists.append(artist_name) elif keep_baseline: s_conds.append(c); s_weights.append(w * baseline_strength); s_artists.append(artist_name) if is_detail: d_conds.append(c); d_weights.append(w); d_artists.append(artist_name) elif keep_baseline: d_conds.append(c); d_weights.append(w * baseline_strength); d_artists.append(artist_name) elif enable_manual: layout_names = [n.strip().lower() for n in layout_artists_str.split(",") if n.strip()] style_names = [n.strip().lower() for n in style_artists_str.split(",") if n.strip()] detail_names = [n.strip().lower() for n in detail_artists_str.split(",") if n.strip()] for idx, (artist_name, w) in enumerate(artists_with_weights): name_lower = artist_name.lower() c = cond_list[idx] if layout_names: is_match = any(ln in name_lower for ln in layout_names) if is_match: l_conds.append(c); l_weights.append(w); l_artists.append(artist_name) elif keep_baseline: l_conds.append(c); l_weights.append(w * baseline_strength); l_artists.append(artist_name) else: l_conds.append(c); l_weights.append(w); l_artists.append(artist_name) if style_names: is_match = any(sn in name_lower for sn in style_names) if is_match: s_conds.append(c); s_weights.append(w); s_artists.append(artist_name) elif keep_baseline: s_conds.append(c); s_weights.append(w * baseline_strength); s_artists.append(artist_name) else: s_conds.append(c); s_weights.append(w); s_artists.append(artist_name) if detail_names: is_match = any(dn in name_lower for dn in detail_names) if is_match: d_conds.append(c); d_weights.append(w); d_artists.append(artist_name) elif keep_baseline: d_conds.append(c); d_weights.append(w * baseline_strength); d_artists.append(artist_name) else: d_conds.append(c); d_weights.append(w); d_artists.append(artist_name) else: l_conds = l_weights = l_artists = None s_conds = s_weights = s_artists = None d_conds = d_weights = d_artists = None return l_conds, l_weights, l_artists, s_conds, s_weights, s_artists, d_conds, d_weights, d_artists layout_pure_conds, layout_weights, layout_artists_list, style_pure_conds, style_weights, style_artists_list, detail_pure_conds, detail_weights, detail_artists_list = partition_conds(pure_conds) layout_mixed_conds, _, _, style_mixed_conds, _, _, detail_mixed_conds, _, _ = partition_conds(mixed_conds) diffusion_model = None if hasattr(p.sd_model, "forge_objects") and hasattr(p.sd_model.forge_objects, "unet"): diffusion_model = p.sd_model.forge_objects.unet.model.diffusion_model elif hasattr(p.sd_model, "model") and hasattr(p.sd_model.model, "diffusion_model"): diffusion_model = p.sd_model.model.diffusion_model if diffusion_model is None: logger.error("Failed to map active diffusion model pipeline.") self.active = False return self.mixer_switch = AnimaArtistMixerSwitch( diffusion_model, 1.0, comp_strength, style_strength, color_strength, start_step_pct, end_step_pct, pure_conds, mixed_conds, user_weights, start_block, end_block, layout_pure_conds, layout_mixed_conds, layout_weights, layout_artists_list, style_pure_conds, style_mixed_conds, style_weights, style_artists_list, detail_pure_conds, detail_mixed_conds, detail_weights, detail_artists_list, sdxl_replication_mode, sanitized_method, curve_type, curve_peak, use_norm_limiting, norm_limit_ratio, style_isolation_strength, isolation_start_step, isolation_end_step, artists_list=artists, artist_schedules=artist_schedules, turbo_mode=turbo_mode, xyz_isolation_mode=xyz_isolation_mode, delta_tensors=delta_tensors, auto_distribute=auto_distribute, enable_manual=enable_manual ) self.mixer_switch.set_patches() def postprocess(self, p, processed, *args): if hasattr(self, "mixer_switch") and self.mixer_switch is not None: self.mixer_switch.set_origin() self.mixer_switch = None logger.info("[AnimaArtistMixer] Reverted cross-attention patches.") # ANCHOR: X/Y/Z Plot Dynamic Axis Injection def find_xyz_grid(): for data in scripts.scripts_data: if os.path.basename(data.path) in ("xyz_grid.py", "xy_grid.py"): return data.module return None def make_axis_on_xyz_grid(): xyz_grid = find_xyz_grid() if xyz_grid is None: return if any(x.label.startswith("[Anima]") for x in xyz_grid.axis_options): return def set_xyz_value(p, x, field): if not hasattr(p, "anima_xyz_overrides"): p.anima_xyz_overrides = {} p.anima_xyz_overrides[field] = x def parse_bool(val): if isinstance(val, str): return val.lower() in ("true", "yes", "1", "on", "enable", "enabled") return bool(val) # Cleaned down to only required user parameters with clean, aesthetic labels extra_axis_options = [ xyz_grid.AxisOption( "[Anima] Enabled", str, lambda p, x, xs: set_xyz_value(p, parse_bool(x), "enabled"), choices=lambda: ["True", "False"] ), xyz_grid.AxisOption( "[Anima] Presets", str, lambda p, x, xs: set_xyz_value(p, str(x), "preset"), choices=lambda: list(PRESETS.keys()) ), xyz_grid.AxisOption( "[Anima] Blending Mathematics", str, lambda p, x, xs: set_xyz_value(p, str(x), "replication_method"), choices=lambda: [ "Spectral (LERoPE)", "Classic LERP", "AdaIN", "Textural Recovery (Anti-Median)", "Decoupled Causal Delta Splicing", "None (Bypass to Classic Parallel Attention)" ] ), xyz_grid.AxisOption( "[Anima] Artist Isolation Mode", str, lambda p, x, xs: set_xyz_value(p, str(x), "xyz_isolation_mode"), choices=lambda: ["First Artist Only", "Second Artist Only", "Merged Blend"] ) ] xyz_grid.axis_options.extend(extra_axis_options) script_callbacks.on_before_ui(make_axis_on_xyz_grid)