File size: 3,614 Bytes
e0dc3e5 b358a50 e0dc3e5 b358a50 e0dc3e5 b358a50 9cbfcc4 b358a50 9cbfcc4 e0dc3e5 9cbfcc4 b358a50 9cbfcc4 b358a50 e0dc3e5 ec1becd b358a50 e0dc3e5 b358a50 e0dc3e5 b358a50 e0dc3e5 9cbfcc4 e0dc3e5 6b18cd3 9cbfcc4 6b18cd3 9cbfcc4 e0dc3e5 9cbfcc4 b358a50 9cbfcc4 b358a50 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """
TTS — Kokoro ONNX (primary) with local edge-tts (secondary/accents).
Runs locally on the HF Space for zero-latency and reliable voice updates.
"""
import os
import asyncio
import tempfile
import threading
# ---------------------------------------------------------------------------
# Kokoro ONNX (High Quality Neural)
# ---------------------------------------------------------------------------
_kokoro = None
def _get_kokoro():
global _kokoro
if _kokoro is None:
try:
from kokoro_onnx import Kokoro
from huggingface_hub import hf_hub_download
onnx_path = hf_hub_download("thewh1teagle/kokoro-onnx", "kokoro-v0_19.onnx")
voices_path = hf_hub_download("thewh1teagle/kokoro-onnx", "voices-v0_19.bin")
_kokoro = Kokoro(onnx_path, voices_path)
print("[TTS] Kokoro ONNX loaded")
except Exception as e:
print(f"[TTS] Kokoro init failed: {e}")
return _kokoro
def _synthesize_kokoro(text: str, voice: str, speed: float) -> str | None:
model = _get_kokoro()
if model is None:
return None
try:
import soundfile as sf
samples, sr = model.create(text, voice=voice, speed=speed, lang="en-us")
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
sf.write(tmp.name, samples, sr)
tmp.close()
return tmp.name
except Exception as e:
print(f"[TTS] Kokoro synthesis error: {e}")
return None
# ---------------------------------------------------------------------------
# Local Edge-TTS (Diverse Accents / Fallback)
# ---------------------------------------------------------------------------
def _run_edge_tts(text: str, voice: str, rate: str, pitch: str, output_path: str):
"""Internal helper to run the async edge-tts communicate."""
import edge_tts
async def _amain():
communicate = edge_tts.Communicate(text, voice, rate=rate, pitch=pitch)
await communicate.save(output_path)
# Run in a new event loop since we're in a threaded environment (Gradio/FastAPI)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(_amain())
finally:
loop.close()
def _synthesize_edge_local(text: str, voice: str, rate: str, pitch: str) -> str | None:
try:
tmp = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False)
tmp_name = tmp.name
tmp.close()
# edge-tts is async, but Gradio is sync-threaded. Run it in a helper.
_run_edge_tts(text, voice, rate, pitch, tmp_name)
return tmp_name
except Exception as e:
print(f"[TTS] Local Edge-TTS error: {e}")
return None
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def synthesize(
text: str,
kokoro_voice: str | None = "af_sky",
kokoro_speed: float = 1.0,
edge_voice: str = "en-US-JennyNeural",
rate: str = "+0%",
pitch: str = "+0Hz",
) -> str | None:
"""Return path to synthesized audio file, or None on failure."""
text = text.strip()
if not text:
return None
# 1. Try Kokoro if specified
if kokoro_voice:
result = _synthesize_kokoro(text, kokoro_voice, kokoro_speed)
if result:
return result
# 2. Otherwise use local Edge-TTS (direct, no remote server)
return _synthesize_edge_local(text, edge_voice, rate, pitch)
|