added audioprocessor node + migrate get_audio + broken nodes fix
This commit is contained in:
SayanoAI
+89
-1
@@ -2,11 +2,13 @@ import base64
|
||||
import io
|
||||
import os
|
||||
import zlib
|
||||
from .utils import get_merge_func
|
||||
from .utils import get_hash, get_merge_func
|
||||
import numpy as np
|
||||
import librosa
|
||||
import soundfile as sf
|
||||
import ffmpeg
|
||||
from scipy.ndimage import uniform_filter1d, median_filter
|
||||
from scipy.interpolate import interp1d
|
||||
|
||||
MAX_INT16 = 32768
|
||||
SUPPORTED_AUDIO = ["mp3","flac","wav"] # ogg breaks soundfile
|
||||
@@ -26,6 +28,92 @@ AUTOTUNE_NOTES = np.array([
|
||||
2959.96, 3135.96, 3322.44, 3520.00, 3729.31, 3951.07
|
||||
])
|
||||
|
||||
class AudioProcessor:
|
||||
def __init__(self, normalize=True, threshold_silence=True, dynamic_threshold=True, sample_size=16000, multiplier=2.0, fill_method="median", kernel_size=5, silence_threshold_db=-50, normalize_threshold_db=-1):
|
||||
self.normalize=normalize
|
||||
self.threshold_silence=threshold_silence
|
||||
self.dynamic_threshold=dynamic_threshold
|
||||
self.sample_size=sample_size
|
||||
self.multiplier=multiplier
|
||||
self.fill_method=fill_method
|
||||
self.kernel_size=kernel_size
|
||||
self.silence_threshold_db=silence_threshold_db
|
||||
self.normalize_threshold_db=normalize_threshold_db
|
||||
|
||||
def __str__(self) -> str:
|
||||
values = [self.normalize, self.threshold_silence, self.dynamic_threshold]
|
||||
if self.normalize: values.append(self.normalize_threshold_db)
|
||||
if self.threshold_silence: values.append(self.silence_threshold_db)
|
||||
if self.dynamic_threshold: values.extend([self.sample_size, self.multiplier, self.fill_method, self.kernel_size])
|
||||
return get_hash(*values)
|
||||
|
||||
def __call__(self, audio):
|
||||
from .karafan.audio_utils import Normalize, Silent
|
||||
|
||||
samples,sr = get_audio(audio)
|
||||
|
||||
if self.threshold_silence:
|
||||
samples = np.atleast_2d([samples])
|
||||
samples = Silent(samples, sample_rate=sr, threshold_dB=self.silence_threshold_db)
|
||||
samples = np.squeeze(samples, 0)
|
||||
if self.dynamic_threshold: samples = self.dynamic_thresholding(
|
||||
samples,
|
||||
multiplier=self.multiplier,
|
||||
sample_size=self.sample_size,
|
||||
method=self.fill_method,
|
||||
kernel_size=self.kernel_size)
|
||||
if self.normalize: samples = Normalize(samples,threshold_dB=self.normalize_threshold_db)
|
||||
|
||||
return samples, sr
|
||||
|
||||
@staticmethod
|
||||
def dynamic_thresholding(samples, multiplier=2., sample_size=16000, method="median", kernel_size=5):
|
||||
# Calculate the local RMS (root mean square) to estimate the local amplitude
|
||||
local_rms = np.sqrt(uniform_filter1d(np.square(samples), size=int(sample_size)))
|
||||
|
||||
# Set dynamic threshold as a multiple of the local RMS
|
||||
dynamic_threshold = multiplier * local_rms
|
||||
|
||||
# Detect clicks where the absolute value exceeds the dynamic threshold
|
||||
clicks = np.abs(samples) > dynamic_threshold
|
||||
|
||||
# Replace clicks using the specified method
|
||||
cleaned_samples = AudioProcessor.replace_clicks(samples, clicks, method=method, kernel_size=kernel_size)
|
||||
return cleaned_samples
|
||||
|
||||
@staticmethod
|
||||
def replace_clicks(samples, clicks, method="median", kernel_size=5):
|
||||
if method == "median":
|
||||
# Replace detected clicks with the median value of surrounding samples
|
||||
cleaned_samples = samples.copy()
|
||||
cleaned_samples[clicks] = median_filter(samples, size=kernel_size)[clicks]
|
||||
elif method == "interpolation":
|
||||
# Replace detected clicks with interpolation
|
||||
cleaned_samples = samples.copy()
|
||||
|
||||
# Indices of non-clicks (valid points for interpolation)
|
||||
non_click_indices = np.where(~clicks)[0]
|
||||
|
||||
# Indices of clicks (points to be replaced)
|
||||
click_indices = np.where(clicks)[0]
|
||||
|
||||
# Interpolation function
|
||||
interp_func = interp1d(non_click_indices, cleaned_samples[non_click_indices], kind='linear', bounds_error=False, fill_value="extrapolate")
|
||||
|
||||
# Replace click samples with interpolated values
|
||||
cleaned_samples[click_indices] = interp_func(click_indices)
|
||||
else:
|
||||
raise ValueError("Method must be 'median' or 'interpolation'")
|
||||
|
||||
return cleaned_samples
|
||||
|
||||
def get_audio(audio):
|
||||
if hasattr(audio,"__call__"): #VHS_AUDIO is a function
|
||||
return bytes_to_audio(audio())
|
||||
elif isinstance(audio,dict): #comfyui AUDIO is a dict
|
||||
return audio["waveform"].squeeze(0).transpose(0,1).numpy(), audio["sample_rate"]
|
||||
else: return audio
|
||||
|
||||
def load_audio(file, sr, **kwargs):
|
||||
try:
|
||||
# https://github.com/openai/whisper/blob/main/whisper/audio.py#L26
|
||||
|
||||
+31
-49
@@ -124,7 +124,7 @@ def Silent(audio_in, sample_rate, threshold_dB = -50):
|
||||
|
||||
start = 0; end = 0
|
||||
audio = audio_in.copy()
|
||||
audio_length = audio_in.shape[1]
|
||||
audio_length = audio_in.shape[-1]
|
||||
|
||||
for i in range(0, audio_length, window_frame):
|
||||
|
||||
@@ -241,71 +241,53 @@ def Change_sample_rate(audio, way, current_cutoff, target_cutoff):
|
||||
|
||||
pitched_audio = librosa.resample(audio, orig_sr = current_cutoff * 2, target_sr = target_cutoff * 2, res_type = 'kaiser_best', axis=1)
|
||||
|
||||
# print(f"SRS input audio shape: {audio.shape}")
|
||||
# print(f"SRS output audio shape: {pitched_audio.shape}")
|
||||
# print (f"ratio : {ratio}")
|
||||
|
||||
return pitched_audio
|
||||
|
||||
# def Remove_High_freq_Noise(audio, threshold_freq):
|
||||
def Remove_High_freq_Noise(audio, threshold_freq):
|
||||
|
||||
# # Calculer la transformée de Fourier
|
||||
# stft = librosa.stft(audio)
|
||||
# Calculer la transformée de Fourier
|
||||
stft = librosa.stft(audio)
|
||||
|
||||
# # Calculer la somme des amplitudes pour chaque fréquence dans le spectre
|
||||
# amplitude_sum = np.sum(np.abs(stft), axis=0)
|
||||
# Calculer la somme des amplitudes pour chaque fréquence dans le spectre
|
||||
amplitude_sum = np.sum(np.abs(stft), axis=0)
|
||||
|
||||
# # Appliquer un masque pour supprimer les fréquences supérieures lorsque la somme des amplitudes est inférieure au seuil
|
||||
# stft[:, amplitude_sum > threshold_freq] = 0.0
|
||||
# Appliquer un masque pour supprimer les fréquences supérieures lorsque la somme des amplitudes est inférieure au seuil
|
||||
stft[:, amplitude_sum > threshold_freq] = 0.0
|
||||
|
||||
# # Reconstruire l'audio à partir du STFT modifié
|
||||
# audio_filtered = librosa.istft(stft)
|
||||
# Reconstruire l'audio à partir du STFT modifié
|
||||
audio_filtered = librosa.istft(stft)
|
||||
|
||||
# return audio_filtered
|
||||
return audio_filtered
|
||||
|
||||
# def Match_Freq_CutOFF(self, audio1, audio2, sample_rate):
|
||||
# # This option matches the Primary stem frequency cut-off to the Secondary stem frequency cut-off
|
||||
# # (if the Primary stem frequency cut-off is lower than the Secondary stem frequency cut-off)
|
||||
# Find the high cut-off frequency of the input audio
|
||||
def Find_Cut_OFF(audio, sample_rate, threshold=0.01):
|
||||
|
||||
# # Get the Primary stem frequency cut-off
|
||||
# freq_cut_off1 = Find_Cut_OFF(audio1, sample_rate)
|
||||
# freq_cut_off2 = Find_Cut_OFF(audio2, sample_rate)
|
||||
# Appliquer un filtre passe-bas pour réduire le bruit
|
||||
cutoff_frequency = sample_rate / 2.0 # Fréquence de Nyquist (la moitié du taux d'échantillonnage)
|
||||
|
||||
# # Match the Primary stem frequency cut-off to the Secondary stem frequency cut-off
|
||||
# if freq_cut_off1 < freq_cut_off2:
|
||||
# audio1 = Resize_Freq_CutOFF(audio1, freq_cut_off2, sample_rate)
|
||||
# Définir l'ordre du filtre passe-bas
|
||||
order = 6
|
||||
|
||||
# return audio1
|
||||
# Calculer les coefficients du filtre passe-bas
|
||||
b, a = signal.butter(order, cutoff_frequency - threshold, btype='low', analog=False, fs=sample_rate)
|
||||
|
||||
# # Find the high cut-off frequency of the input audio
|
||||
# def Find_Cut_OFF(audio, sample_rate, threshold=0.01):
|
||||
# Appliquer le filtre au signal audio
|
||||
filtered_audio = signal.lfilter(b, a, audio, axis=0)
|
||||
|
||||
# # Appliquer un filtre passe-bas pour réduire le bruit
|
||||
# cutoff_frequency = sample_rate / 2.0 # Fréquence de Nyquist (la moitié du taux d'échantillonnage)
|
||||
# Calculer la FFT du signal audio filtré
|
||||
fft_result = np.fft.fft(filtered_audio, axis=0)
|
||||
|
||||
# # Définir l'ordre du filtre passe-bas
|
||||
# order = 6
|
||||
# Calculer les magnitudes du spectre de fréquence
|
||||
magnitudes = np.abs(fft_result)
|
||||
|
||||
# # Calculer les coefficients du filtre passe-bas
|
||||
# b, a = signal.butter(order, cutoff_frequency - threshold, btype='low', analog=False, fs=sample_rate)
|
||||
# Calculer les fréquences correspondant aux bins de la FFT
|
||||
frequencies = np.fft.fftfreq(len(audio), 1.0 / sample_rate)
|
||||
|
||||
# # Appliquer le filtre au signal audio
|
||||
# filtered_audio = signal.lfilter(b, a, audio, axis=0)
|
||||
# Trouver la fréquence de coupure où la magnitude tombe en dessous du seuil
|
||||
cut_off_frequencies = frequencies[np.where(magnitudes > threshold)]
|
||||
|
||||
# # Calculer la FFT du signal audio filtré
|
||||
# fft_result = np.fft.fft(filtered_audio, axis=0)
|
||||
|
||||
# # Calculer les magnitudes du spectre de fréquence
|
||||
# magnitudes = np.abs(fft_result)
|
||||
|
||||
# # Calculer les fréquences correspondant aux bins de la FFT
|
||||
# frequencies = np.fft.fftfreq(len(audio), 1.0 / sample_rate)
|
||||
|
||||
# # Trouver la fréquence de coupure où la magnitude tombe en dessous du seuil
|
||||
# cut_off_frequencies = frequencies[np.where(magnitudes > threshold)]
|
||||
|
||||
# # Trouver la fréquence de coupure maximale parmi toutes les valeurs
|
||||
# return int(max(cut_off_frequencies))
|
||||
# Trouver la fréquence de coupure maximale parmi toutes les valeurs
|
||||
return int(max(cut_off_frequencies))
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user