comfy_api/latest/_io.py

@@ -1004,9 +1004,8 @@ class Autogrow(ComfyTypeI):
             curr_prefix = handle_prefix(curr_prefix, self.id)
             # need to remove self from expected inputs dictionary; replaced by template inputs in frontend
             for inner_dict in d.values():
-                finalized_id = finalize_prefix(curr_prefix, self.id)
-                if finalized_id in inner_dict:
-                    del inner_dict[finalized_id]
+                if self.id in inner_dict:
+                    del inner_dict[self.id]
             self.template.expand_schema_for_dynamic(d, live_inputs, curr_prefix)
 
 @comfytype(io_type="COMFY_DYNAMICCOMBO_V3")
@@ -1032,10 +1031,9 @@ class DynamicCombo(ComfyTypeI):
 
         def expand_schema_for_dynamic(self, d: dict[str, Any], live_inputs: dict[str, Any], curr_prefix: list[str] | None=None):
             # check if dynamic input's id is in live_inputs
-            curr_prefix = handle_prefix(curr_prefix, self.id)
-            finalized_id = finalize_prefix(curr_prefix)
-            if finalized_id in live_inputs:
-                key = live_inputs[finalized_id]
+            if self.id in live_inputs:
+                curr_prefix = handle_prefix(curr_prefix, self.id)
+                key = live_inputs[self.id]
                 selected_option = None
                 for option in self.options:
                     if option.key == key:
@@ -1113,13 +1111,8 @@ def add_dynamic_id_mapping(d: dict[str, Any], inputs: list[Input], curr_prefix:
     if self is not None:
         dynamic[finalize_prefix(curr_prefix[:-1], self.id)] = finalize_prefix(curr_prefix, self.id)
     for i in inputs:
-        if not isinstance(i, DynamicInput):
-            dynamic[finalize_prefix(curr_prefix, i.id)] = finalize_prefix(curr_prefix, i.id)
-
-def add_to_dynamic_dict(dynamic: dict[str, Any], curr_prefix: list[str], id: str, value: str):
-    finalize_key = finalize_prefix(curr_prefix, id)
-    if finalize_key not in dynamic:
-        dynamic[finalize_key] = value
+        # if not isinstance(i, DynamicInput):
+        dynamic[finalize_prefix(curr_prefix, i.id)] = finalize_prefix(curr_prefix, i.id)
 
 class V3Data(TypedDict):
     hidden_inputs: dict[str, Any]

comfy_api/latest/_ui.py

@@ -3,7 +3,6 @@ from __future__ import annotations
 import json
 import os
 import random
-import uuid
 from io import BytesIO
 from typing import Type
 
@@ -319,10 +318,9 @@ class AudioSaveHelper:
             for key, value in metadata.items():
                 output_container.metadata[key] = value
 
-            layout = "mono" if waveform.shape[0] == 1 else "stereo"
             # Set up the output stream with appropriate properties
             if format == "opus":
-                out_stream = output_container.add_stream("libopus", rate=sample_rate, layout=layout)
+                out_stream = output_container.add_stream("libopus", rate=sample_rate)
                 if quality == "64k":
                     out_stream.bit_rate = 64000
                 elif quality == "96k":
@@ -334,7 +332,7 @@ class AudioSaveHelper:
                 elif quality == "320k":
                     out_stream.bit_rate = 320000
             elif format == "mp3":
-                out_stream = output_container.add_stream("libmp3lame", rate=sample_rate, layout=layout)
+                out_stream = output_container.add_stream("libmp3lame", rate=sample_rate)
                 if quality == "V0":
                     # TODO i would really love to support V3 and V5 but there doesn't seem to be a way to set the qscale level, the property below is a bool
                     out_stream.codec_context.qscale = 1
@@ -343,12 +341,12 @@ class AudioSaveHelper:
                 elif quality == "320k":
                     out_stream.bit_rate = 320000
             else:  # format == "flac":
-                out_stream = output_container.add_stream("flac", rate=sample_rate, layout=layout)
+                out_stream = output_container.add_stream("flac", rate=sample_rate)
 
             frame = av.AudioFrame.from_ndarray(
                 waveform.movedim(0, 1).reshape(1, -1).float().numpy(),
                 format="flt",
-                layout=layout,
+                layout="mono" if waveform.shape[0] == 1 else "stereo",
             )
             frame.sample_rate = sample_rate
             frame.pts = 0
@@ -438,19 +436,9 @@ class PreviewUI3D(_UIOutput):
     def __init__(self, model_file, camera_info, **kwargs):
         self.model_file = model_file
         self.camera_info = camera_info
-        self.bg_image_path = None
-        bg_image = kwargs.get("bg_image", None)
-        if bg_image is not None:
-            img_array = (bg_image[0].cpu().numpy() * 255).astype(np.uint8)
-            img = PILImage.fromarray(img_array)
-            temp_dir = folder_paths.get_temp_directory()
-            filename = f"bg_{uuid.uuid4().hex}.png"
-            bg_image_path = os.path.join(temp_dir, filename)
-            img.save(bg_image_path, compress_level=1)
-            self.bg_image_path = f"temp/{filename}"
 
     def as_dict(self):
-        return {"result": [self.model_file, self.camera_info, self.bg_image_path]}
+        return {"result": [self.model_file, self.camera_info]}
 
 
 class PreviewText(_UIOutput):

comfy_extras/nodes_audio.py

@@ -6,80 +6,65 @@ import torch
 import comfy.model_management
 import folder_paths
 import os
+import io
+import json
+import random
 import hashlib
 import node_helpers
 import logging
-from typing_extensions import override
-from comfy_api.latest import ComfyExtension, IO, UI
+from comfy.cli_args import args
+from comfy.comfy_types import FileLocator
 
-class EmptyLatentAudio(IO.ComfyNode):
-    @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="EmptyLatentAudio",
-            display_name="Empty Latent Audio",
-            category="latent/audio",
-            inputs=[
-                IO.Float.Input("seconds", default=47.6, min=1.0, max=1000.0, step=0.1),
-                IO.Int.Input(
-                    "batch_size", default=1, min=1, max=4096, tooltip="The number of latent images in the batch."
-                ),
-            ],
-            outputs=[IO.Latent.Output()],
-        )
+class EmptyLatentAudio:
+    def __init__(self):
+        self.device = comfy.model_management.intermediate_device()
 
     @classmethod
