import gc import math import os import random import sys import traceback from copy import deepcopy from time import time as ttime from typing import Generator, List, Tuple, Union import ffmpeg import librosa import numpy as np import torch import torch.nn.functional as F import yaml from tqdm import tqdm from transformers import AutoModelForMaskedLM, AutoTokenizer from AR.models.t2s_lightning_module import Text2SemanticLightningModule from feature_extractor.cnhubert import CNHubert from module.mel_processing import spectrogram_torch from module.models import SynthesizerTrn from tools.i18n.i18n import I18nAuto, scan_language_list from tools.my_utils import load_audio from TTS_infer_pack.text_segmentation_method import splits from TTS_infer_pack.TextPreprocessor import TextPreprocessor now_dir = os.getcwd() sys.path.append(now_dir) language = os.environ.get("language", "Auto") language = sys.argv[-1] if sys.argv[-1] in scan_language_list() else language i18n = I18nAuto(language=language) # configs/tts_infer.yaml """ custom: bert_base_path: GPT_SoVITS/pretrained_models/chinese-roberta-wwm-ext-large cnhuhbert_base_path: GPT_SoVITS/pretrained_models/chinese-hubert-base device: cpu is_half: false t2s_weights_path: GPT_SoVITS/pretrained_models/gsv-v2final-pretrained/s1bert25hz-5kh-longer-epoch=12-step=369668.ckpt vits_weights_path: GPT_SoVITS/pretrained_models/gsv-v2final-pretrained/s2G2333k.pth version: v2 default: bert_base_path: GPT_SoVITS/pretrained_models/chinese-roberta-wwm-ext-large cnhuhbert_base_path: GPT_SoVITS/pretrained_models/chinese-hubert-base device: cpu is_half: false t2s_weights_path: GPT_SoVITS/pretrained_models/s1bert25hz-2kh-longer-epoch=68e-step=50232.ckpt vits_weights_path: GPT_SoVITS/pretrained_models/s2G488k.pth version: v1 default_v2: bert_base_path: GPT_SoVITS/pretrained_models/chinese-roberta-wwm-ext-large cnhuhbert_base_path: GPT_SoVITS/pretrained_models/chinese-hubert-base device: cuda is_half: false t2s_weights_path: GPT_weights_v2/21hr-e15.ckpt version: v2 vits_weights_path: SoVITS_weights_v2/21hr_e13_s10621.pth """ def set_seed(seed: int): seed = int(seed) seed = seed if seed != -1 else random.randrange(1 << 32) print(f"Set seed to {seed}") os.environ['PYTHONHASHSEED'] = str(seed) random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) try: if torch.cuda.is_available(): torch.cuda.manual_seed(seed) torch.cuda.manual_seed_all(seed) # torch.backends.cudnn.deterministic = True # torch.backends.cudnn.benchmark = False # torch.backends.cudnn.enabled = True # 开启后会影响精度 torch.backends.cuda.matmul.allow_tf32 = False torch.backends.cudnn.allow_tf32 = False except: pass return seed class TTS_Config: default_configs = { "default": { "device": "cpu", "is_half": False, "version": "v1", "t2s_weights_path": "GPT_SoVITS/pretrained_models/s1bert25hz-2kh-longer-epoch=68e-step=50232.ckpt", "vits_weights_path": "GPT_SoVITS/pretrained_models/s2G488k.pth", "cnhuhbert_base_path": "GPT_SoVITS/pretrained_models/chinese-hubert-base", "bert_base_path": "GPT_SoVITS/pretrained_models/chinese-roberta-wwm-ext-large", }, "default_v2": { "device": "cpu", "is_half": False, "version": "v2", "t2s_weights_path": "GPT_SoVITS/pretrained_models/gsv-v2final-pretrained/s1bert25hz-5kh-longer-epoch=12-step=369668.ckpt", "vits_weights_path": "GPT_SoVITS/pretrained_models/gsv-v2final-pretrained/s2G2333k.pth", "cnhuhbert_base_path": "GPT_SoVITS/pretrained_models/chinese-hubert-base", "bert_base_path": "GPT_SoVITS/pretrained_models/chinese-roberta-wwm-ext-large", }, } configs: dict = None v1_languages: list = ["auto", "en", "zh", "ja", "all_zh", "all_ja"] v2_languages: list = ["auto", "auto_yue", "en", "zh", "ja", "yue", "ko", "all_zh", "all_ja", "all_yue", "all_ko"] languages: list = v2_languages # "all_zh",#全部按中文识别 # "en",#全部按英文识别#######不变 # "all_ja",#全部按日文识别 # "all_yue",#全部按中文识别 # "all_ko",#全部按韩文识别 # "zh",#按中英混合识别####不变 # "ja",#按日英混合识别####不变 # "yue",#按粤英混合识别####不变 # "ko",#按韩英混合识别####不变 # "auto",#多语种启动切分识别语种 # "auto_yue",#多语种启动切分识别语种 def __init__(self, configs: Union[dict, str] = None): # 设置默认配置文件路径 configs_base_path: str = "GPT_SoVITS/configs/" os.makedirs(configs_base_path, exist_ok=True) self.configs_path: str = os.path.join( configs_base_path, "tts_infer.yaml") if configs in ["", None]: if not os.path.exists(self.configs_path): self.save_configs() print(f"Create default config file at {self.configs_path}") configs: dict = deepcopy(self.default_configs) if isinstance(configs, str): self.configs_path = configs configs: dict = self._load_configs(self.configs_path) assert isinstance(configs, dict) version = configs.get("version", "v2").lower() assert version in ["v1", "v2"] self.default_configs["default"] = configs.get( "default", self.default_configs["default"]) self.default_configs["default_v2"] = configs.get( "default_v2", self.default_configs["default_v2"]) default_config_key = "default"if version == "v1" else "default_v2" self.configs: dict = configs.get("custom", deepcopy( self.default_configs[default_config_key])) self.device = self.configs.get("device", torch.device("cpu")) self.is_half = self.configs.get("is_half", False) self.version = version self.t2s_weights_path = self.configs.get("t2s_weights_path", None) self.vits_weights_path = self.configs.get("vits_weights_path", None) self.bert_base_path = self.configs.get("bert_base_path", None) self.cnhuhbert_base_path = self.configs.get( "cnhuhbert_base_path", None) self.languages = self.v2_languages if self.version == "v2" else self.v1_languages if (self.t2s_weights_path in [None, ""]) or (not os.path.exists(self.t2s_weights_path)): self.t2s_weights_path = self.default_configs[default_config_key]['t2s_weights_path'] print(f"fall back to default t2s_weights_path: {self.t2s_weights_path}") if (self.vits_weights_path in [None, ""]) or (not os.path.exists(self.vits_weights_path)): self.vits_weights_path = self.default_configs[default_config_key]['vits_weights_path'] print(f"fall back to default vits_weights_path: {self.vits_weights_path}") if (self.bert_base_path in [None, ""]) or (not os.path.exists(self.bert_base_path)): self.bert_base_path = self.default_configs[default_config_key]['bert_base_path'] print(f"fall back to default bert_base_path: {self.bert_base_path}") if (self.cnhuhbert_base_path in [None, ""]) or (not os.path.exists(self.cnhuhbert_base_path)): self.cnhuhbert_base_path = self.default_configs[default_config_key]['cnhuhbert_base_path'] print(f"fall back to default cnhuhbert_base_path: {self.cnhuhbert_base_path}") self.update_configs() self.max_sec = None self.hz: int = 50 self.semantic_frame_rate: str = "25hz" self.segment_size: int = 20480 self.filter_length: int = 2048 self.sampling_rate: int = 32000 self.hop_length: int = 640 self.win_length: int = 2048 self.n_speakers: int = 300 def _load_configs(self, configs_path: str) -> dict: if os.path.exists(configs_path): ... else: print(i18n("路径不存在,使用默认配置")) self.save_configs(configs_path) with open(configs_path, 'r') as f: configs = yaml.load(f, Loader=yaml.FullLoader) return configs def save_configs(self, configs_path: str = None) -> None: configs = deepcopy(self.default_configs) if self.configs is not None: configs["custom"] = self.update_configs() if configs_path is None: configs_path = self.configs_path with open(configs_path, 'w') as f: yaml.dump(configs, f) def update_configs(self): self.config = { "device": str(self.device), "is_half": self.is_half, "version": self.version, "t2s_weights_path": self.t2s_weights_path, "vits_weights_path": self.vits_weights_path, "bert_base_path": self.bert_base_path, "cnhuhbert_base_path": self.cnhuhbert_base_path, } return self.config def update_version(self, version: str) -> None: self.version = version self.languages = self.v2_languages if self.version == "v2" else self.v1_languages def __str__(self): self.configs = self.update_configs() string = "TTS Config".center(100, '-') + '\n' for k, v in self.configs.items(): string += f"{str(k).ljust(20)}: {str(v)}\n" string += "-" * 100 + '\n' return string def __repr__(self): return self.__str__() def __hash__(self): return hash(self.configs_path) def __eq__(self, other): return isinstance(other, TTS_Config) and self.configs_path == other.configs_path class TTS: def __init__(self, configs: Union[dict, str, TTS_Config]): if isinstance(configs, TTS_Config): self.configs = configs else: self.configs: TTS_Config = TTS_Config(configs) self.t2s_model: Text2SemanticLightningModule = None self.vits_model: SynthesizerTrn = None self.bert_tokenizer: AutoTokenizer = None self.bert_model: AutoModelForMaskedLM = None self.cnhuhbert_model: CNHubert = None self._init_models() self.text_preprocessor: TextPreprocessor = \ TextPreprocessor(self.bert_model, self.bert_tokenizer, self.configs.device) self.prompt_cache: dict = { "ref_audio_path": None, "prompt_semantic": None, "refer_spec": [], "prompt_text": None, "prompt_lang": None, "phones": None, "bert_features": None, "norm_text": None, "aux_ref_audio_paths": [], } self.stop_flag: bool = False self.precision: torch.dtype = torch.float16 if self.configs.is_half else torch.float32 def _init_models(self,): self.init_t2s_weights(self.configs.t2s_weights_path) self.init_vits_weights(self.configs.vits_weights_path) self.init_bert_weights(self.configs.bert_base_path) self.init_cnhuhbert_weights(self.configs.cnhuhbert_base_path) # self.enable_half_precision(self.configs.is_half) def init_cnhuhbert_weights(self, base_path: str): print(f"Loading CNHuBERT weights from {base_path}") self.cnhuhbert_model = CNHubert(base_path) self.cnhuhbert_model = self.cnhuhbert_model.eval() self.cnhuhbert_model = self.cnhuhbert_model.to(self.configs.device) if self.configs.is_half and str(self.configs.device) != "cpu": self.cnhuhbert_model = self.cnhuhbert_model.half() def init_bert_weights(self, base_path: str): print(f"Loading BERT weights from {base_path}") self.bert_tokenizer = AutoTokenizer.from_pretrained(base_path) self.bert_model = AutoModelForMaskedLM.from_pretrained(base_path) self.bert_model = self.bert_model.eval() self.bert_model = self.bert_model.to(self.configs.device) if self.configs.is_half and str(self.configs.device) != "cpu": self.bert_model = self.bert_model.half() def init_vits_weights(self, weights_path: str): print(f"Loading VITS weights from {weights_path}") self.configs.vits_weights_path = weights_path dict_s2 = torch.load(weights_path, map_location=self.configs.device) hps = dict_s2["config"] if dict_s2['weight']['enc_p.text_embedding.weight'].shape[0] == 322: self.configs.update_version("v1") else: self.configs.update_version("v2") self.configs.save_configs() hps["model"]["version"] = self.configs.version self.configs.filter_length = hps["data"]["filter_length"] self.configs.segment_size = hps["train"]["segment_size"] self.configs.sampling_rate = hps["data"]["sampling_rate"] self.configs.hop_length = hps["data"]["hop_length"] self.configs.win_length = hps["data"]["win_length"] self.configs.n_speakers = hps["data"]["n_speakers"] self.configs.semantic_frame_rate = "25hz" kwargs = hps["model"] vits_model = SynthesizerTrn( self.configs.filter_length // 2 + 1, self.configs.segment_size // self.configs.hop_length, n_speakers=self.configs.n_speakers, **kwargs ) if hasattr(vits_model, "enc_q"): del vits_model.enc_q vits_model = vits_model.to(self.configs.device) vits_model = vits_model.eval() vits_model.load_state_dict(dict_s2["weight"], strict=False) self.vits_model = vits_model if self.configs.is_half and str(self.configs.device) != "cpu": self.vits_model = self.vits_model.half() def init_t2s_weights(self, weights_path: str): print(f"Loading Text2Semantic weights from {weights_path}") self.configs.t2s_weights_path = weights_path self.configs.save_configs() self.configs.hz = 50 dict_s1 = torch.load(weights_path, map_location=self.configs.device) config = dict_s1["config"] self.configs.max_sec = config["data"]["max_sec"] t2s_model = Text2SemanticLightningModule( config, "****", is_train=False) t2s_model.load_state_dict(dict_s1["weight"]) t2s_model = t2s_model.to(self.configs.device) t2s_model = t2s_model.eval() self.t2s_model = t2s_model if self.configs.is_half and str(self.configs.device) != "cpu": self.t2s_model = self.t2s_model.half() def enable_half_precision(self, enable: bool = True, save: bool = True): ''' To enable half precision for the TTS model. Args: enable: bool, whether to enable half precision. ''' if str(self.configs.device) == "cpu" and enable: print("Half precision is not supported on CPU.") return self.configs.is_half = enable self.precision = torch.float16 if enable else torch.float32 if save: self.configs.save_configs() if enable: if self.t2s_model is not None: self.t2s_model = self.t2s_model.half() if self.vits_model is not None: self.vits_model = self.vits_model.half() if self.bert_model is not None: self.bert_model = self.bert_model.half() if self.cnhuhbert_model is not None: self.cnhuhbert_model = self.cnhuhbert_model.half() else: if self.t2s_model is not None: self.t2s_model = self.t2s_model.float() if self.vits_model is not None: self.vits_model = self.vits_model.float() if self.bert_model is not None: self.bert_model = self.bert_model.float() if self.cnhuhbert_model is not None: self.cnhuhbert_model = self.cnhuhbert_model.float() def set_device(self, device: torch.device, save: bool = True): ''' To set the device for all models. Args: device: torch.device, the device to use for all models. ''' self.configs.device = device if save: self.configs.save_configs() if self.t2s_model is not None: self.t2s_model = self.t2s_model.to(device) if self.vits_model is not None: self.vits_model = self.vits_model.to(device) if self.bert_model is not None: self.bert_model = self.bert_model.to(device) if self.cnhuhbert_model is not None: self.cnhuhbert_model = self.cnhuhbert_model.to(device) def set_ref_audio(self, ref_audio_path: str): ''' To set the reference audio for the TTS model, including the prompt_semantic and refer_spepc. Args: ref_audio_path: str, the path of the reference audio. ''' self._set_prompt_semantic(ref_audio_path) self._set_ref_spec(ref_audio_path) self._set_ref_audio_path(ref_audio_path) def _set_ref_audio_path(self, ref_audio_path): self.prompt_cache["ref_audio_path"] = ref_audio_path def _set_ref_spec(self, ref_audio_path): spec = self._get_ref_spec(ref_audio_path) if self.prompt_cache["refer_spec"] in [[], None]: self.prompt_cache["refer_spec"] = [spec] else: self.prompt_cache["refer_spec"][0] = spec def _get_ref_spec(self, ref_audio_path): audio = load_audio(ref_audio_path, int(self.configs.sampling_rate)) audio = torch.FloatTensor(audio) maxx = audio.abs().max() if (maxx > 1): audio /= min(2, maxx) audio_norm = audio audio_norm = audio_norm.unsqueeze(0) spec = spectrogram_torch( audio_norm, self.configs.filter_length, self.configs.sampling_rate, self.configs.hop_length, self.configs.win_length, center=False, ) spec = spec.to(self.configs.device) if self.configs.is_half: spec = spec.half() return spec def _set_prompt_semantic(self, ref_wav_path: str): zero_wav = np.zeros( int(self.configs.sampling_rate * 0.3), dtype=np.float16 if self.configs.is_half else np.float32, ) with torch.no_grad(): wav16k, sr = librosa.load(ref_wav_path, sr=16000) if (wav16k.shape[0] > 160000 or wav16k.shape[0] < 48000): raise OSError(i18n("参考音频在3~10秒范围外，请更换！")) wav16k = torch.from_numpy(wav16k) zero_wav_torch = torch.from_numpy(zero_wav) wav16k = wav16k.to(self.configs.device) zero_wav_torch = zero_wav_torch.to(self.configs.device) if self.configs.is_half: wav16k = wav16k.half() zero_wav_torch = zero_wav_torch.half() wav16k = torch.cat([wav16k, zero_wav_torch]) hubert_feature = self.cnhuhbert_model.model(wav16k.unsqueeze(0))[ "last_hidden_state" ].transpose( 1, 2 ) # .float() codes = self.vits_model.extract_latent(hubert_feature) prompt_semantic = codes[0, 0].to(self.configs.device) self.prompt_cache["prompt_semantic"] = prompt_semantic def batch_sequences(self, sequences: List[torch.Tensor], axis: int = 0, pad_value: int = 0, max_length: int = None): seq = sequences[0] ndim = seq.dim() if axis < 0: axis += ndim dtype: torch.dtype = seq.dtype pad_value = torch.tensor(pad_value, dtype=dtype) seq_lengths = [seq.shape[axis] for seq in sequences] if max_length is None: max_length = max(seq_lengths) else: max_length = max(seq_lengths) if max_length < max( seq_lengths) else max_length padded_sequences = [] for seq, length in zip(sequences, seq_lengths): padding = [0] * axis + [0, max_length - length] + [0] * (ndim - axis - 1) padded_seq = torch.nn.functional.pad(seq, padding, value=pad_value) padded_sequences.append(padded_seq) batch = torch.stack(padded_sequences) return batch def to_batch(self, data: list, prompt_data: dict = None, batch_size: int = 5, threshold: float = 0.75, split_bucket: bool = True, device: torch.device = torch.device("cpu"), precision: torch.dtype = torch.float32, ): _data: list = [] index_and_len_list = [] for idx, item in enumerate(data): norm_text_len = len(item["norm_text"]) index_and_len_list.append([idx, norm_text_len]) batch_index_list = [] if split_bucket: index_and_len_list.sort(key=lambda x: x[1]) index_and_len_list = np.array(index_and_len_list, dtype=np.int64) batch_index_list_len = 0 pos = 0 while pos < index_and_len_list.shape[0]: # batch_index_list.append(index_and_len_list[pos:min(pos+batch_size,len(index_and_len_list))]) pos_end = min(pos+batch_size, index_and_len_list.shape[0]) while pos < pos_end: batch = index_and_len_list[pos:pos_end, 1].astype( np.float32) score = batch[(pos_end-pos)//2]/(batch.mean()+1e-8) if (score >= threshold) or (pos_end-pos == 1): batch_index = index_and_len_list[pos:pos_end, 0].tolist( ) batch_index_list_len += len(batch_index) batch_index_list.append(batch_index) pos = pos_end break pos_end = pos_end-1 assert batch_index_list_len == len(data) else: for i in range(len(data)): if i % batch_size == 0: batch_index_list.append([]) batch_index_list[-1].append(i) for batch_idx, index_list in enumerate(batch_index_list): item_list = [data[idx] for idx in index_list] phones_list = [] phones_len_list = [] # bert_features_list = [] all_phones_list = [] all_phones_len_list = [] all_bert_features_list = [] norm_text_batch = [] all_bert_max_len = 0 all_phones_max_len = 0 for item in item_list: if prompt_data is not None: all_bert_features = torch.cat([prompt_data["bert_features"], item["bert_features"]], 1)\ .to(dtype=precision, device=device) all_phones = torch.LongTensor( prompt_data["phones"]+item["phones"]).to(device) phones = torch.LongTensor(item["phones"]).to(device) # norm_text = prompt_data["norm_text"]+item["norm_text"] else: all_bert_features = item["bert_features"]\ .to(dtype=precision, device=device) phones = torch.LongTensor(item["phones"]).to(device) all_phones = phones # norm_text = item["norm_text"] all_bert_max_len = max( all_bert_max_len, all_bert_features.shape[-1]) all_phones_max_len = max( all_phones_max_len, all_phones.shape[-1]) phones_list.append(phones) phones_len_list.append(phones.shape[-1]) all_phones_list.append(all_phones) all_phones_len_list.append(all_phones.shape[-1]) all_bert_features_list.append(all_bert_features) norm_text_batch.append(item["norm_text"]) phones_batch = phones_list all_phones_batch = all_phones_list all_bert_features_batch = all_bert_features_list max_len = max(all_bert_max_len, all_phones_max_len) # phones_batch = self.batch_sequences(phones_list, axis=0, pad_value=0, max_length=max_len) # 直接对phones和bert_features进行pad。（padding策略会影响T2S模型生成的结果，但不直接影响复读概率。影响复读概率的主要因素是mask的策略） # all_phones_batch = self.batch_sequences(all_phones_list, axis=0, pad_value=0, max_length=max_len) # all_bert_features_batch = all_bert_features_list # all_bert_features_batch = torch.zeros((len(all_bert_features_list), 1024, max_len), dtype=precision, device=device) # for idx, item in enumerate(all_bert_features_list): # all_bert_features_batch[idx, :, : item.shape[-1]] = item # #### 先对phones进行embedding、对bert_features进行project，再pad到相同长度，（padding策略会影响T2S模型生成的结果，但不直接影响复读概率。影响复读概率的主要因素是mask的策略） # all_phones_list = [self.t2s_model.model.ar_text_embedding(item.to(self.t2s_model.device)) for item in all_phones_list] # all_phones_list = [F.pad(item,(0,0,0,max_len-item.shape[0]),value=0) for item in all_phones_list] # all_phones_batch = torch.stack(all_phones_list, dim=0) # all_bert_features_list = [self.t2s_model.model.bert_proj(item.to(self.t2s_model.device).transpose(0, 1)) for item in all_bert_features_list] # all_bert_features_list = [F.pad(item,(0,0,0,max_len-item.shape[0]), value=0) for item in all_bert_features_list] # all_bert_features_batch = torch.stack(all_bert_features_list, dim=0) batch = { "phones": phones_batch, "phones_len": torch.LongTensor(phones_len_list).to(device), "all_phones": all_phones_batch, "all_phones_len": torch.LongTensor(all_phones_len_list).to(device), "all_bert_features": all_bert_features_batch, "norm_text": norm_text_batch, "max_len": max_len, } _data.append(batch) return _data, batch_index_list def recovery_order(self, data: list, batch_index_list: list) -> list: ''' Recovery the order of the audio according to the batch_index_list. Args: data (List[list(np.ndarray)]): the out of order audio . batch_index_list (List[list[int]]): the batch index list. Returns: list (List[np.ndarray]): the data in the original order. ''' length = len(sum(batch_index_list, [])) _data = [None]*length for i, index_list in enumerate(batch_index_list): for j, index in enumerate(index_list): _data[index] = data[i][j] return _data def stop(self,): ''' Stop the inference process. ''' self.stop_flag = True @torch.no_grad() def run(self, inputs: dict): """ Text to speech inference. Args: inputs (dict): { "text": "", # str.(required) text to be synthesized "text_lang: "", # str.(required) language of the text to be synthesized "ref_audio_path": "", # str.(required) reference audio path "aux_ref_audio_paths": [], # list.(optional) auxiliary reference audio paths for multi-speaker tone fusion "prompt_text": "", # str.(optional) prompt text for the reference audio "prompt_lang": "", # str.(required) language of the prompt text for the reference audio "top_k": 5, # int. top k sampling "top_p": 1, # float. top p sampling "temperature": 1, # float. temperature for sampling "text_split_method": "cut0", # str. text split method, see text_segmentation_method.py for details. "batch_size": 1, # int. batch size for inference "batch_threshold": 0.75, # float. threshold for batch splitting. "split_bucket: True, # bool. whether to split the batch into multiple buckets. "return_fragment": False, # bool. step by step return the audio fragment. "speed_factor":1.0, # float. control the speed of the synthesized audio. "fragment_interval":0.3, # float. to control the interval of the audio fragment. "seed": -1, # int. random seed for reproducibility. "parallel_infer": True, # bool. whether to use parallel inference. "repetition_penalty": 1.35 # float. repetition penalty for T2S model. } returns: Tuple[int, np.ndarray]: sampling rate and audio data. """ ########## variables initialization ########### self.stop_flag: bool = False text: str = inputs.get("text", "") text_lang: str = inputs.get("text_lang", "") ref_audio_path: str = inputs.get("ref_audio_path", "") aux_ref_audio_paths: list = inputs.get("aux_ref_audio_paths", []) prompt_text: str = inputs.get("prompt_text", "") prompt_lang: str = inputs.get("prompt_lang", "") top_k: int = inputs.get("top_k", 5) top_p: float = inputs.get("top_p", 1) temperature: float = inputs.get("temperature", 1) text_split_method: str = inputs.get("text_split_method", "cut0") batch_size = inputs.get("batch_size", 1) batch_threshold = inputs.get("batch_threshold", 0.75) speed_factor = inputs.get("speed_factor", 1.0) split_bucket = inputs.get("split_bucket", True) return_fragment = inputs.get("return_fragment", False) fragment_interval = inputs.get("fragment_interval", 0.3) seed = inputs.get("seed", -1) seed = -1 if seed in ["", None] else seed actual_seed = set_seed(seed) parallel_infer = inputs.get("parallel_infer", True) repetition_penalty = inputs.get("repetition_penalty", 1.35) if parallel_infer: print(i18n("并行推理模式已开启")) self.t2s_model.model.infer_panel = self.t2s_model.model.infer_panel_batch_infer else: print(i18n("并行推理模式已关闭")) self.t2s_model.model.infer_panel = self.t2s_model.model.infer_panel_naive_batched if return_fragment: print(i18n("分段返回模式已开启")) if split_bucket: split_bucket = False print(i18n("分段返回模式不支持分桶处理，已自动关闭分桶处理")) if split_bucket and speed_factor == 1.0: print(i18n("分桶处理模式已开启")) elif speed_factor != 1.0: print(i18n("语速调节不支持分桶处理，已自动关闭分桶处理")) split_bucket = False else: print(i18n("分桶处理模式已关闭")) if fragment_interval < 0.01: fragment_interval = 0.01 print(i18n("分段间隔过小，已自动设置为0.01")) no_prompt_text = False if prompt_text in [None, ""]: no_prompt_text = True assert text_lang in self.configs.languages if not no_prompt_text: assert prompt_lang in self.configs.languages if ref_audio_path in [None, ""] and \ ((self.prompt_cache["prompt_semantic"] is None) or (self.prompt_cache["refer_spec"] in [None, []])): raise ValueError( "ref_audio_path cannot be empty, when the reference audio is not set using set_ref_audio()") ###### setting reference audio and prompt text preprocessing ######## t0 = ttime() if (ref_audio_path is not None) and (ref_audio_path != self.prompt_cache["ref_audio_path"]): if not os.path.exists(ref_audio_path): raise ValueError(f"{ref_audio_path} not exists") self.set_ref_audio(ref_audio_path) aux_ref_audio_paths = aux_ref_audio_paths if aux_ref_audio_paths is not None else [] paths = set(aux_ref_audio_paths) & set( self.prompt_cache["aux_ref_audio_paths"]) if not (len(list(paths)) == len(aux_ref_audio_paths) == len(self.prompt_cache["aux_ref_audio_paths"])): self.prompt_cache["aux_ref_audio_paths"] = aux_ref_audio_paths self.prompt_cache["refer_spec"] = [ self.prompt_cache["refer_spec"][0]] for path in aux_ref_audio_paths: if path in [None, ""]: continue if not os.path.exists(path): print(i18n("音频文件不存在，跳过：{}").format(path)) continue self.prompt_cache["refer_spec"].append( self._get_ref_spec(path)) if not no_prompt_text: prompt_text = prompt_text.strip("\n") if (prompt_text[-1] not in splits): prompt_text += "。" if prompt_lang != "en" else "." print(i18n("实际输入的参考文本:"), prompt_text) if self.prompt_cache["prompt_text"] != prompt_text: self.prompt_cache["prompt_text"] = prompt_text self.prompt_cache["prompt_lang"] = prompt_lang phones, bert_features, norm_text = \ self.text_preprocessor.segment_and_extract_feature_for_text( prompt_text, prompt_lang, self.configs.version) self.prompt_cache["phones"] = phones self.prompt_cache["bert_features"] = bert_features self.prompt_cache["norm_text"] = norm_text ###### text preprocessing ######## t1 = ttime() data: list = None if not return_fragment: data = self.text_preprocessor.preprocess( text, text_lang, text_split_method, self.configs.version) if len(data) == 0: yield self.configs.sampling_rate, np.zeros(int(self.configs.sampling_rate), dtype=np.int16) return batch_index_list: list = None data, batch_index_list = self.to_batch(data, prompt_data=self.prompt_cache if not no_prompt_text else None, batch_size=batch_size, threshold=batch_threshold, split_bucket=split_bucket, device=self.configs.device, precision=self.precision ) else: print(i18n("############ 切分文本 ############")) texts = self.text_preprocessor.pre_seg_text( text, text_lang, text_split_method) data = [] for i in range(len(texts)): if i % batch_size == 0: data.append([]) data[-1].append(texts[i]) def make_batch(batch_texts): batch_data = [] print(i18n("############ 提取文本Bert特征 ############")) for text in tqdm(batch_texts): phones, bert_features, norm_text = self.text_preprocessor.segment_and_extract_feature_for_text( text, text_lang, self.configs.version) if phones is None: continue res = { "phones": phones, "bert_features": bert_features, "norm_text": norm_text, } batch_data.append(res) if len(batch_data) == 0: return None batch, _ = self.to_batch(batch_data, prompt_data=self.prompt_cache if not no_prompt_text else None, batch_size=batch_size, threshold=batch_threshold, split_bucket=False, device=self.configs.device, precision=self.precision ) return batch[0] t2 = ttime() try: print("############ 推理 ############") ###### inference ###### t_34 = 0.0 t_45 = 0.0 audio = [] for item in data: t3 = ttime() if return_fragment: item = make_batch(item) if item is None: continue batch_phones: List[torch.LongTensor] = item["phones"] # batch_phones:torch.LongTensor = item["phones"] batch_phones_len: torch.LongTensor = item["phones_len"] all_phoneme_ids: torch.LongTensor = item["all_phones"] all_phoneme_lens: torch.LongTensor = item["all_phones_len"] all_bert_features: torch.LongTensor = item["all_bert_features"] norm_text: str = item["norm_text"] max_len = item["max_len"] print(i18n("前端处理后的文本(每句):"), norm_text) if no_prompt_text: prompt = None else: prompt = self.prompt_cache["prompt_semantic"].expand( len(all_phoneme_ids), -1).to(self.configs.device) pred_semantic_list, idx_list = self.t2s_model.model.infer_panel( all_phoneme_ids, all_phoneme_lens, prompt, all_bert_features, # prompt_phone_len=ph_offset, top_k=top_k, top_p=top_p, temperature=temperature, early_stop_num=self.configs.hz * self.configs.max_sec, max_len=max_len, repetition_penalty=repetition_penalty, ) t4 = ttime() t_34 += t4 - t3 refer_audio_spec: torch.Tensor = [item.to( dtype=self.precision, device=self.configs.device) for item in self.prompt_cache["refer_spec"]] batch_audio_fragment = [] # ## vits并行推理 method 1 # pred_semantic_list = [item[-idx:] for item, idx in zip(pred_semantic_list, idx_list)] # pred_semantic_len = torch.LongTensor([item.shape[0] for item in pred_semantic_list]).to(self.configs.device) # pred_semantic = self.batch_sequences(pred_semantic_list, axis=0, pad_value=0).unsqueeze(0) # max_len = 0 # for i in range(0, len(batch_phones)): # max_len = max(max_len, batch_phones[i].shape[-1]) # batch_phones = self.batch_sequences(batch_phones, axis=0, pad_value=0, max_length=max_len) # batch_phones = batch_phones.to(self.configs.device) # batch_audio_fragment = (self.vits_model.batched_decode( # pred_semantic, pred_semantic_len, batch_phones, batch_phones_len,refer_audio_spec # )) if speed_factor == 1.0: # ## vits并行推理 method 2 pred_semantic_list = [ item[-idx:] for item, idx in zip(pred_semantic_list, idx_list)] upsample_rate = math.prod(self.vits_model.upsample_rates) audio_frag_idx = [pred_semantic_list[i].shape[0]*2 * upsample_rate for i in range(0, len(pred_semantic_list))] audio_frag_end_idx = [sum(audio_frag_idx[:i+1]) for i in range(0, len(audio_frag_idx))] all_pred_semantic = torch.cat(pred_semantic_list).unsqueeze( 0).unsqueeze(0).to(self.configs.device) _batch_phones = torch.cat(batch_phones).unsqueeze( 0).to(self.configs.device) _batch_audio_fragment = (self.vits_model.decode( all_pred_semantic, _batch_phones, refer_audio_spec, speed=speed_factor ).detach()[0, 0, :]) audio_frag_end_idx.insert(0, 0) batch_audio_fragment = [_batch_audio_fragment[audio_frag_end_idx[i-1]:audio_frag_end_idx[i]] for i in range(1, len(audio_frag_end_idx))] else: # ## vits串行推理 for i, idx in enumerate(idx_list): phones = batch_phones[i].unsqueeze( 0).to(self.configs.device) # .unsqueeze(0)#mq要多unsqueeze一次 _pred_semantic = ( pred_semantic_list[i][-idx:].unsqueeze(0).unsqueeze(0)) audio_fragment = (self.vits_model.decode( _pred_semantic, phones, refer_audio_spec, speed=speed_factor ).detach()[0, 0, :]) batch_audio_fragment.append( audio_fragment ) # 试试重建不带上prompt部分 t5 = ttime() t_45 += t5 - t4 if return_fragment: print("%.3f\t%.3f\t%.3f\t%.3f" % (t1 - t0, t2 - t1, t4 - t3, t5 - t4)) yield self.audio_postprocess([batch_audio_fragment], self.configs.sampling_rate, None, speed_factor, False, fragment_interval ) else: audio.append(batch_audio_fragment) if self.stop_flag: yield self.configs.sampling_rate, np.zeros(int(self.configs.sampling_rate), dtype=np.int16) return if not return_fragment: print("%.3f\t%.3f\t%.3f\t%.3f" % (t1 - t0, t2 - t1, t_34, t_45)) if len(audio) == 0: yield self.configs.sampling_rate, np.zeros(int(self.configs.sampling_rate), dtype=np.int16) return yield self.audio_postprocess(audio, self.configs.sampling_rate, batch_index_list, speed_factor, split_bucket, fragment_interval ) except Exception as e: traceback.print_exc() # 必须返回一个空音频, 否则会导致显存不释放。 yield self.configs.sampling_rate, np.zeros(int(self.configs.sampling_rate), dtype=np.int16) # 重置模型, 否则会导致显存释放不完全。 del self.t2s_model del self.vits_model self.t2s_model = None self.vits_model = None self.init_t2s_weights(self.configs.t2s_weights_path) self.init_vits_weights(self.configs.vits_weights_path) raise e finally: self.empty_cache() def empty_cache(self): try: gc.collect() # 触发gc的垃圾回收。避免内存一直增长。 if "cuda" in str(self.configs.device): torch.cuda.empty_cache() elif str(self.configs.device) == "mps": torch.mps.empty_cache() except: pass def audio_postprocess(self, audio: List[torch.Tensor], sr: int, batch_index_list: list = None, speed_factor: float = 1.0, split_bucket: bool = True, fragment_interval: float = 0.3 ) -> Tuple[int, np.ndarray]: zero_wav = torch.zeros( int(self.configs.sampling_rate * fragment_interval), dtype=self.precision, device=self.configs.device ) for i, batch in enumerate(audio): for j, audio_fragment in enumerate(batch): max_audio = torch.abs(audio_fragment).max() # 简单防止16bit爆音 if max_audio > 1: audio_fragment /= max_audio audio_fragment: torch.Tensor = torch.cat( [audio_fragment, zero_wav], dim=0) audio[i][j] = audio_fragment.cpu().numpy() if split_bucket: audio = self.recovery_order(audio, batch_index_list) else: # audio = [item for batch in audio for item in batch] audio = sum(audio, []) audio = np.concatenate(audio, 0) audio = (audio * 32768).astype(np.int16) # try: # if speed_factor != 1.0: # audio = speed_change(audio, speed=speed_factor, sr=int(sr)) # except Exception as e: # print(f"Failed to change speed of audio: \n{e}") return sr, audio def speed_change(input_audio: np.ndarray, speed: float, sr: int): # 将 NumPy 数组转换为原始 PCM 流 raw_audio = input_audio.astype(np.int16).tobytes() # 设置 ffmpeg 输入流 input_stream = ffmpeg.input( 'pipe:', format='s16le', acodec='pcm_s16le', ar=str(sr), ac=1) # 变速处理 output_stream = input_stream.filter('atempo', speed) # 输出流到管道 out, _ = ( output_stream.output('pipe:', format='s16le', acodec='pcm_s16le') .run(input=raw_audio, capture_stdout=True, capture_stderr=True) ) # 将管道输出解码为 NumPy 数组 processed_audio = np.frombuffer(out, np.int16) return processed_audio