From f19b76b26fea863ef66130222b217043a3d31400 Mon Sep 17 00:00:00 2001 From: spawner Date: Fri, 7 Mar 2025 14:56:45 +0800 Subject: [PATCH 1/3] Add files via upload --- api_role.py | 825 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 825 insertions(+) create mode 100644 api_role.py diff --git a/api_role.py b/api_role.py new file mode 100644 index 0000000..54250f4 --- /dev/null +++ b/api_role.py @@ -0,0 +1,825 @@ +""" +GPT-SoVITS API 实现 + +### 完整请求示例 (/ttsrole POST) +{ + "text": "你好", # str, 必填, 要合成的文本内容 + "role": "role1", # str, 必填, 角色名称，决定使用 roles/{role} 中的配置和音频 + "emotion": "开心", # str, 可选, 情感标签，用于从 roles/{role}/reference_audios 中选择音频 + "text_lang": "auto", # str, 可选, 默认 "auto", 文本语言，"auto" 时根据 emotion 或角色目录动态选择 + "ref_audio_path": "/path/to/ref.wav", # str, 可选, 参考音频路径，若提供则优先使用，跳过自动选择 + "aux_ref_audio_paths": ["/path1.wav", "/path2.wav"], # List[str], 可选, 辅助参考音频路径，用于多说话人融合 + "prompt_lang": "ja", # str, 可选, 提示文本语言，若提供 ref_audio_path 则需指定，"auto" 模式下动态选择 + "prompt_text": "こんにちは", # str, 可选, 提示文本，与 ref_audio_path 配对使用，自动选择时从文件或文件名生成 + "top_k": 10, # int, 可选, Top-K 采样值，覆盖 inference.top_k + "top_p": 0.8, # float, 可选, Top-P 采样值，覆盖 inference.top_p + "temperature": 1.0, # float, 可选, 温度值，覆盖 inference.temperature + "text_split_method": "cut5", # str, 可选, 文本分割方法，覆盖 inference.text_split_method, 具体见text_segmentation_method.py + "batch_size": 2, # int, 可选, 批处理大小，覆盖 inference.batch_size + "batch_threshold": 0.75, # float, 可选, 批处理阈值，覆盖 inference.batch_threshold + "split_bucket": true, # bool, 可选, 是否按桶分割，覆盖 inference.split_bucket + "speed_factor": 1.2, # float, 可选, 语速因子，覆盖 inference.speed_factor + "fragment_interval": 0.3, # float, 可选, 片段间隔（秒），覆盖 inference.fragment_interval + "seed": 42, # int, 可选, 随机种子，覆盖 seed + "media_type": "wav", # str, 可选, 默认 "wav", 输出格式，支持 "wav", "raw", "ogg", "aac" + "streaming_mode": false, # bool, 可选, 默认 false, 是否流式返回 + "parallel_infer": true, # bool, 可选, 默认 true, 是否并行推理 + "repetition_penalty": 1.35, # float, 可选, 重复惩罚值，覆盖 inference.repetition_penalty + "version": "v2", # str, 可选, 配置文件版本，覆盖 version + "languages": ["zh", "ja", "en"], # List[str], 可选, 支持的语言列表，覆盖 languages + "bert_base_path": "/path/to/bert", # str, 可选, BERT 模型路径，覆盖 bert_base_path + "cnhuhbert_base_path": "/path/to/hubert", # str, 可选, HuBERT 模型路径，覆盖 cnhuhbert_base_path + "device": "cpu", # str, 可选, 统一设备，覆盖 device + "is_half": true, # bool, 可选, 是否使用半精度，覆盖 is_half + "t2s_weights_path": "/path/to/gpt.ckpt", # str, 可选, GPT 模型路径，覆盖 t2s_weights_path + "vits_weights_path": "/path/to/sovits.pth", # str, 可选, SoVITS 模型路径，覆盖 vits_weights_path + "t2s_model_path": "/path/to/gpt.ckpt", # str, 可选, GPT 模型路径（与 t2s_weights_path 同义） + "t2s_model_device": "cpu", # str, 可选, GPT 模型设备，覆盖 t2s_model.device，默认检测显卡 + "vits_model_path": "/path/to/sovits.pth", # str, 可选, SoVITS 模型路径（与 vits_weights_path 同义） + "vits_model_device": "cpu" # str, 可选, SoVITS 模型设备，覆盖 vits_model.device，默认检测显卡 +} + +### 参数必要性和优先级 +- 必填参数: + - /ttsrole: text, role + - /tts: text, ref_audio_path, prompt_lang +- 可选参数: 其他均为可选，默认值从 roles/{role}/tts_infer.yaml 或 GPT_SoVITS/configs/tts_infer.yaml 获取 +- 优先级: POST 请求参数 > roles/{role}/tts_infer.yaml > 默认 GPT_SoVITS/configs/tts_infer.yaml + +### 目录结构 +GPT-SoVITS-roleapi/ +├── api_role.py # 本文件, API 主程序 +├── GPT_SoVITS/ # GPT-SoVITS 核心库 +│ └── configs/ +│ └── tts_infer.yaml # 默认配置文件 +├── roles/ # 角色配置目录 +│ ├── role1/ # 示例角色 role1 +│ │ ├── tts_infer.yaml # 角色配置文件（可选） +│ │ ├── model.ckpt # GPT 模型（可选） +│ │ ├── model.pth # SoVITS 模型（可选） +│ │ └── reference_audios/ # 角色参考音频目录 +│ │ ├── zh/ +│ │ │ ├── 【开心】voice1.wav +│ │ │ ├── 【开心】voice1.txt +│ │ ├── ja/ +│ │ │ ├── 【开心】voice2.wav +│ │ │ ├── 【开心】voice2.txt +│ ├── role2/ +│ │ ├── tts_infer.yaml +│ │ ├── model.ckpt +│ │ ├── model.pth +│ │ └── reference_audios/ +│ │ ├── zh/ +│ │ │ ├── 【开心】voice1.wav +│ │ │ ├── 【开心】voice1.txt +│ │ │ ├── 【悲伤】asdafasdas.wav +│ │ │ ├── 【悲伤】asdafasdas.txt +│ │ ├── ja/ +│ │ │ ├── 【开心】voice2.wav +│ │ │ ├── 【开心】voice2.txt + +### text_lang, prompt_lang, prompt_text 选择逻辑 (/ttsrole) +1. text_lang 选择逻辑: + - 默认值: "auto" + - 如果请求未提供 text_lang，视为 "auto" + - 当 text_lang = "auto" 且存在 emotion 参数： + - 从 roles/{role}/reference_audios 下所有语言文件夹中查找以 "【emotion】" 开头的音频 + - 随机选择一个匹配的音频，语言由音频所在文件夹确定 + - 当 text_lang 指定具体语言（如 "zh"）： + - 从 roles/{role}/reference_audios/{text_lang} 中选择音频 + - 如果指定语言无匹配音频，则尝试其他语言文件夹 +2. prompt_lang 选择逻辑: + - 如果提供了 ref_audio_path，则需显式指定 prompt_lang + - 如果未提供 ref_audio_path 且 text_lang = "auto" 且存在 emotion： + - prompt_lang = 随机选择的音频所在语言文件夹名（如 "zh" 或 "ja"） + - 如果未提供 ref_audio_path 且 text_lang 指定具体语言： + - prompt_lang = text_lang（如 "zh"） + - 如果 text_lang 无匹配音频，则为随机选择的音频所在语言 +3. prompt_text 选择逻辑: + - 如果提供了 ref_audio_path（如 "/path/to/ref.wav"）： + - 检查文件名是否包含 "【xxx】" 前缀： + - 如果有（如 "【开心】abc.wav"）： + - 若存在对应 .txt 文件（如 "【开心】abc.txt"），prompt_text = .txt 文件内容 + - 若无对应 .txt 文件，prompt_text = "abc"（去掉 "【开心】" 和 ".wav" 的部分） + - 如果无 "【xxx】" 前缀： + - 若存在对应 .txt 文件（如 "ref.txt"），prompt_text = .txt 文件内容 + - 若无对应 .txt 文件，prompt_text = "ref"（去掉 ".wav" 的部分） + - 如果未提供 ref_audio_path： + - 从 roles/{role}/reference_audios 中选择音频（基于 text_lang 和 emotion）： + - 优先匹配 "【emotion】" 前缀的音频（如 "【开心】voice1.wav"） + - 若存在对应 .txt 文件（如 "【开心】voice1.txt"），prompt_text = .txt 文件内容 + - 若无对应 .txt 文件，prompt_text = "voice1"（去掉 "【开心】" 和 ".wav" 的部分） + - 未匹配 emotion 则随机选择一个音频，逻辑同上 + +### 讲解 +1. 必填参数: + - /ttsrole: text, role + - /tts: text, ref_audio_path, prompt_lang +2. 音频选择 (/ttsrole): + - 若提供 ref_audio_path，则使用它 + - 否则根据 role、text_lang、emotion 从 roles/{role}/reference_audios 中选择 + - text_lang = "auto" 时，若有 emotion，则跨语言匹配 "【emotion】" 前缀音频 + - emotion 匹配 "【emotion】" 前缀音频，未匹配则随机选择 +3. 设备选择: + - 默认尝试检测显卡（torch.cuda.is_available()），若可用则用 "cuda"，否则 "cpu" + - 若缺少 torch 依赖或检测失败，回退到 "cpu" + - POST 参数 device, t2s_model_device, vits_model_device 可强制指定设备，优先级最高 +4. 配置文件: + - 默认加载 GPT_SoVITS/configs/tts_infer.yaml + - 若 roles/{role}/tts_infer.yaml 存在且未被请求参数覆盖，则使用它 (/ttsrole) + - 请求参数（如 top_k, bert_base_path）覆盖所有配置文件 +5. 返回格式: + - 成功时返回音频流 (Response 或 StreamingResponse) + - 失败时返回 JSON，包含错误消息和可能的异常详情 +6. 运行: + - python api_role.py -a 127.0.0.1 -p 9880 + - 检查启动日志确认设备 + +### 调用示例 (/ttsrole) +## 非流式调用，会一次性返回完整的音频数据，适用于需要完整音频文件的场景 +import requests +url = "http://127.0.0.1:9880/ttsrole" +payload = { + "text": "你好，这是一个测试", # 要合成的文本 + "role": "role1", # 角色名称，必填 + "emotion": "开心", # 情感标签，可选 + "text_lang": "zh", # 文本语言，可选，默认为 "zh" + "media_type": "wav" # 输出音频格式，默认 "wav" +} +response = requests.post(url, json=payload) +if response.status_code == 200: + with open("output_non_stream.wav", "wb") as f: + f.write(response.content) + print("非流式音频已生成并保存为 output_non_stream.wav") +else: + print(f"请求失败: {response.json()}") + +## 流式调用，会分块返回音频数据，适用于实时播放或处理大文件的场景 +import requests +url = "http://127.0.0.1:9880/ttsrole" +payload = { + "text": "你好，这是一个测试", # 要合成的文本 + "role": "role1", # 角色名称，必填 + "emotion": "开心", # 情感标签，可选 + "text_lang": "zh", # 文本语言，可选，默认为 "zh" + "media_type": "wav", # 输出音频格式，默认 "wav" + "streaming_mode": True # 启用流式模式 +} +with requests.post(url, json=payload, stream=True) as response: + if response.status_code == 200: + with open("output_stream.wav", "wb") as f: + for chunk in response.iter_content(chunk_size=1024): + if chunk: # 确保 chunk 不为空 + f.write(chunk) + print("流式音频已生成并保存为 output_stream.wav") + else: + print(f"请求失败: {response.json()}") +""" + +import os +import sys +import traceback +from typing import Generator, Optional, List, Dict +import random +import glob +from concurrent.futures import ThreadPoolExecutor +import asyncio + +now_dir = os.getcwd() +sys.path.append(now_dir) +sys.path.append("%s/GPT_SoVITS" % (now_dir)) + +import argparse +import subprocess +import wave +import signal +import numpy as np +import soundfile as sf +from fastapi import FastAPI, HTTPException, Response +from fastapi.responses import StreamingResponse, JSONResponse +from pydantic import BaseModel +import uvicorn +from io import BytesIO +from tools.i18n.i18n import I18nAuto +from GPT_SoVITS.TTS_infer_pack.TTS import TTS, TTS_Config +from GPT_SoVITS.TTS_infer_pack.text_segmentation_method import get_method_names as get_cut_method_names + +# 尝试导入 PyTorch，检测显卡支持 +try: + import torch + cuda_available = torch.cuda.is_available() +except ImportError: + cuda_available = False + print("缺少 PyTorch 依赖，默认使用 CPU") +except Exception as e: + cuda_available = False + print(f"检测显卡时出错: {str(e)}，默认使用 CPU") + +i18n = I18nAuto() +cut_method_names = get_cut_method_names() + +parser = argparse.ArgumentParser(description="GPT-SoVITS api") +parser.add_argument("-c", "--tts_config", type=str, default="GPT_SoVITS/configs/tts_infer.yaml", help="tts_infer路径") +parser.add_argument("-a", "--bind_addr", type=str, default="127.0.0.1", help="default: 127.0.0.1") +parser.add_argument("-p", "--port", type=int, default="9880", help="default: 9880") +args = parser.parse_args() +config_path = args.tts_config +port = args.port +host = args.bind_addr +argv = sys.argv + +if config_path in [None, ""]: + config_path = "GPT_SoVITS/configs/tts_infer.yaml" + +default_device = "cuda" if cuda_available else "cpu" +print(f"默认设备设置为: {default_device}") + +# 初始化 TTS 配置 +tts_config = TTS_Config(config_path) +print(f"TTS_Config contents: {tts_config.__dict__}") +if hasattr(tts_config, 'device'): + tts_config.device = default_device +tts_pipeline = TTS(tts_config) + +# 创建线程池用于异步执行 TTS 任务 +executor = ThreadPoolExecutor(max_workers=1) + +APP = FastAPI() + +class TTS_Request(BaseModel): + text: str + ref_audio_path: str + prompt_lang: str + text_lang: str = "auto" + aux_ref_audio_paths: Optional[List[str]] = None + prompt_text: Optional[str] = "" + top_k: Optional[int] = 5 + top_p: Optional[float] = 1 + temperature: Optional[float] = 1 + text_split_method: Optional[str] = "cut5" + batch_size: Optional[int] = 1 + batch_threshold: Optional[float] = 0.75 + split_bucket: Optional[bool] = True + speed_factor: Optional[float] = 1.0 + fragment_interval: Optional[float] = 0.3 + seed: Optional[int] = -1 + media_type: Optional[str] = "wav" + streaming_mode: Optional[bool] = False + parallel_infer: Optional[bool] = True + repetition_penalty: Optional[float] = 1.35 + device: Optional[str] = None + +class TTSRole_Request(BaseModel): + text: str + role: str + text_lang: Optional[str] = "auto" + ref_audio_path: Optional[str] = None + aux_ref_audio_paths: Optional[List[str]] = None + prompt_lang: Optional[str] = None + prompt_text: Optional[str] = None + emotion: Optional[str] = None + top_k: Optional[int] = 5 + top_p: Optional[float] = 1 + temperature: Optional[float] = 1 + text_split_method: Optional[str] = "cut5" + batch_size: Optional[int] = 1 + batch_threshold: Optional[float] = 0.75 + split_bucket: Optional[bool] = True + speed_factor: Optional[float] = 1.0 + fragment_interval: Optional[float] = 0.3 + seed: Optional[int] = -1 + media_type: Optional[str] = "wav" + streaming_mode: Optional[bool] = False + parallel_infer: Optional[bool] = True + repetition_penalty: Optional[float] = 1.35 + bert_base_path: Optional[str] = None + cnhuhbert_base_path: Optional[str] = None + device: Optional[str] = None + is_half: Optional[bool] = None + t2s_weights_path: Optional[str] = None + version: Optional[str] = None + vits_weights_path: Optional[str] = None + t2s_model_path: Optional[str] = None + vits_model_path: Optional[str] = None + t2s_model_device: Optional[str] = None + vits_model_device: Optional[str] = None + +def pack_ogg(io_buffer: BytesIO, data: np.ndarray, rate: int): + with sf.SoundFile(io_buffer, mode='w', samplerate=rate, channels=1, format='ogg') as audio_file: + audio_file.write(data) + io_buffer.seek(0) + return io_buffer + +def pack_raw(io_buffer: BytesIO, data: np.ndarray, rate: int): + io_buffer.write(data.tobytes()) + io_buffer.seek(0) + return io_buffer + +def pack_wav(io_buffer: BytesIO, data: np.ndarray, rate: int): + sf.write(io_buffer, data, rate, format='wav') + io_buffer.seek(0) + return io_buffer + +def pack_aac(io_buffer: BytesIO, data: np.ndarray, rate: int): + process = subprocess.Popen([ + 'ffmpeg', '-f', 's16le', '-ar', str(rate), '-ac', '1', '-i', 'pipe:0', + '-c:a', 'aac', '-b:a', '192k', '-vn', '-f', 'adts', 'pipe:1' + ], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + out, _ = process.communicate(input=data.tobytes()) + io_buffer.write(out) + io_buffer.seek(0) + return io_buffer + +def pack_audio(data: np.ndarray, rate: int, media_type: str) -> BytesIO: + io_buffer = BytesIO() + if media_type == "ogg": + io_buffer = pack_ogg(io_buffer, data, rate) + elif media_type == "aac": + io_buffer = pack_aac(io_buffer, data, rate) + elif media_type == "wav": + io_buffer = pack_wav(io_buffer, data, rate) + else: + io_buffer = pack_raw(io_buffer, data, rate) + return io_buffer + +def wave_header_chunk(frame_input=b"", channels=1, sample_width=2, sample_rate=32000): + wav_buf = BytesIO() + with wave.open(wav_buf, "wb") as vfout: + vfout.setnchannels(channels) + vfout.setsampwidth(sample_width) + vfout.setframerate(sample_rate) + vfout.writeframes(frame_input) + wav_buf.seek(0) + return wav_buf.read() + +def handle_control(command: str): + if command == "restart": + os.execl(sys.executable, sys.executable, *argv) + elif command == "exit": + os.kill(os.getpid(), signal.SIGTERM) + exit(0) + +def check_params(req: dict, is_ttsrole: bool = False): + text = req.get("text") + text_lang = req.get("text_lang", "auto") + ref_audio_path = req.get("ref_audio_path") + prompt_lang = req.get("prompt_lang") + media_type = req.get("media_type", "wav") + streaming_mode = req.get("streaming_mode", False) + text_split_method = req.get("text_split_method", "cut5") + + if not text: + return {"status": "error", "message": "text is required"} + + if is_ttsrole: + role = req.get("role") + if not role: + return {"status": "error", "message": "role is required for /ttsrole"} + else: + if not ref_audio_path: + return {"status": "error", "message": "ref_audio_path is required"} + if not prompt_lang: + return {"status": "error", "message": "prompt_lang is required"} + + languages = req.get("languages") or tts_config.languages + if text_lang != "auto" and text_lang.lower() not in languages: + return {"status": "error", "message": f"text_lang: {text_lang} is not supported"} + if prompt_lang and prompt_lang.lower() not in languages: + return {"status": "error", "message": f"prompt_lang: {prompt_lang} is not supported"} + + if media_type not in ["wav", "raw", "ogg", "aac"]: + return {"status": "error", "message": f"media_type: {media_type} is not supported"} + if media_type == "ogg" and not streaming_mode: + return {"status": "error", "message": "ogg format is not supported in non-streaming mode"} + if text_split_method not in cut_method_names: + return {"status": "error", "message": f"text_split_method: {text_split_method} is not supported"} + + return None + +def load_role_config(role: str, req: dict): + role_dir = os.path.join(now_dir, "roles", role) + if not os.path.exists(role_dir): + return False + + if not any(req.get(k) for k in ["version", "bert_base_path", "cnhuhbert_base_path", "device", "is_half", "t2s_weights_path", "vits_weights_path"]): + config_path_new = os.path.join(role_dir, "tts_infer.yaml") + if os.path.exists(config_path_new): + global tts_config, tts_pipeline + tts_config = TTS_Config(config_path_new) + if hasattr(tts_config, 'device'): + tts_config.device = default_device + tts_pipeline = TTS(tts_config) + + if not req.get("t2s_weights_path") and not req.get("t2s_model_path"): + gpt_path = glob.glob(os.path.join(role_dir, "*.ckpt")) + if gpt_path: + tts_pipeline.init_t2s_weights(gpt_path[0]) + if not req.get("vits_weights_path") and not req.get("vits_model_path"): + sovits_path = glob.glob(os.path.join(role_dir, "*.pth")) + if sovits_path: + tts_pipeline.init_vits_weights(sovits_path[0]) + + return True + +def select_ref_audio(role: str, text_lang: str, emotion: str = None): + audio_base_dir = os.path.join(now_dir, "roles", role, "reference_audios") + if not os.path.exists(audio_base_dir): + return None, None, None + + if text_lang.lower() == "auto" and emotion: + all_langs = [d for d in os.listdir(audio_base_dir) if os.path.isdir(os.path.join(audio_base_dir, d))] + emotion_files = [] + for lang in all_langs: + lang_dir = os.path.join(audio_base_dir, lang) + emotion_files.extend(glob.glob(os.path.join(lang_dir, f"【{emotion}】*.*"))) + + if emotion_files: + audio_path = random.choice(emotion_files) + txt_path = audio_path.rsplit(".", 1)[0] + ".txt" + if os.path.exists(txt_path): + with open(txt_path, "r", encoding="utf-8") as f: + prompt_text = f.read().strip() + else: + basename = os.path.basename(audio_path) + start_idx = basename.find("】") + 1 + end_idx = basename.rfind(".") + prompt_text = basename[start_idx:end_idx] if end_idx > start_idx else basename + + prompt_lang = os.path.basename(os.path.dirname(audio_path)) + return audio_path, prompt_text, prompt_lang + + lang_dir = os.path.join(audio_base_dir, text_lang.lower()) + all_langs = [d for d in os.listdir(audio_base_dir) if os.path.isdir(os.path.join(audio_base_dir, d))] + + def find_audio_in_dir(dir_path): + if not os.path.exists(dir_path): + return None, None + audio_files = glob.glob(os.path.join(dir_path, "【*】*.*")) + if not audio_files: + audio_files = glob.glob(os.path.join(dir_path, "*.*")) + if not audio_files: + return None, None + + if emotion: + emotion_files = [f for f in audio_files if f"【{emotion}】" in os.path.basename(f)] + if emotion_files: + audio_path = random.choice(emotion_files) + else: + audio_path = random.choice(audio_files) + else: + audio_path = random.choice(audio_files) + + txt_path = audio_path.rsplit(".", 1)[0] + ".txt" + prompt_text = None + if os.path.exists(txt_path): + with open(txt_path, "r", encoding="utf-8") as f: + prompt_text = f.read().strip() + else: + basename = os.path.basename(audio_path) + start_idx = basename.find("】") + 1 + end_idx = basename.rfind(".") + if start_idx > 0 and end_idx > start_idx: + prompt_text = basename[start_idx:end_idx] + else: + prompt_text = basename[:end_idx] if end_idx > 0 else basename + + return audio_path, prompt_text + + audio_path, prompt_text = find_audio_in_dir(lang_dir) + if audio_path: + return audio_path, prompt_text, text_lang.lower() + + for lang in all_langs: + if lang != text_lang.lower(): + audio_path, prompt_text = find_audio_in_dir(os.path.join(audio_base_dir, lang)) + if audio_path: + return audio_path, prompt_text, lang + + return None, None, None + +def set_pipeline_device(pipeline: TTS, device: str): + """将 TTS 管道中的所有模型和相关组件迁移到指定设备，仅在设备变化时执行""" + if not torch.cuda.is_available() and device.startswith("cuda"): + print(f"警告: CUDA 不可用，强制使用 CPU") + device = "cpu" + + target_device = torch.device(device) + + # 检查当前设备是否需要切换 + current_device = None + if hasattr(pipeline, 't2s_model') and pipeline.t2s_model is not None: + current_device = next(pipeline.t2s_model.parameters()).device + elif hasattr(pipeline, 'vits_model') and pipeline.vits_model is not None: + current_device = next(pipeline.vits_model.parameters()).device + + if current_device == target_device: + print(f"设备已是 {device}，无需切换") + return + + # 更新配置中的设备 + if hasattr(pipeline, 'configs') and hasattr(pipeline.configs, 'device'): + pipeline.configs.device = device + + # 迁移所有可能的模型到指定设备 + for attr in ['t2s_model', 'vits_model']: + if hasattr(pipeline, attr) and getattr(pipeline, attr) is not None: + getattr(pipeline, attr).to(target_device) + + for attr in dir(pipeline): + if attr.endswith('_model') and getattr(pipeline, attr) is not None: + try: + getattr(pipeline, attr).to(target_device) + print(f"迁移 {attr} 到 {device}") + except AttributeError: + pass + + # 清理 GPU 缓存 + if torch.cuda.is_available() and not device.startswith("cuda"): + torch.cuda.empty_cache() + + print(f"TTS 管道设备已设置为: {device}") + +def run_tts_pipeline(req): + """在线程池中运行 TTS 任务""" + return tts_pipeline.run(req) + +async def tts_handle(req: dict, is_ttsrole: bool = False): + streaming_mode = req.get("streaming_mode", False) + media_type = req.get("media_type", "wav") + + if "text_lang" not in req: + req["text_lang"] = "auto" + + check_res = check_params(req, is_ttsrole) + if check_res is not None: + return JSONResponse(status_code=400, content=check_res) + + # 如果请求中指定了 device，则覆盖所有与设备相关的参数并更新管道设备 + if "device" in req and req["device"] is not None: + device = req["device"] + req["t2s_model_device"] = device + req["vits_model_device"] = device + if hasattr(tts_config, 'device'): + tts_config.device = device + set_pipeline_device(tts_pipeline, device) + + if is_ttsrole: + role_exists = load_role_config(req["role"], req) + + for key in ["bert_base_path", "cnhuhbert_base_path", "device", "is_half", "t2s_weights_path", "version", "vits_weights_path"]: + if req.get(key) is not None: + setattr(tts_config, key, req[key]) + + if req.get("t2s_model_path"): + tts_config.t2s_weights_path = req["t2s_model_path"] + tts_pipeline.init_t2s_weights(req["t2s_model_path"]) + if req.get("vits_model_path"): + tts_config.vits_weights_path = req["vits_model_path"] + tts_pipeline.init_vits_weights(req["vits_model_path"]) + + if not req.get("ref_audio_path"): + ref_audio_path, prompt_text, prompt_lang = select_ref_audio(req["role"], req["text_lang"], req.get("emotion")) + if ref_audio_path: + req["ref_audio_path"] = ref_audio_path + req["prompt_text"] = prompt_text or "" + req["prompt_lang"] = prompt_lang or req["text_lang"] + elif not role_exists: + return JSONResponse(status_code=400, content={"status": "error", "message": "Role directory not found and no suitable reference audio provided"}) + else: + ref_audio_path = req["ref_audio_path"] + txt_path = ref_audio_path.rsplit(".", 1)[0] + ".txt" + if os.path.exists(txt_path): + with open(txt_path, "r", encoding="utf-8") as f: + req["prompt_text"] = f.read().strip() + else: + basename = os.path.basename(ref_audio_path) + if "【" in basename and "】" in basename: + start_idx = basename.find("】") + 1 + end_idx = basename.rfind(".") + if start_idx > 0 and end_idx > start_idx: + req["prompt_text"] = basename[start_idx:end_idx] + else: + req["prompt_text"] = basename[:end_idx] if end_idx > 0 else basename + else: + end_idx = basename.rfind(".") + req["prompt_text"] = basename[:end_idx] if end_idx > 0 else basename + + if streaming_mode: + req["return_fragment"] = True + + try: + print(f"当前请求设备: {req.get('device')}") + if hasattr(tts_pipeline, 't2s_model') and tts_pipeline.t2s_model is not None: + print(f"t2s_model 设备: {next(tts_pipeline.t2s_model.parameters()).device}") + if hasattr(tts_pipeline, 'vits_model') and tts_pipeline.vits_model is not None: + print(f"vits_model 设备: {next(tts_pipeline.vits_model.parameters()).device}") + + # 异步执行 TTS 任务 + loop = asyncio.get_event_loop() + tts_generator = await loop.run_in_executor(executor, run_tts_pipeline, req) + + if streaming_mode: + def streaming_generator(): + if media_type == "wav": + yield wave_header_chunk() + stream_type = "raw" + else: + stream_type = media_type + for sr, chunk in tts_generator: + buf = pack_audio(chunk, sr, stream_type) + yield buf.getvalue() + return StreamingResponse(streaming_generator(), media_type=f"audio/{media_type}") + else: + sr, audio_data = next(tts_generator) + buf = pack_audio(audio_data, sr, media_type) + return Response(buf.getvalue(), media_type=f"audio/{media_type}") + except Exception as e: + return JSONResponse(status_code=400, content={"status": "error", "message": "tts failed", "exception": str(e)}) + +@APP.get("/control") +async def control(command: str = None): + if command is None: + return JSONResponse(status_code=400, content={"status": "error", "message": "command is required"}) + handle_control(command) + +@APP.get("/tts") +async def tts_get_endpoint( + text: str, + ref_audio_path: str, + prompt_lang: str, + text_lang: str = "auto", + aux_ref_audio_paths: Optional[List[str]] = None, + prompt_text: Optional[str] = "", + top_k: Optional[int] = 5, + top_p: Optional[float] = 1, + temperature: Optional[float] = 1, + text_split_method: Optional[str] = "cut0", + batch_size: Optional[int] = 1, + batch_threshold: Optional[float] = 0.75, + split_bucket: Optional[bool] = True, + speed_factor: Optional[float] = 1.0, + fragment_interval: Optional[float] = 0.3, + seed: Optional[int] = -1, + media_type: Optional[str] = "wav", + streaming_mode: Optional[bool] = False, + parallel_infer: Optional[bool] = True, + repetition_penalty: Optional[float] = 1.35, + device: Optional[str] = None +): + req = { + "text": text, + "text_lang": text_lang.lower(), + "ref_audio_path": ref_audio_path, + "aux_ref_audio_paths": aux_ref_audio_paths, + "prompt_lang": prompt_lang.lower(), + "prompt_text": prompt_text, + "top_k": top_k, + "top_p": top_p, + "temperature": temperature, + "text_split_method": text_split_method, + "batch_size": batch_size, + "batch_threshold": batch_threshold, + "split_bucket": split_bucket, + "speed_factor": speed_factor, + "fragment_interval": fragment_interval, + "seed": seed, + "media_type": media_type, + "streaming_mode": streaming_mode, + "parallel_infer": parallel_infer, + "repetition_penalty": repetition_penalty, + "device": device + } + return await tts_handle(req) + +@APP.post("/tts") +async def tts_post_endpoint(request: TTS_Request): + req = request.dict(exclude_unset=True) + if "text_lang" in req: + req["text_lang"] = req["text_lang"].lower() + if "prompt_lang" in req: + req["prompt_lang"] = req["prompt_lang"].lower() + return await tts_handle(req) + +@APP.get("/ttsrole") +async def ttsrole_get_endpoint( + text: str, + role: str, + text_lang: str = "auto", + ref_audio_path: Optional[str] = None, + aux_ref_audio_paths: Optional[List[str]] = None, + prompt_lang: Optional[str] = None, + prompt_text: Optional[str] = None, + emotion: Optional[str] = None, + top_k: Optional[int] = 5, + top_p: Optional[float] = 1, + temperature: Optional[float] = 1, + text_split_method: Optional[str] = "cut5", + batch_size: Optional[int] = 1, + batch_threshold: Optional[float] = 0.75, + split_bucket: Optional[bool] = True, + speed_factor: Optional[float] = 1.0, + fragment_interval: Optional[float] = 0.3, + seed: Optional[int] = -1, + media_type: Optional[str] = "wav", + streaming_mode: Optional[bool] = False, + parallel_infer: Optional[bool] = True, + repetition_penalty: Optional[float] = 1.35, + bert_base_path: Optional[str] = None, + cnhuhbert_base_path: Optional[str] = None, + device: Optional[str] = None, + is_half: Optional[bool] = None, + t2s_weights_path: Optional[str] = None, + version: Optional[str] = None, + vits_weights_path: Optional[str] = None, + t2s_model_path: Optional[str] = None, + vits_model_path: Optional[str] = None, + t2s_model_device: Optional[str] = None, + vits_model_device: Optional[str] = None +): + req = { + "text": text, + "role": role, + "text_lang": text_lang.lower(), + "ref_audio_path": ref_audio_path, + "aux_ref_audio_paths": aux_ref_audio_paths, + "prompt_lang": prompt_lang.lower() if prompt_lang else None, + "prompt_text": prompt_text, + "emotion": emotion, + "top_k": top_k, + "top_p": top_p, + "temperature": temperature, + "text_split_method": text_split_method, + "batch_size": batch_size, + "batch_threshold": batch_threshold, + "split_bucket": split_bucket, + "speed_factor": speed_factor, + "fragment_interval": fragment_interval, + "seed": seed, + "media_type": media_type, + "streaming_mode": streaming_mode, + "parallel_infer": parallel_infer, + "repetition_penalty": repetition_penalty, + "bert_base_path": bert_base_path, + "cnhuhbert_base_path": cnhuhbert_base_path, + "device": device, + "is_half": is_half, + "t2s_weights_path": t2s_weights_path, + "version": version, + "vits_weights_path": vits_weights_path, + "t2s_model_path": t2s_model_path, + "vits_model_path": vits_model_path, + "t2s_model_device": t2s_model_device, + "vits_model_device": vits_model_device + } + return await tts_handle(req, is_ttsrole=True) + +@APP.post("/ttsrole") +async def ttsrole_post_endpoint(request: TTSRole_Request): + req = request.dict(exclude_unset=True) + if "text_lang" in req: + req["text_lang"] = req["text_lang"].lower() + if "prompt_lang" in req: + req["prompt_lang"] = req["prompt_lang"].lower() + return await tts_handle(req, is_ttsrole=True) + +@APP.get("/set_gpt_weights") +async def set_gpt_weights(weights_path: str = None): + try: + if not weights_path: + return JSONResponse(status_code=400, content={"status": "error", "message": "gpt weight path is required"}) + tts_pipeline.init_t2s_weights(weights_path) + tts_config.t2s_weights_path = weights_path + return JSONResponse(status_code=200, content={"status": "success", "message": "success"}) + except Exception as e: + return JSONResponse(status_code=400, content={"status": "error", "message": f"change gpt weight failed", "exception": str(e)}) + +@APP.get("/set_sovits_weights") +async def set_sovits_weights(weights_path: str = None): + try: + if not weights_path: + return JSONResponse(status_code=400, content={"status": "error", "message": "sovits weight path is required"}) + tts_pipeline.init_vits_weights(weights_path) + tts_config.vits_weights_path = weights_path + return JSONResponse(status_code=200, content={"status": "success", "message": "success"}) + except Exception as e: + return JSONResponse(status_code=400, content={"status": "error", "message": f"change sovits weight failed", "exception": str(e)}) + +@APP.get("/set_refer_audio") +async def set_refer_audio(refer_audio_path: str = None): + try: + if not refer_audio_path: + return JSONResponse(status_code=400, content={"status": "error", "message": "refer audio path is required"}) + tts_pipeline.set_ref_audio(refer_audio_path) + return JSONResponse(status_code=200, content={"status": "success", "message": "success"}) + except Exception as e: + return JSONResponse(status_code=400, content={"status": "error", "message": f"set refer audio failed", "exception": str(e)}) + +if __name__ == "__main__": + try: + if host == 'None': # 在调用时使用 -a None 参数，可以让api监听双栈 + host = None + uvicorn.run(app=APP, host=host, port=port, workers=1) + except Exception as e: + traceback.print_exc() + os.kill(os.getpid(), signal.SIGTERM) + exit(0) \ No newline at end of file From 87c521ea6e81cdfa93c5aed0805e3bb46281742e Mon Sep 17 00:00:00 2001 From: spawner Date: Fri, 7 Mar 2025 14:57:51 +0800 Subject: [PATCH 2/3] Create api_role_v3.py --- api_role_v3.py | 1020 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1020 insertions(+) create mode 100644 api_role_v3.py diff --git a/api_role_v3.py b/api_role_v3.py new file mode 100644 index 0000000..d46ee7a --- /dev/null +++ b/api_role_v3.py @@ -0,0 +1,1020 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +功能： +- 通过 GET 和 POST 请求提供 TTS 推理接口 (`/`)，支持默认参考音频和参数调整。 +- 新增 `/ttsrole` 接口，支持基于角色的 TTS 推理，动态加载角色模型和参考音频，同时支持 GET 和 POST 请求。 +- 支持更换默认参考音频 (`/change_refer`) 和模型权重 (`/set_model`)。 +- 提供控制接口 (`/control`) 用于重启或退出服务。 +- 支持多语言文本处理（中文、英文、日文、韩文等）及自动语言切分。 +- 支持多种音频格式（wav, ogg, aac）和数据类型（int16, int32）。 +- 支持通过 POST 请求动态切换模型版本（v2 或 v3）。 + +使用方法： +1. 安装依赖： + pip install -r requirements.txt +2. 配置环境： + - 确保 GPT 和 SoVITS 模型文件已准备好。 + - 可选：设置默认参考音频路径、文本和语言。 +3. 运行服务： + python api.py -s "path/to/sovits.pth" -g "path/to/gpt.ckpt" -dr "ref.wav" -dt "参考文本" -dl "zh" -p 9880 + +参数说明： +命令行参数： +- -s, --sovits_path: SoVITS 模型路径（默认从 config 获取）。 +- -g, --gpt_path: GPT 模型路径（默认从 config 获取）。 +- -dr, --default_refer_path: 默认参考音频路径。 +- -dt, --default_refer_text: 默认参考音频文本。 +- -dl, --default_refer_language: 默认参考音频语言（zh, en, ja, ko 等）。 +- -d, --device: 设备（cuda 或 cpu，默认从 config 获取）。 +- -a, --bind_addr: 绑定地址（默认 0.0.0.0）。 +- -p, --port: 端口（默认 9880）。 +- -fp, --full_precision: 使用全精度（覆盖默认）。 +- -hp, --half_precision: 使用半精度（覆盖默认）。 +- -sm, --stream_mode: 流式模式（close 或 normal，默认 close）。 +- -mt, --media_type: 音频格式（wav, ogg, aac，默认 wav）。 +- -st, --sub_type: 数据类型（int16 或 int32，默认 int16）。 +- -cp, --cut_punc: 文本切分符号（默认空）。 +- -hb, --hubert_path: HuBERT 模型路径（默认从 config 获取）。 +- -b, --bert_path: BERT 模型路径（默认从 config 获取）。 + +接口参数（/）： +- refer_wav_path: 参考音频路径（可选）。 +- prompt_text: 参考音频文本（可选）。 +- prompt_language: 参考音频语言（可选）。 +- text: 待合成文本（必填）。 +- text_language: 目标文本语言（可选，默认 auto）。 +- cut_punc: 文本切分符号（可选）。 +- top_k: Top-K 采样值（默认 15）。 +- top_p: Top-P 采样值（默认 1.0）。 +- temperature: 温度值（默认 1.0）。 +- speed: 语速因子（默认 1.0）。 +- inp_refs: 辅助参考音频路径列表（默认空）。 +- sample_steps: 采样步数（默认 32，限定 [4, 8, 16, 32]）。 +- if_sr: 是否超分（默认 False）。 + +接口参数（/ttsrole）： +- text: 待合成文本（必填）。 +- role: 角色名称（必填）。 +- text_language: 目标文本语言（默认 auto）。 +- ref_audio_path: 参考音频路径（可选）。 +- prompt_text: 参考音频文本（可选）。 +- prompt_language: 参考音频语言（可选）。 +- emotion: 情感标签（可选）。 +- top_k: Top-K 采样值（默认 15）。 +- top_p: Top-P 采样值（默认 0.6）。 +- temperature: 温度值（默认 0.6）。 +- speed: 语速因子（默认 1.0）。 +- inp_refs: 辅助参考音频路径列表（默认空）。 +- sample_steps: 采样步数（默认 32，限定 [4, 8, 16, 32]）。 +- if_sr: 是否超分（默认 False）。 +- version: 模型版本（可选，v2 或 v3，POST 请求支持动态切换）。 + +### 完整请求示例 (/ttsrole POST) +{ + "text": "你好", # str, 必填, 要合成的文本内容 + "role": "role1", # str, 必填, 角色名称，决定使用 roles/{role} 中的配置和音频 + "emotion": "开心", # str, 可选, 情感标签，用于从 roles/{role}/reference_audios 中选择音频 + "text_lang": "auto", # str, 可选, 默认 "auto", 文本语言，"auto" 时根据 emotion 或角色目录动态选择 + "ref_audio_path": "/path/to/ref.wav", # str, 可选, 参考音频路径，若提供则优先使用，跳过自动选择 + "aux_ref_audio_paths": ["/path1.wav", "/path2.wav"], # List[str], 可选, 辅助参考音频路径，用于多说话人融合 + "prompt_lang": "ja", # str, 可选, 提示文本语言，若提供 ref_audio_path 则需指定，"auto" 模式下动态选择 + "prompt_text": "こんにちは", # str, 可选, 提示文本，与 ref_audio_path 配对使用，自动选择时从文件或文件名生成 + "top_k": 10, # int, 可选, Top-K 采样值，覆盖 inference.top_k + "top_p": 0.8, # float, 可选, Top-P 采样值，覆盖 inference.top_p + "temperature": 1.0, # float, 可选, 温度值，覆盖 inference.temperature + "text_split_method": "cut5", # str, 可选, 文本分割方法，覆盖 inference.text_split_method, 具体见text_segmentation_method.py + "batch_size": 2, # int, 可选, 批处理大小，覆盖 inference.batch_size + "batch_threshold": 0.75, # float, 可选, 批处理阈值，覆盖 inference.batch_threshold + "split_bucket": true, # bool, 可选, 是否按桶分割，覆盖 inference.split_bucket + "speed_factor": 1.2, # float, 可选, 语速因子，覆盖 inference.speed_factor + "fragment_interval": 0.3, # float, 可选, 片段间隔（秒），覆盖 inference.fragment_interval + "seed": 42, # int, 可选, 随机种子，覆盖 seed + "media_type": "wav", # str, 可选, 默认 "wav", 输出格式，支持 "wav", "raw", "ogg", "aac" + "streaming_mode": false, # bool, 可选, 默认 false, 是否流式返回 + "parallel_infer": true, # bool, 可选, 默认 true, 是否并行推理 + "repetition_penalty": 1.35, # float, 可选, 重复惩罚值，覆盖 inference.repetition_penalty + "version": "v2", # str, 可选, 配置文件版本，覆盖 version，动态切换 v2 或 v3 + "languages": ["zh", "ja", "en"], # List[str], 可选, 支持的语言列表，覆盖 languages + "bert_base_path": "/path/to/bert", # str, 可选, BERT 模型路径，覆盖 bert_base_path + "cnhuhbert_base_path": "/path/to/hubert", # str, 可选, HuBERT 模型路径，覆盖 cnhuhbert_base_path + "device": "cpu", # str, 可选, 统一设备，覆盖 device + "is_half": true, # bool, 可选, 是否使用半精度，覆盖 is_half + "t2s_weights_path": "/path/to/gpt.ckpt", # str, 可选, GPT 模型路径，覆盖 t2s_weights_path + "vits_weights_path": "/path/to/sovits.pth", # str, 可选, SoVITS 模型路径，覆盖 vits_weights_path + "t2s_model_path": "/path/to/gpt.ckpt", # str, 可选, GPT 模型路径（与 t2s_weights_path 同义） + "t2s_model_device": "cpu", # str, 可选, GPT 模型设备，覆盖 t2s_model.device，默认检测显卡 + "vits_model_path": "/path/to/sovits.pth", # str, 可选, SoVITS 模型路径（与 vits_weights_path 同义） + "vits_model_device": "cpu" # str, 可选, SoVITS 模型设备，覆盖 vits_model.device，默认检测显卡 +} + +### 参数必要性和优先级 +- 必填参数: + - /ttsrole: text, role + - /tts: text, ref_audio_path, prompt_lang +- 可选参数: 其他均为可选，默认值从 roles/{role}/tts_infer.yaml 或 GPT_SoVITS/configs/tts_infer.yaml 获取 +- 优先级: POST 请求参数 > roles/{role}/tts_infer.yaml > 默认 GPT_SoVITS/configs/tts_infer.yaml + +### 目录结构 +GPT-SoVITS-roleapi/ +├── api.py # 本文件, API 主程序 +├── GPT_SoVITS/ # GPT-SoVITS 核心库 +│ └── configs/ +│ └── tts_infer.yaml # 默认配置文件 +├── roles/ # 角色配置目录 +│ ├── role1/ # 示例角色 role1 +│ │ ├── tts_infer.yaml # 角色配置文件（可选） +│ │ ├── model.ckpt # GPT 模型（可选） +│ │ ├── model.pth # SoVITS 模型（可选） +│ │ └── reference_audios/ # 角色参考音频目录 +│ │ ├── zh/ +│ │ │ ├── 【开心】voice1.wav +│ │ │ ├── 【开心】voice1.txt +│ │ ├── ja/ +│ │ │ ├── 【开心】voice2.wav +│ │ │ ├── 【开心】voice2.txt +│ ├── role2/ +│ │ ├── tts_infer.yaml +│ │ ├── model.ckpt +│ │ ├── model.pth +│ │ └── reference_audios/ +│ │ ├── zh/ +│ │ │ ├── 【开心】voice1.wav +│ │ │ ├── 【开心】voice1.txt +│ │ │ ├── 【悲伤】asdafasdas.wav +│ │ │ ├── 【悲伤】asdafasdas.txt +│ │ ├── ja/ +│ │ │ ├── 【开心】voice2.wav +│ │ │ ├── 【开心】voice2.txt + +### text_lang, prompt_lang, prompt_text 选择逻辑 (/ttsrole) +1. text_lang 选择逻辑: + - 默认值: "auto" + - 如果请求未提供 text_lang，视为 "auto" + - 当 text_lang = "auto" 且存在 emotion 参数： + - 从 roles/{role}/reference_audios 下所有语言文件夹中查找以 "【emotion】" 开头的音频 + - 随机选择一个匹配的音频，语言由音频所在文件夹确定 + - 当 text_lang 指定具体语言（如 "zh"）： + - 从 roles/{role}/reference_audios/{text_lang} 中选择音频 + - 如果指定语言无匹配音频，则尝试其他语言文件夹 +2. prompt_lang 选择逻辑: + - 如果提供了 ref_audio_path，则需显式指定 prompt_lang + - 如果未提供 ref_audio_path 且 text_lang = "auto" 且存在 emotion： + - prompt_lang = 随机选择的音频所在语言文件夹名（如 "zh" 或 "ja"） + - 如果未提供 ref_audio_path 且 text_lang 指定具体语言： + - prompt_lang = text_lang（如 "zh"） + - 如果 text_lang 无匹配音频，则为随机选择的音频所在语言 +3. prompt_text 选择逻辑: + - 如果提供了 ref_audio_path（如 "/path/to/ref.wav"）： + - 检查文件名是否包含 "【xxx】" 前缀： + - 如果有（如 "【开心】abc.wav"）： + - 若存在对应 .txt 文件（如 "【开心】abc.txt"），prompt_text = .txt 文件内容 + - 若无对应 .txt 文件，prompt_text = "abc"（去掉 "【开心】" 和 ".wav" 的部分） + - 如果无 "【xxx】" 前缀： + - 若存在对应 .txt 文件（如 "ref.txt"），prompt_text = .txt 文件内容 + - 若无对应 .txt 文件，prompt_text = "ref"（去掉 ".wav" 的部分） + - 如果未提供 ref_audio_path： + - 从 roles/{role}/reference_audios 中选择音频（基于 text_lang 和 emotion）： + - 优先匹配 "【emotion】" 前缀的音频（如 "【开心】voice1.wav"） + - 若存在对应 .txt 文件（如 "【开心】voice1.txt"），prompt_text = .txt 文件内容 + - 若无对应 .txt 文件，prompt_text = "voice1"（去掉 "【开心】" 和 ".wav" 的部分） + - 未匹配 emotion 则随机选择一个音频，逻辑同上 + +### 讲解 +1. 必填参数: + - /ttsrole: text, role + - /tts: text, ref_audio_path, prompt_lang +2. 音频选择 (/ttsrole): + - 若提供 ref_audio_path，则使用它 + - 否则根据 role、text_lang、emotion 从 roles/{role}/reference_audios 中选择 + - text_lang = "auto" 时，若有 emotion，则跨语言匹配 "【emotion】" 前缀音频 + - emotion 匹配 "【emotion】" 前缀音频，未匹配则随机选择 +3. 设备选择: + - 默认尝试检测显卡（torch.cuda.is_available()），若可用则用 "cuda"，否则 "cpu" + - 若缺少 torch 依赖或检测失败，回退到 "cpu" + - POST 参数 device, t2s_model_device, vits_model_device 可强制指定设备，优先级最高 +4. 配置文件: + - 默认加载 GPT_SoVITS/configs/tts_infer.yaml + - 若 roles/{role}/tts_infer.yaml 存在且未被请求参数覆盖，则使用它 (/ttsrole) + - 请求参数（如 top_k, bert_base_path）覆盖所有配置文件 +5. 返回格式: + - 成功时返回音频流 (Response 或 StreamingResponse) + - 失败时返回 JSON，包含错误消息和可能的异常详情 +6. 运行: + - python api.py -a 127.0.0.1 -p 9880 + - 检查启动日志确认设备 +7. 模型版本切换: + - POST 请求中通过 "version" 参数指定 "v2" 或 "v3"，动态影响推理逻辑。 +""" + +import argparse +import os +import re +import sys +import signal +from time import time as ttime +import torch +import torchaudio +import librosa +import soundfile as sf +from fastapi import FastAPI, Request, Query, HTTPException +from fastapi.responses import StreamingResponse, JSONResponse +import uvicorn +from transformers import AutoModelForMaskedLM, AutoTokenizer +import numpy as np +from feature_extractor import cnhubert +from io import BytesIO +from module.models import SynthesizerTrn, SynthesizerTrnV3 +from peft import LoraConfig, PeftModel, get_peft_model +from AR.models.t2s_lightning_module import Text2SemanticLightningModule +from text import cleaned_text_to_sequence +from text.cleaner import clean_text +from module.mel_processing import spectrogram_torch +from tools.my_utils import load_audio +import config as global_config +import logging +import subprocess +import glob +from typing import Optional, List +from text.LangSegmenter import LangSegmenter +import random + +now_dir = os.getcwd() +sys.path.append(now_dir) +sys.path.append("%s/GPT_SoVITS" % (now_dir)) + +# 日志配置 +logging.config.dictConfig(uvicorn.config.LOGGING_CONFIG) +logger = logging.getLogger('uvicorn') + +# 获取全局配置 +g_config = global_config.Config() + +# 默认参考音频类 +class DefaultRefer: + def __init__(self, path, text, language): + self.path = path + self.text = text + self.language = language + + def is_ready(self) -> bool: + return is_full(self.path, self.text, self.language) + +def is_empty(*items): + for item in items: + if item is not None and item != "": + return False + return True + +def is_full(*items): + for item in items: + if item is None or item == "": + return False + return True + +# 角色和模型定义 +class Speaker: + def __init__(self, name, gpt, sovits, phones=None, bert=None, prompt=None): + self.name = name + self.gpt = gpt + self.sovits = sovits + self.phones = phones + self.bert = bert + self.prompt = prompt + +class Sovits: + def __init__(self, vq_model, hps): + self.vq_model = vq_model + self.hps = hps + +class Gpt: + def __init__(self, max_sec, t2s_model): + self.max_sec = max_sec + self.t2s_model = t2s_model + +# 全局变量 +speaker_list = {} +hz = 50 +bigvgan_model = None + +# BigVGAN 初始化 +def init_bigvgan(): + global bigvgan_model + from BigVGAN import bigvgan + bigvgan_model = bigvgan.BigVGAN.from_pretrained( + "%s/GPT_SoVITS/pretrained_models/models--nvidia--bigvgan_v2_24khz_100band_256x" % (now_dir,), + use_cuda_kernel=False + ) + bigvgan_model.remove_weight_norm() + bigvgan_model = bigvgan_model.eval() + if is_half: + bigvgan_model = bigvgan_model.half().to(device) + else: + bigvgan_model = bigvgan_model.to(device) + +# 模型加载函数 +from process_ckpt import get_sovits_version_from_path_fast, load_sovits_new +def get_sovits_weights(sovits_path): + path_sovits_v3 = "GPT_SoVITS/pretrained_models/s2Gv3.pth" + is_exist_s2gv3 = os.path.exists(path_sovits_v3) + + version, model_version, if_lora_v3 = get_sovits_version_from_path_fast(sovits_path) + if if_lora_v3 and not is_exist_s2gv3: + logger.info("SoVITS V3 底模缺失，无法加载相应 LoRA 权重") + + dict_s2 = load_sovits_new(sovits_path) + hps = dict_s2["config"] + hps = DictToAttrRecursive(hps) + hps.model.semantic_frame_rate = "25hz" + if 'enc_p.text_embedding.weight' not in dict_s2['weight']: + hps.model.version = "v2" + elif dict_s2['weight']['enc_p.text_embedding.weight'].shape[0] == 322: + hps.model.version = "v1" + else: + hps.model.version = "v2" + + if model_version == "v3": + hps.model.version = "v3" + + model_params_dict = vars(hps.model) + if model_version != "v3": + vq_model = SynthesizerTrn( + hps.data.filter_length // 2 + 1, + hps.train.segment_size // hps.data.hop_length, + n_speakers=hps.data.n_speakers, + **model_params_dict + ) + else: + vq_model = SynthesizerTrnV3( + hps.data.filter_length // 2 + 1, + hps.train.segment_size // hps.data.hop_length, + n_speakers=hps.data.n_speakers, + **model_params_dict + ) + init_bigvgan() + logger.info(f"模型版本: {hps.model.version}") + if "pretrained" not in sovits_path: + try: + del vq_model.enc_q + except: + pass + if is_half: + vq_model = vq_model.half().to(device) + else: + vq_model = vq_model.to(device) + vq_model.eval() + if not if_lora_v3: + vq_model.load_state_dict(dict_s2["weight"], strict=False) + else: + vq_model.load_state_dict(load_sovits_new(path_sovits_v3)["weight"], strict=False) + lora_rank = dict_s2["lora_rank"] + lora_config = LoraConfig( + target_modules=["to_k", "to_q", "to_v", "to_out.0"], + r=lora_rank, + lora_alpha=lora_rank, + init_lora_weights=True, + ) + vq_model.cfm = get_peft_model(vq_model.cfm, lora_config) + vq_model.load_state_dict(dict_s2["weight"], strict=False) + vq_model.cfm = vq_model.cfm.merge_and_unload() + vq_model.eval() + + return Sovits(vq_model, hps) + +def get_gpt_weights(gpt_path): + dict_s1 = torch.load(gpt_path, map_location="cpu") + config = dict_s1["config"] + max_sec = config["data"]["max_sec"] + t2s_model = Text2SemanticLightningModule(config, "****", is_train=False) + t2s_model.load_state_dict(dict_s1["weight"]) + if is_half: + t2s_model = t2s_model.half() + t2s_model = t2s_model.to(device) + t2s_model.eval() + return Gpt(max_sec, t2s_model) + +def change_gpt_sovits_weights(gpt_path, sovits_path): + try: + gpt = get_gpt_weights(gpt_path) + sovits = get_sovits_weights(sovits_path) + except Exception as e: + return JSONResponse({"code": 400, "message": str(e)}, status_code=400) + speaker_list["default"] = Speaker(name="default", gpt=gpt, sovits=sovits) + return JSONResponse({"code": 0, "message": "Success"}, status_code=200) + +# 角色配置加载 +def load_role_config(role, vits_weights_path=None, t2s_weights_path=None): + role_dir = os.path.join(now_dir, "roles", role) + if not os.path.exists(role_dir): + return False + gpt_path = t2s_weights_path or (glob.glob(os.path.join(role_dir, "*.ckpt"))[0] if glob.glob(os.path.join(role_dir, "*.ckpt")) else args.gpt_path) + sovits_path = vits_weights_path or (glob.glob(os.path.join(role_dir, "*.pth"))[0] if glob.glob(os.path.join(role_dir, "*.pth")) else args.sovits_path) + speaker_list[role] = Speaker(name=role, gpt=get_gpt_weights(gpt_path), sovits=get_sovits_weights(sovits_path)) + return True + +# 参考音频选择 +def select_ref_audio(role, text_language, emotion=None): + audio_base_dir = os.path.join(now_dir, "roles", role, "reference_audios") + if not os.path.exists(audio_base_dir): + return None, None, None + if text_language.lower() == "auto" and emotion: + all_langs = [d for d in os.listdir(audio_base_dir) if os.path.isdir(os.path.join(audio_base_dir, d))] + emotion_files = [] + for lang in all_langs: + lang_dir = os.path.join(audio_base_dir, lang) + emotion_files.extend(glob.glob(os.path.join(lang_dir, f"【{emotion}】*.*"))) + if emotion_files: + audio_path = random.choice(emotion_files) + txt_path = audio_path.rsplit(".", 1)[0] + ".txt" + prompt_text = open(txt_path, "r", encoding="utf-8").read().strip() if os.path.exists(txt_path) else os.path.basename(audio_path).split("】")[1].rsplit(".", 1)[0] + prompt_language = os.path.basename(os.path.dirname(audio_path)) + return audio_path, prompt_text, prompt_language + lang_dir = os.path.join(audio_base_dir, text_language.lower()) + if os.path.exists(lang_dir): + audio_files = glob.glob(os.path.join(lang_dir, f"【{emotion}】*.*" if emotion else "*.*")) + if audio_files: + audio_path = random.choice(audio_files) + txt_path = audio_path.rsplit(".", 1)[0] + ".txt" + prompt_text = open(txt_path, "r", encoding="utf-8").read().strip() if os.path.exists(txt_path) else os.path.basename(audio_path).rsplit(".", 1)[0] + return audio_path, prompt_text, text_language.lower() + return None, None, None + +# BERT 和文本处理函数 +def get_bert_feature(text, word2ph): + with torch.no_grad(): + inputs = tokenizer(text, return_tensors="pt") + for i in inputs: + inputs[i] = inputs[i].to(device) + res = bert_model(**inputs, output_hidden_states=True) + res = torch.cat(res["hidden_states"][-3:-2], -1)[0].cpu()[1:-1] + assert len(word2ph) == len(text) + phone_level_feature = [] + for i in range(len(word2ph)): + repeat_feature = res[i].repeat(word2ph[i], 1) + phone_level_feature.append(repeat_feature) + phone_level_feature = torch.cat(phone_level_feature, dim=0) + return phone_level_feature.T + +def clean_text_inf(text, language, version): + language = language.replace("all_", "") + phones, word2ph, norm_text = clean_text(text, language, version) + phones = cleaned_text_to_sequence(phones, version) + return phones, word2ph, norm_text + +def get_bert_inf(phones, word2ph, norm_text, language): + language = language.replace("all_", "") + if language == "zh": + bert = get_bert_feature(norm_text, word2ph).to(device) + else: + bert = torch.zeros( + (1024, len(phones)), + dtype=torch.float16 if is_half else torch.float32, + ).to(device) + return bert + +from text import chinese +def get_phones_and_bert(text, language, version, final=False): + if language in {"en", "all_zh", "all_ja", "all_ko", "all_yue"}: + formattext = text + while " " in formattext: + formattext = formattext.replace(" ", " ") + if language == "all_zh": + if re.search(r'[A-Za-z]', formattext): + formattext = re.sub(r'[a-z]', lambda x: x.group(0).upper(), formattext) + formattext = chinese.mix_text_normalize(formattext) + return get_phones_and_bert(formattext, "zh", version) + else: + phones, word2ph, norm_text = clean_text_inf(formattext, language, version) + bert = get_bert_feature(norm_text, word2ph).to(device) + elif language == "all_yue" and re.search(r'[A-Za-z]', formattext): + formattext = re.sub(r'[a-z]', lambda x: x.group(0).upper(), formattext) + formattext = chinese.mix_text_normalize(formattext) + return get_phones_and_bert(formattext, "yue", version) + else: + phones, word2ph, norm_text = clean_text_inf(formattext, language, version) + bert = torch.zeros( + (1024, len(phones)), + dtype=torch.float16 if is_half else torch.float32, + ).to(device) + elif language in {"zh", "ja", "ko", "yue", "auto", "auto_yue"}: + textlist = [] + langlist = [] + if language == "auto": + for tmp in LangSegmenter.getTexts(text): + langlist.append(tmp["lang"]) + textlist.append(tmp["text"]) + elif language == "auto_yue": + for tmp in LangSegmenter.getTexts(text): + if tmp["lang"] == "zh": + tmp["lang"] = "yue" + langlist.append(tmp["lang"]) + textlist.append(tmp["text"]) + else: + for tmp in LangSegmenter.getTexts(text): + if tmp["lang"] == "en": + langlist.append(tmp["lang"]) + else: + langlist.append(language) + textlist.append(tmp["text"]) + phones_list = [] + bert_list = [] + norm_text_list = [] + for i in range(len(textlist)): + lang = langlist[i] + phones, word2ph, norm_text = clean_text_inf(textlist[i], lang, version) + bert = get_bert_inf(phones, word2ph, norm_text, lang) + phones_list.append(phones) + norm_text_list.append(norm_text) + bert_list.append(bert) + bert = torch.cat(bert_list, dim=1) + phones = sum(phones_list, []) + norm_text = ''.join(norm_text_list) + if not final and len(phones) < 6: + return get_phones_and_bert("." + text, language, version, final=True) + return phones, bert.to(torch.float16 if is_half else torch.float32), norm_text + +class DictToAttrRecursive(dict): + def __init__(self, input_dict): + super().__init__(input_dict) + for key, value in input_dict.items(): + if isinstance(value, dict): + value = DictToAttrRecursive(value) + self[key] = value + setattr(self, key, value) + + def __getattr__(self, item): + try: + return self[item] + except KeyError: + raise AttributeError(f"Attribute {item} not found") + + def __setattr__(self, key, value): + if isinstance(value, dict): + value = DictToAttrRecursive(value) + super(DictToAttrRecursive, self).__setitem__(key, value) + super().__setattr__(key, value) + +def get_spepc(hps, filename): + audio, _ = librosa.load(filename, int(hps.data.sampling_rate)) + audio = torch.FloatTensor(audio) + maxx = audio.abs().max() + if maxx > 1: + audio /= min(2, maxx) + audio_norm = audio.unsqueeze(0) + spec = spectrogram_torch(audio_norm, hps.data.filter_length, hps.data.sampling_rate, hps.data.hop_length, + hps.data.win_length, center=False) + return spec + +# 音频处理函数 +def pack_audio(audio_bytes, data, rate): + if media_type == "ogg": + audio_bytes = pack_ogg(audio_bytes, data, rate) + elif media_type == "aac": + audio_bytes = pack_aac(audio_bytes, data, rate) + else: + audio_bytes = pack_raw(audio_bytes, data, rate) + return audio_bytes + +def pack_ogg(audio_bytes, data, rate): + with sf.SoundFile(audio_bytes, mode='w', samplerate=rate, channels=1, format='ogg') as audio_file: + audio_file.write(data) + return audio_bytes + +def pack_raw(audio_bytes, data, rate): + audio_bytes.write(data.tobytes()) + return audio_bytes + +def pack_wav(audio_bytes, rate): + if is_int32: + data = np.frombuffer(audio_bytes.getvalue(), dtype=np.int32) + wav_bytes = BytesIO() + sf.write(wav_bytes, data, rate, format='WAV', subtype='PCM_32') + else: + data = np.frombuffer(audio_bytes.getvalue(), dtype=np.int16) + wav_bytes = BytesIO() + sf.write(wav_bytes, data, rate, format='WAV') + return wav_bytes + +def pack_aac(audio_bytes, data, rate): + pcm = 's32le' if is_int32 else 's16le' + bit_rate = '256k' if is_int32 else '128k' + process = subprocess.Popen([ + 'ffmpeg', '-f', pcm, '-ar', str(rate), '-ac', '1', '-i', 'pipe:0', + '-c:a', 'aac', '-b:a', bit_rate, '-vn', '-f', 'adts', 'pipe:1' + ], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + out, _ = process.communicate(input=data.tobytes()) + audio_bytes.write(out) + return audio_bytes + +def read_clean_buffer(audio_bytes): + audio_chunk = audio_bytes.getvalue() + audio_bytes.truncate(0) + audio_bytes.seek(0) + return audio_bytes, audio_chunk + +# 文本切分 +def cut_text(text, punc): + punc_list = [p for p in punc if p in {",", ".", ";", "?", "!", "、", "，", "。", "？", "！", "；", "：", "…"}] + if len(punc_list) > 0: + punds = r"[" + "".join(punc_list) + r"]" + text = text.strip("\n") + items = re.split(f"({punds})", text) + mergeitems = ["".join(group) for group in zip(items[::2], items[1::2])] + if len(items) % 2 == 1: + mergeitems.append(items[-1]) + text = "\n".join(mergeitems) + while "\n\n" in text: + text = text.replace("\n\n", "\n") + return text + +def only_punc(text): + return not any(t.isalnum() or t.isalpha() for t in text) + +splits = {"，", "。", "？", "！", ",", ".", "?", "!", "~", ":", "：", "—", "…"} + + +# TTS 推理函数 +def get_tts_wav(ref_wav_path, prompt_text, prompt_language, text, text_language, top_k=15, top_p=0.6, temperature=0.6, speed=1, inp_refs=None, sample_steps=32, if_sr=False, spk="default", version=None): + infer_sovits = speaker_list[spk].sovits + vq_model = infer_sovits.vq_model + hps = infer_sovits.hps + + # 如果提供了 version 参数，覆盖默认版本 + if version: + hps.model.version = version + + infer_gpt = speaker_list[spk].gpt + t2s_model = infer_gpt.t2s_model + max_sec = infer_gpt.max_sec + + prompt_text = prompt_text.strip("\n") + if prompt_text[-1] not in splits: + prompt_text += "。" if prompt_language != "en" else "." + prompt_language, text = prompt_language, text.strip("\n") + dtype = torch.float16 if is_half else torch.float32 + zero_wav = np.zeros(int(hps.data.sampling_rate * 0.3), dtype=np.float16 if is_half else np.float32) + with torch.no_grad(): + wav16k, sr = librosa.load(ref_wav_path, sr=16000) + wav16k = torch.from_numpy(wav16k) + zero_wav_torch = torch.from_numpy(zero_wav) + if is_half: + wav16k = wav16k.half().to(device) + zero_wav_torch = zero_wav_torch.half().to(device) + else: + wav16k = wav16k.to(device) + zero_wav_torch = zero_wav_torch.to(device) + wav16k = torch.cat([wav16k, zero_wav_torch]) + ssl_content = ssl_model.model(wav16k.unsqueeze(0))["last_hidden_state"].transpose(1, 2) + codes = vq_model.extract_latent(ssl_content) + prompt_semantic = codes[0, 0] + prompt = prompt_semantic.unsqueeze(0).to(device) + + if hps.model.version != "v3": + refers = [] + if inp_refs: + for path in inp_refs: + try: + refer = get_spepc(hps, path).to(dtype).to(device) + refers.append(refer) + except Exception as e: + logger.error(e) + if not refers: + refers = [get_spepc(hps, ref_wav_path).to(dtype).to(device)] + else: + refer = get_spepc(hps, ref_wav_path).to(device).to(dtype) + + prompt_language = dict_language[prompt_language.lower()] + text_language = dict_language[text_language.lower()] + phones1, bert1, norm_text1 = get_phones_and_bert(prompt_text, prompt_language, hps.model.version) + texts = text.split("\n") + audio_bytes = BytesIO() + + for text in texts: + if only_punc(text): + continue + audio_opt = [] + if text[-1] not in splits: + text += "。" if text_language != "en" else "." + phones2, bert2, norm_text2 = get_phones_and_bert(text, text_language, hps.model.version) + bert = torch.cat([bert1, bert2], 1) + + all_phoneme_ids = torch.LongTensor(phones1 + phones2).to(device).unsqueeze(0) + bert = bert.to(device).unsqueeze(0) + all_phoneme_len = torch.tensor([all_phoneme_ids.shape[-1]]).to(device) + with torch.no_grad(): + pred_semantic, idx = t2s_model.model.infer_panel( + all_phoneme_ids, + all_phoneme_len, + prompt, + bert, + top_k=top_k, + top_p=top_p, + temperature=temperature, + early_stop_num=hz * max_sec) + pred_semantic = pred_semantic[:, -idx:].unsqueeze(0) + + if hps.model.version != "v3": + audio = vq_model.decode(pred_semantic, torch.LongTensor(phones2).to(device).unsqueeze(0), + refers, speed=speed).detach().cpu().numpy()[0, 0] + else: + phoneme_ids0 = torch.LongTensor(phones1).to(device).unsqueeze(0) + phoneme_ids1 = torch.LongTensor(phones2).to(device).unsqueeze(0) + fea_ref, ge = vq_model.decode_encp(prompt.unsqueeze(0), phoneme_ids0, refer) + ref_audio, sr = torchaudio.load(ref_wav_path) + ref_audio = ref_audio.to(device).float() + if ref_audio.shape[0] == 2: + ref_audio = ref_audio.mean(0).unsqueeze(0) + if sr != 24000: + ref_audio = torchaudio.transforms.Resample(sr, 24000).to(device)(ref_audio) + mel_fn = lambda x: torchaudio.transforms.MelSpectrogram( + sample_rate=24000, n_fft=1024, win_length=1024, hop_length=256, n_mels=100, f_min=0, f_max=None, center=False + )(x) + mel2 = mel_fn(ref_audio) + mel2 = (mel2 - (-12)) / (2 - (-12)) * 2 - 1 # 简化的 norm_spec + T_min = min(mel2.shape[2], fea_ref.shape[2]) + mel2 = mel2[:, :, :T_min] + fea_ref = fea_ref[:, :, :T_min] + if T_min > 468: + mel2 = mel2[:, :, -468:] + fea_ref = fea_ref[:, :, -468:] + T_min = 468 + chunk_len = 934 - T_min + fea_todo, ge = vq_model.decode_encp(pred_semantic, phoneme_ids1, refer, ge, speed) + cfm_resss = [] + idx = 0 + while True: + fea_todo_chunk = fea_todo[:, :, idx:idx + chunk_len] + if fea_todo_chunk.shape[-1] == 0: + break + idx += chunk_len + fea = torch.cat([fea_ref, fea_todo_chunk], 2).transpose(2, 1) + cfm_res = vq_model.cfm.inference(fea, torch.LongTensor([fea.size(1)]).to(fea.device), mel2, sample_steps, inference_cfg_rate=0) + cfm_res = cfm_res[:, :, mel2.shape[2]:] + mel2 = cfm_res[:, :, -T_min:] + fea_ref = fea_todo_chunk[:, :, -T_min:] + cfm_resss.append(cfm_res) + cmf_res = torch.cat(cfm_resss, 2) + cmf_res = (cmf_res + 1) / 2 * (2 - (-12)) + (-12) # 简化的 denorm_spec + if bigvgan_model is None: + init_bigvgan() + with torch.inference_mode(): + wav_gen = bigvgan_model(cmf_res) + audio = wav_gen[0][0].cpu().detach().numpy() + + max_audio = np.abs(audio).max() + if max_audio > 1: + audio /= max_audio + audio_opt.append(audio) + audio_opt.append(zero_wav) + audio_opt = np.concatenate(audio_opt, 0) + + sr = hps.data.sampling_rate if hps.model.version != "v3" else 24000 + if if_sr and sr == 24000: + audio_opt = torch.from_numpy(audio_opt).float().to(device) + # 简化为无超分逻辑，需自行实现 audio_sr + audio_opt = audio_opt.cpu().detach().numpy() + sr = 48000 + + if is_int32: + audio_bytes = pack_audio(audio_bytes, (audio_opt * 2147483647).astype(np.int32), sr) + else: + audio_bytes = pack_audio(audio_bytes, (audio_opt * 32768).astype(np.int16), sr) + if stream_mode == "normal": + audio_bytes, audio_chunk = read_clean_buffer(audio_bytes) + yield audio_chunk + + if stream_mode != "normal": + if media_type == "wav": + sr = 48000 if if_sr else 24000 + sr = hps.data.sampling_rate if hps.model.version != "v3" else sr + audio_bytes = pack_wav(audio_bytes, sr) + yield audio_bytes.getvalue() + +# 接口处理函数 +def handle_control(command): + if command == "restart": + os.execl(g_config.python_exec, g_config.python_exec, *sys.argv) + elif command == "exit": + os.kill(os.getpid(), signal.SIGTERM) + exit(0) + +def handle_change(path, text, language): + if is_empty(path, text, language): + return JSONResponse({"code": 400, "message": '缺少任意一项以下参数: "path", "text", "language"'}, status_code=400) + if path: + default_refer.path = path + if text: + default_refer.text = text + if language: + default_refer.language = language + logger.info(f"当前默认参考音频路径: {default_refer.path}") + logger.info(f"当前默认参考音频文本: {default_refer.text}") + logger.info(f"当前默认参考音频语种: {default_refer.language}") + return JSONResponse({"code": 0, "message": "Success"}, status_code=200) + +def handle(refer_wav_path, prompt_text, prompt_language, text, text_language, cut_punc, top_k, top_p, temperature, speed, inp_refs, sample_steps, if_sr): + if not refer_wav_path or not prompt_text or not prompt_language: + refer_wav_path, prompt_text, prompt_language = default_refer.path, default_refer.text, default_refer.language + if not default_refer.is_ready(): + return JSONResponse({"code": 400, "message": "未指定参考音频且接口无预设"}, status_code=400) + if sample_steps not in [4, 8, 16, 32]: + sample_steps = 32 + if cut_punc is None: + text = cut_text(text, default_cut_punc) + else: + text = cut_text(text, cut_punc) + return StreamingResponse(get_tts_wav(refer_wav_path, prompt_text, prompt_language, text, text_language, top_k, top_p, temperature, speed, inp_refs, sample_steps, if_sr), media_type="audio/"+media_type) + +def handle_ttsrole(text, role, text_language="auto", ref_audio_path=None, prompt_text=None, prompt_language=None, emotion=None, top_k=15, top_p=0.6, temperature=0.6, speed=1, inp_refs=None, sample_steps=32, if_sr=False, version=None, vits_weights_path=None, t2s_weights_path=None): + if not text or not role: + return JSONResponse({"code": 400, "message": "text and role are required"}, status_code=400) + if role not in speaker_list: + if not load_role_config(role, vits_weights_path, t2s_weights_path): + return JSONResponse({"code": 400, "message": f"Role {role} not found"}, status_code=400) + if not ref_audio_path: + ref_audio_path, prompt_text_auto, prompt_lang_auto = select_ref_audio(role, text_language, emotion) + if ref_audio_path: + ref_audio_path, prompt_text, prompt_language = ref_audio_path, prompt_text_auto or prompt_text, prompt_lang_auto or prompt_language + else: + ref_audio_path, prompt_text, prompt_language = default_refer.path, default_refer.text, default_refer.language + if not default_refer.is_ready(): + return JSONResponse({"code": 400, "message": "No reference audio provided and default not set"}, status_code=400) + if sample_steps not in [4, 8, 16, 32]: + sample_steps = 32 + text = cut_text(text, default_cut_punc) + return StreamingResponse(get_tts_wav(ref_audio_path, prompt_text, prompt_language, text, text_language, top_k, top_p, temperature, speed, inp_refs, sample_steps, if_sr, spk=role, version=version), media_type="audio/"+media_type) + +# 初始化参数 +dict_language = { + "中文": "all_zh", "粤语": "all_yue", "英文": "en", "日文": "all_ja", "韩文": "all_ko", + "中英混合": "zh", "粤英混合": "yue", "日英混合": "ja", "韩英混合": "ko", "多语种混合": "auto", + "多语种混合(粤语)": "auto_yue", "all_zh": "all_zh", "all_yue": "all_yue", "en": "en", + "all_ja": "all_ja", "all_ko": "all_ko", "zh": "zh", "yue": "yue", "ja": "ja", "ko": "ko", + "auto": "auto", "auto_yue": "auto_yue" +} + +parser = argparse.ArgumentParser(description="GPT-SoVITS api") +parser.add_argument("-s", "--sovits_path", type=str, default=g_config.sovits_path, help="SoVITS模型路径") +parser.add_argument("-g", "--gpt_path", type=str, default=g_config.gpt_path, help="GPT模型路径") +parser.add_argument("-dr", "--default_refer_path", type=str, default="", help="默认参考音频路径") +parser.add_argument("-dt", "--default_refer_text", type=str, default="", help="默认参考音频文本") +parser.add_argument("-dl", "--default_refer_language", type=str, default="", help="默认参考音频语种") +parser.add_argument("-d", "--device", type=str, default=g_config.infer_device, help="cuda / cpu") +parser.add_argument("-a", "--bind_addr", type=str, default="0.0.0.0", help="default: 0.0.0.0") +parser.add_argument("-p", "--port", type=int, default=g_config.api_port, help="default: 9880") +parser.add_argument("-fp", "--full_precision", action="store_true", default=False, help="使用全精度") +parser.add_argument("-hp", "--half_precision", action="store_true", default=False, help="使用半精度") +parser.add_argument("-sm", "--stream_mode", type=str, default="close", help="流式返回模式, close / normal") +parser.add_argument("-mt", "--media_type", type=str, default="wav", help="音频编码格式, wav / ogg / aac") +parser.add_argument("-st", "--sub_type", type=str, default="int16", help="音频数据类型, int16 / int32") +parser.add_argument("-cp", "--cut_punc", type=str, default="", help="文本切分符号设定") +parser.add_argument("-hb", "--hubert_path", type=str, default=g_config.cnhubert_path, help="覆盖config.cnhubert_path") +parser.add_argument("-b", "--bert_path", type=str, default=g_config.bert_path, help="覆盖config.bert_path") + +args = parser.parse_args() +sovits_path = args.sovits_path +gpt_path = args.gpt_path +device = args.device +port = args.port +host = args.bind_addr +cnhubert_base_path = args.hubert_path +bert_path = args.bert_path +default_cut_punc = args.cut_punc + +default_refer = DefaultRefer(args.default_refer_path, args.default_refer_text, args.default_refer_language) +if not default_refer.path or not default_refer.text or not default_refer.language: + default_refer.path, default_refer.text, default_refer.language = "", "", "" + logger.info("未指定默认参考音频") +else: + logger.info(f"默认参考音频路径: {default_refer.path}") + logger.info(f"默认参考音频文本: {default_refer.text}") + logger.info(f"默认参考音频语种: {default_refer.language}") + +is_half = g_config.is_half +if args.full_precision: + is_half = False +if args.half_precision: + is_half = True +if args.full_precision and args.half_precision: + is_half = g_config.is_half +logger.info(f"半精: {is_half}") + +stream_mode = "normal" if args.stream_mode.lower() in ["normal", "n"] else "close" +logger.info(f"流式返回: {'开启' if stream_mode == 'normal' else '关闭'}") + +media_type = args.media_type.lower() if args.media_type.lower() in ["aac", "ogg"] else ("wav" if stream_mode == "close" else "ogg") +logger.info(f"编码格式: {media_type}") + +is_int32 = args.sub_type.lower() == 'int32' +logger.info(f"数据类型: {'int32' if is_int32 else 'int16'}") + +# 模型初始化 +cnhubert.cnhubert_base_path = cnhubert_base_path +tokenizer = AutoTokenizer.from_pretrained(bert_path) +bert_model = AutoModelForMaskedLM.from_pretrained(bert_path) +ssl_model = cnhubert.get_model() +if is_half: + bert_model = bert_model.half().to(device) + ssl_model = ssl_model.half().to(device) +else: + bert_model = bert_model.to(device) + ssl_model = ssl_model.to(device) +change_gpt_sovits_weights(gpt_path=gpt_path, sovits_path=sovits_path) + +# FastAPI 应用 +app = FastAPI() + +@app.post("/set_model") +async def set_model(request: Request): + json_post_raw = await request.json() + return change_gpt_sovits_weights( + gpt_path=json_post_raw.get("gpt_model_path"), + sovits_path=json_post_raw.get("sovits_model_path") + ) + +@app.get("/set_model") +async def set_model(gpt_model_path: str = None, sovits_model_path: str = None): + return change_gpt_sovits_weights(gpt_path=gpt_model_path, sovits_path=sovits_model_path) + +@app.post("/control") +async def control(request: Request): + json_post_raw = await request.json() + return handle_control(json_post_raw.get("command")) + +@app.get("/control") +async def control(command: str = None): + return handle_control(command) + +@app.post("/change_refer") +async def change_refer(request: Request): + json_post_raw = await request.json() + return handle_change( + json_post_raw.get("refer_wav_path"), + json_post_raw.get("prompt_text"), + json_post_raw.get("prompt_language") + ) + +@app.get("/change_refer") +async def change_refer(refer_wav_path: str = None, prompt_text: str = None, prompt_language: str = None): + return handle_change(refer_wav_path, prompt_text, prompt_language) + +@app.post("/") +async def tts_endpoint(request: Request): + json_post_raw = await request.json() + return handle( + json_post_raw.get("refer_wav_path"), + json_post_raw.get("prompt_text"), + json_post_raw.get("prompt_language"), + json_post_raw.get("text"), + json_post_raw.get("text_language"), + json_post_raw.get("cut_punc"), + json_post_raw.get("top_k", 15), + json_post_raw.get("top_p", 1.0), + json_post_raw.get("temperature", 1.0), + json_post_raw.get("speed", 1.0), + json_post_raw.get("inp_refs", []), + json_post_raw.get("sample_steps", 32), + json_post_raw.get("if_sr", False) + ) + +@app.get("/") +async def tts_endpoint( + refer_wav_path: str = None, prompt_text: str = None, prompt_language: str = None, text: str = None, text_language: str = None, + cut_punc: str = None, top_k: int = 15, top_p: float = 1.0, temperature: float = 1.0, speed: float = 1.0, inp_refs: list = Query(default=[]), + sample_steps: int = 32, if_sr: bool = False +): + return handle(refer_wav_path, prompt_text, prompt_language, text, text_language, cut_punc, top_k, top_p, temperature, speed, inp_refs, sample_steps, if_sr) + +@app.post("/ttsrole") +async def ttsrole_endpoint(request: Request): + json_post_raw = await request.json() + return handle_ttsrole( + json_post_raw.get("text"), + json_post_raw.get("role"), + json_post_raw.get("text_lang", "auto"), + json_post_raw.get("ref_audio_path"), + json_post_raw.get("prompt_text"), + json_post_raw.get("prompt_lang"), + json_post_raw.get("emotion"), + json_post_raw.get("top_k", 15), + json_post_raw.get("top_p", 0.6), + json_post_raw.get("temperature", 0.6), + json_post_raw.get("speed_factor", 1.0), + json_post_raw.get("aux_ref_audio_paths", []), + json_post_raw.get("sample_steps", 32), + json_post_raw.get("if_sr", False), + json_post_raw.get("version"), # 支持动态切换版本 + json_post_raw.get("vits_weights_path"), # 支持动态指定模型路径 + json_post_raw.get("t2s_weights_path") + ) + +@app.get("/ttsrole") +async def ttsrole_endpoint( + text: str, role: str, text_language: str = "auto", ref_audio_path: Optional[str] = None, prompt_text: Optional[str] = None, + prompt_language: Optional[str] = None, emotion: Optional[str] = None, top_k: int = 15, top_p: float = 0.6, + temperature: float = 0.6, speed: float = 1.0, inp_refs: list = Query(default=[]), sample_steps: int = 32, if_sr: bool = False, version: Optional[str] = None +): + return handle_ttsrole(text, role, text_language, ref_audio_path, prompt_text, prompt_language, emotion, top_k, top_p, temperature, speed, inp_refs, sample_steps, if_sr, version) + +if __name__ == "__main__": + uvicorn.run(app, host=host, port=port, workers=1) From c9308ec96ba705b01f294db086c7a56907041b25 Mon Sep 17 00:00:00 2001 From: spawner Date: Fri, 7 Mar 2025 15:03:56 +0800 Subject: [PATCH 3/3] Update api_role_v3.py --- api_role_v3.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/api_role_v3.py b/api_role_v3.py index d46ee7a..23a7a69 100644 --- a/api_role_v3.py +++ b/api_role_v3.py @@ -17,7 +17,7 @@ - 确保 GPT 和 SoVITS 模型文件已准备好。 - 可选：设置默认参考音频路径、文本和语言。 3. 运行服务： - python api.py -s "path/to/sovits.pth" -g "path/to/gpt.ckpt" -dr "ref.wav" -dt "参考文本" -dl "zh" -p 9880 + python api_role_v3.py -s "path/to/sovits.pth" -g "path/to/gpt.ckpt" -dr "ref.wav" -dt "参考文本" -dl "zh" -p 9880 参数说明： 命令行参数： @@ -117,7 +117,7 @@ ### 目录结构 GPT-SoVITS-roleapi/ -├── api.py # 本文件, API 主程序 +├── api_role_v3.py # 本文件, API 主程序 ├── GPT_SoVITS/ # GPT-SoVITS 核心库 │ └── configs/ │ └── tts_infer.yaml # 默认配置文件 @@ -201,7 +201,7 @@ GPT-SoVITS-roleapi/ - 成功时返回音频流 (Response 或 StreamingResponse) - 失败时返回 JSON，包含错误消息和可能的异常详情 6. 运行: - - python api.py -a 127.0.0.1 -p 9880 + - python api_role_v3.py -a 127.0.0.1 -p 9880 - 检查启动日志确认设备 7. 模型版本切换: - POST 请求中通过 "version" 参数指定 "v2" 或 "v3"，动态影响推理逻辑。