diff --git a/ksampler_sequence.py b/ksampler_sequence.py index e64951e..e11c11c 100644 --- a/ksampler_sequence.py +++ b/ksampler_sequence.py @@ -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)" }