V3 Nodes: refactor ComfyNodeV3 class; use of ui.SavedResult; ported SaveAnimatedPNG and SaveAnimatedWEBP nodes

This commit is contained in:
bigcat88 2025-07-14 16:24:50 +03:00
parent 371e20494d
commit a580176735
No known key found for this signature in database
GPG Key ID: 1F0BF0EC3CF22721
8 changed files with 219 additions and 114 deletions

16
comfy_api/v3/helpers.py Normal file
View File

@ -0,0 +1,16 @@
from typing import Optional, Callable
def first_real_override(cls: type, name: str, *, base: type) -> Optional[Callable]:
"""Return the *callable* override of `name` visible on `cls`, or None if every
implementation up to (and including) `base` is the placeholder defined on `base`.
"""
base_func = getattr(base, name).__func__
for c in cls.mro(): # NodeB, NodeA, ComfyNodeV3, object …
if c is base: # reached the placeholder we're done
break
if name in c.__dict__: # first class that *defines* the attr
func = getattr(c, name).__func__
if func is not base_func: # real override
return getattr(cls, name) # bound to *cls*
return None

View File

@ -1,5 +1,5 @@
from __future__ import annotations
from typing import Any, Literal, TYPE_CHECKING, TypeVar, Callable, Optional, cast, TypedDict
from typing import Any, Literal, TypeVar, Callable, TypedDict
from typing_extensions import NotRequired
from enum import Enum
from abc import ABC, abstractmethod
@ -1097,28 +1097,24 @@ class ComfyNodeV3:
@classmethod
@abstractmethod
def DEFINE_SCHEMA(cls) -> SchemaV3:
"""
Override this function with one that returns a SchemaV3 instance.
"""
return None
DEFINE_SCHEMA = None
"""Override this function with one that returns a SchemaV3 instance."""
raise NotImplementedError
@classmethod
@abstractmethod
def execute(cls, **kwargs) -> NodeOutput:
pass
execute = None
raise NotImplementedError
@classmethod
def validate_inputs(cls, **kwargs) -> bool:
"""Optionally, define this function to validate inputs; equivalnet to V1's VALIDATE_INPUTS."""
pass
validate_inputs = None
"""Optionally, define this function to validate inputs; equivalent to V1's VALIDATE_INPUTS."""
raise NotImplementedError
@classmethod
def fingerprint_inputs(cls, **kwargs) -> Any:
"""Optionally, define this function to fingerprint inputs; equivalent to V1's IS_CHANGED."""
pass
raise NotImplementedError
fingerprint_inputs = None
@classmethod
@ -1135,8 +1131,8 @@ class ComfyNodeV3:
Comfy Docs: https://docs.comfy.org/custom-nodes/backend/lazy_evaluation#defining-check-lazy-status
"""
need = [name for name in kwargs if kwargs[name] is None]
return need
return [name for name in kwargs if kwargs[name] is None]
check_lazy_status = None
@classmethod

View File

@ -1,7 +1,6 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from comfy_api.v3.io import Image, Mask, FolderType, _UIOutput, ComfyNodeV3
from comfy_api.v3.io import Image, FolderType, _UIOutput, ComfyNodeV3
# used for image preview
from comfy.cli_args import args
import folder_paths
@ -13,33 +12,33 @@ import json
import numpy as np
class SavedResult:
class SavedResult(dict):
def __init__(self, filename: str, subfolder: str, type: FolderType):
self.filename = filename
self.subfolder = subfolder
self.type = type
super().__init__(filename=filename, subfolder=subfolder,type=type.value)
@property
def filename(self) -> str:
return self["filename"]
@property
def subfolder(self) -> str:
return self["subfolder"]
@property
def type(self) -> FolderType:
return FolderType(self["type"])
def as_dict(self):
return {
"filename": self.filename,
"subfolder": self.subfolder,
"type": self.type
}
class PreviewImage(_UIOutput):
def __init__(self, image: Image.Type, animated: bool=False, cls: ComfyNodeV3=None, **kwargs):
output_dir = folder_paths.get_temp_directory()
type = "temp"
prefix_append = "_temp_" + ''.join(random.choice("abcdefghijklmnopqrstupvxyz") for x in range(5))
compress_level = 1
filename_prefix = "ComfyUI"
filename_prefix = "ComfyUI" + prefix_append
filename_prefix += prefix_append
full_output_folder, filename, counter, subfolder, filename_prefix = folder_paths.get_save_image_path(filename_prefix, output_dir, image[0].shape[1], image[0].shape[0])
results = list()
for (batch_number, image) in enumerate(image):
i = 255. * image.cpu().numpy()
img = PILImage.fromarray(np.clip(i, 0, 255).astype(np.uint8))
img = PILImage.fromarray(np.clip(255. * image.cpu().numpy(), 0, 255).astype(np.uint8))
metadata = None
if not args.disable_metadata and cls is not None:
metadata = PngInfo()
@ -51,17 +50,16 @@ class PreviewImage(_UIOutput):
filename_with_batch_num = filename.replace("%batch_num%", str(batch_number))
file = f"{filename_with_batch_num}_{counter:05}_.png"
img.save(os.path.join(full_output_folder, file), pnginfo=metadata, compress_level=compress_level)
results.append(SavedResult(file, subfolder, type))
img.save(os.path.join(full_output_folder, file), pnginfo=metadata, compress_level=1)
results.append(SavedResult(file, subfolder, FolderType.temp))
counter += 1
self.values = results
self.animated = animated
def as_dict(self):
values = [x.as_dict() if isinstance(x, SavedResult) else x for x in self.values]
return {
"images": values,
"images": self.values,
"animated": (self.animated,)
}
@ -113,9 +111,8 @@ class PreviewMask(PreviewImage):
# self.values = values
# def as_dict(self):
# values = [x.as_dict() if isinstance(x, SavedResult) else x for x in self.values]
# return {
# "latents": values,
# "latents": self.values,
# }
class PreviewAudio(_UIOutput):
@ -123,20 +120,14 @@ class PreviewAudio(_UIOutput):
self.values = values
def as_dict(self):
values = [x.as_dict() if isinstance(x, SavedResult) else x for x in self.values]
return {
"audio": values,
}
return {"audio": self.values}
class PreviewUI3D(_UIOutput):
def __init__(self, values: list[SavedResult | dict], **kwargs):
self.values = values
def as_dict(self):
values = [x.as_dict() if isinstance(x, SavedResult) else x for x in self.values]
return {
"3d": values,
}
return {"3d": self.values}
class PreviewText(_UIOutput):
def __init__(self, value: str, **kwargs):

View File

@ -1,11 +1,11 @@
import torch
import time
from comfy_api.v3 import io, ui, resources
import logging
import logging # noqa
import folder_paths
import comfy.utils
import comfy.sd
from typing import Any
@io.comfytype(io_type="XYZ")
class XYZ:
@ -88,11 +88,11 @@ class V3TestNode(io.ComfyNodeV3):
expected_int = 123
if "thing" not in cls.state:
cls.state["thing"] = "hahaha"
yyy = cls.state["thing"]
yyy = cls.state["thing"] # noqa
del cls.state["thing"]
if cls.state.get_value("int2") is None:
cls.state.set_value("int2", 123)
zzz = cls.state.get_value("int2")
zzz = cls.state.get_value("int2") # noqa
cls.state.pop("int2")
if cls.state.my_int is None:
cls.state.my_int = expected_int

View File

@ -37,7 +37,7 @@ class SaveImage_V3(io.ComfyNodeV3):
)
@classmethod
def execute(cls, images, filename_prefix="ComfyUI"):
def execute(cls, images, filename_prefix="ComfyUI") -> io.NodeOutput:
full_output_folder, filename, counter, subfolder, filename_prefix = folder_paths.get_save_image_path(
filename_prefix, folder_paths.get_output_directory(), images[0].shape[1], images[0].shape[0]
)
@ -57,16 +57,122 @@ class SaveImage_V3(io.ComfyNodeV3):
filename_with_batch_num = filename.replace("%batch_num%", str(batch_number))
file = f"{filename_with_batch_num}_{counter:05}_.png"
img.save(os.path.join(full_output_folder, file), pnginfo=metadata, compress_level=4)
results.append({
"filename": file,
"subfolder": subfolder,
"type": "output",
})
results.append(ui.SavedResult(file, subfolder, io.FolderType.output))
counter += 1
return io.NodeOutput(ui={"images": results})
class SaveAnimatedPNG_V3(io.ComfyNodeV3):
@classmethod
def DEFINE_SCHEMA(cls):
return io.SchemaV3(
node_id="SaveAnimatedPNG_V3",
display_name="Save Animated PNG _V3",
category="image/animation",
inputs=[
io.Image.Input("images"),
io.String.Input("filename_prefix", default="ComfyUI"),
io.Float.Input("fps", default=6.0, min=0.01, max=1000.0, step=0.01),
io.Int.Input("compress_level", default=4, min=0, max=9),
],
hidden=[io.Hidden.prompt, io.Hidden.extra_pnginfo],
is_output_node=True,
)
@classmethod
def execute(cls, images, fps, compress_level, filename_prefix="ComfyUI") -> io.NodeOutput:
full_output_folder, filename, counter, subfolder, filename_prefix = (
folder_paths.get_save_image_path(filename_prefix, folder_paths.get_output_directory(), images[0].shape[1], images[0].shape[0])
)
results = []
pil_images = []
for image in images:
img = Image.fromarray(np.clip(255. * image.cpu().numpy(), 0, 255).astype(np.uint8))
pil_images.append(img)
metadata = None
if not args.disable_metadata:
metadata = PngInfo()
if cls.hidden.prompt is not None:
metadata.add(
b"comf", "prompt".encode("latin-1", "strict") + b"\0" + json.dumps(cls.hidden.prompt).encode("latin-1", "strict"), after_idat=True
)
if cls.hidden.extra_pnginfo is not None:
for x in cls.hidden.extra_pnginfo:
metadata.add(
b"comf", x.encode("latin-1", "strict") + b"\0" + json.dumps(cls.hidden.extra_pnginfo[x]).encode("latin-1", "strict"), after_idat=True
)
file = f"{filename}_{counter:05}_.png"
pil_images[0].save(os.path.join(full_output_folder, file), pnginfo=metadata, compress_level=compress_level, save_all=True, duration=int(1000.0/fps), append_images=pil_images[1:])
results.append(ui.SavedResult(file, subfolder, io.FolderType.output))
return io.NodeOutput(ui={"images": results, "animated": (True,) })
class SaveAnimatedWEBP_V3(io.ComfyNodeV3):
COMPRESS_METHODS = {"default": 4, "fastest": 0, "slowest": 6}
@classmethod
def DEFINE_SCHEMA(cls):
return io.SchemaV3(
node_id="SaveAnimatedWEBP_V3",
display_name="Save Animated WEBP _V3",
category="image/animation",
inputs=[
io.Image.Input("images"),
io.String.Input("filename_prefix", default="ComfyUI"),
io.Float.Input("fps", default=6.0, min=0.01, max=1000.0, step=0.01),
io.Boolean.Input("lossless", default=True),
io.Int.Input("quality", default=80, min=0, max=100),
io.Combo.Input("method", options=list(cls.COMPRESS_METHODS.keys())),
# "num_frames": ("INT", {"default": 0, "min": 0, "max": 8192}),
],
hidden=[io.Hidden.prompt, io.Hidden.extra_pnginfo],
is_output_node=True,
)
@classmethod
def execute(cls, images, fps, filename_prefix, lossless, quality, method, num_frames=0) -> io.NodeOutput:
method = cls.COMPRESS_METHODS.get(method)
full_output_folder, filename, counter, subfolder, filename_prefix = folder_paths.get_save_image_path(filename_prefix, folder_paths.get_output_directory(), images[0].shape[1], images[0].shape[0])
results = []
pil_images = []
for image in images:
img = Image.fromarray(np.clip(255. * image.cpu().numpy(), 0, 255).astype(np.uint8))
pil_images.append(img)
metadata = pil_images[0].getexif()
if not args.disable_metadata:
if cls.hidden.prompt is not None:
metadata[0x0110] = "prompt:{}".format(json.dumps(cls.hidden.prompt))
if cls.hidden.extra_pnginfo is not None:
inital_exif = 0x010f
for x in cls.hidden.extra_pnginfo:
metadata[inital_exif] = "{}:{}".format(x, json.dumps(cls.hidden.extra_pnginfo[x]))
inital_exif -= 1
if num_frames == 0:
num_frames = len(pil_images)
for i in range(0, len(pil_images), num_frames):
file = f"{filename}_{counter:05}_.webp"
pil_images[i].save(
os.path.join(full_output_folder, file),
save_all=True, duration=int(1000.0/fps),
append_images=pil_images[i + 1:i + num_frames],
exif=metadata,
lossless=lossless,
quality=quality,
method=method,
)
results.append(ui.SavedResult(file, subfolder, io.FolderType.output))
counter += 1
return io.NodeOutput(ui={"images": results, "animated": (num_frames != 1,)})
class PreviewImage_V3(io.ComfyNodeV3):
@classmethod
def DEFINE_SCHEMA(cls):
@ -76,17 +182,14 @@ class PreviewImage_V3(io.ComfyNodeV3):
description="Preview the input images.",
category="image",
inputs=[
io.Image.Input(
"images",
tooltip="The images to preview.",
),
io.Image.Input("images", tooltip="The images to preview."),
],
hidden=[io.Hidden.prompt, io.Hidden.extra_pnginfo],
is_output_node=True,
)
@classmethod
def execute(cls, images):
def execute(cls, images) -> io.NodeOutput:
return io.NodeOutput(ui=ui.PreviewImage(images, cls=cls))
@ -267,8 +370,9 @@ class LoadImageOutput_V3(io.ComfyNodeV3):
return True
NODES_LIST: list[type[io.ComfyNodeV3]] = [
SaveAnimatedPNG_V3,
SaveAnimatedWEBP_V3,
SaveImage_V3,
PreviewImage_V3,
LoadImage_V3,

View File

@ -28,7 +28,7 @@ from comfy_execution.graph import (
)
from comfy_execution.graph_utils import GraphBuilder, is_link
from comfy_execution.validation import validate_node_input
from comfy_api.v3.io import NodeOutput, ComfyNodeV3, Hidden, NodeStateLocal, ResourcesLocal, AutogrowDynamic, is_class, lock_class
from comfy_api.v3 import io, helpers
class ExecutionResult(Enum):
@ -54,7 +54,7 @@ class IsChangedCache:
class_def = nodes.NODE_CLASS_MAPPINGS[class_type]
has_is_changed = False
is_changed_name = None
if issubclass(class_def, ComfyNodeV3) and getattr(class_def, "fingerprint_inputs", None) is not None:
if issubclass(class_def, io.ComfyNodeV3) and getattr(class_def, "fingerprint_inputs", None) is not None:
has_is_changed = True
is_changed_name = "fingerprint_inputs"
elif hasattr(class_def, "IS_CHANGED"):
@ -127,7 +127,7 @@ class CacheSet:
return result
def get_input_data(inputs, class_def, unique_id, outputs=None, dynprompt=None, extra_data={}):
is_v3 = issubclass(class_def, ComfyNodeV3)
is_v3 = issubclass(class_def, io.ComfyNodeV3)
if is_v3:
valid_inputs, schema = class_def.INPUT_TYPES(include_hidden=False, return_schema=True)
else:
@ -161,18 +161,18 @@ def get_input_data(inputs, class_def, unique_id, outputs=None, dynprompt=None, e
if is_v3:
if schema.hidden:
if Hidden.prompt in schema.hidden:
hidden_inputs_v3[Hidden.prompt] = dynprompt.get_original_prompt() if dynprompt is not None else {}
if Hidden.dynprompt in schema.hidden:
hidden_inputs_v3[Hidden.dynprompt] = dynprompt
if Hidden.extra_pnginfo in schema.hidden:
hidden_inputs_v3[Hidden.extra_pnginfo] = extra_data.get('extra_pnginfo', None)
if Hidden.unique_id in schema.hidden:
hidden_inputs_v3[Hidden.unique_id] = unique_id
if Hidden.auth_token_comfy_org in schema.hidden:
hidden_inputs_v3[Hidden.auth_token_comfy_org] = extra_data.get("auth_token_comfy_org", None)
if Hidden.api_key_comfy_org in schema.hidden:
hidden_inputs_v3[Hidden.api_key_comfy_org] = extra_data.get("api_key_comfy_org", None)
if io.Hidden.prompt in schema.hidden:
hidden_inputs_v3[io.Hidden.prompt] = dynprompt.get_original_prompt() if dynprompt is not None else {}
if io.Hidden.dynprompt in schema.hidden:
hidden_inputs_v3[io.Hidden.dynprompt] = dynprompt
if io.Hidden.extra_pnginfo in schema.hidden:
hidden_inputs_v3[io.Hidden.extra_pnginfo] = extra_data.get('extra_pnginfo', None)
if io.Hidden.unique_id in schema.hidden:
hidden_inputs_v3[io.Hidden.unique_id] = unique_id
if io.Hidden.auth_token_comfy_org in schema.hidden:
hidden_inputs_v3[io.Hidden.auth_token_comfy_org] = extra_data.get("auth_token_comfy_org", None)
if io.Hidden.api_key_comfy_org in schema.hidden:
hidden_inputs_v3[io.Hidden.api_key_comfy_org] = extra_data.get("api_key_comfy_org", None)
else:
if "hidden" in valid_inputs:
h = valid_inputs["hidden"]
@ -224,9 +224,9 @@ def _map_node_over_list(obj, input_data_all, func, allow_interrupt=False, execut
if pre_execute_cb is not None and index is not None:
pre_execute_cb(index)
# V3
if isinstance(obj, ComfyNodeV3) or (is_class(obj) and issubclass(obj, ComfyNodeV3)):
if isinstance(obj, io.ComfyNodeV3) or (io.is_class(obj) and issubclass(obj, io.ComfyNodeV3)):
# if is just a class, then assign no resources or state, just create clone
if is_class(obj):
if io.is_class(obj):
type_obj = obj
obj.VALIDATE_CLASS()
class_clone = obj.PREPARE_CLASS_CLONE(hidden_inputs)
@ -238,16 +238,16 @@ def _map_node_over_list(obj, input_data_all, func, allow_interrupt=False, execut
# NOTE: this is a mock of state management; for local, just stores NodeStateLocal on node instance
if hasattr(obj, "local_state"):
if obj.local_state is None:
obj.local_state = NodeStateLocal(class_clone.hidden.unique_id)
obj.local_state = io.NodeStateLocal(class_clone.hidden.unique_id)
class_clone.state = obj.local_state
# NOTE: this is a mock of resource management; for local, just stores ResourcesLocal on node instance
if hasattr(obj, "local_resources"):
if obj.local_resources is None:
obj.local_resources = ResourcesLocal()
obj.local_resources = io.ResourcesLocal()
class_clone.resources = obj.local_resources
# TODO: delete this when done testing mocking dynamic inputs
for si in obj.SCHEMA.inputs:
if isinstance(si, AutogrowDynamic.Input):
if isinstance(si, io.AutogrowDynamic.Input):
add_key = si.id
dynamic_list = []
real_inputs = {k: v for k, v in inputs.items()}
@ -255,7 +255,7 @@ def _map_node_over_list(obj, input_data_all, func, allow_interrupt=False, execut
dynamic_list.append(real_inputs.pop(d.id, None))
dynamic_list = [x for x in dynamic_list if x is not None]
inputs = {**real_inputs, add_key: dynamic_list}
results.append(getattr(type_obj, func).__func__(lock_class(class_clone), **inputs))
results.append(getattr(type_obj, func).__func__(io.lock_class(class_clone), **inputs))
# V1
else:
results.append(getattr(obj, func)(**inputs))
@ -318,7 +318,7 @@ def get_output_data(obj, input_data_all, execution_block_cb=None, pre_execute_cb
result = tuple([result] * len(obj.RETURN_TYPES))
results.append(result)
subgraph_results.append((None, result))
elif isinstance(r, NodeOutput):
elif isinstance(r, io.NodeOutput):
# V3
if r.ui is not None:
if isinstance(r.ui, dict):
@ -670,11 +670,9 @@ def validate_inputs(prompt, item, validated):
validate_function_inputs = []
validate_has_kwargs = False
validate_function_name = None
validate_function = None
if issubclass(obj_class, ComfyNodeV3):
if issubclass(obj_class, io.ComfyNodeV3):
validate_function_name = "validate_inputs"
validate_function = getattr(obj_class, validate_function_name, None)
validate_function = helpers.first_real_override(obj_class, validate_function_name, base=io.ComfyNodeV3)
else:
validate_function_name = "VALIDATE_INPUTS"
validate_function = getattr(obj_class, validate_function_name, None)