############################################### # DONT CHANGE # These are base imports needed # ############################################### import os import sys import json import time import torch import logging from pathlib import Path from fastapi import (HTTPException) logging.disable(logging.WARNING) ################################################################# # DONT CHANGE # Get Pytorch & Python versions & setup DeepSpeed # ################################################################# pytorch_version = torch.__version__ cuda_version = torch.version.cuda major, minor, micro = sys.version_info[:3] python_version = f"{major}.{minor}.{micro}" deepspeed_available = False ############################################################################################################# ############################################################################################################# # CHANGE ME # Run any specifc imports, requirements or setup any global vaiables needed for this TTS Engine # ############################################################################################################# ############################################################################################################# # In this section you will import any imports that your specific TTS Engine will use. You will provide any # start-up errors for those bits, as if you were starting up a normal Python script. Note the logging.disable # a few lines up from here, you may want to # that out while debugging! import soundfile as sf import tempfile from pydub import AudioSegment, silence import re import torchaudio import numpy as np import subprocess import sys def install_and_restart(): try: print("#################################################################") print("Installing required packages for F5-TTS... This may take a while.") print("################################################################") # Install vocos first print("Installing vocos...") subprocess.check_call([ sys.executable, "-m", "pip", "install", "vocos" ]) # Then install F5-TTS print("Installing F5-TTS...") subprocess.check_call([ sys.executable, "-m", "pip", "install", "git+https://github.com/SWivid/F5-TTS.git" ]) print("##############################################################") print("All packages installed successfully! Restarting application...") print("##############################################################") # Get the current script's path script_path = sys.argv[0] # Restart the current script os.execv(sys.executable, ['python'] + sys.argv) except subprocess.CalledProcessError as e: print("########################################################") print(f"Failed to install required packages: {str(e)}") print("########################################################") raise ImportError("Could not install required packages") try: from f5_tts.model import CFM, DiT from f5_tts.model.utils import ( get_tokenizer, convert_char_to_pinyin, ) from vocos import Vocos except ImportError: install_and_restart() ################################################################################################################################# # DONT CHANGE # Do not change the Class name from tts_class as this is what will be imported into the main tts_server.py script # ################################################################################################################################# class tts_class: def __init__(self): ######################################################################## # DONT CHANGE # Sets up the base variables required for any tts engine # ######################################################################## self.branding = None self.this_dir = Path(__file__).parent.resolve() # Sets up self.this_dir as a variable for the folder THIS script is running in. self.main_dir = Path(__file__).parent.parent.parent.parent.resolve() # Sets up self.main_dir as a variable for the folder AllTalk is running in self.device = "cuda" if torch.cuda.is_available() else "cpu" # Sets up self.device to cuda if torch exists with Nvidia/CUDA, otherwise sets to cpu self.cuda_is_available = torch.cuda.is_available() # Sets up cuda_is_available as a True/False to track if Nvidia/CUDA was found on the system self.tts_generating_lock = False # Used to lock and unlock the tts generation process at the start/end of tts generation. self.tts_stop_generation = False # Used in conjunction with tts_generating_lock to call for a stop to the current generation. If called (set True) it needs to be set back to False when generation has been stopped. self.tts_narrator_generatingtts = False # Used to track if the current tts processes is narrator based. This can be used in conjunction with lowvram and device to avoid moving model between GPU(CUDA)<>RAM(CPU) each chunk of narrated text generated. self.model = None # If loading a model into CUDA/VRAM/RAM "model" is used as the variable name to load and interact with (see the XTTS model_engine script for examples.) self.is_tts_model_loaded = False # Used to track if a model is actually loaded in and error/fail things like TTS generation if its False self.current_model_loaded = None # Stores the name of the currenly loaded in model self.available_models = None # List of available models found by "def scan_models_folder" self.setup_has_run = False # Tracks if async def setup(self) has run, by setting to True, so that the /api/ready endpoint can provide a "Ready" status ############################################################################################## # DONT CHANGE # Load in a list of the available TTS engines and the currently set TTS engine # ############################################################################################## tts_engines_file = os.path.join(self.main_dir, "system", "tts_engines", "tts_engines.json") with open(tts_engines_file, "r") as f: tts_engines_data = json.load(f) self.engines_available = [engine["name"] for engine in tts_engines_data["engines_available"]] # A list of ALL the TTS engines available to be loaded by AllTalk self.engine_loaded = tts_engines_data["engine_loaded"] # In "tts_engines.json" what is the currently set TTS engine loading into AllTalk self.selected_model = tts_engines_data["selected_model"] # In "tts_engines.json" what is the currently set TTS model loading into AllTalk ############################################################################ # DONT CHANGE # Pull out all the settings for the currently set TTS engine # ############################################################################ with open(os.path.join(self.this_dir, "model_settings.json"), "r") as f: tts_model_loaded = json.load(f) # Access the model details self.manufacturer_name = tts_model_loaded["model_details"]["manufacturer_name"] # The company/person/body that generated the TTS engine/models etc self.manufacturer_website = tts_model_loaded["model_details"]["manufacturer_website"] # The website of the company/person/body where people can find more information # Access the features the model is capable of: self.audio_format = tts_model_loaded["model_capabilties"]["audio_format"] # This details the audio format your TTS engine is set to generate TTS in e.g. wav, mp3, flac, opus, acc, pcm. Please use only 1x format. self.deepspeed_capable = tts_model_loaded["model_capabilties"]["deepspeed_capable"] # Is your model capable of DeepSpeed self.deepspeed_available = 'deepspeed' in globals() # When we did the import earlier, at the top of this script, was DeepSpeed available for use self.generationspeed_capable = tts_model_loaded["model_capabilties"]["generationspeed_capable"] # Does this TTS engine support changing the speed of the generated TTS self.languages_capable = tts_model_loaded["model_capabilties"]["languages_capable"] # Are the actual models themselves capable of generating in multiple languages OR is each model language specific self.lowvram_capable = tts_model_loaded["model_capabilties"]["lowvram_capable"] # Is this engine capable of using low VRAM (moving the model between CPU And GPU memory) self.multimodel_capable = tts_model_loaded["model_capabilties"]["multimodel_capable"] # Is there just the one model or are there multiple models this engine supports. self.repetitionpenalty_capable = tts_model_loaded["model_capabilties"]["repetitionpenalty_capable"] # Is this TTS engine capable of changing the repititon penalty self.streaming_capable = tts_model_loaded["model_capabilties"]["streaming_capable"] # Is this TTS engine capabale of generating streaming audio self.temperature_capable = tts_model_loaded["model_capabilties"]["temperature_capable"] # Is this TTS engine capable of changing the temperature of the models self.multivoice_capable = tts_model_loaded["model_capabilties"]["multivoice_capable"] # Are the models multi-voice or single vocice models self.pitch_capable = tts_model_loaded["model_capabilties"]["pitch_capable"] # Is this TTS engine capable of changing the pitch of the genrated TTS # Access the current enginesettings self.def_character_voice = tts_model_loaded["settings"]["def_character_voice"] # What is the current default main/character voice that will be used if no voice specified. self.def_narrator_voice = tts_model_loaded["settings"]["def_narrator_voice"] # What is the current default narrator voice that will be used if no voice specified. self.deepspeed_enabled = tts_model_loaded["settings"]["deepspeed_enabled"] # If its available, is DeepSpeed enabled for the TTS engine self.engine_installed = tts_model_loaded["settings"]["engine_installed"] # Has the TTS engine been setup/installed (not curently used) self.generationspeed_set = tts_model_loaded["settings"]["generationspeed_set"] # What is the set/stored speed for generation. self.lowvram_enabled = tts_model_loaded["settings"]["lowvram_enabled"] # If its available, is LowVRAM enabled for the TTS engine # Check if someone has enabled lowvram on a system that's not CUDA enabled self.lowvram_enabled = False if not torch.cuda.is_available() else self.lowvram_enabled # If LowVRAM is mistakenly set and CUDA is not available, this will force it back off self.repetitionpenalty_set = tts_model_loaded["settings"]["repetitionpenalty_set"] # What is the currenly set repitition policy of the model (If it support repetition) self.temperature_set = tts_model_loaded["settings"]["temperature_set"] # What is the currenly set temperature of the model (If it support temp) self.pitch_set = tts_model_loaded["settings"]["pitch_set"] # What is the currenly set pitch of the model (If it support temp) # Gather the OpenAI API Voice Mappings self.openai_alloy = tts_model_loaded["openai_voices"]["alloy"] # The TTS engine voice that will be mapped to Open AI Alloy voice self.openai_echo = tts_model_loaded["openai_voices"]["echo"] # The TTS engine voice that will be mapped to Open AI Echo voice self.openai_fable = tts_model_loaded["openai_voices"]["fable"] # The TTS engine voice that will be mapped to Open AI Fable voice self.openai_nova = tts_model_loaded["openai_voices"]["nova"] # The TTS engine voice that will be mapped to Open AI Nova voice self.openai_onyx = tts_model_loaded["openai_voices"]["onyx"] # The TTS engine voice that will be mapped to Open AI Onyx voice self.openai_shimmer = tts_model_loaded["openai_voices"]["shimmer"] # The TTS engine voice that will be mapped to Open AI Shimmer voice ################################################################### # DONT CHANGE # Load params and api_defaults from confignew.json # ################################################################### # Define the path to the confignew.json file configfile_path = self.main_dir / "confignew.json" # Load config file and get settings with open(configfile_path, "r") as configfile: configfile_data = json.load(configfile) self.branding = configfile_data.get("branding", "") # Sets up self.branding for outputting the name stored in the "confgnew.json" file, as used in print statements. self.params = configfile_data # Loads in the curent "confgnew.json" file to self.params. self.debug_tts = configfile_data.get("debugging").get("debug_tts") # Can be used within this script as a True/False flag for generally debugging the TTS generation process. self.debug_tts_variables = configfile_data.get("debugging").get("debug_tts_variables") # Can be used within this script as a True/False flag for generally debugging variables (if you wish). ############################################################################ # DONT CHANGE # These settings are specific to the F5-TTS Model/Engine #### ############################################################################ # Add F5-TTS specific parameters self.target_sample_rate = 24000 self.n_mel_channels = 100 self.hop_length = 256 self.win_length = 1024 self.n_fft = 1024 self.mel_spec_type = 'vocos' self.target_rms = 0.1 self.cross_fade_duration = 0.15 self.ode_method = 'euler' self.nfe_step = 32 self.cfg_strength = 2.0 self.sway_sampling_coef = -1.0 self.speed = 1.0 # F5-TTS model configuration self.f5_model_cfg = dict( dim=1024, depth=22, heads=16, ff_mult=2, text_dim=512, conv_layers=4 ) # Add E2-TTS model configuration self.e2_model_cfg = dict( dim=1024, depth=24, heads=16, ff_mult=4 ) ################################################################ # DONT CHANGE # Print out Python, CUDA, DeepSpeed versions #### ################################################################ def printout_versions(self): if deepspeed_available: print(f"[{self.branding}ENG] \033[92mDeepSpeed version :\033[93m",deepspeed.__version__,"\033[0m") else: print(f"[{self.branding}ENG] \033[92mDeepSpeed version :\033[93m Not available\033[0m") print(f"[{self.branding}ENG] \033[92mPython Version :\033[93m {python_version}\033[0m") print(f"[{self.branding}ENG] \033[92mPyTorch Version :\033[93m {pytorch_version}\033[0m") if cuda_version is None: print(f"[{self.branding}ENG] \033[92mCUDA Version :\033[91m Not available\033[0m") else: print(f"[{self.branding}ENG] \033[92mCUDA Version :\033[93m {cuda_version}\033[0m") print(f"[{self.branding}ENG]") return ################################################################################### ################################################################################### # CHANGE ME # Inital setup of the model and engine. Called when the script starts # ################################################################################### ################################################################################### # In here you will add code to load in your model and do its inital setup. So you # may be calling your model loader via handle_tts_method_change or if your TTS # engine doesnt actually load a model into CUDA or System RAM, you may be doing # Something to fake its start-up. async def setup(self): self.printout_versions() # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ Keep everything above this line ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # Record the start time of loading the model generate_start_time = time.time() # Scan for available models self.available_models = self.scan_models_folder() print("in setup - self.selected_model:", self.selected_model) if self.debug_tts else None # Load the model if one is selected if self.selected_model and self.selected_model != "No Models Found": try: await self.api_manual_load_model(self.selected_model) self.current_model_loaded = self.selected_model except Exception as e: print(f"[{self.branding}ENG] \033[91mError loading model: {str(e)}\033[0m") self.is_tts_model_loaded = False generate_end_time = time.time() generate_elapsed_time = generate_end_time - generate_start_time print(f"[{self.branding}ENG]\033[94m Model/Engine :\033[93m F5-TTS\033[94m Ready\033[0m") print(f"[{self.branding}ENG] \033[94mLoad time :\033[93m {generate_elapsed_time:.2f} seconds.\033[0m") # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ Keep everything below this line ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ self.setup_has_run = True # Flag that setup has run, so the /api/ready endpoint will send a "Ready" status and load the webui ################################## ################################## # CHANGE ME # Low VRAM Swapping # ################################## ################################## # If your model does load into CUDA and you want to support LowVRAM, aka moving the model # on the fly between CUDA and System RAM on each generation request, you will be adding # The code in here to do it. See the XTTS tts engine for an example. Piper however, doesnt # Load into VRAM or System RAM, also low vram is set globally disabled by the model_settings.JSON # file, so this would never get called anywyay, however, we still need to keep an empty # function in place. The "pass" tells the function to just exit out cleanly if called. # However, its quite a simple check along the lines of "if CUDA is available and model is # in X place, then send it to Y place (or Y to X). async def handle_lowvram_change(self): """Handle moving model between CPU and GPU for low VRAM operation""" if not self.lowvram_capable or not self.cuda_is_available: return if not self.is_tts_model_loaded: return try: if self.lowvram_enabled and self.device == 'cuda': # Moving to CPU - everything must be float32 print(f"[{self.branding}Debug] Moving models to CPU") if self.debug_tts else None if hasattr(self, 'model'): self.model = self.model.float().to('cpu') torch.cuda.empty_cache() print(f"[{self.branding}Debug] Model moved to CPU") if self.debug_tts else None if hasattr(self, 'vocoder'): # Move vocoder without checking device try: self.vocoder = self.vocoder.float().to('cpu') torch.cuda.empty_cache() print(f"[{self.branding}Debug] Vocoder moved to CPU") if self.debug_tts else None except Exception as e: print(f"[{self.branding}Debug] Vocoder move warning: {str(e)}") if self.debug_tts else None self.device = 'cpu' print(f"[{self.branding}ENG] Models moved to CPU for low VRAM mode") if self.debug_tts else None elif self.lowvram_enabled and self.device == 'cpu': # Moving to GPU - everything must be float16 print(f"[{self.branding}Debug] Moving models to GPU") if self.debug_tts else None if hasattr(self, 'model'): self.model = self.model.to('cuda').half() print(f"[{self.branding}Debug] Model moved to GPU") if self.debug_tts else None if hasattr(self, 'vocoder'): try: self.vocoder = self.vocoder.to('cuda').half() print(f"[{self.branding}Debug] Vocoder moved to GPU") if self.debug_tts else None except Exception as e: print(f"[{self.branding}Debug] Vocoder move warning: {str(e)}") if self.debug_tts else None self.device = 'cuda' print(f"[{self.branding}ENG] Models moved to GPU") if self.debug_tts else None # Verify movements and precision if self.debug_tts: if hasattr(self, 'model'): # Check model parameters only params_dtype = next(self.model.parameters()).dtype print(f"[{self.branding}Debug] Model parameters dtype: {params_dtype}") print(f"[{self.branding}Debug] Final model device: {next(self.model.parameters()).device}") if hasattr(self, 'vocoder'): try: # Try to check vocoder parameters params = next(self.vocoder.parameters()) print(f"[{self.branding}Debug] Vocoder parameters dtype: {params.dtype}") except Exception: pass # Ignore vocoder parameter checking errors except Exception as e: print(f"[{self.branding}ENG] \033[91mWarning during model movement: {str(e)}\033[0m") if self.lowvram_enabled: self.device = 'cpu' else: self.device = 'cuda' if torch.cuda.is_available() else 'cpu' ######################################## ######################################## # CHANGE ME # DeepSpeed model loading # ######################################## ######################################## # If the model supports CUDA and DeepSpeed, this is where you can handle re-loading # the model as/when DeepSpeed is enabled/selected in the user interface. If your # TTS model doesnt support DeepSpeed, then it should be globally set in your # model_settings.JSON and this function will never get called, however it still needs # to exist as a function. async def handle_deepspeed_change(self, value): """F5-TTS currently doesn't support DeepSpeed""" print(f"[{self.branding}ENG] DeepSpeed not supported for F5-TTS") return False ############################################################################################################################################## ############################################################################################################################################## # CHANGE ME # scan for available models/voices that are relevant to this TTS engine # XTTS is very unusal as it has 2x model loading methods # ############################################################################################################################################## ############################################################################################################################################## # This function looks and reports back the list of possible models your TTS engine can # load in. Some TTS engines have multiple models they can load and you will want to use # code for checking if the models are in the correct location/placement within the disk # the correct files exst per model etc (check XTTS for an example of this). Some models # like Piper, the actual models are the voices, so in the Piper scan_models_folder # function, we fake the only model being available as {'piper': 'piper'} aka, model name # then engine name, then we use the voices_file_list to populate the models as available # voices that can be selected in the interface. # If no models are found, we return "No Models Available" and continue on with the script. def scan_models_folder(self): # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ Keep everything above this line ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ """Scan for available F5-TTS models""" self.available_models = {} models_dir = self.main_dir / "models" / "f5tts" if not models_dir.exists(): print(f"[{self.branding}ENG] Models directory not found: {models_dir}") self.available_models["No Models Found"] = "No Models Found" return self.available_models # Look for model directories that match the pattern f5tts_v* for model_dir in models_dir.glob("*tts_v*"): if model_dir.is_dir(): # First try to find model_*.safetensors files model_files = list(model_dir.glob("model_*.safetensors")) if not model_files: # If no model_*.safetensors found, try any .safetensors file model_files = list(model_dir.glob("*.safetensors")) vocab_file = model_dir / "vocab.txt" vocos_dir = model_dir / "vocos" vocos_config = vocos_dir / "config.yaml" vocos_model = vocos_dir / "pytorch_model.bin" # Check if we have at least one model file and all other required files if model_files and all(f.exists() for f in [vocab_file, vocos_config, vocos_model]): model_name = model_dir.name self.available_models[model_name] = model_name # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ Keep everything below this line ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ if not self.available_models: self.available_models["No Models Found"] = "No Models Found" # Return a list with {'No Models Found'} if there are no models found. return self.available_models # Return a list of models in the format {'engine name': 'model 1', 'engine name': 'model 2", etc} ############################################################# ############################################################# # CHANGE ME # POPULATE FILES LIST FROM VOICES DIRECTORY #### ############################################################# ############################################################# # This function looks and reports back the list of possible voice your TTS engine can # load in. Some TTS engines the voices are wav file samples (XTTS), some are models # (Piper) and some are text (Parler) thats stored in a JSON file. We just need to # populate the "voices" variable somehow and if no voices are found, we return # "No Voices Found" back to the interface/api. def voices_file_list(self): try: voices = [] # Function to check if a wav file has a corresponding reference text file def has_reference_text(wav_path): text_path = wav_path.with_suffix('.reference.txt') return text_path.exists() directory = self.main_dir / "voices" # Step 1: Add .wav files in the main "voices" directory to the list (only if they have matching .reference.txt) for f in directory.glob("*.wav"): if has_reference_text(f): voices.append(f.name) else: print(f"[{self.branding}ENG] Warning: {f.name} does not have a matching reference text file") if self.debug_tts else None # Step 2: Walk through subfolders and add subfolder names if they contain valid wav+reference.txt pairs for folder in directory.iterdir(): if folder.is_dir(): valid_pairs = False for wav_file in folder.glob("*.wav"): if has_reference_text(wav_file): valid_pairs = True break if valid_pairs: folder_name = folder.name + "/" voices.append(folder_name) elif self.debug_tts: print(f"[{self.branding}ENG] Warning: Folder {folder.name} has no valid wav+reference.txt pairs") # Remove "voices/" from the list if it somehow got added voices = [v for v in voices if v != "voices/"] if not voices: return ["No Voices Found"] return voices except Exception as e: print(f"[{self.branding}ENG] \033[91mError\033[0m: Voices/Voice Models not found. Cannot load a list of voices.") print(f"[{self.branding}ENG]") return ["No Voices Found"] ################################################################################# ################################################################################# # CHANGE ME # Model loading ##################################################### ################################################################################# ################################################################################# # This function will handle the loading of your model, into VRAM/CUDA, System RAM or whatever. async def api_manual_load_model(self, model_name): if model_name == "No Models Found": print(f"[{self.branding}ENG] \033[91mError\033[0m: No models for this TTS engine were found to load.") raise HTTPException(status_code=400, detail="No models for this TTS engine were found to load.") print(f"[{self.branding}ENG]\033[94m Model/Engine :\033[93m {model_name}\033[94m loading into\033[93m", self.device,"\033[0m") # Split the engine name from model name and get just the model folder name model_folder = model_name.split(" - ")[-1] try: # Set up paths using the correct model folder name model_dir = self.main_dir / "models" / "f5tts" / model_folder vocab_path = model_dir / "vocab.txt" vocos_path = model_dir / "vocos" # Dynamically find the safetensors model file model_files = list(model_dir.glob("model_*.safetensors")) if not model_files: # Try finding any safetensors file as fallback model_files = list(model_dir.glob("*.safetensors")) if not model_files: raise FileNotFoundError("No safetensors model file found in the model directory") # Use the first found model file model_path = model_files[0] print(f"[{self.branding}ENG] Loading model from path: {model_path}") if self.debug_tts else None # Initialize vocoder - always in float32 initially self.vocoder = Vocos.from_hparams(str(vocos_path / "config.yaml")) vocoder_state = torch.load(str(vocos_path / "pytorch_model.bin"), map_location="cpu") self.vocoder.load_state_dict(vocoder_state) self.vocoder = self.vocoder.eval() # Determine if this is an E2-TTS model from the folder name is_e2_model = "e2tts" in model_folder.lower() # Initialize model with appropriate architecture vocab_char_map, vocab_size = get_tokenizer(str(vocab_path), "custom") mel_spec_kwargs = dict( n_fft=self.n_fft, hop_length=self.hop_length, win_length=self.win_length, n_mel_channels=self.n_mel_channels, target_sample_rate=self.target_sample_rate, mel_spec_type=self.mel_spec_type, ) if is_e2_model: print(f"[{self.branding}ENG] Initializing E2-TTS model") if self.debug_tts else None self.model = CFM( transformer=UNetT( **self.e2_model_cfg, text_num_embeds=vocab_size, mel_dim=self.n_mel_channels ), mel_spec_kwargs=mel_spec_kwargs, odeint_kwargs=dict( method=self.ode_method, ), vocab_char_map=vocab_char_map ).float() # Ensure float32 else: print(f"[{self.branding}ENG] Initializing F5-TTS model") if self.debug_tts else None self.model = CFM( transformer=DiT( **self.f5_model_cfg, text_num_embeds=vocab_size, mel_dim=self.n_mel_channels ), mel_spec_kwargs=mel_spec_kwargs, odeint_kwargs=dict( method=self.ode_method, ), vocab_char_map=vocab_char_map ).float() # Ensure float32 # Load model weights if str(model_path).endswith('.safetensors'): from safetensors.torch import load_file checkpoint = load_file(model_path) checkpoint = {"ema_model_state_dict": checkpoint} else: checkpoint = torch.load(model_path, weights_only=True) # Handle backward compatibility before loading weights model_state_dict = { k.replace("ema_model.", ""): v for k, v in checkpoint["ema_model_state_dict"].items() if k not in ["initted", "step"] } # Remove backward compatibility keys before loading backward_compat_keys = [ "mel_spec.mel_stft.mel_scale.fb", "mel_spec.mel_stft.spectrogram.window" ] for key in backward_compat_keys: if key in model_state_dict: del model_state_dict[key] self.model.load_state_dict(model_state_dict) self.model.eval() # Move to device and handle precision if self.device == "cuda": self.model = self.model.to(self.device).half() self.vocoder = self.vocoder.to(self.device).half() # Ensure all parts are in half precision for module in [self.model, self.vocoder]: for param in module.parameters(): param.data = param.data.half() for buffer in module.buffers(): if buffer.dtype == torch.float32: buffer.data = buffer.data.half() else: self.model = self.model.to(self.device) self.vocoder = self.vocoder.to(self.device) self.is_tts_model_loaded = True model_type = "E2-TTS" if is_e2_model else "F5-TTS" print(f"[{self.branding}ENG] {model_type} model loaded successfully") if self.debug_tts else None except Exception as e: print(f"[{self.branding}ENG] \033[91mError loading model: {str(e)}\033[0m") raise HTTPException(status_code=500, detail=f"Error loading model: {str(e)}") ############################### ############################### # CHANGE ME # Model unloading # ############################### ############################### # This function will handle the UN-loading of your model, from VRAM/CUDA, System RAM or whatever. async def unload_model(self): # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ Keep everything above this line ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ """Unload the F5-TTS model from memory""" if hasattr(self, 'model'): del self.model self.model = None if hasattr(self, 'vocoder'): del self.vocoder self.vocoder = None if torch.cuda.is_available(): torch.cuda.empty_cache() # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ Keep everything below this line ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ self.is_tts_model_loaded = False return None ################################################################################################################################### ################################################################################################################################### # CHANGE ME # Model changing. Unload out old model and load in a new one ########################################################## ################################################################################################################################### ################################################################################################################################### # This function is your central model loading/unloading handler that deals with the above functions as necesary, to call loading, unloading, # swappng DeepSpeed, Low vram etc. This function gets called with a "engine name - model name" type call. async def handle_tts_method_change(self, tts_method): generate_start_time = time.time() # Record the start time of loading the model if "No Models Available" in self.available_models: print(f"[{self.branding}ENG] \033[91mError\033[0m: No models for this TTS engine were found to load. Please download a model.") return False # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # ↑↑↑ Keep everything above this line ↑↑↑ # ↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑↑ # Unload current model if one is loaded if self.is_tts_model_loaded: await self.unload_model() # Load the new model try: await self.api_manual_load_model(tts_method) self.current_model_loaded = tts_method except Exception as e: print(f"[{self.branding}ENG] \033[91mError loading model: {str(e)}\033[0m") return False # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ # ↓↓↓ Keep everything below this line ↓↓↓ # ↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓↓ generate_end_time = time.time() # Create an end timer for calculating load times generate_elapsed_time = generate_end_time - generate_start_time # Calculate start time minus end time print(f"[{self.branding}ENG] \033[94mLoad time :\033[93m {generate_elapsed_time:.2f} seconds.\033[0m") # Print out the result of the load time return True async def preprocess_ref_audio_text(self, ref_audio_orig, ref_text, clip_short=True): print(f"[{self.branding}ENG] Converting audio...") if self.debug_tts else None with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as f: aseg = AudioSegment.from_file(ref_audio_orig) if clip_short: # 1. try to find long silence for clipping non_silent_segs = silence.split_on_silence( aseg, min_silence_len=1000, silence_thresh=-50, keep_silence=1000 ) non_silent_wave = AudioSegment.silent(duration=0) for non_silent_seg in non_silent_segs: if len(non_silent_wave) > 6000 and len(non_silent_wave + non_silent_seg) > 15000: print(f"[{self.branding}ENG] Audio is over 15s, clipping short. (1)") if self.debug_tts else None break non_silent_wave += non_silent_seg # 2. try to find short silence for clipping if 1. failed if len(non_silent_wave) > 15000: non_silent_segs = silence.split_on_silence( aseg, min_silence_len=100, silence_thresh=-40, keep_silence=1000 ) non_silent_wave = AudioSegment.silent(duration=0) for non_silent_seg in non_silent_segs: if len(non_silent_wave) > 6000 and len(non_silent_wave + non_silent_seg) > 15000: print(f"[{self.branding}ENG] Audio is over 15s, clipping short. (2)") if self.debug_tts else None break non_silent_wave += non_silent_seg aseg = non_silent_wave # 3. if no proper silence found for clipping if len(aseg) > 15000: aseg = aseg[:15000] print(f"[{self.branding}ENG] Audio is over 15s, clipping short. (3)") if self.debug_tts else None aseg.export(f.name, format="wav") ref_audio = f.name # Ensure ref_text ends with a proper sentence-ending punctuation if not ref_text.endswith(". ") and not ref_text.endswith("。"): if ref_text.endswith("."): ref_text += " " else: ref_text += ". " return ref_audio, ref_text async def infer_process( self, ref_audio, ref_text, gen_text, model_obj, vocoder, target_rms=0.1, cross_fade_duration=0.15, nfe_step=32, cfg_strength=2.0, sway_sampling_coef=-1, speed=1, fix_duration=None, device=None ): """Process text and prepare for batch inference""" # Split the input text into batches audio, sr = torchaudio.load(ref_audio) max_chars = int(len(ref_text.encode("utf-8")) / (audio.shape[-1] / sr) * (25 - audio.shape[-1] / sr)) gen_text_batches = self.chunk_text(gen_text, max_chars=max_chars) for i, gen_text_batch in enumerate(gen_text_batches): print(f"[{self.branding}ENG] gen_text {i}", gen_text_batch) if self.debug_tts else None print(f"[{self.branding}ENG] Generating audio in {len(gen_text_batches)} batches...") if self.debug_tts else None return await self.infer_batch_process( (audio, sr), ref_text, gen_text_batches, model_obj, vocoder, target_rms=target_rms, cross_fade_duration=cross_fade_duration, nfe_step=nfe_step, cfg_strength=cfg_strength, sway_sampling_coef=sway_sampling_coef, speed=speed, fix_duration=fix_duration, device=device ) async def infer_batch_process( self, ref_audio, ref_text, gen_text_batches, model_obj, vocoder, target_rms=0.1, cross_fade_duration=0.15, nfe_step=32, cfg_strength=2.0, sway_sampling_coef=-1, speed=1, fix_duration=None, device=None ): """Process batches for inference""" device = device or self.device audio, sr = ref_audio # Always process initial audio in float32 if audio.shape[0] > 1: audio = torch.mean(audio, dim=0, keepdim=True) audio = audio.to(torch.float32) # Ensure float32 for initial processing # Normalize audio if needed rms = torch.sqrt(torch.mean(torch.square(audio))) if rms < target_rms: audio = audio * target_rms / rms # Resample if needed if sr != self.target_sample_rate: resampler = torchaudio.transforms.Resample(sr, self.target_sample_rate) audio = resampler(audio) # Move to device and adjust precision audio = audio.to(device) if device == 'cuda': audio = audio.half() generated_waves = [] spectrograms = [] if len(ref_text[-1].encode("utf-8")) == 1: ref_text = ref_text + " " for gen_text in gen_text_batches: # Prepare the text text_list = [ref_text + gen_text] final_text_list = convert_char_to_pinyin(text_list) ref_audio_len = audio.shape[-1] // self.hop_length if fix_duration is not None: duration = int(fix_duration * self.target_sample_rate / self.hop_length) else: # Calculate duration ref_text_len = len(ref_text.encode("utf-8")) gen_text_len = len(gen_text.encode("utf-8")) duration = ref_audio_len + int(ref_audio_len / ref_text_len * gen_text_len / speed) # inference with torch.inference_mode(): generated, _ = model_obj.sample( cond=audio, text=final_text_list, duration=duration, steps=nfe_step, cfg_strength=cfg_strength, sway_sampling_coef=sway_sampling_coef, ) # Handle precision for post-processing if self.device == 'cuda': generated = generated.half() else: generated = generated.float() generated = generated[:, ref_audio_len:, :] generated_mel_spec = generated.permute(0, 2, 1) # Convert to the correct precision for vocoder if self.device == 'cuda': generated_mel_spec = generated_mel_spec.half() else: generated_mel_spec = generated_mel_spec.float() generated_wave = vocoder.decode(generated_mel_spec) if rms < target_rms: generated_wave = generated_wave * rms / target_rms # wav -> numpy generated_wave = generated_wave.squeeze().cpu().float().numpy() # Always convert to float32 for numpy generated_waves.append(generated_wave) spectrograms.append(generated_mel_spec[0].cpu().numpy()) # Combine all generated waves with cross-fading if cross_fade_duration <= 0: final_wave = np.concatenate(generated_waves) else: final_wave = generated_waves[0] for i in range(1, len(generated_waves)): prev_wave = final_wave next_wave = generated_waves[i] # Calculate cross-fade samples cross_fade_samples = int(cross_fade_duration * self.target_sample_rate) cross_fade_samples = min(cross_fade_samples, len(prev_wave), len(next_wave)) if cross_fade_samples <= 0: final_wave = np.concatenate([prev_wave, next_wave]) continue # Overlapping parts prev_overlap = prev_wave[-cross_fade_samples:] next_overlap = next_wave[:cross_fade_samples] # Fade out and fade in fade_out = np.linspace(1, 0, cross_fade_samples) fade_in = np.linspace(0, 1, cross_fade_samples) # Cross-faded overlap cross_faded_overlap = prev_overlap * fade_out + next_overlap * fade_in # Combine new_wave = np.concatenate( [prev_wave[:-cross_fade_samples], cross_faded_overlap, next_wave[cross_fade_samples:]] ) final_wave = new_wave # Create a combined spectrogram combined_spectrogram = np.concatenate(spectrograms, axis=1) return final_wave, self.target_sample_rate, combined_spectrogram def chunk_text(self, text, max_chars=135): """ Splits the input text into chunks, each with a maximum number of characters. """ chunks = [] current_chunk = "" # Split the text into sentences based on punctuation followed by whitespace sentences = re.split(r"(?<=[;:,.!?])\s+|(?<=[；：，。！？])", text) for sentence in sentences: if len(current_chunk.encode("utf-8")) + len(sentence.encode("utf-8")) <= max_chars: current_chunk += sentence + " " if sentence and len(sentence[-1].encode("utf-8")) == 1 else sentence else: if current_chunk: chunks.append(current_chunk.strip()) current_chunk = sentence + " " if sentence and len(sentence[-1].encode("utf-8")) == 1 else sentence if current_chunk: chunks.append(current_chunk.strip()) return chunks ############################################################################################################################# ############################################################################################################################# # DONT CHANGE ME # De-Capitalise words (Not used currently) ################################################################# ############################################################################################################################# ############################################################################################################################# def process_text_for_tts(self, text): """ Process text by converting fully capitalized words to lowercase, while preserving other words' capitalization. Args: self: The class instance text (str): Input text string Returns: str: Processed text with fully capitalized words converted to lowercase """ # Split text into words words = text.split() # Process each word processed_words = [] for word in words: # Check if word is fully uppercase and longer than 1 character if word.isupper() and len(word) > 1: processed_words.append(word.lower()) else: processed_words.append(word) # Join words back together return ' '.join(processed_words) ########################################################################################################################################## ########################################################################################################################################## # CHANGE ME # Model changing. Unload out old model and load in a new one ################################################################# ########################################################################################################################################## ########################################################################################################################################## # In here all the possible options are sent over (text, voice to use, lanugage, speed etc etc) and its up to you how you use them, or not. # obviously if your TTS engine doesnt support speed for example, generationspeed_capable should be set False in your model_settings.JSON file # and a fake "generationspeed_set" value should be set. This allows a fake value to be sent over from the main script, even though it # wouldnt actually ever be used in the generation below. Nonethless all these values, real or just whats inside the configuration file # will be sent over for use. # Setting the xxxxxxx_capabale in the model_settings.JSON file, will enable/disable them being selectable by the user. For example, if you # set "generationspeed_capable" as false in the model_settings.JSON file, a user will not be able to select OR set the setting for # generation speed. # One thing to note is that we HAVE to have something in this generation request that is synchronous from the way its called, which means # we have to have an option for Streaming, even if our TTS engine doesnt support streaming. So in that case, we would set streaming_capable # as false in our model_settings.JSON file, meaning streaming will never be called. However, we have to put a fake streaming routine in our # function below (or a real function if it does support streaming of course). Parler has an example of a fake streaming function, which is # very clearly highlighted in its model_engine.py script. async def generate_tts(self, text, voice, language, temperature, repetition_penalty, speed, pitch, output_file, streaming): if voice == "No Voices Found": print(f"[{self.branding}ENG] \033[91mError\033[0m: No voices found to generate TTS.") raise HTTPException(status_code=400, detail="No voices found to generate TTS.") if not self.is_tts_model_loaded: print(f"[{self.branding}ENG] \033[91mError\033[0m: You currently have no TTS model loaded.") raise HTTPException(status_code=400, detail="You currently have no TTS model loaded.") # Language validation - converted to lowercase for case-insensitive comparison language = language.lower() if language not in ['en', 'eng', 'zh', 'zho', 'chi']: print(f"[{self.branding}ENG] \033[91mError\033[0m: F5-TTS only supports English (EN) or Chinese (ZH) language TTS generation.") print(f"[{self.branding}ENG] \033[91mError\033[0m: You may get an error message or gibberish output.") print(f"[{self.branding}ENG] \033[91mError\033[0m: Please use only EN or ZH/ZHO for the language with F5-TTS.") self.tts_generating_lock = True generate_start_time = time.time() try: if self.lowvram_enabled and self.device == "cpu": print(f"[{self.branding}Debug] Moving models to GPU for generation") if self.debug_tts else None await self.handle_lowvram_change() # Verify precision is consistent if self.debug_tts: print(f"[{self.branding}Debug] Verifying model precision...") if hasattr(self, 'model'): params_dtype = {p.dtype for p in self.model.parameters()} buffers_dtype = {b.dtype for b in self.model.buffers()} if len(params_dtype) > 1 or len(buffers_dtype) > 1: print(f"[{self.branding}ENG] Warning: Mixed precision detected in model") print(f"[{self.branding}Debug] Model parameters dtypes: {params_dtype}") print(f"[{self.branding}Debug] Model buffers dtypes: {buffers_dtype}") if hasattr(self, 'vocoder'): params_dtype = {p.dtype for p in self.vocoder.parameters()} buffers_dtype = {b.dtype for b in self.vocoder.buffers()} if len(params_dtype) > 1 or len(buffers_dtype) > 1: print(f"[{self.branding}ENG] Warning: Mixed precision detected in vocoder") print(f"[{self.branding}Debug] Vocoder parameters dtypes: {params_dtype}") print(f"[{self.branding}Debug] Vocoder buffers dtypes: {buffers_dtype}") # Handle voice folder or direct file if voice.endswith('/'): # It's a folder, pick the first valid wav+reference.txt pair voice_dir = self.main_dir / "voices" / voice.rstrip('/') for wav_file in voice_dir.glob("*.wav"): ref_text_path = wav_file.with_suffix('.reference.txt') if ref_text_path.exists(): ref_audio_path = wav_file break else: # It's a direct file ref_audio_path = self.main_dir / "voices" / voice # Get the corresponding reference text file ref_text_path = ref_audio_path.with_suffix('.reference.txt') if not ref_text_path.exists(): raise FileNotFoundError(f"Reference text file not found for {voice}") with open(ref_text_path, 'r', encoding='utf-8') as f: ref_text = f.read().strip() # Process reference audio and text ref_audio, processed_ref_text = await self.preprocess_ref_audio_text( str(ref_audio_path), ref_text ) text = self.process_text_for_tts(text) # Generate the audio using infer_process final_wave, final_sample_rate, _ = await self.infer_process( ref_audio, processed_ref_text, text, self.model, self.vocoder, target_rms=self.target_rms, cross_fade_duration=self.cross_fade_duration, nfe_step=self.nfe_step, cfg_strength=self.cfg_strength, sway_sampling_coef=self.sway_sampling_coef, speed=float(speed), fix_duration=None, device=self.device ) # Save the audio sf.write(output_file, final_wave, final_sample_rate) generate_end_time = time.time() generate_elapsed_time = generate_end_time - generate_start_time print(f"[{self.branding}GEN] \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") if streaming: with open(output_file, 'rb') as f: yield f.read() except FileNotFoundError as e: print(f"[{self.branding}ENG] \033[91mError: {str(e)}\033[0m") raise HTTPException(status_code=404, detail=str(e)) except Exception as e: print(f"[{self.branding}ENG] \033[91mError during TTS generation: {str(e)}\033[0m") raise HTTPException(status_code=500, detail=f"Error during TTS generation: {str(e)}") finally: try: # Handle low VRAM mode - move back to CPU if needed if self.lowvram_enabled and not self.tts_narrator_generatingtts: print(f"[{self.branding}Debug] Moving models back to CPU after generation") if self.debug_tts else None await self.handle_lowvram_change() # Verify final device state in debug mode using parameter checking if self.debug_tts: if hasattr(self, 'model'): model_device = next(self.model.parameters()).device print(f"[{self.branding}Debug] Final model device: {model_device}") if self.debug_tts else None if hasattr(self, 'vocoder'): try: vocoder_device = next(self.vocoder.parameters()).device print(f"[{self.branding}Debug] Final vocoder device: {vocoder_device}") if self.debug_tts else None except Exception: # Skip vocoder device checking if it fails pass except Exception as e: print(f"[{self.branding}Debug] Device verification warning: {str(e)}") if self.debug_tts else None self.tts_generating_lock = False