89 lines
2.3 KiB
Python
89 lines
2.3 KiB
Python
"""
|
|
TTS 双向流测试 - 边发送文本边接收音频
|
|
"""
|
|
|
|
import time
|
|
from pathlib import Path
|
|
from src.Module.tts.tts import StreamingTTS
|
|
|
|
|
|
def simulate_llm_output():
|
|
"""
|
|
模拟 LLM 流式输出
|
|
一句话分成多个部分逐步发送
|
|
"""
|
|
text_chunks = [
|
|
"你好，",
|
|
"我是通义千问",
|
|
"语音合成系统。",
|
|
"现在正在进行",
|
|
"双向流测试！"
|
|
]
|
|
|
|
for chunk in text_chunks:
|
|
print(f" [发送文本] -> {chunk}")
|
|
yield chunk
|
|
time.sleep(0.1) # 模拟 LLM 生成延迟
|
|
|
|
|
|
def test_bidirectional_stream():
|
|
"""测试双向流式合成"""
|
|
print("=" * 60)
|
|
print(" TTS 双向流测试")
|
|
print("=" * 60)
|
|
print("\n一句话分为多个部分发送，同时接收音频\n")
|
|
|
|
tts = StreamingTTS(voice='Cherry')
|
|
|
|
total_bytes = 0
|
|
chunk_count = 0
|
|
audio_data = bytearray()
|
|
|
|
print("开始双向流传输...\n")
|
|
|
|
# 使用 stream_from_generator 实现双向流
|
|
for audio_chunk in tts.stream_from_generator(simulate_llm_output()):
|
|
if audio_chunk.error:
|
|
print(f" [错误] {audio_chunk.error}")
|
|
break
|
|
|
|
if audio_chunk.data:
|
|
chunk_count += 1
|
|
total_bytes += len(audio_chunk.data)
|
|
audio_data.extend(audio_chunk.data)
|
|
print(
|
|
f" [收到音频] Chunk {
|
|
chunk_count:02d}: {
|
|
len(
|
|
audio_chunk.data):5d} 字节 | 累计: {
|
|
total_bytes:6d} 字节")
|
|
|
|
if audio_chunk.is_final:
|
|
print("\n传输完成!")
|
|
break
|
|
|
|
# 保存音频
|
|
output_dir = Path(__file__).parent / 'output'
|
|
output_dir.mkdir(exist_ok=True)
|
|
wav_file = output_dir / 'bidirectional_test.wav'
|
|
|
|
import wave
|
|
with wave.open(str(wav_file), 'wb') as wav:
|
|
wav.setnchannels(1)
|
|
wav.setsampwidth(2)
|
|
wav.setframerate(24000)
|
|
wav.writeframes(audio_data)
|
|
|
|
print(f"\n结果统计:")
|
|
print(f" 音频块: {chunk_count} 个")
|
|
print(f" 总字节: {total_bytes} 字节")
|
|
print(f" 音频时长: {total_bytes / (24000 * 2):.2f} 秒")
|
|
print(f" 已保存: {wav_file}")
|
|
|
|
return chunk_count > 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
success = test_bidirectional_stream()
|
|
print(f"\n测试结果: {'✓ 通过' if success else '✗ 失败'}")
|