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

update: 优化音频播放 文本发送

Browse Source
This commit is contained in:
Sakura-RanChen
2025-07-11 16:36:47 +08:00
parent 23cd453af8
commit 8e5d933745
1 changed files with 49 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/providers/tts/aliyun_stream.py
+49 -76
View File
@@ -123,6 +123,11 @@ class TTSProvider(TTSProviderBase):
self.ws = None self.ws = None
self._monitor_task = None self._monitor_task = None
# 文本获取
self.sentence_queue = queue.Queue()
self.text_buffer = ""
self.sentence_end_chars = {'.', '。', '!', '！', '?', '？', '\n', "~", "；", ";", ":", "：", " ", "，", ","}
# 创建Opus编码器 # 创建Opus编码器
self.opus_encoder = opus_encoder_utils.OpusEncoderUtils( self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
sample_rate=16000, channels=1, frame_size_ms=60 sample_rate=16000, channels=1, frame_size_ms=60
@@ -176,6 +181,7 @@ class TTSProvider(TTSProviderBase):
logger.bind(tag=TAG).warning("Token已过期，正在自动刷新...") logger.bind(tag=TAG).warning("Token已过期，正在自动刷新...")
self._refresh_token() self._refresh_token()
if self.ws: if self.ws:
# 10秒内才可以复用,适合连续对话
logger.bind(tag=TAG).info(f"使用已有链接...") logger.bind(tag=TAG).info(f"使用已有链接...")
return self.ws return self.ws
logger.bind(tag=TAG).info("开始建立新连接...") logger.bind(tag=TAG).info("开始建立新连接...")
@@ -239,6 +245,10 @@ class TTSProvider(TTSProviderBase):
logger.bind(tag=TAG).debug( logger.bind(tag=TAG).debug(
f"开始发送TTS文本: {message.content_detail}" f"开始发送TTS文本: {message.content_detail}"
) )
self.text_buffer += message.content_detail
if message.content_detail in self.sentence_end_chars and len(self.text_buffer) > 6:
self.sentence_queue.put(self.text_buffer)
self.text_buffer = ""
future = asyncio.run_coroutine_threadsafe( future = asyncio.run_coroutine_threadsafe(
self.text_to_speak(message.content_detail, None), self.text_to_speak(message.content_detail, None),
loop=self.conn.loop, loop=self.conn.loop,
@@ -263,6 +273,8 @@ class TTSProvider(TTSProviderBase):
if message.sentence_type == SentenceType.LAST: if message.sentence_type == SentenceType.LAST:
try: try:
logger.bind(tag=TAG).info("开始结束TTS会话...") logger.bind(tag=TAG).info("开始结束TTS会话...")
self.sentence_queue.put(self.text_buffer)
print(list(self.sentence_queue.queue))
future = asyncio.run_coroutine_threadsafe( future = asyncio.run_coroutine_threadsafe(
self.finish_session(self.conn.sentence_id), self.finish_session(self.conn.sentence_id),
loop=self.conn.loop, loop=self.conn.loop,
@@ -315,9 +327,9 @@ class TTSProvider(TTSProviderBase):
try: try:
# 会话开始时检测上个会话的监听状态 # 会话开始时检测上个会话的监听状态
if( if(
self._monitor_task is not None self._monitor_task is not None
and isinstance(self._monitor_task, Task) and isinstance(self._monitor_task, Task)
and not self._monitor_task.done() and not self._monitor_task.done()
): ):
logger.bind(tag=TAG).info("检测到未完成的上个会话，关闭监听任务和连接...") logger.bind(tag=TAG).info("检测到未完成的上个会话，关闭监听任务和连接...")
await self.close() await self.close()
@@ -343,6 +355,7 @@ class TTSProvider(TTSProviderBase):
"volume": self.volume, "volume": self.volume,
"speech_rate": self.speech_rate, "speech_rate": self.speech_rate,
"pitch_rate": self.pitch_rate, "pitch_rate": self.pitch_rate,
"enable_subtitle": True
} }
} }
await self.ws.send(json.dumps(start_request)) await self.ws.send(json.dumps(start_request))
@@ -405,9 +418,9 @@ class TTSProvider(TTSProviderBase):
async def _start_monitor_tts_response(self): async def _start_monitor_tts_response(self):
"""监听TTS响应""" """监听TTS响应"""
opus_datas_cache = [] opus_datas_cache = []
current_sentence_buffer = bytearray() is_first_sentence = True
segment_count = 0 first_sentence_segment_count = 0 # 添加计数器
current_sentence_parts = [] text_buff = ""
try: try:
session_finished = False # 标记会话是否正常结束 session_finished = False # 标记会话是否正常结束
while not self.conn.stop_event.is_set(): while not self.conn.stop_event.is_set():
@@ -422,91 +435,51 @@ class TTSProvider(TTSProviderBase):
data = json.loads(msg) data = json.loads(msg)
header = data.get("header", {}) header = data.get("header", {})
event_name = header.get("name") event_name = header.get("name")
if event_name == "SynthesisStarted": if event_name == "SynthesisStarted":
logger.bind(tag=TAG).debug("TTS合成已启动") logger.bind(tag=TAG).debug("TTS合成已启动")
elif event_name == "SentenceBegin": elif event_name == "SentenceBegin":
logger.bind(tag=TAG).debug(f"句子语音生成开始") try:
current_sentence_buffer = bytearray() text_buff = self.sentence_queue.get_nowait()
segment_count = 0 except queue.Empty:
self.tts_audio_queue.put( text_buff = ""
(SentenceType.FIRST, [], "") logger.bind(tag=TAG).debug(f"句子语音生成开始: {text_buff}")
) opus_datas_cache = []
self.tts_audio_queue.put((SentenceType.FIRST, [], text_buff))
elif event_name == "SentenceSynthesis":
payload = data.get("payload", {})
subtitles = payload.get("subtitles", [])
# 收集所有字幕片段
for sub in subtitles:
text = sub.get("text", "")
if text:
current_sentence_parts.append(text)
elif event_name == "SentenceEnd": elif event_name == "SentenceEnd":
logger.bind(tag=TAG).info(f"句子语音生成成功: {''.join(current_sentence_parts)}") logger.bind(tag=TAG).info(f"句子语音生成成功： {text_buff}")
text_buff = ""
if current_sentence_buffer: if not is_first_sentence or first_sentence_segment_count > 10:
opus_datas = self.opus_encoder.encode_pcm_to_opus( # 发送缓存的数据
bytes(current_sentence_buffer), end_of_stream=True
)
if opus_datas:
if segment_count < 10:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas, None)
)
else:
opus_datas_cache.extend(opus_datas)
if segment_count >= 10 and opus_datas_cache:
self.tts_audio_queue.put( self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_datas_cache, None) (SentenceType.MIDDLE, opus_datas_cache, None)
) )
opus_datas_cache = [] # 第一句话结束后，将标志设置为False
current_sentence_buffer = bytearray() is_first_sentence = False
elif event_name == "SynthesisCompleted": elif event_name == "SynthesisCompleted":
logger.bind(tag=TAG).debug("会话结束") logger.bind(tag=TAG).debug(f"会话结束～～")
self._process_before_stop_play_files() self._process_before_stop_play_files()
session_finished = True session_finished = True
break break
except json.JSONDecodeError: except json.JSONDecodeError:
logger.bind(tag=TAG).warning("收到无效的JSON消息") logger.bind(tag=TAG).warning("收到无效的JSON消息")
# 二进制消息（音频数据） # 二进制消息（音频数据）
elif isinstance(msg, bytes): elif isinstance(msg, (bytes, bytearray)):
# 将音频数据添加到当前句子的缓冲区 logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
current_sentence_buffer.extend(msg) opus_datas = self.opus_encoder.encode_pcm_to_opus(msg, False)
logger.bind(tag=TAG).debug(
# 计算每帧的字节数（60ms） f"推送数据到队列里面帧数～～{len(opus_datas)}"
frame_bytes = int(
self.opus_encoder.sample_rate
* self.opus_encoder.channels
* self.opus_encoder.frame_size_ms
/ 1000
* 2 # 16-bit = 2 bytes
) )
if is_first_sentence:
# 处理完整的音频帧 first_sentence_segment_count += 1
while len(current_sentence_buffer) >= frame_bytes: if first_sentence_segment_count <= 6:
# 取出一帧数据 self.tts_audio_queue.put(
frame = bytes(current_sentence_buffer[:frame_bytes]) (SentenceType.MIDDLE, opus_datas, None)
del current_sentence_buffer[:frame_bytes] )
else:
# 编码为Opus opus_datas_cache = opus_datas_cache + opus_datas
opus_packets = self.opus_encoder.encode_pcm_to_opus(frame, False) else:
# 后续句子缓存
if opus_packets: opus_datas_cache = opus_datas_cache + opus_datas
# 前10个片段直接发送，后续片段缓存
if segment_count < 10:
self.tts_audio_queue.put(
(SentenceType.MIDDLE, opus_packets, None)
)
segment_count += 1
else:
opus_datas_cache.extend(opus_packets)
except websockets.ConnectionClosed: except websockets.ConnectionClosed:
logger.bind(tag=TAG).warning("WebSocket连接已关闭") logger.bind(tag=TAG).warning("WebSocket连接已关闭")