261 lines
10 KiB
Python
261 lines
10 KiB
Python
"""Async Qwen-Audio/Fun-ASR adapter using the DashScope recognition protocol."""
|
|
|
|
import asyncio
|
|
import json
|
|
import uuid
|
|
|
|
import aiohttp
|
|
|
|
from config import settings
|
|
from interfaces.asr import ASR
|
|
from implementations.base_service import BaseService
|
|
from services.interrupt_handler import interrupt_handler
|
|
from utils.logger import session_logger
|
|
|
|
|
|
class ASRError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class AliyunASR(ASR, BaseService):
|
|
def __init__(self, selected_role=None):
|
|
BaseService.__init__(self, "asr", selected_role)
|
|
self.api_key = settings.aliyun_api_key
|
|
self.model = settings.aliyun_asr_model
|
|
self.url = settings.aliyun_asr_url
|
|
# Gummy vocabulary IDs are model-specific; never reuse them implicitly.
|
|
self.vocabulary_id = settings.aliyun_asr_vocabulary_id
|
|
self.sample_rate = 16000
|
|
self.device_id = "unknown"
|
|
self.session_id = "unknown"
|
|
self.connected = False
|
|
self.transcript = ""
|
|
self.error = None
|
|
self.task_id = None
|
|
self.audio_buffer = bytearray()
|
|
self.buffer_size = 3200
|
|
self._client = None
|
|
self._websocket = None
|
|
self._receiver = None
|
|
self._ready = asyncio.Event()
|
|
self._finished = asyncio.Event()
|
|
self._sentences = {}
|
|
self._pending = set()
|
|
self._complete = False
|
|
self._ending = False
|
|
self._closing = False
|
|
self._end_lock = asyncio.Lock()
|
|
self._close_lock = asyncio.Lock()
|
|
|
|
def _fail(self, message):
|
|
if self.error is None:
|
|
self.error = ASRError(message)
|
|
session_logger.error(
|
|
self.device_id, self.session_id,
|
|
f"ASR失败 model={self.model} task_id={self.task_id}: {message}",
|
|
)
|
|
self.connected = False
|
|
self.transcript = ""
|
|
self._ready.set()
|
|
self._finished.set()
|
|
|
|
def _check_error(self):
|
|
if self.error:
|
|
raise self.error
|
|
|
|
def _command(self, action, payload):
|
|
return {
|
|
"header": {"action": action, "task_id": self.task_id, "streaming": "duplex"},
|
|
"payload": payload,
|
|
}
|
|
|
|
async def connect(self):
|
|
if self.connected:
|
|
return
|
|
await self.close()
|
|
self._closing = False
|
|
self._ending = False
|
|
self._complete = False
|
|
self.error = None
|
|
self.transcript = ""
|
|
self.audio_buffer.clear()
|
|
self._sentences.clear()
|
|
self._pending.clear()
|
|
self._ready.clear()
|
|
self._finished.clear()
|
|
self.task_id = uuid.uuid4().hex
|
|
parameters = {
|
|
"format": "mp3",
|
|
"sample_rate": self.sample_rate,
|
|
"semantic_punctuation_enabled": False,
|
|
"max_sentence_silence": 2000,
|
|
"heartbeat": True,
|
|
}
|
|
if self.vocabulary_id:
|
|
parameters["vocabulary_id"] = self.vocabulary_id
|
|
try:
|
|
if not self.api_key:
|
|
raise ASRError("未配置 ALIYUN_API_KEY")
|
|
# Native async I/O allows cancellation and bounded cleanup; the old
|
|
# SDK's stop() joins a worker thread without a timeout.
|
|
self._client = aiohttp.ClientSession(
|
|
timeout=aiohttp.ClientTimeout(total=None, connect=settings.asr_connect_timeout_seconds),
|
|
)
|
|
self._websocket = await asyncio.wait_for(
|
|
self._client.ws_connect(
|
|
self.url, headers={"Authorization": f"Bearer {self.api_key}"},
|
|
heartbeat=20,
|
|
),
|
|
timeout=settings.asr_connect_timeout_seconds,
|
|
)
|
|
self._receiver = asyncio.create_task(self._receive(), name=f"asr:{self.task_id}")
|
|
await asyncio.wait_for(self._websocket.send_json(self._command("run-task", {
|
|
"task_group": "audio", "task": "asr", "function": "recognition",
|
|
"model": self.model, "parameters": parameters, "input": {},
|
|
})), timeout=settings.asr_send_timeout_seconds)
|
|
await asyncio.wait_for(self._ready.wait(), timeout=settings.asr_connect_timeout_seconds)
|
|
self._check_error()
|
|
session_logger.info(
|
|
self.device_id, self.session_id,
|
|
f"ASR任务已就绪 model={self.model} task_id={self.task_id}",
|
|
)
|
|
except asyncio.CancelledError:
|
|
await self.close()
|
|
raise
|
|
except Exception as exc:
|
|
self._fail(f"启动失败: {type(exc).__name__}: {exc}")
|
|
await self.close()
|
|
raise self.error from exc
|
|
|
|
async def _receive(self):
|
|
try:
|
|
async for message in self._websocket:
|
|
if message.type == aiohttp.WSMsgType.ERROR:
|
|
raise ASRError("识别 WebSocket 传输错误")
|
|
if message.type != aiohttp.WSMsgType.TEXT:
|
|
continue
|
|
event = json.loads(message.data)
|
|
header = event.get("header", {})
|
|
if header.get("task_id") != self.task_id:
|
|
continue
|
|
kind = header.get("event")
|
|
if kind == "task-started":
|
|
self.connected = True
|
|
self._ready.set()
|
|
elif kind == "result-generated":
|
|
sentence = event.get("payload", {}).get("output", {}).get("sentence", {})
|
|
if not sentence or sentence.get("heartbeat"):
|
|
continue
|
|
key = sentence.get("sentence_id", sentence.get("begin_time"))
|
|
if key is None:
|
|
raise ASRError("识别结果缺少句子编号和起始时间")
|
|
if sentence.get("sentence_end"):
|
|
self._sentences[key] = sentence.get("text", "")
|
|
self._pending.discard(key)
|
|
elif sentence.get("text") and key not in self._sentences:
|
|
self._pending.add(key)
|
|
elif kind == "task-finished":
|
|
if not self._ending:
|
|
self._fail("识别任务提前结束，设备音频尚未发送完毕")
|
|
return
|
|
self._complete = True
|
|
self.connected = False
|
|
self._finished.set()
|
|
return
|
|
elif kind == "task-failed":
|
|
self._fail(f"{header.get('error_code', 'Unknown')}: {header.get('error_message', '')}")
|
|
return
|
|
except asyncio.CancelledError:
|
|
raise
|
|
except Exception as exc:
|
|
self._fail(f"接收失败: {type(exc).__name__}: {exc}")
|
|
finally:
|
|
if not self._complete and not self._closing and not self.error:
|
|
self._fail("识别连接提前关闭，未收到 task-finished")
|
|
|
|
async def _send_bytes(self, data):
|
|
self._check_error()
|
|
try:
|
|
await asyncio.wait_for(
|
|
self._websocket.send_bytes(data), timeout=settings.asr_send_timeout_seconds,
|
|
)
|
|
except Exception as exc:
|
|
self._fail(f"发送音频失败: {type(exc).__name__}: {exc}")
|
|
raise self.error from exc
|
|
|
|
async def send_audio(self, audio_data: bytes):
|
|
self._check_error()
|
|
if interrupt_handler.is_interrupted((self.device_id, self.session_id)):
|
|
raise ASRError("会话已中断")
|
|
if not self.connected or self._ending:
|
|
raise ASRError("识别任务未就绪或已结束")
|
|
self.audio_buffer.extend(audio_data)
|
|
while len(self.audio_buffer) >= self.buffer_size:
|
|
chunk = bytes(self.audio_buffer[:self.buffer_size])
|
|
await self._send_bytes(chunk)
|
|
del self.audio_buffer[:self.buffer_size]
|
|
|
|
async def send_end(self):
|
|
async with self._end_lock:
|
|
self._check_error()
|
|
if self._complete:
|
|
return
|
|
if not self.connected:
|
|
raise ASRError("识别任务未就绪")
|
|
self._ending = True
|
|
try:
|
|
if self.audio_buffer:
|
|
await self._send_bytes(bytes(self.audio_buffer))
|
|
self.audio_buffer.clear()
|
|
await asyncio.wait_for(
|
|
self._websocket.send_json(self._command("finish-task", {"input": {}})),
|
|
timeout=settings.asr_send_timeout_seconds,
|
|
)
|
|
await asyncio.wait_for(self._finished.wait(), timeout=settings.asr_final_timeout_seconds)
|
|
self._check_error()
|
|
except asyncio.CancelledError:
|
|
await self.close()
|
|
raise
|
|
except Exception as exc:
|
|
self._fail(f"结束识别失败: {type(exc).__name__}: {exc}")
|
|
raise self.error from exc
|
|
|
|
async def receive_results(self):
|
|
self._check_error()
|
|
if not self._complete or self._pending:
|
|
raise ASRError("识别未完整结束，拒绝使用不完整转录")
|
|
self.transcript = "".join(self._sentences[key] for key in sorted(self._sentences))
|
|
session_logger.info(
|
|
self.device_id, self.session_id,
|
|
f"ASR最终结果 model={self.model} task_id={self.task_id}: {self.transcript}",
|
|
)
|
|
return self.transcript
|
|
|
|
async def close(self):
|
|
async with self._close_lock:
|
|
await self._close_connection()
|
|
|
|
async def _close_connection(self):
|
|
self._closing = True
|
|
self.connected = False
|
|
# Wake waiters if the session is interrupted during start or finish.
|
|
if self.task_id and not self._complete and not self.error:
|
|
self.error = ASRError("识别会话已关闭")
|
|
self._ready.set()
|
|
self._finished.set()
|
|
if self._receiver:
|
|
self._receiver.cancel()
|
|
await asyncio.gather(self._receiver, return_exceptions=True)
|
|
self._receiver = None
|
|
try:
|
|
if self._websocket and not self._websocket.closed:
|
|
await asyncio.wait_for(self._websocket.close(), timeout=1.0)
|
|
except (asyncio.TimeoutError, aiohttp.ClientError):
|
|
pass
|
|
finally:
|
|
if self._client:
|
|
await self._client.close()
|
|
self._client = None
|
|
self._websocket = None
|
|
self.audio_buffer.clear()
|