diff --git a/README.md b/README.md index 4766e8d..f2ae0fb 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,10 @@ This repository is based on [Bert-VITS2](https://github.com/fishaudio/Bert-VITS2 +### 動作環境 + +各UIとAPI Serverにおいて、Windows コマンドプロンプト・WSL2・Linux(Ubuntu Desktop)での動作を確認しています(WSLでのパス指定は相対パスなど工夫ください)。 + ### インストール Windows環境で最近のNVIDIA製グラボがあることを前提にしています。 @@ -83,6 +87,12 @@ model_assets 注意: データセットの手動修正やノイズ除去や、より高品質なデータセットを作りたい場合は、[Aivis](https://github.com/tsukumijima/Aivis)や、そのデータセット部分のWindows対応版 [Aivis Dataset](https://github.com/litagin02/Aivis-Dataset) を使うのをおすすめします。 +### API Server + +構築した環境下で`python server_fastapi.py`するとAPIサーバーが起動します。 + +API仕様は起動後に`/docs`にて確認ください。 + ## Bert-VITS2 v2.1との関係 基本的にはBert-VITS2 v2.1のモデル構造を少し改造しただけです。[事前学習モデル](https://huggingface.co/litagin/Style-Bert-VITS2-1.0-base)も、実質Bert-VITS2 v2.1と同じものを使用しています（不要な重みを削ってsafetensorsに変換したもの）。 @@ -102,7 +112,7 @@ model_assets - [ ] LinuxやWSL等、Windowsの通常環境以外でのサポート？ - [ ] 複数話者学習での音声合成対応（学習は現在でも可能） - [ ] 本家のver 2.1, 2.2, 2.3モデルの推論対応？（ver 2.1以外は明らかにめんどいのでたぶんやらない） -- [ ] `server_fastapi.py`の対応、とくにAPIで使えるようになると嬉しい人が増えるのかもしれない +- [x] `server_fastapi.py`の対応、とくにAPIで使えるようになると嬉しい人が増えるのかもしれない - [ ] モデルのマージで声音と感情表現を混ぜる機能の実装 - [ ] 英語等多言語対応？ diff --git a/app.py b/app.py index 09ced91..673580b 100644 --- a/app.py +++ b/app.py @@ -8,12 +8,24 @@ import gradio as gr import numpy as np import torch from gradio.processing_utils import convert_to_16_bit_wav +from typing import Dict, List import utils from config import config from infer import get_net_g, infer from tools.log import logger +languages = ["JP", "EN", "ZH"] + +DEFAULT_SDP_RATIO: float = 0.2 +DEFAULT_NOISE: float = 0.6 +DEFAULT_NOISEW: float = 0.8 +DEFAULT_LENGTH: float = 1 +DEFAULT_LINE_SPLIT: bool = True +DEFAULT_SPLIT_INTERVAL: float = 0.5 +DEFAULT_STYLE_WEIGHT: float = 0.7 +DEFAULT_EMOTION_WEIGHT: float = 1.0 + class Model: def __init__(self, model_path, config_path, style_vec_path, device): @@ -21,11 +33,9 @@ class Model: self.config_path = config_path self.device = device self.style_vec_path = style_vec_path - self.load() - - def load(self): self.hps = utils.get_hparams_from_file(self.config_path) - self.spk2id = self.hps.data.spk2id + self.spk2id: Dict[str, int] = self.hps.data.spk2id + self.id2spk: Dict[int, str] = {v: k for k, v in self.spk2id.items()} self.num_styles = self.hps.data.num_styles if hasattr(self.hps.data, "style2id"): self.style2id = self.hps.data.style2id @@ -63,17 +73,17 @@ class Model: language="JP", sid=0, reference_audio_path=None, - sdp_ratio=0.2, - noise=0.6, - noisew=0.8, - length=1.0, - line_split=True, - split_interval=0.2, + sdp_ratio=DEFAULT_SDP_RATIO, + noise=DEFAULT_NOISE, + noisew=DEFAULT_NOISEW, + length=DEFAULT_LENGTH, + line_split=DEFAULT_LINE_SPLIT, + split_interval=DEFAULT_SPLIT_INTERVAL, style_text="", - style_weight=0.7, + style_weight=DEFAULT_STYLE_WEIGHT, use_style_text=False, style="0", - emotion_weight=1.0, + emotion_weight=DEFAULT_EMOTION_WEIGHT, ): if reference_audio_path == "": reference_audio_path = None @@ -149,8 +159,8 @@ class ModelHolder: self.refresh() def refresh(self): - self.model_files_dict = {} - self.model_names = [] + self.model_files_dict: Dict[str, List[str]] = {} + self.model_names: List[str] = [] self.current_model = None model_dirs = [ d @@ -370,8 +380,6 @@ if __name__ == "__main__": model_holder = ModelHolder(model_dir, device) - languages = ["JP", "EN", "ZH"] - model_names = model_holder.model_names if len(model_names) == 0: logger.error(f"モデルが見つかりませんでした。{model_dir}にモデルを置いてください。") @@ -401,27 +409,27 @@ if __name__ == "__main__": load_button = gr.Button("ロード", scale=1, variant="primary") text_input = gr.TextArea(label="テキスト", value=initial_text) - line_split = gr.Checkbox(label="改行で分けて生成", value=True) + line_split = gr.Checkbox(label="改行で分けて生成", value=DEFAULT_LINE_SPLIT) split_interval = gr.Slider( minimum=0.0, maximum=2, - value=0.5, + value=DEFAULT_SPLIT_INTERVAL, step=0.1, label="分けた場合に挟む無音の長さ（秒）", ) language = gr.Dropdown(choices=languages, value="JP", label="Language") with gr.Accordion(label="詳細設定", open=False): sdp_ratio = gr.Slider( - minimum=0, maximum=1, value=0.2, step=0.1, label="SDP Ratio" + minimum=0, maximum=1, value=DEFAULT_SDP_RATIO, step=0.1, label="SDP Ratio" ) noise_scale = gr.Slider( - minimum=0.1, maximum=2, value=0.6, step=0.1, label="Noise" + minimum=0.1, maximum=2, value=DEFAULT_NOISE, step=0.1, label="Noise" ) noise_scale_w = gr.Slider( - minimum=0.1, maximum=2, value=0.8, step=0.1, label="Noise_W" + minimum=0.1, maximum=2, value=DEFAULT_NOISEW, step=0.1, label="Noise_W" ) length_scale = gr.Slider( - minimum=0.1, maximum=2, value=1.0, step=0.1, label="Length" + minimum=0.1, maximum=2, value=DEFAULT_LENGTH, step=0.1, label="Length" ) use_style_text = gr.Checkbox(label="Style textを使う", value=False) style_text = gr.Textbox( @@ -433,7 +441,7 @@ if __name__ == "__main__": style_text_weight = gr.Slider( minimum=0, maximum=1, - value=0.7, + value=DEFAULT_STYLE_WEIGHT, step=0.1, label="Style textの強さ", visible=False, @@ -459,7 +467,7 @@ if __name__ == "__main__": style_weight = gr.Slider( minimum=0, maximum=50, - value=1, + value=DEFAULT_EMOTION_WEIGHT, step=0.1, label="スタイルの強さ", ) diff --git a/config.py b/config.py index c270abf..c4478e7 100644 --- a/config.py +++ b/config.py @@ -172,11 +172,13 @@ class Webui_config: class Server_config: def __init__( - self, models: List[Dict[str, any]], port: int = 5000, device: str = "cuda" + self, + port: int = 5000, device: str = "cuda", limit: int = 100, language: str = "JP" ): - self.models: List[Dict[str, any]] = models # 需要加载的所有模型的配置 - self.port: int = port # 端口号 - self.device: str = device # 模型默认使用设备 + self.port: int = port + self.device: str = device + self.language: str = language + self.limit: int = limit @classmethod def from_dict(cls, data: Dict[str, any]): diff --git a/default_config.yml b/default_config.yml index 8720d35..7258ab2 100644 --- a/default_config.yml +++ b/default_config.yml @@ -65,17 +65,8 @@ webui: language_identification_library: "langid" # server_fastapi's config -# TODO: `server_fastapi.py` is not implemented yet for this version server: port: 5000 device: "cuda" - models: - - model: "" - config: "" - device: "cuda" - language: "ZH" - - model: "" - config: "" - device: "cpu" - language: "JP" - speakers: [] + language: "JP" + limit: 100 diff --git a/server_fastapi.py b/server_fastapi.py index 16fa9d3..adfe3f7 100644 --- a/server_fastapi.py +++ b/server_fastapi.py @@ -1,370 +1,163 @@ """ -TODO: This file is not supported in this fork. api服务 多版本多模型 fastapi实现 """ -import logging -import gc -import random - -import librosa -import gradio -import numpy as np -import utils -from fastapi import FastAPI, Query, Request, File, UploadFile, Form +import argparse +from fastapi import FastAPI, Query, Request from fastapi.responses import Response, FileResponse -from fastapi.staticfiles import StaticFiles from io import BytesIO from scipy.io import wavfile import uvicorn import torch -import webbrowser import psutil import GPUtil -from typing import Dict, Optional, List, Set, Union -import os +from typing import Dict, Optional, List, Union +import os, sys from tools.log import logger from urllib.parse import unquote - -from infer import infer, get_net_g, latest_version -import tools.translate as trans -from re_matching import cut_sent - - from config import config +from app import ( + Model, + ModelHolder, + languages, + DEFAULT_SDP_RATIO, + DEFAULT_NOISE, + DEFAULT_NOISEW, + DEFAULT_LENGTH, + DEFAULT_LINE_SPLIT, + DEFAULT_SPLIT_INTERVAL, + DEFAULT_STYLE_WEIGHT, + DEFAULT_EMOTION_WEIGHT, +) +from webui_style_vectors import DEFAULT_EMOTION -os.environ["TOKENIZERS_PARALLELISM"] = "false" +ln = config.server_config.language +available_languages = languages + ["mix", "auto"] - -class Model: - """模型封装类""" - - def __init__(self, config_path: str, model_path: str, device: str, language: str): - self.config_path: str = os.path.normpath(config_path) - self.model_path: str = os.path.normpath(model_path) - self.device: str = device - self.language: str = language - self.hps = utils.get_hparams_from_file(config_path) - self.spk2id: Dict[str, int] = self.hps.data.spk2id # spk - id 映射字典 - self.id2spk: Dict[int, str] = dict() # id - spk 映射字典 - for speaker, speaker_id in self.hps.data.spk2id.items(): - self.id2spk[speaker_id] = speaker - self.version: str = ( - self.hps.version if hasattr(self.hps, "version") else latest_version +def load_models(model_holder: ModelHolder): + model_holder.models = [] + for model_name, model_paths in model_holder.model_files_dict.items(): + model = Model( + model_path=model_paths[0], + config_path=os.path.join(model_holder.root_dir, model_name, "config.json"), + style_vec_path=os.path.join(model_holder.root_dir, model_name, "style_vectors.npy"), + device=model_holder.device, ) - self.net_g = get_net_g( - model_path=model_path, - version=self.version, - device=device, - hps=self.hps, - ) - - def to_dict(self) -> Dict[str, any]: - return { - "config_path": self.config_path, - "model_path": self.model_path, - "device": self.device, - "language": self.language, - "spk2id": self.spk2id, - "id2spk": self.id2spk, - "version": self.version, - } - - -class Models: - def __init__(self): - self.models: Dict[int, Model] = dict() - self.num = 0 - # spkInfo[角色名][模型id] = 角色id - self.spk_info: Dict[str, Dict[int, int]] = dict() - self.path2ids: Dict[str, Set[int]] = dict() # 路径指向的model的id - - def init_model( - self, config_path: str, model_path: str, device: str, language: str - ) -> int: - """ - 初始化并添加一个模型 - - :param config_path: 模型config.json路径 - :param model_path: 模型路径 - :param device: 模型推理使用设备 - :param language: 模型推理默认语言 - """ - # 若文件不存在则不进行加载 - if not os.path.isfile(model_path): - if model_path != "": - logger.warning(f"模型文件{model_path} 不存在，不进行初始化") - return self.num - if not os.path.isfile(config_path): - if config_path != "": - logger.warning(f"配置文件{config_path} 不存在，不进行初始化") - return self.num - - # 若路径中的模型已存在，则不添加模型，若不存在，则进行初始化。 - model_path = os.path.realpath(model_path) - if model_path not in self.path2ids.keys(): - self.path2ids[model_path] = {self.num} - self.models[self.num] = Model( - config_path=config_path, - model_path=model_path, - device=device, - language=language, - ) - logger.success(f"添加模型{model_path}，使用配置文件{os.path.realpath(config_path)}") - else: - # 获取一个指向id - m_id = next(iter(self.path2ids[model_path])) - self.models[self.num] = self.models[m_id] - self.path2ids[model_path].add(self.num) - logger.success("模型已存在，添加模型引用。") - # 添加角色信息 - for speaker, speaker_id in self.models[self.num].spk2id.items(): - if speaker not in self.spk_info.keys(): - self.spk_info[speaker] = {self.num: speaker_id} - else: - self.spk_info[speaker][self.num] = speaker_id - # 修改计数 - self.num += 1 - return self.num - 1 - - def del_model(self, index: int) -> Optional[int]: - """删除对应序号的模型，若不存在则返回None""" - if index not in self.models.keys(): - return None - # 删除角色信息 - for speaker, speaker_id in self.models[index].spk2id.items(): - self.spk_info[speaker].pop(index) - if len(self.spk_info[speaker]) == 0: - # 若对应角色的所有模型都被删除，则清除该角色信息 - self.spk_info.pop(speaker) - # 删除路径信息 - model_path = os.path.realpath(self.models[index].model_path) - self.path2ids[model_path].remove(index) - if len(self.path2ids[model_path]) == 0: - self.path2ids.pop(model_path) - logger.success(f"删除模型{model_path}, id = {index}") - else: - logger.success(f"删除模型引用{model_path}, id = {index}") - # 删除模型 - self.models.pop(index) - gc.collect() - if torch.cuda.is_available(): - torch.cuda.empty_cache() - return index - - def get_models(self): - """获取所有模型""" - return self.models + model.load_net_g() + model_holder.models.append(model) if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--cpu", action="store_true", help="Use CPU instead of GPU") + parser.add_argument( + "--dir", "-d", type=str, help="Model directory", default=config.out_dir + ) + args = parser.parse_args() + + if args.cpu: + device = "cpu" + else: + device = "cuda" if torch.cuda.is_available() else "cpu" + + model_dir = args.dir + model_holder = ModelHolder(model_dir, device) + if len(model_holder.model_names) == 0: + logger.error(f"Models not found in {model_dir}.") + sys.exit(1) + + logger.info('Loading models...') + load_models(model_holder) + limit = config.server_config.limit app = FastAPI() app.logger = logger - # 挂载静态文件 - logger.info("开始挂载网页页面") - StaticDir: str = "./Web" - if not os.path.isdir(StaticDir): - logger.warning( - "缺少网页资源，无法开启网页页面，如有需要请在 https://github.com/jiangyuxiaoxiao/Bert-VITS2-UI 或者Bert-VITS对应版本的release页面下载" - ) - else: - dirs = [fir.name for fir in os.scandir(StaticDir) if fir.is_dir()] - files = [fir.name for fir in os.scandir(StaticDir) if fir.is_dir()] - for dirName in dirs: - app.mount( - f"/{dirName}", - StaticFiles(directory=f"./{StaticDir}/{dirName}"), - name=dirName, - ) - loaded_models = Models() - # 加载模型 - logger.info("开始加载模型") - models_info = config.server_config.models - for model_info in models_info: - loaded_models.init_model( - config_path=model_info["config"], - model_path=model_info["model"], - device=model_info["device"], - language=model_info["language"], - ) - - @app.get("/") - async def index(): - return FileResponse("./Web/index.html") async def _voice( text: str, - model_id: int, - speaker_name: str, - speaker_id: int, - sdp_ratio: float, - noise: float, - noisew: float, - length: float, - language: str, - auto_translate: bool, - auto_split: bool, - emotion: Optional[Union[int, str]] = None, - reference_audio=None, + model_id: int = 0, + speaker_name: str = None, + speaker_id: int = 0, + sdp_ratio: float = DEFAULT_SDP_RATIO, + noise: float = DEFAULT_NOISE, + noisew: float = DEFAULT_NOISEW, + length: float = DEFAULT_LENGTH, + language: str = ln, + auto_split: bool = DEFAULT_LINE_SPLIT, + split_interval: float = DEFAULT_SPLIT_INTERVAL, style_text: Optional[str] = None, - style_weight: float = 0.7, + style_weight: float = DEFAULT_STYLE_WEIGHT, + emotion: Optional[Union[int, str]] = DEFAULT_EMOTION, + emotion_weight: float = DEFAULT_EMOTION_WEIGHT, + reference_audio_path: str = None, ) -> Union[Response, Dict[str, any]]: - """TTS实现函数""" - # 检查模型是否存在 - if model_id not in loaded_models.models.keys(): - logger.error(f"/voice 请求错误：模型model_id={model_id}未加载") - return {"status": 10, "detail": f"模型model_id={model_id}未加载"} - # 检查是否提供speaker - if speaker_name is None and speaker_id is None: - logger.error("/voice 请求错误：推理请求未提供speaker_name或speaker_id") - return {"status": 11, "detail": "请提供speaker_name或speaker_id"} - elif speaker_name is None: - # 检查speaker_id是否存在 - if speaker_id not in loaded_models.models[model_id].id2spk.keys(): - logger.error(f"/voice 请求错误：角色speaker_id={speaker_id}不存在") - return {"status": 12, "detail": f"角色speaker_id={speaker_id}不存在"} - speaker_name = loaded_models.models[model_id].id2spk[speaker_id] - # 检查speaker_name是否存在 - if speaker_name not in loaded_models.models[model_id].spk2id.keys(): - logger.error(f"/voice 请求错误：角色speaker_name={speaker_name}不存在") - return {"status": 13, "detail": f"角色speaker_name={speaker_name}不存在"} - # 未传入则使用默认语言 - if language is None: - language = loaded_models.models[model_id].language - # 翻译会破坏mix结构，auto也会变得无意义。不要在这两个模式下使用 - if auto_translate: - if language == "auto" or language == "mix": - logger.error( - f"/voice 请求错误：请勿同时使用language = {language}与auto_translate模式" - ) - return { - "status": 20, - "detail": f"请勿同时使用language = {language}与auto_translate模式", - } - text = trans.translate(Sentence=text, to_Language=language.lower()) - if reference_audio is not None: - ref_audio = BytesIO(await reference_audio.read()) - # 2.2 适配 - if loaded_models.models[model_id].version == "2.2": - ref_audio, _ = librosa.load(ref_audio, 48000) + if model_id >= len(model_holder.models): + return {"status": 10, "detail": f"model_id={model_id} not found"} + elif len(text) > limit: + return {"status": 9, "detail": f"too long text: over {limit}"} + model = model_holder.models[model_id] + if speaker_name is None: + if speaker_id is None: + return {"status": 11, "detail": "Required speaker_name or speaker_id"} + if speaker_id not in model.id2spk.keys(): + return {"status": 12, "detail": f"peaker_id={speaker_id} not found"} else: - ref_audio = reference_audio - if not auto_split: - with torch.no_grad(): - audio = infer( - text=text, - sdp_ratio=sdp_ratio, - noise_scale=noise, - noise_scale_w=noisew, - length_scale=length, - sid=speaker_name, - language=language, - hps=loaded_models.models[model_id].hps, - net_g=loaded_models.models[model_id].net_g, - device=loaded_models.models[model_id].device, - emotion=emotion, - reference_audio=ref_audio, - style_text=style_text, - style_weight=style_weight, - ) - audio = gradio.processing_utils.convert_to_16_bit_wav(audio) - else: - texts = cut_sent(text) - audios = [] - with torch.no_grad(): - for t in texts: - audios.append( - infer( - text=t, - sdp_ratio=sdp_ratio, - noise_scale=noise, - noise_scale_w=noisew, - length_scale=length, - sid=speaker_name, - language=language, - hps=loaded_models.models[model_id].hps, - net_g=loaded_models.models[model_id].net_g, - device=loaded_models.models[model_id].device, - emotion=emotion, - reference_audio=ref_audio, - style_text=style_text, - style_weight=style_weight, - ) - ) - audios.append(np.zeros(int(44100 * 0.2))) - audio = np.concatenate(audios) - audio = gradio.processing_utils.convert_to_16_bit_wav(audio) - with BytesIO() as wavContent: - wavfile.write( - wavContent, loaded_models.models[model_id].hps.data.sampling_rate, audio - ) - response = Response(content=wavContent.getvalue(), media_type="audio/wav") - return response - - @app.post("/voice") - async def voice( - request: Request, # fastapi自动注入 - text: str = Form(...), - model_id: int = Query(..., description="模型ID"), # 模型序号 - speaker_name: str = Query( - None, description="说话人名" - ), # speaker_name与 speaker_id二者选其一 - speaker_id: int = Query(None, description="说话人id，与speaker_name二选一"), - sdp_ratio: float = Query(0.2, description="SDP/DP混合比"), - noise: float = Query(0.2, description="感情"), - noisew: float = Query(0.9, description="音素长度"), - length: float = Query(1, description="语速"), - language: str = Query(None, description="语言"), # 若不指定使用语言则使用默认值 - auto_translate: bool = Query(False, description="自动翻译"), - auto_split: bool = Query(False, description="自动切分"), - emotion: Optional[Union[int, str]] = Query(None, description="emo"), - reference_audio: UploadFile = File(None), - style_text: Optional[str] = Form(None, description="风格文本"), - style_weight: float = Query(0.7, description="风格权重"), - ): - """语音接口，若需要上传参考音频请仅使用post请求""" - logger.info( - f"{request.client.host}:{request.client.port}/voice { unquote(str(request.query_params) )} text={text}" - ) - return await _voice( + if speaker_name not in model.spk2id.keys(): + return {"status": 13, "detail": f"speaker_name={speaker_name} not found"} + speaker_id = model.spk2id[speaker_name] + if emotion not in model.style2id.keys(): + return {"status": 14, "detail": f"emotion={speaker_name} not found"} + if language not in available_languages: + language = ln + sr, audio = model.infer( text=text, - model_id=model_id, - speaker_name=speaker_name, - speaker_id=speaker_id, + language=language, + sid=speaker_id, + reference_audio_path=reference_audio_path, sdp_ratio=sdp_ratio, noise=noise, noisew=noisew, length=length, - language=language, - auto_translate=auto_translate, - auto_split=auto_split, - emotion=emotion, - reference_audio=reference_audio, + line_split=auto_split, + split_interval=split_interval, style_text=style_text, style_weight=style_weight, + use_style_text=bool(style_text), + style=emotion, + emotion_weight=emotion_weight, ) + with BytesIO() as wavContent: + wavfile.write( + wavContent, sr, audio + ) + response = Response(content=wavContent.getvalue(), media_type="audio/wav") + return response @app.get("/voice") async def voice( - request: Request, # fastapi自动注入 - text: str = Query(..., description="输入文字"), - model_id: int = Query(..., description="模型ID"), # 模型序号 + request: Request, + text: str = Query(..., description="セリフ"), + model_id: int = Query(0, description="モデルID。`GET /models/info`のkeyの値を指定ください。"), speaker_name: str = Query( - None, description="说话人名" - ), # speaker_name与 speaker_id二者选其一 - speaker_id: int = Query(None, description="说话人id，与speaker_name二选一"), - sdp_ratio: float = Query(0.2, description="SDP/DP混合比"), - noise: float = Query(0.2, description="感情"), - noisew: float = Query(0.9, description="音素长度"), - length: float = Query(1, description="语速"), - language: str = Query(None, description="语言"), # 若不指定使用语言则使用默认值 - auto_translate: bool = Query(False, description="自动翻译"), - auto_split: bool = Query(False, description="自动切分"), - emotion: Optional[Union[int, str]] = Query(None, description="emo"), - style_text: Optional[str] = Query(None, description="风格文本"), - style_weight: float = Query(0.7, description="风格权重"), + None, description="話者名(speaker_idより優先)。esd.listの2列目記載の文字列を指定。" + ), + speaker_id: int = Query( + 0, description="話者ID。model_assets.[model].config.json内のspk2idを確認。"), + sdp_ratio: float = Query(DEFAULT_SDP_RATIO, description="SDP(Stochastic Duration Predictor)/DP混合比。比率が高くなるほど、トーンのばらつきが大きくなる。"), + noise: float = Query(DEFAULT_NOISE, description="サンプルノイズの割合。大きくするほどランダム性が高まる"), + noisew: float = Query(DEFAULT_NOISEW, description="SDPノイズ。大きくするほど発音の間隔にばらつきが出やすくなる。"), + length: float = Query(DEFAULT_LENGTH, description="話速。基準は1で大きくするほど音声は長くなり読み上げが遅まる。"), + language: str = Query(ln, description=f"{'/'.join(available_languages)}のいずれか。"), + auto_split: bool = Query(True, description="改行で分けて生成"), + split_interval: float = Query(DEFAULT_SPLIT_INTERVAL, description="分けた場合に挟む無音の長さ（秒）"), + style_text: Optional[str] = Query( + None, description="このテキストの読み上げと似た声音・感情になりやすくなる。ただし抑揚やテンポ等が犠牲になる傾向がある。" + ), + style_weight: float = Query(DEFAULT_STYLE_WEIGHT, description="style_textの強さ"), + emotion: Optional[Union[int, str]] = Query(DEFAULT_EMOTION, description="スタイル"), + emotion_weight: float = Query(DEFAULT_EMOTION_WEIGHT, description="emotionの強さ"), + reference_audio_path: Optional[str] = Query(None, description="emotionを音声ファイルで行う"), ): - """语音接口""" + """Infer text to speech(テキストから感情付き音声を生成する)""" logger.info( f"{request.client.host}:{request.client.port}/voice { unquote(str(request.query_params) )}" ) @@ -378,158 +171,41 @@ if __name__ == "__main__": noisew=noisew, length=length, language=language, - auto_translate=auto_translate, auto_split=auto_split, - emotion=emotion, + split_interval=split_interval, style_text=style_text, style_weight=style_weight, + emotion=emotion, + emotion_weight=emotion_weight, + reference_audio_path=reference_audio_path, ) @app.get("/models/info") def get_loaded_models_info(request: Request): - """获取已加载模型信息""" + """ロードされたモデル情報の取得""" result: Dict[str, Dict] = dict() - for key, model in loaded_models.models.items(): - result[str(key)] = model.to_dict() - return result - - @app.get("/models/delete") - def delete_model( - request: Request, model_id: int = Query(..., description="删除模型id") - ): - """删除指定模型""" - logger.info( - f"{request.client.host}:{request.client.port}/models/delete { unquote(str(request.query_params) )}" - ) - result = loaded_models.del_model(model_id) - if result is None: - logger.error(f"/models/delete 模型删除错误：模型{model_id}不存在，删除失败") - return {"status": 14, "detail": f"模型{model_id}不存在，删除失败"} - - return {"status": 0, "detail": "删除成功"} - - @app.get("/models/add") - def add_model( - request: Request, - model_path: str = Query(..., description="添加模型路径"), - config_path: str = Query( - None, description="添加模型配置文件路径，不填则使用./config.json或../config.json" - ), - device: str = Query("cuda", description="推理使用设备"), - language: str = Query("ZH", description="模型默认语言"), - ): - """添加指定模型：允许重复添加相同路径模型，且不重复占用内存""" - logger.info( - f"{request.client.host}:{request.client.port}/models/add { unquote(str(request.query_params) )}" - ) - if config_path is None: - model_dir = os.path.dirname(model_path) - if os.path.isfile(os.path.join(model_dir, "config.json")): - config_path = os.path.join(model_dir, "config.json") - elif os.path.isfile(os.path.join(model_dir, "../config.json")): - config_path = os.path.join(model_dir, "../config.json") - else: - logger.error("/models/add 模型添加失败：未在模型所在目录以及上级目录找到config.json文件") - return { - "status": 15, - "detail": "查询未传入配置文件路径，同时默认路径./与../中不存在配置文件config.json。", - } - try: - model_id = loaded_models.init_model( - config_path=config_path, - model_path=model_path, - device=device, - language=language, - ) - except Exception: - logging.exception("模型加载出错") - return { - "status": 16, - "detail": "模型加载出错，详细查看日志", + for model_id, model in enumerate(model_holder.models): + result[str(model_id)] = { + "config_path": model.config_path, + "model_path": model.model_path, + "device": model.device, + "spk2id": model.spk2id, + "id2spk": model.id2spk, + "style2id": model.style2id, } - return { - "status": 0, - "detail": "模型添加成功", - "Data": { - "model_id": model_id, - "model_info": loaded_models.models[model_id].to_dict(), - }, - } - - def _get_all_models(root_dir: str = "Data", only_unloaded: bool = False): - """从root_dir搜索获取所有可用模型""" - result: Dict[str, List[str]] = dict() - files = os.listdir(root_dir) + ["."] - for file in files: - if os.path.isdir(os.path.join(root_dir, file)): - sub_dir = os.path.join(root_dir, file) - # 搜索 "sub_dir" 、 "sub_dir/models" 两个路径 - result[file] = list() - sub_files = os.listdir(sub_dir) - model_files = [] - for sub_file in sub_files: - relpath = os.path.realpath(os.path.join(sub_dir, sub_file)) - if only_unloaded and relpath in loaded_models.path2ids.keys(): - continue - if sub_file.endswith(".pth") and sub_file.startswith("G_"): - if os.path.isfile(relpath): - model_files.append(sub_file) - # 对模型文件按步数排序 - model_files = sorted( - model_files, - key=lambda pth: int(pth.lstrip("G_").rstrip(".pth")) - if pth.lstrip("G_").rstrip(".pth").isdigit() - else 10**10, - ) - result[file] = model_files - models_dir = os.path.join(sub_dir, "models") - model_files = [] - if os.path.isdir(models_dir): - sub_files = os.listdir(models_dir) - for sub_file in sub_files: - relpath = os.path.realpath(os.path.join(models_dir, sub_file)) - if only_unloaded and relpath in loaded_models.path2ids.keys(): - continue - if sub_file.endswith(".pth") and sub_file.startswith("G_"): - if os.path.isfile(os.path.join(models_dir, sub_file)): - model_files.append(f"models/{sub_file}") - # 对模型文件按步数排序 - model_files = sorted( - model_files, - key=lambda pth: int(pth.lstrip("models/G_").rstrip(".pth")) - if pth.lstrip("models/G_").rstrip(".pth").isdigit() - else 10**10, - ) - result[file] += model_files - if len(result[file]) == 0: - result.pop(file) - return result - @app.get("/models/get_unloaded") - def get_unloaded_models_info( - request: Request, root_dir: str = Query("Data", description="搜索根目录") - ): - """获取未加载模型""" - logger.info( - f"{request.client.host}:{request.client.port}/models/get_unloaded { unquote(str(request.query_params) )}" - ) - return _get_all_models(root_dir, only_unloaded=True) - - @app.get("/models/get_local") - def get_local_models_info( - request: Request, root_dir: str = Query("Data", description="搜索根目录") - ): - """获取全部本地模型""" - logger.info( - f"{request.client.host}:{request.client.port}/models/get_local { unquote(str(request.query_params) )}" - ) - return _get_all_models(root_dir, only_unloaded=False) + @app.get("/models/refresh") + def refresh(): + """モデルをパスに追加/削除した際などに読み込ませる""" + model_holder.refresh() + load_models(model_holder) + return {} @app.get("/status") def get_status(): - """获取电脑运行状态""" + """実行環境のステータスを取得""" cpu_percent = psutil.cpu_percent(interval=1) memory_info = psutil.virtual_memory() memory_total = memory_info.total @@ -563,104 +239,9 @@ if __name__ == "__main__": "gpu": gpuInfo, } - @app.get("/tools/translate") - def translate( - request: Request, - texts: str = Query(..., description="待翻译文本"), - to_language: str = Query(..., description="翻译目标语言"), - ): - """翻译""" - logger.info( - f"{request.client.host}:{request.client.port}/tools/translate { unquote(str(request.query_params) )}" - ) - return {"texts": trans.translate(Sentence=texts, to_Language=to_language)} - - all_examples: Dict[str, Dict[str, List]] = dict() # 存放示例 - - @app.get("/tools/random_example") - def random_example( - request: Request, - language: str = Query(None, description="指定语言，未指定则随机返回"), - root_dir: str = Query("Data", description="搜索根目录"), - ): - """ - 获取一个随机音频+文本，用于对比，音频会从本地目录随机选择。 - """ - logger.info( - f"{request.client.host}:{request.client.port}/tools/random_example { unquote(str(request.query_params) )}" - ) - global all_examples - # 数据初始化 - if root_dir not in all_examples.keys(): - all_examples[root_dir] = {"ZH": [], "JP": [], "EN": []} - - examples = all_examples[root_dir] - - # 从项目Data目录中搜索train/val.list - for root, directories, _files in os.walk(root_dir): - for file in _files: - if file in ["train.list", "val.list"]: - with open( - os.path.join(root, file), mode="r", encoding="utf-8" - ) as f: - lines = f.readlines() - for line in lines: - data = line.split("|") - if len(data) != 7: - continue - # 音频存在 且语言为ZH/EN/JP - if os.path.isfile(data[0]) and data[2] in [ - "ZH", - "JP", - "EN", - ]: - examples[data[2]].append( - { - "text": data[3], - "audio": data[0], - "speaker": data[1], - } - ) - - examples = all_examples[root_dir] - if language is None: - if len(examples["ZH"]) + len(examples["JP"]) + len(examples["EN"]) == 0: - return {"status": 17, "detail": "没有加载任何示例数据"} - else: - # 随机选一个 - rand_num = random.randint( - 0, - len(examples["ZH"]) + len(examples["JP"]) + len(examples["EN"]) - 1, - ) - # ZH - if rand_num < len(examples["ZH"]): - return {"status": 0, "Data": examples["ZH"][rand_num]} - # JP - if rand_num < len(examples["ZH"]) + len(examples["JP"]): - return { - "status": 0, - "Data": examples["JP"][rand_num - len(examples["ZH"])], - } - # EN - return { - "status": 0, - "Data": examples["EN"][ - rand_num - len(examples["ZH"]) - len(examples["JP"]) - ], - } - - else: - if len(examples[language]) == 0: - return {"status": 17, "detail": f"没有加载任何{language}数据"} - return { - "status": 0, - "Data": examples[language][ - random.randint(0, len(examples[language]) - 1) - ], - } - @app.get("/tools/get_audio") - def get_audio(request: Request, path: str = Query(..., description="本地音频路径")): + def get_audio(request: Request, path: str = Query(..., description="local wav path")): + """wavデータを取得する""" logger.info( f"{request.client.host}:{request.client.port}/tools/get_audio { unquote(str(request.query_params) )}" ) @@ -672,10 +253,9 @@ if __name__ == "__main__": return {"status": 19, "detail": "非wav格式文件"} return FileResponse(path=path) - logger.warning("本地服务，请勿将服务端口暴露于外网") - logger.info(f"api文档地址 http://127.0.0.1:{config.server_config.port}/docs") - if os.path.isdir(StaticDir): - webbrowser.open(f"http://127.0.0.1:{config.server_config.port}") + + logger.info(f"server listen: http://127.0.0.1:{config.server_config.port}") + logger.info(f"API docs: http://127.0.0.1:{config.server_config.port}/docs") uvicorn.run( app, port=config.server_config.port, host="0.0.0.0", log_level="warning" ) diff --git a/webui_style_vectors.py b/webui_style_vectors.py index b8696e3..71a5296 100644 --- a/webui_style_vectors.py +++ b/webui_style_vectors.py @@ -9,6 +9,7 @@ from sklearn.manifold import TSNE from config import config MAX_CLUSTER_NUM = 10 +DEFAULT_EMOTION: str = 'Neutral' tsne = TSNE(n_components=2, random_state=42, metric="cosine") @@ -132,7 +133,7 @@ def save_only_mean(model_name): with open(config_path, "r") as f: json_dict = json.load(f) json_dict["data"]["num_styles"] = 1 - json_dict["data"]["style2id"] = {"Neutral": 0} + json_dict["data"]["style2id"] = {DEFAULT_EMOTION: 0} with open(config_path, "w") as f: json.dump(json_dict, f, indent=2) return f"成功!\n{style_vector_path}に保存し{config_path}を更新しました。" @@ -152,7 +153,7 @@ def save_style_vectors(model_name, style_names: str): config_path = os.path.join(result_dir, "config.json") if not os.path.exists(config_path): return f"{config_path}が存在しません。" - style_name_list = ["Neutral"] + style_name_list = [DEFAULT_EMOTION] style_name_list = style_name_list + style_names.split(",") if len(style_name_list) != len(centroids) + 1: return f"スタイルの数が合いません。`,`で正しく{len(centroids)}個に区切られているか確認してください: {style_names}" @@ -201,7 +202,7 @@ def save_style_vectors_from_files(model_name, audio_files_text, style_names_text config_path = os.path.join(result_dir, "config.json") if not os.path.exists(config_path): return f"{config_path}が存在しません。" - style_name_list = ["Neutral"] + style_name_list = [DEFAULT_EMOTION] style_name_list = style_name_list + style_names assert len(style_name_list) == len(style_vectors) @@ -216,7 +217,7 @@ def save_style_vectors_from_files(model_name, audio_files_text, style_names_text return f"成功!\n{style_vector_path}に保存し{config_path}を更新しました。" -initial_md = """ +initial_md = f""" # Style Bert-VITS2 スタイルベクトルの作成 Style-Bert-VITS2で音声合成するには、スタイルベクトルのファイル`style_vectors.npy`が必要です。これをモデルごとに作成する必要があります。 @@ -225,7 +226,7 @@ Style-Bert-VITS2で音声合成するには、スタイルベクトルのファ ## 方法 どうやってスタイルベクトルファイルを作るかはいくつか方法があります。 -- 方法1: めんどくさいから平均スタイルのみを使う（使えるスタイルは標準のNeutralのみ） +- 方法1: めんどくさいから平均スタイルのみを使う（使えるスタイルは標準の{DEFAULT_EMOTION}のみ） - 方法2: 音声ファイルを自動でスタイル別に分け、その各スタイルの平均を取って保存 - 方法3: スタイルを代表する音声ファイルを手動で選んで、その音声のスタイルベクトルを保存 - 方法4: 自分でもっと頑張ってこだわって作る（JVNVコーパスなど、もともとスタイルラベル等が利用可能な場合はこれがよいかも） @@ -233,7 +234,7 @@ Style-Bert-VITS2で音声合成するには、スタイルベクトルのファ 基本的には方法2を使うことを、めんどくさかったりあまり感情に幅がないデータセットなら方法1をおすすめします。 """ -method2 = """ +method2 = f""" 学習の時に取り出したスタイルベクトルを読み込んで、可視化を見ながらスタイルを分けていきます。 手順: @@ -245,102 +246,103 @@ method2 = """ 詳細: スタイルベクトル(256次元)たちを適当なアルゴリズムでクラスタリングして、各クラスタの中心のベクトル（と全体の平均ベクトル）を保存します。 -平均スタイル（Neutral）は自動的に保存されます。 +平均スタイル（{DEFAULT_EMOTION}）は自動的に保存されます。 """ -with gr.Blocks(theme="NoCrypt/miku") as app: - gr.Markdown(initial_md) - with gr.Row(): - model_name = gr.Textbox("your_model_name", label="モデル名") - load_button = gr.Button("スタイルベクトルを読み込む", variant="primary") - output = gr.Plot(label="音声スタイルの可視化") - load_button.click(load, inputs=[model_name], outputs=[output]) - with gr.Tab("方法1: 平均スタイルのみを保存"): - gr.Markdown("平均（Neutral）スタイルのみを保存する場合は、以下のボタンを押してください。") +if __name__ == "__main__": + with gr.Blocks(theme="NoCrypt/miku") as app: + gr.Markdown(initial_md) with gr.Row(): - save_button1 = gr.Button("スタイルベクトルを保存", variant="primary") - info1 = gr.Textbox(label="保存結果") - save_button1.click(save_only_mean, inputs=[model_name], outputs=[info1]) - with gr.Tab("方法2: スタイル分けを自動で行う"): - n_clusters = gr.Slider( - minimum=2, - maximum=10, - step=1, - value=4, - label="作るスタイルの数（平均スタイルを除く）", - info="上の図を見ながらスタイルの数を試行錯誤してください。", - ) - c_method = gr.Radio( - choices=[ - "Agglomerative after t-SNE", - "KMeans after t-SNE", - "Agglomerative", - "KMeans", - ], - label="アルゴリズム", - info="分類する（クラスタリング）アルゴリズムを選択します。いろいろ試してみてください。", - value="Agglomerative after t-SNE", - ) - c_button = gr.Button("スタイル分けを実行") - audio_list = [] - md_list = [] - gr.Markdown("スタイル分けの結果と、各スタイルの特徴的な代表音声（図の黒い x 印）") - gr.Markdown("注意: もともと256次元なものをを2次元に落としているので、正確なベクトルの位置関係ではありません。") - with gr.Row(): - gr_plot = gr.Plot() + model_name = gr.Textbox("your_model_name", label="モデル名") + load_button = gr.Button("スタイルベクトルを読み込む", variant="primary") + output = gr.Plot(label="音声スタイルの可視化") + load_button.click(load, inputs=[model_name], outputs=[output]) + with gr.Tab("方法1: 平均スタイルのみを保存"): + gr.Markdown(f"平均（{DEFAULT_EMOTION}）スタイルのみを保存する場合は、以下のボタンを押してください。") with gr.Row(): - for i in range(MAX_CLUSTER_NUM): - with gr.Column(): - md_list.append(gr.Markdown(visible=False)) - audio_list.append( - gr.Audio( - visible=False, - scale=1, - show_label=True, - label=f"スタイル{i+1}", + save_button1 = gr.Button("スタイルベクトルを保存", variant="primary") + info1 = gr.Textbox(label="保存結果") + save_button1.click(save_only_mean, inputs=[model_name], outputs=[info1]) + with gr.Tab("方法2: スタイル分けを自動で行う"): + n_clusters = gr.Slider( + minimum=2, + maximum=10, + step=1, + value=4, + label="作るスタイルの数（平均スタイルを除く）", + info="上の図を見ながらスタイルの数を試行錯誤してください。", + ) + c_method = gr.Radio( + choices=[ + "Agglomerative after t-SNE", + "KMeans after t-SNE", + "Agglomerative", + "KMeans", + ], + label="アルゴリズム", + info="分類する（クラスタリング）アルゴリズムを選択します。いろいろ試してみてください。", + value="Agglomerative after t-SNE", + ) + c_button = gr.Button("スタイル分けを実行") + audio_list = [] + md_list = [] + gr.Markdown("スタイル分けの結果と、各スタイルの特徴的な代表音声（図の黒い x 印）") + gr.Markdown("注意: もともと256次元なものをを2次元に落としているので、正確なベクトルの位置関係ではありません。") + with gr.Row(): + gr_plot = gr.Plot() + with gr.Row(): + for i in range(MAX_CLUSTER_NUM): + with gr.Column(): + md_list.append(gr.Markdown(visible=False)) + audio_list.append( + gr.Audio( + visible=False, + scale=1, + show_label=True, + label=f"スタイル{i+1}", + ) ) - ) - c_button.click( - do_clustering_gradio, - inputs=[n_clusters, c_method], - outputs=[gr_plot] + audio_list + md_list, - ) - gr.Markdown("結果が良さそうなら、これを保存します。") - style_names = gr.Textbox( - "Angry, Sad, Happy", - label="スタイルの名前", - info="スタイルの名前を`,`で区切って入力してください（日本語可）。例: `Angry, Sad, Happy`や`怒り, 悲しみ, 喜び`など。平均音声はNeutralとして自動的に保存されます。", - ) - with gr.Row(): - save_button = gr.Button("スタイルベクトルを保存", variant="primary") - info2 = gr.Textbox(label="保存結果") + c_button.click( + do_clustering_gradio, + inputs=[n_clusters, c_method], + outputs=[gr_plot] + audio_list + md_list, + ) + gr.Markdown("結果が良さそうなら、これを保存します。") + style_names = gr.Textbox( + "Angry, Sad, Happy", + label="スタイルの名前", + info=f"スタイルの名前を`,`で区切って入力してください（日本語可）。例: `Angry, Sad, Happy`や`怒り, 悲しみ, 喜び`など。平均音声は{DEFAULT_EMOTION}として自動的に保存されます。", + ) + with gr.Row(): + save_button = gr.Button("スタイルベクトルを保存", variant="primary") + info2 = gr.Textbox(label="保存結果") - save_button.click( - save_style_vectors, inputs=[model_name, style_names], outputs=[info2] - ) - with gr.Tab("方法3: 手動でスタイルを選ぶ"): - gr.Markdown("下のテキスト欄に、各スタイルの代表音声のファイル名を`,`区切りで、その横に対応するスタイル名を`,`区切りで入力してください。") - gr.Markdown("例: `angry.wav, sad.wav, happy.wav`と`Angry, Sad, Happy`") - gr.Markdown("注意: Neutralスタイルは自動的に保存されます、手動ではNeutralという名前のスタイルは指定しないでください。") - with gr.Row(): - audio_files_text = gr.Textbox( - label="音声ファイル名", placeholder="angry.wav, sad.wav, happy.wav" + save_button.click( + save_style_vectors, inputs=[model_name, style_names], outputs=[info2] ) - style_names_text = gr.Textbox( - label="スタイル名", placeholder="Angry, Sad, Happy" + with gr.Tab("方法3: 手動でスタイルを選ぶ"): + gr.Markdown("下のテキスト欄に、各スタイルの代表音声のファイル名を`,`区切りで、その横に対応するスタイル名を`,`区切りで入力してください。") + gr.Markdown("例: `angry.wav, sad.wav, happy.wav`と`Angry, Sad, Happy`") + gr.Markdown(f"注意: {DEFAULT_EMOTION}スタイルは自動的に保存されます、手動では{DEFAULT_EMOTION}という名前のスタイルは指定しないでください。") + with gr.Row(): + audio_files_text = gr.Textbox( + label="音声ファイル名", placeholder="angry.wav, sad.wav, happy.wav" + ) + style_names_text = gr.Textbox( + label="スタイル名", placeholder="Angry, Sad, Happy" + ) + with gr.Row(): + save_button3 = gr.Button("スタイルベクトルを保存", variant="primary") + info3 = gr.Textbox(label="保存結果") + save_button3.click( + save_style_vectors_from_files, + inputs=[model_name, audio_files_text, style_names_text], + outputs=[info3], + ) + with gr.Tab("方法4: がんばる"): + gr.Markdown( + "`clustering.ipynb`にjvnvコーパスの場合の作り方とかクラスタ分けのいろいろを書いています。これを参考に自分で頑張って作ってください。" ) - with gr.Row(): - save_button3 = gr.Button("スタイルベクトルを保存", variant="primary") - info3 = gr.Textbox(label="保存結果") - save_button3.click( - save_style_vectors_from_files, - inputs=[model_name, audio_files_text, style_names_text], - outputs=[info3], - ) - with gr.Tab("方法4: がんばる"): - gr.Markdown( - "`clustering.ipynb`にjvnvコーパスの場合の作り方とかクラスタ分けのいろいろを書いています。これを参考に自分で頑張って作ってください。" - ) -app.launch(inbrowser=True) + app.launch(inbrowser=True)