From d2f29f335e9e831b32ad7c14257a977ba189e63a Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Fri, 15 Aug 2025 16:33:46 +0800 Subject: [PATCH] =?UTF-8?q?update:=20=E9=9F=B3=E9=A2=91=E6=B5=81=E5=BC=8F?= =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/config.yaml | 4 - main/xiaozhi-server/core/connection.py | 8 +- .../xiaozhi-server/core/handle/helloHandle.py | 109 ------------ .../core/handle/intentHandler.py | 4 - .../core/handle/receiveAudioHandle.py | 43 +++-- .../core/handle/sendAudioHandle.py | 53 +----- main/xiaozhi-server/core/handle/textHandle.py | 17 +- .../core/providers/tts/aliyun_stream.py | 144 +-------------- .../xiaozhi-server/core/providers/tts/base.py | 164 ++++++++---------- .../providers/tts/huoshan_double_stream.py | 115 +----------- .../core/providers/tts/index_stream.py | 112 ++---------- .../core/providers/tts/linkerai.py | 115 +----------- .../core/utils/audio_flow_control.py | 14 -- .../core/utils/opus_encoder_utils.py | 50 +----- main/xiaozhi-server/core/utils/p3.py | 44 +---- main/xiaozhi-server/core/utils/util.py | 95 +--------- main/xiaozhi-server/core/utils/wakeup_word.py | 140 --------------- 17 files changed, 143 insertions(+), 1088 deletions(-) delete mode 100644 main/xiaozhi-server/core/utils/wakeup_word.py diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index ab662798..b3c1b1a8 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -61,10 +61,6 @@ delete_audio: true close_connection_no_voice_time: 120 # TTS请求超时时间(秒) tts_timeout: 10 -# 开启唤醒词加速 -enable_wakeup_words_response_cache: true -# 开场是否回复唤醒词 -enable_greeting: true # 说完话是否开启提示音 enable_stop_tts_notify: false # 说完话是否开启提示音，音效地址 diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 1a792427..2ca2602a 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -664,16 +664,14 @@ class ConnectionHandler: # 更新系统prompt至上下文 self.dialogue.update_system_message(self.prompt) - def chat(self, query, tool_call=False, depth=0): + def chat(self, query, depth=0): self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}") self.llm_finish_task = False - if not tool_call: - self.dialogue.put(Message(role="user", content=query)) - # 为最顶层时新建会话ID和发送FIRST请求 if depth == 0: self.sentence_id = str(uuid.uuid4().hex) + self.dialogue.put(Message(role="user", content=query)) self.tts.tts_text_queue.put( TTSMessageDTO( sentence_id=self.sentence_id, @@ -878,7 +876,7 @@ class ConnectionHandler: content=text, ) ) - self.chat(text, tool_call=True, depth=depth + 1) + self.chat(text, depth=depth + 1) elif result.action == Action.NOTFOUND or result.action == Action.ERROR: text = result.response if result.response else result.result self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text) diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index c4f55841..de9587b5 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -1,32 +1,13 @@ -import time import json -import random import asyncio -from core.utils.dialogue import Message -from core.utils.util import audio_to_data -from core.handle.sendAudioHandle import sendAudioMessage, send_stt_message -from core.utils.util import remove_punctuation_and_length, opus_datas_to_wav_bytes -from core.providers.tts.dto.dto import ContentType, SentenceType from core.providers.tools.device_mcp import ( MCPClient, send_mcp_initialize_message, send_mcp_tools_list_request, ) -from core.utils.wakeup_word import WakeupWordsConfig TAG = __name__ -WAKEUP_CONFIG = { - "refresh_time": 5, - "words": ["你好", "你好啊", "嘿，你好", "嗨"], -} - -# 创建全局的唤醒词配置管理器 -wakeup_words_config = WakeupWordsConfig() - -# 用于防止并发调用wakeupWordsResponse的锁 -_wakeup_response_lock = asyncio.Lock() - async def handleHelloMessage(conn, msg_json): """处理hello消息""" @@ -49,93 +30,3 @@ async def handleHelloMessage(conn, msg_json): asyncio.create_task(send_mcp_tools_list_request(conn)) await conn.websocket.send(json.dumps(conn.welcome_msg)) - - -async def checkWakeupWords(conn, text): - enable_wakeup_words_response_cache = conn.config[ - "enable_wakeup_words_response_cache" - ] - - if not enable_wakeup_words_response_cache or not conn.tts: - return False - - _, filtered_text = remove_punctuation_and_length(text) - if filtered_text not in conn.config.get("wakeup_words"): - return False - - conn.just_woken_up = True - await send_stt_message(conn, text) - - # 获取当前音色 - voice = getattr(conn.tts, "voice", "default") - if not voice: - voice = "default" - - # 获取唤醒词回复配置 - response = wakeup_words_config.get_wakeup_response(voice) - if not response or not response.get("file_path"): - response = { - "voice": "default", - "file_path": "config/assets/wakeup_words.wav", - "time": 0, - "text": "哈啰啊，我是小智啦，声音好听的台湾女孩一枚，超开心认识你耶，最近在忙啥，别忘了给我来点有趣的料哦，我超爱听八卦的啦", - } - - # 播放唤醒词回复 - conn.client_abort = False - opus_packets = audio_to_data(response.get("file_path")) - - conn.logger.bind(tag=TAG).info(f"播放唤醒词回复: {response.get('text')}") - await sendAudioMessage(conn, SentenceType.FIRST, opus_packets, response.get("text")) - await sendAudioMessage(conn, SentenceType.LAST, [], None) - - # 补充对话 - conn.dialogue.put(Message(role="assistant", content=response.get("text"))) - - # 检查是否需要更新唤醒词回复 - if time.time() - response.get("time", 0) > WAKEUP_CONFIG["refresh_time"]: - if not _wakeup_response_lock.locked(): - asyncio.create_task(wakeupWordsResponse(conn)) - return True - - -async def wakeupWordsResponse(conn): - if not conn.tts or not conn.llm or not conn.llm.response_no_stream: - return - - try: - # 尝试获取锁，如果获取不到就返回 - if not await _wakeup_response_lock.acquire(): - return - - # 生成唤醒词回复 - wakeup_word = random.choice(WAKEUP_CONFIG["words"]) - question = ( - "此刻用户正在和你说```" - + wakeup_word - + "```。\n请你根据以上用户的内容进行20-30字回复。要符合系统设置的角色情感和态度，不要像机器人一样说话。\n" - + "请勿对这条内容本身进行任何解释和回应，请勿返回表情符号，仅返回对用户的内容的回复。" - ) - - result = conn.llm.response_no_stream(conn.config["prompt"], question) - if not result or len(result) == 0: - return - - # 生成TTS音频 - tts_result = await asyncio.to_thread(conn.tts.to_tts, result) - if not tts_result: - return - - # 获取当前音色 - voice = getattr(conn.tts, "voice", "default") - - wav_bytes = opus_datas_to_wav_bytes(tts_result, sample_rate=16000) - file_path = wakeup_words_config.generate_file_path(voice) - with open(file_path, "wb") as f: - f.write(wav_bytes) - # 更新配置 - wakeup_words_config.update_wakeup_response(voice, file_path, result) - finally: - # 确保在任何情况下都释放锁 - if _wakeup_response_lock.locked(): - _wakeup_response_lock.release() diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index 81911f55..58b2cee7 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -2,7 +2,6 @@ import json import asyncio import uuid from core.handle.sendAudioHandle import send_stt_message -from core.handle.helloHandle import checkWakeupWords from core.utils.util import remove_punctuation_and_length from core.providers.tts.dto.dto import ContentType from core.utils.dialogue import Message @@ -27,9 +26,6 @@ async def handle_user_intent(conn, text): filtered_text = remove_punctuation_and_length(text)[1] if await check_direct_exit(conn, filtered_text): return True - # 检查是否是唤醒词 - if await checkWakeupWords(conn, filtered_text): - return True if conn.intent_type == "function_call": # 使用支持function calling的聊天方法,不再进行意图分析 diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 174c8b2b..7d46685b 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -1,12 +1,12 @@ +import time +import asyncio +import json 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.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.utils.util import audio_to_data_stream TAG = __name__ @@ -42,7 +42,7 @@ async def startToChat(conn, text): # 检查输入是否是JSON格式（包含说话人信息） speaker_name = None actual_text = text - + try: # 尝试解析JSON格式的输入 if text.strip().startswith('{') and text.strip().endswith('}'): @@ -51,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 @@ -121,8 +121,9 @@ async def max_out_size(conn): text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！" await send_stt_message(conn, text) file_path = "config/assets/max_output_size.wav" - opus_packets, _ = audio_to_data(file_path) - conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) + conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text)) + play_audio_frames(conn, file_path) + conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) conn.close_after_chat = True @@ -140,16 +141,15 @@ async def check_bind_device(conn): # 播放提示音 music_path = "config/assets/bind_code.wav" - opus_packets = audio_to_data(music_path) - conn.tts.tts_audio_queue.put((SentenceType.FIRST, opus_packets, text)) + conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text)) + play_audio_frames(conn, music_path) # 逐个播放数字 for i in range(6): # 确保只播放6位数字 try: digit = conn.bind_code[i] num_path = f"config/assets/bind_code/{digit}.wav" - num_packets = audio_to_data(num_path) - conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, num_packets, None)) + play_audio_frames(conn, num_path) except Exception as e: conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}") continue @@ -158,5 +158,18 @@ async def check_bind_device(conn): text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。" await send_stt_message(conn, text) music_path = "config/assets/bind_not_found.wav" - opus_packets = audio_to_data(music_path) - conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) + conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text)) + play_audio_frames(conn, music_path) + conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) + + +def play_audio_frames(conn, file_path): + """播放音频文件并处理发送帧数据""" + def handle_audio_frame(frame_data): + conn.tts.tts_audio_queue.put((SentenceType.MIDDLE, frame_data, None)) + + audio_to_data_stream( + file_path, + is_opus=True, + callback=handle_audio_frame + ) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index bf155ee9..c3c50fbc 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -1,6 +1,4 @@ import json -import asyncio -import time from core.providers.tts.dto.dto import SentenceType from core.utils import textUtils @@ -8,20 +6,18 @@ TAG = __name__ async def sendAudioMessage(conn, sentenceType, audios, text): - # 发送句子开始消息 - conn.logger.bind(tag=TAG).info(f"发送音频消息: {sentenceType}, {text}") - - pre_buffer = False if conn.tts.tts_audio_first_sentence: conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}") conn.tts.tts_audio_first_sentence = False - pre_buffer = True await send_tts_message(conn, "start", None) if sentenceType == SentenceType.FIRST: await send_tts_message(conn, "sentence_start", text) - await sendAudio(conn, audios, pre_buffer) + await sendAudio(conn, audios) + # 发送句子开始消息 + if sentenceType is not SentenceType.MIDDLE: + conn.logger.bind(tag=TAG).info(f"发送音频消息: {sentenceType}, {text}") # 发送结束消息（如果是最后一个文本） if conn.llm_finish_task and sentenceType == SentenceType.LAST: @@ -32,46 +28,12 @@ async def sendAudioMessage(conn, sentenceType, audios, text): # 播放音频 -async def sendAudio(conn, audios, pre_buffer=True): +async def sendAudio(conn, audios): if audios is None: return # 如果audios不是opus数组，则不需要进行遍历，可以直接发送;这里需要进行流控管理，防止发送过快引发客户端溢出 - if isinstance(audios ,bytes): + if isinstance(audios, bytes): await conn.websocket.send(audios) - else: - if audios is None or len(audios) == 0: - return - # 流控参数优化 - frame_duration = 60 # 帧时长（毫秒），匹配 Opus 编码 - start_time = time.perf_counter() - play_position = 0 - # 仅当第一句话时执行预缓冲 - if pre_buffer: - pre_buffer_frames = min(3, len(audios)) - for i in range(pre_buffer_frames): - await conn.websocket.send(audios[i]) - remaining_audios = audios[pre_buffer_frames:] - else: - remaining_audios = audios - - # 播放剩余音频帧 - for opus_packet in remaining_audios: - if conn.client_abort: - break - - # 重置没有声音的状态 - conn.last_activity_time = time.time() * 1000 - - # 计算预期发送时间 - expected_time = start_time + (play_position / 1000) - current_time = time.perf_counter() - delay = expected_time - current_time - if delay > 0: - await asyncio.sleep(delay) - - await conn.websocket.send(opus_packet) - - play_position += frame_duration async def send_tts_message(conn, state, text=None): @@ -100,13 +62,12 @@ async def send_tts_message(conn, state, text=None): async def send_stt_message(conn, text): + """发送 STT 状态消息""" end_prompt_str = conn.config.get("end_prompt", {}).get("prompt") if end_prompt_str and end_prompt_str == text: await send_tts_message(conn, "start") return - """发送 STT 状态消息""" - # 解析JSON格式，提取实际的用户说话内容 display_text = text try: diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 1bcf4545..86363217 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -5,7 +5,6 @@ from core.handle.helloHandle import handleHelloMessage from core.providers.tools.device_mcp import handle_mcp_message from core.utils.util import remove_punctuation_and_length, filter_sensitive_info from core.handle.receiveAudioHandle import startToChat, handleAudioMessage -from core.handle.sendAudioHandle import send_stt_message, send_tts_message from core.providers.tools.device_iot import handleIotDescriptors, handleIotStatus from core.handle.reportHandle import enqueue_asr_report import asyncio @@ -51,23 +50,9 @@ async def handleTextMessage(conn, message): filtered_len, filtered_text = remove_punctuation_and_length( original_text ) - # 识别是否是唤醒词 is_wakeup_words = filtered_text in conn.config.get("wakeup_words") - # 是否开启唤醒词回复 - enable_greeting = conn.config.get("enable_greeting", True) - - if is_wakeup_words and not enable_greeting: - # 如果是唤醒词，且关闭了唤醒词回复，就不用回答 - await send_stt_message(conn, original_text) - await send_tts_message(conn, "stop", None) - conn.client_is_speaking = False - elif is_wakeup_words: - conn.just_woken_up = True - # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） - enqueue_asr_report(conn, "嘿，你好呀", []) - await startToChat(conn, "嘿，你好呀") - else: + if not is_wakeup_words: # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） enqueue_asr_report(conn, original_text, []) # 否则需要LLM对文字内容进行答复 diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 4929491c..32e4d50f 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -216,6 +216,7 @@ class TTSProvider(TTSProviderBase): if message.sentence_type == SentenceType.FIRST: self.conn.client_abort = False + self.reset_flow_controller() if self.conn.client_abort: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") @@ -418,8 +419,6 @@ class TTSProvider(TTSProviderBase): async def _start_monitor_tts_response(self): """监听TTS响应""" - is_first_sentence = True - first_sentence_segment_count = 0 # 添加计数器 try: session_finished = False # 标记会话是否正常结束 while not self.conn.stop_event.is_set(): @@ -440,8 +439,6 @@ class TTSProvider(TTSProviderBase): self.tts_audio_queue.put( (SentenceType.FIRST, [], None) ) - elif event_name == "SentenceBegin": - opus_datas_cache = [] elif event_name == "SentenceEnd": # 发送缓存的数据 if self.conn.tts_MessageText: @@ -482,142 +479,3 @@ class TTSProvider(TTSProviderBase): finally: self._monitor_task = None - def to_tts_stream(self, text: str, opus_handler=None) -> None: - """非流式TTS处理，用于测试及保存音频文件的场景""" - if opus_handler is None: - opus_handler = self.handle_opus - try: - # 创建新的事件循环 - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - # 生成会话ID - session_id = uuid.uuid4().hex - # 存储音频数据 - audio_data = [] - - async def _generate_audio(): - # 刷新Token（如果需要） - if self._is_token_expired(): - self._refresh_token() - - # 建立WebSocket连接 - ws = await websockets.connect( - self.ws_url, - additional_headers={"X-NLS-Token": self.token}, - ping_interval=30, - ping_timeout=10, - close_timeout=10, - ) - try: - # 发送StartSynthesis请求 - start_message_id = str(uuid.uuid4().hex) - start_request = { - "header": { - "message_id": start_message_id, - "task_id": session_id, - "namespace": "FlowingSpeechSynthesizer", - "name": "StartSynthesis", - "appkey": self.appkey, - }, - "payload": { - "voice": self.voice, - "format": self.format, - "sample_rate": self.sample_rate, - "volume": self.volume, - "speech_rate": self.speech_rate, - "pitch_rate": self.pitch_rate, - "enable_subtitle": True, - }, - } - await ws.send(json.dumps(start_request)) - - # 等待SynthesisStarted响应 - synthesis_started = False - while not synthesis_started: - msg = await ws.recv() - if isinstance(msg, str): - data = json.loads(msg) - header = data.get("header", {}) - if header.get("name") == "SynthesisStarted": - synthesis_started = True - logger.bind(tag=TAG).debug("TTS合成已启动") - elif header.get("name") == "TaskFailed": - error_info = data.get("payload", {}).get( - "error_info", {} - ) - error_code = error_info.get("error_code") - error_message = error_info.get( - "error_message", "未知错误" - ) - raise Exception( - f"启动合成失败: {error_code} - {error_message}" - ) - - # 发送文本合成请求 - filtered_text = MarkdownCleaner.clean_markdown(text) - run_message_id = str(uuid.uuid4().hex) - run_request = { - "header": { - "message_id": run_message_id, - "task_id": session_id, - "namespace": "FlowingSpeechSynthesizer", - "name": "RunSynthesis", - "appkey": self.appkey, - }, - "payload": {"text": filtered_text}, - } - await ws.send(json.dumps(run_request)) - - # 发送停止合成请求 - stop_message_id = str(uuid.uuid4().hex) - stop_request = { - "header": { - "message_id": stop_message_id, - "task_id": session_id, - "namespace": "FlowingSpeechSynthesizer", - "name": "StopSynthesis", - "appkey": self.appkey, - } - } - await ws.send(json.dumps(stop_request)) - - # 接收音频数据 - synthesis_completed = False - while not synthesis_completed: - msg = await ws.recv() - if isinstance(msg, (bytes, bytearray)): - # 编码为Opus并收集 - self.opus_encoder.encode_pcm_to_opus_stream( - msg, False, opus_handler - ) - # audio_data.extend(opus_frames) - elif isinstance(msg, str): - data = json.loads(msg) - header = data.get("header", {}) - event_name = header.get("name") - if event_name == "SynthesisCompleted": - synthesis_completed = True - logger.bind(tag=TAG).debug("TTS合成完成") - elif event_name == "TaskFailed": - error_info = data.get("payload", {}).get( - "error_info", {} - ) - error_code = error_info.get("error_code") - error_message = error_info.get( - "error_message", "未知错误" - ) - raise Exception( - f"合成失败: {error_code} - {error_message}" - ) - finally: - try: - await ws.close() - except: - pass - - loop.run_until_complete(_generate_audio()) - loop.close() - - except Exception as e: - logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index d55f434e..9e16df97 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -5,14 +5,14 @@ import uuid import asyncio import threading from typing import Callable, Any - from core.utils import p3 import time +from datetime import datetime from core.utils import textUtils from abc import ABC, abstractmethod from config.logger import setup_logging -from core.utils.audio_flow_control import FlowControlConfig, simulate_device_consumption -from core.utils.util import audio_to_data, audio_bytes_to_data, audio_bytes_to_data_stream, audio_to_data_stream +from core.utils.audio_flow_control import FlowControlConfig +from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream from core.utils.tts import MarkdownCleaner from core.utils.output_counter import add_device_output from core.handle.reportHandle import enqueue_tts_report @@ -72,7 +72,6 @@ class TTSProviderBase(ABC): self.processed_chars = 0 self.is_first_sentence = True self.flow_controller = FlowControlConfig.create_flow_controller() - self.flow_control_enabled = config.get("enable_flow_control", True) def generate_filename(self, extension=".wav"): return os.path.join( @@ -93,7 +92,7 @@ class TTSProviderBase(ABC): (file_audio, text) ) - def to_tts_stream(self, text, opus_handler=handle_opus) -> None: + def to_tts_stream(self, text, opus_handler: Callable[[bytes], None] = None) -> None: text = MarkdownCleaner.clean_markdown(text) max_repeat_time = 5 if self.delete_audio_file: @@ -108,7 +107,7 @@ class TTSProviderBase(ABC): audio_bytes_to_data_stream( audio_bytes, file_type=self.audio_file_type, is_opus=True, callback=opus_handler ) - max_repeat_time = 0 + break else: max_repeat_time -= 1 except Exception as e: @@ -160,11 +159,11 @@ class TTSProviderBase(ABC): async def text_to_speak(self, text, output_file): pass - def audio_to_pcm_data_stream(self, audio_file_path, callback: Callable[[Any], Any]=None): + def audio_to_pcm_data_stream(self, audio_file_path, callback: Callable[[Any], Any] = None): """音频文件转换为PCM编码""" return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback) - def audio_to_opus_data_stream(self, audio_file_path, callback: Callable[[Any], Any]=None): + def audio_to_opus_data_stream(self, audio_file_path, callback: Callable[[Any], Any] = None): """音频文件转换为Opus编码""" return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback) @@ -229,6 +228,7 @@ class TTSProviderBase(ABC): self.tts_text_buff = [] self.is_first_sentence = True self.tts_audio_first_sentence = True + self.reset_flow_controller() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() @@ -254,6 +254,9 @@ class TTSProviderBase(ABC): continue def _audio_play_priority_thread(self): + # 需要上报的文本和音频列表 + enqueue_text = None + enqueue_audio = None while not self.conn.stop_event.is_set(): text = None try: @@ -264,108 +267,91 @@ class TTSProviderBase(ABC): break continue - # 如果启用了流控 - if self.flow_control_enabled: - # 计算音频数据的帧数 - if isinstance(audio_datas, bytes): - frame_count = 1 # 单个字节流作为一帧 - elif isinstance(audio_datas, (list, tuple)): - frame_count = len(audio_datas) + if self.conn.client_abort: + logger.bind(tag=TAG).debug("收到打断信号，跳过当前音频数据") + # 打断时丢弃未上报的音频数据 + enqueue_text, enqueue_audio = None, [] + continue + + # 收到下一个文本开始或会话结束时进行上报 + if sentence_type is not SentenceType.MIDDLE: + # 上报TTS数据 + if enqueue_text is not None and enqueue_audio is not None: + enqueue_tts_report(self.conn, enqueue_text, enqueue_audio) + enqueue_audio = [] + enqueue_text = text + + # 计算音频数据的帧数 + if isinstance(audio_datas, bytes): + frame_count = 1 # 单个字节流作为一帧 + enqueue_audio.append(audio_datas) + else: + frame_count = 0 + + # 记录输出和报告 + if self.conn.max_output_size > 0 and text: + add_device_output(self.conn.headers.get("device-id"), len(text)) + + # 流控检查 + if frame_count > 0: + max_wait_time = FlowControlConfig.DEFAULT_MAX_WAIT_TIME + wait_start_time = time.time() + retry_interval = FlowControlConfig.DEFAULT_RETRY_INTERVAL + + while not self.flow_controller.can_send_frames(frame_count): + # 检查是否超时或需要停止 + if (time.time() - wait_start_time > max_wait_time or + self.conn.stop_event.is_set() or + self.conn.client_abort): + logger.bind(tag=TAG).debug("流控等待超时或收到停止信号，跳过音频发送") + break + # 短暂等待后重试 + time.sleep(retry_interval) else: - frame_count = 0 + # 可以发送，记录发送的帧数 + self.flow_controller.record_sent_frames(frame_count) - # 流控检查 - if frame_count > 0: - max_wait_time = FlowControlConfig.DEFAULT_MAX_WAIT_TIME - wait_start_time = time.time() - retry_interval = FlowControlConfig.DEFAULT_RETRY_INTERVAL - - while not self.flow_controller.can_send_frames(frame_count): - # 检查是否超时或需要停止 - if (time.time() - wait_start_time > max_wait_time or - self.conn.stop_event.is_set() or - self.conn.client_abort): - logger.bind(tag=TAG).warning( - f"流控等待超时或收到停止信号，跳过音频发送: {text}" - ) - break - - # 短暂等待后重试 - time.sleep(retry_interval) - - # status = self.flow_controller.get_status() - # logger.bind(tag=TAG).debug( - # f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " - # f"可用令牌={status['available_tokens']}..." - # f"发送帧数={status['sent_frames']}..." - # f"消费帧数={status['consumed_frames']}..." - # f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..." - # ) - else: - # 可以发送，记录发送的帧数 - self.flow_controller.record_sent_frames(frame_count) - - # 发送音频 - future = asyncio.run_coroutine_threadsafe( - self._send_audio_with_flow_control(sentence_type, audio_datas, text), - self.conn.loop, - ) - future.result() - - # 记录输出和报告 - if self.conn.max_output_size > 0 and text: - add_device_output(self.conn.headers.get("device-id"), len(text)) - enqueue_tts_report(self.conn, text, audio_datas) - - # 输出流控状态（调试用） - # if frame_count > 0: # 只在较大的音频块时输出状态 - # status = self.flow_controller.get_status() - # logger.bind(tag=TAG).debug( - # f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " - # f"可用令牌={status['available_tokens']}..." - # f"发送帧数={status['sent_frames']}..." - # f"消费帧数={status['consumed_frames']}..." - # f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..." - # ) - else: - # 没有音频数据，直接发送 + # 发送音频 future = asyncio.run_coroutine_threadsafe( self._send_audio_with_flow_control(sentence_type, audio_datas, text), self.conn.loop, ) future.result() + + # 输出流控状态（调试用） + # status = self.flow_controller.get_status() + # logger.bind(tag=TAG).debug( + # f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, " + # f"可用令牌={status['available_tokens']}..." + # f"发送帧数={status['sent_frames']}..." + # f"消费帧数={status['consumed_frames']}..." + # f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..." + # ) else: - # 未启用流控，直接发送 + # 没有音频数据，直接发送 future = asyncio.run_coroutine_threadsafe( - sendAudioMessage(self.conn, sentence_type, audio_datas, text), + self._send_audio_with_flow_control(sentence_type, audio_datas, text), self.conn.loop, ) future.result() - # 记录输出和报告 - if self.conn.max_output_size > 0 and text: - add_device_output(self.conn.headers.get("device-id"), len(text)) - enqueue_tts_report(self.conn, text, audio_datas) - except Exception as e: logger.bind(tag=TAG).error( f"audio_play_priority_thread: {text} {e}" ) async def _send_audio_with_flow_control(self, sentence_type, audio_datas, text): - """带流控的音频发送方法""" + """ + 带流控的音频发送方法 模拟设备消费音频帧的过程 + 实际应用中应该根据设备反馈来更新消费情况 + """ await sendAudioMessage(self.conn, sentence_type, audio_datas, text) - # 模拟设备消费（实际应用中应该从设备获取反馈） + # 模拟设备消费（实际应用中应该从设备获取反馈）防止音字不同步 if isinstance(audio_datas, bytes): - frame_count = 1 - elif isinstance(audio_datas, (list, tuple)): - frame_count = len(audio_datas) - else: - frame_count = 0 - - if frame_count > 0: - asyncio.create_task(simulate_device_consumption(self.flow_controller, frame_count)) + # 模拟设备播放延迟（60ms per frame）, 实际情况可以低一点（50ms），增加使用体验 + await asyncio.sleep(0.055) + self.flow_controller.update_device_consumption(1) # 在类中添加流控制器重置方法 def reset_flow_controller(self): @@ -452,7 +438,7 @@ class TTSProviderBase(ABC): self.before_stop_play_files.clear() self.tts_audio_queue.put((SentenceType.LAST, [], None)) - def _process_remaining_text_stream(self, opus_handler=handle_opus): + def _process_remaining_text_stream(self, opus_handler: Callable[[bytes], None] = None): """处理剩余的文本并生成语音 Returns: diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index a971215f..ac6301f9 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -5,7 +5,6 @@ import queue import asyncio import traceback from typing import Callable, Any - import websockets from core.utils.tts import MarkdownCleaner from config.logger import setup_logging @@ -214,6 +213,7 @@ class TTSProvider(TTSProviderBase): if message.sentence_type == SentenceType.FIRST: self.conn.client_abort = False + self.reset_flow_controller() if self.conn.client_abort: try: @@ -426,9 +426,6 @@ class TTSProvider(TTSProviderBase): async def _start_monitor_tts_response(self): """监听TTS响应""" - opus_datas_cache = [] - is_first_sentence = True - first_sentence_segment_count = 0 # 添加计数器 try: session_finished = False # 标记会话是否正常结束 while not self.conn.stop_event.is_set(): @@ -449,8 +446,6 @@ class TTSProvider(TTSProviderBase): self.tts_audio_queue.put( (SentenceType.FIRST, [], self.tts_text) ) - opus_datas_cache = [] - first_sentence_segment_count = 0 # 重置计数器 elif ( res.optional.event == EVENT_TTSResponse and res.header.message_type == AUDIO_ONLY_RESPONSE @@ -459,13 +454,6 @@ class TTSProvider(TTSProviderBase): self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus) elif res.optional.event == EVENT_TTSSentenceEnd: logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}") - if not is_first_sentence or first_sentence_segment_count > 10: - # 发送缓存的数据 - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_datas_cache, None) - ) - # 第一句话结束后，将标志设置为False - is_first_sentence = False elif res.optional.event == EVENT_SessionFinished: logger.bind(tag=TAG).debug(f"会话结束～～") self._process_before_stop_play_files() @@ -641,104 +629,3 @@ class TTSProvider(TTSProviderBase): def wav_to_opus_data_audio_raw_stream(self, raw_data_var, is_end=False, callback: Callable[[Any], Any]=None): return self.opus_encoder.encode_pcm_to_opus_stream(raw_data_var, is_end, callback=callback) - - def to_tts_stream(self, text: str, opus_handler=None) -> None: - """非流式生成音频数据，用于生成音频及测试场景 - - Args: - text: 要转换的文本 - opus_handler: opus数据处理方法 - - Returns: - list: 音频数据列表 - """ - if opus_handler is None: - opus_handler = self.handle_opus - try: - # 创建事件循环 - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - - # 生成会话ID - session_id = uuid.uuid4().__str__().replace("-", "") - - async def _generate_audio(): - # 创建新的WebSocket连接 - ws_header = { - "X-Api-App-Key": self.appId, - "X-Api-Access-Key": self.access_token, - "X-Api-Resource-Id": self.resource_id, - "X-Api-Connect-Id": uuid.uuid4(), - } - ws = await websockets.connect( - self.ws_url, additional_headers=ws_header, max_size=1000000000 - ) - - try: - # 启动会话 - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional( - event=EVENT_StartSession, sessionId=session_id - ).as_bytes() - payload = self.get_payload_bytes( - event=EVENT_StartSession, speaker=self.voice - ) - await self.send_event(ws, header, optional, payload) - - # 发送文本 - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional( - event=EVENT_TaskRequest, sessionId=session_id - ).as_bytes() - payload = self.get_payload_bytes( - event=EVENT_TaskRequest, text=text, speaker=self.voice - ) - await self.send_event(ws, header, optional, payload) - - # 发送结束会话请求 - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional( - event=EVENT_FinishSession, sessionId=session_id - ).as_bytes() - payload = str.encode("{}") - await self.send_event(ws, header, optional, payload) - - # 接收音频数据 - while True: - msg = await ws.recv() - res = self.parser_response(msg) - - if ( - res.optional.event == EVENT_TTSResponse - and res.header.message_type == AUDIO_ONLY_RESPONSE - ): - self.wav_to_opus_data_audio_raw_stream(res.payload, callback=opus_handler) - elif res.optional.event == EVENT_SessionFinished: - break - - finally: - # 清理资源 - try: - await ws.close() - except: - pass - - # 运行异步任务 - loop.run_until_complete(_generate_audio()) - loop.close() - - except Exception as e: - logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") - return [] diff --git a/main/xiaozhi-server/core/providers/tts/index_stream.py b/main/xiaozhi-server/core/providers/tts/index_stream.py index 5ea9eb36..d562e802 100644 --- a/main/xiaozhi-server/core/providers/tts/index_stream.py +++ b/main/xiaozhi-server/core/providers/tts/index_stream.py @@ -3,8 +3,6 @@ import queue import asyncio import traceback import aiohttp -import requests -import time from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -27,15 +25,13 @@ class TTSProvider(TTSProviderBase): self.api_url = config.get("api_url", "http://8.138.114.124:11996/tts") self.audio_format = "pcm" self.before_stop_play_files = [] - self.segment_count = 0 # 创建Opus编码器 需注意接口返回的采样率为24000 self.opus_encoder = opus_encoder_utils.OpusEncoderUtils( sample_rate=24000, channels=1, frame_size_ms=60 ) - # 文本缓冲区和PCM缓冲区 - self.text_buffer = "" + # PCM缓冲区 self.pcm_buffer = bytearray() def tts_text_priority_thread(self): @@ -48,8 +44,8 @@ class TTSProvider(TTSProviderBase): self.tts_stop_request = False self.processed_chars = 0 self.tts_text_buff = [] - self.segment_count = 0 self.before_stop_play_files.clear() + self.reset_flow_controller() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() @@ -62,14 +58,11 @@ class TTSProvider(TTSProviderBase): ) if message.content_file and os.path.exists(message.content_file): # 先处理文件音频数据 - file_audio = self._process_audio_file(message.content_file) - self.before_stop_play_files.append( - (file_audio, message.content_detail) - ) + self._process_audio_file_stream(message.content_file, callback=lambda audio_data: self.handle_audio_file(audio_data, message.content_detail)) if message.sentence_type == SentenceType.LAST: # 处理剩余的文本 - self._process_remaining_text(True) + self._process_remaining_text_stream(True) except queue.Empty: continue @@ -78,7 +71,7 @@ class TTSProvider(TTSProviderBase): f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" ) - def _process_remaining_text(self, is_last=False): + def _process_remaining_text_stream(self, is_last=False): """处理剩余的文本并生成语音 Returns: bool: 是否成功处理了文本 @@ -143,8 +136,6 @@ class TTSProvider(TTSProviderBase): return self.pcm_buffer.clear() - opus_datas_cache = [] - self.tts_audio_queue.put((SentenceType.FIRST, [], text)) # 处理音频流数据 @@ -158,41 +149,22 @@ class TTSProvider(TTSProviderBase): while len(self.pcm_buffer) >= frame_bytes: frame = bytes(self.pcm_buffer[:frame_bytes]) del self.pcm_buffer[:frame_bytes] - opus = self.opus_encoder.encode_pcm_to_opus( - frame, end_of_stream=False + + self.opus_encoder.encode_pcm_to_opus_stream( + frame, + end_of_stream=False, + callback=self.handle_opus ) - if opus: - if self.segment_count < 10: # 前10个片段直接发送 - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus, None) - ) - self.segment_count += 1 - else: - opus_datas_cache.extend(opus) # flush 剩余不足一帧的数据 if self.pcm_buffer: - opus = self.opus_encoder.encode_pcm_to_opus( - bytes(self.pcm_buffer), end_of_stream=True + self.opus_encoder.encode_pcm_to_opus_stream( + bytes(self.pcm_buffer), + end_of_stream=True, + callback=self.handle_opus ) - if opus: - if self.segment_count < 10: # 前10个片段直接发送 - # 直接发送 - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus, None) - ) - self.segment_count += 1 - else: - # 后续片段缓存 - opus_datas_cache.extend(opus) self.pcm_buffer.clear() - # 如果不是前10个片段，发送缓存的数据 - if self.segment_count >= 10 and opus_datas_cache: - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_datas_cache, None) - ) - # 如果是最后一段，输出音频获取完毕 if is_last: self._process_before_stop_play_files() @@ -206,59 +178,3 @@ class TTSProvider(TTSProviderBase): await super().close() if hasattr(self, "opus_encoder"): self.opus_encoder.close() - - def to_tts(self, text: str) -> list: - """非流式TTS处理，用于测试及保存音频文件的场景 - - Args: - text: 要转换的文本 - - Returns: - list: 返回opus编码后的音频数据列表 - """ - start_time = time.time() - text = MarkdownCleaner.clean_markdown(text) - - payload = {"text": text, "character": self.character} - - try: - with requests.post(self.api_url, json=payload, timeout=5) as response: - if response.status_code != 200: - logger.bind(tag=TAG).error( - f"TTS请求失败: {response.status_code}, {response.text}" - ) - return [] - - logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒") - - # 使用opus编码器处理PCM数据 - opus_datas = [] - pcm_data = response.content - - # 计算每帧的字节数 - frame_bytes = int( - self.opus_encoder.sample_rate - * self.opus_encoder.channels - * self.opus_encoder.frame_size_ms - / 1000 - * 2 - ) - - # 分帧处理PCM数据 - for i in range(0, len(pcm_data), frame_bytes): - frame = pcm_data[i : i + frame_bytes] - if len(frame) < frame_bytes: - # 最后一帧可能不足，用0填充 - frame = frame + b"\x00" * (frame_bytes - len(frame)) - - opus = self.opus_encoder.encode_pcm_to_opus( - frame, end_of_stream=(i + frame_bytes >= len(pcm_data)) - ) - if opus: - opus_datas.extend(opus) - - return opus_datas - - except Exception as e: - logger.bind(tag=TAG).error(f"TTS请求异常: {e}") - return [] diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index 09d3d82a..a855e8ed 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -3,8 +3,6 @@ import queue import asyncio import traceback import aiohttp -import requests -import time from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -24,23 +22,15 @@ class TTSProvider(TTSProviderBase): self.api_url = config.get("api_url") self.audio_format = "pcm" self.before_stop_play_files = [] - self.segment_count = 0 # 添加片段计数器 # 创建Opus编码器 self.opus_encoder = opus_encoder_utils.OpusEncoderUtils( sample_rate=16000, channels=1, frame_size_ms=60 ) - # 添加文本缓冲区 - self.text_buffer = "" - # PCM缓冲区 self.pcm_buffer = bytearray() - ################################################################################### - # linkerai单流式TTS重写父类的方法--开始 - ################################################################################### - def tts_text_priority_thread(self): """流式文本处理线程""" while not self.conn.stop_event.is_set(): @@ -51,8 +41,8 @@ class TTSProvider(TTSProviderBase): self.tts_stop_request = False self.processed_chars = 0 self.tts_text_buff = [] - self.segment_count = 0 self.before_stop_play_files.clear() + self.reset_flow_controller() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() @@ -172,8 +162,6 @@ class TTSProvider(TTSProviderBase): return self.pcm_buffer.clear() - opus_datas_cache = [] - self.tts_audio_queue.put((SentenceType.FIRST, [], text)) # 兼容 iter_chunked / iter_chunks / iter_any @@ -190,41 +178,21 @@ class TTSProvider(TTSProviderBase): frame = bytes(self.pcm_buffer[:frame_bytes]) del self.pcm_buffer[:frame_bytes] - opus = self.opus_encoder.encode_pcm_to_opus( - frame, end_of_stream=False + self.opus_encoder.encode_pcm_to_opus_stream( + frame, + end_of_stream=False, + callback=self.handle_opus ) - if opus: - if self.segment_count < 10: # 前10个片段直接发送 - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus, None) - ) - self.segment_count += 1 - else: - opus_datas_cache.extend(opus) # flush 剩余不足一帧的数据 if self.pcm_buffer: - opus = self.opus_encoder.encode_pcm_to_opus( - bytes(self.pcm_buffer), end_of_stream=True + self.opus_encoder.encode_pcm_to_opus_stream( + bytes(self.pcm_buffer), + end_of_stream=True, + callback=self.handle_opus ) - if opus: - if self.segment_count < 10: # 前10个片段直接发送 - # 直接发送 - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus, None) - ) - self.segment_count += 1 - else: - # 后续片段缓存 - opus_datas_cache.extend(opus) self.pcm_buffer.clear() - # 如果不是前10个片段，发送缓存的数据 - if self.segment_count >= 10 and opus_datas_cache: - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_datas_cache, None) - ) - # 如果是最后一段，输出音频获取完毕 if is_last: self._process_before_stop_play_files() @@ -232,68 +200,3 @@ class TTSProvider(TTSProviderBase): except Exception as e: logger.bind(tag=TAG).error(f"TTS请求异常: {e}") self.tts_audio_queue.put((SentenceType.LAST, [], None)) - - def to_tts_stream(self, text: str, opus_handler=None) -> list: - """非流式TTS处理，用于测试及保存音频文件的场景 - - Args: - text: 要转换的文本 - opus_handler: opus数据处理方法 - """ - if opus_handler is None: - opus_handler = self.handle_opus - start_time = time.time() - text = MarkdownCleaner.clean_markdown(text) - - params = { - "tts_text": text, - "spk_id": self.voice, - "frame_duration": 60, - "stream": False, - "target_sr": 16000, - "audio_format": self.audio_format, - "instruct_text": "请生成一段自然流畅的语音", - } - headers = { - "Authorization": f"Bearer {self.access_token}", - "Content-Type": "application/json", - } - - try: - with requests.get( - self.api_url, params=params, headers=headers, timeout=5 - ) as response: - if response.status_code != 200: - logger.bind(tag=TAG).error( - f"TTS请求失败: {response.status_code}, {response.text}" - ) - return [] - - logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒") - - # 使用opus编码器处理PCM数据 - opus_datas = [] - pcm_data = response.content - - # 计算每帧的字节数 - frame_bytes = int( - self.opus_encoder.sample_rate - * self.opus_encoder.channels - * self.opus_encoder.frame_size_ms - / 1000 - * 2 - ) - - # 分帧处理PCM数据 - for i in range(0, len(pcm_data), frame_bytes): - frame = pcm_data[i : i + frame_bytes] - if len(frame) < frame_bytes: - # 最后一帧可能不足，用0填充 - frame = frame + b"\x00" * (frame_bytes - len(frame)) - - self.opus_encoder.encode_pcm_to_opus_stream( - frame, end_of_stream=(i + frame_bytes >= len(pcm_data)), callback=opus_handler - ) - - except Exception as e: - logger.bind(tag=TAG).error(f"TTS请求异常: {e}") diff --git a/main/xiaozhi-server/core/utils/audio_flow_control.py b/main/xiaozhi-server/core/utils/audio_flow_control.py index 102914c5..98c29676 100644 --- a/main/xiaozhi-server/core/utils/audio_flow_control.py +++ b/main/xiaozhi-server/core/utils/audio_flow_control.py @@ -151,20 +151,6 @@ class AudioFlowController: ) -async def simulate_device_consumption(flow_controller: AudioFlowController, frame_count: int): - """ - 模拟设备消费音频帧的过程 - 实际应用中应该根据设备反馈来更新消费情况 - - Args: - flow_controller: 流控制器实例 - frame_count: 消费的帧数 - """ - # 模拟设备播放延迟（60ms per frame） - await asyncio.sleep(frame_count * 0.06) - flow_controller.update_device_consumption(frame_count) - - # 流控配置常量 class FlowControlConfig: """流控配置常量""" diff --git a/main/xiaozhi-server/core/utils/opus_encoder_utils.py b/main/xiaozhi-server/core/utils/opus_encoder_utils.py index f9154da1..8ca406d3 100644 --- a/main/xiaozhi-server/core/utils/opus_encoder_utils.py +++ b/main/xiaozhi-server/core/utils/opus_encoder_utils.py @@ -5,9 +5,8 @@ Opus编码工具类 import logging import traceback - import numpy as np -from typing import List, Optional, Callable, Any +from typing import Optional, Callable, Any from opuslib_next import Encoder from opuslib_next import constants @@ -56,53 +55,6 @@ class OpusEncoderUtils: self.encoder.reset_state() self.buffer = np.array([], dtype=np.int16) - def encode_pcm_to_opus(self, pcm_data: bytes, end_of_stream: bool) -> List[bytes]: - """ - 将PCM数据编码为Opus格式 - - Args: - pcm_data: PCM字节数据 - end_of_stream: 是否为流的结束 - - Returns: - Opus数据包列表 - """ - # 将字节数据转换为short数组 - new_samples = self._convert_bytes_to_shorts(pcm_data) - - # 校验PCM数据 - self._validate_pcm_data(new_samples) - - # 将新数据追加到缓冲区 - self.buffer = np.append(self.buffer, new_samples) - - opus_packets = [] - offset = 0 - - # 处理所有完整帧 - while offset <= len(self.buffer) - self.total_frame_size: - frame = self.buffer[offset : offset + self.total_frame_size] - output = self._encode(frame) - if output: - opus_packets.append(output) - offset += self.total_frame_size - - # 保留未处理的样本 - self.buffer = self.buffer[offset:] - - # 流结束时处理剩余数据 - if end_of_stream and len(self.buffer) > 0: - # 创建最后一帧并用0填充 - last_frame = np.zeros(self.total_frame_size, dtype=np.int16) - last_frame[: len(self.buffer)] = self.buffer - - output = self._encode(last_frame) - if output: - opus_packets.append(output) - self.buffer = np.array([], dtype=np.int16) - - return opus_packets - def encode_pcm_to_opus_stream(self, pcm_data: bytes, end_of_stream: bool, callback: Callable[[Any], Any]): """ 将PCM数据编码为Opus格式，以流式方式进行处理 diff --git a/main/xiaozhi-server/core/utils/p3.py b/main/xiaozhi-server/core/utils/p3.py index f2ee3239..415e5366 100644 --- a/main/xiaozhi-server/core/utils/p3.py +++ b/main/xiaozhi-server/core/utils/p3.py @@ -1,30 +1,7 @@ +import io import struct from typing import Callable, Any -def decode_opus_from_file(input_file): - """ - 从p3文件中解码 Opus 数据，并返回一个 Opus 数据包的列表。 - """ - opus_datas = [] - - with open(input_file, 'rb') as f: - while True: - # 读取头部（4字节）：[1字节类型，1字节保留，2字节长度] - header = f.read(4) - if not header: - break - - # 解包头部信息 - _, _, data_len = struct.unpack('>BBH', header) - - # 根据头部指定的长度读取 Opus 数据 - opus_data = f.read(data_len) - if len(opus_data) != data_len: - raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the file.") - - opus_datas.append(opus_data) - - return opus_datas def decode_opus_from_file_stream(input_file, callback: Callable[[Any], Any]): """ @@ -47,25 +24,6 @@ def decode_opus_from_file_stream(input_file, callback: Callable[[Any], Any]): callback(opus_data) -def decode_opus_from_bytes(input_bytes): - """ - 从p3二进制数据中解码 Opus 数据，并返回一个 Opus 数据包的列表。 - """ - import io - opus_datas = [] - - f = io.BytesIO(input_bytes) - while True: - header = f.read(4) - if not header: - break - _, _, data_len = struct.unpack('>BBH', header) - opus_data = f.read(data_len) - if len(opus_data) != data_len: - raise ValueError(f"Data length({len(opus_data)}) mismatch({data_len}) in the bytes.") - opus_datas.append(opus_data) - - return opus_datas def decode_opus_from_bytes_stream(input_bytes, callback: Callable[[Any], Any]): """ diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 8aa2d712..1eec7327 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -3,10 +3,8 @@ import socket import subprocess import re import os -import wave from io import BytesIO from typing import Callable, Any - from core.utils import p3 import numpy as np import requests @@ -213,23 +211,6 @@ def extract_json_from_string(input_string): return None -def audio_to_data(audio_file_path, is_opus=True): - # 获取文件后缀名 - file_type = os.path.splitext(audio_file_path)[1] - if file_type: - file_type = file_type.lstrip(".") - # 读取音频文件，-nostdin 参数：不要从标准输入读取数据，否则FFmpeg会阻塞 - audio = AudioSegment.from_file( - audio_file_path, format=file_type, parameters=["-nostdin"] - ) - - # 转换为单声道/16kHz采样率/16位小端编码（确保与编码器匹配） - audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2) - - # 获取原始PCM数据（16位小端） - raw_data = audio.raw_data - return pcm_to_data(raw_data, is_opus) - def audio_to_data_stream(audio_file_path, is_opus=True, callback: Callable[[Any], Any]=None) -> None: # 获取文件后缀名 file_type = os.path.splitext(audio_file_path)[1] @@ -245,25 +226,9 @@ def audio_to_data_stream(audio_file_path, is_opus=True, callback: Callable[[Any] # 获取原始PCM数据（16位小端） raw_data = audio.raw_data - pcm_to_data_stream(raw_data, is_opus,callback) + pcm_to_data_stream(raw_data, is_opus, callback) -def audio_bytes_to_data(audio_bytes, file_type, is_opus=True): - """ - 直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3 - """ - if file_type == "p3": - # 直接用p3解码 - return p3.decode_opus_from_bytes(audio_bytes) - else: - # 其他格式用pydub - audio = AudioSegment.from_file( - BytesIO(audio_bytes), format=file_type, parameters=["-nostdin"] - ) - audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2) - raw_data = audio.raw_data - return pcm_to_data(raw_data, is_opus) - def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]) -> None: """ 直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3 @@ -281,37 +246,7 @@ def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callab pcm_to_data_stream(raw_data, is_opus, callback) -def pcm_to_data(raw_data, is_opus=True): - # 初始化Opus编码器 - encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO) - - # 编码参数 - frame_duration = 60 # 60ms per frame - frame_size = int(16000 * frame_duration / 1000) # 960 samples/frame - - datas = [] - # 按帧处理所有音频数据（包括最后一帧可能补零） - for i in range(0, len(raw_data), frame_size * 2): # 16bit=2bytes/sample - # 获取当前帧的二进制数据 - chunk = raw_data[i : i + frame_size * 2] - - # 如果最后一帧不足，补零 - if len(chunk) < frame_size * 2: - chunk += b"\x00" * (frame_size * 2 - len(chunk)) - - if is_opus: - # 转换为numpy数组处理 - np_frame = np.frombuffer(chunk, dtype=np.int16) - # 编码Opus数据 - frame_data = encoder.encode(np_frame.tobytes(), frame_size) - else: - frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk) - - datas.append(frame_data) - - return datas - -def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any]=None): +def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] = None): # 初始化Opus编码器 encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO) @@ -338,32 +273,6 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any]=No frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk) callback(frame_data) -def opus_datas_to_wav_bytes(opus_datas, sample_rate=16000, channels=1): - """ - 将opus帧列表解码为wav字节流 - """ - decoder = opuslib_next.Decoder(sample_rate, channels) - pcm_datas = [] - - frame_duration = 60 # ms - frame_size = int(sample_rate * frame_duration / 1000) # 960 - - for opus_frame in opus_datas: - # 解码为PCM（返回bytes，2字节/采样点） - pcm = decoder.decode(opus_frame, frame_size) - pcm_datas.append(pcm) - - pcm_bytes = b"".join(pcm_datas) - - # 写入wav字节流 - wav_buffer = BytesIO() - with wave.open(wav_buffer, "wb") as wf: - wf.setnchannels(channels) - wf.setsampwidth(2) # 16bit - wf.setframerate(sample_rate) - wf.writeframes(pcm_bytes) - return wav_buffer.getvalue() - def check_vad_update(before_config, new_config): if ( diff --git a/main/xiaozhi-server/core/utils/wakeup_word.py b/main/xiaozhi-server/core/utils/wakeup_word.py deleted file mode 100644 index ab58f290..00000000 --- a/main/xiaozhi-server/core/utils/wakeup_word.py +++ /dev/null @@ -1,140 +0,0 @@ -import os -import re -import yaml -import time -import hashlib -import portalocker -from typing import Dict - - -class FileLock: - def __init__(self, file, timeout=5): - self.file = file - self.timeout = timeout - self.start_time = None - - def __enter__(self): - self.start_time = time.time() - while True: - try: - portalocker.lock(self.file, portalocker.LOCK_EX | portalocker.LOCK_NB) - return self.file - except portalocker.LockException: - if time.time() - self.start_time > self.timeout: - raise TimeoutError("获取文件锁超时") - time.sleep(0.1) - - def __exit__(self, exc_type, exc_val, exc_tb): - portalocker.unlock(self.file) - - -class WakeupWordsConfig: - def __init__(self): - self.config_file = "data/.wakeup_words.yaml" - self.assets_dir = "config/assets/wakeup_words" - self._ensure_directories() - self._config_cache = None - self._last_load_time = 0 - self._cache_ttl = 1 # 缓存有效期（秒） - self._lock_timeout = 5 # 文件锁超时时间（秒） - - def _ensure_directories(self): - """确保必要的目录存在""" - os.makedirs(os.path.dirname(self.config_file), exist_ok=True) - os.makedirs(self.assets_dir, exist_ok=True) - - def _load_config(self) -> Dict: - """加载配置文件，使用缓存机制""" - current_time = time.time() - - # 如果缓存有效，直接返回缓存 - if ( - self._config_cache is not None - and current_time - self._last_load_time < self._cache_ttl - ): - return self._config_cache - - try: - with open(self.config_file, "a+") as f: - with FileLock(f, timeout=self._lock_timeout): - f.seek(0) - content = f.read() - config = yaml.safe_load(content) if content else {} - self._config_cache = config - self._last_load_time = current_time - return config - except (TimeoutError, IOError) as e: - print(f"加载配置文件失败: {e}") - return {} - except Exception as e: - print(f"加载配置文件时发生未知错误: {e}") - return {} - - def _save_config(self, config: Dict): - """保存配置到文件，使用文件锁保护""" - try: - with open(self.config_file, "w") as f: - with FileLock(f, timeout=self._lock_timeout): - yaml.dump(config, f, allow_unicode=True) - self._config_cache = config - self._last_load_time = time.time() - except (TimeoutError, IOError) as e: - print(f"保存配置文件失败: {e}") - raise - except Exception as e: - print(f"保存配置文件时发生未知错误: {e}") - raise - - def get_wakeup_response(self, voice: str) -> Dict: - voice = hashlib.md5(voice.encode()).hexdigest() - """获取唤醒词回复配置""" - config = self._load_config() - - if not config or voice not in config: - return None - - # 检查文件大小 - file_path = config[voice]["file_path"] - if not os.path.exists(file_path) or os.stat(file_path).st_size < (15 * 1024): - return None - - return config[voice] - - def update_wakeup_response(self, voice: str, file_path: str, text: str): - """更新唤醒词回复配置""" - try: - # 过滤表情符号 - filtered_text = re.sub(r'[\U0001F600-\U0001F64F\U0001F900-\U0001F9FF]', '', text) - - config = self._load_config() - voice_hash = hashlib.md5(voice.encode()).hexdigest() - config[voice_hash] = { - "voice": voice, - "file_path": file_path, - "time": time.time(), - "text": filtered_text, - } - self._save_config(config) - except Exception as e: - print(f"更新唤醒词回复配置失败: {e}") - raise - - def generate_file_path(self, voice: str) -> str: - """生成音频文件路径，使用voice的哈希值作为文件名""" - try: - # 生成voice的哈希值 - voice_hash = hashlib.md5(voice.encode()).hexdigest() - file_path = os.path.join(self.assets_dir, f"{voice_hash}.wav") - - # 如果文件已存在，先删除 - if os.path.exists(file_path): - try: - os.remove(file_path) - except Exception as e: - print(f"删除已存在的音频文件失败: {e}") - raise - - return file_path - except Exception as e: - print(f"生成音频文件路径失败: {e}") - raise