diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index 1afc3530..c0aa3f93 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -25,8 +25,6 @@ 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 9f1348b7..d80ec862 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,13 @@ 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") + self.logger.bind(tag=TAG).info( + f"WebSocket URL路径: '{request_path}', 来自MQTT连接: {self.conn_from_mqtt_gateway}" + ) + # 初始化活动时间戳 self.last_activity_time = time.time() * 1000 @@ -277,57 +287,64 @@ class ConnectionHandler: if isinstance(message, str): await handleTextMessage(self, message) elif isinstance(message, bytes): - if self.vad is None: + if self.vad is None or self.asr is None: return - if self.asr is None: - return - - # 检查是否需要处理头部（只有当websocket URL以"?from=mqtt"为结尾时才处理头部） - request_path = self.websocket.request.path - need_header_processing = request_path.endswith("?from=mqtt") - - # 调试日志：首次连接时记录配置 - if not hasattr(self, '_logged_mqtt_config'): - self.logger.bind(tag=TAG).info(f"WebSocket URL路径: '{request_path}', 头部处理: {need_header_processing}") - self._logged_mqtt_config = True - 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') - - - # 提取音频数据 - 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}") - + # 处理来自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'): + 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: @@ -346,30 +363,6 @@ class ConnectionHandler: 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: @@ -940,7 +933,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", @@ -1009,9 +1006,9 @@ class ConnectionHandler: """资源清理方法""" try: # 清理音频缓冲区 - if hasattr(self, 'audio_buffer'): + 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 e6f39632..8b635909 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -1,12 +1,11 @@ -from core.handle.sendAudioHandle import send_stt_message +import time +import json +import asyncio +from core.utils.util import audio_to_data +from core.handle.abortHandle import handleAbortMessage from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit -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 +from core.handle.sendAudioHandle import send_stt_message, SentenceType TAG = __name__ @@ -22,7 +21,6 @@ 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) @@ -42,22 +40,22 @@ async def startToChat(conn, text): # 检查输入是否是JSON格式（包含说话人信息） speaker_name = None actual_text = 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格式的文本，不解析 actual_text = text except (json.JSONDecodeError, KeyError): # 如果解析失败，继续使用原始文本 pass - + # 保存说话人信息到连接对象 if speaker_name: conn.current_speaker = speaker_name @@ -118,10 +116,12 @@ 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 @@ -140,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)) # 逐个播放数字 @@ -148,15 +148,17 @@ 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 df18f2a0..810e8803 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -30,6 +30,28 @@ async def sendAudioMessage(conn, sentenceType, audios, text): await conn.close() +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): """ @@ -42,9 +64,6 @@ async def sendAudio(conn, audios, frame_duration=60): """ if audios is None or len(audios) == 0: return - # 检查是否需要处理头部（只有当websocket URL以"?from=mqtt"为结尾时才处理头部） - request_path = conn.websocket.request.path - need_header = request_path.endswith("?from=mqtt") if isinstance(audios, bytes): if conn.client_abort: @@ -71,19 +90,19 @@ async def sendAudio(conn, audios, frame_duration=60): if delay > 0: await asyncio.sleep(delay) - 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) + if conn.conn_from_mqtt_gateway: + # 计算时间戳 + timestamp = int( + ( + flow_control["start_time"] + + flow_control["packet_count"] * frame_duration / 1000 + ) + * 1000 + ) % (2**32) + # 调用通用函数发送带头部的数据包 + await _send_to_mqtt_gateway( + conn, audios, timestamp, flow_control["sequence"] + ) else: # 直接发送opus数据包，不添加头部 await conn.websocket.send(audios) @@ -97,30 +116,21 @@ async def sendAudio(conn, audios, frame_duration=60): start_time = time.perf_counter() play_position = 0 - # 检查是否需要添加头部（只有当websocket URL以"?from=mqtt"为结尾时才添加头部） - request_path = conn.websocket.request.path - need_header = request_path.endswith("?from=mqtt") - # 执行预缓冲 pre_buffer_frames = min(3, len(audios)) for i in range(pre_buffer_frames): - 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) + if conn.conn_from_mqtt_gateway: + # 计算时间戳 + timestamp = int((start_time + i * frame_duration / 1000) * 1000) % ( + 2**32 + ) + # 调用通用函数发送带头部的数据包 + await _send_to_mqtt_gateway(conn, audios[i], timestamp, i) else: # 直接发送预缓冲包，不添加头部 await conn.websocket.send(audios[i]) remaining_audios = audios[pre_buffer_frames:] - + # 播放剩余音频帧 for i, opus_packet in enumerate(remaining_audios): if conn.client_abort: @@ -135,25 +145,19 @@ async def sendAudio(conn, audios, frame_duration=60): delay = expected_time - current_time if delay > 0: await asyncio.sleep(delay) - - if need_header: - # 为opus数据包添加16字节头部 (timestamp at offset 8, length at offset 12) - timestamp = int((start_time + play_position / 1000) * 1000) % (2**32) # 使用播放位置计算时间戳 + + if conn.conn_from_mqtt_gateway: + # 计算时间戳和序列号 + 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) + # 调用通用函数发送带头部的数据包 + await _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence) else: # 直接发送opus数据包，不添加头部 await conn.websocket.send(opus_packet) - + play_position += frame_duration