| import torch
|
| import numpy as np
|
| import comfy.utils, comfy.sample, comfy.samplers, comfy.controlnet, comfy.model_base, comfy.model_management, comfy.sampler_helpers, comfy.supported_models
|
| from comfy.model_patcher import ModelPatcher
|
|
|
| from nodes import RepeatLatentBatch, CLIPTextEncode, VAEEncodeForInpaint
|
| from ..modules.layer_diffuse import LayerMethod
|
| from ..config import *
|
|
|
| from .. import easyCache, sampler
|
|
|
|
|
|
|
| class samplerSettings:
|
|
|
| def __init__(self):
|
| pass
|
|
|
| @classmethod
|
| def INPUT_TYPES(cls):
|
| return {"required":
|
| {"pipe": ("PIPE_LINE",),
|
| "steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
|
| "cfg": ("FLOAT", {"default": 8.0, "min": 0.0, "max": 100.0}),
|
| "sampler_name": (comfy.samplers.KSampler.SAMPLERS,),
|
| "scheduler": (comfy.samplers.KSampler.SCHEDULERS + NEW_SCHEDULERS,),
|
| "denoise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
|
| "seed": ("INT", {"default": 0, "min": 0, "max": MAX_SEED_NUM}),
|
| },
|
| "optional": {
|
| "image_to_latent": ("IMAGE",),
|
| "latent": ("LATENT",),
|
| },
|
| "hidden":
|
| {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "my_unique_id": "UNIQUE_ID"},
|
| }
|
|
|
| RETURN_TYPES = ("PIPE_LINE", )
|
| RETURN_NAMES = ("pipe",)
|
|
|
| FUNCTION = "settings"
|
| CATEGORY = "EasyUse/PreSampling"
|
|
|
| def settings(self, pipe, steps, cfg, sampler_name, scheduler, denoise, seed, image_to_latent=None, latent=None, prompt=None, extra_pnginfo=None, my_unique_id=None):
|
|
|
| vae = pipe["vae"]
|
| batch_size = pipe["loader_settings"]["batch_size"] if "batch_size" in pipe["loader_settings"] else 1
|
| if image_to_latent is not None:
|
| _, height, width, _ = image_to_latent.shape
|
| if height == 1 and width == 1:
|
| samples = pipe["samples"]
|
| images = pipe["images"]
|
| else:
|
| samples = {"samples": vae.encode(image_to_latent[:, :, :, :3])}
|
| samples = RepeatLatentBatch().repeat(samples, batch_size)[0]
|
| images = image_to_latent
|
| elif latent is not None:
|
| samples = latent
|
| images = pipe["images"]
|
| else:
|
| samples = pipe["samples"]
|
| images = pipe["images"]
|
|
|
| new_pipe = {
|
| "model": pipe['model'],
|
| "positive": pipe['positive'],
|
| "negative": pipe['negative'],
|
| "vae": pipe['vae'],
|
| "clip": pipe['clip'],
|
|
|
| "samples": samples,
|
| "images": images,
|
| "seed": seed,
|
|
|
| "loader_settings": {
|
| **pipe["loader_settings"],
|
| "steps": steps,
|
| "cfg": cfg,
|
| "sampler_name": sampler_name,
|
| "scheduler": scheduler,
|
| "denoise": denoise,
|
| "add_noise": "enabled"
|
| }
|
| }
|
|
|
| del pipe
|
|
|
| return {"ui": {"value": [seed]}, "result": (new_pipe,)}
|
|
|
|
|
| class samplerSettingsAdvanced:
|
|
|
| def __init__(self):
|
| pass
|
|
|
| @classmethod
|
| def INPUT_TYPES(cls):
|
| return {"required":
|
| {"pipe": ("PIPE_LINE",),
|
| "steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
|
| "cfg": ("FLOAT", {"default": 8.0, "min": 0.0, "max": 100.0}),
|
| "sampler_name": (comfy.samplers.KSampler.SAMPLERS,),
|
| "scheduler": (comfy.samplers.KSampler.SCHEDULERS + NEW_SCHEDULERS,),
|
| "start_at_step": ("INT", {"default": 0, "min": 0, "max": 10000}),
|
| "end_at_step": ("INT", {"default": 10000, "min": 0, "max": 10000}),
|
| "add_noise": (["enable (CPU)", "enable (GPU=A1111)", "disable"], {"default": "enable (CPU)"}),
|
| "seed": ("INT", {"default": 0, "min": 0, "max": MAX_SEED_NUM}),
|
| "return_with_leftover_noise": (["disable", "enable"], ),
|
| },
|
| "optional": {
|
| "image_to_latent": ("IMAGE",),
|
| "latent": ("LATENT",)
|
| },
|
| "hidden":
|
| {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "my_unique_id": "UNIQUE_ID"},
|
| }
|
|
|
| RETURN_TYPES = ("PIPE_LINE", )
|
| RETURN_NAMES = ("pipe",)
|
|
|
| FUNCTION = "settings"
|
| CATEGORY = "EasyUse/PreSampling"
|
|
|
| def settings(self, pipe, steps, cfg, sampler_name, scheduler, start_at_step, end_at_step, add_noise, seed, return_with_leftover_noise, image_to_latent=None, latent=None, prompt=None, extra_pnginfo=None, my_unique_id=None):
|
|
|
| vae = pipe["vae"]
|
| batch_size = pipe["loader_settings"]["batch_size"] if "batch_size" in pipe["loader_settings"] else 1
|
| if image_to_latent is not None:
|
| _, height, width, _ = image_to_latent.shape
|
| if height == 1 and width == 1:
|
| samples = pipe["samples"]
|
| images = pipe["images"]
|
| else:
|
| samples = {"samples": vae.encode(image_to_latent[:, :, :, :3])}
|
| samples = RepeatLatentBatch().repeat(samples, batch_size)[0]
|
| images = image_to_latent
|
| elif latent is not None:
|
| samples = latent
|
| images = pipe["images"]
|
| else:
|
| samples = pipe["samples"]
|
| images = pipe["images"]
|
|
|
| force_full_denoise = True
|
| if return_with_leftover_noise == "enable":
|
| force_full_denoise = False
|
|
|
| new_pipe = {
|
| "model": pipe['model'],
|
| "positive": pipe['positive'],
|
| "negative": pipe['negative'],
|
| "vae": pipe['vae'],
|
| "clip": pipe['clip'],
|
|
|
| "samples": samples,
|
| "images": images,
|
| "seed": seed,
|
|
|
| "loader_settings": {
|
| **pipe["loader_settings"],
|
| "steps": steps,
|
| "cfg": cfg,
|
| "sampler_name": sampler_name,
|
| "scheduler": scheduler,
|
| "start_step": start_at_step,
|
| "last_step": end_at_step,
|
| "denoise": 1.0,
|
| "add_noise": add_noise,
|
| "force_full_denoise": force_full_denoise
|
| }
|
| }
|
|
|
| del pipe
|
|
|
| return {"ui": {"value": [seed]}, "result": (new_pipe,)}
|
|
|
|
|
| class samplerSettingsNoiseIn:
|
|
|
| def __init__(self):
|
| pass
|
|
|
| @classmethod
|
| def INPUT_TYPES(cls):
|
| return {"required":
|
| {"pipe": ("PIPE_LINE",),
|
| "factor": ("FLOAT", {"default": 0.1, "min": 0.0, "max": 1.0, "step":0.01, "round": 0.01}),
|
| "steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
|
| "cfg": ("FLOAT", {"default": 8.0, "min": 0.0, "max": 100.0}),
|
| "sampler_name": (comfy.samplers.KSampler.SAMPLERS,),
|
| "scheduler": (comfy.samplers.KSampler.SCHEDULERS+NEW_SCHEDULERS,),
|
| "denoise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
|
| "seed": ("INT", {"default": 0, "min": 0, "max": MAX_SEED_NUM}),
|
| },
|
| "optional": {
|
| "optional_noise_seed": ("INT",{"forceInput": True}),
|
| "optional_latent": ("LATENT",),
|
| },
|
| "hidden":
|
| {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "my_unique_id": "UNIQUE_ID"},
|
| }
|
|
|
| RETURN_TYPES = ("PIPE_LINE", )
|
| RETURN_NAMES = ("pipe",)
|
|
|
| FUNCTION = "settings"
|
| CATEGORY = "EasyUse/PreSampling"
|
|
|
| def slerp(self, val, low, high):
|
| dims = low.shape
|
|
|
| low = low.reshape(dims[0], -1)
|
| high = high.reshape(dims[0], -1)
|
|
|
| low_norm = low / torch.norm(low, dim=1, keepdim=True)
|
| high_norm = high / torch.norm(high, dim=1, keepdim=True)
|
|
|
| low_norm[low_norm != low_norm] = 0.0
|
| high_norm[high_norm != high_norm] = 0.0
|
|
|
| omega = torch.acos((low_norm * high_norm).sum(1))
|
| so = torch.sin(omega)
|
| res = (torch.sin((1.0 - val) * omega) / so).unsqueeze(1) * low + (torch.sin(val * omega) / so).unsqueeze(
|
| 1) * high
|
|
|
| return res.reshape(dims)
|
|
|
| def prepare_mask(self, mask, shape):
|
| mask = torch.nn.functional.interpolate(mask.reshape((-1, 1, mask.shape[-2], mask.shape[-1])),
|
| size=(shape[2], shape[3]), mode="bilinear")
|
| mask = mask.expand((-1, shape[1], -1, -1))
|
| if mask.shape[0] < shape[0]:
|
| mask = mask.repeat((shape[0] - 1) // mask.shape[0] + 1, 1, 1, 1)[:shape[0]]
|
| return mask
|
|
|
| def expand_mask(self, mask, expand, tapered_corners):
|
| try:
|
| import scipy
|
|
|
| c = 0 if tapered_corners else 1
|
| kernel = np.array([[c, 1, c],
|
| [1, 1, 1],
|
| [c, 1, c]])
|
| mask = mask.reshape((-1, mask.shape[-2], mask.shape[-1]))
|
| out = []
|
| for m in mask:
|
| output = m.numpy()
|
| for _ in range(abs(expand)):
|
| if expand < 0:
|
| output = scipy.ndimage.grey_erosion(output, footprint=kernel)
|
| else:
|
| output = scipy.ndimage.grey_dilation(output, footprint=kernel)
|
| output = torch.from_numpy(output)
|
| out.append(output)
|
|
|
| return torch.stack(out, dim=0)
|
| except:
|
| return None
|
|
|
| def settings(self, pipe, factor, steps, cfg, sampler_name, scheduler, denoise, seed, optional_noise_seed=None, optional_latent=None, prompt=None, extra_pnginfo=None, my_unique_id=None):
|
| latent = optional_latent if optional_latent is not None else pipe["samples"]
|
| model = pipe["model"]
|
|
|
|
|
| batch_size, _, height, width = latent["samples"].shape
|
| generator = torch.manual_seed(seed)
|
| base_noise = torch.randn((1, 4, height, width), dtype=torch.float32, device="cpu", generator=generator).repeat(batch_size, 1, 1, 1).cpu()
|
|
|
|
|
| if optional_noise_seed is None or optional_noise_seed == seed:
|
| optional_noise_seed = seed+1
|
| generator = torch.manual_seed(optional_noise_seed)
|
| variation_noise = torch.randn((batch_size, 4, height, width), dtype=torch.float32, device="cpu",
|
| generator=generator).cpu()
|
|
|
| slerp_noise = self.slerp(factor, base_noise, variation_noise)
|
|
|
| end_at_step = steps
|
| start_at_step = round(end_at_step - end_at_step * denoise)
|
|
|
| device = comfy.model_management.get_torch_device()
|
| comfy.model_management.load_model_gpu(model)
|
| model_patcher = comfy.model_patcher.ModelPatcher(model.model, load_device=device, offload_device=comfy.model_management.unet_offload_device())
|
| sampler = comfy.samplers.KSampler(model_patcher, steps=steps, device=device, sampler=sampler_name,
|
| scheduler=scheduler, denoise=1.0, model_options=model.model_options)
|
| sigmas = sampler.sigmas
|
| sigma = sigmas[start_at_step] - sigmas[end_at_step]
|
| sigma /= model.model.latent_format.scale_factor
|
| sigma = sigma.cpu().numpy()
|
|
|
| work_latent = latent.copy()
|
| work_latent["samples"] = latent["samples"].clone() + slerp_noise * sigma
|
|
|
| if "noise_mask" in latent:
|
| noise_mask = self.prepare_mask(latent["noise_mask"], latent['samples'].shape)
|
| work_latent["samples"] = noise_mask * work_latent["samples"] + (1-noise_mask) * latent["samples"]
|
| work_latent['noise_mask'] = self.expand_mask(latent["noise_mask"].clone(), 5, True)
|
|
|
| if pipe is None:
|
| pipe = {}
|
|
|
| new_pipe = {
|
| "model": pipe['model'],
|
| "positive": pipe['positive'],
|
| "negative": pipe['negative'],
|
| "vae": pipe['vae'],
|
| "clip": pipe['clip'],
|
|
|
| "samples": work_latent,
|
| "images": pipe['images'],
|
| "seed": seed,
|
|
|
| "loader_settings": {
|
| **pipe["loader_settings"],
|
| "steps": steps,
|
| "cfg": cfg,
|
| "sampler_name": sampler_name,
|
| "scheduler": scheduler,
|
| "denoise": denoise,
|
| "add_noise": "disable"
|
| }
|
| }
|
|
|
| return (new_pipe,)
|
|
|
|
|
| class samplerCustomSettings:
|
|
|
| def __init__(self):
|
| pass
|
|
|
| @classmethod
|
| def INPUT_TYPES(cls):
|
| return {"required": {
|
| "pipe": ("PIPE_LINE",),
|
| "guider": (['CFG','DualCFG','Basic', 'IP2P+CFG', 'IP2P+DualCFG','IP2P+Basic'],{"default":"Basic"}),
|
| "cfg": ("FLOAT", {"default": 3.5, "min": 0.0, "max": 100.0}),
|
| "cfg_negative": ("FLOAT", {"default": 1.5, "min": 0.0, "max": 100.0}),
|
| "sampler_name": (comfy.samplers.KSampler.SAMPLERS + ['inversed_euler'],),
|
| "scheduler": (comfy.samplers.KSampler.SCHEDULERS + ['karrasADV','exponentialADV','polyExponential', 'sdturbo', 'vp', 'alignYourSteps', 'gits'],),
|
| "coeff": ("FLOAT", {"default": 1.20, "min": 0.80, "max": 1.50, "step": 0.05}),
|
| "steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
|
| "sigma_max": ("FLOAT", {"default": 14.614642, "min": 0.0, "max": 1000.0, "step": 0.01, "round": False}),
|
| "sigma_min": ("FLOAT", {"default": 0.0291675, "min": 0.0, "max": 1000.0, "step": 0.01, "round": False}),
|
| "rho": ("FLOAT", {"default": 7.0, "min": 0.0, "max": 100.0, "step": 0.01, "round": False}),
|
| "beta_d": ("FLOAT", {"default": 19.9, "min": 0.0, "max": 1000.0, "step": 0.01, "round": False}),
|
| "beta_min": ("FLOAT", {"default": 0.1, "min": 0.0, "max": 1000.0, "step": 0.01, "round": False}),
|
| "eps_s": ("FLOAT", {"default": 0.001, "min": 0.0, "max": 1.0, "step": 0.0001, "round": False}),
|
| "flip_sigmas": ("BOOLEAN", {"default": False}),
|
| "denoise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
|
| "add_noise": (["enable (CPU)", "enable (GPU=A1111)", "disable"], {"default": "enable (CPU)"}),
|
| "seed": ("INT", {"default": 0, "min": 0, "max": MAX_SEED_NUM}),
|
| },
|
| "optional": {
|
| "image_to_latent": ("IMAGE",),
|
| "latent": ("LATENT",),
|
| "optional_sampler":("SAMPLER",),
|
| "optional_sigmas":("SIGMAS",),
|
| },
|
| "hidden":
|
| {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "my_unique_id": "UNIQUE_ID"},
|
| }
|
|
|
| RETURN_TYPES = ("PIPE_LINE", )
|
| RETURN_NAMES = ("pipe",)
|
|
|
| FUNCTION = "settings"
|
| CATEGORY = "EasyUse/PreSampling"
|
|
|
| def ip2p(self, positive, negative, vae, pixels, latent=None):
|
| if latent is not None:
|
| concat_latent = latent
|
| else:
|
| x = (pixels.shape[1] // 8) * 8
|
| y = (pixels.shape[2] // 8) * 8
|
|
|
| if pixels.shape[1] != x or pixels.shape[2] != y:
|
| x_offset = (pixels.shape[1] % 8) // 2
|
| y_offset = (pixels.shape[2] % 8) // 2
|
| pixels = pixels[:,x_offset:x + x_offset, y_offset:y + y_offset,:]
|
|
|
| concat_latent = vae.encode(pixels)
|
|
|
| out_latent = {}
|
| out_latent["samples"] = torch.zeros_like(concat_latent)
|
|
|
| out = []
|
| for conditioning in [positive, negative]:
|
| c = []
|
| for t in conditioning:
|
| d = t[1].copy()
|
| d["concat_latent_image"] = concat_latent
|
| n = [t[0], d]
|
| c.append(n)
|
| out.append(c)
|
| return (out[0], out[1], out_latent)
|
|
|
|
|
| def settings(self, pipe, guider, cfg, cfg_negative, sampler_name, scheduler, coeff, steps, sigma_max, sigma_min, rho, beta_d, beta_min, eps_s, flip_sigmas, denoise, add_noise, seed, image_to_latent=None, latent=None, optional_sampler=None, optional_sigmas=None, prompt=None, extra_pnginfo=None, my_unique_id=None):
|
|
|
|
|
| vae = pipe["vae"]
|
| model = pipe["model"]
|
| positive = pipe['positive']
|
| negative = pipe['negative']
|
| batch_size = pipe["loader_settings"]["batch_size"] if "batch_size" in pipe["loader_settings"] else 1
|
|
|
| if image_to_latent is not None:
|
| _, height, width, _ = image_to_latent.shape
|
| if height == 1 and width == 1:
|
| samples = pipe["samples"]
|
| images = pipe["images"]
|
| else:
|
| if "IP2P" in guider:
|
| positive, negative, latent = self.ip2p(pipe['positive'], pipe['negative'], vae, image_to_latent)
|
| samples = latent
|
| else:
|
| samples = {"samples": vae.encode(image_to_latent[:, :, :, :3])}
|
| samples = RepeatLatentBatch().repeat(samples, batch_size)[0]
|
| images = image_to_latent
|
| elif latent is not None:
|
| if "IP2P" in guider:
|
| positive, negative, latent = self.ip2p(pipe['positive'], pipe['negative'], latent=latent)
|
| samples = latent
|
| else:
|
| samples = latent
|
| images = pipe["images"]
|
| else:
|
| samples = pipe["samples"]
|
| images = pipe["images"]
|
|
|
|
|
| new_pipe = {
|
| "model": model,
|
| "positive": positive,
|
| "negative": negative,
|
| "vae": pipe['vae'],
|
| "clip": pipe['clip'],
|
|
|
| "samples": samples,
|
| "images": images,
|
| "seed": seed,
|
|
|
| "loader_settings": {
|
| **pipe["loader_settings"],
|
| "middle": pipe['negative'],
|
| "steps": steps,
|
| "cfg": cfg,
|
| "cfg_negative": cfg_negative,
|
| "sampler_name": sampler_name,
|
| "scheduler": scheduler,
|
| "denoise": denoise,
|
| "add_noise": add_noise,
|
| "custom": {
|
| "guider": guider,
|
| "coeff": coeff,
|
| "sigma_max": sigma_max,
|
| "sigma_min": sigma_min,
|
| "rho": rho,
|
| "beta_d": beta_d,
|
| "beta_min": beta_min,
|
| "eps_s": beta_min,
|
| "flip_sigmas": flip_sigmas
|
| },
|
| "optional_sampler": optional_sampler,
|
| "optional_sigmas": optional_sigmas
|
| }
|
| }
|
|
|
| del pipe
|
|
|
| return {"ui": {"value": [seed]}, "result": (new_pipe,)}
|
|
|
|
|
| from ..libs.gradual_latent_hires_fix import sample_dpmpp_2s_ancestral, sample_dpmpp_2m_sde, sample_lcm, sample_euler_ancestral
|
| class sdTurboSettings:
|
| def __init__(self):
|
| pass
|
|
|
| @classmethod
|
| def INPUT_TYPES(cls):
|
| return {"required": {
|
| "pipe": ("PIPE_LINE",),
|
| "steps": ("INT", {"default": 1, "min": 1, "max": 10}),
|
| "cfg": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 100.0}),
|
| "sampler_name": (comfy.samplers.SAMPLER_NAMES,),
|
| "eta": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.01, "round": False}),
|
| "s_noise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.01, "round": False}),
|
| "upscale_ratio": ("FLOAT", {"default": 2.0, "min": 0.0, "max": 16.0, "step": 0.01, "round": False}),
|
| "start_step": ("INT", {"default": 5, "min": 0, "max": 1000, "step": 1}),
|
| "end_step": ("INT", {"default": 15, "min": 0, "max": 1000, "step": 1}),
|
| "upscale_n_step": ("INT", {"default": 3, "min": 0, "max": 1000, "step": 1}),
|
| "unsharp_kernel_size": ("INT", {"default": 3, "min": 1, "max": 21, "step": 1}),
|
| "unsharp_sigma": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 10.0, "step": 0.01, "round": False}),
|
| "unsharp_strength": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 10.0, "step": 0.01, "round": False}),
|
| "seed": ("INT", {"default": 0, "min": 0, "max": MAX_SEED_NUM}),
|
| },
|
| "hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "my_unique_id": "UNIQUE_ID"},
|
| }
|
|
|
| RETURN_TYPES = ("PIPE_LINE",)
|
| RETURN_NAMES = ("pipe",)
|
|
|
| FUNCTION = "settings"
|
| CATEGORY = "EasyUse/PreSampling"
|
|
|
| def settings(self, pipe, steps, cfg, sampler_name, eta, s_noise, upscale_ratio, start_step, end_step, upscale_n_step, unsharp_kernel_size, unsharp_sigma, unsharp_strength, seed, prompt=None, extra_pnginfo=None, my_unique_id=None):
|
| model = pipe['model']
|
|
|
| timesteps = torch.flip(torch.arange(1, 11) * 100 - 1, (0,))[:steps]
|
| sigmas = model.model.model_sampling.sigma(timesteps)
|
| sigmas = torch.cat([sigmas, sigmas.new_zeros([1])])
|
|
|
|
|
| sample_function = None
|
| extra_options = {
|
| "eta": eta,
|
| "s_noise": s_noise,
|
| "upscale_ratio": upscale_ratio,
|
| "start_step": start_step,
|
| "end_step": end_step,
|
| "upscale_n_step": upscale_n_step,
|
| "unsharp_kernel_size": unsharp_kernel_size,
|
| "unsharp_sigma": unsharp_sigma,
|
| "unsharp_strength": unsharp_strength,
|
| }
|
| if sampler_name == "euler_ancestral":
|
| sample_function = sample_euler_ancestral
|
| elif sampler_name == "dpmpp_2s_ancestral":
|
| sample_function = sample_dpmpp_2s_ancestral
|
| elif sampler_name == "dpmpp_2m_sde":
|
| sample_function = sample_dpmpp_2m_sde
|
| elif sampler_name == "lcm":
|
| sample_function = sample_lcm
|
|
|
| if sample_function is not None:
|
| unsharp_kernel_size = unsharp_kernel_size if unsharp_kernel_size % 2 == 1 else unsharp_kernel_size + 1
|
| extra_options["unsharp_kernel_size"] = unsharp_kernel_size
|
| _sampler = comfy.samplers.KSAMPLER(sample_function, extra_options)
|
| else:
|
| _sampler = comfy.samplers.sampler_object(sampler_name)
|
| extra_options = None
|
|
|
| new_pipe = {
|
| "model": pipe['model'],
|
| "positive": pipe['positive'],
|
| "negative": pipe['negative'],
|
| "vae": pipe['vae'],
|
| "clip": pipe['clip'],
|
|
|
| "samples": pipe["samples"],
|
| "images": pipe["images"],
|
| "seed": seed,
|
|
|
| "loader_settings": {
|
| **pipe["loader_settings"],
|
| "extra_options": extra_options,
|
| "sampler": _sampler,
|
| "sigmas": sigmas,
|
| "steps": steps,
|
| "cfg": cfg,
|
| "add_noise": "enabled"
|
| }
|
| }
|
|
|
| del pipe
|
|
|
| return {"ui": {"value": [seed]}, "result": (new_pipe,)}
|
|
|
|
|
|
|
| class cascadeSettings:
|
|
|
| def __init__(self):
|
| pass
|
|
|
| @classmethod
|
| def INPUT_TYPES(cls):
|
| return {"required":
|
| {"pipe": ("PIPE_LINE",),
|
| "encode_vae_name": (["None"] + folder_paths.get_filename_list("vae"),),
|
| "decode_vae_name": (["None"] + folder_paths.get_filename_list("vae"),),
|
| "steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
|
| "cfg": ("FLOAT", {"default": 4.0, "min": 0.0, "max": 100.0}),
|
| "sampler_name": (comfy.samplers.KSampler.SAMPLERS, {"default":"euler_ancestral"}),
|
| "scheduler": (comfy.samplers.KSampler.SCHEDULERS, {"default":"simple"}),
|
| "denoise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
|
| "seed": ("INT", {"default": 0, "min": 0, "max": MAX_SEED_NUM}),
|
| },
|
| "optional": {
|
| "image_to_latent_c": ("IMAGE",),
|
| "latent_c": ("LATENT",),
|
| },
|
| "hidden":{"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "my_unique_id": "UNIQUE_ID"},
|
| }
|
|
|
| RETURN_TYPES = ("PIPE_LINE",)
|
| RETURN_NAMES = ("pipe",)
|
|
|
| FUNCTION = "settings"
|
| CATEGORY = "EasyUse/PreSampling"
|
|
|
| def settings(self, pipe, encode_vae_name, decode_vae_name, steps, cfg, sampler_name, scheduler, denoise, seed, model=None, image_to_latent_c=None, latent_c=None, prompt=None, extra_pnginfo=None, my_unique_id=None):
|
| images, samples_c = None, None
|
| samples = pipe['samples']
|
| batch_size = pipe["loader_settings"]["batch_size"] if "batch_size" in pipe["loader_settings"] else 1
|
|
|
| encode_vae_name = encode_vae_name if encode_vae_name is not None else pipe['loader_settings']['encode_vae_name']
|
| decode_vae_name = decode_vae_name if decode_vae_name is not None else pipe['loader_settings']['decode_vae_name']
|
|
|
| if image_to_latent_c is not None:
|
| if encode_vae_name != 'None':
|
| encode_vae = easyCache.load_vae(encode_vae_name)
|
| else:
|
| encode_vae = pipe['vae'][0]
|
| if "compression" not in pipe["loader_settings"]:
|
| raise Exception("compression is not found")
|
| compression = pipe["loader_settings"]['compression']
|
| width = image_to_latent_c.shape[-2]
|
| height = image_to_latent_c.shape[-3]
|
| out_width = (width // compression) * encode_vae.downscale_ratio
|
| out_height = (height // compression) * encode_vae.downscale_ratio
|
|
|
| s = comfy.utils.common_upscale(image_to_latent_c.movedim(-1, 1), out_width, out_height, "bicubic",
|
| "center").movedim(1,
|
| -1)
|
| c_latent = encode_vae.encode(s[:, :, :, :3])
|
| b_latent = torch.zeros([c_latent.shape[0], 4, height // 4, width // 4])
|
|
|
| samples_c = {"samples": c_latent}
|
| samples_c = RepeatLatentBatch().repeat(samples_c, batch_size)[0]
|
|
|
| samples_b = {"samples": b_latent}
|
| samples_b = RepeatLatentBatch().repeat(samples_b, batch_size)[0]
|
| samples = (samples_c, samples_b)
|
| images = image_to_latent_c
|
| elif latent_c is not None:
|
| samples_c = latent_c
|
| samples = (samples_c, samples[1])
|
| images = pipe["images"]
|
| if samples_c is not None:
|
| samples = (samples_c, samples[1])
|
|
|
| new_pipe = {
|
| "model": pipe['model'],
|
| "positive": pipe['positive'],
|
| "negative": pipe['negative'],
|
| "vae": pipe['vae'],
|
| "clip": pipe['clip'],
|
|
|
| "samples": samples,
|
| "images": images,
|
| "seed": seed,
|
|
|
| "loader_settings": {
|
| **pipe["loader_settings"],
|
| "encode_vae_name": encode_vae_name,
|
| "decode_vae_name": decode_vae_name,
|
| "steps": steps,
|
| "cfg": cfg,
|
| "sampler_name": sampler_name,
|
| "scheduler": scheduler,
|
| "denoise": denoise,
|
| "add_noise": "enabled"
|
| }
|
| }
|
|
|
| sampler.update_value_by_id("pipe_line", my_unique_id, new_pipe)
|
|
|
| del pipe
|
|
|
| return {"ui": {"value": [seed]}, "result": (new_pipe,)}
|
|
|
|
|
| class layerDiffusionSettings:
|
|
|
| def __init__(self):
|
| pass
|
|
|
| @classmethod
|
| def INPUT_TYPES(cls):
|
| return {"required":
|
| {
|
| "pipe": ("PIPE_LINE",),
|
| "method": ([LayerMethod.FG_ONLY_ATTN.value, LayerMethod.FG_ONLY_CONV.value, LayerMethod.EVERYTHING.value, LayerMethod.FG_TO_BLEND.value, LayerMethod.BG_TO_BLEND.value],),
|
| "weight": ("FLOAT",{"default": 1.0, "min": -1, "max": 3, "step": 0.05},),
|
| "steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
|
| "cfg": ("FLOAT", {"default": 8.0, "min": 0.0, "max": 100.0}),
|
| "sampler_name": (comfy.samplers.KSampler.SAMPLERS, {"default": "euler"}),
|
| "scheduler": (comfy.samplers.KSampler.SCHEDULERS+ NEW_SCHEDULERS, {"default": "normal"}),
|
| "denoise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
|
| "seed": ("INT", {"default": 0, "min": 0, "max": MAX_SEED_NUM}),
|
| },
|
| "optional": {
|
| "image": ("IMAGE",),
|
| "blended_image": ("IMAGE",),
|
| "mask": ("MASK",),
|
|
|
|
|
| },
|
| "hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "my_unique_id": "UNIQUE_ID"},
|
| }
|
|
|
| RETURN_TYPES = ("PIPE_LINE",)
|
| RETURN_NAMES = ("pipe",)
|
|
|
| FUNCTION = "settings"
|
| CATEGORY = "EasyUse/PreSampling"
|
|
|
| def get_layer_diffusion_method(self, method, has_blend_latent):
|
| method = LayerMethod(method)
|
| if has_blend_latent:
|
| if method == LayerMethod.BG_TO_BLEND:
|
| method = LayerMethod.BG_BLEND_TO_FG
|
| elif method == LayerMethod.FG_TO_BLEND:
|
| method = LayerMethod.FG_BLEND_TO_BG
|
| return method
|
|
|
| def settings(self, pipe, method, weight, steps, cfg, sampler_name, scheduler, denoise, seed, image=None, blended_image=None, mask=None, prompt=None, extra_pnginfo=None, my_unique_id=None):
|
| blend_samples = pipe['blend_samples'] if "blend_samples" in pipe else None
|
| vae = pipe["vae"]
|
| batch_size = pipe["loader_settings"]["batch_size"] if "batch_size" in pipe["loader_settings"] else 1
|
|
|
| method = self.get_layer_diffusion_method(method, blend_samples is not None or blended_image is not None)
|
|
|
| if image is not None or "image" in pipe:
|
| image = image if image is not None else pipe['image']
|
| if mask is not None:
|
| print('inpaint')
|
| samples, = VAEEncodeForInpaint().encode(vae, image, mask)
|
| else:
|
| samples = {"samples": vae.encode(image[:,:,:,:3])}
|
| samples = RepeatLatentBatch().repeat(samples, batch_size)[0]
|
| images = image
|
| elif "samp_images" in pipe:
|
| samples = {"samples": vae.encode(pipe["samp_images"][:,:,:,:3])}
|
| samples = RepeatLatentBatch().repeat(samples, batch_size)[0]
|
| images = pipe["samp_images"]
|
| else:
|
| if method not in [LayerMethod.FG_ONLY_ATTN, LayerMethod.FG_ONLY_CONV, LayerMethod.EVERYTHING]:
|
| raise Exception("image is missing")
|
|
|
| samples = pipe["samples"]
|
| images = pipe["images"]
|
|
|
| if method in [LayerMethod.BG_BLEND_TO_FG, LayerMethod.FG_BLEND_TO_BG]:
|
| if blended_image is None and blend_samples is None:
|
| raise Exception("blended_image is missing")
|
| elif blended_image is not None:
|
| blend_samples = {"samples": vae.encode(blended_image[:,:,:,:3])}
|
| blend_samples = RepeatLatentBatch().repeat(blend_samples, batch_size)[0]
|
|
|
| new_pipe = {
|
| "model": pipe['model'],
|
| "positive": pipe['positive'],
|
| "negative": pipe['negative'],
|
| "vae": pipe['vae'],
|
| "clip": pipe['clip'],
|
|
|
| "samples": samples,
|
| "blend_samples": blend_samples,
|
| "images": images,
|
| "seed": seed,
|
|
|
| "loader_settings": {
|
| **pipe["loader_settings"],
|
| "steps": steps,
|
| "cfg": cfg,
|
| "sampler_name": sampler_name,
|
| "scheduler": scheduler,
|
| "denoise": denoise,
|
| "add_noise": "enabled",
|
| "layer_diffusion_method": method,
|
| "layer_diffusion_weight": weight,
|
| }
|
| }
|
|
|
| del pipe
|
|
|
| return {"ui": {"value": [seed]}, "result": (new_pipe,)}
|
|
|
|
|
| class layerDiffusionSettingsADDTL:
|
| def __init__(self):
|
| pass
|
|
|
| @classmethod
|
| def INPUT_TYPES(cls):
|
| return {"required":
|
| {
|
| "pipe": ("PIPE_LINE",),
|
| "foreground_prompt": ("STRING", {"default": "", "placeholder": "Foreground Additional Prompt", "multiline": True}),
|
| "background_prompt": ("STRING", {"default": "", "placeholder": "Background Additional Prompt", "multiline": True}),
|
| "blended_prompt": ("STRING", {"default": "", "placeholder": "Blended Additional Prompt", "multiline": True}),
|
| },
|
| "optional": {
|
| "optional_fg_cond": ("CONDITIONING",),
|
| "optional_bg_cond": ("CONDITIONING",),
|
| "optional_blended_cond": ("CONDITIONING",),
|
| },
|
| "hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "my_unique_id": "UNIQUE_ID"},
|
| }
|
|
|
| RETURN_TYPES = ("PIPE_LINE",)
|
| RETURN_NAMES = ("pipe",)
|
|
|
| FUNCTION = "settings"
|
| CATEGORY = "EasyUse/PreSampling"
|
|
|
| def settings(self, pipe, foreground_prompt, background_prompt, blended_prompt, optional_fg_cond=None, optional_bg_cond=None, optional_blended_cond=None, prompt=None, extra_pnginfo=None, my_unique_id=None):
|
| fg_cond, bg_cond, blended_cond = None, None, None
|
| clip = pipe['clip']
|
| if optional_fg_cond is not None:
|
| fg_cond = optional_fg_cond
|
| elif foreground_prompt != "":
|
| fg_cond, = CLIPTextEncode().encode(clip, foreground_prompt)
|
| if optional_bg_cond is not None:
|
| bg_cond = optional_bg_cond
|
| elif background_prompt != "":
|
| bg_cond, = CLIPTextEncode().encode(clip, background_prompt)
|
| if optional_blended_cond is not None:
|
| blended_cond = optional_blended_cond
|
| elif blended_prompt != "":
|
| blended_cond, = CLIPTextEncode().encode(clip, blended_prompt)
|
|
|
| new_pipe = {
|
| **pipe,
|
| "loader_settings": {
|
| **pipe["loader_settings"],
|
| "layer_diffusion_cond": (fg_cond, bg_cond, blended_cond)
|
| }
|
| }
|
|
|
| del pipe
|
|
|
| return (new_pipe,)
|
|
|
|
|
| from ..libs.dynthres_core import DynThresh
|
| class dynamicCFGSettings:
|
| def __init__(self):
|
| pass
|
|
|
| @classmethod
|
| def INPUT_TYPES(cls):
|
| return {"required":
|
| {"pipe": ("PIPE_LINE",),
|
| "steps": ("INT", {"default": 20, "min": 1, "max": 10000}),
|
| "cfg": ("FLOAT", {"default": 8.0, "min": 0.0, "max": 100.0}),
|
| "cfg_mode": (DynThresh.Modes,),
|
| "cfg_scale_min": ("FLOAT", {"default": 3.5, "min": 0.0, "max": 100.0, "step": 0.5}),
|
| "sampler_name": (comfy.samplers.KSampler.SAMPLERS,),
|
| "scheduler": (comfy.samplers.KSampler.SCHEDULERS+NEW_SCHEDULERS,),
|
| "denoise": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
|
| "seed": ("INT", {"default": 0, "min": 0, "max": MAX_SEED_NUM}),
|
| },
|
| "optional":{
|
| "image_to_latent": ("IMAGE",),
|
| "latent": ("LATENT",)
|
| },
|
| "hidden":
|
| {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO", "my_unique_id": "UNIQUE_ID"},
|
| }
|
|
|
| RETURN_TYPES = ("PIPE_LINE",)
|
| RETURN_NAMES = ("pipe",)
|
|
|
| FUNCTION = "settings"
|
| CATEGORY = "EasyUse/PreSampling"
|
|
|
| def settings(self, pipe, steps, cfg, cfg_mode, cfg_scale_min,sampler_name, scheduler, denoise, seed, image_to_latent=None, latent=None, prompt=None, extra_pnginfo=None, my_unique_id=None):
|
|
|
|
|
| dynamic_thresh = DynThresh(7.0, 1.0,"CONSTANT", 0, cfg_mode, cfg_scale_min, 0, 0, 999, False,
|
| "MEAN", "AD", 1)
|
|
|
| def sampler_dyn_thresh(args):
|
| input = args["input"]
|
| cond = input - args["cond"]
|
| uncond = input - args["uncond"]
|
| cond_scale = args["cond_scale"]
|
| time_step = args["timestep"]
|
| dynamic_thresh.step = 999 - time_step[0]
|
|
|
| return input - dynamic_thresh.dynthresh(cond, uncond, cond_scale, None)
|
|
|
| model = pipe['model']
|
|
|
| m = model.clone()
|
| m.set_model_sampler_cfg_function(sampler_dyn_thresh)
|
|
|
|
|
| vae = pipe["vae"]
|
| batch_size = pipe["loader_settings"]["batch_size"] if "batch_size" in pipe["loader_settings"] else 1
|
| if image_to_latent is not None:
|
| samples = {"samples": vae.encode(image_to_latent[:, :, :, :3])}
|
| samples = RepeatLatentBatch().repeat(samples, batch_size)[0]
|
| images = image_to_latent
|
| elif latent is not None:
|
| samples = RepeatLatentBatch().repeat(latent, batch_size)[0]
|
| images = pipe["images"]
|
| else:
|
| samples = pipe["samples"]
|
| images = pipe["images"]
|
|
|
| new_pipe = {
|
| "model": m,
|
| "positive": pipe['positive'],
|
| "negative": pipe['negative'],
|
| "vae": pipe['vae'],
|
| "clip": pipe['clip'],
|
|
|
| "samples": samples,
|
| "images": images,
|
| "seed": seed,
|
|
|
| "loader_settings": {
|
| **pipe["loader_settings"],
|
| "steps": steps,
|
| "cfg": cfg,
|
| "sampler_name": sampler_name,
|
| "scheduler": scheduler,
|
| "denoise": denoise
|
| },
|
| }
|
|
|
| del pipe
|
|
|
| return {"ui": {"value": [seed]}, "result": (new_pipe,)}
|
|
|
|
|
| class dynamicThresholdingFull:
|
| @classmethod
|
| def INPUT_TYPES(s):
|
| return {
|
| "required": {
|
| "model": ("MODEL",),
|
| "mimic_scale": ("FLOAT", {"default": 7.0, "min": 0.0, "max": 100.0, "step": 0.5}),
|
| "threshold_percentile": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
|
| "mimic_mode": (DynThresh.Modes,),
|
| "mimic_scale_min": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 100.0, "step": 0.5}),
|
| "cfg_mode": (DynThresh.Modes,),
|
| "cfg_scale_min": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 100.0, "step": 0.5}),
|
| "sched_val": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 100.0, "step": 0.01}),
|
| "separate_feature_channels": (["enable", "disable"],),
|
| "scaling_startpoint": (DynThresh.Startpoints,),
|
| "variability_measure": (DynThresh.Variabilities,),
|
| "interpolate_phi": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01}),
|
| }
|
| }
|
|
|
| RETURN_TYPES = ("MODEL",)
|
| FUNCTION = "patch"
|
| CATEGORY = "EasyUse/PreSampling"
|
|
|
| def patch(self, model, mimic_scale, threshold_percentile, mimic_mode, mimic_scale_min, cfg_mode, cfg_scale_min,
|
| sched_val, separate_feature_channels, scaling_startpoint, variability_measure, interpolate_phi):
|
| dynamic_thresh = DynThresh(mimic_scale, threshold_percentile, mimic_mode, mimic_scale_min, cfg_mode,
|
| cfg_scale_min, sched_val, 0, 999, separate_feature_channels == "enable",
|
| scaling_startpoint, variability_measure, interpolate_phi)
|
|
|
| def sampler_dyn_thresh(args):
|
| input = args["input"]
|
| cond = input - args["cond"]
|
| uncond = input - args["uncond"]
|
| cond_scale = args["cond_scale"]
|
| time_step = args["timestep"]
|
| dynamic_thresh.step = 999 - time_step[0]
|
|
|
| return input - dynamic_thresh.dynthresh(cond, uncond, cond_scale, None)
|
|
|
| m = model.clone()
|
| m.set_model_sampler_cfg_function(sampler_dyn_thresh)
|
| return (m,)
|
|
|
|
|
|
|
| NODE_CLASS_MAPPINGS = {
|
| "easy preSampling": samplerSettings,
|
| "easy preSamplingAdvanced": samplerSettingsAdvanced,
|
| "easy preSamplingNoiseIn": samplerSettingsNoiseIn,
|
| "easy preSamplingCustom": samplerCustomSettings,
|
| "easy preSamplingSdTurbo": sdTurboSettings,
|
| "easy preSamplingDynamicCFG": dynamicCFGSettings,
|
| "easy preSamplingCascade": cascadeSettings,
|
| "easy preSamplingLayerDiffusion": layerDiffusionSettings,
|
| "easy preSamplingLayerDiffusionADDTL": layerDiffusionSettingsADDTL,
|
| "dynamicThresholdingFull": dynamicThresholdingFull,
|
| }
|
|
|
| NODE_DISPLAY_NAME_MAPPINGS = {
|
| "easy preSampling": "PreSampling",
|
| "easy preSamplingAdvanced": "PreSampling (Advanced)",
|
| "easy preSamplingNoiseIn": "PreSampling (NoiseIn)",
|
| "easy preSamplingCustom": "PreSampling (Custom)",
|
| "easy preSamplingSdTurbo": "PreSampling (SDTurbo)",
|
| "easy preSamplingDynamicCFG": "PreSampling (DynamicCFG)",
|
| "easy preSamplingCascade": "PreSampling (Cascade)",
|
| "easy preSamplingLayerDiffusion": "PreSampling (LayerDiffuse)",
|
| "easy preSamplingLayerDiffusionADDTL": "PreSampling (LayerDiffuse ADDTL)",
|
| "dynamicThresholdingFull": "DynamicThresholdingFull",
|
| } |
