Merge pull request #2 from silveroxides/feature/v3-migration
migrate to v3 schema and add LoRA extraction node
This commit is contained in:
Silver
+50
-55
@@ -1,57 +1,52 @@
|
||||
from .nodes import *
|
||||
from typing_extensions import override
|
||||
from comfy_api.latest import ComfyExtension, io
|
||||
|
||||
NODE_CLASS_MAPPINGS = {
|
||||
"ModelMetaKeys": ModelMetaKeys,
|
||||
"TextEncoderMetaKeys": TextEncoderMetaKeys,
|
||||
"LoRAMetaKeys": LoRAMetaKeys,
|
||||
"CheckpointMetaKeys": CheckpointMetaKeys,
|
||||
"EmbeddingMetaKeys": EmbeddingMetaKeys,
|
||||
"ModelRenameKeys": ModelRenameKeys,
|
||||
"TextEncoderRenameKeys": TextEncoderRenameKeys,
|
||||
"LoRARenameKeys": LoRARenameKeys,
|
||||
"CheckpointRenameKeys": CheckpointRenameKeys,
|
||||
"EmbeddingRenameKeys": EmbeddingRenameKeys,
|
||||
"ModelPruneKeys": ModelPruneKeys,
|
||||
"TextEncoderPruneKeys": TextEncoderPruneKeys,
|
||||
"LoRAPruneKeys": LoRAPruneKeys,
|
||||
"CheckpointPruneKeys": CheckpointPruneKeys,
|
||||
"EmbeddingPruneKeys": EmbeddingPruneKeys,
|
||||
"ModelTwoMerger": ModelTwoMerger,
|
||||
"TextEncoderTwoMerger": TextEncoderTwoMerger,
|
||||
"LoRATwoMerger": LoRATwoMerger,
|
||||
"CheckpointTwoMerger": CheckpointTwoMerger,
|
||||
"EmbeddingTwoMerger": EmbeddingTwoMerger,
|
||||
"ModelThreeMerger": ModelThreeMerger,
|
||||
"TextEncoderThreeMerger": TextEncoderThreeMerger,
|
||||
"LoRAThreeMerger": LoRAThreeMerger,
|
||||
"CheckpointThreeMerger": CheckpointThreeMerger,
|
||||
"EmbeddingThreeMerger": EmbeddingThreeMerger,
|
||||
}
|
||||
from .nodes.metakeys import (
|
||||
ModelMetaKeys, TextEncoderMetaKeys, LoRAMetaKeys,
|
||||
CheckpointMetaKeys, EmbeddingMetaKeys
|
||||
)
|
||||
from .nodes.renamekeys import (
|
||||
ModelRenameKeys, TextEncoderRenameKeys, LoRARenameKeys,
|
||||
CheckpointRenameKeys, EmbeddingRenameKeys
|
||||
)
|
||||
from .nodes.prunekeys import (
|
||||
ModelPruneKeys, TextEncoderPruneKeys, LoRAPruneKeys,
|
||||
CheckpointPruneKeys, EmbeddingPruneKeys
|
||||
)
|
||||
from .nodes.merger import (
|
||||
ModelTwoMerger, TextEncoderTwoMerger, LoRATwoMerger,
|
||||
CheckpointTwoMerger, EmbeddingTwoMerger,
|
||||
ModelThreeMerger, TextEncoderThreeMerger, LoRAThreeMerger,
|
||||
CheckpointThreeMerger, EmbeddingThreeMerger
|
||||
)
|
||||
from .nodes.lora_extract import (
|
||||
LoRASaveFromFile, LoRACheckpointSaveFromFile
|
||||
)
|
||||
|
||||
NODE_DISPLAY_NAME_MAPPINGS = {
|
||||
"ModelMetaKeys": "Get Diffusion Model Metadata & Keys",
|
||||
"TextEncoderMetaKeys": "Get Text Encoder Metadata & Keys",
|
||||
"LoRAMetaKeys": "Get LoRA Metadata & Keys",
|
||||
"CheckpointMetaKeys": "Get Checkpoint Metadata & Keys",
|
||||
"EmbeddingMetaKeys": "Get Embedding Metadata & Keys",
|
||||
"ModelRenameKeys": "ModelRenameKeys",
|
||||
"TextEncoderRenameKeys": "TextEncoderRenameKeys",
|
||||
"LoRARenameKeys": "LoRARenameKeys",
|
||||
"CheckpointRenameKeys": "CheckpointRenameKeys",
|
||||
"EmbeddingRenameKeys": "EmbeddingRenameKeys",
|
||||
"ModelPruneKeys": "ModelPruneKeys",
|
||||
"TextEncoderPruneKeys": "TextEncoderPruneKeys",
|
||||
"LoRAPruneKeys": "LoRAPruneKeys",
|
||||
"CheckpointPruneKeys": "Prune Checkpoint Keys",
|
||||
"EmbeddingPruneKeys": "Prune Embedding Keys",
|
||||
"ModelTwoMerger": "Merge Models (2 Models)",
|
||||
"TextEncoderTwoMerger": "Merge Text Encoders (2 Models)",
|
||||
"LoRATwoMerger": "Merge LoRAs (2 Models)",
|
||||
"CheckpointTwoMerger": "Merge Checkpoints (2 Models)",
|
||||
"EmbeddingTwoMerger": "Merge Embeddings (2 Models)",
|
||||
"ModelThreeMerger": "Merge Models (3 Models)",
|
||||
"TextEncoderThreeMerger": "Merge Text Encoders (3 Models)",
|
||||
"LoRAThreeMerger": "Merge LoRAs (3 Models)",
|
||||
"CheckpointThreeMerger": "Merge Checkpoints (3 Models)",
|
||||
"EmbeddingThreeMerger": "Merge Embeddings (3 Models)",
|
||||
}
|
||||
|
||||
class ModelUtilsExtension(ComfyExtension):
|
||||
@override
|
||||
async def get_node_list(self) -> list[type[io.ComfyNode]]:
|
||||
return [
|
||||
# MetaKeys
|
||||
ModelMetaKeys, TextEncoderMetaKeys, LoRAMetaKeys,
|
||||
CheckpointMetaKeys, EmbeddingMetaKeys,
|
||||
# RenameKeys
|
||||
ModelRenameKeys, TextEncoderRenameKeys, LoRARenameKeys,
|
||||
CheckpointRenameKeys, EmbeddingRenameKeys,
|
||||
# PruneKeys
|
||||
ModelPruneKeys, TextEncoderPruneKeys, LoRAPruneKeys,
|
||||
CheckpointPruneKeys, EmbeddingPruneKeys,
|
||||
# Two-Model Mergers
|
||||
ModelTwoMerger, TextEncoderTwoMerger, LoRATwoMerger,
|
||||
CheckpointTwoMerger, EmbeddingTwoMerger,
|
||||
# Three-Model Mergers
|
||||
ModelThreeMerger, TextEncoderThreeMerger, LoRAThreeMerger,
|
||||
CheckpointThreeMerger, EmbeddingThreeMerger,
|
||||
# LoRA Extraction
|
||||
LoRASaveFromFile, LoRACheckpointSaveFromFile,
|
||||
]
|
||||
|
||||
|
||||
async def comfy_entrypoint() -> ModelUtilsExtension:
|
||||
return ModelUtilsExtension()
|
||||
@@ -1,12 +0,0 @@
|
||||
from .metakeys import *
|
||||
from .renamekeys import *
|
||||
from .prunekeys import *
|
||||
from .merger import *
|
||||
|
||||
__all__ = [
|
||||
"ModelMetaKeys", "TextEncoderMetaKeys", "LoRAMetaKeys", "CheckpointMetaKeys", "EmbeddingMetaKeys",
|
||||
"ModelRenameKeys", "TextEncoderRenameKeys", "LoRARenameKeys", "CheckpointRenameKeys", "EmbeddingRenameKeys",
|
||||
"ModelPruneKeys", "TextEncoderPruneKeys", "LoRAPruneKeys", "CheckpointPruneKeys", "EmbeddingPruneKeys",
|
||||
"CheckpointTwoMerger", "ModelTwoMerger", "TextEncoderTwoMerger", "LoRATwoMerger", "EmbeddingTwoMerger",
|
||||
"CheckpointThreeMerger", "ModelThreeMerger", "TextEncoderThreeMerger", "LoRAThreeMerger", "EmbeddingThreeMerger",
|
||||
]
|
||||
@@ -0,0 +1,350 @@
|
||||
"""
|
||||
Memory-efficient LoRA extraction node.
|
||||
|
||||
This node loads two models from files and extracts a LoRA by computing the difference,
|
||||
without requiring the full models to be loaded into GPU memory via ComfyUI's model management.
|
||||
"""
|
||||
import os
|
||||
import torch
|
||||
import folder_paths
|
||||
import comfy.utils
|
||||
from comfy_api.latest import io
|
||||
from safetensors.torch import save_file
|
||||
from .merger_utils import MemoryEfficientSafeOpen
|
||||
|
||||
|
||||
CLAMP_QUANTILE = 0.99
|
||||
|
||||
|
||||
def extract_lora(diff, rank):
|
||||
"""Extract LoRA weights from a weight difference tensor using SVD."""
|
||||
conv2d = (len(diff.shape) == 4)
|
||||
kernel_size = None if not conv2d else diff.size()[2:4]
|
||||
conv2d_3x3 = conv2d and kernel_size != (1, 1)
|
||||
out_dim, in_dim = diff.size()[0:2]
|
||||
rank = min(rank, in_dim, out_dim)
|
||||
|
||||
if conv2d:
|
||||
if conv2d_3x3:
|
||||
diff = diff.flatten(start_dim=1)
|
||||
else:
|
||||
diff = diff.squeeze()
|
||||
|
||||
U, S, Vh = torch.linalg.svd(diff.float())
|
||||
U = U[:, :rank]
|
||||
S = S[:rank]
|
||||
U = U @ torch.diag(S)
|
||||
Vh = Vh[:rank, :]
|
||||
|
||||
dist = torch.cat([U.flatten(), Vh.flatten()])
|
||||
hi_val = torch.quantile(dist, CLAMP_QUANTILE)
|
||||
low_val = -hi_val
|
||||
|
||||
U = U.clamp(low_val, hi_val)
|
||||
Vh = Vh.clamp(low_val, hi_val)
|
||||
if conv2d:
|
||||
U = U.reshape(out_dim, rank, 1, 1)
|
||||
Vh = Vh.reshape(rank, in_dim, kernel_size[0], kernel_size[1])
|
||||
return (U, Vh)
|
||||
|
||||
|
||||
def extract_lora_from_files(
|
||||
model_a_path: str,
|
||||
model_b_path: str,
|
||||
rank: int,
|
||||
lora_type: str,
|
||||
bias_diff: bool,
|
||||
prefix_lora: str,
|
||||
process_device: str,
|
||||
save_dtype: str,
|
||||
) -> dict[str, torch.Tensor]:
|
||||
"""
|
||||
Memory-efficient LoRA extraction by streaming tensors from safetensors files.
|
||||
|
||||
Args:
|
||||
model_a_path: Path to the finetuned/modified model
|
||||
model_b_path: Path to the base/original model
|
||||
rank: LoRA rank for SVD decomposition
|
||||
lora_type: "standard" for LoRA, "full_diff" for full difference
|
||||
bias_diff: Whether to include bias differences
|
||||
prefix_lora: Prefix for LoRA keys (e.g., "diffusion_model.")
|
||||
process_device: Device for processing ("cuda" or "cpu")
|
||||
save_dtype: Output dtype ("fp16", "bf16", "fp32")
|
||||
|
||||
Returns:
|
||||
Dictionary of LoRA tensors
|
||||
"""
|
||||
output_sd = {}
|
||||
|
||||
# Get dtype for saving
|
||||
save_torch_dtype = {
|
||||
"fp32": torch.float32,
|
||||
"fp16": torch.float16,
|
||||
"bf16": torch.bfloat16
|
||||
}.get(save_dtype, torch.float16)
|
||||
|
||||
# Open both files with memory-efficient handler
|
||||
handler_a = MemoryEfficientSafeOpen(model_a_path)
|
||||
handler_b = MemoryEfficientSafeOpen(model_b_path)
|
||||
|
||||
try:
|
||||
keys_a = set(handler_a.keys())
|
||||
keys_b = set(handler_b.keys())
|
||||
|
||||
# Only process keys that exist in both models
|
||||
common_keys = keys_a & keys_b
|
||||
weight_keys = [k for k in common_keys if k.endswith(".weight")]
|
||||
bias_keys = [k for k in common_keys if k.endswith(".bias")] if bias_diff else []
|
||||
|
||||
# Log key differences
|
||||
missing_in_b = keys_a - keys_b
|
||||
extra_in_b = keys_b - keys_a
|
||||
if missing_in_b:
|
||||
print(f"[LoRA Extract] {len(missing_in_b)} keys in model A not in model B (skipped)")
|
||||
if extra_in_b:
|
||||
print(f"[LoRA Extract] {len(extra_in_b)} keys in model B not in model A (ignored)")
|
||||
|
||||
pbar = comfy.utils.ProgressBar(len(weight_keys) + len(bias_keys))
|
||||
|
||||
for key in weight_keys:
|
||||
# Load tensors one at a time
|
||||
tensor_a = handler_a.get_tensor(key).to(device=process_device, dtype=torch.float32)
|
||||
tensor_b = handler_b.get_tensor(key).to(device=process_device, dtype=torch.float32)
|
||||
|
||||
# Compute difference: finetuned - base
|
||||
weight_diff = tensor_a - tensor_b
|
||||
|
||||
# Clean up immediately
|
||||
del tensor_a, tensor_b
|
||||
|
||||
# Extract LoRA key name
|
||||
lora_key = key[:-7] # Remove ".weight"
|
||||
|
||||
if lora_type == "standard":
|
||||
if weight_diff.ndim < 2:
|
||||
# 1D weights (e.g., LayerNorm) - store as diff if bias_diff enabled
|
||||
if bias_diff:
|
||||
output_sd[f"{prefix_lora}{lora_key}.diff"] = weight_diff.to(save_torch_dtype).cpu()
|
||||
else:
|
||||
try:
|
||||
lora_up, lora_down = extract_lora(weight_diff, rank)
|
||||
output_sd[f"{prefix_lora}{lora_key}.lora_up.weight"] = lora_up.to(save_torch_dtype).cpu()
|
||||
output_sd[f"{prefix_lora}{lora_key}.lora_down.weight"] = lora_down.to(save_torch_dtype).cpu()
|
||||
except Exception as e:
|
||||
print(f"[LoRA Extract] Could not extract LoRA for {key}: {e}")
|
||||
|
||||
elif lora_type == "full_diff":
|
||||
output_sd[f"{prefix_lora}{lora_key}.diff"] = weight_diff.to(save_torch_dtype).cpu()
|
||||
|
||||
del weight_diff
|
||||
pbar.update(1)
|
||||
|
||||
# Process bias keys
|
||||
for key in bias_keys:
|
||||
tensor_a = handler_a.get_tensor(key).to(device=process_device, dtype=torch.float32)
|
||||
tensor_b = handler_b.get_tensor(key).to(device=process_device, dtype=torch.float32)
|
||||
|
||||
bias_diff_tensor = tensor_a - tensor_b
|
||||
del tensor_a, tensor_b
|
||||
|
||||
lora_key = key[:-5] # Remove ".bias"
|
||||
output_sd[f"{prefix_lora}{lora_key}.diff_b"] = bias_diff_tensor.to(save_torch_dtype).cpu()
|
||||
|
||||
del bias_diff_tensor
|
||||
pbar.update(1)
|
||||
|
||||
finally:
|
||||
handler_a.__exit__(None, None, None)
|
||||
handler_b.__exit__(None, None, None)
|
||||
|
||||
return output_sd
|
||||
|
||||
|
||||
class LoRASaveFromFile(io.ComfyNode):
|
||||
"""
|
||||
Memory-efficient LoRA extraction from safetensors files.
|
||||
|
||||
Unlike the standard LoraSave node which requires pre-loaded and subtracted models,
|
||||
this node loads models directly from files one tensor at a time, making it much
|
||||
more memory efficient for extracting LoRAs from large models.
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="LoRASaveFromFile",
|
||||
display_name="Extract LoRA from Files (Memory Efficient)",
|
||||
category="ModelUtils/LoRA",
|
||||
description="Extracts a LoRA by computing the difference between two model files. "
|
||||
"Much more memory efficient than using ModelSubtract + LoraSave.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"model_a",
|
||||
options=folder_paths.get_filename_list("diffusion_models"),
|
||||
tooltip="The finetuned/modified model (A - B = LoRA)"
|
||||
),
|
||||
io.Combo.Input(
|
||||
"model_b",
|
||||
options=folder_paths.get_filename_list("diffusion_models"),
|
||||
tooltip="The base/original model (A - B = LoRA)"
|
||||
),
|
||||
io.Int.Input("rank", default=64, min=1, max=4096, step=1,
|
||||
tooltip="LoRA rank - higher preserves more detail but increases size"),
|
||||
io.Combo.Input("lora_type", options=["standard", "full_diff"], default="standard",
|
||||
tooltip="standard: SVD decomposition to lora_up/lora_down, full_diff: raw weight differences"),
|
||||
io.Boolean.Input("bias_diff", default=True,
|
||||
tooltip="Include bias differences in the LoRA"),
|
||||
io.String.Input("output_filename", default="extracted_lora",
|
||||
tooltip="Output filename (without extension)"),
|
||||
io.Combo.Input("save_dtype", options=["fp16", "bf16", "fp32"], default="fp16"),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"], default="cuda",
|
||||
tooltip="Device for SVD computation"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
is_output_node=True,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(
|
||||
cls,
|
||||
model_a: str,
|
||||
model_b: str,
|
||||
rank: int,
|
||||
lora_type: str,
|
||||
bias_diff: bool,
|
||||
output_filename: str,
|
||||
save_dtype: str,
|
||||
process_device: str,
|
||||
) -> io.NodeOutput:
|
||||
# Get full paths
|
||||
model_a_path = folder_paths.get_full_path_or_raise("diffusion_models", model_a)
|
||||
model_b_path = folder_paths.get_full_path_or_raise("diffusion_models", model_b)
|
||||
|
||||
# Extract LoRA
|
||||
output_sd = extract_lora_from_files(
|
||||
model_a_path=model_a_path,
|
||||
model_b_path=model_b_path,
|
||||
rank=rank,
|
||||
lora_type=lora_type,
|
||||
bias_diff=bias_diff,
|
||||
prefix_lora="diffusion_model.",
|
||||
process_device=process_device,
|
||||
save_dtype=save_dtype,
|
||||
)
|
||||
|
||||
if not output_sd:
|
||||
raise ValueError("No LoRA weights extracted - models may be identical or incompatible")
|
||||
|
||||
# Save to loras folder
|
||||
output_dir = folder_paths.get_folder_paths("loras")[0]
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
output_path = os.path.join(output_dir, f"{output_filename.strip()}.safetensors")
|
||||
|
||||
save_file(output_sd, output_path)
|
||||
print(f"[LoRA Extract] Saved {len(output_sd)} tensors to {output_path}")
|
||||
|
||||
return io.NodeOutput(output_path)
|
||||
|
||||
|
||||
class LoRACheckpointSaveFromFile(io.ComfyNode):
|
||||
"""
|
||||
Memory-efficient LoRA extraction from checkpoint files (which may contain both model and text encoder).
|
||||
"""
|
||||
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="LoRACheckpointSaveFromFile",
|
||||
display_name="Extract LoRA from Checkpoints (Memory Efficient)",
|
||||
category="ModelUtils/LoRA",
|
||||
description="Extracts a LoRA by computing the difference between two checkpoint files. "
|
||||
"Handles both diffusion model and text encoder weights.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"checkpoint_a",
|
||||
options=folder_paths.get_filename_list("checkpoints"),
|
||||
tooltip="The finetuned/modified checkpoint (A - B = LoRA)"
|
||||
),
|
||||
io.Combo.Input(
|
||||
"checkpoint_b",
|
||||
options=folder_paths.get_filename_list("checkpoints"),
|
||||
tooltip="The base/original checkpoint (A - B = LoRA)"
|
||||
),
|
||||
io.Int.Input("rank", default=64, min=1, max=4096, step=1,
|
||||
tooltip="LoRA rank - higher preserves more detail but increases size"),
|
||||
io.Combo.Input("lora_type", options=["standard", "full_diff"], default="standard",
|
||||
tooltip="standard: SVD decomposition, full_diff: raw weight differences"),
|
||||
io.Boolean.Input("bias_diff", default=True,
|
||||
tooltip="Include bias differences in the LoRA"),
|
||||
io.String.Input("output_filename", default="extracted_lora",
|
||||
tooltip="Output filename (without extension)"),
|
||||
io.Combo.Input("save_dtype", options=["fp16", "bf16", "fp32"], default="fp16"),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"], default="cuda",
|
||||
tooltip="Device for SVD computation"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
is_output_node=True,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(
|
||||
cls,
|
||||
checkpoint_a: str,
|
||||
checkpoint_b: str,
|
||||
rank: int,
|
||||
lora_type: str,
|
||||
bias_diff: bool,
|
||||
output_filename: str,
|
||||
save_dtype: str,
|
||||
process_device: str,
|
||||
) -> io.NodeOutput:
|
||||
# Get full paths
|
||||
checkpoint_a_path = folder_paths.get_full_path_or_raise("checkpoints", checkpoint_a)
|
||||
checkpoint_b_path = folder_paths.get_full_path_or_raise("checkpoints", checkpoint_b)
|
||||
|
||||
# Open handlers to detect prefixes
|
||||
handler_a = MemoryEfficientSafeOpen(checkpoint_a_path)
|
||||
keys = handler_a.keys()
|
||||
handler_a.__exit__(None, None, None)
|
||||
|
||||
# Auto-detect key prefixes
|
||||
has_model = any(k.startswith("model.") for k in keys)
|
||||
has_diffusion = any(k.startswith("diffusion_model.") for k in keys)
|
||||
has_text_encoders = any(k.startswith("text_encoders.") or k.startswith("cond_stage_model.") for k in keys)
|
||||
|
||||
output_sd = {}
|
||||
|
||||
# Extract diffusion model LoRA
|
||||
if has_model or has_diffusion:
|
||||
prefix = "model.diffusion_model." if has_model else "diffusion_model."
|
||||
model_lora = extract_lora_from_files(
|
||||
model_a_path=checkpoint_a_path,
|
||||
model_b_path=checkpoint_b_path,
|
||||
rank=rank,
|
||||
lora_type=lora_type,
|
||||
bias_diff=bias_diff,
|
||||
prefix_lora="diffusion_model.",
|
||||
process_device=process_device,
|
||||
save_dtype=save_dtype,
|
||||
)
|
||||
output_sd.update(model_lora)
|
||||
|
||||
# Note: Text encoder extraction would need additional prefix handling
|
||||
# For full checkpoint support, this would need to be expanded
|
||||
|
||||
if not output_sd:
|
||||
raise ValueError("No LoRA weights extracted - checkpoints may be identical or incompatible")
|
||||
|
||||
# Save to loras folder
|
||||
output_dir = folder_paths.get_folder_paths("loras")[0]
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
output_path = os.path.join(output_dir, f"{output_filename.strip()}.safetensors")
|
||||
|
||||
save_file(output_sd, output_path)
|
||||
print(f"[LoRA Extract] Saved {len(output_sd)} tensors to {output_path}")
|
||||
|
||||
return io.NodeOutput(output_path)
|
||||
+522
-106
@@ -3,6 +3,7 @@ import re
|
||||
import torch
|
||||
import folder_paths
|
||||
import comfy.utils
|
||||
from comfy_api.latest import io
|
||||
from safetensors.torch import save_file
|
||||
from .merger_ops import TWO_MODEL_MODES, THREE_MODEL_MODES, MissingTensorBehavior, MissingTensorError
|
||||
from .merger_utils import MemoryEfficientSafeOpen
|
||||
@@ -45,60 +46,24 @@ def _matches_any_pattern(key, patterns):
|
||||
return False
|
||||
|
||||
|
||||
# --- UI Dictionary Helper Functions ---
|
||||
def _create_two_model_inputs(model_folder_name):
|
||||
return {"required": {
|
||||
"execution_mode": (["MERGE", "DOCUMENTATION ONLY"],),
|
||||
"model_a": (["None"] + folder_paths.get_filename_list(model_folder_name),),
|
||||
"model_b": (["None"] + folder_paths.get_filename_list(model_folder_name),),
|
||||
"calc_mode": ([m.name for m in TWO_MODEL_MODES],),
|
||||
"mismatch_mode": (["skip", "zeros", "error"], {"default": "skip"}),
|
||||
"alpha": ("FLOAT", {"default": 0.5, "min": -2.0, "max": 3.0, "step": 0.01}),
|
||||
"beta": ("FLOAT", {"default": 0.5, "min": -2.0, "max": 3.0, "step": 0.01}),
|
||||
"gamma": ("FLOAT", {"default": 0.99, "min": 0.0, "max": 1.0, "step": 0.001}),
|
||||
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
|
||||
"output_filename": ("STRING", {"default": "merged_2_model"}),
|
||||
"save_dtype": (["fp32", "fp16", "bf16"],),
|
||||
"process_device": (["cuda", "cpu"],),
|
||||
"exclude_patterns": ("STRING", {"default": "", "multiline": True, "placeholder": "Regex patterns (space-separated) - matching layers keep Model A only"}),
|
||||
"discard_patterns": ("STRING", {"default": "", "multiline": True, "placeholder": "Regex patterns (space-separated) - matching layers removed from output"}),
|
||||
}}
|
||||
|
||||
|
||||
def _create_three_model_inputs(model_folder_name):
|
||||
return {"required": {
|
||||
"execution_mode": (["MERGE", "DOCUMENTATION ONLY"],),
|
||||
"model_a": (["None"] + folder_paths.get_filename_list(model_folder_name),),
|
||||
"model_b": (["None"] + folder_paths.get_filename_list(model_folder_name),),
|
||||
"model_c": (["None"] + folder_paths.get_filename_list(model_folder_name),),
|
||||
"calc_mode": ([m.name for m in THREE_MODEL_MODES],),
|
||||
"mismatch_mode": (["skip", "zeros", "error"], {"default": "skip"}),
|
||||
"alpha": ("FLOAT", {"default": 0.5, "min": -2.0, "max": 3.0, "step": 0.01}),
|
||||
"beta": ("FLOAT", {"default": 0.5, "min": -2.0, "max": 3.0, "step": 0.01}),
|
||||
"gamma": ("FLOAT", {"default": 0.5, "min": -2.0, "max": 3.0, "step": 0.01}),
|
||||
"delta": ("FLOAT", {"default": 0.5, "min": -2.0, "max": 3.0, "step": 0.01}),
|
||||
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
|
||||
"output_filename": ("STRING", {"default": "merged_3_model"}),
|
||||
"save_dtype": (["fp32", "fp16", "bf16"],),
|
||||
"process_device": (["cuda", "cpu"],),
|
||||
"exclude_patterns": ("STRING", {"default": "", "multiline": True, "placeholder": "Regex patterns (space-separated) - matching layers keep Model A only"}),
|
||||
"discard_patterns": ("STRING", {"default": "", "multiline": True, "placeholder": "Regex patterns (space-separated) - matching layers removed from output"}),
|
||||
}}
|
||||
|
||||
|
||||
# --- Base Classes for Backend Logic ONLY ---
|
||||
class MergerLogic:
|
||||
def _execute_merge(self, model_names, calc_mode, all_modes, recipe_params, model_type):
|
||||
"""Shared merge execution logic for all merger nodes."""
|
||||
|
||||
@staticmethod
|
||||
def execute_merge(model_names, calc_mode, all_modes, recipe_params, model_type):
|
||||
calc_mode_class = next((m for m in all_modes if m.name == calc_mode), None)
|
||||
if not calc_mode_class: raise ValueError(f"Calc mode '{calc_mode}' not found.")
|
||||
if not calc_mode_class:
|
||||
raise ValueError(f"Calc mode '{calc_mode}' not found.")
|
||||
primary_model_name = model_names.get('model_a')
|
||||
if not primary_model_name or primary_model_name == "None": raise ValueError("Model A is required to run the merge.")
|
||||
if not primary_model_name or primary_model_name == "None":
|
||||
raise ValueError("Model A is required to run the merge.")
|
||||
|
||||
handlers = {}
|
||||
for name in model_names.values():
|
||||
if name and name != "None":
|
||||
path = folder_paths.get_full_path(model_type, name)
|
||||
if not path: raise FileNotFoundError(f"Model '{name}' not found.")
|
||||
if not path:
|
||||
raise FileNotFoundError(f"Model '{name}' not found.")
|
||||
handlers[name] = MemoryEfficientSafeOpen(path)
|
||||
|
||||
primary_handler = handlers[primary_model_name]
|
||||
@@ -195,24 +160,58 @@ class MergerLogic:
|
||||
if error_keys:
|
||||
print(f"[Merger] Unexpected errors on {len(error_keys)} keys: {error_keys[:5]}{'...' if len(error_keys) > 5 else ''}")
|
||||
|
||||
for handler in handlers.values(): handler.__exit__(None, None, None)
|
||||
for handler in handlers.values():
|
||||
handler.__exit__(None, None, None)
|
||||
output_folder = "loras" if calc_mode == "SVD LoRA Extraction" else model_type
|
||||
output_dir = folder_paths.get_folder_paths(output_folder)[0]
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
output_filename = recipe_params.get("output_filename")
|
||||
output_path = os.path.join(output_dir, f"{output_filename}.safetensors")
|
||||
save_file(merged_state_dict, output_path, metadata=metadata)
|
||||
return (f"{output_filename}.safetensors",)
|
||||
return f"{output_filename}.safetensors"
|
||||
|
||||
|
||||
class BaseTwoMerger(MergerLogic):
|
||||
FUNCTION = "merge"; CATEGORY = "ModelUtils/Merging"
|
||||
RETURN_TYPES = ("STRING", "STRING",); RETURN_NAMES = ("output_filename", "documentation",)
|
||||
# --- Two-Model Merger Nodes ---
|
||||
|
||||
def merge(self, execution_mode, model_a, model_b, calc_mode, mismatch_mode, alpha, beta, gamma, seed, output_filename, save_dtype, process_device, exclude_patterns, discard_patterns):
|
||||
class CheckpointTwoMerger(io.ComfyNode):
|
||||
MODEL_TYPE = "checkpoints"
|
||||
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="CheckpointTwoMerger",
|
||||
display_name="Merge Checkpoints (2 Models)",
|
||||
category="ModelUtils/Merging",
|
||||
inputs=[
|
||||
io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
|
||||
io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("checkpoints")),
|
||||
io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("checkpoints")),
|
||||
io.Combo.Input("calc_mode", options=[m.name for m in TWO_MODEL_MODES]),
|
||||
io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
|
||||
io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("gamma", default=0.99, min=0.0, max=1.0, step=0.001),
|
||||
io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
|
||||
io.String.Input("output_filename", default="merged_2_checkpoint"),
|
||||
io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"]),
|
||||
io.String.Input("exclude_patterns", default="", multiline=True),
|
||||
io.String.Input("discard_patterns", default="", multiline=True),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_filename"),
|
||||
io.String.Output(display_name="documentation"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, execution_mode: str, model_a: str, model_b: str,
|
||||
calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
|
||||
gamma: float, seed: int, output_filename: str, save_dtype: str,
|
||||
process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
|
||||
doc = load_documentation_from_file('merger_2_model_modes.md')
|
||||
if execution_mode == "DOCUMENTATION ONLY":
|
||||
return ("Documentation mode active. No merge performed.", doc)
|
||||
return io.NodeOutput("Documentation mode active. No merge performed.", doc)
|
||||
|
||||
recipe_params = {
|
||||
"model_a": model_a, "model_b": model_b, "calc_mode": calc_mode,
|
||||
@@ -223,18 +222,265 @@ class BaseTwoMerger(MergerLogic):
|
||||
"exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
|
||||
}
|
||||
model_names = {"model_a": model_a, "model_b": model_b}
|
||||
filename, = self._execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, self.MODEL_TYPE)
|
||||
return (filename, doc)
|
||||
filename = MergerLogic.execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
|
||||
return io.NodeOutput(filename, doc)
|
||||
|
||||
|
||||
class BaseThreeMerger(MergerLogic):
|
||||
FUNCTION = "merge"; CATEGORY = "ModelUtils/Merging"
|
||||
RETURN_TYPES = ("STRING", "STRING",); RETURN_NAMES = ("output_filename", "documentation",)
|
||||
class ModelTwoMerger(io.ComfyNode):
|
||||
MODEL_TYPE = "diffusion_models"
|
||||
|
||||
def merge(self, execution_mode, model_a, model_b, model_c, calc_mode, mismatch_mode, alpha, beta, gamma, delta, seed, output_filename, save_dtype, process_device, exclude_patterns, discard_patterns):
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="ModelTwoMerger",
|
||||
display_name="Merge Models (2 Models)",
|
||||
category="ModelUtils/Merging",
|
||||
inputs=[
|
||||
io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
|
||||
io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("diffusion_models")),
|
||||
io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("diffusion_models")),
|
||||
io.Combo.Input("calc_mode", options=[m.name for m in TWO_MODEL_MODES]),
|
||||
io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
|
||||
io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("gamma", default=0.99, min=0.0, max=1.0, step=0.001),
|
||||
io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
|
||||
io.String.Input("output_filename", default="merged_2_model"),
|
||||
io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"]),
|
||||
io.String.Input("exclude_patterns", default="", multiline=True),
|
||||
io.String.Input("discard_patterns", default="", multiline=True),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_filename"),
|
||||
io.String.Output(display_name="documentation"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, execution_mode: str, model_a: str, model_b: str,
|
||||
calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
|
||||
gamma: float, seed: int, output_filename: str, save_dtype: str,
|
||||
process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
|
||||
doc = load_documentation_from_file('merger_2_model_modes.md')
|
||||
if execution_mode == "DOCUMENTATION ONLY":
|
||||
return io.NodeOutput("Documentation mode active. No merge performed.", doc)
|
||||
|
||||
recipe_params = {
|
||||
"model_a": model_a, "model_b": model_b, "calc_mode": calc_mode,
|
||||
"mismatch_mode": mismatch_mode,
|
||||
"alpha": alpha, "beta": beta, "gamma": gamma, "seed": seed,
|
||||
"output_filename": output_filename, "save_dtype": save_dtype,
|
||||
"device": process_device, "dtype": torch.float32,
|
||||
"exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
|
||||
}
|
||||
model_names = {"model_a": model_a, "model_b": model_b}
|
||||
filename = MergerLogic.execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
|
||||
return io.NodeOutput(filename, doc)
|
||||
|
||||
|
||||
class TextEncoderTwoMerger(io.ComfyNode):
|
||||
MODEL_TYPE = "text_encoders"
|
||||
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="TextEncoderTwoMerger",
|
||||
display_name="Merge Text Encoders (2 Models)",
|
||||
category="ModelUtils/Merging",
|
||||
inputs=[
|
||||
io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
|
||||
io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("text_encoders")),
|
||||
io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("text_encoders")),
|
||||
io.Combo.Input("calc_mode", options=[m.name for m in TWO_MODEL_MODES]),
|
||||
io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
|
||||
io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("gamma", default=0.99, min=0.0, max=1.0, step=0.001),
|
||||
io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
|
||||
io.String.Input("output_filename", default="merged_2_textencoder"),
|
||||
io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"]),
|
||||
io.String.Input("exclude_patterns", default="", multiline=True),
|
||||
io.String.Input("discard_patterns", default="", multiline=True),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_filename"),
|
||||
io.String.Output(display_name="documentation"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, execution_mode: str, model_a: str, model_b: str,
|
||||
calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
|
||||
gamma: float, seed: int, output_filename: str, save_dtype: str,
|
||||
process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
|
||||
doc = load_documentation_from_file('merger_2_model_modes.md')
|
||||
if execution_mode == "DOCUMENTATION ONLY":
|
||||
return io.NodeOutput("Documentation mode active. No merge performed.", doc)
|
||||
|
||||
recipe_params = {
|
||||
"model_a": model_a, "model_b": model_b, "calc_mode": calc_mode,
|
||||
"mismatch_mode": mismatch_mode,
|
||||
"alpha": alpha, "beta": beta, "gamma": gamma, "seed": seed,
|
||||
"output_filename": output_filename, "save_dtype": save_dtype,
|
||||
"device": process_device, "dtype": torch.float32,
|
||||
"exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
|
||||
}
|
||||
model_names = {"model_a": model_a, "model_b": model_b}
|
||||
filename = MergerLogic.execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
|
||||
return io.NodeOutput(filename, doc)
|
||||
|
||||
|
||||
class LoRATwoMerger(io.ComfyNode):
|
||||
MODEL_TYPE = "loras"
|
||||
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="LoRATwoMerger",
|
||||
display_name="Merge LoRAs (2 Models)",
|
||||
category="ModelUtils/Merging",
|
||||
inputs=[
|
||||
io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
|
||||
io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("loras")),
|
||||
io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("loras")),
|
||||
io.Combo.Input("calc_mode", options=[m.name for m in TWO_MODEL_MODES]),
|
||||
io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
|
||||
io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("gamma", default=0.99, min=0.0, max=1.0, step=0.001),
|
||||
io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
|
||||
io.String.Input("output_filename", default="merged_2_lora"),
|
||||
io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"]),
|
||||
io.String.Input("exclude_patterns", default="", multiline=True),
|
||||
io.String.Input("discard_patterns", default="", multiline=True),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_filename"),
|
||||
io.String.Output(display_name="documentation"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, execution_mode: str, model_a: str, model_b: str,
|
||||
calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
|
||||
gamma: float, seed: int, output_filename: str, save_dtype: str,
|
||||
process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
|
||||
doc = load_documentation_from_file('merger_2_model_modes.md')
|
||||
if execution_mode == "DOCUMENTATION ONLY":
|
||||
return io.NodeOutput("Documentation mode active. No merge performed.", doc)
|
||||
|
||||
recipe_params = {
|
||||
"model_a": model_a, "model_b": model_b, "calc_mode": calc_mode,
|
||||
"mismatch_mode": mismatch_mode,
|
||||
"alpha": alpha, "beta": beta, "gamma": gamma, "seed": seed,
|
||||
"output_filename": output_filename, "save_dtype": save_dtype,
|
||||
"device": process_device, "dtype": torch.float32,
|
||||
"exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
|
||||
}
|
||||
model_names = {"model_a": model_a, "model_b": model_b}
|
||||
filename = MergerLogic.execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
|
||||
return io.NodeOutput(filename, doc)
|
||||
|
||||
|
||||
class EmbeddingTwoMerger(io.ComfyNode):
|
||||
MODEL_TYPE = "embeddings"
|
||||
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="EmbeddingTwoMerger",
|
||||
display_name="Merge Embeddings (2 Models)",
|
||||
category="ModelUtils/Merging",
|
||||
inputs=[
|
||||
io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
|
||||
io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("embeddings")),
|
||||
io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("embeddings")),
|
||||
io.Combo.Input("calc_mode", options=[m.name for m in TWO_MODEL_MODES]),
|
||||
io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
|
||||
io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("gamma", default=0.99, min=0.0, max=1.0, step=0.001),
|
||||
io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
|
||||
io.String.Input("output_filename", default="merged_2_embedding"),
|
||||
io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"]),
|
||||
io.String.Input("exclude_patterns", default="", multiline=True),
|
||||
io.String.Input("discard_patterns", default="", multiline=True),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_filename"),
|
||||
io.String.Output(display_name="documentation"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, execution_mode: str, model_a: str, model_b: str,
|
||||
calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
|
||||
gamma: float, seed: int, output_filename: str, save_dtype: str,
|
||||
process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
|
||||
doc = load_documentation_from_file('merger_2_model_modes.md')
|
||||
if execution_mode == "DOCUMENTATION ONLY":
|
||||
return io.NodeOutput("Documentation mode active. No merge performed.", doc)
|
||||
|
||||
recipe_params = {
|
||||
"model_a": model_a, "model_b": model_b, "calc_mode": calc_mode,
|
||||
"mismatch_mode": mismatch_mode,
|
||||
"alpha": alpha, "beta": beta, "gamma": gamma, "seed": seed,
|
||||
"output_filename": output_filename, "save_dtype": save_dtype,
|
||||
"device": process_device, "dtype": torch.float32,
|
||||
"exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
|
||||
}
|
||||
model_names = {"model_a": model_a, "model_b": model_b}
|
||||
filename = MergerLogic.execute_merge(model_names, calc_mode, TWO_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
|
||||
return io.NodeOutput(filename, doc)
|
||||
|
||||
|
||||
# --- Three-Model Merger Nodes ---
|
||||
|
||||
class CheckpointThreeMerger(io.ComfyNode):
|
||||
MODEL_TYPE = "checkpoints"
|
||||
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="CheckpointThreeMerger",
|
||||
display_name="Merge Checkpoints (3 Models)",
|
||||
category="ModelUtils/Merging",
|
||||
inputs=[
|
||||
io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
|
||||
io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("checkpoints")),
|
||||
io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("checkpoints")),
|
||||
io.Combo.Input("model_c", options=["None"] + folder_paths.get_filename_list("checkpoints")),
|
||||
io.Combo.Input("calc_mode", options=[m.name for m in THREE_MODEL_MODES]),
|
||||
io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
|
||||
io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("gamma", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("delta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
|
||||
io.String.Input("output_filename", default="merged_3_checkpoint"),
|
||||
io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"]),
|
||||
io.String.Input("exclude_patterns", default="", multiline=True),
|
||||
io.String.Input("discard_patterns", default="", multiline=True),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_filename"),
|
||||
io.String.Output(display_name="documentation"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, execution_mode: str, model_a: str, model_b: str, model_c: str,
|
||||
calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
|
||||
gamma: float, delta: float, seed: int, output_filename: str, save_dtype: str,
|
||||
process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
|
||||
doc = load_documentation_from_file('merger_3_model_modes.md')
|
||||
if execution_mode == "DOCUMENTATION ONLY":
|
||||
return ("Documentation mode active. No merge performed.", doc)
|
||||
return io.NodeOutput("Documentation mode active. No merge performed.", doc)
|
||||
|
||||
recipe_params = {
|
||||
"model_a": model_a, "model_b": model_b, "model_c": model_c, "calc_mode": calc_mode,
|
||||
@@ -245,55 +491,225 @@ class BaseThreeMerger(MergerLogic):
|
||||
"exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
|
||||
}
|
||||
model_names = {"model_a": model_a, "model_b": model_b, "model_c": model_c}
|
||||
filename, = self._execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, self.MODEL_TYPE)
|
||||
return (filename, doc)
|
||||
filename = MergerLogic.execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
|
||||
return io.NodeOutput(filename, doc)
|
||||
|
||||
|
||||
# --- Concrete 2-Model Nodes ---
|
||||
class CheckpointTwoMerger(BaseTwoMerger):
|
||||
MODEL_TYPE = "checkpoints"
|
||||
@classmethod
|
||||
def INPUT_TYPES(s): return _create_two_model_inputs(s.MODEL_TYPE)
|
||||
class ModelTwoMerger(BaseTwoMerger):
|
||||
class ModelThreeMerger(io.ComfyNode):
|
||||
MODEL_TYPE = "diffusion_models"
|
||||
@classmethod
|
||||
def INPUT_TYPES(s): return _create_two_model_inputs(s.MODEL_TYPE)
|
||||
class TextEncoderTwoMerger(BaseTwoMerger):
|
||||
MODEL_TYPE = "text_encoders"
|
||||
@classmethod
|
||||
def INPUT_TYPES(s): return _create_two_model_inputs(s.MODEL_TYPE)
|
||||
class LoRATwoMerger(BaseTwoMerger):
|
||||
MODEL_TYPE = "loras"
|
||||
@classmethod
|
||||
def INPUT_TYPES(s): return _create_two_model_inputs(s.MODEL_TYPE)
|
||||
class EmbeddingTwoMerger(BaseTwoMerger):
|
||||
MODEL_TYPE = "embeddings"
|
||||
@classmethod
|
||||
def INPUT_TYPES(s): return _create_two_model_inputs(s.MODEL_TYPE)
|
||||
|
||||
# --- Concrete 3-Model Nodes ---
|
||||
class CheckpointThreeMerger(BaseThreeMerger):
|
||||
MODEL_TYPE = "checkpoints"
|
||||
@classmethod
|
||||
def INPUT_TYPES(s): return _create_three_model_inputs(s.MODEL_TYPE)
|
||||
class ModelThreeMerger(BaseThreeMerger):
|
||||
MODEL_TYPE = "diffusion_models"
|
||||
@classmethod
|
||||
def INPUT_TYPES(s): return _create_three_model_inputs(s.MODEL_TYPE)
|
||||
class TextEncoderThreeMerger(BaseThreeMerger):
|
||||
MODEL_TYPE = "text_encoders"
|
||||
@classmethod
|
||||
def INPUT_TYPES(s): return _create_three_model_inputs(s.MODEL_TYPE)
|
||||
class LoRAThreeMerger(BaseThreeMerger):
|
||||
MODEL_TYPE = "loras"
|
||||
@classmethod
|
||||
def INPUT_TYPES(s): return _create_three_model_inputs(s.MODEL_TYPE)
|
||||
class EmbeddingThreeMerger(BaseThreeMerger):
|
||||
MODEL_TYPE = "embeddings"
|
||||
@classmethod
|
||||
def INPUT_TYPES(s): return _create_three_model_inputs(s.MODEL_TYPE)
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="ModelThreeMerger",
|
||||
display_name="Merge Models (3 Models)",
|
||||
category="ModelUtils/Merging",
|
||||
inputs=[
|
||||
io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
|
||||
io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("diffusion_models")),
|
||||
io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("diffusion_models")),
|
||||
io.Combo.Input("model_c", options=["None"] + folder_paths.get_filename_list("diffusion_models")),
|
||||
io.Combo.Input("calc_mode", options=[m.name for m in THREE_MODEL_MODES]),
|
||||
io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
|
||||
io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("gamma", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("delta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
|
||||
io.String.Input("output_filename", default="merged_3_model"),
|
||||
io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"]),
|
||||
io.String.Input("exclude_patterns", default="", multiline=True),
|
||||
io.String.Input("discard_patterns", default="", multiline=True),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_filename"),
|
||||
io.String.Output(display_name="documentation"),
|
||||
],
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"CheckpointTwoMerger", "ModelTwoMerger", "TextEncoderTwoMerger", "LoRATwoMerger", "EmbeddingTwoMerger",
|
||||
"CheckpointThreeMerger", "ModelThreeMerger", "TextEncoderThreeMerger", "LoRAThreeMerger", "EmbeddingThreeMerger"
|
||||
]
|
||||
@classmethod
|
||||
def execute(cls, execution_mode: str, model_a: str, model_b: str, model_c: str,
|
||||
calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
|
||||
gamma: float, delta: float, seed: int, output_filename: str, save_dtype: str,
|
||||
process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
|
||||
doc = load_documentation_from_file('merger_3_model_modes.md')
|
||||
if execution_mode == "DOCUMENTATION ONLY":
|
||||
return io.NodeOutput("Documentation mode active. No merge performed.", doc)
|
||||
|
||||
recipe_params = {
|
||||
"model_a": model_a, "model_b": model_b, "model_c": model_c, "calc_mode": calc_mode,
|
||||
"mismatch_mode": mismatch_mode,
|
||||
"alpha": alpha, "beta": beta, "gamma": gamma, "delta": delta, "seed": seed,
|
||||
"output_filename": output_filename, "save_dtype": save_dtype,
|
||||
"device": process_device, "dtype": torch.float32,
|
||||
"exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
|
||||
}
|
||||
model_names = {"model_a": model_a, "model_b": model_b, "model_c": model_c}
|
||||
filename = MergerLogic.execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
|
||||
return io.NodeOutput(filename, doc)
|
||||
|
||||
|
||||
class TextEncoderThreeMerger(io.ComfyNode):
|
||||
MODEL_TYPE = "text_encoders"
|
||||
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="TextEncoderThreeMerger",
|
||||
display_name="Merge Text Encoders (3 Models)",
|
||||
category="ModelUtils/Merging",
|
||||
inputs=[
|
||||
io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
|
||||
io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("text_encoders")),
|
||||
io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("text_encoders")),
|
||||
io.Combo.Input("model_c", options=["None"] + folder_paths.get_filename_list("text_encoders")),
|
||||
io.Combo.Input("calc_mode", options=[m.name for m in THREE_MODEL_MODES]),
|
||||
io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
|
||||
io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("gamma", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("delta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
|
||||
io.String.Input("output_filename", default="merged_3_textencoder"),
|
||||
io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"]),
|
||||
io.String.Input("exclude_patterns", default="", multiline=True),
|
||||
io.String.Input("discard_patterns", default="", multiline=True),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_filename"),
|
||||
io.String.Output(display_name="documentation"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, execution_mode: str, model_a: str, model_b: str, model_c: str,
|
||||
calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
|
||||
gamma: float, delta: float, seed: int, output_filename: str, save_dtype: str,
|
||||
process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
|
||||
doc = load_documentation_from_file('merger_3_model_modes.md')
|
||||
if execution_mode == "DOCUMENTATION ONLY":
|
||||
return io.NodeOutput("Documentation mode active. No merge performed.", doc)
|
||||
|
||||
recipe_params = {
|
||||
"model_a": model_a, "model_b": model_b, "model_c": model_c, "calc_mode": calc_mode,
|
||||
"mismatch_mode": mismatch_mode,
|
||||
"alpha": alpha, "beta": beta, "gamma": gamma, "delta": delta, "seed": seed,
|
||||
"output_filename": output_filename, "save_dtype": save_dtype,
|
||||
"device": process_device, "dtype": torch.float32,
|
||||
"exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
|
||||
}
|
||||
model_names = {"model_a": model_a, "model_b": model_b, "model_c": model_c}
|
||||
filename = MergerLogic.execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
|
||||
return io.NodeOutput(filename, doc)
|
||||
|
||||
|
||||
class LoRAThreeMerger(io.ComfyNode):
|
||||
MODEL_TYPE = "loras"
|
||||
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="LoRAThreeMerger",
|
||||
display_name="Merge LoRAs (3 Models)",
|
||||
category="ModelUtils/Merging",
|
||||
inputs=[
|
||||
io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
|
||||
io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("loras")),
|
||||
io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("loras")),
|
||||
io.Combo.Input("model_c", options=["None"] + folder_paths.get_filename_list("loras")),
|
||||
io.Combo.Input("calc_mode", options=[m.name for m in THREE_MODEL_MODES]),
|
||||
io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
|
||||
io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("gamma", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("delta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
|
||||
io.String.Input("output_filename", default="merged_3_lora"),
|
||||
io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"]),
|
||||
io.String.Input("exclude_patterns", default="", multiline=True),
|
||||
io.String.Input("discard_patterns", default="", multiline=True),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_filename"),
|
||||
io.String.Output(display_name="documentation"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, execution_mode: str, model_a: str, model_b: str, model_c: str,
|
||||
calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
|
||||
gamma: float, delta: float, seed: int, output_filename: str, save_dtype: str,
|
||||
process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
|
||||
doc = load_documentation_from_file('merger_3_model_modes.md')
|
||||
if execution_mode == "DOCUMENTATION ONLY":
|
||||
return io.NodeOutput("Documentation mode active. No merge performed.", doc)
|
||||
|
||||
recipe_params = {
|
||||
"model_a": model_a, "model_b": model_b, "model_c": model_c, "calc_mode": calc_mode,
|
||||
"mismatch_mode": mismatch_mode,
|
||||
"alpha": alpha, "beta": beta, "gamma": gamma, "delta": delta, "seed": seed,
|
||||
"output_filename": output_filename, "save_dtype": save_dtype,
|
||||
"device": process_device, "dtype": torch.float32,
|
||||
"exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
|
||||
}
|
||||
model_names = {"model_a": model_a, "model_b": model_b, "model_c": model_c}
|
||||
filename = MergerLogic.execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
|
||||
return io.NodeOutput(filename, doc)
|
||||
|
||||
|
||||
class EmbeddingThreeMerger(io.ComfyNode):
|
||||
MODEL_TYPE = "embeddings"
|
||||
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="EmbeddingThreeMerger",
|
||||
display_name="Merge Embeddings (3 Models)",
|
||||
category="ModelUtils/Merging",
|
||||
inputs=[
|
||||
io.Combo.Input("execution_mode", options=["MERGE", "DOCUMENTATION ONLY"]),
|
||||
io.Combo.Input("model_a", options=["None"] + folder_paths.get_filename_list("embeddings")),
|
||||
io.Combo.Input("model_b", options=["None"] + folder_paths.get_filename_list("embeddings")),
|
||||
io.Combo.Input("model_c", options=["None"] + folder_paths.get_filename_list("embeddings")),
|
||||
io.Combo.Input("calc_mode", options=[m.name for m in THREE_MODEL_MODES]),
|
||||
io.Combo.Input("mismatch_mode", options=["skip", "zeros", "error"], default="skip"),
|
||||
io.Float.Input("alpha", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("beta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("gamma", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Float.Input("delta", default=0.5, min=-2.0, max=3.0, step=0.01),
|
||||
io.Int.Input("seed", default=0, min=0, max=0xffffffffffffffff),
|
||||
io.String.Input("output_filename", default="merged_3_embedding"),
|
||||
io.Combo.Input("save_dtype", options=["fp32", "fp16", "bf16"]),
|
||||
io.Combo.Input("process_device", options=["cuda", "cpu"]),
|
||||
io.String.Input("exclude_patterns", default="", multiline=True),
|
||||
io.String.Input("discard_patterns", default="", multiline=True),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_filename"),
|
||||
io.String.Output(display_name="documentation"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, execution_mode: str, model_a: str, model_b: str, model_c: str,
|
||||
calc_mode: str, mismatch_mode: str, alpha: float, beta: float,
|
||||
gamma: float, delta: float, seed: int, output_filename: str, save_dtype: str,
|
||||
process_device: str, exclude_patterns: str, discard_patterns: str) -> io.NodeOutput:
|
||||
doc = load_documentation_from_file('merger_3_model_modes.md')
|
||||
if execution_mode == "DOCUMENTATION ONLY":
|
||||
return io.NodeOutput("Documentation mode active. No merge performed.", doc)
|
||||
|
||||
recipe_params = {
|
||||
"model_a": model_a, "model_b": model_b, "model_c": model_c, "calc_mode": calc_mode,
|
||||
"mismatch_mode": mismatch_mode,
|
||||
"alpha": alpha, "beta": beta, "gamma": gamma, "delta": delta, "seed": seed,
|
||||
"output_filename": output_filename, "save_dtype": save_dtype,
|
||||
"device": process_device, "dtype": torch.float32,
|
||||
"exclude_patterns": exclude_patterns, "discard_patterns": discard_patterns,
|
||||
}
|
||||
model_names = {"model_a": model_a, "model_b": model_b, "model_c": model_c}
|
||||
filename = MergerLogic.execute_merge(model_names, calc_mode, THREE_MODEL_MODES, recipe_params, cls.MODEL_TYPE)
|
||||
return io.NodeOutput(filename, doc)
|
||||
+137
-76
@@ -1,98 +1,159 @@
|
||||
import os
|
||||
import folder_paths
|
||||
from comfy_api.latest import io
|
||||
from .utils import convert_pt_to_safetensors, load_metadata_from_safetensors
|
||||
|
||||
|
||||
def _get_metakeys(model_name: str, model_type: str) -> tuple[str, str]:
|
||||
"""Shared logic for all MetaKeys nodes."""
|
||||
model_path = folder_paths.get_full_path_or_raise(model_type, model_name)
|
||||
model_weights = {}
|
||||
metadata = {}
|
||||
layer_shapes = ""
|
||||
|
||||
class BaseMetaKeys:
|
||||
RETURN_TYPES = ("STRING", "STRING")
|
||||
RETURN_NAMES = ("metadata", "keys")
|
||||
FUNCTION = "get_metakeys"
|
||||
CATEGORY = "ModelUtils/Keys"
|
||||
DESCRIPTION = "Loads a model and returns its metadata and layer names."
|
||||
if model_path.endswith(('.pt', '.pth', '.bin', '.ckpt')):
|
||||
temp_safe_path = model_path + ".safetensors"
|
||||
if not os.path.exists(temp_safe_path):
|
||||
conversion_successful, error_message = convert_pt_to_safetensors(model_path, temp_safe_path)
|
||||
if not conversion_successful:
|
||||
return f"Conversion failed: {error_message}", ""
|
||||
model_path = temp_safe_path
|
||||
|
||||
def _get_metakeys(self, model_name, model_type):
|
||||
model_path = folder_paths.get_full_path_or_raise(model_type, model_name)
|
||||
model_weights = {}
|
||||
metadata = {}
|
||||
layer_shapes = ""
|
||||
|
||||
if model_path.endswith(('.pt', '.pth', '.bin', '.ckpt')):
|
||||
temp_safe_path = model_path + ".safetensors"
|
||||
if not os.path.exists(temp_safe_path):
|
||||
conversion_successful, error_message = convert_pt_to_safetensors(model_path, temp_safe_path)
|
||||
if not conversion_successful:
|
||||
return f"Conversion failed: {error_message}", ""
|
||||
model_path = temp_safe_path
|
||||
|
||||
model_weights, metadata = load_metadata_from_safetensors(model_path)
|
||||
metadata_str = str(metadata)
|
||||
for layer_name, tensor in model_weights.items():
|
||||
shape = str(tensor.shape)
|
||||
shape = shape.replace("torch.Size(", "").replace(")", "")
|
||||
name_shape = f"{layer_name}, {shape}\n"
|
||||
layer_shapes += name_shape
|
||||
return metadata_str, layer_shapes
|
||||
model_weights, metadata = load_metadata_from_safetensors(model_path)
|
||||
metadata_str = str(metadata)
|
||||
for layer_name, tensor in model_weights.items():
|
||||
shape = str(tensor.shape)
|
||||
shape = shape.replace("torch.Size(", "").replace(")", "")
|
||||
name_shape = f"{layer_name}, {shape}\n"
|
||||
layer_shapes += name_shape
|
||||
return metadata_str, layer_shapes
|
||||
|
||||
|
||||
|
||||
class ModelMetaKeys(BaseMetaKeys):
|
||||
class ModelMetaKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"diffusionmodel_name": (folder_paths.get_filename_list("diffusion_models"), ),
|
||||
}
|
||||
}
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="ModelMetaKeys",
|
||||
display_name="Get Diffusion Model Metadata & Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a diffusion model and returns its metadata and layer names.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"diffusionmodel_name",
|
||||
options=folder_paths.get_filename_list("diffusion_models"),
|
||||
),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="metadata"),
|
||||
io.String.Output(display_name="keys"),
|
||||
],
|
||||
)
|
||||
|
||||
def get_metakeys(self, diffusionmodel_name):
|
||||
return self._get_metakeys(diffusionmodel_name, "diffusion_models")
|
||||
|
||||
class TextEncoderMetaKeys(BaseMetaKeys):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"textencoder_name": (folder_paths.get_filename_list("text_encoders"), ),
|
||||
}
|
||||
}
|
||||
def execute(cls, diffusionmodel_name: str) -> io.NodeOutput:
|
||||
metadata, keys = _get_metakeys(diffusionmodel_name, "diffusion_models")
|
||||
return io.NodeOutput(metadata, keys)
|
||||
|
||||
def get_metakeys(self, textencoder_name):
|
||||
return self._get_metakeys(textencoder_name, "text_encoders")
|
||||
|
||||
class LoRAMetaKeys(BaseMetaKeys):
|
||||
class TextEncoderMetaKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"lora_name": (folder_paths.get_filename_list("loras"), ),
|
||||
}
|
||||
}
|
||||
def get_metakeys(self, lora_name):
|
||||
return self._get_metakeys(lora_name, "loras")
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="TextEncoderMetaKeys",
|
||||
display_name="Get Text Encoder Metadata & Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a text encoder and returns its metadata and layer names.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"textencoder_name",
|
||||
options=folder_paths.get_filename_list("text_encoders"),
|
||||
),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="metadata"),
|
||||
io.String.Output(display_name="keys"),
|
||||
],
|
||||
)
|
||||
|
||||
class CheckpointMetaKeys(BaseMetaKeys):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"ckpt_name": (folder_paths.get_filename_list("checkpoints"), ),
|
||||
}
|
||||
}
|
||||
def execute(cls, textencoder_name: str) -> io.NodeOutput:
|
||||
metadata, keys = _get_metakeys(textencoder_name, "text_encoders")
|
||||
return io.NodeOutput(metadata, keys)
|
||||
|
||||
def get_metakeys(self, ckpt_name):
|
||||
return self._get_metakeys(ckpt_name, "checkpoints")
|
||||
|
||||
class EmbeddingMetaKeys(BaseMetaKeys):
|
||||
class LoRAMetaKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"embedding": (folder_paths.get_filename_list("embeddings"), ),
|
||||
}
|
||||
}
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="LoRAMetaKeys",
|
||||
display_name="Get LoRA Metadata & Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a LoRA and returns its metadata and layer names.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"lora_name",
|
||||
options=folder_paths.get_filename_list("loras"),
|
||||
),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="metadata"),
|
||||
io.String.Output(display_name="keys"),
|
||||
],
|
||||
)
|
||||
|
||||
def get_metakeys(self, embedding):
|
||||
return self._get_metakeys(embedding, "embeddings")
|
||||
@classmethod
|
||||
def execute(cls, lora_name: str) -> io.NodeOutput:
|
||||
metadata, keys = _get_metakeys(lora_name, "loras")
|
||||
return io.NodeOutput(metadata, keys)
|
||||
|
||||
__all__ = ("ModelMetaKeys", "TextEncoderMetaKeys", "LoRAMetaKeys", "CheckpointMetaKeys", "EmbeddingMetaKeys")
|
||||
|
||||
class CheckpointMetaKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="CheckpointMetaKeys",
|
||||
display_name="Get Checkpoint Metadata & Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a checkpoint and returns its metadata and layer names.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"ckpt_name",
|
||||
options=folder_paths.get_filename_list("checkpoints"),
|
||||
),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="metadata"),
|
||||
io.String.Output(display_name="keys"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, ckpt_name: str) -> io.NodeOutput:
|
||||
metadata, keys = _get_metakeys(ckpt_name, "checkpoints")
|
||||
return io.NodeOutput(metadata, keys)
|
||||
|
||||
|
||||
class EmbeddingMetaKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="EmbeddingMetaKeys",
|
||||
display_name="Get Embedding Metadata & Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads an embedding and returns its metadata and layer names.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"embedding",
|
||||
options=folder_paths.get_filename_list("embeddings"),
|
||||
),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="metadata"),
|
||||
io.String.Output(display_name="keys"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, embedding: str) -> io.NodeOutput:
|
||||
metadata, keys = _get_metakeys(embedding, "embeddings")
|
||||
return io.NodeOutput(metadata, keys)
|
||||
|
||||
+173
-101
@@ -1,127 +1,199 @@
|
||||
import os
|
||||
import re
|
||||
import folder_paths
|
||||
from .utils import convert_pt_to_safetensors, load_metadata_from_safetensors
|
||||
from comfy_api.latest import io
|
||||
from safetensors.torch import save_file
|
||||
from .utils import convert_pt_to_safetensors, load_metadata_from_safetensors
|
||||
|
||||
class BasePruneKeys:
|
||||
RETURN_TYPES = ("STRING",)
|
||||
RETURN_NAMES = ("output_path",)
|
||||
FUNCTION = "prune_keys"
|
||||
CATEGORY = "ModelUtils/Keys"
|
||||
DESCRIPTION = "Loads a model, removes specified keys, and saves it as a new safetensors file."
|
||||
|
||||
def _prune_keys(self, model_name, model_type, keys_to_prune_str, use_regex, output_filename):
|
||||
model_path = folder_paths.get_full_path_or_raise(model_type, model_name)
|
||||
def _prune_keys(model_name: str, model_type: str, keys_to_prune_str: str,
|
||||
use_regex: bool, output_filename: str) -> str:
|
||||
"""Shared logic for all PruneKeys nodes."""
|
||||
model_path = folder_paths.get_full_path_or_raise(model_type, model_name)
|
||||
|
||||
if model_path.endswith(('.pt', '.pth', '.bin', '.ckpt')):
|
||||
temp_safe_path = model_path + ".safetensors"
|
||||
if not os.path.exists(temp_safe_path):
|
||||
conversion_successful, error_message = convert_pt_to_safetensors(model_path, temp_safe_path)
|
||||
if not conversion_successful:
|
||||
raise Exception(f"Conversion failed: {error_message}")
|
||||
model_path_to_load = temp_safe_path
|
||||
if model_path.endswith(('.pt', '.pth', '.bin', '.ckpt')):
|
||||
temp_safe_path = model_path + ".safetensors"
|
||||
if not os.path.exists(temp_safe_path):
|
||||
conversion_successful, error_message = convert_pt_to_safetensors(model_path, temp_safe_path)
|
||||
if not conversion_successful:
|
||||
raise Exception(f"Conversion failed: {error_message}")
|
||||
model_path_to_load = temp_safe_path
|
||||
else:
|
||||
model_path_to_load = model_path
|
||||
|
||||
model_weights, metadata = load_metadata_from_safetensors(model_path_to_load)
|
||||
|
||||
patterns = [p.strip() for p in keys_to_prune_str.strip().split('\n') if p.strip()]
|
||||
|
||||
if not patterns:
|
||||
raise ValueError("No keys/patterns provided to prune.")
|
||||
|
||||
pruned_tensors = {}
|
||||
for key, tensor in model_weights.items():
|
||||
is_match = False
|
||||
if use_regex:
|
||||
if any(re.search(pattern, key) for pattern in patterns):
|
||||
is_match = True
|
||||
else:
|
||||
model_path_to_load = model_path
|
||||
if any(pattern in key for pattern in patterns):
|
||||
is_match = True
|
||||
|
||||
model_weights, metadata = load_metadata_from_safetensors(model_path_to_load)
|
||||
if not is_match:
|
||||
pruned_tensors[key] = tensor
|
||||
|
||||
patterns = [p.strip() for p in keys_to_prune_str.strip().split('\n') if p.strip()]
|
||||
model_dir = folder_paths.get_folder_paths(model_type)[0]
|
||||
output_path = os.path.join(model_dir, f"{output_filename.strip()}.safetensors")
|
||||
|
||||
if not patterns:
|
||||
raise ValueError("No keys/patterns provided to prune.")
|
||||
save_file(pruned_tensors, output_path, metadata)
|
||||
|
||||
pruned_tensors = {}
|
||||
for key, tensor in model_weights.items():
|
||||
is_match = False
|
||||
if use_regex:
|
||||
if any(re.search(pattern, key) for pattern in patterns):
|
||||
is_match = True
|
||||
else:
|
||||
if any(pattern in key for pattern in patterns):
|
||||
is_match = True
|
||||
return output_path
|
||||
|
||||
if not is_match:
|
||||
pruned_tensors[key] = tensor
|
||||
|
||||
model_dir = folder_paths.get_folder_paths(model_type)[0]
|
||||
output_path = os.path.join(model_dir, f"{output_filename.strip()}.safetensors")
|
||||
|
||||
save_file(pruned_tensors, output_path, metadata)
|
||||
|
||||
return (output_path,)
|
||||
|
||||
class ModelPruneKeys(BasePruneKeys):
|
||||
class ModelPruneKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"diffusionmodel_name": (folder_paths.get_filename_list("diffusion_models"), ),
|
||||
"keys_to_prune": ("STRING", {"multiline": True, "default": ""}),
|
||||
"use_regex": ("BOOLEAN", {"default": False}),
|
||||
"output_filename": ("STRING", {"default": "pruned_model"}),
|
||||
}
|
||||
}
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="ModelPruneKeys",
|
||||
display_name="Prune Diffusion Model Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a diffusion model, removes specified keys, and saves it as a new safetensors file.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"diffusionmodel_name",
|
||||
options=folder_paths.get_filename_list("diffusion_models"),
|
||||
),
|
||||
io.String.Input("keys_to_prune", multiline=True, default=""),
|
||||
io.Boolean.Input("use_regex", default=False),
|
||||
io.String.Input("output_filename", default="pruned_model"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
)
|
||||
|
||||
def prune_keys(self, diffusionmodel_name, keys_to_prune, use_regex, output_filename):
|
||||
return self._prune_keys(diffusionmodel_name, "diffusion_models", keys_to_prune, use_regex, output_filename)
|
||||
|
||||
class TextEncoderPruneKeys(BasePruneKeys):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"textencoder_name": (folder_paths.get_filename_list("text_encoders"), ),
|
||||
"keys_to_prune": ("STRING", {"multiline": True, "default": ""}),
|
||||
"use_regex": ("BOOLEAN", {"default": False}),
|
||||
"output_filename": ("STRING", {"default": "pruned_textencoder"}),
|
||||
}
|
||||
}
|
||||
def execute(cls, diffusionmodel_name: str, keys_to_prune: str,
|
||||
use_regex: bool, output_filename: str) -> io.NodeOutput:
|
||||
path = _prune_keys(diffusionmodel_name, "diffusion_models",
|
||||
keys_to_prune, use_regex, output_filename)
|
||||
return io.NodeOutput(path)
|
||||
|
||||
def prune_keys(self, textencoder_name, keys_to_prune, use_regex, output_filename):
|
||||
return self._prune_keys(textencoder_name, "text_encoders", keys_to_prune, use_regex, output_filename)
|
||||
|
||||
class LoRAPruneKeys(BasePruneKeys):
|
||||
class TextEncoderPruneKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"lora_name": (folder_paths.get_filename_list("loras"), ),
|
||||
"keys_to_prune": ("STRING", {"multiline": True, "default": ""}),
|
||||
"use_regex": ("BOOLEAN", {"default": False}),
|
||||
"output_filename": ("STRING", {"default": "pruned_lora"}),
|
||||
}
|
||||
}
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="TextEncoderPruneKeys",
|
||||
display_name="Prune Text Encoder Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a text encoder, removes specified keys, and saves it as a new safetensors file.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"textencoder_name",
|
||||
options=folder_paths.get_filename_list("text_encoders"),
|
||||
),
|
||||
io.String.Input("keys_to_prune", multiline=True, default=""),
|
||||
io.Boolean.Input("use_regex", default=False),
|
||||
io.String.Input("output_filename", default="pruned_textencoder"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
)
|
||||
|
||||
def prune_keys(self, lora_name, keys_to_prune, use_regex, output_filename):
|
||||
return self._prune_keys(lora_name, "loras", keys_to_prune, use_regex, output_filename)
|
||||
|
||||
class CheckpointPruneKeys(BasePruneKeys):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"ckpt_name": (folder_paths.get_filename_list("checkpoints"), ),
|
||||
"keys_to_prune": ("STRING", {"multiline": True, "default": ""}),
|
||||
"use_regex": ("BOOLEAN", {"default": False}),
|
||||
"output_filename": ("STRING", {"default": "pruned_checkpoint"}),
|
||||
}
|
||||
}
|
||||
def execute(cls, textencoder_name: str, keys_to_prune: str,
|
||||
use_regex: bool, output_filename: str) -> io.NodeOutput:
|
||||
path = _prune_keys(textencoder_name, "text_encoders",
|
||||
keys_to_prune, use_regex, output_filename)
|
||||
return io.NodeOutput(path)
|
||||
|
||||
def prune_keys(self, ckpt_name, keys_to_prune, use_regex, output_filename):
|
||||
return self._prune_keys(ckpt_name, "checkpoints", keys_to_prune, use_regex, output_filename)
|
||||
|
||||
class EmbeddingPruneKeys(BasePruneKeys):
|
||||
class LoRAPruneKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"embedding": (folder_paths.get_filename_list("embeddings"), ),
|
||||
"keys_to_prune": ("STRING", {"multiline": True, "default": ""}),
|
||||
"use_regex": ("BOOLEAN", {"default": False}),
|
||||
"output_filename": ("STRING", {"default": "pruned_embedding"}),
|
||||
}
|
||||
}
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="LoRAPruneKeys",
|
||||
display_name="Prune LoRA Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a LoRA, removes specified keys, and saves it as a new safetensors file.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"lora_name",
|
||||
options=folder_paths.get_filename_list("loras"),
|
||||
),
|
||||
io.String.Input("keys_to_prune", multiline=True, default=""),
|
||||
io.Boolean.Input("use_regex", default=False),
|
||||
io.String.Input("output_filename", default="pruned_lora"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
)
|
||||
|
||||
def prune_keys(self, embedding, keys_to_prune, use_regex, output_filename):
|
||||
return self._prune_keys(embedding, "embeddings", keys_to_prune, use_regex, output_filename)
|
||||
@classmethod
|
||||
def execute(cls, lora_name: str, keys_to_prune: str,
|
||||
use_regex: bool, output_filename: str) -> io.NodeOutput:
|
||||
path = _prune_keys(lora_name, "loras",
|
||||
keys_to_prune, use_regex, output_filename)
|
||||
return io.NodeOutput(path)
|
||||
|
||||
|
||||
class CheckpointPruneKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="CheckpointPruneKeys",
|
||||
display_name="Prune Checkpoint Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a checkpoint, removes specified keys, and saves it as a new safetensors file.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"ckpt_name",
|
||||
options=folder_paths.get_filename_list("checkpoints"),
|
||||
),
|
||||
io.String.Input("keys_to_prune", multiline=True, default=""),
|
||||
io.Boolean.Input("use_regex", default=False),
|
||||
io.String.Input("output_filename", default="pruned_checkpoint"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, ckpt_name: str, keys_to_prune: str,
|
||||
use_regex: bool, output_filename: str) -> io.NodeOutput:
|
||||
path = _prune_keys(ckpt_name, "checkpoints",
|
||||
keys_to_prune, use_regex, output_filename)
|
||||
return io.NodeOutput(path)
|
||||
|
||||
|
||||
class EmbeddingPruneKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="EmbeddingPruneKeys",
|
||||
display_name="Prune Embedding Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads an embedding, removes specified keys, and saves it as a new safetensors file.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"embedding",
|
||||
options=folder_paths.get_filename_list("embeddings"),
|
||||
),
|
||||
io.String.Input("keys_to_prune", multiline=True, default=""),
|
||||
io.Boolean.Input("use_regex", default=False),
|
||||
io.String.Input("output_filename", default="pruned_embedding"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, embedding: str, keys_to_prune: str,
|
||||
use_regex: bool, output_filename: str) -> io.NodeOutput:
|
||||
path = _prune_keys(embedding, "embeddings",
|
||||
keys_to_prune, use_regex, output_filename)
|
||||
return io.NodeOutput(path)
|
||||
+167
-95
@@ -1,122 +1,194 @@
|
||||
import os
|
||||
import folder_paths
|
||||
from .utils import convert_pt_to_safetensors, load_metadata_from_safetensors
|
||||
from comfy_api.latest import io
|
||||
from safetensors.torch import save_file
|
||||
from .utils import convert_pt_to_safetensors, load_metadata_from_safetensors
|
||||
|
||||
class BaseRenameKeys:
|
||||
RETURN_TYPES = ("STRING",)
|
||||
RETURN_NAMES = ("output_path",)
|
||||
FUNCTION = "rename_keys"
|
||||
CATEGORY = "ModelUtils/Keys"
|
||||
DESCRIPTION = "Loads a model, renames specified keys, and saves it as a new safetensors file."
|
||||
|
||||
def _rename_keys(self, model_name, model_type, old_keys_str, new_keys_str, output_filename):
|
||||
model_path = folder_paths.get_full_path_or_raise(model_type, model_name)
|
||||
def _rename_keys(model_name: str, model_type: str, old_keys_str: str,
|
||||
new_keys_str: str, output_filename: str) -> str:
|
||||
"""Shared logic for all RenameKeys nodes."""
|
||||
model_path = folder_paths.get_full_path_or_raise(model_type, model_name)
|
||||
|
||||
if model_path.endswith(('.pt', '.pth', '.bin', '.ckpt')):
|
||||
temp_safe_path = model_path + ".safetensors"
|
||||
if not os.path.exists(temp_safe_path):
|
||||
conversion_successful, error_message = convert_pt_to_safetensors(model_path, temp_safe_path)
|
||||
if not conversion_successful:
|
||||
raise Exception(f"Conversion failed: {error_message}")
|
||||
model_path_to_load = temp_safe_path
|
||||
else:
|
||||
model_path_to_load = model_path
|
||||
if model_path.endswith(('.pt', '.pth', '.bin', '.ckpt')):
|
||||
temp_safe_path = model_path + ".safetensors"
|
||||
if not os.path.exists(temp_safe_path):
|
||||
conversion_successful, error_message = convert_pt_to_safetensors(model_path, temp_safe_path)
|
||||
if not conversion_successful:
|
||||
raise Exception(f"Conversion failed: {error_message}")
|
||||
model_path_to_load = temp_safe_path
|
||||
else:
|
||||
model_path_to_load = model_path
|
||||
|
||||
model_weights, metadata = load_metadata_from_safetensors(model_path_to_load)
|
||||
model_weights, metadata = load_metadata_from_safetensors(model_path_to_load)
|
||||
|
||||
old_keys = [key.strip() for key in old_keys_str.strip().split('\n') if key.strip()]
|
||||
new_keys = [key.strip() for key in new_keys_str.strip().split('\n') if key.strip()]
|
||||
old_keys = [key.strip() for key in old_keys_str.strip().split('\n') if key.strip()]
|
||||
new_keys = [key.strip() for key in new_keys_str.strip().split('\n') if key.strip()]
|
||||
|
||||
if len(old_keys) != len(new_keys):
|
||||
raise ValueError("The number of old keys must match the number of new keys.")
|
||||
if len(old_keys) != len(new_keys):
|
||||
raise ValueError("The number of old keys must match the number of new keys.")
|
||||
|
||||
key_map = dict(zip(old_keys, new_keys))
|
||||
key_map = dict(zip(old_keys, new_keys))
|
||||
|
||||
original_keys = list(model_weights.keys())
|
||||
renamed_tensors = {}
|
||||
for key in original_keys:
|
||||
new_key = key_map.get(key, key)
|
||||
renamed_tensors[new_key] = model_weights[key]
|
||||
original_keys = list(model_weights.keys())
|
||||
renamed_tensors = {}
|
||||
for key in original_keys:
|
||||
new_key = key_map.get(key, key)
|
||||
renamed_tensors[new_key] = model_weights[key]
|
||||
|
||||
model_dir = folder_paths.get_folder_paths(model_type)[0]
|
||||
output_path = os.path.join(model_dir, f"{output_filename.strip()}.safetensors")
|
||||
model_dir = folder_paths.get_folder_paths(model_type)[0]
|
||||
output_path = os.path.join(model_dir, f"{output_filename.strip()}.safetensors")
|
||||
|
||||
save_file(renamed_tensors, output_path, metadata)
|
||||
save_file(renamed_tensors, output_path, metadata)
|
||||
|
||||
return (output_path,)
|
||||
return output_path
|
||||
|
||||
class ModelRenameKeys(BaseRenameKeys):
|
||||
|
||||
class ModelRenameKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"diffusionmodel_name": (folder_paths.get_filename_list("diffusion_models"), ),
|
||||
"old_keys": ("STRING", {"multiline": True, "default": ""}),
|
||||
"new_keys": ("STRING", {"multiline": True, "default": ""}),
|
||||
"output_filename": ("STRING", {"default": "renamed_model"}),
|
||||
}
|
||||
}
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="ModelRenameKeys",
|
||||
display_name="Rename Diffusion Model Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a diffusion model, renames specified keys, and saves it as a new safetensors file.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"diffusionmodel_name",
|
||||
options=folder_paths.get_filename_list("diffusion_models"),
|
||||
),
|
||||
io.String.Input("old_keys", multiline=True, default=""),
|
||||
io.String.Input("new_keys", multiline=True, default=""),
|
||||
io.String.Input("output_filename", default="renamed_model"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
)
|
||||
|
||||
def rename_keys(self, diffusionmodel_name, old_keys, new_keys, output_filename):
|
||||
return self._rename_keys(diffusionmodel_name, "diffusion_models", old_keys, new_keys, output_filename)
|
||||
|
||||
class TextEncoderRenameKeys(BaseRenameKeys):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"textencoder_name": (folder_paths.get_filename_list("text_encoders"), ),
|
||||
"old_keys": ("STRING", {"multiline": True, "default": ""}),
|
||||
"new_keys": ("STRING", {"multiline": True, "default": ""}),
|
||||
"output_filename": ("STRING", {"default": "renamed_textencoder"}),
|
||||
}
|
||||
}
|
||||
def execute(cls, diffusionmodel_name: str, old_keys: str,
|
||||
new_keys: str, output_filename: str) -> io.NodeOutput:
|
||||
path = _rename_keys(diffusionmodel_name, "diffusion_models",
|
||||
old_keys, new_keys, output_filename)
|
||||
return io.NodeOutput(path)
|
||||
|
||||
def rename_keys(self, textencoder_name, old_keys, new_keys, output_filename):
|
||||
return self._rename_keys(textencoder_name, "text_encoders", old_keys, new_keys, output_filename)
|
||||
|
||||
class LoRARenameKeys(BaseRenameKeys):
|
||||
class TextEncoderRenameKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"lora_name": (folder_paths.get_filename_list("loras"), ),
|
||||
"old_keys": ("STRING", {"multiline": True, "default": ""}),
|
||||
"new_keys": ("STRING", {"multiline": True, "default": ""}),
|
||||
"output_filename": ("STRING", {"default": "renamed_lora"}),
|
||||
}
|
||||
}
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="TextEncoderRenameKeys",
|
||||
display_name="Rename Text Encoder Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a text encoder, renames specified keys, and saves it as a new safetensors file.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"textencoder_name",
|
||||
options=folder_paths.get_filename_list("text_encoders"),
|
||||
),
|
||||
io.String.Input("old_keys", multiline=True, default=""),
|
||||
io.String.Input("new_keys", multiline=True, default=""),
|
||||
io.String.Input("output_filename", default="renamed_textencoder"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
)
|
||||
|
||||
def rename_keys(self, lora_name, old_keys, new_keys, output_filename):
|
||||
return self._rename_keys(lora_name, "loras", old_keys, new_keys, output_filename)
|
||||
|
||||
class CheckpointRenameKeys(BaseRenameKeys):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"ckpt_name": (folder_paths.get_filename_list("checkpoints"), ),
|
||||
"old_keys": ("STRING", {"multiline": True, "default": ""}),
|
||||
"new_keys": ("STRING", {"multiline": True, "default": ""}),
|
||||
"output_filename": ("STRING", {"default": "renamed_checkpoint"}),
|
||||
}
|
||||
}
|
||||
def execute(cls, textencoder_name: str, old_keys: str,
|
||||
new_keys: str, output_filename: str) -> io.NodeOutput:
|
||||
path = _rename_keys(textencoder_name, "text_encoders",
|
||||
old_keys, new_keys, output_filename)
|
||||
return io.NodeOutput(path)
|
||||
|
||||
def rename_keys(self, ckpt_name, old_keys, new_keys, output_filename):
|
||||
return self._rename_keys(ckpt_name, "checkpoints", old_keys, new_keys, output_filename)
|
||||
|
||||
class EmbeddingRenameKeys(BaseRenameKeys):
|
||||
class LoRARenameKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def INPUT_TYPES(s):
|
||||
return {
|
||||
"required": {
|
||||
"embedding": (folder_paths.get_filename_list("embeddings"), ),
|
||||
"old_keys": ("STRING", {"multiline": True, "default": ""}),
|
||||
"new_keys": ("STRING", {"multiline": True, "default": ""}),
|
||||
"output_filename": ("STRING", {"default": "renamed_embedding"}),
|
||||
}
|
||||
}
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="LoRARenameKeys",
|
||||
display_name="Rename LoRA Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a LoRA, renames specified keys, and saves it as a new safetensors file.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"lora_name",
|
||||
options=folder_paths.get_filename_list("loras"),
|
||||
),
|
||||
io.String.Input("old_keys", multiline=True, default=""),
|
||||
io.String.Input("new_keys", multiline=True, default=""),
|
||||
io.String.Input("output_filename", default="renamed_lora"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
)
|
||||
|
||||
def rename_keys(self, embedding, old_keys, new_keys, output_filename):
|
||||
return self._rename_keys(embedding, "embeddings", old_keys, new_keys, output_filename)
|
||||
@classmethod
|
||||
def execute(cls, lora_name: str, old_keys: str,
|
||||
new_keys: str, output_filename: str) -> io.NodeOutput:
|
||||
path = _rename_keys(lora_name, "loras",
|
||||
old_keys, new_keys, output_filename)
|
||||
return io.NodeOutput(path)
|
||||
|
||||
|
||||
class CheckpointRenameKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="CheckpointRenameKeys",
|
||||
display_name="Rename Checkpoint Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads a checkpoint, renames specified keys, and saves it as a new safetensors file.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"ckpt_name",
|
||||
options=folder_paths.get_filename_list("checkpoints"),
|
||||
),
|
||||
io.String.Input("old_keys", multiline=True, default=""),
|
||||
io.String.Input("new_keys", multiline=True, default=""),
|
||||
io.String.Input("output_filename", default="renamed_checkpoint"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, ckpt_name: str, old_keys: str,
|
||||
new_keys: str, output_filename: str) -> io.NodeOutput:
|
||||
path = _rename_keys(ckpt_name, "checkpoints",
|
||||
old_keys, new_keys, output_filename)
|
||||
return io.NodeOutput(path)
|
||||
|
||||
|
||||
class EmbeddingRenameKeys(io.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls):
|
||||
return io.Schema(
|
||||
node_id="EmbeddingRenameKeys",
|
||||
display_name="Rename Embedding Keys",
|
||||
category="ModelUtils/Keys",
|
||||
description="Loads an embedding, renames specified keys, and saves it as a new safetensors file.",
|
||||
inputs=[
|
||||
io.Combo.Input(
|
||||
"embedding",
|
||||
options=folder_paths.get_filename_list("embeddings"),
|
||||
),
|
||||
io.String.Input("old_keys", multiline=True, default=""),
|
||||
io.String.Input("new_keys", multiline=True, default=""),
|
||||
io.String.Input("output_filename", default="renamed_embedding"),
|
||||
],
|
||||
outputs=[
|
||||
io.String.Output(display_name="output_path"),
|
||||
],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, embedding: str, old_keys: str,
|
||||
new_keys: str, output_filename: str) -> io.NodeOutput:
|
||||
path = _rename_keys(embedding, "embeddings",
|
||||
old_keys, new_keys, output_filename)
|
||||
return io.NodeOutput(path)
|
||||
Reference in New Issue
Block a user