Demo/test.py

# coding:utf-8
from base64 import b64encode
from hashlib import md5, sha1
from hmac import new
import json
import re
import sys
from threading import Thread
import time
from urllib.parse import quote

from pyaudio import paInt16, PyAudio
import websocket
from PyQt5.QtCore import Qt, QSize, QUrl, pyqtSlot, pyqtSignal
from PyQt5.QtGui import QIcon, QFont, QColor
from PyQt5.QtWidgets import QApplication, QWidget, QHBoxLayout, QVBoxLayout, QSizePolicy, QScrollArea
from qfluentwidgets import (SmoothScrollArea, IconWidget, BodyLabel, CaptionLabel, TransparentToolButton, FluentIcon,
                            ImageLabel, SimpleCardWidget, LineEdit,
                            HeaderCardWidget, InfoBar, InfoBarPosition, InfoBarIcon, HyperlinkLabel, PrimaryPushButton, TitleLabel, setFont,
                            ScrollArea,
                            VerticalSeparator, MSFluentWindow)
import ahocorasick
import pandas as pd


def isWin11():
    return sys.platform == 'win32' and sys.getwindowsversion().build >= 22000


class StatisticsWidget(QWidget):
    """ Statistics widget """

    def __init__(self, title: str, value: str, parent=None):
        super().__init__(parent=parent)
        self.titleLabel = CaptionLabel(title, self)
        self.valueLabel = BodyLabel(value, self)
        self.vBoxLayout = QVBoxLayout(self)

        self.vBoxLayout.setContentsMargins(16, 0, 16, 0)
        self.vBoxLayout.addWidget(self.valueLabel, 0, Qt.AlignTop)
        self.vBoxLayout.addWidget(self.titleLabel, 0, Qt.AlignBottom)

        setFont(self.valueLabel, 18, QFont.DemiBold)
        self.titleLabel.setTextColor(QColor(96, 96, 96), QColor(206, 206, 206))


class ConfigCard(HeaderCardWidget):
    """ Config card for API settings """
    
    configChanged = pyqtSignal(str, str)  # Signal for when config changes (app_id, api_key)

    def __init__(self, parent=None):
        super().__init__(parent)
        self.setTitle('API 配置')
        self.setBorderRadius(8)

        # Create layout
        self.mainLayout = QVBoxLayout()
        
        # App ID input
        self.appIdLayout = QHBoxLayout()
        self.appIdLabel = BodyLabel('App ID:', self)
        self.appIdInput = LineEdit(self)
        self.appIdInput.setPlaceholderText('输入 App ID')
        self.appIdLayout.addWidget(self.appIdLabel)
        self.appIdLayout.addWidget(self.appIdInput)
        
        # API Key input
        self.apiKeyLayout = QHBoxLayout()
        self.apiKeyLabel = BodyLabel('API Key:', self)
        self.apiKeyInput = LineEdit(self)
        self.apiKeyInput.setPlaceholderText('输入 API Key')
        self.apiKeyLayout.addWidget(self.apiKeyLabel)
        self.apiKeyLayout.addWidget(self.apiKeyInput)
        
        # Test connection button
        self.buttonLayout = QHBoxLayout()
        self.testButton = PrimaryPushButton('测试连接', self)
        self.saveButton = PrimaryPushButton('保存配置', self)
        self.buttonLayout.addWidget(self.testButton)
        self.buttonLayout.addWidget(self.saveButton)
        
        # Add all layouts to main layout
        self.mainLayout.addLayout(self.appIdLayout)
        self.mainLayout.addLayout(self.apiKeyLayout)
        self.mainLayout.addLayout(self.buttonLayout)
        
        # Set main layout
        self.viewLayout.addLayout(self.mainLayout)
        
        # Connect signals
        self.testButton.clicked.connect(self.test_connection)
        self.saveButton.clicked.connect(self.save_config)
        
        # Load existing config if any
        self.load_config()

    def load_config(self):
        """Load existing configuration if available"""
        # Here you might want to load from a config file
        # For now, we'll use default values
        self.appIdInput.setText("f3a9a1bc")
        self.apiKeyInput.setText("3ba5a497a68a930546fb95ef750abf90")

    def save_config(self):
        """Save the configuration"""
        app_id = self.appIdInput.text()
        api_key = self.apiKeyInput.text()
        
        if not app_id or not api_key:
            InfoBar.error(
                title='错误',
                content="App ID 和 API Key 不能为空",
                orient=Qt.Horizontal,
                isClosable=True,
                position=InfoBarPosition.TOP,
                duration=2000,
                parent=self
            )
            return
            
        # Emit the config changed signal
        self.configChanged.emit(app_id, api_key)
        
        InfoBar.success(
            title='成功',
            content="配置已保存",
            orient=Qt.Horizontal,
            isClosable=True,
            position=InfoBarPosition.TOP,
            duration=2000,
            parent=self
        )

    def test_connection(self):
        """Test the API connection"""
        app_id = self.appIdInput.text()
        api_key = self.apiKeyInput.text()
        
        if not app_id or not api_key:
            InfoBar.error(
                title='错误',
                content="请先填写 App ID 和 API Key",
                orient=Qt.Horizontal,
                isClosable=True,
                position=InfoBarPosition.TOP,
                duration=2000,
                parent=self
            )
            return
            
        try:
            # Prepare connection parameters
            base_url = "ws://rtasr.xfyun.cn/v1/ws"
            ts = str(int(time.time()))
            tt = (app_id + ts).encode('utf-8')
            md5_ = md5()
            md5_.update(tt)
            baseString = md5_.hexdigest()
            baseString = bytes(baseString, encoding='utf-8')
            
            apiKey = api_key.encode('utf-8')
            signa = new(apiKey, baseString, sha1).digest()
            signa = b64encode(signa)
            signa = str(signa, 'utf-8')
            
            # Create WebSocket connection
            ws_url = f"{base_url}?appid={app_id}&ts={ts}&signa={quote(signa)}"
            ws = websocket.create_connection(ws_url)
            
            # Check the response
            result = ws.recv()
            ws.close()
            
            if "success" in result:
                InfoBar.success(
                    title='成功',
                    content="API连接测试成功",
                    orient=Qt.Horizontal,
                    isClosable=True,
                    position=InfoBarPosition.TOP,
                    duration=2000,
                    parent=self
                )
            else:
                InfoBar.error(
                    title='错误',
                    content="API连接测试失败",
                    orient=Qt.Horizontal,
                    isClosable=True,
                    position=InfoBarPosition.TOP,
                    duration=2000,
                    parent=self
                )
                
        except Exception as e:
            InfoBar.error(
                title='错误',
                content=f"连接测试失败: {str(e)}",
                orient=Qt.Horizontal,
                isClosable=True,
                position=InfoBarPosition.TOP,
                duration=2000,
                parent=self
            )


