import math
from copy import deepcopy
from torch.nn import Upsample
import comfy.model_management as model_management
from comfy.model_patcher import set_model_options_patch_replace
from comfy.ldm.modules.attention import attention_basic, attention_xformers, attention_pytorch, attention_split, attention_sub_quad, optimized_attention_for_device
from .experimental_temperature import temperature_patcher
import comfy.samplers
import comfy.utils
import numpy as np
import torch
import torch.nn.functional as F
from colorama import Fore, Style
import json
import os
import random
import base64
original_sampling_function = None
current_dir = os.path.dirname(os.path.realpath(__file__))
json_preset_path = os.path.join(current_dir, 'presets')
attnfunc = optimized_attention_for_device(model_management.get_torch_device())
check_string = "UEFUUkVPTi50eHQ="
support_string = b'CgoKClRoYW5rIHlvdSBmb3IgdXNpbmcgbXkgbm9kZXMhCgpJZiB5b3UgZW5qb3kgaXQsIHBsZWFzZSBjb25zaWRlciBzdXBwb3J0aW5nIG1lIG9uIFBhdHJlb24gdG8ga2VlcCB0aGUgbWFnaWMgZ29pbmchCgpWaXNpdDoKCmh0dHBzOi8vd3d3LnBhdHJlb24uY29tL2V4dHJhbHRvZGV1cwoKCgo='
def support_function():
if base64.b64decode(check_string).decode('utf8') not in os.listdir(current_dir):
print(base64.b64decode(check_string).decode('utf8'))
print(base64.b64decode(support_string).decode('utf8'))
def sampling_function_patched(model, x, timestep, uncond, cond, cond_scale, model_options={}, seed=None, **kwargs):
cond_copy = cond
uncond_copy = uncond
for fn in model_options.get("sampler_patch_model_pre_cfg_function", []):
args = {"model": model, "sigma": timestep, "model_options": model_options}
model, model_options = fn(args)
if "sampler_pre_cfg_automatic_cfg_function" in model_options:
uncond, cond, cond_scale = model_options["sampler_pre_cfg_automatic_cfg_function"](
sigma=timestep, uncond=uncond, cond=cond, cond_scale=cond_scale
)
if math.isclose(cond_scale, 1.0) and model_options.get("disable_cfg1_optimization", False) == False:
uncond_ = None
else:
uncond_ = uncond
conds = [cond, uncond_]
out = comfy.samplers.calc_cond_batch(model, conds, x, timestep, model_options)
for fn in model_options.get("sampler_pre_cfg_function", []):
args = {"conds":conds, "conds_out": out, "cond_scale": cond_scale, "timestep": timestep,
"input": x, "sigma": timestep, "model": model, "model_options": model_options}
out = fn(args)
cond_pred = out[0]
uncond_pred = out[1]
if "sampler_cfg_function" in model_options:
args = {"cond": x - cond_pred, "uncond": x - uncond_pred, "cond_scale": cond_scale, "timestep": timestep, "input": x, "sigma": timestep,
"cond_denoised": cond_pred, "uncond_denoised": uncond_pred, "model": model, "model_options": model_options, "cond_pos": cond_copy, "cond_neg": uncond_copy}
cfg_result = x - model_options["sampler_cfg_function"](args)
else:
cfg_result = uncond_pred + (cond_pred - uncond_pred) * cond_scale
for fn in model_options.get("sampler_post_cfg_function", []):
args = {"denoised": cfg_result, "cond": cond_copy, "uncond": uncond_copy, "model": model, "uncond_denoised": uncond_pred, "cond_denoised": cond_pred,
"sigma": timestep, "model_options": model_options, "input": x}
cfg_result = fn(args)
return cfg_result
def monkey_patching_comfy_sampling_function():
global original_sampling_function
if original_sampling_function is None:
original_sampling_function = comfy.samplers.sampling_function
# Make sure to only patch once
if hasattr(comfy.samplers.sampling_function, '_automatic_cfg_decorated'):
return
comfy.samplers.sampling_function = sampling_function_patched
comfy.samplers.sampling_function._automatic_cfg_decorated = True # flag to check monkey patch
def make_sampler_pre_cfg_automatic_cfg_function(minimum_sigma_to_disable_uncond=0, maximum_sigma_to_enable_uncond=1000000, disabled_cond_start=10000,disabled_cond_end=10000):
def sampler_pre_cfg_automatic_cfg_function(sigma, uncond, cond, cond_scale, **kwargs):
if sigma[0] < minimum_sigma_to_disable_uncond or sigma[0] > maximum_sigma_to_enable_uncond:
uncond = None
if sigma[0] <= disabled_cond_start and sigma[0] > disabled_cond_end:
cond = None
return uncond, cond, cond_scale
return sampler_pre_cfg_automatic_cfg_function
def get_entropy(tensor):
hist = np.histogram(tensor.cpu(), bins=100)[0]
hist = hist / hist.sum()
hist = hist[hist > 0]
return -np.sum(hist * np.log2(hist))
def map_sigma(sigma, sigmax, sigmin):
return 1 + ((sigma - sigmax) * (0 - 1)) / (sigmin - sigmax)
def center_latent_mean_values(latent, per_channel, mult):
for b in range(len(latent)):
if per_channel:
for c in range(len(latent[b])):
latent[b][c] -= latent[b][c].mean() * mult
else:
latent[b] -= latent[b].mean() * mult
return latent
def get_denoised_ranges(latent, measure="hard", top_k=0.25):
chans = []
for x in range(len(latent)):
max_values = torch.topk(latent[x] - latent[x].mean() if measure == "range" else latent[x], k=int(len(latent[x])*top_k), largest=True).values
min_values = torch.topk(latent[x] - latent[x].mean() if measure == "range" else latent[x], k=int(len(latent[x])*top_k), largest=False).values
max_val = torch.mean(max_values).item()
min_val = abs(torch.mean(min_values).item()) if measure == "soft" else torch.mean(torch.abs(min_values)).item()
denoised_range = (max_val + min_val) / 2
chans.append(denoised_range**2 if measure == "hard_squared" else denoised_range)
return chans
def get_sigmin_sigmax(model):
model_sampling = model.model.model_sampling
sigmin = model_sampling.sigma(model_sampling.timestep(model_sampling.sigma_min))
sigmax = model_sampling.sigma(model_sampling.timestep(model_sampling.sigma_max))
return sigmin, sigmax
def gaussian_similarity(x, y, sigma=1.0):
diff = (x - y) ** 2
return torch.exp(-diff / (2 * sigma ** 2))
def check_skip(sigma, high_sigma_threshold, low_sigma_threshold):
return sigma > high_sigma_threshold or sigma < low_sigma_threshold
def max_abs(tensors):
shape = tensors.shape
tensors = tensors.reshape(shape[0], -1)
tensors_abs = torch.abs(tensors)
max_abs_idx = torch.argmax(tensors_abs, dim=0)
result = tensors[max_abs_idx, torch.arange(tensors.shape[1])]
return result.reshape(shape[1:])
def gaussian_kernel(size: int, sigma: float):
x = torch.arange(size) - size // 2
gauss = torch.exp(-x**2 / (2 * sigma**2))
kernel = gauss / gauss.sum()
return kernel.view(1, size) * kernel.view(size, 1)
def blur_tensor(tensor, kernel_size = 9, sigma = 2.0):
tensor = tensor.unsqueeze(0)
C = tensor.size(1)
kernel = gaussian_kernel(kernel_size, sigma)
kernel = kernel.expand(C, 1, kernel_size, kernel_size).to(tensor.device).to(dtype=tensor.dtype, device=tensor.device)
padding = kernel_size // 2
tensor = F.pad(tensor, (padding, padding, padding, padding), mode='reflect')
blurred_tensor = F.conv2d(tensor, kernel, groups=C)
return blurred_tensor.squeeze(0)
def smallest_distances(tensors):
if all(torch.equal(tensors[0], tensor) for tensor in tensors[1:]):
return tensors[0]
set_device = tensors.device
min_val = torch.full(tensors[0].shape, float("inf")).to(set_device)
result = torch.zeros_like(tensors[0])
for idx1, t1 in enumerate(tensors):
temp_diffs = torch.zeros_like(tensors[0])
for idx2, t2 in enumerate(tensors):
if idx1 != idx2:
temp_diffs += torch.abs(torch.sub(t1, t2))
min_val = torch.minimum(min_val, temp_diffs)
mask = torch.eq(min_val,temp_diffs)
result[mask] = t1[mask]
return result
def rescale(tensor, multiplier=2):
batch, seq_length, features = tensor.shape
H = W = int(seq_length**0.5)
tensor_reshaped = tensor.view(batch, features, H, W)
new_H = new_W = int(H * multiplier)
resized_tensor = F.interpolate(tensor_reshaped, size=(new_H, new_W), mode='bilinear', align_corners=False)
return resized_tensor.view(batch, new_H * new_W, features)
# from https://discuss.pytorch.org/t/help-regarding-slerp-function-for-generative-model-sampling/32475
def slerp(high, low, val):
dims = low.shape
#flatten to batches
low = low.reshape(dims[0], -1)
high = high.reshape(dims[0], -1)
low_norm = low/torch.norm(low, dim=1, keepdim=True)
high_norm = high/torch.norm(high, dim=1, keepdim=True)
# in case we divide by zero
low_norm[low_norm != low_norm] = 0.0
high_norm[high_norm != high_norm] = 0.0
omega = torch.acos((low_norm*high_norm).sum(1))
so = torch.sin(omega)
res = (torch.sin((1.0-val)*omega)/so).unsqueeze(1)*low + (torch.sin(val*omega)/so).unsqueeze(1) * high
return res.reshape(dims)
normalize_tensor = lambda x: x / x.norm()
def random_swap(tensors, proportion=1):
num_tensors = tensors.shape[0]
if num_tensors < 2: return tensors[0],0
tensor_size = tensors[0].numel()
if tensor_size < 100: return tensors[0],0
true_count = int(tensor_size * proportion)
mask = torch.cat((torch.ones(true_count, dtype=torch.bool, device=tensors[0].device),
torch.zeros(tensor_size - true_count, dtype=torch.bool, device=tensors[0].device)))
mask = mask[torch.randperm(tensor_size)].reshape(tensors[0].shape)
if num_tensors == 2 and proportion < 1:
index_tensor = torch.ones_like(tensors[0], dtype=torch.int64, device=tensors[0].device)
else:
index_tensor = torch.randint(1 if proportion < 1 else 0, num_tensors, tensors[0].shape, device=tensors[0].device)
for i, t in enumerate(tensors):
if i == 0: continue
merge_mask = index_tensor == i & mask
tensors[0][merge_mask] = t[merge_mask]
return tensors[0]
def multi_tensor_check_mix(tensors):
if tensors[0].numel() < 2 or len(tensors) < 2:
return tensors[0]
ref_tensor_shape = tensors[0].shape
sequence_tensor = torch.arange(tensors[0].numel(), device=tensors[0].device) % len(tensors)
reshaped_sequence = sequence_tensor.view(ref_tensor_shape)
for i in range(len(tensors)):
if i == 0: continue
mask = reshaped_sequence == i
tensors[0][mask] = tensors[i][mask]
return tensors[0]
def sspow(input_tensor, p=2):
return input_tensor.abs().pow(p) * input_tensor.sign()
def sspown(input_tensor, p=2):
abs_t = input_tensor.abs()
abs_t = (abs_t - abs_t.min()) / (abs_t.max() - abs_t.min())
return abs_t.pow(p) * input_tensor.sign()
def gradient_merge(tensor1, tensor2, start_value=0, dim=0):
if torch.numel(tensor1) <= 1: return tensor1
if dim >= tensor1.dim(): dim = 0
size = tensor1.size(dim)
alpha = torch.linspace(start_value, 1-start_value, steps=size, device=tensor1.device).view([-1 if i == dim else 1 for i in range(tensor1.dim())])
return tensor1 * alpha + tensor2 * (1 - alpha)
def save_tensor(input_tensor,name):
if "rndnum" in name:
rndnum = str(random.randint(100000,999999))
name = name.replace("rndnum", rndnum)
output_directory = os.path.join(current_dir, 'saved_tensors')
os.makedirs(output_directory, exist_ok=True)
output_file_path = os.path.join(output_directory, f"{name}.pt")
torch.save(input_tensor, output_file_path)
return input_tensor
def print_and_return(input_tensor, *args):
for what_to_print in args:
print(" ",what_to_print)
return input_tensor
# Experimental testings
def normal_attention(q, k, v, mask=None):
attention_scores = torch.matmul(q, k.transpose(-2, -1))
d_k = k.size(-1)
attention_scores = attention_scores / torch.sqrt(torch.tensor(d_k, dtype=torch.float32))
if mask is not None:
attention_scores = attention_scores.masked_fill(mask == 0, float('-inf'))
attention_weights = F.softmax(attention_scores, dim=-1)
output = torch.matmul(attention_weights, v)
return output
def split_heads(x, n_heads):
batch_size, seq_length, hidden_dim = x.size()
head_dim = hidden_dim // n_heads
x = x.view(batch_size, seq_length, n_heads, head_dim)
return x.permute(0, 2, 1, 3)
def combine_heads(x, n_heads):
batch_size, n_heads, seq_length, head_dim = x.size()
hidden_dim = n_heads * head_dim
x = x.permute(0, 2, 1, 3).contiguous()
return x.view(batch_size, seq_length, hidden_dim)
def sparsemax(logits):
logits_sorted, _ = torch.sort(logits, descending=True, dim=-1)
cumulative_sum = torch.cumsum(logits_sorted, dim=-1) - 1
rho = (logits_sorted > cumulative_sum / (torch.arange(logits.size(-1)) + 1).to(logits.device)).float()
tau = (cumulative_sum / rho.sum(dim=-1, keepdim=True)).gather(dim=-1, index=rho.sum(dim=-1, keepdim=True).long() - 1)
return torch.max(torch.zeros_like(logits), logits - tau)
def attnfunc_custom(q, k, v, n_heads, eval_string = ""):
q = split_heads(q, n_heads)
k = split_heads(k, n_heads)
v = split_heads(v, n_heads)
d_k = q.size(-1)
scores = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(d_k)
if eval_string == "":
attn_weights = F.softmax(scores, dim=-1)
else:
attn_weights = eval(eval_string)
output = torch.matmul(attn_weights, v)
output = combine_heads(output, n_heads)
return output
def min_max_norm(t):
return (t - t.min()) / (t.max() - t.min())
class attention_modifier():
def __init__(self, self_attn_mod_eval, conds = None):
self.self_attn_mod_eval = self_attn_mod_eval
self.conds = conds
def modified_attention(self, q, k, v, extra_options, mask=None):
"""extra_options contains: {'cond_or_uncond': [1, 0], 'sigmas': tensor([14.6146], device='cuda:0'),
'original_shape': [2, 4, 128, 128], 'transformer_index': 4, 'block': ('middle', 0),
'block_index': 3, 'n_heads': 20, 'dim_head': 64, 'attn_precision': None}"""
if "attnbc" in self.self_attn_mod_eval:
attnbc = attention_basic(q, k, v, extra_options['n_heads'], mask)
if "normattn" in self.self_attn_mod_eval:
normattn = normal_attention(q, k, v, mask)
if "attnxf" in self.self_attn_mod_eval:
attnxf = attention_xformers(q, k, v, extra_options['n_heads'], mask)
if "attnpy" in self.self_attn_mod_eval:
attnpy = attention_pytorch(q, k, v, extra_options['n_heads'], mask)
if "attnsp" in self.self_attn_mod_eval:
attnsp = attention_split(q, k, v, extra_options['n_heads'], mask)
if "attnsq" in self.self_attn_mod_eval:
attnsq = attention_sub_quad(q, k, v, extra_options['n_heads'], mask)
if "attnopt" in self.self_attn_mod_eval:
attnopt = attnfunc(q, k, v, extra_options['n_heads'], mask)
n_heads = extra_options['n_heads']
if self.conds is not None:
cond_pos_l = self.conds[0][..., :768].cuda()
cond_neg_l = self.conds[1][..., :768].cuda()
if self.conds[0].shape[-1] > 768:
cond_pos_g = self.conds[0][..., 768:2048].cuda()
cond_neg_g = self.conds[1][..., 768:2048].cuda()
return eval(self.self_attn_mod_eval)
def experimental_functions(cond_input, method, exp_value, exp_normalize, pcp, psi, sigma, sigmax, attention_modifiers_input, args, model_options_copy, eval_string = ""):
"""
There may or may not be an actual reasoning behind each of these methods.
Some like the sine value have interesting properties. Enabled for both cond and uncond preds it somehow make them stronger.
Note that there is a "normalize" toggle and it may change greatly the end result since some operation will totaly butcher the values.
"theDaRkNeSs" for example without normalizing seems to darken if used for cond/uncond (not with the cond as the uncond or something).
Maybe just with the positive. I don't remember. I leave it for now if you want to play around.
The eval_string can be used to create the uncond replacement.
I made it so it's split by semicolons and only the last split is the value in used.
What is before is added in an array named "v".
pcp is previous cond_pred
psi is previous sigma
args is the CFG function input arguments with the added cond/unconds (like the actual activation conditionings) named respectively "cond_pos" and "cond_neg"
So if you write:
pcp if sigma < 7 else -pcp;
print("it works too just don't use the output I guess");
v[0] if sigma < 14 else torch.zeros_like(cond);
v[-1]*2
Well the first line becomes v[0], second v[1] etc.
The last one becomes the result.
Note that it's just an example, I don't see much interest in that one.
Using comfy.samplers.calc_cond_batch(args["model"], [args["cond_pos"], None], args["input"], args["timestep"], args["model_options"])[0]
can work too.
This whole mess has for initial goal to attempt to find the best way (or have some bruteforcing fun) to replace the uncond pred for as much as possible.
Or simply to try things around :)
"""
if method == "cond_pred":
return cond_input
default_device = cond_input.device
# print()
# print(get_entropy(cond))
cond = cond_input.clone()
cond_norm = cond.norm()
if method == "amplify":
mask = torch.abs(cond) >= 1
cond_copy = cond.clone()
cond = torch.pow(torch.abs(cond), ( 1 / exp_value)) * cond.sign()
cond[mask] = torch.pow(torch.abs(cond_copy[mask]), exp_value) * cond[mask].sign()
elif method == "root":
cond = torch.pow(torch.abs(cond), ( 1 / exp_value)) * cond.sign()
elif method == "power":
cond = torch.pow(torch.abs(cond), exp_value) * cond.sign()
elif method == "erf":
cond = torch.erf(cond)
elif method == "exp_erf":
cond = torch.pow(torch.erf(cond), exp_value)
elif method == "root_erf":
cond = torch.erf(cond)
cond = torch.pow(torch.abs(cond), 1 / exp_value ) * cond.sign()
elif method == "erf_amplify":
cond = torch.erf(cond)
mask = torch.abs(cond) >= 1
cond_copy = cond.clone()
cond = torch.pow(torch.abs(cond), 1 / exp_value ) * cond.sign()
cond[mask] = torch.pow(torch.abs(cond_copy[mask]), exp_value) * cond[mask].sign()
elif method == "sine":
cond = torch.sin(torch.abs(cond)) * cond.sign()
elif method == "sine_exp":
cond = torch.sin(torch.abs(cond)) * cond.sign()
cond = torch.pow(torch.abs(cond), exp_value) * cond.sign()
elif method == "sine_exp_diff":
cond = torch.sin(torch.abs(cond)) * cond.sign()
cond = torch.pow(torch.abs(cond_input), exp_value) * cond.sign() - cond
elif method == "sine_exp_diff_to_sine":
cond = torch.sin(torch.abs(cond)) * cond.sign()
cond = torch.pow(torch.abs(cond), exp_value) * cond.sign() - cond
elif method == "sine_root":
cond = torch.sin(torch.abs(cond)) * cond.sign()
cond = torch.pow(torch.abs(cond), ( 1 / exp_value)) * cond.sign()
elif method == "sine_root_diff":
cond = torch.sin(torch.abs(cond)) * cond.sign()
cond = torch.pow(torch.abs(cond_input), 1 / exp_value) * cond.sign() - cond
elif method == "sine_root_diff_to_sine":
cond = torch.sin(torch.abs(cond)) * cond.sign()
cond = torch.pow(torch.abs(cond), 1 / exp_value) * cond.sign() - cond
elif method == "theDaRkNeSs":
cond = torch.sin(cond)
cond = torch.pow(torch.abs(cond), 1 / exp_value) * cond.sign() - cond
elif method == "cosine":
cond = torch.cos(torch.abs(cond)) * cond.sign()
elif method == "sign":
cond = cond.sign()
elif method == "zero":
cond = torch.zeros_like(cond)
elif method in ["attention_modifiers_input_using_cond","attention_modifiers_input_using_uncond","subtract_attention_modifiers_input_using_cond","subtract_attention_modifiers_input_using_uncond"]:
cond_to_use = args["cond_pos"] if method in ["attention_modifiers_input_using_cond","subtract_attention_modifiers_input_using_cond"] else args["cond_neg"]
tmp_model_options = deepcopy(model_options_copy)
for atm in attention_modifiers_input:
if sigma <= atm['sigma_start'] and sigma > atm['sigma_end']:
block_layers = {"input": atm['unet_block_id_input'], "middle": atm['unet_block_id_middle'], "output": atm['unet_block_id_output']}
for unet_block in block_layers:
for unet_block_id in block_layers[unet_block].split(","):
if unet_block_id != "":
unet_block_id = int(unet_block_id)
tmp_model_options = set_model_options_patch_replace(tmp_model_options, attention_modifier(atm['self_attn_mod_eval'], [args["cond_pos"][0]["cross_attn"], args["cond_neg"][0]["cross_attn"]]if "cond" in atm['self_attn_mod_eval'] else None).modified_attention, atm['unet_attn'], unet_block, unet_block_id)
cond = comfy.samplers.calc_cond_batch(args["model"], [cond_to_use], args["input"], args["timestep"], tmp_model_options)[0]
if method in ["subtract_attention_modifiers_input_using_cond","subtract_attention_modifiers_input_using_uncond"]:
cond = cond_input + (cond_input - cond) * exp_value
elif method == "previous_average":
if sigma > (sigmax - 1):
cond = torch.zeros_like(cond)
else:
cond = (pcp / psi * sigma + cond) / 2
elif method == "eval":
if "condmix" in eval_string:
def condmix(args, mult=2):
cond_pos_tmp = deepcopy(args["cond_pos"])
cond_pos_tmp[0]["cross_attn"] += (args["cond_pos"][0]["cross_attn"] - args["cond_neg"][0]["cross_attn"]*-1) * mult
return cond_pos_tmp
v = []
evals_strings = eval_string.split(";")
if len(evals_strings) > 1:
for i in range(len(evals_strings[:-1])):
v.append(eval(evals_strings[i]))
cond = eval(evals_strings[-1])
if exp_normalize and torch.all(cond != 0):
cond = cond * cond_norm / cond.norm()
# print(get_entropy(cond))
return cond.to(device=default_device)
class advancedDynamicCFG:
def __init__(self):
self.last_cfg_ht_one = 8
self.previous_cond_pred = None
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"automatic_cfg" : (["None", "soft", "hard", "hard_squared", "range"], {"default": "hard"},),
"skip_uncond" : ("BOOLEAN", {"default": True}),
"fake_uncond_start" : ("BOOLEAN", {"default": False}),
"uncond_sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"uncond_sigma_end": ("FLOAT", {"default": 1, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"lerp_uncond" : ("BOOLEAN", {"default": False}),
"lerp_uncond_strength": ("FLOAT", {"default": 2, "min": 0.0, "max": 10.0, "step": 0.1, "round": 0.1}),
"lerp_uncond_sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"lerp_uncond_sigma_end": ("FLOAT", {"default": 1, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"subtract_latent_mean" : ("BOOLEAN", {"default": False}),
"subtract_latent_mean_sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"subtract_latent_mean_sigma_end": ("FLOAT", {"default": 1, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"latent_intensity_rescale" : ("BOOLEAN", {"default": False}),
"latent_intensity_rescale_method" : (["soft","hard","range"], {"default": "hard"},),
"latent_intensity_rescale_cfg": ("FLOAT", {"default": 8, "min": 0.0, "max": 100.0, "step": 0.1, "round": 0.1}),
"latent_intensity_rescale_sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"latent_intensity_rescale_sigma_end": ("FLOAT", {"default": 3, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"cond_exp": ("BOOLEAN", {"default": False}),
"cond_exp_normalize": ("BOOLEAN", {"default": False}),
"cond_exp_sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"cond_exp_sigma_end": ("FLOAT", {"default": 1, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"cond_exp_method": (["amplify", "root", "power", "erf", "erf_amplify", "exp_erf", "root_erf", "sine", "sine_exp", "sine_exp_diff", "sine_exp_diff_to_sine", "sine_root", "sine_root_diff", "sine_root_diff_to_sine", "theDaRkNeSs", "cosine", "sign", "zero", "previous_average", "eval",
"attention_modifiers_input_using_cond","attention_modifiers_input_using_uncond",
"subtract_attention_modifiers_input_using_cond","subtract_attention_modifiers_input_using_uncond"],),
"cond_exp_value": ("FLOAT", {"default": 2, "min": 0, "max": 100, "step": 0.1, "round": 0.01}),
"uncond_exp": ("BOOLEAN", {"default": False}),
"uncond_exp_normalize": ("BOOLEAN", {"default": False}),
"uncond_exp_sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"uncond_exp_sigma_end": ("FLOAT", {"default": 1, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"uncond_exp_method": (["amplify", "root", "power", "erf", "erf_amplify", "exp_erf", "root_erf", "sine", "sine_exp", "sine_exp_diff", "sine_exp_diff_to_sine", "sine_root", "sine_root_diff", "sine_root_diff_to_sine", "theDaRkNeSs", "cosine", "sign", "zero", "previous_average", "eval",
"subtract_attention_modifiers_input_using_cond","subtract_attention_modifiers_input_using_uncond"],),
"uncond_exp_value": ("FLOAT", {"default": 2, "min": 0, "max": 100, "step": 0.1, "round": 0.01}),
"fake_uncond_exp": ("BOOLEAN", {"default": False}),
"fake_uncond_exp_normalize": ("BOOLEAN", {"default": False}),
"fake_uncond_exp_method" : (["cond_pred", "previous_average",
"amplify", "root", "power", "erf", "erf_amplify", "exp_erf", "root_erf", "sine", "sine_exp", "sine_exp_diff", "sine_exp_diff_to_sine", "sine_root", "sine_root_diff",
"sine_root_diff_to_sine", "theDaRkNeSs", "cosine", "sign", "zero", "eval",
"subtract_attention_modifiers_input_using_cond","subtract_attention_modifiers_input_using_uncond",
"attention_modifiers_input_using_cond","attention_modifiers_input_using_uncond"],),
"fake_uncond_exp_value": ("FLOAT", {"default": 2, "min": 0, "max": 1000, "step": 0.1, "round": 0.01}),
"fake_uncond_multiplier": ("INT", {"default": 1, "min": -1, "max": 1, "step": 1}),
"fake_uncond_sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"fake_uncond_sigma_end": ("FLOAT", {"default": 1, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"auto_cfg_topk": ("FLOAT", {"default": 0.25, "min": 0.0, "max": 0.5, "step": 0.05, "round": 0.01}),
"auto_cfg_ref": ("FLOAT", {"default": 8, "min": 0.0, "max": 100, "step": 0.5, "round": 0.01}),
"attention_modifiers_global_enabled": ("BOOLEAN", {"default": False}),
"disable_cond": ("BOOLEAN", {"default": False}),
"disable_cond_sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"disable_cond_sigma_end": ("FLOAT", {"default": 0, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"save_as_preset": ("BOOLEAN", {"default": False}),
"preset_name": ("STRING", {"multiline": False}),
},
"optional":{
"eval_string_cond": ("STRING", {"multiline": True}),
"eval_string_uncond": ("STRING", {"multiline": True}),
"eval_string_fake": ("STRING", {"multiline": True}),
"args_filter": ("STRING", {"multiline": True, "forceInput": True}),
"attention_modifiers_positive": ("ATTNMOD", {"forceInput": True}),
"attention_modifiers_negative": ("ATTNMOD", {"forceInput": True}),
"attention_modifiers_fake_negative": ("ATTNMOD", {"forceInput": True}),
"attention_modifiers_global": ("ATTNMOD", {"forceInput": True}),
}
}
RETURN_TYPES = ("MODEL","STRING",)
FUNCTION = "patch"
CATEGORY = "model_patches/Automatic_CFG"
def patch(self, model, automatic_cfg = "None",
skip_uncond = False, fake_uncond_start = False, uncond_sigma_start = 1000, uncond_sigma_end = 0,
lerp_uncond = False, lerp_uncond_strength = 1, lerp_uncond_sigma_start = 1000, lerp_uncond_sigma_end = 1,
subtract_latent_mean = False, subtract_latent_mean_sigma_start = 1000, subtract_latent_mean_sigma_end = 1,
latent_intensity_rescale = False, latent_intensity_rescale_sigma_start = 1000, latent_intensity_rescale_sigma_end = 1,
cond_exp = False, cond_exp_sigma_start = 1000, cond_exp_sigma_end = 1000, cond_exp_method = "amplify", cond_exp_value = 2, cond_exp_normalize = False,
uncond_exp = False, uncond_exp_sigma_start = 1000, uncond_exp_sigma_end = 1000, uncond_exp_method = "amplify", uncond_exp_value = 2, uncond_exp_normalize = False,
fake_uncond_exp = False, fake_uncond_exp_method = "amplify", fake_uncond_exp_value = 2, fake_uncond_exp_normalize = False, fake_uncond_multiplier = 1, fake_uncond_sigma_start = 1000, fake_uncond_sigma_end = 1,
latent_intensity_rescale_cfg = 8, latent_intensity_rescale_method = "hard",
ignore_pre_cfg_func = False, args_filter = "", auto_cfg_topk = 0.25, auto_cfg_ref = 8,
eval_string_cond = "", eval_string_uncond = "", eval_string_fake = "",
attention_modifiers_global_enabled = False,
attention_modifiers_positive = [], attention_modifiers_negative = [], attention_modifiers_fake_negative = [], attention_modifiers_global = [],
disable_cond=False, disable_cond_sigma_start=1000,disable_cond_sigma_end=1000, save_as_preset = False, preset_name = "", **kwargs
):
# support_function()
model_options_copy = deepcopy(model.model_options)
monkey_patching_comfy_sampling_function()
if args_filter != "":
args_filter = args_filter.split(",")
else:
args_filter = [k for k, v in locals().items()]
not_in_filter = ['self','model','args','args_filter','save_as_preset','preset_name','model_options_copy']
if fake_uncond_exp_method != "eval":
not_in_filter.append("eval_string")
if save_as_preset and preset_name != "":
preset_parameters = {key: value for key, value in locals().items() if key not in not_in_filter}
with open(os.path.join(json_preset_path, preset_name+".json"), 'w', encoding='utf-8') as f:
json.dump(preset_parameters, f)
print(f"Preset saved with the name: {Fore.GREEN}{preset_name}{Fore.RESET}")
print(f"{Fore.RED}Don't forget to turn the save toggle OFF to not overwrite!{Fore.RESET}")
args_str = '\n'.join(f'{k}: {v}' for k, v in locals().items() if k not in not_in_filter and k in args_filter)
sigmin, sigmax = get_sigmin_sigmax(model)
lerp_start, lerp_end = lerp_uncond_sigma_start, lerp_uncond_sigma_end
subtract_start, subtract_end = subtract_latent_mean_sigma_start, subtract_latent_mean_sigma_end
rescale_start, rescale_end = latent_intensity_rescale_sigma_start, latent_intensity_rescale_sigma_end
print(f"Model maximum sigma: {sigmax} / Model minimum sigma: {sigmin}")
m = model.clone()
if skip_uncond or disable_cond:
# set model_options sampler_pre_cfg_automatic_cfg_function
m.model_options["sampler_pre_cfg_automatic_cfg_function"] = make_sampler_pre_cfg_automatic_cfg_function(uncond_sigma_end if skip_uncond else 0, uncond_sigma_start if skip_uncond else 100000,\
disable_cond_sigma_start if disable_cond else 100000, disable_cond_sigma_end if disable_cond else 100000)
print(f"Sampling function patched. Uncond enabled from {round(uncond_sigma_start,2)} to {round(uncond_sigma_end,2)}")
elif not ignore_pre_cfg_func:
m.model_options.pop("sampler_pre_cfg_automatic_cfg_function", None)
uncond_sigma_start, uncond_sigma_end = 1000000, 0
top_k = auto_cfg_topk
previous_cond_pred = None
previous_sigma = None
def automatic_cfg_function(args):
nonlocal previous_sigma
cond_scale = args["cond_scale"]
input_x = args["input"]
cond_pred = args["cond_denoised"]
uncond_pred = args["uncond_denoised"]
sigma = args["sigma"][0]
model_options = args["model_options"]
if self.previous_cond_pred is None:
self.previous_cond_pred = cond_pred.clone().detach().to(device=cond_pred.device)
if previous_sigma is None:
previous_sigma = sigma.item()
reference_cfg = auto_cfg_ref if auto_cfg_ref > 0 else cond_scale
def fake_uncond_step():
return fake_uncond_start and skip_uncond and (sigma > uncond_sigma_start or sigma < uncond_sigma_end) and sigma <= fake_uncond_sigma_start and sigma >= fake_uncond_sigma_end
if fake_uncond_step():
uncond_pred = cond_pred.clone().detach().to(device=cond_pred.device) * fake_uncond_multiplier
if cond_exp and sigma <= cond_exp_sigma_start and sigma >= cond_exp_sigma_end:
cond_pred = experimental_functions(cond_pred, cond_exp_method, cond_exp_value, cond_exp_normalize, self.previous_cond_pred, previous_sigma, sigma.item(), sigmax, attention_modifiers_positive, args, model_options_copy, eval_string_cond)
if uncond_exp and sigma <= uncond_exp_sigma_start and sigma >= uncond_exp_sigma_end and not fake_uncond_step():
uncond_pred = experimental_functions(uncond_pred, uncond_exp_method, uncond_exp_value, uncond_exp_normalize, self.previous_cond_pred, previous_sigma, sigma.item(), sigmax, attention_modifiers_negative, args, model_options_copy, eval_string_uncond)
if fake_uncond_step() and fake_uncond_exp:
uncond_pred = experimental_functions(uncond_pred, fake_uncond_exp_method, fake_uncond_exp_value, fake_uncond_exp_normalize, self.previous_cond_pred, previous_sigma, sigma.item(), sigmax, attention_modifiers_fake_negative, args, model_options_copy, eval_string_fake)
self.previous_cond_pred = cond_pred.clone().detach().to(device=cond_pred.device)
if sigma >= sigmax or cond_scale > 1:
self.last_cfg_ht_one = cond_scale
target_intensity = self.last_cfg_ht_one / 10
if ((check_skip(sigma, uncond_sigma_start, uncond_sigma_end) and skip_uncond) and not fake_uncond_step()) or cond_scale == 1:
return input_x - cond_pred
if lerp_uncond and not check_skip(sigma, lerp_start, lerp_end) and lerp_uncond_strength != 1:
uncond_pred_norm = uncond_pred.norm()
uncond_pred = torch.lerp(cond_pred, uncond_pred, lerp_uncond_strength)
uncond_pred = uncond_pred * uncond_pred_norm / uncond_pred.norm()
cond = input_x - cond_pred
uncond = input_x - uncond_pred
if automatic_cfg == "None":
return uncond + cond_scale * (cond - uncond)
denoised_tmp = input_x - (uncond + reference_cfg * (cond - uncond))
for b in range(len(denoised_tmp)):
denoised_ranges = get_denoised_ranges(denoised_tmp[b], automatic_cfg, top_k)
for c in range(len(denoised_tmp[b])):
fixeds_scale = reference_cfg * target_intensity / denoised_ranges[c]
denoised_tmp[b][c] = uncond[b][c] + fixeds_scale * (cond[b][c] - uncond[b][c])
return denoised_tmp
def center_mean_latent_post_cfg(args):
denoised = args["denoised"]
sigma = args["sigma"][0]
if check_skip(sigma, subtract_start, subtract_end):
return denoised
denoised = center_latent_mean_values(denoised, False, 1)
return denoised
def rescale_post_cfg(args):
denoised = args["denoised"]
sigma = args["sigma"][0]
if check_skip(sigma, rescale_start, rescale_end):
return denoised
target_intensity = latent_intensity_rescale_cfg / 10
for b in range(len(denoised)):
denoised_ranges = get_denoised_ranges(denoised[b], latent_intensity_rescale_method)
for c in range(len(denoised[b])):
scale_correction = target_intensity / denoised_ranges[c]
denoised[b][c] = denoised[b][c] * scale_correction
return denoised
tmp_model_options = deepcopy(m.model_options)
if attention_modifiers_global_enabled:
# print(f"{Fore.GREEN}Sigma timings are ignored for global modifiers.{Fore.RESET}")
for atm in attention_modifiers_global:
block_layers = {"input": atm['unet_block_id_input'], "middle": atm['unet_block_id_middle'], "output": atm['unet_block_id_output']}
for unet_block in block_layers:
for unet_block_id in block_layers[unet_block].split(","):
if unet_block_id != "":
unet_block_id = int(unet_block_id)
tmp_model_options = set_model_options_patch_replace(tmp_model_options, attention_modifier(atm['self_attn_mod_eval']).modified_attention, atm['unet_attn'], unet_block, unet_block_id)
m.model_options = tmp_model_options
if not ignore_pre_cfg_func:
m.set_model_sampler_cfg_function(automatic_cfg_function, disable_cfg1_optimization = False)
if subtract_latent_mean:
m.set_model_sampler_post_cfg_function(center_mean_latent_post_cfg)
if latent_intensity_rescale:
m.set_model_sampler_post_cfg_function(rescale_post_cfg)
return (m, args_str, )
class attentionModifierParametersNode:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"sigma_end": ("FLOAT", {"default": 0, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"self_attn_mod_eval": ("STRING", {"multiline": True }, {"default": ""}),
"unet_block_id_input": ("STRING", {"multiline": False}, {"default": ""}),
"unet_block_id_middle": ("STRING", {"multiline": False}, {"default": ""}),
"unet_block_id_output": ("STRING", {"multiline": False}, {"default": ""}),
"unet_attn": (["attn1","attn2","both"],),
},
"optional":{
"join_parameters": ("ATTNMOD", {"forceInput": True}),
}}
RETURN_TYPES = ("ATTNMOD","STRING",)
RETURN_NAMES = ("Attention modifier", "Parameters as string")
FUNCTION = "exec"
CATEGORY = "model_patches/Automatic_CFG/experimental_attention_modifiers"
def exec(self, join_parameters=None, **kwargs):
info_string = "\n".join([f"{k}: {v}" for k,v in kwargs.items() if v != ""])
if kwargs['unet_attn'] == "both":
copy_kwargs = kwargs.copy()
kwargs['unet_attn'] = "attn1"
copy_kwargs['unet_attn'] = "attn2"
out_modifiers = [kwargs, copy_kwargs]
else:
out_modifiers = [kwargs]
return (out_modifiers if join_parameters is None else join_parameters + out_modifiers, info_string, )
class attentionModifierBruteforceParametersNode:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
"sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"sigma_end": ("FLOAT", {"default": 0, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"self_attn_mod_eval": ("STRING", {"multiline": True , "default": ""}),
"unet_block_id_input": ("STRING", {"multiline": False, "default": "4,5,7,8"}),
"unet_block_id_middle": ("STRING", {"multiline": False, "default": "0"}),
"unet_block_id_output": ("STRING", {"multiline": False, "default": "0,1,2,3,4,5"}),
"unet_attn": (["attn1","attn2","both"],),
},
"optional":{
"join_parameters": ("ATTNMOD", {"forceInput": True}),
}}
RETURN_TYPES = ("ATTNMOD","STRING",)
RETURN_NAMES = ("Attention modifier", "Parameters as string")
FUNCTION = "exec"
CATEGORY = "model_patches/Automatic_CFG/experimental_attention_modifiers"
def create_sequence_parameters(self, input_str, middle_str, output_str):
input_values = input_str.split(",") if input_str else []
middle_values = middle_str.split(",") if middle_str else []
output_values = output_str.split(",") if output_str else []
result = []
result.extend([{"unet_block_id_input": val, "unet_block_id_middle": "", "unet_block_id_output": ""} for val in input_values])
result.extend([{"unet_block_id_input": "", "unet_block_id_middle": val, "unet_block_id_output": ""} for val in middle_values])
result.extend([{"unet_block_id_input": "", "unet_block_id_middle": "", "unet_block_id_output": val} for val in output_values])
return result
def exec(self, seed, join_parameters=None, **kwargs):
sequence_parameters = self.create_sequence_parameters(kwargs['unet_block_id_input'],kwargs['unet_block_id_middle'],kwargs['unet_block_id_output'])
lenseq = len(sequence_parameters)
current_index = seed % lenseq
current_sequence = sequence_parameters[current_index]
kwargs["unet_block_id_input"] = current_sequence["unet_block_id_input"]
kwargs["unet_block_id_middle"] = current_sequence["unet_block_id_middle"]
kwargs["unet_block_id_output"] = current_sequence["unet_block_id_output"]
if current_sequence["unet_block_id_input"] != "":
current_block_string = f"unet_block_id_input: {current_sequence['unet_block_id_input']}"
elif current_sequence["unet_block_id_middle"] != "":
current_block_string = f"unet_block_id_middle: {current_sequence['unet_block_id_middle']}"
elif current_sequence["unet_block_id_output"] != "":
current_block_string = f"unet_block_id_output: {current_sequence['unet_block_id_output']}"
info_string = f"Progress: {current_index+1}/{lenseq}\n{kwargs['self_attn_mod_eval']}\n{kwargs['unet_attn']} {current_block_string}"
if kwargs['unet_attn'] == "both":
copy_kwargs = kwargs.copy()
kwargs['unet_attn'] = "attn1"
copy_kwargs['unet_attn'] = "attn2"
out_modifiers = [kwargs, copy_kwargs]
else:
out_modifiers = [kwargs]
return (out_modifiers if join_parameters is None else join_parameters + out_modifiers, info_string, )
class attentionModifierConcatNode:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"parameters_1": ("ATTNMOD", {"forceInput": True}),
"parameters_2": ("ATTNMOD", {"forceInput": True}),
}}
RETURN_TYPES = ("ATTNMOD",)
FUNCTION = "exec"
CATEGORY = "model_patches/Automatic_CFG/experimental_attention_modifiers"
def exec(self, parameters_1, parameters_2):
output_parms = parameters_1 + parameters_2
return (output_parms, )
class simpleDynamicCFG:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"hard_mode" : ("BOOLEAN", {"default": True}),
"boost" : ("BOOLEAN", {"default": True}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Automatic_CFG/presets"
def patch(self, model, hard_mode, boost):
advcfg = advancedDynamicCFG()
m = advcfg.patch(model,
skip_uncond = boost,
uncond_sigma_start = 1000, uncond_sigma_end = 1,
automatic_cfg = "hard" if hard_mode else "soft"
)[0]
return (m, )
class presetLoader:
@classmethod
def INPUT_TYPES(s):
presets_files = [pj.replace(".json","") for pj in os.listdir(json_preset_path) if ".json" in pj and pj not in ["Experimental_temperature.json","do_not_delete.json"]]
presets_files = sorted(presets_files, key=str.lower)
return {"required": {
"model": ("MODEL",),
"preset" : (presets_files, {"default": "Excellent_attention"}),
"uncond_sigma_end": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"use_uncond_sigma_end_from_preset" : ("BOOLEAN", {"default": True}),
"automatic_cfg" : (["From preset","None", "soft", "hard", "hard_squared", "range"],),
},
"optional":{
"join_global_parameters": ("ATTNMOD", {"forceInput": True}),
}}
RETURN_TYPES = ("MODEL", "STRING", "STRING",)
RETURN_NAMES = ("Model", "Preset name", "Parameters as string",)
FUNCTION = "patch"
CATEGORY = "model_patches/Automatic_CFG"
def patch(self, model, preset, uncond_sigma_end, use_uncond_sigma_end_from_preset, automatic_cfg, join_global_parameters=None):
with open(os.path.join(json_preset_path, preset+".json"), 'r', encoding='utf-8') as f:
preset_args = json.load(f)
if not use_uncond_sigma_end_from_preset:
preset_args["uncond_sigma_end"] = uncond_sigma_end
preset_args["fake_uncond_sigma_end"] = uncond_sigma_end
preset_args["fake_uncond_exp_sigma_end"] = uncond_sigma_end
preset_args["uncond_exp_sigma_end"] = uncond_sigma_end
if join_global_parameters is not None:
preset_args["attention_modifiers_global"] = preset_args["attention_modifiers_global"] + join_global_parameters
preset_args["attention_modifiers_global_enabled"] = True
if automatic_cfg != "From preset":
preset_args["automatic_cfg"] = automatic_cfg
advcfg = advancedDynamicCFG()
m = advcfg.patch(model, **preset_args)[0]
info_string = ",\n".join([f"\"{k}\": {v}" for k,v in preset_args.items() if v != ""])
print(f"Preset {Fore.GREEN}{preset}{Fore.RESET} loaded successfully!")
return (m, preset, info_string,)
class simpleDynamicCFGlerpUncond:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"boost" : ("BOOLEAN", {"default": True}),
"negative_strength": ("FLOAT", {"default": 1, "min": 0.0, "max": 5.0, "step": 0.1, "round": 0.1}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Automatic_CFG/presets"
def patch(self, model, boost, negative_strength):
advcfg = advancedDynamicCFG()
m = advcfg.patch(model=model,
automatic_cfg="hard", skip_uncond=boost,
uncond_sigma_start = 15, uncond_sigma_end = 1,
lerp_uncond=negative_strength != 1, lerp_uncond_strength=negative_strength,
lerp_uncond_sigma_start = 15, lerp_uncond_sigma_end = 1
)[0]
return (m, )
class postCFGrescaleOnly:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"subtract_latent_mean" : ("BOOLEAN", {"default": True}),
"subtract_latent_mean_sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.1}),
"subtract_latent_mean_sigma_end": ("FLOAT", {"default": 7.5, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.1}),
"latent_intensity_rescale" : ("BOOLEAN", {"default": True}),
"latent_intensity_rescale_method" : (["soft","hard","range"], {"default": "hard"},),
"latent_intensity_rescale_cfg" : ("FLOAT", {"default": 8, "min": 0.0, "max": 100.0, "step": 0.1, "round": 0.1}),
"latent_intensity_rescale_sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.1}),
"latent_intensity_rescale_sigma_end": ("FLOAT", {"default": 5, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.1}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Automatic_CFG/utils"
def patch(self, model,
subtract_latent_mean, subtract_latent_mean_sigma_start, subtract_latent_mean_sigma_end,
latent_intensity_rescale, latent_intensity_rescale_method, latent_intensity_rescale_cfg, latent_intensity_rescale_sigma_start, latent_intensity_rescale_sigma_end
):
advcfg = advancedDynamicCFG()
m = advcfg.patch(model=model,
subtract_latent_mean = subtract_latent_mean,
subtract_latent_mean_sigma_start = subtract_latent_mean_sigma_start, subtract_latent_mean_sigma_end = subtract_latent_mean_sigma_end,
latent_intensity_rescale = latent_intensity_rescale, latent_intensity_rescale_cfg = latent_intensity_rescale_cfg, latent_intensity_rescale_method = latent_intensity_rescale_method,
latent_intensity_rescale_sigma_start = latent_intensity_rescale_sigma_start, latent_intensity_rescale_sigma_end = latent_intensity_rescale_sigma_end,
ignore_pre_cfg_func = True
)[0]
return (m, )
class simpleDynamicCFGHighSpeed:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Automatic_CFG/presets"
def patch(self, model):
advcfg = advancedDynamicCFG()
m = advcfg.patch(model=model, automatic_cfg = "hard",
skip_uncond = True, uncond_sigma_start = 7.5, uncond_sigma_end = 1)[0]
return (m, )
class simpleDynamicCFGwarpDrive:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"uncond_sigma_start": ("FLOAT", {"default": 5.5, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"uncond_sigma_end": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"fake_uncond_sigma_end": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Automatic_CFG/presets"
def patch(self, model, uncond_sigma_start, uncond_sigma_end, fake_uncond_sigma_end):
advcfg = advancedDynamicCFG()
print(f" {Fore.CYAN}WARP DRIVE MODE ENGAGED!{Style.RESET_ALL}\n Settings suggestions:\n"
f" {Fore.GREEN}1/1/1: {Fore.YELLOW}Maaaxxxiiimum speeeeeed.{Style.RESET_ALL} {Fore.RED}Uncond disabled.{Style.RESET_ALL} {Fore.MAGENTA}Fasten your seatbelt!{Style.RESET_ALL}\n"
f" {Fore.GREEN}3/1/1: {Fore.YELLOW}Risky space-time continuum distortion.{Style.RESET_ALL} {Fore.MAGENTA}Awesome for prompts with a clear subject!{Style.RESET_ALL}\n"
f" {Fore.GREEN}5.5/1/1: {Fore.YELLOW}Frameshift Drive Autopilot: {Fore.GREEN}Engaged.{Style.RESET_ALL} {Fore.MAGENTA}Should work with anything but do it better and faster!{Style.RESET_ALL}")
m = advcfg.patch(model=model, automatic_cfg = "hard",
skip_uncond = True, uncond_sigma_start = uncond_sigma_start, uncond_sigma_end = uncond_sigma_end,
fake_uncond_sigma_end = fake_uncond_sigma_end, fake_uncond_sigma_start = 1000, fake_uncond_start=True,
fake_uncond_exp=True,fake_uncond_exp_normalize=True,fake_uncond_exp_method="previous_average",
cond_exp = False, cond_exp_sigma_start = 9, cond_exp_sigma_end = uncond_sigma_start, cond_exp_method = "erf", cond_exp_normalize = True,
)[0]
return (m, )
class simpleDynamicCFGunpatch:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "unpatch"
CATEGORY = "model_patches/Automatic_CFG/utils"
def unpatch(self, model):
m = model.clone()
m.model_options.pop("sampler_pre_cfg_automatic_cfg_function", None)
return (m, )
class simpleDynamicCFGExcellentattentionPatch:
@classmethod
def INPUT_TYPES(s):
inputs = {"required": {
"model": ("MODEL",),
"Auto_CFG": ("BOOLEAN", {"default": True}),
"patch_multiplier": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 100.0, "step": 1.0, "round": 0.01}),
"patch_cond": ("BOOLEAN", {"default": True}),
"patch_uncond": ("BOOLEAN", {"default": True}),
"light_patch": ("BOOLEAN", {"default": False}),
"mute_self_input_layer_8_cond": ("BOOLEAN", {"default": False}),
"mute_cross_input_layer_8_cond": ("BOOLEAN", {"default": False}),
"mute_self_input_layer_8_uncond": ("BOOLEAN", {"default": True}),
"mute_cross_input_layer_8_uncond": ("BOOLEAN", {"default": False}),
"uncond_sigma_end": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"bypass_layer_8_instead_of_mute": ("BOOLEAN", {"default": False}),
"save_as_preset": ("BOOLEAN", {"default": False}),
"preset_name": ("STRING", {"multiline": False}),
},
"optional":{
"attn_mod_for_positive_operation": ("ATTNMOD", {"forceInput": True}),
"attn_mod_for_negative_operation": ("ATTNMOD", {"forceInput": True}),
},
}
if "dev_env.txt" in os.listdir(current_dir):
inputs['optional'].update({"attn_mod_for_global_operation": ("ATTNMOD", {"forceInput": True})})
return inputs
RETURN_TYPES = ("MODEL","STRING",)
RETURN_NAMES = ("Model", "Parameters as string",)
FUNCTION = "patch"
CATEGORY = "model_patches/Automatic_CFG"
def patch(self, model, Auto_CFG, patch_multiplier, patch_cond, patch_uncond, light_patch,
mute_self_input_layer_8_cond, mute_cross_input_layer_8_cond,
mute_self_input_layer_8_uncond, mute_cross_input_layer_8_uncond,
uncond_sigma_end,bypass_layer_8_instead_of_mute, save_as_preset, preset_name,
attn_mod_for_positive_operation = None, attn_mod_for_negative_operation = None, attn_mod_for_global_operation = None):
parameters_as_string = "Excellent attention:\n" + "\n".join([f"{k}: {v}" for k, v in locals().items() if k not in ["self", "model"]])
with open(os.path.join(json_preset_path, "Excellent_attention.json"), 'r', encoding='utf-8') as f:
patch_parameters = json.load(f)
attn_patch = {"sigma_start": 1000, "sigma_end": 0,
"self_attn_mod_eval": f"normalize_tensor(q+(q-attention_basic(attnbc, k, v, extra_options['n_heads'])))*attnbc.norm()*{patch_multiplier}",
"unet_block_id_input": "", "unet_block_id_middle": "0", "unet_block_id_output": "", "unet_attn": "attn2"}
attn_patch_light = {"sigma_start": 1000, "sigma_end": 0,
"self_attn_mod_eval": f"q*{patch_multiplier}",
"unet_block_id_input": "", "unet_block_id_middle": "0", "unet_block_id_output": "", "unet_attn": "attn2"}
kill_self_input_8 = {
"sigma_start": 1000,
"sigma_end": 0,
"self_attn_mod_eval": "q" if bypass_layer_8_instead_of_mute else "torch.zeros_like(q)",
"unet_block_id_input": "8",
"unet_block_id_middle": "",
"unet_block_id_output": "",
"unet_attn": "attn1"}
kill_cross_input_8 = kill_self_input_8.copy()
kill_cross_input_8['unet_attn'] = "attn2"
attention_modifiers_positive = []
attention_modifiers_fake_negative = []
if patch_cond: attention_modifiers_positive.append(attn_patch) if not light_patch else attention_modifiers_positive.append(attn_patch_light)
if mute_self_input_layer_8_cond: attention_modifiers_positive.append(kill_self_input_8)
if mute_cross_input_layer_8_cond: attention_modifiers_positive.append(kill_cross_input_8)
if patch_uncond: attention_modifiers_fake_negative.append(attn_patch) if not light_patch else attention_modifiers_fake_negative.append(attn_patch_light)
if mute_self_input_layer_8_uncond: attention_modifiers_fake_negative.append(kill_self_input_8)
if mute_cross_input_layer_8_uncond: attention_modifiers_fake_negative.append(kill_cross_input_8)
patch_parameters['attention_modifiers_positive'] = attention_modifiers_positive
patch_parameters['attention_modifiers_fake_negative'] = attention_modifiers_fake_negative
if attn_mod_for_positive_operation is not None:
patch_parameters['attention_modifiers_positive'] = patch_parameters['attention_modifiers_positive'] + attn_mod_for_positive_operation
if attn_mod_for_negative_operation is not None:
patch_parameters['attention_modifiers_fake_negative'] = patch_parameters['attention_modifiers_fake_negative'] + attn_mod_for_negative_operation
if attn_mod_for_global_operation is not None:
patch_parameters["attention_modifiers_global_enabled"] = True
patch_parameters['attention_modifiers_global'] = attn_mod_for_global_operation
patch_parameters["uncond_sigma_end"] = uncond_sigma_end
patch_parameters["fake_uncond_sigma_end"] = uncond_sigma_end
patch_parameters["automatic_cfg"] = "hard" if Auto_CFG else "None"
if save_as_preset:
patch_parameters["save_as_preset"] = save_as_preset
patch_parameters["preset_name"] = preset_name
advcfg = advancedDynamicCFG()
m = advcfg.patch(model, **patch_parameters)[0]
return (m, parameters_as_string, )
class simpleDynamicCFGCustomAttentionPatch:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"Auto_CFG": ("BOOLEAN", {"default": True}),
"cond_mode" : (["replace_by_custom","normal+(normal-custom_cond)*multiplier","normal+(normal-custom_uncond)*multiplier"],),
"uncond_mode" : (["replace_by_custom","normal+(normal-custom_cond)*multiplier","normal+(normal-custom_uncond)*multiplier"],),
"cond_diff_multiplier": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.1, "round": 0.01}),
"uncond_diff_multiplier": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.1, "round": 0.01}),
"uncond_sigma_end": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10000, "step": 0.1, "round": 0.01}),
"save_as_preset": ("BOOLEAN", {"default": False}),
"preset_name": ("STRING", {"multiline": False}),
},
"optional":{
"attn_mod_for_positive_operation": ("ATTNMOD", {"forceInput": True}),
"attn_mod_for_negative_operation": ("ATTNMOD", {"forceInput": True}),
}}
RETURN_TYPES = ("MODEL",)
RETURN_NAMES = ("Model",)
FUNCTION = "patch"
CATEGORY = "model_patches/Automatic_CFG/experimental_attention_modifiers"
def patch(self, model, Auto_CFG, cond_mode, uncond_mode, cond_diff_multiplier, uncond_diff_multiplier, uncond_sigma_end, save_as_preset, preset_name,
attn_mod_for_positive_operation = [], attn_mod_for_negative_operation = []):
with open(os.path.join(json_preset_path, "do_not_delete.json"), 'r', encoding='utf-8') as f:
patch_parameters = json.load(f)
patch_parameters["cond_exp_value"] = cond_diff_multiplier
patch_parameters["uncond_exp_value"] = uncond_diff_multiplier
if cond_mode != "replace_by_custom":
patch_parameters["disable_cond"] = False
if cond_mode == "normal+(normal-custom_cond)*multiplier":
patch_parameters["cond_exp_method"] = "subtract_attention_modifiers_input_using_cond"
elif cond_mode == "normal+(normal-custom_uncond)*multiplier":
patch_parameters["cond_exp_method"] = "subtract_attention_modifiers_input_using_uncond"
if uncond_mode != "replace_by_custom":
patch_parameters["uncond_sigma_start"] = 1000.0
patch_parameters["fake_uncond_exp"] = False
patch_parameters["uncond_exp"] = True
if uncond_mode == "normal+(normal-custom_cond)*multiplier":
patch_parameters["uncond_exp_method"] = "subtract_attention_modifiers_input_using_cond"
elif uncond_mode == "normal+(normal-custom_uncond)*multiplier":
patch_parameters["uncond_exp_method"] = "subtract_attention_modifiers_input_using_uncond"
if cond_mode != "replace_by_custom" and attn_mod_for_positive_operation != []:
smallest_sigma = min([float(x['sigma_end']) for x in attn_mod_for_positive_operation])
patch_parameters["disable_cond_sigma_end"] = smallest_sigma
patch_parameters["cond_exp_sigma_end"] = smallest_sigma
if uncond_mode != "replace_by_custom" and attn_mod_for_negative_operation != []:
smallest_sigma = min([float(x['sigma_end']) for x in attn_mod_for_negative_operation])
patch_parameters["uncond_exp_sigma_end"] = smallest_sigma
patch_parameters["fake_uncond_start"] = False
# else:
# biggest_sigma = max([float(x['sigma_start']) for x in attn_mod_for_negative_operation])
# patch_parameters["fake_uncond_sigma_start"] = biggest_sigma
patch_parameters["automatic_cfg"] = "hard" if Auto_CFG else "None"
patch_parameters['attention_modifiers_positive'] = attn_mod_for_positive_operation
patch_parameters['attention_modifiers_negative'] = attn_mod_for_negative_operation
patch_parameters['attention_modifiers_fake_negative'] = attn_mod_for_negative_operation
patch_parameters["uncond_sigma_end"] = uncond_sigma_end
patch_parameters["fake_uncond_sigma_end"] = uncond_sigma_end
patch_parameters["save_as_preset"] = save_as_preset
patch_parameters["preset_name"] = preset_name
advcfg = advancedDynamicCFG()
m = advcfg.patch(model, **patch_parameters)[0]
return (m, )
class attentionModifierSingleLayerBypassNode:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"sigma_end": ("FLOAT", {"default": 0, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"block_name": (["input","middle","output"],),
"block_number": ("INT", {"default": 0, "min": 0, "max": 12, "step": 1}),
"unet_attn": (["attn1","attn2","both"],),
},
"optional":{
"join_parameters": ("ATTNMOD", {"forceInput": True}),
}}
RETURN_TYPES = ("ATTNMOD","STRING",)
RETURN_NAMES = ("Attention modifier", "Parameters as string")
FUNCTION = "exec"
CATEGORY = "model_patches/Automatic_CFG/experimental_attention_modifiers"
def exec(self, sigma_start, sigma_end, block_name, block_number, unet_attn, join_parameters=None):
attn_modifier_dict = {
"sigma_start": sigma_start, "sigma_end": sigma_end,
"self_attn_mod_eval": "q",
"unet_block_id_input": str(block_number) if block_name == "input" else "",
"unet_block_id_middle": str(block_number) if block_name == "middle" else "",
"unet_block_id_output": str(block_number) if block_name == "output" else "",
"unet_attn": f"{unet_attn}"
}
info_string = "\n".join([f"{k}: {v}" for k,v in attn_modifier_dict.items() if v != ""])
if unet_attn == "both":
attn_modifier_dict['unet_attn'] = "attn1"
copy_attn_modifier_dict = attn_modifier_dict.copy()
copy_attn_modifier_dict['unet_attn'] = "attn2"
out_modifiers = [attn_modifier_dict, copy_attn_modifier_dict]
else:
out_modifiers = [attn_modifier_dict]
return (out_modifiers if join_parameters is None else join_parameters + out_modifiers, info_string, )
class attentionModifierSingleLayerTemperatureNode:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"sigma_start": ("FLOAT", {"default": 1000, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"sigma_end": ("FLOAT", {"default": 0, "min": 0.0, "max": 10000.0, "step": 0.1, "round": 0.01}),
"block_name": (["input","middle","output"],),
"block_number": ("INT", {"default": 0, "min": 0, "max": 12, "step": 1}),
"unet_attn": (["attn1","attn2","both"],),
"temperature": ("FLOAT", {"default": 1, "min": 0.0, "max": 10000.0, "step": 0.01, "round": 0.01}),
},
"optional":{
"join_parameters": ("ATTNMOD", {"forceInput": True}),
}}
RETURN_TYPES = ("ATTNMOD","STRING",)
RETURN_NAMES = ("Attention modifier", "Parameters as string")
FUNCTION = "exec"
CATEGORY = "model_patches/Automatic_CFG/experimental_attention_modifiers"
def exec(self, sigma_start, sigma_end, block_name, block_number, unet_attn, temperature, join_parameters=None):
attn_modifier_dict = {
"sigma_start": sigma_start, "sigma_end": sigma_end,
"self_attn_mod_eval": f"temperature_patcher({temperature}).attention_basic_with_temperature(q, k, v, extra_options)",
"unet_block_id_input": str(block_number) if block_name == "input" else "",
"unet_block_id_middle": str(block_number) if block_name == "middle" else "",
"unet_block_id_output": str(block_number) if block_name == "output" else "",
"unet_attn": f"{unet_attn}"
}
info_string = "\n".join([f"{k}: {v}" for k,v in attn_modifier_dict.items() if v != ""])
if unet_attn == "both":
attn_modifier_dict['unet_attn'] = "attn1"
copy_attn_modifier_dict = attn_modifier_dict.copy()
copy_attn_modifier_dict['unet_attn'] = "attn2"
out_modifiers = [attn_modifier_dict, copy_attn_modifier_dict]
else:
out_modifiers = [attn_modifier_dict]
return (out_modifiers if join_parameters is None else join_parameters + out_modifiers, info_string, )
class uncondZeroNode:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"scale": ("FLOAT", {"default": 1.2, "min": 0.0, "max": 10.0, "step": 0.01, "round": 0.01}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Automatic_CFG"
def patch(self, model, scale):
def custom_patch(args):
cond_pred = args["cond_denoised"]
input_x = args["input"]
if args["sigma"][0] <= 1:
return input_x - cond_pred
cond = input_x - cond_pred
uncond = input_x - torch.zeros_like(cond)
return uncond + scale * (cond - uncond)
m = model.clone()
m.set_model_sampler_cfg_function(custom_patch)
return (m, )