-    def execute(cls, seconds, batch_size) -> IO.NodeOutput:
+    def INPUT_TYPES(s):
+        return {"required": {"seconds": ("FLOAT", {"default": 47.6, "min": 1.0, "max": 1000.0, "step": 0.1}),
+                             "batch_size": ("INT", {"default": 1, "min": 1, "max": 4096, "tooltip": "The number of latent images in the batch."}),
+                             }}
+    RETURN_TYPES = ("LATENT",)
+    FUNCTION = "generate"
+
+    CATEGORY = "latent/audio"
+
+    def generate(self, seconds, batch_size):
         length = round((seconds * 44100 / 2048) / 2) * 2
-        latent = torch.zeros([batch_size, 64, length], device=comfy.model_management.intermediate_device())
-        return IO.NodeOutput({"samples":latent, "type": "audio"})
+        latent = torch.zeros([batch_size, 64, length], device=self.device)
+        return ({"samples":latent, "type": "audio"}, )
 
-    generate = execute  # TODO: remove
-
-
-class ConditioningStableAudio(IO.ComfyNode):
+class ConditioningStableAudio:
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="ConditioningStableAudio",
-            category="conditioning",
-            inputs=[
-                IO.Conditioning.Input("positive"),
-                IO.Conditioning.Input("negative"),
-                IO.Float.Input("seconds_start", default=0.0, min=0.0, max=1000.0, step=0.1),
-                IO.Float.Input("seconds_total", default=47.0, min=0.0, max=1000.0, step=0.1),
-            ],
-            outputs=[
-                IO.Conditioning.Output(display_name="positive"),
-                IO.Conditioning.Output(display_name="negative"),
-            ],
-        )
+    def INPUT_TYPES(s):
+        return {"required": {"positive": ("CONDITIONING", ),
+                             "negative": ("CONDITIONING", ),
+                             "seconds_start": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1000.0, "step": 0.1}),
+                             "seconds_total": ("FLOAT", {"default": 47.0, "min": 0.0, "max": 1000.0, "step": 0.1}),
+                             }}
 
-    @classmethod
-    def execute(cls, positive, negative, seconds_start, seconds_total) -> IO.NodeOutput:
+    RETURN_TYPES = ("CONDITIONING","CONDITIONING")
+    RETURN_NAMES = ("positive", "negative")
+
+    FUNCTION = "append"
+
+    CATEGORY = "conditioning"
+
+    def append(self, positive, negative, seconds_start, seconds_total):
         positive = node_helpers.conditioning_set_values(positive, {"seconds_start": seconds_start, "seconds_total": seconds_total})
         negative = node_helpers.conditioning_set_values(negative, {"seconds_start": seconds_start, "seconds_total": seconds_total})
-        return IO.NodeOutput(positive, negative)
+        return (positive, negative)
 
-    append = execute  # TODO: remove
-
-
-class VAEEncodeAudio(IO.ComfyNode):
+class VAEEncodeAudio:
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="VAEEncodeAudio",
-            display_name="VAE Encode Audio",
-            category="latent/audio",
-            inputs=[
-                IO.Audio.Input("audio"),
-                IO.Vae.Input("vae"),
-            ],
-            outputs=[IO.Latent.Output()],
-        )
+    def INPUT_TYPES(s):
+        return {"required": { "audio": ("AUDIO", ), "vae": ("VAE", )}}
+    RETURN_TYPES = ("LATENT",)
+    FUNCTION = "encode"
 
-    @classmethod
-    def execute(cls, vae, audio) -> IO.NodeOutput:
+    CATEGORY = "latent/audio"
+
+    def encode(self, vae, audio):
         sample_rate = audio["sample_rate"]
         if 44100 != sample_rate:
             waveform = torchaudio.functional.resample(audio["waveform"], sample_rate, 44100)
@@ -87,134 +72,213 @@ class VAEEncodeAudio(IO.ComfyNode):
             waveform = audio["waveform"]
 
         t = vae.encode(waveform.movedim(1, -1))
-        return IO.NodeOutput({"samples":t})
+        return ({"samples":t}, )
 
-    encode = execute  # TODO: remove
-
-
-class VAEDecodeAudio(IO.ComfyNode):
+class VAEDecodeAudio:
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="VAEDecodeAudio",
-            display_name="VAE Decode Audio",
-            category="latent/audio",
-            inputs=[
-                IO.Latent.Input("samples"),
-                IO.Vae.Input("vae"),
-            ],
-            outputs=[IO.Audio.Output()],
-        )
+    def INPUT_TYPES(s):
+        return {"required": { "samples": ("LATENT", ), "vae": ("VAE", )}}
+    RETURN_TYPES = ("AUDIO",)
+    FUNCTION = "decode"
 
-    @classmethod
-    def execute(cls, vae, samples) -> IO.NodeOutput:
+    CATEGORY = "latent/audio"
+
+    def decode(self, vae, samples):
         audio = vae.decode(samples["samples"]).movedim(-1, 1)
         std = torch.std(audio, dim=[1,2], keepdim=True) * 5.0
         std[std < 1.0] = 1.0
         audio /= std
