|
import logging |
|
logging.getLogger('numba').setLevel(logging.WARNING) |
|
logging.getLogger('matplotlib').setLevel(logging.WARNING) |
|
logging.getLogger('urllib3').setLevel(logging.WARNING) |
|
from text import text_to_sequence |
|
import numpy as np |
|
from scipy.io import wavfile |
|
import torch |
|
import json |
|
import commons |
|
import utils |
|
import sys |
|
import pathlib |
|
import onnxruntime as ort |
|
import gradio as gr |
|
import argparse |
|
import time |
|
import os |
|
import io |
|
from scipy.io.wavfile import write |
|
from flask import Flask, request |
|
from threading import Thread |
|
import openai |
|
import requests |
|
class VitsGradio: |
|
def __init__(self): |
|
self.lan = ["中文","日文","自动"] |
|
self.chatapi = ["gpt-3.5-turbo","gpt3"] |
|
self.modelPaths = [] |
|
for root,dirs,files in os.walk("checkpoints"): |
|
for dir in dirs: |
|
self.modelPaths.append(dir) |
|
with gr.Blocks() as self.Vits: |
|
with gr.Tab("调试用"): |
|
with gr.Row(): |
|
with gr.Column(): |
|
with gr.Row(): |
|
with gr.Column(): |
|
self.text = gr.TextArea(label="Text", value="你好") |
|
with gr.Accordion(label="测试api", open=False): |
|
self.local_chat1 = gr.Checkbox(value=False, label="使用网址+文本进行模拟") |
|
self.url_input = gr.TextArea(label="键入测试", value="http://127.0.0.1:8080/chat?Text=") |
|
butto = gr.Button("模拟前端抓取语音文件") |
|
btnVC = gr.Button("测试tts+对话程序") |
|
with gr.Column(): |
|
output2 = gr.TextArea(label="回复") |
|
output1 = gr.Audio(label="采样率22050") |
|
output3 = gr.outputs.File(label="44100hz: output.wav") |
|
butto.click(self.Simul, inputs=[self.text, self.url_input], outputs=[output2,output3]) |
|
btnVC.click(self.tts_fn, inputs=[self.text], outputs=[output1,output2]) |
|
with gr.Tab("控制面板"): |
|
with gr.Row(): |
|
with gr.Column(): |
|
with gr.Row(): |
|
with gr.Column(): |
|
self.api_input1 = gr.TextArea(label="输入api-key或本地存储说话模型的路径", value="https://platform.openai.com/account/api-keys") |
|
with gr.Accordion(label="chatbot选择", open=False): |
|
self.api_input2 = gr.Checkbox(value=True, label="采用gpt3.5") |
|
self.local_chat1 = gr.Checkbox(value=False, label="启动本地chatbot") |
|
self.local_chat2 = gr.Checkbox(value=True, label="是否量化") |
|
res = gr.TextArea() |
|
Botselection = gr.Button("完成chatbot设定") |
|
Botselection.click(self.check_bot, inputs=[self.api_input1,self.api_input2,self.local_chat1,self.local_chat2], outputs = [res]) |
|
self.input1 = gr.Dropdown(label = "模型", choices = self.modelPaths, value = self.modelPaths[0], type = "value") |
|
self.input2 = gr.Dropdown(label="Language", choices=self.lan, value="自动", interactive=True) |
|
with gr.Column(): |
|
btnVC = gr.Button("完成vits TTS端设定") |
|
self.input3 = gr.Dropdown(label="Speaker", choices=list(range(101)), value=0, interactive=True) |
|
self.input4 = gr.Slider(minimum=0, maximum=1.0, label="更改噪声比例(noise scale)，以控制情感", value=0.267) |
|
self.input5 = gr.Slider(minimum=0, maximum=1.0, label="更改噪声偏差(noise scale w)，以控制音素长短", value=0.7) |
|
self.input6 = gr.Slider(minimum=0.1, maximum=10, label="duration", value=1) |
|
statusa = gr.TextArea() |
|
btnVC.click(self.create_tts_fn, inputs=[self.input1, self.input2, self.input3, self.input4, self.input5, self.input6], outputs = [statusa]) |
|
|
|
def Simul(self,text,url_input): |
|
web = url_input + text |
|
res = requests.get(web) |
|
music = res.content |
|
with open('output.wav', 'wb') as code: |
|
code.write(music) |
|
file_path = "output.wav" |
|
return web,file_path |
|
|
|
|
|
def chatgpt(self,text): |
|
self.messages.append({"role": "user", "content": text},) |
|
chat = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages= self.messages) |
|
reply = chat.choices[0].message.content |
|
return reply |
|
|
|
def ChATGLM(self,text): |
|
if text == 'clear': |
|
self.history = [] |
|
response, new_history = self.model.chat(self.tokenizer, text, self.history) |
|
response = response.replace(" ",'').replace("\n",'.') |
|
self.history = new_history |
|
return response |
|
|
|
def gpt3_chat(self,text): |
|
call_name = "Waifu" |
|
openai.api_key = args.key |
|
identity = "" |
|
start_sequence = '\n'+str(call_name)+':' |
|
restart_sequence = "\nYou: " |
|
if 1 == 1: |
|
prompt0 = text |
|
if text == 'quit': |
|
return prompt0 |
|
prompt = identity + prompt0 + start_sequence |
|
response = openai.Completion.create( |
|
model="text-davinci-003", |
|
prompt=prompt, |
|
temperature=0.5, |
|
max_tokens=1000, |
|
top_p=1.0, |
|
frequency_penalty=0.5, |
|
presence_penalty=0.0, |
|
stop=["\nYou:"] |
|
) |
|
return response['choices'][0]['text'].strip() |
|
|
|
def check_bot(self,api_input1,api_input2,local_chat1,local_chat2): |
|
if local_chat1: |
|
from transformers import AutoTokenizer, AutoModel |
|
self.tokenizer = AutoTokenizer.from_pretrained(api_input1, trust_remote_code=True) |
|
if local_chat2: |
|
self.model = AutoModel.from_pretrained(api_input1, trust_remote_code=True).half().quantize(4).cuda() |
|
else: |
|
self.model = AutoModel.from_pretrained(api_input1, trust_remote_code=True) |
|
self.history = [] |
|
else: |
|
self.messages = [] |
|
openai.api_key = api_input1 |
|
return "Finished" |
|
|
|
def is_japanese(self,string): |
|
for ch in string: |
|
if ord(ch) > 0x3040 and ord(ch) < 0x30FF: |
|
return True |
|
return False |
|
|
|
def is_english(self,string): |
|
import re |
|
pattern = re.compile('^[A-Za-z0-9.,:;!?()_*"\' ]+$') |
|
if pattern.fullmatch(string): |
|
return True |
|
else: |
|
return False |
|
|
|
def get_symbols_from_json(self,path): |
|
assert os.path.isfile(path) |
|
with open(path, 'r') as f: |
|
data = json.load(f) |
|
return data['symbols'] |
|
|
|
def sle(self,language,text): |
|
text = text.replace('\n','。').replace(' ',',') |
|
if language == "中文": |
|
tts_input1 = "[ZH]" + text + "[ZH]" |
|
return tts_input1 |
|
elif language == "自动": |
|
tts_input1 = f"[JA]{text}[JA]" if self.is_japanese(text) else f"[ZH]{text}[ZH]" |
|
return tts_input1 |
|
elif language == "日文": |
|
tts_input1 = "[JA]" + text + "[JA]" |
|
return tts_input1 |
|
|
|
def get_text(self,text,hps_ms): |
|
text_norm = text_to_sequence(text,hps_ms.data.text_cleaners) |
|
if hps_ms.data.add_blank: |
|
text_norm = commons.intersperse(text_norm, 0) |
|
text_norm = torch.LongTensor(text_norm) |
|
return text_norm |
|
|
|
def create_tts_fn(self,path, input2, input3, n_scale= 0.667,n_scale_w = 0.8, l_scale = 1 ): |
|
self.symbols = self.get_symbols_from_json(f"checkpoints/{path}/config.json") |
|
self.hps = utils.get_hparams_from_file(f"checkpoints/{path}/config.json") |
|
phone_dict = { |
|
symbol: i for i, symbol in enumerate(self.symbols) |
|
} |
|
self.ort_sess = ort.InferenceSession(f"checkpoints/{path}/model.onnx") |
|
self.language = input2 |
|
self.speaker_id = input3 |
|
self.n_scale = n_scale |
|
self.n_scale_w = n_scale_w |
|
self.l_scale = l_scale |
|
print(self.language,self.speaker_id,self.n_scale) |
|
return 'success' |
|
|
|
def tts_fn(self,text): |
|
if self.local_chat1: |
|
text = self.chatgpt(text) |
|
elif self.api_input2: |
|
text = self.ChATGLM(text) |
|
else: |
|
text = self.gpt3_chat(text) |
|
print(text) |
|
text =self.sle(self.language,text) |
|
seq = text_to_sequence(text, cleaner_names=self.hps.data.text_cleaners) |
|
if self.hps.data.add_blank: |
|
seq = commons.intersperse(seq, 0) |
|
with torch.no_grad(): |
|
x = np.array([seq], dtype=np.int64) |
|
x_len = np.array([x.shape[1]], dtype=np.int64) |
|
sid = np.array([self.speaker_id], dtype=np.int64) |
|
scales = np.array([self.n_scale, self.n_scale_w, self.l_scale], dtype=np.float32) |
|
scales.resize(1, 3) |
|
ort_inputs = { |
|
'input': x, |
|
'input_lengths': x_len, |
|
'scales': scales, |
|
'sid': sid |
|
} |
|
t1 = time.time() |
|
audio = np.squeeze(self.ort_sess.run(None, ort_inputs)) |
|
audio *= 32767.0 / max(0.01, np.max(np.abs(audio))) * 0.6 |
|
audio = np.clip(audio, -32767.0, 32767.0) |
|
t2 = time.time() |
|
spending_time = "推理时间："+str(t2-t1)+"s" |
|
print(spending_time) |
|
bytes_wav = bytes() |
|
byte_io = io.BytesIO(bytes_wav) |
|
wavfile.write('moe/temp1.wav',self.hps.data.sampling_rate, audio.astype(np.int16)) |
|
cmd = 'ffmpeg -y -i ' + 'moe/temp1.wav' + ' -ar 44100 ' + 'moe/temp2.wav' |
|
os.system(cmd) |
|
return (self.hps.data.sampling_rate, audio),text.replace('[JA]','').replace('[ZH]','') |
|
|
|
app = Flask(__name__) |
|
print("开始部署") |
|
grVits = VitsGradio() |
|
|
|
@app.route('/chat') |
|
def text_api(): |
|
message = request.args.get('Text','') |
|
audio,text = grVits.tts_fn(message) |
|
text = text.replace('[JA]','').replace('[ZH]','') |
|
with open('moe/temp2.wav','rb') as bit: |
|
wav_bytes = bit.read() |
|
headers = { |
|
'Content-Type': 'audio/wav', |
|
'Text': text.encode('utf-8')} |
|
return wav_bytes, 200, headers |
|
|
|
def gradio_interface(): |
|
return grVits.Vits.launch() |
|
|
|
if __name__ == '__main__': |
|
api_thread = Thread(target=app.run, args=("0.0.0.0", 8080)) |
|
gradio_thread = Thread(target=gradio_interface) |
|
api_thread.start() |
|
gradio_thread.start() |
