nodes/Extraltodeus-pre_cfg_comfy_nodes_for_ComfyUI
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
1e0c3209492f56b17fabf50855e35d2af9f3103a
Extraltodeus-pre_cfg_comfy_…/nodes.py
T
Extraltodeus 64314b1726 Add files via upload
2024-09-03 13:22:59 +02:00

2138 lines
100 KiB
Python
Raw Blame History

import torch
import torch.nn.functional as F
from math import ceil, floor
from copy import deepcopy
import comfy.model_patcher
from comfy.sampler_helpers import convert_cond
from comfy.samplers import calc_cond_batch, encode_model_conds
from comfy.ldm.modules.attention import optimized_attention_for_device
from nodes import ConditioningConcat, ConditioningSetTimestepRange
import comfy.model_management as model_management
from comfy.latent_formats import SDXL as SDXL_Latent
import os
from comfy.taesd import taesd as taesd_class
from comfy.sample import prepare_noise
from .imported_functions import skimmed_CFG_patch_wrap
import numpy as np
import random
taesd = taesd_class.TAESD()
current_dir = os.path.dirname(os.path.realpath(__file__))
SDXL_Latent = SDXL_Latent()
sdxl_latent_rgb_factors = SDXL_Latent.latent_rgb_factors
ConditioningConcat = ConditioningConcat()
ConditioningSetTimestepRange = ConditioningSetTimestepRange()
default_attention = optimized_attention_for_device(model_management.get_torch_device())
default_device = model_management.get_torch_device()
def get_sigma_min_max(model):
model_sampling = model.model.model_sampling
sigma_min = model_sampling.sigma(model_sampling.timestep(model_sampling.sigma_min)).item()
sigma_max = model_sampling.sigma(model_sampling.timestep(model_sampling.sigma_max)).item()
return sigma_min, sigma_max
def selfnorm(x):
return x / x.norm()
def weighted_average(tensor1, tensor2, weight1):
return (weight1 * tensor1 + (1 - weight1) * tensor2)
def minmaxnorm(x):
return torch.nan_to_num((x - x.min()) / (x.max() - x.min()), nan=0.0, posinf=1.0, neginf=0.0)
def normlike(x,y):
return x / x.norm() * y.norm()
def make_new_uncond_at_scale(cond,uncond,cond_scale,new_scale):
new_scale_ratio = (new_scale - 1) / (cond_scale - 1)
return cond * (1 - new_scale_ratio) + uncond * new_scale_ratio
def make_new_uncond_at_scale_co(conds_out,cond_scale,new_scale):
new_scale_ratio = (new_scale - 1) / (cond_scale - 1)
return conds_out[0] * (1 - new_scale_ratio) + conds_out[1] * new_scale_ratio
def get_denoised_at_scale(x_orig,cond,uncond,cond_scale):
return x_orig - ((x_orig - uncond) + cond_scale * ((x_orig - cond) - (x_orig - uncond)))
@torch.no_grad()
def gaussian_kernel(size: int, sigma: float, device=default_device):
coords = torch.arange(size, dtype=torch.float32, device=device)
coords -= (size - 1) / 2.0
g = torch.exp(-(coords ** 2) / (2 * sigma ** 2))
g /= g.sum()
kernel = g[:, None] * g[None, :]
return kernel
def blur_tensor(tensor, kernel_size=9, sigma=1.0):
device = tensor.device
kernel = gaussian_kernel(kernel_size, sigma, device=device).unsqueeze(0).unsqueeze(0)
padding = kernel_size // 2
tensor = tensor.unsqueeze(0).unsqueeze(0)
blurred_tensor = F.conv2d(tensor, kernel, padding=padding)
return blurred_tensor.squeeze()
def roll_channel(tensor, channel_index, shift, dim):
tensor[:, channel_index, :, :] = torch.roll(tensor[:, channel_index, :, :], shifts=shift, dims=dim)
return tensor
def mirror_from_middle(tensor, vertical):
dim = 2 if vertical else 3
middle_index = tensor.size(dim) // 2
left_part = tensor.index_select(dim, torch.arange(middle_index - 1, -1, -1, device=tensor.device))
right_part = tensor.index_select(dim, torch.arange(middle_index, tensor.size(dim), 1, device=tensor.device))
mirrored_tensor = torch.cat([right_part, left_part], dim=dim).to(device=tensor.device)
return mirrored_tensor
def mirror_flip(tensor, vertical):
dim = 2 if vertical else 3
return torch.flip(tensor, dims=[dim])
class pre_cfg_perp_neg:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"clip": ("CLIP",),
"neg_scale": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 1/10, "round": 0.01}),
"set_context_length" : ("BOOLEAN", {"default": False,"tooltip":"For static tensor rt engines with a set context length."}),
"context_length": ("INT", {"default": 1, "min": 1, "max": 100, "step": 1}),
"start_at_sigma": ("FLOAT", {"default": 15, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
# "cond_or_uncond": (["both","uncond"], {"default":"uncond"}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, clip, neg_scale, set_context_length, context_length, start_at_sigma, end_at_sigma, cond_or_uncond="uncond"):
empty_cond, pooled = clip.encode_from_tokens(clip.tokenize(""), return_pooled=True)
nocond = [[empty_cond, {"pooled_output": pooled}]]
if context_length > 1 and set_context_length:
short_nocond = deepcopy(nocond)
for x in range(context_length - 1):
(nocond,) = ConditioningConcat.concat(nocond, short_nocond)
nocond = convert_cond(nocond)
@torch.no_grad()
def pre_cfg_perp_neg_function(args):
conds_out = args["conds_out"]
noise_pred_pos = conds_out[0]
if args["sigma"][0] > start_at_sigma or args["sigma"][0] <= end_at_sigma or not torch.any(conds_out[1]):
return conds_out
noise_pred_neg = conds_out[1]
model_options = args["model_options"]
timestep = args["timestep"]
model = args["model"]
x = args["input"]
nocond_processed = encode_model_conds(model.extra_conds, nocond, x, x.device, "negative")
(noise_pred_nocond,) = calc_cond_batch(model, [nocond_processed], x, timestep, model_options)
pos = noise_pred_pos - noise_pred_nocond
neg = noise_pred_neg - noise_pred_nocond
perp = neg - ((torch.mul(neg, pos).sum())/(torch.norm(pos)**2)) * pos
perp_neg = perp * neg_scale
if cond_or_uncond == "both":
perp_p = pos - ((torch.mul(neg, pos).sum())/(torch.norm(neg)**2)) * neg
perp_pos = perp_p * neg_scale
conds_out[0] = noise_pred_nocond + perp_pos
else:
conds_out[0] = noise_pred_nocond + pos
conds_out[1] = noise_pred_nocond + perp_neg
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_perp_neg_function)
return (m, )
class pre_cfg_re_negative:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"clip": ("CLIP",),
"empty_proportion": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 1/20, "round": 0.01,"tooltip":"How much of the empty prediction will be mixed with the negative.",}),
"progressive_scale" : ("BOOLEAN", {"default": False, "tooltip":"If turned on:\nThe proportion of empty prediction will vary along the sampling relatively to the sigma.\nThe proportion slider will set the starting value of the empty prediction.\nThe end value will be 1 - the proportion."}),
"set_context_length" : ("BOOLEAN", {"default": False,"tooltip":"For static tensor rt engines with a set context length."}),
"context_length": ("INT", {"default": 1, "min": 1, "max": 100, "step": 1}),
"end_at_sigma": ("FLOAT", {"default": 5.42, "min": 0.0, "max": 10000.0, "step": 1/100, "round": 1/100}),
},
"optional": {
"optional_text": ("STRING", {"forceInput": True,"tooltip":"If used, instead of an empty prediction, it will use this."}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, clip, empty_proportion, progressive_scale, set_context_length, context_length, end_at_sigma, optional_text=""):
sigma_min, sigma_max = get_sigma_min_max(model)
empty_cond, pooled = clip.encode_from_tokens(clip.tokenize(optional_text), return_pooled=True)
nocond = [[empty_cond, {"pooled_output": pooled}]]
if context_length > 1 and set_context_length:
short_nocond = deepcopy(nocond)
for x in range(context_length - 1):
(nocond,) = ConditioningConcat.concat(nocond, short_nocond)
nocond[0][0] = nocond[0][0][:,:77*context_length,...]
nocond = convert_cond(nocond)
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
sigma = args["sigma"][0]
# cond_scale = args["cond_scale"]
if sigma <= end_at_sigma or not torch.any(conds_out[1]):
return conds_out
model_options = args["model_options"]
timestep = args["timestep"]
model = args["model"]
x_orig = args["input"]
nocond_processed = encode_model_conds(model.extra_conds, nocond, x_orig, x_orig.device, "negative")
(noise_pred_nocond,) = calc_cond_batch(model, [nocond_processed], x_orig, timestep, model_options)
if progressive_scale:
progression = (sigma - sigma_min) / (sigma_max - sigma_min)
current_scale = progression * empty_proportion + (1 - progression) * (1 - empty_proportion)
current_scale = torch.clamp(current_scale, min=0, max=1)
conds_out[1] = current_scale * noise_pred_nocond + conds_out[1] * (1 - current_scale)
else:
conds_out[1] = empty_proportion * noise_pred_nocond + conds_out[1] * (1 - empty_proportion)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
@torch.no_grad()
def normalize_adjust(a,b,strength=1):
norm_a = torch.linalg.norm(a)
a = selfnorm(a)
b = selfnorm(b)
res = b - a * (a * b).sum()
if res.isnan().any():
res = torch.nan_to_num(res, nan=0.0)
a = a - res * strength
return a * norm_a
class condDiffSharpeningNode:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"do_on": (["both","cond","uncond"], {"default": "cond"},),
"scale": ("FLOAT", {"default": 0.75, "min": -10.0, "max": 10.0, "step": 1/20, "round": 1/100}),
"normalized": ("BOOLEAN", {"default": False}),
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 1000000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 01.0, "min": 0.0, "max": 1000000.0, "step": 1/100, "round": 1/100}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, do_on, scale, normalized, start_at_sigma, end_at_sigma):
model_sampling = model.model.model_sampling
sigma_max = model_sampling.sigma(model_sampling.timestep(model_sampling.sigma_max)).item()
prev_cond = None
prev_uncond = None
@torch.no_grad()
def sharpen_conds_pre_cfg(args):
nonlocal prev_cond, prev_uncond
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
sigma = args["sigma"][0].item()
first_step = sigma > (sigma_max - 1)
if first_step:
prev_cond = None
prev_uncond = None
if normalized:
n0 = conds_out[0].norm()
if uncond:
n1 = conds_out[1].norm()
prev_cond_tmp = conds_out[0].clone()
prev_uncond_tmp = conds_out[1].clone()
if not first_step and sigma > end_at_sigma and sigma <= start_at_sigma:
for b in range(len(conds_out[0])):
for c in range(len(conds_out[0][b])):
if prev_cond is not None and do_on in ['both','cond']:
conds_out[0][b][c] = normalize_adjust(conds_out[0][b][c], prev_cond[b][c], scale)
if prev_uncond is not None and uncond and do_on in ['both','uncond']:
conds_out[1][b][c] = normalize_adjust(conds_out[1][b][c], prev_uncond[b][c], scale)
prev_cond = prev_cond_tmp
if uncond:
prev_uncond = prev_uncond_tmp
if normalized:
conds_out[0] = selfnorm(conds_out[0]) * n0
if uncond:
conds_out[1] = selfnorm(conds_out[1]) * n1
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(sharpen_conds_pre_cfg)
return (m, )
class condBlurSharpeningNode:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"do_on": (["both","cond","uncond"], {"default": "both"},),
"operation": (["sharpen","sharpen_rescale","blur","blur_rescale"],),
"sharpening_scale": ("FLOAT", {"default": 0.5, "min": -10.0, "max": 10.0, "step": 1/20, "round": 1/100}),
"blur_sigma": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 10.0, "step": 1/20, "round": 1/100}),
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 1000000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 01.0, "min": 0.0, "max": 1000000.0, "step": 1/100, "round": 1/100}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, do_on, operation, sharpening_scale, blur_sigma, start_at_sigma, end_at_sigma):
model_sampling = model.model.model_sampling
sigma_max = model_sampling.sigma(model_sampling.timestep(model_sampling.sigma_max)).item()
operations = {
"sharpen": lambda x, y, z: x + (x - blur_tensor(x,sigma=z)) * y,
"blur": lambda x, y, z: blur_tensor(x, sigma=z),
"sharpen_rescale": lambda x, y, z: normlike(x + (x - blur_tensor(x,sigma=z)) * y, x),
"blur_rescale": lambda x, y, z: normlike(blur_tensor(x, sigma=z), x),
}
@torch.no_grad()
def sharpen_conds_pre_cfg(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
sigma = args["sigma"][0].item()
first_step = sigma > (sigma_max - 1)
if not first_step and sigma > end_at_sigma and sigma <= start_at_sigma:
for b in range(len(conds_out[0])):
for c in range(len(conds_out[0][b])):
if do_on in ['both','cond']:
conds_out[0][b][c] = operations[operation](conds_out[0][b][c],sharpening_scale,blur_sigma)
if uncond and do_on in ['both','uncond']:
conds_out[1][b][c] = operations[operation](conds_out[1][b][c],sharpening_scale,blur_sigma)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(sharpen_conds_pre_cfg)
return (m, )
@torch.no_grad()
def normalized_pow(t,p):
t_norm = t.norm()
t_sign = t.sign()
t_pow = (t / t_norm).abs().pow(p)
t_pow = selfnorm(t_pow) * t_norm * t_sign
return t_pow
class condExpNode:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"do_on": (["both","cond","uncond"], {"default": "both"},),
"exponent": ("FLOAT", {"default": 0.8, "min": 0.0, "max": 10.0, "step": 1/20, "round": 1/100}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, do_on, exponent):
@torch.no_grad()
def exponentiate_conds_pre_cfg(args):
if args["sigma"][0] <= 1: return args["conds_out"]
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
if do_on in ['both','uncond'] and not uncond:
return conds_out
for b in range(len(conds_out[0])):
if do_on in ['both','cond']:
conds_out[0][b] = normalized_pow(conds_out[0][b], exponent)
if uncond and do_on in ['both','uncond']:
conds_out[1][b] = normalized_pow(conds_out[1][b], exponent)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(exponentiate_conds_pre_cfg)
return (m, )
@torch.no_grad()
def topk_average(latent, top_k=0.25, measure="average"):
max_values = torch.topk(latent.flatten(), k=ceil(latent.numel()*top_k), largest=True ).values
min_values = torch.topk(latent.flatten(), k=ceil(latent.numel()*top_k), largest=False).values
value_range = measuring_methods[measure](max_values, min_values)
return value_range
apply_scaling_methods = {
"individual": lambda c, m: c * torch.tensor(m).view(c.shape[0],1,1).to(c.device),
"all_as_one": lambda c, m: c * m[0],
"average_of_all_channels" : lambda c, m: c * (sum(m) / len(m)),
"smallest_of_all_channels": lambda c, m: c * min(m),
"biggest_of_all_channels" : lambda c, m: c * max(m),
}
measuring_methods = {
"difference": lambda x, y: (x.mean() - y.mean()).abs() / 2,
"average": lambda x, y: (x.mean() + y.abs().mean()) / 2,
"biggest": lambda x, y: max(x.mean(), y.abs().mean()),
}
class automatic_pre_cfg:
@classmethod
def INPUT_TYPES(s):
scaling_methods_names = [k for k in apply_scaling_methods]
measuring_methods_names = [k for k in measuring_methods]
return {"required": {
"model": ("MODEL",),
"scaling_method": (scaling_methods_names, {"default": scaling_methods_names[0]}),
"min_max_method": ([m for m in measuring_methods], {"default": measuring_methods_names[1]}),
"reference_CFG": ("FLOAT", {"default": 8, "min": 0.0, "max": 100, "step": 1/10, "round": 1/100}),
"scale_multiplier": ("FLOAT", {"default": 0.8, "min": 0.0, "max": 100, "step": 1/100, "round": 1/100}),
"top_k": ("FLOAT", {"default": 0.25, "min": 0.0, "max": 0.5, "step": 1/20, "round": 1/100}),
},
"optional": {
"channels_selection": ("CHANS",),
}
}
RETURN_TYPES = ("MODEL","STRING",)
RETURN_NAMES = ("MODEL","parameters",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, scaling_method, min_max_method="difference", reference_CFG=8, scale_multiplier=0.8, top_k=0.25, channels_selection=None):
parameters_string = f"scaling_method: {scaling_method}\nmin_max_method: {min_max_method}"
if channels_selection is not None:
for x in range(channels_selection):
parameters_string += f"\nchannel {x+1}: {channels_selection[x]}"
scaling_methods_names = [k for k in apply_scaling_methods]
@torch.no_grad()
def automatic_pre_cfg(args):
conds_out = args["conds_out"]
cond_scale = args["cond_scale"]
uncond = torch.any(conds_out[1])
if reference_CFG == 0:
reference_scale = cond_scale
else:
reference_scale = reference_CFG
if not uncond:
return conds_out
if channels_selection is None:
channels = [True for _ in range(conds_out[0].shape[-3])]
else:
channels = channels_selection
for b in range(len(conds_out[0])):
chans = []
if scaling_method == scaling_methods_names[1]:
if all(channels):
mes = topk_average(reference_scale * conds_out[0][b] - (reference_scale - 1) * conds_out[1][b], top_k=top_k, measure=min_max_method)
else:
cond_for_measure = torch.stack([conds_out[0][b][j] for j in range(len(channels)) if channels[j]])
uncond_for_measure = torch.stack([conds_out[1][b][j] for j in range(len(channels)) if channels[j]])
mes = topk_average(reference_scale * cond_for_measure - (reference_scale - 1) * uncond_for_measure, top_k=top_k, measure=min_max_method)
chans.append(scale_multiplier / max(mes,0.01))
else:
for c in range(len(conds_out[0][b])):
if not channels[c]:
if scaling_method == scaling_methods_names[0]:
chans.append(1)
continue
mes = topk_average(reference_scale * conds_out[0][b][c] - (reference_scale - 1) * conds_out[1][b][c], top_k=top_k, measure=min_max_method)
new_scale = scale_multiplier / max(mes,0.01)
chans.append(new_scale)
conds_out[0][b] = apply_scaling_methods[scaling_method](conds_out[0][b],chans)
conds_out[1][b] = apply_scaling_methods[scaling_method](conds_out[1][b],chans)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(automatic_pre_cfg)
return (m, parameters_string,)
class channel_selection_node:
CHANNELS_AMOUNT = 4
@classmethod
def INPUT_TYPES(s):
toggles = {f"channel_{x}" : ("BOOLEAN", {"default": True}) for x in range(s.CHANNELS_AMOUNT)}
return {"required": toggles}
RETURN_TYPES = ("CHANS",)
FUNCTION = "exec"
CATEGORY = "model_patches/Pre CFG/channels_selectors"
def exec(self, **kwargs):
chans = []
for k, v in kwargs.items():
if "channel_" in k:
chans.append(v)
return (chans, )
class individual_channel_selection_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"exclude" : ("BOOLEAN", {"default": False}),
"selected_channel": ("INT", {"default": 1, "min": 1, "max": 128}),
"total_channels" : ("INT", {"default": 4, "min": 1, "max": 128}),
}
}
RETURN_TYPES = ("CHANS",)
FUNCTION = "exec"
CATEGORY = "model_patches/Pre CFG/channels_selectors"
def exec(self, exclude, selected_channel, total_channels):
chans = [exclude for _ in range(total_channels)]
chans[selected_channel - 1] = not exclude
return (chans, )
class channel_multiplier_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"channel_1": ("FLOAT", {"default": 1, "min": -10.0, "max": 10.0, "step": 1/100, "round": 1/100}),
"channel_2": ("FLOAT", {"default": 1, "min": -10.0, "max": 10.0, "step": 1/100, "round": 1/100}),
"channel_3": ("FLOAT", {"default": 1, "min": -10.0, "max": 10.0, "step": 1/100, "round": 1/100}),
"channel_4": ("FLOAT", {"default": 1, "min": -10.0, "max": 10.0, "step": 1/100, "round": 1/100}),
"selection": (["both","cond","uncond"],),
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 100.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 01.0, "min": 0.0, "max": 100.0, "step": 1/100, "round": 1/100}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, channel_1, channel_2, channel_3, channel_4, selection, start_at_sigma, end_at_sigma):
chans = [channel_1, channel_2, channel_3, channel_4]
@torch.no_grad()
def channel_multiplier_function(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
sigma = args["sigma"]
if sigma[0] <= end_at_sigma or sigma[0] > start_at_sigma:
return conds_out
for b in range(len(conds_out[0])):
for c in range(len(conds_out[0][b])):
if selection in ["both","cond"]:
conds_out[0][b][c] *= chans[c]
if uncond and selection in ["both","uncond"]:
conds_out[1][b][c] *= chans[c]
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(channel_multiplier_function)
return (m, )
class support_empty_uncond_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"method": (["from cond","divide by CFG"],),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, method):
@torch.no_grad()
def support_empty_uncond(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
cond_scale = args["cond_scale"]
if not uncond and cond_scale > 1:
if method == "divide by CFG":
conds_out[0] /= cond_scale
else:
conds_out[1] = conds_out[0].clone()
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(support_empty_uncond)
return (m, )
def replace_timestep(cond):
cond = deepcopy(cond)
cond[0]['timestep_start'] = 999999999.9
cond[0]['timestep_end'] = 0.0
return cond
def check_if_in_timerange(conds,timestep_in):
for c in conds:
all_good = True
if 'timestep_start' in c:
timestep_start = c['timestep_start']
if timestep_in[0] > timestep_start:
all_good = False
if 'timestep_end' in c:
timestep_end = c['timestep_end']
if timestep_in[0] < timestep_end:
all_good = False
if all_good: return True
return False
class zero_attention_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"do_on": (["cond","uncond"], {"default": "uncond"},),
"mix_scale": ("FLOAT", {"default": 1.5, "min": -2.0, "max": 2.0, "step": 1/2, "round": 1/100}),
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
# "attention": (["both","self","cross"],),
# "unet_block": (["input","middle","output"],),
# "unet_block_id": ("INT", {"default": 8, "min": 0, "max": 20}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, do_on, mix_scale, start_at_sigma, end_at_sigma, attention="both", unet_block="input", unet_block_id=8):
cond_index = 1 if do_on == "uncond" else 0
attn = {"both":["attn1","attn2"],"self":["attn1"],"cross":["attn2"]}[attention]
def zero_attention_function(q, k, v, extra_options, mask=None):
return torch.zeros_like(q)
@torch.no_grad()
def zero_attention_pre_cfg_patch(args):
conds_out = args["conds_out"]
sigma = args["sigma"][0].item()
if sigma > start_at_sigma or sigma <= end_at_sigma:
return conds_out
conds = args["conds"]
cond_to_process = conds[cond_index]
cond_generated = torch.any(conds_out[cond_index])
if not cond_generated:
cond_to_process = replace_timestep(cond_to_process)
elif mix_scale == 1:
print(" Mix scale at one!\nPrediction not generated.\nUse the node ConditioningSetTimestepRange to avoid generating if you want to use this node.")
return conds_out
model_options = deepcopy(args["model_options"])
for att in attn:
model_options = comfy.model_patcher.set_model_options_patch_replace(model_options, zero_attention_function, att, unet_block, unet_block_id)
(noise_pred,) = calc_cond_batch(args['model'], [cond_to_process], args['input'], args['timestep'], model_options)
if mix_scale == 1 or not cond_generated:
conds_out[cond_index] = noise_pred
elif cond_generated:
conds_out[cond_index] = weighted_average(noise_pred,conds_out[cond_index],mix_scale)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(zero_attention_pre_cfg_patch)
return (m, )
class perturbed_attention_guidance_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"scale": ("FLOAT", {"default": 0.5, "min": -2.0, "max": 10.0, "step": 1/20, "round": 1/100}),
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, scale, start_at_sigma, end_at_sigma, do_on="cond", attention="self", unet_block="middle", unet_block_id=0):
cond_index = 1 if do_on == "uncond" else 0
attn = {"both":["attn1","attn2"],"self":["attn1"],"cross":["attn2"]}[attention]
def perturbed_attention_guidance(q, k, v, extra_options, mask=None):
return v
@torch.no_grad()
def perturbed_attention_guidance_pre_cfg_patch(args):
conds_out = args["conds_out"]
sigma = args["sigma"][0].item()
if sigma > start_at_sigma or sigma <= end_at_sigma:
return conds_out
conds = args["conds"]
cond_to_process = conds[cond_index]
cond_generated = torch.any(conds_out[cond_index])
if not cond_generated:
return conds_out
model_options = deepcopy(args["model_options"])
for att in attn:
model_options = comfy.model_patcher.set_model_options_patch_replace(model_options, perturbed_attention_guidance, att, unet_block, unet_block_id)
(noise_pred,) = calc_cond_batch(args['model'], [cond_to_process], args['input'], args['timestep'], model_options)
conds_out[cond_index] = conds_out[cond_index] + (conds_out[cond_index] - noise_pred) * scale
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(perturbed_attention_guidance_pre_cfg_patch)
return (m, )
def sigma_to_percent(model_sampling, sigma_value):
if sigma_value >= 999999999.9:
return 0.0
if sigma_value <= 0.0:
return 1.0
sigma_tensor = torch.tensor([sigma_value], dtype=torch.float32)
timestep = model_sampling.timestep(sigma_tensor)
percent = 1.0 - (timestep.item() / 999.0)
return percent
class ConditioningSetTimestepRangeFromSigma:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"conditioning": ("CONDITIONING", ),
"sigma_start" : ("FLOAT", {"default": 15.0, "min": 0.0, "max": 10000.0, "step": 0.01}),
"sigma_end" : ("FLOAT", {"default": 0.0, "min": 0.0, "max": 10000.0, "step": 0.01})
}}
RETURN_TYPES = ("CONDITIONING",)
FUNCTION = "set_range"
CATEGORY = "advanced/conditioning"
def set_range(self, model, conditioning, sigma_start, sigma_end):
model_sampling = model.model.model_sampling
(c, ) = ConditioningSetTimestepRange.set_range(conditioning,sigma_to_percent(model_sampling, sigma_start),sigma_to_percent(model_sampling, sigma_end))
return (c, )
class ShapeAttentionNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": ("MODEL",),
"scale": ("FLOAT", {"default": 1.5, "min": 0.0, "max": 10.0, "step": 1/10, "round": 1/100}),
# "start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
# "end_at_sigma": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
# "enabled" : ("BOOLEAN", {"default": True}),
# "attention": (["both","self","cross"],),
# "unet_block": (["input","middle","output"],),
# "unet_block_id": ("INT", {"default": 8, "min": 0, "max": 20}), # uncomment these lines if you want to have fun with the other layers
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches"
def patch(self, model, scale, start_at_sigma=999999999.9, end_at_sigma=0.0, enabled=True, attention="self", unet_block="input", unet_block_id=8):
attn = {"both":["attn1","attn2"],"self":["attn1"],"cross":["attn2"]}[attention]
if scale == 1:
print(" Shape attention disabled (scale is one)")
if not enabled or scale == 1:
return (model,)
m = model.clone()
def shape_attention(q, k, v, extra_options, mask=None):
sigma = extra_options['sigmas'][0]
if sigma > start_at_sigma or sigma <= end_at_sigma:
return default_attention(q, k, v, extra_options['n_heads'], mask)
if scale != 0:
return default_attention(q, k, v, extra_options['n_heads'], mask) * scale
else:
return torch.zeros_like(q)
for att in attn:
m.model_options = comfy.model_patcher.set_model_options_patch_replace(m.model_options, shape_attention, att, unet_block, unet_block_id)
return (m,)
class ExlAttentionNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": ("MODEL",),
"scale": ("FLOAT", {"default": 2, "min": -1.0, "max": 10.0, "step": 1/10, "round": 1/100}),
"enabled": ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches"
def patch(self, model, scale, enabled):
if not enabled:
return (model,)
m = model.clone()
def cross_patch(q, k, v, extra_options, mask=None):
first_attention = default_attention(q, k, v, extra_options['n_heads'], mask)
second_attention = normlike(q+(q-default_attention(first_attention, k, v, extra_options['n_heads'])), first_attention) * scale
return second_attention
m.model_options = comfy.model_patcher.set_model_options_patch_replace(m.model_options, cross_patch, "attn2", "middle", 0)
return (m,)
class PreCFGRollLatentNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": ("MODEL",),
"shift_per_step" : ("INT", {"default": 1, "min": -10000, "max": 10000, "step": 1}),
"vertical" : ("BOOLEAN", {"default": False}),
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 100000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 100000.0, "step": 1/100, "round": 1/100}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, shift_per_step, vertical, start_at_sigma, end_at_sigma):
m = model.clone()
def pre_cfg_function(args):
conds_out = args["conds_out"]
sigma = args["sigma"][0].item()
if sigma > start_at_sigma or sigma <= end_at_sigma:
return conds_out
for x in range(len(conds_out)):
if torch.any(conds_out[x]):
for c in range(conds_out[x].shape[-3]):
conds_out[x] = roll_channel(conds_out[x], c, shift_per_step, -(1 + vertical))
return conds_out
m.set_model_sampler_pre_cfg_function(pre_cfg_function)
return (m,)
class PreCFGMirrorFlipLatentNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": ("MODEL",),
"vertical" : ("BOOLEAN", {"default": False}),
# "do_a_flip" : ("BOOLEAN", {"default": False}),
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 100000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 2.0, "min": 0.0, "max": 100000.0, "step": 1/100, "round": 1/100}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, vertical, start_at_sigma, end_at_sigma, do_a_flip=True):
m = model.clone()
def pre_cfg_function(args):
conds_out = args["conds_out"]
sigma = args["sigma"][0].item()
if sigma > start_at_sigma or sigma <= end_at_sigma:
return conds_out
for x in range(len(conds_out)):
if torch.any(conds_out[x]):
if do_a_flip:
conds_out[x] = mirror_flip(conds_out[x], vertical)
else:
conds_out[x] = mirror_from_middle(conds_out[x], vertical)
return conds_out
m.set_model_sampler_pre_cfg_function(pre_cfg_function)
return (m,)
class PreCFGsubtractMeanNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": ("MODEL",),
# "per_channel" : ("BOOLEAN", {"default": False}), #It's just not good
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"enabled" : ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, start_at_sigma, end_at_sigma, enabled, per_channel=False):
if not enabled: return (model,)
m = model.clone()
def pre_cfg_function(args):
conds_out = args["conds_out"]
sigma = args["sigma"][0].item()
if sigma > start_at_sigma or sigma <= end_at_sigma:
return conds_out
for x in range(len(conds_out)):
if torch.any(conds_out[x]):
for b in range(len(conds_out[x])):
if per_channel:
for c in range(len(conds_out[x][b])):
conds_out[x][b][c] -= conds_out[x][b][c].mean()
else:
conds_out[x][b] -= conds_out[x][b].mean()
return conds_out
m.set_model_sampler_pre_cfg_function(pre_cfg_function)
return (m,)
class PostCFGsubtractMeanNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": ("MODEL",),
# "per_channel" : ("BOOLEAN", {"default": False}), #It's just not good
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"enabled" : ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches"
def patch(self, model, start_at_sigma, end_at_sigma, enabled, per_channel=False):
if not enabled: return (model,)
m = model.clone()
def post_cfg_function(args):
cfg_result = args["denoised"]
sigma = args["sigma"][0].item()
if sigma > start_at_sigma or sigma <= end_at_sigma:
return cfg_result
for b in range(len(cfg_result)):
if per_channel:
for c in range(len(cfg_result[b])):
cfg_result[b][c] -= cfg_result[b][c].mean()
else:
cfg_result[b] -= cfg_result[b].mean()
return cfg_result
m.set_model_sampler_post_cfg_function(post_cfg_function)
return (m,)
class PostCFGDotNode:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": ("MODEL",),
"batch": ("INT", {"default": 0, "min": 0, "max": 100, "step": 1}),
"channel": ("INT", {"default": 0, "min": 0, "max": 100, "step": 1}),
"coord_x": ("INT", {"default": 64, "min": 0, "max": 1000, "step": 1}),
"coord_y": ("INT", {"default": 64, "min": 0, "max": 1000, "step": 1}),
"value": ("FLOAT", {"default": 1, "min": -10.0, "max": 10.0, "step": 1/10, "round": 1/100}),
"start_at_sigma": ("FLOAT", {"default": 0.1, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"enabled" : ("BOOLEAN", {"default": True}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches"
def patch(self, model, batch, channel, coord_x, coord_y, value, start_at_sigma, end_at_sigma, enabled):
if not enabled: return (model,)
m = model.clone()
def post_cfg_function(args):
cfg_result = args["denoised"]
sigma = args["sigma"][0].item()
if sigma > start_at_sigma or sigma <= end_at_sigma:
return cfg_result
channel_norm = cfg_result[batch][channel].norm()
cfg_result[batch][channel] /= channel_norm
cfg_result[batch][channel][coord_y][coord_x] = value
cfg_result[batch][channel] *= channel_norm
return cfg_result
m.set_model_sampler_post_cfg_function(post_cfg_function)
return (m,)
class uncondZeroPreCFGNode:
@classmethod
def INPUT_TYPES(s):
scaling_methods_names = [k for k in apply_scaling_methods]
return {"required": {
"model": ("MODEL",),
"scale": ("FLOAT", {"default": 0.75, "min": 0.0, "max": 10.0, "step": 1/20, "round": 0.01}),
"start_at_sigma": ("FLOAT", {"default": 100, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"scaling_method": (scaling_methods_names, {"default": scaling_methods_names[2]}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, scale, start_at_sigma, end_at_sigma, scaling_method):
scaling_methods_names = [k for k in apply_scaling_methods]
@torch.no_grad()
def uncond_zero_pre_cfg(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
sigma = args["sigma"][0].item()
if uncond or sigma <= end_at_sigma or sigma > start_at_sigma:
return conds_out
for b in range(len(conds_out[0])):
chans = []
if scaling_method == scaling_methods_names[1]:
mes = topk_average(8 * conds_out[0][b] - 7 * conds_out[1][b], measure="difference")
for c in range(len(conds_out[0][b])):
mes = topk_average(conds_out[0][b][c], measure="difference") ** 0.5
chans.append(scale / mes)
conds_out[0][b] = apply_scaling_methods[scaling_method](conds_out[0][b],chans)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(uncond_zero_pre_cfg)
return (m, )
class latent_color_control_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"model": ("MODEL",),
"Red": ("FLOAT", {"default": 0, "min": -2.0, "max": 2.0, "step": 1/100, "round": 1/100}),
"Green": ("FLOAT", {"default": 0, "min": -2.0, "max": 2.0, "step": 1/100, "round": 1/100}),
"Blue": ("FLOAT", {"default": 0, "min": -2.0, "max": 2.0, "step": 1/100, "round": 1/100}),
"selection": (["both","cond","uncond"],),
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 100.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 01.0, "min": 0.0, "max": 100.0, "step": 1/100, "round": 1/100}),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, Red, Green, Blue, selection, start_at_sigma, end_at_sigma):
latent_rgb_factors = sdxl_latent_rgb_factors
rgb = [Red, Green, Blue]
@torch.no_grad()
def latent_control_pre_cfg_function(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
sigma = args["sigma"][0]
if sigma <= end_at_sigma or sigma > start_at_sigma or all(c == 0 for c in rgb):
return conds_out
conds_index = []
if selection in ["both","cond"]:
conds_index.append(0)
if uncond and selection in ["both","uncond"]:
conds_index.append(1)
for i in conds_index:
for b in range(len(conds_out[i])):
cond_norm = conds_out[i][b].norm()
color_cond = torch.zeros_like(conds_out[i][b])
for c in range(len(conds_out[i][b])):
for r in range(len(rgb)):
if rgb[r] != 0:
color_cond[c] += rgb[r] * latent_rgb_factors[c][r] / 0.13025
conds_out[i][b] = selfnorm(conds_out[i][b] + color_cond) * cond_norm
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(latent_control_pre_cfg_function)
return (m, )
class variable_scale_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"target_scale": ("FLOAT", {"default": 5.0, "min": 1.0, "max": 100.0, "step": 1/2, "round": 1/100}),
"target_as_start": ("BOOLEAN", {"default": True}),
"proportional_to": (["sigma","steps progression"],),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, target_scale, target_as_start, proportional_to):
model_sampling = model.model.model_sampling
sigma_max = model_sampling.sigma(model_sampling.timestep(model_sampling.sigma_max)).item()
@torch.no_grad()
def variable_scale_pre_cfg_patch(args):
conds_out = args["conds_out"]
cond_scale = args["cond_scale"]
sigma = args["sigma"][0].item()
scales = [cond_scale,target_scale]
if not torch.any(conds_out[1]):
return conds_out
if proportional_to == "steps progression":
progression = sigma_to_percent(model_sampling, sigma)
else:
progression = 1 - sigma / sigma_max
progression = max(min(progression, 1), 0)
current_scale = scales[target_as_start] * (1 - progression) + scales[not target_as_start] * progression
new_scale = (current_scale - 1) / (cond_scale - 1)
conds_out[1] = weighted_average(conds_out[1], conds_out[0], new_scale)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(variable_scale_pre_cfg_patch)
return (m, )
selfsquare = lambda x: x.abs().pow(2) * x.sign()
class boost_std_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"enabled" : ("BOOLEAN", {"default": True})
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, enabled):
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
if not uncond or not enabled:
return conds_out
for b in range(len(conds_out[0])):
pred = 2 * conds_out[0][b] - conds_out[1][b]
pred_std = pred.std()
pred_std = selfsquare(pred_std)
pred_std_cond = conds_out[0][b].std() / pred_std
pred_std_uncond = conds_out[1][b].std() / pred_std
pred_std_cond = selfsquare(pred_std_cond)
pred_std_uncond = selfsquare(pred_std_uncond)
conds_out[0][b] = conds_out[0][b] * pred_std_cond
conds_out[1][b] = conds_out[1][b] * pred_std_uncond
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
@torch.no_grad()
def clamp_uncond_relative_to_sign(conds_out, same_sign, btst, value_filter, use_uncond_sign, multiplier):
if same_sign:
sign_mask = conds_out[0].sign() == conds_out[1].sign()
else:
sign_mask = conds_out[0].sign() != conds_out[1].sign()
if btst == "smaller_than":
abs_mask = conds_out[1].abs() < conds_out[0].abs()
elif btst == "bigger_than":
abs_mask = conds_out[1].abs() > conds_out[0].abs()
elif btst == "any":
abs_mask = sign_mask
if value_filter == "disabled":
value_mask = sign_mask
elif value_filter == "only_positive_cond":
value_mask = conds_out[0] > 0
elif value_filter == "only_negative_cond":
value_mask = conds_out[0] < 0
elif value_filter == "only_positive_uncond":
value_mask = conds_out[1] > 0
elif value_filter == "only_negative_uncond":
value_mask = conds_out[1] < 0
sign_mask = sign_mask == abs_mask
abs_sign_mask = sign_mask == value_mask
if not same_sign and use_uncond_sign:
result = conds_out[0][abs_sign_mask].abs() * conds_out[1][abs_sign_mask].sign()
else:
result = conds_out[0][abs_sign_mask]
conds_out[1][abs_sign_mask] = conds_out[1][abs_sign_mask] + (result - conds_out[1][abs_sign_mask]) * multiplier
return conds_out
class clamp_sign_uncond_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"same_sign" : (["disabled","bigger_than","smaller_than","any"],),
"opposite_sign" : (["disabled","bigger_than","smaller_than","any"],),
"value_filter" : (["disabled","only_positive_cond","only_negative_cond","only_positive_uncond","only_negative_uncond"],),
"use_uncond_sign" : ("BOOLEAN", {"default": True}),
"multiplier": ("FLOAT", {"default": 1.0, "min": -2.0, "max": 2.0, "step": 1/2, "round": 1/100}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, same_sign, opposite_sign, value_filter, use_uncond_sign, multiplier):
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
if not uncond or (same_sign == "disabled" and opposite_sign == "disabled"):
return conds_out
if same_sign != "disabled":
conds_out = clamp_uncond_relative_to_sign(conds_out, True, same_sign, value_filter, use_uncond_sign, multiplier)
if opposite_sign != "disabled":
conds_out = clamp_uncond_relative_to_sign(conds_out, False, opposite_sign, value_filter, use_uncond_sign, multiplier)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
@torch.no_grad()
def uncond_limiter(x_orig, cond, uncond, cond_scale, same_sign, opposite_sign, multiplier):
denoised = ((x_orig - uncond) + cond_scale * ((x_orig - cond) - (x_orig - uncond)))
denoised_diff = x_orig - denoised
cond_relation = cond_scale * cond - (cond_scale - 1) * uncond
denoised_relation = cond_relation - denoised_diff
uncond = uncond - denoised_relation * multiplier
return uncond
class clamp_uncond_to_denoised_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"same_sign" : (["disabled","bigger_than","smaller_than","any"],),
"opposite_sign" : (["disabled","bigger_than","smaller_than","any"],),
"multiplier": ("FLOAT", {"default": 1.0, "min": -2.0, "max": 2.0, "step": 1/2, "round": 1/100}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, same_sign, opposite_sign, multiplier):
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
if not torch.any(conds_out[1]) or (same_sign == "disabled" and opposite_sign == "disabled"):
return conds_out
cond = conds_out[0]
uncond = conds_out[1]
x_orig = args['input']
cond_scale = args["cond_scale"]
conds_out[1] = uncond_limiter(x_orig,cond,uncond,cond_scale,same_sign,opposite_sign,multiplier)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
class latent_noise_subtract_mean_node:
def __init__(self):
pass
@classmethod
def INPUT_TYPES(s):
return {"required": {
"latent_input": ("LATENT", {"forceInput": True}),
"enabled" : ("BOOLEAN", {"default": True}),
}}
FUNCTION = "exec"
RETURN_TYPES = ("LATENT",)
CATEGORY = "latent"
def exec(self, latent_input, enabled):
if not enabled:
return (latent_input,)
new_latents = deepcopy(latent_input)
for x in range(len(new_latents['samples'])):
new_latents['samples'][x] -= torch.mean(new_latents['samples'][x])
return (new_latents,)
class flip_flip_conds_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"enabled" : ("BOOLEAN", {"default": True})
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, enabled):
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
if not uncond or not enabled:
return conds_out
conds_out[0], conds_out[1] = conds_out[1], conds_out[0]
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
class norm_uncond_to_cond_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"enabled" : ("BOOLEAN", {"default": True})
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, enabled):
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
if not uncond or not enabled:
return conds_out
conds_out[1] = conds_out[1] / conds_out[1].norm() * conds_out[0].norm()
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
class replace_uncond_channel_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"channel": ("INT", {"default": 1, "min": 1, "max": 128, "step": 1}),
"enabled" : ("BOOLEAN", {"default": True})
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, channel, enabled):
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
if not uncond or not enabled:
return conds_out
for b in range(len(conds_out[0])):
if len(conds_out[1][b]) < channel:
print(F" WRONG CHANNEL SELECTED. THE LATENT SPACE ONLY HAS {len(conds_out[1][b])} CHANNELS")
else:
conds_out[1][b][channel - 1] = conds_out[0][b][channel - 1]
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
class merge_uncond_channel_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"channel": ("INT", {"default": 1, "min": 1, "max": 128, "step": 1}),
"CFG_scale": ("FLOAT", {"default": 5, "min": 2.0, "max": 100.0, "step": 1/2, "round": 1/100}),
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 100.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 01.0, "min": 0.0, "max": 100.0, "step": 1/100, "round": 1/100}),
"enabled" : ("BOOLEAN", {"default": True})
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, channel, CFG_scale, start_at_sigma, end_at_sigma, enabled):
if not enabled: return model,
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
cond_scale = args["cond_scale"]
sigma = args["sigma"][0].item()
if not torch.any(conds_out[1]) or sigma <= end_at_sigma or sigma > start_at_sigma:
return conds_out
for b in range(len(conds_out[0])):
if len(conds_out[1][b]) < channel:
print(F" WRONG CHANNEL SELECTED. THE LATENT SPACE ONLY HAS {len(conds_out[1][b])} CHANNELS")
else:
new_scale = (CFG_scale - 1) / (cond_scale - 1)
conds_out[1][b][channel - 1] = weighted_average(conds_out[1][b][channel - 1], conds_out[0][b][channel - 1], new_scale)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
class rescale_cfg_during_sigma_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"CFG_scale": ("FLOAT", {"default": 5, "min": 2.0, "max": 100.0, "step": 1/2, "round": 1/100}),
"start_at_sigma": ("FLOAT", {"default": 15.0, "min": 0.0, "max": 100.0, "step": 1/100, "round": 1/100}),
"end_at_sigma": ("FLOAT", {"default": 01.0, "min": 0.0, "max": 100.0, "step": 1/100, "round": 1/100}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, CFG_scale, start_at_sigma, end_at_sigma):
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
cond_scale = args["cond_scale"]
sigma = args["sigma"][0].item()
if not torch.any(conds_out[1]) or sigma <= end_at_sigma or sigma > start_at_sigma:
return conds_out
if cond_scale == 1:
cond_scale += 1e-08
conds_out[1] = make_new_uncond_at_scale_co(conds_out,cond_scale,CFG_scale)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
class multiply_cond_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"selection": (["both","cond","uncond"],),
"value": ("FLOAT", {"default": 0, "min": -100.0, "max": 100.0, "step": 1/100, "round": 1/100}),
"enabled" : ("BOOLEAN", {"default": True})
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, selection, value, enabled):
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
if (not uncond and selection in ["both","uncond"]) or not enabled:
return conds_out
if selection in ["both","cond"]:
conds_out[0] = conds_out[0] * value
if selection in ["both","uncond"]:
conds_out[1] = conds_out[1] * value
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
class minmax_clamp_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"cond_to_clamp": (["uncond","cond"],),
"enabled" : ("BOOLEAN", {"default": True})
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, cond_to_clamp, enabled):
clamp_neg = cond_to_clamp == "uncond"
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
if not uncond or not enabled:
return conds_out
for b in range(len(conds_out[0])):
for c in range(len(conds_out[0][b])):
conds_out[clamp_neg][b][c] = torch.clamp(conds_out[clamp_neg][b][c],min=conds_out[not clamp_neg][b][c].min(),max=conds_out[not clamp_neg][b][c].max())
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
class lerp_conds_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"selection": (["both","uncond","cond"],),
"scale": ("FLOAT", {"default": 1, "min": 0.0, "max": 10.0, "step": 1/10, "round": 1/100}),
"enabled" : ("BOOLEAN", {"default": True})
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, selection, scale, enabled):
@torch.no_grad()
def pre_cfg_patch(args):
conds_out = args["conds_out"]
uncond = torch.any(conds_out[1])
if not uncond or not enabled or scale == 1:
return conds_out
if selection in ["both"]:
conds_out[0], conds_out[1] = torch.lerp(conds_out[1], conds_out[0], scale), torch.lerp(conds_out[0], conds_out[1], scale)
elif selection in ["cond"]:
conds_out[0] = torch.lerp(conds_out[1], conds_out[0], scale)
elif selection in ["uncond"]:
conds_out[1] = torch.lerp(conds_out[0], conds_out[1], scale)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
def generate_gradient_mask(tensor, horizontal=False):
dim = 3 if horizontal else 2
gradient = torch.linspace(0, 1, steps=tensor.size(dim), device=tensor.device)
if horizontal:
merging_gradient = gradient.repeat(tensor.size(0), tensor.size(1), tensor.size(2), 1)
else:
merging_gradient = gradient.unsqueeze(1).repeat(tensor.size(0), tensor.size(1), 1, tensor.size(3))
return merging_gradient
@torch.no_grad()
def random_swap(tensors, proportion=1):
# torch.manual_seed(seed)
num_tensors = tensors.shape[0]
tensor_size = tensors[0].numel()
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],true_count
class gradient_scaling_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"maximum_scale": ("FLOAT", {"default": 80, "min": 0.0, "max": 1000.0, "step": 1, "round": 1/100, "tooltip":"It is an equivalent to the CFG scale."}),
"minimum_scale": ("FLOAT", {"default": 4.5, "min": 0.0, "max": 10.0, "step": 1/2, "round": 1/100, "tooltip":"It is an equivalent to the CFG scale."}),
"strength": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 10.0, "step": 1/10, "round": 1/10}),
"end_at_sigma": ("FLOAT", {"default": 0.28, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"converging_scales" : ("BOOLEAN", {"default": True}),
# "noise_add_diff" : ("BOOLEAN", {"default": True}),
# "split_channels" : ("BOOLEAN", {"default": False}),
"invert_mask" : ("BOOLEAN", {"default": False}),
# ,"black_mean","rev_mean","cond_mean"
"no_input" : (["rand","rev","cond","uncond","swap","r_swap","rev_swap","rev_r_swap","black","black_cond","black_uncond","black_CFG","black_CFG_diff_noise","black_CFG_x2_diff_noise","black_x2_CFG_diff_noise","black_swap","black_r_swap","all_avg","all_avg_x2","all_avg_321","all_avg_312","diff","CFG_diff","CFG_diff_noise","add_diff","rand_rev","rev_cond","rand_cond","rev_cond_sp","cond_rev_sp"],),
# "start_at_sigma": ("FLOAT", {"default": 15, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
# "end_at_sigma": ("FLOAT", {"default": 0.28, "min": 0.0, "max": 1000.0, "step": 1/100, "round": 1/100}),
"free_scale" : (["disabled","sharp","sharp_clamp","full","full_blur","full_sharp","full_clamp","full_blur_clamp","full_sharp_clamp"],),
# "free_scale" : (["disabled","full","per_layer","full_and_blur","per_layer_and_blur","full_clamp","per_layer_clamp","full_and_blur_clamp","per_layer_and_blur_clamp"],),
},
"optional":{
"input_mask": ("MASK", {"tooltip":"If only a mask is connected the scale becomes a CFG scale of what is being masked.\nWhen a latent is connected the mask defines what will be modified by the node."},),
"input_latent": ("LATENT", {"tooltip":"If a latent is connected the scale becomes the maximum scale allowed in which to seek similarity."},),
}
}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def get_latent_guidance_mask_channel(self,x_orig,cond,uncond,guide,minimum_scale,maximum_scale,noise_add_diff):
scales = torch.zeros_like(x_orig, device=x_orig.device)
for b in range(cond.shape[0]):
for c in range(cond.shape[1]):
scales[b][c] = self.get_latent_guidance_mask(x_orig[b][c],cond[b][c],uncond[b][c],guide[0][c],minimum_scale,maximum_scale,noise_add_diff)
return scales
@torch.no_grad()
def get_latent_guidance_mask(self,x_orig,cond,uncond,guide,minimum_scale,maximum_scale,noise_add_diff):
low_denoised = get_denoised_at_scale(x_orig,cond,uncond,minimum_scale)
high_denoised = get_denoised_at_scale(x_orig,cond,uncond,maximum_scale)
if noise_add_diff:
guide = guide + (guide - (x_orig * guide.norm() / x_orig.norm()))
guide = guide / guide.norm()
low_diff = (low_denoised - guide * low_denoised.norm()).abs()
high_diff = (high_denoised - guide * high_denoised.norm()).abs()
return torch.clamp(low_diff / high_diff, min=0, max=1)
def get_black_latent(self,):
latent_path = os.path.join(current_dir,"latents","sdxl_black.pt")
latent_space = torch.load(latent_path).to(device=default_device)
return latent_space
def patch(self, model, maximum_scale, minimum_scale, strength, end_at_sigma, start_at_sigma=99999, invert_mask=False, no_input="black", noise_add_diff=True, converging_scales=False, split_channels=False, free_scale="disabled", input_mask=None, input_latent=None):
black_avg = [-21.758529663085938, 3.8702831268310547, 2.311274766921997, 2.559422016143799]
black_latent = self.get_black_latent()
# if input_mask is None and input_latent is None:
# return (model,)
sigma_min, sigma_max = get_sigma_min_max(model)
model_sampling = model.model.model_sampling
scaling_function = self.get_latent_guidance_mask_channel if split_channels else self.get_latent_guidance_mask
mask_as_weight = None
latent_as_guidance = None
random_guidance = False
if input_mask is not None:
mask_as_weight = input_mask.clone().to(device=default_device)
if invert_mask:
mask_as_weight = 1 - mask_as_weight
if mask_as_weight.dim() == 3:
mask_as_weight = mask_as_weight.unsqueeze(1)
if input_latent is not None:
latent_as_guidance = input_latent["samples"].clone().to(device=default_device)
elif input_mask is None:
random_guidance = True
@torch.no_grad()
def snc(x): return x / x.norm()
trl = lambda x: torch.randn_like(x,device=x.device)
# p y t h o n i c a s f u c k >.<
no_input_operations = {
"rand": lambda x, y, o, z, s, c: snc(trl(x)) * x.norm(),
"rev": lambda x, y, o, z, s, c: x * -1,
"cond": lambda x, y, o, z, s, c: snc(y) * x.norm(),
"uncond": lambda x, y, o, z, s, c: snc(o) * x.norm() * -1,
"swap": lambda x, y, o, z, s, c: no_input_operations["cond"](x, y, o, z, s, c) if s > 0.37 else no_input_operations["uncond"](x, y, o, z, s, c),
"r_swap": lambda x, y, o, z, s, c: no_input_operations["cond"](x, y, o, z, s, c) if s <= 0.37 else no_input_operations["uncond"](x, y, o, z, s, c),
"rev_swap": lambda x, y, o, z, s, c: snc(no_input_operations["rev"](x, y, o, z, s, c)+no_input_operations["swap"](x, y, o, z, s, c)) * x.norm(),
"rev_r_swap": lambda x, y, o, z, s, c: snc(no_input_operations["rev"](x, y, o, z, s, c)+no_input_operations["r_swap"](x, y, o, z, s, c)) * x.norm(),
# "black": lambda x, y, o, z, s, c: snc(torch.tensor(black_avg).view(1, len(black_avg), 1, 1).expand(1, len(black_avg), x.shape[2], x.shape[3]).to(device=x.device)) * x.norm(),
"black": lambda x, y, o, z, s, c: snc(black_latent.clone()) * x.norm(),
"black_cond": lambda x, y, o, z, s, c: (no_input_operations["cond"](x, y, o, z, s, c) + no_input_operations["black"](x, y, o, z, s, c)) / 2,
"black_uncond": lambda x, y, o, z, s, c: (no_input_operations["uncond"](x, y, o, z, s, c) + no_input_operations["black"](x, y, o, z, s, c)) / 2,
"black_CFG": lambda x, y, o, z, s, c: (snc(y * c - o * (c - 1)) * x.norm() + no_input_operations["black"](x, y, o, z, s, c)) / 2,
"black_denoised": lambda x, y, o, z, s, c: (snc(x - ( (x - o) + c * ( (x - y) - (x - o) ))) * x.norm() + no_input_operations["black"](x, y, o, z, s, c)) / 2,
"black_denoised_x2": lambda x, y, o, z, s, c: (snc(x - ( (x - o) + c * ( (x - y) - (x - o) ))) * x.norm() * 0.5 + no_input_operations["black"](x, y, o, z, s, c) * 1.5) / 2,
"black_CFG_diff_noise": lambda x, y, o, z, s, c: snc(no_input_operations["black_CFG"](x, y, o, z, s, c)-x) * x.norm(),
"black_CFG_x2_diff_noise": lambda x, y, o, z, s, c: snc(2 * no_input_operations["black_CFG"](x, y, o, z, s, c) - x) * x.norm(),
"black_x2_CFG_diff_noise": lambda x, y, o, z, s, c: snc(no_input_operations["black"](x, y, o, z, s, c) + no_input_operations["black_CFG"](x, y, o, z, s, c) - x) * x.norm(),
"black_swap": lambda x, y, o, z, s, c: (no_input_operations["black"](x, y, o, z, s, c) + no_input_operations["swap"](x, y, o, z, s, c)) / 2,
"black_r_swap": lambda x, y, o, z, s, c: (no_input_operations["black"](x, y, o, z, s, c) + no_input_operations["r_swap"](x, y, o, z, s, c)) / 2,
"all_avg": lambda x, y, o, z, s, c: snc(y-x-o) * x.norm(),
"all_avg_x2": lambda x, y, o, z, s, c: snc(2*y-x-o) * x.norm(),
"all_avg_321": lambda x, y, o, z, s, c: snc(3*y-2*x-o) * x.norm(),
"all_avg_312": lambda x, y, o, z, s, c: snc(3*y-2*o-x) * x.norm(),
"diff": lambda x, y, o, z, s, c: snc(y - o) * x.norm(),
"CFG_diff": lambda x, y, o, z, s, c: snc(y * c - o * (c - 1)) * x.norm(),
"CFG_diff_noise": lambda x, y, o, z, s, c: (snc(y * c - o * (c - 1)) + snc(-x)) / 2 * x.norm(),
"add_diff": lambda x, y, o, z, s, c: snc(y + (y - o) / 2) * x.norm(),
"rand_rev": lambda x, y, o, z, s, c: snc(trl(x)) * x.norm(),
"rev_cond": lambda x, y, o, z, s, c: (snc(x) * -1 + snc(y) * 0.5) * x.norm() / 1.5,
"rand_cond": lambda x, y, o, z, s, c: (snc(x) * -1 + snc(trl(x)) * 0.5) * x.norm() / 1.5,
"rev_cond_sp": lambda x, y, o, z, s, c: no_input_operations["rev"](x, y, o, z, s, c) * z + (1 - z) * no_input_operations["cond"](x, y, o, z, s, c),
"cond_rev_sp": lambda x, y, o, z, s, c: no_input_operations["rev"](x, y, o, z, s, c) * (1 - z) + z * no_input_operations["cond"](x, y, o, z, s, c),
}
@torch.no_grad()
def pre_cfg_patch(args):
nonlocal mask_as_weight, latent_as_guidance, black_latent
conds_out = args["conds_out"]
cond_scale = args["cond_scale"]
x_orig = args['input']
sigma = args["sigma"][0]
sp = min(1,max(0,sigma_to_percent(model_sampling, sigma - sigma_min * 3) + 1 / 100)) ** 2
sp2 = (sigma - sigma_min) / (sigma_max - sigma_min)
if not torch.any(conds_out[1]) or sigma <= end_at_sigma or sigma > start_at_sigma or (converging_scales and sp == 1):
return conds_out
if converging_scales:
current_maximum_scale = sp * cond_scale + (1 - sp) * maximum_scale
current_minimum_scale = sp * cond_scale + (1 - sp) * minimum_scale
else:
current_maximum_scale = maximum_scale
current_minimum_scale = minimum_scale
if mask_as_weight is not None and mask_as_weight.shape[-2:] != conds_out[1].shape[-2:]:
mask_as_weight = F.interpolate(mask_as_weight, size=(conds_out[1].shape[-2], conds_out[1].shape[-1]), mode='bilinear', align_corners=False)
if "black" in no_input and random_guidance:
if black_latent.shape[-2:] != conds_out[1].shape[-2:]:
black_latent = F.interpolate(black_latent, size=(conds_out[1].shape[-2], conds_out[1].shape[-1]), mode='bilinear', align_corners=False)
if random_guidance:
latent_as_guidance = no_input_operations[no_input](x_orig.clone(),conds_out[0].clone(),conds_out[1].clone(),sp,sp2,cond_scale)
if latent_as_guidance is not None:
if latent_as_guidance.shape[-2:] != conds_out[1].shape[-2:]:
latent_as_guidance = F.interpolate(latent_as_guidance, size=(conds_out[1].shape[-2], conds_out[1].shape[-1]), mode='bilinear', align_corners=False)
scaling_weight = scaling_function(x_orig,conds_out[0],conds_out[1],latent_as_guidance.clone(),current_minimum_scale,current_maximum_scale,noise_add_diff)
target_scales = scaling_weight * current_maximum_scale + (1 - scaling_weight) * current_minimum_scale
if "blur" in free_scale:
for b in range(target_scales.shape[0]):
for c in range(target_scales.shape[1]):
target_scales[b][c] = blur_tensor(target_scales[b][c], kernel_size=9, sigma=1.0)
elif "sharp" in free_scale:
for b in range(target_scales.shape[0]):
for c in range(target_scales.shape[1]):
target_scales[b][c] = target_scales[b][c] + (target_scales[b][c] - blur_tensor(target_scales[b][c], kernel_size=9, sigma=0.5)) * 0.5
if "full" in free_scale:
target_scales = target_scales * cond_scale / target_scales.mean()
elif "per_layer" in free_scale:
for b in range(target_scales.shape[0]):
for c in range(target_scales.shape[1]):
target_scales[b][c] = target_scales[b][c] * cond_scale / target_scales[b][c].mean()
if "clamp" in free_scale:
target_scales = torch.clamp(target_scales,min=current_minimum_scale,max=current_maximum_scale)
global_multiplier = strength
if input_mask is not None:
global_multiplier = global_multiplier * mask_as_weight
target_scales = target_scales * global_multiplier + torch.full_like(target_scales, cond_scale) * (1 - global_multiplier)
conds_out[1] = make_new_uncond_at_scale(conds_out[0],conds_out[1],cond_scale,target_scales)
return conds_out
else:
target_scales = maximum_scale * mask_as_weight * strength + torch.full_like(conds_out[1], cond_scale) * (1 - mask_as_weight * strength)
conds_out[1] = make_new_uncond_at_scale(conds_out[0],conds_out[1],cond_scale,target_scales)
return conds_out
m = model.clone()
m.set_model_sampler_pre_cfg_function(pre_cfg_patch)
return (m, )
dArK_ScRiPtUrE = lambda s: ''.join(random.choice([c.upper(), c.lower()]) for c in s)
dArK_mEtHoDs = {"ThE dArKnEsS":["black",1000,"will come and remaiiiin",True,"full_blur",[]],
"ThE CoNtRaCt":["black_denoised",1000,"has been siiiiiiigned.",True,"full_blur",[]],
"ThE ExChAnGe":["black_swap",1000,"will be done with his souuuuuuul",True,"full_blur",[]],
"ThE rEdEmPtiOn":["swap",16,"is never pooossiiiibleeeee.",False,"disabled",[]],
"ThE ShAdOw":["black",12,"of his mom covers the eaaaarth",False,"disabled",[]],
"ThE AbYsS":["black",32,"is deeeeeeeeeeeeeeeeeep",False,"disabled",[]],
"ThE WhIsPeR oF tHe DrOwNeD":["swap",12,"sooooundeeeeed liiiiiiiiiiike 'blub'.",False,"disabled",[["ThE ShAdOw", False]]],
}
dArK_nAmEs = [n for n in dArK_mEtHoDs]
class dark_guidance_pre_cfg_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {"model": ("MODEL",),
"RiTuaL": (dArK_nAmEs, {"default": dArK_nAmEs[0]}),
"AsK_fOr_ForGiVeNeSs" : ("BOOLEAN", {"default": False}),
}}
RETURN_TYPES = ("MODEL",)
FUNCTION = "patch"
CATEGORY = "model_patches/Pre CFG"
def patch(self, model, RiTuaL, AsK_fOr_ForGiVeNeSs, as_extra=False):
if not as_extra:
decided = dArK_ScRiPtUrE(f"{RiTuaL} {dArK_mEtHoDs[RiTuaL][2]}")
print(f" \033[91m\033[4m{decided}\033[0m")
simi,sima = get_sigma_min_max(model)
# short_end = (sima-simi)*0.0172+simi
guided_guidance_scaling = gradient_scaling_pre_cfg_node()
m, = guided_guidance_scaling.patch(model=model, end_at_sigma=0,maximum_scale=dArK_mEtHoDs[RiTuaL][1],minimum_scale=4.75,
strength=1,converging_scales=True,no_input=dArK_mEtHoDs[RiTuaL][0],free_scale=dArK_mEtHoDs[RiTuaL][4])
if AsK_fOr_ForGiVeNeSs:
m, = skimmed_CFG_patch_wrap(m,end_proportion=0.35)
if dArK_mEtHoDs[RiTuaL][3]:
cds = condDiffSharpeningNode()
m, = cds.patch(model=m,do_on="both",start_at_sigma=99999,end_at_sigma=0,scale=0.3,normalized=True)
cdsb = condBlurSharpeningNode()
m, = cdsb.patch(model=m,do_on="both",start_at_sigma=99999,end_at_sigma=(sima-simi)*0.37+simi,
sharpening_scale=0.5,blur_sigma=0.5,operation="sharpen_rescale")
for rit in dArK_mEtHoDs[RiTuaL][5]:
dcg1 = dark_guidance_pre_cfg_node()
m, = dcg1.patch(m,RiTuaL=rit[0],AsK_fOr_ForGiVeNeSs=rit[1],as_extra=True)
return (m, )
def adjust_vibrance(images: torch.Tensor, vibrance_factor: float) -> torch.Tensor:
grayscale_images = images.mean(dim=-1, keepdim=True)
vibrance_adjustment = (images - grayscale_images) * vibrance_factor
low_saturation_mask = (images - grayscale_images).abs() < 0.5
adjusted_images = images + vibrance_adjustment * low_saturation_mask.float()
return torch.clamp(adjusted_images, 0, 1)
def adjust_saturation(images: torch.Tensor, vibrance_factor: float) -> torch.Tensor:
grayscale_images = images.mean(dim=-1, keepdim=True)
adjusted_images = grayscale_images + vibrance_factor * (images - grayscale_images)
return torch.clamp(adjusted_images, 0, 1)
class EmptyRGBImage:
@classmethod
def INPUT_TYPES(s):
return {"required": { "width": ("INT", {"default": 1024, "min": 1, "max": 16384, "step": 1}),
"height": ("INT", {"default": 1024, "min": 1, "max": 16384, "step": 1}),
"batch_size": ("INT", {"default": 1, "min": 1, "max": 4096}),
"r": ("INT", {"default": 0, "min": 0, "max": 255, "step": 1}),
"g": ("INT", {"default": 0, "min": 0, "max": 255, "step": 1}),
"b": ("INT", {"default": 0, "min": 0, "max": 255, "step": 1}),
},
"optional": {
"grayscale_to_color": ("IMAGE",),
}}
RETURN_TYPES = ("IMAGE",)
FUNCTION = "generate"
CATEGORY = "image"
def generate(self, width, height, batch_size=1, r=0, g=0, b=0, grayscale_to_color=None):
if grayscale_to_color is not None:
grayscale_to_color = grayscale_to_color.permute(0, 3, 1, 2).mean(dim=1).unsqueeze(-1)
height = grayscale_to_color.shape[1]
width = grayscale_to_color.shape[2]
r_normalized = torch.full([batch_size, height, width, 1], r / 255.0)
g_normalized = torch.full([batch_size, height, width, 1], g / 255.0)
b_normalized = torch.full([batch_size, height, width, 1], b / 255.0)
rgb_image = torch.cat((r_normalized, g_normalized, b_normalized), dim=-1)
if grayscale_to_color is not None:
rgb_image = rgb_image * grayscale_to_color
return (rgb_image,)
gradient_patterns = {
"linear": lambda x, y: x,
"sine": lambda x, y: torch.sin(x * torch.pi * y),
"triangle": lambda x, y: 2 * torch.abs(torch.round(x % (1 / max(y, 1)) * y) - (x % (1 / max(y, 1)) * y)),
}
# red = [-19.2851, -19.4045, 11.3942, -11.8191]
# green = [-3.2465, 14.3003, 27.0502, 9.1685]
# blue = [0.5476, 16.1999, -17.2144, 4.3121]
class GradientRGBImage:
@classmethod
def INPUT_TYPES(s):
return {"required": { "width": ("INT", {"default": 1024, "min": 0, "max": 16384, "step": 64}),
"height": ("INT", {"default": 1024, "min": 0, "max": 16384, "step": 64}),
"r1": ("INT", {"default": 0, "min": 0, "max": 255, "step": 1}),
"g1": ("INT", {"default": 0, "min": 0, "max": 255, "step": 1}),
"b1": ("INT", {"default": 0, "min": 0, "max": 255, "step": 1}),
"r2": ("INT", {"default": 255, "min": 0, "max": 255, "step": 1}),
"g2": ("INT", {"default": 255, "min": 0, "max": 255, "step": 1}),
"b2": ("INT", {"default": 255, "min": 0, "max": 255, "step": 1}),
"axis" : (["vertical","horizontal","circular"],),
"power_to": ("INT", {"default": 1, "min": 1, "max": 16, "step": 1}),
"reverse_power" : ("BOOLEAN", {"default": False}),
},
"optional":{
"mask": ("MASK",),
}
}
RETURN_TYPES = ("IMAGE","MASK",)
FUNCTION = "generate"
CATEGORY = "image"
def get_gradient_mask(self,width,height,horizontal):
if horizontal:
return torch.linspace(0, 1, width).view(1, 1, width).repeat(1, height, 1)
return torch.linspace(0, 1, height).view(1, height, 1).repeat(1, 1, width)
def generate(self, width, height, batch_size=1, r1=0, g1=0, b1=0, r2=255, g2=255, b2=255, pattern_value=1, power_to=1, reverse_power=False, axis="vertical", mask=None):
gradient = self.get_gradient_mask(width, height, axis in ["horizontal","circular"])
gradient = gradient_patterns["linear" if axis != "circular" else "sine"](gradient, pattern_value)
if axis == "circular":
gradient2 = self.get_gradient_mask(width, height, False)
gradient2 = gradient_patterns["sine"](gradient2, pattern_value)
gradient = gradient * gradient2
if power_to > 1:
if reverse_power: gradient = 1 - gradient
gradient = gradient ** power_to
if reverse_power: gradient = 1 - gradient
if mask is not None:
if mask.shape != gradient.shape:
mask = F.interpolate(mask.unsqueeze(1), size=(gradient.shape[-2], gradient.shape[-1]), mode='nearest').squeeze(1)
gradient = gradient * mask
gradient = gradient.squeeze(0).unsqueeze(-1)
r_gradient = r1 / 255.0 + gradient * (r2 - r1) / 255.0
g_gradient = g1 / 255.0 + gradient * (g2 - g1) / 255.0
b_gradient = b1 / 255.0 + gradient * (b2 - b1) / 255.0
r_image = r_gradient.expand(batch_size, height, width, 1)
g_image = g_gradient.expand(batch_size, height, width, 1)
b_image = b_gradient.expand(batch_size, height, width, 1)
rgb_image = torch.cat((r_image, g_image, b_image), dim=-1)
mask_gradient = gradient.expand(1, height, width, 1).squeeze(-1)
return (rgb_image,mask_gradient,)
def loglinear_interp(t_steps, num_steps):
"""
Performs log-linear interpolation of a given array of decreasing numbers.
"""
xs = np.linspace(0, 1, len(t_steps))
ys = np.log(t_steps[::-1])
new_xs = np.linspace(0, 1, num_steps)
new_ys = np.interp(new_xs, xs, ys)
interped_ys = np.exp(new_ys)[::-1].copy()
return interped_ys
class gradientNoisyLatentMaskBatch:
@classmethod
def INPUT_TYPES(s):
return {
"required": {
"model": ("MODEL",),
"latent": ("LATENT",),
"denoising_start": ("INT", {"default": 8, "min": 1, "max": 256, "step": 1}),
"denoising_end": ("INT", {"default": 16, "min": 1, "max": 256, "step": 1}),
"inject_noise" : (["disabled","gradient","full_for_first_batch"],),
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
"sigmas": ("SIGMAS",),
}
}
RETURN_TYPES = ("MODEL","LATENT",)
RETURN_NAMES = ("model","noisy_latents_for_sampler",)
FUNCTION = "generate"
CATEGORY = "model_patches"
def generate(self, model, latent, denoising_start, denoising_end, inject_noise, seed, sigmas):
sigma_min, sigma_max = get_sigma_min_max(model)
input_sigma_min = (sigmas[sigmas>0]).min()
samples = latent["samples"].clone()
if samples.shape[0] < denoising_end:
additional_latents = latent["samples"][-1:].repeat(denoising_end - samples.shape[0], 1, 1, 1)
samples = torch.cat((samples, additional_latents), dim=0)
if inject_noise != "disabled":
noisy_latents = prepare_noise(torch.zeros_like(samples, device=samples.device), seed)
# .unsqueeze(0)
# .repeat(samples.shape[0], 1, 1, 1)
batched_sigmas = []
if inject_noise == "gradient":
cropped_sigmas = sigmas[:-1]
for x in range(samples.shape[0]):
if x < denoising_start:
batched_sigmas.append(torch.zeros_like(sigmas).to(device=default_device))
continue
first_step = (x - denoising_start + 1) / (denoising_end - denoising_start)
batch_sigmas = loglinear_interp(cropped_sigmas.tolist(),floor(cropped_sigmas.shape[0] / first_step))[::-1][:cropped_sigmas.shape[0]][::-1]
batch_sigmas = torch.tensor(batch_sigmas)
batch_sigmas = torch.cat([batch_sigmas, torch.tensor([0.])])
sigmasf = float((batch_sigmas[0]-batch_sigmas[-1])/model.model.latent_format.scale_factor)
samples[x] = samples[x] + noisy_latents[x] * sigmasf
batched_sigmas.append(batch_sigmas.to(device=default_device))
# for cb in batched_sigmas:
# print(cb,cb.shape)
elif inject_noise == "full_for_first_batch":
sigmasf = float((sigmas[0]-sigmas[-1])/model.model.latent_format.scale_factor)
samples = (noisy_latents * sigmasf)
current_step = 0
@torch.no_grad()
def cfg_patch(args):
nonlocal current_step
cond = args["cond"]
uncond = args["uncond"]
scale = args["cond_scale"]
sigma = args["sigma"][0].item()
denoised = uncond + scale * (cond - uncond)
if inject_noise != "gradient" or sigma == 0:
return denoised
if sigma > (sigma_max - 1):
current_step = 0
for b in range(len(cond)):
if b == len(cond) - 1:
continue
print("-"*40)
print(sigma,batched_sigmas[-1][current_step].item(),batched_sigmas[b][current_step].item())
batch_ratio = batched_sigmas[b][current_step] / batched_sigmas[-1][current_step]
denoised[b] = denoised[b] * batch_ratio
# denoised[b] = (denoised[b] - sigma_min) * batch_ratio + sigma_min
# denoised[b] = (denoised[b] - input_sigma_min) * batch_ratio + input_sigma_min
current_step += 1
return denoised
m = model.clone()
m.set_model_sampler_cfg_function(cfg_patch)
return (m,{"samples":samples},)
gradient_patterns = {
"linear": lambda x, y: x,
"sine": lambda x, y: torch.sin(x * torch.pi * y),
"triangle": lambda x, y: 2 * torch.abs(torch.round(x % (1 / max(y, 1)) * y) - (x % (1 / max(y, 1)) * y)),
}
class load_latent_for_guidance:
@classmethod
def INPUT_TYPES(s):
latents_folder = os.path.join(current_dir,"latents")
latents_names = os.listdir(latents_folder)
return {"required": {
"latent_name": (latents_names,)
}}
RETURN_TYPES = ("LATENT",)
FUNCTION = "exec"
CATEGORY = "latent"
def exec(self, latent_name):
latent_path = os.path.join(current_dir,"latents",latent_name)
latent_space = torch.load(latent_path).cpu()
latent = {"samples": latent_space.unsqueeze(0)}
return (latent, )
# >>> x=torch.rand(1,50,50,3)
# >>> x.permute(0, 3, 1, 2).shape
# torch.Size([1, 3, 50, 50])
# >>> x.permute(0, 3, 1, 2).permute(0, 2, 3, 1).shape
# torch.Size([1, 50, 50, 3])
# rgb black white
sdxl_latent_full_colors = torch.tensor([
[-19.467044830322266, -19.590354919433594, 10.634221076965332, -12.227653503417969],
[-3.2475805282592773, 14.238205909729004, 26.689006805419922, 9.03557300567627],
[0.5756417512893677, 16.23186492919922, -16.991405487060547, 4.4524736404418945],
[-21.758529663085938, 3.8702831268310547, 2.311274766921997, 2.559422016143799],
[18.211639404296875, 1.7760906219482422, 9.4437255859375, -7.949795722961426]])
# def make_latent_color(lat,r,g,b):
# col = sdxl_latent_full_colors.clone()
# w = (r+g+b) / 3
# bw = col[3] * (1 - w) + col[4] * w
# lat = lat + col[0]*r + col[1] * g + col[2] * b + bw
# for b in range(lat.shape[0]):
# for c in range(lat.shape[1]):
# bw = col[3][c] * (1 - w) + col[4][c] * w
# lat[b][c] = lat[b][c] + col[0][c] * r + col[1][c] * g + col[2][c] * b + bw
# return lat
def make_latent_color(lat, r, g, b):
col = sdxl_latent_full_colors.clone()
w = (r + g + b) / 3
# bw = col[3].view(1, 4, 1, 1) * (1 - w) + col[4].view(1, 4, 1, 1) * w
# lat = lat + col[0].view(1, 4, 1, 1) * r + col[1].view(1, 4, 1, 1) * g + col[2].view(1, 4, 1, 1) * b + bw
black = col[3].view(1, 4, 1, 1) * (1 - w)
white = col[4].view(1, 4, 1, 1) * w
red = col[0].view(1, 4, 1, 1) * r
green = col[1].view(1, 4, 1, 1) * g
blue = col[2].view(1, 4, 1, 1) * b
lat = lat + (red/1.45 + green/1.45 + blue) / 3
return lat
class colors_test_node:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"batch_size": ("INT", {"default": 1, "min": 1, "max": 4096}),
"width": ("INT", {"default": 1024, "min": 0, "max": 16384, "step": 64}),
"height": ("INT", {"default": 1024, "min": 0, "max": 16384, "step": 64}),
"r": ("INT", {"default": 0, "min": 0, "max": 255, "step": 1}),
"g": ("INT", {"default": 0, "min": 0, "max": 255, "step": 1}),
"b": ("INT", {"default": 0, "min": 0, "max": 255, "step": 1}),
}}
RETURN_TYPES = ("LATENT",)
FUNCTION = "exec"
CATEGORY = "latent"
def exec(self, batch_size, width, height, r,g,b):
new_latent = torch.zeros([batch_size,4,height//8,width//8])
new_latent = make_latent_color(new_latent,r,g,b)
# latent_path = os.path.join(current_dir,"latents",latent_name)
# latent_space = torch.load(latent_path).cpu()
# latent = {"samples": latent_space.unsqueeze(0)}
return ({"samples": new_latent}, )
class latent_recombine_channels:
@classmethod
def INPUT_TYPES(s):
return {"required": {
"channel1": ("LATENT",),
"channel2": ("LATENT",),
"channel3": ("LATENT",),
"channel4": ("LATENT",),
}}
RETURN_TYPES = ("LATENT",)
FUNCTION = "exec"
CATEGORY = "latent"
def exec(self, channel1, channel2, channel3, channel4):
min_batch = min(channel1["samples"].shape[0],channel2["samples"].shape[0],
channel3["samples"].shape[0],channel4["samples"].shape[0])
latents1 = channel1["samples"].clone()
latents1[0:min_batch,1,:,:] = channel2["samples"][0:min_batch,1,:,:]
latents1[0:min_batch,2,:,:] = channel3["samples"][0:min_batch,2,:,:]
latents1[0:min_batch,3,:,:] = channel4["samples"][0:min_batch,3,:,:]
output_samples = {"samples":latents1}
return (output_samples, )
Reference in New Issue View Git Blame Copy Permalink