1200 lines
60 KiB
Python
1200 lines
60 KiB
Python
"""
|
|
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(str(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)}")
|