diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 66a8ed54..6460643d 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -15,6 +15,7 @@ from plugins_func.loadplugins import auto_import_modules from config.logger import setup_logging from core.utils.dialogue import Message, Dialogue from core.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType +from core.providers.tts.default import DefaultTTS from core.handle.textHandle import handleTextMessage from core.utils.util import ( extract_json_from_string, @@ -312,7 +313,7 @@ class ConnectionHandler: if self.asr is None: self.asr = self._asr if self.tts is None: - self.tts = self._tts + self.tts = DefaultTTS(self.config, delete_audio_file=True) # 使用事件循环运行异步方法 asyncio.run_coroutine_threadsafe(self.tts.open_audio_channels(self), self.loop) diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 49ba3d07..14e1bc5a 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -5,6 +5,7 @@ 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.reportHandle import enqueue_asr_report +from core.handle.sendAudioHandle import SentenceType from core.utils.util import audio_to_data TAG = __name__ @@ -112,7 +113,7 @@ async def max_out_size(conn): 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((opus_packets, text, 0)) + conn.tts.tts_audio_queue.put((SentenceType.LAST, opus_packets, text)) conn.close_after_chat = True @@ -131,7 +132,7 @@ 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((opus_packets, text, 0)) + conn.tts.tts_audio_queue.put((SentenceType.FIRST, opus_packets, text)) # 逐个播放数字 for i in range(6): # 确保只播放6位数字 @@ -139,13 +140,14 @@ async def check_bind_device(conn): 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((num_packets, None, i + 1)) + 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: 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((opus_packets, text, 0)) + 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 f3561efd..b32d63ab 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -3,6 +3,7 @@ import asyncio import time from core.providers.tts.dto.dto import SentenceType from core.utils.util import get_string_no_punctuation_or_emoji, analyze_emotion +from loguru import logger TAG = __name__ @@ -49,7 +50,7 @@ async def sendAudioMessage(conn, sentenceType, audios, text): await send_tts_message(conn, "sentence_start", text) - await sendAudio(conn, audios, True) + await sendAudio(conn, audios, False) await send_tts_message(conn, "sentence_end", text) diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 44c2aacb..2ae01acc 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -11,6 +11,7 @@ from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType, ContentType from abc import ABC, abstractmethod from core.utils.tts import MarkdownCleaner from core.utils.util import audio_to_data +import traceback TAG = __name__ logger = setup_logging() @@ -24,8 +25,6 @@ class TTSProviderBase(ABC): self.output_file = config.get("output_dir") self.tts_text_queue = queue.Queue() self.tts_audio_queue = queue.Queue() - self.stop_event = threading.Event() - self.loop = asyncio.get_event_loop() self.tts_text_buff = [] self.punctuations = ( @@ -148,9 +147,9 @@ class TTSProviderBase(ABC): # 这里默认是非流式的处理方式 # 流式处理方式请在子类中重写 def _tts_text_priority_thread(self): - while not self.stop_event.is_set(): + while not self.conn.stop_event.is_set(): try: - message = self.tts_text_queue.get() + message = self.tts_text_queue.get(timeout=1) if message.sentence_type == SentenceType.FIRST: # 初始化参数 self.tts_stop_request = False @@ -162,30 +161,35 @@ class TTSProviderBase(ABC): segment_text = self._get_segment_text() if segment_text: tts_file = self.to_tts(segment_text) - audio_datas = self._process_audio_file(tts_file) - self.tts_audio_queue.put( - (message.sentence_type, audio_datas, 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) + ) elif ContentType.FILE == message.content_type: self._process_remaining_text() - tts_file = message.content_file - audio_datas = self._process_audio_file(tts_file) - self.tts_audio_queue.put( - (message.sentence_type, audio_datas, message.content_detail) - ) + 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) + ) + if message.sentence_type == SentenceType.LAST: self._process_remaining_text() - self.tts_audio_queue.put( (message.sentence_type, [], message.content_detail) ) + except queue.Empty: + continue except Exception as e: - logger.bind(tag=TAG).error(f"Failed to process TTS text: {e}") + logger.bind(tag=TAG).error( + f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" + ) def _audio_play_priority_thread(self): - while not self.stop_event.is_set(): + while not self.conn.stop_event.is_set(): text = None try: try: @@ -193,14 +197,17 @@ class TTSProviderBase(ABC): timeout=1 ) except queue.Empty: - if self.stop_event.is_set(): + if self.conn.stop_event.is_set(): break continue future = asyncio.run_coroutine_threadsafe( sendAudioMessage(self.conn, sentence_type, audio_datas, text), - self.loop, + self.conn.loop, + ) + result = future.result() + logger.bind(tag=TAG).info( + f"audio_play_priority result: {text} {result}" ) - future.result() except Exception as e: logger.bind(tag=TAG).error( f"audio_play_priority priority_thread: {text} {e}" diff --git a/main/xiaozhi-server/core/providers/tts/huoshan.py b/main/xiaozhi-server/core/providers/tts/huoshan.py index a5bcfd76..1b58ec00 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan.py @@ -145,7 +145,6 @@ class TTSProvider(TTSProviderBase): self.authorization = config.get("authorization") self.speaker = config.get("speaker") self.header = {"Authorization": f"{self.authorization}{self.access_token}"} - self.stop_event_response = threading.Event() self.enable_two_way = True self.start_connection_flag = False self.tts_text = "" @@ -162,7 +161,7 @@ class TTSProvider(TTSProviderBase): self.ws_url, additional_headers=ws_header, max_size=1000000000 ) tts_priority = threading.Thread( - target=self._start_monitor_tts_response_thread(), daemon=True + target=self._start_monitor_tts_response_thread, daemon=True ) tts_priority.start() @@ -174,15 +173,15 @@ class TTSProvider(TTSProviderBase): def to_tts(self, text): future = asyncio.run_coroutine_threadsafe( - self.start_session(self.conn.sentence_id), loop=self.loop + self.start_session(self.conn.sentence_id), loop=self.conn.loop ) future.result() future = asyncio.run_coroutine_threadsafe( - self.text_to_speak(text, None), loop=self.loop + self.text_to_speak(text, None), loop=self.conn.loop ) future.result() future = asyncio.run_coroutine_threadsafe( - self.finish_session(self.conn.sentence_id), loop=self.loop + self.finish_session(self.conn.sentence_id), loop=self.conn.loop ) future.result() return self.interface_type.value @@ -332,7 +331,6 @@ class TTSProvider(TTSProviderBase): return async def start_session(self, session_id): - self.stop_event_response.clear() header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, @@ -345,7 +343,6 @@ class TTSProvider(TTSProviderBase): async def finish_session(self, session_id): logger.bind(tag=TAG).info(f"会话结束～～{session_id}") - self.stop_event_response.set() header = Header( message_type=FULL_CLIENT_REQUEST, message_type_specific_flags=MsgTypeFlagWithEvent, @@ -380,11 +377,11 @@ class TTSProvider(TTSProviderBase): def _start_monitor_tts_response_thread(self): # 初始化链接 asyncio.run_coroutine_threadsafe( - self._start_monitor_tts_response(), loop=self.loop + self._start_monitor_tts_response(), loop=self.conn.loop ) async def _start_monitor_tts_response(self): - while not self.stop_event.is_set(): + while not self.conn.stop_event.is_set(): try: msg = await self.ws.recv() # 确保 `recv()` 运行在同一个 event loop res = self.parser_response(msg) diff --git a/main/xiaozhi-server/core/websocket_server.py b/main/xiaozhi-server/core/websocket_server.py index 578e1a01..81974b61 100644 --- a/main/xiaozhi-server/core/websocket_server.py +++ b/main/xiaozhi-server/core/websocket_server.py @@ -2,7 +2,6 @@ import asyncio import websockets from config.logger import setup_logging from core.connection import ConnectionHandler -from core.providers.tts.default import DefaultTTS from core.utils.util import initialize_modules, check_vad_update, check_asr_update from config.config_loader import get_config_from_api @@ -31,9 +30,6 @@ class WebSocketServer: self._intent = modules["intent"] if "intent" in modules else None self._memory = modules["memory"] if "memory" in modules else None - if self._tts is None: - self._tts = DefaultTTS(self.config, delete_audio_file=True) - self.active_connections = set() async def start(self):