From eee907b1a23156d712f2c282cbade8faf65de38e Mon Sep 17 00:00:00 2001 From: FAN-yeB <1442100690@qq.com> Date: Tue, 9 Sep 2025 09:37:31 +0800 Subject: [PATCH] =?UTF-8?q?=E5=85=BC=E5=AE=B9udp=E5=92=8Cwebsocket?= =?UTF-8?q?=E5=8D=8F=E8=AE=AE=E4=BC=A0=E8=BE=93=E9=9F=B3=E9=A2=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/impl/DeviceServiceImpl.java | 4 +- main/xiaozhi-server/config.yaml | 2 + main/xiaozhi-server/core/connection.py | 12 ++- .../core/handle/sendAudioHandle.py | 94 ++++++++++++------- 4 files changed, 73 insertions(+), 39 deletions(-) diff --git a/main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/DeviceServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/DeviceServiceImpl.java index eeacce9a..b4343276 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/DeviceServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/DeviceServiceImpl.java @@ -181,9 +181,9 @@ public class DeviceServiceImpl extends BaseServiceImpl response.setWebsocket(websocket); // 添加MQTT UDP配置 - // 从系统参数获取MQTT Gateway地址，如果未配置不使用默认值 + // 从系统参数获取MQTT Gateway地址，仅在配置有效时使用 String mqttUdpConfig = sysParamsService.getValue(Constant.SERVER_MQTT_GATEWAY, false); - if(!StringUtils.isBlank(mqttUdpConfig) && deviceById != null) { + if(mqttUdpConfig != null && !mqttUdpConfig.equals("null") && !mqttUdpConfig.isEmpty() && deviceById != null) { try { DeviceReportRespDTO.MQTT mqtt = buildMqttConfig(macAddress, clientId, deviceById); if (mqtt != null) { diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index c0aa3f93..1afc3530 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -25,6 +25,8 @@ 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 # 认证配置 diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 3e6e8633..e23826a1 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -282,8 +282,17 @@ class ConnectionHandler: if self.asr is None: return + # 检查是否需要处理头部（只有当mqtt_gateway有实际值时才处理头部） + mqtt_gateway = self.config.get("server", {}).get("mqtt_gateway") + # 当mqtt_gateway为None, "null", "", 或实际的null值时，不处理头部 + need_header_processing = mqtt_gateway and mqtt_gateway not in [None, "null", ""] and str(mqtt_gateway).strip() != "" + + # 调试日志：首次连接时记录配置 + if not hasattr(self, '_logged_mqtt_config'): + self.logger.bind(tag=TAG).info(f"MQTT Gateway配置: '{mqtt_gateway}', 头部处理: {need_header_processing}") + self._logged_mqtt_config = True - if len(message) >= 16: + 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') @@ -304,6 +313,7 @@ class ConnectionHandler: except Exception as e: self.logger.bind(tag=TAG).error(f"解析WebSocket音频包失败: {e}") + # 不需要头部处理或没有头部时，直接处理原始消息 self.asr_audio_queue.put(message) def _process_websocket_audio(self, audio_data, timestamp): diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index c8a05ded..56d21932 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -43,6 +43,11 @@ async def sendAudio(conn, audios, frame_duration=60): if audios is None or len(audios) == 0: return + # 检查是否需要添加头部（只有当mqtt_gateway有实际值时才添加头部） + mqtt_gateway = conn.config.get("server", {}).get("mqtt_gateway") + # 当mqtt_gateway为None, "null", "", 或实际的null值时，不添加头部 + need_header = mqtt_gateway and mqtt_gateway not in [None, "null", ""] and str(mqtt_gateway).strip() != "" + if isinstance(audios, bytes): if conn.client_abort: return @@ -68,18 +73,22 @@ async def sendAudio(conn, audios, frame_duration=60): if delay > 0: await asyncio.sleep(delay) - # 为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 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) + else: + # 直接发送opus数据包，不添加头部 + await conn.websocket.send(audios) # 更新流控状态 flow_control["packet_count"] += 1 @@ -90,20 +99,29 @@ async def sendAudio(conn, audios, frame_duration=60): start_time = time.perf_counter() play_position = 0 + # 检查是否需要添加头部（只有当mqtt_gateway有实际值时才添加头部） + mqtt_gateway = conn.config.get("server", {}).get("mqtt_gateway") + # 当mqtt_gateway为None, "null", "", 或实际的null值时，不添加头部 + need_header = mqtt_gateway and mqtt_gateway not in [None, "null", ""] and str(mqtt_gateway).strip() != "" + # 执行预缓冲 pre_buffer_frames = min(3, len(audios)) for i in range(pre_buffer_frames): - # 为预缓冲包添加头部 - 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 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) + else: + # 直接发送预缓冲包，不添加头部 + await conn.websocket.send(audios[i]) remaining_audios = audios[pre_buffer_frames:] # 播放剩余音频帧 @@ -121,19 +139,23 @@ async def sendAudio(conn, audios, frame_duration=60): if delay > 0: await asyncio.sleep(delay) - # 为opus数据包添加16字节头部 (timestamp at offset 8, length at offset 12) - 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) + if need_header: + # 为opus数据包添加16字节头部 (timestamp at offset 8, length at offset 12) + 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) + else: + # 直接发送opus数据包，不添加头部 + await conn.websocket.send(opus_packet) play_position += frame_duration