diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index e4e836b4..75b9fcb2 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -4,15 +4,15 @@ import random import asyncio from core.utils.dialogue import Message from core.utils.util import audio_to_data +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 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__ @@ -56,7 +56,16 @@ async def checkWakeupWords(conn, text): "enable_wakeup_words_response_cache" ] - if not enable_wakeup_words_response_cache or not conn.tts: + # 等待tts初始化，最多等待3秒 + start_time = time.time() + while time.time() - start_time < 3: + if conn.tts: + break + await asyncio.sleep(0.1) + else: + return False + + if not enable_wakeup_words_response_cache: return False _, filtered_text = remove_punctuation_and_length(text) @@ -81,9 +90,10 @@ async def checkWakeupWords(conn, text): "text": "哈啰啊，我是小智啦，声音好听的台湾女孩一枚，超开心认识你耶，最近在忙啥，别忘了给我来点有趣的料哦，我超爱听八卦的啦", } + # 获取音频数据 + opus_packets = audio_to_data(response.get("file_path")) # 播放唤醒词回复 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")) @@ -138,4 +148,4 @@ async def wakeupWordsResponse(conn): finally: # 确保在任何情况下都释放锁 if _wakeup_response_lock.locked(): - _wakeup_response_lock.release() + _wakeup_response_lock.release() \ 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 81911f55..72424968 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -1,12 +1,12 @@ 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 core.handle.helloHandle import checkWakeupWords 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,9 +24,10 @@ 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 diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index e6f39632..8db50633 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) @@ -31,18 +29,16 @@ async def handleAudioMessage(conn, audio): # 接收音频 await conn.asr.receive_audio(conn, audio, have_voice) - async def resume_vad_detection(conn): # 等待2秒后恢复VAD检测 await asyncio.sleep(1) conn.just_woken_up = False - async def startToChat(conn, text): # 检查输入是否是JSON格式（包含说话人信息） speaker_name = None actual_text = text - + try: # 尝试解析JSON格式的输入 if text.strip().startswith('{') and text.strip().endswith('}'): @@ -51,13 +47,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 @@ -118,10 +114,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 +138,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 +146,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 0ca36e1e..661c4e21 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -1,25 +1,26 @@ import json -import asyncio import time -from core.providers.tts.dto.dto import SentenceType +import asyncio from core.utils import textUtils +from core.utils.util import audio_to_data +from core.providers.tts.dto.dto import SentenceType 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) - await send_tts_message(conn, "sentence_start", text) + 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: @@ -30,45 +31,83 @@ async def sendAudioMessage(conn, sentenceType, audios, text): # 播放音频 -async def sendAudio(conn, audios, pre_buffer=True): +async def sendAudio(conn, audios, frame_duration=60): + """ + 发送单个opus包，支持流控 + Args: + conn: 连接对象 + opus_packet: 单个opus数据包 + pre_buffer: 快速发送音频 + frame_duration: 帧时长（毫秒），匹配 Opus 编码 + """ 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 isinstance(audios, bytes): if conn.client_abort: - break + return - # 重置没有声音的状态 conn.last_activity_time = time.time() * 1000 - # 计算预期发送时间 - expected_time = start_time + (play_position / 1000) + # 获取或初始化流控状态 + if not hasattr(conn, "audio_flow_control"): + conn.audio_flow_control = { + "last_send_time": 0, + "packet_count": 0, + "start_time": time.perf_counter(), + } + + flow_control = conn.audio_flow_control current_time = time.perf_counter() + # 计算预期发送时间 + expected_time = flow_control["start_time"] + ( + flow_control["packet_count"] * frame_duration / 1000 + ) delay = expected_time - current_time if delay > 0: await asyncio.sleep(delay) - await conn.websocket.send(opus_packet) + # 发送数据包 + await conn.websocket.send(audios) - play_position += frame_duration + # 更新流控状态 + flow_control["packet_count"] += 1 + flow_control["last_send_time"] = time.perf_counter() + else: + # 文件型音频走普通播放 + 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): """发送 TTS 状态消息""" + if text is None and state == "sentence_start": + return message = {"type": "tts", "state": state, "session_id": conn.session_id} if text is not None: message["text"] = textUtils.check_emoji(text) @@ -81,7 +120,7 @@ async def send_tts_message(conn, state, text=None): stop_tts_notify_voice = conn.config.get( "stop_tts_notify_voice", "config/assets/tts_notify.mp3" ) - audios, _ = conn.tts.audio_to_opus_data(stop_tts_notify_voice) + audios = audio_to_data(stop_tts_notify_voice, is_opus=True) await sendAudio(conn, audios) # 清除服务端讲话状态 conn.clearSpeakStatus() @@ -91,18 +130,17 @@ 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: # 尝试解析JSON格式 - if text.strip().startswith('{') and text.strip().endswith('}'): + if text.strip().startswith("{") and text.strip().endswith("}"): parsed_data = json.loads(text) if isinstance(parsed_data, dict) and "content" in parsed_data: # 如果是包含说话人信息的JSON格式，只显示content部分 diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 972818d4..250a25f2 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -1,18 +1,18 @@ 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 -from typing import Optional, Tuple, List, Dict, Any +from typing import Optional, Tuple, List from core.handle.receiveAudioHandle import startToChat from core.handle.reportHandle import enqueue_asr_report from core.utils.util import remove_punctuation_and_length @@ -87,11 +87,9 @@ class ASRProviderBase(ABC): # 预先准备WAV数据 wav_data = None - # 使用连接的声纹识别提供者 if conn.voiceprint_provider and combined_pcm_data: wav_data = self._pcm_to_wav(combined_pcm_data) - # 定义ASR任务 def run_asr(): start_time = time.monotonic() @@ -132,8 +130,6 @@ class ASRProviderBase(ABC): return None # 使用线程池执行器并行运行 - parallel_start_time = time.monotonic() - with concurrent.futures.ThreadPoolExecutor(max_workers=2) as thread_executor: asr_future = thread_executor.submit(run_asr) @@ -151,7 +147,7 @@ class ASRProviderBase(ABC): # 处理结果 - raw_text, file_path = results.get("asr", ("", None)) + raw_text, _ = results.get("asr", ("", None)) speaker_name = results.get("voiceprint", None) # 记录识别结果 diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 0bfb367b..38e70927 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -268,11 +268,7 @@ 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: try: logger.bind(tag=TAG).info("开始结束TTS会话...") @@ -422,9 +418,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(): @@ -445,28 +438,16 @@ class TTSProvider(TTSProviderBase): self.tts_audio_queue.put( (SentenceType.FIRST, [], None) ) - elif event_name == "SentenceBegin": - opus_datas_cache = [] elif event_name == "SentenceEnd": - if ( - not is_first_sentence - or first_sentence_segment_count > 10 - ): - # 发送缓存的数据 - if self.conn.tts_MessageText: - logger.bind(tag=TAG).info( - f"句子语音生成成功： {self.conn.tts_MessageText}" - ) - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_datas_cache, self.conn.tts_MessageText) - ) - self.conn.tts_MessageText = None - else: - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_datas_cache, None) - ) - # 第一句话结束后，将标志设置为False - is_first_sentence = False + # 发送缓存的数据 + if self.conn.tts_MessageText: + logger.bind(tag=TAG).info( + f"句子语音生成成功： {self.conn.tts_MessageText}" + ) + self.tts_audio_queue.put( + (SentenceType.FIRST, [], self.conn.tts_MessageText) + ) + self.conn.tts_MessageText = None elif event_name == "SynthesisCompleted": logger.bind(tag=TAG).debug(f"会话结束～～") self._process_before_stop_play_files() @@ -477,22 +458,7 @@ class TTSProvider(TTSProviderBase): # 二进制消息（音频数据） elif isinstance(msg, (bytes, bytearray)): logger.bind(tag=TAG).debug(f"推送数据到队列里面～～") - opus_datas = self.opus_encoder.encode_pcm_to_opus(msg, False) - logger.bind(tag=TAG).debug( - f"推送数据到队列里面帧数～～{len(opus_datas)}" - ) - if is_first_sentence: - first_sentence_segment_count += 1 - if first_sentence_segment_count <= 6: - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_datas, None) - ) - else: - opus_datas_cache.extend(opus_datas) - else: - # 后续句子缓存 - opus_datas_cache.extend(opus_datas) - + self.opus_encoder.encode_pcm_to_opus_stream(msg, False, self.handle_opus) except websockets.ConnectionClosed: logger.bind(tag=TAG).warning("WebSocket连接已关闭") break @@ -615,11 +581,11 @@ class TTSProvider(TTSProviderBase): while not synthesis_completed: msg = await ws.recv() if isinstance(msg, (bytes, bytearray)): - # 编码为Opus并收集 - opus_frames = self.opus_encoder.encode_pcm_to_opus( - msg, False + self.opus_encoder.encode_pcm_to_opus_stream( + msg, + end_of_stream=False, + callback=lambda opus: audio_data.append(opus) ) - audio_data.extend(opus_frames) elif isinstance(msg, str): data = json.loads(msg) header = data.get("header", {}) @@ -650,4 +616,4 @@ class TTSProvider(TTSProviderBase): return audio_data except Exception as e: logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}") - return [] + return [] \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 40570326..04a7fa36 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -1,19 +1,22 @@ import os import re -import queue +import time import uuid +import queue import asyncio import threading +import traceback from core.utils import p3 from datetime import datetime from core.utils import textUtils +from typing import Callable, Any from abc import ABC, abstractmethod from config.logger import setup_logging -from core.utils.util import audio_to_data, audio_bytes_to_data from core.utils.tts import MarkdownCleaner from core.utils.output_counter import add_device_output from core.handle.reportHandle import enqueue_tts_report from core.handle.sendAudioHandle import sendAudioMessage +from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream from core.providers.tts.dto.dto import ( TTSMessageDTO, SentenceType, @@ -21,8 +24,6 @@ from core.providers.tts.dto.dto import ( InterfaceType, ) -import traceback - TAG = __name__ logger = setup_logging() @@ -31,7 +32,6 @@ class TTSProviderBase(ABC): def __init__(self, config, delete_audio_file): self.interface_type = InterfaceType.NON_STREAM self.conn = None - self.tts_timeout = 10 self.delete_audio_file = delete_audio_file self.audio_file_type = "wav" self.output_file = config.get("output_dir", "tmp/") @@ -50,11 +50,9 @@ class TTSProviderBase(ABC): "；", ";", "：", - "~", ) self.first_sentence_punctuations = ( "，", - "～", "~", "、", ",", @@ -77,6 +75,75 @@ class TTSProviderBase(ABC): f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}", ) + def handle_opus(self, opus_data: bytes): + logger.bind(tag=TAG).debug(f"推送数据到队列里面帧数～～ {len(opus_data)}") + self.tts_audio_queue.put((SentenceType.MIDDLE, opus_data, None)) + + def handle_audio_file(self, file_audio: bytes, text): + self.before_stop_play_files.append((file_audio, text)) + + 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: + # 需要删除文件的直接转为音频数据 + while max_repeat_time > 0: + try: + audio_bytes = asyncio.run(self.text_to_speak(text, None)) + if audio_bytes: + self.tts_audio_queue.put((SentenceType.FIRST, None, text)) + audio_bytes_to_data_stream( + audio_bytes, + file_type=self.audio_file_type, + is_opus=True, + callback=opus_handler, + ) + break + else: + max_repeat_time -= 1 + except Exception as e: + logger.bind(tag=TAG).warning( + f"语音生成失败{5 - max_repeat_time + 1}次: {text}，错误: {e}" + ) + max_repeat_time -= 1 + if max_repeat_time > 0: + logger.bind(tag=TAG).info( + f"语音生成成功: {text}，重试{5 - max_repeat_time}次" + ) + else: + logger.bind(tag=TAG).error( + f"语音生成失败: {text}，请检查网络或服务是否正常" + ) + return None + else: + tmp_file = self.generate_filename() + try: + while not os.path.exists(tmp_file) and max_repeat_time > 0: + try: + asyncio.run(self.text_to_speak(text, tmp_file)) + except Exception as e: + logger.bind(tag=TAG).warning( + f"语音生成失败{5 - max_repeat_time + 1}次: {text}，错误: {e}" + ) + # 未执行成功，删除文件 + if os.path.exists(tmp_file): + os.remove(tmp_file) + max_repeat_time -= 1 + + if max_repeat_time > 0: + logger.bind(tag=TAG).info( + f"语音生成成功: {text}:{tmp_file}，重试{5 - max_repeat_time}次" + ) + else: + logger.bind(tag=TAG).error( + f"语音生成失败: {text}，请检查网络或服务是否正常" + ) + self.tts_audio_queue.put((SentenceType.FIRST, None, text)) + self._process_audio_file_stream(tmp_file, callback=opus_handler) + except Exception as e: + logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}") + return None + def to_tts(self, text): text = MarkdownCleaner.clean_markdown(text) max_repeat_time = 5 @@ -86,8 +153,12 @@ class TTSProviderBase(ABC): try: audio_bytes = asyncio.run(self.text_to_speak(text, None)) if audio_bytes: - audio_datas, _ = audio_bytes_to_data( - audio_bytes, file_type=self.audio_file_type, is_opus=True + audio_datas = [] + audio_bytes_to_data_stream( + audio_bytes, + file_type=self.audio_file_type, + is_opus=True, + callback=lambda data: audio_datas.append(data) ) return audio_datas else: @@ -139,13 +210,17 @@ class TTSProviderBase(ABC): async def text_to_speak(self, text, output_file): pass - def audio_to_pcm_data(self, audio_file_path): + def audio_to_pcm_data_stream( + self, audio_file_path, callback: Callable[[Any], Any] = None + ): """音频文件转换为PCM编码""" - return audio_to_data(audio_file_path, is_opus=False) + return audio_to_data_stream(audio_file_path, is_opus=False, callback=callback) - def audio_to_opus_data(self, audio_file_path): + def audio_to_opus_data_stream( + self, audio_file_path, callback: Callable[[Any], Any] = None + ): """音频文件转换为Opus编码""" - return audio_to_data(audio_file_path, is_opus=True) + return audio_to_data_stream(audio_file_path, is_opus=True, callback=callback) def tts_one_sentence( self, @@ -177,7 +252,6 @@ class TTSProviderBase(ABC): async def open_audio_channels(self, conn): self.conn = conn - self.tts_timeout = conn.config.get("tts_timeout", 10) # tts 消化线程 self.tts_priority_thread = threading.Thread( target=self.tts_text_priority_thread, daemon=True @@ -212,30 +286,16 @@ class TTSProviderBase(ABC): self.tts_text_buff.append(message.content_detail) segment_text = self._get_segment_text() if segment_text: - if self.delete_audio_file: - audio_datas = self.to_tts(segment_text) - if audio_datas: - self.tts_audio_queue.put( - (message.sentence_type, audio_datas, segment_text) - ) - else: - tts_file = self.to_tts(segment_text) - if tts_file: - audio_datas = self._process_audio_file(tts_file) - self.tts_audio_queue.put( - (message.sentence_type, audio_datas, segment_text) - ) + self.to_tts_stream(segment_text, opus_handler=self.handle_opus) elif ContentType.FILE == message.content_type: - self._process_remaining_text() + self._process_remaining_text_stream(opus_handler=self.handle_opus) tts_file = message.content_file if tts_file and os.path.exists(tts_file): - audio_datas = self._process_audio_file(tts_file) - self.tts_audio_queue.put( - (message.sentence_type, audio_datas, message.content_detail) + self._process_audio_file_stream( + tts_file, callback=self.handle_opus ) - if message.sentence_type == SentenceType.LAST: - self._process_remaining_text() + self._process_remaining_text_stream(opus_handler=self.handle_opus) self.tts_audio_queue.put( (message.sentence_type, [], message.content_detail) ) @@ -249,29 +309,59 @@ 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: try: sentence_type, audio_datas, text = self.tts_audio_queue.get( - timeout=1 + timeout=0.1 ) except queue.Empty: if self.conn.stop_event.is_set(): break continue + + if self.conn.client_abort: + logger.bind(tag=TAG).debug("收到打断信号，跳过当前音频数据") + enqueue_text, enqueue_audio = None, [] + continue + + # 收到下一个文本开始或会话结束时进行上报 + if sentence_type is not SentenceType.MIDDLE: + # 重置音频流控状态（新句子开始或者结束） + if hasattr(self.conn, 'audio_flow_control'): + self.conn.audio_flow_control = { + 'last_send_time': 0, + 'packet_count': 0, + 'start_time': time.perf_counter() + } + + # 上报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) and enqueue_audio is not None: + enqueue_audio.append(audio_datas) + + # 发送音频 future = asyncio.run_coroutine_threadsafe( sendAudioMessage(self.conn, 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 priority_thread: {text} {e}" - ) + logger.bind(tag=TAG).error(f"audio_play_priority_thread: {text} {e}") async def start_session(self, session_id): pass @@ -323,22 +413,21 @@ class TTSProviderBase(ABC): else: return None - def _process_audio_file(self, tts_file): + def _process_audio_file_stream( + self, tts_file, callback: Callable[[Any], Any] + ) -> None: """处理音频文件并转换为指定格式 Args: tts_file: 音频文件路径 - content_detail: 内容详情 - - Returns: - tuple: (sentence_type, audio_datas, content_detail) + callback: 文件处理函数 """ if tts_file.endswith(".p3"): - audio_datas, _ = p3.decode_opus_from_file(tts_file) + p3.decode_opus_from_file_stream(tts_file, callback=callback) elif self.conn.audio_format == "pcm": - audio_datas, _ = self.audio_to_pcm_data(tts_file) + self.audio_to_pcm_data_stream(tts_file, callback=callback) else: - audio_datas, _ = self.audio_to_opus_data(tts_file) + self.audio_to_opus_data_stream(tts_file, callback=callback) if ( self.delete_audio_file @@ -347,7 +436,6 @@ class TTSProviderBase(ABC): and tts_file.startswith(self.output_file) ): os.remove(tts_file) - return audio_datas def _process_before_stop_play_files(self): for audio_datas, text in self.before_stop_play_files: @@ -355,7 +443,9 @@ class TTSProviderBase(ABC): self.before_stop_play_files.clear() self.tts_audio_queue.put((SentenceType.LAST, [], None)) - def _process_remaining_text(self): + def _process_remaining_text_stream( + self, opus_handler: Callable[[bytes], None] = None + ): """处理剩余的文本并生成语音 Returns: @@ -366,18 +456,7 @@ class TTSProviderBase(ABC): if remaining_text: segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text) if segment_text: - if self.delete_audio_file: - audio_datas = self.to_tts(segment_text) - if audio_datas: - self.tts_audio_queue.put( - (SentenceType.MIDDLE, audio_datas, segment_text) - ) - else: - tts_file = self.to_tts(segment_text) - audio_datas = self._process_audio_file(tts_file) - self.tts_audio_queue.put( - (SentenceType.MIDDLE, audio_datas, segment_text) - ) + self.to_tts_stream(segment_text, opus_handler=opus_handler) self.processed_chars += len(full_text) return True return False 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 33e2750d..ff295581 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -4,6 +4,7 @@ import json 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 @@ -266,11 +267,7 @@ 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: try: logger.bind(tag=TAG).info("开始结束TTS会话...") @@ -428,9 +425,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(): @@ -451,37 +445,14 @@ 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 ): logger.bind(tag=TAG).debug(f"推送数据到队列里面～～") - opus_datas = self.wav_to_opus_data_audio_raw(res.payload) - logger.bind(tag=TAG).debug( - f"推送数据到队列里面帧数～～{len(opus_datas)}" - ) - if is_first_sentence: - first_sentence_segment_count += 1 - if first_sentence_segment_count <= 6: - self.tts_audio_queue.put( - (SentenceType.MIDDLE, opus_datas, None) - ) - else: - opus_datas_cache.extend(opus_datas) - else: - # 后续句子缓存 - opus_datas_cache.extend(opus_datas) + 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() @@ -655,16 +626,13 @@ class TTSProvider(TTSProviderBase): ) ) - def wav_to_opus_data_audio_raw(self, raw_data_var, is_end=False): - opus_datas = self.opus_encoder.encode_pcm_to_opus(raw_data_var, is_end) - return opus_datas + 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(self, text: str) -> list: """非流式生成音频数据，用于生成音频及测试场景 - Args: text: 要转换的文本 - Returns: list: 音频数据列表 """ @@ -741,8 +709,7 @@ class TTSProvider(TTSProviderBase): res.optional.event == EVENT_TTSResponse and res.header.message_type == AUDIO_ONLY_RESPONSE ): - opus_datas = self.wav_to_opus_data_audio_raw(res.payload) - audio_data.extend(opus_datas) + self.wav_to_opus_data_audio_raw_stream(res.payload, callback=lambda opus_frame: audio_data.append(opus_frame)) elif res.optional.event == EVENT_SessionFinished: break diff --git a/main/xiaozhi-server/core/providers/tts/index_stream.py b/main/xiaozhi-server/core/providers/tts/index_stream.py index 5ea9eb36..8abb06f2 100644 --- a/main/xiaozhi-server/core/providers/tts/index_stream.py +++ b/main/xiaozhi-server/core/providers/tts/index_stream.py @@ -1,10 +1,10 @@ import os -import queue -import asyncio -import traceback -import aiohttp -import requests import time +import queue +import aiohttp +import asyncio +import requests +import traceback from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -27,15 +27,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,7 +46,6 @@ 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() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) @@ -62,14 +59,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 +72,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 +137,6 @@ class TTSProvider(TTSProviderBase): return self.pcm_buffer.clear() - opus_datas_cache = [] - self.tts_audio_queue.put((SentenceType.FIRST, [], text)) # 处理音频流数据 @@ -158,41 +150,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() @@ -209,10 +182,8 @@ class TTSProvider(TTSProviderBase): def to_tts(self, text: str) -> list: """非流式TTS处理，用于测试及保存音频文件的场景 - Args: text: 要转换的文本 - Returns: list: 返回opus编码后的音频数据列表 """ @@ -251,14 +222,14 @@ class TTSProvider(TTSProviderBase): # 最后一帧可能不足，用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)) + 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) ) - if opus: - opus_datas.extend(opus) return opus_datas except Exception as e: logger.bind(tag=TAG).error(f"TTS请求异常: {e}") - return [] + return [] \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/linkerai.py b/main/xiaozhi-server/core/providers/tts/linkerai.py index d1be586c..9b5e946f 100644 --- a/main/xiaozhi-server/core/providers/tts/linkerai.py +++ b/main/xiaozhi-server/core/providers/tts/linkerai.py @@ -1,10 +1,10 @@ import os -import queue -import asyncio -import traceback -import aiohttp -import requests import time +import queue +import aiohttp +import asyncio +import requests +import traceback from config.logger import setup_logging from core.utils.tts import MarkdownCleaner from core.providers.tts.base import TTSProviderBase @@ -24,23 +24,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,7 +43,6 @@ 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() elif ContentType.TEXT == message.content_type: self.tts_text_buff.append(message.content_detail) @@ -65,14 +56,10 @@ 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 @@ -81,7 +68,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: @@ -124,10 +111,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) @@ -176,8 +159,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 @@ -194,41 +175,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() @@ -239,10 +200,8 @@ class TTSProvider(TTSProviderBase): def to_tts(self, text: str) -> list: """非流式TTS处理，用于测试及保存音频文件的场景 - Args: text: 要转换的文本 - Returns: list: 返回opus编码后的音频数据列表 """ @@ -295,14 +254,14 @@ class TTSProvider(TTSProviderBase): # 最后一帧可能不足，用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)) + 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) ) - if opus: - opus_datas.extend(opus) return opus_datas except Exception as e: logger.bind(tag=TAG).error(f"TTS请求异常: {e}") - return [] + return [] \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/tts/paddle_speech.py b/main/xiaozhi-server/core/providers/tts/paddle_speech.py index 7f1c6a23..864cf505 100644 --- a/main/xiaozhi-server/core/providers/tts/paddle_speech.py +++ b/main/xiaozhi-server/core/providers/tts/paddle_speech.py @@ -1,13 +1,15 @@ -import asyncio -import json -import base64 -import aiohttp -import numpy as np import io import wave +import json +import base64 +import asyncio import websockets -from core.providers.tts.base import TTSProviderBase +import numpy as np +from datetime import datetime from config.logger import setup_logging +from core.providers.tts.base import TTSProviderBase + + TAG = __name__ logger = setup_logging() @@ -18,11 +20,12 @@ class TTSProvider(TTSProviderBase): super().__init__(config, delete_audio_file) self.url = config.get("url", "ws://192.168.1.10:8092/paddlespeech/tts/streaming") self.protocol = config.get("protocol", "websocket") + if config.get("private_voice"): self.spk_id = int(config.get("private_voice")) else: - self.spk_id = int(config.get("spk_id", "0")) - + self.spk_id = int(config.get("spk_id", "0")) + sample_rate = config.get("sample_rate", 24000) self.sample_rate = float(sample_rate) if sample_rate else 24000 @@ -32,7 +35,21 @@ class TTSProvider(TTSProviderBase): volume = config.get("volume", 1.0) self.volume = float(volume) if volume else 1.0 - self.save_path = config.get("save_path", "./streaming_tts.wav") + self.delete_audio_file = config.get("delete_audio", True) + if not self.delete_audio_file: + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + save_path = config.get("save_path") + if save_path: + if not save_path.endswith('.wav'): + save_path = f"{save_path}_{timestamp}.wav" + else: + other_path = save_path[:-4] + save_path = f"{other_path}_{timestamp}.wav" + self.save_path = save_path + else: + self.save_path = f"./streaming_tts_{timestamp}.wav" + else: + self.save_path = None async def pcm_to_wav(self, pcm_data: bytes, sample_rate: int = 24000, num_channels: int = 1, bits_per_sample: int = 16) -> bytes: @@ -58,43 +75,9 @@ class TTSProvider(TTSProviderBase): async def text_to_speak(self, text, output_file): if self.protocol == "websocket": return await self.text_streaming(text, output_file) - elif self.protocol == "http": - return await self.text(text, output_file) else: raise ValueError("Unsupported protocol. Please use 'websocket' or 'http'.") - async def text(self, text, output_file): - request_json = { - "text": text, - "spk_id": self.spk_id, - "speed": self.speed, - "volume": self.volume, - "sample_rate": self.sample_rate, - "save_path": self.save_path - } - - try: - async with aiohttp.ClientSession() as session: - async with session.post(self.url, json=request_json) as resp: - if resp.status == 200: - resp_json = await resp.json() - if resp_json.get("success"): - data = resp_json["result"] - audio_bytes = base64.b64decode(data["audio"]) - if output_file: - with open(output_file, "wb") as file_to_save: - file_to_save.write(audio_bytes) - else: - return audio_bytes - else: - raise Exception( - f"Error: {resp_json.get('message', 'Unknown error')} while processing text: {text}") - else: - raise Exception( - f"HTTP Error: {resp.status} - {await resp.text()} while processing text: {text}") - except Exception as e: - raise Exception(f"Error during TTS HTTP request: {e} while processing text: {text}") - async def text_streaming(self, text, output_file): try: # 使用 websockets 异步连接到 WebSocket 服务器 @@ -151,6 +134,12 @@ class TTSProvider(TTSProviderBase): # 接收结束响应避免服务抛出异常 await ws.recv() + # 根据配置决定是否保存文件 + if not self.delete_audio_file and self.save_path: + with open(self.save_path, "wb") as f: + f.write(wav_data) + logger.bind(tag=TAG).info(f"音频文件已保存到: {self.save_path}") + # 返回或保存音频数据 if output_file: with open(output_file, "wb") as file_to_save: @@ -159,4 +148,4 @@ class TTSProvider(TTSProviderBase): return wav_data except Exception as e: - raise Exception(f"Error during TTS WebSocket request: {e} while processing text: {text}") + raise Exception(f"Error during TTS WebSocket request: {e} while processing text: {text}") \ No newline at end of file diff --git a/main/xiaozhi-server/core/providers/vad/silero.py b/main/xiaozhi-server/core/providers/vad/silero.py index 80e65b55..b516d8fb 100644 --- a/main/xiaozhi-server/core/providers/vad/silero.py +++ b/main/xiaozhi-server/core/providers/vad/silero.py @@ -34,7 +34,7 @@ class VADProvider(VADProviderBase): ) # 至少要多少帧才算有语音 - self.frame_window_threshold = 1 + self.frame_window_threshold = 3 def is_vad(self, conn, opus_packet): try: @@ -70,9 +70,7 @@ class VADProvider(VADProviderBase): # 更新滑动窗口 conn.client_voice_window.append(is_voice) - client_have_voice = ( - conn.client_voice_window.count(True) >= self.frame_window_threshold - ) + client_have_voice = (conn.client_voice_window.count(True) >= self.frame_window_threshold) # 如果之前有声音，但本次没有声音，且与上次有声音的时间差已经超过了静默阈值，则认为已经说完一句话 if conn.client_have_voice and not client_have_voice: diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index ce6f20ea..27471e58 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -1,16 +1,17 @@ -import json -import socket -import subprocess import re import os +import json +import copy import wave +import socket +import requests +import subprocess +import numpy as np +import opuslib_next from io import BytesIO from core.utils import p3 -import numpy as np -import requests -import opuslib_next from pydub import AudioSegment -import copy +from typing import Callable, Any TAG = __name__ emoji_map = { @@ -211,7 +212,7 @@ def extract_json_from_string(input_string): return None -def audio_to_data(audio_file_path, is_opus=True): +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] if file_type: @@ -224,33 +225,32 @@ def audio_to_data(audio_file_path, is_opus=True): # 转换为单声道/16kHz采样率/16位小端编码（确保与编码器匹配） audio = audio.set_channels(1).set_frame_rate(16000).set_sample_width(2) - # 音频时长(秒) - duration = len(audio) / 1000.0 + # 获取原始PCM数据（16位小端） + raw_data = audio.raw_data + pcm_to_data_stream(raw_data, is_opus, callback) + +def audio_to_data(audio_file_path: str, is_opus: bool = True) -> list[bytes]: + """ + 将音频文件转换为Opus/PCM编码的帧列表 + Args: + audio_file_path: 音频文件路径 + is_opus: 是否进行Opus编码 + """ + # 获取文件后缀名 + 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), duration - -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) - duration = len(audio) / 1000.0 - raw_data = audio.raw_data - return pcm_to_data(raw_data, is_opus), duration - - -def pcm_to_data(raw_data, is_opus=True): # 初始化Opus编码器 encoder = opuslib_next.Encoder(16000, 1, opuslib_next.APPLICATION_AUDIO) @@ -280,6 +280,49 @@ def pcm_to_data(raw_data, is_opus=True): return datas +def audio_bytes_to_data_stream(audio_bytes, file_type, is_opus, callback: Callable[[Any], Any]) -> None: + """ + 直接用音频二进制数据转为opus/pcm数据，支持wav、mp3、p3 + """ + if file_type == "p3": + # 直接用p3解码 + return p3.decode_opus_from_bytes_stream(audio_bytes, callback) + 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 + pcm_to_data_stream(raw_data, is_opus, callback) + + +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) + + # 编码参数 + frame_duration = 60 # 60ms per frame + frame_size = int(16000 * frame_duration / 1000) # 960 samples/frame + + # 按帧处理所有音频数据（包括最后一帧可能补零） + 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) + callback(frame_data) + else: + 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): """ @@ -307,7 +350,6 @@ def opus_datas_to_wav_bytes(opus_datas, sample_rate=16000, channels=1): wf.writeframes(pcm_bytes) return wav_buffer.getvalue() - def check_vad_update(before_config, new_config): if ( new_config.get("selected_module") is None diff --git a/main/xiaozhi-server/core/utils/wakeup_word.py b/main/xiaozhi-server/core/utils/wakeup_word.py index ab58f290..d2f4fb32 100644 --- a/main/xiaozhi-server/core/utils/wakeup_word.py +++ b/main/xiaozhi-server/core/utils/wakeup_word.py @@ -137,4 +137,4 @@ class WakeupWordsConfig: return file_path except Exception as e: print(f"生成音频文件路径失败: {e}") - raise + raise \ No newline at end of file