@classmethod

def INPUT_TYPES(s) -> InputTypeDict:

return {

"required": {

"text": (IO.STRING, {"multiline": True, "dynamicPrompts": True, "tooltip": "The text to be encoded."}),

"clip": (IO.CLIP, {"tooltip": "The CLIP model used for encoding the text."})

}

}

RETURN_TYPES = (IO.CONDITIONING,)

OUTPUT_TOOLTIPS = ("A conditioning containing the embedded text used to guide the diffusion model.",)

FUNCTION = "encode"

CATEGORY = "conditioning"

DESCRIPTION = "Encodes a text prompt using a CLIP model into an embedding that can be used to guide the diffusion model towards generating specific images."

def encode(self, clip, text):

if clip is None:

raise RuntimeError("ERROR: clip input is invalid: None\n\nIf the clip is from a checkpoint loader node your checkpoint does not contain a valid clip or text encoder model.")

tokens = clip.tokenize(text)

@classmethod

def INPUT_TYPES(s):

return {"required": {"conditioning_1": ("CONDITIONING", ), "conditioning_2": ("CONDITIONING", )}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "combine"

CATEGORY = "conditioning"

def combine(self, conditioning_1, conditioning_2):

@classmethod

def INPUT_TYPES(s):

return {"required": {"conditioning_to": ("CONDITIONING", ), "conditioning_from": ("CONDITIONING", ),

"conditioning_to_strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.01})

}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "addWeighted"

CATEGORY = "conditioning"

def addWeighted(self, conditioning_to, conditioning_from, conditioning_to_strength):

out = []

if len(conditioning_from) > 1:

logging.warning("Warning: ConditioningAverage conditioning_from contains more than 1 cond, only the first one will actually be applied to conditioning_to.")

cond_from = conditioning_from[0][0]

pooled_output_from = conditioning_from[0][1].get("pooled_output", None)

for i in range(len(conditioning_to)):

t1 = conditioning_to[i][0]

pooled_output_to = conditioning_to[i][1].get("pooled_output", pooled_output_from)

t0 = cond_from[:,:t1.shape[1]]

if t0.shape[1] < t1.shape[1]:

t0 = torch.cat([t0] + [torch.zeros((1, (t1.shape[1] - t0.shape[1]), t1.shape[2]))], dim=1)

tw = torch.mul(t1, conditioning_to_strength) + torch.mul(t0, (1.0 - conditioning_to_strength))

t_to = conditioning_to[i][1].copy()

if pooled_output_from is not None and pooled_output_to is not None:

t_to["pooled_output"] = torch.mul(pooled_output_to, conditioning_to_strength) + torch.mul(pooled_output_from, (1.0 - conditioning_to_strength))

elif pooled_output_from is not None:

t_to["pooled_output"] = pooled_output_from

n = [tw, t_to]

out.append(n)

@classmethod

def INPUT_TYPES(s):

return {"required": {

"conditioning_to": ("CONDITIONING",),

"conditioning_from": ("CONDITIONING",),

}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "concat"

CATEGORY = "conditioning"

def concat(self, conditioning_to, conditioning_from):

out = []

if len(conditioning_from) > 1:

logging.warning("Warning: ConditioningConcat conditioning_from contains more than 1 cond, only the first one will actually be applied to conditioning_to.")

cond_from = conditioning_from[0][0]

for i in range(len(conditioning_to)):

t1 = conditioning_to[i][0]

tw = torch.cat((t1, cond_from),1)

n = [tw, conditioning_to[i][1].copy()]

out.append(n)

@classmethod

def INPUT_TYPES(s):

return {"required": {"conditioning": ("CONDITIONING", ),

"width": ("INT", {"default": 64, "min": 64, "max": MAX_RESOLUTION, "step": 8}),

"height": ("INT", {"default": 64, "min": 64, "max": MAX_RESOLUTION, "step": 8}),

"x": ("INT", {"default": 0, "min": 0, "max": MAX_RESOLUTION, "step": 8}),

"y": ("INT", {"default": 0, "min": 0, "max": MAX_RESOLUTION, "step": 8}),

"strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.01}),

}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "append"

CATEGORY = "conditioning"

def append(self, conditioning, width, height, x, y, strength):

c = node_helpers.conditioning_set_values(conditioning, {"area": (height // 8, width // 8, y // 8, x // 8),

"strength": strength,

"set_area_to_bounds": False})

@classmethod

def INPUT_TYPES(s):

return {"required": {"conditioning": ("CONDITIONING", ),

"width": ("FLOAT", {"default": 1.0, "min": 0, "max": 1.0, "step": 0.01}),

"height": ("FLOAT", {"default": 1.0, "min": 0, "max": 1.0, "step": 0.01}),

"x": ("FLOAT", {"default": 0, "min": 0, "max": 1.0, "step": 0.01}),

"y": ("FLOAT", {"default": 0, "min": 0, "max": 1.0, "step": 0.01}),

"strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.01}),

}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "append"

CATEGORY = "conditioning"

def append(self, conditioning, width, height, x, y, strength):

c = node_helpers.conditioning_set_values(conditioning, {"area": ("percentage", height, width, y, x),

"strength": strength,

"set_area_to_bounds": False})

@classmethod

def INPUT_TYPES(s):

return {"required": {"conditioning": ("CONDITIONING", ),

"strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.01}),

}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "append"

CATEGORY = "conditioning"

def append(self, conditioning, strength):

c = node_helpers.conditioning_set_values(conditioning, {"strength": strength})

@classmethod

def INPUT_TYPES(s):

return {"required": {"conditioning": ("CONDITIONING", ),

"mask": ("MASK", ),

"strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.01}),

"set_cond_area": (["default", "mask bounds"],),

}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "append"

CATEGORY = "conditioning"

def append(self, conditioning, mask, set_cond_area, strength):

set_area_to_bounds = False

if set_cond_area != "default":

set_area_to_bounds = True

if len(mask.shape) < 3:

mask = mask.unsqueeze(0)

c = node_helpers.conditioning_set_values(conditioning, {"mask": mask,

"set_area_to_bounds": set_area_to_bounds,

"mask_strength": strength})

@classmethod

def INPUT_TYPES(s):

return {"required": {"conditioning": ("CONDITIONING", )}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "zero_out"

CATEGORY = "advanced/conditioning"

def zero_out(self, conditioning):

c = []

for t in conditioning:

d = t[1].copy()

pooled_output = d.get("pooled_output", None)

if pooled_output is not None:

d["pooled_output"] = torch.zeros_like(pooled_output)

n = [torch.zeros_like(t[0]), d]

c.append(n)

@classmethod

def INPUT_TYPES(s):

return {"required": {"conditioning": ("CONDITIONING", ),

"start": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.001}),

"end": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.001})

}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "set_range"

CATEGORY = "advanced/conditioning"

def set_range(self, conditioning, start, end):

c = node_helpers.conditioning_set_values(conditioning, {"start_percent": start,

"end_percent": end})

@classmethod

def INPUT_TYPES(s):

return {

"required": {

"samples": ("LATENT", {"tooltip": "The latent to be decoded."}),

"vae": ("VAE", {"tooltip": "The VAE model used for decoding the latent."})

}

}

RETURN_TYPES = ("IMAGE",)

OUTPUT_TOOLTIPS = ("The decoded image.",)

FUNCTION = "decode"

CATEGORY = "latent"

DESCRIPTION = "Decodes latent images back into pixel space images."

def decode(self, vae, samples):

images = vae.decode(samples["samples"])

if len(images.shape) == 5: #Combine batches

images = images.reshape(-1, images.shape[-3], images.shape[-2], images.shape[-1])

@classmethod

def INPUT_TYPES(s):

return {"required": {"samples": ("LATENT", ), "vae": ("VAE", ),

"tile_size": ("INT", {"default": 512, "min": 64, "max": 4096, "step": 32}),

"overlap": ("INT", {"default": 64, "min": 0, "max": 4096, "step": 32}),

"temporal_size": ("INT", {"default": 64, "min": 8, "max": 4096, "step": 4, "tooltip": "Only used for video VAEs: Amount of frames to decode at a time."}),

"temporal_overlap": ("INT", {"default": 8, "min": 4, "max": 4096, "step": 4, "tooltip": "Only used for video VAEs: Amount of frames to overlap."}),

}}

RETURN_TYPES = ("IMAGE",)

FUNCTION = "decode"

CATEGORY = "_for_testing"

def decode(self, vae, samples, tile_size, overlap=64, temporal_size=64, temporal_overlap=8):

if tile_size < overlap * 4:

overlap = tile_size // 4

if temporal_size < temporal_overlap * 2:

temporal_overlap = temporal_overlap // 2

temporal_compression = vae.temporal_compression_decode()

if temporal_compression is not None:

temporal_size = max(2, temporal_size // temporal_compression)

temporal_overlap = max(1, min(temporal_size // 2, temporal_overlap // temporal_compression))

else:

temporal_size = None

temporal_overlap = None

compression = vae.spacial_compression_decode()

images = vae.decode_tiled(samples["samples"], tile_x=tile_size // compression, tile_y=tile_size // compression, overlap=overlap // compression, tile_t=temporal_size, overlap_t=temporal_overlap)

if len(images.shape) == 5: #Combine batches

images = images.reshape(-1, images.shape[-3], images.shape[-2], images.shape[-1])

@classmethod

def INPUT_TYPES(s):

return {"required": { "pixels": ("IMAGE", ), "vae": ("VAE", )}}

RETURN_TYPES = ("LATENT",)

FUNCTION = "encode"

CATEGORY = "latent"

def encode(self, vae, pixels):

t = vae.encode(pixels[:,:,:,:3])

@classmethod

def INPUT_TYPES(s):

return {"required": {"pixels": ("IMAGE", ), "vae": ("VAE", ),

"tile_size": ("INT", {"default": 512, "min": 64, "max": 4096, "step": 64}),

"overlap": ("INT", {"default": 64, "min": 0, "max": 4096, "step": 32}),

"temporal_size": ("INT", {"default": 64, "min": 8, "max": 4096, "step": 4, "tooltip": "Only used for video VAEs: Amount of frames to encode at a time."}),

"temporal_overlap": ("INT", {"default": 8, "min": 4, "max": 4096, "step": 4, "tooltip": "Only used for video VAEs: Amount of frames to overlap."}),

}}

RETURN_TYPES = ("LATENT",)

FUNCTION = "encode"

CATEGORY = "_for_testing"

def encode(self, vae, pixels, tile_size, overlap, temporal_size=64, temporal_overlap=8):

t = vae.encode_tiled(pixels[:,:,:,:3], tile_x=tile_size, tile_y=tile_size, overlap=overlap, tile_t=temporal_size, overlap_t=temporal_overlap)

@classmethod

def INPUT_TYPES(s):

return {"required": { "pixels": ("IMAGE", ), "vae": ("VAE", ), "mask": ("MASK", ), "grow_mask_by": ("INT", {"default": 6, "min": 0, "max": 64, "step": 1}),}}

RETURN_TYPES = ("LATENT",)

FUNCTION = "encode"

CATEGORY = "latent/inpaint"

def encode(self, vae, pixels, mask, grow_mask_by=6):

x = (pixels.shape[1] // vae.downscale_ratio) * vae.downscale_ratio

y = (pixels.shape[2] // vae.downscale_ratio) * vae.downscale_ratio

mask = torch.nn.functional.interpolate(mask.reshape((-1, 1, mask.shape[-2], mask.shape[-1])), size=(pixels.shape[1], pixels.shape[2]), mode="bilinear")

pixels = pixels.clone()

if pixels.shape[1] != x or pixels.shape[2] != y:

x_offset = (pixels.shape[1] % vae.downscale_ratio) // 2

y_offset = (pixels.shape[2] % vae.downscale_ratio) // 2

pixels = pixels[:,x_offset:x + x_offset, y_offset:y + y_offset,:]

mask = mask[:,:,x_offset:x + x_offset, y_offset:y + y_offset]

#grow mask by a few pixels to keep things seamless in latent space

if grow_mask_by == 0:

mask_erosion = mask

else:

kernel_tensor = torch.ones((1, 1, grow_mask_by, grow_mask_by))

padding = math.ceil((grow_mask_by - 1) / 2)

mask_erosion = torch.clamp(torch.nn.functional.conv2d(mask.round(), kernel_tensor, padding=padding), 0, 1)

m = (1.0 - mask.round()).squeeze(1)

for i in range(3):

pixels[:,:,:,i] -= 0.5

pixels[:,:,:,i] *= m

pixels[:,:,:,i] += 0.5

t = vae.encode(pixels)

@classmethod

def INPUT_TYPES(s):

return {"required": {"positive": ("CONDITIONING", ),

"negative": ("CONDITIONING", ),

"vae": ("VAE", ),

"pixels": ("IMAGE", ),

"mask": ("MASK", ),

"noise_mask": ("BOOLEAN", {"default": True, "tooltip": "Add a noise mask to the latent so sampling will only happen within the mask. Might improve results or completely break things depending on the model."}),

}}

RETURN_TYPES = ("CONDITIONING","CONDITIONING","LATENT")

RETURN_NAMES = ("positive", "negative", "latent")

FUNCTION = "encode"

CATEGORY = "conditioning/inpaint"

def encode(self, positive, negative, pixels, vae, mask, noise_mask=True):

x = (pixels.shape[1] // 8) * 8

y = (pixels.shape[2] // 8) * 8

mask = torch.nn.functional.interpolate(mask.reshape((-1, 1, mask.shape[-2], mask.shape[-1])), size=(pixels.shape[1], pixels.shape[2]), mode="bilinear")

orig_pixels = pixels

pixels = orig_pixels.clone()

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,:]

mask = mask[:,:,x_offset:x + x_offset, y_offset:y + y_offset]

m = (1.0 - mask.round()).squeeze(1)

for i in range(3):

pixels[:,:,:,i] -= 0.5

pixels[:,:,:,i] *= m

pixels[:,:,:,i] += 0.5

concat_latent = vae.encode(pixels)

orig_latent = vae.encode(orig_pixels)

out_latent = {}

out_latent["samples"] = orig_latent

if noise_mask:

out_latent["noise_mask"] = mask

out = []

for conditioning in [positive, negative]:

c = node_helpers.conditioning_set_values(conditioning, {"concat_latent_image": concat_latent,

"concat_mask": mask})

out.append(c)

def __init__(self):

self.output_dir = folder_paths.get_output_directory()

@classmethod

def INPUT_TYPES(s):

return {"required": { "samples": ("LATENT", ),

"filename_prefix": ("STRING", {"default": "latents/ComfyUI"})},

"hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO"},

}

RETURN_TYPES = ()

FUNCTION = "save"

OUTPUT_NODE = True

CATEGORY = "_for_testing"

def save(self, samples, filename_prefix="ComfyUI", prompt=None, extra_pnginfo=None):

full_output_folder, filename, counter, subfolder, filename_prefix = folder_paths.get_save_image_path(filename_prefix, self.output_dir)

# support save metadata for latent sharing

prompt_info = ""

if prompt is not None:

prompt_info = json.dumps(prompt)

metadata = None

if not args.disable_metadata:

metadata = {"prompt": prompt_info}

if extra_pnginfo is not None:

for x in extra_pnginfo:

metadata[x] = json.dumps(extra_pnginfo[x])

file = f"{filename}_{counter:05}_.latent"

results = list()

results.append({

"filename": file,

"subfolder": subfolder,

"type": "output"

})

file = os.path.join(full_output_folder, file)

output = {}

output["latent_tensor"] = samples["samples"]

output["latent_format_version_0"] = torch.tensor([])

comfy.utils.save_torch_file(output, file, metadata=metadata)

@classmethod

def INPUT_TYPES(s):

input_dir = folder_paths.get_input_directory()

files = [f for f in os.listdir(input_dir) if os.path.isfile(os.path.join(input_dir, f)) and f.endswith(".latent")]

return {"required": {"latent": [sorted(files), ]}, }

CATEGORY = "_for_testing"

RETURN_TYPES = ("LATENT", )

FUNCTION = "load"

def load(self, latent):

latent_path = folder_paths.get_annotated_filepath(latent)

latent = safetensors.torch.load_file(latent_path, device="cpu")

multiplier = 1.0

if "latent_format_version_0" not in latent:

multiplier = 1.0 / 0.18215

samples = {"samples": latent["latent_tensor"].float() * multiplier}

return (samples, )

@classmethod

def IS_CHANGED(s, latent):

image_path = folder_paths.get_annotated_filepath(latent)

m = hashlib.sha256()

with open(image_path, 'rb') as f:

m.update(f.read())

return m.digest().hex()

@classmethod

def VALIDATE_INPUTS(s, latent):

if not folder_paths.exists_annotated_filepath(latent):

return "Invalid latent file: {}".format(latent)

@classmethod

def INPUT_TYPES(s):

return {"required": { "config_name": (folder_paths.get_filename_list("configs"), ),

"ckpt_name": (folder_paths.get_filename_list("checkpoints"), )}}

RETURN_TYPES = ("MODEL", "CLIP", "VAE")

FUNCTION = "load_checkpoint"

CATEGORY = "advanced/loaders"

DEPRECATED = True

def load_checkpoint(self, config_name, ckpt_name):

config_path = folder_paths.get_full_path("configs", config_name)

ckpt_path = folder_paths.get_full_path_or_raise("checkpoints", ckpt_name)

@classmethod

def INPUT_TYPES(s):

return {

"required": {

"ckpt_name": (folder_paths.get_filename_list("checkpoints"), {"tooltip": "The name of the checkpoint (model) to load."}),

}

}

RETURN_TYPES = ("MODEL", "CLIP", "VAE")

OUTPUT_TOOLTIPS = ("The model used for denoising latents.",

"The CLIP model used for encoding text prompts.",

"The VAE model used for encoding and decoding images to and from latent space.")

FUNCTION = "load_checkpoint"

CATEGORY = "loaders"

DESCRIPTION = "Loads a diffusion model checkpoint, diffusion models are used to denoise latents."

def load_checkpoint(self, ckpt_name):

ckpt_path = folder_paths.get_full_path_or_raise("checkpoints", ckpt_name)

out = comfy.sd.load_checkpoint_guess_config(ckpt_path, output_vae=True, output_clip=True, embedding_directory=folder_paths.get_folder_paths("embeddings"))

@classmethod

def INPUT_TYPES(cls):

paths = []

for search_path in folder_paths.get_folder_paths("diffusers"):

if os.path.exists(search_path):

for root, subdir, files in os.walk(search_path, followlinks=True):

if "model_index.json" in files:

paths.append(os.path.relpath(root, start=search_path))

return {"required": {"model_path": (paths,), }}

RETURN_TYPES = ("MODEL", "CLIP", "VAE")

FUNCTION = "load_checkpoint"

CATEGORY = "advanced/loaders/deprecated"

def load_checkpoint(self, model_path, output_vae=True, output_clip=True):

for search_path in folder_paths.get_folder_paths("diffusers"):

if os.path.exists(search_path):

path = os.path.join(search_path, model_path)

if os.path.exists(path):

model_path = path

break

@classmethod

def INPUT_TYPES(s):

return {"required": { "ckpt_name": (folder_paths.get_filename_list("checkpoints"), ),

}}

RETURN_TYPES = ("MODEL", "CLIP", "VAE", "CLIP_VISION")

FUNCTION = "load_checkpoint"

CATEGORY = "loaders"

def load_checkpoint(self, ckpt_name, output_vae=True, output_clip=True):

ckpt_path = folder_paths.get_full_path_or_raise("checkpoints", ckpt_name)

out = comfy.sd.load_checkpoint_guess_config(ckpt_path, output_vae=True, output_clip=True, output_clipvision=True, embedding_directory=folder_paths.get_folder_paths("embeddings"))

@classmethod

def INPUT_TYPES(s):

return {"required": { "clip": ("CLIP", ),

"stop_at_clip_layer": ("INT", {"default": -1, "min": -24, "max": -1, "step": 1}),

}}

RETURN_TYPES = ("CLIP",)

FUNCTION = "set_last_layer"

CATEGORY = "conditioning"

def set_last_layer(self, clip, stop_at_clip_layer):

clip = clip.clone()

clip.clip_layer(stop_at_clip_layer)

def __init__(self):

self.loaded_lora = None

@classmethod

def INPUT_TYPES(s):

return {

"required": {

"model": ("MODEL", {"tooltip": "The diffusion model the LoRA will be applied to."}),

"clip": ("CLIP", {"tooltip": "The CLIP model the LoRA will be applied to."}),

"lora_name": (folder_paths.get_filename_list("loras"), {"tooltip": "The name of the LoRA."}),

"strength_model": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.01, "tooltip": "How strongly to modify the diffusion model. This value can be negative."}),

"strength_clip": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.01, "tooltip": "How strongly to modify the CLIP model. This value can be negative."}),

}

}

RETURN_TYPES = ("MODEL", "CLIP")

OUTPUT_TOOLTIPS = ("The modified diffusion model.", "The modified CLIP model.")

FUNCTION = "load_lora"

CATEGORY = "loaders"

DESCRIPTION = "LoRAs are used to modify diffusion and CLIP models, altering the way in which latents are denoised such as applying styles. Multiple LoRA nodes can be linked together."

def load_lora(self, model, clip, lora_name, strength_model, strength_clip):

if strength_model == 0 and strength_clip == 0:

return (model, clip)

lora_path = folder_paths.get_full_path_or_raise("loras", lora_name)

lora = None

if self.loaded_lora is not None:

if self.loaded_lora[0] == lora_path:

lora = self.loaded_lora[1]

else:

self.loaded_lora = None

if lora is None:

lora = comfy.utils.load_torch_file(lora_path, safe_load=True)

self.loaded_lora = (lora_path, lora)

model_lora, clip_lora = comfy.sd.load_lora_for_models(model, clip, lora, strength_model, strength_clip)

@classmethod

def INPUT_TYPES(s):

return {"required": { "model": ("MODEL",),

"lora_name": (folder_paths.get_filename_list("loras"), ),

"strength_model": ("FLOAT", {"default": 1.0, "min": -100.0, "max": 100.0, "step": 0.01}),

}}

RETURN_TYPES = ("MODEL",)

FUNCTION = "load_lora_model_only"

def load_lora_model_only(self, model, lora_name, strength_model):

@staticmethod

def vae_list():

vaes = folder_paths.get_filename_list("vae")

approx_vaes = folder_paths.get_filename_list("vae_approx")

sdxl_taesd_enc = False

sdxl_taesd_dec = False

sd1_taesd_enc = False

sd1_taesd_dec = False

sd3_taesd_enc = False

sd3_taesd_dec = False

f1_taesd_enc = False

f1_taesd_dec = False

for v in approx_vaes:

if v.startswith("taesd_decoder."):

sd1_taesd_dec = True

elif v.startswith("taesd_encoder."):

sd1_taesd_enc = True

elif v.startswith("taesdxl_decoder."):

sdxl_taesd_dec = True

elif v.startswith("taesdxl_encoder."):

sdxl_taesd_enc = True

elif v.startswith("taesd3_decoder."):

sd3_taesd_dec = True

elif v.startswith("taesd3_encoder."):

sd3_taesd_enc = True

elif v.startswith("taef1_encoder."):

f1_taesd_dec = True

elif v.startswith("taef1_decoder."):

f1_taesd_enc = True

if sd1_taesd_dec and sd1_taesd_enc:

vaes.append("taesd")

if sdxl_taesd_dec and sdxl_taesd_enc:

vaes.append("taesdxl")

if sd3_taesd_dec and sd3_taesd_enc:

vaes.append("taesd3")

if f1_taesd_dec and f1_taesd_enc:

vaes.append("taef1")

return vaes

@staticmethod

def load_taesd(name):

sd = {}

approx_vaes = folder_paths.get_filename_list("vae_approx")

encoder = next(filter(lambda a: a.startswith("{}_encoder.".format(name)), approx_vaes))

decoder = next(filter(lambda a: a.startswith("{}_decoder.".format(name)), approx_vaes))

enc = comfy.utils.load_torch_file(folder_paths.get_full_path_or_raise("vae_approx", encoder))

for k in enc:

sd["taesd_encoder.{}".format(k)] = enc[k]

dec = comfy.utils.load_torch_file(folder_paths.get_full_path_or_raise("vae_approx", decoder))

for k in dec:

sd["taesd_decoder.{}".format(k)] = dec[k]

if name == "taesd":

sd["vae_scale"] = torch.tensor(0.18215)

sd["vae_shift"] = torch.tensor(0.0)

elif name == "taesdxl":

sd["vae_scale"] = torch.tensor(0.13025)

sd["vae_shift"] = torch.tensor(0.0)

elif name == "taesd3":

sd["vae_scale"] = torch.tensor(1.5305)

sd["vae_shift"] = torch.tensor(0.0609)

elif name == "taef1":

sd["vae_scale"] = torch.tensor(0.3611)

sd["vae_shift"] = torch.tensor(0.1159)

return sd

@classmethod

def INPUT_TYPES(s):

return {"required": { "vae_name": (s.vae_list(), )}}

RETURN_TYPES = ("VAE",)

FUNCTION = "load_vae"

CATEGORY = "loaders"

#TODO: scale factor?

def load_vae(self, vae_name):

if vae_name in ["taesd", "taesdxl", "taesd3", "taef1"]:

sd = self.load_taesd(vae_name)

else:

vae_path = folder_paths.get_full_path_or_raise("vae", vae_name)

sd = comfy.utils.load_torch_file(vae_path)

vae = comfy.sd.VAE(sd=sd)

@classmethod

def INPUT_TYPES(s):

return {"required": { "control_net_name": (folder_paths.get_filename_list("controlnet"), )}}

RETURN_TYPES = ("CONTROL_NET",)

FUNCTION = "load_controlnet"

CATEGORY = "loaders"

def load_controlnet(self, control_net_name):

controlnet_path = folder_paths.get_full_path_or_raise("controlnet", control_net_name)

controlnet = comfy.controlnet.load_controlnet(controlnet_path)

@classmethod

def INPUT_TYPES(s):

return {"required": { "model": ("MODEL",),

"control_net_name": (folder_paths.get_filename_list("controlnet"), )}}

RETURN_TYPES = ("CONTROL_NET",)

FUNCTION = "load_controlnet"

CATEGORY = "loaders"

def load_controlnet(self, model, control_net_name):

controlnet_path = folder_paths.get_full_path_or_raise("controlnet", control_net_name)

controlnet = comfy.controlnet.load_controlnet(controlnet_path, model)

@classmethod

def INPUT_TYPES(s):

return {"required": {"conditioning": ("CONDITIONING", ),

"control_net": ("CONTROL_NET", ),

"image": ("IMAGE", ),

"strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.01})

}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "apply_controlnet"

DEPRECATED = True

CATEGORY = "conditioning/controlnet"

def apply_controlnet(self, conditioning, control_net, image, strength):

if strength == 0:

return (conditioning, )

c = []

control_hint = image.movedim(-1,1)

for t in conditioning:

n = [t[0], t[1].copy()]

c_net = control_net.copy().set_cond_hint(control_hint, strength)

if 'control' in t[1]:

c_net.set_previous_controlnet(t[1]['control'])

n[1]['control'] = c_net

n[1]['control_apply_to_uncond'] = True

c.append(n)

@classmethod

def INPUT_TYPES(s):

return {"required": {"positive": ("CONDITIONING", ),

"negative": ("CONDITIONING", ),

"control_net": ("CONTROL_NET", ),

"image": ("IMAGE", ),

"strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.01}),

"start_percent": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.001}),

"end_percent": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.001})

},

"optional": {"vae": ("VAE", ),

}

}

RETURN_TYPES = ("CONDITIONING","CONDITIONING")

RETURN_NAMES = ("positive", "negative")

FUNCTION = "apply_controlnet"

CATEGORY = "conditioning/controlnet"

def apply_controlnet(self, positive, negative, control_net, image, strength, start_percent, end_percent, vae=None, extra_concat=[]):

if strength == 0:

return (positive, negative)

control_hint = image.movedim(-1,1)

cnets = {}

out = []

for conditioning in [positive, negative]:

c = []

for t in conditioning:

d = t[1].copy()

prev_cnet = d.get('control', None)

if prev_cnet in cnets:

c_net = cnets[prev_cnet]

else:

c_net = control_net.copy().set_cond_hint(control_hint, strength, (start_percent, end_percent), vae=vae, extra_concat=extra_concat)

c_net.set_previous_controlnet(prev_cnet)

cnets[prev_cnet] = c_net

d['control'] = c_net

d['control_apply_to_uncond'] = False

n = [t[0], d]

c.append(n)

out.append(c)

@classmethod

def INPUT_TYPES(s):

return {"required": { "unet_name": (folder_paths.get_filename_list("diffusion_models"), ),

"weight_dtype": (["default", "fp8_e4m3fn", "fp8_e4m3fn_fast", "fp8_e5m2"],)

}}

RETURN_TYPES = ("MODEL",)

FUNCTION = "load_unet"

CATEGORY = "advanced/loaders"

def load_unet(self, unet_name, weight_dtype):

model_options = {}

if weight_dtype == "fp8_e4m3fn":

model_options["dtype"] = torch.float8_e4m3fn

elif weight_dtype == "fp8_e4m3fn_fast":

model_options["dtype"] = torch.float8_e4m3fn

model_options["fp8_optimizations"] = True

elif weight_dtype == "fp8_e5m2":

model_options["dtype"] = torch.float8_e5m2

unet_path = folder_paths.get_full_path_or_raise("diffusion_models", unet_name)

model = comfy.sd.load_diffusion_model(unet_path, model_options=model_options)

@classmethod

def INPUT_TYPES(s):

return {"required": { "clip_name": (folder_paths.get_filename_list("text_encoders"), ),

"type": (["stable_diffusion", "stable_cascade", "sd3", "stable_audio", "mochi", "ltxv", "pixart", "cosmos", "lumina2"], ),

},

"optional": {

"device": (["default", "cpu"], {"advanced": True}),

}}

RETURN_TYPES = ("CLIP",)

FUNCTION = "load_clip"

CATEGORY = "advanced/loaders"

DESCRIPTION = "[Recipes]\n\nstable_diffusion: clip-l\nstable_cascade: clip-g\nsd3: t5 / clip-g / clip-l\nstable_audio: t5\nmochi: t5\ncosmos: old t5 xxl"

def load_clip(self, clip_name, type="stable_diffusion", device="default"):

if type == "stable_cascade":

clip_type = comfy.sd.CLIPType.STABLE_CASCADE

elif type == "sd3":

clip_type = comfy.sd.CLIPType.SD3

elif type == "stable_audio":

clip_type = comfy.sd.CLIPType.STABLE_AUDIO

elif type == "mochi":

clip_type = comfy.sd.CLIPType.MOCHI

elif type == "ltxv":

clip_type = comfy.sd.CLIPType.LTXV

elif type == "pixart":

clip_type = comfy.sd.CLIPType.PIXART

elif type == "cosmos":

clip_type = comfy.sd.CLIPType.COSMOS

elif type == "lumina2":

clip_type = comfy.sd.CLIPType.LUMINA2

else:

clip_type = comfy.sd.CLIPType.STABLE_DIFFUSION

model_options = {}

if device == "cpu":

model_options["load_device"] = model_options["offload_device"] = torch.device("cpu")

clip_path = folder_paths.get_full_path_or_raise("text_encoders", clip_name)

clip = comfy.sd.load_clip(ckpt_paths=[clip_path], embedding_directory=folder_paths.get_folder_paths("embeddings"), clip_type=clip_type, model_options=model_options)