-        return IO.NodeOutput({"waveform": audio, "sample_rate": 44100})
-
-    decode = execute  # TODO: remove
+        return ({"waveform": audio, "sample_rate": 44100}, )
 
 
-class SaveAudio(IO.ComfyNode):
-    @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="SaveAudio",
-            display_name="Save Audio (FLAC)",
-            category="audio",
-            inputs=[
-                IO.Audio.Input("audio"),
-                IO.String.Input("filename_prefix", default="audio/ComfyUI"),
-            ],
-            hidden=[IO.Hidden.prompt, IO.Hidden.extra_pnginfo],
-            is_output_node=True,
-        )
+def save_audio(self, audio, filename_prefix="ComfyUI", format="flac", prompt=None, extra_pnginfo=None, quality="128k"):
+
+    filename_prefix += self.prefix_append
+    full_output_folder, filename, counter, subfolder, filename_prefix = folder_paths.get_save_image_path(filename_prefix, self.output_dir)
+    results: list[FileLocator] = []
+
+    # Prepare metadata dictionary
+    metadata = {}
+    if not args.disable_metadata:
+        if prompt is not None:
+            metadata["prompt"] = json.dumps(prompt)
+        if extra_pnginfo is not None:
+            for x in extra_pnginfo:
+                metadata[x] = json.dumps(extra_pnginfo[x])
+
+    # Opus supported sample rates
+    OPUS_RATES = [8000, 12000, 16000, 24000, 48000]
+
+    for (batch_number, waveform) in enumerate(audio["waveform"].cpu()):
+        filename_with_batch_num = filename.replace("%batch_num%", str(batch_number))
+        file = f"{filename_with_batch_num}_{counter:05}_.{format}"
+        output_path = os.path.join(full_output_folder, file)
+
+        # Use original sample rate initially
+        sample_rate = audio["sample_rate"]
+
+        # Handle Opus sample rate requirements
+        if format == "opus":
+            if sample_rate > 48000:
+                sample_rate = 48000
+            elif sample_rate not in OPUS_RATES:
+                # Find the next highest supported rate
+                for rate in sorted(OPUS_RATES):
+                    if rate > sample_rate:
+                        sample_rate = rate
+                        break
+                if sample_rate not in OPUS_RATES:  # Fallback if still not supported
+                    sample_rate = 48000
+
+            # Resample if necessary
+            if sample_rate != audio["sample_rate"]:
+                waveform = torchaudio.functional.resample(waveform, audio["sample_rate"], sample_rate)
+
+        # Create output with specified format
+        output_buffer = io.BytesIO()
+        output_container = av.open(output_buffer, mode='w', format=format)
+
+        # Set metadata on the container
+        for key, value in metadata.items():
+            output_container.metadata[key] = value
+
+        layout = 'mono' if waveform.shape[0] == 1 else 'stereo'
+        # Set up the output stream with appropriate properties
+        if format == "opus":
+            out_stream = output_container.add_stream("libopus", rate=sample_rate, layout=layout)
+            if quality == "64k":
+                out_stream.bit_rate = 64000
+            elif quality == "96k":
+                out_stream.bit_rate = 96000
+            elif quality == "128k":
+                out_stream.bit_rate = 128000
+            elif quality == "192k":
+                out_stream.bit_rate = 192000
+            elif quality == "320k":
+                out_stream.bit_rate = 320000
+        elif format == "mp3":
+            out_stream = output_container.add_stream("libmp3lame", rate=sample_rate, layout=layout)
+            if quality == "V0":
+                #TODO i would really love to support V3 and V5 but there doesn't seem to be a way to set the qscale level, the property below is a bool
+                out_stream.codec_context.qscale = 1
+            elif quality == "128k":
+                out_stream.bit_rate = 128000
+            elif quality == "320k":
+                out_stream.bit_rate = 320000
+        else: #format == "flac":
+            out_stream = output_container.add_stream("flac", rate=sample_rate, layout=layout)
+
+        frame = av.AudioFrame.from_ndarray(waveform.movedim(0, 1).reshape(1, -1).float().numpy(), format='flt', layout=layout)
+        frame.sample_rate = sample_rate
+        frame.pts = 0
+        output_container.mux(out_stream.encode(frame))
+
+        # Flush encoder
+        output_container.mux(out_stream.encode(None))
+
+        # Close containers
+        output_container.close()
+
+        # Write the output to file
+        output_buffer.seek(0)
+        with open(output_path, 'wb') as f:
+            f.write(output_buffer.getbuffer())
+
+        results.append({
+            "filename": file,
+            "subfolder": subfolder,
+            "type": self.type
+        })
+        counter += 1
+
+    return { "ui": { "audio": results } }
+
+class SaveAudio:
+    def __init__(self):
+        self.output_dir = folder_paths.get_output_directory()
+        self.type = "output"
+        self.prefix_append = ""
 
     @classmethod
-    def execute(cls, audio, filename_prefix="ComfyUI", format="flac") -> IO.NodeOutput:
-        return IO.NodeOutput(
-            ui=UI.AudioSaveHelper.get_save_audio_ui(audio, filename_prefix=filename_prefix, cls=cls, format=format)
-        )
+    def INPUT_TYPES(s):
+        return {"required": { "audio": ("AUDIO", ),
+                            "filename_prefix": ("STRING", {"default": "audio/ComfyUI"}),
+                            },
+                "hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO"},
+                }
 
-    save_flac = execute  # TODO: remove
+    RETURN_TYPES = ()
+    FUNCTION = "save_flac"
 
+    OUTPUT_NODE = True
 
-class SaveAudioMP3(IO.ComfyNode):
-    @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="SaveAudioMP3",
-            display_name="Save Audio (MP3)",
-            category="audio",
-            inputs=[
-                IO.Audio.Input("audio"),
-                IO.String.Input("filename_prefix", default="audio/ComfyUI"),
-                IO.Combo.Input("quality", options=["V0", "128k", "320k"], default="V0"),
-            ],
-            hidden=[IO.Hidden.prompt, IO.Hidden.extra_pnginfo],
-            is_output_node=True,
-        )
+    CATEGORY = "audio"
+
+    def save_flac(self, audio, filename_prefix="ComfyUI", format="flac", prompt=None, extra_pnginfo=None):
+        return save_audio(self, audio, filename_prefix, format, prompt, extra_pnginfo)
+
+class SaveAudioMP3:
+    def __init__(self):
+        self.output_dir = folder_paths.get_output_directory()
+        self.type = "output"
+        self.prefix_append = ""
 
     @classmethod
-    def execute(cls, audio, filename_prefix="ComfyUI", format="mp3", quality="128k") -> IO.NodeOutput:
-        return IO.NodeOutput(
-            ui=UI.AudioSaveHelper.get_save_audio_ui(
-                audio, filename_prefix=filename_prefix, cls=cls, format=format, quality=quality
-            )
-        )
+    def INPUT_TYPES(s):
+        return {"required": { "audio": ("AUDIO", ),
+                            "filename_prefix": ("STRING", {"default": "audio/ComfyUI"}),
+                            "quality": (["V0", "128k", "320k"], {"default": "V0"}),
+                            },
+                "hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO"},
+                }
 
