diff --git a/modeldownload.py b/modeldownload.py index b4edae5..e69de29 100644 --- a/modeldownload.py +++ b/modeldownload.py @@ -1,145 +0,0 @@ -import os -from pathlib import Path -import requests -from tqdm import tqdm -import importlib.metadata as metadata -import json -from packaging import version - -################################################################# -#### LOAD PARAMS FROM confignew.json - REQUIRED FOR BRANDING #### -################################################################# -# STARTUP VARIABLE - Create "this_dir" variable as the current script directory -this_dir = Path(__file__).parent.resolve() - - -# 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 / "confignew.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 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 - base_path = Path(settings.get("base_path", "")) - model_path = Path(settings.get("model_path", "")) - files_to_download = settings.get("files_to_download", {}) -else: - # Default settings if the JSON file doesn't exist or is empty - 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_nvidia.txt -with open(this_dir / "system" / "requirements" / "requirements_standalone.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] - break - -if tts_version_required is None: - raise ValueError( - f"[{params['branding']}Startup] \033[91mWarning\033[0m Could not find TTS version specifier in requirements file" - ) - - -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)) - block_size = 1024 # 1 Kibibyte - - 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"[{params['branding']}Startup] TTS version installed: \033[93m" - + tts_version - + "\033[0m" - ) - - if version.parse(tts_version) < version.parse(tts_version_required): - 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(f"[{params['branding']}Startup] TTS version is up to date.") - except metadata.PackageNotFoundError: - 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( - 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("[" + 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 - else: - destination = Path(base_path) / model_path / filename - - if not destination.exists(): - print(f"[{params['branding']}Startup] Downloading " + filename + "...") - download_file(url, destination) - -check_tts_version() diff --git a/script.py b/script.py index 5b4066d..e69de29 100644 --- a/script.py +++ b/script.py @@ -1,1073 +0,0 @@ -import html -import json -import random -import subprocess -import time -import os -import requests -import threading -import signal -import sys -import atexit -from pathlib import Path -from datetime import datetime, timedelta -import re -import numpy as np -import soundfile as sf -import uuid - -###################################### -#### ALLTALK ALLOWED STARTUP TIME #### -###################################### -startup_wait_time = 120 - -# You can change the above setting to a larger number to allow AllTAlk more time to start up. The default setting is 120 seconds (2 minutes). -# On some older systems you may need to allow AllTalk more time. So you could set it to 240 (4 minutes) which will give AllTalk more to load. - -################################################################# -#### LOAD PARAMS FROM confignew.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 / "confignew.json" -# Load the params dictionary from the confignew.json file -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( - f"[{params['branding']}]\033[91mWarning\033[0m Could not find the TTS module. Make sure to install the requirements for the {params['branding']} extension." - f"[{params['branding']}]\033[91mWarning\033[0m Linux / Mac:\npip install -r /alltalk_tts/requirements.txt\n" - f"[{params['branding']}]\033[91mWarning\033[0m Windows:\npip install -r \\alltalk_tts\\requirements.txt\n" - f"[{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 (required for displaying Deepspeed checkbox in gradio) -try: - import deepspeed - - deepspeed_installed = True -except ImportError: - deepspeed_installed = False - -# Import gradio if being used within text generation webUI -try: - import gradio as gr - - from modules import chat, shared, ui_chat - from modules.logging_colors import logger - from modules.ui import create_refresh_button - from modules.utils import gradio - - # This is set to check if the script is being run within text generation webui or as a standalone script. False is running as part of text gen web ui or a gradio interface - running_in_standalone = False - output_folder_wav = params["output_folder_wav"] -except ModuleNotFoundError: - output_folder_wav = params["output_folder_wav_standalone"] - print(f"[{params['branding']}Startup] Running script.py in standalone mode") - # This is set to check if the script is being run within text generation webui or as a standalone script. true means standalone - running_in_standalone = True - -########################### -#### STARTUP VARIABLES #### -########################### -# STARTUP VARIABLE - Import languges file for Gradio to be able to display them in the interface -with open(this_dir / "system" / "config" / "languages.json", encoding="utf8") as f: - languages = json.load(f) -# Create a global lock -process_lock = threading.Lock() -# Base setting for a possible FineTuned model existing and the loader being available -tts_method_xtts_ft = False - - -# Gather the voice files -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(f"[{params['branding']}Startup] \033[94mCoqui Public Model License\033[0m") -print(f"[{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( - f"[{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): - file_creation_time = datetime.fromtimestamp(os.path.getctime(file_path)) - age = current_time - file_creation_time - 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 = os.path.normpath(output_folder_wav) - -# Check and perform file deletion -if delete_output_wavs_setting.lower() == "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( - f"[{params['branding']}Startup] Invalid setting for deleting old wav files. Please use 'Disabled' or 'X Days' format." - ) - - -######################## -#### STARTUP CHECKS #### -######################## -# 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(f"[{params['branding']}Startup] All required files are present.") - - -# STARTUP Call Check routine -check_required_files() - -################################################## -#### Check to see if a finetuned model exists #### -################################################## -# Set the path to the directory -trained_model_directory = this_dir / "models" / "trainedmodel" -# Check if the directory "trainedmodel" exists -finetuned_model = trained_model_directory.exists() -# If the directory exists, check for the existence of the required files -# If true, this will add a extra option in the Gradio interface for loading Xttsv2 FT -if finetuned_model: - required_files = ["model.pth", "config.json", "vocab.json"] - finetuned_model = all( - (trained_model_directory / file).exists() for file in required_files - ) -if finetuned_model: - print(f"[{params['branding']}Startup] Finetuned model \033[93mDetected\033[0m") - -#################################################### -#### SET GRADIO BUTTONS BASED ON confignew.json #### -#################################################### - -if params["tts_method_api_tts"] == True: - gr_modelchoice = "API TTS" -elif params["tts_method_api_local"] == True: - gr_modelchoice = "API Local" -elif params["tts_method_xtts_local"] == True: - gr_modelchoice = "XTTSv2 Local" - -# Set the default for Narrated text without asterisk or quotes to be Narrator -non_quoted_text_is = True - - -###################### -#### GRADIO STUFF #### -###################### -def remove_tts_from_history(history): - 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"]): - visible_reply = entry[1] - if visible_reply.startswith("')[0]}\n\n{reply}", - ] - else: - 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>"), - ] - - return history - - -###################################### -#### SUBPROCESS/WEBSERVER STARTUP #### -###################################### -base_url = f"http://{params['ip_address']}:{params['port_number']}" -script_path = this_dir / "tts_server.py" - - -def signal_handler(sig, frame): - 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) -# Check if we're running in docker -if os.path.isfile("/.dockerenv"): - print( - f"[{params['branding']}Startup] \033[94mRunning in Docker. Please wait.\033[0m" - ) -else: - # Start the subprocess - process = subprocess.Popen(["python", script_path]) - # Check if the subprocess has started successfully - if process.poll() is None: - print(f"[{params['branding']}Startup] TTS Subprocess starting") - print(f"[{params['branding']}Startup]") - print( - f"[{params['branding']}Startup] \033[94m {params['branding']}Settings & Documentation:\033[00m", - f"\033[92mhttp://{params['ip_address']}:{params['port_number']}\033[00m", - ) - print(f"[{params['branding']}Startup]") - else: - 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 = startup_wait_time # Gather timeout setting from startup_wait_time - - # Introduce a delay before starting the check loop - time.sleep(26) # Wait 26 secs before checking if the tts_server.py has started up. - start_time = time.time() - while time.time() - start_time < timeout: - try: - response = requests.get(f"{base_url}/ready") - if response.status_code == 200: - break - except requests.RequestException as e: - # Print the exception for debugging purposes - print( - f"[{params['branding']}Startup] \033[91mWarning\033[0m TTS Subprocess has NOT started up yet, Will keep trying for {timeout} seconds maximum. Please wait." - ) - time.sleep(5) - else: - print( - f"\n[{params['branding']}Startup] Startup timed out. Full help available here \033[92mhttps://github.com/erew123/alltalk_tts#-help-with-problems\033[0m" - ) - print( - f"[{params['branding']}Startup] On older system you may wish to open and edit \033[94mscript.py\033[0m with a text editor and changing the" - ) - print( - f"[{params['branding']}Startup] \033[94mstartup_wait_time = 120\033[0m setting to something like \033[94mstartup_wait_time = 240\033[0m as this will allow" - ) - print( - f"[{params['branding']}Startup] AllTalk more time to try load the model into your VRAM. Otherise please visit the Github for" - ) - print( - f"[{params['branding']}Startup] a list of other possible troubleshooting options." - ) - # 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 -def send_reload_request(tts_method): - global tts_method_xtts_ft - try: - params["tts_model_loaded"] = False - url = f"{base_url}/api/reload" - payload = {"tts_method": tts_method} - response = requests.post(url, params=payload) - response.raise_for_status() # Raises an HTTPError for bad responses - json_response = response.json() - # Check if the reload operation was successful - if json_response.get("status") == "model-success": - # Update tts_tts_model_loaded to True if the reload was successful - params["tts_model_loaded"] = True - # Update local script parameters based on the tts_method - if tts_method == "API TTS": - params["tts_method_api_local"] = False - params["tts_method_xtts_local"] = False - params["tts_method_api_tts"] = True - params["deepspeed_activate"] = False - audio_path = this_dir / "system" / "at_sounds" / "apitts.wav" - tts_method_xtts_ft = False - elif tts_method == "API Local": - params["tts_method_api_tts"] = False - params["tts_method_xtts_local"] = False - params["tts_method_api_local"] = True - params["deepspeed_activate"] = False - audio_path = this_dir / "system" / "at_sounds" / "apilocal.wav" - tts_method_xtts_ft = False - elif tts_method == "XTTSv2 Local": - params["tts_method_api_tts"] = False - params["tts_method_api_local"] = False - params["tts_method_xtts_local"] = True - audio_path = this_dir / "system" / "at_sounds" / "xttslocal.wav" - tts_method_xtts_ft = False - elif tts_method == "XTTSv2 FT": - params["tts_method_api_tts"] = False - params["tts_method_api_local"] = False - params["tts_method_xtts_local"] = False - audio_path = this_dir / "system" / "at_sounds" / "xttsfinetuned.wav" - tts_method_xtts_ft = True - return f'' - except requests.exceptions.RequestException as e: - # Handle the HTTP request error - 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 -def send_lowvram_request(low_vram): - try: - params["tts_model_loaded"] = False - if low_vram: - audio_path = this_dir / "system" / "at_sounds" / "lowvramenabled.wav" - else: - audio_path = this_dir / "system" / "at_sounds" / "lowvramdisabled.wav" - url = f"{base_url}/api/lowvramsetting?new_low_vram_value={low_vram}" - headers = {"Content-Type": "application/json"} - response = requests.post(url, headers=headers) - response.raise_for_status() # Raises an HTTPError for bad responses - json_response = response.json() - # Check if the low VRAM request was successful - if json_response.get("status") == "lowvram-success": - # Update any relevant variables or perform other actions on success - params["tts_model_loaded"] = True - return f'' - except requests.exceptions.RequestException as e: - # Handle the HTTP request error - 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 -def send_deepspeed_request(deepspeed_param): - try: - params["tts_model_loaded"] = False - if deepspeed_param: - audio_path = this_dir / "system" / "at_sounds" / "deepspeedenabled.wav" - else: - audio_path = this_dir / "system" / "at_sounds" / "deepspeeddisabled.wav" - - url = f"{base_url}/api/deepspeed?new_deepspeed_value={deepspeed_param}" - headers = {"Content-Type": "application/json"} - response = requests.post(url, headers=headers) - response.raise_for_status() # Raises an HTTPError for bad responses - json_response = response.json() - # Check if the deepspeed request was successful - if json_response.get("status") == "deepspeed-success": - # Update any relevant variables or perform other actions on success - params["tts_model_loaded"] = True - return f'' - except requests.exceptions.RequestException as e: - # Handle the HTTP request error - 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 STRING CLEANING & PROCESSING PRE SENDING TO TTS #### -############################################################# -def new_split_into_sentences(self, text): - sentences = self.seg.segment(text) - if params["remove_trailing_dots"]: - sentences_without_dots = [] - for sentence in sentences: - if sentence.endswith(".") and not sentence.endswith("..."): - sentence = sentence[:-1] - - sentences_without_dots.append(sentence) - - return sentences_without_dots - else: - return sentences - - -Synthesizer.split_into_sentences = new_split_into_sentences - - -# Check model is loaded and string isnt empty, before sending a TTS request. -def before_audio_generation(string, params): - # Check Model is loaded into cuda or cpu and error if not - if not params["tts_model_loaded"]: - 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() - if string == "": - return "*Empty string*" - return string - - -################## -#### Narrator #### -################## -def combine(audio_files, output_folder, state): - audio = np.array([]) - - for audio_file in audio_files: - audio_data, sample_rate = sf.read(audio_file) - # Ensure all audio files have the same sample rate - if audio.size == 0: - audio = audio_data - else: - audio = np.concatenate((audio, audio_data)) - - # Save the combined audio to a file with a specified sample rate - if "character_menu" in state: - output_file_path = os.path.join( - output_folder, f'{state["character_menu"]}_{int(time.time())}_combined.wav' - ) - else: - output_file_path = os.path.join( - output_folder, f"TTSOUT_{int(time.time())}_combined.wav" - ) - sf.write(output_file_path, audio, samplerate=sample_rate) - # Clean up unnecessary files - for audio_file in audio_files: - os.remove(audio_file) - - return output_file_path - - -################################ -#### TTS PREVIEW GENERATION #### -################################ -# PREVIEW VOICE - Generate Random Sentence if Voice Preview box is empty -def random_sentence(): - with open(this_dir / "system" / "config" / "harvard_sentences.txt") as f: - return random.choice(list(f)) - - -# PREVIEW VOICE- Generate TTS Function -def voice_preview(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) - if cleaned_string is None: - 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"]) - temperature = params["local_temperature"] - repetition_penalty = params["local_repetition_penalty"] - # 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 - with process_lock: - generate_response = send_generate_request( - string, - params["voice"], - language_code, - temperature, - repetition_penalty, - output_file_str, - ) - # Check if lock is already acquired - if process_lock.locked(): - 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 "" - return f'' - else: - # Handle the case where audio generation was not successful - return f"[{params['branding']}Server] Audio generation failed. Status: {generate_response.get('status')}" - - -####################### -#### TEXT CLEANING #### -####################### - -def process_text(text): - # Normalize HTML encoded quotes - text = html.unescape(text) - - # Replace ellipsis with a single dot - text = re.sub(r"\.{3,}", ".", text) - - # Pattern to identify combined narrator and character speech - combined_pattern = r'(\*[^*"]+\*|"[^"*]+")' - - # List to hold parts of speech along with their type - ordered_parts = [] - - # Track the start of the next segment - start = 0 - - # Find all matches - for match in re.finditer(combined_pattern, text): - # Add the text before the match, if any, as ambiguous - if start < match.start(): - ambiguous_text = text[start : match.start()].strip() - if ambiguous_text: - ordered_parts.append(("ambiguous", ambiguous_text)) - - # Add the matched part as either narrator or character - matched_text = match.group(0) - if matched_text.startswith("*") and matched_text.endswith("*"): - ordered_parts.append(("narrator", matched_text.strip("*").strip())) - elif matched_text.startswith('"') and matched_text.endswith('"'): - ordered_parts.append(("character", matched_text.strip('"').strip())) - else: - # In case of mixed or improperly formatted parts - if "*" in matched_text: - ordered_parts.append(("narrator", matched_text.strip("*").strip('"'))) - else: - ordered_parts.append(("character", matched_text.strip('"').strip("*"))) - - # Update the start of the next segment - start = match.end() - - # Add any remaining text after the last match as ambiguous - if start < len(text): - ambiguous_text = text[start:].strip() - if ambiguous_text: - ordered_parts.append(("ambiguous", ambiguous_text)) - - return ordered_parts - - -######################## -#### IMAGE CLEANING #### -######################## - -img_pattern = r']*src\s*=\s*["\'][^"\'>]+["\'][^>]*>' - -def extract_and_remove_images(text): - """ - Extracts all image data from the text and removes it for clean TTS processing. - Returns the cleaned text and the extracted image data. - """ - img_matches = re.findall(img_pattern, text) - img_info = "\n".join(img_matches) # Store extracted image data - cleaned_text = re.sub(img_pattern, '', text) # Remove images from text - return cleaned_text, img_info - -def reinsert_images(text, img_info): - """ - Reinserts the previously extracted image data back into the text. - """ - if img_info: # Check if there are images to reinsert - text += f"\n\n{img_info}" - return text - -################################# -#### TTS STANDARD GENERATION #### -################################# -# STANDARD VOICE - Generate TTS Function -def output_modifier(string, state): - if not params["activate"]: - return string - img_info = "" - cleaned_text, img_info = extract_and_remove_images(string) - # print("Cleaned STRING IS:", cleaned_text) - cleaned_string = before_audio_generation(cleaned_text, params) - if cleaned_string is None: - return - language_code = languages.get(params["language"]) - temperature = params["local_temperature"] - repetition_penalty = params["local_repetition_penalty"] - # Create a list to store generated audio paths - audio_files = [] - if process_lock.acquire(blocking=False): - try: - if params["narrator_enabled"]: - processed_parts = process_text(cleaned_string) - - audio_files_all_paragraphs = [] - for part_type, part in processed_parts: - # Skip parts that are too short - if len(part.strip()) <= 3: - continue - - # Determine the voice to use based on the part type - if part_type == "narrator": - voice_to_use = params["narrator_voice"] - print( - f"[{params['branding']}TTSGen] \033[92mNarrator\033[0m" - ) # Green - elif part_type == "character": - voice_to_use = params["voice"] - print( - f"[{params['branding']}TTSGen] \033[36mCharacter\033[0m" - ) # Yellow - else: - # Handle ambiguous parts based on user preference - voice_to_use = ( - params["voice"] - if non_quoted_text_is - else params["narrator_voice"] - ) - voice_description = ( - "\033[36mCharacter (Text-not-inside)\033[0m" - if non_quoted_text_is - else "\033[92mNarrator (Text-not-inside)\033[0m" - ) - print(f"[{params['branding']}TTSGen] {voice_description}") - - # Replace multiple exclamation marks, question marks, or other punctuation with a single instance - cleaned_part = re.sub(r"([!?.])\1+", r"\1", part) - # Further clean to remove any other unwanted characters - cleaned_part = re.sub(r'[^a-zA-Z0-9\s\.,;:!?\-\'"\u0100-\u017F\u0400-\u04FFÀ-ÿ\u0150\u0151\u0170\u0171\u4E00-\u9FFF\u3400-\u4DBF\u20000-\u2A6DF\u3000-\u303F\$]', '', cleaned_part) - # Remove all newline characters (single or multiple) - cleaned_part = re.sub(r"\n+", " ", cleaned_part) - - # Generate TTS and output to a file - output_filename = get_output_filename(state) - generate_response = send_generate_request( - cleaned_part, - voice_to_use, - language_code, - temperature, - repetition_penalty, - output_filename, - ) - audio_path = generate_response.get("data", {}).get("audio_path") - audio_files_all_paragraphs.append(audio_path) - - # Combine audio files across paragraphs - final_output_file = combine( - audio_files_all_paragraphs, params["output_folder_wav"], state - ) - else: - # Decode HTML entities first - cleaned_part = html.unescape(cleaned_string) - # Replace multiple instances of certain punctuation marks with a single instance - cleaned_part = re.sub(r"([!?.])\1+", r"\1", cleaned_part) - # Further clean to remove any other unwanted characters - cleaned_part = re.sub(r'[^a-zA-Z0-9\s\.,;:!?\-\'"\u0100-\u017F\u0400-\u04FFÀ-ÿ\u0150\u0151\u0170\u0171\u4E00-\u9FFF\u3400-\u4DBF\u20000-\u2A6DF\u3000-\u303F\$]', '', cleaned_part) - # Remove all newline characters (single or multiple) - cleaned_part = re.sub(r"\n+", " ", cleaned_part) - # Process the part and give it a non-character name if being used vai API or standalone. - if "character_menu" in state: - output_file = Path( - f'{params["output_folder_wav"]}/{state["character_menu"]}_{int(time.time())}.wav' - ) - else: - output_file = Path( - f'{params["output_folder_wav"]}/TTSOUT_{int(time.time())}.wav' - ) - output_file_str = output_file.as_posix() - output_file = get_output_filename(state) - generate_response = send_generate_request( - cleaned_part, - params["voice"], - language_code, - temperature, - repetition_penalty, - output_file_str, - ) - audio_path = generate_response.get("data", {}).get("audio_path") - final_output_file = audio_path - finally: - # Always release the lock, whether an exception occurs or not - process_lock.release() - else: - # The lock is already acquired - 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 "" - string = (f'') - if params["show_text"]: - string += reinsert_images(cleaned_string, img_info) - shared.processing_message = "*Is typing...*" - return string - else: - print(f"[{params['branding']}Server] \033[91mWarning\033[0m No audio path in the response.") - else: - print(f"[{params['branding']}Server] \033[91mWarning\033[0m Audio generation failed. Status:", generate_response.get("message"),) - -def get_output_filename(state): - if "character_menu" in state: - return Path( - f'{params["output_folder_wav"]}/{state["character_menu"]}_{str(uuid.uuid4())[:8]}.wav' - ).as_posix() - else: - return Path( - f'{params["output_folder_wav"]}/TTSOUT_{str(uuid.uuid4())[:8]}.wav' - ).as_posix() - - -############################################### -#### SEND GENERATION REQUEST TO TTS ENGINE #### -############################################### -def send_generate_request( - text, voice, language, temperature, repetition_penalty, output_file -): - url = f"{base_url}/api/generate" - payload = { - "text": text, - "voice": voice, - "language": language, - "temperature": temperature, - "repetition_penalty": repetition_penalty, - "output_file": output_file, - } - headers = {"Content-Type": "application/json"} - response = requests.post(url, json=payload, headers=headers) - return response.json() - - -################################ -#### SUBPORCESS TERMINATION #### -################################ -# 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"]: - return state - - state["stream"] = False - return state - - -def update_narrator_enabled(value): - if value == "Enabled": - params["narrator_enabled"] = True - elif value == "Disabled": - params["narrator_enabled"] = False - - -def update_non_quoted_text_is(value): - global non_quoted_text_is - if value == "Narrator": - non_quoted_text_is = False - elif value == "Char": - non_quoted_text_is = True - - -def input_modifier(string, state): - if not params["activate"]: - return string - - shared.processing_message = "*Is recording a voice message...*" - return string - - -def ui(): - with gr.Accordion(params["branding"] + " TTS (XTTSv2)"): - # Activate alltalk_tts, Enable autoplay, Show text - with gr.Row(): - activate = gr.Checkbox(value=params["activate"], label="Enable TTS") - autoplay = gr.Checkbox(value=params["autoplay"], label="Autoplay TTS") - show_text = gr.Checkbox(value=params["show_text"], label="Show Text") - - # Low vram enable, Deepspeed enable, Remove trailing dots - with gr.Row(): - low_vram = gr.Checkbox( - value=params["low_vram"], label="Enable Low VRAM Mode" - ) - low_vram_play = gr.HTML(visible=False) - deepspeed_checkbox = gr.Checkbox( - value=params["deepspeed_activate"], - label="Enable DeepSpeed", - visible=deepspeed_installed, - ) - deepspeed_checkbox_play = gr.HTML(visible=False) - remove_trailing_dots = gr.Checkbox( - value=params["remove_trailing_dots"], label='Remove trailing "."' - ) - - # TTS method, Character voice selection - with gr.Row(): - model_loader_choices = ["API TTS", "API Local", "XTTSv2 Local"] - if finetuned_model: - model_loader_choices.append("XTTSv2 FT") - tts_radio_buttons = gr.Radio( - choices=model_loader_choices, - label="TTS Method (Each method sounds slightly different)", - value=gr_modelchoice, # Set the default value - ) - tts_radio_buttons_play = gr.HTML(visible=False) - with gr.Row(): - available_voices = get_available_voices() - default_voice = params[ - "voice" - ] # Check if the default voice is in the list of available voices - - if default_voice not in available_voices: - default_voice = available_voices[ - 0 - ] # Choose the first available voice as the default - # Add allow_custom_value=True to the Dropdown - voice = gr.Dropdown( - available_voices, - label="Character Voice", - value=default_voice, - allow_custom_value=True, - ) - create_refresh_button( - voice, - lambda: None, - lambda: { - "choices": get_available_voices(), - "value": params["voice"], - }, - "refresh-button", - ) - - # Language, Narrator voice - with gr.Row(): - language = gr.Dropdown( - languages.keys(), label="Language", value=params["language"] - ) - with gr.Row(): - narrator_voice_gr = gr.Dropdown( - get_available_voices(), - label="Narrator Voice", - allow_custom_value=True, - value=params["narrator_voice"], - ) - create_refresh_button( - narrator_voice_gr, - lambda: None, - lambda: { - "choices": get_available_voices(), - "value": params["narrator_voice"], - }, - "refresh-button", - ) - - # Temperature, Repetition Penalty - with gr.Row(): - local_temperature_gr = gr.Slider( - minimum=0.05, - maximum=1, - step=0.05, - label="Temperature", - value=params["local_temperature"], - ) - local_repetition_penalty_gr = gr.Slider( - minimum=0.5, - maximum=20, - step=0.5, - label="Repetition Penalty", - value=params["local_repetition_penalty"], - ) - - # Narrator enable, Non quoted text, Explanation text - with gr.Row(): - with gr.Row(): - narrator_enabled_gr = gr.Radio( - choices={"Enabled": "true", "Disabled": "false"}, - label="Narrator", - value="Enabled" if params.get("narrator_enabled") else "Disabled", - ) - non_quoted_text_is_gr = gr.Radio( - choices={"Character": "true", "Narrator": "false"}, - label='Unmarked text NOT inside of * or " is', - value="Character" if non_quoted_text_is else "Narrator", - ) - explanation_text = gr.HTML( - f"

