import os import sys import torch import faiss import numpy as np import torch.nn.functional as F from scipy import signal sys.path.append(os.getcwd()) from main.app.variables import translations from main.library.utils import extract_features from main.library.predictors.Generator import Generator from main.inference.extracting.rms import RMSEnergyExtractor from main.inference.conversion.utils import change_rms, clear_gpu_cache, get_onnx_argument bh, ah = signal.butter(N=5, Wn=48, btype="high", fs=16000) class Pipeline: def __init__(self, tgt_sr, config): self.x_pad = config.x_pad self.x_query = config.x_query self.x_center = config.x_center self.x_max = config.x_max self.sample_rate = 16000 self.window = 160 self.t_pad = self.sample_rate * self.x_pad self.t_pad_tgt = tgt_sr * self.x_pad self.t_pad2 = self.t_pad * 2 self.t_query = self.sample_rate * self.x_query self.t_center = self.sample_rate * self.x_center self.t_max = self.sample_rate * self.x_max self.f0_min = 50 self.f0_max = 1100 self.device = config.device self.is_half = config.is_half def voice_conversion(self, model, net_g, sid, audio0, pitch, pitchf, index, big_npy, index_rate, version, protect, energy): pitch_guidance = pitch != None and pitchf != None energy_use = energy != None feats = torch.from_numpy(audio0) feats = feats.half() if self.is_half else feats.float() feats = feats.mean(-1) if feats.dim() == 2 else feats assert feats.dim() == 1, feats.dim() feats = feats.view(1, -1) with torch.no_grad(): if self.embed_suffix == ".pt": padding_mask = torch.BoolTensor(feats.shape).to(self.device).fill_(False) logits = model.extract_features(**{"source": feats.to(self.device), "padding_mask": padding_mask, "output_layer": 9 if version == "v1" else 12}) feats = model.final_proj(logits[0]) if version == "v1" else logits[0] elif self.embed_suffix == ".onnx": feats = extract_features(model, feats.to(self.device), version).to(self.device) elif self.embed_suffix == ".safetensors": logits = model(feats.to(self.device))["last_hidden_state"] feats = model.final_proj(logits[0]).unsqueeze(0) if version == "v1" else logits else: raise ValueError(translations["option_not_valid"]) feats0 = feats.clone() if protect < 0.5 and pitch_guidance else None if (not isinstance(index, type(None)) and not isinstance(big_npy, type(None)) and index_rate != 0): npy = feats[0].cpu().numpy() if self.is_half: npy = npy.astype(np.float32) score, ix = index.search(npy, k=8) weight = np.square(1 / score) npy = np.sum(big_npy[ix] * np.expand_dims(weight / weight.sum(axis=1, keepdims=True), axis=2), axis=1) if self.is_half: npy = npy.astype(np.float16) feats = (torch.from_numpy(npy).unsqueeze(0).to(self.device) * index_rate + (1 - index_rate) * feats) feats = F.interpolate(feats.permute(0, 2, 1), scale_factor=2).permute(0, 2, 1) p_len = min(audio0.shape[0] // self.window, feats.shape[1]) if pitch_guidance: pitch, pitchf = pitch[:, :p_len], pitchf[:, :p_len] if energy_use: energy = energy[:, :p_len] if feats0 is not None: pitchff = pitchf.clone() pitchff[pitchf > 0] = 1 pitchff[pitchf < 1] = protect pitchff = pitchff.unsqueeze(-1) feats0 = F.interpolate(feats0.permute(0, 2, 1), scale_factor=2).permute(0, 2, 1) feats = (feats * pitchff + feats0 * (1 - pitchff)).to(feats0.dtype) p_len = torch.tensor([p_len], device=self.device).long() feats = feats.half() if self.is_half else feats.float() if not pitch_guidance: pitch, pitchf = None, None else: pitchf = pitchf.half() if self.is_half else pitchf.float() if not energy_use: energy = None else: energy = energy.half() if self.is_half else energy.float() audio1 = ( ( net_g.infer( feats, p_len, pitch, pitchf, sid, energy )[0][0, 0] ).data.cpu().float().numpy() ) if self.suffix == ".pth" else ( net_g.run( [net_g.get_outputs()[0].name], ( get_onnx_argument( net_g, feats, p_len, sid, pitch, pitchf, energy, pitch_guidance, energy_use ) ) )[0][0, 0] ) if self.embed_suffix == ".pt": del padding_mask del feats, feats0, p_len clear_gpu_cache() return audio1 def pipeline(self, logger, model, net_g, sid, audio, f0_up_key, f0_method, file_index, index_rate, pitch_guidance, filter_radius, rms_mix_rate, version, protect, hop_length, f0_autotune, f0_autotune_strength, suffix, embed_suffix, f0_file=None, f0_onnx=False, pbar=None, proposal_pitch=False, proposal_pitch_threshold=255.0, energy_use=False): self.suffix = suffix self.embed_suffix = embed_suffix if file_index != "" and os.path.exists(file_index) and index_rate != 0: try: index = faiss.read_index(file_index) big_npy = index.reconstruct_n(0, index.ntotal) except Exception as e: logger.error(translations["read_faiss_index_error"].format(e=e)) index = big_npy = None else: index = big_npy = None if pbar: pbar.update(1) opt_ts, audio_opt = [], [] audio = signal.filtfilt(bh, ah, audio) audio_pad = np.pad(audio, (self.window // 2, self.window // 2), mode="reflect") if audio_pad.shape[0] > self.t_max: audio_sum = np.zeros_like(audio) for i in range(self.window): audio_sum += audio_pad[i : i - self.window] for t in range(self.t_center, audio.shape[0], self.t_center): opt_ts.append(t - self.t_query + np.where(np.abs(audio_sum[t - self.t_query : t + self.t_query]) == np.abs(audio_sum[t - self.t_query : t + self.t_query]).min())[0][0]) s = 0 t, inp_f0 = None, None audio_pad = np.pad(audio, (self.t_pad, self.t_pad), mode="reflect") sid = torch.tensor(sid, device=self.device).unsqueeze(0).long() p_len = audio_pad.shape[0] // self.window if hasattr(f0_file, "name"): try: with open(f0_file.name, "r") as f: raw_lines = f.read() if len(raw_lines) > 0: inp_f0 = [] for line in raw_lines.strip("\n").split("\n"): inp_f0.append([float(i) for i in line.split(",")]) inp_f0 = np.array(inp_f0, dtype=np.float32) except: logger.error(translations["error_readfile"]) inp_f0 = None if pbar: pbar.update(1) if pitch_guidance: if not hasattr(self, "f0_generator"): self.f0_generator = Generator(self.sample_rate, hop_length, self.f0_min, self.f0_max, self.is_half, self.device, f0_onnx, f0_onnx) pitch, pitchf = self.f0_generator.calculator(self.x_pad, f0_method, audio_pad, f0_up_key, p_len, filter_radius, f0_autotune, f0_autotune_strength, manual_f0=inp_f0, proposal_pitch=proposal_pitch, proposal_pitch_threshold=proposal_pitch_threshold) if self.device == "mps": pitchf = pitchf.astype(np.float32) pitch, pitchf = torch.tensor(pitch[:p_len], device=self.device).unsqueeze(0).long(), torch.tensor(pitchf[:p_len], device=self.device).unsqueeze(0).float() if pbar: pbar.update(1) if energy_use: if not hasattr(self, "rms_extract"): self.rms_extract = RMSEnergyExtractor(frame_length=2048, hop_length=self.window, center=True, pad_mode = "reflect").to(self.device).eval() energy = self.rms_extract(torch.from_numpy(audio_pad).to(self.device).unsqueeze(0)).cpu().numpy() if self.device == "mps": energy = energy.astype(np.float32) energy = torch.tensor(energy[:p_len], device=self.device).unsqueeze(0).float() if pbar: pbar.update(1) for t in opt_ts: t = t // self.window * self.window audio_opt.append( self.voice_conversion( model, net_g, sid, audio_pad[s : t + self.t_pad2 + self.window], pitch[:, s // self.window : (t + self.t_pad2) // self.window] if pitch_guidance else None, pitchf[:, s // self.window : (t + self.t_pad2) // self.window] if pitch_guidance else None, index, big_npy, index_rate, version, protect, energy[:, s // self.window : (t + self.t_pad2) // self.window] if energy_use else None )[self.t_pad_tgt : -self.t_pad_tgt] ) s = t audio_opt.append( self.voice_conversion( model, net_g, sid, audio_pad[t:], (pitch[:, t // self.window :] if t is not None else pitch) if pitch_guidance else None, (pitchf[:, t // self.window :] if t is not None else pitchf) if pitch_guidance else None, index, big_npy, index_rate, version, protect, (energy[:, t // self.window :] if t is not None else energy) if energy_use else None )[self.t_pad_tgt : -self.t_pad_tgt] ) audio_opt = np.concatenate(audio_opt) if pbar: pbar.update(1) if rms_mix_rate != 1: audio_opt = change_rms(audio, self.sample_rate, audio_opt, self.sample_rate, rms_mix_rate) audio_max = np.abs(audio_opt).max() / 0.99 if audio_max > 1: audio_opt /= audio_max if pitch_guidance: del pitch, pitchf del sid clear_gpu_cache() return audio_opt