main.py

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 #当期prompt
        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()