⚙️ Settings and Documentation Page⚙️
- Low VRAM Mode and Deepspeed take 15 seconds to be enabled or disabled.
- The DeepSpeed checkbox is only visible if DeepSpeed is present.

" - ) - - # Preview speech - with gr.Row(): - 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 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_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") - ) - - 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") - ) - - # Event functions to update the parameters in the backend - activate.change(lambda x: params.update({"activate": x}), activate, None) - autoplay.change(lambda x: params.update({"autoplay": x}), autoplay, None) - low_vram.change(lambda x: params.update({"low_vram": x}), low_vram, None) - low_vram.change(lambda x: send_lowvram_request(x), low_vram, low_vram_play, None) - tts_radio_buttons.change( - send_reload_request, tts_radio_buttons, tts_radio_buttons_play, None - ) - deepspeed_checkbox.change( - send_deepspeed_request, deepspeed_checkbox, deepspeed_checkbox_play, 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) - - # TSS Settings - local_temperature_gr.change( - lambda x: params.update({"local_temperature": x}), local_temperature_gr, None - ) - local_repetition_penalty_gr.change( - lambda x: params.update({"local_repetition_penalty": x}), - local_repetition_penalty_gr, - None, - ) - - # Narrator selection actions - narrator_enabled_gr.change(update_narrator_enabled, narrator_enabled_gr, None) - non_quoted_text_is_gr.change(update_non_quoted_text_is, non_quoted_text_is_gr, None) - narrator_voice_gr.change( - lambda x: params.update({"narrator_voice": x}), narrator_voice_gr, None - ) - - # Play preview - preview_text.submit(voice_preview, preview_text, preview_audio) - preview_play.click(voice_preview, preview_text, preview_audio) - - -########################################## -#### STANDALONE MODE LOOP TERMINATION #### -########################################## -## Loop to keep the script from exiting out if its being run as a standalone script and not part of text-generation-webui -if running_in_standalone: - while True: - try: - time.sleep(1) # Add a small delay to avoid high CPU usage - except KeyboardInterrupt: - break # Allow graceful exit on Ctrl+C diff --git a/tts_server.py b/tts_server.py index 1f7cba3..e69de29 100644 --- a/tts_server.py +++ b/tts_server.py @@ -1,1247 +0,0 @@ -import json -import time -import os -from pathlib import Path -import torch -import torchaudio -from TTS.tts.configs.xtts_config import XttsConfig -from TTS.tts.models.xtts import Xtts -import io -import wave - -########################## -#### Webserver Imports#### -########################## -from fastapi import ( - FastAPI, - Form, - Request, - Response, - Depends, - HTTPException, - Query, -) -from fastapi.responses import JSONResponse, HTMLResponse, RedirectResponse, FileResponse, StreamingResponse -from fastapi.middleware.cors import CORSMiddleware -from fastapi.templating import Jinja2Templates -from fastapi.staticfiles import StaticFiles -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 / "system" / "config" / "languages.json", encoding="utf8") as f: - languages = json.load(f) -# Base setting for a possible FineTuned model existing and the loader being available -tts_method_xtts_ft = False - -################################################################# -#### LOAD PARAMS FROM confignew.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 confignew.json file -configfile_path = this_dir / "confignew.json" - -# Load confignew.json and assign it to a different variable (config_data) -params = load_config(configfile_path) -# check someone hasnt enabled lowvram on a system thats not cuda enabled -params["low_vram"] = "false" if not torch.cuda.is_available() else params["low_vram"] - -# Load values for temperature and repetition_penalty -temperature = params["local_temperature"] -repetition_penalty = params["local_repetition_penalty"] - -# 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 alltalk_tts main folder." - ) - -################################################## -#### Check to see if a finetuned model exists #### -################################################## -# Set the path to the directory -trained_model_directory = this_dir / "models" / "trainedmodel" -# Check if the directory "trainedmodel" exists -finetuned_model = trained_model_directory.exists() -# If the directory exists, check for the existence of the required files -if finetuned_model: - required_files = ["model.pth", "config.json", "vocab.json"] - finetuned_model = all((trained_model_directory / file).exists() for file in required_files) - -######################## -#### STARTUP CHECKS #### -######################## -try: - from TTS.api import TTS - from TTS.utils.synthesizer import Synthesizer -except ModuleNotFoundError: - print( - 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_available = False -try: - import deepspeed - deepspeed_available = True -except ImportError: - pass -if deepspeed_available: - print(f"[{params['branding']}Startup] DeepSpeed \033[93mDetected\033[0m") - print(f"[{params['branding']}Startup] Activate DeepSpeed in {params['branding']}settings") -else: - 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 - # Shutdown logic - - -# Create FastAPI app with lifespan -app = FastAPI(lifespan=startup_shutdown) -# Allow all origins, and set other CORS options -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], # Set this to the specific origins you want to allow - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], -) - -##################################### -#### MODEL LOADING AND UNLOADING #### -##################################### -# MODEL LOADERS Picker For API TTS, API Local, XTTSv2 Local, XTTSv2 FT -async def setup(): - global device - # 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 - if params["tts_method_api_tts"]: - print( - f"[{params['branding']}Model] \033[94mAPI TTS Loading\033[0m {params['tts_model_name']} \033[94minto\033[93m", - device, - "\033[0m", - ) - model = await api_load_model() - elif params["tts_method_api_local"]: - 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"[{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() - elif tts_method_xtts_ft: - print( - f"[{params['branding']}Model] \033[94mXTTSv2 FT Loading\033[0m /models/fintuned/model.pth \033[94minto\033[93m", - device, - "\033[0m", - ) - model = await xtts_ft_manual_load_model() - # Create an end timer for calculating load times - generate_end_time = time.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"[{params['branding']}Model] \033[94mModel Loaded in \033[93m{generate_elapsed_time:.2f} seconds.\033[0m" - ) - # Set "tts_model_loaded" to true - params["tts_model_loaded"] = True - # Set the output path for wav files - output_directory = this_dir / params["output_folder_wav_standalone"] - output_directory.mkdir(parents=True, exist_ok=True) - #Path(f'this_folder/outputs/').mkdir(parents=True, exist_ok=True) - - -# MODEL LOADER For "API TTS" -async def api_load_model(): - global model - model = TTS(params["tts_model_name"]).to(device) - return model - - -# 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 - 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) - else: - 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" -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 - if str(modeldownload_base_path) == "models": - config_path = this_dir / "models" / modeldownload_model_path / "config.json" - vocab_path_dir = this_dir / "models" / modeldownload_model_path / "vocab.json" - checkpoint_dir = this_dir / "models" / modeldownload_model_path - else: - 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" - vocab_path_dir = modeldownload_base_path / modeldownload_model_path / "vocab.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), - vocab_path=str(vocab_path_dir), - use_deepspeed=params["deepspeed_activate"], - ) - model.to(device) - return model - -# MODEL LOADER For "XTTSv2 FT" -async def xtts_ft_manual_load_model(): - global model - config = XttsConfig() - config_path = this_dir / "models" / "trainedmodel" / "config.json" - vocab_path_dir = this_dir / "models" / "trainedmodel" / "vocab.json" - checkpoint_dir = this_dir / "models" / "trainedmodel" - config.load_json(str(config_path)) - model = Xtts.init_from_config(config) - model.load_checkpoint( - config, - checkpoint_dir=str(checkpoint_dir), - vocab_path=str(vocab_path_dir), - use_deepspeed=params["deepspeed_activate"], - ) - model.to(device) - return model - -# MODEL UNLOADER -async def unload_model(model): - print(f"[{params['branding']}Model] \033[94mUnloading model \033[0m") - del model - if torch.cuda.is_available(): - torch.cuda.empty_cache() - params["tts_model_loaded"] = False - return None - - -# MODEL - Swap model based on Gradio selection API TTS, API Local, XTTSv2 Local -async def handle_tts_method_change(tts_method): - global model - global tts_method_xtts_ft - # 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 - params["tts_method_api_tts"] = True - params["deepspeed_activate"] = False - tts_method_xtts_ft = False - elif tts_method == "API Local": - params["tts_method_api_tts"] = False - params["tts_method_xtts_local"] = False - params["tts_method_api_local"] = True - params["deepspeed_activate"] = False - tts_method_xtts_ft = False - elif tts_method == "XTTSv2 Local": - params["tts_method_api_tts"] = False - params["tts_method_api_local"] = False - params["tts_method_xtts_local"] = True - tts_method_xtts_ft = False - elif tts_method == "XTTSv2 FT": - tts_method_xtts_ft = True - params["tts_method_api_tts"] = False - params["tts_method_api_local"] = False - params["tts_method_xtts_local"] = False - - # Unload the current model - model = await unload_model(model) - - # Load the correct model based on the updated params - await setup() - - -# 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", "XTTSv2 FT"]: - 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" - ) - - -################## -#### LOW VRAM #### -################## -# LOW VRAM - MODEL MOVER VRAM(cuda)<>System RAM(cpu) for Low VRAM setting -async def switch_device(): - global model, device - # Check if CUDA is available before performing GPU-related operations - if torch.cuda.is_available(): - if device == "cuda": - device = "cpu" - model.to(device) - torch.cuda.empty_cache() - 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 - try: - if new_low_vram_value is None: - raise ValueError("Missing 'low_vram' parameter") - - if params["low_vram"] == new_low_vram_value: - return Response( - 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) - if torch.cuda.is_available(): - device = "cpu" - 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: - # Handle the case where CUDA is not available - print( - f"[{params['branding']}Model] \033[91mError:\033[0m Nvidia CUDA is not available on this system. Unable to use LowVRAM mode." - ) - params["low_vram"] = False - else: - await unload_model(model) - if torch.cuda.is_available(): - device = "cuda" - 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() - else: - # Handle the case where CUDA is not available - print( - f"[{params['branding']}Model] \033[91mError:\033[0m Nvidia CUDA is not available on this system. Unable to use LowVRAM mode." - ) - params["low_vram"] = False - 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 -async def handle_deepspeed_change(value): - global model - if value: - # DeepSpeed enabled - 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 - params["tts_method_xtts_local"] = True - params["deepspeed_activate"] = True - await setup() - else: - # DeepSpeed disabled - 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 - - -# DEEPSPEED WEBSERVER- API Enable/Disable DeepSpeed -@app.post("/api/deepspeed") -async def deepspeed(request: Request, new_deepspeed_value: bool): - try: - if new_deepspeed_value is None: - 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'}.", - } - ) - ) - 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) -async def generate_audio(text, voice, language, temperature, repetition_penalty, output_file, streaming=False): - # Get the async generator from the internal function - response = generate_audio_internal(text, voice, language, temperature, repetition_penalty, output_file, streaming) - # If streaming, then return the generator as-is, otherwise just exhaust it and return - if streaming: - return response - async for _ in response: - pass - -async def generate_audio_internal(text, voice, language, temperature, repetition_penalty, output_file, streaming): - 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 & Xttsv2 FT Method - if params["tts_method_xtts_local"] or tts_method_xtts_ft: - print(f"[{params['branding']}TTSGen] {text}") - gpt_cond_latent, speaker_embedding = model.get_conditioning_latents( - audio_path=[f"{this_dir}/voices/{voice}"], - gpt_cond_len=model.config.gpt_cond_len, - max_ref_length=model.config.max_ref_len, - sound_norm_refs=model.config.sound_norm_refs, - ) - - # Common arguments for both functions - common_args = { - "text": text, - "language": language, - "gpt_cond_latent": gpt_cond_latent, - "speaker_embedding": speaker_embedding, - "temperature": float(temperature), - "length_penalty": float(model.config.length_penalty), - "repetition_penalty": float(repetition_penalty), - "top_k": int(model.config.top_k), - "top_p": float(model.config.top_p), - "enable_text_splitting": True - } - - # Determine the correct inference function and add streaming specific argument if needed - inference_func = model.inference_stream if streaming else model.inference - if streaming: - common_args["stream_chunk_size"] = 20 - - # Call the appropriate function - output = inference_func(**common_args) - - # Process the output based on streaming or non-streaming - if streaming: - # Streaming-specific operations - file_chunks = [] - wav_buf = io.BytesIO() - with wave.open(wav_buf, "wb") as vfout: - vfout.setnchannels(1) - vfout.setsampwidth(2) - vfout.setframerate(24000) - vfout.writeframes(b"") - wav_buf.seek(0) - yield wav_buf.read() - - for i, chunk in enumerate(output): - file_chunks.append(chunk) - if isinstance(chunk, list): - chunk = torch.cat(chunk, dim=0) - chunk = chunk.clone().detach().cpu().numpy() - chunk = chunk[None, : int(chunk.shape[0])] - chunk = np.clip(chunk, -1, 1) - chunk = (chunk * 32767).astype(np.int16) - yield chunk.tobytes() - else: - # Non-streaming-specific operation - torchaudio.save(output_file, torch.tensor(output["wav"]).unsqueeze(0), 24000) - - - # API LOCAL Methods - elif params["tts_method_api_local"]: - # Streaming only allowed for XTTSv2 local - if streaming: - raise ValueError("Streaming is only supported in XTTSv2 local") - - # Set the correct output path (different from the if statement) - print(f"[{params['branding']}TTSGen] Using API Local") - model.tts_to_file( - text=text, - file_path=output_file, - speaker_wav=[f"{this_dir}/voices/{voice}"], - language=language, - temperature=temperature, - length_penalty=model.config.length_penalty, - repetition_penalty=repetition_penalty, - top_k=model.config.top_k, - top_p=model.config.top_p, - ) - - # API TTS - elif params["tts_method_api_tts"]: - # Streaming only allowed for XTTSv2 local - if streaming: - raise ValueError("Streaming is only supported in XTTSv2 local") - - print(f"[{params['branding']}TTSGen] Using API TTS") - model.tts_to_file( - text=text, - file_path=output_file, - speaker_wav=[f"{this_dir}/voices/{voice}"], - language=language, - ) - - # 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"[{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 -@app.route("/api/generate", methods=["POST"]) -async def generate(request: Request): - try: - # Get parameters from JSON body - data = await request.json() - text = data["text"] - voice = data["voice"] - language = data["language"] - temperature = data["temperature"] - repetition_penalty = data["repetition_penalty"] - output_file = data["output_file"] - streaming = False - # Generation logic - response = await generate_audio(text, voice, language, temperature, repetition_penalty, output_file, streaming) - if streaming: - return StreamingResponse(response, media_type="audio/wav") - return JSONResponse( - content={"status": "generate-success", "data": {"audio_path": output_file}} - ) - except Exception as e: - return JSONResponse(content={"status": "error", "message": str(e)}) - - -################################################### -#### POPULATE FILES LIST FROM VOICES DIRECTORY #### -################################################### -# List files in the "voices" directory -def list_files(directory): - files = [ - f - for f in os.listdir(directory) - if os.path.isfile(os.path.join(directory, f)) and f.endswith(".wav") - ] - return files - -############################# -#### JSON CONFIG UPDATER #### -############################# - -# Create an instance of Jinja2Templates for rendering HTML templates -templates = Jinja2Templates(directory=this_dir / "system") - -# Create a dependency to get the current JSON data -def get_json_data(): - with open(this_dir / "confignew.json", "r") as json_file: - data = json.load(json_file) - return data - - -# Define an endpoint function -@app.get("/settings") -async def get_settings(request: Request): - wav_files = list_files(this_dir / "voices") - # Render the template with the current JSON data and list of WAV files - return templates.TemplateResponse( - "/at_admin/at_settings.html", - { - "request": request, - "data": get_json_data(), - "modeldownload_model_path": modeldownload_model_path, - "wav_files": wav_files, - }, - ) - -# Define an endpoint to serve static files -app.mount("/static", StaticFiles(directory=str(this_dir / "system")), name="static") - -@app.post("/update-settings") -async def update_settings( - request: Request, - activate: bool = Form(...), - autoplay: bool = Form(...), - deepspeed_activate: bool = Form(...), - delete_output_wavs: str = Form(...), - ip_address: str = Form(...), - language: str = Form(...), - local_temperature: str = Form(...), - local_repetition_penalty: str = Form(...), - low_vram: bool = Form(...), - tts_model_loaded: bool = Form(...), - tts_model_name: str = Form(...), - narrator_enabled: bool = Form(...), - narrator_voice: str = Form(...), - output_folder_wav: str = Form(...), - port_number: str = Form(...), - remove_trailing_dots: bool = Form(...), - show_text: bool = Form(...), - tts_method: str = Form(...), - voice: str = Form(...), - data: dict = Depends(get_json_data), -): - # Update the settings based on the form values - data["activate"] = activate - data["autoplay"] = autoplay - data["deepspeed_activate"] = deepspeed_activate - data["delete_output_wavs"] = delete_output_wavs - data["ip_address"] = ip_address - data["language"] = language - data["local_temperature"] = local_temperature - data["local_repetition_penalty"] = local_repetition_penalty - data["low_vram"] = low_vram - data["tts_model_loaded"] = tts_model_loaded - data["tts_model_name"] = tts_model_name - data["narrator_enabled"] = narrator_enabled - data["narrator_voice"] = narrator_voice - data["output_folder_wav"] = output_folder_wav - data["port_number"] = port_number - data["remove_trailing_dots"] = remove_trailing_dots - data["show_text"] = show_text - data["tts_method_api_local"] = tts_method == "api_local" - data["tts_method_api_tts"] = tts_method == "api_tts" - 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 / "confignew.json", "w") as json_file: - json.dump(data, json_file) - - # Redirect to the settings page to display the updated settings - return RedirectResponse(url="/settings", status_code=303) - - -################################## -#### SETTINGS PAGE DEMO VOICE #### -################################## - -@app.get("/tts-demo-request", response_class=StreamingResponse) -async def tts_demo_request_streaming(text: str, voice: str, language: str, output_file: str): - try: - output_file_path = this_dir / "outputs" / output_file - stream = await generate_audio(text, voice, language, temperature, repetition_penalty, output_file_path, streaming=True) - return StreamingResponse(stream, media_type="audio/wav") - except Exception as e: - print(f"An error occurred: {e}") - return JSONResponse(content={"error": "An error occurred"}, status_code=500) - -@app.post("/tts-demo-request", response_class=JSONResponse) -async def tts_demo_request(request: Request, text: str = Form(...), voice: str = Form(...), language: str = Form(...), output_file: str = Form(...)): - try: - output_file_path = this_dir / "outputs" / output_file - await generate_audio(text, voice, language, temperature, repetition_penalty, output_file_path, streaming=False) - return JSONResponse(content={"output_file_path": str(output_file)}, status_code=200) - except Exception as e: - print(f"An error occurred: {e}") - return JSONResponse(content={"error": "An error occurred"}, status_code=500) - - -##################### -#### Audio feeds #### -##################### - -# Gives web access to the output files -@app.get("/audio/{filename}") -async def get_audio(filename: str): - audio_path = this_dir / "outputs" / filename - return FileResponse(audio_path) - -@app.get("/audiocache/{filename}") -async def get_audio(filename: str): - audio_path = Path("outputs") / filename - if not audio_path.is_file(): - raise HTTPException(status_code=404, detail="File not found") - - response = FileResponse( - path=audio_path, - media_type='audio/wav', - filename=filename - ) - # Set caching headers - response.headers["Cache-Control"] = "public, max-age=604800" # Cache for one week - response.headers["ETag"] = str(audio_path.stat().st_mtime) # Use the file's last modified time as a simple ETag - - return response - -######################### -#### VOICES LIST API #### -######################### -# Define the new endpoint -@app.get("/api/voices") -async def get_voices(): - wav_files = list_files(this_dir / "voices") - return {"voices": wav_files} - -########################### -#### PREVIEW VOICE API #### -########################### -@app.post("/api/previewvoice/", response_class=JSONResponse) -async def preview_voice(request: Request, voice: str = Form(...)): - try: - # Hardcoded settings - language = "en" - output_file_name = "api_preview_voice" - - # Clean the voice filename for inclusion in the text - clean_voice_filename = re.sub(r'\.wav$', '', voice.replace(' ', '_')) - clean_voice_filename = re.sub(r'[^a-zA-Z0-9]', ' ', clean_voice_filename) - - # Generate the audio - text = f"Hello, this is a preview of voice {clean_voice_filename}." - - # Generate the audio - output_file_path = this_dir / "outputs" / f"{output_file_name}.wav" - await generate_audio(text, voice, language, temperature, repetition_penalty, output_file_path, streaming=False) - - # Generate the URL - output_file_url = f'http://{params["ip_address"]}:{params["port_number"]}/audio/{output_file_name}.wav' - - # Return the response with both local file path and URL - return JSONResponse( - content={ - "status": "generate-success", - "output_file_path": str(output_file_path), - "output_file_url": str(output_file_url), - }, - status_code=200, - ) - except Exception as e: - print(f"An error occurred: {e}") - return JSONResponse(content={"error": "An error occurred"}, status_code=500) - -######################## -#### GENERATION API #### -######################## -import html -import re -import uuid -import numpy as np -import soundfile as sf -import sys -import hashlib - -############################## -#### Streaming Generation #### -############################## - -@app.get("/api/tts-generate-streaming", response_class=StreamingResponse) -async def tts_generate_streaming(text: str, voice: str, language: str, output_file: str): - try: - output_file_path = this_dir / "outputs" / output_file - stream = await generate_audio(text, voice, language, temperature, repetition_penalty, output_file_path, streaming=True) - return StreamingResponse(stream, media_type="audio/wav") - except Exception as e: - print(f"An error occurred: {e}") - return JSONResponse(content={"error": "An error occurred"}, status_code=500) - -@app.post("/api/tts-generate-streaming", response_class=JSONResponse) -async def tts_generate_streaming(request: Request, text: str = Form(...), voice: str = Form(...), language: str = Form(...), output_file: str = Form(...)): - try: - output_file_path = this_dir / "outputs" / output_file - await generate_audio(text, voice, language, temperature, repetition_penalty, output_file_path, streaming=False) - return JSONResponse(content={"output_file_path": str(output_file)}, status_code=200) - except Exception as e: - print(f"An error occurred: {e}") - return JSONResponse(content={"error": "An error occurred"}, status_code=500) - -############################## -#### Standard Generation #### -############################## - -# Check for PortAudio library on Linux -try: - import sounddevice as sd - sounddevice_installed=True -except OSError: - print(f"[{params['branding']}Startup] \033[91mInfo\033[0m PortAudio library not found. If you wish to play TTS in standalone mode through the API suite") - print(f"[{params['branding']}Startup] \033[91mInfo\033[0m please install PortAudio. This will not affect any other features or use of Alltalk.") - print(f"[{params['branding']}Startup] \033[91mInfo\033[0m If you don't know what the API suite is, then this message is nothing to worry about.") - sounddevice_installed=False - if sys.platform.startswith('linux'): - print(f"[{params['branding']}Startup] \033[91mInfo\033[0m On Linux, you can use the following command to install PortAudio:") - print(f"[{params['branding']}Startup] \033[91mInfo\033[0m sudo apt-get install portaudio19-dev") - -from typing import Union, Dict, List -from pydantic import BaseModel, ValidationError, Field - -def play_audio(file_path, volume): - data, fs = sf.read(file_path) - sd.play(volume * data, fs) - sd.wait() - -class Request(BaseModel): - # Define the structure of the 'Request' class if needed - pass - -class JSONInput(BaseModel): - text_input: str = Field(..., max_length=2000, description="text_input needs to be 2000 characters or less.") - text_filtering: str = Field(..., pattern="^(none|standard|html)$", description="text_filtering needs to be 'none', 'standard' or 'html'.") - character_voice_gen: str = Field(..., pattern="^.*\.wav$", description="character_voice_gen needs to be the name of a wav file e.g. mysample.wav.") - narrator_enabled: bool = Field(..., description="narrator_enabled needs to be true or false.") - narrator_voice_gen: str = Field(..., pattern="^.*\.wav$", description="narrator_voice_gen needs to be the name of a wav file e.g. mysample.wav.") - text_not_inside: str = Field(..., pattern="^(character|narrator)$", description="text_not_inside needs to be 'character' or 'narrator'.") - language: str = Field(..., pattern="^(ar|zh-cn|cs|nl|en|fr|de|hu|it|ja|ko|pl|pt|ru|es|tr)$", description="language needs to be one of the following ar|zh-cn|cs|nl|en|fr|de|hu|it|ja|ko|pl|pt|ru|es|tr.") - output_file_name: str = Field(..., pattern="^[a-zA-Z0-9_]+$", description="output_file_name needs to be the name without any special characters or file extension e.g. 'filename'") - output_file_timestamp: bool = Field(..., description="output_file_timestamp needs to be true or false.") - autoplay: bool = Field(..., description="autoplay needs to be a true or false value.") - autoplay_volume: float = Field(..., ge=0.1, le=1.0, description="autoplay_volume needs to be from 0.1 to 1.0") - - @classmethod - def validate_autoplay_volume(cls, value): - if not (0.1 <= value <= 1.0): - raise ValueError("Autoplay volume must be between 0.1 and 1.0") - return value - - -class TTSGenerator: - @staticmethod - def validate_json_input(json_data: Union[Dict, str]) -> Union[None, str]: - try: - if isinstance(json_data, str): - json_data = json.loads(json_data) - JSONInput(**json_data) - return None # JSON is valid - except ValidationError as e: - return str(e) - -def process_text(text): - # Normalize HTML encoded quotes - text = html.unescape(text) - # Replace ellipsis with a single dot - text = re.sub(r'\.{3,}', '.', text) - # Pattern to identify combined narrator and character speech - combined_pattern = r'(\*[^*"]+\*|"[^"*]+")' - # List to hold parts of speech along with their type - ordered_parts = [] - # Track the start of the next segment - start = 0 - # Find all matches - for match in re.finditer(combined_pattern, text): - # Add the text before the match, if any, as ambiguous - if start < match.start(): - ambiguous_text = text[start:match.start()].strip() - if ambiguous_text: - ordered_parts.append(('ambiguous', ambiguous_text)) - # Add the matched part as either narrator or character - matched_text = match.group(0) - if matched_text.startswith('*') and matched_text.endswith('*'): - ordered_parts.append(('narrator', matched_text.strip('*').strip())) - elif matched_text.startswith('"') and matched_text.endswith('"'): - ordered_parts.append(('character', matched_text.strip('"').strip())) - else: - # In case of mixed or improperly formatted parts - if '*' in matched_text: - ordered_parts.append(('narrator', matched_text.strip('*').strip('"'))) - else: - ordered_parts.append(('character', matched_text.strip('"').strip('*'))) - # Update the start of the next segment - start = match.end() - # Add any remaining text after the last match as ambiguous - if start < len(text): - ambiguous_text = text[start:].strip() - if ambiguous_text: - ordered_parts.append(('ambiguous', ambiguous_text)) - return ordered_parts - -def standard_filtering(text_input): - text_output = (text_input - .replace("***", "") - .replace("**", "") - .replace("*", "") - .replace("\n\n", "\n") - .replace("'", "'") - ) - return text_output - -def combine(output_file_timestamp, output_file_name, audio_files): - audio = np.array([]) - sample_rate = None - try: - for audio_file in audio_files: - audio_data, current_sample_rate = sf.read(audio_file) - if audio.size == 0: - audio = audio_data - sample_rate = current_sample_rate - elif sample_rate == current_sample_rate: - audio = np.concatenate((audio, audio_data)) - else: - raise ValueError("Sample rates of input files are not consistent.") - except Exception as e: - # Handle exceptions (e.g., file not found, invalid audio format) - return None, None - if output_file_timestamp: - timestamp = int(time.time()) - output_file_path = os.path.join(this_dir / "outputs" / f'{output_file_name}_{timestamp}_combined.wav') - output_file_url = f'http://{params["ip_address"]}:{params["port_number"]}/audio/{output_file_name}_{timestamp}_combined.wav' - output_cache_url = f'http://{params["ip_address"]}:{params["port_number"]}/audiocache/{output_file_name}_{timestamp}_combined.wav' - else: - output_file_path = os.path.join(this_dir / "outputs" / f'{output_file_name}_combined.wav') - output_file_url = f'http://{params["ip_address"]}:{params["port_number"]}/audio/{output_file_name}_combined.wav' - output_cache_url = f'http://{params["ip_address"]}:{params["port_number"]}/audiocache/{output_file_name}_combined.wav' - try: - sf.write(output_file_path, audio, samplerate=sample_rate) - # Clean up unnecessary files - for audio_file in audio_files: - os.remove(audio_file) - except Exception as e: - # Handle exceptions (e.g., failed to write output file) - return None, None - return output_file_path, output_file_url, output_cache_url - -# Generation API (separate from text-generation-webui) -@app.post("/api/tts-generate", response_class=JSONResponse) -async def tts_generate( - text_input: str = Form(...), - text_filtering: str = Form(...), - character_voice_gen: str = Form(...), - narrator_enabled: bool = Form(...), - narrator_voice_gen: str = Form(...), - text_not_inside: str = Form(...), - language: str = Form(...), - output_file_name: str = Form(...), - output_file_timestamp: bool = Form(...), - autoplay: bool = Form(...), - autoplay_volume: float = Form(...), - streaming: bool = Form(False), -): - try: - json_input_data = { - "text_input": text_input, - "text_filtering": text_filtering, - "character_voice_gen": character_voice_gen, - "narrator_enabled": narrator_enabled, - "narrator_voice_gen": narrator_voice_gen, - "text_not_inside": text_not_inside, - "language": language, - "output_file_name": output_file_name, - "output_file_timestamp": output_file_timestamp, - "autoplay": autoplay, - "autoplay_volume": autoplay_volume, - "streaming": streaming, - } - JSONresult = TTSGenerator.validate_json_input(json_input_data) - if JSONresult is None: - pass - else: - return JSONResponse(content={"error": JSONresult}, status_code=400) - if narrator_enabled: - processed_parts = process_text(text_input) - audio_files_all_paragraphs = [] - for part_type, part in processed_parts: - # Skip parts that are too short - if len(part.strip()) <= 3: - continue - # Determine the voice to use based on the part type - if part_type == 'narrator': - voice_to_use = narrator_voice_gen - print(f"[{params['branding']}TTSGen] \033[92mNarrator\033[0m") # Green - elif part_type == 'character': - voice_to_use = character_voice_gen - print(f"[{params['branding']}TTSGen] \033[36mCharacter\033[0m") # Yellow - else: - # Handle ambiguous parts based on user preference - voice_to_use = character_voice_gen if text_not_inside == "character" else narrator_voice_gen - voice_description = "\033[36mCharacter (Text-not-inside)\033[0m" if text_not_inside == "character" else "\033[92mNarrator (Text-not-inside)\033[0m" - print(f"[{params['branding']}TTSGen] {voice_description}") - # Replace multiple exclamation marks, question marks, or other punctuation with a single instance - cleaned_part = re.sub(r'([!?.])\1+', r'\1', part) - # Further clean to remove any other unwanted characters - cleaned_part = re.sub(r'[^a-zA-Z0-9\s\.,;:!?\-\'"\u0400-\u04FFÀ-ÿ\u0150\u0151\u0170\u0171]\$', '', cleaned_part) - # Remove all newline characters (single or multiple) - cleaned_part = re.sub(r'\n+', ' ', cleaned_part) - output_file = this_dir / "outputs" / f"{output_file_name}_{uuid.uuid4()}_{int(time.time())}.wav" - output_file_str = output_file.as_posix() - response = await generate_audio(cleaned_part, voice_to_use, language,temperature, repetition_penalty, output_file_str, streaming) - audio_path = output_file_str - audio_files_all_paragraphs.append(audio_path) - # Combine audio files across paragraphs - output_file_path, output_file_url, output_cache_url = combine(output_file_timestamp, output_file_name, audio_files_all_paragraphs) - else: - if output_file_timestamp: - timestamp = int(time.time()) - # Generate a standard UUID - original_uuid = uuid.uuid4() - # Hash the UUID using SHA-256 - hash_object = hashlib.sha256(str(original_uuid).encode()) - hashed_uuid = hash_object.hexdigest() - # Truncate to the desired length, for example, 16 characters - short_uuid = hashed_uuid[:5] - output_file_path = this_dir / "outputs" / f"{output_file_name}_{timestamp}{short_uuid}.wav" - output_file_url = f'http://{params["ip_address"]}:{params["port_number"]}/audio/{output_file_name}_{timestamp}{short_uuid}.wav' - output_cache_url = f'http://{params["ip_address"]}:{params["port_number"]}/audiocache/{output_file_name}_{timestamp}{short_uuid}.wav' - else: - output_file_path = this_dir / "outputs" / f"{output_file_name}.wav" - output_file_url = f'http://{params["ip_address"]}:{params["port_number"]}/audio/{output_file_name}.wav' - output_cache_url = f'http://{params["ip_address"]}:{params["port_number"]}/audiocache/{output_file_name}.wav' - if text_filtering == "html": - cleaned_string = html.unescape(standard_filtering(text_input)) - cleaned_string = re.sub(r'([!?.])\1+', r'\1', text_input) - # Further clean to remove any other unwanted characters - cleaned_string = re.sub(r'[^a-zA-Z0-9\s\.,;:!?\-\'"\u0400-\u04FFÀ-ÿ\u0150\u0151\u0170\u0171]\$', '', cleaned_string) - # Remove all newline characters (single or multiple) - cleaned_string = re.sub(r'\n+', ' ', cleaned_string) - elif text_filtering == "standard": - cleaned_string = re.sub(r'([!?.])\1+', r'\1', text_input) - # Further clean to remove any other unwanted characters - cleaned_string = re.sub(r'[^a-zA-Z0-9\s\.,;:!?\-\'"\u0400-\u04FFÀ-ÿ\u0150\u0151\u0170\u0171]\$', '', cleaned_string) - # Remove all newline characters (single or multiple) - cleaned_string = re.sub(r'\n+', ' ', cleaned_string) - else: - cleaned_string = text_input - response = await generate_audio(cleaned_string, character_voice_gen, language, temperature, repetition_penalty, output_file_path, streaming) - if sounddevice_installed == False or streaming == True: - autoplay = False - if autoplay: - play_audio(output_file_path, autoplay_volume) - if streaming: - return StreamingResponse(response, media_type="audio/wav") - return JSONResponse(content={"status": "generate-success", "output_file_path": str(output_file_path), "output_file_url": str(output_file_url), "output_cache_url": str(output_cache_url)}, status_code=200) - except Exception as e: - return JSONResponse(content={"status": "generate-failure", "error": "An error occurred"}, status_code=500) - - -########################## -#### Current Settings #### -########################## -# Define the available models -models_available = [ - {"name": "Coqui", "model_name": "API TTS"}, - {"name": "Coqui", "model_name": "API Local"}, - {"name": "Coqui", "model_name": "XTTSv2 Local"} -] - -@app.get('/api/currentsettings') -def get_current_settings(): - # Determine the current model loaded - if params["tts_method_api_tts"]: - current_model_loaded = "API TTS" - elif params["tts_method_api_local"]: - current_model_loaded = "API Local" - elif params["tts_method_xtts_local"]: - current_model_loaded = "XTTSv2 Local" - else: - current_model_loaded = None # or a default value if no method is active - - settings = { - "models_available": models_available, - "current_model_loaded": current_model_loaded, - "deepspeed_available": deepspeed_available, - "deepspeed_status": params["deepspeed_activate"], - "low_vram_status": params["low_vram"], - "finetuned_model": finetuned_model - } - return settings # Automatically converted to JSON by Fas - -############################# -#### Word Add-in Sharing #### -############################# -# Mount the static files from the 'word_addin' directory -app.mount("/api/word_addin", StaticFiles(directory=os.path.join(this_dir / 'system' / 'word_addin')), name="word_addin") - -############################################# -#### TTS Generator Comparision Endpoints #### -############################################# -import subprocess -import aiofiles - -class TTSItem(BaseModel): - id: int - fileUrl: str - text: str - characterVoice: str - language: str - -class TTSData(BaseModel): - ttsList: List[TTSItem] - -@app.post("/api/save-tts-data") -async def save_tts_data(tts_data: List[TTSItem]): - # Convert the list of Pydantic models to a list of dictionaries - tts_data_list = [item.dict() for item in tts_data] - # Serialize the list of dictionaries to a JSON string - tts_data_json = json.dumps(tts_data_list, indent=4) - async with aiofiles.open(this_dir / "outputs" / "ttsList.json", 'w') as f: - await f.write(tts_data_json) - return {"message": "Data saved successfully"} - -import sys - -@app.get("/api/trigger-analysis") -async def trigger_analysis(threshold: int = Query(default=98)): - venv_path = sys.prefix - env = os.environ.copy() - env["PATH"] = os.path.join(venv_path, "bin") + ":" + env["PATH"] - ttslist_path = this_dir / "outputs" / "ttsList.json" - wavfile_path = this_dir / "outputs" - subprocess.run(["python", "tts_diff.py", f"--threshold={threshold}", f"--ttslistpath={ttslist_path}", f"--wavfilespath={wavfile_path}"], cwd=this_dir / "system" / "tts_diff", env=env) - # Read the analysis summary - try: - with open(this_dir / "outputs" / "analysis_summary.json", "r") as summary_file: - summary_data = json.load(summary_file) - except FileNotFoundError: - summary_data = {"error": "Analysis summary file not found."} - return {"message": "Analysis Completed", "summary": summary_data} - -################################################# -#### TTS Generator SRT Subtitiles generation #### -################################################# -@app.get("/api/srt-generation") -async def srt_generation(): - venv_path = sys.prefix - env = os.environ.copy() - env["PATH"] = os.path.join(venv_path, "bin") + ":" + env["PATH"] - ttslist_path = this_dir / "outputs" / "ttsList.json" - wavfile_path = this_dir / "outputs" - subprocess.run(["python", "tts_srt.py", f"--ttslistpath={ttslist_path}", f"--wavfilespath={wavfile_path}"], cwd=this_dir / "system" / "tts_srt", env=env) - - srt_file_path = this_dir / "outputs" / "subtitles.srt" - if not srt_file_path.exists(): - raise HTTPException(status_code=404, detail="Subtitle file not found.") - - return FileResponse(path=srt_file_path, filename="subtitles.srt", media_type='application/octet-stream') - -################################################### -#### Webserver Startup & Initial model Loading #### -################################################### - -# Get the admin interface template -template = templates.get_template("admin.html") -# Render the template with the dynamic values -rendered_html = template.render(params=params) - -############################### -#### Internal script ready #### -############################### -@app.get("/ready") -async def ready(): - return Response("Ready endpoint") - -############################ -#### External API ready #### -############################ -@app.get("/api/ready") -async def ready(): - return Response("Ready") - -@app.get("/") -async def read_root(): - return HTMLResponse(content=rendered_html, status_code=200) - -# Start Uvicorn Webserver -host_parameter = params["ip_address"] -port_parameter = int(params["port_number"]) - -if __name__ == "__main__": - import uvicorn - - uvicorn.run(app, host=host_parameter, port=port_parameter, log_level="warning")