feat: Add OpenAI-compatible TTS endpoint and engines
- Implements POST /v1/audio/speech endpoint (OpenAI API compatible). - Integrates Kokoro and XTTS engines (including dependencies and implementations). - Updates main application to register new engines and router. - Adds unit tests for OpenAI compatibility. - Updates requirements.txt for new engines.
This commit is contained in:
177
app/engines/xtts.py
Normal file
177
app/engines/xtts.py
Normal file
@ -0,0 +1,177 @@
|
||||
"""
|
||||
NovaAi – TTS-Engine-Hub
|
||||
engines/xtts.py
|
||||
Version: v0.1.0
|
||||
|
||||
Description:
|
||||
Coqui XTTS v2 engine adapter.
|
||||
Supports multilingual synthesis and voice cloning via reference audio.
|
||||
"""
|
||||
|
||||
import os
|
||||
import asyncio
|
||||
import logging
|
||||
import torch
|
||||
from .engine_base import TTSEngineBase
|
||||
from app.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class XTTSEngine(TTSEngineBase):
|
||||
def __init__(self):
|
||||
logger.debug("XTTSEngine __init__ started.")
|
||||
self.device = "cpu"
|
||||
if torch.cuda.is_available():
|
||||
logger.debug("CUDA is available.")
|
||||
if settings.XTTS_DEVICE == "cuda":
|
||||
self.device = "cuda"
|
||||
logger.debug(f"XTTS_DEVICE setting is 'cuda'. Using CUDA.")
|
||||
else:
|
||||
logger.debug(f"XTTS_DEVICE setting is '{settings.XTTS_DEVICE}'. Falling back to CPU despite CUDA availability.")
|
||||
else:
|
||||
logger.debug("CUDA is not available. Using CPU.")
|
||||
|
||||
self.model = None
|
||||
self.tts = None
|
||||
|
||||
# Verify license acceptance
|
||||
if not settings.XTTS_ACCEPT_LICENSE:
|
||||
logger.warning("XTTS license not accepted. Engine will not load. Set XTTS_ACCEPT_LICENSE=true in .env")
|
||||
return
|
||||
|
||||
try:
|
||||
from TTS.api import TTS
|
||||
logger.debug("Coqui TTS library imported successfully.")
|
||||
except ImportError:
|
||||
logger.error("Coqui TTS library not found. Install 'TTS' via pip.")
|
||||
return
|
||||
|
||||
logger.info(f"Initializing XTTS v2 on {self.device}...")
|
||||
try:
|
||||
# Set environment variable to bypass TTS library's interactive license prompt
|
||||
# This tells the TTS library that we agree to the terms
|
||||
os.environ['COQUI_TOS_AGREED'] = '1'
|
||||
|
||||
# Initialize TTS with the model name.
|
||||
# This will download the model if not present.
|
||||
# We use the official model name.
|
||||
logger.debug(f"Calling TTS('tts_models/multilingual/multi-dataset/xtts_v2').to({self.device})...")
|
||||
self.tts = TTS("tts_models/multilingual/multi-dataset/xtts_v2").to(self.device)
|
||||
logger.info("XTTS v2 model loaded successfully.")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load XTTS model: {e}", exc_info=True) # exc_info=True to log traceback
|
||||
self.tts = None
|
||||
logger.debug("XTTSEngine __init__ finished.")
|
||||
|
||||
def list_models(self):
|
||||
return ["xtts_v2"]
|
||||
|
||||
def list_voices(self, model: str = None):
|
||||
"""
|
||||
Returns a list of available reference audio files (speakers)
|
||||
found in the VOICES_DIR.
|
||||
"""
|
||||
voices_dir = settings.VOICES_DIR
|
||||
if not os.path.exists(voices_dir):
|
||||
return ["default"]
|
||||
|
||||
# List .wav files in the voices directory
|
||||
voices = [f for f in os.listdir(voices_dir) if f.lower().endswith(".wav")]
|
||||
return sorted(voices) if voices else ["default"]
|
||||
|
||||
def healthcheck(self):
|
||||
if not settings.XTTS_ACCEPT_LICENSE:
|
||||
return {"status": "license_not_accepted", "detail": "Set XTTS_ACCEPT_LICENSE=true"}
|
||||
if self.tts is None:
|
||||
return {"status": "error", "detail": "Model not loaded"}
|
||||
return {"status": "ok", "device": self.device}
|
||||
|
||||
async def synthesize(self, text: str, speaker: str = None, model: str = None, fmt: str = "wav"):
|
||||
"""
|
||||
Synthesize speech using XTTS v2.
|
||||
|
||||
Args:
|
||||
text: Text to synthesize.
|
||||
speaker: Filename of the reference audio in VOICES_DIR (e.g., "my_voice.wav").
|
||||
model: Ignored (only xtts_v2 supported).
|
||||
fmt: Output format (wav by default).
|
||||
"""
|
||||
if not self.tts:
|
||||
raise RuntimeError("XTTS engine is not initialized or license not accepted.")
|
||||
|
||||
# Resolve speaker/reference audio
|
||||
voices_dir = settings.VOICES_DIR
|
||||
if not os.path.exists(voices_dir):
|
||||
os.makedirs(voices_dir, exist_ok=True)
|
||||
|
||||
# precise path handling
|
||||
speaker_wav = None
|
||||
if speaker and speaker != "default":
|
||||
potential_path = os.path.join(voices_dir, speaker)
|
||||
if os.path.exists(potential_path):
|
||||
speaker_wav = potential_path
|
||||
else:
|
||||
# Check if speaker has extension, if not try adding .wav
|
||||
if not speaker.lower().endswith(".wav"):
|
||||
potential_path_ext = os.path.join(voices_dir, f"{speaker}.wav")
|
||||
if os.path.exists(potential_path_ext):
|
||||
speaker_wav = potential_path_ext
|
||||
|
||||
# Fallback if no valid speaker provided - XTTS NEEDS a speaker reference.
|
||||
# We'll use a default sample if provided, or fail.
|
||||
# Ideally, we should ship a default reference.
|
||||
if not speaker_wav:
|
||||
# Try to find *any* wav file in the dir to use as default
|
||||
available = self.list_voices()
|
||||
if available and available[0] != "default":
|
||||
speaker_wav = os.path.join(voices_dir, available[0])
|
||||
logger.warning(f"No valid speaker '{speaker}' found. Using first available: {available[0]}")
|
||||
else:
|
||||
raise ValueError("XTTS requires a reference audio file (speaker). Please upload a .wav file to app/asset/voices/")
|
||||
|
||||
# Output file
|
||||
import tempfile
|
||||
fd, output_path = tempfile.mkstemp(suffix=".wav", prefix="xtts_")
|
||||
os.close(fd)
|
||||
|
||||
# Run synthesis in thread pool to avoid blocking event loop
|
||||
# XTTS API: tts.tts_to_file(text=..., speaker_wav=..., language=..., file_path=...)
|
||||
# We need to detect language or default to English ("en")
|
||||
# For now, we hardcode "en" or try to auto-detect if the library supports it,
|
||||
# but tts_to_file usually requires language for multilingual models.
|
||||
language = "en" # TODO: Add language parameter to API or auto-detect
|
||||
|
||||
logger.info(f"Synthesizing with XTTS. Speaker: {os.path.basename(speaker_wav)}, Lang: {language}")
|
||||
|
||||
try:
|
||||
await asyncio.to_thread(
|
||||
self.tts.tts_to_file,
|
||||
text=text,
|
||||
speaker_wav=speaker_wav,
|
||||
language=language,
|
||||
file_path=output_path
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"XTTS synthesis failed: {e}")
|
||||
if os.path.exists(output_path):
|
||||
os.remove(output_path)
|
||||
raise RuntimeError(f"XTTS synthesis failed: {str(e)}")
|
||||
|
||||
return output_path
|
||||
|
||||
async def selftest(self):
|
||||
try:
|
||||
# Check if we have at least one reference voice
|
||||
voices = self.list_voices()
|
||||
if not voices or voices == ["default"]:
|
||||
return {"selftest": False, "error": "No reference voices found in asset/voices", "engine": "xtts"}
|
||||
|
||||
test_voice = voices[0]
|
||||
output = await self.synthesize("XTTS selftest.", speaker=test_voice)
|
||||
|
||||
if os.path.exists(output) and os.path.getsize(output) > 0:
|
||||
os.remove(output)
|
||||
return {"selftest": True, "engine": "xtts"}
|
||||
return {"selftest": False, "error": "Output file empty or missing", "engine": "xtts"}
|
||||
except Exception as e:
|
||||
return {"selftest": False, "error": str(e), "engine": "xtts"}
|
||||
Reference in New Issue
Block a user