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..eeacce9a 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,12 +1,16 @@ 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; 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; @@ -175,7 +179,22 @@ public class DeviceServiceImpl extends BaseServiceImpl } response.setWebsocket(websocket); - + + // 添加MQTT UDP配置 + // 从系统参数获取MQTT Gateway地址，如果未配置不使用默认值 + String mqttUdpConfig = sysParamsService.getValue(Constant.SERVER_MQTT_GATEWAY, false); + if(!StringUtils.isBlank(mqttUdpConfig) && 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 +456,75 @@ 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 = System.getenv("MQTT_SIGNATURE_KEY"); + if (StringUtils.isBlank(signatureKey)) { + // 如果环境变量没有，尝试从系统参数获取 + signatureKey = sysParamsService.getValue("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/202509080921.sql b/main/manager-api/src/main/resources/db/changelog/202509080921.sql new file mode 100644 index 00000000..4d89ee57 --- /dev/null +++ b/main/manager-api/src/main/resources/db/changelog/202509080921.sql @@ -0,0 +1,7 @@ +delete from `sys_params` where id = 108; +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 (108, 'server.mqtt_gateway', 'null', 'string', 1, 'mqtt gateway 配置'); + +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 (109, 'server.udp_gateway', 'null', 'string', 1, 'udp gateway 配置'); + diff --git a/main/manager-api/src/main/resources/db/changelog/202509080927.sql b/main/manager-api/src/main/resources/db/changelog/202509080927.sql new file mode 100644 index 00000000..4741af88 --- /dev/null +++ b/main/manager-api/src/main/resources/db/changelog/202509080927.sql @@ -0,0 +1,2 @@ +delete from `sys_params` where param_code = 'mqtt.signature_key'; +INSERT INTO `sys_params` (id, param_code, param_value, value_type, param_type, remark) VALUES (120, 'mqtt.signature_key', 'null', 'string', 1, 'mqtt 密钥 配置'); 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 0e8d3dc3..ed66ebb7 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 @@ -303,3 +303,19 @@ databaseChangeLog: - sqlFile: encoding: utf8 path: classpath:db/changelog/202508131557.sql + + - changeSet: + id: 202509080921 + author: fan + changes: + - sqlFile: + encoding: utf8 + path: classpath:db/changelog/202509080921.sql + + - changeSet: + id: 202509080927 + author: fan + changes: + - sqlFile: + encoding: utf8 + path: classpath:db/changelog/202509080927.sql \ No newline at end of file diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 4b3b9d1c..3e6e8633 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -281,8 +281,86 @@ class ConnectionHandler: return if self.asr is None: return + + + if 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}") + self.asr_audio_queue.put(message) + 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) + + 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: @@ -921,6 +999,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..e6f39632 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -1,11 +1,12 @@ -import time -import json -import asyncio -from core.utils.util import audio_to_data -from core.handle.abortHandle import handleAbortMessage +from core.handle.sendAudioHandle import send_stt_message from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit -from core.handle.sendAudioHandle import send_stt_message, SentenceType +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 TAG = __name__ @@ -21,6 +22,7 @@ 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) @@ -29,16 +31,18 @@ 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 actual_text = text - + try: # 尝试解析JSON格式的输入 if text.strip().startswith('{') and text.strip().endswith('}'): @@ -47,13 +51,13 @@ async def startToChat(conn, text): 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 @@ -114,12 +118,10 @@ 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 @@ -138,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)) # 逐个播放数字 @@ -146,17 +148,15 @@ 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)) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 661c4e21..c8a05ded 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -55,6 +55,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 +68,22 @@ async def sendAudio(conn, audios, frame_duration=60): if delay > 0: await asyncio.sleep(delay) - # 发送数据包 - await conn.websocket.send(audios) + # 为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) # 更新流控状态 flow_control["packet_count"] += 1 + flow_control["sequence"] += 1 flow_control["last_send_time"] = time.perf_counter() else: # 文件型音频走普通播放 @@ -81,11 +93,21 @@ 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]) + # 为预缓冲包添加头部 + 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) 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 @@ -98,9 +120,21 @@ async def sendAudio(conn, audios, frame_duration=60): delay = expected_time - current_time if delay > 0: await asyncio.sleep(delay) - - await conn.websocket.send(opus_packet) - + + # 为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) + 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数据