Files
alltalk_tts/finetune.py
T

4810 lines
190 KiB
Python

"""
XTTS finetune module for training and customizing text-to-speech models.
Provides functionality for dataset creation, model training, and inference.
"""
# Standard Library Imports
import argparse
import datetime
import gc
import glob
import logging
import math
import os
import re
import shutil
import signal
import string
import subprocess
import sys
import tempfile
import traceback
from pathlib import Path
# Third-Party Imports
import warnings
from importlib import metadata
import gradio as gr
import pandas as pd
import psutil
import torchaudio
import torchaudio.transforms as T
import torch
from packaging import version
from tqdm import tqdm
from tokenizers import ByteLevelBPETokenizer
from tokenizers.pre_tokenizers import Whitespace
from word2number import w2n
from pynvml import nvmlInit, nvmlDeviceGetHandleByIndex, nvmlDeviceGetMemoryInfo
# TTS Package Imports
from TTS.config.shared_configs import BaseDatasetConfig
from TTS.tts.configs.xtts_config import XttsConfig
from TTS.tts.datasets import load_tts_samples
from TTS.tts.models.xtts import Xtts
from TTS.tts.layers.xtts.trainer.gpt_trainer import (
GPTArgs,
GPTTrainer,
GPTTrainerConfig,
XttsAudioConfig,
)
# Trainer Imports
from trainer_alltalk.trainer import TrainerArgs, Trainer
# Local Module Imports
from trainer_alltalk.metrics_logger import MetricsLogger
from system.ft_tokenizer.tokenizer import multilingual_cleaners
# Help documentation
from trainer_alltalk.finetune_content import FinetuneContent
# Suppress Warnings
warnings.filterwarnings(
"ignore",
message="1Torch was not compiled with flash attention")
warnings.filterwarnings(
"ignore",
message="Failed to launch Triton kernels, likely due to missing CUDA toolkit")
# Disable Gradio Analytics
os.environ["GRADIO_ANALYTICS_ENABLED"] = "False"
# Try to Import Whisper, Install if Not Found
try:
import whisper
except ImportError:
print("[FINETUNE] OpenAI Whisper not found. Attempting to install...")
try:
subprocess.check_call(
[sys.executable, "-m", "pip", "install", "openai-whisper"])
import whisper
print("[FINETUNE] Successfully installed OpenAI Whisper! Continuing.")
except Exception as e:
print("[FINETUNE] Failed to install OpenAI Whisper:")
print(f"[FINETUNE] Error: {str(e)}")
print(
"[FINETUNE] Please try manually installing with: pip install openai-whisper")
sys.exit(1)
# STARTUP VARIABLES
# Paths and Directories
this_dir = Path(__file__).parent.resolve()
audio_folder = this_dir / "finetune" / "put-voice-samples-in-here"
default_path = this_dir / "finetune" / "tmp-trn"
base_path = this_dir / "models" / "xtts"
# Gradio Configuration
theme = gr.themes.Default()
gradio_temp_dir = this_dir / "finetune" / "gradio_temp"
os.environ["GRADIO_TEMP_DIR"] = str(gradio_temp_dir)
# Environment Variables
os.environ["TRAINER_TELEMETRY"] = "0"
out_path = default_path
# Progress Tracking
progress = 0
# Validation Globals (Set to None initially)
VALIDATE_TRAIN_METADATA_PATH = None
VALIDATE_EVAL_METADATA_PATH = None
VALIDATE_AUDIO_FOLDER = None
VALIDATE_WHISPER_MODEL = None
VALIDATE_TARGET_LANGUAGE = None
XTTS_MODEL = None
#################################
#### Define the Logger class ####
#################################
class Logger:
"""
Singleton class to handle logging output to both the terminal and a log file.
"""
_instance = None
def __new__(cls, *args, **kwargs):
"""
Ensure a single instance of the Logger class is created.
Initializes the log file and other necessary attributes.
"""
if cls._instance is None:
cls._instance = super(Logger, cls).__new__(cls)
cls._instance.log_file = "finetune.log"
cls._instance.terminal = sys.stdout
cls._instance.current_model_path = None # To store current training path
# Open in append mode
cls._instance.log = open(cls._instance.log_file, "a", encoding="utf-8")
return cls._instance
def __init__(self, *args, **kwargs):
"""Initialize logger instance."""
pass
def set_model_path(self, path):
"""
Set the current model training path for logging context.
:param path: Path to the current model directory.
"""
self.current_model_path = path
def write(self, message):
"""
Write a message to both the terminal and the log file.
Filters out non-printable characters.
:param message: The message to be logged.
"""
filtered_message = ''.join(char for char in message
if char.isprintable() or char in '\n\r\t')
self.terminal.write(filtered_message)
try:
self.log.write(filtered_message)
self.log.flush()
except:
pass
def flush(self):
"""
Flush any buffered log content to the log file and terminal.
"""
self.terminal.flush()
try:
self.log.flush()
except:
pass
def isatty(self):
"""
Mimic the isatty method to comply with terminal-like behavior.
:return: Always returns False.
"""
return False
def clear_log(self):
"""
Delete the existing log file and recreate it to clear its contents.
"""
try:
# Close the current append-only file handle
self.log.close()
# Delete the log file
os.remove(self.log_file)
# Reopen the file in append mode for further logging
self.log = open(self.log_file, "a", encoding="utf-8")
except Exception as e:
print(f"Failed to delete and recreate log file: {e}")
_logging_setup_done = False
def setup_logging():
global _logging_setup_done
if _logging_setup_done:
return
# redirect stdout and stderr to a file
sys.stdout = Logger()
sys.stderr = sys.stdout
logging.basicConfig(
level=logging.INFO,
format="[FINETUNE] %(message)s",
handlers=[
logging.StreamHandler(sys.stdout)
]
)
_logging_setup_done = True
# Call setup at module level
setup_logging()
c_logger = MetricsLogger()
def load_metrics():
return c_logger.plot_metrics(), f"Running Time: {c_logger.format_duration(c_logger.total_duration)} - Estimated Completion: {c_logger.format_duration(c_logger.estimated_duration)}"
def read_logs():
sys.stdout.flush()
with open(sys.stdout.log_file, "r", encoding="utf-8") as f:
content = f.read()
# Additional filtering when reading the file
return ''.join(char for char in content
if char.isprintable() or char in '\n\r\t')
##############################
#### Debugging management ####
##############################
class DebugLevels:
"""
Defines debug levels for Stage 1 Dataset Creation
Each debug level is a boolean flag that controls whether debug information
for a specific area of functionality is logged.
"""
GPU_MEMORY = False # GPU memory and CUDA related debugging
MODEL_OPS = True # Model loading, transcription, cleanup operations
DATA_PROCESS = False # Data processing, words, sentences
GENERAL = True # General flow, file operations, metadata
AUDIO = False # Audio processing statistics and info
SEGMENTS = False # Detailed segment information
DUPLICATES = False # Duplicate handling information
VALIDATION = False # DataSet validation handling
def debug_print(
message,
level,
is_error=False,
is_warning=False,
is_info=False):
"""Enhanced debug printing with categorization and formatting"""
prefix = "[FINETUNE]"
if is_error:
prefix += " ERROR:"
elif is_warning:
prefix += " WARNING:"
elif is_info:
prefix += ""
if level == "GPU_MEMORY" and (DebugLevels.GPU_MEMORY or is_error):
print(f"{prefix} [GPU] {message}")
elif level == "MODEL_OPS" and (DebugLevels.MODEL_OPS or is_error):
print(f"{prefix} [MODEL] {message}")
elif level == "DATA_PROCESS" and (DebugLevels.DATA_PROCESS or is_error):
print(f"{prefix} [DATA] {message}")
elif level == "GENERAL" and (DebugLevels.GENERAL or is_error):
print(f"{prefix} [INFO] {message}")
elif level == "AUDIO" and (DebugLevels.AUDIO or is_error):
print(f"{prefix} [AUDIO] {message}")
elif level == "SEGMENTS" and (DebugLevels.SEGMENTS or is_error):
print(f"{prefix} [SEG] {message}")
elif level == "DUPLICATES" and (DebugLevels.DUPLICATES or is_error):
print(f"{prefix} [DUP] {message}")
elif level == "VALIDATION" and (DebugLevels.VALIDATION or is_error):
print(f"{prefix} [VAL] {message}")
class AudioStats:
"""Track audio processing statistics"""
def __init__(self):
self.total_segments = 0
self.segments_under_min = 0
self.segments_over_max = 0
self.total_duration = 0
self.segment_durations = []
def add_segment(self, duration):
"""
Adds a new audio segment duration to tracking statistics.
Args:
duration (float): Duration of audio segment in seconds
"""
self.total_segments += 1
self.total_duration += duration
self.segment_durations.append(duration)
def print_stats(self):
"""
Prints summary statistics for all processed audio segments.
Shows total segments, average duration, segments under minimum length,
and segments over maximum length.
"""
if not self.segment_durations:
return
avg_duration = self.total_duration / self.total_segments
debug_print("Audio Processing Statistics:", "AUDIO")
debug_print(f"Total segments: {self.total_segments}", "AUDIO")
debug_print(f"Average duration: {avg_duration:.2f}s", "AUDIO")
debug_print(
f"Segments under minimum: {self.segments_under_min}",
"AUDIO")
debug_print(
f"Segments over maximum: {self.segments_over_max}",
"AUDIO")
if self.segments_under_min > 0:
debug_print(
f"{self.segments_under_min} segments are under minimum duration!",
"AUDIO",
is_warning=True,
)
def get_gpu_memory():
"""Enhanced GPU memory reporting"""
if DebugLevels.GPU_MEMORY:
try:
nvmlInit()
handle = nvmlDeviceGetHandleByIndex(0)
info = nvmlDeviceGetMemoryInfo(handle)
debug_print("GPU Memory Status:", "GPU_MEMORY")
debug_print(f"Total: {info.total / 1024**2:.2f} MB", "GPU_MEMORY")
debug_print(f"Used: {info.used / 1024**2:.2f} MB", "GPU_MEMORY")
debug_print(f"Free: {info.free / 1024**2:.2f} MB", "GPU_MEMORY")
# Add warning if memory is low
if info.free / info.total < 0.1: # Less than 10% free
debug_print(
"Low GPU memory available!",
"GPU_MEMORY",
is_warning=True)
except Exception as e:
debug_print(
f"NVML not available: {e}",
"GPU_MEMORY",
is_error=True)
########################
#### Find All Model ####
########################
BASE_MODEL_DETECTED = False
def scan_models_folder():
"""
Scans the models folder to detect available XTTS models and verifies required files.
This function checks each subfolder within the base models directory to see if it contains
all the required files for a valid model. It updates the `BASE_MODEL_DETECTED` global variable
based on whether at least one valid model is found.
Returns:
dict: A dictionary where keys are model names (subfolder names) and values are booleans
indicating whether the model folder contains all the required files.
Globals:
BASE_MODEL_DETECTED (bool): Set to `True` if at least one valid model is detected,
otherwise `False`.
Raises:
FileNotFoundError: If the models folder is not found, prints an error message and exits
the script.
"""
global BASE_MODEL_DETECTED
models_folder = base_path
scan_available_models = {}
scan_required_files = [
"config.json",
"model.pth",
"mel_stats.pth",
"speakers_xtts.pth",
"vocab.json",
"dvae.pth",
]
try:
for subfolder in models_folder.iterdir():
if subfolder.is_dir():
model_name = subfolder.name
if all(subfolder.joinpath(file).exists()
for file in scan_required_files):
scan_available_models[model_name] = True
BASE_MODEL_DETECTED = True
else:
debug_print(
f"Model folder '{model_name}' is missing required files",
level="GENERAL",
is_warning=True)
if not scan_available_models:
scan_available_models["No Model Available"] = False
BASE_MODEL_DETECTED = False
except FileNotFoundError:
debug_print(
"No XTTS models folder found. You have not yet downloaded any models or no XTTS",
"GENERAL",
is_error=True)
debug_print(
"models can be found. Please run AllTalk and download an XTTS model that can be",
"GENERAL",
is_error=True)
debug_print(
"used for training. Or place a full model in the following location",
"GENERAL",
is_error=True)
debug_print(
"\\models\\xtts\\{modelfolderhere}",
"GENERAL",
is_error=True)
sys.exit(1) # Exit the script with an error status
return scan_available_models
# Get available models
available_models = scan_models_folder()
#################################################
#### Pre-Flight Checklist Functions & Gradio ####
#################################################
class SystemChecks:
"""Centralized system check management"""
def __init__(self):
self.status = {
"overall": True,
"disk_space": True,
"ram": True,
"gpu": True,
"cuda": True,
"pytorch": True,
"tts": True,
"base_model": True,
}
self.results = {}
def check_disk_space(self, required_gb=18):
"""Check available disk space"""
try:
disk_usage = shutil.disk_usage(os.getcwd())
free_space_gb = disk_usage.free / (1 << 30)
status = free_space_gb > required_gb
self.status["disk_space"] = status
self.results["disk_space"] = {
"status": "✅ Pass" if status else "❌ Fail",
"details": f"{free_space_gb:.2f} GB available",
"icon": "✅" if status else "❌",
"message": (
"Sufficient disk space available"
if status
else f"Insufficient disk space. Need {required_gb}GB, have {free_space_gb:.2f}GB"
),
}
except Exception as e:
self.handle_check_error("disk_space", str(e))
def check_system_ram(self):
"""Check system RAM and GPU memory with refined thresholds"""
try:
vm = psutil.virtual_memory()
total_ram_gb = vm.total / (1024**3)
available_ram_gb = vm.available / (1024**3)
# RAM check
ram_status = available_ram_gb >= 8
self.status["ram"] = ram_status
# GPU memory check
if torch.cuda.is_available():
gpu_id = torch.cuda.current_device()
gpu_name = torch.cuda.get_device_name(gpu_id)
total_gpu_mem = torch.cuda.get_device_properties(
gpu_id).total_memory / (1024**3)
used_gpu_mem = torch.cuda.memory_reserved(gpu_id) / (1024**3)
available_gpu_mem = total_gpu_mem - used_gpu_mem
# New GPU status logic with better thresholds
if total_gpu_mem >= 13:
gpu_status = "pass" # More than 13GB - Clear pass
elif total_gpu_mem >= 11.5:
gpu_status = "warning" # Between 11.5GB and 13GB - Warning
else:
gpu_status = "fail" # Less than 11.5GB - Fail
# Both pass and warning are considered "okay"
self.status["gpu"] = gpu_status in ["pass", "warning"]
else:
gpu_name = "No GPU detected"
total_gpu_mem = 0
available_gpu_mem = 0
gpu_status = "fail"
self.status["gpu"] = False
self.results["memory"] = {
"ram_status": "✅ Pass" if ram_status else "❌ Fail",
"ram_details": f"{available_ram_gb:.2f}GB available of {total_ram_gb:.2f}GB",
"gpu_status": {
"pass": "✅ Pass",
"warning": "⚠️ Warning",
"fail": "❌ Fail"}[gpu_status],
"gpu_details": f"{gpu_name}: {available_gpu_mem:.2f}GB available of {total_gpu_mem:.2f}GB",
}
except Exception as e:
self.handle_check_error("memory", str(e))
def check_cuda_pytorch(self):
"""Check CUDA and PyTorch setup"""
try:
cuda_available = torch.cuda.is_available()
pytorch_version = torch.__version__
if cuda_available:
cuda_version = torch.version.cuda
# Check if CUDA is actually working
try:
torch.tensor([1.0, 2.0]).cuda()
cuda_working = True
except BaseException:
cuda_working = False
pytorch_cuda_status = cuda_version in ["11.8", "12.1"]
else:
cuda_working = False
cuda_version = "N/A"
pytorch_cuda_status = False
self.status["cuda"] = cuda_working
self.status["pytorch"] = pytorch_cuda_status
# Separate results for CUDA and PyTorch
self.results["cuda"] = {
"status": "✅ Pass" if cuda_working else "❌ Fail",
"details": f"CUDA {cuda_version}" if cuda_working else "CUDA not working",
}
self.results["pytorch"] = {
"status": "✅ Pass" if pytorch_cuda_status else "❌ Fail",
"details": f"PyTorch {pytorch_version} with CUDA {cuda_version}",
}
except Exception as e:
self.handle_check_error("cuda_pytorch", str(e))
def check_tts_version(self, required_version="0.24.0"):
"""Check TTS version"""
try:
installed_version = metadata.version("coqui-tts")
status = version.parse(
installed_version) >= version.parse(required_version)
self.status["tts"] = status
self.results["tts"] = {
"status": "✅ Pass" if status else "❌ Fail",
"details": f"TTS version {installed_version} installed",
"meets_requirement": status,
}
except Exception as e:
self.handle_check_error("tts", str(e))
def check_base_model(self):
"""Check XTTS base model"""
try:
BASE_MODEL_DETECTED = any(
available_models.values()) if available_models else False
self.status["base_model"] = BASE_MODEL_DETECTED
self.results["base_model"] = {
"status": "✅ Pass" if BASE_MODEL_DETECTED else "❌ Fail",
"details": "Base model detected" if BASE_MODEL_DETECTED else "No base model found",
}
except Exception as e:
self.handle_check_error("base_model", str(e))
def handle_check_error(self, check_name: str, error_msg: str):
"""Handle errors in checks"""
self.status[check_name] = False
self.status["overall"] = False
self.results[check_name] = {
"status": "❌ Error",
"details": f"Check failed: {error_msg}"}
def run_all_checks(self):
"""Run all system checks"""
self.check_disk_space()
self.check_system_ram()
self.check_cuda_pytorch()
self.check_tts_version()
self.check_base_model()
# Update overall status
self.status["overall"] = all(self.status.values())
return self.status["overall"]
def get_markdown_report(self):
"""Generate markdown report of all checks"""
report = []
# Overall Status
overall_icon = "✅" if self.status["overall"] else "❌"
report.append(f"## System Check Results {overall_icon}\n")
# Individual Checks
if "disk_space" in self.results:
report.append(
f"### Storage\n{self.results['disk_space']['status']} {self.results['disk_space']['details']}\n"
)
if "memory" in self.results:
report.append("### Memory")
report.append(
f"- RAM: {self.results['memory']['ram_status']} {self.results['memory']['ram_details']}"
)
report.append(
f"- GPU: {self.results['memory']['gpu_status']} {self.results['memory']['gpu_details']}\n"
)
if "cuda_pytorch" in self.results:
report.append(
f"### CUDA & PyTorch\n{self.results['cuda_pytorch']['status']} {self.results['cuda_pytorch']['details']}\n"
)
if "tts" in self.results:
report.append(
f"### TTS\n{self.results['tts']['status']} {self.results['tts']['details']}\n"
)
if "base_model" in self.results:
report.append(
f"### Base Model\n{self.results['base_model']['status']} {self.results['base_model']['details']}\n"
)
return "\n".join(report)
def get_ui_updates(self, _components):
"""Map check results to UI components"""
# Get status for all checks
status_updates = []
# Overall status first
overall_status = "✅ All Systems Go!" if self.status["overall"] else "⚠️ Some Checks Failed"
status_updates.append(overall_status)
# Add status for each check in the order they appear in
# CHECK_REQUIREMENTS
for check_id in CHECK_REQUIREMENTS:
if check_id in self.results:
status_updates.append(self.results[check_id]["status"])
else:
status_updates.append("❌ Check not run")
return status_updates
# PFC Help Content and Requirements
HELP_CONTENT = {
"disk_space_details": """
### Disk Space Requirements
#### Minimum Requirements
- **18GB** free disk space for training
- SSD recommended for better performance
#### Space Usage Breakdown
- Model files: ~4GB
- Training data: ~5-10GB
- Temporary files: ~4GB
- Safety margin: ~2GB
#### Troubleshooting
1. **Insufficient Space**
- Clear space on the current partition
- Run fine-tuning on another partition with more space
- Keep at least 20% free space on drive
2. **Performance Issues**
- Mechanical HDDs will be significantly slower
- Consider using an SSD for training
""",
"gpu_requirements": """
### GPU Memory Requirements
#### System Requirements
- **NVIDIA GPU Required**
- **12GB+ VRAM** recommended for optimal performance
- **8GB+ VRAM** minimum requirement
#### Platform Differences
- **Windows Systems:**
* Can use system RAM as extended VRAM
* 12GB VRAM or less requires 24GB+ System RAM and may fail.
* 12GB VRAM cards will have a warning because of the above.
* Performance may vary when using extended VRAM
- **Linux Systems:**
* Cannot use extended VRAM
* Limited to physical GPU VRAM only
* Minimum 12GB VRAM recommended
""",
"pytorch_cuda": """
### PyTorch and CUDA
- CUDA-enabled PyTorch installation required
- Compatible with CUDA 11.8/12.1
- Current PyTorch version recommended
""",
"model_setup": """
### Base Model Requirements
#### Model Installation
- Download XTTS v2.0.3 model (recommended)
- BPE Tokenize needs a v2.0.3 model
- Use AllTalk's main interface:
* TTS Engine Settings > XTTS > Model/Voices Download
#### Model Location
- Place in `/models/xtts/{modelname}/`
- Required files:
* model.pth
* config.json
* vocab.json
* dvae.pth
* mel_stats.pth
* speakers_xtts.pth
""",
}
CHECK_REQUIREMENTS = {
"disk_space": {
"label": "Disk Space",
"requirement": "18GB+ Required",
"description": "Required for temporary files and model storage",
"priority": "high",
"check_function": "check_disk_space",
},
"ram": {
"label": "System RAM",
"requirement": "16GB+ Recommended",
"description": "Affects overall processing speed and stability",
"priority": "medium",
"check_function": "check_system_ram",
},
"vram": {
"label": "GPU VRAM",
"requirement": "12GB+ Recommended",
"description": "Crucial for model training performance",
"priority": "high",
"check_function": "check_gpu_memory",
},
"cuda": {
"label": "CUDA Support",
"requirement": "CUDA 11.8/12.1",
"description": "Required for GPU acceleration",
"priority": "high",
"check_function": "check_cuda_pytorch",
},
"pytorch": {
"label": "PyTorch",
"requirement": "CUDA enabled",
"description": "Must be CUDA compatible version",
"priority": "high",
"check_function": "check_cuda_pytorch",
},
"tts": {
"label": "TTS Version",
"requirement": "0.24.0+",
"description": "Required for fine-tuning",
"priority": "high",
"check_function": "check_tts_version",
},
"base_model": {
"label": "Base Model",
"requirement": "Must be present",
"description": "XTTS base model required",
"priority": "high",
"check_function": "check_base_model",
},
}
# Display categories for organizing checks
CHECK_CATEGORIES = {
"hardware": {
"title": "💻 Hardware Requirements",
"description": "System hardware capabilities",
"checks": ["disk_space", "ram", "vram"],
},
"cuda": {
"title": "🔧 CUDA Setup",
"description": "CUDA and PyTorch configuration",
"checks": ["cuda", "pytorch"],
},
"software": {
"title": "🤖 Software Setup",
"description": "Required software and models",
"checks": ["tts", "base_model"],
},
}
class PFCComponents:
"""Store all PFC UI components"""
def __init__(self):
self.overall_status = None
self.status_boxes = {}
self.accordions = {}
def create_status_box(self, check_id):
"""Create a status box for a specific check"""
check_info = CHECK_REQUIREMENTS[check_id]
with gr.Group():
status = gr.Label(
label=check_info["label"],
value="Ready to Check",
elem_classes="status-indicator")
gr.Markdown(f"*Required: {check_info['requirement']}*")
self.status_boxes[check_id] = status
return status
def create_pfc_interface():
"""
Creates a pre-flight checklist Gradio interface for validating system requirements before XTTS model training.
Validates:
- Disk space (min 18GB)
- System RAM (min 16GB recommended)
- GPU VRAM (min 12GB recommended)
- CUDA support
- PyTorch configuration
- TTS version (min 0.24.0)
- Base model presence
Returns:
PFCComponents: Container with all UI component references for status updates
"""
components = PFCComponents()
system_checks = SystemChecks()
with gr.Column():
# Status and button at top
with gr.Row():
components.overall_status = gr.Label(
value="Click 'Run System Checks' to begin",
label="System Status",
elem_classes="status-label",
scale=2,
)
with gr.Row():
refresh_btn = gr.Button("Run System Checks", scale=1)
# Checks in 3x2 grid
with gr.Row():
for check_id in CHECK_REQUIREMENTS:
with gr.Column(scale=1):
components.create_status_box(check_id)
# Help sections in 2x2 grid
with gr.Group("💡 Help & Troubleshooting"):
with gr.Row():
with gr.Column():
with gr.Accordion("💻 Hardware Requirements", open=False):
gr.Markdown(HELP_CONTENT["gpu_requirements"], elem_classes="custom-markdown")
with gr.Column():
with gr.Accordion("🖥️ Storage Requirements", open=False):
gr.Markdown(HELP_CONTENT["disk_space_details"], elem_classes="custom-markdown")
with gr.Row():
with gr.Column():
with gr.Accordion("🤖 PyTorch & CUDA", open=False):
gr.Markdown(HELP_CONTENT["pytorch_cuda"], elem_classes="custom-markdown")
with gr.Column():
with gr.Accordion("📦 Model Setup", open=False):
gr.Markdown(HELP_CONTENT["model_setup"], elem_classes="custom-markdown")
def run_checks():
system_checks.run_all_checks()
# Prepare updates for all components
updates = []
# Overall status
updates.append(
"✅ All Systems Go!"
if system_checks.status["overall"]
else "⚠️ Some Checks Failed"
)
# Add updates for each category's checks
for category in CHECK_CATEGORIES.values():
for check_id in category["checks"]:
if check_id == "ram":
updates.append(
system_checks.results["memory"]["ram_status"])
elif check_id == "vram":
updates.append(
system_checks.results["memory"]["gpu_status"])
elif check_id == "cuda":
updates.append(
system_checks.results["cuda"]["status"])
elif check_id == "pytorch":
updates.append(
system_checks.results["pytorch"]["status"])
else:
updates.append(
system_checks.results.get(check_id, {}).get(
"status", "❌ Check not run"
)
)
return updates
refresh_btn.click(
fn=run_checks,
outputs=[
components.overall_status] +
list(
components.status_boxes.values()),
)
return components
###########################################
#### STEP 1 Dataset Creation Functions ####
###########################################
def format_audio_list(
fal_target_language,
fal_whisper_model,
fal_max_sample_length,
fal_min_sample_length,
fal_eval_split_number,
fal_speaker_name_input,
fal_create_bpe_tokenizer,
fal_gradio_progress=gr.Progress(),
fal_use_vad=True,
fal_precision="mixed",
):
"""
Process and format audio files for XTTS training. Handles audio segmentation, transcription,
and metadata creation with optional VAD and precision settings.
Returns:
tuple: (train_metadata_path, eval_metadata_path, audio_total_size)
"""
global validate_train_metadata_path, validate_eval_metadata_path, validate_audio_folder
global validate_whisper_model, validate_target_language, out_path, torch, whisper # pylint: disable=no-member
# Clear down the finetune.log file
Logger().clear_log()
# Initialize statistics tracker
stats = AudioStats()
# Basic setup
buffer = 0.3
max_duration = float(fal_max_sample_length)
min_duration = float(fal_min_sample_length)
eval_percentage = fal_eval_split_number / 100.0
speaker_name = fal_speaker_name_input
audio_total_size = 0
metadata = {"audio_file": [], "text": [], "speaker_name": []}
too_long_files = []
# Initialize directories
if speaker_name and speaker_name != 'personsname':
out_path = this_dir / "finetune" / speaker_name
else:
out_path = default_path
debug_print(f"Initializing output directory: {out_path}", "GENERAL")
os.makedirs(out_path, exist_ok=True)
temp_folder = os.path.join(out_path, "temp")
os.makedirs(temp_folder, exist_ok=True)
audio_folder = os.path.join(out_path, "wavs")
os.makedirs(audio_folder, exist_ok=True)
original_samples_folder = os.path.join(
out_path, "..", "put-voice-samples-in-here")
# Load Whisper model with specified precision
fal_gradio_progress((1, 10), desc="Loading Whisper Model")
device = "cuda" if torch.cuda.is_available() else "cpu"
debug_print(f"Using device: {device}", "MODEL_OPS")
# Monitor GPU memory before model loading
get_gpu_memory()
debug_print(f"Loading Whisper model: {fal_whisper_model}", "MODEL_OPS")
asr_model = whisper.load_model(fal_whisper_model, device=device)
if fal_precision == "float16" and device == "cuda":
debug_print("Using FP16 precision", "MODEL_OPS")
asr_model = asr_model.half()
elif fal_precision == "mixed" and device == "cuda":
debug_print("Using mixed precision", "MODEL_OPS")
else:
debug_print("Using FP32 precision", "MODEL_OPS")
# Initialize SileroVAD if requested
vad_model = None
if fal_use_vad:
fal_gradio_progress((2, 10), desc="Loading VAD Model")
debug_print("Initializing Silero VAD", "MODEL_OPS")
model, utils = torch.hub.load(repo_or_dir='snakers4/silero-vad',
model='silero_vad',
force_reload=False)
vad_model = model.to(device)
get_speech_timestamps, collect_chunks = utils[0], utils[4]
# Monitor GPU memory after model loading
get_gpu_memory()
# Load existing metadata
fal_gradio_progress((3, 10), desc="Checking for Existing Metadata")
train_metadata_path = os.path.join(out_path, "metadata_train.csv")
eval_metadata_path = os.path.join(out_path, "metadata_eval.csv")
existing_metadata = {'train': None, 'eval': None}
# Handle language file
lang_file_path = os.path.join(out_path, "lang.txt")
current_language = None
if os.path.exists(train_metadata_path):
existing_metadata['train'] = pd.read_csv(train_metadata_path, sep="|")
debug_print("Loaded existing training metadata", "DATA_PROCESS")
if os.path.exists(eval_metadata_path):
existing_metadata['eval'] = pd.read_csv(eval_metadata_path, sep="|")
debug_print("Loaded existing evaluation metadata", "DATA_PROCESS")
if os.path.exists(train_metadata_path) and os.path.exists(eval_metadata_path):
# If dataset exists, read but don't modify the language
if os.path.exists(lang_file_path):
with open(lang_file_path, 'r', encoding='utf-8') as existing_lang_file:
current_language = existing_lang_file.read().strip()
debug_print(f"Using existing dataset language: {current_language}", "GENERAL")
fal_target_language = current_language # Update our target language to match existing
else:
# Only create/update language file for new datasets
if os.path.exists(lang_file_path):
with open(lang_file_path, 'r', encoding='utf-8') as existing_lang_file:
current_language = existing_lang_file.read().strip()
if current_language != fal_target_language:
with open(lang_file_path, 'w', encoding='utf-8') as lang_file:
lang_file.write(fal_target_language + '\n')
debug_print(f"Updated language to: {fal_target_language}", "GENERAL")
else:
debug_print("Using existing language setting", "GENERAL")
# Get audio files list
original_audio_files = [os.path.join(original_samples_folder, file)
for file in os.listdir(original_samples_folder)
if file.endswith(('.mp3', '.flac', '.wav'))]
fal_gradio_progress((4, 10), desc="Scanning for Audio Files")
if not original_audio_files:
debug_print(
f"No audio files found in {original_samples_folder}",
"AUDIO",
is_error=True)
return None, None, 0
debug_print(
f"Found {len(original_audio_files)} audio files to process",
"AUDIO")
# Initialize processing
whisper_words = []
audio_steps = (0, len(original_audio_files))
gradio_progress_duration = 0
fal_gradio_progress(
audio_steps,
desc="Processing Audio Files",
unit="files")
for audio_path in original_audio_files:
start = datetime.datetime.now()
audio_file_name_without_ext, _ = os.path.splitext(
os.path.basename(audio_path))
temp_audio_path = os.path.join(
temp_folder, f"{audio_file_name_without_ext}.wav")
try:
shutil.copy2(audio_path, temp_audio_path)
fal_gradio_progress(
audio_steps,
desc=f"Processing {audio_file_name_without_ext}",
unit="files")
debug_print(
f"Processing: {audio_file_name_without_ext}",
"GENERAL")
except Exception as e:
debug_print(
f"Error copying file {audio_path}: {str(e)}",
"GENERAL",
is_error=True)
continue
# Check if already processed
prefix_check = f"wavs/{audio_file_name_without_ext}_"
skip_processing = False
for key in ['train', 'eval']:
if existing_metadata[key] is not None:
mask = existing_metadata[key]['audio_file'].str.startswith(
prefix_check)
if mask.any():
debug_print(
f"Skipping previously processed file: {audio_file_name_without_ext}",
"GENERAL")
skip_processing = True
audio_total_size = 121
break
if skip_processing:
continue
# Load and process audio
wav, sr = torchaudio.load(str(audio_path))
if wav.size(0) != 1:
wav = torch.mean(wav, dim=0, keepdim=True)
wav = wav.squeeze()
audio_total_size += (wav.size(-1) / sr)
original_duration = wav.size(-1) / sr
debug_print(
f"Original audio duration: {original_duration:.2f}s",
"AUDIO")
# Process with VAD if enabled
if fal_use_vad and vad_model is not None:
debug_print("Processing with VAD", "AUDIO")
# Get VAD segments with resampling
vad_segments = process_audio_with_vad(
wav, sr, vad_model, get_speech_timestamps)
# Group short segments that are close together
merged_segments = merge_short_segments(
vad_segments, min_duration, max_gap=0.3)
debug_print(
f"Merged {len(vad_segments)-len(merged_segments)} short segments",
"SEGMENTS")
# Convert VAD segments to audio chunks
speech_chunks = []
for segment in merged_segments:
chunk = wav[segment['start']:segment['end']]
duration = chunk.size(-1) / sr
if duration < min_duration:
debug_print(
f"Segment too short ({duration:.2f}s), attempting to extend",
"SEGMENTS",
is_warning=True)
# Try to extend segment if possible
chunk = extend_segment(
wav, segment['start'], segment['end'], sr, min_duration)
duration = chunk.size(-1) / sr
if chunk.numel() > 0:
speech_chunks.append((chunk, duration))
stats.add_segment(duration)
# Process each speech chunk
for chunk_idx, (chunk, duration) in enumerate(speech_chunks):
if duration < min_duration:
stats.segments_under_min += 1
debug_print(
f"Short segment: {duration:.2f}s",
"SEGMENTS",
is_warning=True)
elif duration > max_duration:
stats.segments_over_max += 1
debug_print(
f"Long segment: {duration:.2f}s",
"SEGMENTS",
is_warning=True)
chunk_path = os.path.join(
temp_folder, f"{audio_file_name_without_ext}_chunk_{chunk_idx}.wav")
torchaudio.save(str(chunk_path), chunk.unsqueeze(0), sr)
# Transcribe with appropriate precision
if fal_precision == "mixed" and device == "cuda":
with torch.cuda.amp.autocast():
fal_gradio_progress((5, 10), desc="Transcribing Audio")
result = asr_model.transcribe(
chunk_path,
language=fal_target_language,
word_timestamps=True,
verbose=None
)
else:
fal_gradio_progress((5, 10), desc="Transcribing Audio")
result = asr_model.transcribe(
chunk_path,
language=fal_target_language,
word_timestamps=True,
verbose=None
)
if not result.get("text", "").strip():
debug_print(
f"Empty transcription for chunk {chunk_idx}",
"DATA_PROCESS",
is_warning=True)
continue
# Process transcription result
process_transcription_result(
result,
chunk,
sr,
chunk_idx,
audio_file_name_without_ext,
metadata,
whisper_words,
max_duration,
buffer,
speaker_name,
audio_folder,
too_long_files,
fal_create_bpe_tokenizer,
fal_target_language)
os.remove(chunk_path)
debug_print(
f"Processed chunk {chunk_idx} ({duration:.2f}s)",
"SEGMENTS")
else:
# Regular processing without VAD
debug_print("Processing without VAD", "AUDIO")
if fal_precision == "mixed" and device == "cuda":
with torch.cuda.amp.autocast():
result = asr_model.transcribe(
audio_path,
language=fal_target_language,
word_timestamps=True,
verbose=None
)
else:
result = asr_model.transcribe(
audio_path,
language=fal_target_language,
word_timestamps=True,
verbose=None
)
# Process transcription result
process_transcription_result(
result,
wav,
sr,
0,
audio_file_name_without_ext,
metadata,
whisper_words,
max_duration,
buffer,
speaker_name,
audio_folder,
too_long_files,
fal_create_bpe_tokenizer,
fal_target_language)
os.remove(temp_audio_path)
# Update progress
end = datetime.datetime.now()
gradio_progress_duration += (end - start).total_seconds()
audio_steps = (audio_steps[0] + 1, audio_steps[1])
additional_data_points_needed = audio_steps[1] - audio_steps[0]
avg_duration = gradio_progress_duration / audio_steps[0]
gradio_estimated_duration = (
avg_duration * additional_data_points_needed)
fal_gradio_progress(
audio_steps,
desc=f"Processing. Estimated Completion: {c_logger.format_duration(gradio_estimated_duration)}",
unit="files")
# Print final statistics
stats.print_stats()
# Verify processed files exist
audio_files = [os.path.join(audio_folder, file) for file in os.listdir(
audio_folder) if file.endswith('.wav')]
if not audio_files:
debug_print("No processed audio files found", "AUDIO", is_error=True)
return None, None, 0
# Final statistics before metadata handling
stats.print_stats()
# Handle existing metadata case
if os.path.exists(train_metadata_path) and os.path.exists(
eval_metadata_path):
debug_print("Using existing metadata files", "DATA_PROCESS")
_set_validation_paths(
train_metadata_path, eval_metadata_path, audio_folder,
fal_whisper_model, fal_target_language
)
_cleanup_resources(asr_model, existing_metadata)
return train_metadata_path, eval_metadata_path, audio_total_size
# Check for new metadata
if not metadata["audio_file"]:
debug_print(
"No new audio files to process",
"DATA_PROCESS",
is_warning=True)
_set_validation_paths(
train_metadata_path, eval_metadata_path, audio_folder,
fal_whisper_model, fal_target_language
)
_cleanup_resources(asr_model, existing_metadata)
return train_metadata_path, eval_metadata_path, audio_total_size
# Process metadata and handle duplicates
debug_print("Processing metadata and handling duplicates", "DATA_PROCESS")
new_data_df = pd.DataFrame(metadata)
# Duplicate detection and handling
duplicate_files = new_data_df['audio_file'].value_counts()
duplicates_found = duplicate_files[duplicate_files > 1]
fal_gradio_progress((6, 10), desc="Handling Duplicate Transcriptions")
if not duplicates_found.empty:
debug_print(
f"Found {len(duplicates_found)} files with multiple transcriptions",
"DUPLICATES")
for file, count in duplicates_found.items():
debug_print(f"{file}: {count} occurrences", "DUPLICATES")
# Re-transcribe duplicates
best_transcriptions = handle_duplicates(
duplicates_found.index,
audio_folder,
fal_target_language,
fal_whisper_model)
# Update transcriptions and remove duplicates
for file_path, trans_info in best_transcriptions.items():
new_data_df.loc[new_data_df['audio_file'] ==
file_path, 'text'] = trans_info['text']
debug_print(f"Updated transcription for {file_path}", "DUPLICATES")
new_data_df = new_data_df.drop_duplicates(
subset='audio_file', keep='first')
debug_print(
f"Cleaned up {len(duplicates_found)} duplicate entries",
"DUPLICATES")
# Handle evaluation split
debug_print("Creating train/eval split", "DATA_PROCESS")
eval_percentage = _adjust_eval_percentage(fal_eval_split_number)
# Create and validate splits
fal_gradio_progress((7, 10), desc="Creating Train/Eval Split")
train_eval_split = create_dataset_splits(
new_data_df,
eval_percentage,
random_seed=42
)
if train_eval_split is None:
debug_print(
"Failed to create valid dataset splits",
"DATA_PROCESS",
is_error=True)
return None, None, 0
final_training_set, final_eval_set = train_eval_split
# Write metadata files
debug_print(
f"Writing {len(final_training_set)} training and {len(final_eval_set)} eval samples",
"DATA_PROCESS")
try:
fal_gradio_progress((8, 10), desc="Saving Metadata Files")
_write_metadata_files(
final_training_set, final_eval_set,
train_metadata_path, eval_metadata_path
)
except Exception as e:
debug_print(
f"Error writing metadata: {str(e)}",
"DATA_PROCESS",
is_error=True)
raise
# Handle BPE tokenizer
if fal_create_bpe_tokenizer:
fal_gradio_progress((9, 10), desc="Training BPE Tokenizer")
debug_print("Training BPE Tokenizer", "MODEL_OPS")
_create_bpe_tokenizer(whisper_words, out_path, base_path)
debug_print("Finalizing processing", "GENERAL")
fal_gradio_progress((10, 10), desc="Finalizing Process")
# Cleanup and set validation paths
_cleanup_resources(
asr_model,
final_eval_set,
final_training_set,
new_data_df,
existing_metadata)
_set_validation_paths(
train_metadata_path, eval_metadata_path, audio_folder,
fal_whisper_model, fal_target_language
)
# Log final statistics
if too_long_files:
debug_print("Files that were split due to length:", "SEGMENTS")
for file_name, length in too_long_files:
debug_print(f" {file_name}: {length:.2f} seconds", "SEGMENTS")
return train_metadata_path, eval_metadata_path, audio_total_size
def _create_bpe_tokenizer(bpe_whisper_words, bpe_out_path, bpe_base_path):
"""Create and train BPE tokenizer"""
vocab_path = bpe_base_path / "xttsv2_2.0.3" / "vocab.json"
if not vocab_path.exists():
debug_print(
"BPE tokenizer will not be created for this dataset creation run.",
"GENERAL",
is_warning=True,
)
debug_print(
"XTTS v2.0.3 base model not found for BPE Tokenizer. Please download it first using AllTalk's",
"GENERAL",
is_warning=True,
)
debug_print(
"main Gradio interface > TTS Engine Settings > XTTS > Model/Voices Downloads.",
"GENERAL",
is_warning=True,
)
debug_print(
"Dataset creation will continue without the BPE tokenizer.",
"GENERAL",
is_warning=True)
raise FileNotFoundError(f"Missing required file: {vocab_path}")
try:
debug_print("Initializing BPE tokenizer training", "MODEL_OPS")
tokenizer = ByteLevelBPETokenizer(str(vocab_path))
tokenizer.pre_tokenizer = Whitespace()
# Add special tokens
special_tokens = [
"[PAD]",
"[UNK]",
"[CLS]",
"[SEP]",
"[MASK]",
"[STOP]",
"[SPACE]"]
tokenizer.add_tokens(special_tokens)
debug_print(
f"Training tokenizer on {len(bpe_whisper_words)} words",
"MODEL_OPS")
tokenizer.train_from_iterator(
bpe_whisper_words,
vocab_size=30000,
show_progress=True,
min_frequency=2,
special_tokens=special_tokens,
)
# Save tokenizer
tokenizer_path = str(bpe_out_path / "bpe_tokenizer-vocab.json")
tokenizer.save(path=tokenizer_path, pretty=True)
debug_print(f"Saved BPE tokenizer to {tokenizer_path}", "MODEL_OPS")
except Exception as e:
debug_print(
f"Failed to create BPE tokenizer: {str(e)}",
"MODEL_OPS",
is_error=True)
raise
def merge_short_segments(segments, min_duration, max_gap=0.5):
"""
More aggressive merge strategy for short segments
- Increases max_gap to 0.5s (from 0.3s)
- Looks ahead multiple segments for potential merges
- Considers surrounding context
"""
if not segments:
return segments
merged = []
current_group = []
target_duration = (min_duration + 10.0) / 2 # Target middle of range
for i, segment in enumerate(segments):
current_duration = sum(s["end"] - s["start"]
for s in current_group) if current_group else 0
# If this is a continuation of current group
if current_group and (segment["start"] - current_group[-1]["end"]) <= max_gap:
# Check if adding this segment gets us closer to target duration
new_duration = current_duration + (segment["end"] - segment["start"])
if abs(new_duration - target_duration) < abs(current_duration - target_duration):
current_group.append(segment)
else:
# Save current group and start new one
merged_segment = {
"start": current_group[0]["start"],
"end": current_group[-1]["end"]
}
merged.append(merged_segment)
current_group = [segment]
else:
# Save previous group if it exists
if current_group:
merged_segment = {
"start": current_group[0]["start"],
"end": current_group[-1]["end"]
}
merged.append(merged_segment)
current_group = [segment]
# Handle last group
if current_group:
merged_segment = {
"start": current_group[0]["start"],
"end": current_group[-1]["end"]
}
merged.append(merged_segment)
debug_print(
f"Merged {len(segments) - len(merged)} segments into {len(merged)} segments with mid-range preference",
"SEGMENTS"
)
return merged
def extend_segment(wav, start, end, sr, min_duration, context_window=1.0):
"""
Improved segment extension with better context handling
- Adds context_window parameter for smoother extensions
- More balanced extension on both sides
- Checks audio content when extending
"""
current_duration = (end - start) / sr
if current_duration >= min_duration:
return wav[start:end]
samples_needed = int((min_duration - current_duration) * sr)
# Try to extend equally on both sides
extend_left = samples_needed // 2
extend_right = samples_needed - extend_left
# Add some context window
context_samples = int(context_window * sr)
new_start = max(0, start - extend_left - context_samples)
new_end = min(wav.size(-1), end + extend_right + context_samples)
# Check if we got enough duration
if (new_end - new_start) / sr < min_duration:
# If still too short, try to compensate from the other side
if new_start == 0:
new_end = min(wav.size(-1), end + samples_needed + context_samples)
elif new_end == wav.size(-1):
new_start = max(0, start - samples_needed - context_samples)
debug_print(
f"Extended segment from {current_duration:.2f}s to {(new_end - new_start) / sr:.2f}s",
"SEGMENTS",
)
return wav[new_start:new_end]
# Helper functions for better organization
def _adjust_eval_percentage(aep_eval_split_number):
"""Adjust evaluation percentage to be within acceptable bounds"""
eval_percentage = aep_eval_split_number / 100.0
min_eval_percentage = 0.1
max_eval_percentage = 0.3
if eval_percentage < min_eval_percentage:
debug_print(
f"Adjusting eval split from {eval_percentage:.1%} to {min_eval_percentage:.1%}",
"DATA_PROCESS",
is_warning=True,
)
return min_eval_percentage
if eval_percentage > max_eval_percentage: # Changed from elif to if
debug_print(
f"Adjusting eval split from {eval_percentage:.1%} to {max_eval_percentage:.1%}",
"DATA_PROCESS",
is_warning=True,
)
return max_eval_percentage
return eval_percentage
def _set_validation_paths(vp_train_path, vp_eval_path, vp_audio_folder,
vp_whisper_model, vp_target_language):
"""Set global validation paths"""
global VALIDATE_TRAIN_METADATA_PATH, VALIDATE_EVAL_METADATA_PATH, VALIDATE_AUDIO_FOLDER # pylint: disable=no-member
global VALIDATE_WHISPER_MODEL, VALIDATE_TARGET_LANGUAGE # pylint: disable=no-member
VALIDATE_TRAIN_METADATA_PATH = vp_train_path
VALIDATE_EVAL_METADATA_PATH = vp_eval_path
VALIDATE_AUDIO_FOLDER = vp_audio_folder
VALIDATE_WHISPER_MODEL = vp_whisper_model
VALIDATE_TARGET_LANGUAGE = vp_target_language
def _cleanup_resources(*resources):
"""Clean up resources and free memory"""
for resource in resources:
del resource
gc.collect()
if torch.cuda.is_available():
torch.cuda.empty_cache()
def _write_metadata_files(
wm_train_set,
wm_eval_set,
wm_train_path,
wm_eval_path):
"""Write metadata files with error handling"""
try:
wm_train_set.sort_values("audio_file").to_csv(
wm_train_path, sep="|", index=False)
wm_eval_set.sort_values("audio_file").to_csv(
wm_eval_path, sep="|", index=False)
debug_print("Successfully wrote metadata files", "DATA_PROCESS")
except Exception as e:
debug_print(
f"Failed to write metadata files: {str(e)}",
"DATA_PROCESS",
is_error=True)
raise
def create_dataset_splits(df, eval_percentage, random_seed=42):
"""Create training and evaluation splits with validation"""
if df.empty:
debug_print(
"No data available for splitting",
"DATA_PROCESS",
is_error=True)
return None
shuffled_df = df.sample(frac=1, random_state=random_seed)
num_val_samples = max(1, int(len(shuffled_df) * eval_percentage))
if num_val_samples >= len(shuffled_df):
debug_print(
"Not enough samples for valid split",
"DATA_PROCESS",
is_error=True)
return None
return (shuffled_df[num_val_samples:],
shuffled_df[:num_val_samples]) # training # eval
def save_audio_segment(
sas_audio,
sas_sr,
sas_start_time,
sas_end_time,
sas_sentence,
sas_audio_file_name_without_ext,
sas_segment_idx,
sas_speaker_name,
sas_audio_folder,
sas_metadata,
sas_max_duration,
_sas_buffer,
sas_too_long_files,
sas_target_language,
):
"""Helper function to save audio segments and update metadata"""
sas_sentence = sas_sentence.strip()
sas_sentence = multilingual_cleaners(sas_sentence, sas_target_language)
sas_audio_file_name = f"{sas_audio_file_name_without_ext}_{str(sas_segment_idx).zfill(8)}.wav"
sas_absolute_path = os.path.join(sas_audio_folder, sas_audio_file_name)
os.makedirs(os.path.dirname(sas_absolute_path), exist_ok=True)
# Extract audio segment
sas_audio_start = int(sas_sr * sas_start_time)
sas_audio_end = int(sas_sr * sas_end_time)
sas_audio_segment = sas_audio[sas_audio_start:sas_audio_end].unsqueeze(0)
# Handle long audio segments
if sas_audio_segment.size(-1) > sas_max_duration * sas_sr:
sas_too_long_files.append(
(sas_audio_file_name, sas_audio_segment.size(-1) / sas_sr))
while sas_audio_segment.size(-1) > sas_max_duration * sas_sr:
sas_split_audio = sas_audio_segment[:, : int(
sas_max_duration * sas_sr)]
sas_audio_segment = sas_audio_segment[:, int(
sas_max_duration * sas_sr):]
sas_split_file_name = f"{sas_audio_file_name_without_ext}_{str(sas_segment_idx).zfill(8)}.wav"
sas_split_relative_path = os.path.join(sas_split_file_name)
sas_split_absolute_path = os.path.normpath(
os.path.join(sas_audio_folder, sas_split_relative_path))
os.makedirs(
os.path.dirname(sas_split_absolute_path),
exist_ok=True)
torchaudio.save(str(sas_split_absolute_path), sas_split_audio, sas_sr)
sas_metadata["audio_file"].append(
f"wavs/{sas_split_relative_path}")
sas_metadata["text"].append(sas_sentence)
sas_metadata["speaker_name"].append(sas_speaker_name)
sas_segment_idx += 1
# Only save if segment is at least 1 second
if sas_audio_segment.size(-1) >= sas_sr:
torchaudio.save(str(sas_absolute_path), sas_audio_segment, sas_sr)
sas_metadata["audio_file"].append(f"wavs/{sas_audio_file_name}")
sas_metadata["text"].append(sas_sentence)
sas_metadata["speaker_name"].append(sas_speaker_name)
def process_transcription_result(
ptr_result,
ptr_audio,
ptr_sr,
ptr_segment_idx,
ptr_audio_file_name_without_ext,
ptr_metadata,
ptr_whisper_words,
ptr_max_duration,
ptr_buffer,
ptr_speaker_name,
ptr_audio_folder,
ptr_too_long_files,
ptr_create_bpe_tokenizer,
ptr_target_language,
):
"""Helper function to process transcription results and save audio segments"""
ptr_i = ptr_segment_idx + 1
ptr_sentence = ""
ptr_sentence_start = None
ptr_first_word = True
ptr_current_words = []
for ptr_segment in ptr_result["segments"]:
if "words" not in ptr_segment:
continue
for ptr_word_info in ptr_segment["words"]:
ptr_word = ptr_word_info.get("word", "").strip()
if not ptr_word:
continue
ptr_start_time = ptr_word_info.get("start", 0)
ptr_end_time = ptr_word_info.get("end", 0)
if ptr_create_bpe_tokenizer:
ptr_whisper_words.append(ptr_word)
if ptr_first_word:
ptr_sentence_start = ptr_start_time
if len(ptr_current_words) == 0:
ptr_sentence_start = max(
ptr_sentence_start - ptr_buffer, 0)
else:
ptr_previous_end = ptr_current_words[-1].get(
"end", 0) if ptr_current_words else 0
ptr_sentence_start = max(
ptr_sentence_start - ptr_buffer,
(ptr_previous_end + ptr_start_time) / 2)
ptr_sentence = ptr_word
ptr_first_word = False
else:
ptr_sentence += " " + ptr_word
ptr_current_words.append(
{"word": ptr_word, "start": ptr_start_time, "end": ptr_end_time})
# Handle sentence splitting and audio saving
if ptr_word[-1] in ["!", ".",
"?"] or (ptr_end_time - ptr_sentence_start) > ptr_max_duration:
save_audio_segment(
ptr_audio,
ptr_sr,
ptr_sentence_start,
ptr_end_time,
ptr_sentence,
ptr_audio_file_name_without_ext,
ptr_i,
ptr_speaker_name,
ptr_audio_folder,
ptr_metadata,
ptr_max_duration,
ptr_buffer,
ptr_too_long_files,
ptr_target_language,
)
ptr_i += 1
ptr_first_word = True
ptr_current_words = []
ptr_sentence = ""
def process_audio_with_vad(wav, sr, vad_model, get_speech_timestamps):
"""
Enhanced VAD processing with better end-of-speech detection
"""
device = "cuda" if torch.cuda.is_available() else "cpu"
wav = wav.to(device)
resampler = T.Resample(sr, 16000).to(device)
wav_16k = resampler(wav)
# Adjusted VAD parameters
vad_segments = get_speech_timestamps(
wav_16k,
vad_model,
sampling_rate=16000,
threshold=0.2, # Lower threshold to be more sensitive to speech
min_speech_duration_ms=200, # Shorter to catch brief utterances
max_speech_duration_s=float("inf"),
min_silence_duration_ms=300, # Shorter silence duration
window_size_samples=1024, # Smaller window for more precise detection
speech_pad_ms=300, # Add padding to end of speech segments
)
# Scale timestamps back to original sample rate
scale_factor = sr / 16000
for segment in vad_segments:
segment["start"] = int(segment["start"] * scale_factor)
# Add extra padding at the end
segment["end"] = int(segment["end"] * scale_factor) + \
int(0.2 * sr) # Add 200ms padding
merged_segments = merge_short_segments(
vad_segments, min_duration=6.0, max_gap=0.5)
debug_print(
f"VAD processing: {len(vad_segments)} original segments, {len(merged_segments)} after merging",
"SEGMENTS",
)
return merged_segments
def handle_duplicates(
duplicate_files,
dup_audio_folder,
dup_target_language,
dup_whisper_model):
"""Re-transcribe duplicate files to get best transcription"""
debug_print(
"Re-transcribing duplicate files to get best transcription",
"DUPLICATES")
best_transcriptions = {}
device = "cuda" if torch.cuda.is_available() else "cpu"
asr_model = whisper.load_model(dup_whisper_model, device=device)
for file_path in duplicate_files:
debug_print(f"Re-transcribing {file_path}", "DUPLICATES")
# Get full path
full_path = os.path.join(
dup_audio_folder, os.path.basename(
file_path.replace(
"wavs/", "")))
# Re-transcribe with highest quality settings
result = asr_model.transcribe(
full_path,
language=dup_target_language,
word_timestamps=True,
verbose=None)
# Store the new transcription
best_transcriptions[file_path] = {
"text": result["text"].strip(),
"confidence": sum(s.get("confidence", 0) for s in result["segments"])
/ len(result["segments"]),
}
return best_transcriptions
#############################################
#### STEP 1 Dataset Validation Functions ####
#############################################
def normalize_text(text):
"""
Normalizes text by converting to lowercase, removing punctuation, converting written numbers
to digits, and standardizing whitespace.
"""
# Convert text to lowercase
text = text.lower()
# Remove punctuation
text = text.translate(str.maketrans("", "", string.punctuation))
# Replace multiple spaces with a single space
text = re.sub(r"\s+", " ", text)
# Convert written numbers to digits
words = text.split()
normalized_words = []
for word in words:
try:
# Try to convert word to a number
normalized_word = str(w2n.word_to_num(word))
except ValueError:
# If it fails, keep the original word
normalized_word = word
normalized_words.append(normalized_word)
return " ".join(normalized_words)
def get_audio_file_list(mismatches):
"""Gets list of audio file paths from mismatched transcriptions DataFrame."""
if mismatches.empty:
return ["No bad transcriptions"]
return mismatches["Audio Path"].tolist()
def load_and_display_mismatches():
"""
Loads training/eval metadata CSVs, validates transcriptions against Whisper,
and displays mismatches in Gradio interface.
Returns:
tuple: (
mismatches_df: Full DataFrame with mismatch details,
display_df: DataFrame with visible columns,
message: Status message
)
"""
def validate_audio_transcriptions(
vat_csv_paths,
vat_audio_folder,
vat_whisper_model,
vat_target_language,
vat_progress=None):
# Load and combine metadata from CSV files
metadata_dfs = []
for csv_path in vat_csv_paths:
debug_print(f"Reading CSV file: {csv_path}", "VALIDATION", is_info=True)
metadata_df = pd.read_csv(csv_path, sep="|")
debug_print(f"CSV columns: {metadata_df.columns.tolist()}", "VALIDATION", is_info=True)
debug_print(f"Number of rows: {len(metadata_df)}", "VALIDATION", is_info=True)
# Add source CSV tracking
metadata_df["source_csv"] = csv_path
metadata_df["row_index"] = metadata_df.index
metadata_dfs.append(metadata_df)
metadata_df = pd.concat(metadata_dfs, ignore_index=True)
debug_print(f"Total combined rows: {len(metadata_df)}", "VALIDATION", is_info=True)
# Load Whisper model
device = "cuda" if torch.cuda.is_available() else "cpu"
asr_model = whisper.load_model(vat_whisper_model, device=device)
mismatches = []
missing_files = []
total_files = metadata_df.shape[0]
if vat_progress is not None:
vat_progress((0, total_files), desc="Processing files")
for index, row in tqdm(metadata_df.iterrows(
), total=total_files, unit="file", disable=False, leave=True):
audio_file = row["audio_file"]
expected_text = row["text"]
debug_print(f"Processing file {index + 1}/{total_files}: {audio_file}", "VALIDATION", is_info=True)
debug_print(f"Expected text length: {len(expected_text)}", "VALIDATION", is_info=True)
audio_file_name = audio_file.replace("wavs/", "")
audio_path = os.path.normpath(
os.path.join(
vat_audio_folder,
audio_file_name))
if not os.path.exists(audio_path):
missing_files.append(audio_file_name)
debug_print(f"File not found: {audio_path}", "GENERAL", is_warning=True)
if vat_progress is not None:
vat_progress((index + 1, total_files),
desc="Processing files")
continue
# Transcribe with OpenAI Whisper
result = asr_model.transcribe(
audio_path,
language=vat_target_language,
word_timestamps=True,
verbose=None)
# Get the full transcription from the result
transcribed_text = result["text"].strip()
debug_print(f"Transcribed text length: {len(transcribed_text)}", "VALIDATION", is_info=True)
# Normalize and compare texts
normalized_expected_text = normalize_text(expected_text)
normalized_transcribed_text = normalize_text(transcribed_text)
debug_print(f"Normalized expected text length: {len(normalized_expected_text)}", "VALIDATION", is_info=True)
debug_print(f"Normalized transcribed text length: {len(normalized_transcribed_text)}", "VALIDATION", is_info=True)
if normalized_transcribed_text != normalized_expected_text:
debug_print("Mismatch found! Adding to mismatches list", "VALIDATION", is_info=True)
mismatch_entry = {
"expected_text": row["text"],
"transcribed_text": transcribed_text,
"filename": audio_file_name,
"full_path": audio_path,
"row_index": row["row_index"],
"source_csv": row["source_csv"],
}
debug_print(f"Mismatch entry keys: {mismatch_entry.keys()}", "VALIDATION", is_info=True)
mismatches.append(mismatch_entry)
if vat_progress is not None:
vat_progress((index + 1, total_files), desc="Processing files")
debug_print(f"Total mismatches found: {len(mismatches)}", "GENERAL", is_info=True)
if mismatches:
debug_print("Sample mismatch entry:", "VALIDATION", is_info=True)
debug_print(str(mismatches[0]), "VALIDATION", is_info=True)
if missing_files:
debug_print("Missing files:", "GENERAL", is_warning=True)
for file_name in missing_files:
debug_print(f"- {file_name}", "GENERAL", is_warning=True)
if missing_files:
debug_print("", "GENERAL")
debug_print(
"The following files are missing and should be removed from the CSV files:",
"GENERAL")
for file_name in missing_files:
debug_print(f"- {file_name}", "GENERAL")
return mismatches
vat_progress = gr.Progress(track_tqdm=True)
if (
VALIDATE_TRAIN_METADATA_PATH
and VALIDATE_EVAL_METADATA_PATH
and VALIDATE_AUDIO_FOLDER
and VALIDATE_WHISPER_MODEL
and VALIDATE_TARGET_LANGUAGE
):
mismatches = validate_audio_transcriptions(
[VALIDATE_TRAIN_METADATA_PATH, VALIDATE_EVAL_METADATA_PATH],
VALIDATE_AUDIO_FOLDER,
VALIDATE_WHISPER_MODEL,
VALIDATE_TARGET_LANGUAGE,
vat_progress,
)
if not mismatches:
debug_print("No transcription mismatches found!", "GENERAL", is_info=True)
empty_df = pd.DataFrame(columns=["expected_text", "transcribed_text", "filename",
"full_path", "row_index", "source_csv"])
display_df = pd.DataFrame(columns=["expected_text", "transcribed_text", "filename"])
display_df.loc[0] = ["No bad transcriptions", "No bad transcriptions", "N/A"]
return empty_df, display_df, "No transcription mismatches found - all transcriptions match!"
# Convert mismatches list to DataFrame
df = pd.DataFrame(mismatches)
# Ensure all fields are single values, not series
for col in df.columns:
if isinstance(df[col].iloc[0], pd.Series):
df[col] = df[col].apply(
lambda x: x.iloc[0] if isinstance(
x, pd.Series) else x)
# Clean all text columns
df["expected_text"] = df["expected_text"].astype(
str).apply(lambda x: x.strip())
df["transcribed_text"] = df["transcribed_text"].astype(
str).apply(lambda x: x.strip())
df["full_path"] = df["full_path"].astype(
str).apply(lambda x: x.strip())
# Create display version with only visible columns
display_df = df[["expected_text",
"transcribed_text", "filename"]].copy()
return df, display_df, ""
else:
empty_df = pd.DataFrame(
columns=[
"Expected Text",
"Transcribed Text",
"Filename"])
return empty_df, empty_df, "Please generate your dataset first"
def save_correction_to_csv(csv_path, row_index, new_text):
"""Update the relevant parts of the CSV files"""
try:
# Ensure we have single values, not Series
if isinstance(csv_path, pd.Series):
csv_path = str(csv_path.iloc[0])
else:
csv_path = str(csv_path)
if isinstance(row_index, pd.Series):
row_index = int(row_index.iloc[0])
else:
row_index = int(row_index)
if isinstance(new_text, pd.Series):
new_text = str(new_text.iloc[0])
else:
new_text = str(new_text)
# Read the CSV file
df = pd.read_csv(csv_path, sep="|")
# Update the text
df.loc[row_index, "text"] = new_text
# Save back to CSV
df.to_csv(csv_path, sep="|", index=False)
# Verify the save
df_check = pd.read_csv(csv_path, sep="|")
if not df_check.loc[row_index, "text"] == new_text:
debug_print(
"Save verification failed. Text mismatch.",
"GENERAL",
is_error=True)
return "Error: Save verification failed"
return f"Successfully updated transcription in {os.path.basename(csv_path)}"
except Exception as e:
debug_print(
f"Error saving correction: {str(e)}",
"GENERAL",
is_error=True)
debug_print(f"CSV path: {csv_path}", "GENERAL", is_error=True)
debug_print(f"Row index: {row_index}", "GENERAL", is_error=True)
debug_print("Full error traceback:", "GENERAL", is_error=True)
traceback.print_exc()
return f"Error updating CSV: {str(e)}"
def save_audio_and_correction(
choice,
manual_text,
audio_data,
df,
current_idx):
"""Handle both audio and transcription saves"""
if current_idx is None:
return {
mismatch_table: df[["expected_text", "transcribed_text", "filename"]],
current_expected: "",
save_status: "Please select a row first",
audio_player: None,
}
try:
# Handle current_idx coming as a list from Gradio
if isinstance(current_idx, list):
current_idx = current_idx[0]
row = df.iloc[int(current_idx)]
audio_path = str(row["full_path"]).strip()
debug_print(f"Processing file: {audio_path}", "DATA_PROCESS")
save_status_msg = []
# Handle audio save if audio was edited
if audio_data is not None and isinstance(
audio_data, tuple) and len(audio_data) == 2:
try:
sr, audio = audio_data
debug_print(
f"Saving edited audio: {sr}Hz, length: {len(audio)}",
"DATA_PROCESS")
audio_tensor = torch.tensor(audio).unsqueeze(0)
torchaudio.save(str(audio_path), audio_tensor, sr)
save_status_msg.append("Audio saved successfully")
debug_print(
f"Saved edited audio to {audio_path}",
"DATA_PROCESS")
except Exception as e:
save_status_msg.append(f"Error saving audio: {str(e)}")
debug_print(
f"Error saving audio: {str(e)}",
"DATA_PROCESS",
is_error=True)
# Handle text correction
if choice == "Use Original":
new_text = str(row["expected_text"])
elif choice == "Use Whisper":
new_text = str(row["transcribed_text"])
elif choice == "Edit Manually":
new_text = str(manual_text)
# Save text correction to CSV
result = save_correction_to_csv(
str(row["source_csv"]), int(row["row_index"]), new_text)
save_status_msg.append(result)
# Update both text and expected_text in DataFrame
if "Successfully" in result:
df.loc[current_idx, "text"] = new_text
df.loc[current_idx, "expected_text"] = new_text
debug_print(
f"Updated DataFrame with new text: {new_text}",
"DATA_PROCESS")
# Create updated display DataFrame
display_df = df[["expected_text",
"transcribed_text", "filename"]].copy()
return {
mismatch_table: display_df,
current_expected: new_text if "Successfully" in result else row["expected_text"],
save_status: " | ".join(save_status_msg),
audio_player: audio_path,
}
except Exception as e:
error_msg = f"Error saving correction: {str(e)}"
debug_print(error_msg, "DATA_PROCESS", is_error=True)
traceback.print_exc()
return {
mismatch_table: df[["expected_text", "transcribed_text", "filename"]],
current_expected: row["expected_text"] if "row" in locals() else "",
save_status: error_msg,
audio_player: None,
}
#########################################
#### STEP 2 Model Training Functions ####
#########################################
def basemodel_or_finetunedmodel_choice(value):
"""update basemodel"""
global basemodel_or_finetunedmodel
if value == "Base Model":
basemodel_or_finetunedmodel = True
elif value == "Existing finetuned model":
basemodel_or_finetunedmodel = False
def check_model_requirements(model_folder):
"""Check if all required files exist in the model folder"""
required_files = {
"model.pth": False,
"config.json": False,
"vocab.json": False,
"dvae.pth": False,
"mel_stats.pth": False,
"speakers_xtts.pth": False
}
if model_folder.exists():
for file in required_files:
required_files[file] = (model_folder / file).exists()
return required_files
def train_gpt(
language,
num_epochs,
batch_size,
grad_acumm,
train_csv,
eval_csv,
learning_rate,
model_to_train,
continue_run,
disable_shared_memory,
learning_rate_scheduler,
optimizer,
num_workers,
warm_up,
max_audio_length=255995,
progress=gr.Progress()):
# First check if a model was selected
if "No Models Available" in model_to_train:
debug_print("No XTTS model selected for training.", "MODEL_OPS", is_error=True)
debug_print("Please download a model using AllTalk's main interface > TTS Engine Settings > XTTS > Model/Voices Download", "MODEL_OPS", is_info=True)
return
# Check if selected model exists and has required files
model_path = this_dir / "models" / "xtts" / model_to_train
if not model_path.exists():
debug_print(f"Selected model folder not found: {model_path}", "MODEL_OPS", is_error=True)
debug_print("Please ensure you have downloaded the model correctly", "MODEL_OPS", is_info=True)
return
# Check for required files
files = check_model_requirements(model_path)
missing_files = [file for file, exists in files.items() if not exists]
if missing_files:
debug_print(f"Missing required files in {model_to_train}:", "MODEL_OPS", is_error=True)
for file in missing_files:
debug_print(f"❌ {file}", "MODEL_OPS", is_error=True)
debug_print("\nPlease redownload the model using AllTalk's interface", "MODEL_OPS", is_info=True)
return
# Confirm all model files found and continue with training
debug_print(f"✓ All required files found for model: {model_to_train}", "MODEL_OPS", is_info=True)
# Dataset validation
if not train_csv or not eval_csv:
debug_print("Missing training or evaluation CSV files", "MODEL_OPS", is_error=True)
debug_print(f"Train CSV: {train_csv}", "MODEL_OPS", is_error=True)
debug_print(f"Eval CSV: {eval_csv}", "MODEL_OPS", is_error=True)
return
# Training parameter validation
debug_print("***********************", "MODEL_OPS", is_info=True)
debug_print("Training Configuration:", "MODEL_OPS", is_info=True)
debug_print("***********************", "MODEL_OPS", is_info=True)
debug_print(f"- Language: {language}", "MODEL_OPS", is_info=True)
debug_print(f"- Epochs: {num_epochs}", "MODEL_OPS", is_info=True)
debug_print(f"- Batch Size: {batch_size}", "MODEL_OPS", is_info=True)
debug_print(f"- Gradient Accumulation: {grad_acumm}", "MODEL_OPS", is_info=True)
debug_print(f"- Learning Rate: {learning_rate}", "MODEL_OPS", is_info=True)
debug_print(f"- Learning Rate Scheduler: {learning_rate_scheduler}", "MODEL_OPS", is_info=True)
debug_print(f"- Optimizer: {optimizer}", "MODEL_OPS", is_info=True)
debug_print(f"- Number of Workers: {num_workers}", "MODEL_OPS", is_info=True)
debug_print(f"- Warm Up: {warm_up}", "MODEL_OPS", is_info=True)
debug_print(f"- Max Audio Length: {max_audio_length}", "MODEL_OPS", is_info=True)
# GPU/Memory information
if torch.cuda.is_available():
gpu_id = torch.cuda.current_device()
debug_print("****************", "GPU_MEMORY", is_info=True)
debug_print("GPU Information:", "GPU_MEMORY", is_info=True)
debug_print("****************", "GPU_MEMORY", is_info=True)
debug_print(f"- Device: {torch.cuda.get_device_name(gpu_id)}", "GPU_MEMORY", is_info=True)
debug_print(f"- CUDA Version: {torch.version.cuda}", "GPU_MEMORY", is_info=True)
total_memory = torch.cuda.get_device_properties(gpu_id).total_memory / (1024**3)
free_memory = total_memory - (torch.cuda.memory_allocated(gpu_id) / (1024**3))
debug_print(f"- Total VRAM: {total_memory:.2f}GB", "GPU_MEMORY", is_info=True)
debug_print(f"- Free VRAM: {free_memory:.2f}GB", "GPU_MEMORY", is_info=True)
# Memory warnings
if free_memory < 3:
debug_print("WARNING: Very low available VRAM!", "GPU_MEMORY", is_warning=True)
debug_print("Consider reducing batch size or using gradient accumulation", "GPU_MEMORY", is_warning=True)
else:
debug_print("No GPU detected - training will be very slow!", "GPU_MEMORY", is_warning=True)
if torch.cuda.is_available():
torch.cuda.empty_cache()
# Get the current device ID
gpu_device_id = torch.cuda.current_device()
gpu_available_mem_gb = (torch.cuda.get_device_properties(
gpu_device_id).total_memory - torch.cuda.memory_allocated(gpu_device_id)) / (1024 ** 3)
if gpu_available_mem_gb < 12:
debug_print(
"******************************",
level="GPU_MEMORY",
is_warning=True)
debug_print(
"IMPORTANT MEMORY CONSIDERATION",
level="GPU_MEMORY",
is_warning=True)
debug_print(
"******************************",
level="GPU_MEMORY",
is_warning=True)
debug_print(
"Your available VRAM is below the recommended 12GB threshold.",
level="GPU_MEMORY",
is_warning=True)
# Empty line for formatting
debug_print("", level="GPU_MEMORY", is_warning=True)
debug_print(
"System-Specific Considerations:",
level="GPU_MEMORY",
is_warning=True)
debug_print(
"- Windows: Will utilize system RAM as extended VRAM",
level="GPU_MEMORY",
is_warning=True)
debug_print(
" * Ensure sufficient system RAM is available",
level="GPU_MEMORY",
is_warning=True)
debug_print(
" * Recommended minimum: 24GB system RAM",
level="GPU_MEMORY",
is_warning=True)
debug_print(
"- Linux: Limited to physical VRAM only",
level="GPU_MEMORY",
is_warning=True)
debug_print(
" * Training may fail with insufficient VRAM",
level="GPU_MEMORY",
is_warning=True)
debug_print(
" * Consider reducing batch size or using gradient accumulation",
level="GPU_MEMORY",
is_warning=True)
# Empty line for formatting
debug_print("", level="GPU_MEMORY", is_warning=True)
debug_print(
"For detailed memory management strategies and optimization tips:",
level="GPU_MEMORY",
is_warning=True)
debug_print(
"1. Refer to the 'Memory Management' section in the Training Guide",
level="GPU_MEMORY",
is_warning=True)
debug_print(
"2. Review the Pre-flight Check tab for system requirements",
level="GPU_MEMORY",
is_warning=True)
# Dataset statistics
try:
train_df = pd.read_csv(train_csv, sep="|")
eval_df = pd.read_csv(eval_csv, sep="|")
debug_print("*******************", "DATA_PROCESS", is_info=True)
debug_print("Dataset Statistics:", "DATA_PROCESS", is_info=True)
debug_print("*******************", "DATA_PROCESS", is_info=True)
debug_print(f"- Training samples: {len(train_df)}", "DATA_PROCESS", is_info=True)
debug_print(f"- Evaluation samples: {len(eval_df)}", "DATA_PROCESS", is_info=True)
if (out_path / "bpe_tokenizer-vocab.json").exists():
debug_print(
"- Using custom BPE tokenizer",
level="DATA_PROCESS",
is_info=True)
training_assets = {
'Tokenizer': str(out_path / "bpe_tokenizer-vocab.json")
}
# Check for potential issues
if len(train_df) < 100:
debug_print("Very small training dataset", "DATA_PROCESS", is_warning=True)
if len(eval_df) < 10:
debug_print("Very small evaluation dataset", "DATA_PROCESS", is_warning=True)
except Exception as e:
debug_print(f"Error reading dataset files: {str(e)}", "DATA_PROCESS", is_error=True)
return
# Logging parameters
project_run_name = "XTTS_FT"
project_name = "XTTS_trainer"
dashboard_logger = "tensorboard"
logger_uri = None
# Check for lang.txt in the same directory as train_csv
dataset_dir = os.path.dirname(train_csv)
lang_file = os.path.join(dataset_dir, "lang.txt")
# Set here the path that the checkpoints will be saved. Default:
# ./training/
project_path = os.path.join(out_path, "training")
# Project path information
debug_print("**********************", "GENERAL", is_info=True)
debug_print("Project Configuration:", "GENERAL", is_info=True)
debug_print("**********************", "GENERAL", is_info=True)
debug_print(f"- Project Path: {project_path}", "GENERAL", is_info=True)
debug_print(f"- Model Path: {model_path}", "GENERAL", is_info=True)
debug_print(f"- Training Data: {train_csv}", level="GENERAL", is_info=True)
debug_print(
f"- Evaluation Data: {eval_csv}",
level="GENERAL",
is_info=True)
debug_print(f"- Language: {language}", level="GENERAL", is_info=True)
if os.path.exists(lang_file):
try:
with open(lang_file, 'r', encoding='utf-8') as f:
dataset_language = f.read().strip()
debug_print(f"- Found language file, using language: {dataset_language}", "GENERAL", is_info=True)
# Override the input language with the one from lang.txt
language = dataset_language
except Exception as e:
debug_print(f"- Error reading lang.txt: {str(e)}", "GENERAL", is_warning=True)
debug_print(f"- Falling back to provided language: {language}", "GENERAL", is_warning=True)
else:
debug_print("- No lang.txt found, using provided language setting", "GENERAL", is_warning=True)
debug_print(f"- Batch Size: {batch_size}", level="GENERAL", is_info=True)
debug_print(
f"- Grad Steps: {grad_acumm}",
level="GENERAL",
is_info=True)
debug_print(
f"- Training Epochs: {num_epochs}",
level="GENERAL",
is_info=True)
# Create the directory
os.makedirs(project_path, exist_ok=True)
# Training Parameters
# For multi-GPU training, set to False
param_optimizer_wd_only_on_weights = True
# If True, it will start with evaluation
param_start_with_eval = False
param_batch_size = batch_size # Set the batch size here
# Set the gradient accumulation steps here
param_grad_acumm_steps = grad_acumm
# Define here the dataset that you want to use for the fine-tuning on.
config_dataset = BaseDatasetConfig(
formatter="coqui",
dataset_name="ft_dataset",
path=os.path.dirname(train_csv),
meta_file_train=train_csv,
meta_file_val=eval_csv,
language=language,
)
# Dataset Configuration list
dataset_config_list = [config_dataset]
dataset_tokenizer_file = str(model_path / "vocab.json")
dataset_xtts_checkpoint = str(model_path / "model.pth")
dataset_xtts_config_file = str(model_path / "config.json")
dataset_dvae_checkpoint = model_path / "dvae.pth"
dataset_mel_norm_file = model_path / "mel_stats.pth"
dataset_speakers_file = model_path / "speakers_xtts.pth"
training_assets = None
continue_path = None
if continue_run:
folders = glob.glob(os.path.join(project_path, '*/'))
if folders:
last_run = max(folders, key=os.path.getmtime)
if last_run:
checkpoints = glob.glob(
os.path.join(
last_run,
"best_model_*.pth"))
if checkpoints:
latest_checkpoint = max(checkpoints, key=os.path.getmtime)
if latest_checkpoint:
dataset_xtts_checkpoint = None
continue_path = last_run
print(
f"[FINETUNE] - Continuing previous fine tuning {latest_checkpoint}")
# Copy the supporting files
destination_dir = out_path / "chkptandnorm"
destination_dir.mkdir(parents=True, exist_ok=True)
shutil.copy2(
dataset_dvae_checkpoint,
destination_dir /
dataset_dvae_checkpoint.name)
shutil.copy2(
dataset_mel_norm_file,
destination_dir /
dataset_mel_norm_file.name)
shutil.copy2(
dataset_speakers_file,
destination_dir /
dataset_speakers_file.name)
# init args and config
model_args = GPTArgs(
max_conditioning_length=132300, # 6 secs
min_conditioning_length=66150, # 3 secs
debug_loading_failures=False,
max_wav_length=max_audio_length, # ~11.6 seconds
max_text_length=200,
mel_norm_file=dataset_mel_norm_file,
dvae_checkpoint=dataset_dvae_checkpoint,
xtts_checkpoint=dataset_xtts_checkpoint,
# checkpoint path of the model that you want to fine-tune
tokenizer_file=dataset_tokenizer_file,
gpt_num_audio_tokens=1026,
gpt_start_audio_token=1024,
gpt_stop_audio_token=1025,
gpt_use_masking_gt_prompt_approach=True,
gpt_use_perceiver_resampler=True,
)
# define audio config
audio_config = XttsAudioConfig(
sample_rate=22050,
dvae_sample_rate=22050,
output_sample_rate=24000)
# Resolve Japanese threading issue
number_of_workers = int(num_workers)
if language == "ja":
number_of_workers = 0
lr_scheduler = None
lr_scheduler_params = {}
if learning_rate_scheduler and learning_rate_scheduler != "None":
lr_gamma_mapping = {
1e-6: 0.9,
5e-6: 0.8,
1e-5: 0.3,
5e-5: 0.3,
1e-4: 0.3,
5e-4: 0.1,
1e-3: 0.1
}
lr_scheduler = learning_rate_scheduler
if lr_scheduler == "StepLR":
lr_scheduler_params = {
'step_size': 30,
'gamma': 0.1,
'last_epoch': -1}
elif lr_scheduler == "MultiStepLR":
exponent = 3 - int(math.log2(num_epochs) / 2)
base = 2
num_milestones = min(num_epochs, int(math.pow(base, exponent)))
milestone_interval = num_epochs // (num_milestones + 1)
milestones = [milestone_interval *
(i + 1) for i in range(num_milestones)]
lr_scheduler_params = {
'milestones': milestones,
'gamma': lr_gamma_mapping[learning_rate],
'last_epoch': -1}
elif lr_scheduler == "ExponentialLR":
lr_scheduler_params = {'gamma': 0.5, 'last_epoch': -1}
elif lr_scheduler == "CosineAnnealingLR":
lr_scheduler_params = {
'T_max': num_epochs,
'eta_min': 1e-6,
'last_epoch': -1}
elif lr_scheduler == "ReduceLROnPlateau":
lr_scheduler_params = {
'mode': 'min',
'factor': 0.8,
'patience': 1,
'threshold': 0.0001,
'threshold_mode': 'rel',
'cooldown': 0,
'min_lr': 1e-8,
'eps': 1e-08,
}
elif lr_scheduler == "CyclicLR":
lr_scheduler_params = {
'base_lr': learning_rate,
'max_lr': 0.1,
'step_size_up': 2000,
'step_size_down': None,
'mode': 'triangular',
'gamma': 1.0,
'scale_fn': None,
'scale_mode': 'cycle',
'cycle_momentum': True,
'base_momentum': 0.8,
'max_momentum': 0.9,
'last_epoch': -1}
elif lr_scheduler == "OneCycleLR":
lr_scheduler_params = {
'max_lr': learning_rate,
'total_steps': None,
'epochs_up': None,
'steps_per_epoch': None,
'anneal_strategy': 'cos',
'cycle_momentum': True,
'base_momentum': 0.85,
'max_momentum': 0.95,
'div_factor': 25.0,
'final_div_factor': 10000.0,
'last_epoch': -1}
elif lr_scheduler == "CosineAnnealingWarmRestarts":
if num_epochs < 4:
error_message = "For Cosine Annealing Warm Restarts, epochs must be at least 4. Please set a minimum of 4 epochs."
progress(1.0, desc=f"Error: {error_message}")
raise ValueError(error_message)
# Set 4 learning rate restarts
lr_scheduler_params = {
'T_0': int(
num_epochs / 4),
'T_mult': 1,
'eta_min': 1e-6,
'last_epoch': -1}
optimizer_params = None
OPTIMIZER_PARAMS = {
"AdamW": {
"betas": [0.9, 0.96],
"eps": 1e-8,
"weight_decay": 1e-2
},
"RMSprop": {
"alpha": 0.99,
"eps": 1e-8,
"weight_decay": 1e-4
},
"SGD": {
"momentum": 0.9,
"weight_decay": 1e-4
},
"Adam": {
"betas": [0.9, 0.999],
"eps": 1e-8,
"weight_decay": 1e-4
},
"Adagrad": {
"lr_decay": 0,
"weight_decay": 1e-4,
"eps": 1e-10
},
"RAdam": {
"betas": [0.9, 0.999],
"eps": 1e-8,
"weight_decay": 1e-2
},
"stepwisegraduallr": {},
"noamlr": {}
}
optimizer_params = OPTIMIZER_PARAMS.get(optimizer, {})
debug_print(
f"- Learning Scheduler {lr_scheduler} Parameters",
level="GENERAL",
is_info=True)
debug_print(
f"- {lr_scheduler_params}",
level="GENERAL",
is_info=True)
# training parameters config
config = GPTTrainerConfig(
epochs=num_epochs,
output_path=project_path,
model_args=model_args,
run_name=project_run_name,
project_name=project_name,
run_description="GPT XTTS training",
dashboard_logger=dashboard_logger,
logger_uri=logger_uri,
audio=audio_config,
batch_size=param_batch_size,
batch_group_size=48,
eval_batch_size=param_batch_size,
num_loader_workers=number_of_workers,
eval_split_max_size=256,
print_step=50,
plot_step=100,
log_model_step=100,
save_step=1000,
save_n_checkpoints=1,
save_checkpoints=True,
# target_loss="loss",
print_eval=False,
# Optimizer values like tortoise, pytorch implementation with
# modifications to not apply WD to non-weight parameters.
optimizer=optimizer,
optimizer_wd_only_on_weights=param_optimizer_wd_only_on_weights,
optimizer_params=optimizer_params,
lr=learning_rate, # learning rate
lr_scheduler=lr_scheduler,
# it was adjusted accordly for the new step scheme
lr_scheduler_params=lr_scheduler_params,
test_sentences=[],
)
progress(0, desc="Model is currently training. See console for more information")
# init the model from config
model = GPTTrainer.init_from_config(config)
# load training samples
debug_print("Loading training samples...", "MODEL_OPS", is_info=True)
train_samples, eval_samples = load_tts_samples(
dataset_config_list,
eval_split=True,
eval_split_max_size=config.eval_split_max_size,
eval_split_size=config.eval_split_size,
)
debug_print(f"Loaded {len(train_samples)} training and {len(eval_samples)} eval samples\n", "MODEL_OPS", is_info=True)
global c_logger
c_logger = MetricsLogger()
# init the trainer
trainer = Trainer(
TrainerArgs(
restore_path=None, # xtts checkpoint is restored via xtts_checkpoint key so no need of restore it using Trainer restore_path parameter
skip_train_epoch=False,
continue_path=continue_path,
start_with_eval=param_start_with_eval,
grad_accum_steps=param_grad_acumm_steps,
),
config,
output_path=project_path,
model=model,
train_samples=train_samples,
eval_samples=eval_samples,
training_assets=training_assets,
c_logger=c_logger,
warmup=warm_up,
)
c_logger.update_model_path(project_path)
if disable_shared_memory:
# Limit training to GPU memory instead of shared memory
debug_print("Limiting GPU memory to 95%", "GPU_MEMORY", is_info=True)
torch.cuda.set_per_process_memory_fraction(0.95)
print("\n")
debug_print("********************************", "MODEL_OPS", is_info=True)
debug_print("Starting training the XTTS model", "MODEL_OPS", is_info=True)
debug_print("********************************\n", "MODEL_OPS", is_info=True)
trainer.fit()
print("\n")
debug_print("********************************", "MODEL_OPS", is_info=True)
debug_print("Training completed successfully", "MODEL_OPS", is_info=True)
debug_print("********************************", "MODEL_OPS", is_info=True)
# get the longest text audio file to use as speaker reference
samples_len = [len(item["text"].split(" ")) for item in train_samples]
longest_text_idx = samples_len.index(max(samples_len))
speaker_ref = train_samples[longest_text_idx]["audio_file"]
trainer_out_path = trainer.output_path
# Before cleanup
debug_print("GPU Memory Before Cleanup:", "GPU_MEMORY")
get_gpu_memory()
# deallocate VRAM and RAM
del model, trainer, train_samples, eval_samples, config, model_args, config_dataset
gc.collect()
train_samples = None
eval_samples = None
config_dataset = None
trainer = None
model = None
model_args = None
try:
return dataset_xtts_config_file, dataset_xtts_checkpoint, dataset_tokenizer_file, trainer_out_path, speaker_ref
except Exception as e:
debug_print(f"Error during training: {str(e)}", "MODEL_OPS", is_error=True)
debug_print(traceback.format_exc(), "MODEL_OPS", is_error=True)
return
##########################
#### STEP 3 AND OTHER ####
##########################
def clear_gpu_cache():
"""clear the GPU cache"""
if torch.cuda.is_available():
torch.cuda.empty_cache()
debug_print("GPU Memory After Cleanup:", "GPU_MEMORY")
get_gpu_memory()
def find_a_speaker_file(folder_path):
"""locate a speakers_xtts.pth file"""
search_path = folder_path / "*" / "speakers_xtts.pth"
files = glob.glob(str(search_path), recursive=True)
latest_file = max(files, key=os.path.getctime, default=None)
return latest_file
def load_model(xtts_checkpoint, xtts_config, xtts_vocab):
"""Load in the XTTS model for testing"""
global XTTS_MODEL
clear_gpu_cache()
if not all([xtts_checkpoint, xtts_config, xtts_vocab]):
return "No Models were selected. Click the Refresh Dropdowns button and try again."
xtts_speakers_pth = find_a_speaker_file(this_dir / "models")
config = XttsConfig()
config.load_json(xtts_config)
XTTS_MODEL = Xtts.init_from_config(config)
debug_print("Starting Step 3 - Loading XTTS model!", level="GENERAL", is_info=True)
XTTS_MODEL.load_checkpoint(
config,
checkpoint_path=xtts_checkpoint,
vocab_path=xtts_vocab,
use_deepspeed=False,
speaker_file_path=xtts_speakers_pth,
)
if torch.cuda.is_available():
XTTS_MODEL.cuda()
debug_print("Model Loaded!", level="GENERAL", is_info=True)
return "Model Loaded!"
def run_tts(lang, tts_text, speaker_audio_file):
"""Generate the TTS for testing"""
if XTTS_MODEL is None or not speaker_audio_file:
return "You need to run the previous step to load the model !!", None, None
speaker_audio_file = str(speaker_audio_file)
wavs_files = [speaker_audio_file]
if os.path.isdir(speaker_audio_file):
wavs_files = glob.glob(os.path.join(speaker_audio_file, "*.wav"))
speaker_audio_file = wavs_files[0]
gpt_cond_latent, speaker_embedding = XTTS_MODEL.get_conditioning_latents(
audio_path=wavs_files,
gpt_cond_len=XTTS_MODEL.config.gpt_cond_len,
max_ref_length=XTTS_MODEL.config.max_ref_len,
sound_norm_refs=XTTS_MODEL.config.sound_norm_refs,
)
out = XTTS_MODEL.inference(
text=tts_text,
language=lang,
gpt_cond_latent=gpt_cond_latent,
speaker_embedding=speaker_embedding,
temperature=XTTS_MODEL.config.temperature, # Add custom parameters here
length_penalty=XTTS_MODEL.config.length_penalty,
repetition_penalty=XTTS_MODEL.config.repetition_penalty,
top_k=XTTS_MODEL.config.top_k,
top_p=XTTS_MODEL.config.top_p,
)
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as fp:
out["wav"] = torch.tensor(out["wav"]).unsqueeze(0)
out_path = fp.name
torchaudio.save(str(out_path), out["wav"], 24000)
return "Speech generated !", out_path, speaker_audio_file
def get_available_voices(min_duration_seconds=6, speaker_name=None):
"""Get available voice files based on minimum duration."""
directory = this_dir / "finetune" / speaker_name if (speaker_name and speaker_name != "personsname") else out_path
valid_files = []
wav_files = Path(f"{directory}/wavs").glob("*.wav")
for voice_file in wav_files:
try:
waveform, sample_rate = torchaudio.load(str(voice_file))
duration = waveform.size(1) / sample_rate
if duration >= float(min_duration_seconds):
valid_files.append(str(voice_file))
except Exception as e:
debug_print(f"Error processing {voice_file}: {str(e)}", level="GENERAL", is_error=True)
return sorted(valid_files)
def find_best_models(directory, speaker_name=None):
"""Find the best_model.pth file for the correct project name and last training run"""
if speaker_name and speaker_name != "personsname":
directory = this_dir / "finetune" / speaker_name
# Look in both the base directory and the training subdirectory
model_files = []
# Check the training directory first
training_dir = directory / "training"
if training_dir.exists():
# Look in all subdirectories of training
for subdir in training_dir.glob("*"):
if subdir.is_dir():
model_path = subdir / "best_model.pth"
if model_path.exists():
model_files.append(str(model_path))
# Also check the base directory
for model_path in directory.glob("**/best_model.pth"):
if "training" not in str(model_path): # Avoid duplicates
model_files.append(str(model_path))
return sorted(model_files)
def find_jsons(directory, filename, speaker_name=None):
"""Locate JSON's for the correct project name and last training run"""
if speaker_name and speaker_name != "personsname":
directory = this_dir / "finetune" / speaker_name
return [str(file) for file in Path(directory).rglob(filename)]
# XTTS checkpoint files (best_model.pth)
xtts_checkpoint_files = find_best_models(out_path)
# XTTS config files (config.json)
xtts_config_files = find_jsons(out_path, "config.json")
# XTTS vocab files (vocab.json)
xtts_vocab_files = find_jsons(out_path, "vocab.json")
##########################
#### STEP 4 AND OTHER ####
##########################
def compact_custom_model(
xtts_checkpoint_copy,
folder_path,
overwrite_existing):
"""Compact and move all the files for the correct project name and last training run"""
this_dir = Path(__file__).parent.resolve()
# Early validation checks
if not xtts_checkpoint_copy:
error_message = "No trained model was selected. Please click Refresh Dropdowns and try again."
debug_print(error_message, level="GENERAL", is_error=True)
return error_message
target_dir = this_dir / "models" / "xtts" / folder_path
if overwrite_existing == "Do not overwrite existing files" and target_dir.exists():
error_message = "The target folder already exists. Please change folder name or allow overwrites."
debug_print(error_message, level="GENERAL", is_error=True)
return error_message
xtts_checkpoint_copy = Path(xtts_checkpoint_copy)
# Get the source directory (either tmp-trn or custom named directory)
source_dir = xtts_checkpoint_copy.parent.parent.parent # Go up to the base directory
debug_print(
"=== File Copy Operations ===",
level="DATA_PROCESS",
is_info=True)
debug_print(
f"Source base directory: {source_dir}",
level="DATA_PROCESS",
is_info=True)
debug_print(
f"Target directory: {target_dir}",
level="DATA_PROCESS",
is_info=True)
try:
checkpoint = torch.load(
xtts_checkpoint_copy,
map_location=torch.device("cpu"))
except Exception as e:
debug_print(
f"Error loading checkpoint: {e}",
level="GENERAL",
is_error=True)
raise
del checkpoint["optimizer"]
target_dir.mkdir(parents=True, exist_ok=True)
# Remove dvae-related keys from checkpoint
for key in list(checkpoint["model"].keys()):
if "dvae" in key:
del checkpoint["model"][key]
debug_print("Processing model.pth:", level="DATA_PROCESS", is_info=True)
debug_print(
f" From: {xtts_checkpoint_copy}",
level="DATA_PROCESS",
is_info=True)
debug_print(
f" To: {target_dir / 'model.pth'}",
level="DATA_PROCESS",
is_info=True)
torch.save(checkpoint, target_dir / "model.pth")
# Copy first set of files
folder_path_new = xtts_checkpoint_copy.parent
debug_print("Copying config files:", level="DATA_PROCESS", is_info=True)
for file_name in ["vocab.json", "config.json"]:
src_path = folder_path_new / file_name
dest_path = target_dir / file_name
if src_path.exists():
shutil.copy2(src_path, dest_path)
debug_print(f" {file_name}:", level="DATA_PROCESS", is_info=True)
debug_print(
f" From: {src_path}",
level="DATA_PROCESS",
is_info=True)
debug_print(
f" To: {dest_path}",
level="DATA_PROCESS",
is_info=True)
else:
debug_print(
f"Warning: {src_path} not found",
level="DATA_PROCESS",
is_warning=True)
# Copy second set of files from chkptandnorm directory
chkptandnorm_path = source_dir / "chkptandnorm"
debug_print("Copying support files:", level="DATA_PROCESS", is_info=True)
for file_name in ["speakers_xtts.pth", "mel_stats.pth", "dvae.pth"]:
src_path = chkptandnorm_path / file_name
dest_path = target_dir / file_name
if src_path.exists():
shutil.copy2(src_path, dest_path)
debug_print(f" {file_name}:", level="DATA_PROCESS", is_info=True)
debug_print(
f" From: {src_path}",
level="DATA_PROCESS",
is_info=True)
debug_print(
f" To: {dest_path}",
level="DATA_PROCESS",
is_info=True)
else:
debug_print(
f"Warning: {src_path} not found",
level="DATA_PROCESS",
is_warning=True)
# Create directories for different categories of WAV files
target_wavs_dir = target_dir / "wavs"
target_wavs_dir.mkdir(parents=True, exist_ok=True)
too_short_dir = target_wavs_dir / "too_short"
too_long_dir = target_wavs_dir / "too_long"
suitable_dir = target_wavs_dir / "suitable"
for dir_path in [too_short_dir, too_long_dir, suitable_dir]:
dir_path.mkdir(parents=True, exist_ok=True)
# Process WAV files
source_wavs_dir = source_dir / "wavs"
debug_print("Processing WAV files:", level="DATA_PROCESS", is_info=True)
debug_print(
f" From: {source_wavs_dir}",
level="DATA_PROCESS",
is_info=True)
debug_print(f" To: {target_wavs_dir}", level="DATA_PROCESS", is_info=True)
file_stats = {"too_short": [], "too_long": [], "suitable": []}
if not source_wavs_dir.exists():
debug_print(
f"Warning: Source WAV directory {source_wavs_dir} does not exist",
level="DATA_PROCESS",
is_warning=True)
return f"Model files copied to '/models/xtts/{folder_path}/' but no WAV files were found to process"
for file_path in source_wavs_dir.iterdir():
if file_path.is_file() and file_path.suffix.lower() == ".wav":
try:
# Load audio file and get duration
waveform, sample_rate = torchaudio.load(str(file_path))
duration = waveform.size(
1) / sample_rate # Duration in seconds
# Determine category and target directory
if duration < 6:
category = "too_short"
target_subdir = too_short_dir
elif duration > 30:
category = "too_long"
target_subdir = too_long_dir
else:
category = "suitable"
target_subdir = suitable_dir
# Copy file to appropriate directory
shutil.copy2(file_path, target_subdir / file_path.name)
# Store file info
file_stats[category].append(
{"name": file_path.name, "duration": round(duration, 2)}
)
except Exception as e:
print(
f"[FINETUNE] Error processing {file_path.name}: {str(e)}")
debug_print("WAV File Statistics:", level="DATA_PROCESS", is_info=True)
debug_print(
f" Suitable files (6-30s): {len(file_stats['suitable'])}",
level="DATA_PROCESS",
is_info=True)
debug_print(
f" Too short files (<6s): {len(file_stats['too_short'])}",
level="DATA_PROCESS",
is_info=True)
debug_print(
f" Too long files (>30s): {len(file_stats['too_long'])}",
level="DATA_PROCESS",
is_info=True)
debug_print(
"=== File Copy Operations Complete ===",
level="DATA_PROCESS",
is_info=True)
# Create report file
report_content = FinetuneContent.report_content # pylint: disable=no-member
for category, files in file_stats.items():
report_content += f"\n{category.replace('_', ' ').title()} files ({len(files)}):\n"
for file_info in files:
report_content += f"- {file_info['name']}: {file_info['duration']} seconds\n"
report_content += FinetuneContent.report_content2 # pylint: disable=no-member
with open(target_wavs_dir / "audio_report.txt", "w", encoding="utf-8") as f:
f.write(report_content)
# Model & WAV processing log
debug_print(
f"Model & WAV samples processed and copied to '/models/xtts/{folder_path}/'",
level="DATA_PROCESS",
is_info=True)
return f"Model & WAV samples processed and copied to '/models/xtts/{folder_path}/'"
def delete_training_data():
"""Deletes the specified project name folder"""
# Define the folder to be deleted
folder_to_delete = Path(out_path)
# Check if the folder exists before deleting
if not folder_to_delete.exists():
debug_print(
f"Project Name folder > {folder_to_delete} < does not exist.",
level="GENERAL",
is_warning=True)
return "Specified Project Name folder could not be found."
# Iterate over all files and subdirectories
for item in folder_to_delete.iterdir():
# Exclude trainer_0_log.txt from deletion
if item.name != "trainer_0_log.txt":
try:
if item.is_file():
item.unlink()
elif item.is_dir():
shutil.rmtree(item)
except PermissionError:
debug_print(
f"PermissionError: Could not delete {item}. Skipping.",
level="GENERAL",
is_error=True)
debug_print(
f"Project Name folder > {folder_to_delete} < was deleted successfully.",
level="GENERAL",
is_info=True)
return "Specified Project Name folder & tmp data was deleted successfully."
def clear_folder_contents(folder_path):
"""Deletes the contents of the supplied folder"""
if not folder_path.exists() or not folder_path.is_dir():
debug_print(
f"Folder {folder_path} does not exist.",
level="GENERAL",
is_warning=True)
return f"Folder '{folder_path}' does not exist."
# List all files and subdirectories in the folder
for item in os.listdir(folder_path):
item_path = folder_path / item
if item_path.is_file():
# If it's a file, remove it
os.remove(item_path)
elif item_path.is_dir():
# If it's a subdirectory, remove it recursively
shutil.rmtree(item_path)
debug_print(
f"Contents of {folder_path} deleted successfully.",
level="GENERAL",
is_info=True)
return f"Contents of '{folder_path}' deleted successfully."
def delete_voice_sample_contents():
"""Clears out the specfied folders"""
# Define the folders to be cleared
voice_samples_folder = this_dir / "finetune" / "put-voice-samples-in-here"
gradio_temp_folder = this_dir / "finetune" / "gradio_temp"
# Clear the contents of the gradio_temp folder
clear_folder_contents(gradio_temp_folder)
# Clear the contents of the voice samples folder
voice_samples_message = clear_folder_contents(voice_samples_folder)
return voice_samples_message
#######################
#### OTHER Generic ####
#######################
def cleanup_before_exit(_signum, _frame):
"""Handle cleanup operations before exiting the program.""" # pylint: disable=no-member
debug_print(
"Received interrupt signal. Cleaning up and exiting...",
level="GENERAL",
is_warning=True)
# Perform cleanup operations here if necessary
sys.exit(0)
def create_refresh_button(
refresh_components,
refresh_methods,
elem_class,
interactive=True):
"""Create a refresh button with specified components and methods."""
def refresh(speaker_name, min_duration_seconds):
updates = {}
for component, method in zip(refresh_components, refresh_methods):
# Pass both speaker_name and min_duration_seconds to the method
args = (
method(
speaker_name=speaker_name,
min_duration_seconds=min_duration_seconds) if callable(method) else method)
if args and "choices" in args:
args["value"] = args["choices"][-1] if args["choices"] else ""
for k, v in args.items():
setattr(component, k, v)
updates[component] = gr.update(**(args or {}))
return updates
refresh_button = gr.Button(
"Refresh Dropdowns", elem_classes=elem_class, interactive=interactive
)
refresh_button.click(
fn=refresh,
inputs=[speaker_name_input_testing, min_audio_length],
outputs=refresh_components,
)
return refresh_button
def create_refresh_button_next(
refresh_components,
refresh_methods,
elem_class,
interactive=True):
"""Create a refresh button with specified components and methods."""
def refresh_export(speaker_name):
global out_path
if speaker_name and speaker_name != "personsname":
out_path = this_dir / "finetune" / speaker_name
else:
out_path = this_dir / "finetune" / "tmp-trn"
updates = {}
for component, method in zip(refresh_components, refresh_methods):
args = method(speaker_name=speaker_name) if callable(
method) else method
if args and "choices" in args:
args["value"] = args["choices"][-1] if args["choices"] else ""
for k, v in args.items():
setattr(component, k, v)
updates[component] = gr.update(**(args or {}))
return updates
refresh_button = gr.Button(
"Refresh Dropdowns", elem_classes=elem_class, interactive=interactive
)
refresh_button.click(
fn=refresh_export,
inputs=[speaker_name_input_export],
outputs=refresh_components)
return refresh_button
if __name__ == "__main__":
# Register the signal handler
signal.signal(signal.SIGINT, cleanup_before_exit)
################
#### GRADIO ####
################
parser = argparse.ArgumentParser(
description="""XTTS fine-tuning demo\n\n"""
"""
Example runs:
python3 TTS/demos/xtts_ft_demo/xtts_demo.py --port
""",
formatter_class=argparse.RawTextHelpFormatter,
)
parser.add_argument(
"--num_epochs",
type=int,
help="Number of epochs to train. Default: 10",
default=10,
)
parser.add_argument(
"--batch_size",
type=int,
help="Batch size. Default: 4",
default=4,
)
parser.add_argument(
"--grad_acumm",
type=int,
help="Grad accumulation steps. Default: 1",
default=1,
)
parser.add_argument(
"--max_audio_length",
type=int,
help="Max permitted audio size in seconds. Default: 11",
default=11,
)
args = parser.parse_args()
#####################
#### GRADIO INFO ####
#####################
with gr.Blocks(theme=gr.themes.Base(), css=FinetuneContent.custom_css) as demo:
with gr.Row():
gr.Markdown("## XTTS Models Finetuning")
gr.Markdown("")
gr.Markdown("")
dark_mode_btn = gr.Button(
"Light/Dark Mode", variant="primary", size="sm")
dark_mode_btn.click(
None,
None,
None,
js="""() => {
if (document.querySelectorAll('.dark').length) {
document.querySelectorAll('.dark').forEach(el => el.classList.remove('dark'));
// localStorage.setItem('darkMode', 'disabled');
} else {
document.querySelector('body').classList.add('dark');
// localStorage.setItem('darkMode', 'enabled');
}
}""",
show_api=False,
)
with gr.Tab("🚀 Pre-flight Checklist"):
create_pfc_interface()
#######################
#### GRADIO STEP 1 ####
#######################
with gr.Tab("📁 Step 1 - Generating the dataset"):
with gr.Tab("Generate Dataset"):
# Define directories
this_dir = Path(__file__).parent.resolve()
voice_samples_dir = this_dir / "finetune" / "put-voice-samples-in-here"
training_data_dir = this_dir / "finetune" / "tmp-trn" / "wavs"
metadata_files = [
this_dir / "finetune" / "tmp-trn" / "metadata_eval.csv",
this_dir / "finetune" / "tmp-trn" / "metadata_train.csv",
this_dir / "finetune" / "tmp-trn" / "lang.txt",
]
# Ensure directories exist
voice_samples_dir.mkdir(parents=True, exist_ok=True)
training_data_dir.mkdir(parents=True, exist_ok=True)
def upload_audio(files):
"""Upload audio files to the voice samples directory."""
for file in files:
shutil.copy(file.name, voice_samples_dir)
return f"Uploaded {len(files)} files to {voice_samples_dir}"
def delete_existing_audio():
"""Delete all files in the voice samples directory."""
for file in voice_samples_dir.iterdir():
if file.is_file():
file.unlink()
return f"Deleted all files in {voice_samples_dir}"
def delete_existing_training_data():
"""Delete all files from the project name directory."""
if training_data_dir.exists():
for file in training_data_dir.iterdir():
if file.is_file():
file.unlink()
for metadata_file in metadata_files:
if metadata_file.exists():
metadata_file.unlink()
return f"Deleted all files in {training_data_dir} and related metadata files"
with gr.Row():
with gr.Column(scale=1):
audio_files_upload = gr.Files(
label="Upload Audio Files")
with gr.Column(scale=3):
with gr.Row():
with gr.Column(scale=1):
audio_upload_button = gr.Button(
"Upload New Audio Samples")
delete_audio_button = gr.Button(
"Delete Existing Audio Samples")
delete_dataset_button = gr.Button(
"Delete Existing Training Dataset"
)
with gr.Column(scale=2):
gr.Markdown(
"""
You can manually copy your audio files to `/finetune/put-voice-samples-in-here/` or use the upload to the left and click "Upload New Audio Samples". Once you have uploaded files, you can start creating your dataset.
- If you wish to delete previously uploaded audio samples files then use 'Delete Existing Audio Samples'.
- If you wish to delete previously generated training datasets, please use 'Delete Existing Training Dataset'.
- If you wish to re-use your previously created training data, fill in the `Training Project Name` corectly and click 'Create Dataset'.
"""
)
with gr.Row():
output_text = gr.Textbox(
label="Audio File Management Result", interactive=False)
# Define actions for buttons
audio_upload_button.click(
upload_audio,
inputs=audio_files_upload,
outputs=output_text)
delete_audio_button.click(
delete_existing_audio, outputs=output_text)
delete_dataset_button.click(
delete_existing_training_data, outputs=output_text)
def update_language_options(model):
# English-only models
if model in [
"tiny.en",
"base.en",
"small.en",
"medium.en"]:
languages = ["en"]
else:
# Multilingual models
languages = [
"en",
"es",
"fr",
"de",
"hi",
"it",
"pt",
"pl",
"tr",
"ru",
"nl",
"cs",
"ar",
"zh",
"hu",
"ko",
"ja",
]
return gr.Dropdown(choices=languages, value=languages[0])
with gr.Row():
speaker_name_input = gr.Textbox(
label="Training Project Name",
value="personsname",
visible=True,
scale=2,
)
whisper_model = gr.Dropdown(
label="Whisper Model",
value="large-v3",
choices=[
("tiny.en", "tiny.en"),
("tiny", "tiny"),
("base.en", "base.en"),
("base", "base"),
("small.en", "small.en"),
("small", "small"),
("medium.en", "medium.en"),
("medium", "medium"),
("large-v1", "large-v1"),
("large-v2", "large-v2"),
("large-v3", "large-v3"),
("large", "large"),
("large-v3-turbo", "large-v3-turbo"),
("turbo", "turbo"),
],
scale=1,
)
precision = gr.Dropdown(
label="Model Precision", value="mixed", choices=[
("Mixed", "mixed"), ("FP16", "float16"), ("FP32", "float32")], scale=1)
lang = gr.Dropdown(
label="Dataset Language",
value="en",
choices=[
"en",
"es",
"fr",
"de",
"hi",
"it",
"pt",
"pl",
"tr",
"ru",
"nl",
"cs",
"ar",
"zh",
"hu",
"ko",
"ja",
],
scale=1,
)
eval_split_number = gr.Number(
label="Evaluation Data Split",
value=15,
minimum=5,
maximum=95,
step=1,
scale=1,
)
with gr.Row():
create_bpe_tokenizer = gr.Dropdown(
label="BPE Tokenizer",
value="False",
choices=[("Enabled", "True"), ("Disabled", "False")],
info="Custom Tokenizer for training",
scale=1,
)
use_vad = gr.Dropdown(
label="VAD",
value="True",
choices=[("Enabled", "True"), ("Disabled", "False")],
info="Enable Silero VAD for better speech detection",
scale=1,
)
min_sample_length = gr.Dropdown(
label="Min Audio Length (seconds)",
value="2",
choices=["1", "2", "3", "4", "5"],
info="Split large audio into a minimum of",
scale=1,
)
max_sample_length = gr.Dropdown(
label="Max Audio Length (seconds)",
value="10",
choices=["8","9","10", "11", "12", "13", "14", "15", "16", "17", "18", "19", "20"],
info="Split large audio into a maximum of",
scale=1,
)
with gr.Accordion("🔍 Dataset Creation Debug Settings", open=False):
def update_debug_levels(
gpu, model, data, validation, general, audio, segments, duplicates):
DebugLevels.GPU_MEMORY = gpu
DebugLevels.MODEL_OPS = model
DebugLevels.DATA_PROCESS = data
DebugLevels.VALIDATION = validation
DebugLevels.GENERAL = general
DebugLevels.AUDIO = audio
DebugLevels.SEGMENTS = segments
DebugLevels.DUPLICATES = duplicates
return "Debug settings updated"
def select_all_debug():
return {
debug_gpu: True,
debug_model: True,
debug_data: True,
debug_validation: True,
debug_general: True,
debug_audio: True,
debug_segments: True,
debug_duplicates: True,
}
def clear_all_debug():
return {
debug_gpu: False,
debug_model: False,
debug_data: False,
debug_validation: False,
debug_general: False,
debug_audio: False,
debug_segments: False,
debug_duplicates: False,
}
with gr.Row():
gr.Markdown(
"""
Enable or disable different types of debug messages during dataset creation.
These settings will apply to the current session only.
"""
)
with gr.Row():
with gr.Column(scale=1):
debug_gpu = gr.Checkbox(
label="GPU Memory",
value=DebugLevels.GPU_MEMORY,
info="GPU memory and CUDA related debugging",
)
debug_model = gr.Checkbox(
label="Model Operations",
value=DebugLevels.MODEL_OPS,
info="Model loading, transcription, cleanup operations",
)
debug_data = gr.Checkbox(
label="Data Processing",
value=DebugLevels.DATA_PROCESS,
info="Data processing, words, sentences",
)
debug_validation = gr.Checkbox(
label="Dataset Validation",
value=DebugLevels.VALIDATION,
info="Dataset Validation, amount, sentences, files",
)
with gr.Column(scale=1):
debug_general = gr.Checkbox(
label="General",
value=DebugLevels.GENERAL,
info="General flow, file operations, metadata",
)
debug_audio = gr.Checkbox(
label="Audio",
value=DebugLevels.AUDIO,
info="Audio processing statistics and info",
)
debug_segments = gr.Checkbox(
label="Segments",
value=DebugLevels.SEGMENTS,
info="Detailed segment information",
)
debug_duplicates = gr.Checkbox(
label="Duplicates",
value=DebugLevels.DUPLICATES,
info="Duplicate handling information",
)
with gr.Row():
debug_select_all = gr.Button("Select All")
debug_clear_all = gr.Button("Clear All")
# Debug update functions
for checkbox in [
debug_gpu,
debug_model,
debug_data,
debug_validation,
debug_general,
debug_audio,
debug_segments,
debug_duplicates,
]:
checkbox.change(
fn=update_debug_levels,
inputs=[
debug_gpu,
debug_model,
debug_data,
debug_validation,
debug_general,
debug_audio,
debug_segments,
debug_duplicates,
],
# Hidden status output
outputs=[gr.Textbox(visible=False)],
)
debug_select_all.click(
fn=select_all_debug,
inputs=[],
outputs=[
debug_gpu,
debug_model,
debug_data,
debug_validation,
debug_general,
debug_audio,
debug_segments,
debug_duplicates,
],
)
debug_clear_all.click(
fn=clear_all_debug,
inputs=[],
outputs=[
debug_gpu,
debug_model,
debug_data,
debug_validation,
debug_general,
debug_audio,
debug_segments,
debug_duplicates,
],
)
progress_data = gr.Label(label="Progress:")
logs = gr.Textbox(
label="Logs:",
interactive=False,
lines=10,
)
demo.load(read_logs, None, logs, every=1)
# Update `lang` options when the `whisper_model` changes
whisper_model.change(
fn=update_language_options,
inputs=whisper_model,
outputs=lang)
prompt_compute_btn = gr.Button(value="Step 1 - Create dataset")
def preprocess_dataset(
pd_language,
pd_whisper_model,
pd_max_sample_length,
pd_min_sample_length,
pd_eval_split_number,
pd_speaker_name_input,
pd_create_bpe_tokenizer,
pd_use_vad,
pd_precision,
pd_progress=gr.Progress(),
):
"""Preprocess the dataset by validating audio files, formatting data, and splitting into training and evaluation sets."""
clear_gpu_cache()
# Check for audio files in the specified folder
pd_test_for_audio_files = [
file for file in os.listdir(audio_folder) if any(
file.lower().endswith(ext) for ext in [
'.wav', '.mp3', '.flac'])]
if not pd_test_for_audio_files:
return (
"I cannot find any mp3, wav or flac files in the folder called 'put-voice-samples-in-here'",
"",
"",
)
try:
# Format audio list and split into training and
# evaluation datasets
pd_train_meta, pd_eval_meta, pd_audio_total_size = format_audio_list(
fal_target_language=pd_language,
fal_whisper_model=pd_whisper_model,
fal_max_sample_length=pd_max_sample_length,
fal_min_sample_length=pd_min_sample_length,
fal_eval_split_number=pd_eval_split_number,
fal_speaker_name_input=pd_speaker_name_input,
fal_create_bpe_tokenizer= True if pd_create_bpe_tokenizer == "True" else False,
fal_gradio_progress=pd_progress,
fal_use_vad= True if pd_use_vad == "True" else False,
fal_precision=pd_precision,
)
except Exception:
traceback.print_exc()
pd_error = traceback.format_exc()
return (
f"The data processing was interrupted due to an error!! Please check the console to verify the full error message! \n Error summary: {pd_error}",
"",
"",
)
clear_gpu_cache()
# Check total audio size
if pd_audio_total_size < 120:
pd_message = (
"The total duration of the audio file or files you provided was less than 2 minutes in length. "
"Please add more audio samples.")
debug_print(
pd_message,
level="DATA_PROCESS",
is_warning=True)
return pd_message, "", ""
# Final GPU cleanup
get_gpu_memory()
debug_print(
"Dataset Generated. Either run Dataset Validation or move to Step 2",
level="DATA_PROCESS",
is_info=True,
)
# Return metadata and speaker name inputs for further
# processing in Gradio interface
return (
"Dataset Generated. Either run Dataset Validation or move to Step 2",
pd_train_meta,
pd_eval_meta,
pd_speaker_name_input,
pd_speaker_name_input,
pd_speaker_name_input,
)
with gr.Tab("Dataset Generation Guide"):
gr.Markdown(
"""
# Dataset Generation Guide
Below you'll find comprehensive instructions and information about generating your dataset.
Click each section to expand its contents.
"""
)
with gr.Accordion("🎯 Quick Start Guide", open=False):
gr.Markdown(
FinetuneContent.STEP1_QUICKSTART, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("📋 Detailed Instructions", open=False):
gr.Markdown(
FinetuneContent.STEP1_DETAILED_INSTRUCTIONS, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("🔧 Process Overview", open=False):
gr.Markdown(
FinetuneContent.STEP1_PROCESS_OVERVIEW, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("🔍 Whisper Model Selection", open=False):
gr.Markdown(
FinetuneContent.STEP1_WHISPER_MODEL_SELECTION, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("⚙️ Advanced Settings", open=False):
gr.Markdown(
FinetuneContent.STEP1_ADVANCED_SETTINGS, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("🔍 Dataset Creation Debug Settings", open=False):
gr.Markdown(
FinetuneContent.STEP1_DEBUG_SETTINGS, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("❗ Troubleshooting", open=False):
gr.Markdown(
FinetuneContent.STEP1_TROUBLESHOOTING, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.TabItem("📊 Dataset Validation"):
with gr.Row():
gr.Markdown("""# Audio Transcription Validation""")
with gr.Row():
with gr.Accordion("🎯 Audio Transcription Validation Help", open=False):
with gr.Row():
with gr.Column():
gr.Markdown(
FinetuneContent.DATASET_VALIDATION_1, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Column():
gr.Markdown(
FinetuneContent.DATASET_VALIDATION_2, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Row():
progress_box = gr.Textbox(label="Progress", interactive=False)
load_button = gr.Button("Run Validation")
with gr.Row():
with gr.Column(scale=2):
# Store full DataFrame in state
state = gr.State()
# Display DataFrame shows only visible columns
mismatch_table = gr.DataFrame(
headers=["Original Text", "Whisper Text", "Filename"],
datatype=["str", "str", "str"],
interactive=False,
wrap=True,
)
with gr.Column(scale=1):
audio_player = gr.Audio(
label="Audio Player (Edit enabled)", interactive=True)
current_expected = gr.Textbox(
label="Original Text", interactive=False)
current_transcribed = gr.Textbox(
label="Whisper Text", interactive=False)
text_choice = gr.Radio(
choices=[
"Use Original",
"Use Whisper",
"Edit Manually"],
label="Choose Transcription",
value="Use Original",
)
manual_edit = gr.Textbox(
label="Manual Edit", interactive=True, visible=False)
current_index = gr.Number(visible=False)
save_button = gr.Button("Save Audio and Correction")
save_status = gr.Textbox(
label="Save Status", interactive=False)
def update_audio_player(evt: gr.SelectData, df):
"""Update audio player with selected file"""
try:
selected_row = df.iloc[evt.index]
audio_path = selected_row["full_path"]
# If it's a Series, get the first value
if isinstance(audio_path, pd.Series):
audio_path = audio_path.iloc[0]
# Clean the path string
audio_path = str(audio_path).strip()
# Check if file exists
if not os.path.exists(audio_path):
debug_print(
f"Audio file not found: {audio_path}",
level="DATA_PROCESS",
is_warning=True)
return {
audio_player: None,
current_expected: "",
current_transcribed: "",
current_index: None,
save_status: "Error: Audio file not found",
}
# Get text values directly from the row
expected = (
str(selected_row["expected_text"]).strip()
if isinstance(selected_row["expected_text"], str)
else selected_row["expected_text"].iloc[0]
)
transcribed = (
str(selected_row["transcribed_text"]).strip()
if isinstance(selected_row["transcribed_text"], str)
else selected_row["transcribed_text"].iloc[0]
)
return {
audio_player: audio_path,
current_expected: expected,
current_transcribed: transcribed,
current_index: evt.index,
save_status: "Ready to save correction",
}
except Exception as e:
debug_print(
f"Error in update_audio_player: {str(e)}",
level="GENERAL",
is_error=True)
debug_print(
f"Selected row data: {selected_row if 'selected_row' in locals() else 'Not available'}",
level="GENERAL",
is_info=True,
)
return {
audio_player: None,
current_expected: "",
current_transcribed: "",
current_index: None,
save_status: f"Error: {str(e)}",
}
# Event handlers
text_choice.change(
lambda x: gr.update(
visible=x == "Edit Manually"),
text_choice,
manual_edit)
mismatch_table.select(
update_audio_player,
[state], # Use full DataFrame from state
[audio_player,
current_expected,
current_transcribed,
current_index,
save_status],
)
save_button.click(
save_audio_and_correction,
inputs=[
text_choice,
manual_edit,
audio_player,
state,
current_index],
outputs=[
mismatch_table,
current_expected,
save_status,
audio_player],
)
# Store both display and full DataFrame
load_button.click(
load_and_display_mismatches,
# state gets full df, mismatch_table gets display_df
outputs=[state, mismatch_table, progress_box],
)
#######################
#### GRADIO STEP 2 ####
#######################
with gr.Tab("💻 Step 2 - Training"):
with gr.Tab("Training the model"):
with gr.Row():
speaker_name_input_training = gr.Textbox(
label="Project Name",
value="personsname",
visible=True,
scale=1,
)
with gr.Group():
continue_run = gr.Checkbox(
value=False,
label="Continue Previous Project",
scale=1,
visible=False,
)
# Continue Run has been set to invisible from the interface as its setting
# a 1000 epoch run when used. Have been unable to track down where this is
# coming from.
disable_shared_memory = gr.Checkbox(
value=False,
label="Disable Shared Memory Use",
scale=1,
)
warm_up = gr.Checkbox(
value=False,
label="Perform Warmup Learning",
scale=1,
)
train_csv = gr.Textbox(
label="Train CSV file path:",
scale=2,
)
eval_csv = gr.Textbox(
label="Eval CSV file path:",
scale=2,
)
with gr.Row():
model_to_train_choices = list(available_models.keys())
model_to_train = gr.Dropdown(
choices=model_to_train_choices,
label="Select the Model to train",
value=model_to_train_choices[0] if model_to_train_choices else None,
scale=2,
)
learning_rates = gr.Dropdown(
value=5e-6,
label="Learning Rate",
choices=[
("1e-6", 1e-6),
("5e-6", 5e-6),
("1e-5", 1e-5),
("5e-5", 5e-5),
("1e-4", 1e-4),
("5e-4", 5e-4),
("1e-3", 1e-3),
],
type="value",
allow_custom_value=True,
scale=1,
)
learning_rate_scheduler = gr.Dropdown(
value="CosineAnnealingWarmRestarts",
label="Learning Rate Scheduler(s)",
choices=[
("None", "None"),
("Cosine Annealing", "CosineAnnealingLR"),
("Cosine Annealing Warm Restarts",
"CosineAnnealingWarmRestarts"),
("Cyclic", "CyclicLR"),
("Exponential", "ExponentialLR"),
("Multi Step", "MultiStepLR"),
("Reduce on Plateau", "ReduceLROnPlateau"),
("Step", "StepLR"),
# ("OneCycleLR", "OneCycleLR"),
],
type="value",
allow_custom_value=False,
scale=2,
)
optimizer = gr.Dropdown(
value="AdamW",
label="Optimizer",
choices=[
("AdamW", "AdamW"),
("SGD with Momentum", "SGD"),
("RMSprop", "RMSprop"),
("Adagrad", "Adagrad"),
("Adam", "Adam"),
("Rectified Adam", "RAdam"),
("Step Wise", "stepwisegraduallr"),
("Noam", "noamlr"),
],
type="value",
allow_custom_value=False,
scale=2,
)
num_workers = gr.Dropdown(
value="8",
label="Workers/Threads",
choices=[
"0",
"1",
"2",
"3",
"4",
"5",
"6",
"7",
"8",
"9",
"10"],
allow_custom_value=False,
interactive=True,
scale=0,
)
with gr.Row():
num_epochs = gr.Slider(
label="Number of epochs:",
minimum=1,
maximum=100,
step=1,
value=args.num_epochs,
)
batch_size = gr.Slider(
label="Batch size:",
minimum=2,
maximum=512,
step=1,
value=args.batch_size,
)
grad_acumm = gr.Slider(
label="Grad accumulation steps:",
minimum=1,
maximum=128,
step=1,
value=args.grad_acumm,
)
max_audio_length = gr.Slider(
label="Max permitted audio size in seconds:",
minimum=2,
maximum=20,
step=1,
value=args.max_audio_length,
)
with gr.Accordion("🔍 Training Debug Settings", open=False):
gr.Markdown("""
Enable or disable different types of debug messages during model training.
These settings will apply to the current training session only.
""")
with gr.Row():
with gr.Column(scale=1):
debug_gpu = gr.Checkbox(
label="GPU Memory",
value=DebugLevels.GPU_MEMORY,
info="GPU memory and CUDA related debugging",
)
debug_model = gr.Checkbox(
label="Model Operations",
value=DebugLevels.MODEL_OPS,
info="Model loading, training operations, cleanup",
)
with gr.Column(scale=1):
debug_general = gr.Checkbox(
label="General",
value=DebugLevels.GENERAL,
info="Training flow, file operations",
)
debug_data = gr.Checkbox(
label="Data Processing",
value=DebugLevels.DATA_PROCESS,
info="Data processing, files, folders",
)
with gr.Row():
debug_select_all = gr.Button("Select All")
debug_clear_all = gr.Button("Clear All")
# Debug update functions
def train_update_debug_levels(gpu, model, general, data):
DebugLevels.GPU_MEMORY = gpu
DebugLevels.MODEL_OPS = model
DebugLevels.GENERAL = general
DebugLevels.DATA_PROCESS = data
return "Debug settings updated"
def train_select_all_debug():
return {
debug_gpu: True,
debug_model: True,
debug_general: True,
debug_data: True,
}
def train_clear_all_debug():
return {
debug_gpu: False,
debug_model: False,
debug_general: False,
debug_data: False,
}
# Connect the debugging controls
for checkbox in [debug_gpu, debug_model, debug_general, debug_data]:
checkbox.change(
fn=train_update_debug_levels,
inputs=[debug_gpu, debug_model, debug_general, debug_data],
outputs=[gr.Textbox(visible=False)],
)
debug_select_all.click(
fn=train_select_all_debug,
inputs=[],
outputs=[debug_gpu, debug_model, debug_general, debug_data],
)
debug_clear_all.click(
fn=train_clear_all_debug,
inputs=[],
outputs=[debug_gpu, debug_model, debug_general, debug_data],
)
progress_train = gr.Label(label="Progress:")
with gr.Row():
train_time = gr.Label(
"Estimated Total Training Time", show_label=False, scale=2)
train_btn = gr.Button(
value="Step 2 - Run the training", scale=1)
with gr.Row():
model_data = gr.Image(
c_logger.plot_metrics(), show_label=False)
logs_tts_train = gr.Textbox(
label="Logs:",
interactive=False,
lines=10,
)
demo.load(
load_metrics, None, [
model_data, train_time], every=1)
demo.load(read_logs, None, logs_tts_train, every=1)
def train_model(
language,
train_csv,
eval_csv,
learning_rates,
model_to_train,
num_epochs,
batch_size,
grad_acumm,
max_audio_length,
speaker_name_input_training,
continue_run,
disable_shared_memory,
learning_rate_scheduler,
optimizer,
num_workers,
warm_up,
progress=gr.Progress()):
"""
Trains XTTS model with specified parameters and returns model artifacts.
Returns:
tuple: Status message, config path, vocab file, checkpoint path, speaker reference, speaker name
"""
clear_gpu_cache()
global out_path
if speaker_name_input_training and speaker_name_input_training != 'personsname':
out_path = this_dir / "finetune" / speaker_name_input_training
else:
out_path = default_path
if not train_csv or not eval_csv:
if (out_path /
"metadata_eval.csv").exists() and (out_path /
"metadata_train.csv").exists():
train_csv = out_path / "metadata_train.csv"
eval_csv = out_path / "metadata_eval.csv"
debug_print(
"Using existing metadata and training csv.",
level="GENERAL",
is_info=True)
else:
return (
"You need to run the data processing step or manually set `Train CSV` and `Eval CSV` fields !",
"",
"",
"",
"",
)
try:
# convert seconds to waveform frames
max_audio_length = int(max_audio_length * 22050)
# Convert the learning rate value to a float
learning_rate = float(learning_rates)
progress(0, "Initializing training...")
config_path, return_xtts_checkpoint, vocab_file, exp_path, speaker_wav = train_gpt(
language, num_epochs, batch_size, grad_acumm, train_csv, eval_csv, learning_rate, model_to_train, continue_run, disable_shared_memory, learning_rate_scheduler, optimizer, num_workers, warm_up, max_audio_length=max_audio_length, progress=gr.Progress())
# copy original files to avoid parameters changes
# issues
shutil.copy(config_path, exp_path)
shutil.copy(vocab_file, exp_path)
ft_xtts_checkpoint=return_xtts_checkpoint
ft_xtts_checkpoint = os.path.join(
exp_path, "best_model.pth")
debug_print(
"Model training done. Move to Step 3",
level="GENERAL",
is_info=True)
clear_gpu_cache()
return (
"Model training done. Move to Step 3",
config_path,
vocab_file,
ft_xtts_checkpoint,
speaker_wav,
speaker_name_input_training,
)
except ValueError as ve:
error_message = str(ve)
debug_print(
f"{error_message}",
level="GENERAL",
is_error=True)
return f"Training error: {error_message}", "", "", "", "", ""
except Exception as e:
# This will catch any other unexpected errors
error_message = f"An unexpected error occurred: {str(e)}"
debug_print(
f"{error_message}",
level="GENERAL",
is_error=True)
return f"Training error: {error_message}", "", "", "", "", ""
with gr.Tab("Training Guide"):
gr.Markdown(
"""
# Comprehensive Training Guide
Below you'll find detailed explanations of all training parameters and processes.
Each section contains both conceptual explanations and specific configuration guidance.
"""
)
with gr.Accordion("🎯 Quick Start Training Guide", open=False):
gr.Markdown(
FinetuneContent.STEP2_QUICKSTART, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("📁 Using Your Own Training Dataset", open=False):
gr.Markdown(
FinetuneContent.STEP2_YOUR_OWN_DATASET, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("📊 Training Metrics and Logs", open=False):
gr.Markdown(
FinetuneContent.STEP2_TRAINING_METRICS, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("💾 Memory Management", open=False):
gr.Markdown(
FinetuneContent.STEP2_MEMORY_MANAGEMENT, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("⚙️ Batch Size & Gradient Accumulation", open=False):
gr.Markdown(
FinetuneContent.STEP2_BATCH_SIZE, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("📊 Learning Rate & Schedulers", open=False):
gr.Markdown(
FinetuneContent.STEP2_LEARNING_RATE, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("🔧 Optimizers", open=False):
gr.Markdown(
FinetuneContent.STEP2_OPTIMIZERS, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("🔄 Training Epochs", open=False):
gr.Markdown(
FinetuneContent.STEP2_EPOCHS, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("📈 Max Audio Length", open=False):
gr.Markdown(
FinetuneContent.STEP2_AUDIO_LENGTH, elem_classes="custom-markdown") # pylint: disable=no-member
#######################
#### GRADIO STEP 3 ####
#######################
with gr.Tab("✅ Step 3 - Testing"):
with gr.Tab("Testing"):
with gr.Row():
with gr.Column() as col1:
xtts_checkpoint = gr.Dropdown(
[str(file) for file in xtts_checkpoint_files],
label="XTTS checkpoint path (best_model.pth):",
value="",
allow_custom_value=True,
)
xtts_config = gr.Dropdown(
[str(file) for file in xtts_config_files],
label="XTTS config path (config.json):",
value="",
allow_custom_value=True,
)
xtts_vocab = gr.Dropdown(
[str(file) for file in xtts_vocab_files],
label="XTTS vocab path (vocab.json):",
value="",
allow_custom_value=True,
)
progress_load = gr.Label(label="Progress:")
load_btn = gr.Button(
value="Step 3 - Load Fine-tuned XTTS model")
with gr.Column() as col2:
with gr.Row():
# Gather the voice files
available_speaker_audios = get_available_voices()
# Create Dropdown for speaker reference audio
speaker_reference_audio = gr.Dropdown(
available_speaker_audios,
label="Speaker reference audio (Press Refresh Dropdowns):",
value="", # Set the default value if needed
allow_custom_value=True, # Allow custom values
scale=2,
)
min_audio_length = gr.Dropdown(
label="Min Audio Length (seconds)",
value="6",
choices=["3", "4", "5", "6", "7", "8", "9", "10"],
type="value",
allow_custom_value=False,
scale=1,
)
with gr.Row():
speaker_name_input_testing = gr.Textbox(
label="Project Name",
value="personsname",
visible=True,
scale=1,
)
tts_language = gr.Dropdown(
label="Language",
value="en",
choices=[
"en",
"es",
"fr",
"de",
"hi",
"it",
"pt",
"pl",
"tr",
"ru",
"nl",
"cs",
"ar",
"zh",
"hu",
"ko",
"ja",
],
)
# Create refresh button
refresh_button = create_refresh_button(
[
xtts_checkpoint,
xtts_config,
xtts_vocab,
speaker_reference_audio,
speaker_name_input_testing,
],
[
lambda speaker_name, min_duration_seconds: {
"choices": find_best_models(
out_path, speaker_name=speaker_name
),
"value": "",
},
lambda speaker_name, min_duration_seconds: {
"choices": find_jsons(
out_path, "config.json", speaker_name=speaker_name
),
"value": "",
},
lambda speaker_name, min_duration_seconds: {
"choices": find_jsons(
out_path, "vocab.json", speaker_name=speaker_name
),
"value": "",
},
lambda speaker_name, min_duration_seconds: {
"choices": get_available_voices(
min_duration_seconds=int(min_duration_seconds),
speaker_name=speaker_name,
),
"value": "",
},
],
elem_class="refresh-button-class",
)
tts_text = gr.Textbox(
label="Input Text:",
value="I've just fine tuned a text to speech language model and this is how it sounds. If it doesn't sound right, I will try a different Speaker Reference Audio file.",
lines=5,
)
tts_btn = gr.Button(
value="Step 4 - Inference (Generate TTS)")
with gr.Row():
progress_gen = gr.Label(label="Progress:")
with gr.Row():
tts_output_audio = gr.Audio(label="TTS Generated Speech.")
reference_audio = gr.Audio(
label="Speaker Reference Audio Sample.")
with gr.Tab("Testing Guide"):
gr.Markdown(
"""
# Testing Your Fine-tuned Model
This section guides you through evaluating the quality of your fine-tuned model using various reference audios. Testing ensures your model produces high-quality TTS that matches your target speaker characteristics. Follow the steps below for effective testing and validation.
"""
)
with gr.Accordion("🎯 Testing Overview", open=False):
gr.Markdown(
FinetuneContent.STEP3_TESTING_OVERVIEW, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("⚠️ Important Notes", open=False):
gr.Markdown(
FinetuneContent.STEP3_IMPORTANT, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("📝 Testing Instructions", open=False):
gr.Markdown(
FinetuneContent.STEP3_INSTRUCTIONS, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("🔍 What the Testing Step Does", open=False):
gr.Markdown(
FinetuneContent.STEP3_WHAT_IT_DOES, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Tab("📦 Model Export"):
with gr.Accordion("🎯 Export Overview", open=False):
gr.Markdown(
FinetuneContent.EXPORT_OVERVIEW, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("📊 Voice Sample Organization", open=False):
gr.Markdown(
FinetuneContent.EXPORT_VOICE_SAMPLES, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("🔄 Export Options", open=False):
gr.Markdown(
FinetuneContent.EXPORT_OPTIONS, elem_classes="custom-markdown") # pylint: disable=no-member
with gr.Accordion("💾 Storage Management", open=False):
gr.Markdown(
FinetuneContent.EXPORT_STORAGE, elem_classes="custom-markdown") # pylint: disable=no-member
final_progress_data = gr.Label(label="Progress:")
with gr.Row():
xtts_checkpoint_copy = gr.Dropdown(
[str(file) for file in xtts_checkpoint_files],
label="XTTS checkpoint path (Click the refresh button to populate):",
value="",
allow_custom_value=True,
scale=2,
)
speaker_name_input_export = gr.Textbox(
label="Project Name (Refresh Dropdowns on change)",
value="personsname",
visible=True,
scale=1,
)
# Create refresh button
refresh_button = create_refresh_button_next(
[xtts_checkpoint_copy, speaker_name_input_export],
[
lambda speaker_name: {
"choices": find_best_models(out_path, speaker_name),
"value": "",
},
],
elem_class="refresh-button-class",
)
with gr.Row():
overwrite_existing = gr.Dropdown(
value="Do not overwrite existing files",
choices=[
"Overwrite existing files",
"Do not overwrite existing files"],
label="File Overwrite Options",
)
folder_path = gr.Textbox(
label="Enter a new folder name (will be sub the models folder)",
lines=1,
value="mycustomfolder",
)
compact_custom_btn = gr.Button(
value="Compact and move model to a folder name of your choosing")
with gr.Row():
gr.Textbox(
value="This will DELETE your training data and the raw finetuned model from the specified Project Name (above)",
scale=2,
show_label=False,
interactive=False,
)
delete_training_btn = gr.Button(
value="Delete generated training data")
with gr.Row():
gr.Textbox(
value="This will DELETE your original voice samples from /finetune/put-voice-samples-in-here/.",
scale=2,
show_label=False,
interactive=False,
)
delete_voicesamples_btn = gr.Button(
value="Delete original voice samples")
prompt_compute_btn.click(
fn=preprocess_dataset,
inputs=[
lang,
whisper_model,
max_sample_length,
min_sample_length,
eval_split_number,
speaker_name_input,
create_bpe_tokenizer,
use_vad,
precision,
],
outputs=[
progress_data,
train_csv,
eval_csv,
speaker_name_input_training,
speaker_name_input_testing,
speaker_name_input_export,
],
)
train_btn.click(
fn=train_model,
inputs=[
lang,
train_csv,
eval_csv,
learning_rates,
model_to_train,
num_epochs,
batch_size,
grad_acumm,
max_audio_length,
speaker_name_input_training,
continue_run,
disable_shared_memory,
learning_rate_scheduler,
optimizer,
num_workers,
warm_up,
],
outputs=[
progress_train,
xtts_config,
xtts_vocab,
xtts_checkpoint,
speaker_reference_audio,
speaker_name_input_testing,
],
)
load_btn.click(
fn=load_model,
inputs=[xtts_checkpoint, xtts_config, xtts_vocab],
outputs=[progress_load],
)
tts_btn.click(
fn=run_tts,
inputs=[
tts_language,
tts_text,
speaker_reference_audio,
],
outputs=[progress_gen, tts_output_audio, reference_audio],
)
compact_custom_btn.click(
fn=compact_custom_model,
inputs=[xtts_checkpoint_copy, folder_path, overwrite_existing],
outputs=[final_progress_data],
)
delete_training_btn.click(
fn=delete_training_data,
outputs=[final_progress_data],
)
delete_voicesamples_btn.click(
fn=delete_voice_sample_contents,
outputs=[final_progress_data],
)
model_to_train.change(
basemodel_or_finetunedmodel_choice,
model_to_train,
None)
demo.queue().launch(
show_api=False,
inbrowser=True,
share=False,
debug=False,
server_port=7052,
server_name="0.0.0.0",
)