Spaces:
Running on Zero
Running on Zero
| """FastAPI server wrapping edge-tts for use with Flutter or any HTTP client.""" | |
| import asyncio | |
| import hashlib | |
| import os | |
| import socket | |
| import sys | |
| import unicodedata | |
| import uuid | |
| import aiohttp | |
| from collections import OrderedDict | |
| from contextlib import asynccontextmanager | |
| from pathlib import Path | |
| from typing import Literal | |
| from fastapi import FastAPI, File, Form, HTTPException, Query, Request, Response, UploadFile | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from fastapi.responses import RedirectResponse | |
| from pydantic import BaseModel, Field | |
| from audio_cache_store import atomic_write, backup_audio, read_audio | |
| from tts_delivery import AudioEncodingError, audio_response, boost_model_wav, change_wav_tempo, encode_mp3, normalize_voice_wav, tempo_from_rate | |
| # To run the server, use the following command: | |
| # uvicorn server:app --host 0.0.0.0 --port 8000 | |
| # | |
| # LDPlayer / Android emulator: connect from inside emulator via http://10.0.2.2:8000 | |
| # Local browser / same machine: http://127.0.0.1:8000 | |
| # Use the repository's patched Edge client for every entry point, including uvicorn. | |
| sys.path.insert(0, str(Path(__file__).resolve().parent / "src")) | |
| import edge_tts | |
| from edge_tts.exceptions import EdgeTTSException, NoAudioReceived | |
| from vtts_backend import ( | |
| VTTSBackendError, | |
| VTTSInvalidVoiceIdError, | |
| VTTSVoiceExistsError, | |
| VTTSVoiceNotFoundError, | |
| vtts_backend, | |
| ) | |
| from zerotts_backend import ( | |
| ZeroTTSBackendError, | |
| ZeroTTSInvalidVoiceIdError, | |
| ZeroTTSVoiceNotFoundError, | |
| zerotts_backend, | |
| ) | |
| from vieneu_backend import ( | |
| VieNeuBackendError, | |
| VieNeuVoiceNotFoundError, | |
| vieneu_backend, | |
| ) | |
| from gwen_backend import GwenBackendError, GwenVoiceNotFoundError, gwen_backend | |
| # Set True to override client voice IDs; set FIXED_VOICE_ID to your installed ID. | |
| # FIXED_PROVIDER also overrides /tts provider when FORCE_VOICE_ID is True. | |
| # None preserves automatic/client routing; otherwise use edge/vtts/zerotts/vieneu/gwen. | |
| # Restart the server after changing these settings. | |
| FORCE_VOICE_ID = False | |
| FIXED_VOICE_ID = "Mỹ Duyên" | |
| FIXED_PROVIDER: str | None = "vieneu" | |
| def _effective_voice_id(requested_voice_id: str | None) -> str | None: | |
| if not FORCE_VOICE_ID: | |
| return requested_voice_id | |
| fixed_voice_id = FIXED_VOICE_ID.strip() | |
| if not fixed_voice_id: | |
| raise HTTPException( | |
| status_code=503, | |
| detail="FIXED_VOICE_ID must be set when FORCE_VOICE_ID is enabled", | |
| ) | |
| return fixed_voice_id | |
| async def lifespan(_app: FastAPI): | |
| port = int(os.getenv("PORT", "8000")) | |
| try: | |
| s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) | |
| s.connect(("8.8.8.8", 80)) | |
| lan_ip = s.getsockname()[0] | |
| s.close() | |
| except Exception: | |
| lan_ip = "unavailable" | |
| print("\n" + "=" * 50) | |
| print(" edge-tts server is running!") | |
| print("=" * 50) | |
| print(f" Local (same machine) : http://127.0.0.1:{port}") | |
| print(f" LDPlayer / Emulator : http://10.0.2.2:{port}") | |
| print(f" LAN (other devices) : http://{lan_ip}:{port}") | |
| from zerotts_suffix import HiddenSuffixConfig | |
| suffix_enabled = HiddenSuffixConfig.from_env().enabled | |
| print(f" ZeroTTS hidden suffix: {'ON' if suffix_enabled else 'OFF'}") | |
| print("=" * 50 + "\n") | |
| # Load persisted audio cache from disk. | |
| _load_disk_cache() | |
| # Pre-warm connection to Microsoft TTS servers by fully completing a request. | |
| # This puts the TCP/TLS connection back into aiohttp's pool for reuse. | |
| async def warm_edge(): | |
| communicate = edge_tts.Communicate(text=".", voice="vi-VN-HoaiMyNeural") | |
| async for _ in communicate.stream(): | |
| pass # Consume all chunks so the connection closes cleanly and is pooled | |
| print(" [warmup] Connection to TTS server pre-warmed.\n") | |
| if os.getenv("EDGE_WARMUP", "1").lower() not in {"0", "false", "no"}: | |
| try: | |
| await asyncio.wait_for(warm_edge(), timeout=5) | |
| except Exception: | |
| pass # Warmup failure must not block startup. | |
| try: | |
| yield | |
| finally: | |
| await asyncio.to_thread(gwen_backend.close) | |
| app = FastAPI(title="edge-tts API vCache", lifespan=lifespan) | |
| async def api_home(): | |
| return RedirectResponse("/docs") | |
| async def health(): | |
| """Liveness check that does not load models or call external services.""" | |
| return {"status": "ok"} | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| expose_headers=["ETag", "Accept-Ranges", "Content-Range", "Content-Length", "X-Audio-Duration-Seconds"], | |
| ) | |
| # --------------------------------------------------------------------------- | |
| # Audio LRU cache — persisted to disk at models/cache/<key>.mp3 | |
| # Key: MD5 of (text, voice, rate, volume, pitch). | |
| # - Survives server restarts (loaded from disk on startup). | |
| # - Hot LRU in RAM is limited to 64 MiB; disk retains generated audio. | |
| # --------------------------------------------------------------------------- | |
| _CACHE_DIR = Path(os.getenv("EDGE_AUDIO_CACHE_DIR", "models/cache")) | |
| _EDGE_RAM_MAX_BYTES = 64 * 1024 * 1024 | |
| _audio_cache: OrderedDict[str, bytes] = OrderedDict() # in-RAM hot cache | |
| _cache_locks: dict[str, asyncio.Lock] = {} | |
| def _make_cache_key(text: str, voice: str, rate: str, volume: str, pitch: str) -> str: | |
| return hashlib.md5(f"{text}|{voice}|{rate}|{volume}|{pitch}".encode()).hexdigest() | |
| def _cache_path(key: str) -> Path: | |
| return _CACHE_DIR / f"{key}.mp3" | |
| def _load_disk_cache() -> None: | |
| """On startup, read existing cache files into the in-RAM OrderedDict. | |
| Files are ordered by modification time (oldest first) to respect LRU eviction.""" | |
| _CACHE_DIR.mkdir(parents=True, exist_ok=True) | |
| files = sorted(_CACHE_DIR.glob("*.mp3"), key=lambda p: p.stat().st_mtime) | |
| budget = _EDGE_RAM_MAX_BYTES | |
| selected = [] | |
| for f in reversed(files): | |
| size = f.stat().st_size | |
| if size <= budget: | |
| selected.append(f) | |
| budget -= size | |
| for f in reversed(selected): | |
| _remember_edge_audio(f.stem, f.read_bytes()) | |
| if _audio_cache: | |
| print(f" [cache] Loaded {len(_audio_cache)} cached audio file(s) from disk.\n") | |
| def _save_to_disk(key: str, data: bytes) -> None: | |
| atomic_write(_cache_path(key), data) | |
| backup_audio(_cache_path(key), "edge", data) | |
| def _remember_edge_audio(key: str, data: bytes) -> None: | |
| _audio_cache.pop(key, None) | |
| if len(data) > _EDGE_RAM_MAX_BYTES: | |
| return | |
| _audio_cache[key] = data | |
| while sum(map(len, _audio_cache.values())) > _EDGE_RAM_MAX_BYTES: | |
| _audio_cache.popitem(last=False) | |
| async def _get_audio(text: str, voice: str, rate: str, volume: str, pitch: str) -> bytes: | |
| """Return cached audio (RAM → disk → generate), persist new results to disk.""" | |
| key = _make_cache_key(text, voice, rate, volume, pitch) | |
| # 1. RAM hit | |
| if key in _audio_cache: | |
| _audio_cache.move_to_end(key) | |
| return _audio_cache[key] | |
| # One lock per key prevents duplicate generation for concurrent requests. | |
| if key not in _cache_locks: | |
| _cache_locks[key] = asyncio.Lock() | |
| async with _cache_locks[key]: | |
| # Re-check inside the lock. | |
| if key in _audio_cache: | |
| _audio_cache.move_to_end(key) | |
| return _audio_cache[key] | |
| # 2. Disk hit (entry evicted from RAM but file still present) | |
| disk_file = _cache_path(key) | |
| try: | |
| audio_data = await asyncio.to_thread(read_audio, disk_file, "edge") | |
| # Touch the file so its mtime reflects recent use | |
| if disk_file.exists(): | |
| os.utime(disk_file) | |
| except OSError: | |
| # 3. Generate from Microsoft TTS | |
| for attempt in range(3): | |
| try: | |
| communicate = edge_tts.Communicate( | |
| text=text, voice=voice, rate=rate, volume=volume, pitch=pitch | |
| ) | |
| chunks: list[bytes] = [] | |
| async for chunk in communicate.stream(): | |
| if chunk["type"] == "audio": | |
| chunks.append(chunk["data"]) | |
| audio_data = b"".join(chunks) | |
| if not audio_data: | |
| raise NoAudioReceived("Edge TTS returned empty audio") | |
| break | |
| except (NoAudioReceived, aiohttp.ClientConnectionError, asyncio.TimeoutError): | |
| if attempt == 2: | |
| raise | |
| await asyncio.sleep(0.3 * (attempt + 1)) | |
| await asyncio.to_thread(_save_to_disk, key, audio_data) | |
| _remember_edge_audio(key, audio_data) | |
| _cache_locks.pop(key, None) | |
| return audio_data | |
| _TTS_HEADERS = { | |
| "Content-Disposition": "inline; filename=tts.mp3", | |
| "Cache-Control": "no-store", | |
| } | |
| _VTTS_HEADERS = { | |
| "Content-Disposition": "inline; filename=vtts-cloned.wav", | |
| "Cache-Control": "no-store", | |
| } | |
| _VTTS_TEMP_DIR = Path(os.getenv("VTTS_TEMP_DIR", "models/vtts/tmp")) | |
| _VTTS_MAX_REFERENCE_BYTES = int( | |
| os.getenv("VTTS_MAX_REFERENCE_BYTES", str(25 * 1024 * 1024)) | |
| ) | |
| _VTTS_ALLOWED_SUFFIXES = {".wav", ".mp3", ".flac", ".ogg"} | |
| _vtts_inference_lock = asyncio.Lock() | |
| # ZeroTTS serializes cache misses in its backend; cached audio bypasses inference. | |
| # Applied after source-cache lookup for both old and newly synthesized audio. | |
| MODEL_AUDIO_GAIN = float(os.getenv("MODEL_AUDIO_GAIN", "1.5")) | |
| VOICE_LOUDNESS_LUFS = float(os.getenv("VOICE_LOUDNESS_LUFS", "-11.0")) | |
| def _balanced_voice(provider: str | None, voice_id: str | None) -> bool: | |
| name = unicodedata.normalize("NFC", voice_id or "").strip().casefold() | |
| name = "".join(char for char in name if char not in " _-") | |
| return ( | |
| provider == "zerotts" and name in { | |
| "hamy", "hàmy", "tiendatchild", "quangvinhchild", "vpyllcqegcohgycjstnpchild", | |
| } | |
| ) or (provider == "vieneu" and name in {"mỹduyên", "myduyen"}) | |
| async def _boost_model_audio(audio: bytes, provider: str | None = None, voice_id: str | None = None) -> bytes: | |
| try: | |
| if MODEL_AUDIO_GAIN != 1.0 and _balanced_voice(provider, voice_id): | |
| cache_dir = (zerotts_backend._audio_cache_dir() if provider == "zerotts" | |
| else vieneu_backend.audio_cache_dir()) | |
| return await asyncio.to_thread(normalize_voice_wav, audio, cache_dir, VOICE_LOUDNESS_LUFS) | |
| return await asyncio.to_thread(boost_model_wav, audio, MODEL_AUDIO_GAIN) | |
| except AudioEncodingError as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| async def _zerotts_audio_response( | |
| audio: bytes, format: str, request: Request | None, voice_id: str | None = None, | |
| ) -> Response: | |
| audio = await _boost_model_audio(audio, "zerotts", voice_id) | |
| if format == "mp3": | |
| try: | |
| audio = await asyncio.to_thread(encode_mp3, audio, zerotts_backend._audio_cache_dir()) | |
| except AudioEncodingError as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| return audio_response( | |
| audio, "audio/mpeg" if format == "mp3" else "audio/wav", | |
| f"zerotts.{format}", request, | |
| ) | |
| async def _save_vtts_reference(reference_audio: UploadFile) -> Path: | |
| """Validate one uploaded reference clip and save it to a unique temp file.""" | |
| suffix = Path(reference_audio.filename or "reference.wav").suffix.lower() | |
| if suffix not in _VTTS_ALLOWED_SUFFIXES: | |
| allowed = ", ".join(sorted(_VTTS_ALLOWED_SUFFIXES)) | |
| raise HTTPException( | |
| status_code=400, | |
| detail=f"reference_audio must use one of these formats: {allowed}", | |
| ) | |
| reference_data = await reference_audio.read(_VTTS_MAX_REFERENCE_BYTES + 1) | |
| await reference_audio.close() | |
| if not reference_data: | |
| raise HTTPException(status_code=400, detail="reference_audio must not be empty") | |
| if len(reference_data) > _VTTS_MAX_REFERENCE_BYTES: | |
| raise HTTPException(status_code=413, detail="reference_audio is too large") | |
| _VTTS_TEMP_DIR.mkdir(parents=True, exist_ok=True) | |
| reference_path = _VTTS_TEMP_DIR / f"{uuid.uuid4().hex}{suffix}" | |
| reference_path.write_bytes(reference_data) | |
| return reference_path | |
| class TTSRequest(BaseModel): | |
| text: str | |
| voice: str = "vi-VN-HoaiMyNeural" | |
| voiceid: str | None = None | |
| voice_id: str | None = None | |
| provider: str | None = None | |
| format: Literal["wav", "mp3"] | None = None | |
| eoa_extra_frames: int = Field(default=1, ge=1, le=12) | |
| seed: int | None = Field(default=None, ge=0, le=4294967295, | |
| description="ZeroTTS sampling seed; omit to use the server default") | |
| rate: str = "-25%" | |
| volume: str = "+100%" | |
| pitch: str = "+0Hz" | |
| class VTTSSynthesisRequest(BaseModel): | |
| text: str | |
| voice_id: str | |
| length_scale: float = Field(default=1.0, ge=0.25, le=4.0) | |
| noise_scale: float = Field(default=0.667, ge=0.0, le=2.0) | |
| noise_scale_w: float = Field(default=0.8, ge=0.0, le=2.0) | |
| class ZeroTTSSynthesisRequest(BaseModel): | |
| text: str | |
| voice_id: str | |
| format: Literal["wav", "mp3"] = "wav" | |
| cfg_scale: float = Field(default=1.0, ge=0.0, le=5.0) | |
| audio_temperature: float = Field(default=0.8, gt=0.0, le=2.0) | |
| audio_repetition_penalty: float = Field(default=1.2, ge=1.0, le=2.0) | |
| max_chunk_sec: float = Field( | |
| default=15.0, ge=1.0, le=30.0, | |
| description="Estimated maximum seconds per sentence chunk; actual duration may vary", | |
| ) | |
| max_frames: int = Field(default=1500, ge=4, le=1500) | |
| eoa_extra_frames: int = Field(default=1, ge=1, le=12) | |
| seed: int | None = Field(default=None, ge=0, le=4294967295, | |
| description="ZeroTTS sampling seed; omit to use the server default") | |
| async def _synthesize_tts_response( | |
| *, | |
| text: str, | |
| voice: str, | |
| voiceid: str | None, | |
| provider: str | None, | |
| rate: str, | |
| volume: str, | |
| pitch: str, | |
| eoa_extra_frames: int, | |
| seed: int | None = None, | |
| format: Literal["wav", "mp3"] | None = None, | |
| request: Request | None = None, | |
| rate_is_explicit: bool = False, | |
| ) -> Response: | |
| """Route a unified /tts request to the selected speech engine.""" | |
| if not text.strip(): | |
| raise HTTPException(status_code=400, detail="text must not be empty") | |
| voiceid = _effective_voice_id(voiceid) | |
| provider_is_fixed = FORCE_VOICE_ID and FIXED_PROVIDER is not None | |
| if provider_is_fixed: | |
| provider = FIXED_PROVIDER | |
| elif FORCE_VOICE_ID and provider and provider.strip().lower() == "edge": | |
| # Edge cannot use registered IDs; preserve explicit ZeroTTS/V-TTS routing. | |
| provider = None | |
| selected_provider = provider.strip().lower() if provider else None | |
| if (selected_provider not in {None, "edge", "vtts", "zerotts", "vieneu", "gwen"} | |
| or (provider_is_fixed and selected_provider is None)): | |
| raise HTTPException( | |
| status_code=503 if provider_is_fixed else 400, | |
| detail=("FIXED_PROVIDER" if provider_is_fixed else "provider") | |
| + " must be edge, vtts, zerotts, vieneu, or gwen", | |
| ) | |
| if provider_is_fixed and selected_provider == "edge": | |
| voice = voiceid | |
| resolved_voiceid = voiceid.strip() if voiceid else None | |
| if selected_provider in {"vieneu", "gwen"}: | |
| backend = gwen_backend if selected_provider == "gwen" else vieneu_backend | |
| tempo = 1.0 | |
| if selected_provider == "gwen" and rate_is_explicit: | |
| try: | |
| tempo = tempo_from_rate(rate) | |
| except ValueError as exc: | |
| raise HTTPException(status_code=400, detail=str(exc)) from exc | |
| if not resolved_voiceid and voice != "vi-VN-HoaiMyNeural": | |
| resolved_voiceid = voice.strip() or None | |
| try: | |
| audio = await asyncio.to_thread( | |
| backend.synthesize_wav, text.strip(), resolved_voiceid | |
| ) | |
| if tempo != 1.0: | |
| audio = await asyncio.to_thread(change_wav_tempo, audio, tempo, backend.audio_cache_dir()) | |
| audio = await _boost_model_audio(audio, selected_provider, resolved_voiceid) | |
| output_format = format or "wav" | |
| if output_format == "mp3": | |
| audio = await asyncio.to_thread( | |
| encode_mp3, audio, backend.audio_cache_dir() | |
| ) | |
| except (VieNeuVoiceNotFoundError, GwenVoiceNotFoundError) as exc: | |
| raise HTTPException(status_code=404, detail=str(exc)) from exc | |
| except (VieNeuBackendError, GwenBackendError, AudioEncodingError) as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| return audio_response( | |
| audio, "audio/mpeg" if output_format == "mp3" else "audio/wav", | |
| f"{selected_provider}.{output_format}", request, | |
| ) | |
| if resolved_voiceid is None and selected_provider in {"vtts", "zerotts"}: | |
| if voice != "vi-VN-HoaiMyNeural": | |
| resolved_voiceid = voice | |
| if ( | |
| resolved_voiceid is None | |
| and selected_provider is None | |
| and zerotts_backend.has_local_voice(voice) | |
| ): | |
| resolved_voiceid = voice | |
| if selected_provider in {"vtts", "zerotts"} and not resolved_voiceid: | |
| raise HTTPException( | |
| status_code=400, detail=f"voiceid is required for provider={selected_provider}" | |
| ) | |
| use_zerotts = bool( | |
| resolved_voiceid | |
| and ( | |
| selected_provider == "zerotts" | |
| or ( | |
| selected_provider is None | |
| and zerotts_backend.has_local_voice(resolved_voiceid) | |
| ) | |
| ) | |
| ) | |
| if use_zerotts: | |
| try: | |
| audio = await asyncio.to_thread( | |
| zerotts_backend.synthesize_wav, | |
| text.strip(), | |
| resolved_voiceid, | |
| eoa_extra_frames=eoa_extra_frames, | |
| **({"seed": seed} if seed is not None else {}), | |
| ) | |
| except ZeroTTSVoiceNotFoundError as exc: | |
| raise HTTPException(status_code=404, detail=str(exc)) from exc | |
| except ZeroTTSInvalidVoiceIdError as exc: | |
| raise HTTPException(status_code=400, detail=str(exc)) from exc | |
| except ZeroTTSBackendError as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| return await _zerotts_audio_response(audio, format or "wav", request, resolved_voiceid) | |
| if format is not None: | |
| raise HTTPException(status_code=400, detail="format is supported only for ZeroTTS, VieNeu, and Gwen") | |
| if resolved_voiceid and selected_provider != "edge": | |
| try: | |
| async with _vtts_inference_lock: | |
| audio = await asyncio.to_thread( | |
| vtts_backend.synthesize_registered_voice, | |
| text.strip(), | |
| resolved_voiceid, | |
| ) | |
| except VTTSVoiceNotFoundError as exc: | |
| raise HTTPException(status_code=404, detail=str(exc)) from exc | |
| except VTTSInvalidVoiceIdError as exc: | |
| raise HTTPException(status_code=400, detail=str(exc)) from exc | |
| except VTTSBackendError as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| audio = await _boost_model_audio(audio) | |
| return Response(content=audio, media_type="audio/wav", headers=_VTTS_HEADERS) | |
| try: | |
| audio = await _get_audio(text, voice, rate, volume, pitch) | |
| except ValueError as exc: | |
| raise HTTPException(status_code=400, detail=f"Invalid Edge TTS settings: {exc}") from exc | |
| except (EdgeTTSException, aiohttp.ClientError, asyncio.TimeoutError) as exc: | |
| raise HTTPException( | |
| status_code=502, | |
| detail=f"Edge TTS could not generate audio: {type(exc).__name__}: {exc}", | |
| ) from exc | |
| return Response(content=audio, media_type="audio/mpeg", headers=_TTS_HEADERS) | |
| async def text_to_speech(req: TTSRequest, request: Request): | |
| """Unified JSON endpoint for Edge, V-TTS, ZeroTTS, VieNeu, and Gwen.""" | |
| if not FORCE_VOICE_ID and req.voiceid and req.voice_id and req.voiceid != req.voice_id: | |
| raise HTTPException(status_code=400, detail="voiceid and voice_id must match") | |
| return await _synthesize_tts_response( | |
| text=req.text, | |
| voice=req.voice, | |
| voiceid=req.voiceid or req.voice_id, | |
| provider=req.provider, | |
| rate=req.rate, | |
| rate_is_explicit="rate" in req.model_fields_set, | |
| volume=req.volume, | |
| pitch=req.pitch, | |
| eoa_extra_frames=req.eoa_extra_frames, | |
| seed=req.seed, | |
| format=req.format, | |
| request=request, | |
| ) | |
| async def text_to_speech_get( | |
| request: Request, | |
| text: str = Query(description="Text to synthesize"), | |
| voice: str = Query(default="vi-VN-HoaiMyNeural"), | |
| voiceid: str | None = Query( | |
| default=None, description="V-TTS/ZeroTTS voice ID or VieNeu/Gwen preset name" | |
| ), | |
| voice_id: str | None = Query(default=None, description="Alias of voiceid"), | |
| provider: str | None = Query( | |
| default=None, description="Optional provider: edge, vtts, zerotts, vieneu, or gwen" | |
| ), | |
| rate: str = Query(default="-10%"), | |
| volume: str = Query(default="+100%"), | |
| pitch: str = Query(default="+0Hz"), | |
| eoa_extra_frames: int = Query(default=1, ge=1, le=12), | |
| seed: int | None = Query(default=None, ge=0, le=4294967295, | |
| description="ZeroTTS sampling seed; omit to use the server default"), | |
| format: Literal["wav", "mp3"] | None = Query(default=None, description="ZeroTTS/VieNeu/Gwen output format; default wav"), | |
| ): | |
| """Unified query endpoint for Edge, V-TTS, ZeroTTS, VieNeu, and Gwen.""" | |
| if not FORCE_VOICE_ID and voiceid and voice_id and voiceid != voice_id: | |
| raise HTTPException(status_code=400, detail="voiceid and voice_id must match") | |
| return await _synthesize_tts_response( | |
| text=text, | |
| voice=voice, | |
| voiceid=voiceid or voice_id, | |
| provider=provider, | |
| rate=rate, | |
| rate_is_explicit="rate" in request.query_params, | |
| volume=volume, | |
| pitch=pitch, | |
| eoa_extra_frames=eoa_extra_frames, | |
| seed=seed, | |
| format=format, | |
| request=request, | |
| ) | |
| async def clone_voice_vtts( | |
| text: str = Form(...), | |
| reference_audio: UploadFile = File(...), | |
| length_scale: float = Form(default=1.0), | |
| noise_scale: float = Form(default=0.667), | |
| noise_scale_w: float = Form(default=0.8), | |
| ): | |
| """Clone a voice from a short reference clip and return one complete WAV.""" | |
| text = text.strip() | |
| if not text: | |
| raise HTTPException(status_code=400, detail="text must not be empty") | |
| if not 0.25 <= length_scale <= 4.0: | |
| raise HTTPException( | |
| status_code=400, detail="length_scale must be between 0.25 and 4.0" | |
| ) | |
| if not 0.0 <= noise_scale <= 2.0 or not 0.0 <= noise_scale_w <= 2.0: | |
| raise HTTPException( | |
| status_code=400, | |
| detail="noise_scale and noise_scale_w must be between 0.0 and 2.0", | |
| ) | |
| reference_path = await _save_vtts_reference(reference_audio) | |
| try: | |
| async with _vtts_inference_lock: | |
| audio = await asyncio.to_thread( | |
| vtts_backend.synthesize_wav, | |
| text, | |
| reference_path, | |
| length_scale=length_scale, | |
| noise_scale=noise_scale, | |
| noise_scale_w=noise_scale_w, | |
| ) | |
| except VTTSBackendError as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| finally: | |
| reference_path.unlink(missing_ok=True) | |
| audio = await _boost_model_audio(audio) | |
| return Response(content=audio, media_type="audio/wav", headers=_VTTS_HEADERS) | |
| async def register_vtts_voice( | |
| reference_audio: UploadFile = File(...), | |
| voice_id: str | None = Form(default=None), | |
| ): | |
| """Register a reference clip once and persist its voice embeddings.""" | |
| reference_path = await _save_vtts_reference(reference_audio) | |
| try: | |
| async with _vtts_inference_lock: | |
| registered_id = await asyncio.to_thread( | |
| vtts_backend.register_voice, reference_path, voice_id | |
| ) | |
| except VTTSVoiceExistsError as exc: | |
| raise HTTPException(status_code=409, detail=str(exc)) from exc | |
| except VTTSInvalidVoiceIdError as exc: | |
| raise HTTPException(status_code=400, detail=str(exc)) from exc | |
| except VTTSBackendError as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| finally: | |
| reference_path.unlink(missing_ok=True) | |
| return {"voice_id": registered_id} | |
| async def list_vtts_voices(): | |
| """List voice profiles registered on this server.""" | |
| return {"voices": vtts_backend.list_voices()} | |
| async def synthesize_registered_vtts_voice(req: VTTSSynthesisRequest): | |
| """Synthesize a complete WAV using text and a registered voice ID.""" | |
| text = req.text.strip() | |
| if not text: | |
| raise HTTPException(status_code=400, detail="text must not be empty") | |
| voice_id = _effective_voice_id(req.voice_id) | |
| try: | |
| async with _vtts_inference_lock: | |
| audio = await asyncio.to_thread( | |
| vtts_backend.synthesize_registered_voice, | |
| text, | |
| voice_id, | |
| length_scale=req.length_scale, | |
| noise_scale=req.noise_scale, | |
| noise_scale_w=req.noise_scale_w, | |
| ) | |
| except VTTSVoiceNotFoundError as exc: | |
| raise HTTPException(status_code=404, detail=str(exc)) from exc | |
| except VTTSInvalidVoiceIdError as exc: | |
| raise HTTPException(status_code=400, detail=str(exc)) from exc | |
| except VTTSBackendError as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| audio = await _boost_model_audio(audio) | |
| return Response(content=audio, media_type="audio/wav", headers=_VTTS_HEADERS) | |
| async def list_zerotts_voices(): | |
| """List installed ZeroTTS packs without forcing a model download.""" | |
| return {"voices": zerotts_backend.list_voices()} | |
| async def synthesize_zerotts_voice(req: ZeroTTSSynthesisRequest, request: Request): | |
| """Synthesize a complete WAV or optional MP3 using an installed voice pack.""" | |
| text = req.text.strip() | |
| if not text: | |
| raise HTTPException(status_code=400, detail="text must not be empty") | |
| voice_id = _effective_voice_id(req.voice_id) | |
| try: | |
| audio = await asyncio.to_thread( | |
| zerotts_backend.synthesize_wav, | |
| text, | |
| voice_id, | |
| cfg_scale=req.cfg_scale, | |
| audio_temperature=req.audio_temperature, | |
| audio_repetition_penalty=req.audio_repetition_penalty, | |
| max_chunk_sec=req.max_chunk_sec, | |
| max_frames=req.max_frames, | |
| eoa_extra_frames=req.eoa_extra_frames, | |
| **({"seed": req.seed} if req.seed is not None else {}), | |
| ) | |
| except ZeroTTSVoiceNotFoundError as exc: | |
| raise HTTPException(status_code=404, detail=str(exc)) from exc | |
| except ZeroTTSInvalidVoiceIdError as exc: | |
| raise HTTPException(status_code=400, detail=str(exc)) from exc | |
| except ZeroTTSBackendError as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| return await _zerotts_audio_response(audio, req.format, request, voice_id) | |
| async def list_gwen_voices(): | |
| """List bundled Gwen speakers without loading or downloading the model.""" | |
| try: | |
| return {"voices": gwen_backend.list_voices()} | |
| except GwenBackendError as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| async def list_vieneu_voices(): | |
| """List VieNeu v3 Turbo presets; the first call loads the model.""" | |
| try: | |
| voices = await asyncio.to_thread(vieneu_backend.list_voices) | |
| except VieNeuBackendError as exc: | |
| raise HTTPException(status_code=503, detail=str(exc)) from exc | |
| return {"voices": voices} | |
| async def list_voices(locale: str = Query(default=None, description="Filter by locale, e.g. vi-VN")): | |
| """List all available voices, optionally filtered by locale.""" | |
| voices = await edge_tts.list_voices() | |
| if locale: | |
| voices = [v for v in voices if v["Locale"].lower().startswith(locale.lower())] | |
| return voices | |