mirror of
https://github.com/BoardWare-Genius/jarvis-models.git
synced 2025-12-13 16:53:24 +00:00
47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
import logging
|
|
from fastapi import Request, Response,status
|
|
from fastapi.responses import JSONResponse
|
|
from injector import inject, singleton
|
|
|
|
from .asr import ASR
|
|
from .tesou import Tesou
|
|
from .tts import TTS
|
|
|
|
from .blackbox import Blackbox
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
@singleton
|
|
class AudioChat(Blackbox):
|
|
|
|
@inject
|
|
def __init__(self, asr: ASR, gpt: Tesou, tts: TTS):
|
|
logger.info("audio chat initint")
|
|
self.asr = asr
|
|
self.gpt = gpt
|
|
self.tts = tts
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
return self.processing(*args, **kwargs)
|
|
|
|
def valid(self, *args, **kwargs) -> bool :
|
|
data = args[0]
|
|
if isinstance(data, bytes):
|
|
return True
|
|
return False
|
|
|
|
async def processing(self, *args, **kwargs):
|
|
data = args[0]
|
|
text = await self.asr(data)
|
|
# TODO: ID
|
|
text = self.gpt("123", " " + text)
|
|
audio = self.tts(text)
|
|
return audio
|
|
|
|
async def fast_api_handler(self, request: Request) -> Response:
|
|
data = (await request.form()).get("audio")
|
|
if data is None:
|
|
return JSONResponse(content={"error": "data is required"}, status_code=status.HTTP_400_BAD_REQUEST)
|
|
d = await data.read()
|
|
by = await self.processing(d)
|
|
return Response(content=by.read(), media_type="audio/x-wav", headers={"Content-Disposition": "attachment; filename=audio.wav"}) |