class AppInfoCard(SimpleCardWidget):
    """ App information card """

    def __init__(self, parent=None):
        super().__init__(parent)
        self.iconLabel = ImageLabel(":/qfluentwidgets/images/logo.png", self)
        self.iconLabel.setBorderRadius(8, 8, 8, 8)
        self.iconLabel.scaledToWidth(120)

        self.nameLabel = TitleLabel('实时语音识别', self)
        self.installButton = PrimaryPushButton('开始', self)
        self.stopButton = PrimaryPushButton('停止', self)
        self.companyLabel = HyperlinkLabel(
            QUrl('https://blog.csdn.net/fjh1997/article/details/104449110'), '电脑音频设置教程', self)
        self.installButton.setFixedWidth(160)
        self.stopButton.setFixedWidth(160)

        self.scoreWidget = StatisticsWidget('版本', '1.0', self)
        self.separator = VerticalSeparator(self)

        self.descriptionLabel = BodyLabel(
            '一个实时语音识别工具，能够实时识别语言内容，并提供帮助提示。', self)
        self.descriptionLabel.setWordWrap(True)

        self.shareButton = TransparentToolButton(FluentIcon.ROBOT, self)
        self.shareButton.setFixedSize(32, 32)
        self.shareButton.setIconSize(QSize(14, 14))

        self.hBoxLayout = QHBoxLayout(self)
        self.vBoxLayout = QVBoxLayout()
        self.topLayout = QHBoxLayout()
        self.statisticsLayout = QHBoxLayout()
        self.buttonLayout = QHBoxLayout()

        self.initLayout()
        self.setBorderRadius(8)

    def initLayout(self):
        self.hBoxLayout.setSpacing(30)
        self.hBoxLayout.setContentsMargins(34, 24, 24, 24)
        self.hBoxLayout.addWidget(self.iconLabel)
        self.hBoxLayout.addLayout(self.vBoxLayout)

        self.vBoxLayout.setContentsMargins(0, 0, 0, 0)
        self.vBoxLayout.setSpacing(0)

        # name label and install button
        self.vBoxLayout.addLayout(self.topLayout)
        self.topLayout.setContentsMargins(0, 0, 0, 0)
        self.topLayout.addWidget(self.nameLabel)
        self.topLayout.addWidget(self.installButton, 0, Qt.AlignRight)
        self.topLayout.addWidget(self.stopButton, 0, Qt.AlignRight)

        # company label
        self.vBoxLayout.addSpacing(3)
        self.vBoxLayout.addWidget(self.companyLabel)

        # statistics widgets
        self.vBoxLayout.addSpacing(20)
        self.vBoxLayout.addLayout(self.statisticsLayout)
        self.statisticsLayout.setContentsMargins(0, 0, 0, 0)
        self.statisticsLayout.setSpacing(10)
        self.statisticsLayout.addWidget(self.scoreWidget)
        self.statisticsLayout.addWidget(self.separator)
        self.statisticsLayout.setAlignment(Qt.AlignLeft)

        # description label
        self.vBoxLayout.addSpacing(20)
        self.vBoxLayout.addWidget(self.descriptionLabel)

        # button
        self.vBoxLayout.addSpacing(12)
        self.buttonLayout.setContentsMargins(0, 0, 0, 0)
        self.vBoxLayout.addLayout(self.buttonLayout)

        self.buttonLayout.addWidget(self.shareButton, 0, Qt.AlignRight)


