diff --git a/docs/performance_tester.md b/docs/performance_tester.md index abe57458..85987120 100644 --- a/docs/performance_tester.md +++ b/docs/performance_tester.md @@ -1,8 +1,8 @@ -# 语音识别、大语言模型、非流式语音合成、视觉模型的性能测试工具使用指南 +# 语音识别、大语言模型、非流式语音合成、流式语音合成、视觉模型的性能测试工具使用指南 1.在main/xiaozhi-server目录下创建data目录 2.在data目录下创建.config.yaml文件 -3.在.data/config.yaml中，写入你的语音识别、大语言模型、非流式语音合成、视觉模型的参数 +3.在.data/config.yaml中，写入你的语音识别、大语言模型、流式语音合成、视觉模型的参数 例如： ``` LLM: diff --git a/main/xiaozhi-server/performance_tester/performance_tester_asr.py b/main/xiaozhi-server/performance_tester/performance_tester_asr.py index e4d8c6d8..db359e28 100644 --- a/main/xiaozhi-server/performance_tester/performance_tester_asr.py +++ b/main/xiaozhi-server/performance_tester/performance_tester_asr.py @@ -2,35 +2,56 @@ import asyncio import logging import os import time -from typing import Dict - +import concurrent.futures +from typing import Dict, Optional import aiohttp from tabulate import tabulate from core.utils.asr import create_instance as create_stt_instance -from config.settings import load_config # 设置全局日志级别为WARNING，抑制INFO级别日志 logging.basicConfig(level=logging.WARNING) description = "语音识别模型性能测试" - class ASRPerformanceTester: def __init__(self): - self.config = load_config() + self.config = self._load_config_from_data_dir() self.test_wav_list = self._load_test_wav_files() self.results = {"stt": {}} - + # 调试日志 print(f"[DEBUG] 加载的ASR配置: {self.config.get('ASR', {})}") print(f"[DEBUG] 音频文件数量: {len(self.test_wav_list)}") + def _load_config_from_data_dir(self) -> Dict: + """从 data 目录加载所有 .config.yaml 文件的配置""" + config = {"ASR": {}} + data_dir = os.path.join(os.getcwd(), "data") + print(f"[DEBUG] 扫描配置文件目录: {data_dir}") + + for root, _, files in os.walk(data_dir): + for file in files: + if file.endswith(".config.yaml"): + file_path = os.path.join(root, file) + try: + with open(file_path, "r", encoding="utf-8") as f: + import yaml + file_config = yaml.safe_load(f) + # 兼容大小写的 ASR/asr 配置 + asr_config = file_config.get("ASR") or file_config.get("asr") + if asr_config: + config["ASR"].update(asr_config) + print(f"[DEBUG] 从 {file_path} 加载 ASR 配置成功") + except Exception as e: + print(f" 加载配置文件 {file_path} 失败: {str(e)}") + return config + def _load_test_wav_files(self) -> list: """加载测试用的音频文件（添加路径调试）""" wav_root = os.path.join(os.getcwd(), "config", "assets") print(f"[DEBUG] 音频文件目录: {wav_root}") test_wav_list = [] - + if os.path.exists(wav_root): file_list = os.listdir(wav_root) print(f"[DEBUG] 找到音频文件: {file_list}") @@ -43,18 +64,46 @@ class ASRPerformanceTester: print(f" 目录不存在: {wav_root}") return test_wav_list - async def _test_stt(self, stt_name: str, config: Dict) -> Dict: - """异步测试单个STT性能（跳过无效配置）""" + async def _test_single_audio(self, stt_name: str, stt, audio_data: bytes) -> Optional[float]: + """测试单个音频文件的性能""" try: + start_time = time.time() + text, _ = await stt.speech_to_text([audio_data], "1", stt.audio_format) + if text is None: + return None + + duration = time.time() - start_time + + # 检测0.000s的异常时间 + if abs(duration) < 0.001: # 小于1毫秒视为异常 + print(f"{stt_name} 检测到异常时间: {duration:.6f}s (视为错误)") + return None + + return duration + except Exception as e: + error_msg = str(e).lower() + if "502" in error_msg or "bad gateway" in error_msg: + print(f"{stt_name} 遇到502错误") + return None + return None + + async def _test_stt_with_timeout(self, stt_name: str, config: Dict) -> Dict: + """异步测试单个STT性能，带超时控制""" + try: + # 检查配置有效性 token_fields = ["access_token", "api_key", "token"] - # 忽略值为 "none" 的情况（需根据实际需求调整） if any( field in config - and str(config[field]).lower() in ["你的", "placeholder"] + and str(config[field]).lower() in ["你的", "placeholder", "none", "null", ""] for field in token_fields ): - print(f" STT {stt_name} 未配置access_token/api_key，已跳过") - return {"name": stt_name, "type": "stt", "errors": 1} + print(f" STT {stt_name} 未配置有效access_token/api_key，已跳过") + return { + "name": stt_name, + "type": "stt", + "errors": 1, + "error_type": "配置错误" + } module_type = config.get("type", stt_name) stt = create_stt_instance(module_type, config, delete_audio_file=True) @@ -62,56 +111,203 @@ class ASRPerformanceTester: print(f" 测试 STT: {stt_name}") - # 测试第一个音频文件 - text, _ = await stt.speech_to_text( - [self.test_wav_list[0]], "1", stt.audio_format - ) - if text is None: - print(f" {stt_name} 连接失败") - return {"name": stt_name, "type": "stt", "errors": 1} + # 使用线程池和超时控制 + loop = asyncio.get_event_loop() + + # 测试第一个音频文件作为连通性检查 + try: + with concurrent.futures.ThreadPoolExecutor() as executor: + future = executor.submit( + lambda: asyncio.run(self._test_single_audio(stt_name, stt, self.test_wav_list[0])) + ) + first_result = await asyncio.wait_for( + asyncio.wrap_future(future), timeout=10.0 + ) + + if first_result is None: + print(f" {stt_name} 连接失败") + return { + "name": stt_name, + "type": "stt", + "errors": 1, + "error_type": "网络错误" + } + except asyncio.TimeoutError: + print(f" {stt_name} 连接超时（10秒），跳过") + return { + "name": stt_name, + "type": "stt", + "errors": 1, + "error_type": "超时连接" + } + except Exception as e: + error_msg = str(e).lower() + if "502" in error_msg or "bad gateway" in error_msg: + print(f" {stt_name} 遇到502错误，跳过") + return { + "name": stt_name, + "type": "stt", + "errors": 1, + "error_type": "502网络错误" + } + print(f" {stt_name} 连接异常: {str(e)}") + return { + "name": stt_name, + "type": "stt", + "errors": 1, + "error_type": "网络错误" + } - # 全量测试 + # 全量测试，带超时控制 total_time = 0 + valid_tests = 0 test_count = len(self.test_wav_list) - for i, sentence in enumerate(self.test_wav_list, 1): - start = time.time() - text, _ = await stt.speech_to_text([sentence], "1", stt.audio_format) - duration = time.time() - start - total_time += duration - print(f" {stt_name} [{i}/{test_count}] 耗时: {duration:.2f}s") + + for i, audio_data in enumerate(self.test_wav_list, 1): + try: + with concurrent.futures.ThreadPoolExecutor() as executor: + future = executor.submit( + lambda: asyncio.run(self._test_single_audio(stt_name, stt, audio_data)) + ) + duration = await asyncio.wait_for( + asyncio.wrap_future(future), timeout=10.0 + ) + + if duration is not None and duration > 0.001: + total_time += duration + valid_tests += 1 + print(f" {stt_name} [{i}/{test_count}] 耗时: {duration:.2f}s") + else: + print(f" {stt_name} [{i}/{test_count}] 测试失败(含0.000s异常)") + + except asyncio.TimeoutError: + print(f" {stt_name} [{i}/{test_count}] 超时（10秒），跳过") + continue + except Exception as e: + error_msg = str(e).lower() + if "502" in error_msg or "bad gateway" in error_msg: + print(f" {stt_name} [{i}/{test_count}] 502错误，跳过") + return { + "name": stt_name, + "type": "stt", + "errors": 1, + "error_type": "502网络错误" + } + print(f" {stt_name} [{i}/{test_count}] 异常: {str(e)}") + continue + # 检查有效测试数量 + if valid_tests < test_count * 0.3: # 至少30%成功率 + print(f" {stt_name} 成功测试过少({valid_tests}/{test_count})，可能网络不稳定") + return { + "name": stt_name, + "type": "stt", + "errors": 1, + "error_type": "网络错误" + } + if valid_tests == 0: + return { + "name": stt_name, + "type": "stt", + "errors": 1, + "error_type": "网络错误" + } + + avg_time = total_time / valid_tests return { "name": stt_name, "type": "stt", - "avg_time": total_time / test_count, + "avg_time": avg_time, + "success_rate": f"{valid_tests}/{test_count}", "errors": 0, } + except Exception as e: + error_msg = str(e).lower() + if "502" in error_msg or "bad gateway" in error_msg: + error_type = "502网络错误" + elif "timeout" in error_msg: + error_type = "超时连接" + else: + error_type = "网络错误" print(f"⚠️ {stt_name} 测试失败: {str(e)}") - return {"name": stt_name, "type": "stt", "errors": 1} + return { + "name": stt_name, + "type": "stt", + "errors": 1, + "error_type": error_type + } def _print_results(self): - """打印测试结果""" - stt_table = [] + """打印测试结果，按响应时间排序""" + print("\n" + "=" * 50) + print("ASR 性能测试结果") + print("=" * 50) + + if not self.results.get("stt"): + print("没有可用的测试结果") + return + + headers = ["模型名称", "平均耗时(s)", "成功率", "状态"] + table_data = [] + + # 收集所有数据并分类 + valid_results = [] + error_results = [] + for name, data in self.results["stt"].items(): if data["errors"] == 0: - stt_table.append([name, f"{data['avg_time']:.3f}秒"]) + # 正常结果 + avg_time = f"{data['avg_time']:.3f}" + success_rate = data.get("success_rate", "N/A") + status = "✅ 正常" + + # 保存用于排序的值 + sort_key = data["avg_time"] + + valid_results.append({ + "name": name, + "avg_time": avg_time, + "success_rate": success_rate, + "status": status, + "sort_key": sort_key, + }) + else: + # 错误结果 + avg_time = "-" + success_rate = "0/N" + + # 获取具体错误类型 + error_type = data.get("error_type", "网络错误") + status = f"❌ {error_type}" + + error_results.append([name, avg_time, success_rate, status]) - if stt_table: - print("\nASR 性能排行:\n") - print( - tabulate( - stt_table, - headers=["模型名称", "平均耗时"], - tablefmt="github", - colalign=("left", "right"), - ) - ) - else: - print("\n 没有可用的ASR模块进行测试。") + # 按响应时间升序排序（从快到慢） + valid_results.sort(key=lambda x: x["sort_key"]) + + # 将排序后的有效结果转换为表格数据 + for result in valid_results: + table_data.append([ + result["name"], + result["avg_time"], + result["success_rate"], + result["status"], + ]) + + # 将错误结果添加到表格数据末尾 + table_data.extend(error_results) + + print(tabulate(table_data, headers=headers, tablefmt="grid")) + print("\n测试说明:") + print("- 超时控制：单个音频最大等待时间为10秒") + print("- 错误处理：自动跳过502错误、超时和网络异常的模型") + print("- 成功率：成功识别的音频数量/总测试音频数量") + print("- 排序规则：按平均耗时从快到慢排序，错误模型排最后") + print("\n测试完成！") async def run(self): - """执行全量异步测试""" + """执行全量异步测试""" print("开始筛选可用ASR模块...") if not self.config.get("ASR"): print("配置中未找到 ASR 模块") @@ -119,24 +315,33 @@ class ASRPerformanceTester: all_tasks = [] for stt_name, config in self.config["ASR"].items(): - print(f"[DEBUG] 检查 ASR 模块: {stt_name}, 配置: {config}") - all_tasks.append(self._test_stt(stt_name, config)) + # 检查配置有效性 + token_fields = ["access_token", "api_key", "token"] + if any( + field in config + and str(config[field]).lower() in ["你的", "placeholder", "none", "null", ""] + for field in token_fields + ): + print(f"ASR {stt_name} 未配置有效access_token/api_key，已跳过") + continue + + print(f"添加 ASR 测试任务: {stt_name}") + all_tasks.append(self._test_stt_with_timeout(stt_name, config)) if not all_tasks: print("没有可用的ASR模块进行测试。") return + print(f"\n找到 {len(all_tasks)} 个可用ASR模块") print("\n开始并发测试所有ASR模块...") all_results = await asyncio.gather(*all_tasks, return_exceptions=True) # 处理结果 for result in all_results: if isinstance(result, dict) and result.get("type") == "stt": - if result["errors"] == 0: - self.results["stt"][result["name"]] = result + self.results["stt"][result["name"]] = result # 打印结果 - print("\n测试完成") self._print_results() @@ -146,4 +351,4 @@ async def main(): if __name__ == "__main__": - asyncio.run(main()) + asyncio.run(main()) \ No newline at end of file diff --git a/main/xiaozhi-server/performance_tester/performance_tester_stream_tts.py b/main/xiaozhi-server/performance_tester/performance_tester_stream_tts.py new file mode 100644 index 00000000..309ad339 --- /dev/null +++ b/main/xiaozhi-server/performance_tester/performance_tester_stream_tts.py @@ -0,0 +1,388 @@ +import asyncio +import time +import json +import uuid +import aiohttp +import websockets +from tabulate import tabulate +from config.settings import load_config + +description = "流式TTS语音合成首词耗时测试" +class StreamTTSPerformanceTester: + def __init__(self): + self.config = load_config() + self.test_texts = [ + "你好，这是一句话。" + ] + self.results = [] + + async def test_aliyun_tts(self, text=None, test_count=5): + """测试阿里云流式TTS首词延迟（测试多次取平均）""" + text = text or self.test_texts[0] + latencies = [] + + for i in range(test_count): + try: + tts_config = self.config["TTS"]["AliyunStreamTTS"] + appkey = tts_config["appkey"] + token = tts_config["token"] + voice = tts_config["voice"] + host = tts_config["host"] + ws_url = f"wss://{host}/ws/v1" + + start_time = time.time() + async with websockets.connect(ws_url, extra_headers={"X-NLS-Token": token}) as ws: + task_id = str(uuid.uuid4()) + message_id = str(uuid.uuid4()) + + start_request = { + "header": { + "message_id": message_id, + "task_id": task_id, + "namespace": "FlowingSpeechSynthesizer", + "name": "StartSynthesis", + "appkey": appkey, + }, + "payload": { + "voice": voice, + "format": "pcm", + "sample_rate": 16000, + "volume": 50, + "speech_rate": 0, + "pitch_rate": 0, + } + } + await ws.send(json.dumps(start_request)) + + start_response = json.loads(await ws.recv()) + if start_response["header"]["name"] != "SynthesisStarted": + raise Exception("启动合成失败") + + run_request = { + "header": { + "message_id": str(uuid.uuid4()), + "task_id": task_id, + "namespace": "FlowingSpeechSynthesizer", + "name": "RunSynthesis", + "appkey": appkey, + }, + "payload": {"text": text} + } + await ws.send(json.dumps(run_request)) + + while True: + response = await ws.recv() + if isinstance(response, bytes): + latency = time.time() - start_time + latencies.append(latency) + break + elif isinstance(response, str): + data = json.loads(response) + if data["header"]["name"] == "TaskFailed": + raise Exception(f"合成失败: {data['payload']['error_info']}") + + except Exception as e: + latencies.append(0) + + return self._calculate_result("阿里云TTS", latencies, test_count) + + async def test_doubao_tts(self, text=None, test_count=5): + """测试火山引擎流式TTS首词延迟（测试多次取平均）""" + text = text or self.test_texts[0] + latencies = [] + + for i in range(test_count): + try: + tts_config = self.config["TTS"]["HuoshanDoubleStreamTTS"] + ws_url = tts_config["ws_url"] + app_id = tts_config["appid"] + access_token = tts_config["access_token"] + resource_id = tts_config["resource_id"] + speaker = tts_config["speaker"] + + start_time = time.time() + ws_header = { + "X-Api-App-Key": app_id, + "X-Api-Access-Key": access_token, + "X-Api-Resource-Id": resource_id, + "X-Api-Connect-Id": str(uuid.uuid4()), + } + async with websockets.connect(ws_url, additional_headers=ws_header, max_size=1000000000) as ws: + session_id = uuid.uuid4().hex + + # 发送会话启动请求 + header = bytes([ + (0b0001 << 4) | 0b0001, + 0b0001 << 4 | 0b100, + 0b0001 << 4 | 0b0000, + 0 + ]) + optional = bytearray() + optional.extend((1).to_bytes(4, "big", signed=True)) + session_id_bytes = session_id.encode() + optional.extend(len(session_id_bytes).to_bytes(4, "big", signed=True)) + optional.extend(session_id_bytes) + payload = json.dumps({"speaker": speaker}).encode() + await ws.send(header + optional + len(payload).to_bytes(4, "big", signed=True) + payload) + + # 发送文本 + header = bytes([ + (0b0001 << 4) | 0b0001, + 0b0001 << 4 | 0b100, + 0b0001 << 4 | 0b0000, + 0 + ]) + optional = bytearray() + optional.extend((200).to_bytes(4, "big", signed=True)) + session_id_bytes = session_id.encode() + optional.extend(len(session_id_bytes).to_bytes(4, "big", signed=True)) + optional.extend(session_id_bytes) + payload = json.dumps({"text": text, "speaker": speaker}).encode() + await ws.send(header + optional + len(payload).to_bytes(4, "big", signed=True) + payload) + + first_chunk = await ws.recv() + latency = time.time() - start_time + latencies.append(latency) + + except Exception as e: + latencies.append(0) + + return self._calculate_result("火山引擎TTS", latencies, test_count) + + async def test_paddlespeech_tts(self, text=None, test_count=5): + """测试PaddleSpeech流式TTS首词延迟（测试多次取平均）""" + text = text or self.test_texts[0] + latencies = [] + + for i in range(test_count): + try: + tts_config = self.config["TTS"]["PaddleSpeechTTS"] + tts_url = tts_config["url"] + spk_id = tts_config["spk_id"] + speed = tts_config["speed"] + volume = tts_config["volume"] + + start_time = time.time() + async with websockets.connect(tts_url) as ws: + # 发送开始请求 + await ws.send(json.dumps({ + "task": "tts", + "signal": "start" + })) + + start_response = json.loads(await ws.recv()) + if start_response.get("status") != 0: + raise Exception("连接失败") + + # 发送文本数据 + await ws.send(json.dumps({ + "text": text, + "spk_id": spk_id, + "speed": speed, + "volume": volume + })) + + # 接收第一个数据块 + first_chunk = await ws.recv() + latency = time.time() - start_time + latencies.append(latency) + + # 发送结束请求 + end_request = { + "task": "tts", + "signal": "end" + } + await ws.send(json.dumps(end_request)) + + # 确保连接正常关闭 + try: + await ws.recv() + except websockets.exceptions.ConnectionClosedOK: + pass + + except Exception as e: + latencies.append(0) + + return self._calculate_result("PaddleSpeechTTS", latencies, test_count) + + async def test_indexstream_tts(self, text=None, test_count=5): + """测试IndexStream流式TTS首词延迟（测试多次取平均）""" + text = text or self.test_texts[0] + latencies = [] + + for i in range(test_count): + try: + tts_config = self.config["TTS"]["IndexStreamTTS"] + api_url = tts_config.get("api_url") + voice = tts_config.get("voice") + + start_time = time.time() + + async with aiohttp.ClientSession() as session: + payload = {"text": text, "character": voice} + async with session.post(api_url, json=payload, timeout=10) as resp: + if resp.status != 200: + raise Exception(f"请求失败: {resp.status}, {await resp.text()}") + + async for chunk in resp.content.iter_any(): + data = chunk[0] if isinstance(chunk, (list, tuple)) else chunk + if not data: + continue + + latency = time.time() - start_time + latencies.append(latency) + resp.close() + break + else: + latencies.append(0) + + except Exception as e: + latencies.append(0) + + return self._calculate_result("IndexStreamTTS", latencies, test_count) + + async def test_linkerai_tts(self, text=None, test_count=5): + """测试Linkerai流式TTS首词延迟（测试多次取平均）""" + text = text or self.test_texts[0] + latencies = [] + + for i in range(test_count): + try: + tts_config = self.config["TTS"]["LinkeraiTTS"] + api_url = tts_config["api_url"] + access_token = tts_config["access_token"] + voice = tts_config["voice"] + + start_time = time.time() + async with aiohttp.ClientSession() as session: + params = { + "tts_text": text, + "spk_id": voice, + "frame_durition": 60, + "stream": "true", + "target_sr": 16000, + "audio_format": "pcm", + "instruct_text": "请生成一段自然流畅的语音", + } + headers = { + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + } + + async with session.get(api_url, params=params, headers=headers, timeout=10) as resp: + if resp.status != 200: + raise Exception(f"请求失败: {resp.status}, {await resp.text()}") + + # 接收第一个数据块 + async for _ in resp.content.iter_any(): + latency = time.time() - start_time + latencies.append(latency) + break + else: + latencies.append(0) + + except Exception as e: + latencies.append(0) + + return self._calculate_result("LinkeraiTTS", latencies, test_count) + + + def _calculate_result(self, service_name, latencies, test_count): + """计算测试结果""" + valid_latencies = [l for l in latencies if l > 0] + if valid_latencies: + avg_latency = sum(valid_latencies) / len(valid_latencies) + status = f"成功（{len(valid_latencies)}/{test_count}次有效）" + else: + avg_latency = 0 + status = "失败: 所有测试均失败" + return {"name": service_name, "latency": avg_latency, "status": status} + + def _print_results(self, test_text, test_count): + """打印测试结果""" + if not self.results: + print("没有有效的TTS测试结果") + return + + print(f"\n{'='*60}") + print("流式TTS首词延迟测试结果") + print(f"{'='*60}") + print(f"测试文本: {test_text}") + print(f"测试次数: 每个TTS服务测试 {test_count} 次") + + # 排序结果：成功优先，按延迟升序 + success_results = sorted( + [r for r in self.results if "成功" in r["status"]], + key=lambda x: x["latency"] + ) + failed_results = [r for r in self.results if "成功" not in r["status"]] + + table_data = [ + [r["name"], f"{r['latency']:.3f}", r["status"]] + for r in success_results + failed_results + ] + + print(tabulate(table_data, headers=["TTS服务", "首词延迟(秒)", "状态"], tablefmt="grid")) + print("\n测试说明：测量从发送请求到接收第一个音频数据块的时间，取多次测试平均值") + print("- 超时控制: 单个请求最大等待时间为10秒") + print("- 错误处理: 无法连接和超时的列为网络错误") + print("- 排序规则: 按平均耗时从快到慢排序") + + + async def run(self, test_text=None, test_count=5): + """执行测试 + + Args: + test_text: 要测试的文本，如果为None则使用默认文本 + test_count: 每个TTS服务的测试次数 + """ + test_text = test_text or self.test_texts[0] + print(f"开始流式TTS首词延迟测试...") + print(f"测试文本: {test_text}") + print(f"每个TTS服务测试次数: {test_count}次") + + if not self.config.get("TTS"): + print("配置文件中未找到TTS配置") + return + + # 测试每种TTS服务 + self.results = [] + + # 测试阿里云TTS + result = await self.test_aliyun_tts(test_text, test_count) + self.results.append(result) + + # 测试火山引擎TTS + result = await self.test_doubao_tts(test_text, test_count) + self.results.append(result) + + # 测试PaddleSpeech TTS + result = await self.test_paddlespeech_tts(test_text, test_count) + self.results.append(result) + + # 测试Linkerai TTS + result = await self.test_linkerai_tts(test_text, test_count) + self.results.append(result) + + # 测试IndexStreamTTS + result = await self.test_indexstream_tts(test_text, test_count) + self.results.append(result) + + # 打印结果 + self._print_results(test_text, test_count) + + +async def main(): + import argparse + + parser = argparse.ArgumentParser(description="流式TTS首词延迟测试工具") + parser.add_argument("--text", help="要测试的文本内容") + parser.add_argument("--count", type=int, default=5, help="每个TTS服务的测试次数") + + args = parser.parse_args() + await StreamTTSPerformanceTester().run(args.text, args.count) + + +if __name__ == "__main__": + import asyncio + asyncio.run(main()) \ No newline at end of file diff --git a/main/xiaozhi-server/performance_tester/performance_tester_tts.py b/main/xiaozhi-server/performance_tester/performance_tester_tts.py index 4702c5a9..61e49682 100644 --- a/main/xiaozhi-server/performance_tester/performance_tester_tts.py +++ b/main/xiaozhi-server/performance_tester/performance_tester_tts.py @@ -86,22 +86,67 @@ class TTSPerformanceTester: print("没有有效的TTS测试结果") return - table = [] + headers = ["TTS模块", "平均耗时(秒)", "测试句子数", "状态"] + table_data = [] + + # 收集所有数据并分类 + valid_results = [] + error_results = [] + for name, data in self.results.items(): if data["errors"] == 0: - table.append( - [name, f"{data['avg_time']:.3f}秒/句", len(self.test_sentences[:3])] - ) + # 正常结果 + avg_time = f"{data['avg_time']:.3f}" + test_count = len(self.test_sentences[:3]) + status = "✅ 正常" + + # 保存用于排序的值 + valid_results.append({ + "name": name, + "avg_time": avg_time, + "test_count": test_count, + "status": status, + "sort_key": data['avg_time'] + }) + else: + # 错误结果 + avg_time = "-" + test_count = "0/3" + + # 默认错误类型为网络错误 + error_type = "网络错误" + status = f"❌ {error_type}" + + error_results.append([name, avg_time, test_count, status]) + + # 按平均耗时升序排序 + valid_results.sort(key=lambda x: x["sort_key"]) + + # 将排序后的有效结果转换为表格数据 + for result in valid_results: + table_data.append([ + result["name"], + result["avg_time"], + result["test_count"], + result["status"] + ]) + + # 将错误结果添加到表格数据末尾 + table_data.extend(error_results) print("\nTTS性能测试结果:") print( tabulate( - table, - headers=["TTS模块", "平均耗时", "测试句子数"], - tablefmt="github", - colalign=("left", "right", "right"), + table_data, + headers=headers, + tablefmt="grid", + colalign=("left", "right", "right", "left"), ) ) + print("\n测试说明:") + print("- 超时控制: 单个请求最大等待时间为10秒") + print("- 错误处理: 无法连接和超时的列为网络错误") + print("- 排序规则: 按平均耗时从快到慢排序") async def run(self): """执行测试""" @@ -119,10 +164,9 @@ class TTSPerformanceTester: # 并发执行测试 results = await asyncio.gather(*tasks) - # 保存有效结果 + # 保存所有结果，包括错误 for result in results: - if result["errors"] == 0: - self.results[result["name"]] = result + self.results[result["name"]] = result # 打印结果 self._print_results()