diff --git a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java index afd7addf..6d4e7ef0 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java +++ b/main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java @@ -91,6 +91,12 @@ public interface Constant { */ String SERVER_WEBSOCKET = "server.websocket"; + /** + * mqtt gateway 配置 + */ + String SERVER_MQTT_GATEWAY = "server.mqtt_gateway"; + + /** * ota地址 */ diff --git a/main/manager-api/src/main/java/xiaozhi/modules/device/controller/OTAController.java b/main/manager-api/src/main/java/xiaozhi/modules/device/controller/OTAController.java index 023510cd..26e5dbe6 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/device/controller/OTAController.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/device/controller/OTAController.java @@ -77,6 +77,10 @@ public class OTAController { @GetMapping @Hidden public ResponseEntity getOTA() { + String mqttUdpConfig = sysParamsService.getValue(Constant.SERVER_MQTT_GATEWAY, false); + if(StringUtils.isBlank(mqttUdpConfig)) { + return ResponseEntity.ok("OTA接口不正常，缺少mqtt_gateway地址，请登录智控台，在参数管理找到【server.mqtt_gateway】配置"); + } String wsUrl = sysParamsService.getValue(Constant.SERVER_WEBSOCKET, true); if (StringUtils.isBlank(wsUrl) || wsUrl.equals("null")) { return ResponseEntity.ok("OTA接口不正常，缺少websocket地址，请登录智控台，在参数管理找到【server.websocket】配置"); diff --git a/main/manager-api/src/main/java/xiaozhi/modules/device/dto/DeviceReportRespDTO.java b/main/manager-api/src/main/java/xiaozhi/modules/device/dto/DeviceReportRespDTO.java index f938001e..423868ca 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/device/dto/DeviceReportRespDTO.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/device/dto/DeviceReportRespDTO.java @@ -23,6 +23,9 @@ public class DeviceReportRespDTO { @Schema(description = "WebSocket配置") private Websocket websocket; + @Schema(description = "MQTT Gateway配置") + private MQTT mqtt; + @Getter @Setter public static class Firmware { @@ -70,4 +73,21 @@ public class DeviceReportRespDTO { @Schema(description = "WebSocket服务器地址") private String url; } -} + + @Getter + @Setter + public static class MQTT { + @Schema(description = "MQTT 配置网址") + private String endpoint; + @Schema(description = "MQTT 客户端唯一标识符") + private String client_id; + @Schema(description = "MQTT 认证用户名") + private String username; + @Schema(description = "MQTT 认证密码") + private String password; + @Schema(description = "ESP32 发布消息的主题") + private String publish_topic; + @Schema(description = "ESP32 订阅的主题") + private String subscribe_topic; + } +} \ No newline at end of file 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 3cff111d..66380041 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 @@ -1,6 +1,8 @@ package xiaozhi.modules.device.service.impl; +import java.nio.charset.StandardCharsets; import java.time.Instant; +import java.util.Base64; import java.util.Date; import java.util.HashMap; import java.util.List; @@ -8,6 +10,9 @@ import java.util.Map; import java.util.TimeZone; import java.util.UUID; +import javax.crypto.Mac; +import javax.crypto.spec.SecretKeySpec; + import org.apache.commons.lang3.StringUtils; import org.springframework.aop.framework.AopContext; import org.springframework.scheduling.annotation.Async; @@ -33,6 +38,7 @@ import xiaozhi.common.user.UserDetail; import xiaozhi.common.utils.ConvertUtils; import xiaozhi.common.utils.DateUtils; import xiaozhi.modules.device.dao.DeviceDao; +import xiaozhi.modules.device.dto.DeviceManualAddDTO; import xiaozhi.modules.device.dto.DevicePageUserDTO; import xiaozhi.modules.device.dto.DeviceReportReqDTO; import xiaozhi.modules.device.dto.DeviceReportRespDTO; @@ -44,7 +50,6 @@ import xiaozhi.modules.device.vo.UserShowDeviceListVO; import xiaozhi.modules.security.user.SecurityUser; import xiaozhi.modules.sys.service.SysParamsService; import xiaozhi.modules.sys.service.SysUserUtilService; -import xiaozhi.modules.device.dto.DeviceManualAddDTO; @Slf4j @Service @@ -176,6 +181,21 @@ public class DeviceServiceImpl extends BaseServiceImpl response.setWebsocket(websocket); + // 添加MQTT UDP配置 + // 从系统参数获取MQTT Gateway地址，仅在配置有效时使用 + String mqttUdpConfig = sysParamsService.getValue(Constant.SERVER_MQTT_GATEWAY, false); + if (mqttUdpConfig != null && !mqttUdpConfig.equals("null") && !mqttUdpConfig.isEmpty() && deviceById != null) { + try { + DeviceReportRespDTO.MQTT mqtt = buildMqttConfig(macAddress, clientId, deviceById); + if (mqtt != null) { + mqtt.setEndpoint(mqttUdpConfig); + response.setMqtt(mqtt); + } + } catch (Exception e) { + log.error("生成MQTT配置失败: {}", e.getMessage()); + } + } + if (deviceById != null) { // 如果设备存在，则异步更新上次连接时间和版本信息 String appVersion = deviceReport.getApplication() != null ? deviceReport.getApplication().getVersion() @@ -437,4 +457,74 @@ public class DeviceServiceImpl extends BaseServiceImpl entity.setAutoUpdate(1); baseDao.insert(entity); } + + /** + * 生成MQTT密码签名 + * + * @param content 签名内容 (clientId + '|' + username) + * @param secretKey 密钥 + * @return Base64编码的HMAC-SHA256签名 + */ + private String generatePasswordSignature(String content, String secretKey) throws Exception { + Mac hmac = Mac.getInstance("HmacSHA256"); + SecretKeySpec keySpec = new SecretKeySpec(secretKey.getBytes(StandardCharsets.UTF_8), "HmacSHA256"); + hmac.init(keySpec); + byte[] signature = hmac.doFinal(content.getBytes(StandardCharsets.UTF_8)); + return Base64.getEncoder().encodeToString(signature); + } + + /** + * 构建MQTT配置信息 + * + * @param macAddress MAC地址 + * @param clientId 客户端ID (UUID) + * @param device 设备信息 + * @return MQTT配置对象 + */ + private DeviceReportRespDTO.MQTT buildMqttConfig(String macAddress, String clientId, DeviceEntity device) + throws Exception { + // 从环境变量或系统参数获取签名密钥 + String signatureKey = sysParamsService.getValue("server.mqtt_signature_key", false); + if (StringUtils.isBlank(signatureKey)) { + log.warn("缺少MQTT_SIGNATURE_KEY，跳过MQTT配置生成"); + return null; + } + + // 构建客户端ID格式：groupId@@@macAddress_without_colon@@@uuid + String groupId = device.getBoard() != null ? device.getBoard() : "GID_default"; + String deviceIdNoColon = macAddress.replace(":", "_"); + String mqttClientId = String.format("%s@@@%s@@@%s", groupId, deviceIdNoColon, clientId); + + // 构建用户数据（包含IP等信息） + Map userData = new HashMap<>(); + // 尝试获取客户端IP + try { + ServletRequestAttributes attributes = (ServletRequestAttributes) RequestContextHolder + .getRequestAttributes(); + if (attributes != null) { + HttpServletRequest request = attributes.getRequest(); + String clientIp = request.getRemoteAddr(); + userData.put("ip", clientIp); + } + } catch (Exception e) { + userData.put("ip", "unknown"); + } + + // 将用户数据编码为Base64 JSON + String userDataJson = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(userData); + String username = Base64.getEncoder().encodeToString(userDataJson.getBytes(StandardCharsets.UTF_8)); + + // 生成密码签名 + String password = generatePasswordSignature(mqttClientId + "|" + username, signatureKey); + + // 构建MQTT配置 + DeviceReportRespDTO.MQTT mqtt = new DeviceReportRespDTO.MQTT(); + mqtt.setClient_id(mqttClientId); + mqtt.setUsername(username); + mqtt.setPassword(password); + mqtt.setPublish_topic("device-server"); + mqtt.setSubscribe_topic("devices/p2p/" + deviceIdNoColon); + + return mqtt; + } } diff --git a/main/manager-api/src/main/resources/db/changelog/202509080922.sql b/main/manager-api/src/main/resources/db/changelog/202509080922.sql new file mode 100644 index 00000000..424b3cd8 --- /dev/null +++ b/main/manager-api/src/main/resources/db/changelog/202509080922.sql @@ -0,0 +1,8 @@ +delete from `sys_params` where param_code = 'server.mqtt_gateway'; +INSERT INTO `sys_params` (id, param_code, param_value, value_type, param_type, remark) VALUES (116, 'server.mqtt_gateway', 'null', 'string', 1, 'mqtt gateway 配置'); + +delete from `sys_params` where param_code = 'server.mqtt_signature_key'; +INSERT INTO `sys_params` (id, param_code, param_value, value_type, param_type, remark) VALUES (117, 'server.mqtt_signature_key', 'null', 'string', 1, 'mqtt 密钥 配置'); + +delete from `sys_params` where param_code = 'server.udp_gateway'; +INSERT INTO `sys_params` (id, param_code, param_value, value_type, param_type, remark) VALUES (118, 'server.udp_gateway', 'null', 'string', 1, 'udp gateway 配置'); \ No newline at end of file diff --git a/main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml b/main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml index 0bf29d2a..2890424d 100755 --- a/main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml +++ b/main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml @@ -331,7 +331,6 @@ databaseChangeLog: - sqlFile: encoding: utf8 path: classpath:db/changelog/202509091042.sql - - changeSet: id: 202509091633 author: fyb @@ -339,3 +338,10 @@ databaseChangeLog: - sqlFile: encoding: utf8 path: classpath:db/changelog/202509091633.sql + - changeSet: + id: 202509080922 + author: fyb + changes: + - sqlFile: + encoding: utf8 + path: classpath:db/changelog/202509080922.sql diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 324237dd..8db70fbf 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/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,12 @@ 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_gateway") + if self.conn_from_mqtt_gateway: + self.logger.bind(tag=TAG).info("连接来自:MQTT网关") + # 初始化活动时间戳 self.last_activity_time = time.time() * 1000 @@ -277,12 +286,82 @@ class ConnectionHandler: if isinstance(message, str): await handleTextMessage(self, message) elif isinstance(message, bytes): - if self.vad is None: - return - if self.asr is None: + if self.vad is None or self.asr is None: return + + # 处理来自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"): + 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: + processed_any = False + for ts in sorted(self.audio_timestamp_buffer.keys()): + if ts > self.last_processed_timestamp: + buffered_audio = self.audio_timestamp_buffer.pop(ts) + self.asr_audio_queue.put(buffered_audio) + self.last_processed_timestamp = ts + processed_any = True + break + else: + # 乱序包，暂存 + if len(self.audio_timestamp_buffer) < self.max_timestamp_buffer_size: + self.audio_timestamp_buffer[timestamp] = audio_data + else: + self.asr_audio_queue.put(audio_data) + async def handle_restart(self, message): """处理服务器重启请求""" try: @@ -857,7 +936,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", @@ -925,6 +1008,10 @@ class ConnectionHandler: async def close(self, ws=None): """资源清理方法""" try: + # 清理音频缓冲区 + if hasattr(self, "audio_buffer"): + self.audio_buffer.clear() + # 取消超时任务 if self.timeout_task and not self.timeout_task.done(): self.timeout_task.cancel() diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 8db50633..8b635909 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -29,11 +29,13 @@ async def handleAudioMessage(conn, audio): # 接收音频 await conn.asr.receive_audio(conn, audio, have_voice) + async def resume_vad_detection(conn): # 等待2秒后恢复VAD检测 await asyncio.sleep(1) conn.just_woken_up = False + async def startToChat(conn, text): # 检查输入是否是JSON格式（包含说话人信息） speaker_name = None @@ -41,11 +43,11 @@ async def startToChat(conn, 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格式的文本，不解析 diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 661c4e21..d159d3ff 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -30,6 +30,53 @@ async def sendAudioMessage(conn, sentenceType, audios, text): await conn.close() +def calculate_timestamp_and_sequence(conn, start_time, packet_index, frame_duration=60): + """ + 计算音频数据包的时间戳和序列号 + Args: + conn: 连接对象 + start_time: 起始时间（性能计数器值） + packet_index: 数据包索引 + frame_duration: 帧时长（毫秒），匹配 Opus 编码 + Returns: + tuple: (timestamp, sequence) + """ + # 计算时间戳（使用播放位置计算） + timestamp = int((start_time + packet_index * frame_duration / 1000) * 1000) % ( + 2**32 + ) + + # 计算序列号 + if hasattr(conn, "audio_flow_control"): + sequence = conn.audio_flow_control["sequence"] + else: + sequence = packet_index # 如果没有流控状态，直接使用索引 + + return timestamp, sequence + + +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): """ @@ -55,6 +102,7 @@ async def sendAudio(conn, audios, frame_duration=60): "last_send_time": 0, "packet_count": 0, "start_time": time.perf_counter(), + "sequence": 0, # 添加序列号 } flow_control = conn.audio_flow_control @@ -67,11 +115,23 @@ async def sendAudio(conn, audios, frame_duration=60): if delay > 0: await asyncio.sleep(delay) - # 发送数据包 - await conn.websocket.send(audios) + if conn.conn_from_mqtt_gateway: + # 计算时间戳和序列号 + timestamp, sequence = calculate_timestamp_and_sequence( + conn, + flow_control["start_time"], + flow_control["packet_count"], + frame_duration, + ) + # 调用通用函数发送带头部的数据包 + await _send_to_mqtt_gateway(conn, audios, timestamp, sequence) + else: + # 直接发送opus数据包，不添加头部 + await conn.websocket.send(audios) # 更新流控状态 flow_control["packet_count"] += 1 + flow_control["sequence"] += 1 flow_control["last_send_time"] = time.perf_counter() else: # 文件型音频走普通播放 @@ -81,11 +141,20 @@ async def sendAudio(conn, audios, frame_duration=60): # 执行预缓冲 pre_buffer_frames = min(3, len(audios)) for i in range(pre_buffer_frames): - await conn.websocket.send(audios[i]) + if conn.conn_from_mqtt_gateway: + # 计算时间戳和序列号 + timestamp, sequence = calculate_timestamp_and_sequence( + conn, start_time, i, frame_duration + ) + # 调用通用函数发送带头部的数据包 + await _send_to_mqtt_gateway(conn, audios[i], timestamp, sequence) + else: + # 直接发送预缓冲包，不添加头部 + await conn.websocket.send(audios[i]) remaining_audios = audios[pre_buffer_frames:] # 播放剩余音频帧 - for opus_packet in remaining_audios: + for i, opus_packet in enumerate(remaining_audios): if conn.client_abort: break @@ -99,7 +168,17 @@ async def sendAudio(conn, audios, frame_duration=60): if delay > 0: await asyncio.sleep(delay) - await conn.websocket.send(opus_packet) + if conn.conn_from_mqtt_gateway: + # 计算时间戳和序列号（使用当前的数据包索引确保连续性） + packet_index = pre_buffer_frames + i + timestamp, sequence = calculate_timestamp_and_sequence( + conn, start_time, packet_index, frame_duration + ) + # 调用通用函数发送带头部的数据包 + await _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence) + else: + # 直接发送opus数据包，不添加头部 + await conn.websocket.send(opus_packet) play_position += frame_duration diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 04a7fa36..c4b32355 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -336,7 +336,8 @@ class TTSProviderBase(ABC): self.conn.audio_flow_control = { 'last_send_time': 0, 'packet_count': 0, - 'start_time': time.perf_counter() + 'start_time': time.perf_counter(), + 'sequence': 0 # 添加序列号 } # 上报TTS数据