import json
|
import uuid
|
import asyncio
|
import websockets
|
from typing import List, TYPE_CHECKING
|
|
if TYPE_CHECKING:
|
from core.connection import ConnectionHandler
|
|
from config.logger import setup_logging
|
from core.providers.asr.base import ASRProviderBase
|
from core.providers.asr.dto.dto import InterfaceType
|
|
TAG = __name__
|
logger = setup_logging()
|
|
|
class ASRProvider(ASRProviderBase):
|
def __init__(self, config, delete_audio_file):
|
super().__init__()
|
self.interface_type = InterfaceType.STREAM
|
self.config = config
|
self.text = ""
|
self.asr_ws = None
|
self.forward_task = None
|
self.is_processing = False
|
self.server_ready = False # 服务器准备状态
|
self.task_id = None # 当前任务ID
|
|
# 阿里百炼配置
|
self.api_key = config.get("api_key")
|
self.model = config.get("model", "paraformer-realtime-v2")
|
self.sample_rate = config.get("sample_rate", 16000)
|
self.format = config.get("format", "pcm")
|
|
# 可选参数
|
self.vocabulary_id = config.get("vocabulary_id")
|
self.disfluency_removal_enabled = config.get("disfluency_removal_enabled", False)
|
self.language_hints = config.get("language_hints")
|
self.semantic_punctuation_enabled = config.get("semantic_punctuation_enabled", False)
|
max_sentence_silence = config.get("max_sentence_silence")
|
self.max_sentence_silence = int(max_sentence_silence) if max_sentence_silence else 200
|
self.multi_threshold_mode_enabled = config.get("multi_threshold_mode_enabled", False)
|
self.punctuation_prediction_enabled = config.get("punctuation_prediction_enabled", True)
|
self.inverse_text_normalization_enabled = config.get("inverse_text_normalization_enabled", True)
|
|
# WebSocket URL
|
self.ws_url = "wss://dashscope.aliyuncs.com/api-ws/v1/inference"
|
|
self.output_dir = config.get("output_dir", "./audio_output")
|
self.delete_audio_file = delete_audio_file
|
|
async def open_audio_channels(self, conn):
|
await super().open_audio_channels(conn)
|
|
async def receive_audio(self, conn, pcm_frame, audio_have_voice):
|
# 先调用父类方法处理基础逻辑
|
await super().receive_audio(conn, pcm_frame, audio_have_voice)
|
|
# 只在有声音且没有连接时建立连接
|
if audio_have_voice and not self.is_processing and not self.asr_ws:
|
try:
|
await self._start_recognition(conn)
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"开始识别失败: {str(e)}")
|
await self._cleanup()
|
return
|
|
# 发送音频数据
|
if self.asr_ws and self.is_processing and self.server_ready:
|
try:
|
await self.asr_ws.send(pcm_frame)
|
except Exception as e:
|
logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}")
|
await self._cleanup()
|
|
async def _start_recognition(self, conn: "ConnectionHandler"):
|
"""开始识别会话"""
|
try:
|
# 如果为手动模式,设置超时时长为最大值
|
if conn.client_listen_mode == "manual":
|
self.max_sentence_silence = 6000
|
|
self.is_processing = True
|
self.task_id = uuid.uuid4().hex
|
|
# 建立WebSocket连接
|
headers = {
|
"Authorization": f"Bearer {self.api_key}"
|
}
|
|
logger.bind(tag=TAG).debug(f"正在连接阿里百炼ASR服务, task_id: {self.task_id}")
|
|
self.asr_ws = await websockets.connect(
|
self.ws_url,
|
additional_headers=headers,
|
max_size=1000000000,
|
ping_interval=None,
|
ping_timeout=None,
|
close_timeout=5,
|
)
|
|
logger.bind(tag=TAG).debug("WebSocket连接建立成功")
|
|
self.server_ready = False
|
self.forward_task = asyncio.create_task(self._forward_results(conn))
|
|
# 发送run-task指令
|
run_task_msg = self._build_run_task_message()
|
await self.asr_ws.send(json.dumps(run_task_msg, ensure_ascii=False))
|
logger.bind(tag=TAG).debug("已发送run-task指令，等待服务器准备...")
|
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"建立ASR连接失败: {str(e)}")
|
if self.asr_ws:
|
await self.asr_ws.close()
|
self.asr_ws = None
|
self.is_processing = False
|
raise
|
|
def _build_run_task_message(self) -> dict:
|
"""构建run-task指令"""
|
message = {
|
"header": {
|
"action": "run-task",
|
"task_id": self.task_id,
|
"streaming": "duplex"
|
},
|
"payload": {
|
"task_group": "audio",
|
"task": "asr",
|
"function": "recognition",
|
"model": self.model,
|
"parameters": {
|
"format": self.format,
|
"sample_rate": self.sample_rate,
|
"disfluency_removal_enabled": self.disfluency_removal_enabled,
|
"semantic_punctuation_enabled": self.semantic_punctuation_enabled,
|
"max_sentence_silence": self.max_sentence_silence,
|
"multi_threshold_mode_enabled": self.multi_threshold_mode_enabled,
|
"punctuation_prediction_enabled": self.punctuation_prediction_enabled,
|
"inverse_text_normalization_enabled": self.inverse_text_normalization_enabled,
|
},
|
"input": {}
|
}
|
}
|
|
# 只有当模型名称以v2结尾时才添加vocabulary_id参数
|
if self.model.lower().endswith("v2"):
|
message["payload"]["parameters"]["vocabulary_id"] = self.vocabulary_id
|
|
if self.language_hints:
|
message["payload"]["parameters"]["language_hints"] = self.language_hints
|
|
return message
|
|
async def _forward_results(self, conn: "ConnectionHandler"):
|
"""转发识别结果"""
|
try:
|
while not conn.stop_event.is_set():
|
# 获取当前连接的音频数据
|
audio_data = conn.asr_audio
|
try:
|
response = await asyncio.wait_for(self.asr_ws.recv(), timeout=1.0)
|
result = json.loads(response)
|
|
header = result.get("header", {})
|
payload = result.get("payload", {})
|
event = header.get("event", "")
|
|
# 处理task-started事件
|
if event == "task-started":
|
self.server_ready = True
|
logger.bind(tag=TAG).debug("服务器已准备，开始发送缓存音频...")
|
|
# 发送缓存音频
|
if conn.asr_audio:
|
for cached_pcm in conn.asr_audio[-10:]:
|
try:
|
await self.asr_ws.send(cached_pcm)
|
except Exception as e:
|
logger.bind(tag=TAG).warning(f"发送缓存音频失败: {e}")
|
break
|
continue
|
|
# 处理result-generated事件
|
elif event == "result-generated":
|
output = payload.get("output", {})
|
sentence = output.get("sentence", {})
|
|
text = sentence.get("text", "")
|
sentence_end = sentence.get("sentence_end", False)
|
end_time = sentence.get("end_time")
|
|
# 判断是否为最终结果(sentence_end为True且end_time不为null)
|
is_final = sentence_end and end_time is not None
|
|
if is_final:
|
logger.bind(tag=TAG).info(f"识别到文本: {text}")
|
|
# 手动模式下累积识别结果
|
if conn.client_listen_mode == "manual":
|
if self.text:
|
self.text += text
|
else:
|
self.text = text
|
|
# 手动模式下,只有在收到stop信号后才触发处理
|
if conn.client_voice_stop:
|
logger.bind(tag=TAG).debug("收到最终识别结果，触发处理")
|
await self.handle_voice_stop(conn, audio_data)
|
break
|
else:
|
# 自动模式下直接覆盖
|
self.text = text
|
await self.handle_voice_stop(conn, audio_data)
|
break
|
|
# 处理task-finished事件
|
elif event == "task-finished":
|
logger.bind(tag=TAG).debug("任务已完成")
|
break
|
|
# 处理task-failed事件
|
elif event == "task-failed":
|
error_code = header.get("error_code", "UNKNOWN")
|
error_message = header.get("error_message", "未知错误")
|
logger.bind(tag=TAG).error(f"任务失败: {error_code} - {error_message}")
|
break
|
|
except asyncio.TimeoutError:
|
continue
|
except websockets.ConnectionClosed:
|
logger.bind(tag=TAG).info("ASR服务连接已关闭")
|
self.is_processing = False
|
break
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"处理结果失败: {str(e)}")
|
break
|
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"结果转发失败: {str(e)}")
|
finally:
|
# 清理连接的音频缓存
|
await self._cleanup()
|
conn.reset_audio_states()
|
|
async def _send_stop_request(self):
|
"""发送停止请求(用于手动模式停止录音)"""
|
if self.asr_ws:
|
try:
|
# 先停止音频发送
|
self.is_processing = False
|
|
logger.bind(tag=TAG).debug("收到停止请求，发送finish-task指令")
|
await self._send_finish_task()
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"发送停止请求失败: {e}")
|
|
async def _send_finish_task(self):
|
"""发送finish-task指令"""
|
if self.asr_ws and self.task_id:
|
try:
|
finish_msg = {
|
"header": {
|
"action": "finish-task",
|
"task_id": self.task_id,
|
"streaming": "duplex"
|
},
|
"payload": {
|
"input": {}
|
}
|
}
|
await self.asr_ws.send(json.dumps(finish_msg, ensure_ascii=False))
|
logger.bind(tag=TAG).debug("已发送finish-task指令")
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"发送finish-task指令失败: {e}")
|
|
async def _cleanup(self):
|
"""清理资源"""
|
logger.bind(tag=TAG).debug(f"开始ASR会话清理 | 当前状态: processing={self.is_processing}, server_ready={self.server_ready}")
|
|
# 状态重置
|
self.is_processing = False
|
self.server_ready = False
|
logger.bind(tag=TAG).debug("ASR状态已重置")
|
|
# 关闭连接
|
if self.asr_ws:
|
try:
|
# 先发送finish-task指令
|
await self._send_finish_task()
|
# 等待一小段时间让服务器处理
|
await asyncio.sleep(0.1)
|
|
logger.bind(tag=TAG).debug("正在关闭WebSocket连接")
|
await asyncio.wait_for(self.asr_ws.close(), timeout=2.0)
|
logger.bind(tag=TAG).debug("WebSocket连接已关闭")
|
except Exception as e:
|
logger.bind(tag=TAG).error(f"关闭WebSocket连接失败: {e}")
|
finally:
|
self.asr_ws = None
|
|
# 清理任务引用
|
self.forward_task = None
|
self.task_id = None
|
|
logger.bind(tag=TAG).debug("ASR会话清理完成")
|
|
async def speech_to_text(self, opus_data, session_id, artifacts=None):
|
"""获取识别结果"""
|
result = self.text
|
self.text = ""
|
return result, None
|
|
async def close(self):
|
"""关闭资源"""
|
await self._cleanup()
|
