mirror of
https://github.com/BoardWare-Genius/jarvis-models.git
synced 2025-12-13 16:53:24 +00:00
45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
from io import BytesIO
|
|
from typing import Any, Coroutine
|
|
|
|
from fastapi import Request, Response, status
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from ..asr.rapid_paraformer.utils import read_yaml
|
|
from ..asr.rapid_paraformer import RapidParaformer
|
|
from .blackbox import Blackbox
|
|
from injector import singleton, inject
|
|
|
|
@singleton
|
|
class ASR(Blackbox):
|
|
|
|
@inject
|
|
def __init__(self,path = ".env.yaml") -> None:
|
|
config = read_yaml(path)
|
|
self.paraformer = RapidParaformer(config)
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
return self.processing(*args, **kwargs)
|
|
|
|
async def processing(self, *args, **kwargs):
|
|
data = args[0]
|
|
results = self.paraformer([BytesIO(data)])
|
|
if len(results) == 0:
|
|
return None
|
|
return results[0]
|
|
|
|
def valid(self, data: any) -> bool:
|
|
if isinstance(data, bytes):
|
|
return True
|
|
return False
|
|
|
|
async def fast_api_handler(self, request: Request) -> Response:
|
|
data = (await request.form()).get("audio")
|
|
if data is None:
|
|
self.logger.warn("asr bag request","type", "fast_api_handler", "api", "asr")
|
|
return JSONResponse(content={"error": "data is required"}, status_code=status.HTTP_400_BAD_REQUEST)
|
|
d = await data.read()
|
|
try:
|
|
txt = await self.processing(d)
|
|
except ValueError as e:
|
|
return JSONResponse(content={"error": str(e)}, status_code=status.HTTP_400_BAD_REQUEST)
|
|
return JSONResponse(content={"text": txt}, status_code=status.HTTP_200_OK) |