-    save_mp3 = execute  # TODO: remove
+    RETURN_TYPES = ()
+    FUNCTION = "save_mp3"
 
+    OUTPUT_NODE = True
 
-class SaveAudioOpus(IO.ComfyNode):
-    @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="SaveAudioOpus",
-            display_name="Save Audio (Opus)",
-            category="audio",
-            inputs=[
-                IO.Audio.Input("audio"),
-                IO.String.Input("filename_prefix", default="audio/ComfyUI"),
-                IO.Combo.Input("quality", options=["64k", "96k", "128k", "192k", "320k"], default="128k"),
-            ],
-            hidden=[IO.Hidden.prompt, IO.Hidden.extra_pnginfo],
-            is_output_node=True,
-        )
+    CATEGORY = "audio"
+
+    def save_mp3(self, audio, filename_prefix="ComfyUI", format="mp3", prompt=None, extra_pnginfo=None, quality="128k"):
+        return save_audio(self, audio, filename_prefix, format, prompt, extra_pnginfo, quality)
+
+class SaveAudioOpus:
+    def __init__(self):
+        self.output_dir = folder_paths.get_output_directory()
+        self.type = "output"
+        self.prefix_append = ""
 
     @classmethod
-    def execute(cls, audio, filename_prefix="ComfyUI", format="opus", quality="V3") -> IO.NodeOutput:
-        return IO.NodeOutput(
-            ui=UI.AudioSaveHelper.get_save_audio_ui(
-                audio, filename_prefix=filename_prefix, cls=cls, format=format, quality=quality
-            )
-        )
+    def INPUT_TYPES(s):
+        return {"required": { "audio": ("AUDIO", ),
+                            "filename_prefix": ("STRING", {"default": "audio/ComfyUI"}),
+                            "quality": (["64k", "96k", "128k", "192k", "320k"], {"default": "128k"}),
+                            },
+                "hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO"},
+                }
 
-    save_opus = execute  # TODO: remove
+    RETURN_TYPES = ()
+    FUNCTION = "save_opus"
 
+    OUTPUT_NODE = True
 
-class PreviewAudio(IO.ComfyNode):
-    @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="PreviewAudio",
-            display_name="Preview Audio",
-            category="audio",
-            inputs=[
-                IO.Audio.Input("audio"),
-            ],
-            hidden=[IO.Hidden.prompt, IO.Hidden.extra_pnginfo],
-            is_output_node=True,
-        )
+    CATEGORY = "audio"
+
+    def save_opus(self, audio, filename_prefix="ComfyUI", format="opus", prompt=None, extra_pnginfo=None, quality="V3"):
+        return save_audio(self, audio, filename_prefix, format, prompt, extra_pnginfo, quality)
+
+class PreviewAudio(SaveAudio):
+    def __init__(self):
+        self.output_dir = folder_paths.get_temp_directory()
+        self.type = "temp"
+        self.prefix_append = "_temp_" + ''.join(random.choice("abcdefghijklmnopqrstupvxyz") for x in range(5))
 
     @classmethod
-    def execute(cls, audio) -> IO.NodeOutput:
-        return IO.NodeOutput(ui=UI.PreviewAudio(audio, cls=cls))
-
-    save_flac = execute  # TODO: remove
-
+    def INPUT_TYPES(s):
+        return {"required":
+                    {"audio": ("AUDIO", ), },
+                "hidden": {"prompt": "PROMPT", "extra_pnginfo": "EXTRA_PNGINFO"},
+                }
 
 def f32_pcm(wav: torch.Tensor) -> torch.Tensor:
     """Convert audio to float 32 bits PCM format."""
@@ -252,30 +316,26 @@ def load(filepath: str) -> tuple[torch.Tensor, int]:
         wav = f32_pcm(wav)
         return wav, sr
 
-class LoadAudio(IO.ComfyNode):
+class LoadAudio:
     @classmethod
-    def define_schema(cls):
+    def INPUT_TYPES(s):
         input_dir = folder_paths.get_input_directory()
         files = folder_paths.filter_files_content_types(os.listdir(input_dir), ["audio", "video"])
-        return IO.Schema(
-            node_id="LoadAudio",
-            display_name="Load Audio",
-            category="audio",
-            inputs=[
-                IO.Combo.Input("audio", upload=IO.UploadType.audio, options=sorted(files)),
-            ],
-            outputs=[IO.Audio.Output()],
-        )
+        return {"required": {"audio": (sorted(files), {"audio_upload": True})}}
 
-    @classmethod
-    def execute(cls, audio) -> IO.NodeOutput:
+    CATEGORY = "audio"
+
+    RETURN_TYPES = ("AUDIO", )
+    FUNCTION = "load"
+
+    def load(self, audio):
         audio_path = folder_paths.get_annotated_filepath(audio)
         waveform, sample_rate = load(audio_path)
         audio = {"waveform": waveform.unsqueeze(0), "sample_rate": sample_rate}
-        return IO.NodeOutput(audio)
+        return (audio, )
 
     @classmethod
-    def fingerprint_inputs(cls, audio):
+    def IS_CHANGED(s, audio):
         image_path = folder_paths.get_annotated_filepath(audio)
         m = hashlib.sha256()
         with open(image_path, 'rb') as f:
@@ -283,69 +343,46 @@ class LoadAudio(IO.ComfyNode):
         return m.digest().hex()
 
     @classmethod
-    def validate_inputs(cls, audio):
+    def VALIDATE_INPUTS(s, audio):
         if not folder_paths.exists_annotated_filepath(audio):
             return "Invalid audio file: {}".format(audio)
         return True
 
-    load = execute  # TODO: remove
-
-
-class RecordAudio(IO.ComfyNode):
+class RecordAudio:
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="RecordAudio",
-            display_name="Record Audio",
-            category="audio",
-            inputs=[
-                IO.Custom("AUDIO_RECORD").Input("audio"),
-            ],
-            outputs=[IO.Audio.Output()],
-        )
+    def INPUT_TYPES(s):
+        return {"required": {"audio": ("AUDIO_RECORD", {})}}
 