class InnerAudioCard(HeaderCardWidget):
    """ Inner audio card """

    # 定义自定义信号
    textChanged = pyqtSignal(str)

    def __init__(self, parent=None):
        super().__init__(parent)

        self.descriptionLabel = BodyLabel(self)
        self.descriptionLabel.setWordWrap(True)
        self.viewLayout.addWidget(self.descriptionLabel)

        self.setTitle('音频识别结果')
        self.setBorderRadius(8)
        self.init_asr()

    def init_asr(self):
        self.app_id = "f3a9a1bc"
        self.api_key = "3ba5a497a68a930546fb95ef750abf90"
        self.base_url = "ws://rtasr.xfyun.cn/v1/ws"
        self.ts = str(int(time.time()))
        tt = (self.app_id + self.ts).encode('utf-8')
        md5_ = md5()
        md5_.update(tt)
        baseString = md5_.hexdigest()
        baseString = bytes(baseString, encoding='utf-8')

        apiKey = self.api_key.encode('utf-8')
        signa = new(apiKey, baseString, sha1).digest()
        signa = b64encode(signa)
        self.signa = str(signa, 'utf-8')
        self.end_tag = "{\"end\": true}"

        self.ws = None
        self.p = None
        self.stream = None
        self.tsend = None
        self.trecv = None
        self.running = False

    def findInternalRecordingDevice(self, p):
        target = 'CABLE Output'
        for i in range(p.get_device_count()):
            devInfo = p.get_device_info_by_index(i)
            if devInfo['name'].find(target) >= 0 and devInfo['hostApi'] == 0:
                return i
        print('无法找到内录设备!')
        return -1

    def start_asr(self):
        if not self.running:
            self.running = True

            # 创建WebSocket连接
            self.ws = websocket.create_connection(
                self.base_url + "?appid=" + self.app_id + "&ts=" + self.ts + "&signa=" + quote(self.signa))

            # 启动发送和接收线程
            self.tsend = Thread(target=self.send)
            self.tsend.start()
            self.trecv = Thread(target=self.recv)
            self.trecv.start()

    def stop_asr(self):
        if self.running:
            self.running = False

            try:
                # 发送结束标签
                if self.ws and self.ws.connected:
                    self.ws.send(self.end_tag.encode('utf-8'))
                    print("send end tag success")

                # 停止并关闭音频流
                if self.stream:
                    self.stream.stop_stream()
                    self.stream.close()
                    self.stream = None
                
                if self.p:
                    self.p.terminate()
                    self.p = None

                # 等待线程结束
                if self.tsend and self.tsend.is_alive():
                    self.tsend.join(timeout=2)  # 给2秒钟超时
                if self.trecv and self.trecv.is_alive():
                    self.trecv.join(timeout=2)  # 给2秒钟超时

                # 关闭WebSocket连接
                if self.ws:
                    self.ws.close()
                    self.ws = None

                print("ASR stopped successfully")

            except Exception as e:
                print(f"Error stopping ASR: {str(e)}")
                # 确保资源被释放
                self.stream = None
                self.p = None
                self.ws = None

    def send(self):
        chunk_size = 1280
        audio_format = paInt16
        channels = 1
        rate = 16000

        self.p = PyAudio()
        dev_idx = self.findInternalRecordingDevice(self.p)
        if dev_idx < 0:
            return
        self.stream = self.p.open(input_device_index=dev_idx,
                                  format=audio_format,
                                  channels=channels,
                                  rate=rate,
                                  input=True,
                                  frames_per_buffer=chunk_size)

        try:
            while self.running and self.ws and self.ws.connected:
                data = self.stream.read(chunk_size)
                self.ws.send(data, opcode=websocket.ABNF.OPCODE_BINARY)
                time.sleep(0.04)
        except Exception as e:
            print(f"Send thread error: {e}")
        finally:
            if self.stream:
                self.stream.stop_stream()
                self.stream.close()
            if self.p:
                self.p.terminate()

    def recv(self):
        try:
            while self.running and self.ws and self.ws.connected:
                result = self.ws.recv()
                if len(result) == 0:
                    print("receive result end")
                    break
                result_dict = json.loads(result)
                if result_dict["action"] == "started":
                    print("handshake success, result: " + result)
                if result_dict["action"] == "result":
                    pattern = r'"w":"(.*?)"'
                    extracted_words = ''.join(re.findall(pattern, result_dict["data"]))
                    self.updateLabel(extracted_words)
                if result_dict["action"] == "error":
                    print("rtasr error: " + result)
                    self.ws.close()
                    self.running = False
                    return
        except Exception as e:
            print(f"Recv thread error: {e}")

    def updateLabel(self, text):
        pre = self.descriptionLabel.text()
        pre_list = pre.split("\n")
        if len(text) > len(pre_list[-1]):
            pre_list[-1] = text
        else:
            pre_list.append(text)
        new_text = "\n".join(pre_list)
        self.descriptionLabel.setText(new_text)

        # 发射信号，传递新的文本内容
        self.textChanged.emit(new_text)


