src/application.py

import asyncio
import json
import logging
import threading
import time
import sys
import traceback
from pathlib import Path

from src.utils.logging_config import get_logger
# 在导入 opuslib 之前处理 opus 动态库
from src.utils.system_info import setup_opus
from src.constants.constants import (
    DeviceState, EventType, AudioConfig, 
    AbortReason, ListeningMode
)
from src.display import gui_display, cli_display
from src.utils.config_manager import ConfigManager

setup_opus()

# 配置日志
logger = get_logger(__name__)

# 现在导入 opuslib
try:
    import opuslib  # noqa: F401
    from src.utils.tts_utility import TtsUtility
except Exception as e:
    logger.critical("导入 opuslib 失败: %s", e, exc_info=True)
    logger.critical("请确保 opus 动态库已正确安装或位于正确的位置")
    sys.exit(1)

from src.protocols.mqtt_protocol import MqttProtocol
from src.protocols.websocket_protocol import WebsocketProtocol


class Application:
    _instance = None

    @classmethod
    def get_instance(cls):
        """获取单例实例"""
        if cls._instance is None:
            logger.debug("创建Application单例实例")
            cls._instance = Application()
        return cls._instance

    def __init__(self):
        """初始化应用程序"""
        # 确保单例模式
        if Application._instance is not None:
            logger.error("尝试创建Application的多个实例")
            raise Exception("Application是单例类，请使用get_instance()获取实例")
        Application._instance = self

        logger.debug("初始化Application实例")
        # 获取配置管理器实例
        self.config = ConfigManager.get_instance()

        # 状态变量
        self.device_state = DeviceState.IDLE
        self.voice_detected = False
        self.keep_listening = False
        self.aborted = False
        self.current_text = ""
        self.current_emotion = "neutral"

        # 音频处理相关
        self.audio_codec = None  # 将在 _initialize_audio 中初始化
        self._tts_lock = threading.Lock()
        self.is_tts_playing = False  # 因为Display的播放状态只是GUI使用，不方便Music_player使用，所以加了这个标志位表示是TTS在说话

        # 事件循环和线程
        self.loop = asyncio.new_event_loop()
        self.loop_thread = None
        self.running = False
        self.input_event_thread = None
        self.output_event_thread = None

        # 任务队列和锁
        self.main_tasks = []
        self.mutex = threading.Lock()

        # 协议实例
        self.protocol = None

        # 回调函数
        self.on_state_changed_callbacks = []

        # 初始化事件对象
        self.events = {
            EventType.SCHEDULE_EVENT: threading.Event(),
            EventType.AUDIO_INPUT_READY_EVENT: threading.Event(),
            EventType.AUDIO_OUTPUT_READY_EVENT: threading.Event()
        }

        # 创建显示界面
        self.display = None

        # 添加唤醒词检测器
        self.wake_word_detector = None
        logger.debug("Application实例初始化完成")

    def run(self, **kwargs):
        """启动应用程序"""
        logger.info("启动应用程序，参数: %s", kwargs)
        mode = kwargs.get('mode', 'gui')
        protocol = kwargs.get('protocol', 'websocket')

        # 启动主循环线程
        logger.debug("启动主循环线程")
        main_loop_thread = threading.Thread(target=self._main_loop)
        main_loop_thread.daemon = True
        main_loop_thread.start()

        # 初始化通信协议
        logger.debug("设置协议类型: %s", protocol)
        self.set_protocol_type(protocol)

        # 创建并启动事件循环线程
        logger.debug("启动事件循环线程")
        self.loop_thread = threading.Thread(target=self._run_event_loop)
        self.loop_thread.daemon = True
        self.loop_thread.start()

        # 等待事件循环准备就绪
        time.sleep(0.1)

        # 初始化应用程序（移除自动连接）
        logger.debug("初始化应用程序组件")
        asyncio.run_coroutine_threadsafe(
            self._initialize_without_connect(), 
            self.loop
        )

        # 初始化物联网设备
        self._initialize_iot_devices()

        logger.debug("设置显示类型: %s", mode)
        self.set_display_type(mode)
        # 启动GUI
        logger.debug("启动显示界面")
        self.display.start()

    def _run_event_loop(self):
        """运行事件循环的线程函数"""
        logger.debug("设置并启动事件循环")
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()

    def set_is_tts_playing(self, value: bool):
        with self._tts_lock:
            self.is_tts_playing = value

    def get_is_tts_playing(self) -> bool:
        with self._tts_lock:
            return self.is_tts_playing

    async def _initialize_without_connect(self):
        """初始化应用程序组件（不建立连接）"""
        logger.info("正在初始化应用程序组件...")

        # 设置设备状态为待命
        logger.debug("设置初始设备状态为IDLE")
        self.schedule(lambda: self.set_device_state(DeviceState.IDLE))

        # 初始化音频编解码器
        logger.debug("初始化音频编解码器")
        self._initialize_audio()

        # 初始化并启动唤醒词检测
        self._initialize_wake_word_detector()
        
        # 设置联网协议回调（MQTT AND WEBSOCKET）
        logger.debug("设置协议回调函数")
        self.protocol.on_network_error = self._on_network_error
        self.protocol.on_incoming_audio = self._on_incoming_audio
        self.protocol.on_incoming_json = self._on_incoming_json
        self.protocol.on_audio_channel_opened = self._on_audio_channel_opened
        self.protocol.on_audio_channel_closed = self._on_audio_channel_closed

        logger.info("应用程序组件初始化完成")

    def _initialize_audio(self):
        """初始化音频设备和编解码器"""
        try:
            logger.debug("开始初始化音频编解码器")
            from src.audio_codecs.audio_codec import AudioCodec
            self.audio_codec = AudioCodec()
            logger.info("音频编解码器初始化成功")

            # 记录音量控制状态
            has_volume_control = (
                hasattr(self.display, 'volume_controller') and
                self.display.volume_controller
            )
            if has_volume_control:
                logger.info("系统音量控制已启用")
            else:
                logger.info("系统音量控制未启用，将使用模拟音量控制")

        except Exception as e:
            logger.error("初始化音频设备失败: %s", e, exc_info=True)
            self.alert("错误", f"初始化音频设备失败: {e}")

    def set_protocol_type(self, protocol_type: str):
        """设置协议类型"""
        logger.debug("设置协议类型: %s", protocol_type)
        if protocol_type == 'mqtt':
            self.protocol = MqttProtocol(self.loop)
            logger.debug("已创建MQTT协议实例")
        else:  # websocket
            self.protocol = WebsocketProtocol()
            logger.debug("已创建WebSocket协议实例")

    def set_display_type(self, mode: str):
        """初始化显示界面"""
        logger.debug("设置显示界面类型: %s", mode)
        # 通过适配器的概念管理不同的显示模式
        if mode == 'gui':
            self.display = gui_display.GuiDisplay()
            logger.debug("已创建GUI显示界面")
            self.display.set_callbacks(
                press_callback=self.start_listening,
                release_callback=self.stop_listening,
                status_callback=self._get_status_text,
                text_callback=self._get_current_text,
                emotion_callback=self._get_current_emotion,
                mode_callback=self._on_mode_changed,
                auto_callback=self.toggle_chat_state,
                abort_callback=lambda: self.abort_speaking(
                    AbortReason.WAKE_WORD_DETECTED
                ),
                send_text_callback=self._send_text_tts
            )
        else:
            self.display = cli_display.CliDisplay()
            logger.debug("已创建CLI显示界面")
            self.display.set_callbacks(
                auto_callback=self.toggle_chat_state,
                abort_callback=lambda: self.abort_speaking(
                    AbortReason.WAKE_WORD_DETECTED
                ),
                status_callback=self._get_status_text,
                text_callback=self._get_current_text,
                emotion_callback=self._get_current_emotion,
                send_text_callback=self._send_text_tts
            )
        logger.debug("显示界面回调函数设置完成")

    def _main_loop(self):
        """应用程序主循环"""
        logger.info("主循环已启动")
        self.running = True

        while self.running:
            # 等待事件
            for event_type, event in self.events.items():
                if event.is_set():
                    event.clear()
                    logger.debug("处理事件: %s", event_type)

                    if event_type == EventType.AUDIO_INPUT_READY_EVENT:
                        self._handle_input_audio()
                    elif event_type == EventType.AUDIO_OUTPUT_READY_EVENT:
                        self._handle_output_audio()
                    elif event_type == EventType.SCHEDULE_EVENT:
                        self._process_scheduled_tasks()

            # 短暂休眠以避免CPU占用过高
            time.sleep(0.01)

    def _process_scheduled_tasks(self):
        """处理调度任务"""
        with self.mutex:
            tasks = self.main_tasks.copy()
            self.main_tasks.clear()

        logger.debug("处理%d个调度任务", len(tasks))
        for task in tasks:
            try:
                task()
            except Exception as e:
                logger.error("执行调度任务时出错: %s", e, exc_info=True)

    def schedule(self, callback):
        """调度任务到主循环"""
        with self.mutex:
            self.main_tasks.append(callback)
        self.events[EventType.SCHEDULE_EVENT].set()

    def _handle_input_audio(self):
        """处理音频输入"""
        if self.device_state != DeviceState.LISTENING:
            return

        # 读取并发送音频数据
        encoded_data = self.audio_codec.read_audio()
        if (encoded_data and self.protocol and
                self.protocol.is_audio_channel_opened()):
            asyncio.run_coroutine_threadsafe(
                self.protocol.send_audio(encoded_data),
                self.loop
            )

    async def _send_text_tts(self, text):
        """将文本转换为语音并发送"""
        try:
            tts_utility = TtsUtility(AudioConfig)

            # 生成 Opus 音频数据包
            opus_frames = await tts_utility.text_to_opus_audio(text)

            # 尝试打开音频通道
            if (not self.protocol.is_audio_channel_opened() and
                    DeviceState.IDLE == self.device_state):
                # 打开音频通道
                success = await self.protocol.open_audio_channel()
                if not success:
                    logger.error("打开音频通道失败")
                    return

            # 确认 opus 帧生成成功
            if opus_frames:
                logger.info(f"生成了 {len(opus_frames)} 个 Opus 音频帧")

                # 设置状态为说话中
                self.schedule(lambda: self.set_device_state(DeviceState.SPEAKING))

                # 发送音频数据
                for i, frame in enumerate(opus_frames):
                    await self.protocol.send_audio(frame)
                    await asyncio.sleep(0.06)

                # 设置聊天消息
                self.set_chat_message("user", text)
                await self.protocol.send_text(
                    json.dumps({"session_id": "", "type": "listen", "state": "stop"}))
                await self.protocol.send_text(b'')

                return True
            else:
                logger.error("生成音频失败")
                return False

        except Exception as e:
            logger.error(f"发送文本到TTS时出错: {e}")
            logger.error(traceback.format_exc())
            return False

    def _handle_output_audio(self):
        """处理音频输出"""
        if self.device_state != DeviceState.SPEAKING:
            return
        self.set_is_tts_playing(True)   # 开始播放
        self.audio_codec.play_audio()

    def _on_network_error(self, error_message=None):
        """网络错误回调"""
        if error_message:
            logger.error(f"网络错误: {error_message}")
            
        self.keep_listening = False
        self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
        # 恢复唤醒词检测
        if self.wake_word_detector and self.wake_word_detector.paused:
            self.wake_word_detector.resume()

        if self.device_state != DeviceState.CONNECTING:
            logger.info("检测到连接断开")
            self.schedule(lambda: self.set_device_state(DeviceState.IDLE))

            # 关闭现有连接，但不关闭音频流
            if self.protocol:
                asyncio.run_coroutine_threadsafe(
                    self.protocol.close_audio_channel(),
                    self.loop
                )

    def _on_incoming_audio(self, data):
        """接收音频数据回调"""
        if self.device_state == DeviceState.SPEAKING:
            self.audio_codec.write_audio(data)
            self.events[EventType.AUDIO_OUTPUT_READY_EVENT].set()

    def _on_incoming_json(self, json_data):
        """接收JSON数据回调"""
        try:
            if not json_data:
                return

            # 解析JSON数据
            if isinstance(json_data, str):
                data = json.loads(json_data)
            else:
                data = json_data
            # 处理不同类型的消息
            msg_type = data.get("type", "")
            if msg_type == "tts":
                self._handle_tts_message(data)
            elif msg_type == "stt":
                self._handle_stt_message(data)
            elif msg_type == "llm":
                self._handle_llm_message(data)
            elif msg_type == "iot":
                self._handle_iot_message(data)
            else:
                logger.warning(f"收到未知类型的消息: {msg_type}")
        except Exception as e:
            logger.error(f"处理JSON消息时出错: {e}")

    def _handle_tts_message(self, data):
        """处理TTS消息"""
        state = data.get("state", "")
        if state == "start":
            self.schedule(lambda: self._handle_tts_start())
        elif state == "stop":
            self.schedule(lambda: self._handle_tts_stop())
        elif state == "sentence_start":
            text = data.get("text", "")
            if text:
                logger.info(f"<< {text}")
                self.schedule(lambda: self.set_chat_message("assistant", text))

                # 检查是否包含验证码信息
                if "请登录到控制面板添加设备，输入验证码" in text:
                    self.schedule(lambda: self._handle_verification_code(text))

    def _handle_tts_start(self):
        """处理TTS开始事件"""
        self.aborted = False
        self.set_is_tts_playing(True)   # 开始播放
        # 清空可能存在的旧音频数据
        self.audio_codec.clear_audio_queue()

        if self.device_state == DeviceState.IDLE or self.device_state == DeviceState.LISTENING:
            self.schedule(lambda: self.set_device_state(DeviceState.SPEAKING))

        # 注释掉恢复VAD检测器的代码
        # if hasattr(self, 'vad_detector') and self.vad_detector:
        #     self.vad_detector.resume()

    def _handle_tts_stop(self):
        """处理TTS停止事件"""
        if self.device_state == DeviceState.SPEAKING:
            # 给音频播放一个缓冲时间，确保所有音频都播放完毕
            def delayed_state_change():
                # 等待音频队列清空
                # 增加等待重试次数，确保音频可以完全播放完毕
                max_wait_attempts = 30  # 增加等待尝试次数
                wait_interval = 0.1  # 每次等待的时间间隔
                attempts = 0

                # 等待直到队列为空或超过最大尝试次数
                while (not self.audio_codec.audio_decode_queue.empty() and 
                       attempts < max_wait_attempts):
                    time.sleep(wait_interval)
                    attempts += 1

                # 确保所有数据都被播放出来
                # 再额外等待一点时间确保最后的数据被处理
                if self.get_is_tts_playing():
                    time.sleep(0.5)

                # 设置TTS播放状态为False
                self.set_is_tts_playing(False)

                # 状态转换
                if self.keep_listening:
                    asyncio.run_coroutine_threadsafe(
                        self.protocol.send_start_listening(ListeningMode.AUTO_STOP),
                        self.loop
                    )
                    self.schedule(lambda: self.set_device_state(DeviceState.LISTENING))
                else:
                    self.schedule(lambda: self.set_device_state(DeviceState.IDLE))

            # 安排延迟执行
            # threading.Thread(target=delayed_state_change, daemon=True).start()
            self.schedule(delayed_state_change)

    def _handle_stt_message(self, data):
        """处理STT消息"""
        text = data.get("text", "")
        if text:
            logger.info(f">> {text}")
            self.schedule(lambda: self.set_chat_message("user", text))

    def _handle_llm_message(self, data):
        """处理LLM消息"""
        emotion = data.get("emotion", "")
        if emotion:
            self.schedule(lambda: self.set_emotion(emotion))

    async def _on_audio_channel_opened(self):
        """音频通道打开回调"""
        logger.info("音频通道已打开")
        self.schedule(lambda: self._start_audio_streams())

        # 发送物联网设备描述符
        from src.iot.thing_manager import ThingManager
        thing_manager = ThingManager.get_instance()
        asyncio.run_coroutine_threadsafe(
            self.protocol.send_iot_descriptors(thing_manager.get_descriptors_json()),
            self.loop
        )
        self._update_iot_states(False)


    def _start_audio_streams(self):
        """启动音频流"""
        try:
            # 不再关闭和重新打开流，只确保它们处于活跃状态
            if self.audio_codec.input_stream and not self.audio_codec.input_stream.is_active():
                try:
                    self.audio_codec.input_stream.start_stream()
                except Exception as e:
                    logger.warning(f"启动输入流时出错: {e}")
                    # 只有在出错时才重新初始化
                    self.audio_codec._reinitialize_input_stream()

            if self.audio_codec.output_stream and not self.audio_codec.output_stream.is_active():
                try:
                    self.audio_codec.output_stream.start_stream()
                except Exception as e:
                    logger.warning(f"启动输出流时出错: {e}")
                    # 只有在出错时才重新初始化
                    self.audio_codec._reinitialize_output_stream()

            # 设置事件触发器
            if self.input_event_thread is None or not self.input_event_thread.is_alive():
                self.input_event_thread = threading.Thread(
                    target=self._audio_input_event_trigger, daemon=True)
                self.input_event_thread.start()
                logger.info("已启动输入事件触发线程")

            # 检查输出事件线程
            if self.output_event_thread is None or not self.output_event_thread.is_alive():
                self.output_event_thread = threading.Thread(
                    target=self._audio_output_event_trigger, daemon=True)
                self.output_event_thread.start()
                logger.info("已启动输出事件触发线程")

            logger.info("音频流已启动")
        except Exception as e:
            logger.error(f"启动音频流失败: {e}")

    def _audio_input_event_trigger(self):
        """音频输入事件触发器"""
        while self.running:
            try:
                # 只有在主动监听状态下才触发输入事件
                if self.device_state == DeviceState.LISTENING and self.audio_codec.input_stream:
                    self.events[EventType.AUDIO_INPUT_READY_EVENT].set()
            except OSError as e:
                logger.error(f"音频输入流错误: {e}")
                # 不要退出循环，继续尝试
                time.sleep(0.5)
            except Exception as e:
                logger.error(f"音频输入事件触发器错误: {e}")
                time.sleep(0.5)
            
            # 确保触发频率足够高，即使帧长度较大
            # 使用20ms作为最大触发间隔，确保即使帧长度为60ms也能有足够的采样率
            sleep_time = min(20, AudioConfig.FRAME_DURATION) / 1000
            time.sleep(sleep_time)  # 按帧时长触发，但确保最小触发频率

    def _audio_output_event_trigger(self):
        """音频输出事件触发器"""
        while self.running:
            try:
                # 确保输出流是活跃的
                if (self.device_state == DeviceState.SPEAKING and
                    self.audio_codec and
                    self.audio_codec.output_stream):

                    # 如果输出流不活跃，尝试重新激活
                    if not self.audio_codec.output_stream.is_active():
                        try:
                            self.audio_codec.output_stream.start_stream()
                        except Exception as e:
                            logger.warning(f"启动输出流失败，尝试重新初始化: {e}")
                            self.audio_codec._reinitialize_output_stream()

                    # 当队列中有数据时才触发事件
                    if not self.audio_codec.audio_decode_queue.empty():
                        self.events[EventType.AUDIO_OUTPUT_READY_EVENT].set()
            except Exception as e:
                logger.error(f"音频输出事件触发器错误: {e}")

            time.sleep(0.02)  # 稍微延长检查间隔

    async def _on_audio_channel_closed(self):
        """音频通道关闭回调"""
        logger.info("音频通道已关闭")
        # 设置为空闲状态但不关闭音频流
        self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
        self.keep_listening = False

        # 确保唤醒词检测正常工作
        if self.wake_word_detector:
            if not self.wake_word_detector.is_running():
                logger.info("在空闲状态下启动唤醒词检测")
                # 直接使用AudioCodec实例，而不是尝试获取共享流
                if hasattr(self, 'audio_codec') and self.audio_codec:
                    self.wake_word_detector.start(self.audio_codec)
                else:
                    self.wake_word_detector.start()
            elif self.wake_word_detector.paused:
                logger.info("在空闲状态下恢复唤醒词检测")
                self.wake_word_detector.resume()

    def set_device_state(self, state):
        """设置设备状态"""
        if self.device_state == state:
            return

        self.device_state = state

        # 根据状态执行相应操作
        if state == DeviceState.IDLE:
            self.display.update_status("待命")
            # self.display.update_emotion("😶")
            self.set_emotion("neutral")
            # 恢复唤醒词检测（添加安全检查）
            if self.wake_word_detector and hasattr(self.wake_word_detector, 'paused') and self.wake_word_detector.paused:
                self.wake_word_detector.resume()
                logger.info("唤醒词检测已恢复")
            # 恢复音频输入流
            if self.audio_codec and self.audio_codec.is_input_paused():
                self.audio_codec.resume_input()
        elif state == DeviceState.CONNECTING:
            self.display.update_status("连接中...")
        elif state == DeviceState.LISTENING:
            self.display.update_status("聆听中...")
            self.set_emotion("neutral")
            self._update_iot_states(True)
            # 暂停唤醒词检测（添加安全检查）
            if self.wake_word_detector and hasattr(self.wake_word_detector, 'is_running') and self.wake_word_detector.is_running():
                self.wake_word_detector.pause()
                logger.info("唤醒词检测已暂停")
            # 确保音频输入流活跃
            if self.audio_codec:
                if self.audio_codec.is_input_paused():
                    self.audio_codec.resume_input()
        elif state == DeviceState.SPEAKING:
            self.display.update_status("说话中...")
            if self.wake_word_detector and hasattr(self.wake_word_detector, 'paused') and self.wake_word_detector.paused:
                self.wake_word_detector.resume()
            # 暂停唤醒词检测（添加安全检查）
            # if self.wake_word_detector and hasattr(self.wake_word_detector, 'is_running') and self.wake_word_detector.is_running():
                # self.wake_word_detector.pause()
                # logger.info("唤醒词检测已暂停")
            # 暂停音频输入流以避免自我监听
            # if self.audio_codec and not self.audio_codec.is_input_paused():
            #     self.audio_codec.pause_input()

        # 通知状态变化
        for callback in self.on_state_changed_callbacks:
            try:
                callback(state)
            except Exception as e:
                logger.error(f"执行状态变化回调时出错: {e}")

    def _get_status_text(self):
        """获取当前状态文本"""
        states = {
            DeviceState.IDLE: "待命",
            DeviceState.CONNECTING: "连接中...",
            DeviceState.LISTENING: "聆听中...",
            DeviceState.SPEAKING: "说话中..."
        }
        return states.get(self.device_state, "未知")

    def _get_current_text(self):
        """获取当前显示文本"""
        return self.current_text

    def _get_current_emotion(self):
        """获取当前表情"""
        # 如果表情没有变化，直接返回缓存的路径
        if hasattr(self, '_last_emotion') and self._last_emotion == self.current_emotion:
            return self._last_emotion_path
        
        # 获取基础路径
        if getattr(sys, 'frozen', False):
            # 打包环境
            if hasattr(sys, '_MEIPASS'):
                base_path = Path(sys._MEIPASS)
            else:
                base_path = Path(sys.executable).parent
        else:
            # 开发环境
            base_path = Path(__file__).parent.parent
            
        emotion_dir = base_path / "assets" / "emojis"
            
        emotions = {
            "neutral": str(emotion_dir / "neutral.gif"),
            "happy": str(emotion_dir / "happy.gif"),
            "laughing": str(emotion_dir / "laughing.gif"),
            "funny": str(emotion_dir / "funny.gif"),
            "sad": str(emotion_dir / "sad.gif"),
            "angry": str(emotion_dir / "angry.gif"),
            "crying": str(emotion_dir / "crying.gif"),
            "loving": str(emotion_dir / "loving.gif"),
            "embarrassed": str(emotion_dir / "embarrassed.gif"),
            "surprised": str(emotion_dir / "surprised.gif"),
            "shocked": str(emotion_dir / "shocked.gif"),
            "thinking": str(emotion_dir / "thinking.gif"),
            "winking": str(emotion_dir / "winking.gif"),
            "cool": str(emotion_dir / "cool.gif"),
            "relaxed": str(emotion_dir / "relaxed.gif"),
            "delicious": str(emotion_dir / "delicious.gif"),
            "kissy": str(emotion_dir / "kissy.gif"),
            "confident": str(emotion_dir / "confident.gif"),
            "sleepy": str(emotion_dir / "sleepy.gif"),
            "silly": str(emotion_dir / "silly.gif"),
            "confused": str(emotion_dir / "confused.gif")
        }
        
        # 保存当前表情和对应的路径
        self._last_emotion = self.current_emotion
        self._last_emotion_path = emotions.get(self.current_emotion, str(emotion_dir / "neutral.gif"))
        
        logger.debug(f"表情路径: {self._last_emotion_path}")
        return self._last_emotion_path

    def set_chat_message(self, role, message):
        """设置聊天消息"""
        self.current_text = message
        # 更新显示
        if self.display:
            self.display.update_text(message)

    def set_emotion(self, emotion):
        """设置表情"""
        self.current_emotion = emotion
        # 更新显示
        if self.display:
            self.display.update_emotion(self._get_current_emotion())

    def start_listening(self):
        """开始监听"""
        self.schedule(self._start_listening_impl)

    def _start_listening_impl(self):
        """开始监听的实现"""
        if not self.protocol:
            logger.error("协议未初始化")
            return

        self.keep_listening = False

        # 检查唤醒词检测器是否存在
        if self.wake_word_detector:
            self.wake_word_detector.pause()

        if self.device_state == DeviceState.IDLE:
            self.schedule(lambda: self.set_device_state(DeviceState.CONNECTING))  # 设置设备状态为连接中
            # 尝试打开音频通道
            if not self.protocol.is_audio_channel_opened():
                try:
                    # 等待异步操作完成
                    future = asyncio.run_coroutine_threadsafe(
                        self.protocol.open_audio_channel(),
                        self.loop
                    )
                    # 等待操作完成并获取结果
                    success = future.result(timeout=10.0)  # 添加超时时间

                    if not success:
                        self.alert("错误", "打开音频通道失败")  # 弹出错误提示
                        self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
                        return

                except Exception as e:
                    logger.error(f"打开音频通道时发生错误: {e}")
                    self.alert("错误", f"打开音频通道失败: {str(e)}")
                    self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
                    return
                
            # --- 强制重新初始化输入流 --- 
            try:
                if self.audio_codec:
                     self.audio_codec._reinitialize_input_stream() # 调用重新初始化
                else:
                     logger.warning("Cannot force reinitialization, audio_codec is None.")
            except Exception as force_reinit_e:
                logger.error(f"Forced reinitialization failed: {force_reinit_e}", exc_info=True)
                self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
                if self.wake_word_detector and self.wake_word_detector.paused:
                     self.wake_word_detector.resume()
                return
            # --- 强制重新初始化结束 --- 

            asyncio.run_coroutine_threadsafe(
                self.protocol.send_start_listening(ListeningMode.MANUAL),
                self.loop
            )
            self.schedule(lambda: self.set_device_state(DeviceState.LISTENING))
        elif self.device_state == DeviceState.SPEAKING:
            if not self.aborted:
                self.abort_speaking(AbortReason.WAKE_WORD_DETECTED)

    async def _open_audio_channel_and_start_manual_listening(self):
        """打开音频通道并开始手动监听"""
        if not await self.protocol.open_audio_channel():
            self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
            self.alert("错误", "打开音频通道失败")
            return

        await self.protocol.send_start_listening(ListeningMode.MANUAL)
        self.schedule(lambda: self.set_device_state(DeviceState.LISTENING))

    def toggle_chat_state(self):
        """切换聊天状态"""
        # 检查唤醒词检测器是否存在
        if self.wake_word_detector:
            self.wake_word_detector.pause()
        self.schedule(self._toggle_chat_state_impl)

    def _toggle_chat_state_impl(self):
        """切换聊天状态的具体实现"""
        # 检查协议是否已初始化
        if not self.protocol:
            logger.error("协议未初始化")
            return

        # 如果设备当前处于空闲状态，尝试连接并开始监听
        if self.device_state == DeviceState.IDLE:
            self.schedule(lambda: self.set_device_state(DeviceState.CONNECTING))  # 设置设备状态为连接中
            # 使用线程来处理连接操作，避免阻塞
            def connect_and_listen():
                # 尝试打开音频通道
                if not self.protocol.is_audio_channel_opened():
                    try:
                        # 等待异步操作完成
                        future = asyncio.run_coroutine_threadsafe(
                            self.protocol.open_audio_channel(),
                            self.loop
                        )
                        # 等待操作完成并获取结果，使用较短的超时时间
                        try:
                            success = future.result(timeout=5.0)
                        except asyncio.TimeoutError:
                            logger.error("打开音频通道超时")
                            self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
                            self.alert("错误", "打开音频通道超时")
                            return
                        except Exception as e:
                            logger.error(f"打开音频通道时发生未知错误: {e}")
                            self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
                            self.alert("错误", f"打开音频通道失败: {str(e)}")
                            return

                        if not success:
                            self.alert("错误", "打开音频通道失败")  # 弹出错误提示
                            self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
                            return

                    except Exception as e:
                        logger.error(f"打开音频通道时发生错误: {e}")
                        self.alert("错误", f"打开音频通道失败: {str(e)}")
                        self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
                        return

                self.keep_listening = True  # 开始监听
                # 启动自动停止的监听模式
                try:
                    asyncio.run_coroutine_threadsafe(
                        self.protocol.send_start_listening(ListeningMode.AUTO_STOP),
                        self.loop
                    )
                    self.schedule(lambda: self.set_device_state(DeviceState.LISTENING))
                except Exception as e:
                    logger.error(f"启动监听时发生错误: {e}")
                    self.set_device_state(DeviceState.IDLE)
                    self.alert("错误", f"启动监听失败: {str(e)}")

            # 启动连接线程
            threading.Thread(target=connect_and_listen, daemon=True).start()

        # 如果设备正在说话，停止当前说话
        elif self.device_state == DeviceState.SPEAKING:
            self.abort_speaking(AbortReason.NONE)  # 中止说话

        # 如果设备正在监听，关闭音频通道
        elif self.device_state == DeviceState.LISTENING:
            # 使用线程处理关闭操作，避免阻塞
            def close_audio_channel():
                try:
                    future = asyncio.run_coroutine_threadsafe(
                        self.protocol.close_audio_channel(),
                        self.loop
                    )
                    future.result(timeout=3.0)  # 使用较短的超时
                except Exception as e:
                    logger.error(f"关闭音频通道时发生错误: {e}")
            
            threading.Thread(target=close_audio_channel, daemon=True).start()
            # 立即设置为空闲状态，不等待关闭完成
            self.schedule(lambda: self.set_device_state(DeviceState.IDLE))

    def stop_listening(self):
        """停止监听"""
        self.schedule(self._stop_listening_impl)

    def _stop_listening_impl(self):
        """停止监听的实现"""
        if self.device_state == DeviceState.LISTENING:
            asyncio.run_coroutine_threadsafe(
                self.protocol.send_stop_listening(),
                self.loop
            )
            self.set_device_state(DeviceState.IDLE)

    def abort_speaking(self, reason):
        """中止语音输出"""
        # 如果已经中止，不要重复处理
        if self.aborted:
            logger.debug(f"已经中止，忽略重复的中止请求: {reason}")
            return

        logger.info(f"中止语音输出，原因: {reason}")
        self.aborted = True

        # 设置TTS播放状态为False
        self.set_is_tts_playing(False)
        
        # 立即清空音频队列
        if self.audio_codec:
            self.audio_codec.clear_audio_queue()

        # 注释掉确保VAD检测器暂停的代码
        # if hasattr(self, 'vad_detector') and self.vad_detector:
        #     self.vad_detector.pause()

        # 使用线程来处理状态变更和异步操作，避免阻塞主线程
        def process_abort():
            # 先发送中止指令
            try:
                future = asyncio.run_coroutine_threadsafe(
                    self.protocol.send_abort_speaking(reason),
                    self.loop
                )
                # 使用较短的超时确保不会长时间阻塞
                future.result(timeout=1.0)
            except Exception as e:
                logger.error(f"发送中止指令时出错: {e}")
            
            # 然后设置状态
            # self.set_device_state(DeviceState.IDLE)
            self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
            # 如果是唤醒词触发的中止，并且启用了自动聆听，则自动进入录音模式
            if (reason == AbortReason.WAKE_WORD_DETECTED and 
                    self.keep_listening and 
                    self.protocol.is_audio_channel_opened()):
                # 短暂延迟确保abort命令被处理
                time.sleep(0.1)  # 缩短延迟时间
                self.schedule(lambda: self.toggle_chat_state())
        
        # 启动处理线程
        threading.Thread(target=process_abort, daemon=True).start()

    def alert(self, title, message):
        """显示警告信息"""
        logger.warning(f"警告: {title}, {message}")
        # 在GUI上显示警告
        if self.display:
            self.display.update_text(f"{title}: {message}")

    def on_state_changed(self, callback):
        """注册状态变化回调"""
        self.on_state_changed_callbacks.append(callback)

    def shutdown(self):
        """关闭应用程序"""
        logger.info("正在关闭应用程序...")
        self.running = False

        # 关闭音频编解码器
        if self.audio_codec:
            self.audio_codec.close()

        # 关闭协议
        if self.protocol:
            asyncio.run_coroutine_threadsafe(
                self.protocol.close_audio_channel(),
                self.loop
            )

        # 停止事件循环
        if self.loop and self.loop.is_running():
            self.loop.call_soon_threadsafe(self.loop.stop)

        # 等待事件循环线程结束
        if self.loop_thread and self.loop_thread.is_alive():
            self.loop_thread.join(timeout=1.0)

        # 停止唤醒词检测
        if self.wake_word_detector:
            self.wake_word_detector.stop()

        # 关闭VAD检测器
        # if hasattr(self, 'vad_detector') and self.vad_detector:
        #     self.vad_detector.stop()

        logger.info("应用程序已关闭")

    def _handle_verification_code(self, text):
        """处理验证码信息"""
        try:
            # 提取验证码
            import re
            verification_code = re.search(r'验证码：(\d+)', text)
            if verification_code:
                code = verification_code.group(1)

                # 尝试复制到剪贴板
                try:
                    import pyperclip
                    pyperclip.copy(code)
                    logger.info(f"验证码 {code} 已复制到剪贴板")
                except Exception as e:
                    logger.warning(f"无法复制验证码到剪贴板: {e}")

                # 尝试打开浏览器
                try:
                    import webbrowser
                    if webbrowser.open("https://xiaozhi.me/login"):
                        logger.info("已打开登录页面")
                    else:
                        logger.warning("无法打开浏览器")
                except Exception as e:
                    logger.warning(f"打开浏览器时出错: {e}")

                # 无论如何都显示验证码
                self.alert("验证码", f"您的验证码是: {code}")

        except Exception as e:
            logger.error(f"处理验证码时出错: {e}")

    def _on_mode_changed(self, auto_mode):
        """处理对话模式变更"""
        # 只有在IDLE状态下才允许切换模式
        if self.device_state != DeviceState.IDLE:
            self.alert("提示", "只有在待命状态下才能切换对话模式")
            return False

        self.keep_listening = auto_mode
        logger.info(f"对话模式已切换为: {'自动' if auto_mode else '手动'}")
        return True

    def _initialize_wake_word_detector(self):
        """初始化唤醒词检测器"""
        # 首先检查配置中是否启用了唤醒词功能
        if not self.config.get_config('WAKE_WORD_OPTIONS.USE_WAKE_WORD', False):
            logger.info("唤醒词功能已在配置中禁用，跳过初始化")
            self.wake_word_detector = None
            return

        try:
            from src.audio_processing.wake_word_detect import WakeWordDetector

            # 创建检测器实例
            self.wake_word_detector = WakeWordDetector()

            # 如果唤醒词检测器被禁用（内部故障），则更新配置
            if not getattr(self.wake_word_detector, 'enabled', True):
                logger.warning("唤醒词检测器被禁用（内部故障）")
                self.config.update_config("WAKE_WORD_OPTIONS.USE_WAKE_WORD", False)
                self.wake_word_detector = None
                return

            # 注册唤醒词检测回调和错误处理
            self.wake_word_detector.on_detected(self._on_wake_word_detected)
            
            # 使用lambda捕获self，而不是单独定义函数
            self.wake_word_detector.on_error = lambda error: (
                self._handle_wake_word_error(error)
            )
            
            logger.info("唤醒词检测器初始化成功")

            # 启动唤醒词检测器
            self._start_wake_word_detector()

        except Exception as e:
            logger.error(f"初始化唤醒词检测器失败: {e}")
            import traceback
            logger.error(traceback.format_exc())

            # 禁用唤醒词功能，但不影响程序其他功能
            self.config.update_config("WAKE_WORD_OPTIONS.USE_WAKE_WORD", False)
            logger.info("由于初始化失败，唤醒词功能已禁用，但程序将继续运行")
            self.wake_word_detector = None

    def _handle_wake_word_error(self, error):
        """处理唤醒词检测器错误"""
        logger.error(f"唤醒词检测错误: {error}")
        # 尝试重新启动检测器
        if self.device_state == DeviceState.IDLE:
            self.schedule(lambda: self._restart_wake_word_detector())

    def _start_wake_word_detector(self):
        """启动唤醒词检测器"""
        if not self.wake_word_detector:
            return
        
        # 确保音频编解码器已初始化
        if hasattr(self, 'audio_codec') and self.audio_codec:
            logger.info("使用音频编解码器启动唤醒词检测器")
            self.wake_word_detector.start(self.audio_codec)
        else:
            # 如果没有音频编解码器，使用独立模式
            logger.info("使用独立模式启动唤醒词检测器")
            self.wake_word_detector.start()

    def _on_wake_word_detected(self, wake_word, full_text):
        """唤醒词检测回调"""
        logger.info(f"检测到唤醒词: {wake_word} (完整文本: {full_text})")
        self.schedule(lambda: self._handle_wake_word_detected(wake_word))

    def _handle_wake_word_detected(self, wake_word):
        """处理唤醒词检测事件"""
        if self.device_state == DeviceState.IDLE:
            # 暂停唤醒词检测
            if self.wake_word_detector:
                self.wake_word_detector.pause()

            # 开始连接并监听
            self.schedule(lambda: self.set_device_state(DeviceState.CONNECTING))
            # 尝试连接并打开音频通道
            asyncio.run_coroutine_threadsafe(
                self._connect_and_start_listening(wake_word),
                self.loop
            )
        elif self.device_state == DeviceState.SPEAKING:
            self.abort_speaking(AbortReason.WAKE_WORD_DETECTED)

    async def _connect_and_start_listening(self, wake_word):
        """连接服务器并开始监听"""
        # 首先尝试连接服务器
        if not await self.protocol.connect():
            logger.error("连接服务器失败")
            self.alert("错误", "连接服务器失败")
            self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
            # 恢复唤醒词检测
            if self.wake_word_detector:
                self.wake_word_detector.resume()
            return

        # 然后尝试打开音频通道
        if not await self.protocol.open_audio_channel():
            logger.error("打开音频通道失败")
            self.schedule(lambda: self.set_device_state(DeviceState.IDLE))
            self.alert("错误", "打开音频通道失败")
            # 恢复唤醒词检测
            if self.wake_word_detector:
                self.wake_word_detector.resume()
            return

        await self.protocol.send_wake_word_detected(wake_word)
        # 设置为自动监听模式
        self.keep_listening = True
        await self.protocol.send_start_listening(ListeningMode.AUTO_STOP)
        self.schedule(lambda: self.set_device_state(DeviceState.LISTENING))

    def _restart_wake_word_detector(self):
        """重新启动唤醒词检测器"""
        logger.info("尝试重新启动唤醒词检测器")
        try:
            # 停止现有的检测器
            if self.wake_word_detector:
                self.wake_word_detector.stop()
                time.sleep(0.5)  # 给予一些时间让资源释放

            # 直接使用音频编解码器
            if hasattr(self, 'audio_codec') and self.audio_codec:
                self.wake_word_detector.start(self.audio_codec)
                logger.info("使用音频编解码器重新启动唤醒词检测器")
            else:
                # 如果没有音频编解码器，使用独立模式
                self.wake_word_detector.start()
                logger.info("使用独立模式重新启动唤醒词检测器")

            logger.info("唤醒词检测器重新启动成功")
        except Exception as e:
            logger.error(f"重新启动唤醒词检测器失败: {e}")

    def _initialize_iot_devices(self):
        """初始化物联网设备"""
        from src.iot.thing_manager import ThingManager
        from src.iot.things.lamp import Lamp
        from src.iot.things.speaker import Speaker
        from src.iot.things.music_player import MusicPlayer
        from src.iot.things.CameraVL.Camera import Camera
        from src.iot.things.query_bridge_rag import QueryBridgeRAG
        from src.iot.things.temperature_sensor import TemperatureSensor
        # 导入Home Assistant设备控制类
        from src.iot.things.ha_control import HomeAssistantLight, HomeAssistantSwitch, HomeAssistantNumber, HomeAssistantButton
        # 导入新的倒计时器设备
        from src.iot.things.countdown_timer import CountdownTimer
        
        # 获取物联网设备管理器实例
        thing_manager = ThingManager.get_instance()

        # 添加设备
        thing_manager.add_thing(Lamp())
        thing_manager.add_thing(Speaker())
        thing_manager.add_thing(MusicPlayer())
        # 默认不启用以下示例
        thing_manager.add_thing(Camera())
        # thing_manager.add_thing(QueryBridgeRAG())
        # thing_manager.add_thing(TemperatureSensor())

        # 添加倒计时器设备
        thing_manager.add_thing(CountdownTimer())
        logger.info("已添加倒计时器设备,用于计时执行命令用")

        # 添加Home Assistant设备
        ha_devices = self.config.get_config("HOME_ASSISTANT.DEVICES", [])
        for device in ha_devices:
            entity_id = device.get("entity_id")
            friendly_name = device.get("friendly_name")
            if entity_id:
                # 根据实体ID判断设备类型
                if entity_id.startswith("light."):
                    # 灯设备
                    thing_manager.add_thing(HomeAssistantLight(entity_id, friendly_name))
                    logger.info(f"已添加Home Assistant灯设备: {friendly_name or entity_id}")
                elif entity_id.startswith("switch."):
                    # 开关设备
                    thing_manager.add_thing(HomeAssistantSwitch(entity_id, friendly_name))
                    logger.info(f"已添加Home Assistant开关设备: {friendly_name or entity_id}")
                elif entity_id.startswith("number."):
                    # 数值设备（如音量控制）
                    thing_manager.add_thing(HomeAssistantNumber(entity_id, friendly_name))
                    logger.info(f"已添加Home Assistant数值设备: {friendly_name or entity_id}")
                elif entity_id.startswith("button."):
                    # 按钮设备
                    thing_manager.add_thing(HomeAssistantButton(entity_id, friendly_name))
                    logger.info(f"已添加Home Assistant按钮设备: {friendly_name or entity_id}")
                else:
                    # 默认作为灯设备处理
                    thing_manager.add_thing(HomeAssistantLight(entity_id, friendly_name))
                    logger.info(f"已添加Home Assistant设备(默认作为灯处理): {friendly_name or entity_id}")

        logger.info("物联网设备初始化完成")

    def _handle_iot_message(self, data):
        """处理物联网消息"""
        from src.iot.thing_manager import ThingManager
        thing_manager = ThingManager.get_instance()

        commands = data.get("commands", [])
        for command in commands:
            try:
                result = thing_manager.invoke(command)
                logger.info(f"执行物联网命令结果: {result}")
                # self.schedule(lambda: self._update_iot_states())
            except Exception as e:
                logger.error(f"执行物联网命令失败: {e}")

    def _update_iot_states(self, delta=None):
        """
        更新物联网设备状态

        Args:
            delta: 是否只发送变化的部分
                   - None: 使用原始行为，总是发送所有状态
                   - True: 只发送变化的部分
                   - False: 发送所有状态并重置缓存
        """
        from src.iot.thing_manager import ThingManager
        thing_manager = ThingManager.get_instance()

        # 处理向下兼容
        if delta is None:
            # 保持原有行为：获取所有状态并发送
            states_json = thing_manager.get_states_json_str()  # 调用旧方法

            # 发送状态更新
            asyncio.run_coroutine_threadsafe(
                self.protocol.send_iot_states(states_json),
                self.loop
            )
            logger.info("物联网设备状态已更新")
            return

        # 使用新方法获取状态
        changed, states_json = thing_manager.get_states_json(delta=delta)
        # delta=False总是发送，delta=True只在有变化时发送
        if not delta or changed:
            asyncio.run_coroutine_threadsafe(
                self.protocol.send_iot_states(states_json),
                self.loop
            )
            if delta:
                logger.info("物联网设备状态已更新(增量)")
            else:
                logger.info("物联网设备状态已更新(完整)")
        else:
            logger.debug("物联网设备状态无变化，跳过更新")

    def _update_wake_word_detector_stream(self):
        """更新唤醒词检测器的音频流"""
        if self.wake_word_detector and self.audio_codec and self.wake_word_detector.is_running():
            # 直接引用AudioCodec实例中的输入流
            if self.audio_codec.input_stream and self.audio_codec.input_stream.is_active():
                self.wake_word_detector.stream = self.audio_codec.input_stream
                self.wake_word_detector.external_stream = True
                logger.info("已更新唤醒词检测器的音频流引用")