| | import os |
| | import re |
| | import sys |
| | import math |
| | import torch |
| | import parselmouth |
| |
|
| | import numba as nb |
| | import numpy as np |
| |
|
| | from scipy.signal import medfilt |
| | from librosa import yin, pyin, piptrack |
| |
|
| | sys.path.append(os.getcwd()) |
| |
|
| | from infer.lib.predictors.CREPE.filter import mean, median |
| | from infer.lib.predictors.WORLD.SWIPE import swipe, stonemask |
| | from infer.lib.variables import config, configs, logger, translations |
| | from infer.lib.utils import autotune_f0, proposal_f0_up_key, circular_write |
| |
|
| | @nb.jit(nopython=True) |
| | def post_process( |
| | tf0, |
| | f0, |
| | f0_up_key, |
| | manual_x_pad, |
| | f0_mel_min, |
| | f0_mel_max, |
| | manual_f0 = None |
| | ): |
| | f0 *= pow(2, f0_up_key / 12) |
| |
|
| | if manual_f0 is not None: |
| | replace_f0 = np.interp( |
| | list( |
| | range( |
| | np.round( |
| | (manual_f0[:, 0].max() - manual_f0[:, 0].min()) * tf0 + 1 |
| | ).astype(np.int16) |
| | ) |
| | ), |
| | manual_f0[:, 0] * 100, |
| | manual_f0[:, 1] |
| | ) |
| |
|
| | f0[ |
| | manual_x_pad * tf0 : manual_x_pad * tf0 + len(replace_f0) |
| | ] = replace_f0[ |
| | :f0[ |
| | manual_x_pad * tf0 : manual_x_pad * tf0 + len(replace_f0) |
| | ].shape[0] |
| | ] |
| |
|
| | f0_mel = 1127 * np.log(1 + f0 / 700) |
| | f0_mel[f0_mel > 0] = (f0_mel[f0_mel > 0] - f0_mel_min) * 254 / (f0_mel_max - f0_mel_min) + 1 |
| | f0_mel[f0_mel <= 1] = 1 |
| | f0_mel[f0_mel > 255] = 255 |
| |
|
| | return np.rint(f0_mel).astype(np.int32), f0 |
| |
|
| | def realtime_post_process( |
| | f0, |
| | pitch, |
| | pitchf, |
| | f0_up_key = 0, |
| | f0_mel_min = 50.0, |
| | f0_mel_max = 1100.0 |
| | ): |
| | f0 *= 2 ** (f0_up_key / 12) |
| |
|
| | f0_mel = 1127.0 * (1.0 + f0 / 700.0).log() |
| | f0_mel = torch.clip((f0_mel - f0_mel_min) * 254 / (f0_mel_max - f0_mel_min) + 1, 1, 255, out=f0_mel) |
| | f0_coarse = torch.round(f0_mel, out=f0_mel).long() |
| |
|
| | if pitch is not None and pitchf is not None: |
| | circular_write(f0_coarse, pitch) |
| | circular_write(f0, pitchf) |
| | else: |
| | pitch = f0_coarse |
| | pitchf = f0 |
| |
|
| | return pitch.unsqueeze(0), pitchf.unsqueeze(0) |
| |
|
| | class Generator: |
| | def __init__( |
| | self, |
| | sample_rate = 16000, |
| | hop_length = 160, |
| | f0_min = 50, |
| | f0_max = 1100, |
| | alpha = 0.5, |
| | is_half = False, |
| | device = "cpu", |
| | predictor_onnx = False, |
| | delete_predictor_onnx = True |
| | ): |
| | self.sample_rate = sample_rate |
| | self.hop_length = hop_length |
| | self.f0_min = f0_min |
| | self.f0_max = f0_max |
| | self.is_half = is_half |
| | self.device = device |
| | self.providers = config.providers |
| | self.predictor_onnx = predictor_onnx |
| | self.delete_predictor_onnx = delete_predictor_onnx |
| | self.window = 160 |
| | self.batch_size = 512 |
| | self.alpha = alpha |
| | self.ref_freqs = [ |
| | 49.00, |
| | 51.91, |
| | 55.00, |
| | 58.27, |
| | 61.74, |
| | 65.41, |
| | 69.30, |
| | 73.42, |
| | 77.78, |
| | 82.41, |
| | 87.31, |
| | 92.50, |
| | 98.00, |
| | 103.83, |
| | 110.00, |
| | 116.54, |
| | 123.47, |
| | 130.81, |
| | 138.59, |
| | 146.83, |
| | 155.56, |
| | 164.81, |
| | 174.61, |
| | 185.00, |
| | 196.00, |
| | 207.65, |
| | 220.00, |
| | 233.08, |
| | 246.94, |
| | 261.63, |
| | 277.18, |
| | 293.66, |
| | 311.13, |
| | 329.63, |
| | 349.23, |
| | 369.99, |
| | 392.00, |
| | 415.30, |
| | 440.00, |
| | 466.16, |
| | 493.88, |
| | 523.25, |
| | 554.37, |
| | 587.33, |
| | 622.25, |
| | 659.25, |
| | 698.46, |
| | 739.99, |
| | 783.99, |
| | 830.61, |
| | 880.00, |
| | 932.33, |
| | 987.77, |
| | 1046.50 |
| | ] |
| |
|
| | def calculator( |
| | self, |
| | x_pad, |
| | f0_method, |
| | x, |
| | f0_up_key = 0, |
| | p_len = None, |
| | filter_radius = 3, |
| | f0_autotune = False, |
| | f0_autotune_strength = 1, |
| | manual_f0 = None, |
| | proposal_pitch = False, |
| | proposal_pitch_threshold = 255.0 |
| | ): |
| | if p_len is None: p_len = x.shape[0] // self.window |
| | if "hybrid" in f0_method: logger.debug(translations["hybrid_calc"].format(f0_method=f0_method)) |
| |
|
| | compute_fn = ( |
| | self.get_f0_hybrid if "hybrid" in f0_method else self.compute_f0 |
| | ) |
| |
|
| | f0 = compute_fn( |
| | f0_method, |
| | x, |
| | p_len, |
| | filter_radius if filter_radius % 2 != 0 else filter_radius + 1 |
| | ) |
| | |
| | if proposal_pitch: |
| | up_key = proposal_f0_up_key( |
| | f0, |
| | proposal_pitch_threshold, |
| | configs["limit_f0"] |
| | ) |
| |
|
| | logger.debug(translations["proposal_f0"].format(up_key=up_key)) |
| | f0_up_key += up_key |
| |
|
| | if f0_autotune: |
| | logger.debug(translations["startautotune"]) |
| |
|
| | f0 = autotune_f0( |
| | self.ref_freqs, |
| | f0, |
| | f0_autotune_strength |
| | ) |
| |
|
| | return post_process( |
| | self.sample_rate // self.window, |
| | f0, |
| | f0_up_key, |
| | x_pad, |
| | 1127 * math.log(1 + self.f0_min / 700), |
| | 1127 * math.log(1 + self.f0_max / 700), |
| | manual_f0 |
| | ) |
| |
|
| | def realtime_calculator( |
| | self, |
| | audio, |
| | f0_method, |
| | pitch, |
| | pitchf, |
| | f0_up_key = 0, |
| | filter_radius = 3, |
| | f0_autotune = False, |
| | f0_autotune_strength = 1, |
| | proposal_pitch = False, |
| | proposal_pitch_threshold = 255.0 |
| | ): |
| | if torch.is_tensor(audio): audio = audio.cpu().numpy() |
| | p_len = audio.shape[0] // self.window |
| |
|
| | f0 = self.compute_f0( |
| | f0_method, |
| | audio, |
| | p_len, |
| | filter_radius if filter_radius % 2 != 0 else filter_radius + 1 |
| | ) |
| |
|
| | if f0_autotune: |
| | f0 = autotune_f0( |
| | self.ref_freqs, |
| | f0, |
| | f0_autotune_strength |
| | ) |
| |
|
| | if proposal_pitch: |
| | up_key = proposal_f0_up_key( |
| | f0, |
| | proposal_pitch_threshold, |
| | configs["limit_f0"] |
| | ) |
| |
|
| | f0_up_key += up_key |
| |
|
| | return realtime_post_process( |
| | torch.from_numpy(f0).float().to(self.device), |
| | pitch, |
| | pitchf, |
| | f0_up_key, |
| | self.f0_min, |
| | self.f0_max |
| | ) |
| |
|
| | def _resize_f0(self, x, target_len): |
| | if len(x) == target_len: return x |
| |
|
| | source = np.array(x) |
| | source[source < 0.001] = np.nan |
| |
|
| | return np.nan_to_num( |
| | np.interp( |
| | np.arange(0, len(source) * target_len, len(source)) / target_len, |
| | np.arange(0, len(source)), |
| | source |
| | ) |
| | ) |
| | |
| | def compute_f0(self, f0_method, x, p_len, filter_radius): |
| | if "pm" in f0_method: |
| | f0 = self.get_f0_pm( |
| | x, |
| | p_len, |
| | filter_radius=filter_radius, |
| | mode=f0_method.split("-")[1] |
| | ) |
| | elif f0_method.split("-")[0] in ["harvest", "dio"]: |
| | f0 = self.get_f0_pyworld( |
| | x, |
| | p_len, |
| | filter_radius, |
| | f0_method.split("-")[0], |
| | use_stonemask="stonemask" in f0_method |
| | ) |
| | elif "crepe" in f0_method: |
| | split_f0 = f0_method.split("-") |
| | f0 = ( |
| | self.get_f0_mangio_crepe( |
| | x, |
| | p_len, |
| | split_f0[2] |
| | ) |
| | ) if split_f0[0] == "mangio" else ( |
| | self.get_f0_crepe( |
| | x, |
| | p_len, |
| | split_f0[1], |
| | filter_radius=filter_radius |
| | ) |
| | ) |
| | elif "fcpe" in f0_method: |
| | f0 = self.get_f0_fcpe( |
| | x, |
| | p_len, |
| | legacy="legacy" in f0_method and "previous" not in f0_method, |
| | previous="previous" in f0_method, |
| | filter_radius=filter_radius |
| | ) |
| | elif "rmvpe" in f0_method: |
| | f0 = self.get_f0_rmvpe( |
| | x, |
| | p_len, |
| | clipping="clipping" in f0_method, |
| | filter_radius=filter_radius, |
| | hpa="hpa" in f0_method, |
| | previous="previous" in f0_method |
| | ) |
| | elif f0_method in ["yin", "pyin", "piptrack"]: |
| | f0 = self.get_f0_librosa( |
| | x, |
| | p_len, |
| | mode=f0_method, |
| | filter_radius=filter_radius |
| | ) |
| | |
| | elif "djcm" in f0_method: |
| | f0 = self.get_f0_djcm( |
| | x, |
| | p_len, |
| | clipping="clipping" in f0_method, |
| | svs="svs" in f0_method, |
| | filter_radius=filter_radius |
| | ) |
| | |
| | else: |
| | raise ValueError(translations["option_not_valid"]) |
| | |
| | if isinstance(f0, tuple): f0 = f0[0] |
| | if "medfilt" in f0_method or "svs" in f0_method: f0 = medfilt(f0, kernel_size=5) |
| |
|
| | return f0 |
| | |
| | def get_f0_hybrid(self, methods_str, x, p_len, filter_radius): |
| | methods_str = re.search(r"hybrid\[(.+)\]", methods_str) |
| | if methods_str: |
| | methods = [ |
| | method.strip() |
| | for method in methods_str.group(1).split("+") |
| | ] |
| |
|
| | n = len(methods) |
| | f0_stack = [] |
| |
|
| | for method in methods: |
| | f0_stack.append( |
| | self._resize_f0( |
| | self.compute_f0( |
| | method, |
| | x, |
| | p_len, |
| | filter_radius |
| | ), |
| | p_len |
| | ) |
| | ) |
| | |
| | f0_mix = np.zeros(p_len) |
| |
|
| | if not f0_stack: return f0_mix |
| | if len(f0_stack) == 1: return f0_stack[0] |
| |
|
| | weights = (1 - np.abs(np.arange(n) / (n - 1) - (1 - self.alpha))) ** 2 |
| | weights /= weights.sum() |
| |
|
| | stacked = np.vstack(f0_stack) |
| | voiced_mask = np.any(stacked > 0, axis=0) |
| |
|
| | f0_mix[voiced_mask] = np.exp( |
| | np.nansum( |
| | np.log(stacked + 1e-6) * weights[:, None], axis=0 |
| | )[voiced_mask] |
| | ) |
| |
|
| | return f0_mix |
| |
|
| | def get_f0_pm(self, x, p_len, filter_radius=3, mode="ac"): |
| | time_step = self.window / self.sample_rate * 1000 / 1000 |
| |
|
| | pm = parselmouth.Sound( |
| | x, |
| | self.sample_rate |
| | ) |
| | pm_fn = { |
| | "ac": pm.to_pitch_ac, |
| | "cc": pm.to_pitch_cc, |
| | "shs": pm.to_pitch_shs |
| | }.get(mode, pm.to_pitch_ac) |
| |
|
| | pitch = ( |
| | pm_fn( |
| | time_step=time_step, |
| | voicing_threshold=filter_radius / 10 * 2, |
| | pitch_floor=self.f0_min, |
| | pitch_ceiling=self.f0_max |
| | ) |
| | ) if mode != "shs" else ( |
| | pm_fn( |
| | time_step=time_step, |
| | minimum_pitch=self.f0_min, |
| | maximum_frequency_component=self.f0_max |
| | ) |
| | ) |
| |
|
| | f0 = pitch.selected_array["frequency"] |
| | pad_size = (p_len - len(f0) + 1) // 2 |
| |
|
| | if pad_size > 0 or p_len - len(f0) - pad_size > 0: |
| | f0 = np.pad( |
| | f0, |
| | [[pad_size, p_len - len(f0) - pad_size]], |
| | mode="constant" |
| | ) |
| |
|
| | return f0 |
| | |
| | def get_f0_mangio_crepe(self, x, p_len, model="full"): |
| | if not hasattr(self, "mangio_crepe"): |
| | from infer.lib.predictors.CREPE.CREPE import CREPE |
| |
|
| | self.mangio_crepe = CREPE( |
| | os.path.join( |
| | configs["predictors_path"], |
| | f"crepe_{model}.{'onnx' if self.predictor_onnx else 'pth'}" |
| | ), |
| | model_size=model, |
| | hop_length=self.hop_length, |
| | batch_size=self.hop_length * 2, |
| | f0_min=self.f0_min, |
| | f0_max=self.f0_max, |
| | device=self.device, |
| | sample_rate=self.sample_rate, |
| | providers=self.providers, |
| | onnx=self.predictor_onnx, |
| | return_periodicity=False |
| | ) |
| |
|
| | x = x.astype(np.float32) |
| | x /= np.quantile(np.abs(x), 0.999) |
| |
|
| | audio = torch.from_numpy(x).to(self.device, copy=True).unsqueeze(dim=0) |
| | if audio.ndim == 2 and audio.shape[0] > 1: audio = audio.mean(dim=0, keepdim=True).detach() |
| |
|
| | f0 = self.mangio_crepe.compute_f0(audio.detach(), pad=True) |
| | if self.predictor_onnx and self.delete_predictor_onnx: del self.mangio_crepe.model, self.mangio_crepe |
| |
|
| | return self._resize_f0(f0.squeeze(0).cpu().float().numpy(), p_len) |
| | |
| | def get_f0_crepe(self, x, p_len, model="full", filter_radius=3): |
| | if not hasattr(self, "crepe"): |
| | from infer.lib.predictors.CREPE.CREPE import CREPE |
| |
|
| | self.crepe = CREPE( |
| | os.path.join( |
| | configs["predictors_path"], |
| | f"crepe_{model}.{'onnx' if self.predictor_onnx else 'pth'}" |
| | ), |
| | model_size=model, |
| | hop_length=self.window, |
| | batch_size=self.batch_size, |
| | f0_min=self.f0_min, |
| | f0_max=self.f0_max, |
| | device=self.device, |
| | sample_rate=self.sample_rate, |
| | providers=self.providers, |
| | onnx=self.predictor_onnx, |
| | return_periodicity=True |
| | ) |
| |
|
| | f0, pd = self.crepe.compute_f0(torch.tensor(np.copy(x))[None].float(), pad=True) |
| | if self.predictor_onnx and self.delete_predictor_onnx: del self.crepe.model, self.crepe |
| |
|
| | f0, pd = mean(f0, filter_radius), median(pd, filter_radius) |
| | f0[pd < 0.1] = 0 |
| |
|
| | return self._resize_f0(f0[0].cpu().numpy(), p_len) |
| | |
| | def get_f0_fcpe(self, x, p_len, legacy=False, previous=False, filter_radius=3): |
| | if not hasattr(self, "fcpe"): |
| | from infer.lib.predictors.FCPE.FCPE import FCPE |
| |
|
| | self.fcpe = FCPE( |
| | configs, |
| | os.path.join( |
| | configs["predictors_path"], |
| | ( |
| | "fcpe_legacy" |
| | if legacy else |
| | ("fcpe" if previous else "ddsp_200k") |
| | ) + (".onnx" if self.predictor_onnx else ".pt") |
| | ), |
| | hop_length=self.hop_length, |
| | f0_min=self.f0_min, |
| | f0_max=self.f0_max, |
| | dtype=torch.float32, |
| | device=self.device, |
| | sample_rate=self.sample_rate, |
| | threshold=( |
| | filter_radius / 100 |
| | ) if legacy else ( |
| | filter_radius / 1000 * 2 |
| | ), |
| | providers=self.providers, |
| | onnx=self.predictor_onnx, |
| | legacy=legacy |
| | ) |
| | |
| | f0 = self.fcpe.compute_f0(x, p_len) |
| | if self.predictor_onnx and self.delete_predictor_onnx: del self.fcpe.fcpe.model, self.fcpe |
| |
|
| | return f0 |
| | |
| | def get_f0_rmvpe(self, x, p_len, clipping=False, filter_radius=3, hpa=False, previous=False): |
| | if not hasattr(self, "rmvpe"): |
| | from infer.lib.predictors.RMVPE.RMVPE import RMVPE |
| |
|
| | self.rmvpe = RMVPE( |
| | os.path.join( |
| | configs["predictors_path"], |
| | ( |
| | ( |
| | "hpa-rmvpe-76000" |
| | if previous else |
| | "hpa-rmvpe-112000" |
| | ) if hpa else "rmvpe" |
| | ) + (".onnx" if self.predictor_onnx else ".pt") |
| | ), |
| | is_half=self.is_half, |
| | device=self.device, |
| | onnx=self.predictor_onnx, |
| | providers=self.providers, |
| | hpa=hpa |
| | ) |
| |
|
| | filter_radius = filter_radius / 100 |
| |
|
| | f0 = ( |
| | self.rmvpe.infer_from_audio_with_pitch( |
| | x, |
| | thred=filter_radius, |
| | f0_min=self.f0_min, |
| | f0_max=self.f0_max |
| | ) |
| | ) if clipping else ( |
| | self.rmvpe.infer_from_audio( |
| | x, |
| | thred=filter_radius |
| | ) |
| | ) |
| | |
| | if self.predictor_onnx and self.delete_predictor_onnx: del self.rmvpe.model, self.rmvpe |
| | return self._resize_f0(f0, p_len) |
| | |
| | |
| | def get_f0_librosa(self, x, p_len, mode="yin", filter_radius=3): |
| | if mode != "piptrack": |
| | self.if_yin = mode == "yin" |
| | self.yin = yin if self.if_yin else pyin |
| |
|
| | f0 = self.yin( |
| | x.astype(np.float32), |
| | sr=self.sample_rate, |
| | fmin=self.f0_min, |
| | fmax=self.f0_max, |
| | hop_length=self.hop_length |
| | ) |
| |
|
| | if not self.if_yin: f0 = f0[0] |
| | else: |
| | pitches, magnitudes = piptrack( |
| | y=x.astype(np.float32), |
| | sr=self.sample_rate, |
| | fmin=self.f0_min, |
| | fmax=self.f0_max, |
| | hop_length=self.hop_length, |
| | threshold=filter_radius / 10 |
| | ) |
| |
|
| | max_indexes = np.argmax(magnitudes, axis=0) |
| | f0 = pitches[max_indexes, range(magnitudes.shape[1])] |
| |
|
| | return self._resize_f0(f0, p_len) |
| |
|
| | |
| | |
| | def get_f0_djcm(self, x, p_len, clipping=False, svs=False, filter_radius=3): |
| | if not hasattr(self, "djcm"): |
| | from main.library.predictors.DJCM.DJCM import DJCM |
| | |
| | self.djcm = DJCM( |
| | os.path.join( |
| | configs["predictors_path"], |
| | ( |
| | "djcm-svs" |
| | if svs else |
| | "djcm" |
| | ) + (".onnx" if self.predictor_onnx else ".pt") |
| | ), |
| | is_half=self.is_half, |
| | device=self.device, |
| | onnx=self.predictor_onnx, |
| | svs=svs, |
| | providers=self.providers |
| | ) |
| |
|
| | filter_radius /= 10 |
| |
|
| | f0 = ( |
| | self.djcm.infer_from_audio_with_pitch( |
| | x, |
| | thred=filter_radius, |
| | f0_min=self.f0_min, |
| | f0_max=self.f0_max |
| | ) |
| | ) if clipping else ( |
| | self.djcm.infer_from_audio( |
| | x, |
| | thred=filter_radius |
| | ) |
| | ) |
| | |
| | if self.predictor_onnx and self.delete_predictor_onnx: del self.djcm.model, self.djcm |
| | return self._resize_f0(f0, p_len) |
| | |
| | |
| | |
