diff --git a/lollms/server/endpoints/lollms_asr.py b/lollms/server/endpoints/lollms_asr.py new file mode 100644 index 0000000..11b1e4c --- /dev/null +++ b/lollms/server/endpoints/lollms_asr.py @@ -0,0 +1,156 @@ +""" +project: lollms_webui +file: lollms_asr.py +author: ParisNeo +description: + This module contains a set of FastAPI routes that provide information about the Lord of Large Language and Multimodal Systems (LoLLMs) Web UI + application. These routes allow users to + +""" +from fastapi import APIRouter, Request, UploadFile, File, HTTPException +from lollms_webui import LOLLMSWebUI +from pydantic import BaseModel +from starlette.responses import StreamingResponse +from lollms.types import MSG_TYPE +from lollms.main_config import BaseConfig +from lollms.utilities import detect_antiprompt, remove_text_from_string, trace_exception, find_first_available_file_index, add_period, PackageManager +from lollms.security import sanitize_path, validate_path +from pathlib import Path +from ascii_colors import ASCIIColors +import os +import platform + +# ----------------------- Defining router and main class ------------------------------ + +router = APIRouter() +lollmsElfServer:LOLLMSWebUI = LOLLMSWebUI.get_instance() + +class LollmsAudio2TextRequest(BaseModel): + text: str + voice: str = None + fn:str = None +# ----------------------- voice ------------------------------ +@router.post("/asr/audio2test") +async def text2Audio(request: LollmsAudio2TextRequest): + """ + Executes Python code and returns the output. + + :param request: The HTTP request object. + :return: A JSON response with the status of the operation. + """ + if lollmsElfServer.config.headless_server_mode: + return {"status":False,"error":"Code execution is blocked when in headless mode for obvious security reasons!"} + + if lollmsElfServer.config.host!="localhost" and lollmsElfServer.config.host!="127.0.0.1": + return {"status":False,"error":"Code execution is blocked when the server is exposed outside for very obvious reasons!"} + + if request.fn: + request.fn = os.path.realpath(str((lollmsElfServer.lollms_paths.personal_outputs_path/"audio_out")/request.fn)) + validate_path(request.fn,[str(lollmsElfServer.lollms_paths.personal_outputs_path/"audio_out")]) + + try: + # Get the JSON data from the POST request. + try: + from lollms.services.asr.lollms_asr import LollmsASR + voice=lollmsElfServer.config.asr_current_voice + if lollmsElfServer.asr is None: + voice=lollmsElfServer.config.asr_current_voice + if voice!="main_voice": + voices_folder = lollmsElfServer.lollms_paths.custom_voices_path + else: + voices_folder = Path(__file__).parent.parent.parent/"services/asr/voices" + + lollmsElfServer.asr = LollmsASR( + lollmsElfServer, + voices_folder=voices_folder, + voice_samples_path=Path(__file__).parent/"voices", + asr_base_url= lollmsElfServer.config.asr_base_url + ) + except Exception as ex: + return {"url": None, "error":f"{ex}"} + + voice=lollmsElfServer.config.asr_current_voice if request.voice is None else request.voice + index = find_first_available_file_index(lollmsElfServer.asr.output_folder, "voice_sample_",".wav") + output_fn=f"voice_sample_{index}.wav" if request.fn is None else request.fn + if voice is None: + voice = "main_voice" + lollmsElfServer.info("Starting to build voice") + try: + from lollms.services.asr.lollms_asr import LollmsASR + # If the personality has a voice, then use it + personality_audio:Path = lollmsElfServer.personality.personality_package_path/"audio" + if personality_audio.exists() and len([v for v in personality_audio.iterdir()])>0: + voices_folder = personality_audio + elif voice!="main_voice": + voices_folder = lollmsElfServer.lollms_paths.custom_voices_path + else: + voices_folder = Path(__file__).parent.parent.parent/"services/asr/voices" + if lollmsElfServer.asr is None: + lollmsElfServer.asr = LollmsASR( + lollmsElfServer, + voices_folder=voices_folder, + voice_samples_path=Path(__file__).parent/"voices", + asr_base_url= lollmsElfServer.config.asr_base_url, + ) + if lollmsElfServer.asr.ready: + language = lollmsElfServer.config.asr_current_language# convert_language_name() + lollmsElfServer.asr.set_speaker_folder(voices_folder) + preprocessed_text= add_period(request.text) + voice_file = [v for v in voices_folder.iterdir() if v.stem==voice and v.suffix==".wav"] + if len(voice_file)==0: + return {"status":False,"error":"Voice not found"} + lollmsElfServer.asr.tts_to_audio(preprocessed_text, voice_file[0].name, f"{output_fn}", language=language) + else: + lollmsElfServer.InfoMessage("asr is not up yet.\nPlease wait for it to load then try again. This may take some time.") + return {"status":False, "error":"Service not ready yet"} + except Exception as ex: + trace_exception(ex) + return {"url": None} + except Exception as ex: + trace_exception(ex) + lollmsElfServer.error(ex) + return {"status":False,"error":str(ex)} + +@router.get("/install_asr") +def install_asr(): + try: + if lollmsElfServer.config.headless_server_mode: + return {"status":False,"error":"Service installation is blocked when in headless mode for obvious security reasons!"} + + if lollmsElfServer.config.host!="localhost" and lollmsElfServer.config.host!="127.0.0.1": + return {"status":False,"error":"Service installation is blocked when the server is exposed outside for very obvious reasons!"} + + from lollms.services.asr.lollms_asr import install_asr + lollmsElfServer.ShowBlockingMessage("Installing ASR api server\nPlease stand by") + install_asr(lollmsElfServer) + lollmsElfServer.HideBlockingMessage() + return {"status":True} + except Exception as ex: + trace_exception(ex) + lollmsElfServer.HideBlockingMessage() + return {"status":False, 'error':str(ex)} + +@router.get("/start_asr") +def start_asr(): + try: + lollmsElfServer.ShowBlockingMessage("Starting ASR api server\nPlease stand by") + from lollms.services.asr.lollms_asr import LollmsASR + if lollmsElfServer.asr is None: + lollmsElfServer.asr = LollmsASR( + lollmsElfServer, + voice_samples_path=Path(__file__).parent/"voices", + asr_base_url= lollmsElfServer.config.asr_base_url, + ) + lollmsElfServer.HideBlockingMessage() + except Exception as ex: + trace_exception(ex) + lollmsElfServer.HideBlockingMessage() + return {"url": None, "error":f"{ex}"} + + +@router.get("/asr_is_ready") +def asr_is_ready(): + if hasattr(lollmsElfServer,'sd') and lollmsElfServer.sd is not None: + if lollmsElfServer.sd.ready: + return {"status":True} + return {"status":False} diff --git a/lollms/services/asr/lollms_asr.py b/lollms/services/asr/lollms_asr.py index ead2aeb..3500f84 100644 --- a/lollms/services/asr/lollms_asr.py +++ b/lollms/services/asr/lollms_asr.py @@ -110,14 +110,9 @@ class LollmsASR: asr_base_url=None, share=False, max_retries=20, - voices_folder=None, - voice_samples_path="", - wait_for_service=True, - use_deep_speed=False, - use_streaming_mode = True + wait_for_service=True ): self.generation_threads = [] - self.voices_folder = voices_folder self.ready = False if asr_base_url=="" or asr_base_url=="http://127.0.0.1:9000": asr_base_url = None @@ -125,9 +120,6 @@ class LollmsASR: lollms_paths = app.lollms_paths self.app = app root_dir = lollms_paths.personal_path - self.voice_samples_path = voice_samples_path - self.use_deep_speed = use_deep_speed - self.use_streaming_mode = use_streaming_mode # Store the path to the script if asr_base_url is None: