update:优化代码

main/xiaozhi-server/config.yaml

@@ -25,8 +25,6 @@ server:
   # 所以如果你使用docker部署时，将vision_explain设置成局域网地址
   # 如果你使用公网部署时，将vision_explain设置成公网地址
   vision_explain: http://你的ip或者域名:端口号/mcp/vision/explain
-  # mqtt网关地址，当这个值为null时，mqtt网关桥接功能不开启，使用websocket双向通信，不使用mqtt和udp协议
-  mqtt_gateway: null
   # OTA返回信息时区偏移量
   timezone_offset: +8
   # 认证配置

main/xiaozhi-server/core/connection.py

@@ -154,6 +154,9 @@ class ConnectionHandler:
         # {"mcp":true} 表示启用MCP功能
         self.features = None
 
+        # 标记连接是否来自MQTT
+        self.conn_from_mqtt_gateway = False
+
         # 初始化提示词管理器
         self.prompt_manager = PromptManager(config, self.logger)
 
@@ -198,6 +201,13 @@ class ConnectionHandler:
             self.websocket = ws
             self.device_id = self.headers.get("device-id", None)
 
+            # 检查是否来自MQTT连接
+            request_path = ws.request.path
+            self.conn_from_mqtt_gateway = request_path.endswith("?from=mqtt")
+            self.logger.bind(tag=TAG).info(
+                f"WebSocket URL路径: '{request_path}', 来自MQTT连接: {self.conn_from_mqtt_gateway}"
+            )
+
             # 初始化活动时间戳
             self.last_activity_time = time.time() * 1000
 
@@ -277,57 +287,64 @@ class ConnectionHandler:
         if isinstance(message, str):
             await handleTextMessage(self, message)
         elif isinstance(message, bytes):
-            if self.vad is None:
+            if self.vad is None or self.asr is None:
                 return
-            if self.asr is None:
-                return
-            
-            # 检查是否需要处理头部（只有当websocket URL以"?from=mqtt"为结尾时才处理头部）
-            request_path = self.websocket.request.path
-            need_header_processing = request_path.endswith("?from=mqtt")
-            
-            # 调试日志：首次连接时记录配置
-            if not hasattr(self, '_logged_mqtt_config'):
-                self.logger.bind(tag=TAG).info(f"WebSocket URL路径: '{request_path}', 头部处理: {need_header_processing}")
-                self._logged_mqtt_config = True
 
-            if need_header_processing and len(message) >= 16:
-                try:
-                    timestamp = int.from_bytes(message[8:12], 'big')
-                    audio_length = int.from_bytes(message[12:16], 'big')
-                    
-                    
-                    # 提取音频数据
-                    if audio_length > 0 and len(message) >= 16 + audio_length:
-                        audio_data = message[16:16 + audio_length]
-                        
-                        # 基于时间戳进行简单排序
-                        self._process_websocket_audio(audio_data, timestamp)
-                        return
-                    elif len(message) > 16:
-                        # 去掉16字节头部
-                        audio_data = message[16:]
-                        self.asr_audio_queue.put(audio_data)
-                        return
-                except Exception as e:
-                    self.logger.bind(tag=TAG).error(f"解析WebSocket音频包失败: {e}")
-            
+            # 处理来自MQTT网关的音频包
+            if self.conn_from_mqtt_gateway and len(message) >= 16:
+                handled = await self._process_mqtt_audio_message(message)
+                if handled:
+                    return
+
             # 不需要头部处理或没有头部时，直接处理原始消息
             self.asr_audio_queue.put(message)
 
+    async def _process_mqtt_audio_message(self, message):
+        """
+        处理来自MQTT网关的音频消息，解析16字节头部并提取音频数据
+
+        Args:
+            message: 包含头部的音频消息
+
+        Returns:
+            bool: 是否成功处理了消息
+        """
+        try:
+            # 提取头部信息
+            timestamp = int.from_bytes(message[8:12], "big")
+            audio_length = int.from_bytes(message[12:16], "big")
+
+            # 提取音频数据
+            if audio_length > 0 and len(message) >= 16 + audio_length:
+                # 有指定长度，提取精确的音频数据
+                audio_data = message[16 : 16 + audio_length]
+                # 基于时间戳进行排序处理
+                self._process_websocket_audio(audio_data, timestamp)
+                return True
+            elif len(message) > 16:
+                # 没有指定长度或长度无效，去掉头部后处理剩余数据
+                audio_data = message[16:]
+                self.asr_audio_queue.put(audio_data)
+                return True
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"解析WebSocket音频包失败: {e}")
+
+        # 处理失败，返回False表示需要继续处理
+        return False
+
     def _process_websocket_audio(self, audio_data, timestamp):
         """处理WebSocket格式的音频包"""
         # 初始化时间戳序列管理
-        if not hasattr(self, 'audio_timestamp_buffer'):
+        if not hasattr(self, "audio_timestamp_buffer"):
             self.audio_timestamp_buffer = {}
             self.last_processed_timestamp = 0
             self.max_timestamp_buffer_size = 20
-            
+
         # 如果时间戳是递增的，直接处理
         if timestamp >= self.last_processed_timestamp:
             self.asr_audio_queue.put(audio_data)
             self.last_processed_timestamp = timestamp
-            
+
             # 处理缓冲区中的后续包
             processed_any = True
             while processed_any:
@@ -346,30 +363,6 @@ class ConnectionHandler:
             else:
                 self.asr_audio_queue.put(audio_data)
 
-    def _process_sequenced_audio(self, audio_data, sequence, timestamp):
-        """处理有序的音频包"""
-        # 初始化音频缓冲区
-        if not hasattr(self, 'audio_buffer'):
-            self.audio_buffer = {}
-            self.expected_sequence = sequence
-            self.max_buffer_size = 20  # 最大缓冲20个包
-            
-        # 如果是下一个期望的包，直接处理
-        if sequence == self.expected_sequence:
-            self.asr_audio_queue.put(audio_data)
-            self.expected_sequence += 1
-            
-            # 检查缓冲区中是否有后续的连续包
-            while self.expected_sequence in self.audio_buffer:
-                buffered_audio = self.audio_buffer.pop(self.expected_sequence)
-                self.asr_audio_queue.put(buffered_audio)
-                self.expected_sequence += 1
-                
-        elif sequence > self.expected_sequence:
-            # 乱序包，暂存到缓冲区
-            if len(self.audio_buffer) < self.max_buffer_size:
-                self.audio_buffer[sequence] = audio_data
-
     async def handle_restart(self, message):
         """处理服务器重启请求"""
         try:
@@ -940,7 +933,11 @@ class ConnectionHandler:
                             {
                                 "id": function_id,
                                 "function": {
-                                    "arguments": "{}" if function_arguments == "" else function_arguments,
+                                    "arguments": (
+                                        "{}"
+                                        if function_arguments == ""
+                                        else function_arguments
+                                    ),
                                     "name": function_name,
                                 },
                                 "type": "function",
@@ -1009,9 +1006,9 @@ class ConnectionHandler:
         """资源清理方法"""
         try:
             # 清理音频缓冲区
-            if hasattr(self, 'audio_buffer'):
+            if hasattr(self, "audio_buffer"):
                 self.audio_buffer.clear()
-                
+
             # 取消超时任务
             if self.timeout_task and not self.timeout_task.done():
                 self.timeout_task.cancel()

main/xiaozhi-server/core/handle/receiveAudioHandle.py

@@ -1,12 +1,11 @@
-from core.handle.sendAudioHandle import send_stt_message
+import time
+import json
+import asyncio
+from core.utils.util import audio_to_data
+from core.handle.abortHandle import handleAbortMessage
 from core.handle.intentHandler import handle_user_intent
 from core.utils.output_counter import check_device_output_limit
-from core.handle.abortHandle import handleAbortMessage
-import time
-import asyncio
-import json
-from core.handle.sendAudioHandle import SentenceType
-from core.utils.util import audio_to_data
+from core.handle.sendAudioHandle import send_stt_message, SentenceType
 
 TAG = __name__
 
@@ -22,7 +21,6 @@ async def handleAudioMessage(conn, audio):
         if not hasattr(conn, "vad_resume_task") or conn.vad_resume_task.done():
             conn.vad_resume_task = asyncio.create_task(resume_vad_detection(conn))
         return
-
     if have_voice:
         if conn.client_is_speaking:
             await handleAbortMessage(conn)
@@ -42,22 +40,22 @@ async def startToChat(conn, text):
     # 检查输入是否是JSON格式（包含说话人信息）
     speaker_name = None
     actual_text = text
-    
+
     try:
         # 尝试解析JSON格式的输入
-        if text.strip().startswith('{') and text.strip().endswith('}'):
+        if text.strip().startswith("{") and text.strip().endswith("}"):
             data = json.loads(text)
-            if 'speaker' in data and 'content' in data:
-                speaker_name = data['speaker']
-                actual_text = data['content']
+            if "speaker" in data and "content" in data:
+                speaker_name = data["speaker"]
+                actual_text = data["content"]
                 conn.logger.bind(tag=TAG).info(f"解析到说话人信息: {speaker_name}")
-                
+
                 # 直接使用JSON格式的文本，不解析
                 actual_text = text
     except (json.JSONDecodeError, KeyError):
         # 如果解析失败，继续使用原始文本
         pass
-    
+
     # 保存说话人信息到连接对象
     if speaker_name:
         conn.current_speaker = speaker_name
@@ -118,10 +116,12 @@ async def no_voice_close_connect(conn, have_voice):
 
 
 async def max_out_size(conn):
+    # 播放超出最大输出字数的提示
+    conn.client_abort = False
     text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！"
     await send_stt_message(conn, text)
     file_path = "config/assets/max_output_size.wav"
-    opus_packets, _ = audio_to_data(file_path)
+    opus_packets = audio_to_data(file_path)
     conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text))
     conn.close_after_chat = True
 