-    @classmethod
-    def execute(cls, audio) -> IO.NodeOutput:
+    CATEGORY = "audio"
+
+    RETURN_TYPES = ("AUDIO", )
+    FUNCTION = "load"
+
+    def load(self, audio):
         audio_path = folder_paths.get_annotated_filepath(audio)
 
         waveform, sample_rate = load(audio_path)
         audio = {"waveform": waveform.unsqueeze(0), "sample_rate": sample_rate}
-        return IO.NodeOutput(audio)
-
-    load = execute  # TODO: remove
+        return (audio, )
 
 
-class TrimAudioDuration(IO.ComfyNode):
+class TrimAudioDuration:
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="TrimAudioDuration",
-            display_name="Trim Audio Duration",
-            description="Trim audio tensor into chosen time range.",
-            category="audio",
-            inputs=[
-                IO.Audio.Input("audio"),
-                IO.Float.Input(
-                    "start_index",
-                    default=0.0,
-                    min=-0xffffffffffffffff,
-                    max=0xffffffffffffffff,
-                    step=0.01,
-                    tooltip="Start time in seconds, can be negative to count from the end (supports sub-seconds).",
-                ),
-                IO.Float.Input(
-                    "duration",
-                    default=60.0,
-                    min=0.0,
-                    step=0.01,
-                    tooltip="Duration in seconds",
-                ),
-            ],
-            outputs=[IO.Audio.Output()],
-        )
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "audio": ("AUDIO",),
+                "start_index": ("FLOAT", {"default": 0.0, "min": -0xffffffffffffffff, "max": 0xffffffffffffffff, "step": 0.01, "tooltip": "Start time in seconds, can be negative to count from the end (supports sub-seconds)."}),
+                "duration": ("FLOAT", {"default": 60.0, "min": 0.0, "step": 0.01, "tooltip": "Duration in seconds"}),
+            },
+        }
 
-    @classmethod
-    def execute(cls, audio, start_index, duration) -> IO.NodeOutput:
+    FUNCTION = "trim"
+    RETURN_TYPES = ("AUDIO",)
+    CATEGORY = "audio"
+    DESCRIPTION = "Trim audio tensor into chosen time range."
+
+    def trim(self, audio, start_index, duration):
         waveform = audio["waveform"]
         sample_rate = audio["sample_rate"]
         audio_length = waveform.shape[-1]
@@ -362,30 +399,23 @@ class TrimAudioDuration(IO.ComfyNode):
         if start_frame >= end_frame:
             raise ValueError("AudioTrim: Start time must be less than end time and be within the audio length.")
 
-        return IO.NodeOutput({"waveform": waveform[..., start_frame:end_frame], "sample_rate": sample_rate})
-
-    trim = execute  # TODO: remove
+        return ({"waveform": waveform[..., start_frame:end_frame], "sample_rate": sample_rate},)
 
 
-class SplitAudioChannels(IO.ComfyNode):
+class SplitAudioChannels:
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="SplitAudioChannels",
-            display_name="Split Audio Channels",
-            description="Separates the audio into left and right channels.",
-            category="audio",
-            inputs=[
-                IO.Audio.Input("audio"),
-            ],
-            outputs=[
-                IO.Audio.Output(display_name="left"),
-                IO.Audio.Output(display_name="right"),
-            ],
-        )
+    def INPUT_TYPES(s):
+        return {"required": {
+            "audio": ("AUDIO",),
+        }}
 
-    @classmethod
-    def execute(cls, audio) -> IO.NodeOutput:
+    RETURN_TYPES = ("AUDIO", "AUDIO")
+    RETURN_NAMES = ("left", "right")
+    FUNCTION = "separate"
+    CATEGORY = "audio"
+    DESCRIPTION = "Separates the audio into left and right channels."
+
+    def separate(self, audio):
         waveform = audio["waveform"]
         sample_rate = audio["sample_rate"]
 
@@ -395,9 +425,7 @@ class SplitAudioChannels(IO.ComfyNode):
         left_channel = waveform[..., 0:1, :]
         right_channel = waveform[..., 1:2, :]
 
-        return IO.NodeOutput({"waveform": left_channel, "sample_rate": sample_rate}, {"waveform": right_channel, "sample_rate": sample_rate})
-
-    separate = execute  # TODO: remove
+        return ({"waveform": left_channel, "sample_rate": sample_rate}, {"waveform": right_channel, "sample_rate": sample_rate})
 
 
 def match_audio_sample_rates(waveform_1, sample_rate_1, waveform_2, sample_rate_2):
@@ -415,29 +443,21 @@ def match_audio_sample_rates(waveform_1, sample_rate_1, waveform_2, sample_rate_
     return waveform_1, waveform_2, output_sample_rate
 
 
-class AudioConcat(IO.ComfyNode):
+class AudioConcat:
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="AudioConcat",
-            display_name="Audio Concat",
-            description="Concatenates the audio1 to audio2 in the specified direction.",
-            category="audio",
-            inputs=[
-                IO.Audio.Input("audio1"),
-                IO.Audio.Input("audio2"),
-                IO.Combo.Input(
-                    "direction",
-                    options=['after', 'before'],
-                    default="after",
-                    tooltip="Whether to append audio2 after or before audio1.",
-                )
-            ],
-            outputs=[IO.Audio.Output()],
-        )
+    def INPUT_TYPES(s):
+        return {"required": {
+            "audio1": ("AUDIO",),
+            "audio2": ("AUDIO",),
+            "direction": (['after', 'before'], {"default": 'after', "tooltip": "Whether to append audio2 after or before audio1."}),
+        }}
 
-    @classmethod
-    def execute(cls, audio1, audio2, direction) -> IO.NodeOutput:
+    RETURN_TYPES = ("AUDIO",)
+    FUNCTION = "concat"
+    CATEGORY = "audio"
+    DESCRIPTION = "Concatenates the audio1 to audio2 in the specified direction."
+
+    def concat(self, audio1, audio2, direction):
         waveform_1 = audio1["waveform"]
         waveform_2 = audio2["waveform"]
         sample_rate_1 = audio1["sample_rate"]
