diff --git a/main/xiaozhi-server/config.yaml b/main/xiaozhi-server/config.yaml index ab662798..4812cd2a 100644 --- a/main/xiaozhi-server/config.yaml +++ b/main/xiaozhi-server/config.yaml @@ -61,8 +61,6 @@ delete_audio: true close_connection_no_voice_time: 120 # TTS请求超时时间(秒) tts_timeout: 10 -# 开启唤醒词加速 -enable_wakeup_words_response_cache: true # 开场是否回复唤醒词 enable_greeting: true # 说完话是否开启提示音 diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index 9374303d..8147db1a 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -1,16 +1,5 @@ -import time import json -import random import asyncio -from core.utils.dialogue import Message -from core.providers.tts.dto.dto import SentenceType -from core.utils.wakeup_word import WakeupWordsConfig -from core.handle.sendAudioHandle import sendAudioMessage, send_stt_message -from core.utils.util import ( - audio_to_data_stream, - remove_punctuation_and_length, - opus_datas_to_wav_bytes -) from core.providers.tools.device_mcp import ( MCPClient, send_mcp_initialize_message, @@ -19,18 +8,6 @@ from core.providers.tools.device_mcp import ( 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消息""" audio_params = msg_json.get("audio_params") @@ -51,100 +28,4 @@ async def handleHelloMessage(conn, msg_json): # 发送mcp消息,获取tools列表 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": "哈啰啊,我是小智啦,声音好听的台湾女孩一枚,超开心认识你耶,最近在忙啥,别忘了给我来点有趣的料哦,我超爱听八卦的啦", - } - - # 获取音频数据 - opus_packets = [] - def handle_audio_frame(frame_data): - opus_packets.append(frame_data) - - audio_to_data_stream(response.get("file_path"), is_opus=True, callback=handle_audio_frame) - - # 播放唤醒词回复 - conn.client_abort = False - - 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() \ No newline at end of file + await conn.websocket.send(json.dumps(conn.welcome_msg)) \ No newline at end of file diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index 66102a97..2ddf03d6 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -1,12 +1,11 @@ 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 +import asyncio from core.utils.dialogue import Message +from core.providers.tts.dto.dto import ContentType from plugins_func.register import Action, ActionResponse +from core.handle.sendAudioHandle import send_stt_message +from core.utils.util import remove_punctuation_and_length from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType TAG = __name__ @@ -24,13 +23,9 @@ async def handle_user_intent(conn, text): pass # 检查是否有明确的退出命令 - filtered_text = remove_punctuation_and_length(text)[1] + _, filtered_text = remove_punctuation_and_length(text) 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 b0166fe3..15a9e434 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -1,26 +1,21 @@ import time import json -import asyncio -from core.utils.util import audio_to_data_stream 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.utils.util import play_audio_frames, play_audio_response from core.handle.sendAudioHandle import send_stt_message, SentenceType TAG = __name__ async def handleAudioMessage(conn, audio): + # 检查是否在唤醒处理锁定期内 + if getattr(conn, 'wakeup_processing_lock', 0) > time.monotonic(): + return + # 当前片段是否有人说话 have_voice = conn.vad.is_vad(conn, audio) - # 如果设备刚刚被唤醒,短暂忽略VAD检测 - if have_voice and hasattr(conn, "just_woken_up") and conn.just_woken_up: - have_voice = False - # 设置一个短暂延迟后恢复VAD检测 - conn.asr_audio.clear() - 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,13 +24,6 @@ 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 @@ -119,9 +107,7 @@ async def max_out_size(conn): text = "不好意思,我现在有点事情要忙,明天这个时候我们再聊,约好了哦!明天不见不散,拜拜!" await send_stt_message(conn, text) file_path = "config/assets/max_output_size.wav" - conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], text)) - play_audio_frames(conn, file_path) - conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) + play_audio_response(conn, {"text": text, "file_path": file_path}) conn.close_after_chat = True @@ -156,17 +142,4 @@ async def check_bind_device(conn): text = f"没有找到该设备的版本信息,请正确配置 OTA地址,然后重新编译固件。" await send_stt_message(conn, text) music_path = "config/assets/bind_not_found.wav" - 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 - ) + play_audio_response(conn, {"text": text, "file_path": music_path}) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index fb62bd32..c61b0d0d 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -30,7 +30,7 @@ async def sendAudioMessage(conn, sentenceType, audios, text): # 播放音频 -async def sendAudio(conn, audios, pre_buffer=False, frame_duration=60): +async def sendAudio(conn, audios, pre_buffer=False): """ 发送单个opus包,支持流控 Args: @@ -38,7 +38,7 @@ async def sendAudio(conn, audios, pre_buffer=False, frame_duration=60): opus_packet: 单个opus数据包 pre_buffer: 快速发送音频 """ - if audios is None: + if audios is None or len(audios) == 0: return if isinstance(audios, bytes): @@ -60,6 +60,7 @@ async def sendAudio(conn, audios, pre_buffer=False, frame_duration=60): "start_time": time.perf_counter(), } + frame_duration=60 flow_control = conn.audio_flow_control current_time = time.perf_counter() # 计算预期发送时间 @@ -76,38 +77,6 @@ async def sendAudio(conn, audios, pre_buffer=False, frame_duration=60): # 更新流控状态 flow_control["packet_count"] += 1 flow_control["last_send_time"] = time.perf_counter() - else: - if audios is None or len(audios) == 0: - return - # 流控参数优化 - frame_duration = 60 # 帧时长(毫秒),匹配 Opus 编码 - start_time = time.perf_counter() - play_position = 0 - - # 执行预缓冲 - 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:] - - # 播放剩余音频帧 - 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): diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index d5dd1338..c1e8f6be 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -1,13 +1,11 @@ import json -import time import asyncio +from core.utils.util import filter_sensitive_info from core.handle.abortHandle import handleAbortMessage from core.handle.helloHandle import handleHelloMessage -from core.handle.reportHandle import enqueue_asr_report +from core.handle.receiveAudioHandle import handleAudioMessage from core.providers.tools.device_mcp import handle_mcp_message -from core.handle.receiveAudioHandle import startToChat, handleAudioMessage from core.handle.sendAudioHandle import send_stt_message, send_tts_message -from core.utils.util import remove_punctuation_and_length, filter_sensitive_info from core.providers.tools.device_iot import handleIotDescriptors, handleIotStatus TAG = __name__ @@ -46,14 +44,9 @@ async def handleTextMessage(conn, message): conn.client_have_voice = False conn.asr_audio.clear() if "text" in msg_json: - conn.last_activity_time = time.time() * 1000 - original_text = msg_json["text"] # 保留原始文本 - filtered_len, filtered_text = remove_punctuation_and_length( - original_text - ) - + original_text = msg_json["text"] # 保留设备上传的文本 # 识别是否是唤醒词 - is_wakeup_words = filtered_text in conn.config.get("wakeup_words") + is_wakeup_words = original_text in conn.config.get("wakeup_words") # 是否开启唤醒词回复 enable_greeting = conn.config.get("enable_greeting", True) @@ -62,16 +55,10 @@ async def handleTextMessage(conn, message): 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: - # 上报纯文字数据(复用ASR上报功能,但不提供音频数据) - enqueue_asr_report(conn, original_text, []) - # 否则需要LLM对文字内容进行答复 - await startToChat(conn, original_text) + # 检测到唤醒词,开始等待后续进行声纹识别 + conn.wakeup_mode = True + conn.logger.bind(tag=TAG).info(f"检测到唤醒词~") elif msg_json["type"] == "iot": conn.logger.bind(tag=TAG).info(f"收到iot消息:{message}") if "descriptors" in msg_json: diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index f4803d31..7885261d 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -1,14 +1,14 @@ import os +import io import wave import uuid +import json +import time import queue import asyncio import traceback import threading import opuslib_next -import json -import io -import time import concurrent.futures from abc import ABC, abstractmethod from config.logger import setup_logging @@ -53,6 +53,10 @@ class ASRProviderBase(ABC): # 接收音频 async def receive_audio(self, conn, audio, audio_have_voice): + # 检查是否在唤醒处理锁定期内 + if getattr(conn, 'wakeup_processing_lock', 0) > time.monotonic(): + return + if conn.client_listen_mode == "auto" or conn.client_listen_mode == "realtime": have_voice = audio_have_voice else: @@ -62,6 +66,14 @@ class ASRProviderBase(ABC): if not have_voice and not conn.client_have_voice: conn.asr_audio = conn.asr_audio[-10:] return + + # 检查是否处于唤醒模式 + if getattr(conn, 'wakeup_mode', False) and len(conn.asr_audio) >= 10: + asr_audio_task = conn.asr_audio.copy() + conn.reset_vad_states() + conn.asr_audio.clear() + + await self.handle_voice_stop(conn, asr_audio_task) if conn.client_voice_stop: asr_audio_task = conn.asr_audio.copy() @@ -87,32 +99,13 @@ class ASRProviderBase(ABC): # 预先准备WAV数据 wav_data = None - # 使用连接的声纹识别提供者 if conn.voiceprint_provider and combined_pcm_data: wav_data = self._pcm_to_wav(combined_pcm_data) + # 检查是否处于唤醒模式 + wakeup_mode = getattr(conn, 'wakeup_mode', False) - # 定义ASR任务 - def run_asr(): - start_time = time.monotonic() - try: - loop = asyncio.new_event_loop() - asyncio.set_event_loop(loop) - try: - result = loop.run_until_complete( - self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format) - ) - end_time = time.monotonic() - logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s") - return result - finally: - loop.close() - except Exception as e: - end_time = time.monotonic() - logger.bind(tag=TAG).error(f"ASR失败: {e}") - return ("", None) - - # 定义声纹识别任务 + # 声纹识别任务(公共逻辑) def run_voiceprint(): if not wav_data: return None @@ -131,48 +124,83 @@ class ASRProviderBase(ABC): logger.bind(tag=TAG).error(f"声纹识别失败: {e}") return None - # 使用线程池执行器并行运行 - with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor: - asr_future = thread_executor.submit(run_asr) + if wakeup_mode and conn.voiceprint_provider and wav_data: + conn.wakeup_mode = False + # 设置处理锁,防止后续音频片段重复处理 + conn.wakeup_processing_lock = time.monotonic() + 3 # 3秒锁定期 - if conn.voiceprint_provider and wav_data: + # 唤醒模式:只执行声纹识别 + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as thread_executor: voiceprint_future = thread_executor.submit(run_voiceprint) - - # 等待两个线程都完成 - asr_result = asr_future.result(timeout=15) voiceprint_result = voiceprint_future.result(timeout=15) - - results = {"asr": asr_result, "voiceprint": voiceprint_result} - else: - asr_result = asr_future.result(timeout=15) - results = {"asr": asr_result, "voiceprint": None} - - - # 处理结果 - raw_text, file_path = results.get("asr", ("", None)) - speaker_name = results.get("voiceprint", None) - - # 记录识别结果 - if raw_text: - logger.bind(tag=TAG).info(f"识别文本: {raw_text}") - if speaker_name: + + speaker_name = voiceprint_result logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}") - - # 性能监控 - total_time = time.monotonic() - total_start_time - logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s") - - # 检查文本长度 - text_len, _ = remove_punctuation_and_length(raw_text) - self.stop_ws_connection() - - if text_len > 0: - # 构建包含说话人信息的JSON字符串 - enhanced_text = self._build_enhanced_text(raw_text, speaker_name) + + fixed_text = "嘿,你好啊" + enhanced_text = self._build_enhanced_text(fixed_text, speaker_name) - # 使用自定义模块进行上报 + # 性能监控 + total_time = time.monotonic() - total_start_time + logger.bind(tag=TAG).info(f"唤醒模式总处理耗时: {total_time:.3f}s") + await startToChat(conn, enhanced_text) enqueue_asr_report(conn, enhanced_text, asr_audio_task) + else: + # 正常模式:执行声纹识别和ASR + def run_asr(): + start_time = time.monotonic() + try: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + try: + result = loop.run_until_complete( + self.speech_to_text(asr_audio_task, conn.session_id, conn.audio_format) + ) + end_time = time.monotonic() + logger.bind(tag=TAG).info(f"ASR耗时: {end_time - start_time:.3f}s") + return result + finally: + loop.close() + except Exception as e: + end_time = time.monotonic() + logger.bind(tag=TAG).error(f"ASR失败: {e}") + return ("", None) + + # 使用线程池执行器并行运行 + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor: + asr_future = thread_executor.submit(run_asr) + + if conn.voiceprint_provider and wav_data: + voiceprint_future = thread_executor.submit(run_voiceprint) + asr_result = asr_future.result(timeout=15) + voiceprint_result = voiceprint_future.result(timeout=15) + results = {"asr": asr_result, "voiceprint": voiceprint_result} + else: + asr_result = asr_future.result(timeout=15) + results = {"asr": asr_result, "voiceprint": None} + + # 处理结果 + raw_text, _ = results.get("asr", ("", None)) + speaker_name = results.get("voiceprint", None) + + if raw_text: + logger.bind(tag=TAG).info(f"识别文本: {raw_text}") + if speaker_name: + logger.bind(tag=TAG).info(f"识别说话人: {speaker_name}") + + # 性能监控 + total_time = time.monotonic() - total_start_time + logger.bind(tag=TAG).info(f"总处理耗时: {total_time:.3f}s") + + # 检查文本长度 + text_len, _ = remove_punctuation_and_length(raw_text) + self.stop_ws_connection() + + if text_len > 0: + enhanced_text = self._build_enhanced_text(raw_text, speaker_name) + await startToChat(conn, enhanced_text) + enqueue_asr_report(conn, enhanced_text, asr_audio_task) except Exception as e: logger.bind(tag=TAG).error(f"处理语音停止失败: {e}") diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index 3306942b..04e83148 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -1,9 +1,7 @@ import os -import time import queue import aiohttp import asyncio -import requests import traceback from config.logger import setup_logging from core.utils.tts import MarkdownCleaner @@ -111,10 +109,6 @@ class TTSProvider(TTSProviderBase): finally: return None - ################################################################################### - # linkerai单流式TTS重写父类的方法--结束 - ################################################################################### - async def text_to_speak(self, text, is_last): """流式处理TTS音频,每句只推送一次音频列表""" await self._tts_request(text, is_last) @@ -201,71 +195,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(self, text: str) -> list: - """非流式TTS处理,用于测试及保存音频文件的场景 - Args: - text: 要转换的文本 - Returns: - list: 返回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=lambda opus: opus_datas.append(opus) - ) - - return opus_datas - - except Exception as e: - logger.bind(tag=TAG).error(f"TTS请求异常: {e}") - return [] \ No newline at end of file diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 26484b41..a608049c 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -1,7 +1,6 @@ import re import os import json -import wave import copy import socket import requests @@ -12,6 +11,7 @@ from io import BytesIO from core.utils import p3 from pydub import AudioSegment from typing import Callable, Any +from core.providers.tts.dto.dto import SentenceType TAG = __name__ emoji_map = { @@ -274,33 +274,23 @@ def pcm_to_data_stream(raw_data, is_opus=True, callback: Callable[[Any], Any] = frame_data = chunk if isinstance(chunk, bytes) else bytes(chunk) callback(frame_data) +def play_audio_response(conn, response): + """音频响应处理""" + conn.tts.tts_audio_queue.put((SentenceType.FIRST, [], response.get("text"))) + play_audio_frames(conn, response.get("file_path")) + conn.tts.tts_audio_queue.put((SentenceType.LAST, [], None)) -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 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 + ) 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 70274dd4..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