class OuterAudioCard(HeaderCardWidget):
    """ Outer audio card """

    def __init__(self, parent=None):
        super().__init__(parent)

        self.descriptionLabel = BodyLabel(self)
        self.descriptionLabel.setWordWrap(True)
        self.viewLayout.addWidget(self.descriptionLabel)

        self.setTitle('帮助提示')
        self.setBorderRadius(8)

        df = pd.read_csv('tips.csv')
        self.automaton = ahocorasick.Automaton()
        for index, row in df.iterrows():
            keyword = row['关键字']
            self.automaton.add_word(keyword, (index, keyword, row['解决方案']))
        self.automaton.make_automaton()

    def updateLabel(self, text):
        text = text.split("\n")[-1]
        results = set()
        for end_index, (index, keyword, solution) in self.automaton.iter(text):
            results.add((keyword, solution))
        print(results)
        if len(results) != 0:
            content = []
            for keyword, solution in results:
                content.append(f"关键字: {keyword}, 解决方案: {solution}")

            pre = self.descriptionLabel.text()
            content_str = pre + "\n" + "\n".join(content)
            self.descriptionLabel.setText(content_str)


class SystemRequirementCard(HeaderCardWidget):
    """ System requirements card """

    def __init__(self, parent=None):
        super().__init__(parent)
        self.setTitle('系统要求')
        self.setBorderRadius(8)

        self.infoLabel = BodyLabel('此产品适用于你的设备。具有复选标记的项目符合开发人员的系统要求。', self)
        self.successIcon = IconWidget(InfoBarIcon.SUCCESS, self)
        self.detailButton = HyperlinkLabel('详细信息', self)

        self.vBoxLayout = QVBoxLayout()
        self.hBoxLayout = QHBoxLayout()

        self.successIcon.setFixedSize(16, 16)
        self.hBoxLayout.setSpacing(10)
        self.vBoxLayout.setSpacing(16)
        self.hBoxLayout.setContentsMargins(0, 0, 0, 0)
        self.vBoxLayout.setContentsMargins(0, 0, 0, 0)

        self.hBoxLayout.addWidget(self.successIcon)
        self.hBoxLayout.addWidget(self.infoLabel)
        self.vBoxLayout.addLayout(self.hBoxLayout)
        self.vBoxLayout.addWidget(self.detailButton)

        self.viewLayout.addLayout(self.vBoxLayout)


class ConfigInterface(ScrollArea):
    """ Configuration interface """

    def __init__(self, parent=None):
        super().__init__(parent=parent)
        self.view = QWidget(self)
        self.vBoxLayout = QVBoxLayout(self.view)
        
        # Create config card
        self.configCard = ConfigCard(self)
        
        # Setup layout
        self.vBoxLayout.setSpacing(30)
        self.vBoxLayout.setContentsMargins(36, 20, 36, 36)
        self.vBoxLayout.addWidget(self.configCard)
        self.vBoxLayout.addStretch(1)
        
        # Setup scroll area
        self.setWidget(self.view)
        self.setWidgetResizable(True)
        self.setObjectName("configInterface")
        
        self.enableTransparentBackground()


