From ce11e4846e9aaddaf13ffbd44209b369a66edf8b Mon Sep 17 00:00:00 2001 From: erew123 <35898566+erew123@users.noreply.github.com> Date: Fri, 8 Dec 2023 14:54:45 +0000 Subject: [PATCH] Branding Change --- config.json | 2 +- modeldownload.py | 102 ++++++++--- script.py | 453 +++++++++++++++++++++++++++++++--------------- tts_server.py | 456 ++++++++++++++++++++++++++++++----------------- 4 files changed, 683 insertions(+), 330 deletions(-) diff --git a/config.json b/config.json index 7201e05..cd6371e 100644 --- a/config.json +++ b/config.json @@ -1 +1 @@ -{"activate": true, "autoplay": true, "deepspeed_activate": false, "delete_output_wavs": "Disabled", "ip_address": "127.0.0.1", "language": "English", "low_vram": false, "model_loaded": true, "model_name": "tts_models/multilingual/multi-dataset/xtts_v2", "output_folder_wav": "extensions/coqui_tts/outputs/", "port_number": "7851", "remove_trailing_dots": false, "show_text": true, "tts_method_api_local": false, "tts_method_api_tts": false, "tts_method_xtts_local": true, "voice": "female_01.wav"} \ No newline at end of file +{"activate": true, "autoplay": true, "branding": "AllTalk ", "deepspeed_activate": false, "delete_output_wavs": "Disabled", "ip_address": "127.0.0.1", "language": "English", "low_vram": false, "model_loaded": true, "model_name": "tts_models/multilingual/multi-dataset/xtts_v2", "output_folder_wav": "extensions/coqui_tts/outputs/", "port_number": "7851", "remove_trailing_dots": false, "show_text": true, "tts_method_api_local": false, "tts_method_api_tts": false, "tts_method_xtts_local": true, "voice": "female_01.wav"} \ No newline at end of file diff --git a/modeldownload.py b/modeldownload.py index cdbf9d9..3896abf 100644 --- a/modeldownload.py +++ b/modeldownload.py @@ -6,15 +6,29 @@ import importlib.metadata as metadata import json from packaging import version -# Use this_dir in the downloader script +############################################################## +#### LOAD PARAMS FROM CONFIG.JSON - REQUIRED FOR BRANDING #### +############################################################## +# STARTUP VARIABLE - Create "this_dir" variable as the current script directory this_dir = Path(__file__).parent.resolve() -# Define the path to the JSON file -config_file_path = this_dir / 'modeldownload.json' + +# load standard config file in and get settings +def load_config(file_path): + with open(file_path, "r") as config_file: + config = json.load(config_file) + return config + + +config_file_path = this_dir / "config.json" +params = load_config(config_file_path) + +# Define the path to the modeldownload config file file +modeldownload_config_file_path = this_dir / "modeldownload.json" # Check if the JSON file exists -if config_file_path.exists(): - with open(config_file_path, 'r') as config_file: +if modeldownload_config_file_path.exists(): + with open(modeldownload_config_file_path, "r") as config_file: settings = json.load(config_file) # Extract settings from the loaded JSON @@ -23,65 +37,101 @@ if config_file_path.exists(): files_to_download = settings.get("files_to_download", {}) else: # Default settings if the JSON file doesn't exist or is empty - print("[CoquiTTS Startup] \033[91mWarning\033[0m modeldownload.json is missing so please re-download it and save it in the coquii_tts main folder.") - print("[CoquiTTS Startup] \033[91mWarning\033[0m API Local and XTTSv2 Local will error unless this is corrected.") + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m modeldownload.json is missing so please re-download it and save it in the coquii_tts main folder." + ) + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m API Local and XTTSv2 Local will error unless this is corrected." + ) # Read the version specifier from requirements.txt -with open(this_dir / 'requirements.txt', 'r') as req_file: +with open(this_dir / "requirements.txt", "r") as req_file: requirements = req_file.readlines() tts_version_required = None for req in requirements: - if req.startswith('TTS=='): - tts_version_required = req.strip().split('==')[1] + if req.startswith("TTS=="): + tts_version_required = req.strip().split("==")[1] break if tts_version_required is None: - raise ValueError("[CoquiTTS Startup] \033[91mWarning\033[0m Could not find TTS version specifier in requirements.txt") + raise ValueError( + f"[{params['branding']}Startup] \033[91mWarning\033[0m Could not find TTS version specifier in requirements.txt" + ) + def create_directory_if_not_exists(directory): if not directory.exists(): directory.mkdir(parents=True) + def download_file(url, destination): response = requests.get(url, stream=True) - total_size_in_bytes = int(response.headers.get('content-length', 0)) + total_size_in_bytes = int(response.headers.get("content-length", 0)) block_size = 1024 # 1 Kibibyte - progress_bar = tqdm(total=total_size_in_bytes, unit='iB', unit_scale=True) - - with open(destination, 'wb') as file: + progress_bar = tqdm(total=total_size_in_bytes, unit="iB", unit_scale=True) + + with open(destination, "wb") as file: for data in response.iter_content(block_size): progress_bar.update(len(data)) file.write(data) progress_bar.close() + def check_tts_version(): try: tts_version = metadata.version("tts") - print(f"[CoquiTTS Startup] TTS version installed: \033[93m{tts_version}\033[0m") + print( + f"[{params['branding']}Startup] TTS version installed: \033[93m" + + tts_version + + "\033[0m" + ) if version.parse(tts_version) < version.parse(tts_version_required): - print(f"[CoquiTTS Startup] \033[91mWarning\033[0m TTS version is too old. Please upgrade to version \033[93m{tts_version_required}\033[0m or later.\033[0m") - print("[CoquiTTS Startup] \033[91mWarning\033[0m At your terminal/command prompt \033[94mpip install --upgrade tts\033[0m") + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m TTS version is too old. Please upgrade to version \033[93m" + + tts_version_required + + "\033[0m or later.\033[0m" + ) + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m At your terminal/command prompt \033[94mpip install --upgrade tts\033[0m" + ) else: - print("[CoquiTTS Startup] TTS version is up to date.") + print(f"[{params['branding']}Startup] TTS version is up to date.") except metadata.PackageNotFoundError: - print("[CoquiTTS Startup] \033[91mWarning\033[0m TTS is not installed.") + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m TTS is not installed." + ) + # Check and create directories if str(base_path) == "models": create_directory_if_not_exists(this_dir / base_path / model_path) else: create_directory_if_not_exists(base_path / model_path) - print("[CoquiTTS Startup] \033[94mInfo\033[0m Custom path set in \033[93mmodeldownload.json\033[0m. Using the following settings:") - print("[CoquiTTS Startup] \033[94mInfo\033[0m Base folder Path:\033[93m", base_path , "\033[0m") - print("[CoquiTTS Startup] \033[94mInfo\033[0m Model folder Path:\033[93m", model_path , "\033[0m") - print("[CoquiTTS Startup] \033[94mInfo\033[0m Full Path:\033[93m", base_path / model_path , "\033[0m") + print( + f"[{params['branding']}Startup] \033[94mInfo\033[0m Custom path set in \033[93mmodeldownload.json\033[0m. Using the following settings:" + ) + print( + f"[{params['branding']}Startup] \033[94mInfo\033[0m Base folder Path:\033[93m", + base_path, + "\033[0m", + ) + print( + f"[{params['branding']}Startup] \033[94mInfo\033[0m Model folder Path:\033[93m", + model_path, + "\033[0m", + ) + print( + f"[{params['branding']}Startup] \033[94mInfo\033[0m Full Path:\033[93m", + base_path / model_path, + "\033[0m", + ) # Download files if they don't exist -print("[CoquiTTS Startup] Checking Model is Downloaded.") +print("[" + params["branding"] + "Startup] Checking Model is Downloaded.") for filename, url in files_to_download.items(): if str(base_path) == "models": destination = this_dir / base_path / model_path / filename @@ -89,7 +139,7 @@ for filename, url in files_to_download.items(): destination = Path(base_path) / model_path / filename if not destination.exists(): - print(f"[CoquiTTS Startup] Downloading {filename}...") + print(f"[{params['branding']}Startup] Downloading " + filename + "...") download_file(url, destination) check_tts_version() diff --git a/script.py b/script.py index ef21aab..454bac1 100644 --- a/script.py +++ b/script.py @@ -19,22 +19,43 @@ from modules.logging_colors import logger from modules.ui import create_refresh_button from modules.utils import gradio -#Required for sentence splitting +############################################################## +#### LOAD PARAMS FROM CONFIG.JSON - REQUIRED FOR BRANDING #### +############################################################## +# STARTUP VARIABLE - Create "this_dir" variable as the current script directory +this_dir = Path(__file__).parent.resolve() + + +# load config file in and get settings +def load_config(file_path): + with open(file_path, "r") as config_file: + config = json.load(config_file) + return config + + +config_file_path = this_dir / "config.json" +params = load_config(config_file_path) + + +# Required for sentence splitting try: from TTS.api import TTS from TTS.utils.synthesizer import Synthesizer except ModuleNotFoundError: logger.error( - "[CoquiTTS Startup] \033[91mWarning\033[0m Could not find the TTS module. Make sure to install the requirements for the coqui_tts extension." - "[CoquiTTS Startup] \033[91mWarning\033[0m Linux / Mac:\npip install -r extensions/coqui_tts/requirements.txt\n" - "[CoquiTTS Startup] \033[91mWarning\033[0m Windows:\npip install -r extensions\\coqui_tts\\requirements.txt\n" - "[CoquiTTS Startup] \033[91mWarning\033[0m If you used the one-click installer, paste the command above in the terminal window launched after running the \"cmd_\" script. On Windows, that's \"cmd_windows.bat\"." + "[params['branding']]\033[91mWarning\033[0m Could not find the TTS module. Make sure to install the requirements for the " + + params["branding"] + + " extension." + "[params['branding']]\033[91mWarning\033[0m Linux / Mac:\npip install -r extensions/alltalk_tts/requirements.txt\n" + "[params['branding']]\033[91mWarning\033[0m Windows:\npip install -r extensions\\alltalk_tts\\requirements.txt\n" + "[params['branding']]\033[91mWarning\033[0m If you used the one-click installer, paste the command above in the terminal window launched after running the cmd_ script. On Windows, that's cmd_windows.bat." ) raise -#IMPORT - Attempt Importing DeepSpeed +# IMPORT - Attempt Importing DeepSpeed try: import deepspeed + deepspeed_installed = True except ImportError: deepspeed_installed = False @@ -42,41 +63,36 @@ except ImportError: ########################### #### STARTUP VARIABLES #### ########################### -#STARTUP VARIABLE - Create "this_dir" variable as the current script directory -this_dir = Path(__file__).parent.resolve() -#STARTUP VARIABLE - Import languges file for Gradio to be able to display them in the interface -with open(this_dir / 'languages.json', encoding='utf8') as f: +# STARTUP VARIABLE - Import languges file for Gradio to be able to display them in the interface +with open(this_dir / "languages.json", encoding="utf8") as f: languages = json.load(f) # Create a global lock process_lock = threading.Lock() -####################################### -#### LOAD PARAMS FROM CONFIG.JSON #### -####################################### -def load_config(file_path): - with open(file_path, 'r') as config_file: - config = json.load(config_file) - return config - -config_file_path = (this_dir / 'config.json') -params = load_config(config_file_path) def get_available_voices(): return sorted([voice.name for voice in Path(f"{this_dir}/voices").glob("*.wav")]) + ######################### #### LICENSE DISPLAY #### ######################### -#STARTUP Display Licence Information -print("[CoquiTTS Startup] \033[94mCoqui Public Model License\033[0m") -print("[CoquiTTS Startup] \033[94mhttps://coqui.ai/cpml.txt\033[0m") +# STARTUP Display Licence Information +print("[" + params["branding"] + "Startup] \033[94mCoqui Public Model License\033[0m") +print("[" + params["branding"] + "Startup] \033[94mhttps://coqui.ai/cpml.txt\033[0m") + ############################################ #### DELETE OLD OUTPUT WAV FILES IF SET #### ############################################ def delete_old_files(folder_path, days_to_keep): current_time = datetime.now() - print("[CoquiTTS Startup] Deletion of old output folder WAV files is currently enabled and set at", delete_output_wavs_setting) + print( + "[" + + params["branding"] + + "Startup] Deletion of old output folder WAV files is currently enabled and set at", + delete_output_wavs_setting, + ) for file_name in os.listdir(folder_path): file_path = os.path.join(folder_path, file_name) if os.path.isfile(file_path): @@ -85,6 +101,7 @@ def delete_old_files(folder_path, days_to_keep): if age > timedelta(days=days_to_keep): os.remove(file_path) + # Extract settings using params dictionary delete_output_wavs_setting = params["delete_output_wavs"] output_folder_wav = params["output_folder_wav"] @@ -92,25 +109,33 @@ output_folder_wav = os.path.normpath(output_folder_wav) # Check and perform file deletion if delete_output_wavs_setting.lower() == "disabled": - print("[CoquiTTS Startup] Old output wav file deletion is set to disabled.") + print( + "[" + + params["branding"] + + "Startup] Old output wav file deletion is set to disabled." + ) else: try: days_to_keep = int(delete_output_wavs_setting.split()[0]) delete_old_files(output_folder_wav, days_to_keep) except ValueError: - print("[CoquiTTS Startup] Invalid setting for deleting old wav files. Please use 'Disabled' or 'X Days' format.") + print( + f"[{params['branding']}Startup] Invalid setting for deleting old wav files. Please use 'Disabled' or 'X Days' format." + ) + ######################## #### STARTUP CHECKS #### ######################## -#STARTUP Checks routine +# STARTUP Checks routine def check_required_files(): this_dir = Path(__file__).parent.resolve() - download_script_path = this_dir / 'modeldownload.py' - subprocess.run(['python', str(download_script_path)]) - print("[CoquiTTS Startup] All required files are present.") + download_script_path = this_dir / "modeldownload.py" + subprocess.run(["python", str(download_script_path)]) + print(f"[{params['branding']}Startup] All required files are present.") -#STARTUP Call Check routine + +# STARTUP Call Check routine check_required_files() ################################################# @@ -124,72 +149,102 @@ elif params["tts_method_api_local"] == True: elif params["tts_method_xtts_local"] == True: gr_modelchoice = "XTTSv2 Local" + ###################### #### GRADIO STUFF #### ###################### def remove_tts_from_history(history): - for i, entry in enumerate(history['internal']): - history['visible'][i] = [history['visible'][i][0], entry[1]] + for i, entry in enumerate(history["internal"]): + history["visible"][i] = [history["visible"][i][0], entry[1]] return history + def toggle_text_in_history(history): - for i, entry in enumerate(history['visible']): + for i, entry in enumerate(history["visible"]): visible_reply = entry[1] - if visible_reply.startswith('')[0]}\n\n{reply}"] + if visible_reply.startswith("')[0]}\n\n{reply}", + ] else: - history['visible'][i] = [history['visible'][i][0], f"{visible_reply.split('')[0]}"] + history["visible"][i] = [ + history["visible"][i][0], + f"{visible_reply.split('')[0]}", + ] return history + def history_modifier(history): # Remove autoplay from the last reply - if len(history['internal']) > 0: - history['visible'][-1] = [ - history['visible'][-1][0], - history['visible'][-1][1].replace('controls autoplay>', 'controls>') + if len(history["internal"]) > 0: + history["visible"][-1] = [ + history["visible"][-1][0], + history["visible"][-1][1].replace("controls autoplay>", "controls>"), ] return history + ###################################### #### SUBPROCESS/WEBSERVER STARTUP #### ###################################### base_url = f"http://{params['ip_address']}:{params['port_number']}" -script_path = this_dir / 'tts_server.py' +script_path = this_dir / "tts_server.py" + def signal_handler(sig, frame): - print("[CoquiTTS Shutdown] \033[94mReceived Ctrl+C, terminating subprocess\033[92m") + print( + f"[{params['branding']}Shutdown] \033[94mReceived Ctrl+C, terminating subprocess\033[92m" + ) if process.poll() is None: process.terminate() process.wait() # Wait for the subprocess to finish sys.exit(0) + # Attach the signal handler to the SIGINT signal (Ctrl+C) signal.signal(signal.SIGINT, signal_handler) # Start the subprocess -process = subprocess.Popen(['python', script_path]) +process = subprocess.Popen(["python", script_path]) # Check if the subprocess has started successfully if process.poll() is None: - print("[CoquiTTS Startup] TTS Subprocess starting") - print("[CoquiTTS Startup] Readme available here:", f"http://{params['ip_address']}:{params['port_number']}") + print(f"[{params['branding']}Startup] TTS Subprocess starting") + print( + f"[{params['branding']}Startup] Readme available here:", + f"http://{params['ip_address']}:{params['port_number']}", + ) else: - print("[CoquiTTS Startup] \033[91mWarning\033[0m TTS Subprocess Webserver failing to start process") - print("[CoquiTTS Startup] \033[91mWarning\033[0m It could be that you have something on port:", params['port_number']) - print("[CoquiTTS Startup] \033[91mWarning\033[0m Or you have not started in a Python environement with all the necesssary bits installed") - print("[CoquiTTS Startup] \033[91mWarning\033[0m Check you are starting Text-generation-webui with either the start_xxxxx file or the Python environment with cmd_xxxxx file.") - print("[CoquiTTS Startup] \033[91mWarning\033[0m xxxxx is the type of OS you are on e.g. windows, linux or mac.") - print("[CoquiTTS Startup] \033[91mWarning\033[0m Alternatively, you could check no other Python processes are running that shouldnt be e.g. Restart your computer is the simple way.") - #Cleanly kill off this script, but allow text-generation-webui to keep running, albeit without this coquii_tts + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m TTS Subprocess Webserver failing to start process" + ) + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m It could be that you have something on port:", + params["port_number"], + ) + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m Or you have not started in a Python environement with all the necesssary bits installed" + ) + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m Check you are starting Text-generation-webui with either the start_xxxxx file or the Python environment with cmd_xxxxx file." + ) + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m xxxxx is the type of OS you are on e.g. windows, linux or mac." + ) + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m Alternatively, you could check no other Python processes are running that shouldnt be e.g. Restart your computer is the simple way." + ) + # Cleanly kill off this script, but allow text-generation-webui to keep running, albeit without this alltalk_tts sys.exit(1) timeout = 60 # Adjust the timeout as needed # Introduce a delay before starting the check loop -time.sleep(25) #Wait 25 secs before checking if the tts_server.py has started up. +time.sleep(25) # Wait 25 secs before checking if the tts_server.py has started up. start_time = time.time() while time.time() - start_time < timeout: try: @@ -198,17 +253,22 @@ while time.time() - start_time < timeout: break except requests.RequestException as e: # Print the exception for debugging purposes - print(f"[CoquiTTS Startup] \033[91mWarning\033[0m TTS Subprocess has NOT started up yet, Will keep trying for 60 seconds maximum") + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m TTS Subprocess has NOT started up yet, Will keep trying for 60 seconds maximum" + ) time.sleep(1) else: - print("[CoquiTTS Startup] Startup timed out. Check the server logs for more information.") - #Cleanly kill off this script, but allow text-generation-webui to keep running, albeit without this coquii_tts + print( + f"[{params['branding']}Startup] Startup timed out. Check the server logs for more information." + ) + # Cleanly kill off this script, but allow text-generation-webui to keep running, albeit without this alltalk_tts sys.exit(1) + ##################################### #### MODEL LOADING AND UNLOADING #### ##################################### -#MODEL - Swap model based on Gradio selection API TTS, API Local, XTTSv2 Local +# MODEL - Swap model based on Gradio selection API TTS, API Local, XTTSv2 Local def send_reload_request(tts_method): try: params["model_loaded"] = False @@ -224,13 +284,16 @@ def send_reload_request(tts_method): return json_response except requests.exceptions.RequestException as e: # Handle the HTTP request error - print(f"[CoquiTTS Server] \033[91mWarning\033[0m Error during request to webserver process: {e}") + print( + f"[{params['branding']}Server] \033[91mWarning\033[0m Error during request to webserver process: {e}" + ) return {"status": "error", "message": str(e)} + ################## #### LOW VRAM #### ################## -#LOW VRAM - Gradio Checkbox handling +# LOW VRAM - Gradio Checkbox handling def send_lowvram_request(low_vram): try: params["model_loaded"] = False @@ -246,13 +309,16 @@ def send_lowvram_request(low_vram): return json_response except requests.exceptions.RequestException as e: # Handle the HTTP request error - print(f"[CoquiTTS Server] \033[91mWarning\033[0m Error during request to webserver process: {e}") + print( + f"[{params['branding']}Server] \033[91mWarning\033[0m Error during request to webserver process: {e}" + ) return {"status": "error", "message": str(e)} + ################### #### DeepSpeed #### ################### -#DEEPSPEED - Reload the model when DeepSpeed checkbox is enabled/disabled +# DEEPSPEED - Reload the model when DeepSpeed checkbox is enabled/disabled def send_deepspeed_request(deepspeed_param): try: params["model_loaded"] = False @@ -268,54 +334,63 @@ def send_deepspeed_request(deepspeed_param): return json_response except requests.exceptions.RequestException as e: # Handle the HTTP request error - print(f"[CoquiTTS Server] \033[91mWarning\033[0m Error during request to webserver process: {e}") + print( + f"[{params['branding']}Server] \033[91mWarning\033[0m Error during request to webserver process: {e}" + ) return {"status": "error", "message": str(e)} # DEEPSPEED - Display DeepSpeed Checkbox Yes or No deepspeed_condition = params["tts_method_xtts_local"] == "True" and deepspeed_installed + ######################## #### TTS GENERATION #### ######################## -#PREVIEW VOICE - Generate Random Sentence if Voice Preview box is empty +# PREVIEW VOICE - Generate Random Sentence if Voice Preview box is empty def random_sentence(): - with open(Path("extensions/coqui_tts/harvard_sentences.txt")) as f: + with open(Path("extensions/alltalk_tts/harvard_sentences.txt")) as f: return random.choice(list(f)) -#PREVIEW VOICE- Generate TTS Function + +# PREVIEW VOICE- Generate TTS Function def voice_preview(string): - if not params['activate']: - return string + if not params["activate"]: + return string # Clean the string, capture model not loaded, and move model to cuda if needed - cleaned_string = before_audio_generation(string, params) + cleaned_string = before_audio_generation(string, params) if cleaned_string is None: - return - string = cleaned_string + return + string = cleaned_string # Setup the output file output_file = Path(params["output_folder_wav"]) / "voice_preview.wav" # Generate the audio - language_code = languages.get(params["language"]) + language_code = languages.get(params["language"]) # Convert the WindowsPath object to a string before using it in JSON payload - output_file_str = output_file.as_posix() + output_file_str = output_file.as_posix() # Lock before making the generate request with process_lock: - generate_response = send_generate_request(string, params["voice"], language_code, output_file_str) + generate_response = send_generate_request( + string, params["voice"], language_code, output_file_str + ) # Check if lock is already acquired if process_lock.locked(): - print("[CoquiTTS Model] \033[91mWarning\033[0m Audio generation is already in progress. Please wait.") + print( + f"[{params['branding']}Model] \033[91mWarning\033[0m Audio generation is already in progress. Please wait." + ) return if generate_response.get("status") == "generate-success": # Handle Gradio and playback - autoplay = 'autoplay' if params['autoplay'] else '' + autoplay = "autoplay" if params["autoplay"] else "" return f'' else: # Handle the case where audio generation was not successful - return f"Audio generation failed. Status: {generate_response.get('status')}" + return f"[{params['branding']}Server] Audio generation failed. Status: {generate_response.get('status')}" -#STANDARD VOICE - Generate TTS Function + +# STANDARD VOICE - Generate TTS Function def output_modifier(string, state): - if not params['activate']: + if not params["activate"]: return string original_string = string cleaned_string = before_audio_generation(string, params) @@ -323,50 +398,63 @@ def output_modifier(string, state): return string = cleaned_string # Setup the output file - output_file = Path(f'{params["output_folder_wav"]}/{state["character_menu"]}_{int(time.time())}.wav') + output_file = Path( + f'{params["output_folder_wav"]}/{state["character_menu"]}_{int(time.time())}.wav' + ) # Generate the audio and handle the response language_code = languages.get(params["language"]) # Convert the WindowsPath object to a string before using it in JSON payload output_file_str = output_file.as_posix() - # Lock before making the generate request + # Lock before making the generate request with process_lock: - generate_response = send_generate_request(string, params["voice"], language_code, output_file_str) - + generate_response = send_generate_request( + string, params["voice"], language_code, output_file_str + ) + # Check if lock is already acquired if process_lock.locked(): - print("[CoquiTTS Model] \033[91mWarning\033[0m Audio generation is already in progress. Please wait.") + print( + f"[{params['branding']}Model] \033[91mWarning\033[0m Audio generation is already in progress. Please wait." + ) return - + if generate_response.get("status") == "generate-success": audio_path = generate_response.get("data", {}).get("audio_path") if audio_path: # Handle Gradio and playback - autoplay = 'autoplay' if params['autoplay'] else '' + autoplay = "autoplay" if params["autoplay"] else "" string = f'' - - if params['show_text']: - string += f'\n\n{original_string}' + + if params["show_text"]: + string += f"\n\n{original_string}" shared.processing_message = "*Is typing...*" - + return string else: - print("No audio path in the response.") + print( + f"[{params['branding']}Server] \033[91mWarning\033[0m No audio path in the response." + ) else: - print("[CoquiTTS Server] \033[91mWarning\033[0m Audio generation failed. Status:", generate_response.get("message")) - + print( + f"[{params['branding']}Server] \033[91mWarning\033[0m Audio generation failed. Status:", + generate_response.get("message"), + ) + + def send_generate_request(text, voice, language, output_file): url = f"{base_url}/api/generate" payload = { "text": text, "voice": voice, "language": language, - "output_file": output_file + "output_file": output_file, } headers = {"Content-Type": "application/json"} response = requests.post(url, json=payload, headers=headers) return response.json() + ############################################################# #### TTS STRING CLEANING & PROCESSING PRE SENDING TO TTS #### ############################################################# @@ -375,12 +463,13 @@ def preprocess(raw_input): # raw_input = raw_input.strip("\"") return raw_input + def new_split_into_sentences(self, text): sentences = self.seg.segment(text) - if params['remove_trailing_dots']: + if params["remove_trailing_dots"]: sentences_without_dots = [] for sentence in sentences: - if sentence.endswith('.') and not sentence.endswith('...'): + if sentence.endswith(".") and not sentence.endswith("..."): sentence = sentence[:-1] sentences_without_dots.append(sentence) @@ -389,104 +478,182 @@ def new_split_into_sentences(self, text): else: return sentences + Synthesizer.split_into_sentences = new_split_into_sentences -#TTS STRING CLEANING - Clean String before TTS generation (Called from voice_preview and output_modifer) + +# TTS STRING CLEANING - Clean String before TTS generation (Called from voice_preview and output_modifer) def before_audio_generation(string, params): - #Check Model is loaded into cuda or cpu and error if not + # Check Model is loaded into cuda or cpu and error if not if not params["model_loaded"]: - print("[CoquiTTS Model] \033[91mWarning\033[0m Model is still loading, please wait before trying to generate TTS") + print( + f"[{params['branding']}Model] \033[91mWarning\033[0m Model is still loading, please wait before trying to generate TTS" + ) return string = html.unescape(string) or random_sentence() # Replace double quotes with single, asterisks, carriage returns, and line feeds - string = string.replace('"', "'").replace(".'", "'.").replace('*', '').replace('\r', '').replace('\n', '') - if string == '': - return '*Empty string*' + string = ( + string.replace('"', "'") + .replace(".'", "'.") + .replace("*", "") + .replace("\r", "") + .replace("\n", "") + ) + if string == "": + return "*Empty string*" return string + ################################ #### SUBPORCESS TERMINATION #### ################################ -#Register the termination code to be executed at exit +# Register the termination code to be executed at exit atexit.register(lambda: process.terminate() if process.poll() is None else None) + ###################### #### GRADIO STUFF #### ###################### def state_modifier(state): - if not params['activate']: + if not params["activate"]: return state - state['stream'] = False + state["stream"] = False return state + def input_modifier(string, state): - if not params['activate']: + if not params["activate"]: return string shared.processing_message = "*Is recording a voice message...*" return string + def custom_css(): path_to_css = Path(f"{this_dir}/style.css") - return open(path_to_css, 'r').read() + return open(path_to_css, "r").read() + def ui(): - with gr.Accordion("Coqui TTS (XTTSv2)"): + with gr.Accordion(params["branding"] + " TTS (XTTSv2)"): with gr.Row(): - activate = gr.Checkbox(value=params['activate'], label='Activate TTS') - autoplay = gr.Checkbox(value=params['autoplay'], label='Play TTS automatically') + activate = gr.Checkbox(value=params["activate"], label="Activate TTS") + autoplay = gr.Checkbox( + value=params["autoplay"], label="Play TTS automatically" + ) with gr.Row(): - show_text = gr.Checkbox(value=params['show_text'], label='Show message text under audio player') - remove_trailing_dots = gr.Checkbox(value=params['remove_trailing_dots'], label='Remove trailing "." from text segments before generation') - + show_text = gr.Checkbox( + value=params["show_text"], label="Show message text under audio player" + ) + remove_trailing_dots = gr.Checkbox( + value=params["remove_trailing_dots"], + label='Remove trailing "." from text segments before generation', + ) + with gr.Row(): - low_vram = gr.Checkbox(value=params['low_vram'], label='Low VRAM mode (Read NOTE)') - deepspeed_checkbox = gr.Checkbox(value=params['deepspeed_activate'], label='Activate DeepSpeed (Read NOTE)', visible=deepspeed_installed) - + low_vram = gr.Checkbox( + value=params["low_vram"], label="Low VRAM mode (Read NOTE)" + ) + deepspeed_checkbox = gr.Checkbox( + value=params["deepspeed_activate"], + label="Activate DeepSpeed (Read NOTE)", + visible=deepspeed_installed, + ) + with gr.Row(): tts_radio_buttons = gr.Radio( - choices=["API TTS", "API Local", "XTTSv2 Local"], - label="Select TTS Generation Method (Read NOTE)", - value=gr_modelchoice # Set the default value + choices=["API TTS", "API Local", "XTTSv2 Local"], + label="Select TTS Generation Method (Read NOTE)", + value=gr_modelchoice, # Set the default value ) - explanation_text = gr.HTML(f"