@@ -457,33 +477,26 @@ class AudioConcat(IO.ComfyNode):
         elif direction == 'before':
             concatenated_audio = torch.cat((waveform_2, waveform_1), dim=2)
 
-        return IO.NodeOutput({"waveform": concatenated_audio, "sample_rate": output_sample_rate})
-
-    concat = execute  # TODO: remove
+        return ({"waveform": concatenated_audio, "sample_rate": output_sample_rate},)
 
 
-class AudioMerge(IO.ComfyNode):
+class AudioMerge:
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="AudioMerge",
-            display_name="Audio Merge",
-            description="Combine two audio tracks by overlaying their waveforms.",
-            category="audio",
-            inputs=[
-                IO.Audio.Input("audio1"),
-                IO.Audio.Input("audio2"),
-                IO.Combo.Input(
-                    "merge_method",
-                    options=["add", "mean", "subtract", "multiply"],
-                    tooltip="The method used to combine the audio waveforms.",
-                )
-            ],
-            outputs=[IO.Audio.Output()],
-        )
+    def INPUT_TYPES(cls):
+        return {
+            "required": {
+                "audio1": ("AUDIO",),
+                "audio2": ("AUDIO",),
+                "merge_method": (["add", "mean", "subtract", "multiply"], {"tooltip": "The method used to combine the audio waveforms."}),
+            },
+        }
 
-    @classmethod
-    def execute(cls, audio1, audio2, merge_method) -> IO.NodeOutput:
+    FUNCTION = "merge"
+    RETURN_TYPES = ("AUDIO",)
+    CATEGORY = "audio"
+    DESCRIPTION = "Combine two audio tracks by overlaying their waveforms."
+
+    def merge(self, audio1, audio2, merge_method):
         waveform_1 = audio1["waveform"]
         waveform_2 = audio2["waveform"]
         sample_rate_1 = audio1["sample_rate"]
@@ -517,108 +530,85 @@ class AudioMerge(IO.ComfyNode):
         if max_val > 1.0:
             waveform = waveform / max_val
 
-        return IO.NodeOutput({"waveform": waveform, "sample_rate": output_sample_rate})
-
-    merge = execute  # TODO: remove
+        return ({"waveform": waveform, "sample_rate": output_sample_rate},)
 
 
-class AudioAdjustVolume(IO.ComfyNode):
+class AudioAdjustVolume:
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="AudioAdjustVolume",
-            display_name="Audio Adjust Volume",
-            category="audio",
-            inputs=[
-                IO.Audio.Input("audio"),
-                IO.Int.Input(
-                    "volume",
-                    default=1,
-                    min=-100,
-                    max=100,
-                    tooltip="Volume adjustment in decibels (dB). 0 = no change, +6 = double, -6 = half, etc",
-                )
-            ],
-            outputs=[IO.Audio.Output()],
-        )
+    def INPUT_TYPES(s):
+        return {"required": {
+            "audio": ("AUDIO",),
+            "volume": ("INT", {"default": 1.0, "min": -100, "max": 100, "tooltip": "Volume adjustment in decibels (dB). 0 = no change, +6 = double, -6 = half, etc"}),
+        }}
 
-    @classmethod
-    def execute(cls, audio, volume) -> IO.NodeOutput:
+    RETURN_TYPES = ("AUDIO",)
+    FUNCTION = "adjust_volume"
+    CATEGORY = "audio"
+
+    def adjust_volume(self, audio, volume):
         if volume == 0:
-            return IO.NodeOutput(audio)
+            return (audio,)
         waveform = audio["waveform"]
         sample_rate = audio["sample_rate"]
 
         gain = 10 ** (volume / 20)
         waveform = waveform * gain
 
-        return IO.NodeOutput({"waveform": waveform, "sample_rate": sample_rate})
-
-    adjust_volume = execute  # TODO: remove
+        return ({"waveform": waveform, "sample_rate": sample_rate},)
 
 
-class EmptyAudio(IO.ComfyNode):
+class EmptyAudio:
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="EmptyAudio",
-            display_name="Empty Audio",
-            category="audio",
-            inputs=[
-                IO.Float.Input(
-                    "duration",
-                    default=60.0,
-                    min=0.0,
-                    max=0xffffffffffffffff,
-                    step=0.01,
-                    tooltip="Duration of the empty audio clip in seconds",
-                ),
-                IO.Float.Input(
-                    "sample_rate",
-                    default=44100,
-                    tooltip="Sample rate of the empty audio clip.",
-                ),
-                IO.Float.Input(
-                    "channels",
-                    default=2,
-                    min=1,
-                    max=2,
-                    tooltip="Number of audio channels (1 for mono, 2 for stereo).",
-                ),
-            ],
-            outputs=[IO.Audio.Output()],
-        )
+    def INPUT_TYPES(s):
+        return {"required": {
+            "duration": ("FLOAT", {"default": 60.0, "min": 0.0, "max": 0xffffffffffffffff, "step": 0.01, "tooltip": "Duration of the empty audio clip in seconds"}),
+            "sample_rate": ("INT", {"default": 44100, "tooltip": "Sample rate of the empty audio clip."}),
+            "channels": ("INT", {"default": 2, "min": 1, "max": 2, "tooltip": "Number of audio channels (1 for mono, 2 for stereo)."}),
+        }}
 
-    @classmethod
-    def execute(cls, duration, sample_rate, channels) -> IO.NodeOutput:
+    RETURN_TYPES = ("AUDIO",)
+    FUNCTION = "create_empty_audio"
+    CATEGORY = "audio"
+
+    def create_empty_audio(self, duration, sample_rate, channels):
         num_samples = int(round(duration * sample_rate))
         waveform = torch.zeros((1, channels, num_samples), dtype=torch.float32)
