nodes/c0ffymachyne-ComfyUI_SignalProcessing
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

added more nodes

Browse Source
This commit is contained in:
c0ffymachyne
2024-11-20 22:27:33 -08:00
parent 80069bf10c
commit 448f868b67
16 changed files with 1799 additions and 451 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+2
View File
@@ -1,5 +1,7 @@
# ComfyUI Signal Processing
## THIS IS WORK IN PROGRESS REPOSITORY
This repo contains 4 nodes for [ComfyUI](https://github.com/comfyanonymous/ComfyUI) that allows for audio filtering
## Nodes
SignalProcessingFilter.py
+91
View File
@@ -0,0 +1,91 @@
import torch, torchaudio
import torch
from typing import Dict
class SignalProcessingFilter:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"audio": ("AUDIO", {"forceInput": True}),
"cutoff": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.01}),
"filter_type": (["lowpass", "highpass", "bandpass", "bandstop"], {"default": "lowpass"}),
"q_factor": ("FLOAT", {"default": 0.707, "min": 0.1, "max": 5.0, "step": 0.01}), # For resonance/bandwidth
}
}
RETURN_TYPES = ("AUDIO","INT")
RETURN_NAMES = ("audio","sample_rate")
CATEGORY = "Signal Processing"
FUNCTION = "apply_filter"
def apply_filter(self, audio: Dict[str, torch.Tensor], cutoff: float, filter_type: str, q_factor: float):
"""
Apply a specified filter to the input audio.
Parameters:
audio (Dict[str, torch.Tensor]): Input audio with 'waveform' and 'sample_rate'.
cutoff (float): Normalized cutoff frequency (0.0 to 1.0).
filter_type (str): Type of filter ('lowpass', 'highpass', 'bandpass', 'bandstop').
q_factor (float): Quality factor determining the filter's bandwidth.
Returns:
Tuple[Dict[str, torch.Tensor]]: Filtered audio.
"""
waveform = audio['waveform'].squeeze(0)
sample_rate = audio['sample_rate']
nyquist = sample_rate / 2.0
# Define minimum and maximum frequencies for mapping
log_min = 20.0 # 20 Hz, typical lower bound of human hearing
log_max = nyquist - 100.0 # Slightly below Nyquist to prevent instability
# Avoid log(0) by ensuring cutoff is within (0,1)
cutoff = min(max(cutoff, 1e-6), 1.0 - 1e-6)
# Logarithmic mapping
log_min = torch.log(torch.tensor(log_min))
log_max = torch.log(torch.tensor(log_max))
log_cutoff = log_min + cutoff * (log_max - log_min)
cutoff_freq = torch.exp(log_cutoff).item()
# Debug: Print mapped cutoff frequency
print(f"Normalized cutoff: {cutoff} mapped to frequency: {cutoff_freq:.2f} Hz")
# Choose filter type
if filter_type == "lowpass":
filtered_waveform = torchaudio.functional.lowpass_biquad(
waveform, sample_rate, cutoff_freq, Q=q_factor
)
elif filter_type == "highpass":
filtered_waveform = torchaudio.functional.highpass_biquad(
waveform, sample_rate, cutoff_freq, Q=q_factor
)
elif filter_type in ["bandpass", "bandstop"]:
center_freq = cutoff_freq
# Ensure that the bandwidth does not exceed the Nyquist frequency
bandwidth = center_freq / q_factor
lower_freq = max(center_freq - bandwidth / 2.0, 20.0) # Prevent dropping below 20 Hz
upper_freq = min(center_freq + bandwidth / 2.0, nyquist - 100.0) # Prevent exceeding Nyquist
# Debug: Print bandwidth details
print(f"Band filter with center frequency: {center_freq:.2f} Hz, "
f"lower: {lower_freq:.2f} Hz, upper: {upper_freq:.2f} Hz")
if filter_type == "bandpass":
filtered_waveform = torchaudio.functional.bandpass_biquad(
waveform, sample_rate, center_freq, Q=q_factor
)
else: # bandstop
filtered_waveform = torchaudio.functional.band_biquad(
waveform, sample_rate, center_freq, Q=q_factor
)
else:
raise ValueError(f"Unsupported filter type: {filter_type}")
filtered_waveform = filtered_waveform.unsqueeze(0)
print('SignalProcessingFilter.filtered_waveform',filtered_waveform.shape)
return {'waveform': filtered_waveform, 'sample_rate': sample_rate}, sample_rate
SignalProcessingLoadAudio.py
+56
View File
@@ -0,0 +1,56 @@
import sys,os
import torch, torchaudio
import hashlib
sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), "comfy"))
import folder_paths
class SignalProcessingLoadAudio():
supported_formats = ['wav','mp3','ogg','m4a','flac','mp4']
@classmethod
def INPUT_TYPES(s):
supported_extensions = tuple(f".{fmt.lower()}" for fmt in SignalProcessingLoadAudio.supported_formats)
input_dir = folder_paths.get_input_directory()
all_items = os.listdir(input_dir)
filtered_files = [
x for x in all_items
if x.lower().endswith(supported_extensions)
]
files = [os.path.join(input_dir,x) for x in filtered_files]
return {
"required": {"audio_file": (sorted(files), {"image_upload": True})},
"optional" : {"seek_seconds": ("FLOAT", {"default": 0, "min": 0})}
}
RETURN_TYPES = ("AUDIO","INT")
RETURN_NAMES = ("audio","sample_rate")
CATEGORY = "Signal Processing"
FUNCTION = "process"
def process(self, audio_file, seek_seconds):
audio_file_path = folder_paths.get_annotated_filepath(audio_file)
waveform, sample_rate = torchaudio.load(audio_file_path)
waveform = waveform.unsqueeze(0).contiguous()
print('SignalProcessingLoadAudio.waveform',waveform.shape)
return {'waveform': waveform, 'sample_rate': sample_rate} , sample_rate
@classmethod
def IS_CHANGED(s, audio_file, seek_seconds):
audio_file_path = folder_paths.get_annotated_filepath(audio_file)
m = hashlib.sha256()
with open(audio_file_path, 'rb') as f:
m.update(f.read())
return m.digest().hex()
@classmethod
def VALIDATE_INPUTS(s, audio_file, seek_seconds):
if not folder_paths.exists_annotated_filepath(audio_file):
return "Invalid image file: {}".format(audio_file)
return True
SignalProcessingMixdown.py
+128
View File
@@ -0,0 +1,128 @@
import torch
import os, sys
from typing import Tuple, List, Dict, Union
import torchaudio
class SignalProcessingMixdown:
@classmethod
def INPUT_TYPES(cls) -> Dict:
return {
"required": {
"audio_inputs": ("AUDIO_LIST", {"default": []}), # List of audio inputs
"output_normalization": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 1.0, "step": 0.1}),
},
"optional": {
"gain_factors": ("FLOAT_LIST", {"default": [], "min": 0.0, "max": 2.0, "step": 0.1}),
# If empty, default to [1.0] * num_audios
}
}
RETURN_TYPES = ("AUDIO", "INT")
RETURN_NAMES = ("mixed_audio", "sample_rate")
CATEGORY = "Audio Processing"
FUNCTION = "process"
def process(
self,
audio_inputs: List[Dict[str, Union[torch.Tensor, int]]],
output_normalization: float,
gain_factors: List[float] = [],
) -> Tuple[Dict[str, torch.Tensor], int]:
"""
Mix down multiple audio inputs into a single audio output with optional individual volume controls.
Parameters:
audio_inputs (List[Dict]): List of audio inputs, each containing 'waveform' and 'sample_rate'.
output_normalization (float): Normalization factor for the mixed audio (0.0 to 1.0).
gain_factors (List[float], optional): List of gain factors for each audio input.
Returns:
Tuple[Dict[str, torch.Tensor], int]: Mixed audio with waveform and sample rate.
"""
if not audio_inputs:
raise ValueError("No audio inputs provided for mixing.")
num_audios = len(audio_inputs)
print(f"SignalProcessingMixdown: Number of audio inputs: {num_audios}")
# Handle gain_factors
if not gain_factors:
gain_factors = [1.0] * num_audios
print("SignalProcessingMixdown: No gain_factors provided. Defaulting to 1.0 for all inputs.")
elif len(gain_factors) != num_audios:
raise ValueError(f"Number of gain factors ({len(gain_factors)}) does not match number of audio inputs ({num_audios}).")
else:
print(f"SignalProcessingMixdown: Gain factors: {gain_factors}")
# Extract sample rates and verify consistency
sample_rates = [audio['sample_rate'] for audio in audio_inputs]
target_sample_rate = sample_rates[0]
for idx, sr in enumerate(sample_rates):
if sr != target_sample_rate:
print(f"SignalProcessingMixdown: Resampling audio {idx} from {sr} Hz to {target_sample_rate} Hz")
resampler = torchaudio.transforms.Resample(orig_freq=sr, new_freq=target_sample_rate)
audio_inputs[idx]['waveform'] = resampler(audio_inputs[idx]['waveform'])
audio_inputs[idx]['sample_rate'] = target_sample_rate
# Determine the maximum length among all audio inputs
lengths = [audio['waveform'].shape[-1] for audio in audio_inputs]
max_length = max(lengths)
print(f"SignalProcessingMixdown: Maximum waveform length: {max_length} samples")
# Pad or truncate each audio to match the maximum length and apply gain
for idx, audio in enumerate(audio_inputs):
waveform = audio['waveform']
current_length = waveform.shape[-1]
gain = gain_factors[idx]
print(f"SignalProcessingMixdown: Processing audio {idx} with gain factor {gain}")
if current_length < max_length:
padding = max_length - current_length
# Pad with zeros (silence) at the end
waveform = torch.nn.functional.pad(waveform, (0, padding))
print(f"SignalProcessingMixdown: Padded audio {idx} with {padding} zeros")
elif current_length > max_length:
# Truncate the waveform to max_length
waveform = waveform[:, :, :max_length]
print(f"SignalProcessingMixdown: Truncated audio {idx} to {max_length} samples")
# Apply gain
waveform = waveform * gain
print(f"SignalProcessingMixdown: Applied gain to audio {idx}")
audio['waveform'] = waveform
# Sum all waveforms to create the mix
mixed_waveform = torch.zeros_like(audio_inputs[0]['waveform'])
for idx, audio in enumerate(audio_inputs):
mixed_waveform += audio['waveform']
print(f"SignalProcessingMixdown: Summed audio {idx} into mixed waveform")
# Normalize the mixed audio based on the maximum absolute value
max_val = torch.max(torch.abs(mixed_waveform))
if max_val < 1e-5:
max_val = 1e-5 # Prevent division by zero
print("SignalProcessingMixdown: Maximum value too low, setting to 1e-5 to prevent division by zero")
# Apply output normalization as a multiplier
mixed_waveform = (mixed_waveform / max_val) * output_normalization
print(f"SignalProcessingMixdown: Applied output normalization factor {output_normalization}")
# Clamp to [-1.0, 1.0] to ensure no clipping occurs
mixed_waveform = torch.clamp(mixed_waveform, -1.0, 1.0)
print("SignalProcessingMixdown: Clamped mixed waveform to [-1.0, 1.0]")
# Ensure the tensor is of type float32
mixed_waveform = mixed_waveform.float()
# Ensure the tensor is on CPU and contiguous
mixed_waveform = mixed_waveform.to('cpu').contiguous()
print(f"SignalProcessingMixdown: Mixed waveform shape: {mixed_waveform.shape}")
# Return the mixed audio and sample rate
return ({'waveform': mixed_waveform, 'sample_rate': target_sample_rate}, target_sample_rate)
SignalProcessingMultiBandEQ.py
+284
View File
@@ -0,0 +1,284 @@
import torch
import torchaudio
def normalize_waveform(waveform):
"""
Normalize the waveform to ensure its maximum absolute value is 1.0.
Parameters:
waveform (torch.Tensor): Input waveform tensor with shape [batch, channels, samples].
Returns:
torch.Tensor: Normalized waveform tensor with the same shape.
"""
max_val = torch.max(torch.abs(waveform))
if max_val > 1.0:
waveform = waveform / max_val
return waveform
class SignalProcessingMultibandEQ:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"audio_input": ("AUDIO",),
"method" : (['hann','rfft'],)
},
"optional": {
"sub_bass_gain_db": ("FLOAT", {"default": 0.0, "min": -12.0, "max": 12.0, "step": 0.1}),
"bass_gain_db": ("FLOAT", {"default": 0.0, "min": -12.0, "max": 12.0, "step": 0.1}),
"low_mid_gain_db": ("FLOAT", {"default": 0.0, "min": -12.0, "max": 12.0, "step": 0.1}),
"mid_gain_db": ("FLOAT", {"default": 0.0, "min": -12.0, "max": 12.0, "step": 0.1}),
"upper_mid_gain_db": ("FLOAT", {"default": 0.0, "min": -12.0, "max": 12.0, "step": 0.1}),
"presence_gain_db": ("FLOAT", {"default": 0.0, "min": -12.0, "max": 12.0, "step": 0.1}),
"brilliance_gain_db": ("FLOAT", {"default": 0.0, "min": -12.0, "max": 12.0, "step": 0.1}),
}
}
RETURN_TYPES = ("AUDIO",)
RETURN_NAMES = ("equalized_audio",)
CATEGORY = "Audio Processing"
FUNCTION = "process"
def process(
self,
audio_input,
method,
sub_bass_gain_db=0.0,
bass_gain_db=0.0,
low_mid_gain_db=0.0,
mid_gain_db=0.0,
upper_mid_gain_db=0.0,
presence_gain_db=0.0,
brilliance_gain_db=0.0
):
if method == 'rfft':
return self.process_rfft(audio_input,sub_bass_gain_db,bass_gain_db,low_mid_gain_db,mid_gain_db,upper_mid_gain_db,presence_gain_db,brilliance_gain_db)
if method == 'hann':
return self.process_hann(audio_input,sub_bass_gain_db,bass_gain_db,low_mid_gain_db,mid_gain_db,upper_mid_gain_db,presence_gain_db,brilliance_gain_db)
def process_rfft(
self,
audio_input,
sub_bass_gain_db=0.0,
bass_gain_db=0.0,
low_mid_gain_db=0.0,
mid_gain_db=0.0,
upper_mid_gain_db=0.0,
presence_gain_db=0.0,
brilliance_gain_db=0.0
):
"""
Apply multiband equalization to the input audio.
Parameters:
audio_input (Dict): Dictionary containing 'waveform' and 'sample_rate'.
sub_bass_gain_db (float): Gain for the Sub-Bass band in decibels.
bass_gain_db (float): Gain for the Bass band in decibels.
low_mid_gain_db (float): Gain for the Low-Mid band in decibels.
mid_gain_db (float): Gain for the Mid band in decibels.
upper_mid_gain_db (float): Gain for the Upper-Mid band in decibels.
presence_gain_db (float): Gain for the Presence band in decibels.
brilliance_gain_db (float): Gain for the Brilliance band in decibels.
Returns:
Tuple[Dict[str, torch.Tensor]]: Dictionary with equalized 'waveform' and 'sample_rate'.
"""
waveform = audio_input.get('waveform') # [batch, channels, samples]
sample_rate = audio_input.get('sample_rate')
if waveform is None or sample_rate is None:
raise ValueError("Input audio must contain 'waveform' and 'sample_rate'.")
if not isinstance(waveform, torch.Tensor):
raise TypeError("Waveform must be a torch.Tensor.")
if waveform.ndim != 3:
raise ValueError("Waveform must be a 3D tensor with shape (batch, channels, samples).")
batch_size, channels, num_samples = waveform.shape
if channels not in [1, 2]:
raise ValueError(f"Unsupported number of channels: {channels}. Only mono and stereo are supported.")
# Convert gains from dB to linear scale
sub_bass_gain = 10 ** (sub_bass_gain_db / 20)
bass_gain = 10 ** (bass_gain_db / 20)
low_mid_gain = 10 ** (low_mid_gain_db / 20)
mid_gain = 10 ** (mid_gain_db / 20)
upper_mid_gain = 10 ** (upper_mid_gain_db / 20)
presence_gain = 10 ** (presence_gain_db / 20)
brilliance_gain = 10 ** (brilliance_gain_db / 20)
# Perform FFT
waveform_fft = torch.fft.rfft(waveform, dim=-1) # [batch, channels, freq_bins]
# Frequency bins
freq_bins = torch.linspace(0, sample_rate / 2, steps=waveform_fft.size(-1), device=waveform.device)
# Define frequency bands
bands = {
'sub_bass': (20, 60),
'bass': (60, 250),
'low_mid': (250, 500),
'mid': (500, 2000),
'upper_mid': (2000, 4000),
'presence': (4000, 6000),
'brilliance': (6000, 20000)
}
# Initialize gain factors tensor
gain_factors = torch.ones_like(waveform_fft) # [batch, channels, freq_bins]
# Apply gains to each band
for band, (f_start, f_end) in bands.items():
if band == 'sub_bass':
gain = sub_bass_gain
elif band == 'bass':
gain = bass_gain
elif band == 'low_mid':
gain = low_mid_gain
elif band == 'mid':
gain = mid_gain
elif band == 'upper_mid':
gain = upper_mid_gain
elif band == 'presence':
gain = presence_gain
elif band == 'brilliance':
gain = brilliance_gain
else:
gain = 1.0 # Default gain
# Create mask for the current band
mask = (freq_bins >= f_start) & (freq_bins < f_end)
gain_factors[:, :, mask] *= gain
# Apply gain factors
waveform_fft_eq = waveform_fft * gain_factors # [batch, channels, freq_bins]
# Perform inverse FFT
waveform_eq = torch.fft.irfft(waveform_fft_eq, n=num_samples, dim=-1) # [batch, channels, samples]
# Normalize the waveform
waveform_eq = normalize_waveform(waveform_eq) # [batch, channels, samples]
# Prepare the output dictionary
equalized_audio = {
'waveform': waveform_eq,
'sample_rate': sample_rate
}
return (equalized_audio, )
def process_hann(
self,
audio_input,
sub_bass_gain_db=0.0,
bass_gain_db=0.0,
low_mid_gain_db=0.0,
mid_gain_db=0.0,
upper_mid_gain_db=0.0,
presence_gain_db=0.0,
brilliance_gain_db=0.0
):
waveform = audio_input.get('waveform') # [batch, channels, samples]
sample_rate = audio_input.get('sample_rate')
if waveform is None or sample_rate is None:
raise ValueError("Input audio must contain 'waveform' and 'sample_rate'.")
if not isinstance(waveform, torch.Tensor):
raise TypeError("Waveform must be a torch.Tensor.")
if waveform.ndim != 3:
raise ValueError("Waveform must be a 3D tensor with shape (batch, channels, samples).")
batch_size, channels, num_samples = waveform.shape
if channels not in [1, 2]:
raise ValueError(f"Unsupported number of channels: {channels}. Only mono and stereo are supported.")
# Limit maximum gain to prevent artifacts
max_gain_db = 12.0
gains_db = {
'sub_bass': min(sub_bass_gain_db, max_gain_db),
'bass': min(bass_gain_db, max_gain_db),
'low_mid': min(low_mid_gain_db, max_gain_db),
'mid': min(mid_gain_db, max_gain_db),
'upper_mid': min(upper_mid_gain_db, max_gain_db),
'presence': min(presence_gain_db, max_gain_db),
'brilliance': min(brilliance_gain_db, max_gain_db)
}
gains_linear = {band: 10 ** (gain_db / 20) for band, gain_db in gains_db.items()}
# FFT parameters
n_fft = 2048
hop_length = n_fft // 2
window = torch.hann_window(n_fft, device=waveform.device, periodic=True)
# Perform STFT
waveform_stft = torch.stft(
waveform.view(-1, num_samples),
n_fft=n_fft,
hop_length=hop_length,
window=window,
return_complex=True,
center=True,
pad_mode='reflect'
) # [batch*channels, freq_bins, time_frames]
# Frequency bins
freq_bins = torch.linspace(0, sample_rate / 2, steps=waveform_stft.size(1), device=waveform.device)
# Initialize gain factors tensor
gain_factors = torch.ones_like(waveform_stft, device=waveform.device)
# Define frequency bands
bands = {
'sub_bass': (20, 60),
'bass': (60, 250),
'low_mid': (250, 500),
'mid': (500, 2000),
'upper_mid': (2000, 4000),
'presence': (4000, 6000),
'brilliance': (6000, sample_rate / 2)
}
# Apply gains to each band with smoother transitions
for band, (f_start, f_end) in bands.items():
gain = gains_linear[band]
# Smooth transition using Gaussian profile
center_freq = (f_start + f_end) / 2
bandwidth = (f_end - f_start) / 2
gain_profile = torch.exp(-0.5 * ((freq_bins - center_freq) / (bandwidth / 2)) ** 2)
gain_profile = 1 + (gain - 1) * gain_profile
gain_factors *= gain_profile.unsqueeze(1)
# Apply gain factors
waveform_stft_eq = waveform_stft * gain_factors
# Perform inverse STFT
waveform_eq = torch.istft(
waveform_stft_eq,
n_fft=n_fft,
hop_length=hop_length,
window=window,
length=num_samples,
center=True
)
# Reshape back to original dimensions
waveform_eq = waveform_eq.view(batch_size, channels, num_samples)
# Normalize the waveform
waveform_eq = normalize_waveform(waveform_eq)
# Prepare the output dictionary
equalized_audio = {
'waveform': waveform_eq,
'sample_rate': sample_rate
}
return (equalized_audio,)
SignalProcessingMultiBandEQExperimental.py
+191
View File
@@ -0,0 +1,191 @@
import torch
import torchaudio
def normalize_waveform(waveform):
max_val = torch.max(torch.abs(waveform))
if max_val > 1.0:
waveform = waveform / max_val
return waveform
class SignalProcessingMultiBandEQExperimental:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"audio_input": ("AUDIO",),
},
"optional": {
"sub_bass_gain_db": ("FLOAT", {"default": 0.0, "min": -24.0, "max": 24.0, "step": 0.1}),
"bass_gain_db": ("FLOAT", {"default": 0.0, "min": -24.0, "max": 24.0, "step": 0.1}),
"low_mid_gain_db": ("FLOAT", {"default": 0.0, "min": -24.0, "max": 24.0, "step": 0.1}),
"mid_gain_db": ("FLOAT", {"default": 0.0, "min": -24.0, "max": 24.0, "step": 0.1}),
"upper_mid_gain_db": ("FLOAT", {"default": 0.0, "min": -24.0, "max": 24.0, "step": 0.1}),
"presence_gain_db": ("FLOAT", {"default": 0.0, "min": -24.0, "max": 24.0, "step": 0.1}),
"brilliance_gain_db": ("FLOAT", {"default": 0.0, "min": -24.0, "max": 24.0, "step": 0.1}),
}
}
RETURN_TYPES = ("AUDIO",)
RETURN_NAMES = ("equalized_audio",)
CATEGORY = "Audio Processing"
FUNCTION = "process"
def process(
self,
audio_input,
sub_bass_gain_db=0.0,
bass_gain_db=0.0,
low_mid_gain_db=0.0,
mid_gain_db=0.0,
upper_mid_gain_db=0.0,
presence_gain_db=0.0,
brilliance_gain_db=0.0
):
waveform = audio_input.get('waveform') # [batch, channels, samples]
sample_rate = audio_input.get('sample_rate')
if waveform is None or sample_rate is None:
raise ValueError("Input audio must contain 'waveform' and 'sample_rate'.")
if not isinstance(waveform, torch.Tensor):
raise TypeError("Waveform must be a torch.Tensor.")
if waveform.ndim != 3:
raise ValueError("Waveform must be a 3D tensor with shape (batch, channels, samples).")
batch_size, channels, num_samples = waveform.shape
if channels not in [1, 2]:
raise ValueError(f"Unsupported number of channels: {channels}. Only mono and stereo are supported.")
# Limit gain to prevent artifacts
max_gain_db = 24.0
min_gain_db = -24.0
gains_db = {
'sub_bass': max(min(sub_bass_gain_db, max_gain_db), min_gain_db),
'bass': max(min(bass_gain_db, max_gain_db), min_gain_db),
'low_mid': max(min(low_mid_gain_db, max_gain_db), min_gain_db),
'mid': max(min(mid_gain_db, max_gain_db), min_gain_db),
'upper_mid': max(min(upper_mid_gain_db, max_gain_db), min_gain_db),
'presence': max(min(presence_gain_db, max_gain_db), min_gain_db),
'brilliance': max(min(brilliance_gain_db, max_gain_db), min_gain_db),
}
# Convert gains from dB to linear scale
gains_linear = {band: 10 ** (gain_db / 20) for band, gain_db in gains_db.items()}
# STFT parameters
n_fft = 2048
hop_length = n_fft // 2
window = torch.hann_window(n_fft, device=waveform.device, periodic=True)
# Perform STFT
waveform_stft = torch.stft(
waveform.view(-1, num_samples),
n_fft=n_fft,
hop_length=hop_length,
window=window,
return_complex=True,
center=True,
pad_mode='reflect'
) # [batch*channels, freq_bins, time_frames]
# Frequency bins
freq_bins = torch.linspace(0, sample_rate / 2, steps=waveform_stft.size(1), device=waveform.device)
# Initialize gain factors tensor
gain_factors = torch.ones_like(waveform_stft)
# Define frequency bands
bands = {
'sub_bass': (20, 60),
'bass': (60, 250),
'low_mid': (250, 500),
'mid': (500, 2000),
'upper_mid': (2000, 4000),
'presence': (4000, 6000),
'brilliance': (6000, sample_rate / 2)
}
# Apply gains to each band with smoother transitions
for band, (f_start, f_end) in bands.items():
gain = gains_linear[band]
center_freq = (f_start + f_end) / 2
bandwidth = (f_end - f_start)
if bandwidth == 0:
bandwidth = f_end * 0.01
gain_profile = 1 + (gain - 1) * torch.exp(-0.5 * ((freq_bins - center_freq) / (bandwidth / 2)) ** 2)
gain_profile = gain_profile.unsqueeze(0).unsqueeze(0)
gain_factors = gain_factors * gain_profile
# Apply gain factors
waveform_stft_eq = waveform_stft * gain_factors
# Bass enhancement using compression
# Define compressor parameters
threshold = -20.0 # in dB
ratio = 4.0 # compression ratio
knee = 5.0 # in dB
makeup_gain = 5.0 # in dB
# Convert to linear scale
threshold_linear = 10 ** (threshold / 20)
makeup_gain_linear = 10 ** (makeup_gain / 20)
# Apply compression to sub-bass and bass bands
bass_mask = torch.zeros_like(freq_bins)
bass_indices = ((freq_bins >= bands['sub_bass'][0]) & (freq_bins <= bands['bass'][1]))
bass_mask[bass_indices] = 1.0
bass_mask = bass_mask.unsqueeze(0).unsqueeze(0)
# Get magnitude and phase
magnitude = torch.abs(waveform_stft_eq)
phase = torch.angle(waveform_stft_eq)
# Compress magnitude in bass frequencies
magnitude_compressed = magnitude.clone()
bass_magnitude = magnitude * bass_mask
bass_magnitude_db = 20 * torch.log10(bass_magnitude + 1e-8)
# Apply soft knee compression
over_threshold = bass_magnitude_db - threshold
compression_amount = torch.zeros_like(over_threshold)
compression_amount[over_threshold > -knee/2] = (1 / ratio - 1) * (over_threshold[over_threshold > -knee/2] + knee/2) ** 2 / (2 * knee)
compression_amount[over_threshold > knee/2] = (1 / ratio - 1) * over_threshold[over_threshold > knee/2]
gain_reduction_db = compression_amount
gain_reduction_linear = 10 ** (gain_reduction_db / 20)
magnitude_compressed = magnitude_compressed * (1 + gain_reduction_linear * bass_mask)
# Apply makeup gain
magnitude_compressed = magnitude_compressed * makeup_gain_linear
# Reconstruct the compressed STFT
waveform_stft_eq = magnitude_compressed * torch.exp(1j * phase)
# Perform inverse STFT
waveform_eq = torch.istft(
waveform_stft_eq,
n_fft=n_fft,
hop_length=hop_length,
window=window,
length=num_samples,
center=True
)
# Reshape back to original dimensions
waveform_eq = waveform_eq.view(batch_size, channels, num_samples)
# Normalize the waveform
waveform_eq = normalize_waveform(waveform_eq)
# Prepare the output dictionary
equalized_audio = {
'waveform': waveform_eq,
'sample_rate': sample_rate
}
return (equalized_audio, )
SignalProcessingPadSynth.py
+142
View File
@@ -0,0 +1,142 @@
import torch
import math, json
from typing import Tuple, List, Dict
class SignalProcessingPadSynth:
@classmethod
def INPUT_TYPES(cls) -> Dict:
return {
"required": {
"samplerate": ("INT", {"default": 44100, "min": 8000, "max": 96000, "step": 1}),
"fundamental_freq": ("FLOAT", {"default": 261.0, "min": 20.0, "max": 2000.0, "step": 1.0}),
"bandwidth_cents": ("FLOAT", {"default": 40.0, "min": 10.0, "max": 100.0, "step": 1.0}),
"number_harmonics": ("INT", {"default": 64, "min": 1, "max": 128, "step": 1}),
"amplitude_per_harmonic": ("STRING", {"default": "", "multiline": False, "hint": "Provide a JSON array of amplitudes per harmonic, e.g., [0.0, 1.0, 0.5, ...] or leave empty for default."})
}
}
RETURN_TYPES = ("AUDIO","INT")
RETURN_NAMES = ("audio","sample_rate")
CATEGORY = "Signal Processing"
FUNCTION = "process"
def process(
self,
samplerate: int,
fundamental_freq: float,
bandwidth_cents: float,
number_harmonics: int,
amplitude_per_harmonic: str,
) -> Tuple[Dict[str, torch.Tensor]]:
"""
Apply PADsynth algorithm to generate audio.
Parameters:
samplerate (int): Sampling rate in Hz.
fundamental_freq (float): Fundamental frequency in Hz.
bandwidth_cents (float): Bandwidth in cents for Gaussian profile.
number_harmonics (int): Number of harmonics to generate.
amplitude_per_harmonic (str): JSON-formatted string specifying amplitudes per harmonic. Leave empty for defaults.
Returns:
Tuple[Dict[str, torch.Tensor]]: Generated audio with waveform and sample rate.
"""
# Define FFT size
N = 262144 # As per C++ code
# Initialize amplitude array A
if not amplitude_per_harmonic.strip():
# Use default amplitude distribution
A = torch.zeros(number_harmonics, dtype=torch.double)
A[0] = 0.0 # A[0] is not used
for i in range(1, number_harmonics):
A[i] = 1.0 / i
if (i % 2) == 0:
A[i] *= 2.0
else:
# Parse JSON string to list
try:
amplitude_list = json.loads(amplitude_per_harmonic)
if not isinstance(amplitude_list, list):
raise ValueError("amplitude_per_harmonic must be a JSON array.")
if len(amplitude_list) != number_harmonics:
raise ValueError("Length of amplitude_per_harmonic must match number_harmonics.")
# Convert to torch tensor
A = torch.tensor(amplitude_list, dtype=torch.double)
except json.JSONDecodeError as e:
raise ValueError("amplitude_per_harmonic must be a valid JSON array.") from e
except Exception as e:
raise
# Initialize frequency amplitude and phase arrays
freq_amp = torch.zeros(N // 2, dtype=torch.double)
freq_phase = torch.rand(N // 2, dtype=torch.double) * 2.0 * math.pi # Random phases between 0 and 2pi
# Define Gaussian profile function
def profile(fi: torch.Tensor, bwi: torch.Tensor) -> torch.Tensor:
x = fi / bwi
x_sq = x ** 2
# Avoid computing exp(-x^2) for x_sq > 14.71280603
mask = x_sq <= 14.71280603
result = torch.zeros_like(x_sq)
result[mask] = torch.exp(-x_sq[mask]) / bwi[mask]
return result
# Convert bandwidth from cents to Hz
# bw_Hz = (2^(bw/1200) -1) * f * nh
# Convert bandwidth_cents to multiplier
bw_multiplier = 2.0 ** (bandwidth_cents / 1200.0) - 1.0
# Populate frequency amplitude array
for nh in range(1, number_harmonics):
f_nh = fundamental_freq * nh
bw_Hz = bw_multiplier * f_nh
bwi = bw_Hz / (2.0 * samplerate)
fi = f_nh / samplerate # Normalized frequency
# Create tensors for frequency bins
i = torch.arange(N // 2, dtype=torch.double)
# Normalized frequency for each bin
normalized_freq = i / N # Equivalent to i * (samplerate / N) / samplerate = i / N
# Compute profile
fi_tensor = torch.full_like(i, fi)
bwi_tensor = torch.full_like(i, bwi)
profile_values = profile(normalized_freq - fi_tensor, bwi_tensor)
# Update frequency amplitude
freq_amp += profile_values * A[nh]
# Construct complex frequency domain tensor
real = freq_amp * torch.cos(freq_phase)
imag = freq_amp * torch.sin(freq_phase)
freq_complex = torch.complex(real, imag) # Shape: (N//2,)
# Perform IFFT using torch.fft.irfft
smp = torch.fft.irfft(freq_complex, n=N) # Shape: (N,)
# Normalize the signal to prevent clipping
max_val = torch.max(torch.abs(smp))
if max_val < 1e-5:
max_val = 1e-5 # Prevent division by zero
smp = smp / (max_val * math.sqrt(2)) # Normalize to 1/sqrt(2) as in C++ code
# Convert to float32 for saving
smp = smp.float()
# Prepare waveform tensor: (C, N)
waveform_out = smp.unsqueeze(0) # Mono audio
# Reshape waveform_out to include batch dimension: (1, C, N)
waveform_out = waveform_out.unsqueeze(0) # Shape: (1, C, N)
#audios = []
#print('SignalProcessingPadSynth.waveform_out',waveform_out.shape)
#audios.append({'waveform': waveform_out, 'sample_rate': samplerate})
# Return the synthesized audio
return {'waveform': waveform_out, 'sample_rate': samplerate}, samplerate
SignalProcessingPadSynthChoir.py
+141
View File
@@ -0,0 +1,141 @@
import torch
import math
from typing import Tuple, List, Dict
class SignalProcessingPadSynthChoir:
@classmethod
def INPUT_TYPES(cls) -> Dict:
return {
"required": {
"samplerate": ("INT", {"default": 44100, "min": 8000, "max": 96000, "step": 1}),
"base_freq": ("FLOAT", {"default": 130.81, "min": 20.0, "max": 2000.0, "step": 1.0}),
"step_size": ("INT", {"default": 4, "min": 1, "max": 24, "step": 1}),
"num_notes": ("INT", {"default": 7, "min": 1, "max": 24, "step": 1}),
"bandwidth_cents": ("FLOAT", {"default": 60.0, "min": 10.0, "max": 100.0, "step": 1.0}),
"number_harmonics": ("INT", {"default": 64, "min": 1, "max": 128, "step": 1})
}
}
RETURN_TYPES = ("AUDIO_LIST", "INT")
RETURN_NAMES = ("audios", "sample_rate")
CATEGORY = "Signal Processing"
FUNCTION = "process"
def process(
self,
samplerate: int,
base_freq: float,
step_size: int,
num_notes: int,
bandwidth_cents: float,
number_harmonics: int,
) -> Tuple[List[Dict[str, torch.Tensor]]]:
"""
Apply PADsynth choir algorithm to generate multiple audio files.
Parameters:
samplerate (int): Sampling rate in Hz.
base_freq (float): Base frequency in Hz.
step_size (int): Step size in semitones between notes.
num_notes (int): Number of notes to generate.
bandwidth_cents (float): Bandwidth in cents for Gaussian profile.
number_harmonics (int): Number of harmonics to generate.
Returns:
Tuple[List[Dict[str, torch.Tensor]]]: List of generated audios with waveform and sample rate.
"""
# Define FFT size
N = 262144 # As per C++ code
audios = []
for note_index in range(num_notes):
note_semitones = step_size * note_index
f1 = base_freq * (2.0 ** (note_semitones / 12.0))
# Compute amplitude_per_harmonic with formants
A = torch.zeros(number_harmonics, dtype=torch.double)
A[0] = 0.0 # A[0] is not used
for i in range(1, number_harmonics):
# Calculate formants based on the C++ choir implementation
formants = (
math.exp(-((i * f1 - 600.0) / 150.0) ** 2) +
math.exp(-((i * f1 - 900.0) / 250.0) ** 2) +
math.exp(-((i * f1 - 2200.0) / 200.0) ** 2) +
math.exp(-((i * f1 - 2600.0) / 250.0) ** 2) +
math.exp(-((i * f1) / 3000.0) ** 2) * 0.1
)
A[i] = (1.0 / i) * formants
# Optionally, you can debug amplitude values
# logger.debug(f"Harmonic {i}: A[{i}]={A[i]:.4f}")
# Initialize frequency amplitude and phase arrays
freq_amp = torch.zeros(N // 2, dtype=torch.double)
freq_phase = torch.rand(N // 2, dtype=torch.double) * 2.0 * math.pi # Random phases between 0 and 2pi
# Define Gaussian profile function
def profile(fi: torch.Tensor, bwi: torch.Tensor) -> torch.Tensor:
x = fi / bwi
x_sq = x ** 2
# Avoid computing exp(-x^2) for x_sq > 14.71280603
mask = x_sq <= 14.71280603
result = torch.zeros_like(x_sq)
result[mask] = torch.exp(-x_sq[mask]) / bwi[mask]
return result
# Convert bandwidth from cents to Hz
# bw_Hz = (2^(bw/1200) -1) * f * nh
bw_multiplier = 2.0 ** (bandwidth_cents / 1200.0) - 1.0
# Create tensors for frequency bins
i = torch.arange(N // 2, dtype=torch.double)
normalized_freq = i / N # Equivalent to i / N
# Compute and accumulate frequency amplitudes for each harmonic
for nh in range(1, number_harmonics):
f_nh = f1 * nh
bw_Hz = bw_multiplier * f_nh
bwi = bw_Hz / (2.0 * samplerate)
fi = f_nh / samplerate # Normalized frequency
fi_tensor = torch.full_like(i, fi)
bwi_tensor = torch.full_like(i, bwi)
profile_values = profile(normalized_freq - fi_tensor, bwi_tensor)
# Update frequency amplitude
freq_amp += profile_values * A[nh]
# Construct complex frequency domain tensor
real = freq_amp * torch.cos(freq_phase)
imag = freq_amp * torch.sin(freq_phase)
freq_complex = torch.complex(real, imag) # Shape: (N//2,)
# Perform IFFT using torch.fft.irfft
smp = torch.fft.irfft(freq_complex, n=N) # Shape: (N,)
# Normalize the signal to prevent clipping
max_val = torch.max(torch.abs(smp))
if max_val < 1e-5:
max_val = 1e-5 # Prevent division by zero
smp = smp / (max_val * math.sqrt(2)) # Normalize to 1/sqrt(2) as in C++ code
# Convert to float32 for saving
smp = smp.float()
# Prepare waveform tensor: (C, N)
waveform_out = smp.unsqueeze(0) # Mono audio
# Reshape waveform_out to include batch dimension: (1, C, N)
waveform_out = waveform_out.unsqueeze(0) # Shape: (1, C, N)
print('SignalProcessingPadSynthChoir.waveform_out',waveform_out.shape)
# Append to audios list
audios.append({'waveform':waveform_out,'sample_rate': samplerate})
# Return the list of generated audios
return audios, samplerate
SignalProcessingPadSynthChoir2.py
+201
View File
@@ -0,0 +1,201 @@
import torch
import math
from typing import Tuple, List, Dict
class SignalProcessingPadSynthChoir2:
@classmethod
def INPUT_TYPES(cls) -> Dict:
return {
"required": {
"samplerate": ("INT", {"default": 44100, "min": 8000, "max": 96000, "step": 1}),
"base_freq": ("FLOAT", {"default": 130.81, "min": 20.0, "max": 2000.0, "step": 1.0}),
"step_size": ("INT", {"default": 4, "min": 1, "max": 24, "step": 1}),
"num_notes": ("INT", {"default": 7, "min": 1, "max": 24, "step": 1}),
"bandwidth_cents": ("FLOAT", {"default": 60.0, "min": 10.0, "max": 100.0, "step": 1.0}),
"number_harmonics": ("INT", {"default": 64, "min": 1, "max": 128, "step": 1}),
}
}
RETURN_TYPES = ("AUDIO_LIST","INT")
RETURN_NAMES = ("audios","sample_rate")
CATEGORY = "Signal Processing"
FUNCTION = "process"
def process(
self,
samplerate: int,
base_freq: float,
step_size: int,
num_notes: int,
bandwidth_cents: float,
number_harmonics: int,
) -> Tuple[List[Dict[str, torch.Tensor]]]:
"""
Apply PADsynth choir algorithm to generate multiple audio outputs with aliasing mitigation.
Parameters:
samplerate (int): Sampling rate in Hz.
base_freq (float): Base frequency in Hz.
step_size (int): Step size in semitones between notes.
num_notes (int): Number of notes to generate.
bandwidth_cents (float): Bandwidth in cents for Gaussian profile shaping.
number_harmonics (int): Number of harmonics per note.
Returns:
Tuple[List[Dict[str, torch.Tensor]]]: List containing a single dictionary with batched audios.
"""
audios = []
nyquist = samplerate / 2.0
N = 262144 # FFT size for high frequency resolution
for note_index in range(num_notes):
# Calculate the fundamental frequency for the current note
note_semitones = step_size * note_index
f1 = base_freq * (2.0 ** (note_semitones / 12.0))
# Compute amplitude_per_harmonic with formants
A = torch.zeros(number_harmonics, dtype=torch.double)
A[0] = 0.0 # A[0] is not used
# Calculate maximum allowed harmonics to prevent aliasing
max_harmonics = int(math.floor(nyquist / f1))
effective_harmonics = min(number_harmonics, max_harmonics)
for i in range(1, effective_harmonics):
# Calculate formants based on the C++ choir implementation
formants = (
math.exp(-((i * f1 - 600.0) / 150.0) ** 2) +
math.exp(-((i * f1 - 900.0) / 250.0) ** 2) +
math.exp(-((i * f1 - 2200.0) / 200.0) ** 2) +
math.exp(-((i * f1 - 2600.0) / 250.0) ** 2) +
math.exp(-((i * f1) / 3000.0) ** 2) * 0.1
)
A[i] = (1.0 / i) * formants
# Initialize frequency amplitude and phase arrays
freq_amp = torch.zeros(N // 2, dtype=torch.double)
freq_phase = torch.rand(N // 2, dtype=torch.double) * 2.0 * math.pi # Random phases between 0 and 2pi
# Define Gaussian profile function
def profile(fi: torch.Tensor, bwi: torch.Tensor) -> torch.Tensor:
x = fi / bwi
x_sq = x ** 2
# Avoid computing exp(-x^2) for x_sq > 14.71280603
mask = x_sq <= 14.71280603
result = torch.zeros_like(x_sq)
result[mask] = torch.exp(-x_sq[mask]) / bwi[mask]
return result
# Convert bandwidth from cents to Hz
bw_multiplier = 2.0 ** (bandwidth_cents / 1200.0) - 1.0
# Create tensors for frequency bins
i = torch.arange(N // 2, dtype=torch.double)
normalized_freq = i / N # Equivalent to i / N
# Compute and accumulate frequency amplitudes for each harmonic
for nh in range(1, effective_harmonics):
f_nh = f1 * nh
bw_Hz = bw_multiplier * f_nh
bwi = bw_Hz / (2.0 * samplerate)
fi = f_nh / samplerate # Normalized frequency
fi_tensor = torch.full_like(i, fi)
bwi_tensor = torch.full_like(i, bwi)
profile_values = profile(normalized_freq - fi_tensor, bwi_tensor)
# Update frequency amplitude
freq_amp += profile_values * A[nh]
# Construct complex frequency domain tensor
real = freq_amp * torch.cos(freq_phase)
imag = freq_amp * torch.sin(freq_phase)
freq_complex = torch.complex(real, imag) # Shape: (N//2,)
# Perform IFFT using torch.fft.irfft
smp = torch.fft.irfft(freq_complex, n=N) # Shape: (N,)
# Normalize the signal to prevent clipping
max_val = torch.max(torch.abs(smp))
if max_val < 1e-5:
max_val = 1e-5 # Prevent division by zero
smp = smp / (max_val * math.sqrt(2)) # Normalize to 1/sqrt(2) as in C++ code
# Convert to float32 for further processing
smp = smp.float()
# Prepare waveform tensor: (C, N)
waveform_out = smp.unsqueeze(0) # Mono audio: (1, N)
# Implement a custom low-pass FIR filter using windowed sinc
def lowpass_fir(waveform: torch.Tensor, samplerate: int, cutoff_freq: float, num_taps: int = 101) -> torch.Tensor:
"""
Apply a low-pass FIR filter to the waveform.
Parameters:
waveform (torch.Tensor): Input waveform tensor of shape (C, N).
samplerate (int): Sampling rate in Hz.
cutoff_freq (float): Cutoff frequency in Hz.
num_taps (int): Number of filter taps (must be odd for symmetric FIR).
Returns:
torch.Tensor: Filtered waveform tensor of shape (C, N).
"""
nyquist = samplerate / 2.0
normalized_cutoff = cutoff_freq / nyquist
# Create sinc filter
taps = torch.sinc(2 * normalized_cutoff * (torch.arange(num_taps, dtype=torch.float32) - (num_taps - 1) / 2.0))
# Apply window (Hamming)
window = torch.hamming_window(num_taps)
taps = taps * window
# Normalize to get unity gain
taps = taps / taps.sum()
# Reshape taps for convolution: (C, 1, num_taps)
taps = taps.unsqueeze(0).unsqueeze(0)
# Pad the waveform to handle borders
pad_width = num_taps // 2
waveform_padded = torch.nn.functional.pad(waveform, (pad_width, pad_width), mode='reflect')
# Apply convolution
filtered_waveform = torch.nn.functional.conv1d(waveform_padded, taps)
return filtered_waveform
# Determine cutoff frequency with a margin to prevent aliasing
highest_harmonic_freq = f1 * effective_harmonics
cutoff_freq = min(highest_harmonic_freq * 1.1, nyquist * 0.95) # 10% margin
cutoff_freq = max(cutoff_freq, 2000.0) # Ensure a minimum cutoff frequency
# Apply custom low-pass FIR filter
waveform_out = lowpass_fir(waveform_out, samplerate, cutoff_freq)
# Further normalize after filtering
max_val_filtered = torch.max(torch.abs(waveform_out))
if max_val_filtered < 1e-5:
max_val_filtered = 1e-5 # Prevent division by zero
waveform_out = waveform_out / max_val_filtered
# Clamp to [-1.0, 1.0] to ensure no clipping occurs
waveform_out = torch.clamp(waveform_out, -1.0, 1.0)
# Move waveform to CPU and ensure it's contiguous
waveform_out = waveform_out.cpu().contiguous()
stereo_waveform = waveform_out.unsqueeze(1).repeat(1, 2, 1)
#print('SignalProcessingPadSynthChoir2.waveform_out',stereo_waveform.shape)
# Append to audios list
audios.append({"waveform":stereo_waveform,"sample_rate": samplerate})
# Stack audios into a batched Tensor: (batch, channels, samples)
#waveforms = torch.stack(audios, dim=0) # Shape: (num_notes, 1, N)
# Return as a list containing a single dictionary
return audios, samplerate
SignalProcessingPaulStretch.py
+172
View File
@@ -0,0 +1,172 @@
import torch
import math
from typing import Tuple, List, Dict
import numpy as np
#https://github.com/paulnasca/paulstretch_python
#https://github.com/paulnasca/paulstretch_python/blob/master/paulstretch_stereo.py
class SignalProcessingPaulStretch:
@classmethod
def INPUT_TYPES(cls) -> Dict:
return {
"required": {
"audio": ("AUDIO", {"forceInput": True}),
"stretch_factor": ("FLOAT", {"default": 8.0, "min": 1.0, "max": 100.0, "step": 0.1}),
"window_size_seconds": ("FLOAT", {"default": 0.25, "min": 0.05, "max": 10.0, "step": 0.05}),
}
}
RETURN_TYPES = ("AUDIO","INT")
RETURN_NAMES = ("audio","sample_rate")
CATEGORY = "Signal Processing"
FUNCTION = "process"
def process(
self,
audio: Dict[str, torch.Tensor],
stretch_factor: float,
window_size_seconds: float
) -> Tuple[Dict[str, torch.Tensor]]:
"""
Apply the PaulStretch algorithm to time-stretch the input audio without altering its pitch.
Parameters:
audio (Dict[str, torch.Tensor]): Input audio with 'waveform' and 'sample_rate'.
stretch_factor (float): Factor by which to stretch the audio (1.0 = no stretch).
window_size_seconds (float): Window size in seconds.
Returns:
Tuple[Dict[str, torch.Tensor]]: PaulStretched audio with original sample rate.
"""
# Conditional processing: If stretch_factor is 1.0, return original audio
if stretch_factor == 1.0:
return ({'waveform': audio['waveform'], 'sample_rate': audio['sample_rate']},)
# Squeeze the first dimension (e.g., batch dimension) to get shape (C, N) or (N,)
waveform = audio['waveform'].squeeze(0).cpu().numpy() # Shape: (C, N) or (N,) for mono
sample_rate = audio['sample_rate']
# Handle mono audio by converting to stereo by duplicating the channel
if waveform.ndim == 1:
waveform = np.tile(waveform, (2, 1)) # Shape: (2, N)
nchannels, nsamples = waveform.shape
# Optimize window size to be divisible by 2, 3, and 5
window_size = int(window_size_seconds * sample_rate)
if window_size < 16:
window_size = 16
window_size = self.optimize_windowsize(window_size)
window_size = int(window_size / 2) * 2 # Ensure even window size
half_window_size = int(window_size / 2)
# Correct the end of the waveform by applying a fade-out
end_size = int(sample_rate * 0.05)
if end_size < 16:
end_size = 16
waveform[:, -end_size:] *= np.linspace(1, 0, end_size)
# Compute displacement inside the input file
start_pos = 0.0
displace_pos = (window_size * 0.5) / stretch_factor
# Create custom window function as in original code
window = np.power(1.0 - np.power(np.linspace(-1.0, 1.0, window_size), 2.0), 1.25)
# Initialize old windowed buffer
old_windowed_buf = np.zeros((nchannels, window_size))
# Initialize list to store output frames
output_frames = []
# Processing loop
frame_count = 0
while True:
# Get the windowed buffer
istart_pos = int(math.floor(start_pos))
buf = waveform[:, istart_pos:istart_pos + window_size]
if buf.shape[1] < window_size:
buf = np.pad(buf, ((0, 0), (0, window_size - buf.shape[1])), 'constant')
buf = buf * window
# Get the amplitudes of the frequency components and discard the phases
freqs = np.abs(np.fft.rfft(buf, axis=1))
# Randomize the phases by multiplication with a random complex number with modulus=1
ph = np.random.uniform(0, 2 * np.pi, (nchannels, freqs.shape[1])) * 1j
freqs = freqs * np.exp(ph)
# Do the inverse FFT
buf_ifft = np.fft.irfft(freqs, n=window_size, axis=1)
# Window again the output buffer
buf_ifft *= window
# Overlap-add the output
output = buf_ifft[:, 0:half_window_size] + old_windowed_buf[:, half_window_size:window_size]
old_windowed_buf = buf_ifft
# Clamp the values to -1..1
output = np.clip(output, -1.0, 1.0)
# Append to output_frames
output_frames.append(output)
# Increment start_pos
start_pos += displace_pos
frame_count += 1
# Check if we have reached the end of the input
if start_pos >= nsamples:
break
# Optional: Log progress every 10%
num_frames = max(int(math.ceil((nsamples - window_size) / (window_size * 0.5)) + 1), 1)
if frame_count % max(int(num_frames / 10), 1) == 0:
progress = (start_pos / nsamples) * 100
# Concatenate all output frames horizontally
output_array = np.hstack(output_frames)
# Clamp final output to ensure no clipping
output_array = np.clip(output_array, -1.0, 1.0)
# Convert back to torch tensor
output_tensor = torch.from_numpy(output_array).float()
# Unsqueeze to add the first dimension back (e.g., batch dimension)
output_tensor = output_tensor.unsqueeze(0) # Shape: (1, C, N_stretched)
print('SignalProcessingPaulStretch.waveform_out',output_tensor.shape)
return {'waveform': output_tensor, 'sample_rate': sample_rate}, sample_rate
@staticmethod
def optimize_windowsize(n: int) -> int:
"""
Optimize the window size to be divisible by 2, 3, and 5.
Parameters:
n (int): Initial window size.
Returns:
int: Optimized window size.
"""
orig_n = n
while True:
n = orig_n
while (n % 2) == 0:
n /= 2
while (n % 3) == 0:
n /= 3
while (n % 5) == 0:
n /= 5
if n < 2:
break
orig_n += 1
return orig_n
SignalProcessingSpectrogram.py
+104
View File
@@ -0,0 +1,104 @@
import torch
from typing import List, Dict
from PIL import Image
import numpy as np
import torchaudio
import matplotlib.pyplot as plt
class SignalProcessingSpectrogram:
@classmethod
def INPUT_TYPES(cls):
cmaps = ['viridis', 'plasma', 'inferno', 'magma', 'cividis']
return {
"required": {
"audio_input": ("AUDIO",),
"color_map": (cmaps,),
},
"optional": {
"n_fft": ("INT", {"default": 4096, "min": 512, "max": 8192, "step": 256}),
"hop_length": ("INT", {"default": 128, "min": 64, "max": 4096, "step": 128}),
"n_mels": ("INT", {"default": 512, "min": 32, "max": 2048, "step": 32}),
"top_db": ("FLOAT", {"default": 80.0, "min": 10.0, "max": 100.0, "step": 5.0}),
}
}
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("spectrogram_image",)
CATEGORY = "Audio Processing"
FUNCTION = "process"
def process(self, audio_input, color_map, n_fft=2048, hop_length=512, n_mels=128, top_db=80.0):
waveform = audio_input.get('waveform') # [channels, samples] or [batch, channels, samples]
sample_rate = audio_input.get('sample_rate')
# Convert to mono by averaging channels
if waveform.ndim == 3:
# [batch, channels, samples]
waveform = waveform.mean(dim=1, keepdim=True) # [batch, 1, samples]
waveform = waveform.squeeze(0) # [1, samples]
elif waveform.ndim == 2:
# [channels, samples]
if waveform.shape[0] > 1:
waveform = waveform.mean(dim=0, keepdim=True) # [1, samples]
else:
waveform = waveform.unsqueeze(0) # [1, samples]
elif waveform.ndim == 1:
# [samples]
waveform = waveform.unsqueeze(0) # [1, samples]
else:
raise ValueError(f"Unsupported waveform shape: {waveform.shape}")
# Generate Mel Spectrogram
spectrogram_transform = torchaudio.transforms.MelSpectrogram(
sample_rate=sample_rate,
n_fft=n_fft,
hop_length=hop_length,
n_mels=n_mels,
power=2.0,
norm='slaney',
mel_scale='htk',
)
spectrogram = spectrogram_transform(waveform) # [1, n_mels, time_frames]
# Convert to decibel scale
amplitude_to_db = torchaudio.transforms.AmplitudeToDB(top_db=top_db)
spectrogram_db = amplitude_to_db(spectrogram) # [1, n_mels, time_frames]
# Convert to numpy
spectrogram_db = spectrogram_db.squeeze().detach().cpu().numpy() # [n_mels, time_frames]
# Clip spectrogram to a range for better contrast
spectrogram_db = np.clip(spectrogram_db, -top_db, 0.0)
# Normalize spectrogram to [0,1]
spectrogram_normalized = (spectrogram_db + top_db) / top_db # [0,1]
# Apply a colormap (e.g., 'inferno') using matplotlib
cmap = plt.get_cmap(color_map)
spectrogram_colored = cmap(spectrogram_normalized) # [n_mels, time_frames, 4] RGBA
# Convert to RGB by removing alpha channel
spectrogram_rgb = (spectrogram_colored[:, :, :3] * 255).astype(np.uint8) # [n_mels, time_frames, 3]
# Create PIL Image
spectrogram_image = Image.fromarray(spectrogram_rgb).convert("RGB")
# Optionally resize for better resolution
spectrogram_image = spectrogram_image.resize(
(spectrogram_image.width * 2, spectrogram_image.height * 2),
Image.BILINEAR
)
# Convert to numpy array and normalize to [0,1]
image_np = np.array(spectrogram_image).astype(np.float32) / 255.0 # [H, W, 3]
# Convert to torch tensor and add batch dimension
#image_tensor = torch.from_numpy(image_np).permute(2, 0, 1).unsqueeze(0) # [1, 3, H, W]
image = torch.from_numpy(image_np)[None,]
return (image, )
SignalProcessingStereoWidening.py
+99
View File
@@ -0,0 +1,99 @@
import torch
#https://github.com/AudioEmerge/Widener
class SignalProcessingStereoWidening:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"audio_input": ("AUDIO",),
"gain": ("FLOAT", {"default": 2.0, "min": 0, "max": 5.0, "step": 0.1}),
},
"optional": {
"width": ("FLOAT", {"default": 2.0, "min": 1.0, "max": 8.0, "step": 0.1}),
}
}
RETURN_TYPES = ("AUDIO",)
RETURN_NAMES = ("widened_audio",)
CATEGORY = "Audio Processing"
FUNCTION = "process"
def process(self, audio_input, gain, width=1.2):
"""
Widen stereo audio or convert mono audio to wide stereo using the provided widening algorithm.
Parameters:
audio_input (Dict): Dictionary containing 'waveform' and 'sample_rate'.
width (float): Width factor (>1.0).
Returns:
Tuple[Dict[str, torch.Tensor]]: Dictionary with widened 'waveform' and 'sample_rate'.
"""
waveform = audio_input.get('waveform') # [batch, channels, samples]
sample_rate = audio_input.get('sample_rate')
if waveform is None or sample_rate is None:
raise ValueError("Input audio must contain 'waveform' and 'sample_rate'.")
if not isinstance(waveform, torch.Tensor):
raise TypeError("Waveform must be a torch.Tensor.")
if waveform.ndim != 3:
raise ValueError("Waveform must be a 3D tensor with shape (batch, channels, samples).")
batch_size, channels, num_samples = waveform.shape
if channels not in [1, 2]:
raise ValueError(f"Unsupported number of channels: {channels}. Only mono and stereo are supported.")
# Calculate coefficients based on the provided width parameter
tmp = 1.0 / max(1.0 + width, 2.0) # Scalar
coef_M = 1.0 * tmp # Coefficient for mid
coef_S = width * tmp # Coefficient for sides
if channels == 2:
# Stereo to Widened Stereo
L = waveform[:, 0, :] # Left channel [batch, samples]
R = waveform[:, 1, :] # Right channel [batch, samples]
# Apply the widening algorithm
mid = (L + R) * coef_M # Mid signal
sides = (R - L) * coef_S # Side signal
widened_L = mid - sides # New Left channel
widened_R = mid + sides # New Right channel
# Stack the widened channels back into a stereo waveform
widened_waveform = torch.stack((widened_L, widened_R), dim=1) # [batch, 2, samples]
elif channels == 1:
# Mono to Wide Stereo
L = waveform[:, 0, :].clone() # Duplicate mono channel to Left
R = waveform[:, 0, :].clone() # Duplicate mono channel to Right
# Apply the widening algorithm
mid = (L + R) * coef_M # Mid signal
sides = (R - L) * coef_S # Side signal
widened_L = mid - sides # New Left channel
widened_R = mid + sides # New Right channel
# Stack the widened channels into a stereo waveform
widened_waveform = torch.stack((widened_L, widened_R), dim=1) # [batch, 2, samples]
# Clamp the waveform to prevent clipping
widened_waveform = widened_waveform * gain
widened_waveform = torch.clamp(widened_waveform, -1.0, 1.0)
# Prepare the output dictionary
widened_audio = {
'waveform': widened_waveform,
'sample_rate': sample_rate
}
return widened_audio, sample_rate
SignalProcessingVolumeControl.py
+71
View File
@@ -0,0 +1,71 @@
import torch
import os, sys
from typing import Tuple, List, Dict, Union
class SignalProcessingVolumeControl:
@classmethod
def INPUT_TYPES(cls) -> Dict:
return {
"required": {
"audio_input": ("AUDIO",), # Single audio input
"gain": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 2.0, "step": 0.01}),
}
}
RETURN_TYPES = ("AUDIO","INT")
RETURN_NAMES = ("adjusted_audio","sample_rate")
CATEGORY = "Audio Processing"
FUNCTION = "process"
def process(
self,
audio_input: Dict[str, Union[torch.Tensor, int]],
gain: float,
) -> Tuple[Dict[str, torch.Tensor]]:
"""
Adjust the volume of the input audio by applying a gain factor.
Parameters:
audio_input (Dict): Dictionary containing 'waveform' and 'sample_rate'.
gain (float): Gain factor to adjust the volume. >1.0 amplifies, <1.0 attenuates.
Returns:
Tuple[Dict[str, torch.Tensor]]: Dictionary with adjusted 'waveform' and 'sample_rate'.
"""
# Extract waveform and sample rate
waveform = audio_input.get('waveform')
sample_rate = audio_input.get('sample_rate')
if waveform is None or sample_rate is None:
raise ValueError("Input audio must contain 'waveform' and 'sample_rate'.")
# Validate waveform tensor
if not isinstance(waveform, torch.Tensor):
raise TypeError("Waveform must be a torch.Tensor.")
if waveform.ndim != 3:
raise ValueError("Waveform must be a 3D tensor with shape (batch, channels, samples).")
if waveform.dtype != torch.float32:
waveform = waveform.float()
if not waveform.is_contiguous():
waveform = waveform.contiguous()
# Apply gain as a multiplier
adjusted_waveform = waveform * gain
print(f"AudioVolumeControl: Applied gain factor {gain}")
# Clamp to [-1.0, 1.0] to prevent clipping
adjusted_waveform = torch.clamp(adjusted_waveform, -1.0, 1.0)
print("AudioVolumeControl: Clamped adjusted waveform to [-1.0, 1.0]")
# Ensure the tensor is on CPU
if adjusted_waveform.device != torch.device('cpu'):
adjusted_waveform = adjusted_waveform.to('cpu')
# Prepare the output dictionary
adjusted_audio = {
'waveform': adjusted_waveform,
'sample_rate': sample_rate
}
return adjusted_audio, sample_rate
SignalProcessingWaveform.py
+81
View File
@@ -0,0 +1,81 @@
import torch
from io import BytesIO
from PIL import Image
import numpy as np
import matplotlib.pyplot as plt
class SignalProcessingWaveform:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"audio_input": ("AUDIO",),
},
"optional": {
"color": ("STRING", {"default": "black"}),
"background_color": ("STRING", {"default": "white"}),
"width": ("INT", {"default": 800, "min": 100, "max": 4000, "step": 100}),
"height": ("INT", {"default": 200, "min": 50, "max": 1000, "step": 50}),
"line_width": ("FLOAT", {"default": 1.0, "min": 0.1, "max": 10.0, "step": 0.1}),
}
}
RETURN_TYPES = ("IMAGE",)
RETURN_NAMES = ("waveform_image",)
CATEGORY = "Audio Processing"
FUNCTION = "process"
def process(self, audio_input, color="white", background_color="black", width=800, height=200, line_width=1.0):
waveform = audio_input.get('waveform') # [channels, samples] or [batch, channels, samples]
sample_rate = audio_input.get('sample_rate')
# Convert to mono by averaging channels
if waveform.ndim == 3:
# [batch, channels, samples]
waveform = waveform.mean(dim=1, keepdim=True) # [batch, 1, samples]
waveform = waveform.squeeze(0) # [1, samples]
elif waveform.ndim == 2:
# [channels, samples]
if waveform.shape[0] > 1:
waveform = waveform.mean(dim=0, keepdim=True) # [1, samples]
else:
waveform = waveform.unsqueeze(0) # [1, samples]
elif waveform.ndim == 1:
# [samples]
waveform = waveform.unsqueeze(0) # [1, samples]
else:
raise ValueError(f"Unsupported waveform shape: {waveform.shape}")
# Convert waveform to numpy
waveform_np = waveform.squeeze().detach().cpu().numpy() # [samples]
# Create a matplotlib figure without axes
plt.figure(figsize=(width/100, height/100), dpi=100)
plt.axis('off')
plt.margins(0,0)
plt.gca().set_facecolor(background_color)
plt.gca().set_position([0, 0, 1, 1])
# Plot the waveform
plt.plot(waveform_np, color=color, linewidth=line_width)
plt.tight_layout(pad=0)
# Save the plot to a buffer
buf = BytesIO()
plt.savefig(buf, format='png', bbox_inches='tight', pad_inches=0)
plt.close()
# Load the image from the buffer
buf.seek(0)
waveform_image = Image.open(buf).convert("RGB")
# Resize if necessary
waveform_image = waveform_image.resize((width, height), Image.BILINEAR)
# Convert to numpy array and normalize to [0,1]
image_np = np.array(waveform_image).astype(np.float32) / 255.0 # [H, W, 3]
# Convert to torch tensor and add batch dimension
image = torch.from_numpy(image_np)[None,]
return (image, )
__init__.py
+1 -2
View File
@@ -1,3 +1,2 @@
from .nodes import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS
__all__ = ['NODE_CLASS_MAPPINGS', 'NODE_DISPLAY_NAME_MAPPINGS']
__all__ = ['NODE_CLASS_MAPPINGS', 'NODE_DISPLAY_NAME_MAPPINGS','SignalProcessingFilter','SignalProcessingLoadAudio']
nodes.py
+35 -449
View File
@@ -1,464 +1,50 @@
import torch
import os, sys, re, shutil, subprocess
import hashlib
import numpy as np
import math
from typing import Tuple, List, Dict
import torchaudio.functional as F
from scipy.signal import get_window
import os, sys
sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), "comfy"))
sys.path.insert(0, os.path.join(os.path.dirname(os.path.realpath(__file__)), "comfy"))
import torchaudio
import folder_paths
class SignalProcessingLoadAudio():
supported_formats = ['wav','mp3','ogg','m4a','flac']
@classmethod
def INPUT_TYPES(s):
supported_extensions = tuple(f".{fmt.lower()}" for fmt in SignalProcessingLoadAudio.supported_formats)
from .SignalProcessingLoadAudio import SignalProcessingLoadAudio
from .SignalProcessingFilter import SignalProcessingFilter
from .SignalProcessingMultiBandEQ import SignalProcessingMultibandEQ
from .SignalProcessingPaulStretch import SignalProcessingPaulStretch
from .SignalProcessingPadSynth import SignalProcessingPadSynth
from .SignalProcessingPadSynthChoir import SignalProcessingPadSynthChoir
from .SignalProcessingMixdown import SignalProcessingMixdown
from .SignalProcessingPadSynthChoir2 import SignalProcessingPadSynthChoir2
from .SignalProcessingVolumeControl import SignalProcessingVolumeControl
from .SignalProcessingSpectrogram import SignalProcessingSpectrogram
from .SignalProcessingWaveform import SignalProcessingWaveform
from .SignalProcessingStereoWidening import SignalProcessingStereoWidening
from .SignalProcessingMultiBandEQExperimental import SignalProcessingMultiBandEQExperimental
input_dir = folder_paths.get_input_directory()
all_items = os.listdir(input_dir)
filtered_files = [
x for x in all_items
if x.lower().endswith(supported_extensions)
]
files = [os.path.join(input_dir,x) for x in filtered_files]
return {
"required": {"audio_file": (sorted(files), {"image_upload": True})},
"optional" : {"seek_seconds": ("FLOAT", {"default": 0, "min": 0})}
}
RETURN_TYPES = ("AUDIO",)
RETURN_NAMES = ("audio",)
CATEGORY = "Signal Processing"
FUNCTION = "process"
def process(self, audio_file, seek_seconds):
audio_file_path = folder_paths.get_annotated_filepath(audio_file)
waveform, sample_rate = torchaudio.load(audio_file_path)
waveform = waveform.unsqueeze(0).contiguous()
return ({'waveform': waveform, 'sample_rate': sample_rate},)
@classmethod
def IS_CHANGED(s, audio_file, seek_seconds):
audio_file_path = folder_paths.get_annotated_filepath(audio_file)
m = hashlib.sha256()
with open(audio_file_path, 'rb') as f:
m.update(f.read())
return m.digest().hex()
@classmethod
def VALIDATE_INPUTS(s, audio_file, seek_seconds):
if not folder_paths.exists_annotated_filepath(audio_file):
return "Invalid image file: {}".format(audio_file)
return True
class SignalProcessingFilter:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"audio": ("AUDIO", {"forceInput": True}),
"cutoff": ("FLOAT", {"default": 0.5, "min": 0.0, "max": 1.0, "step": 0.01}),
"filter_type": (["lowpass", "highpass", "bandpass", "bandstop"], {"default": "lowpass"}),
"q_factor": ("FLOAT", {"default": 0.707, "min": 0.1, "max": 5.0, "step": 0.01}), # For resonance/bandwidth
}
}
RETURN_TYPES = ("AUDIO",)
RETURN_NAMES = ("audio",)
CATEGORY = "Signal Processing"
FUNCTION = "apply_filter"
def apply_filter(self, audio: Dict[str, torch.Tensor], cutoff: float, filter_type: str, q_factor: float):
"""
Apply a specified filter to the input audio.
Parameters:
audio (Dict[str, torch.Tensor]): Input audio with 'waveform' and 'sample_rate'.
cutoff (float): Normalized cutoff frequency (0.0 to 1.0).
filter_type (str): Type of filter ('lowpass', 'highpass', 'bandpass', 'bandstop').
q_factor (float): Quality factor determining the filter's bandwidth.
Returns:
Tuple[Dict[str, torch.Tensor]]: Filtered audio.
"""
waveform = audio['waveform'].squeeze(0)
sample_rate = audio['sample_rate']
nyquist = sample_rate / 2.0
# Define minimum and maximum frequencies for mapping
log_min = 20.0 # 20 Hz, typical lower bound of human hearing
log_max = nyquist - 100.0 # Slightly below Nyquist to prevent instability
# Avoid log(0) by ensuring cutoff is within (0,1)
cutoff = min(max(cutoff, 1e-6), 1.0 - 1e-6)
# Logarithmic mapping
log_min = torch.log(torch.tensor(log_min))
log_max = torch.log(torch.tensor(log_max))
log_cutoff = log_min + cutoff * (log_max - log_min)
cutoff_freq = torch.exp(log_cutoff).item()
# Debug: Print mapped cutoff frequency
print(f"Normalized cutoff: {cutoff} mapped to frequency: {cutoff_freq:.2f} Hz")
# Choose filter type
if filter_type == "lowpass":
filtered_waveform = torchaudio.functional.lowpass_biquad(
waveform, sample_rate, cutoff_freq, Q=q_factor
)
elif filter_type == "highpass":
filtered_waveform = torchaudio.functional.highpass_biquad(
waveform, sample_rate, cutoff_freq, Q=q_factor
)
elif filter_type in ["bandpass", "bandstop"]:
center_freq = cutoff_freq
# Ensure that the bandwidth does not exceed the Nyquist frequency
bandwidth = center_freq / q_factor
lower_freq = max(center_freq - bandwidth / 2.0, 20.0) # Prevent dropping below 20 Hz
upper_freq = min(center_freq + bandwidth / 2.0, nyquist - 100.0) # Prevent exceeding Nyquist
# Debug: Print bandwidth details
print(f"Band filter with center frequency: {center_freq:.2f} Hz, "
f"lower: {lower_freq:.2f} Hz, upper: {upper_freq:.2f} Hz")
if filter_type == "bandpass":
filtered_waveform = torchaudio.functional.bandpass_biquad(
waveform, sample_rate, center_freq, Q=q_factor
)
else: # bandstop
filtered_waveform = torchaudio.functional.band_biquad(
waveform, sample_rate, center_freq, Q=q_factor
)
else:
raise ValueError(f"Unsupported filter type: {filter_type}")
return ({'waveform': filtered_waveform.unsqueeze(0), 'sample_rate': sample_rate},)
#https://github.com/paulnasca/paulstretch_python
#https://github.com/paulnasca/paulstretch_python/blob/master/paulstretch_stereo.py
class SignalProcessingPaulStretch:
@classmethod
def INPUT_TYPES(cls) -> Dict:
return {
"required": {
"audio": ("AUDIO", {"forceInput": True}),
"stretch_factor": ("FLOAT", {"default": 8.0, "min": 1.0, "max": 100.0, "step": 0.1}),
"window_size_seconds": ("FLOAT", {"default": 0.25, "min": 0.05, "max": 10.0, "step": 0.05}),
}
}
RETURN_TYPES = ("AUDIO",)
RETURN_NAMES = ("audio",)
CATEGORY = "Signal Processing"
FUNCTION = "process"
def process(
self,
audio: Dict[str, torch.Tensor],
stretch_factor: float,
window_size_seconds: float
) -> Tuple[Dict[str, torch.Tensor]]:
"""
Apply the PaulStretch algorithm to time-stretch the input audio without altering its pitch.
Parameters:
audio (Dict[str, torch.Tensor]): Input audio with 'waveform' and 'sample_rate'.
stretch_factor (float): Factor by which to stretch the audio (1.0 = no stretch).
window_size_seconds (float): Window size in seconds.
Returns:
Tuple[Dict[str, torch.Tensor]]: PaulStretched audio with original sample rate.
"""
# Initialize logging
# Conditional processing: If stretch_factor is 1.0, return original audio
if stretch_factor == 1.0:
return ({'waveform': audio['waveform'], 'sample_rate': audio['sample_rate']},)
# Squeeze the first dimension (e.g., batch dimension) to get shape (C, N) or (N,)
waveform = audio['waveform'].squeeze(0).cpu().numpy() # Shape: (C, N) or (N,) for mono
sample_rate = audio['sample_rate']
# Handle mono audio by converting to stereo by duplicating the channel
if waveform.ndim == 1:
waveform = np.tile(waveform, (2, 1)) # Shape: (2, N)
nchannels, nsamples = waveform.shape
# Optimize window size to be divisible by 2, 3, and 5
window_size = int(window_size_seconds * sample_rate)
if window_size < 16:
window_size = 16
window_size = self.optimize_windowsize(window_size)
window_size = int(window_size / 2) * 2 # Ensure even window size
half_window_size = int(window_size / 2)
# Correct the end of the waveform by applying a fade-out
end_size = int(sample_rate * 0.05)
if end_size < 16:
end_size = 16
waveform[:, -end_size:] *= np.linspace(1, 0, end_size)
# Compute displacement inside the input file
start_pos = 0.0
displace_pos = (window_size * 0.5) / stretch_factor
# Create custom window function as in original code
window = np.power(1.0 - np.power(np.linspace(-1.0, 1.0, window_size), 2.0), 1.25)
# Initialize old windowed buffer
old_windowed_buf = np.zeros((nchannels, window_size))
# Initialize list to store output frames
output_frames = []
# Processing loop
frame_count = 0
while True:
# Get the windowed buffer
istart_pos = int(math.floor(start_pos))
buf = waveform[:, istart_pos:istart_pos + window_size]
if buf.shape[1] < window_size:
buf = np.pad(buf, ((0, 0), (0, window_size - buf.shape[1])), 'constant')
buf = buf * window
# Get the amplitudes of the frequency components and discard the phases
freqs = np.abs(np.fft.rfft(buf, axis=1))
# Randomize the phases by multiplication with a random complex number with modulus=1
ph = np.random.uniform(0, 2 * np.pi, (nchannels, freqs.shape[1])) * 1j
freqs = freqs * np.exp(ph)
# Do the inverse FFT
buf_ifft = np.fft.irfft(freqs, n=window_size, axis=1)
# Window again the output buffer
buf_ifft *= window
# Overlap-add the output
output = buf_ifft[:, 0:half_window_size] + old_windowed_buf[:, half_window_size:window_size]
old_windowed_buf = buf_ifft
# Clamp the values to -1..1
output = np.clip(output, -1.0, 1.0)
# Append to output_frames
output_frames.append(output)
# Increment start_pos
start_pos += displace_pos
frame_count += 1
# Check if we have reached the end of the input
if start_pos >= nsamples:
break
# Optional: Log progress every 10%
num_frames = max(int(math.ceil((nsamples - window_size) / (window_size * 0.5)) + 1), 1)
if frame_count % max(int(num_frames / 10), 1) == 0:
progress = (start_pos / nsamples) * 100
# Concatenate all output frames horizontally
output_array = np.hstack(output_frames)
# Clamp final output to ensure no clipping
output_array = np.clip(output_array, -1.0, 1.0)
# Convert back to torch tensor
output_tensor = torch.from_numpy(output_array).float()
# Unsqueeze to add the first dimension back (e.g., batch dimension)
output_tensor = output_tensor.unsqueeze(0) # Shape: (1, C, N_stretched)
return ({'waveform': output_tensor, 'sample_rate': sample_rate},)
@staticmethod
def optimize_windowsize(n: int) -> int:
"""
Optimize the window size to be divisible by 2, 3, and 5.
Parameters:
n (int): Initial window size.
Returns:
int: Optimized window size.
"""
orig_n = n
while True:
n = orig_n
while (n % 2) == 0:
n /= 2
while (n % 3) == 0:
n /= 3
while (n % 5) == 0:
n /= 5
if n < 2:
break
orig_n += 1
return orig_n
class SignalProcessingPadSynth:
@classmethod
def INPUT_TYPES(cls) -> Dict:
return {
"required": {
"samplerate": ("INT", {"default": 44100, "min": 8000, "max": 96000, "step": 1}),
"fundamental_freq": ("FLOAT", {"default": 261.0, "min": 20.0, "max": 2000.0, "step": 1.0}),
"bandwidth_cents": ("FLOAT", {"default": 40.0, "min": 10.0, "max": 100.0, "step": 1.0}),
"number_harmonics": ("INT", {"default": 64, "min": 1, "max": 128, "step": 1}),
"amplitude_per_harmonic": ("STRING", {"default": "", "multiline": False, "hint": "Provide a JSON array of amplitudes per harmonic, e.g., [0.0, 1.0, 0.5, ...] or leave empty for default."})
}
}
RETURN_TYPES = ("AUDIO",)
RETURN_NAMES = ("audio",)
CATEGORY = "Signal Processing"
FUNCTION = "process"
def process(
self,
samplerate: int,
fundamental_freq: float,
bandwidth_cents: float,
number_harmonics: int,
amplitude_per_harmonic: str,
) -> Tuple[Dict[str, torch.Tensor]]:
"""
Apply PADsynth algorithm to generate audio.
Parameters:
samplerate (int): Sampling rate in Hz.
fundamental_freq (float): Fundamental frequency in Hz.
bandwidth_cents (float): Bandwidth in cents for Gaussian profile.
number_harmonics (int): Number of harmonics to generate.
amplitude_per_harmonic (str): JSON-formatted string specifying amplitudes per harmonic. Leave empty for defaults.
Returns:
Tuple[Dict[str, torch.Tensor]]: Generated audio with waveform and sample rate.
"""
import json
# Initialize logging
# Define FFT size
N = 262144 # As per C++ code
# Initialize amplitude array A
if not amplitude_per_harmonic.strip():
# Use default amplitude distribution
A = torch.zeros(number_harmonics, dtype=torch.double)
A[0] = 0.0 # A[0] is not used
for i in range(1, number_harmonics):
A[i] = 1.0 / i
if (i % 2) == 0:
A[i] *= 2.0
else:
# Parse JSON string to list
try:
amplitude_list = json.loads(amplitude_per_harmonic)
if not isinstance(amplitude_list, list):
raise ValueError("amplitude_per_harmonic must be a JSON array.")
if len(amplitude_list) != number_harmonics:
raise ValueError("Length of amplitude_per_harmonic must match number_harmonics.")
# Convert to torch tensor
A = torch.tensor(amplitude_list, dtype=torch.double)
except json.JSONDecodeError as e:
raise ValueError("amplitude_per_harmonic must be a valid JSON array.") from e
except Exception as e:
raise
# Initialize frequency amplitude and phase arrays
freq_amp = torch.zeros(N // 2, dtype=torch.double)
freq_phase = torch.rand(N // 2, dtype=torch.double) * 2.0 * math.pi # Random phases between 0 and 2pi
# Define Gaussian profile function
def profile(fi: torch.Tensor, bwi: torch.Tensor) -> torch.Tensor:
x = fi / bwi
x_sq = x ** 2
# Avoid computing exp(-x^2) for x_sq > 14.71280603
mask = x_sq <= 14.71280603
result = torch.zeros_like(x_sq)
result[mask] = torch.exp(-x_sq[mask]) / bwi[mask]
return result
# Convert bandwidth from cents to Hz
# bw_Hz = (2^(bw/1200) -1) * f * nh
# Convert bandwidth_cents to multiplier
bw_multiplier = 2.0 ** (bandwidth_cents / 1200.0) - 1.0
# Populate frequency amplitude array
for nh in range(1, number_harmonics):
f_nh = fundamental_freq * nh
bw_Hz = bw_multiplier * f_nh
bwi = bw_Hz / (2.0 * samplerate)
fi = f_nh / samplerate # Normalized frequency
# Create tensors for frequency bins
i = torch.arange(N // 2, dtype=torch.double)
# Normalized frequency for each bin
normalized_freq = i / N # Equivalent to i * (samplerate / N) / samplerate = i / N
# Compute profile
fi_tensor = torch.full_like(i, fi)
bwi_tensor = torch.full_like(i, bwi)
profile_values = profile(normalized_freq - fi_tensor, bwi_tensor)
# Update frequency amplitude
freq_amp += profile_values * A[nh]
# Construct complex frequency domain tensor
real = freq_amp * torch.cos(freq_phase)
imag = freq_amp * torch.sin(freq_phase)
freq_complex = torch.complex(real, imag) # Shape: (N//2,)
# Perform IFFT using torch.fft.irfft
smp = torch.fft.irfft(freq_complex, n=N) # Shape: (N,)
# Normalize the signal to prevent clipping
max_val = torch.max(torch.abs(smp))
if max_val < 1e-5:
max_val = 1e-5 # Prevent division by zero
smp = smp / (max_val * math.sqrt(2)) # Normalize to 1/sqrt(2) as in C++ code
# Convert to float32 for saving
smp = smp.float()
# Prepare waveform tensor: (C, N)
waveform_out = smp.unsqueeze(0) # Mono audio
# Reshape waveform_out to include batch dimension: (1, C, N)
waveform_out = waveform_out.unsqueeze(0) # Shape: (1, C, N)
# Return the synthesized audio
return ({'waveform': waveform_out, 'sample_rate': samplerate},)
NODE_CLASS_MAPPINGS = {
"SignalProcessingLoadAudio": SignalProcessingLoadAudio,
"SignalProcessingFilter": SignalProcessingFilter,
"SignalProcessingPaulStretch": SignalProcessingPaulStretch,
"SignalProcessingPadSynth": SignalProcessingPadSynth
"SignalProcessingPadSynth": SignalProcessingPadSynth,
"SignalProcessingPadSynthChoir": SignalProcessingPadSynthChoir,
"SignalProcessingPadSynthChoir2":SignalProcessingPadSynthChoir2,
"SignalProcessingMixdown": SignalProcessingMixdown,
"SignalProcessingVolumeControl" : SignalProcessingVolumeControl,
"SignalProcessingSpectrogram" : SignalProcessingSpectrogram,
"SignalProcessingWaveform" : SignalProcessingWaveform,
"SignalProcessingStereoWidening" : SignalProcessingStereoWidening,
"SignalProcessingMultibandEQ": SignalProcessingMultibandEQ,
"SignalProcessingMultiBandEQExperimental": SignalProcessingMultiBandEQExperimental
}
print('NODE_CLASS_MAPPINGS',NODE_CLASS_MAPPINGS)
NODE_DISPLAY_NAME_MAPPINGS = {
"SignalProcessingLoadAudio": "(SP) Load Audio",
"SignalProcessingFilter": "(SP) Filter",
"SignalProcessingPaulStretch" : "(SP) PaulStretch",
"SignalProcessingPadSynth": "(SP) PadSynth"
"SignalProcessingPadSynth": "(SP) PadSynth",
"SignalProcessingPadSynthChoir": "(SP) PadSynth Choir",
"SignalProcessingPadSynthChoir2":"(SP) PadSynth Choir 2",
"SignalProcessingMixdown": "(SP) Mix Down",
"SignalProcessingVolumeControl" : "(SP) Volume",
"SignalProcessingSpectrogram" : "(SP) Spectogram",
"SignalProcessingWaveform" : "(SP) Waveform",
"SignalProcessingStereoWidening" : "(SP) Stereo Widening",
"SignalProcessingMultibandEQ" : "(SP) Equalizer",
"SignalProcessingMultiBandEQExperimental": "(SP) Equalizer Experimental"
}
print('NODE_DISPLAY_NAME_MAPPINGS',NODE_DISPLAY_NAME_MAPPINGS)