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

Merge pull request #2044 from xinnan-tech/py_test_audio

Browse Source 
Py test audio
This commit is contained in:
hrz
2025-08-15 17:59:21 +08:00
committed by GitHub
parent c395c31e46 d2f29f335e
commit e2324d3c37
17 changed files with 440 additions and 1022 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/providers/tts/aliyun_stream.py
+12 -184
View File
@@ -216,6 +216,7 @@ class TTSProvider(TTSProviderBase):
if message.sentence_type == SentenceType.FIRST:
self.conn.client_abort = False
self.reset_flow_controller()
if self.conn.client_abort:
logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
@@ -268,11 +269,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 +419,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 +439,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 +459,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
@@ -512,142 +479,3 @@ class TTSProvider(TTSProviderBase):
finally:
self._monitor_task = None
def to_tts(self, text: str) -> list:
"""非流式TTS处理，用于测试及保存音频文件的场景"""
try:
# 创建新的事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 生成会话ID
session_id = uuid.uuid4().hex
# 存储音频数据
audio_data = []
async def _generate_audio():
# 刷新Token（如果需要）
if self._is_token_expired():
self._refresh_token()
# 建立WebSocket连接
ws = await websockets.connect(
self.ws_url,
additional_headers={"X-NLS-Token": self.token},
ping_interval=30,
ping_timeout=10,
close_timeout=10,
)
try:
# 发送StartSynthesis请求
start_message_id = str(uuid.uuid4().hex)
start_request = {
"header": {
"message_id": start_message_id,
"task_id": session_id,
"namespace": "FlowingSpeechSynthesizer",
"name": "StartSynthesis",
"appkey": self.appkey,
},
"payload": {
"voice": self.voice,
"format": self.format,
"sample_rate": self.sample_rate,
"volume": self.volume,
"speech_rate": self.speech_rate,
"pitch_rate": self.pitch_rate,
"enable_subtitle": True,
},
}
await ws.send(json.dumps(start_request))
# 等待SynthesisStarted响应
synthesis_started = False
while not synthesis_started:
msg = await ws.recv()
if isinstance(msg, str):
data = json.loads(msg)
header = data.get("header", {})
if header.get("name") == "SynthesisStarted":
synthesis_started = True
logger.bind(tag=TAG).debug("TTS合成已启动")
elif header.get("name") == "TaskFailed":
error_info = data.get("payload", {}).get(
"error_info", {}
)
error_code = error_info.get("error_code")
error_message = error_info.get(
"error_message", "未知错误"
)
raise Exception(
f"启动合成失败: {error_code} - {error_message}"
)
# 发送文本合成请求
filtered_text = MarkdownCleaner.clean_markdown(text)
run_message_id = str(uuid.uuid4().hex)
run_request = {
"header": {
"message_id": run_message_id,
"task_id": session_id,
"namespace": "FlowingSpeechSynthesizer",
"name": "RunSynthesis",
"appkey": self.appkey,
},
"payload": {"text": filtered_text},
}
await ws.send(json.dumps(run_request))
# 发送停止合成请求
stop_message_id = str(uuid.uuid4().hex)
stop_request = {
"header": {
"message_id": stop_message_id,
"task_id": session_id,
"namespace": "FlowingSpeechSynthesizer",
"name": "StopSynthesis",
"appkey": self.appkey,
}
}
await ws.send(json.dumps(stop_request))
# 接收音频数据
synthesis_completed = False
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
)
audio_data.extend(opus_frames)
elif isinstance(msg, str):
data = json.loads(msg)
header = data.get("header", {})
event_name = header.get("name")
if event_name == "SynthesisCompleted":
synthesis_completed = True
logger.bind(tag=TAG).debug("TTS合成完成")
elif event_name == "TaskFailed":
error_info = data.get("payload", {}).get(
"error_info", {}
)
error_code = error_info.get("error_code")
error_message = error_info.get(
"error_message", "未知错误"
)
raise Exception(
f"合成失败: {error_code} - {error_message}"
)
finally:
try:
await ws.close()
except:
pass
loop.run_until_complete(_generate_audio())
loop.close()
return audio_data
except Exception as e:
logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
return []
main/xiaozhi-server/core/providers/tts/base.py
+137 -65
View File
@@ -4,12 +4,15 @@ import queue
import uuid
import asyncio
import threading
from typing import Callable, Any
from core.utils import p3
import time
from datetime import datetime
from core.utils import textUtils
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.audio_flow_control import FlowControlConfig
from core.utils.util import audio_bytes_to_data_stream, audio_to_data_stream
from core.utils.tts import MarkdownCleaner
from core.utils.output_counter import add_device_output
from core.handle.reportHandle import enqueue_tts_report
@@ -50,11 +53,9 @@ class TTSProviderBase(ABC):
"；",
";",
"：",
"~",
)
self.first_sentence_punctuations = (
"，",
"～",
"~",
"、",
",",
@@ -70,6 +71,7 @@ class TTSProviderBase(ABC):
self.tts_stop_request = False
self.processed_chars = 0
self.is_first_sentence = True
self.flow_controller = FlowControlConfig.create_flow_controller()
def generate_filename(self, extension=".wav"):
return os.path.join(
@@ -77,7 +79,20 @@ class TTSProviderBase(ABC):
f"tts-{datetime.now().date()}@{uuid.uuid4().hex}{extension}",
)
def to_tts(self, text):
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:
@@ -86,10 +101,13 @@ 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
self.tts_audio_queue.put(
(SentenceType.FIRST, None, text)
)
return audio_datas
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:
@@ -129,8 +147,10 @@ class TTSProviderBase(ABC):
logger.bind(tag=TAG).error(
f"语音生成失败: {text}，请检查网络或服务是否正常"
)
return tmp_file
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
@@ -139,13 +159,13 @@ 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,
@@ -208,34 +228,19 @@ class TTSProviderBase(ABC):
self.tts_text_buff = []
self.is_first_sentence = True
self.tts_audio_first_sentence = True
self.reset_flow_controller()
elif ContentType.TEXT == message.content_type:
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,30 +254,112 @@ 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
)
sentence_type, audio_datas, text = self.tts_audio_queue.get(timeout=0.1)
except queue.Empty:
if self.conn.stop_event.is_set():
break
continue
future = asyncio.run_coroutine_threadsafe(
sendAudioMessage(self.conn, sentence_type, audio_datas, text),
self.conn.loop,
)
future.result()
if self.conn.client_abort:
logger.bind(tag=TAG).debug("收到打断信号，跳过当前音频数据")
# 打断时丢弃未上报的音频数据
enqueue_text, enqueue_audio = None, []
continue
# 收到下一个文本开始或会话结束时进行上报
if sentence_type is not SentenceType.MIDDLE:
# 上报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):
frame_count = 1 # 单个字节流作为一帧
enqueue_audio.append(audio_datas)
else:
frame_count = 0
# 记录输出和报告
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)
# 流控检查
if frame_count > 0:
max_wait_time = FlowControlConfig.DEFAULT_MAX_WAIT_TIME
wait_start_time = time.time()
retry_interval = FlowControlConfig.DEFAULT_RETRY_INTERVAL
while not self.flow_controller.can_send_frames(frame_count):
# 检查是否超时或需要停止
if (time.time() - wait_start_time > max_wait_time or
self.conn.stop_event.is_set() or
self.conn.client_abort):
logger.bind(tag=TAG).debug("流控等待超时或收到停止信号，跳过音频发送")
break
# 短暂等待后重试
time.sleep(retry_interval)
else:
# 可以发送，记录发送的帧数
self.flow_controller.record_sent_frames(frame_count)
# 发送音频
future = asyncio.run_coroutine_threadsafe(
self._send_audio_with_flow_control(sentence_type, audio_datas, text),
self.conn.loop,
)
future.result()
# 输出流控状态（调试用）
# status = self.flow_controller.get_status()
# logger.bind(tag=TAG).debug(
# f"流控状态: 缓冲区使用率={status['buffer_usage_percent']:.1f}%, "
# f"可用令牌={status['available_tokens']}..."
# f"发送帧数={status['sent_frames']}..."
# f"消费帧数={status['consumed_frames']}..."
# f"代播放帧数={status['sent_frames'] - status['consumed_frames']}..."
# )
else:
# 没有音频数据，直接发送
future = asyncio.run_coroutine_threadsafe(
self._send_audio_with_flow_control(sentence_type, audio_datas, text),
self.conn.loop,
)
future.result()
except Exception as e:
logger.bind(tag=TAG).error(
f"audio_play_priority priority_thread: {text} {e}"
f"audio_play_priority_thread: {text} {e}"
)
async def _send_audio_with_flow_control(self, sentence_type, audio_datas, text):
"""
带流控的音频发送方法 模拟设备消费音频帧的过程
实际应用中应该根据设备反馈来更新消费情况
"""
await sendAudioMessage(self.conn, sentence_type, audio_datas, text)
# 模拟设备消费（实际应用中应该从设备获取反馈）防止音字不同步
if isinstance(audio_datas, bytes):
# 模拟设备播放延迟（60ms per frame）, 实际情况可以低一点（50ms），增加使用体验
await asyncio.sleep(0.055)
self.flow_controller.update_device_consumption(1)
# 在类中添加流控制器重置方法
def reset_flow_controller(self):
"""重置流控制器状态，通常在新会话开始时调用"""
if hasattr(self, 'flow_controller'):
self.flow_controller.reset()
logger.bind(tag=TAG).info("流控制器状态已重置")
async def start_session(self, session_id):
pass
@@ -323,22 +410,19 @@ 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 +431,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 +438,7 @@ 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 +449,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
main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py
+6 -139
View File
@@ -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
@@ -212,6 +213,7 @@ class TTSProvider(TTSProviderBase):
if message.sentence_type == SentenceType.FIRST:
self.conn.client_abort = False
self.reset_flow_controller()
if self.conn.client_abort:
try:
@@ -266,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会话...")
@@ -428,9 +426,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 +446,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,110 +627,5 @@ 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 to_tts(self, text: str) -> list:
"""非流式生成音频数据，用于生成音频及测试场景
Args:
text: 要转换的文本
Returns:
list: 音频数据列表
"""
try:
# 创建事件循环
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# 生成会话ID
session_id = uuid.uuid4().__str__().replace("-", "")
# 存储音频数据
audio_data = []
async def _generate_audio():
# 创建新的WebSocket连接
ws_header = {
"X-Api-App-Key": self.appId,
"X-Api-Access-Key": self.access_token,
"X-Api-Resource-Id": self.resource_id,
"X-Api-Connect-Id": uuid.uuid4(),
}
ws = await websockets.connect(
self.ws_url, additional_headers=ws_header, max_size=1000000000
)
try:
# 启动会话
header = Header(
message_type=FULL_CLIENT_REQUEST,
message_type_specific_flags=MsgTypeFlagWithEvent,
serial_method=JSON,
).as_bytes()
optional = Optional(
event=EVENT_StartSession, sessionId=session_id
).as_bytes()
payload = self.get_payload_bytes(
event=EVENT_StartSession, speaker=self.voice
)
await self.send_event(ws, header, optional, payload)
# 发送文本
header = Header(
message_type=FULL_CLIENT_REQUEST,
message_type_specific_flags=MsgTypeFlagWithEvent,
serial_method=JSON,
).as_bytes()
optional = Optional(
event=EVENT_TaskRequest, sessionId=session_id
).as_bytes()
payload = self.get_payload_bytes(
event=EVENT_TaskRequest, text=text, speaker=self.voice
)
await self.send_event(ws, header, optional, payload)
# 发送结束会话请求
header = Header(
message_type=FULL_CLIENT_REQUEST,
message_type_specific_flags=MsgTypeFlagWithEvent,
serial_method=JSON,
).as_bytes()
optional = Optional(
event=EVENT_FinishSession, sessionId=session_id
).as_bytes()
payload = str.encode("{}")
await self.send_event(ws, header, optional, payload)
# 接收音频数据
while True:
msg = await ws.recv()
res = self.parser_response(msg)
if (
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)
elif res.optional.event == EVENT_SessionFinished:
break
finally:
# 清理资源
try:
await ws.close()
except:
pass
# 运行异步任务
loop.run_until_complete(_generate_audio())
loop.close()
return audio_data
except Exception as e:
logger.bind(tag=TAG).error(f"生成音频数据失败: {str(e)}")
return []
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)
main/xiaozhi-server/core/providers/tts/index_stream.py
+14 -98
View File
@@ -3,8 +3,6 @@ import queue
import asyncio
import traceback
import aiohttp
import requests
import time
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
from core.providers.tts.base import TTSProviderBase
@@ -27,15 +25,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,8 +44,8 @@ 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()
self.reset_flow_controller()
elif ContentType.TEXT == message.content_type:
self.tts_text_buff.append(message.content_detail)
segment_text = self._get_segment_text()
@@ -62,14 +58,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 +71,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 +136,6 @@ class TTSProvider(TTSProviderBase):
return
self.pcm_buffer.clear()
opus_datas_cache = []
self.tts_audio_queue.put((SentenceType.FIRST, [], text))
# 处理音频流数据
@@ -158,41 +149,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()
@@ -206,59 +178,3 @@ class TTSProvider(TTSProviderBase):
await super().close()
if hasattr(self, "opus_encoder"):
self.opus_encoder.close()
def to_tts(self, text: str) -> list:
"""非流式TTS处理，用于测试及保存音频文件的场景
Args:
text: 要转换的文本
Returns:
list: 返回opus编码后的音频数据列表
"""
start_time = time.time()
text = MarkdownCleaner.clean_markdown(text)
payload = {"text": text, "character": self.character}
try:
with requests.post(self.api_url, json=payload, 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))
opus = self.opus_encoder.encode_pcm_to_opus(
frame, end_of_stream=(i + frame_bytes >= len(pcm_data))
)
if opus:
opus_datas.extend(opus)
return opus_datas
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
return []
main/xiaozhi-server/core/providers/tts/linkerai.py
+12 -118
View File
@@ -3,8 +3,6 @@ import queue
import asyncio
import traceback
import aiohttp
import requests
import time
from config.logger import setup_logging
from core.utils.tts import MarkdownCleaner
from core.providers.tts.base import TTSProviderBase
@@ -24,23 +22,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,8 +41,8 @@ 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()
self.reset_flow_controller()
elif ContentType.TEXT == message.content_type:
self.tts_text_buff.append(message.content_detail)
segment_text = self._get_segment_text()
@@ -65,14 +55,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 +67,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:
@@ -176,8 +162,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 +178,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()
@@ -236,73 +200,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))
opus = self.opus_encoder.encode_pcm_to_opus(
frame, end_of_stream=(i + frame_bytes >= len(pcm_data))
)
if opus:
opus_datas.extend(opus)
return opus_datas
except Exception as e:
logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
return []