import asyncio import json import logging import threading import time from xiaozhi.services.display import gui_display from xiaozhi.services.protocols.typing import ( AbortReason, AudioConfig, DeviceState, EventType, ListeningMode, ) from xiaozhi.services.protocols.websocket_protocol import WebsocketProtocol from xiaozhi.utils.config_manager import ConfigManager from xiaozhi.xiaoai import XiaoAi # 配置日志 logger = logging.getLogger("XiaoZhi") class XiaoZhi: """智能音箱应用程序主类""" _instance = None @classmethod def instance(cls): """获取单例实例""" if cls._instance is None: cls._instance = XiaoZhi() return cls._instance def __init__(self): """初始化应用程序""" # 确保单例模式 if XiaoZhi._instance is not None: raise Exception("XiaoZhi是单例类，请使用instance()获取实例") XiaoZhi._instance = self # 获取配置管理器实例 self.config = ConfigManager.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 # 事件循环和线程 self.loop = asyncio.new_event_loop() self.loop_thread = None self.running = False # 任务队列和锁 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 def run(self): self.protocol = WebsocketProtocol() # 创建并启动事件循环线程 self.loop_thread = threading.Thread(target=self._run_event_loop) self.loop_thread.daemon = True self.loop_thread.start() # 等待事件循环准备就绪 time.sleep(0.1) # 初始化应用程序（移除自动连接） asyncio.run_coroutine_threadsafe(XiaoAi.init_xiaoai(), self.loop) asyncio.run_coroutine_threadsafe(self._initialize_without_connect(), self.loop) # 启动主循环线程 main_loop_thread = threading.Thread(target=self._main_loop) main_loop_thread.daemon = True main_loop_thread.start() # 启动 GUI self._initialize_display() self.display.start() def _run_event_loop(self): """运行事件循环的线程函数""" asyncio.set_event_loop(self.loop) self.loop.run_forever() async def _initialize_without_connect(self): """初始化应用程序组件（不建立连接）""" logger.info("正在初始化应用程序...") # 设置设备状态为待命 self.set_device_state(DeviceState.IDLE) # 初始化音频编解码器 self._initialize_audio() # 设置协议回调 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: from xiaozhi.services.audio.codec import AudioCodec self.audio_codec = AudioCodec() logger.info("音频编解码器初始化成功") except Exception as e: logger.error(f"初始化音频设备失败: {e}") self.alert("错误", f"初始化音频设备失败: {e}") def _initialize_display(self): """初始化显示界面""" self.display = gui_display.GuiDisplay() # 设置回调函数 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), ) 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() 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() for task in tasks: try: task() except Exception as e: logger.error(f"执行调度任务时出错: {e}") def schedule(self, callback): """调度任务到主循环""" with self.mutex: # 如果是中止语音的任务，检查是否已经存在相同类型的任务 if "abort_speaking" in str(callback): # 如果已经有中止任务在队列中，就不再添加 if any("abort_speaking" in str(task) for task in self.main_tasks): return 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 ) def _handle_output_audio(self): """处理音频输出""" if self.device_state != DeviceState.SPEAKING: return self.audio_codec.play_audio() def _on_network_error(self, message): """网络错误回调""" self.keep_listening = False self.set_device_state(DeviceState.IDLE) if self.device_state != DeviceState.CONNECTING: logger.info("检测到连接断开") self.set_device_state(DeviceState.IDLE) # 关闭现有连接 if self.protocol: asyncio.run_coroutine_threadsafe( self.protocol.close_audio_channel(), self.loop ) def _attempt_reconnect(self): """尝试重新连接服务器""" if self.device_state != DeviceState.CONNECTING: logger.info("检测到连接断开，尝试重新连接...") self.set_device_state(DeviceState.CONNECTING) # 关闭现有连接 if self.protocol: asyncio.run_coroutine_threadsafe( self.protocol.close_audio_channel(), self.loop ) # 延迟一秒后尝试重新连接 def delayed_reconnect(): time.sleep(1) asyncio.run_coroutine_threadsafe(self._reconnect(), self.loop) threading.Thread(target=delayed_reconnect, daemon=True).start() async def _reconnect(self): """重新连接到服务器""" # 设置协议回调 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 # 连接到服务器 retry_count = 0 max_retries = 3 while retry_count < max_retries: logger.info(f"尝试重新连接 (尝试 {retry_count + 1}/{max_retries})...") if await self.protocol.connect(): logger.info("重新连接成功") self.set_device_state(DeviceState.IDLE) return True retry_count += 1 await asyncio.sleep(2) # 等待2秒后重试 logger.error(f"重新连接失败，已尝试 {max_retries} 次") self.schedule(lambda: self.alert("连接错误", "无法重新连接到服务器")) self.set_device_state(DeviceState.IDLE) return False 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) 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.audio_codec.clear_audio_queue() if ( self.device_state == DeviceState.IDLE or self.device_state == DeviceState.LISTENING ): self.set_device_state(DeviceState.SPEAKING) def _handle_tts_stop(self): """处理TTS停止事件""" if self.device_state == DeviceState.SPEAKING: # 给音频播放一个缓冲时间，确保所有音频都播放完毕 def delayed_state_change(): # 等待音频队列清空 self.audio_codec.wait_for_audio_complete() # 状态转换 if self.keep_listening: asyncio.run_coroutine_threadsafe( self.protocol.send_start_listening(ListeningMode.AUTO_STOP), self.loop, ) self.set_device_state(DeviceState.LISTENING) else: self.set_device_state(DeviceState.IDLE) # 安排延迟执行 threading.Thread(target=delayed_state_change, daemon=True).start() 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()) def _start_audio_streams(self): """启动音频流""" try: # 确保流已关闭后再重新打开 if ( self.audio_codec.input_stream and self.audio_codec.input_stream.is_active() ): self.audio_codec.input_stream.stop_stream() # 重新打开流 self.audio_codec.input_stream.start_stream() if ( self.audio_codec.output_stream and self.audio_codec.output_stream.is_active() ): self.audio_codec.output_stream.stop_stream() # 重新打开流 self.audio_codec.output_stream.start_stream() # 设置事件触发器 threading.Thread( target=self._audio_input_event_trigger, daemon=True ).start() threading.Thread( target=self._audio_output_event_trigger, daemon=True ).start() logger.info("音频流已启动") except Exception as e: logger.error(f"启动音频流失败: {e}") def _audio_input_event_trigger(self): """音频输入事件触发器""" while self.running: try: if ( self.audio_codec.input_stream and self.audio_codec.input_stream.is_active() ): self.events[EventType.AUDIO_INPUT_READY_EVENT].set() except OSError as e: logger.error(f"音频输入流错误: {e}") # 如果流已关闭，尝试重新打开或者退出循环 if "Stream not open" in str(e): break except Exception as e: logger.error(f"音频输入事件触发器错误: {e}") time.sleep(AudioConfig.FRAME_DURATION / 1000) # 按帧时长触发 def _audio_output_event_trigger(self): """音频输出事件触发器""" while ( self.running and self.audio_codec.output_stream and self.audio_codec.output_stream.is_active() ): # 当队列中有数据时才触发事件 if ( not self.audio_codec.audio_decode_queue.empty() ): # 修改为使用 audio_codec 的队列 self.events[EventType.AUDIO_OUTPUT_READY_EVENT].set() time.sleep(0.02) # 稍微延长检查间隔 async def _on_audio_channel_closed(self): """音频通道关闭回调""" logger.info("音频通道已关闭") self.set_device_state(DeviceState.IDLE) self.keep_listening = False self.schedule(lambda: self._stop_audio_streams()) def _stop_audio_streams(self): """停止音频流""" try: if ( self.audio_codec.input_stream and self.audio_codec.input_stream.is_active() ): self.audio_codec.input_stream.stop_stream() if ( self.audio_codec.output_stream and self.audio_codec.output_stream.is_active() ): self.audio_codec.output_stream.stop_stream() logger.info("音频流已停止") except Exception as e: logger.error(f"停止音频流失败: {e}") def set_device_state(self, state): """设置设备状态""" if self.device_state == state: return old_state = self.device_state # 如果从 SPEAKING 状态切换出去，确保音频播放完成 if old_state == DeviceState.SPEAKING: self.audio_codec.wait_for_audio_complete() self.device_state = state logger.info(f"状态变更: {old_state} -> {state}") # 根据状态执行相应操作 if state == DeviceState.IDLE: self.display.update_status("待命") self.display.update_emotion("😶") # 停止输出流但不关闭它 if ( self.audio_codec.output_stream and self.audio_codec.output_stream.is_active() ): try: self.audio_codec.output_stream.stop_stream() except Exception as e: logger.warning(f"停止输出流时出错: {e}") elif state == DeviceState.CONNECTING: self.display.update_status("连接中...") elif state == DeviceState.LISTENING: self.display.update_status("聆听中...") self.display.update_emotion("🙂") 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}") # 使用 AudioCodec 类中的方法重新初始化 self.audio_codec._reinitialize_input_stream() elif state == DeviceState.SPEAKING: self.display.update_status("说话中...") # 确保输出流处于活跃状态 if 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}") # 使用 AudioCodec 类中的方法重新初始化 self.audio_codec._reinitialize_output_stream() # 停止输入流 if ( self.audio_codec.input_stream and self.audio_codec.input_stream.is_active() ): try: self.audio_codec.input_stream.stop_stream() except Exception as e: logger.warning(f"停止输入流时出错: {e}") # 通知状态变化 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): """获取当前表情""" emotions = { "neutral": "😶", "happy": "🙂", "laughing": "😆", "funny": "😂", "sad": "😔", "angry": "😠", "crying": "😭", "loving": "😍", "embarrassed": "😳", "surprised": "😲", "shocked": "😱", "thinking": "🤔", "winking": "😉", "cool": "😎", "relaxed": "😌", "delicious": "🤤", "kissy": "😘", "confident": "😏", "sleepy": "😴", "silly": "😜", "confused": "🙄", } return emotions.get(self.current_emotion, "😶") 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.device_state == DeviceState.IDLE: 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.set_device_state(DeviceState.IDLE) # 设置设备状态为空闲 return except Exception as e: logger.error(f"打开音频通道时发生错误: {e}") self.alert("错误", f"打开音频通道失败: {str(e)}") self.set_device_state(DeviceState.IDLE) return asyncio.run_coroutine_threadsafe( self.protocol.send_start_listening(ListeningMode.MANUAL), self.loop ) 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.set_device_state(DeviceState.IDLE) self.alert("错误", "打开音频通道失败") return await self.protocol.send_start_listening(ListeningMode.MANUAL) self.set_device_state(DeviceState.LISTENING) def toggle_chat_state(self): """切换聊天状态""" 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.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.set_device_state(DeviceState.IDLE) # 设置设备状态为空闲 return except Exception as e: logger.error(f"打开音频通道时发生错误: {e}") self.alert("错误", f"打开音频通道失败: {str(e)}") self.set_device_state(DeviceState.IDLE) return self.keep_listening = True # 开始监听 # 启动自动停止的监听模式 asyncio.run_coroutine_threadsafe( self.protocol.send_start_listening(ListeningMode.AUTO_STOP), self.loop ) self.set_device_state(DeviceState.LISTENING) # 设置设备状态为监听中 # 如果设备正在说话，停止当前说话 elif self.device_state == DeviceState.SPEAKING: self.abort_speaking(AbortReason.NONE) # 中止说话 # 如果设备正在监听，关闭音频通道 elif self.device_state == DeviceState.LISTENING: asyncio.run_coroutine_threadsafe( self.protocol.close_audio_channel(), self.loop ) 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): """中止语音输出""" logger.info(f"中止语音输出，原因: {reason}") self.aborted = True asyncio.run_coroutine_threadsafe( self.protocol.send_abort_speaking(reason), self.loop ) self.set_device_state(DeviceState.IDLE) # 添加此代码：当用户主动打断时自动进入录音模式 if reason == AbortReason.WAKE_WORD_DETECTED and self.keep_listening: # 短暂延迟确保abort命令被处理 def start_listening_after_abort(): time.sleep(0.2) # 短暂延迟 self.set_device_state(DeviceState.IDLE) self.schedule(lambda: self.toggle_chat_state()) threading.Thread(target=start_listening_after_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) 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 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