@@ -140,7 +140,7 @@ async def check_bind_device(conn):
 
         # 播放提示音
         music_path = "config/assets/bind_code.wav"
-        opus_packets, _ = audio_to_data(music_path)
+        opus_packets = audio_to_data(music_path)
         conn.tts.tts_audio_queue.put((SentenceType.FIRST, opus_packets, text))
 
         # 逐个播放数字
@@ -148,15 +148,17 @@ async def check_bind_device(conn):
             try:
                 digit = conn.bind_code[i]
                 num_path = f"config/assets/bind_code/{digit}.wav"
-                num_packets, _ = audio_to_data(num_path)
+                num_packets = audio_to_data(num_path)
                 conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, num_packets, None))
             except Exception as e:
                 conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}")
                 continue
         conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None))
     else:
+        # 播放未绑定提示
+        conn.client_abort = False
         text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。"
         await send_stt_message(conn, text)
         music_path = "config/assets/bind_not_found.wav"
-        opus_packets, _ = audio_to_data(music_path)
+        opus_packets = audio_to_data(music_path)
         conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text))

main/xiaozhi-server/core/handle/sendAudioHandle.py

@@ -30,6 +30,28 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
             await conn.close()
 
 
+async def _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence):
+    """
+    发送带16字节头部的opus数据包给mqtt_gateway
+    Args:
+        conn: 连接对象
+        opus_packet: opus数据包
+        timestamp: 时间戳
+        sequence: 序列号
+    """
+    # 为opus数据包添加16字节头部
+    header = bytearray(16)
+    header[0] = 1  # type
+    header[2:4] = len(opus_packet).to_bytes(2, "big")  # payload length
+    header[4:8] = sequence.to_bytes(4, "big")  # sequence
+    header[8:12] = timestamp.to_bytes(4, "big")  # 时间戳
+    header[12:16] = len(opus_packet).to_bytes(4, "big")  # opus长度
+
+    # 发送包含头部的完整数据包
+    complete_packet = bytes(header) + opus_packet
+    await conn.websocket.send(complete_packet)
+
+
 # 播放音频
 async def sendAudio(conn, audios, frame_duration=60):
     """
@@ -42,9 +64,6 @@ async def sendAudio(conn, audios, frame_duration=60):
     """
     if audios is None or len(audios) == 0:
         return
-    # 检查是否需要处理头部（只有当websocket URL以"?from=mqtt"为结尾时才处理头部）
-    request_path = conn.websocket.request.path
-    need_header = request_path.endswith("?from=mqtt")
 
     if isinstance(audios, bytes):
         if conn.client_abort:
@@ -71,19 +90,19 @@ async def sendAudio(conn, audios, frame_duration=60):
         if delay > 0:
             await asyncio.sleep(delay)
 