-        return IO.NodeOutput({"waveform": waveform, "sample_rate": sample_rate})
-
-    create_empty_audio = execute  # TODO: remove
+        return ({"waveform": waveform, "sample_rate": sample_rate},)
 
 
-class AudioExtension(ComfyExtension):
-    @override
-    async def get_node_list(self) -> list[type[IO.ComfyNode]]:
-        return [
-            EmptyLatentAudio,
-            VAEEncodeAudio,
-            VAEDecodeAudio,
-            SaveAudio,
-            SaveAudioMP3,
-            SaveAudioOpus,
-            LoadAudio,
-            PreviewAudio,
-            ConditioningStableAudio,
-            RecordAudio,
-            TrimAudioDuration,
-            SplitAudioChannels,
-            AudioConcat,
-            AudioMerge,
-            AudioAdjustVolume,
-            EmptyAudio,
-        ]
+NODE_CLASS_MAPPINGS = {
+    "EmptyLatentAudio": EmptyLatentAudio,
+    "VAEEncodeAudio": VAEEncodeAudio,
+    "VAEDecodeAudio": VAEDecodeAudio,
+    "SaveAudio": SaveAudio,
+    "SaveAudioMP3": SaveAudioMP3,
+    "SaveAudioOpus": SaveAudioOpus,
+    "LoadAudio": LoadAudio,
+    "PreviewAudio": PreviewAudio,
+    "ConditioningStableAudio": ConditioningStableAudio,
+    "RecordAudio": RecordAudio,
+    "TrimAudioDuration": TrimAudioDuration,
+    "SplitAudioChannels": SplitAudioChannels,
+    "AudioConcat": AudioConcat,
+    "AudioMerge": AudioMerge,
+    "AudioAdjustVolume": AudioAdjustVolume,
+    "EmptyAudio": EmptyAudio,
+}
 
-async def comfy_entrypoint() -> AudioExtension:
-    return AudioExtension()
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "EmptyLatentAudio": "Empty Latent Audio",
+    "VAEEncodeAudio": "VAE Encode Audio",
+    "VAEDecodeAudio": "VAE Decode Audio",
+    "PreviewAudio": "Preview Audio",
+    "LoadAudio": "Load Audio",
+    "SaveAudio": "Save Audio (FLAC)",
+    "SaveAudioMP3": "Save Audio (MP3)",
+    "SaveAudioOpus": "Save Audio (Opus)",
+    "RecordAudio": "Record Audio",
+    "TrimAudioDuration": "Trim Audio Duration",
+    "SplitAudioChannels": "Split Audio Channels",
+    "AudioConcat": "Audio Concat",
+    "AudioMerge": "Audio Merge",
+    "AudioAdjustVolume": "Audio Adjust Volume",
+    "EmptyAudio": "Empty Audio",
+}

comfy_extras/nodes_load_3d.py

@@ -2,18 +2,22 @@ import nodes
 import folder_paths
 import os
 
-from typing_extensions import override
-from comfy_api.latest import IO, ComfyExtension, InputImpl, UI
+from comfy.comfy_types import IO
+from comfy_api.input_impl import VideoFromFile
 
 from pathlib import Path
 
+from PIL import Image
+import numpy as np
+
+import uuid
 
 def normalize_path(path):
     return path.replace('\\', '/')
 
-class Load3D(IO.ComfyNode):
+class Load3D():
     @classmethod
-    def define_schema(cls):
+    def INPUT_TYPES(s):
         input_dir = os.path.join(folder_paths.get_input_directory(), "3d")
 
         os.makedirs(input_dir, exist_ok=True)
@@ -26,29 +30,23 @@ class Load3D(IO.ComfyNode):
             for file_path in input_path.rglob("*")
             if file_path.suffix.lower() in {'.gltf', '.glb', '.obj', '.fbx', '.stl'}
         ]
-        return IO.Schema(
-            node_id="Load3D",
-            display_name="Load 3D & Animation",
-            category="3d",
-            is_experimental=True,
-            inputs=[
-                IO.Combo.Input("model_file", options=sorted(files), upload=IO.UploadType.model),
-                IO.Load3D.Input("image"),
-                IO.Int.Input("width", default=1024, min=1, max=4096, step=1),
-                IO.Int.Input("height", default=1024, min=1, max=4096, step=1),
-            ],
-            outputs=[
-                IO.Image.Output(display_name="image"),
-                IO.Mask.Output(display_name="mask"),
-                IO.String.Output(display_name="mesh_path"),
-                IO.Image.Output(display_name="normal"),
-                IO.Load3DCamera.Output(display_name="camera_info"),
-                IO.Video.Output(display_name="recording_video"),
-            ],
-        )
 
-    @classmethod
-    def execute(cls, model_file, image, **kwargs) -> IO.NodeOutput:
+        return {"required": {
+            "model_file": (sorted(files), {"file_upload": True}),
+            "image": ("LOAD_3D", {}),
+            "width": ("INT", {"default": 1024, "min": 1, "max": 4096, "step": 1}),
+            "height": ("INT", {"default": 1024, "min": 1, "max": 4096, "step": 1}),
+        }}
+
+    RETURN_TYPES = ("IMAGE", "MASK", "STRING", "IMAGE", "LOAD3D_CAMERA", IO.VIDEO)
+    RETURN_NAMES = ("image", "mask", "mesh_path", "normal", "camera_info", "recording_video")
+
+    FUNCTION = "process"
+    EXPERIMENTAL = True
+
+    CATEGORY = "3d"
+
+    def process(self, model_file, image, **kwargs):
         image_path = folder_paths.get_annotated_filepath(image['image'])
         mask_path = folder_paths.get_annotated_filepath(image['mask'])
         normal_path = folder_paths.get_annotated_filepath(image['normal'])
@@ -63,47 +61,58 @@ class Load3D(IO.ComfyNode):
         if image['recording'] != "":
             recording_video_path = folder_paths.get_annotated_filepath(image['recording'])
 
-            video = InputImpl.VideoFromFile(recording_video_path)
+            video = VideoFromFile(recording_video_path)
 
-        return IO.NodeOutput(output_image, output_mask, model_file, normal_image, image['camera_info'], video)
+        return output_image, output_mask, model_file, normal_image, image['camera_info'], video
 
-    process = execute  # TODO: remove
-
-
-class Preview3D(IO.ComfyNode):
+class Preview3D():
     @classmethod
