""" XTTS (Coqui TTS) Engine Implementation for AllTalk TTS Version: 1.0 Last Updated: 2024 This implementation serves as both a functional XTTS engine and a reference example for implementing new TTS engines in AllTalk. It demonstrates proper: - Variable initialization and management - Debug message handling - Model loading/unloading patterns - Resource management - Generation handling with streaming support Implementation Notes: - All core system variables must be maintained - All functions prefixed with "DONT CHANGE" must remain unmodified - Code additions should be placed between the marked sections in each function - Debug messages use self.print_message() with appropriate message types - All functions must implement self.debug_func_entry() for trace logging Note: Text between "↑↑↑ Keep everything above this line ↑↑↑" and "↓↓↓ Keep everything below this line ↓↓↓" markers **TYPICALLY** must remain unchanged as it contains critical system integration code. Note: You can add new functions, just DONT remove the functions that are already there, even if they are doing nothing as `tts_server.py` will still look for their existance and fail if they are missing. """ ######################################## # Default imports # Do not change this # ######################################## import os import gc import sys import glob import json import time import inspect import torch import logging from pathlib import Path from fastapi import (HTTPException) logging.disable(logging.WARNING) # Confguration file management for confignew.json try: from .config import AlltalkConfig, AlltalkTTSEnginesConfig, AlltalkNewEnginesConfig # TGWUI import except ImportError: from config import AlltalkConfig, AlltalkTTSEnginesConfig, AlltalkNewEnginesConfig # Standalone import def initialize_configs(): """Initialize all configuration instances""" config = AlltalkConfig.get_instance() tts_engines_config = AlltalkTTSEnginesConfig.get_instance() new_engines_config = AlltalkNewEnginesConfig.get_instance() return config, tts_engines_config, new_engines_config # Load in the central config management config, tts_engines_config, new_engines_config = initialize_configs() ###################################################### # Get Pytorch & Python versions # Do not change this # ###################################################### pytorch_version = torch.__version__ cuda_version = torch.version.cuda major, minor, micro = sys.version_info[:3] python_version = f"{major}.{minor}.{micro}" ############################################ # DeepSpeed imports # POSSIBLY change this # ############################################ """ If the new TTS engine you are importing doesnt support DeepSpeed, then you can simply change `model_supports_deepspeed_true_or_false` to `False` This will be much faster when starting up the engine. This is seperate from what is stored in your `model_settings.json` file. """ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ MODIFY THIS LINE ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ model_supports_deepspeed_true_or_false = True # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ MODIFY THIS LINE ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ if model_supports_deepspeed_true_or_false: try: import deepspeed deepspeed_available = True except ImportError: deepspeed_available = False pass ####################################################### # TTS Engine-Specific Imports and Setup # Change this # ####################################################### """ This section is for TTS engine specific imports and global variable setup. Guidelines: 1. Import all required modules for your TTS engine 2. Handle import errors with appropriate error messages 3. Set up any global variables or configurations needed by your engine 4. Use try/except blocks to gracefully handle missing dependencies 5. Ensure all error messages follow the AllTalk format for print messages. At this stage the `def print_messages` is not available, so please use standard `print("[AllTalk ENG] some message here")` messages. Example structure: try: # Import your TTS engine's required modules from your_tts_engine import required_modules except ModuleNotFoundError: # Handle missing dependencies with helpful error messages print("Missing required modules. Please install...") raise For XTTS specifically, we need: - torchaudio: For audio processing - TTS modules: Core Coqui TTS functionality - Support libraries: wave, io, random, numpy for audio handling """ try: import torchaudio import wave import io import random import numpy as np from TTS.tts.configs.xtts_config import XttsConfig from TTS.tts.models.xtts import Xtts from TTS.api import TTS from TTS.utils.synthesizer import Synthesizer except ModuleNotFoundError: brand = "[AllTalk ENG]" print(f"{brand} \033[91mError\033[0m Could not find the Coqui TTS modules.") print(f"{brand} \033[91mError\033[0m Please re-install the requirements.") raise ######################################################### # Class setup # Change the relevant functions as needed # ######################################################### class tts_class: """ TTS Engine Implementation Class This class provides the interface between `tts_server.py` and whatever TTS engine you install. It handles model loading, voice management, and TTS generation in both streaming and non-streaming modes. Streaming will only be supported if the underlying TTS engine actually supports streaming. Key Responsibilities: 1. Model Management: - Loading/unloading models - Managing model state between CPU and GPU - Handling DeepSpeed integration 2. Voice Management: - Managing voice samples or model files 3. TTS Generation: - Converting text to speech - Supporting streaming output (If the engine supports it) - Managing generation parameters 4. System Integration: - Implementing standard AllTalk interfaces - Managing engine state and configuration - Handling resource allocation """ ############################################### # Central print function # Do not change this # ############################################### def print_message(self, message, message_type="standard", component="ENG"): """ Centralized print function for messages. Use this for print output to console. As this is the model Engine, all `component` printouts are set to ENG as default. Args: message (str): The message to print message_type (str): Type of message (standard/warning/error/debug_*/debug) component (str): Component identifier (TTS/ENG/GEN/API/etc.) Example Use: self.print_message("This is a standard print out mesage to a user) self.print_message("This is a debug_tts message, message_type="debug_tts") self.print_message("This is an error message to a user, message_type="error") self.print_message("This is an warning message to a user, message_type="warning") Debug Types: debug_func: Tracks function entry debug_tts: Enable TTS process debugging debug_tts_variables: Enable variable state debugging WARNING: This is a core system function. Do not modify its implementation as it provides standardized version reporting across all engines. """ # ANSI color codes BLUE = "\033[94m" MAGENTA = "\033[95m" YELLOW = "\033[93m" RED = "\033[91m" GREEN = "\033[92m" RESET = "\033[0m" prefix = f"[{config.branding}{component}] " if message_type.startswith("debug_"): debug_flag = getattr(config.debugging, message_type, False) if not debug_flag: return if message_type == "debug_func" and "Function entry:" in message: message_parts = message.split("Function entry:", 1) print(f"{prefix}{BLUE}Debug{RESET} {YELLOW}{message_type}{RESET} Function entry:{GREEN}{message_parts[1]}{RESET} in model_engine") else: print(f"{prefix}{BLUE}Debug{RESET} {YELLOW}{message_type}{RESET} {message}") elif message_type == "debug": print(f"{prefix}{BLUE}Debug{RESET} {message}") elif message_type == "warning": print(f"{prefix}{YELLOW}Warning{RESET} {message}") elif message_type == "error": print(f"{prefix}{RED}Error{RESET} {message}") else: print(f"{prefix}{message}") def debug_func_entry(self): """Log function entry if debug_func is enabled.""" if config.debugging.debug_func: current_func = inspect.currentframe().f_back.f_code.co_name self.print_message(f"Function entry: {current_func}", "debug_func") ############################################# # Script initalisation # Do not change this # ############################################# def __init__(self): """ Initialize the XTTS engine instance. WARNING: This class requires specific variables to interface with AllTalk's main system (tts_server.py). Do not remove or rename any of the predefined variables as they are required for proper system integration. Required System Interface Variables: 1. Core System Variables (DO NOT MODIFY): Base Configuration: - self.this_dir: Engine directory path (where this script is located) - self.main_dir: AllTalk root directory - self.device: Processing device ("cuda" or "cpu") - self.cuda_is_available: Whether GPU/CUDA is available State Tracking: - self.tts_generating_lock: Prevents concurrent generation requests - self.tts_stop_generation: Signals generation stop request - self.tts_narrator_generatingtts: Tracks narrator mode for optimization - self.model: Active TTS model instance - self.is_tts_model_loaded: Whether a model is currently loaded - self.current_model_loaded: Name of currently loaded model - self.available_models: List of models found by scan_models_folder - self.setup_has_run: Tracks if setup() has completed 2. Engine Configuration Variables (DO NOT MODIFY): - self.engines_available: List of all available TTS engines - self.engine_loaded: Currently selected TTS engine - self.selected_model: Currently selected model name 3. Model Settings (SET VIA model_settings.json): Capability Flags: - self.audio_format: Output audio format (wav, mp3, etc.) - self.deepspeed_capable: DeepSpeed acceleration support - self.generationspeed_capable: Speed adjustment support - self.languages_capable: Multi-language support - self.lowvram_capable: Low VRAM mode support - self.multimodel_capable: Multiple model support - self.repetitionpenalty_capable: Repetition penalty support - self.streaming_capable: Audio streaming support - self.temperature_capable: Temperature adjustment support - self.multivoice_capable: Multiple voice support - self.pitch_capable: Pitch adjustment support Engine Settings: - self.def_character_voice: Default character voice - self.def_narrator_voice: Default narrator voice - self.deepspeed_enabled: DeepSpeed status - self.engine_installed: Engine installation status - self.generationspeed_set: Current speed setting - self.lowvram_enabled: Low VRAM mode status - self.repetitionpenalty_set: Current repetition penalty - self.temperature_set: Current temperature setting - self.pitch_set: Current pitch setting OpenAI Voice Mappings: - self.openai_alloy: Alloy voice mapping - self.openai_echo: Echo voice mapping - self.openai_fable: Fable voice mapping - self.openai_nova: Nova voice mapping - self.openai_onyx: Onyx voice mapping - self.openai_shimmer: Shimmer voice mapping Integration Requirements: - All variables must be present even if unused by your engine - Capability flags should accurately reflect engine features - Settings should have sensible defaults even if not used - OpenAI mappings should be set even if not supporting OpenAI compatibility Note: Variables marked (DO NOT MODIFY) are critical system integration points. Other variables should be configured through their respective JSON files or you can add new central variables in the section provided down below. """ # DO NOT MODIFY - Sets up the base variables required for any tts engine # self.this_dir = Path(__file__).parent.resolve() self.main_dir = Path(__file__).parent.parent.parent.parent.resolve() self.device = "cuda" if torch.cuda.is_available() else "cpu" self.cuda_is_available = torch.cuda.is_available() self.tts_generating_lock = False self.tts_stop_generation = False self.tts_narrator_generatingtts = False self.model = None self.is_tts_model_loaded = False self.current_model_loaded = None self.available_models = None self.setup_has_run = False self.engines_available = tts_engines_config.get_engine_names_available() self.engine_loaded = tts_engines_config.engine_loaded self.selected_model = tts_engines_config.selected_model # DO NOT MODIFY - Load in the current TTS Engines model_settings.json file with open(os.path.join(self.this_dir, "model_settings.json"), "r") as f: model_settings_file = json.load(f) # DO NOT MODIFY - Model details from model_settings.json self.manufacturer_name = model_settings_file["model_details"]["manufacturer_name"] self.manufacturer_website = model_settings_file["model_details"]["manufacturer_website"] # DO NOT MODIFY - Model capabilities from model_settings.json self.audio_format = model_settings_file["model_capabilties"]["audio_format"] self.deepspeed_capable = model_settings_file["model_capabilties"]["deepspeed_capable"] self.deepspeed_available = 'deepspeed' in globals() self.generationspeed_capable = model_settings_file["model_capabilties"]["generationspeed_capable"] self.languages_capable = model_settings_file["model_capabilties"]["languages_capable"] self.lowvram_capable = model_settings_file["model_capabilties"]["lowvram_capable"] self.multimodel_capable = model_settings_file["model_capabilties"]["multimodel_capable"] self.repetitionpenalty_capable = model_settings_file["model_capabilties"]["repetitionpenalty_capable"] self.streaming_capable = model_settings_file["model_capabilties"]["streaming_capable"] self.temperature_capable = model_settings_file["model_capabilties"]["temperature_capable"] self.multivoice_capable = model_settings_file["model_capabilties"]["multivoice_capable"] self.pitch_capable = model_settings_file["model_capabilties"]["pitch_capable"] # DO NOT MODIFY - Engine settings from model_settings.json self.def_character_voice = model_settings_file["settings"]["def_character_voice"] self.def_narrator_voice = model_settings_file["settings"]["def_narrator_voice"] self.deepspeed_enabled = model_settings_file["settings"]["deepspeed_enabled"] self.engine_installed = model_settings_file["settings"]["engine_installed"] self.generationspeed_set = model_settings_file["settings"]["generationspeed_set"] self.lowvram_enabled = model_settings_file["settings"]["lowvram_enabled"] self.lowvram_enabled = False if not torch.cuda.is_available() else self.lowvram_enabled self.repetitionpenalty_set = model_settings_file["settings"]["repetitionpenalty_set"] self.temperature_set = model_settings_file["settings"]["temperature_set"] self.pitch_set = model_settings_file["settings"]["pitch_set"] # DO NOT MODIFY - OpenAI voice mappings from model_settings.json self.openai_alloy = model_settings_file["openai_voices"]["alloy"] self.openai_echo = model_settings_file["openai_voices"]["echo"] self.openai_fable = model_settings_file["openai_voices"]["fable"] self.openai_nova = model_settings_file["openai_voices"]["nova"] self.openai_onyx = model_settings_file["openai_voices"]["onyx"] self.openai_shimmer = model_settings_file["openai_voices"]["shimmer"] """ Below is the name of the folder that will be created-used under `/models/{folder}` And is further used by the base functions of this script. Change the name stored in `self.model_folder_name` to the model folder name you will be using. Ensure to use the correct CAPS/Non-CAPS spelling as Linux OS is CAPS specific on folder names. """ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ MODIFY THIS LINE ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ self.model_folder_name = "xtts" # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ MODIFY THIS LINE ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ Add your own central `self.myvariable` variables in here if needed for your engine ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ """ If you need globally accessable variables of your own for your own purposes, you can put them in here as self.myvariable = "whatever". """ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ Add your own central `self.myvariable` variables in here if needed for your engine ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # DO NOT MODIFY - log the function call to this function self.debug_func_entry() ##################################################### # Printout engine loading bits # Do not change this # ##################################################### def printout_versions(self): """ Print Python, DeepSpeed, Pytorch and CUDA version on start-up. WARNING: This is a core system function. Do not modify its implementation as it provides standardized version reporting across all engines. """ self.debug_func_entry() if not model_supports_deepspeed_true_or_false: self.print_message(f"\033[92mDeepSpeed version :\033[93m Not supported on {self.model_folder_name}\033[0m", message_type="standard") else: if deepspeed_available: self.print_message("\033[92mDeepSpeed version :\033[93m " + str(deepspeed.__version__) + "\033[0m", message_type="standard") else: self.print_message("\033[92mDeepSpeed version :\033[93m Not available\033[0m", message_type="standard") self.print_message(f"\033[92mPython Version :\033[93m {python_version}\033[0m", message_type="standard") self.print_message(f"\033[92mPyTorch Version :\033[93m {pytorch_version}\033[0m", message_type="standard") if cuda_version is None: self.print_message("\033[92mCUDA Version :\033[91m Not available\033[0m", message_type="standard") else: self.print_message(f"\033[92mCUDA Version :\033[93m {cuda_version}\033[0m", message_type="standard") self.print_message("", message_type="standard") return ################################################################ # Handle low VRAM change between CUDA/CPU # Do not change this # ################################################################ async def handle_lowvram_change(self): """ Manage model location between CPU and GPU memory for low VRAM operation. This function handles the movement of models between CPU and GPU memory to support systems with limited VRAM. It's called automatically during generation when low VRAM mode is enabled. Operation: 1. Checks CUDA availability 2. Moves model between devices based on current location: - GPU (cuda) -> CPU - CPU -> GPU (cuda) 3. Manages CUDA cache to optimize memory usage States Affected: - self.device: Updated to reflect current processing device - self.model.device: Model's current memory location Requirements: - CUDA must be available for GPU operations - Model must be loaded (self.model is not None) - lowvram_enabled must be `True` in the `model_settings.json` file Note: This function is only called when self.lowvram_enabled is True meaning the engine does or doesnt support the call, hence if its not True, then this function would never be called anyway, so doesnt need changing. """ self.debug_func_entry() # Initial validation if not self.is_tts_model_loaded: self.print_message("No model is currently loaded. Please select a model to load.", message_type="error") raise HTTPException(status_code=400, detail="No model is currently loaded. Please select a model to load.") if torch.cuda.is_available(): if self.device == "cuda": self.print_message("Moving model to CPU", message_type="debug_tts") self.device = "cpu" self.model.to(self.device) torch.cuda.empty_cache() gc.collect() else: self.device = "cuda" self.print_message("Moving model to GPU", message_type="debug_tts") self.model.to(self.device) gc.collect() ################################################ # Handle DeepSpeed change # Do not change this # ################################################ async def handle_deepspeed_change(self, value): """ Handle enabling/disabling of DeepSpeed acceleration. This function manages the process of reloading the model with or without DeepSpeed acceleration. DeepSpeed can significantly improve performance on supported hardware. Args: value (bool): True to enable DeepSpeed, False to disable Operation: 1. Unloads current model 2. Updates DeepSpeed settings 3. Reloads model with new configuration States Affected: - self.deepspeed_enabled: Updated to reflect new state - self.model: Reloaded with new configuration Returns: bool: The new DeepSpeed state (same as input value) Note: DeepSpeed must be installed and available in the system for this functionality to work. `deepspeed_capable` must be set `True` in the `model_settings.json` file """ self.debug_func_entry() # Initial validation if self.current_model_loaded.startswith("apitts"): # Specific only to the XTTS engine self.print_message("\033[93mDeepSpeed not supported in API mode\033[0m", message_type="error") self.deepspeed_enabled = False return False if not self.is_tts_model_loaded: self.print_message("No model is currently loaded. Please select a model to load.", message_type="error") raise HTTPException(status_code=400, detail="No model is currently loaded. Please select a model to load.") if value: self.print_message("\033[93mDeepSpeed Activating\033[0m", message_type="standard") await self.unload_model() self.deepspeed_enabled = True await self.setup() else: self.print_message("\033[93mDeepSpeed De-Activating\033[0m", message_type="standard") self.deepspeed_enabled = False await self.unload_model() await self.setup() return value ################################################################ # Unload models from VRAM/RAM if possible # Do not change this # ################################################################ async def unload_model(self): """ Unload the current model and free associated resources. This function handles the cleanup of model resources, including: 1. Setting model loaded flag to False 2. Deleting the model instance 3. Clearing CUDA cache if available Operation: 1. Updates model loading status 2. Logs unloading process if a model is loaded 3. Removes model from memory 4. Cleans up CUDA cache if using GPU States Affected: - self.is_tts_model_loaded: Set to False - self.model: Set to None after unloading """ self.debug_func_entry() self.is_tts_model_loaded = False if not self.current_model_loaded == None: self.print_message("Unloading model", message_type="debug_tts") if hasattr(self, 'model'): del self.model if torch.cuda.is_available(): torch.cuda.empty_cache() return None ############################################################ # On start-up, perform these actions # Change as necessary # ############################################################ async def setup(self): """ Initialize the XTTS engine and load initial model configuration. This function is called during system startup and handles: 1. Version information display 2. Model scanning and availability check 3. Initial model loading if specified The setup sequence ensures: - Proper version reporting - Model availability verification - Graceful handling of missing models - Correct initial model loading state States Set: - self.available_models: Updated with found models - self.current_model_loaded: Set to loaded model name or None - self.setup_has_run: Set True when complete Returns: None Note: Custom initialization code should be placed between the marked sections. """ self.debug_func_entry() self.print_message("Initializing XTTS engine", message_type="debug_tts") self.printout_versions() self.available_models = self.scan_models_folder() # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ Keep everything above this line ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ if self.selected_model: tts_model = f"{self.selected_model}" if tts_model in self.available_models: self.print_message(f"Loading selected model: {tts_model}", message_type="debug_tts") await self.handle_tts_method_change(tts_model) self.current_model_loaded = tts_model # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ Keep everything below this line ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ else: self.current_model_loaded = "No Models Available" self.print_message(f"Selected model '{self.selected_model}' not found in models folder.", message_type="error") self.print_message(f"Please download a model or select a different model file.", message_type="error") self.setup_has_run = True ########################################################################### # Scan your models folder for models OR voice files # Change as necessary # ########################################################################### def scan_models_folder(self): """ Scan for available XTTS models in the models directory. This function searches the models directory for valid XTTS model installations. Each model must contain all required files to be considered valid. Required Files for Each Model: - config.json: Model configuration - model.pth: Model weights - mel_stats.pth: Mel spectrogram statistics - speakers_xtts.pth: Speaker embeddings - vocab.json: Tokenizer vocabulary - dvae.pth: Discrete VAE weights Operation: 1. Scans the models/xtts directory 2. Checks each subfolder for required files 3. Registers valid models in two formats: - "xtts - {model_name}": For local inference - "apitts - {model_name}": For API-based inference States Affected: - self.available_models: Updated with found models Returns: dict: Dictionary of available models in format: {model_identifier: engine_type} Note: If no valid models are found, returns {"No Models Available": "xtts"} """ self.debug_func_entry() models_folder = self.main_dir / "models" / self.model_folder_name self.available_models = {} # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ Keep everything above this line ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ required_files = ["config.json", "model.pth", "mel_stats.pth", "speakers_xtts.pth", "vocab.json", "dvae.pth"] for subfolder in models_folder.iterdir(): if subfolder.is_dir(): model_name = subfolder.name self.print_message(f"Checking model folder: {model_name}", message_type="debug_tts") if all(subfolder.joinpath(file).exists() for file in required_files): self.print_message(f"Found valid model: {model_name}", message_type="debug_tts") self.available_models[f"xtts - {model_name}"] = "xtts" self.available_models[f"apitts - {model_name}"] = "apitts" # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ Keep everything below this line ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ else: self.available_models = {'No Models Available': self.model_folder_name} self.print_message(f"Model folder '{model_name}' is missing required files", message_type="warning") self.print_message("Required files or folder does not exist", message_type="warning") self.print_message("Please download some models/voices for this engine", message_type="warning") return self.available_models ################################################################ # Scan your voice folder for voice files # Change as necessary # ################################################################ def voices_file_list(self): """ Scan and compile a list of available voice samples and latents. This function scans multiple directories to find voice samples in different formats: 1. Individual WAV files in the main voices directory 2. Collections of WAV files in the xtts_multi_voice_sets directory 3. Pre-computed voice latents in the xtts_latents directory Directory Structure: - voices/: Individual WAV files - voices/xtts_multi_voice_sets/: Folders containing multiple WAV files - voices/xtts_latents/: JSON files containing pre-computed latents Voice Types: - Standard: Direct WAV files - voiceset: Multiple WAV files for one voice - latent: Pre-computed speaker embeddings Returns: list: Available voices with appropriate prefixes: - Standard WAV: filename.wav - Voice sets: "voiceset:foldername" - Latents: "latent:filename.json" Note: Returns ["No Voices Found"] if no valid voices are detected """ self.debug_func_entry() try: voices = [] # An empy variable for the list of voices to be put into. directory = self.main_dir / "voices" # Base directory that voices are stored in. # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ Keep everything above this line ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ json_latents_dir = directory / "xtts_latents" # XTTS specific multi_voice_dir = directory / "xtts_multi_voice_sets" # XTTS specific # Scan for individual WAV files self.print_message("Scanning for individual voice files", message_type="debug_tts") voices.extend([f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f)) and f.endswith(".wav")]) # Scan for voice sets if os.path.exists(multi_voice_dir): self.print_message("Found Multi_Voice_Sets directory", message_type="debug_tts") for voice_set in os.listdir(multi_voice_dir): voice_set_path = multi_voice_dir / voice_set if os.path.isdir(voice_set_path): if any(f.endswith(".wav") for f in os.listdir(voice_set_path)): voices.append(f"voiceset:{voice_set}") self.print_message(f"Added voice set: {voice_set}", message_type="debug_tts_variables") # Scan for JSON latents if not self.current_model_loaded.startswith("apitts"): # APITTS doesnt support latents if os.path.exists(json_latents_dir): self.print_message("Found JSON_Latents directory", message_type="debug_tts") json_files = [f for f in os.listdir(json_latents_dir) if f.endswith('.json')] for json_file in json_files: voices.append(f"latent:{json_file}") if json_files: self.print_message(f"Added {len(json_files)} JSON latent files", message_type="debug_tts") # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ Keep everything below this line ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # Sort voices by type alphabetically voices.sort(key=lambda x: (x.startswith("voiceset:"), x.startswith("latent:"), x)) if not voices: return ["No Voices Found"] return voices except Exception as e: self.print_message(f"Error scanning for voices: {str(e)}", message_type="error") return ["No Voices Found"] ############################################ # Load in your model # Change as necessary # ############################################ async def load_model(self, model_name): """ Load a model using the Coqui TTS API interface. This is one of two model loading methods for XTTS. This method uses the high-level TTS API which provides a simpler interface but less control over model parameters. Args: model_name (str): Name of the model to load Operation: 1. Validates model availability 2. Constructs model and config paths 3. Initializes model using TTS API 4. Moves model to appropriate device (CPU/GPU) States Affected: - self.model: Updated with loaded model - self.is_tts_model_loaded: Set to True on success Returns: The loaded model instance Raises: HTTPException: If no models are available to load """ self.debug_func_entry() if "No Models Available" in self.available_models: self.print_message("No models for this TTS engine were found to load", message_type="error") return model_path = self.main_dir / "models" / self.model_folder_name / model_name # You may need to edit/modfy depending on how your TTS engine works with models self.print_message(f"Loading model from: {model_path}", message_type="debug_tts") # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ Keep everything above this line ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ self.model = TTS( model_path=model_path, config_path=model_path / "config.json", ).to(self.device) self.print_message("\033[94mModel License : \033[93mhttps://coqui.ai/cpml.txt\033[0m") # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ Keep everything below this line ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ self.is_tts_model_loaded = True return self.model ################################################################################################################### # Load in your model # XTTS specific as XTTS has 2x loader types so you wouldnt normally have 2x loader functions # ################################################################################################################### async def xtts_manual_load_model(self, model_name): """ Load a model using the direct XTTS interface. This is the second model loading method for XTTS. It provides direct access to the XTTS model interface, allowing for more detailed control over model parameters and DeepSpeed integration. Args: model_name (str): Name of the model to load Operation: 1. Validates model availability 2. Initializes XttsConfig with model settings 3. Loads model with appropriate configurations 4. Handles DeepSpeed integration if enabled 5. Moves model to appropriate device States Affected: - self.model: Updated with loaded model - self.is_tts_model_loaded: Set to True on success Returns: The loaded model instance Raises: HTTPException: If no models are available to load """ self.debug_func_entry() if "No Models Available" in self.available_models: self.print_message("No models for this TTS engine were found to load", message_type="error") return # Enhanced debugging for paths and configuration config = XttsConfig() model_path = self.main_dir / "models" / self.model_folder_name / model_name config_path = model_path / "config.json" vocab_path_dir = model_path / "vocab.json" checkpoint_dir = model_path self.print_message(f"Model path: {model_path}", message_type="debug_tts_variables") self.print_message(f"Config path: {config_path}", message_type="debug_tts_variables") self.print_message(f"Vocab path: {vocab_path_dir}", message_type="debug_tts_variables") # Load and debug configuration config.load_json(str(config_path)) self.print_message("Model configuration:", message_type="debug_tts_variables") self.print_message(f"├─ Model dimension: {config.model_args.gpt_n_model_channels}", message_type="debug_tts_variables") self.print_message(f"├─ Number of layers: {config.model_args.gpt_layers}", message_type="debug_tts_variables") self.print_message(f"├─ Number of heads: {config.model_args.gpt_n_heads}", message_type="debug_tts_variables") self.print_message(f"├─ Max audio tokens: {config.model_args.gpt_max_audio_tokens}", message_type="debug_tts_variables") self.print_message(f"├─ Max text tokens: {config.model_args.gpt_max_text_tokens}", message_type="debug_tts_variables") self.print_message(f"└─ Using DeepSpeed: {self.deepspeed_enabled}", message_type="debug_tts_variables") # Initialize model self.print_message("Initializing model from config...", message_type="debug_tts") self.model = Xtts.init_from_config(config) # Load checkpoint with detailed progress self.print_message("Loading model checkpoint...", message_type="debug_tts") self.model.load_checkpoint( config, checkpoint_dir=str(checkpoint_dir), vocab_path=str(vocab_path_dir), use_deepspeed=self.deepspeed_enabled, ) # Device management debugging self.print_message(f"Moving model to device: {self.device}", message_type="debug_tts") self.model.to(self.device) if self.device == "cuda": memory_allocated = torch.cuda.memory_allocated() / 1024**2 memory_reserved = torch.cuda.memory_reserved() / 1024**2 self.print_message("CUDA Memory Status:", message_type="debug_tts_variables") self.print_message(f"├─ Allocated: {memory_allocated:.2f} MB", message_type="debug_tts_variables") self.print_message(f"└─ Reserved: {memory_reserved:.2f} MB", message_type="debug_tts_variables") self.is_tts_model_loaded = True self.print_message("\033[94mModel License : \033[93mhttps://coqui.ai/cpml.txt\033[0m") return self.model async def handle_tts_method_change(self, tts_method): """ Handle switching between different XTTS model types and loading methods. This function manages the process of changing between different model loading methods (XTTS local vs API) and handles the actual model loading process. Args: tts_method (str): Format "type - modelname" where type is either "xtts" or "apitts" Operation: 1. Validates model availability 2. Unloads current model if any 3. Parses method string to determine loader type 4. Calls appropriate model loader 5. Updates current model tracking States Affected: - self.current_model_loaded: Updated to new model identifier - self.model: Updated with newly loaded model Returns: bool: True if model loaded successfully, False otherwise Timing: Records and reports model loading time """ self.debug_func_entry() # Track loading time generate_start_time = time.time() # Validate model availability if "No Models Available" in self.available_models: self.print_message("No models for this TTS engine were found to load", message_type="error") return False # Unload current model await self.unload_model() # Handle different loading methods if tts_method.startswith("xtts"): model_name = tts_method.split(" - ")[1] self.print_message(f"\033[94mLoading XTTS model \033[93m{model_name} \033[94mon \033[93m{self.device}\033[0m") self.model = await self.xtts_manual_load_model(model_name) self.current_model_loaded = f"xtts - {model_name}" elif tts_method.startswith("apitts"): model_name = tts_method.split(" - ")[1] self.print_message(f"\033[94mLoading API model \033[93m{model_name} \033[94mon \033[93m{self.device}\033[0m") self.model = await self.load_model(model_name) self.current_model_loaded = f"apitts - {model_name}" else: self.print_message(f"Unknown model type in: {tts_method}", message_type="error") self.current_model_loaded = None return False # Report loading time generate_end_time = time.time() generate_elapsed_time = generate_end_time - generate_start_time self.print_message(f"\033[94mModel Loadtime: \033[93m{generate_elapsed_time:.2f}\033[94m seconds\033[0m") return True async def generate_tts(self, text, voice, language, temperature, repetition_penalty, speed, pitch, output_file, streaming): """ Generate speech from text using the XTTS model. This core function handles all TTS generation, supporting both streaming and non-streaming output, multiple voice input types (WAV, voice sets, latents), and various generation parameters. Args: text (str): Text to convert to speech voice (str): Voice identifier (WAV file, voice set, or latent) language (str): Target language code temperature (float): Generation temperature (0.0-1.0) repetition_penalty (float): Penalty for repetitive generation speed (float): Speech speed multiplier pitch (float): Voice pitch adjustment (not used in XTTS) output_file (str): Path for output audio file streaming (bool): Whether to stream audio chunks Returns: For streaming=True: Generator yielding audio chunks For streaming=False: None (saves to output_file) States Used: - self.model: Active TTS model - self.device: Current processing device - self.lowvram_enabled: Low VRAM mode status - self.current_model_loaded: Current model type """ self.debug_func_entry() # Initial validation if not self.is_tts_model_loaded: self.print_message("No TTS model loaded", message_type="error") raise HTTPException(status_code=400, detail="You currently have no TTS model loaded.") # Lock generation and track start time self.tts_generating_lock = True self.print_message("Starting TTS generation process", message_type="debug_tts") self.print_message(f"Generation parameters: temperature={temperature}, speed={speed}, streaming={streaming}", message_type="debug_tts_variables") # Handle low VRAM mode if needed if self.lowvram_enabled and self.device == "cpu": self.print_message("Low VRAM mode: Moving model to GPU", message_type="debug_tts") await self.handle_lowvram_change() generate_start_time = time.time() try: # Voice input processing self.print_message(f"Processing voice input: {voice}", message_type="debug_tts") gpt_cond_latent = None speaker_embedding = None # Handle different voice types if voice.startswith('latent:'): if self.current_model_loaded.startswith("xtts"): gpt_cond_latent, speaker_embedding = self._load_latents(voice) elif voice.startswith('voiceset:'): voice_set = voice.replace("voiceset:", "") voice_set_path = os.path.join(self.main_dir, "voices", "xtts_multi_voice_sets", voice_set) self.print_message(f"Processing voice set from: {voice_set_path}", message_type="debug_tts") wavs_files = glob.glob(os.path.join(voice_set_path, "*.wav")) if not wavs_files: self.print_message(f"No WAV files found in voice set: {voice_set}", message_type="error") raise HTTPException(status_code=400, detail=f"No WAV files found in voice set: {voice_set}") if len(wavs_files) > 5: wavs_files = random.sample(wavs_files, 5) self.print_message(f"Using 5 random samples from voice set", message_type="debug_tts") if self.current_model_loaded.startswith("xtts"): self.print_message("Generating conditioning latents from voice set", message_type="debug_tts") gpt_cond_latent, speaker_embedding = self._generate_conditioning_latents(wavs_files) else: normalized_path = os.path.normpath(os.path.join(self.main_dir, "voices", voice)) wavs_files = [normalized_path] self.print_message(f"Using single voice sample: {normalized_path}", message_type="debug_tts") if self.current_model_loaded.startswith("xtts"): self.print_message("Generating conditioning latents from single sample", message_type="debug_tts") gpt_cond_latent, speaker_embedding = self._generate_conditioning_latents(wavs_files) # Generate speech if self.current_model_loaded.startswith("xtts"): self.print_message(f"Generating speech for text: {text}", message_type="debug_tts") common_args = { "text": text, "language": language, "gpt_cond_latent": gpt_cond_latent, "speaker_embedding": speaker_embedding, "temperature": float(temperature), "length_penalty": float(self.model.config.length_penalty), "repetition_penalty": float(repetition_penalty), "top_k": int(self.model.config.top_k), "top_p": float(self.model.config.top_p), "speed": float(speed), "enable_text_splitting": True } self.print_message("Generation settings:", message_type="debug_tts_variables") self.print_message(f"├─ Temperature: {temperature}", message_type="debug_tts_variables") self.print_message(f"├─ Speed: {speed}", message_type="debug_tts_variables") self.print_message(f"├─ Language: {language}", message_type="debug_tts_variables") self.print_message(f"└─ Text length: {len(text)} characters", message_type="debug_tts_variables") # Handle streaming vs non-streaming if streaming: self.print_message("Starting streaming generation", message_type="debug_tts") self.print_message(f"Using streaming-based generation and files {wavs_files}") output = self.model.inference_stream(**common_args, stream_chunk_size=20) 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): if self.tts_stop_generation: self.print_message("Generation stopped by user", message_type="debug_tts") self.tts_stop_generation = False self.tts_generating_lock = False break self.print_message(f"Processing chunk {i+1}", message_type="debug_tts") 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: self.print_message("Starting non-streaming generation", message_type="debug_tts") output = self.model.inference(**common_args) torchaudio.save(output_file, torch.tensor(output["wav"]).unsqueeze(0), 24000) self.print_message(f"Saved audio to: {output_file}", message_type="debug_tts") elif self.current_model_loaded.startswith("apitts"): if streaming: raise ValueError("Streaming is only supported in XTTSv2 local mode") # Common arguments for both error and normal cases common_args = { "file_path": output_file, "language": language, "temperature": temperature, "length_penalty": self.model.config.length_penalty, "repetition_penalty": repetition_penalty, "top_k": self.model.config.top_k, "top_p": self.model.config.top_p, "speed": speed } if voice.startswith('latent:'): self.print_message("API TTS method does not support latent files - Please use an audio reference file", message_type="error") self.model.tts_to_file( text="The API TTS method only supports audio files not latents. Please select an audio reference file instead.", speaker="Ana Florence", **common_args ) else: self.print_message("Using API-based generation", message_type="debug_tts") self.model.tts_to_file( text=text, speaker_wav=wavs_files, **common_args ) self.print_message(f"API generation completed, saved to: {output_file}", message_type="debug_tts") finally: # Generation complete generate_end_time = time.time() generate_elapsed_time = generate_end_time - generate_start_time # Standard output message (not debug) self.print_message( f"\033[94mTTS Generate: \033[93m{generate_elapsed_time:.2f} seconds. \033[94mLowVRAM: \033[33m{self.lowvram_enabled} \033[94mDeepSpeed: \033[33m{self.deepspeed_enabled}\033[0m", message_type="standard" ) # Handle low VRAM cleanup if self.lowvram_enabled and self.device == "cuda" and not self.tts_narrator_generatingtts: self.print_message("Low VRAM mode: Moving model back to CPU", message_type="debug_tts") await self.handle_lowvram_change() self.tts_generating_lock = False ############################################################################## # Helper Functions that are specific to this script & not generically needed # ############################################################################## def _generate_conditioning_latents(self, audio_paths): """Generate conditioning latents from audio files.""" self.debug_func_entry() self.print_message(f"Generating latents from {len(audio_paths)} audio files", message_type="debug_tts") return self.model.get_conditioning_latents( audio_path=audio_paths, gpt_cond_len=self.model.config.gpt_cond_len, max_ref_length=self.model.config.max_ref_len, sound_norm_refs=self.model.config.sound_norm_refs, ) def _load_latents(self, voice): """Load speaker latents from JSON file.""" self.debug_func_entry() try: json_file = voice.replace("latent:", "") json_path = os.path.join(self.main_dir, "voices", "xtts_latents", json_file) self.print_message(f"Loading latents from: {json_path}", message_type="debug_tts") with open(json_path) as f: latent_data = json.load(f) gpt_cond_latent = torch.tensor(latent_data['gpt_cond_latent']) speaker_embedding = torch.tensor(latent_data['speaker_embedding']) self.print_message("Successfully loaded speaker latents", message_type="debug_tts") return gpt_cond_latent, speaker_embedding except Exception as e: self.print_message(f"Failed to load speaker latents: {str(e)}", message_type="error") raise HTTPException(status_code=400, detail=f"Failed to load voice latents: {str(e)}")