Spaces:
Sleeping
Sleeping
File size: 3,839 Bytes
558c90a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 |
import io, wave
import os, json, sys
import threading
from Synthesizers.base import Base_TTS_Synthesizer ,load_config
from .remote_task import Remote_TTS_Task as TTS_Task, set_based_synthesizer, get_ui_config
import requests
from urllib import parse
from datetime import datetime
from typing import Union, Generator, Tuple, Any, Optional, Dict, Literal
import numpy as np
import soundfile as sf
class Remote_Synthesizer(Base_TTS_Synthesizer):
url :str = "http://127.0.0.1:5000"
tts_endpoint:str = "/tts"
character_endpoint:str = "/character_list"
based_synthesizer :str = "gsv_fast"
class Config:
extra = "ignore"
def __init__(self, config_path:str = None, **kwargs):
super().__init__(**kwargs)
if config_path is None:
config_path = os.path.join(os.path.dirname(__file__), "configs", "config.json")
config_dict = load_config(config_path)
config_dict.update(kwargs)
for key, value in config_dict.items():
if hasattr(self, key):
setattr(self, key, value)
set_based_synthesizer(self.based_synthesizer)
self.ui_config = get_ui_config(self.based_synthesizer)
def get_characters(self)-> dict:
url = self.url + self.character_endpoint
res = requests.get(url)
return json.loads(res.text)
@staticmethod
def stream_audio(url, data: Dict[str, Any]) -> Generator[Tuple[int, np.ndarray], None, None]:
headers = {"Content-Type": "application/json"}
# 发起POST请求，获取响应流
response = requests.post(
url, data=json.dumps(data), headers=headers, stream=True
)
chunk_size = 1024
# 确保请求成功
if response.status_code == 200:
# 循环读取音频流
for chunk in response.iter_content(chunk_size):
# 将二进制数据转换为numpy数组，这里假设音频数据是16位整数格式
audiodata = np.frombuffer(chunk, dtype=np.int16)
yield 32000, audiodata
else:
raise Exception(
f"Failed to get audio stream, status code: {response.status_code}"
)
def generate(
self,
task: TTS_Task,
return_type: Literal["filepath", "numpy"] = "numpy",
save_path: Optional[str] = None,
) -> Union[str, Generator[Tuple[int, np.ndarray], None, None], Any]:
url = self.url + self.tts_endpoint
data = task.data
print(return_type)
if self.debug_mode:
print(f"generate task: \n{data}")
headers = {"Content-Type": "application/json"}
if return_type == "filepath" or (
return_type == "numpy" and not task.stream
):
if save_path is None:
save_path = f"tmp_audio/{datetime.now().strftime('%Y%m%d%H%M%S')}.wav"
res = requests.post(url, data=json.dumps(data), headers=headers)
if res.status_code == 200:
with open(save_path, "wb") as f:
f.write(res.content)
if return_type == "filepath":
return save_path
else:
audiodata, sr = sf.read(save_path)
return ((sr, audiodata) for _ in range(1))
else:
raise Exception(f"remote synthesizer error: {res.text}")
elif return_type == "numpy" and task.stream:
return self.stream_audio(url, data)
def params_parser(self, data) -> TTS_Task:
task = TTS_Task(based_synthesizer=self.based_synthesizer, **data)
return task
def ms_like_parser(self,data) -> TTS_Task:
task = TTS_Task(based_synthesizer=self.based_synthesizer, **data)
return task
|