-    def define_schema(cls):
-        return IO.Schema(
-            node_id="Preview3D",
-            display_name="Preview 3D & Animation",
-            category="3d",
-            is_experimental=True,
-            is_output_node=True,
-            inputs=[
-                IO.String.Input("model_file", default="", multiline=False),
-                IO.Load3DCamera.Input("camera_info", optional=True),
-                IO.Image.Input("bg_image", optional=True),
-            ],
-            outputs=[],
-        )
+    def INPUT_TYPES(s):
+        return {"required": {
+            "model_file": ("STRING", {"default": "", "multiline": False}),
+        },
+        "optional": {
+            "camera_info": ("LOAD3D_CAMERA", {}),
+            "bg_image": ("IMAGE", {})
+        }}
 
-    @classmethod
-    def execute(cls, model_file, **kwargs) -> IO.NodeOutput:
+    OUTPUT_NODE = True
+    RETURN_TYPES = ()
+
+    CATEGORY = "3d"
+
+    FUNCTION = "process"
+    EXPERIMENTAL = True
+
+    def process(self, model_file, **kwargs):
         camera_info = kwargs.get("camera_info", None)
         bg_image = kwargs.get("bg_image", None)
-        return IO.NodeOutput(ui=UI.PreviewUI3D(model_file, camera_info, bg_image=bg_image))
 
-    process = execute  # TODO: remove
+        bg_image_path = None
+        if bg_image is not None:
 
+            img_array = (bg_image[0].cpu().numpy() * 255).astype(np.uint8)
+            img = Image.fromarray(img_array)
 
-class Load3DExtension(ComfyExtension):
-    @override
-    async def get_node_list(self) -> list[type[IO.ComfyNode]]:
-        return [
-            Load3D,
-            Preview3D,
-        ]
+            temp_dir = folder_paths.get_temp_directory()
+            filename = f"bg_{uuid.uuid4().hex}.png"
+            bg_image_path = os.path.join(temp_dir, filename)
+            img.save(bg_image_path, compress_level=1)
 
+            bg_image_path = f"temp/{filename}"
 
-async def comfy_entrypoint() -> Load3DExtension:
-    return Load3DExtension()
+        return {
+            "ui": {
+                "result": [model_file, camera_info, bg_image_path]
+            }
+        }
+
+NODE_CLASS_MAPPINGS = {
+    "Load3D": Load3D,
+    "Preview3D": Preview3D,
+}
+
+NODE_DISPLAY_NAME_MAPPINGS = {
+    "Load3D": "Load 3D & Animation",
+    "Preview3D": "Preview 3D & Animation",
+}

comfy_extras/nodes_logic.py

@@ -3,8 +3,6 @@ from typing_extensions import override
 from comfy_api.latest import ComfyExtension, io
 from comfy_api.latest import _io
 
-# sentinel for missing inputs
-MISSING = object()
 
 
 class SwitchNode(io.ComfyNode):
@@ -27,14 +25,14 @@ class SwitchNode(io.ComfyNode):
         )
 
     @classmethod
-    def check_lazy_status(cls, switch, on_false=MISSING, on_true=MISSING):
-        # We use MISSING instead of None, as None is passed for connected-but-unevaluated inputs.
+    def check_lazy_status(cls, switch, on_false=..., on_true=...):
+        # We use ... instead of None, as None is passed for connected-but-unevaluated inputs.
         # This trick allows us to ignore the value of the switch and still be able to run execute().
 
         # One of the inputs may be missing, in which case we need to evaluate the other input
-        if on_false is MISSING:
+        if on_false is ...:
             return ["on_true"]
-        if on_true is MISSING:
+        if on_true is ...:
             return ["on_false"]
         # Normal lazy switch operation
         if switch and on_true is None:
@@ -43,43 +41,22 @@ class SwitchNode(io.ComfyNode):
             return ["on_false"]
 
     @classmethod
-    def validate_inputs(cls, switch, on_false=MISSING, on_true=MISSING):
+    def validate_inputs(cls, switch, on_false=..., on_true=...):
         # This check happens before check_lazy_status(), so we can eliminate the case where
         # both inputs are missing.
-        if on_false is MISSING and on_true is MISSING:
+        if on_false is ... and on_true is ...:
             return "At least one of on_false or on_true must be connected to Switch node"
         return True
 
     @classmethod
-    def execute(cls, switch, on_true=MISSING, on_false=MISSING) -> io.NodeOutput:
-        if on_true is MISSING:
+    def execute(cls, switch, on_true=..., on_false=...) -> io.NodeOutput:
+        if on_true is ...:
             return io.NodeOutput(on_false)
-        if on_false is MISSING:
+        if on_false is ...:
             return io.NodeOutput(on_true)
         return io.NodeOutput(on_true if switch else on_false)
 
 
-class CustomComboNode(io.ComfyNode):
-    """
-    Frontend node that allows user to write their own options for a combo.
-    This is here to make sure the node has a backend-representation to avoid some annoyances.
-    """
-    @classmethod
-    def define_schema(cls):
-        return io.Schema(
-            node_id="CustomCombo",
-            display_name="Custom Combo",
-            category="util",
-            is_experimental=True,
-            inputs=[io.Combo.Input("choice", options=[])],
-            outputs=[io.String.Output()]
-        )
-
-    @classmethod
-    def execute(cls, choice: io.Combo.Type) -> io.NodeOutput:
-        return io.NodeOutput(choice)
-
-
 class DCTestNode(io.ComfyNode):
     class DCValues(TypedDict):
         combo: str
@@ -185,7 +162,6 @@ class LogicExtension(ComfyExtension):
     async def get_node_list(self) -> list[type[io.ComfyNode]]:
         return [
             SwitchNode,
-            CustomComboNode,
             DCTestNode,
             AutogrowNamesTestNode,
             AutogrowPrefixTestNode,

execution.py

@@ -756,7 +756,6 @@ async def validate_inputs(prompt_id, prompt, item, validated):
     validate_function_inputs = []
     validate_has_kwargs = False
     if issubclass(obj_class, _ComfyNodeInternal):
-        obj_class: _io._ComfyNodeBaseInternal
         class_inputs, _, _ = obj_class.INPUT_TYPES(include_hidden=False, return_schema=True, live_inputs=inputs)
         validate_function_name = "validate_inputs"
         validate_function = first_real_override(obj_class, validate_function_name)