Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-22 07:03:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

update:优化线程

Browse Source
This commit is contained in:
hrz
2025-05-26 10:44:35 +08:00
parent 24526ad206
commit 5be65216e2
6 changed files with 42 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/connection.py
+2 -1
View File
@@ -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)
main/xiaozhi-server/core/handle/receiveAudioHandle.py
+6 -4
View File
@@ -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))
main/xiaozhi-server/core/handle/sendAudioHandle.py
+2 -1
View File
@@ -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)
main/xiaozhi-server/core/providers/tts/base.py
+26 -19
View File
@@ -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}"
main/xiaozhi-server/core/providers/tts/huoshan.py
+6 -9
View File
@@ -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)
main/xiaozhi-server/core/websocket_server.py
-4
View File
@@ -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):