import asyncio
|
import time
|
import json
|
import uuid
|
import aiohttp
|
import websockets
|
import hmac
|
import base64
|
import hashlib
|
import asyncio
|
from urllib.parse import urlparse, urlencode
|
from tabulate import tabulate
|
from config.settings import load_config
|
|
description = "流式TTS语音合成首词耗时测试"
|
class StreamTTSPerformanceTester:
|
def __init__(self, config):
|
self.config = 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,
|
"enable_subtitle": True,
|
}
|
}
|
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)
|
print(f"[阿里云TTS] 第{i+1}次 首词延迟: {latency:.3f}s")
|
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:
|
print(f"[阿里云TTS] 第{i+1}次测试失败: {str(e)}")
|
latencies.append(None)
|
|
return self._calculate_result("阿里云TTS", latencies, test_count)
|
|
async def test_alibl_tts(self, text=None, test_count=5):
|
"""测试阿里云百炼CosyVoice流式TTS首词延迟"""
|
text = text or self.test_texts[0]
|
latencies = []
|
|
for i in range(test_count):
|
try:
|
tts_config = self.config["TTS"]["AliBLTTS"]
|
api_key = tts_config["api_key"]
|
model = tts_config.get("model", "cosyvoice-v2")
|
voice = tts_config.get("voice", "longxiaochun_v2")
|
format_type = tts_config.get("format", "pcm")
|
sample_rate = int(tts_config.get("sample_rate", "24000"))
|
|
ws_url = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/"
|
headers = {
|
"Authorization": f"Bearer {api_key}",
|
"X-DashScope-DataInspection": "enable",
|
}
|
|
start_time = time.time()
|
|
async with websockets.connect(
|
ws_url,
|
additional_headers=headers,
|
ping_interval=30,
|
ping_timeout=10,
|
close_timeout=10,
|
max_size=10 * 1024 * 1024,
|
) as ws:
|
session_id = uuid.uuid4().hex
|
|
# 1. 发送 run-task（启动任务）
|
run_task_message = {
|
"header": {
|
"action": "run-task",
|
"task_id": session_id,
|
"streaming": "duplex",
|
},
|
"payload": {
|
"task_group": "audio",
|
"task": "tts",
|
"function": "SpeechSynthesizer",
|
"model": model,
|
"parameters": {
|
"text_type": "PlainText",
|
"voice": voice,
|
"format": format_type,
|
"sample_rate": sample_rate,
|
"volume": 50,
|
"rate": 1.0,
|
"pitch": 1.0,
|
},
|
"input": {}
|
},
|
}
|
await ws.send(json.dumps(run_task_message))
|
|
# 2. 等待 task-started 事件（关键！必须等这个再发文本）
|
task_started = False
|
while not task_started:
|
msg = await ws.recv()
|
if isinstance(msg, str):
|
data = json.loads(msg)
|
header = data.get("header", {})
|
event = header.get("event")
|
if event == "task-started":
|
task_started = True
|
print(f"[阿里云百炼TTS] 第{i+1}次 任务启动成功")
|
elif event == "task-failed":
|
raise Exception(f"启动失败: {header.get('error_message', '未知错误')}")
|
|
# 3. 发送 continue-task（发送文本！这是正确动作）
|
continue_task_message = {
|
"header": {
|
"action": "continue-task", # 改回 continue-task
|
"task_id": session_id,
|
"streaming": "duplex",
|
},
|
"payload": {"input": {"text": text}},
|
}
|
await ws.send(json.dumps(continue_task_message))
|
|
# 4. 发送 finish-task（结束任务）
|
finish_task_message = {
|
"header": {
|
"action": "finish-task",
|
"task_id": session_id,
|
"streaming": "duplex",
|
},
|
"payload": {"input": {}}
|
}
|
await ws.send(json.dumps(finish_task_message))
|
|
# 5. 等待第一个音频数据块
|
while True:
|
msg = await asyncio.wait_for(ws.recv(), timeout=15.0)
|
if isinstance(msg, (bytes, bytearray)) and len(msg) > 0:
|
latency = time.time() - start_time
|
print(f"[阿里云百炼TTS] 第{i+1}次 首词延迟: {latency:.3f}s")
|
latencies.append(latency)
|
break
|
elif isinstance(msg, str):
|
data = json.loads(msg)
|
event = data.get("header", {}).get("event")
|
if event == "task-failed":
|
raise Exception(f"合成失败: {data}")
|
elif event == "task-finished":
|
if not latencies or latencies[-1] is None:
|
raise Exception("任务结束但未收到音频")
|
|
except Exception as e:
|
print(f"[阿里云百炼TTS] 第{i+1}次失败: {str(e)}")
|
latencies.append(None)
|
|
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 | 0b1011,
|
0b0001 << 4 | 0b0000,
|
])
|
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 | 0b1011,
|
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)
|
print(f"[火山引擎TTS] 第{i+1}次 首词延迟: {latency:.3f}s")
|
|
except Exception as e:
|
print(f"[火山引擎TTS] 第{i+1}次测试失败: {str(e)}")
|
latencies.append(None)
|
|
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)
|
print(f"[PaddleSpeechTTS] 第{i+1}次 首词延迟: {latency:.3f}s")
|
|
# 发送结束请求
|
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:
|
print(f"[PaddleSpeechTTS] 第{i+1}次测试失败: {str(e)}")
|
latencies.append(None)
|
|
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)
|
print(f"[IndexStreamTTS] 第{i+1}次 首词延迟: {latency:.3f}s")
|
resp.close()
|
break
|
else:
|
latencies.append(None)
|
|
except Exception as e:
|
print(f"[IndexStreamTTS] 第{i+1}次测试失败: {str(e)}")
|
latencies.append(None)
|
|
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)
|
print(f"[LinkeraiTTS] 第{i+1}次 首词延迟: {latency:.3f}s")
|
break
|
else:
|
latencies.append(None)
|
|
except Exception as e:
|
print(f"[LinkeraiTTS] 第{i+1}次测试失败: {str(e)}")
|
latencies.append(None)
|
|
return self._calculate_result("LinkeraiTTS", latencies, test_count)
|
|
async def test_xunfei_tts(self, text=None, test_count=5):
|
"""测试讯飞流式TTS首词延迟（测试多次取平均）"""
|
text = text or self.test_texts[0]
|
latencies = []
|
|
for i in range(test_count):
|
try:
|
# 修正配置节点名称，与配置文件中的XunFeiTTS匹配
|
tts_config = self.config["TTS"]["XunFeiTTS"]
|
app_id = tts_config["app_id"]
|
api_key = tts_config["api_key"]
|
api_secret = tts_config["api_secret"]
|
api_url = tts_config.get("api_url", "wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6")
|
voice = tts_config.get("voice", "x5_lingxiaoxuan_flow")
|
# 生成认证URL
|
auth_url = self._create_xunfei_auth_url(api_key, api_secret, api_url)
|
start_time = time.time()
|
async with websockets.connect(
|
auth_url,
|
ping_interval=30,
|
ping_timeout=10,
|
close_timeout=10,
|
max_size=1000000000
|
) as ws:
|
# 构造请求
|
request = self._build_xunfei_request(app_id, text, voice)
|
await ws.send(json.dumps(request))
|
# 等待第一个音频数据块
|
first_audio_received = False
|
while not first_audio_received:
|
msg = await asyncio.wait_for(ws.recv(), timeout=10)
|
data = json.loads(msg)
|
header = data.get("header", {})
|
code = header.get("code")
|
|
if code != 0:
|
message = header.get("message", "未知错误")
|
raise Exception(f"合成失败: {code} - {message}")
|
|
payload = data.get("payload", {})
|
audio_payload = payload.get("audio", {})
|
|
if audio_payload:
|
status = audio_payload.get("status", 0)
|
audio_data = audio_payload.get("audio", "")
|
if status == 1 and audio_data:
|
# 收到第一个音频数据块
|
latency = time.time() - start_time
|
latencies.append(latency)
|
print(f"[讯飞TTS] 第{i+1}次 首词延迟: {latency:.3f}s")
|
first_audio_received = True
|
break
|
except Exception as e:
|
print(f"[讯飞TTS] 第{i+1}次测试失败: {str(e)}")
|
latencies.append(None)
|
|
return self._calculate_result("讯飞TTS", latencies, test_count)
|
|
def _create_xunfei_auth_url(self, api_key, api_secret, api_url):
|
"""生成讯飞WebSocket认证URL"""
|
parsed_url = urlparse(api_url)
|
host = parsed_url.netloc
|
path = parsed_url.path
|
|
# 获取UTC时间，讯飞要求使用RFC1123格式
|
now = time.gmtime()
|
date = time.strftime('%a, %d %b %Y %H:%M:%S GMT', now)
|
|
# 构造签名字符串
|
signature_origin = f"host: {host}\ndate: {date}\nGET {path} HTTP/1.1"
|
|
# 计算签名
|
signature_sha = hmac.new(
|
api_secret.encode('utf-8'),
|
signature_origin.encode('utf-8'),
|
digestmod=hashlib.sha256
|
).digest()
|
signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')
|
|
# 构造authorization
|
authorization_origin = f'api_key="{api_key}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'
|
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
|
|
# 构造最终的WebSocket URL
|
v = {
|
"authorization": authorization,
|
"date": date,
|
"host": host
|
}
|
url = api_url + '?' + urlencode(v)
|
return url
|
|
def _build_xunfei_request(self, app_id, text, voice):
|
"""构建讯飞TTS请求结构"""
|
return {
|
"header": {
|
"app_id": app_id,
|
"status": 2,
|
},
|
"parameter": {
|
"oral": {
|
"oral_level": "mid",
|
"spark_assist": 1,
|
"stop_split": 0,
|
"remain": 0
|
},
|
"tts": {
|
"vcn": voice,
|
"speed": 50,
|
"volume": 50,
|
"pitch": 50,
|
"bgs": 0,
|
"reg": 0,
|
"rdn": 0,
|
"rhy": 0,
|
"audio": {
|
"encoding": "raw",
|
"sample_rate": 24000,
|
"channels": 1,
|
"bit_depth": 16,
|
"frame_size": 0
|
}
|
}
|
},
|
"payload": {
|
"text": {
|
"encoding": "utf8",
|
"compress": "raw",
|
"format": "plain",
|
"status": 2,
|
"seq": 1,
|
"text": base64.b64encode(text.encode('utf-8')).decode('utf-8')
|
}
|
}
|
}
|
|
|
def _calculate_result(self, service_name, latencies, test_count):
|
"""计算测试结果（正确处理None值，剔除失败测试）"""
|
# 剔除失败的测试（None值和<=0延迟），只统计有效延迟
|
valid_latencies = [l for l in latencies if l is not None and 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("- 计时起点: 建立WebSocket/HTTP连接前（统一包含网络建连、握手、发送文本全流程）")
|
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
|
if self.config.get("TTS", {}).get("AliBLTTS"):
|
result = await self.test_alibl_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)
|
|
# 测试讯飞TTS
|
if self.config.get("TTS", {}).get("XunFeiTTS"):
|
result = await self.test_xunfei_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()
|
config = await load_config()
|
await StreamTTSPerformanceTester(config).run(args.text, args.count)
|
|
|
if __name__ == "__main__":
|
import asyncio
|
asyncio.run(main())
|
