558 lines
27 KiB
Python
558 lines
27 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright (c) 2025 Salvador E. Tropea
|
|
# Copyright (c) 2025 Instituto Nacional de Tecnologïa Industrial
|
|
# License: GPL-3.0
|
|
# Project: ComfyUI-AudioBatch
|
|
# From code generated by Gemini 2.5 Pro
|
|
import torch
|
|
import torchaudio.transforms as T
|
|
from .utils.logger import main_logger
|
|
from .utils.misc import parse_time_to_seconds
|
|
|
|
logger = main_logger
|
|
BASE_CATEGORY = "audio"
|
|
BATCH_CATEGORY = "batch"
|
|
CONV_CATEGORY = "conversion"
|
|
MANIPULATION_CATEGORY = "manipulation"
|
|
|
|
|
|
def convert_batch_to_stereo_tensor(audio_waveform_mono_batch: torch.Tensor) -> torch.Tensor:
|
|
""" Converts a batch of mono audio tensors (B, 1, N) to stereo (B, 2, N). """
|
|
if audio_waveform_mono_batch.ndim != 3 or audio_waveform_mono_batch.shape[1] != 1:
|
|
# This could also happen if an input was (N) or (1,N) and wasn't unsqueezed to (B,1,N) yet
|
|
raise ValueError("Input for stereo conversion must be a batch of mono audio (B, 1, N), "
|
|
f"got {audio_waveform_mono_batch.shape}")
|
|
return audio_waveform_mono_batch.repeat(1, 2, 1) # (B, 1, N) -> (B, 2, N)
|
|
|
|
|
|
class AudioBatch:
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"audio1": ("AUDIO", {"tooltip": "The first audio input. Can be a single audio item or a batch"}),
|
|
"audio2": ("AUDIO", {"tooltip": "The second audio input. Can be a single audio item or a batch"}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("AUDIO",)
|
|
RETURN_NAMES = ("audio_batch",)
|
|
FUNCTION = "batch_audio"
|
|
CATEGORY = BASE_CATEGORY + "/" + BATCH_CATEGORY
|
|
DESCRIPTION = "Audio batch creator"
|
|
UNIQUE_NAME = "SET_AudioBatch"
|
|
DISPLAY_NAME = "Batch Audios"
|
|
|
|
def _preprocess_waveform_batch(
|
|
self,
|
|
waveform: torch.Tensor, # (B, C_in, N_in)
|
|
original_sr: int,
|
|
target_sr: int,
|
|
target_channels: int,
|
|
target_samples: int,
|
|
reference_dtype: torch.dtype,
|
|
reference_device: torch.device
|
|
) -> torch.Tensor:
|
|
"""
|
|
Helper function to resample, change channels, and pad a batch of waveforms.
|
|
Returns a tensor of shape (B, target_channels, target_samples).
|
|
"""
|
|
# Ensure correct device and dtype
|
|
processed_wf = waveform.to(device=reference_device, dtype=reference_dtype)
|
|
current_batch_size, current_channels, current_samples = processed_wf.shape
|
|
|
|
# 1. Resample if necessary
|
|
if original_sr != target_sr:
|
|
logger.debug(f"Resampling batch from {original_sr} to {target_sr}. Input shape: {processed_wf.shape}")
|
|
# Resample expects (..., time)
|
|
# For (B, C, N), we can reshape to (B*C, N), resample, then reshape back.
|
|
# Or, if T.Resample handles batch dims appropriately (some versions might if C=1 or applied per channel).
|
|
# Let's reshape for robustness with T.Resample.
|
|
resampler = T.Resample(orig_freq=original_sr, new_freq=target_sr, dtype=reference_dtype).to(reference_device)
|
|
|
|
if current_channels == 1:
|
|
# Reshape (B, 1, N) to (B, N) for resampler, then unsqueeze back
|
|
processed_wf_reshaped = processed_wf.squeeze(1) # (B, N)
|
|
resampled_wf_reshaped = resampler(processed_wf_reshaped) # (B, N_new)
|
|
processed_wf = resampled_wf_reshaped.unsqueeze(1) # (B, 1, N_new)
|
|
else: # Multi-channel (e.g., stereo)
|
|
# Resample each channel in the batch separately
|
|
# This is more complex if T.Resample doesn't broadcast correctly.
|
|
# A common way: permute to (C, B, N), reshape to (C*B, N), resample, reshape back.
|
|
# Or loop (less efficient for GPU tensors).
|
|
# Let's try reshaping to (B*C, N)
|
|
original_shape = processed_wf.shape
|
|
processed_wf_flat_batch_channel = processed_wf.reshape(-1, current_samples) # (B*C, N)
|
|
resampled_wf_flat = resampler(processed_wf_flat_batch_channel) # (B*C, N_new)
|
|
# Reshape back to (B, C, N_new)
|
|
processed_wf = resampled_wf_flat.reshape(original_shape[0], original_shape[1], -1)
|
|
|
|
current_samples = processed_wf.shape[2] # Update sample count
|
|
logger.debug(f"Batch after resampling: shape={processed_wf.shape}")
|
|
|
|
# 2. Adjust number of channels
|
|
if current_channels == 1 and target_channels == 2:
|
|
processed_wf = convert_batch_to_stereo_tensor(processed_wf) # (B, 1, N) -> (B, 2, N)
|
|
logger.debug(f"Batch converted to stereo: shape={processed_wf.shape}")
|
|
elif current_channels == 2 and target_channels == 1:
|
|
# Example: Convert stereo to mono by averaging (can make this an option)
|
|
processed_wf = processed_wf.mean(dim=1, keepdim=True) # (B, 2, N) -> (B, 1, N)
|
|
logger.debug(f"Batch converted to mono (avg): shape={processed_wf.shape}")
|
|
elif current_channels != target_channels:
|
|
# Fallback or error for other unsupported channel conversions (e.g. 5.1 to stereo)
|
|
# For now, if shapes don't match after mono/stereo adjustment, it might error later.
|
|
# A more robust node would handle various conversions or error clearly.
|
|
logger.warning(f"Unhandled channel conversion from {current_channels} to {target_channels}."
|
|
f" Resulting channels: {processed_wf.shape[1]}")
|
|
# If, for example, target is 2, and current is 5, how to downmix?
|
|
# For this node's purpose (batching existing audio), major downmixing is out of scope.
|
|
# We primarily handle mono <-> stereo alignment.
|
|
if processed_wf.shape[1] != target_channels:
|
|
raise ValueError(f"Cannot align channels: input has {processed_wf.shape[1]} after initial processing, "
|
|
f"target is {target_channels}")
|
|
|
|
# 3. Pad length if necessary
|
|
if current_samples < target_samples:
|
|
padding_needed = target_samples - current_samples
|
|
# Pad only the last dimension (samples)
|
|
processed_wf = torch.nn.functional.pad(processed_wf, (0, padding_needed))
|
|
logger.debug(f"Batch padded: shape={processed_wf.shape}")
|
|
elif current_samples > target_samples: # Should not happen if target_samples is max length
|
|
logger.warning(f"Waveform has {current_samples} samples, but target is {target_samples}. "
|
|
"This indicates an issue in target_samples calculation.")
|
|
# Truncate as a fallback, though logic should prevent this.
|
|
processed_wf = processed_wf[..., :target_samples]
|
|
|
|
# Final check for shape
|
|
if processed_wf.shape[1] != target_channels or processed_wf.shape[2] != target_samples:
|
|
raise RuntimeError(f"Internal Preprocessing Error: Waveform shape {processed_wf.shape} "
|
|
f"does not match target ({current_batch_size}, {target_channels}, {target_samples}).")
|
|
|
|
return processed_wf
|
|
|
|
def batch_audio(self, audio1: dict, audio2: dict):
|
|
waveform1_orig = audio1['waveform'] # (B1, C1, N1)
|
|
sr1 = audio1['sample_rate']
|
|
waveform2_orig = audio2['waveform'] # (B2, C2, N2)
|
|
sr2 = audio2['sample_rate']
|
|
|
|
logger.debug(f"Audio1 input: shape={waveform1_orig.shape}, sr={sr1}, dtype={waveform1_orig.dtype}, "
|
|
f"device={waveform1_orig.device}")
|
|
logger.debug(f"Audio2 input: shape={waveform2_orig.shape}, sr={sr2}, dtype={waveform2_orig.dtype}, "
|
|
f"device={waveform2_orig.device}")
|
|
|
|
# Use properties of audio1 as the reference for the output batch
|
|
# (e.g., device, dtype, and target sample rate)
|
|
reference_device = waveform1_orig.device
|
|
reference_dtype = waveform1_orig.dtype
|
|
target_sr = sr1 # All audio will be resampled to sr1
|
|
|
|
# Determine target number of channels for the batch (max of inputs, ensure mono/stereo alignment)
|
|
c1 = waveform1_orig.shape[1]
|
|
c2 = waveform2_orig.shape[1]
|
|
# If one is mono and other is stereo, output will be stereo. Otherwise, max (handles both mono or both stereo).
|
|
if (c1 == 1 and c2 == 2) or (c1 == 2 and c2 == 1) or (c1 == 2 and c2 == 2):
|
|
target_channels = 2
|
|
elif c1 == 1 and c2 == 1:
|
|
target_channels = 1
|
|
else:
|
|
# For other multi-channel counts (e.g. 5.1), this simple logic might not be ideal.
|
|
# For now, default to max, but this could be an error or require downmixing node.
|
|
logger.warning(f"Complex channel counts detected (C1={c1}, C2={c2}). Defaulting to max channels ({max(c1,c2)}) "
|
|
"and hoping for downstream compatibility or further processing. "
|
|
"Explicit mono/stereo alignment is preferred for this node.")
|
|
target_channels = max(c1, c2)
|
|
|
|
# Determine target number of samples (length) for the batch
|
|
# First, calculate lengths *after* potential resampling
|
|
n1_orig = waveform1_orig.shape[2]
|
|
n2_orig = waveform2_orig.shape[2]
|
|
|
|
n1_after_resample = n1_orig # No resampling for audio1 as it's the target_sr reference
|
|
n2_after_resample = n2_orig
|
|
if sr2 != target_sr:
|
|
n2_after_resample = int(n2_orig * (target_sr / sr2))
|
|
|
|
target_samples = max(n1_after_resample, n2_after_resample)
|
|
logger.debug(f"Unified Batch Params: SR={target_sr}, Channels={target_channels}, Samples={target_samples}")
|
|
|
|
# Preprocess both waveform batches
|
|
wf1_processed = self._preprocess_waveform_batch(waveform1_orig, sr1, target_sr, target_channels, target_samples,
|
|
reference_dtype, reference_device)
|
|
wf2_processed = self._preprocess_waveform_batch(waveform2_orig, sr2, target_sr, target_channels, target_samples,
|
|
reference_dtype, reference_device)
|
|
|
|
# wf1_processed is (B1, target_channels, target_samples)
|
|
# wf2_processed is (B2, target_channels, target_samples)
|
|
|
|
# Concatenate along the batch dimension (dim=0)
|
|
try:
|
|
batched_waveform_final = torch.cat((wf1_processed, wf2_processed), dim=0)
|
|
except RuntimeError as e:
|
|
logger.error(f"Error during final torch.cat: {e}")
|
|
logger.error(f"Processed Waveform1 shape: {wf1_processed.shape}, dtype: {wf1_processed.dtype}, "
|
|
f"device: {wf1_processed.device}")
|
|
logger.error(f"Processed Waveform2 shape: {wf2_processed.shape}, dtype: {wf2_processed.dtype}, "
|
|
f"device: {wf2_processed.device}")
|
|
raise # Re-raise to make error visible
|
|
|
|
logger.info(f"Final batched audio: shape={batched_waveform_final.shape}, sr={target_sr}")
|
|
|
|
output_audio = {
|
|
"waveform": batched_waveform_final,
|
|
"sample_rate": target_sr
|
|
}
|
|
return (output_audio,)
|
|
|
|
|
|
class SelectAudioFromBatch:
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"audio_batch": ("AUDIO",), # Expects {'waveform': (B, C, N), 'sample_rate': int}
|
|
"index": ("INT", {
|
|
"default": 0,
|
|
"min": 0,
|
|
"max": 0xffffffffffffffff, # Effectively unbounded, but UI might cap
|
|
"step": 1,
|
|
"display": "number",
|
|
"tootip": "The 0-based index of the audio stream to select from the batch"
|
|
}),
|
|
"behavior_out_of_range": (["silence_original_length", "silence_fixed_length", "error"], {
|
|
"default": "silence_original_length",
|
|
"tootip": ("silence_original_length: Output silent audio with the same channel count and duration as "
|
|
"items in the original batch.\n"
|
|
"silence_fixed_length: Output silent audio with a duration specified by "
|
|
" silence_duration_seconds.\n"
|
|
"error: Raise an error (which will halt the workflow and display an error in ComfyUI)")
|
|
}),
|
|
"silence_duration_seconds": ("FLOAT", { # Only used if behavior is "silence_fixed_length"
|
|
"default": 1.0,
|
|
"min": 0.01,
|
|
"max": 3600.0, # 1 hour
|
|
"step": 0.1,
|
|
"display": "number",
|
|
"tootip": "The duration of the silent audio if `behavior_out_of_range` is set to `silence_fixed_length`"
|
|
}),
|
|
}
|
|
}
|
|
|
|
RETURN_TYPES = ("AUDIO",)
|
|
RETURN_NAMES = ("selected_audio",)
|
|
FUNCTION = "select_audio"
|
|
CATEGORY = BASE_CATEGORY + "/" + BATCH_CATEGORY
|
|
DESCRIPTION = "Selects an audio from a batch"
|
|
UNIQUE_NAME = "SET_SelectAudioFromBatch"
|
|
DISPLAY_NAME = "Select Audio from Batch"
|
|
|
|
def select_audio(self, audio_batch: dict, index: int, behavior_out_of_range: str, silence_duration_seconds: float):
|
|
|
|
waveform_batch = audio_batch['waveform'] # (B, C, N)
|
|
sample_rate = audio_batch['sample_rate']
|
|
|
|
batch_size, num_channels, num_samples_per_item = waveform_batch.shape
|
|
|
|
logger.debug(f"Input audio batch: shape={waveform_batch.shape}, sr={sample_rate}, index={index}")
|
|
logger.debug(f"Out of range behavior: {behavior_out_of_range}, silence duration: {silence_duration_seconds}s")
|
|
|
|
selected_waveform = None
|
|
|
|
if 0 <= index < batch_size:
|
|
# Valid index, select the audio
|
|
# Slicing with index:index+1 keeps the batch dimension, resulting in (1, C, N)
|
|
selected_waveform = waveform_batch[index:index+1, :, :]
|
|
logger.info(f"Selected audio at index {index}. Shape: {selected_waveform.shape}")
|
|
else:
|
|
# Index is out of range
|
|
msg = f"Index {index} is out of range for batch of size {batch_size}."
|
|
logger.warning(msg)
|
|
|
|
if behavior_out_of_range == "error":
|
|
raise ValueError(msg)
|
|
elif behavior_out_of_range == "silence_original_length":
|
|
logger.info(f"Outputting silence with original length: {num_samples_per_item} samples, "
|
|
f"{num_channels} channels.")
|
|
selected_waveform = torch.zeros((1, num_channels if num_channels > 0 else 1, num_samples_per_item),
|
|
dtype=waveform_batch.dtype, device=waveform_batch.device)
|
|
elif behavior_out_of_range == "silence_fixed_length":
|
|
silence_samples = int(silence_duration_seconds * sample_rate)
|
|
if silence_samples <= 0: # Ensure positive sample count
|
|
silence_samples = 1
|
|
logger.warning(f"Calculated silence samples is <=0 ({silence_samples} from "
|
|
f"{silence_duration_seconds}s). Using 1 sample.")
|
|
logger.info(f"Outputting silence with fixed length: {silence_samples} samples, {num_channels} channels.")
|
|
selected_waveform = torch.zeros((1, num_channels if num_channels > 0 else 1, silence_samples),
|
|
dtype=waveform_batch.dtype, device=waveform_batch.device)
|
|
|
|
if selected_waveform is None: # Should not happen if logic above is complete
|
|
logger.error("Selected waveform is None unexpectedly. Defaulting to silence.")
|
|
selected_waveform = torch.zeros((1, num_channels if num_channels > 0 else 1, 1),
|
|
dtype=waveform_batch.dtype, device=waveform_batch.device)
|
|
|
|
output_audio = {
|
|
"waveform": selected_waveform, # Shape (1, C, N_selected)
|
|
"sample_rate": sample_rate
|
|
}
|
|
|
|
return (output_audio,)
|
|
|
|
|
|
class AudioChannelConverter:
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"audio": ("AUDIO",),
|
|
"channel_conversion": (["keep", "stereo_to_mono", "mono_to_stereo", "force_mono", "force_stereo"],
|
|
{"default": "keep",
|
|
"tooltip": "keep: maintain same channels,\n"
|
|
"stereo_to_mono/force_mono: 1 channel,\n"
|
|
"mono_to_stereo/force_stereo: 2 channels"}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("AUDIO",)
|
|
RETURN_NAMES = ("audio_out",)
|
|
FUNCTION = "convert_channels"
|
|
CATEGORY = BASE_CATEGORY + "/" + CONV_CATEGORY
|
|
DESCRIPTION = "Converts audio channels (mono/stereo/multi-channel handling)."
|
|
UNIQUE_NAME = "SET_AudioChannelConverter"
|
|
DISPLAY_NAME = "Audio Channel Converter"
|
|
|
|
def convert_channels(self, audio: dict, channel_conversion: str):
|
|
waveform = audio['waveform'] # (B, C, T)
|
|
sample_rate = audio['sample_rate']
|
|
|
|
original_batch_size, original_channels, original_samples = waveform.shape
|
|
logger.debug(f"Input audio: {original_batch_size}B, {original_channels}C, {original_samples}T @ {sample_rate}Hz")
|
|
logger.debug(f"Channel conversion mode: {channel_conversion}")
|
|
|
|
output_waveform = waveform.clone() # Start with a copy
|
|
|
|
if channel_conversion == "keep":
|
|
# No change needed, but log if > 2 channels
|
|
if original_channels > 2:
|
|
logger.warning(f"Channel mode 'keep': Input has {original_channels} channels. Outputting all "
|
|
f"{original_channels} channels.")
|
|
# output_waveform remains as is
|
|
|
|
elif channel_conversion == "stereo_to_mono" or channel_conversion == "force_mono":
|
|
if original_channels == 1:
|
|
logger.debug("Input is already mono. No change for stereo_to_mono/force_mono.")
|
|
# output_waveform remains as is
|
|
else: # Stereo or Multi-channel to Mono
|
|
if original_channels > 2 and channel_conversion == "stereo_to_mono":
|
|
logger.warning(f"Channel mode 'stereo_to_mono': Input has {original_channels} channels. "
|
|
"Averaging all to mono.")
|
|
elif channel_conversion == "force_mono":
|
|
logger.info(f"Channel mode 'force_mono': Input has {original_channels} channels. Averaging all to mono.")
|
|
# Average across the channel dimension (dim=1)
|
|
output_waveform = torch.mean(waveform, dim=1, keepdim=True)
|
|
logger.debug(f"Converted to mono. New shape: {output_waveform.shape}")
|
|
|
|
elif channel_conversion == "mono_to_stereo" or channel_conversion == "force_stereo":
|
|
if original_channels == 2:
|
|
logger.debug("Input is already stereo. No change for mono_to_stereo/force_stereo.")
|
|
# output_waveform remains as is
|
|
elif original_channels == 1: # Mono to Stereo
|
|
# Duplicate the mono channel
|
|
output_waveform = waveform.repeat(1, 2, 1)
|
|
logger.debug(f"Converted mono to stereo by duplication. New shape: {output_waveform.shape}")
|
|
else: # Multi-channel (>2) to Stereo
|
|
if channel_conversion == "mono_to_stereo":
|
|
logger.warning(f"Channel mode 'mono_to_stereo': Input has {original_channels} channels. "
|
|
"Taking the first channel and duplicating it to create stereo.")
|
|
elif channel_conversion == "force_stereo":
|
|
logger.info(f"Channel mode 'force_stereo': Input has {original_channels} channels. "
|
|
"Taking the first channel and duplicating it to create stereo.")
|
|
output_waveform = waveform[:, 0:1, :].repeat(1, 2, 1) # Take first channel, make it (B,1,T), then repeat
|
|
logger.debug(f"Converted {original_channels}ch to stereo. New shape: {output_waveform.shape}")
|
|
else:
|
|
logger.error(f"Unknown channel_conversion mode: {channel_conversion}. Returning original.")
|
|
# output_waveform remains as is (original waveform)
|
|
|
|
processed_audio_dict = {"waveform": output_waveform, "sample_rate": sample_rate}
|
|
return (processed_audio_dict,)
|
|
|
|
|
|
class AudioResampler:
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"audio": ("AUDIO",),
|
|
"target_sample_rate": ("INT", {"default": 0, "min": 0, "max": 192000, "step": 100}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("AUDIO",)
|
|
RETURN_NAMES = ("audio_out",)
|
|
FUNCTION = "resample_audio"
|
|
CATEGORY = BASE_CATEGORY + "/" + CONV_CATEGORY
|
|
DESCRIPTION = "Resamples audio to a target sample rate using torchaudio."
|
|
UNIQUE_NAME = "SET_AudioResampler"
|
|
DISPLAY_NAME = "Audio Resampler"
|
|
|
|
def resample_audio(self, audio: dict, target_sample_rate: int):
|
|
waveform = audio['waveform'] # (B, C, T)
|
|
original_sample_rate = audio['sample_rate']
|
|
|
|
logger.debug(f"Input audio SR: {original_sample_rate}Hz, Target SR: {target_sample_rate}Hz")
|
|
|
|
if target_sample_rate == 0 or target_sample_rate == original_sample_rate:
|
|
logger.info(f"Target sample rate ({target_sample_rate}Hz) is 0 or matches original ({original_sample_rate}Hz). "
|
|
"Skipping resampling.")
|
|
return (audio,) # Return original audio dict
|
|
|
|
try:
|
|
# Ensure waveform is on the CPU for torchaudio transforms if it might be on GPU
|
|
# Though many torchaudio ops support GPU tensors. Resample does.
|
|
# For consistency or if issues arise:
|
|
# device = waveform.device
|
|
# waveform_cpu = waveform.cpu()
|
|
# resampler = T.Resample(orig_freq=original_sample_rate, new_freq=target_sample_rate).to(device)
|
|
# resampled_waveform = resampler(waveform)
|
|
resampler = T.Resample(orig_freq=original_sample_rate, new_freq=target_sample_rate,
|
|
dtype=waveform.dtype # Preserve dtype
|
|
).to(waveform.device) # Perform resampling on the tensor's current device
|
|
|
|
resampled_waveform = resampler(waveform)
|
|
logger.info(f"Resampled audio from {original_sample_rate}Hz to {target_sample_rate}Hz. "
|
|
f"Original shape: {waveform.shape}, New shape: {resampled_waveform.shape}")
|
|
processed_audio_dict = {"waveform": resampled_waveform, "sample_rate": target_sample_rate}
|
|
return (processed_audio_dict,)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error during resampling: {e}", exc_info=True)
|
|
# Fallback: return original audio if resampling fails
|
|
return (audio,)
|
|
|
|
|
|
class AudioProcessAdvanced:
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"audio": ("AUDIO",),
|
|
"channel_conversion": (["keep", "stereo_to_mono", "mono_to_stereo", "force_mono", "force_stereo"],
|
|
{"default": "keep",
|
|
"tooltip": "keep: maintain same channels,\n"
|
|
"stereo_to_mono/force_mono: 1 channel,\n"
|
|
"mono_to_stereo/force_stereo: 2 channels"}),
|
|
"target_sample_rate": ("INT", {"default": 0, "min": 0, "max": 192000, "step": 100.,
|
|
"tooltip": "Output sample rate"}),
|
|
},
|
|
}
|
|
RETURN_TYPES = ("AUDIO",)
|
|
RETURN_NAMES = ("audio_out",)
|
|
FUNCTION = "process_audio"
|
|
CATEGORY = BASE_CATEGORY + "/" + CONV_CATEGORY
|
|
DESCRIPTION = "Applies channel conversion and resampling to audio."
|
|
UNIQUE_NAME = "SET_AudioChannelConvResampler"
|
|
DISPLAY_NAME = "Audio Channel Conv and Resampler"
|
|
|
|
def process_audio(self, audio: dict, channel_conversion: str, target_sample_rate: int):
|
|
# Instantiate helper classes (or move their logic directly here)
|
|
channel_converter_node = AudioChannelConverter()
|
|
resampler_node = AudioResampler()
|
|
|
|
# 1. Channel Conversion
|
|
(audio_after_channels,) = channel_converter_node.convert_channels(audio, channel_conversion)
|
|
|
|
# 2. Resampling
|
|
(audio_after_resample,) = resampler_node.resample_audio(audio_after_channels, target_sample_rate)
|
|
|
|
return (audio_after_resample,)
|
|
|
|
|
|
class AudioInfo:
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"audio": ("AUDIO",),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("AUDIO", "INT", "INT", "INT", "INT")
|
|
RETURN_NAMES = ("audio_bypass", "batch_size", "channels", "num_samples", "sample_rate")
|
|
FUNCTION = "show_info"
|
|
CATEGORY = BASE_CATEGORY + "/" + CONV_CATEGORY
|
|
DESCRIPTION = "Shows information about the audio."
|
|
UNIQUE_NAME = "SET_AudioInfo"
|
|
DISPLAY_NAME = "Audio Information"
|
|
|
|
def show_info(self, audio: dict):
|
|
wav = audio['waveform'] # (B, C, T)
|
|
sample_rate = audio['sample_rate']
|
|
return audio, wav.shape[0], wav.shape[1], wav.shape[2], sample_rate
|
|
|
|
|
|
class AudioForceChannels:
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"audio": ("AUDIO",),
|
|
"channels": ("INT", {"default": 0, "min": 0, "max": 2}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("AUDIO",)
|
|
RETURN_NAMES = ("audio",)
|
|
FUNCTION = "force_channels"
|
|
CATEGORY = BASE_CATEGORY + "/" + CONV_CATEGORY
|
|
DESCRIPTION = "Forces the number of channels. 0 means keep same."
|
|
UNIQUE_NAME = "SET_AudioForceChannels"
|
|
DISPLAY_NAME = "Audio Force Channels"
|
|
CHANNELS_TO_MODE = ["keep", "force_mono", "force_stereo"]
|
|
|
|
def force_channels(self, audio: dict, channels: int):
|
|
return AudioChannelConverter().convert_channels(audio, self.CHANNELS_TO_MODE[channels])
|
|
|
|
|
|
class AudioCut:
|
|
""" From https://github.com/christian-byrne/audio-separation-nodes-comfyui/ """
|
|
@classmethod
|
|
def INPUT_TYPES(cls):
|
|
return {
|
|
"required": {
|
|
"audio": ("AUDIO",),
|
|
"start_time": ("STRING", {
|
|
"default": "0:00",
|
|
"tooltip": "Start time in HH:MM:SS.ss format, or just a float."
|
|
}),
|
|
"end_time": ("STRING", {
|
|
"default": "1:00",
|
|
"tooltip": "End time in HH:MM:SS.ss format, or just a float."
|
|
}),
|
|
},
|
|
}
|
|
|
|
RETURN_TYPES = ("AUDIO",)
|
|
RETURN_NAMES = ("audio_out",)
|
|
FUNCTION = "cut"
|
|
CATEGORY = BASE_CATEGORY + "/" + MANIPULATION_CATEGORY
|
|
DESCRIPTION = "Cuts a portion of the input audio. Can use seconds or HH:MM:SS.ss."
|
|
UNIQUE_NAME = "SET_AudioCut"
|
|
DISPLAY_NAME = "Audio Cut"
|
|
|
|
def cut(self, audio: dict, start_time: str, end_time: str):
|
|
waveform = audio["waveform"]
|
|
sample_rate = audio["sample_rate"]
|
|
length = waveform.shape[-1] - 1
|
|
|
|
start_frame = int(parse_time_to_seconds(start_time) * sample_rate)
|
|
start_frame = max(0, min(start_frame, length))
|
|
|
|
end_frame = int(parse_time_to_seconds(end_time) * sample_rate)
|
|
end_frame = max(0, min(end_frame, length))
|
|
|
|
logger.debug(f"Cutting audio from {start_frame} to {end_frame} (SR: {sample_rate})")
|
|
if start_frame > end_frame:
|
|
raise ValueError("Audio Cut: Start time must be smaller than end time and both be within the audio length.")
|
|
|
|
return ({"waveform": waveform[..., start_frame:end_frame],
|
|
"sample_rate": sample_rate},)
|