Blackbox MeloTTS created

src/blackbox/blackbox_factory.py

@@ -13,6 +13,7 @@ from .text_and_image import TextAndImage
 from .chroma_query import ChromaQuery
 from .chroma_upsert import ChromaUpsert
 from .chroma_chat import ChromaChat
+from .melotts import MeloTTS
 from .vlms import VLMS
 from injector import inject, singleton
 
@@ -36,6 +37,7 @@ class BlackboxFactory:
                  chroma_query: ChromaQuery,
                  chroma_upsert: ChromaUpsert,
                  chroma_chat: ChromaChat,
+                 melotts: MeloTTS,
                  vlms: VLMS) -> None:
         self.models["audio_to_text"] = audio_to_text
         self.models["text_to_audio"] = text_to_audio
@@ -51,6 +53,7 @@ class BlackboxFactory:
         self.models["chroma_query"] = chroma_query
         self.models["chroma_upsert"] = chroma_upsert
         self.models["chroma_chat"] = chroma_chat
+        self.models["melotts"] = melotts
         self.models["vlms"] = vlms
 
     def __call__(self, *args, **kwargs):
@@ -59,5 +62,5 @@ class BlackboxFactory:
     def call_blackbox(self, blackbox_name: str) -> Blackbox:
         model = self.models.get(blackbox_name)
         if model is None:
-            raise ValueError("Invalid blockbox type")
+            raise ValueError("Invalid Blackbox Type...")
         return model

src/blackbox/melotts.py

@@ -0,0 +1,47 @@
+import io
+import time
+
+import requests
+from fastapi import Request, Response, status
+from fastapi.responses import JSONResponse
+from injector import inject
+from injector import singleton
+
+from ..configuration import MeloConf
+from .blackbox import Blackbox
+
+
+@singleton
+class MeloTTS(Blackbox):
+    melotts: str
+
+    @inject
+    def __init__(self, melo_config: MeloConf) -> None:
+        self.melotts = melo_config.melotts
+
+    def __call__(self, *args, **kwargs):
+        return self.processing(*args, **kwargs)
+
+    def valid(self, *args, **kwargs) -> bool:
+        text = args[0]
+        return isinstance(text, str)
+
+    def processing(self, *args, **kwargs) -> io.BytesIO:
+        text = args[0]
+        current_time = time.time()
+        audio = self.tts_service.read(text)
+        print("#### MeloTTS Service consume : ", (time.time()-current_time))
+        return audio
+
+    async def fast_api_handler(self, request: Request) -> Response:
+        try:
+            data = await request.json()
+        except:
+            return JSONResponse(content={"error": "json parse error"}, status_code=status.HTTP_400_BAD_REQUEST)
+        text = data.get("text")
+        if text is None:
+            return JSONResponse(content={"error": "text is required"}, status_code=status.HTTP_400_BAD_REQUEST)
+        by = self.processing(text)
+        return Response(content=by.read(), media_type="audio/mp3", headers={"Content-Disposition": "attachment; filename=audio.mp3"})
+
+