client3-vad/main_aiui.py

"""
调用讯飞演示笔获取录音
QObject::connect: Cannot queue arguments of type 'QTextCursor'
(Make sure 'QTextCursor' is registered using qRegisterMetaType().)
"""
import os
import struct
import sys
from functools import partial
from pathlib import Path
import json

from PyQt5 import uic, QtCore
from PyQt5.QtCore import QUrl
from PyQt5.QtGui import QIcon, QDesktopServices
from PyQt5.QtWidgets import QMainWindow, QApplication

import mspeech_ui_thr as mspeech
from libs import lib_opus, lib_aiui, lib_player
from util import constants
from BusinessHadler import BusinessHadler

socketHadler = None  # 全局变量

OPUS_FLAG = False
TTS_FLAG = False


class MainUI(QMainWindow):

    def __init__(self):
        super().__init__()
        self._run_flag = False
        self.tts_player = None

        self.init_thr()

    def init_thr(self):
        # 子线程
        self._mspeech = mspeech.Mspeech()
        self._mspeech.daemon = True
        self._mspeech.start()
        # 处理子线程发送的数据
        self._mspeech.sign_thread_send.connect(self._proc)

    # 业务处理（处理子进程和子线程发送的数据）
    def _proc(self, result):
        try:
            code = result["code"]
            if code != constants.MSPEECH_AIUI_SEND_DATA:
                print("UI_main_proc_real", code)
            if code == constants.MSPEECH_AIUI_SEND_DATA:  # aiui
                QtCore.QTimer.singleShot(0, partial(self._ist.audio_write, result["data"]))
            elif code == constants.MSPEECH_AIUI_RESET_DICTATION:  # aiui
                self._start_record()
            elif code == constants.MSPEECH_AIUI_STOP_WS:  # aiui
                self._stop_record()
        except Exception as e:
            print(e)
            pass

    def _load_ui(self):
        self.ui = uic.loadUi("ui/form.ui")
        self.ui.setWindowTitle("测试工具")
        self.ui.setWindowIcon(QIcon(str(Path.cwd().joinpath("images", "logo.ico"))))
        # self.ui.btn_filedialog.clicked.connect(self._open_dir)
        # self.ui.btn_start.clicked.connect(self._start_record)
        # self.ui.btn_stop.clicked.connect(self._stop_record)
        self.ui.show()

    def _open_dir(self):
        _dir = self.gen_home_doc_path()
        if sys.platform == "win32":
            filepath = _dir.replace("\\", "/")
            QDesktopServices.openUrl(QUrl(filepath, QUrl.StrictMode))
        else:
            os.popen("open %s" % _dir)

    def _logger(self, text):
        if self.ui:
            self.ui.textBrowser.append(text)

    def gen_home_doc_path(self):
        dir_name = "VoiceAssistant"
        if sys.platform == "win32":
            import ctypes.wintypes
            path_id = 5  # 5:文档 0:桌面
            buf = ctypes.create_unicode_buffer(ctypes.wintypes.MAX_PATH)
            ctypes.windll.shell32.SHGetFolderPathW(None, path_id, None, 0, buf)
            path = buf.value
        elif sys.platform == "darwin":
            path = os.path.join(os.environ[self.get_home_env()], dir_name)
        elif sys.platform == "linux":
            path = os.path.join(os.environ[self.get_home_env()], dir_name)
        else:
            path = os.path.join(os.environ[self.get_home_env()], dir_name)
        dir_path = os.path.join(path, dir_name)
        if not os.path.exists(dir_path):
            os.makedirs(dir_path)
        return dir_path

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

    # 初始化ist引擎
    def _init_ist(self):
        self._ist = lib_aiui.AiuiManager()
        self._ist.trigger.connect(self._proc_aiui)
        self._logger("ISR引擎初始化成功！")

    # 初始化opus压缩工具
    def _init_opus(self):
        self.opus = lib_opus.Opus()
        self.enc = self.opus.create_state(constants.SAMPLE_RATE, constants.CHANNEL_NUM, constants.OPUS_APPLICATION_VOIP)
        self.opus.encoder_ctl(self.enc, lib_opus.set_signal, lib_opus.OPUS_SIGNAL_VOICE)
        self.frame_size = int(constants.SAMPLE_RATE * 0.02)
        self.pcm_length = int(constants.SAMPLE_RATE / 8 * 16 * 1 * 0.02)
        self._logger("OPUS始化成功！")

    def _tts_player_callback(self, msg):
        print(msg)
        if msg == "--end--":
            # self._tts_play_end = True
            # self._send_data(my_util.gen_q_data(constants.TTS_STEAM_END, None))
            self.tts_player.quit()
        elif msg == "--stop--":
            self._tts_play_end = False

    def tts_play(self, text, vcn=0, speed_mode=0):
        _duration = lib_player.calculate_duration(text, speed_mode)
        # self._send_data(my_util.gen_q_data(constants.TTS_STEAM_DURATION, _duration))
        _info = {
            "text": text,
            "vcn": vcn,
            "speed": speed_mode
        }
        # 上一个播放器处于暂停状态
        if self.tts_player and self.tts_player.is_paused():
            self.tts_player.resume()
            return
        self.tts_player = lib_player.TTSSteamPlayer(self._tts_player_callback)
        self.tts_player.play(_info)

    def tts_pause(self):
        if self.tts_player:
            self.tts_player.pause()

    def tts_stop(self):
        if self.tts_player:
            self.tts_player.stop()
            self.tts_player.quit()

    # 开启录音
    def _start_record(self):
        # self._logger("开始录音")
        print("开始录音")
        if not self._run_flag:
            self._run_flag = True
            self._start_ist()

    # 停止录音
    def _stop_record(self):
        # self._logger("结束录音")
        print("结束录音")
        self._run_flag = False
        self._start_ist(stop=True)

    def _proc_aiui(self, msg):
        code = msg.get("code")
        data = msg.get("data")
        # IST引擎开始
        if code == constants.AITEST_AIUI_START:
            if data:
                self._logger(f"IST引擎启动成功！")
        # IST引擎异常
        elif code == constants.AITEST_AIUI_ERROR:
            self._logger(f"IST错误：{data}")
        # IST引擎结果
        elif code == constants.AITEST_AIUI_RESULT:
            self._logger(f"IST转写结果：{data}")
            self._stop_record()
            if TTS_FLAG:
                self.tts_play(data)
        elif code == constants.AITEST_AIUI_LOG:
            self._logger(data)
        elif code == constants.AITEST_AIUI_NLP:
            self._logger(data)
            print('开始业务逻辑')
            nlp_date = json.loads(str(data))
            intent = nlp_date['intent']
            if intent != {} and intent['rc'] == 0:
                intent_action = intent['semantic'][0]['intent']
                print("intent_action: ", intent_action)
                # 遍历词槽
                intent_solts = intent['semantic'][0]['slots']
                print("intent_solts: ", intent_solts)

                # # 方式1：使用这种方式，当语音命令为打开浏览器操作时，
                # # webbrowser.open()会阻塞_proc方法，导致AIUI重新发送消息给websocket，会导致_proc再执行一次，从而打开2次浏览器的BUG
                # # tts_text = socketHadler.handler(intent_action, intent_solts)

                # 方式2：为解决方式1的BUG，在调用业务方式时，先断开再重连的方式
                # 断开信号连接
                # PYQT6 中测试没有这个问题，断开再重连的代码注释掉
                self._ist.trigger.disconnect(self._proc_aiui)
                tts_text = socketHadler.handler(intent_action, intent_solts)
                # 重新连接信号
                self._ist.trigger.connect(self._proc_aiui)

                # # 方式3：在socketHadler里面使用以下方式打开，也可以避免打开2次web页面问题
                # # import os
                # # os.system('start http://101.37.148.192:8080/')

                if len(tts_text) != 0:
                    self.tts_stop()
                    self.tts_play(tts_text)
            else:
                self.tts_stop()
                self.tts_play("我没有理解您说的话")
                print("我没有理解你说的话啊")
        else:
            pass

    def _start_ist(self, stop=False):
        if stop:
            print("stop")
            self._ist.stop_speech()
        else:
            print("start")
            if OPUS_FLAG:
                ext_params = "opus-wb"
            else:
                ext_params = "raw"
            self._ist.regist_engine(ext_params)
            self._ist.start_speech()

    # 压缩音频
    def _compress_buf(self, data):
        out = self.opus.encode(self.enc, bytes(data), self.frame_size, self.pcm_length)
        data = bytearray(len(out) + 2)
        data[0:2] = struct.pack(">H", len(out))
        data[2:] = out
        buf = bytes(data)
        return buf

    def show(self):
        self._load_ui()
        self._init_ist()
        if OPUS_FLAG:
            self._init_opus()

    def closeEvent(self, event):
        super().closeEvent(event)
        QApplication.quit()


def create_socket_handler():
    print('create_socket_handler')
    global socketHadler
    if socketHadler is None:
        socketHadler = BusinessHadler()
        app.aboutToQuit.connect(socketHadler.cleanup)  # 关闭应用时进行清理操作


if __name__ == '__main__':
    app = QApplication(sys.argv)
    win = MainUI()
    win.show()
    create_socket_handler()
    sys.exit(app.exec_())