-        if need_header:
-            # 为opus数据包添加16字节头部
-            timestamp = int((flow_control["start_time"] + flow_control["packet_count"] * frame_duration / 1000) * 1000) % (2**32)
-            header = bytearray(16)
-            header[0] = 1  # type
-            header[2:4] = len(audios).to_bytes(2, 'big')  # payload length
-            header[4:8] = flow_control["sequence"].to_bytes(4, 'big')  # connection id/sequence
-            header[8:12] = timestamp.to_bytes(4, 'big')  # 时间戳
-            header[12:16] = len(audios).to_bytes(4, 'big')  # opus长度
-            
-            # 发送包含头部的完整数据包
-            complete_packet = bytes(header) + audios
-            await conn.websocket.send(complete_packet)
+        if conn.conn_from_mqtt_gateway:
+            # 计算时间戳
+            timestamp = int(
+                (
+                    flow_control["start_time"]
+                    + flow_control["packet_count"] * frame_duration / 1000
+                )
+                * 1000
+            ) % (2**32)
+            # 调用通用函数发送带头部的数据包
+            await _send_to_mqtt_gateway(
+                conn, audios, timestamp, flow_control["sequence"]
+            )
         else:
             # 直接发送opus数据包，不添加头部
             await conn.websocket.send(audios)
@@ -97,30 +116,21 @@ async def sendAudio(conn, audios, frame_duration=60):
         start_time = time.perf_counter()
         play_position = 0
 
-        # 检查是否需要添加头部（只有当websocket URL以"?from=mqtt"为结尾时才添加头部）
-        request_path = conn.websocket.request.path
-        need_header = request_path.endswith("?from=mqtt")
-
         # 执行预缓冲
         pre_buffer_frames = min(3, len(audios))
         for i in range(pre_buffer_frames):
-            if need_header:
-                # 为预缓冲包添加头部
-                timestamp = int((start_time + i * frame_duration / 1000) * 1000) % (2**32)
-                header = bytearray(16)
-                header[0] = 1  # type
-                header[2:4] = len(audios[i]).to_bytes(2, 'big')  # payload length
-                header[4:8] = i.to_bytes(4, 'big')  # sequence
-                header[8:12] = timestamp.to_bytes(4, 'big')  # 时间戳
-                header[12:16] = len(audios[i]).to_bytes(4, 'big')  # opus长度
-                
-                complete_packet = bytes(header) + audios[i]
-                await conn.websocket.send(complete_packet)
+            if conn.conn_from_mqtt_gateway:
+                # 计算时间戳
+                timestamp = int((start_time + i * frame_duration / 1000) * 1000) % (
+                    2**32
+                )
+                # 调用通用函数发送带头部的数据包
+                await _send_to_mqtt_gateway(conn, audios[i], timestamp, i)
             else:
                 # 直接发送预缓冲包，不添加头部
                 await conn.websocket.send(audios[i])
         remaining_audios = audios[pre_buffer_frames:]
-        
+
         # 播放剩余音频帧
         for i, opus_packet in enumerate(remaining_audios):
             if conn.client_abort:
@@ -135,25 +145,19 @@ async def sendAudio(conn, audios, frame_duration=60):
             delay = expected_time - current_time
             if delay > 0:
                 await asyncio.sleep(delay)
-            
-            if need_header:
-                # 为opus数据包添加16字节头部 (timestamp at offset 8, length at offset 12)
-                timestamp = int((start_time + play_position / 1000) * 1000) % (2**32)  # 使用播放位置计算时间戳
+
+            if conn.conn_from_mqtt_gateway:
+                # 计算时间戳和序列号
+                timestamp = int((start_time + play_position / 1000) * 1000) % (
+                    2**32
+                )  # 使用播放位置计算时间戳
                 sequence = pre_buffer_frames + i  # 确保序列号连续
-                header = bytearray(16)
-                header[0] = 1  # type
-                header[2:4] = len(opus_packet).to_bytes(2, 'big')  # payload length
-                header[4:8] = sequence.to_bytes(4, 'big')  # sequence
-                header[8:12] = timestamp.to_bytes(4, 'big')  # 时间戳在第8-11字节
-                header[12:16] = len(opus_packet).to_bytes(4, 'big')  # opus长度在第12-15字节
-                
-                # 发送包含头部的完整数据包
-                complete_packet = bytes(header) + opus_packet
-                await conn.websocket.send(complete_packet)
+                # 调用通用函数发送带头部的数据包
+                await _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence)
             else:
                 # 直接发送opus数据包，不添加头部
                 await conn.websocket.send(opus_packet)
-            
+
             play_position += frame_duration