# -*- coding: utf-8 -*-
|
"""配送视频门头/交付时刻检测推理服务（ONNX + ASR）。"""
|
import logging
|
import os
|
|
from fastapi import FastAPI
|
from fastapi.responses import JSONResponse
|
from pydantic import BaseModel
|
|
from app.asr import asr_available
|
from app.pipeline import get_registry, run_analyze
|
from app.schemas import AnalyzeRequest, HealthResponse
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s")
|
|
app = FastAPI(title="snapshot-infer", version="2.0.0")
|
|
|
def _camel_json(model: BaseModel) -> JSONResponse:
|
return JSONResponse(content=model.model_dump(by_alias=True, exclude_none=True))
|
|
|
@app.get("/health")
|
def health():
|
reg = get_registry()
|
body = HealthResponse(
|
status="ok" if reg.ready else "degraded",
|
model_version=reg.version,
|
onnx_storefront_loaded=reg.storefront.loaded if reg.storefront else False,
|
onnx_handover_loaded=reg.handover.loaded if reg.handover else False,
|
asr_available=asr_available(),
|
)
|
return _camel_json(body)
|
|
|
@app.post("/analyze")
|
def analyze(req: AnalyzeRequest):
|
return _camel_json(run_analyze(req))
|
