nodes/set-soft-ComfyUI-AudioBatch
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Migrated to SeCoNoHe [Experimental]

Browse Source
This commit is contained in:
Salvador E. Tropea
2025-07-22 13:55:11 -03:00
parent df94b6b419
commit 8599693fba
29 changed files with 25 additions and 538 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.pre-commit-config.yaml
+1 -1
View File
@@ -51,7 +51,7 @@ repos:
# The command to execute. It's expected to be found in the PATH
# of the environment where you run `git commit`.
entry: mypy
args: ["--explicit-package-bases", "source/"]
args: ["--explicit-package-bases", "src/"]
# 'system' tells pre-commit to find the command in the current environment
# instead of building an isolated one.
language: system
__init__.py
+5 -20
View File
@@ -3,26 +3,11 @@
# Copyright (c) 2025 Instituto Nacional de Tecnologïa Industrial
# License: GPL-3.0
# Project: ComfyUI-AudioBatch
from .source.nodes import nodes_audio
import inspect
import logging
from .source.nodes.utils.misc import NODES_NAME
from .src.nodes import nodes_audio, main_logger
from seconohe.register_nodes import register_nodes
from seconohe import JS_PATH
init_logger = logging.getLogger(NODES_NAME + ".__init__")
NODE_CLASS_MAPPINGS = {}
NODE_DISPLAY_NAME_MAPPINGS = {}
for name, obj in inspect.getmembers(nodes_audio):
# We skip nodes imported from the ComfyUI main nodes
if not inspect.isclass(obj) or not hasattr(obj, "INPUT_TYPES") or obj.__module__ == "nodes":
continue
assert hasattr(obj, "UNIQUE_NAME"), f"No name for {obj.__name__}"
NODE_CLASS_MAPPINGS[obj.UNIQUE_NAME] = obj
NODE_DISPLAY_NAME_MAPPINGS[obj.UNIQUE_NAME] = obj.DISPLAY_NAME
init_logger.info(f"Registering {len(NODE_CLASS_MAPPINGS)} node(s).")
init_logger.debug(f"{list(NODE_DISPLAY_NAME_MAPPINGS.values())}")
WEB_DIRECTORY = "./js"
NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS = register_nodes(main_logger, [nodes_audio])
WEB_DIRECTORY = JS_PATH
__all__ = ['NODE_CLASS_MAPPINGS', 'NODE_DISPLAY_NAME_MAPPINGS']
mypy.ini
+1 -1
View File
@@ -3,7 +3,7 @@
python_version = 3.11
warn_return_any = True
warn_unused_ignores = True
mypy_path = source/
mypy_path = src/
# This is often needed when starting out, especially with libraries that lack stubs
ignore_missing_imports = True
source/nodes/utils/__init__.py
View File
source/nodes/utils/ansi.py
-113
View File
@@ -1,113 +0,0 @@
# Copyright Jonathan Hartley 2013. BSD 3-Clause license, see LICENSE file.
'''
This module generates ANSI character codes to printing colors to terminals.
See: http://en.wikipedia.org/wiki/ANSI_escape_code
'''
import sys
import os
CSI = '\033['
OSC = '\033]'
BEL = '\a'
is_a_tty = sys.stderr.isatty() and os.name == 'posix'
def code_to_chars(code):
return CSI + str(code) + 'm' if is_a_tty else ''
def set_title(title):
return OSC + '2;' + title + BEL
def clear_screen(mode=2):
return CSI + str(mode) + 'J'
def clear_line(mode=2):
return CSI + str(mode) + 'K'
class AnsiCodes(object):
def __init__(self):
# the subclasses declare class attributes which are numbers.
# Upon instantiation we define instance attributes, which are the same
# as the class attributes but wrapped with the ANSI escape sequence
for name in dir(self):
if not name.startswith('_'):
value = getattr(self, name)
setattr(self, name, code_to_chars(value))
class AnsiCursor(object):
def UP(self, n=1):
return CSI + str(n) + 'A'
def DOWN(self, n=1):
return CSI + str(n) + 'B'
def FORWARD(self, n=1):
return CSI + str(n) + 'C'
def BACK(self, n=1):
return CSI + str(n) + 'D'
def POS(self, x=1, y=1):
return CSI + str(y) + ';' + str(x) + 'H'
class AnsiFore(AnsiCodes):
BLACK = 30
RED = 31
GREEN = 32
YELLOW = 33
BLUE = 34
MAGENTA = 35
CYAN = 36
WHITE = 37
RESET = 39
# These are fairly well supported, but not part of the standard.
LIGHTBLACK_EX = 90
LIGHTRED_EX = 91
LIGHTGREEN_EX = 92
LIGHTYELLOW_EX = 93
LIGHTBLUE_EX = 94
LIGHTMAGENTA_EX = 95
LIGHTCYAN_EX = 96
LIGHTWHITE_EX = 97
class AnsiBack(AnsiCodes):
BLACK = 40
RED = 41
GREEN = 42
YELLOW = 43
BLUE = 44
MAGENTA = 45
CYAN = 46
WHITE = 47
RESET = 49
# These are fairly well supported, but not part of the standard.
LIGHTBLACK_EX = 100
LIGHTRED_EX = 101
LIGHTGREEN_EX = 102
LIGHTYELLOW_EX = 103
LIGHTBLUE_EX = 104
LIGHTMAGENTA_EX = 105
LIGHTCYAN_EX = 106
LIGHTWHITE_EX = 107
class AnsiStyle(AnsiCodes):
BRIGHT = 1
DIM = 2
NORMAL = 22
RESET_ALL = 0
Fore = AnsiFore()
Back = AnsiBack()
Style = AnsiStyle()
Cursor = AnsiCursor()
source/nodes/utils/comfy_notification.py
-47
View File
@@ -1,47 +0,0 @@
# Copyright (c) 2025 Salvador E. Tropea
# Copyright (c) 2025 Instituto Nacional de Tecnología Industrial
# License: GPLv3
# Project: ComfyUI-AudioBatch
#
# ComfyUI Toast API messages
# Original code from Gemini 2.5 Pro, which was really outdated
# Took ideas from Easy Use nodes and looking at ComfyUI code
import logging
from typing import Optional
# ComfyUI imports
try:
from server import PromptServer
with_comfy = True
except Exception:
with_comfy = False
# Local imports
from .misc import NODES_NAME
logger = logging.getLogger(f"{NODES_NAME}.comfy_notification")
def send_toast_notification(message: str, summary: str = "Warning", severity: str = "warn", sid: Optional[str] = None):
"""
Sends a toast notification event to the ComfyUI client.
Args:
message (str): The message content of the toast.
severity (str): The type of toast. Can be 'success' | 'info' | 'warn' | 'error' | 'secondary' | 'contrast'
summary (str): Short explanation
sid (str, optional): The session ID of the client to send to.
If None, broadcasts to all clients. Defaults to None.
"""
if not with_comfy:
return
try:
PromptServer.instance.send_sync(
"set-audiobatch-toast", # This is our custom event name
{
'message': message,
'summary': summary,
'severity': severity
},
sid
)
except Exception as e:
logger.error(f"when trying to use ComfyUI PromptServer: {e}")
source/nodes/utils/downloader.py
-206
View File
@@ -1,206 +0,0 @@
# Copyright (c) 2025 Salvador E. Tropea
# Copyright (c) 2025 Instituto Nacional de Tecnología Industrial
# License: GPLv3
# Project: ComfyUI-AudioSeparation and ComfyUI-AudioBatch
#
# Model downloader w/TQDM and ComfyUI progress
# Original code from Gemini 2.5 Pro
import logging
import os
# Requests is better than the core Python urllib, and is a really common package
# But we don't really need it. Lets make it optional:
try:
import requests
with_requests = True
except Exception:
with_requests = False
import urllib
from tqdm import tqdm
# ComfyUI imports
try:
import comfy.utils
with_comfy = True
except Exception:
with_comfy = False
# Local imports
from .misc import NODES_NAME
logger = logging.getLogger(f"{NODES_NAME}.downloader")
def download_model_requests(url: str, save_dir: str, file_name: str):
"""
Downloads a file from a URL with progress bars for both console and ComfyUI.
Args:
url (str): The direct download URL for the file.
save_dir (str): The directory where the file will be saved.
file_name (str): The name of the file to be saved on disk.
"""
full_path = os.path.join(save_dir, file_name)
# Ensure the save directory exists
os.makedirs(save_dir, exist_ok=True)
try:
# Use a streaming request to handle large files and get content length
with requests.get(url, stream=True, timeout=10) as r:
r.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx)
# Get total file size from headers
total_size_in_bytes = int(r.headers.get('content-length', 0))
block_size = 1024 # 1 Kibibyte
# --- Setup Progress Bars ---
# Console progress bar using tqdm
progress_bar_console = tqdm(
total=total_size_in_bytes,
unit='iB',
unit_scale=True,
desc=f"Downloading {file_name}"
)
# ComfyUI progress bar
progress_bar_ui = comfy.utils.ProgressBar(total_size_in_bytes) if with_comfy else None
# --- Download Loop ---
downloaded_size = 0
with open(full_path, 'wb') as f:
for chunk in r.iter_content(chunk_size=block_size):
if chunk: # filter out keep-alive new chunks
chunk_size = len(chunk)
# Update console progress bar
progress_bar_console.update(chunk_size)
# Update ComfyUI progress bar
downloaded_size += chunk_size
if progress_bar_ui:
progress_bar_ui.update(chunk_size) # ProgressBar takes absolute value, but update is incremental
# Write chunk to file
f.write(chunk)
# --- Cleanup ---
progress_bar_console.close()
# Final check to see if download was complete
if total_size_in_bytes != 0 and progress_bar_console.n != total_size_in_bytes:
logger.error("Download failed: Size mismatch.")
# Optional: remove partial file
# os.remove(full_path)
raise IOError(f"Download failed for {file_name}. Expected {total_size_in_bytes} but got "
f"{progress_bar_console.n}")
return full_path
except requests.exceptions.RequestException as e:
logger.error(f"Network error while downloading {file_name}: {e}")
# Clean up partial file if it exists
if os.path.exists(full_path):
try:
os.remove(full_path)
except OSError:
pass
raise
except Exception as e:
logger.error(f"An error occurred during download: {e}")
if os.path.exists(full_path):
try:
os.remove(full_path)
except OSError:
pass
raise
# A simple version implemented using the Python urllib
class Downloader:
def __init__(self, model_path, model_name):
self.model_path = model_path
self.model_name = model_name
self.model_full_name = os.path.join(self.model_path, self.model_name)
# Ensure the directory for the model_path exists before __init__ if used elsewhere
# or create it at the start of download_model
# A TQDM helper class for urlretrieve reporthook
# This is a common pattern for this use case.
class TqdmUpTo(tqdm):
"""
Provides `update_to(block_num, block_size, total_size)`
and updates the TQDM bar.
"""
def __init__(self, unit, unit_scale, unit_divisor, miniters, desc):
super().__init__(unit=unit, unit_scale=unit_scale, unit_divisor=unit_divisor, miniters=miniters, desc=desc)
self.ui_bar = None
self.total = None
def update_to(self, block_num=1, block_size=1, total_size=None):
"""
block_num : int, optional
Number of blocks transferred so far [default: 1].
block_size : int, optional
Size of each block (in tqdm units) [default: 1].
total_size : int, optional
Total size (in tqdm units). If [default: None] remains unchanged.
"""
if total_size is not None and self.total is None:
self.total = total_size
# ComfyUI progress bar
if self.ui_bar is None and with_comfy:
self.ui_bar = comfy.utils.ProgressBar(total_size)
# self.update() will take the *difference* from the last call.
# So we pass the number of new blocks * block_size.
# Since block_num is cumulative, we calculate the new amount.
chunk_size = block_num * block_size - self.n
self.update(chunk_size) # self.n is current progress
if self.ui_bar:
self.ui_bar.update(chunk_size) # ProgressBar takes absolute value, but update is incremental
def download_model(self, url: str):
try:
# Ensure the directory exists
# Use or '.' for current dir if dirname is empty
os.makedirs(self.model_path or '.', exist_ok=True)
# Get filename for tqdm description
filename = self.model_name
# Use TqdmUpTo as a context manager
with self.TqdmUpTo(unit='iB', unit_scale=True, unit_divisor=1024, miniters=1,
desc=f"Downloading {filename}") as t:
# urlretrieve(url, filename=None, reporthook=None, data=None)
# reporthook is called with (block_num, block_size, total_size)
urllib.request.urlretrieve(url, self.model_full_name, reporthook=t.update_to)
# The 'with' statement ensures t.close() is called.
return filename
except urllib.error.URLError as e: # More specific exception for network issues
# Clean up partially downloaded file if an error occurs
if os.path.exists(self.model_full_name):
os.remove(self.model_full_name)
raise Exception(f"An error occurred while downloading the model (URL Error): {e.reason} from {url}")
except Exception as e:
# Clean up partially downloaded file if an error occurs
if os.path.exists(self.model_full_name):
os.remove(self.model_full_name)
raise Exception(f"An unexpected error occurred while downloading the model: {e}")
def download_model_urllib(url: str, save_dir: str, file_name: str):
return Downloader(save_dir, file_name).download_model(url)
def download_model(url: str, save_dir: str, file_name: str, force_urllib: bool = False, kind: str = "model"):
logger.info(f"Downloading {kind}: {file_name}")
logger.info(f"Source URL: {url}")
full_name = os.path.join(save_dir, file_name)
logger.info(f"Destination: {full_name}")
if with_requests and not force_urllib:
download_model_requests(url, save_dir, file_name)
else:
download_model_urllib(url, save_dir, file_name)
logger.info(f"Successfully downloaded {full_name}")
return full_name
source/nodes/utils/logger.py
-136
View File
@@ -1,136 +0,0 @@
# Copyright (c) 2025 Salvador E. Tropea
# Copyright (c) 2025 Instituto Nacional de Tecnologïa Industrial
# License: GPL-3.0
# Project: ComfyUI-AudioBatch
from __future__ import annotations # Good practice
import logging
import os
import sys
from typing import Any, Callable
from .misc import NODES_NAME, NODES_DEBUG_VAR
from .comfy_notification import send_toast_notification
# 1. Initialize variables with the `Any` type.
# This tells mypy not to make assumptions about their specific class.
Fore: Any
Back: Any
Style: Any
# 2. Perform the runtime import logic as before.
try:
from colorama import init as colorama_init, Fore, Back, Style
colorama_init()
except ImportError:
# If colorama is not available, import our fallback.
# mypy will now allow this assignment because the variables were declared as Any.
from .ansi import Fore, Back, Style
class CustomFormatter(logging.Formatter):
"""Logging Formatter to add colors"""
def __init__(self):
super(logging.Formatter, self).__init__()
white = Fore.WHITE + Style.BRIGHT
yellow = Fore.YELLOW + Style.BRIGHT
red = Fore.RED + Style.BRIGHT
red_alarm = Fore.RED + Back.WHITE + Style.BRIGHT
cyan = Fore.CYAN + Style.BRIGHT
reset = Style.RESET_ALL
# format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s "
# "(%(filename)s:%(lineno)d)"
format = f"[{NODES_NAME} %(levelname)s] %(message)s (%(name)s - %(filename)s:%(lineno)d)"
format_simple = f"[{NODES_NAME}] %(message)s"
self.FORMATS = {
logging.DEBUG: cyan + format + reset,
logging.INFO: white + format_simple + reset,
logging.WARNING: yellow + format + reset,
logging.ERROR: red + format + reset,
logging.CRITICAL: red_alarm + format + reset
}
def format(self, record):
log_fmt = self.FORMATS.get(record.levelno)
formatter = logging.Formatter(log_fmt)
return formatter.format(record)
def on_log_error_or_warning(record: logging.LogRecord) -> None:
"""
This function is called whenever a log with level WARNING or higher is emitted.
The 'record' object contains all information about the log event.
"""
if record.levelno == logging.WARNING:
summary = "Warning"
severity = "warn"
else:
summary = "Error"
severity = "error"
send_toast_notification(record.getMessage(), summary=summary, severity=severity)
class WarningAndErrorFilter(logging.Filter):
"""
A custom log filter that intercepts logs of a certain level.
"""
def __init__(self, callback: Callable, level: int = logging.WARNING):
"""
Initializes the filter.
Args:
callback: The function to call when a log record meets the level criteria.
level: The minimum level to trigger the callback.
"""
super().__init__()
self._callback = callback
self._level = level
def filter(self, record: logging.LogRecord) -> bool:
"""
This method is called for every log record.
"""
# Check if the log level is WARNING or higher
if record.levelno >= self._level:
self._callback(record)
# Always return True to ensure the log is always processed
# by the handlers after this filter.
return True
# Create a new logger
logger = logging.getLogger(NODES_NAME)
logger.propagate = False
# Add the custom filter to the logger.
logger.addFilter(WarningAndErrorFilter(callback=on_log_error_or_warning))
# Add handler if we don't have one.
if not logger.handlers:
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(CustomFormatter())
logger.addHandler(handler)
# ######################
# Logger setup
# ######################
# 1. Determine the ComfyUI global log level (influenced by --verbose)
main_logger = logger
comfy_root_logger = logging.getLogger('comfy')
effective_comfy_level = logging.getLogger().getEffectiveLevel()
# 2. Check our custom environment variable for more verbosity
try:
nodes_debug_env = int(os.environ.get(NODES_DEBUG_VAR, "0"))
except ValueError:
nodes_debug_env = 0
# 3. Set node's logger level
if nodes_debug_env:
main_logger.setLevel(logging.DEBUG - (nodes_debug_env - 1))
final_level_str = f"DEBUG (due to {NODES_DEBUG_VAR}={nodes_debug_env})"
else:
main_logger.setLevel(effective_comfy_level)
final_level_str = logging.getLevelName(effective_comfy_level) + " (matching ComfyUI global)"
_initial_setup_logger = logging.getLogger(NODES_NAME + ".setup") # A temporary logger for this message
_initial_setup_logger.debug(f"{NODES_NAME} logger level set to: {final_level_str}")
source/Makefile → src/Makefile
+1 -1
View File
@@ -67,4 +67,4 @@ format:
# Target to run all checks together
check: lint type-check test
@echo "All checks completed."
@echo "All checks completed."
source/__init__.py → src/__init__.py
View File
src/nodes/__init__.py
+9
View File
@@ -0,0 +1,9 @@
# Copyright (c) 2025 Salvador E. Tropea
# Copyright (c) 2025 Instituto Nacional de Tecnologïa Industrial
# License: GPL-3.0
# Project: ComfyUI-AudioBatch
from seconohe.logger import initialize_logger
__version__ = "1.3.0"
NODES_NAME = "AudioBatch"
main_logger = initialize_logger(NODES_NAME)
source/nodes/nodes_audio.py → src/nodes/nodes_audio.py
+5 -7
View File
@@ -10,11 +10,11 @@ import torchaudio
import torchaudio.transforms as T
from typing import Optional, Dict, Any
from .utils.aligner import AudioBatchAligner
from .utils.logger import main_logger
from seconohe.downloader import download_file
# We are the main source, so we use the main_logger
from . import main_logger
from .utils.misc import parse_time_to_seconds, parse_note_to_frequency
from .utils.downmix import spectral_downmix
from .utils.downloader import download_model
from .utils.comfy_notification import send_toast_notification
try:
from folder_paths import get_input_directory # To get the ComfyUI input directory
except ModuleNotFoundError:
@@ -33,7 +33,7 @@ DOWNMIX_OPTIONS = (["average", "standard_gain", "spectral"],
{"default": "standard_gain",
"tooltip": ("Method for stereo/multi-channel to mono conversion:\n"
"- average: Simple average ((L+R)/2). Can reduce volume.\n"
"- standard_gain: Sums channels with -3dB gain (0.707). "
"- standard_gain: Sums channels with -3 dB gain (0.707). "
"Better preserves perceived loudness."
"- spectral: Averages frequency magnitudes to prevent phase cancellation.")})
DOWNMIX_NFFT = ("INT", {
@@ -1093,14 +1093,12 @@ class AudioDownload:
if not base_url.endswith('/'):
base_url += '/'
download_url = base_url + filename
send_toast_notification(f"Downloading `{filename}`", "Download")
try:
download_model(url=download_url, save_dir=save_dir, file_name=filename, kind="audio")
download_file(logger, url=download_url, save_dir=save_dir, file_name=filename, kind="audio")
except Exception as e:
logger.error(f"Download failed for {download_url}: {e}", exc_info=True)
raise # Re-raise to stop the workflow and show the error
send_toast_notification("Finished downloading", "Download", 'success')
else:
logger.info(f"Found existing file, skipping download: '{local_filepath}'")
source/nodes/__init__.py → src/nodes/utils/__init__.py
View File
source/nodes/utils/aligner.py → src/nodes/utils/aligner.py
+1 -1
View File
@@ -8,7 +8,7 @@
import logging
import torch
import torchaudio.transforms as T
from .misc import NODES_NAME
from .. import NODES_NAME
logger = logging.getLogger(f"{NODES_NAME}.aligner")
source/nodes/utils/downmix.py → src/nodes/utils/downmix.py
+2 -2
View File
@@ -5,9 +5,9 @@
#
# Audio batch aligner
# Original code from Gemini 2.5 Pro
import torch
import logging
from .misc import NODES_NAME
import torch
from .. import NODES_NAME
logger = logging.getLogger(f"{NODES_NAME}.DownmixUtil")
source/nodes/utils/misc.py → src/nodes/utils/misc.py
-3
View File
@@ -4,9 +4,6 @@
# Project: ComfyUI-AudioBatch
import re
NODES_NAME = "AudioBatch"
NODES_DEBUG_VAR = NODES_NAME.upper() + "_NODES_DEBUG"
def parse_time_to_seconds(time_str: str) -> float:
"""
source/requirements-dev.txt → src/requirements-dev.txt
View File
source/tests/bootstrap/__init__.py → src/tests/bootstrap/__init__.py
View File
source/tests/test_audio_batch.py → src/tests/test_audio_batch.py
View File
source/tests/test_audio_blend.py → src/tests/test_audio_blend.py
View File
source/tests/test_audio_concatenate.py → src/tests/test_audio_concatenate.py
View File
source/tests/test_audio_join_split.py → src/tests/test_audio_join_split.py
View File
source/tests/test_audio_normalize.py → src/tests/test_audio_normalize.py
View File
source/tests/test_audio_note_node.py → src/tests/test_audio_note_node.py
View File
source/tests/test_audio_processing_nodes.py → src/tests/test_audio_processing_nodes.py
View File
source/tests/test_audio_test_signal_generator.py → src/tests/test_audio_test_signal_generator.py
View File
source/tests/test_downmix_util.py → src/tests/test_downmix_util.py
View File
source/tests/test_misc_utils.py → src/tests/test_misc_utils.py
View File
source/tests/test_select_audio_from_batch.py → src/tests/test_select_audio_from_batch.py
View File