@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 = "model/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."

SEARCH_ALIASES = ["text", "prompt", "text prompt", "positive prompt", "negative prompt", "encode text", "text encoder", "encode prompt"]

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)

ESSENTIALS_CATEGORY = "Image Generation"

@classmethod

def INPUT_TYPES(s):

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

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "combine"

CATEGORY = "model/conditioning/transform"

SEARCH_ALIASES = ["combine", "merge conditioning", "combine prompts", "merge prompts", "mix prompts", "add prompt"]

def combine(self, conditioning_1, conditioning_2):

SEARCH_ALIASES = ["blend prompts", "interpolate conditioning", "mix prompts", "style fusion", "weighted blend"]

@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 = "model/conditioning/transform"

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 = "model/conditioning/transform"

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)

SEARCH_ALIASES = ["scale conditioning", "scale prompt", "multiply conditioning", "multiply prompt"]

@classmethod

def INPUT_TYPES(cls):

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

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

}}

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "multiply"

CATEGORY = "model/conditioning/transform"

def multiply(self, conditioning, multiplier):

c = []

for t in conditioning:

values = {}

pooled_output = t[1].get("pooled_output", None)

if pooled_output is not None:

values["pooled_output"] = pooled_output * multiplier

scaled = node_helpers.conditioning_set_values([[t[0] * multiplier, t[1]]], values)[0]

c.append(scaled)

SEARCH_ALIASES = ["regional prompt", "area prompt", "spatial conditioning", "localized prompt"]

@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 = "model/conditioning/transform"

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 = "model/conditioning/transform"

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 = "model/conditioning/transform"

def append(self, conditioning, strength):

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

SEARCH_ALIASES = ["masked prompt", "regional inpaint conditioning", "mask conditioning"]

@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 = "model/conditioning/transform"

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})

SEARCH_ALIASES = ["null conditioning", "clear conditioning"]

@classmethod

def INPUT_TYPES(s):

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

RETURN_TYPES = ("CONDITIONING",)

FUNCTION = "zero_out"

CATEGORY = "model/conditioning/transform"

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)

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

if conditioning_lyrics is not None:

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

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

if conditioning_scale is not None:

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

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

if direct_context is not None:

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

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 = "model/conditioning/transform"

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 = "model/latent"

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

SEARCH_ALIASES = ["decode", "decode latent", "latent to image", "render latent"]

def decode(self, vae, samples):

latent = samples["samples"]

if latent.is_nested:

latent = latent.unbind()[0]

images = vae.decode(latent)

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, "advanced": True}),

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

"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.", "advanced": True}),

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

}}

RETURN_TYPES = ("IMAGE",)

FUNCTION = "decode"

CATEGORY = "model/latent"

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

latent = samples["samples"]

if latent.is_nested:

latent = latent.unbind()[0]

compression = vae.spacial_compression_decode()

images = vae.decode_tiled(latent, 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 = "model/latent"

SEARCH_ALIASES = ["encode", "encode image", "image to latent"]

def encode(self, vae, pixels):

t = vae.encode(pixels)

@classmethod

def INPUT_TYPES(s):

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

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

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

"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.", "advanced": True}),

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

}}

RETURN_TYPES = ("LATENT",)

FUNCTION = "encode"

CATEGORY = "model/latent"

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

t = vae.encode_tiled(pixels, 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 = "model/latent"

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

downscale_ratio = vae.spacial_compression_encode()

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

y = (pixels.shape[2] // downscale_ratio) * 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] % downscale_ratio) // 2

y_offset = (pixels.shape[2] % 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 = "model/conditioning"

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)

SEARCH_ALIASES = ["export latent"]

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 = ("LATENT",)

RETURN_NAMES = ("samples",)

FUNCTION = "save"

OUTPUT_NODE = True

CATEGORY = "model/latent"

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[FileLocator] = []

results.append({

"filename": file,

"subfolder": subfolder,

"type": "output"

})

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

output = {}

output["latent_tensor"] = samples["samples"].contiguous()

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

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

SEARCH_ALIASES = ["import latent", "open latent"]

@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 = "model/latent"

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)

SEARCH_ALIASES = ["load model", "model loader"]

@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 = "model/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 = "model/loaders"

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

SEARCH_ALIASES = ["load model", "checkpoint", "model loader", "load checkpoint", "ckpt", "model"]

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"))

SEARCH_ALIASES = ["load diffusers model"]

@classmethod

def _model_paths(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 paths

@classmethod

def INPUT_TYPES(cls):

return {"required": {"model_path": (cls._model_paths(),), }}

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

FUNCTION = "load_checkpoint"

DEPRECATED = True

CATEGORY = "model/loaders"

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

if model_path not in self._model_paths():

raise ValueError(f"Invalid diffusers model path: {model_path!r}")

resolved_model_path = None

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.isfile(os.path.join(path, "model_index.json")):

resolved_model_path = path

break

if resolved_model_path is None:

raise FileNotFoundError(f"Diffusers model {model_path!r} not found.")

@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 = "model/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, "advanced": True}),

}}