NOTE: Switching Model Type, Low VRAM & DeepSpeed takes 15 seconds. Each TTS generation method has a slightly different sound. DeepSpeed checkbox is only visible if DeepSpeed is present. Readme & Settings: http://{params['ip_address']}:{params['port_number']}") - + explanation_text = gr.HTML( + f"

NOTE: Switching Model Type, Low VRAM & DeepSpeed takes 15 seconds. Each TTS generation method has a slightly different sound. DeepSpeed checkbox is only visible if DeepSpeed is present. Readme & Settings: http://{params['ip_address']}:{params['port_number']}" + ) + with gr.Row(): with gr.Row(): - voice = gr.Dropdown(get_available_voices(), label="Voice wav", value=params["voice"]) - create_refresh_button(voice, lambda: None, lambda: {'choices': get_available_voices(), 'value': params["voice"]}, 'refresh-button') + voice = gr.Dropdown( + get_available_voices(), label="Voice wav", value=params["voice"] + ) + create_refresh_button( + voice, + lambda: None, + lambda: { + "choices": get_available_voices(), + "value": params["voice"], + }, + "refresh-button", + ) - language = gr.Dropdown(languages.keys(), label="Language", value=params["language"]) + language = gr.Dropdown( + languages.keys(), label="Language", value=params["language"] + ) with gr.Row(): - preview_text = gr.Text(show_label=False, placeholder="Preview text", elem_id="silero_preview_text") + preview_text = gr.Text( + show_label=False, + placeholder="Preview text", + elem_id="silero_preview_text", + ) preview_play = gr.Button("Preview") preview_audio = gr.HTML(visible=False) with gr.Row(): - convert = gr.Button('Permanently replace audios with the message texts') - convert_cancel = gr.Button('Cancel', visible=False) - convert_confirm = gr.Button('Confirm (cannot be undone)', variant="stop", visible=False) + convert = gr.Button("Permanently replace audios with the message texts") + convert_cancel = gr.Button("Cancel", visible=False) + convert_confirm = gr.Button( + "Confirm (cannot be undone)", variant="stop", visible=False + ) # Convert history with confirmation convert_arr = [convert_confirm, convert, convert_cancel] - convert.click(lambda: [gr.update(visible=True), gr.update(visible=False), gr.update(visible=True)], None, convert_arr) + convert.click( + lambda: [ + gr.update(visible=True), + gr.update(visible=False), + gr.update(visible=True), + ], + None, + convert_arr, + ) convert_confirm.click( - lambda: [gr.update(visible=False), gr.update(visible=True), gr.update(visible=False)], None, convert_arr).then( - remove_tts_from_history, gradio('history'), gradio('history')).then( - chat.save_history, gradio('history', 'unique_id', 'character_menu', 'mode'), None).then( - chat.redraw_html, gradio(ui_chat.reload_arr), gradio('display')) + lambda: [ + gr.update(visible=False), + gr.update(visible=True), + gr.update(visible=False), + ], + None, + convert_arr, + ).then(remove_tts_from_history, gradio("history"), gradio("history")).then( + chat.save_history, + gradio("history", "unique_id", "character_menu", "mode"), + None, + ).then( + chat.redraw_html, gradio(ui_chat.reload_arr), gradio("display") + ) - convert_cancel.click(lambda: [gr.update(visible=False), gr.update(visible=True), gr.update(visible=False)], None, convert_arr) + convert_cancel.click( + lambda: [ + gr.update(visible=False), + gr.update(visible=True), + gr.update(visible=False), + ], + None, + convert_arr, + ) # Toggle message text in history - show_text.change( - lambda x: params.update({"show_text": x}), show_text, None).then( - toggle_text_in_history, gradio('history'), gradio('history')).then( - chat.save_history, gradio('history', 'unique_id', 'character_menu', 'mode'), None).then( - chat.redraw_html, gradio(ui_chat.reload_arr), gradio('display')) + show_text.change(lambda x: params.update({"show_text": x}), show_text, None).then( + toggle_text_in_history, gradio("history"), gradio("history") + ).then( + chat.save_history, + gradio("history", "unique_id", "character_menu", "mode"), + None, + ).then( + chat.redraw_html, gradio(ui_chat.reload_arr), gradio("display") + ) # Event functions to update the parameters in the backend activate.change(lambda x: params.update({"activate": x}), activate, None) @@ -495,7 +662,9 @@ def ui(): low_vram.change(lambda x: send_lowvram_request(x), low_vram, None) tts_radio_buttons.change(send_reload_request, tts_radio_buttons, None) deepspeed_checkbox.change(send_deepspeed_request, deepspeed_checkbox, None) - remove_trailing_dots.change(lambda x: params.update({"remove_trailing_dots": x}), remove_trailing_dots, None) + remove_trailing_dots.change( + lambda x: params.update({"remove_trailing_dots": x}), remove_trailing_dots, None + ) voice.change(lambda x: params.update({"voice": x}), voice, None) language.change(lambda x: params.update({"language": x}), language, None) diff --git a/tts_server.py b/tts_server.py index 54e4072..538ac6a 100644 --- a/tts_server.py +++ b/tts_server.py @@ -11,11 +11,75 @@ from TTS.tts.models.xtts import Xtts ########################## #### Webserver Imports#### ########################## -from fastapi import FastAPI, Form, Request, HTTPException, BackgroundTasks, Response, Depends +from fastapi import ( + FastAPI, + Form, + Request, + HTTPException, + BackgroundTasks, + Response, + Depends, +) from fastapi.responses import JSONResponse, HTMLResponse, RedirectResponse from fastapi.templating import Jinja2Templates from contextlib import asynccontextmanager + +########################### +#### STARTUP VARIABLES #### +########################### +# STARTUP VARIABLE - Create "this_dir" variable as the current script directory +this_dir = Path(__file__).parent.resolve() +# STARTUP VARIABLE - Set "device" to cuda if exists, otherwise cpu +device = "cuda" if torch.cuda.is_available() else "cpu" +# STARTUP VARIABLE - Import languges file for Gradio to be able to display them in the interface +with open(this_dir / "languages.json", encoding="utf8") as f: + languages = json.load(f) + + +############################################################## +#### LOAD PARAMS FROM CONFIG.JSON - REQUIRED FOR BRANDING #### +############################################################## + + +# Load config file and get settings +def load_config(file_path): + with open(file_path, "r") as configfile_path: + configfile_data = json.load(configfile_path) + return configfile_data + + +# Define the path to the config.json file +configfile_path = this_dir / "config.json" + +# Load config.json and assign it to a different variable (config_data) +params = load_config(configfile_path) + +# Define the path to the JSON file +config_file_path = this_dir / "modeldownload.json" + +############################################# +#### LOAD PARAMS FROM MODELDOWNLOAD.JSON #### +############################################ +# This is used only in the instance that someone has changed their model path +# Define the path to the JSON file +modeldownload_config_file_path = this_dir / "modeldownload.json" + +# Check if the JSON file exists +if modeldownload_config_file_path.exists(): + with open(modeldownload_config_file_path, "r") as modeldownload_config_file: + modeldownload_settings = json.load(modeldownload_config_file) + + # Extract settings from the loaded JSON + modeldownload_base_path = Path(modeldownload_settings.get("base_path", "")) + modeldownload_model_path = Path(modeldownload_settings.get("model_path", "")) +else: + # Default settings if the JSON file doesn't exist or is empty + print( + f"[{params['branding']}Startup] \033[91mWarning\033[0m modeldownload.config is missing so please re-download it and save it in the coquii_tts main folder." + ) + + ######################## #### STARTUP CHECKS #### ######################## @@ -24,151 +88,160 @@ try: from TTS.utils.synthesizer import Synthesizer except ModuleNotFoundError: logger.error( - "[CoquiTTS Startup] \033[91mWarning\033[0m Could not find the TTS module. Make sure to install the requirements for the coqui_tts extension." - "[CoquiTTS Startup] \033[91mWarning\033[0m Linux / Mac:\npip install -r extensions/coqui_tts/requirements.txt\n" - "[CoquiTTS Startup] \033[91mWarning\033[0m Windows:\npip install -r extensions\\coqui_tts\\requirements.txt\n" - "[CoquiTTS Startup] \033[91mWarning\033[0m If you used the one-click installer, paste the command above in the terminal window launched after running the \"cmd_\" script. On Windows, that's \"cmd_windows.bat\"." + f"[{params['branding']}Startup] \033[91mWarning\033[0m Could not find the TTS module. Make sure to install the requirements for the alltalk_tts extension." + f"[{params['branding']}Startup] \033[91mWarning\033[0m Linux / Mac:\npip install -r extensions/alltalk_tts/requirements.txt\n" + f"[{params['branding']}Startup] \033[91mWarning\033[0m Windows:\npip install -r extensions\\alltalk_tts\\requirements.txt\n" + f"[{params['branding']}Startup] \033[91mWarning\033[0m If you used the one-click installer, paste the command above in the terminal window launched after running the cmd_ script. On Windows, that's cmd_windows.bat." ) raise -#DEEPSPEED Import - Check for DeepSpeed and import it if it exists +# DEEPSPEED Import - Check for DeepSpeed and import it if it exists try: import deepspeed + deepspeed_installed = True - print("[CoquiTTS Startup] DeepSpeed \033[93mDetected\033[0m") - print("[CoquiTTS Startup] Activate DeepSpeed in Coqui settings") + print(f"[{params['branding']}Startup] DeepSpeed \033[93mDetected\033[0m") + print( + f"[{params['branding']}Startup] Activate DeepSpeed in " + + params["branding"] + + " settings" + ) except ImportError: deepspeed_installed = False - print("[CoquiTTS Startup] DeepSpeed \033[93mNot Detected\033[0m. See https://github.com/microsoft/DeepSpeed") + print( + f"[{params['branding']}Startup] DeepSpeed \033[93mNot Detected\033[0m. See https://github.com/microsoft/DeepSpeed" + ) + @asynccontextmanager async def startup_shutdown(no_actual_value_it_demanded_something_be_here): await setup() - yield + yield # Shutdown logic -########################### -#### STARTUP VARIABLES #### -########################### -#STARTUP VARIABLE - Set "device" to cuda if exists, otherwise cpu -device = "cuda" if torch.cuda.is_available() else "cpu" -#STARTUP VARIABLE - Create "this_dir" variable as the current script directory -this_dir = Path(__file__).parent.resolve() -#STARTUP VARIABLE - Import languges file for Gradio to be able to display them in the interface -with open(this_dir / 'languages.json', encoding='utf8') as f: - languages = json.load(f) + # Create FastAPI app with lifespan app = FastAPI(lifespan=startup_shutdown) -###################################### -#### LOAD PARAMS FROM CONFIG.JSON #### -###################################### -def load_config(file_path): - with open(file_path, 'r') as config_file: - config = json.load(config_file) - return config - -config_file_path = (this_dir / 'config.json') -params = load_config(config_file_path) - -############################################# -#### LOAD PARAMS FROM MODELDOWNLOAD.JSON #### -############################################ -#This is used only in the instance that someone has changed their model path -# Define the path to the JSON file -modeldownload_config_file_path = this_dir / 'modeldownload.json' - -#Check if the JSON file exists -if modeldownload_config_file_path.exists(): - with open(modeldownload_config_file_path, 'r') as modeldownload_config_file: - modeldownload_settings = json.load(modeldownload_config_file) - - #Extract settings from the loaded JSON - modeldownload_base_path = Path(modeldownload_settings.get("base_path", "")) - modeldownload_model_path = Path(modeldownload_settings.get("model_path", "")) -else: - #Default settings if the JSON file doesn't exist or is empty - print("[CoquiTTS Startup] \033[91mWarning\033[0m modeldownload.config is missing so please re-download it and save it in the coquii_tts main folder.") ##################################### #### MODEL LOADING AND UNLOADING #### ##################################### -#MODEL LOADERS Picker For API TTS, API Local, XTTSv2 Local +# MODEL LOADERS Picker For API TTS, API Local, XTTSv2 Local async def setup(): global device - #Set a timer to calculate load times + # Set a timer to calculate load times generate_start_time = time.time() # Record the start time of loading the model - #Start loading the correct model as set by "tts_method_api_tts", "tts_method_api_local" or "tts_method_xtts_local" being True/False + # Start loading the correct model as set by "tts_method_api_tts", "tts_method_api_local" or "tts_method_xtts_local" being True/False if params["tts_method_api_tts"]: - print(f"[CoquiTTS Model] \033[94mAPI TTS Loading\033[0m {params['model_name']} \033[94minto\033[93m", device, "\033[0m") + print( + f"[{params['branding']}Model] \033[94mAPI TTS Loading\033[0m {params['model_name']} \033[94minto\033[93m", + device, + "\033[0m", + ) model = await api_load_model() elif params["tts_method_api_local"]: - print(f"[CoquiTTS Model] \033[94mAPI Local Loading\033[0m {modeldownload_model_path} \033[94minto\033[93m", device, "\033[0m") + print( + f"[{params['branding']}Model] \033[94mAPI Local Loading\033[0m {modeldownload_model_path} \033[94minto\033[93m", + device, + "\033[0m", + ) model = await api_manual_load_model() elif params["tts_method_xtts_local"]: - print(f"[CoquiTTS Model] \033[94mXTTSv2 Local Loading\033[0m {modeldownload_model_path} \033[94minto\033[93m", device, "\033[0m") + print( + f"[{params['branding']}Model] \033[94mXTTSv2 Local Loading\033[0m {modeldownload_model_path} \033[94minto\033[93m", + device, + "\033[0m", + ) model = await xtts_manual_load_model() - #Create an end timer for calculating load times + # Create an end timer for calculating load times generate_end_time = time.time() - #Calculate start time minus end time + # Calculate start time minus end time generate_elapsed_time = generate_end_time - generate_start_time - #Print out the result of the load time - print(f"[CoquiTTS Model] \033[94mModel Loaded in \033[93m{generate_elapsed_time:.2f} seconds.\033[0m") - #Set "model_loaded" to true + # Print out the result of the load time + print( + f"[{params['branding']}Model] \033[94mModel Loaded in \033[93m{generate_elapsed_time:.2f} seconds.\033[0m" + ) + # Set "model_loaded" to true params["model_loaded"] = True - #Set the output path for wav files + # Set the output path for wav files Path(f'{params["output_folder_wav"]}outputs/').mkdir(parents=True, exist_ok=True) -#MODEL LOADER For "API TTS" + +# MODEL LOADER For "API TTS" async def api_load_model(): global model model = TTS(params["model_name"]).to(device) return model -#MODEL LOADER For "API Local" + +# MODEL LOADER For "API Local" async def api_manual_load_model(): global model - #check to see if a custom path has been set in modeldownload.json and use that path to load the model if so + # check to see if a custom path has been set in modeldownload.json and use that path to load the model if so if str(modeldownload_base_path) == "models": - model = TTS(model_path=this_dir / 'models' / modeldownload_model_path,config_path=this_dir / 'models' / modeldownload_model_path / 'config.json').to(device) + model = TTS( + model_path=this_dir / "models" / modeldownload_model_path, + config_path=this_dir / "models" / modeldownload_model_path / "config.json", + ).to(device) else: - print("[CoquiTTS Model] \033[94mInfo\033[0m Loading your custom model set in \033[93mmodeldownload.json\033[0m:", modeldownload_base_path / modeldownload_model_path) - model = TTS(model_path=modeldownload_base_path / modeldownload_model_path,config_path=modeldownload_base_path / modeldownload_model_path / 'config.json').to(device) + print( + f"[{params['branding']}Model] \033[94mInfo\033[0m Loading your custom model set in \033[93mmodeldownload.json\033[0m:", + modeldownload_base_path / modeldownload_model_path, + ) + model = TTS( + model_path=modeldownload_base_path / modeldownload_model_path, + config_path=modeldownload_base_path + / modeldownload_model_path + / "config.json", + ).to(device) return model -#MODEL LOADER For "XTTSv2 Local" + +# MODEL LOADER For "XTTSv2 Local" async def xtts_manual_load_model(): global model config = XttsConfig() - #check to see if a custom path has been set in modeldownload.json and use that path to load the model if so + # check to see if a custom path has been set in modeldownload.json and use that path to load the model if so if str(modeldownload_base_path) == "models": - config_path = this_dir / 'models' / modeldownload_model_path / 'config.json' - checkpoint_dir = this_dir / 'models' / modeldownload_model_path + config_path = this_dir / "models" / modeldownload_model_path / "config.json" + checkpoint_dir = this_dir / "models" / modeldownload_model_path else: - print("[CoquiTTS Model] \033[94mInfo\033[0m Loading your custom model set in \033[93mmodeldownload.json\033[0m:", modeldownload_base_path / modeldownload_model_path) - config_path = modeldownload_base_path / modeldownload_model_path / 'config.json' + print( + f"[{params['branding']}Model] \033[94mInfo\033[0m Loading your custom model set in \033[93mmodeldownload.json\033[0m:", + modeldownload_base_path / modeldownload_model_path, + ) + config_path = modeldownload_base_path / modeldownload_model_path / "config.json" checkpoint_dir = modeldownload_base_path / modeldownload_model_path config.load_json(str(config_path)) model = Xtts.init_from_config(config) - model.load_checkpoint(config, checkpoint_dir=str(checkpoint_dir), use_deepspeed=params['deepspeed_activate']) + model.load_checkpoint( + config, + checkpoint_dir=str(checkpoint_dir), + use_deepspeed=params["deepspeed_activate"], + ) model.cuda() model.to(device) return model -#MODEL UNLOADER + +# MODEL UNLOADER async def unload_model(model): - print("[CoquiTTS Model] \033[94mUnloading model \033[0m") + print(f"[{params['branding']}Model] \033[94mUnloading model \033[0m") del model torch.cuda.empty_cache() params["model_loaded"] = False return None -#MODEL - Swap model based on Gradio selection API TTS, API Local, XTTSv2 Local + +# MODEL - Swap model based on Gradio selection API TTS, API Local, XTTSv2 Local async def handle_tts_method_change(tts_method): global model - #Update the params dictionary based on the selected radio button - print("[CoquiTTS Model] \033[94mChanging model \033[92m(Please wait 15 seconds)\033[0m") - #Set other parameters to False + # Update the params dictionary based on the selected radio button + print( + f"[{params['branding']}Model] \033[94mChanging model \033[92m(Please wait 15 seconds)\033[0m" + ) + # Set other parameters to False if tts_method == "API TTS": params["tts_method_api_local"] = False params["tts_method_xtts_local"] = False @@ -184,39 +257,41 @@ async def handle_tts_method_change(tts_method): params["tts_method_api_local"] = False params["tts_method_xtts_local"] = True - #Unload the current model + # Unload the current model model = await unload_model(model) - #Load the correct model based on the updated params + # Load the correct model based on the updated params await setup() -#MODEL WEBSERVER- API Swap Between Models + +# MODEL WEBSERVER- API Swap Between Models @app.route("/api/reload", methods=["POST"]) async def reload(request: Request): tts_method = request.query_params.get("tts_method") if tts_method not in ["API TTS", "API Local", "XTTSv2 Local"]: - return {"status": "error", "message": "Invalid TTS method specified"} + return {"status": "error", "message": "Invalid TTS method specified"} await handle_tts_method_change(tts_method) return Response( - content=json.dumps({"status": "model-success"}), - media_type="application/json" + content=json.dumps({"status": "model-success"}), media_type="application/json" ) + ################## #### LOW VRAM #### ################## -#LOW VRAM - MODEL MOVER VRAM(cuda)<>System RAM(cpu) for Low VRAM setting +# LOW VRAM - MODEL MOVER VRAM(cuda)<>System RAM(cpu) for Low VRAM setting async def switch_device(): global model, device if device == "cuda": device = "cpu" - model.to(device) + model.to(device) torch.cuda.empty_cache() - else: + else: device == "cpu" device = "cuda" model.to(device) + @app.post("/api/lowvramsetting") async def set_low_vram(request: Request, new_low_vram_value: bool): global device @@ -225,38 +300,64 @@ async def set_low_vram(request: Request, new_low_vram_value: bool): raise ValueError("Missing 'low_vram' parameter") if params["low_vram"] == new_low_vram_value: return Response( - content=json.dumps({"status": "success", "message": f"[CoquiTTS Model] LowVRAM is already {'enabled' if new_low_vram_value else 'disabled'}."})) + content=json.dumps( + { + "status": "success", + "message": f"[{params['branding']}Model] LowVRAM is already {'enabled' if new_low_vram_value else 'disabled'}.", + } + ) + ) params["low_vram"] = new_low_vram_value if params["low_vram"]: await unload_model(model) device = "cpu" - print("[CoquiTTS Model] \033[94mChanging model \033[92m(Please wait 15 seconds)\033[0m") - print("[CoquiTTS Model] \033[94mLowVRAM Enabled.\033[0m Model will move between \033[93mVRAM(cuda) <> System RAM(cpu)\033[0m") + print( + f"[{params['branding']}Model] \033[94mChanging model \033[92m(Please wait 15 seconds)\033[0m" + ) + print( + f"[{params['branding']}Model] \033[94mLowVRAM Enabled.\033[0m Model will move between \033[93mVRAM(cuda) <> System RAM(cpu)\033[0m" + ) await setup() else: await unload_model(model) device = "cuda" - print("[CoquiTTS Model] \033[94mChanging model \033[92m(Please wait 15 seconds)\033[0m") - print("[CoquiTTS Model] \033[94mLowVRAM Disabled.\033[0m Model will stay in \033[93mVRAM(cuda)\033[0m") + print( + f"[{params['branding']}Model] \033[94mChanging model \033[92m(Please wait 15 seconds)\033[0m" + ) + print( + f"[{params['branding']}Model] \033[94mLowVRAM Disabled.\033[0m Model will stay in \033[93mVRAM(cuda)\033[0m" + ) await setup() return Response(content=json.dumps({"status": "lowvram-success"})) except Exception as e: return Response(content=json.dumps({"status": "error", "message": str(e)})) + ################### #### DeepSpeed #### ################### -#DEEPSPEED - Reload the model when DeepSpeed checkbox is enabled/disabled +# DEEPSPEED - Reload the model when DeepSpeed checkbox is enabled/disabled async def handle_deepspeed_change(value): global model if value: # DeepSpeed enabled - print("[CoquiTTS Model] \033[93mDeepSpeed Activating\033[0m") - print("[CoquiTTS Model] \033[94mChanging model \033[92m(DeepSpeed can take 30 seconds to activate)\033[0m") - print("[CoquiTTS Model] \033[91mInformation\033[0m If you have not set CUDA_HOME path, DeepSpeed may fail to load/activate") - print("[CoquiTTS Model] \033[91mInformation\033[0m DeepSpeed needs to find nvcc from the CUDA Toolkit. Please check your CUDA_HOME path is") - print("[CoquiTTS Model] \033[91mInformation\033[0m pointing to the correct location and use 'set CUDA_HOME=putyoutpathhere' (Windows) or") - print("[CoquiTTS Model] \033[91mInformation\033[0m 'export CUDA_HOME=putyoutpathhere' (Linux) within your Python Environment") + print(f"[{params['branding']}Model] \033[93mDeepSpeed Activating\033[0m") + + print( + f"[{params['branding']}Model] \033[94mChanging model \033[92m(DeepSpeed can take 30 seconds to activate)\033[0m" + ) + print( + f"[{params['branding']}Model] \033[91mInformation\033[0m If you have not set CUDA_HOME path, DeepSpeed may fail to load/activate" + ) + print( + f"[{params['branding']}Model] \033[91mInformation\033[0m DeepSpeed needs to find nvcc from the CUDA Toolkit. Please check your CUDA_HOME path is" + ) + print( + f"[{params['branding']}Model] \033[91mInformation\033[0m pointing to the correct location and use 'set CUDA_HOME=putyoutpathhere' (Windows) or" + ) + print( + f"[{params['branding']}Model] \033[91mInformation\033[0m 'export CUDA_HOME=putyoutpathhere' (Linux) within your Python Environment" + ) model = await unload_model(model) params["tts_method_api_tts"] = False params["tts_method_api_local"] = False @@ -265,15 +366,18 @@ async def handle_deepspeed_change(value): await setup() else: # DeepSpeed disabled - print("[CoquiTTS Model] \033[93mDeepSpeed De-Activating\033[0m") - print("[CoquiTTS Model] \033[94mChanging model \033[92m(Please wait 15 seconds)\033[0m") - params["deepspeed_activate"] = False + print(f"[{params['branding']}Model] \033[93mDeepSpeed De-Activating\033[0m") + print( + f"[{params['branding']}Model] \033[94mChanging model \033[92m(Please wait 15 seconds)\033[0m" + ) + params["deepspeed_activate"] = False model = await unload_model(model) await setup() - return value # Return new checkbox value + return value # Return new checkbox value -#DEEPSPEED WEBSERVER- API Enable/Disable DeepSpeed + +# DEEPSPEED WEBSERVER- API Enable/Disable DeepSpeed @app.post("/api/deepspeed") async def deepspeed(request: Request, new_deepspeed_value: bool): try: @@ -281,55 +385,67 @@ async def deepspeed(request: Request, new_deepspeed_value: bool): raise ValueError("Missing 'deepspeed' parameter") if params["deepspeed_activate"] == new_deepspeed_value: return Response( - content=json.dumps({"status": "success", "message": f"DeepSpeed is already {'enabled' if new_deepspeed_value else 'disabled'}."})) + content=json.dumps( + { + "status": "success", + "message": f"DeepSpeed is already {'enabled' if new_deepspeed_value else 'disabled'}.", + } + ) + ) params["deepspeed_activate"] = new_deepspeed_value await handle_deepspeed_change(params["deepspeed_activate"]) return Response(content=json.dumps({"status": "deepspeed-success"})) except Exception as e: return Response(content=json.dumps({"status": "error", "message": str(e)})) + ######################## #### TTS GENERATION #### ######################## -#TTS VOICE GENERATION METHODS (called from voice_preview and output_modifer) +# TTS VOICE GENERATION METHODS (called from voice_preview and output_modifer) async def generate_audio(text, voice, language, output_file): global model if params["low_vram"] and device == "cpu": await switch_device() generate_start_time = time.time() # Record the start time of generating TTS - #XTTSv2 LOCAL Method - if params["tts_method_xtts_local"]: - print("[CoquiTTS TTSGen] {}".format(text)) - gpt_cond_latent, speaker_embedding = model.get_conditioning_latents(audio_path=[f"{this_dir}/voices/{voice}"]) + # XTTSv2 LOCAL Method + if params["tts_method_xtts_local"]: + print("[" + params["branding"] + "TTSGen] {}".format(text)) + gpt_cond_latent, speaker_embedding = model.get_conditioning_latents( + audio_path=[f"{this_dir}/voices/{voice}"] + ) out = model.inference( - text, - language, - gpt_cond_latent=gpt_cond_latent, - speaker_embedding=speaker_embedding, - temperature=0.7, - enable_text_splitting=True + text, + language, + gpt_cond_latent=gpt_cond_latent, + speaker_embedding=speaker_embedding, + temperature=0.7, + enable_text_splitting=True, ) torchaudio.save(output_file, torch.tensor(out["wav"]).unsqueeze(0), 24000) - #API TTS and API LOCAL Methods + # API TTS and API LOCAL Methods elif params["tts_method_api_tts"] or params["tts_method_api_local"]: - #Set the correct output path (different from the if statement) - print("[CoquiTTS TTSGen] Using API TTS/Local Method") + # Set the correct output path (different from the if statement) + print("[" + params["branding"] + "TTSGen] Using API TTS/Local Method") model.tts_to_file( text=text, file_path=output_file, speaker_wav=[f"{this_dir}/voices/{voice}"], language=language, ) - #Print Generation time and settings + # Print Generation time and settings generate_end_time = time.time() # Record the end time to generate TTS generate_elapsed_time = generate_end_time - generate_start_time - print(f"[CoquiTTS TTSGen] \033[93m{generate_elapsed_time:.2f} seconds. \033[94mLowVRAM: \033[33m{params['low_vram']} \033[94mDeepSpeed: \033[33m{params['deepspeed_activate']}\033[0m") - #Move model back to cpu system ram if needed. + print( + f"[{params['branding']}TTSGen] \033[93m{generate_elapsed_time:.2f} seconds. \033[94mLowVRAM: \033[33m{params['low_vram']} \033[94mDeepSpeed: \033[33m{params['deepspeed_activate']}\033[0m" + ) + # Move model back to cpu system ram if needed. if params["low_vram"] and device == "cuda": await switch_device() return -#TTS VOICE GENERATION METHODS - generate TTS API + +# TTS VOICE GENERATION METHODS - generate TTS API @app.route("/api/generate", methods=["POST"]) async def generate(request: Request): try: @@ -338,35 +454,48 @@ async def generate(request: Request): text = data["text"] voice = data["voice"] language = data["language"] - output_file = data["output_file"] + output_file = data["output_file"] # Generation logic - await generate_audio(text, voice, language, output_file) - return JSONResponse(content={"status": "generate-success", "data": {"audio_path": output_file}}) + await generate_audio(text, voice, language, output_file) + return JSONResponse( + content={"status": "generate-success", "data": {"audio_path": output_file}} + ) except Exception as e: return JSONResponse(content={"status": "error", "message": str(e)}) + ############################# #### JSON CONFIG UPDATER #### ############################# -#Create an instance of Jinja2Templates for rendering HTML templates +# Create an instance of Jinja2Templates for rendering HTML templates templates = Jinja2Templates(directory=this_dir / "templates") -#Create a dependency to get the current JSON data + +# Create a dependency to get the current JSON data def get_json_data(): - with open(this_dir / 'config.json', "r") as json_file: + with open(this_dir / "config.json", "r") as json_file: data = json.load(json_file) return data -#Define an endpoint function + +# Define an endpoint function @app.get("/settings") async def get_settings(request: Request): # Render the template with the current JSON data - return templates.TemplateResponse("generate_form.html", {"request": request, "data": get_json_data(), "modeldownload_model_path": modeldownload_model_path}) + return templates.TemplateResponse( + "generate_form.html", + { + "request": request, + "data": get_json_data(), + "modeldownload_model_path": modeldownload_model_path, + }, + ) + @app.post("/update-settings") async def update_settings( - request: Request, + request: Request, activate: bool = Form(...), autoplay: bool = Form(...), deepspeed_activate: bool = Form(...), @@ -403,13 +532,14 @@ async def update_settings( data["tts_method_xtts_local"] = tts_method == "xtts_local" data["voice"] = voice - #Save the updated settings back to the JSON file - with open(this_dir / 'config.json', "w") as json_file: + # Save the updated settings back to the JSON file + with open(this_dir / "config.json", "w") as json_file: json.dump(data, json_file) - #Redirect to the settings page to display the updated settings + # Redirect to the settings page to display the updated settings return RedirectResponse(url="/settings", status_code=303) + ############################################################# #### DOCUMENTATION - README ETC - PRESENTED AS A WEBPAGE #### ############################################################# @@ -421,7 +551,7 @@ simple_webpage = """ - CoquiTTS for Text-generation-WebUI + AllTalk TTS for Text-generation-WebUI