From 448f868b676e6f325d0a9a30553e80ebccd52515 Mon Sep 17 00:00:00 2001 From: c0ffymachyne Date: Wed, 20 Nov 2024 22:27:33 -0800 Subject: [PATCH] added more nodes --- README.md | 2 + SignalProcessingFilter.py | 91 ++++ SignalProcessingLoadAudio.py | 56 +++ SignalProcessingMixdown.py | 128 ++++++ SignalProcessingMultiBandEQ.py | 284 ++++++++++++ SignalProcessingMultiBandEQExperimental.py | 191 ++++++++ SignalProcessingPadSynth.py | 142 ++++++ SignalProcessingPadSynthChoir.py | 141 ++++++ SignalProcessingPadSynthChoir2.py | 201 +++++++++ SignalProcessingPaulStretch.py | 172 ++++++++ SignalProcessingSpectrogram.py | 104 +++++ SignalProcessingStereoWidening.py | 99 +++++ SignalProcessingVolumeControl.py | 71 +++ SignalProcessingWaveform.py | 81 ++++ __init__.py | 3 +- nodes.py | 484 ++------------------- 16 files changed, 1799 insertions(+), 451 deletions(-) create mode 100644 SignalProcessingFilter.py create mode 100644 SignalProcessingLoadAudio.py create mode 100644 SignalProcessingMixdown.py create mode 100644 SignalProcessingMultiBandEQ.py create mode 100644 SignalProcessingMultiBandEQExperimental.py create mode 100644 SignalProcessingPadSynth.py create mode 100644 SignalProcessingPadSynthChoir.py create mode 100644 SignalProcessingPadSynthChoir2.py create mode 100644 SignalProcessingPaulStretch.py create mode 100644 SignalProcessingSpectrogram.py create mode 100644 SignalProcessingStereoWidening.py create mode 100644 SignalProcessingVolumeControl.py create mode 100644 SignalProcessingWaveform.py diff --git a/README.md b/README.md index 5005809..d506d2e 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/SignalProcessingFilter.py b/SignalProcessingFilter.py new file mode 100644 index 0000000..3f5fbda --- /dev/null +++ b/SignalProcessingFilter.py @@ -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 \ No newline at end of file diff --git a/SignalProcessingLoadAudio.py b/SignalProcessingLoadAudio.py new file mode 100644 index 0000000..9b1d50d --- /dev/null +++ b/SignalProcessingLoadAudio.py @@ -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 \ No newline at end of file diff --git a/SignalProcessingMixdown.py b/SignalProcessingMixdown.py new file mode 100644 index 0000000..2f1eaea --- /dev/null +++ b/SignalProcessingMixdown.py @@ -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) \ No newline at end of file diff --git a/SignalProcessingMultiBandEQ.py b/SignalProcessingMultiBandEQ.py new file mode 100644 index 0000000..8d69140 --- /dev/null +++ b/SignalProcessingMultiBandEQ.py @@ -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,) diff --git a/SignalProcessingMultiBandEQExperimental.py b/SignalProcessingMultiBandEQExperimental.py new file mode 100644 index 0000000..4cbf984 --- /dev/null +++ b/SignalProcessingMultiBandEQExperimental.py @@ -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, ) \ No newline at end of file diff --git a/SignalProcessingPadSynth.py b/SignalProcessingPadSynth.py new file mode 100644 index 0000000..dbdbd9d --- /dev/null +++ b/SignalProcessingPadSynth.py @@ -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 \ No newline at end of file diff --git a/SignalProcessingPadSynthChoir.py b/SignalProcessingPadSynthChoir.py new file mode 100644 index 0000000..2961a00 --- /dev/null +++ b/SignalProcessingPadSynthChoir.py @@ -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 \ No newline at end of file diff --git a/SignalProcessingPadSynthChoir2.py b/SignalProcessingPadSynthChoir2.py new file mode 100644 index 0000000..74e54c5 --- /dev/null +++ b/SignalProcessingPadSynthChoir2.py @@ -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 \ No newline at end of file diff --git a/SignalProcessingPaulStretch.py b/SignalProcessingPaulStretch.py new file mode 100644 index 0000000..10a1dc6 --- /dev/null +++ b/SignalProcessingPaulStretch.py @@ -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 \ No newline at end of file diff --git a/SignalProcessingSpectrogram.py b/SignalProcessingSpectrogram.py new file mode 100644 index 0000000..dcb36d5 --- /dev/null +++ b/SignalProcessingSpectrogram.py @@ -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, ) \ No newline at end of file diff --git a/SignalProcessingStereoWidening.py b/SignalProcessingStereoWidening.py new file mode 100644 index 0000000..3a46174 --- /dev/null +++ b/SignalProcessingStereoWidening.py @@ -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 \ No newline at end of file diff --git a/SignalProcessingVolumeControl.py b/SignalProcessingVolumeControl.py new file mode 100644 index 0000000..c1e3d53 --- /dev/null +++ b/SignalProcessingVolumeControl.py @@ -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 \ No newline at end of file diff --git a/SignalProcessingWaveform.py b/SignalProcessingWaveform.py new file mode 100644 index 0000000..50d3fec --- /dev/null +++ b/SignalProcessingWaveform.py @@ -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, ) \ No newline at end of file diff --git a/__init__.py b/__init__.py index d721463..1e43cf8 100644 --- a/__init__.py +++ b/__init__.py @@ -1,3 +1,2 @@ from .nodes import NODE_CLASS_MAPPINGS, NODE_DISPLAY_NAME_MAPPINGS - -__all__ = ['NODE_CLASS_MAPPINGS', 'NODE_DISPLAY_NAME_MAPPINGS'] \ No newline at end of file +__all__ = ['NODE_CLASS_MAPPINGS', 'NODE_DISPLAY_NAME_MAPPINGS','SignalProcessingFilter','SignalProcessingLoadAudio'] \ No newline at end of file diff --git a/nodes.py b/nodes.py index 62c7b99..1911b67 100644 --- a/nodes.py +++ b/nodes.py @@ -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)