Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-22 07:03:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

兼容udp和websocket协议传输音频

Browse Source
This commit is contained in:
FAN-yeB
2025-09-09 09:37:31 +08:00
parent eed9503391
commit eee907b1a2
4 changed files with 73 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/DeviceServiceImpl.java
+2 -2
View File
@@ -181,9 +181,9 @@ public class DeviceServiceImpl extends BaseServiceImpl<DeviceDao, DeviceEntity>
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) {
main/xiaozhi-server/config.yaml
+2
View File
@@ -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
# 认证配置
main/xiaozhi-server/core/connection.py
+11 -1
View File
@@ -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):
main/xiaozhi-server/core/handle/sendAudioHandle.py
+58 -36
View File
@@ -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