| 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) |
| |
| |
| |
