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)