410 lines
15 KiB
Python
410 lines
15 KiB
Python
import logging
|
||
import os
|
||
import threading
|
||
import time
|
||
from typing import Dict, List, Optional
|
||
|
||
import requests
|
||
from flask import Flask, abort, request
|
||
from pydantic import BaseModel, Field, ValidationError, field_validator
|
||
|
||
from schemas.dataset import QueryData
|
||
from schemas.stream import StreamDataModel
|
||
from utils.evaluator_plus import evaluate_editops, evaluate_punctuation
|
||
|
||
from .logger import log
|
||
|
||
MY_POD_IP = os.environ["MY_POD_IP"]
|
||
|
||
|
||
class StopException(Exception): ...
|
||
|
||
|
||
class EvaluateResult(BaseModel):
|
||
lang: str
|
||
cer: float
|
||
align_start: Dict[int, int] = Field(
|
||
description="句首字对齐时间差值(ms) -> 对齐数"
|
||
)
|
||
align_end: Dict[int, int] = Field(
|
||
description="句尾字对齐时间差值(ms) -> 对齐数"
|
||
)
|
||
first_word_distance_sum: float = Field(description="句首字距离总和(s)")
|
||
last_word_distance_sum: float = Field(description="句尾字距离总和(s)")
|
||
rtf: float = Field(description="翻译速度")
|
||
first_receive_delay: float = Field(description="首包接收延迟(s)")
|
||
query_count: int = Field(description="音频数")
|
||
voice_count: int = Field(description="句子数")
|
||
pred_punctuation_num: int = Field(description="预测标点数")
|
||
label_punctuation_num: int = Field(description="标注标点数")
|
||
pred_sentence_punctuation_num: int = Field(description="预测句子标点数")
|
||
label_setence_punctuation_num: int = Field(description="标注句子标点数")
|
||
preds: List[StreamDataModel] = Field(description="预测结果")
|
||
label: QueryData = Field(description="标注结果")
|
||
|
||
|
||
class ResultModel(BaseModel):
|
||
taskId: str
|
||
status: str
|
||
message: str = Field("")
|
||
recognition_results: Optional[StreamDataModel] = Field(None)
|
||
|
||
@field_validator("recognition_results", mode="after")
|
||
def convert_to_seconds(cls, v: Optional[StreamDataModel], values):
|
||
# 在这里处理除以1000的逻辑
|
||
if v is None:
|
||
return v
|
||
v.end_time = v.end_time / 1000
|
||
v.start_time = v.start_time / 1000
|
||
for word in v.words:
|
||
word.start_time /= 1000
|
||
word.end_time /= 1000
|
||
return v
|
||
|
||
|
||
class ClientCallback:
|
||
def __init__(self, sut_url: str, port: int):
|
||
self.sut_url = sut_url #sut_url：ASR 服务的 URL（如 http://asr-service:8080）
|
||
self.port = port #port：当前客户端监听的端口（用于接收回调）
|
||
|
||
#创建 Flask 应用并注册路由
|
||
self.app = Flask(__name__)
|
||
self.app.add_url_rule(
|
||
"/api/asr/batch-callback/<taskId>",
|
||
view_func=self.asr_callback,
|
||
methods=["POST"],
|
||
)
|
||
self.app.add_url_rule(
|
||
"/api/asr-runner/report",
|
||
view_func=self.heartbeat,
|
||
methods=["POST"],
|
||
)
|
||
"""
|
||
路由 1：/api/asr/batch-callback/<taskId>
|
||
接收 ASR 服务的识别结果回调（self.asr_callback 处理）。
|
||
taskId 是路径参数，用于标识具体任务。
|
||
路由 2：/api/asr-runner/report
|
||
接收 ASR 服务的心跳检测请求（self.heartbeat 处理）。
|
||
"""
|
||
|
||
logging.getLogger("werkzeug").disabled = True
|
||
threading.Thread(
|
||
target=self.app.run, args=("0.0.0.0", port), daemon=True
|
||
).start()
|
||
self.mutex = threading.Lock()
|
||
self.finished = threading.Event()
|
||
self.product_avaiable = True
|
||
|
||
self.reset()
|
||
|
||
def reset(self):
|
||
self.begin_time = None
|
||
self.end_time = None
|
||
self.first_receive_time = None
|
||
self.last_heartbeat_time = None
|
||
self.app_on = False
|
||
self.para_seq = 0
|
||
self.finished.clear()
|
||
self.error: Optional[str] = None
|
||
self.last_recognition_result: Optional[StreamDataModel] = None
|
||
self.recognition_results: List[StreamDataModel] = []
|
||
|
||
def asr_callback(self, taskId: str):
|
||
if self.app_on is False:
|
||
abort(400)
|
||
body = request.get_json(silent=True) # 静默解析JSON，失败时返回None
|
||
if body is None:
|
||
abort(404)
|
||
try:
|
||
result = ResultModel.model_validate(body) #将回调的 JSON 数据解析为 ResultModel 对象，确保结构符合预期。
|
||
except ValidationError as e:
|
||
log.error("asr_callback: 结果格式错误: %s", e)
|
||
abort(404)
|
||
|
||
#处理任务完成状态（FINISHED）
|
||
if result.status == "FINISHED":
|
||
with self.mutex:
|
||
self.stop()
|
||
return "ok"
|
||
#处理非运行状态（非 RUNNING）
|
||
if result.status != "RUNNING":
|
||
log.error(
|
||
"asr_callback: 结果状态错误: %s, message: %s",
|
||
result.status,
|
||
result.message,
|
||
)
|
||
abort(404)
|
||
|
||
recognition_result = result.recognition_results
|
||
if recognition_result is None:
|
||
log.error("asr_callback: 结果中没有recognition_results字段")
|
||
abort(404)
|
||
|
||
with self.mutex:
|
||
if not self.app_on:
|
||
log.error("asr_callback: 应用已结束")
|
||
abort(400)
|
||
|
||
if recognition_result.para_seq < self.para_seq:
|
||
error = "asr_callback: 结果中para_seq小于上一次的: %d < %d" % (
|
||
recognition_result.para_seq,
|
||
self.para_seq,
|
||
)
|
||
log.error(error)
|
||
if self.error is None:
|
||
self.error = error
|
||
self.stop()
|
||
abort(404)
|
||
if recognition_result.para_seq > self.para_seq + 1:
|
||
error = (
|
||
"asr_callback: 结果中para_seq大于上一次的+1 \
|
||
说明存在para_seq = %d没有final_result为True确认"
|
||
% (self.para_seq + 1,)
|
||
)
|
||
log.error(error)
|
||
if self.error is None:
|
||
self.error = error
|
||
self.stop()
|
||
abort(404)
|
||
if (
|
||
self.last_recognition_result is not None
|
||
and recognition_result.start_time
|
||
< self.last_recognition_result.end_time
|
||
):
|
||
error = "asr_callback: 结果中start_time小于上一次的end_time: %s < %s" % (
|
||
recognition_result.start_time,
|
||
self.last_recognition_result.end_time,
|
||
)
|
||
log.error(error)
|
||
if self.error is None:
|
||
self.error = error
|
||
self.stop()
|
||
abort(404)
|
||
|
||
self.recognition_results.append(recognition_result)
|
||
if recognition_result.final_result is True:
|
||
self.para_seq = recognition_result.para_seq
|
||
if self.last_recognition_result is None:
|
||
self.first_receive_time = time.time()
|
||
self.last_recognition_result = recognition_result
|
||
|
||
return "ok"
|
||
|
||
"""
|
||
def heartbeat(self):
|
||
if self.app_on is False:
|
||
abort(400)
|
||
body = request.get_json(silent=True)
|
||
if body is None:
|
||
abort(404)
|
||
status = body.get("status")
|
||
if status != "RUNNING":
|
||
message = body.get("message", "")
|
||
if message:
|
||
message = ", message: " + message
|
||
log.error("heartbeat: 状态错误: %s%s", status, message)
|
||
return "ok"
|
||
|
||
with self.mutex:
|
||
self.last_heartbeat_time = time.time()
|
||
return "ok"
|
||
|
||
"""
|
||
|
||
def predict(
|
||
self,
|
||
language: Optional[str],
|
||
audio_file: str,
|
||
audio_duration: float,
|
||
task_id: str,
|
||
):
|
||
#使用互斥锁确保线程安全
|
||
with self.mutex:
|
||
if self.app_on:
|
||
log.error("上一音频尚未完成处理，流程出现异常")
|
||
raise StopException()
|
||
self.reset()
|
||
self.app_on = True
|
||
|
||
#请求URL：self.sut_url + "/predict"（如 http://localhost:8080/predict）
|
||
resp = requests.post(
|
||
self.sut_url + "/predict",
|
||
data={
|
||
"language": language,
|
||
"taskId": task_id,
|
||
"progressCallbackUrl": "http://%s:%d/api/asr/batch-callback/%s"
|
||
% (MY_POD_IP, self.port, task_id),
|
||
"heartbeatUrl": "http://%s:%d/api/asr-runner/report" % (MY_POD_IP, self.port),
|
||
},
|
||
files={"file": (audio_file, open(audio_file, "rb"))},
|
||
timeout=60,
|
||
)
|
||
|
||
#响应处理
|
||
if resp.status_code != 200:
|
||
log.error("/predict接口返回http code %s", resp.status_code)
|
||
raise StopException()
|
||
resp.raise_for_status()
|
||
|
||
status = resp.json().get("status")
|
||
if status != "OK":
|
||
log.error("/predict接口返回非OK状态: %s", status)
|
||
raise StopException()
|
||
#辅助线程
|
||
threading.Thread(
|
||
target=self.dead_line_check, args=(audio_duration,), daemon=True
|
||
).start()
|
||
threading.Thread(target=self.heartbeat_check, daemon=True).start()
|
||
|
||
def dead_line_check(self, audio_duration: float):
|
||
begin_time = time.time()
|
||
self.begin_time = begin_time
|
||
|
||
# 初始化 10s 延迟检测
|
||
self.sleep_to(begin_time + 10)
|
||
with self.mutex:
|
||
if self.last_recognition_result is None:
|
||
error = "首包延迟内未收到返回"
|
||
log.error(error)
|
||
if self.error is None:
|
||
self.error = error
|
||
self.stop()
|
||
return
|
||
|
||
# 第一次30s检测
|
||
next_checktime = begin_time + 30
|
||
ddl = begin_time + max((audio_duration / 6) + 10, 30)
|
||
while time.time() < ddl:
|
||
self.sleep_to(next_checktime)
|
||
with self.mutex:
|
||
if self.finished.is_set():
|
||
return
|
||
if self.last_recognition_result is None:
|
||
error = "检测追赶线过程中获取最后一次识别结果异常"
|
||
log.error(error)
|
||
if self.error is None:
|
||
self.error = error
|
||
self.stop()
|
||
return
|
||
last_end_time = self.last_recognition_result.end_time
|
||
expect_end_time = (next_checktime - begin_time - 30) * 5.4
|
||
if last_end_time < expect_end_time:
|
||
log.warning(
|
||
"识别时间位置 %s 被死亡追赶线 %s 已追上，将置为产品不可用",
|
||
last_end_time,
|
||
expect_end_time,
|
||
)
|
||
self.product_avaiable = False
|
||
self.sleep_to(ddl)
|
||
break
|
||
next_checktime = last_end_time / 5.4 + begin_time + 30 + 1
|
||
next_checktime = min(next_checktime, ddl)
|
||
with self.mutex:
|
||
if self.finished.is_set():
|
||
return
|
||
|
||
log.warning("识别速度rtf低于1/6, 将置为产品不可用")
|
||
self.product_avaiable = False
|
||
self.sleep_to(begin_time + max((audio_duration / 3) + 10, 30))
|
||
with self.mutex:
|
||
if self.finished.is_set():
|
||
return
|
||
error = "处理时间超过ddl %s " % (ddl - begin_time)
|
||
log.error(error)
|
||
if self.error is None:
|
||
self.error = error
|
||
self.stop()
|
||
return
|
||
|
||
def heartbeat_check(self):
|
||
self.last_heartbeat_time = time.time()
|
||
while True:
|
||
with self.mutex:
|
||
if self.finished.is_set():
|
||
return
|
||
if time.time() - self.last_heartbeat_time > 30:
|
||
error = "asr_runner 心跳超时 %s" % (
|
||
time.time() - self.last_heartbeat_time
|
||
)
|
||
log.error(error)
|
||
if self.error is None:
|
||
self.error = error
|
||
self.stop()
|
||
return
|
||
time.sleep(5)
|
||
|
||
def sleep_to(self, to: float):
|
||
seconds = to - time.time()
|
||
if seconds <= 0:
|
||
return
|
||
time.sleep(seconds)
|
||
|
||
def stop(self):
|
||
self.end_time = time.time()
|
||
self.finished.set()
|
||
self.app_on = False
|
||
|
||
def evaluate(self, query_data: QueryData):
|
||
log.info("开始评估")
|
||
if (
|
||
self.begin_time is None
|
||
or self.end_time is None
|
||
or self.first_receive_time is None
|
||
):
|
||
if self.begin_time is None:
|
||
log.error("评估流程异常 无开始时间")
|
||
if self.end_time is None:
|
||
log.error("评估流程异常 无结束时间")
|
||
if self.first_receive_time is None:
|
||
log.error("评估流程异常 无首次接收时间")
|
||
raise StopException()
|
||
rtf = max(self.end_time - self.begin_time - 10, 0) / query_data.duration
|
||
first_receive_delay = max(self.first_receive_time - self.begin_time, 0)
|
||
query_count = 1
|
||
voice_count = len(query_data.voice)
|
||
preds = self.recognition_results
|
||
self.recognition_results = list(
|
||
filter(lambda x: x.final_result, self.recognition_results)
|
||
)
|
||
(
|
||
pred_punctuation_num,
|
||
label_punctuation_num,
|
||
pred_sentence_punctuation_num,
|
||
label_setence_punctuation_num,
|
||
) = evaluate_punctuation(query_data, self.recognition_results)
|
||
|
||
(
|
||
cer,
|
||
_,
|
||
align_start,
|
||
align_end,
|
||
first_word_distance_sum,
|
||
last_word_distance_sum,
|
||
) = evaluate_editops(query_data, self.recognition_results)
|
||
|
||
if align_start[300] / voice_count < 0.8:
|
||
log.warning(
|
||
"评估结果首字300ms对齐率 %s < 0.8, 将置为产品不可用",
|
||
align_start[300] / voice_count,
|
||
)
|
||
self.product_avaiable = False
|
||
|
||
return EvaluateResult(
|
||
lang=query_data.lang,
|
||
cer=cer,
|
||
align_start=align_start,
|
||
align_end=align_end,
|
||
first_word_distance_sum=first_word_distance_sum,
|
||
last_word_distance_sum=last_word_distance_sum,
|
||
rtf=rtf,
|
||
first_receive_delay=first_receive_delay,
|
||
query_count=query_count,
|
||
voice_count=voice_count,
|
||
pred_punctuation_num=pred_punctuation_num,
|
||
label_punctuation_num=label_punctuation_num,
|
||
pred_sentence_punctuation_num=pred_sentence_punctuation_num,
|
||
label_setence_punctuation_num=label_setence_punctuation_num,
|
||
preds=preds,
|
||
label=query_data,
|
||
)
|