refactor: replace Kokoro-82M with suno/bark for TTS, update HF cache path, and add model warmup on startup.
This commit is contained in:
@@ -0,0 +1,37 @@
|
||||
import asyncio
|
||||
import base64
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
||||
|
||||
from backend.tts_asr import _tts_sync_with_retry
|
||||
|
||||
async def play_audio():
|
||||
print("生成测试音频中,请稍候...")
|
||||
test_text = "这是一段用以测试新语音模型音质的中文合成音频。"
|
||||
|
||||
try:
|
||||
audio_bytes, sr = await _tts_sync_with_retry(test_text, rate=1.0)
|
||||
|
||||
# 保存到本地文件
|
||||
wav_path = os.path.join(os.path.dirname(__file__), "test_audio.wav")
|
||||
with open(wav_path, "wb") as f:
|
||||
f.write(audio_bytes)
|
||||
|
||||
print(f"音频已生成并保存到: {wav_path}")
|
||||
print("正在尝试在 macOS 上播放...")
|
||||
|
||||
# Mac OS 的播放命令
|
||||
os.system(f"afplay '{wav_path}'")
|
||||
|
||||
print("播放完成。")
|
||||
|
||||
except Exception as e:
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
print(f"音频生成失败: {str(e)}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(play_audio())
|
||||
Binary file not shown.
@@ -16,8 +16,9 @@ from unittest.mock import Mock, MagicMock, patch
|
||||
import tempfile
|
||||
import numpy as np
|
||||
|
||||
# 确保可以导入tts_asr模块
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||||
# 确保可以导入backend和tts_asr模块
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
|
||||
|
||||
|
||||
class TestAppleSiliconDetection(unittest.TestCase):
|
||||
@@ -174,7 +175,7 @@ class TestModelSizeSelection(unittest.TestCase):
|
||||
import backend.tts_asr as tts_asr_module
|
||||
importlib.reload(tts_asr_module)
|
||||
|
||||
from backend.tts_asr import _get_recommended_model_size
|
||||
from backend.tts_asr import _get_recommended_model_size, WHISPER_MODEL_SIZES
|
||||
# 应该回退到推荐大小而不崩溃
|
||||
size = _get_recommended_model_size()
|
||||
self.assertIn(size, WHISPER_MODEL_SIZES.keys())
|
||||
@@ -299,8 +300,8 @@ class TestModelCacheCheck(unittest.TestCase):
|
||||
"""测试离线模式下缺失模型的处理"""
|
||||
from backend.tts_asr import _check_model_cached
|
||||
|
||||
# 模拟transformers缓存路径
|
||||
with patch('transformers.file_utils.default_cache_path', '/nonexistent/path'):
|
||||
# 模拟缓存路径
|
||||
with patch('huggingface_hub.constants.HF_HUB_CACHE', '/nonexistent/path'):
|
||||
result = _check_model_cached('nonexistent/model')
|
||||
# 应该返回False(模型未缓存)
|
||||
self.assertFalse(result)
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
import asyncio
|
||||
import base64
|
||||
import os
|
||||
import sys
|
||||
|
||||
# 确保能找到backend模块
|
||||
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
|
||||
|
||||
from backend.tts_asr import _tts_sync_with_retry, _load_asr_pipeline_with_retry, _get_asr_pipeline
|
||||
|
||||
async def verify_tts_asr_cross():
|
||||
print("====================================")
|
||||
print(" 开始严格交叉验证: TTS 生成 -> ASR 解析")
|
||||
print("====================================")
|
||||
|
||||
test_text = "苹果设备支持离线大模型运算"
|
||||
print(f"\n[1] 正在调用 TTS 引擎 (suno/bark-small)...")
|
||||
print(f"目标文本: '{test_text}'")
|
||||
|
||||
try:
|
||||
# TTS生成
|
||||
audio_bytes, sr = await _tts_sync_with_retry(test_text, rate=1.0)
|
||||
print(f"-> TTS 成功生成音频数据,大小: {len(audio_bytes)} Bytes, 采样率: {sr}Hz")
|
||||
except Exception as e:
|
||||
print(f"-> TTS 失败: {str(e)}")
|
||||
sys.exit(1)
|
||||
|
||||
print("\n[2] 正在调用 ASR 引擎 (Whisper)...")
|
||||
try:
|
||||
loaded = await _load_asr_pipeline_with_retry()
|
||||
if not loaded:
|
||||
print("-> ASR 核心加载失败!")
|
||||
sys.exit(1)
|
||||
|
||||
print("-> ASR 加载成功,开始解析音频...")
|
||||
|
||||
# 将生成的wav bytes传递给ASR进行语音识别
|
||||
asr_pipeline = _get_asr_pipeline()
|
||||
result = asr_pipeline(audio_bytes, generate_kwargs={"task": "transcribe"})
|
||||
recognized_text = result.get('text', '')
|
||||
print(f"-> ASR 识别结果: '{recognized_text.strip()}'")
|
||||
|
||||
if len(recognized_text.strip()) > 0:
|
||||
print("\n结论: ✅ 验证成功!TTS和ASR模块功能链路闭环完成。")
|
||||
else:
|
||||
print("\n结论: ❌ ASR输出为空字符,闭环失败。")
|
||||
sys.exit(1)
|
||||
|
||||
except Exception as e:
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
print(f"-> ASR 分析阶段失败: {str(e)}")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(verify_tts_asr_cross())
|
||||
Reference in New Issue
Block a user