class AppInterface(ScrollArea):
    def __init__(self, parent=None):
        super().__init__(parent)
        
        self.view = QWidget(self)
        self.vBoxLayout = QVBoxLayout(self.view)
        
        # Create all cards (removed config card)
        self.appCard = AppInfoCard(self)
        self.systemCard = SystemRequirementCard(self)
        
        # Create scroll areas for audio cards
        self.innerAudioScrollArea = ScrollArea(self)
        self.outerAudioScrollArea = ScrollArea(self)
        
        self.innerAudioCard = InnerAudioCard(self)
        self.outerAudioCard = OuterAudioCard(self)
        
        # Setup scroll areas
        self.innerAudioScrollArea.setWidget(self.innerAudioCard)
        self.outerAudioScrollArea.setWidget(self.outerAudioCard)
        
        self.innerAudioScrollArea.setWidgetResizable(True)
        self.outerAudioScrollArea.setWidgetResizable(True)
        
        self.innerAudioScrollArea.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
        self.outerAudioScrollArea.setSizePolicy(QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed)
        
        self.innerAudioScrollArea.setFixedHeight(200)
        self.outerAudioScrollArea.setFixedHeight(200)
        
        # Create horizontal layout for audio cards
        self.hBoxLayout = QHBoxLayout()
        self.hBoxLayout.addWidget(self.innerAudioScrollArea, 1)
        self.hBoxLayout.addWidget(self.outerAudioScrollArea, 1)
        
        # Setup main layout
        self.vBoxLayout.setSpacing(10)
        self.vBoxLayout.setContentsMargins(0, 0, 10, 30)
        self.vBoxLayout.addWidget(self.appCard, 0, Qt.AlignTop)
        self.vBoxLayout.addLayout(self.hBoxLayout)
        self.vBoxLayout.addWidget(self.systemCard, 0, Qt.AlignTop)
        
        self.setWidget(self.view)
        self.setWidgetResizable(True)
        self.setObjectName("appInterface")
        
        # Connect signals
        self.appCard.installButton.clicked.connect(self.start_asr)
        self.appCard.stopButton.clicked.connect(self.stop_asr)
        self.innerAudioCard.textChanged.connect(self.onTextChanged)
        
        self.enableTransparentBackground()

    def start_asr(self):
        self.innerAudioCard.start_asr()

    def stop_asr(self):
        self.innerAudioCard.stop_asr()

    def resizeEvent(self, e):
        super().resizeEvent(e)

    @pyqtSlot(str)
    def onTextChanged(self, text):
        self.outerAudioCard.updateLabel(text)
        print(f"QLabel 内容已更新为: {text}")

    def closeEvent(self, event):
        # 确保所有子组件的线程被正确关闭
        self.innerAudioCard.stop_asr()
        event.accept()


class Demo3(MSFluentWindow):
    def __init__(self, parent=None):
        super().__init__(parent)

        # 创建界面
        self.appInterface = AppInterface(self)
        self.configInterface = ConfigInterface(self)

        # 添加子界面到侧边栏，只使用基础图标
        self.addSubInterface(self.appInterface, FluentIcon.HOME, "主页", FluentIcon.HOME, isTransparent=True)
        self.addSubInterface(self.configInterface, FluentIcon.SETTING, "配置", FluentIcon.SETTING, isTransparent=True)

        # 连接配置信号
        self.configInterface.configCard.configChanged.connect(self.onConfigChanged)

        self.resize(880, 760)
        self.setWindowTitle('语音识别工具')
        self.setWindowIcon(QIcon(':/qfluentwidgets/images/logo.png'))

        self.titleBar.raise_()

    def onConfigChanged(self, app_id, api_key):
        """处理配置变更"""
        # 将配置更新传递给主页面的InnerAudioCard
        self.appInterface.innerAudioCard.app_id = app_id
        self.appInterface.innerAudioCard.api_key = api_key
        self.appInterface.innerAudioCard.init_asr()


if __name__ == '__main__':
    # enable dpi scale
    QApplication.setHighDpiScaleFactorRoundingPolicy(
        Qt.HighDpiScaleFactorRoundingPolicy.PassThrough)
    QApplication.setAttribute(Qt.AA_EnableHighDpiScaling)
    QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps)

    app = QApplication(sys.argv)
    w3 = Demo3()
    w3.show()
    app.exec_()