Update ksampler_sequence.py
This commit is contained in:
WAS
+298
-8
@@ -4,6 +4,7 @@ import random
|
||||
import re
|
||||
import torch
|
||||
import torch.nn.functional as F
|
||||
import numpy as np
|
||||
|
||||
import comfy.sample
|
||||
import comfy.samplers
|
||||
@@ -86,11 +87,11 @@ def unsample(model, seed, cfg, sampler_name, steps, end_at_step, scheduler, norm
|
||||
noise = torch.zeros(noise_shape, dtype=latent_image.dtype, layout=latent_image.layout, device=device)
|
||||
noise_mask = comfy.sample.prepare_mask(latent.get("noise_mask"), noise, device) if "noise_mask" in latent else None
|
||||
|
||||
positive_copy = comfy.sample.broadcast_cond(positive, noise.shape[0], device)
|
||||
negative_copy = comfy.sample.broadcast_cond(negative, noise.shape[0], device)
|
||||
positive_copy = comfy.sample.convert_cond(positive)
|
||||
negative_copy = comfy.sample.convert_cond(negative)
|
||||
|
||||
models, inference_memory = comfy.sample.get_additional_models(positive, negative, model.model_dtype())
|
||||
comfy.model_management.load_models_gpu([model] + models, comfy.model_management.batch_area_memory(noise.numel() // noise.shape[0]) + inference_memory)
|
||||
comfy.model_management.load_models_gpu([model] + models, model.memory_required(noise.shape) + inference_memory)
|
||||
|
||||
real_model = model.model
|
||||
sampler = comfy.samplers.KSampler(real_model, steps=steps, device=device, sampler=sampler_name, scheduler=scheduler, denoise=1.0, model_options=model.model_options)
|
||||
@@ -163,18 +164,98 @@ class CLIPTextEncodeSequence:
|
||||
|
||||
return (conditionings, )
|
||||
|
||||
class CLIPTextEncodeSequence2:
|
||||
@classmethod
|
||||
def INPUT_TYPES(cls):
|
||||
return {
|
||||
"required": {
|
||||
"clip": ("CLIP", ),
|
||||
"token_normalization": (["none", "mean", "length", "length+mean"],),
|
||||
"weight_interpretation": (["comfy", "A1111", "compel", "comfy++"],),
|
||||
"cond_keyframes_type": (["linear", "sinus", "sinus_inverted", "half_sinus", "half_sinus_inverted"],),
|
||||
"frame_count": ("INT", {"default": 100, "min": 1, "max": 1024, "step": 1}),
|
||||
"text": ("STRING", {"multiline": True, "default": '''A portrait of a rosebud
|
||||
A portrait of a blooming rosebud
|
||||
A portrait of a blooming rose
|
||||
A portrait of a rose'''}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("CONDITIONING", "INT", "INT")
|
||||
RETURN_NAMES = ("conditioning_sequence", "cond_keyframes", "frame_count")
|
||||
IS_LIST_OUTPUT = (True, True, False)
|
||||
|
||||
FUNCTION = "encode"
|
||||
CATEGORY = "conditioning"
|
||||
|
||||
def encode(self, clip, text, cond_keyframes_type, frame_count, token_normalization, weight_interpretation):
|
||||
text = text.strip()
|
||||
conditionings = []
|
||||
for line in text.splitlines():
|
||||
if USE_BLK:
|
||||
encoded = blk_adv.encode(clip=clip, text=line, token_normalization=token_normalization, weight_interpretation=weight_interpretation)
|
||||
else:
|
||||
encoded = CLIPTextEncode.encode(clip=clip, text=line)
|
||||
|
||||
conditionings.append([encoded[0][0][0], encoded[0][0][1]])
|
||||
|
||||
conditioning_count = len(conditionings)
|
||||
cond_keyframes = self.calculate_cond_keyframes(cond_keyframes_type, frame_count, conditioning_count)
|
||||
|
||||
return (conditionings, cond_keyframes, frame_count)
|
||||
|
||||
def calculate_cond_keyframes(self, type, frame_count, conditioning_count):
|
||||
if type == "linear":
|
||||
return np.linspace(frame_count // conditioning_count, frame_count, conditioning_count, dtype=int).tolist()
|
||||
|
||||
elif type == "sinus":
|
||||
# Create a sinusoidal distribution
|
||||
t = np.linspace(0, np.pi, conditioning_count)
|
||||
sinus_values = np.sin(t)
|
||||
# Normalize the sinusoidal values to 0-1 range
|
||||
normalized_values = (sinus_values - sinus_values.min()) / (sinus_values.max() - sinus_values.min())
|
||||
# Scale to frame count and shift to avoid starting at frame 0
|
||||
scaled_values = normalized_values * (frame_count - 1) + 1
|
||||
# Ensure unique keyframes by rounding and converting to integer
|
||||
unique_keyframes = np.round(scaled_values).astype(int)
|
||||
# Deduplicate while preserving order
|
||||
unique_keyframes = np.unique(unique_keyframes, return_index=True)[1]
|
||||
return sorted(unique_keyframes.tolist())
|
||||
|
||||
elif type == "sinus_inverted":
|
||||
return (np.cos(np.linspace(0, np.pi, conditioning_count)) * (frame_count - 1) + 1).astype(int).tolist()
|
||||
|
||||
elif type == "half_sinus":
|
||||
return (np.sin(np.linspace(0, np.pi / 2, conditioning_count)) * (frame_count - 1) + 1).astype(int).tolist()
|
||||
|
||||
elif type == "half_sinus_inverted":
|
||||
return (np.cos(np.linspace(0, np.pi / 2, conditioning_count)) * (frame_count - 1) + 1).astype(int).tolist()
|
||||
|
||||
else:
|
||||
raise ValueError("Unsupported cond_keyframes_type: " + type)
|
||||
|
||||
class KSamplerSeq:
|
||||
|
||||
def __init__(self):
|
||||
self.previous_seed = None
|
||||
self.current_seed = None
|
||||
|
||||
def initialize_seeds(self, initial_seed):
|
||||
self.previous_seed = initial_seed
|
||||
self.current_seed = initial_seed
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {"required":
|
||||
{"model": ("MODEL",),
|
||||
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
|
||||
"seed_mode_seq": (["increment", "decrement", "random", "fixed"],),
|
||||
"alternate_values": ("BOOLEAN", {"default": True}),
|
||||
"steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
|
||||
"cfg": ("FLOAT", {"default": 8.0, "min": 0.0, "max": 100.0, "step":0.5, "round": 0.01}),
|
||||
"sampler_name": (comfy.samplers.KSampler.SAMPLERS, ),
|
||||
"scheduler": (comfy.samplers.KSampler.SCHEDULERS, ),
|
||||
"sequence_loop_count": ("INT", {"default": 20, "min": 1, "max": 100, "step": 1}),
|
||||
"sequence_loop_count": ("INT", {"default": 20, "min": 1, "max": 1024, "step": 1}),
|
||||
"positive_seq": ("CONDITIONING_SEQ", ),
|
||||
"negative_seq": ("CONDITIONING_SEQ", ),
|
||||
"use_conditioning_slerp": ("BOOLEAN", {"default": False}),
|
||||
@@ -216,16 +297,33 @@ class KSamplerSeq:
|
||||
break
|
||||
return matching_conditioning if matching_conditioning else (last_conditioning if last_conditioning else None)
|
||||
|
||||
def sample(self, model, seed, seed_mode_seq, steps, cfg, sampler_name, scheduler, sequence_loop_count, positive_seq, negative_seq, cond_slerp_strength, latent_image, use_latent_interpolation, latent_interpolation_mode, latent_interp_strength, denoise_start=1.0, denoise_seq=0.5, use_conditioning_slerp=False, unsample_latents=False):
|
||||
def update_alternate_seed(self, loop_count):
|
||||
if loop_count % 3 == 0:
|
||||
if self.previous_seed is None:
|
||||
self.previous_seed = self.current_seed
|
||||
else:
|
||||
self.previous_seed, self.current_seed = self.current_seed, self.previous_seed + 1 if loop_count // 2 % 2 == 0 else self.previous_seed - 1
|
||||
return self.current_seed
|
||||
|
||||
def alternate_denoise(self, current_denoise):
|
||||
return 0.95 if current_denoise == 0.75 else 0.75
|
||||
|
||||
def sample(self, model, seed, seed_mode_seq, alternate_values, steps, cfg, sampler_name, scheduler, sequence_loop_count, positive_seq, negative_seq, cond_slerp_strength, latent_image, use_latent_interpolation, latent_interpolation_mode, latent_interp_strength, denoise_start=1.0, denoise_seq=0.5, use_conditioning_slerp=False, unsample_latents=False, alternate_mode=False):
|
||||
positive_seq = positive_seq
|
||||
negative_seq = negative_seq
|
||||
|
||||
results = []
|
||||
positive_conditioning = None
|
||||
negative_conditioning = None
|
||||
for loop_count in range(sequence_loop_count):
|
||||
|
||||
seq_seed = seed if loop_count <= 0 else self.update_seed(seq_seed, seed_mode_seq)
|
||||
self.initialize_seeds(seed)
|
||||
|
||||
for loop_count in range(sequence_loop_count):
|
||||
if alternate_values and loop_count % 2 == 0:
|
||||
seq_seed = self.update_alternate_seed(seed) if seed_mode_seq != "fixed" else seed
|
||||
#denoise_seq = self.alternate_denoise(denoise_seq)
|
||||
else:
|
||||
seq_seed = seed if loop_count <= 0 else self.update_seed(seq_seed, seed_mode_seq)
|
||||
|
||||
|
||||
print(f"Loop count: {loop_count}, Seed: {seq_seed}")
|
||||
|
||||
@@ -295,15 +393,207 @@ class KSamplerSeq:
|
||||
return (results,)
|
||||
|
||||
|
||||
class KSamplerSeq2:
|
||||
|
||||
def __init__(self):
|
||||
self.previous_seed = None
|
||||
self.current_seed = None
|
||||
|
||||
def initialize_seeds(self, initial_seed):
|
||||
self.previous_seed = initial_seed
|
||||
self.current_seed = initial_seed
|
||||
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {"required":
|
||||
{"model": ("MODEL",),
|
||||
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
|
||||
"seed_mode_seq": (["increment", "decrement", "random", "fixed"],),
|
||||
"alternate_values": ("BOOLEAN", {"default": True}),
|
||||
"steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
|
||||
"cfg": ("FLOAT", {"default": 8.0, "min": 0.0, "max": 100.0, "step":0.5, "round": 0.01}),
|
||||
"sampler_name": (comfy.samplers.KSampler.SAMPLERS, ),
|
||||
"scheduler": (comfy.samplers.KSampler.SCHEDULERS, ),
|
||||
"frame_count": ("INT", {"default": 0, "min": 0, "max": 1024, "step": 1}),
|
||||
"cond_keyframes": ("INT", {"default": 0, "min": 0, "max": 1024, "step": 1}),
|
||||
"positive_seq": ("CONDITIONING", ),
|
||||
"negative_seq": ("CONDITIONING", ),
|
||||
"use_conditioning_slerp": ("BOOLEAN", {"default": False}),
|
||||
"cond_slerp_strength": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.001}),
|
||||
"latent_image": ("LATENT", ),
|
||||
"use_latent_interpolation": ("BOOLEAN", {"default": False}),
|
||||
"latent_interpolation_mode": (["Blend", "Slerp", "Cosine Interp"],),
|
||||
"latent_interp_strength": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.001}),
|
||||
"denoise_start": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
|
||||
"denoise_seq": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.01}),
|
||||
"unsample_latents": ("BOOLEAN", {"default": False}),
|
||||
"inject_noise": ("BOOLEAN", {"default": True}),
|
||||
"noise_strength": ("FLOAT", {"default": 0.1, "max": 1.0, "min": 0.001, "step": 0.001}),
|
||||
"denoise_sine": ("BOOLEAN", {"default": True}),
|
||||
"denoise_max": ("FLOAT", {"default": 0.9, "max": 1.0, "min": 0.0, "step": 0.001}),
|
||||
"seed_keying": ("BOOLEAN", {"default": True}),
|
||||
"seed_keying_mode": (["sine", "modulo"],),
|
||||
"seed_divisor": ("INT", {"default": 4, "max": 1024, "min": 2, "step": 1}),
|
||||
}
|
||||
}
|
||||
|
||||
RETURN_TYPES = ("LATENT",)
|
||||
FUNCTION = "sample"
|
||||
|
||||
CATEGORY = "sampling"
|
||||
|
||||
def update_seed(self, seed, seed_mode):
|
||||
if seed_mode == "increment":
|
||||
return seed + 1
|
||||
elif seed_mode == "decrement":
|
||||
return seed - 1
|
||||
elif seed_mode == "random":
|
||||
return random.randint(0, 0xffffffffffffffff)
|
||||
elif seed_mode == "fixed":
|
||||
return seed
|
||||
|
||||
def alternate_seed_modulo(self, current, seed, divisor):
|
||||
if current % divisor == 0:
|
||||
new_seed = (seed + current) % 0xffffffffffffffff
|
||||
else:
|
||||
new_seed = seed
|
||||
return new_seed
|
||||
|
||||
def alternate_seed_sine(self, current, start_seed, divisor):
|
||||
seed = 1000 * np.sin(2 * math.pi * current / divisor) + start_seed
|
||||
return seed
|
||||
|
||||
def alternate_denoise(self, curent, total, start_denoise=0.5, max_denoise=0.95):
|
||||
amplitude = (max_denoise - start_denoise) / 2
|
||||
mid_point = (max_denoise + start_denoise) / 2
|
||||
cycle_position = (math.pi * 2 * curent) / total
|
||||
current_denoise = amplitude * math.sin(cycle_position) + mid_point
|
||||
return current_denoise
|
||||
|
||||
def inject_noise(self, latent_image, noise_strength):
|
||||
noise = torch.randn_like(latent_image) * noise_strength
|
||||
return latent_image + noise
|
||||
|
||||
def sample(self, model, seed, seed_mode_seq, alternate_values, steps, cfg, sampler_name, scheduler,
|
||||
frame_count, cond_keyframes, positive_seq, negative_seq, cond_slerp_strength, latent_image,
|
||||
use_latent_interpolation, latent_interpolation_mode, latent_interp_strength, denoise_start=1.0,
|
||||
denoise_seq=0.5, use_conditioning_slerp=False, unsample_latents=False, alternate_mode=False,
|
||||
inject_noise=True, noise_strength=0.1, denoise_sine=True, denoise_max=0.9, seed_keying=True,
|
||||
seed_keying_mode="sine", seed_divisor=4):
|
||||
|
||||
if not isinstance(positive_seq, list):
|
||||
positive_seq = [positive_seq]
|
||||
if not isinstance(negative_seq, list):
|
||||
negative_seq = [negative_seq]
|
||||
if not isinstance(cond_keyframes, list):
|
||||
cond_keyframes = [cond_keyframes]
|
||||
cond_keyframes.sort()
|
||||
|
||||
positive_cond_idx = 0
|
||||
negative_cond_idx = 0
|
||||
results = []
|
||||
|
||||
self.initialize_seeds(seed)
|
||||
sequence_loop_count = max(frame_count, len(positive_seq)) if cond_keyframes else len(positive_seq)
|
||||
|
||||
print(f"Starting loop sequence with {sequence_loop_count} frames.")
|
||||
print(f"Using {len(positive_seq)} positive conditionings and {len(negative_seq)} negative conditionings")
|
||||
print(f"Conditioning keyframe schedule is: {', '.join(map(str, cond_keyframes))}")
|
||||
|
||||
|
||||
for loop_count in range(sequence_loop_count):
|
||||
if loop_count in cond_keyframes:
|
||||
positive_cond_idx = min(positive_cond_idx + 1, len(positive_seq) - 1)
|
||||
negative_cond_idx = min(negative_cond_idx + 1, len(negative_seq) - 1)
|
||||
|
||||
positive_conditioning = positive_seq[positive_cond_idx]
|
||||
negative_conditioning = negative_seq[negative_cond_idx]
|
||||
|
||||
if seed_keying:
|
||||
if seed_keying_mode == "sine":
|
||||
seq_seed = seed if loop_count <= 0 else self.alternate_seed_sine(loop_count, seed, seed_divisor)
|
||||
else:
|
||||
seq_seed = seed if loop_count <= 0 else self.alternate_seed_modulo(loop_count, seed, seed_divisor)
|
||||
else:
|
||||
seq_seed = seed if loop_count <= 0 else self.update_seed(seq_seed, seed_mode_seq)
|
||||
|
||||
seq_seed = seed if loop_count <= 0 else self.update_seed(seq_seed, seed_mode_seq)
|
||||
print(f"Loop count: {loop_count}, Seed: {seq_seed}")
|
||||
|
||||
last_positive_conditioning = positive_conditioning if positive_conditioning else None
|
||||
last_negative_conditioning = negative_conditioning if negative_conditioning else None
|
||||
|
||||
if use_conditioning_slerp and (last_positive_conditioning and last_negative_conditioning):
|
||||
a, b = last_positive_conditioning[0].clone(), positive_conditioning[0].clone()
|
||||
na, nb = last_negative_conditioning[0].clone(), negative_conditioning[0].clone()
|
||||
pa, pb = last_positive_conditioning[1]["pooled_output"].clone(), positive_conditioning[1]["pooled_output"].clone()
|
||||
npa, npb = last_negative_conditioning[1]["pooled_output"].clone(), negative_conditioning[1]["pooled_output"].clone()
|
||||
pos_cond = slerp(cond_slerp_strength, a, b)
|
||||
pos_pooled = slerp(cond_slerp_strength, pa, pb)
|
||||
neg_cond = slerp(cond_slerp_strength, na, nb)
|
||||
neg_pooled = slerp(cond_slerp_strength, npa, npb)
|
||||
positive_conditioning = [pos_cond, {"pooled_output": pos_pooled}]
|
||||
negative_conditioning = [neg_cond, {"pooled_output": neg_pooled}]
|
||||
|
||||
positive_conditioning = [positive_conditioning]
|
||||
negative_conditioning = [negative_conditioning]
|
||||
|
||||
end_at_step = steps
|
||||
if results and len(results) > 0:
|
||||
latent_input = {'samples': results[-1]}
|
||||
denoise = self.alternate_denoise(loop_count, sequence_loop_count, denoise_seq, denoise_max) if denoise_sine else denoise_seq
|
||||
start_at_step = round((1 - denoise) * steps)
|
||||
end_at_step = steps
|
||||
else:
|
||||
latent_input = latent_image
|
||||
denoise = denoise_start
|
||||
|
||||
if unsample_latents and loop_count > 0:
|
||||
force_full_denoise = not (loop_count > 0 or loop_count <= steps - 1)
|
||||
disable_noise = False
|
||||
if seed_keying:
|
||||
if seed_keying_mode == "modulo" and loop_count % seed_divisor == 0:
|
||||
unsampled_latent = latent_input
|
||||
else:
|
||||
unsampled_latent = unsample(model=model, seed=seq_seed, cfg=cfg, sampler_name=sampler_name, steps=steps, end_at_step=end_at_step, scheduler=scheduler, normalize=False, positive=positive_conditioning, negative=negative_conditioning, latent_image=latent_input)[0]
|
||||
else:
|
||||
unsampled_latent = unsample(model=model, seed=seq_seed, cfg=cfg, sampler_name=sampler_name, steps=steps, end_at_step=end_at_step, scheduler=scheduler, normalize=False, positive=positive_conditioning, negative=negative_conditioning, latent_image=latent_input)[0]
|
||||
if inject_noise and loop_count > 0:
|
||||
print(f"Injecting noise at {noise_strength} strength.")
|
||||
unsampled_latent['samples'] = self.inject_noise(unsampled_latent['samples'], noise_strength)
|
||||
sample = nodes.common_ksampler(model, seq_seed, steps, cfg, sampler_name, scheduler, positive_conditioning, negative_conditioning, unsampled_latent, denoise=denoise, disable_noise=disable_noise, start_step=start_at_step, last_step=end_at_step, force_full_denoise=force_full_denoise)[0]['samples']
|
||||
else:
|
||||
if inject_noise and loop_count > 0:
|
||||
print(f"Injecting noise at {noise_strength} strength.")
|
||||
latent_input['samples'] = self.inject_noise(latent_input['samples'], noise_strength)
|
||||
sample = nodes.common_ksampler(model, seq_seed, steps, cfg, sampler_name, scheduler, positive_conditioning, negative_conditioning, latent_input, denoise=denoise)[0]['samples']
|
||||
|
||||
if use_latent_interpolation and results and loop_count > 0:
|
||||
if latent_interpolation_mode == "Blend":
|
||||
sample = blend_latents(latent_interp_strength, results[-1], sample)
|
||||
elif latent_interpolation_mode == "Slerp":
|
||||
sample = slerp_latents(latent_interp_strength, results[-1], sample)
|
||||
elif latent_interpolation_mode == "Cosine Interp":
|
||||
sample = cosine_interp_latents(latent_interp_strength, results[-1], sample)
|
||||
|
||||
results.append(sample)
|
||||
|
||||
results = torch.cat(results, dim=0)
|
||||
results = {'samples': results}
|
||||
return (results,)
|
||||
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"CLIPTextEncodeList": CLIPTextEncodeSequence,
|
||||
"CLIPTextEncodeSequence2": CLIPTextEncodeSequence2,
|
||||
"KSamplerSeq": KSamplerSeq,
|
||||
"KSamplerSeq2": KSamplerSeq2,
|
||||
}
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"CLIPTextEncodeList": "CLIP Text Encode Sequence (Advanced)",
|
||||
"CLIPTextEncodeSequence2": "CLIP Text Encode Sequence (v2)",
|
||||
"KSamplerSeq": "KSampler Sequence",
|
||||
"KSamplerSeq2": "KSampler Sequence (v2)"
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user