RETURN_TYPES = ("CLIP",)

FUNCTION = "set_last_layer"

CATEGORY = "model/conditioning"

def set_last_layer(self, clip, stop_at_clip_layer):

clip = clip.clone()

clip.clip_layer(stop_at_clip_layer)

ESSENTIALS_CATEGORY = "Image Generation"

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 = "model/loaders"

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

SEARCH_ALIASES = ["lora", "load lora", "apply lora", "lora loader", "lora model"]

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

lora_metadata = None

if self.loaded_lora is not None:

if self.loaded_lora[0] == lora_path:

lora = self.loaded_lora[1]

lora_metadata = self.loaded_lora[2] if len(self.loaded_lora) > 2 else None

else:

self.loaded_lora = None

if lora is None:

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

self.loaded_lora = (lora_path, lora, lora_metadata)

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

@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",)

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

SEARCH_ALIASES = ["lora", "load lora", "apply lora", "lora loader", "lora model"]

FUNCTION = "load_lora_model_only"

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

video_taes = ["taehv", "lighttaew2_2", "lighttaew2_1", "lighttaehy1_5", "taeltx_2", "taeh3"]

image_taes = ["taesd", "taesdxl", "taesd3", "taef1", "taef2", "taeqi2_1"]

@staticmethod

def vae_list(s):

vaes = folder_paths.get_filename_list("vae")

approx_vaes = folder_paths.get_filename_list("vae_approx")

have_img_encoder, have_img_decoder = set(), set()

for v in approx_vaes:

parts = v.rsplit("_", 1)

if len(parts) != 2 or parts[0] not in s.image_taes:

for tae in s.video_taes:

if v.startswith(tae):

vaes.append(v)

break

continue

if parts[1].startswith("encoder."):

have_img_encoder.add(parts[0])

elif parts[1].startswith("decoder."):

have_img_decoder.add(parts[0])

vaes += [k for k in have_img_decoder if k in have_img_encoder]

vaes.append("pixel_space")

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)

elif name == "taeqi2_1":

latent_format = comfy.latent_formats.QwenImage21()

sd["vae_scale"] = 1.0 / latent_format.latents_std[0]

sd["vae_shift"] = latent_format.latents_mean[0]

return sd

@classmethod

def INPUT_TYPES(s):

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

RETURN_TYPES = ("VAE",)

FUNCTION = "load_vae"

CATEGORY = "model/loaders"

#TODO: scale factor?

def load_vae(self, vae_name):

metadata = None

vae_path = None

if vae_name == "pixel_space":

sd = {}

sd["pixel_space_vae"] = torch.tensor(1.0)

elif vae_name in self.image_taes:

sd = self.load_taesd(vae_name)

else:

if os.path.splitext(vae_name)[0] in self.video_taes:

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

else:

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

sd, metadata = comfy.utils.load_torch_file(vae_path, return_metadata=True)

if vae_name == "taef2":

if metadata is None:

metadata = {"tae_latent_channels": 128}

else:

metadata["tae_latent_channels"] = 128

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

vae.throw_exception_if_invalid()

# Register a reload factory on the patcher so multigpu deepclones

# (Select VAE Device, future MultiGPU VAE work-units) can produce

# per-device clones from the same loader context. Only set when we

# actually have a single backing file -- pixel_space and the

# image TAESDs (composed from separate encoder/decoder files via

# load_taesd) are not addressable by a single vae_path.

if vae_path is not None:

vae.patcher.cached_patcher_init = (comfy.sd.load_vae_patcher, (vae_path, metadata, None))

@classmethod

def INPUT_TYPES(s):

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

RETURN_TYPES = ("CONTROL_NET",)

FUNCTION = "load_controlnet"

CATEGORY = "model/loaders"

SEARCH_ALIASES = ["controlnet", "control net", "cn", "load controlnet", "controlnet loader"]

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)

if controlnet is None:

raise RuntimeError("ERROR: controlnet file is invalid and does not contain a valid controlnet model.")

@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 = "model/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 = "model/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", ),