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

mqtt

Browse Source
This commit is contained in:
FAN-yeB
2025-09-08 11:04:42 +08:00
parent d04ec9d510
commit eed9503391
11 changed files with 290 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java
+6
View File
@@ -91,6 +91,12 @@ public interface Constant {
*/
String SERVER_WEBSOCKET = "server.websocket";
/**
* mqtt gateway 配置
*/
String SERVER_MQTT_GATEWAY = "server.mqtt_gateway";
/**
* ota地址
*/
main/manager-api/src/main/java/xiaozhi/modules/device/controller/OTAController.java
+4
View File
@@ -77,6 +77,10 @@ public class OTAController {
@GetMapping
@Hidden
public ResponseEntity<String> 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】配置");
main/manager-api/src/main/java/xiaozhi/modules/device/dto/DeviceReportRespDTO.java
+21 -1
View File
@@ -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;
}
}
main/manager-api/src/main/java/xiaozhi/modules/device/service/impl/DeviceServiceImpl.java
+91 -1
View File
@@ -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<DeviceDao, DeviceEntity>
}
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<DeviceDao, DeviceEntity>
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<String, String> 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;
}
}
main/manager-api/src/main/resources/db/changelog/202509080921.sql
+7
View File
@@ -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 配置');
main/manager-api/src/main/resources/db/changelog/202509080927.sql
+2
View File
@@ -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 密钥 配置');
main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml
+16
View File
@@ -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
main/xiaozhi-server/core/connection.py
+82
View File
@@ -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()
main/xiaozhi-server/core/handle/receiveAudioHandle.py
+17 -17
View File
@@ -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))
main/xiaozhi-server/core/handle/sendAudioHandle.py
+42 -8
View File
@@ -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
main/xiaozhi-server/core/providers/tts/base.py
+2 -1
View File
@@ -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数据