| import os |
| os.environ.setdefault("OMP_NUM_THREADS", "4") |
| os.environ.setdefault("COQUI_TOS_AGREED", "1") |
|
|
| import io |
| import base64 |
| import tempfile |
| import logging |
| import wave |
| import numpy as np |
| import torch |
| import pyrubberband as pyrb |
| from contextlib import asynccontextmanager |
| from pathlib import Path |
| from fastapi import FastAPI, Request, HTTPException |
| from fastapi.responses import Response, JSONResponse, HTMLResponse |
| from pydantic import BaseModel, Field |
| from typing import Optional |
|
|
| logging.basicConfig(level=logging.INFO) |
| logger = logging.getLogger("xttsv2-engine") |
|
|
| BEARER_TOKEN = os.environ.get("API_KEY", "124CC717-7517-47A2-BBD6-54FCAE310297") |
| MODEL_NAME = "tts_models/multilingual/multi-dataset/xtts_v2" |
| SAMPLE_RATE = 24000 |
| BIT_DEPTH = 16 |
| CHANNELS = 1 |
| MAX_SECONDS = 30 |
|
|
| CANONICAL_EMOTIONS = [ |
| "neutral", "happy", "sad", "angry", "calm", "excited", |
| "fear", "surprise", "disgust", "confused", "anxious", |
| "hopeful", "melancholy", "fearful", |
| ] |
|
|
| EMOTION_SPEED_MAP = { |
| "neutral": 1.0, |
| "happy": 1.05, |
| "sad": 0.93, |
| "angry": 1.08, |
| "calm": 0.92, |
| "excited": 1.10, |
| "fear": 1.06, |
| "surprise": 1.07, |
| "disgust": 0.97, |
| "confused": 0.96, |
| "anxious": 1.04, |
| "hopeful": 1.02, |
| "melancholy": 0.91, |
| "fearful": 1.06, |
| } |
|
|
| EMOTION_PITCH_MAP = { |
| "neutral": 0.0, |
| "happy": 0.6, |
| "sad": -0.5, |
| "angry": -0.3, |
| "calm": 0.0, |
| "excited": 0.8, |
| "fear": 0.4, |
| "surprise": 0.7, |
| "disgust": -0.4, |
| "confused": 0.3, |
| "anxious": 0.3, |
| "hopeful": 0.4, |
| "melancholy": -0.5, |
| "fearful": 0.4, |
| } |
|
|
| tts_model = None |
|
|
|
|
| def load_model(): |
| global tts_model |
| import torch.serialization |
| _original_load = torch.load |
| def _patched_load(*args, **kwargs): |
| kwargs.setdefault("weights_only", False) |
| return _original_load(*args, **kwargs) |
| torch.load = _patched_load |
|
|
| from TTS.api import TTS |
|
|
| device = "cuda" if torch.cuda.is_available() else "cpu" |
| logger.info(f"Loading XTTSv2 model on {device}...") |
| tts_model = TTS(model_name=MODEL_NAME, progress_bar=True).to(device) |
| logger.info("XTTSv2 model loaded successfully.") |
|
|
|
|
| @asynccontextmanager |
| async def lifespan(app: FastAPI): |
| load_model() |
| yield |
|
|
|
|
| app = FastAPI(title="XTTSv2 TTS Engine", lifespan=lifespan) |
|
|
|
|
| def verify_auth(request: Request): |
| if not BEARER_TOKEN: |
| return |
| auth = request.headers.get("Authorization", "") |
| if auth != f"Bearer {BEARER_TOKEN}": |
| raise HTTPException(status_code=401, detail="Unauthorized") |
|
|
|
|
| def numpy_to_wav_bytes(audio_np: np.ndarray, sample_rate: int) -> bytes: |
| audio_np = np.clip(audio_np, -1.0, 1.0) |
| audio_int16 = (audio_np * 32767).astype(np.int16) |
|
|
| buf = io.BytesIO() |
| with wave.open(buf, "wb") as wf: |
| wf.setnchannels(CHANNELS) |
| wf.setsampwidth(2) |
| wf.setframerate(sample_rate) |
| wf.writeframes(audio_int16.tobytes()) |
| return buf.getvalue() |
|
|
|
|
| class ConvertRequest(BaseModel): |
| input_text: str |
| builtin_voice_id: Optional[str] = None |
| voice_to_clone_sample: Optional[str] = None |
| random_seed: Optional[int] = None |
| emotion_set: list[str] = Field(default_factory=lambda: ["neutral"]) |
| intensity: int = Field(default=50, ge=1, le=100) |
| volume: int = Field(default=75, ge=1, le=100) |
| speed_adjust: float = Field(default=0.0, ge=-5.0, le=5.0) |
| pitch_adjust: float = Field(default=0.0, ge=-5.0, le=5.0) |
|
|
|
|
| @app.post("/GetEngineDetails") |
| async def get_engine_details(request: Request): |
| verify_auth(request) |
|
|
| return { |
| "engine_id": "xttsv2", |
| "engine_name": "Coqui XTTSv2", |
| "sample_rate": SAMPLE_RATE, |
| "bit_depth": BIT_DEPTH, |
| "channels": CHANNELS, |
| "max_seconds_per_conversion": MAX_SECONDS, |
| "supports_voice_cloning": True, |
| "builtin_voices": [], |
| "supported_emotions": CANONICAL_EMOTIONS, |
| "extra_properties": { |
| "model": MODEL_NAME, |
| "languages": [ |
| "en", "es", "fr", "de", "it", "pt", "pl", "tr", |
| "ru", "nl", "cs", "ar", "zh-cn", "ja", "hu", "ko" |
| ] |
| } |
| } |
|
|
|
|
| @app.post("/ConvertTextToSpeech") |
| async def convert_text_to_speech(request: Request): |
| verify_auth(request) |
|
|
| try: |
| body = await request.json() |
| req = ConvertRequest(**body) |
| except Exception as e: |
| return JSONResponse( |
| status_code=400, |
| content={"error": str(e), "error_code": "INVALID_REQUEST"} |
| ) |
|
|
| if not req.input_text.strip(): |
| return JSONResponse( |
| status_code=400, |
| content={"error": "Input text is empty", "error_code": "INVALID_REQUEST"} |
| ) |
|
|
| if req.random_seed is not None: |
| torch.manual_seed(req.random_seed) |
| if torch.cuda.is_available(): |
| torch.cuda.manual_seed(req.random_seed) |
|
|
| speaker_wav_path = None |
| temp_files = [] |
|
|
| try: |
| if req.voice_to_clone_sample: |
| wav_bytes = base64.b64decode(req.voice_to_clone_sample) |
| tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) |
| tmp.write(wav_bytes) |
| tmp.close() |
| speaker_wav_path = tmp.name |
| temp_files.append(tmp.name) |
|
|
| dominant_emotion = req.emotion_set[0].lower() if req.emotion_set else "neutral" |
| if dominant_emotion not in EMOTION_SPEED_MAP: |
| dominant_emotion = "neutral" |
|
|
| intensity_scale = req.intensity / 50.0 |
|
|
| emotion_speed_raw = EMOTION_SPEED_MAP[dominant_emotion] |
| emotion_speed = 1.0 + (emotion_speed_raw - 1.0) * intensity_scale |
|
|
| emotion_pitch_raw = EMOTION_PITCH_MAP[dominant_emotion] |
| emotion_pitch = emotion_pitch_raw * intensity_scale |
|
|
| base_speed = emotion_speed * (1.0 + (req.speed_adjust / 100.0)) |
| speed = max(0.5, min(2.0, base_speed)) |
|
|
| total_pitch = emotion_pitch + (req.pitch_adjust * 0.24) |
|
|
| logger.info( |
| f"Emotion: {dominant_emotion}, intensity: {req.intensity}, " |
| f"emotion_speed: {emotion_speed:.3f}, emotion_pitch: {emotion_pitch:.2f}, " |
| f"final_speed: {speed:.3f}, final_pitch: {total_pitch:.2f}" |
| ) |
|
|
| synth_text = req.input_text |
|
|
| language = "en" |
|
|
| if speaker_wav_path: |
| audio = tts_model.tts( |
| text=synth_text, |
| speaker_wav=speaker_wav_path, |
| language=language, |
| speed=speed, |
| ) |
| else: |
| audio = tts_model.tts( |
| text=synth_text, |
| language=language, |
| speed=speed, |
| ) |
|
|
| audio_np = np.array(audio, dtype=np.float32) |
|
|
| if abs(total_pitch) > 0.01: |
| audio_np = pyrb.pitch_shift(audio_np, SAMPLE_RATE, total_pitch) |
|
|
| vol_factor = req.volume / 75.0 |
| audio_np = audio_np * vol_factor |
|
|
| wav_bytes = numpy_to_wav_bytes(audio_np, SAMPLE_RATE) |
|
|
| return Response(content=wav_bytes, media_type="audio/wav") |
|
|
| except Exception as e: |
| logger.exception("TTS generation failed") |
| return JSONResponse( |
| status_code=500, |
| content={ |
| "error": "Audio generation failed", |
| "error_code": "GENERATION_FAILED", |
| "details": str(e) |
| } |
| ) |
| finally: |
| for f in temp_files: |
| try: |
| os.unlink(f) |
| except OSError: |
| pass |
|
|
|
|
| @app.get("/", response_class=HTMLResponse) |
| async def root(): |
| html_path = Path(__file__).parent / "index.html" |
| return HTMLResponse(content=html_path.read_text()) |
|
|
|
|
| @app.get("/health") |
| async def health(): |
| return {"status": "ok", "model_loaded": tts_model is not None} |
|
|
|
|
| if __name__ == "__main__": |
| import uvicorn |
| uvicorn.run(app, host="0.0.0.0", port=7860) |
|
|
