client3/lib_tts_grpc.py

# -*- encoding: utf-8 -*-
"""
@Author: 程尧
@Desc: 语音合成（朗读）功能
#  合成小语种需要传输小语种文本、使用小语种发音人vcn、tte=unicode以及修改文本编码方式
#  错误码链接：https://www.xfyun.cn/document/error-code （code返回错误码时必看）

"""
import time

import grpc
import tts_pb2
import tts_pb2_grpc
import datetime
import hashlib
import base64
import hmac
import json
import ssl
import _thread as thread
import os
from threading import Thread
import sys
import random
from urllib.parse import quote
import config

URL = '124.243.226.40:17009'


def get_proxy():
    _proxy = ""
    proxy_type = ""
    host = ""
    port = ""
    user = ""
    pswd = ""
    try:
        _proxy = os.environ["http_proxy"]
    except:
        pass
    else:
        try:
            addr = _proxy.split("//")[1]
            proxy_type_str = _proxy.split("//")[0]
            proxy_type = proxy_type_str[:-1]
            port = addr.split(":")[-1]
            if "@" in addr:  # 有用户密码
                host_str = addr.split("@")[-1]
                host = host_str.split(f":{port}")[0]
                auth_str = addr.split(host_str)[0]
                user = auth_str.split(":")[0]
                pswd_str = auth_str.split(":")[-1]
                pswd = pswd_str[:-1]
            else:
                host = addr.split(":")[0]
        except:
            pass
    return _proxy, proxy_type, host, port, user, pswd


def get_home_env():
    if sys.platform == "win32":
        return 'APPDATA'
    return 'HOME'


def gen_tmp_mp3_file():
    _dir = os.path.join(os.environ[get_home_env()], "iFLYAssistant")
    if not os.path.exists(_dir):
        os.makedirs(_dir)
    _tmp_dir = os.path.join(_dir, "tmp")
    if not os.path.exists(_tmp_dir):
        os.makedirs(_tmp_dir)
    _path = os.path.join(_tmp_dir, f"{int(time.time())}.mp3")
    return _path


def get_nonce():
    random_number = random.randint(1, 100)
    return random_number


# TTS工作线程
class TTSManagerNew(Thread):
    def __init__(self, text, vcn, speed_mode, audio_queue):
        super(TTSManagerNew, self).__init__()
        self.daemon = True
        self.audio_queue = audio_queue
        self.sid = str(random.randint(100000, 999999))
        self.text = text

        self.sessionParam = {"sid": self.sid, "sample_rate": "16000", "frame_size": "640", "audio_coding": "raw",
                             "native_voice_name": "bingjie",
                             "speed": "0", "volume": "10", "pitch": "10"}

    def generate_requests(self):
        yield tts_pb2.TtsRequest(
            sessionParam=self.sessionParam, endFlag=False)

        yield tts_pb2.TtsRequest(
            text=self.text,
            endFlag=False)

        yield tts_pb2.TtsRequest(endFlag=True)

    def run(self):
        thread.start_new_thread(self._grpc_request, ())

    def _grpc_request(self):
        try:
            with grpc.insecure_channel('124.243.226.40:17009') as channel:
                stub = tts_pb2_grpc.TtsStub(channel)

                responses = stub.createRec(self.generate_requests())

                for response in responses:
                    # print("Received response:", response)
                    if response.data:
                        self.audio_queue.put(response.data)
                    if response.endFlag:
                        # 合成结束，音频流添加--end--，通知lib_player类中的_write_stream结束_tts_thread线程
                        self.audio_queue.put("--end--")
                        print("Audio synthesis finished.")
                        break
        except Exception as e:
            print("Exception occurred:", e)
        finally:
            channel.close()

    def stop(self):
        pass
        # self._ws.close()