Marius Mutu
00ac971044
Kyutai pocket-tts nu suportă română (vezi eval anterior). Codul rula oricum sinteza pe orice text primit; acum _synthesize_pockettts respinge diacriticele RO înainte de request, fără fallback silențios pe altă voce. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
264 lines
8.9 KiB
Python
264 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
||
"""Text-to-speech — rutează pe engine (Supertonic sau pocket-tts) pe baza catalogului de voci.
|
||
|
||
CLI:
|
||
python3 tools/tts.py --text "Salut Marius" [--voice M1] [--lang ro]
|
||
→ stdout: {"ok": true, "path": "/tmp/echo-tts-xxx.wav", "size_bytes": 12345, "engine_used": "supertonic"}
|
||
→ stdout: {"ok": false, "error": "..."}
|
||
|
||
Module:
|
||
from tools.tts import synthesize
|
||
result = synthesize("text", voice="M1", lang="ro")
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
|
||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
if str(REPO_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(REPO_ROOT))
|
||
|
||
SUPERTONIC_URL = "http://127.0.0.1:7788"
|
||
# Safety net dacă tts_voices.json lipsește/e gol — normal, catalogul e sursa de adevăr.
|
||
_FALLBACK_SUPERTONIC_VOICES = {"M1", "M2", "M3", "M4", "M5", "F1", "F2", "F3", "F4", "F5"}
|
||
DEFAULT_VOICE = "M2"
|
||
DEFAULT_LANG = "ro"
|
||
DEFAULT_ENGINE = "supertonic"
|
||
|
||
_VOICES_CATALOG_PATH = REPO_ROOT / "tts_voices.json"
|
||
|
||
# Punctuation Supertonic synthesis rejects with HTTP 500 (Romanian curly quotes,
|
||
# smart dashes, ellipsis, angle quotes). Mapped to ASCII so a stray „foo" in
|
||
# any caller's text doesn't kill the whole request.
|
||
_TTS_PUNCT_MAP = {
|
||
'„': '"', '“': '"', '”': '"',
|
||
'‘': "'", '’': "'", '‚': "'",
|
||
'«': '"', '»': '"',
|
||
'–': '-', '—': '-',
|
||
'…': '...',
|
||
}
|
||
|
||
|
||
# Supertonic ONNX model hard limit: inputs longer than this trigger
|
||
# Mul node dimension mismatches in attention layers.
|
||
_MAX_TTS_CHARS = 400
|
||
|
||
|
||
class _PocketTTSUnavailable(Exception):
|
||
"""pocket-tts server e nereachable sau eșuează server-side (5xx) — caller-ul trebuie să facă fallback pe Supertonic."""
|
||
|
||
|
||
# pocket-tts e monolingv engleză per deployment — diacriticele RO sunt un semnal
|
||
# sigur că textul n-a fost tradus (vezi personality/TOOLS.md).
|
||
_RO_DIACRITICS = set("ăâîșțĂÂÎȘȚşţŞŢ")
|
||
|
||
|
||
def _looks_romanian(text: str) -> bool:
|
||
return any(ch in _RO_DIACRITICS for ch in text)
|
||
|
||
|
||
def sanitize_for_supertonic(text: str) -> str:
|
||
"""Replace Unicode punctuation and strip chars that crash Supertonic's ONNX model."""
|
||
for src, dst in _TTS_PUNCT_MAP.items():
|
||
text = text.replace(src, dst)
|
||
# Strip emoji and high-codepoint chars (keep ASCII printable + Latin/Romanian diacritice)
|
||
cleaned = []
|
||
for ch in text:
|
||
cp = ord(ch)
|
||
if (32 <= cp <= 126) or (128 <= cp <= 591):
|
||
cleaned.append(ch)
|
||
else:
|
||
cleaned.append(' ')
|
||
text = ' '.join(''.join(cleaned).split())
|
||
if len(text) > _MAX_TTS_CHARS:
|
||
text = text[:_MAX_TTS_CHARS]
|
||
return text
|
||
|
||
|
||
def _load_voice_catalog() -> dict:
|
||
from src.jsonlock import read_locked
|
||
|
||
try:
|
||
return read_locked(str(_VOICES_CATALOG_PATH))
|
||
except FileNotFoundError:
|
||
return {}
|
||
|
||
|
||
def _catalog_entry(voice: str) -> dict | None:
|
||
catalog = _load_voice_catalog()
|
||
return catalog.get(voice) or catalog.get(voice.upper())
|
||
|
||
|
||
def list_voice_names(engine: str | None = None) -> list[str]:
|
||
"""Nume de voci din tts_voices.json, opțional filtrate pe `engine`.
|
||
|
||
Sursă live pentru autocomplete Discord (/audio, /voice setvoice) — o voce
|
||
nouă adăugată în catalog apare imediat, fără redeploy.
|
||
"""
|
||
catalog = _load_voice_catalog()
|
||
if engine is not None:
|
||
return sorted(name for name, entry in catalog.items() if entry.get("engine") == engine)
|
||
return sorted(catalog.keys())
|
||
|
||
|
||
def _config_get(key: str, default: str) -> str:
|
||
from src.config import Config
|
||
|
||
return Config().get(key, default)
|
||
|
||
|
||
def _write_temp_wav(content: bytes) -> dict:
|
||
try:
|
||
fd, path = tempfile.mkstemp(prefix="echo-tts-", suffix=".wav")
|
||
with open(fd, "wb") as f:
|
||
f.write(content)
|
||
return {"ok": True, "path": path, "size_bytes": len(content)}
|
||
except Exception as e:
|
||
return {"ok": False, "error": f"Scriere fișier: {e}"}
|
||
|
||
|
||
def _synthesize_supertonic(text: str, voice: str = DEFAULT_VOICE, lang: str = DEFAULT_LANG) -> dict:
|
||
"""Call Supertonic server and save audio to a temp WAV file.
|
||
|
||
Returns:
|
||
{"ok": True, "path": "/tmp/echo-tts-xxx.wav", "size_bytes": N}
|
||
{"ok": False, "error": "mesaj eroare"}
|
||
"""
|
||
text = sanitize_for_supertonic(text)
|
||
|
||
voice = voice.upper()
|
||
supertonic_voices = set(list_voice_names(engine="supertonic")) or _FALLBACK_SUPERTONIC_VOICES
|
||
if voice not in supertonic_voices:
|
||
voice = DEFAULT_VOICE
|
||
|
||
try:
|
||
resp = httpx.post(
|
||
f"{SUPERTONIC_URL}/v1/audio/speech",
|
||
json={
|
||
"model": "supertonic-3",
|
||
"input": text,
|
||
"voice": voice,
|
||
"response_format": "wav",
|
||
"lang": lang,
|
||
},
|
||
timeout=60.0,
|
||
)
|
||
resp.raise_for_status()
|
||
except httpx.ConnectError:
|
||
return {
|
||
"ok": False,
|
||
"error": (
|
||
"Serverul Supertonic nu rulează pe :7788. "
|
||
"Pornește cu: systemctl --user start supertonic-tts"
|
||
),
|
||
}
|
||
except httpx.HTTPStatusError as e:
|
||
body = e.response.text[:300]
|
||
# Fallback: dacă lang=ro eșuează, încearcă na (language-agnostic)
|
||
if lang != "na":
|
||
return _synthesize_supertonic(text, voice=voice, lang="na")
|
||
return {"ok": False, "error": f"HTTP {e.response.status_code}: {body}"}
|
||
except Exception as e:
|
||
return {"ok": False, "error": str(e)}
|
||
|
||
# Salvează în fișier temp
|
||
return _write_temp_wav(resp.content)
|
||
|
||
|
||
def _synthesize_pockettts(text: str, entry: dict) -> dict:
|
||
"""POST multipart la serverul pocket-tts (:7789/tts). Ignoră lang (modelul e monolingv per deployment).
|
||
|
||
`entry` e intrarea din tts_voices.json (deja rezolvată de caller — evită un al doilea read+lock
|
||
al catalogului). state_path (voce clonată) -> încarcă fișierul ca voice_wav. voice_url (voce
|
||
predefinită, ex. 'alba') -> trimite voice_url. Nici unul -> nu trimite voice_wav/voice_url
|
||
(serverul alege default-ul limbii).
|
||
|
||
Raises:
|
||
_PocketTTSUnavailable: eșec tehnic (connect error/timeout/5xx) — caller-ul face fallback.
|
||
"""
|
||
if _looks_romanian(text):
|
||
return {
|
||
"ok": False,
|
||
"error": "pocket-tts nu suportă română — textul trebuie să fie în engleză.",
|
||
}
|
||
|
||
state_path = entry.get("state_path")
|
||
voice_url = entry.get("voice_url")
|
||
|
||
data = {"text": text}
|
||
files = None
|
||
if state_path:
|
||
full_path = REPO_ROOT / state_path
|
||
if not full_path.exists():
|
||
return {"ok": False, "error": f"Fișier voce lipsă: {state_path}"}
|
||
files = {"voice_wav": (full_path.name, full_path.read_bytes())}
|
||
elif voice_url:
|
||
data["voice_url"] = voice_url
|
||
|
||
try:
|
||
resp = httpx.post(
|
||
f"{_config_get('tts.pockettts_url', 'http://127.0.0.1:7789')}/tts",
|
||
data=data,
|
||
files=files,
|
||
timeout=60.0,
|
||
)
|
||
except (httpx.ConnectError, httpx.TimeoutException) as e:
|
||
raise _PocketTTSUnavailable(str(e)) from e
|
||
|
||
if resp.status_code >= 500:
|
||
raise _PocketTTSUnavailable(f"HTTP {resp.status_code}: {resp.text[:300]}")
|
||
if resp.status_code >= 400:
|
||
return {"ok": False, "error": f"HTTP {resp.status_code}: {resp.text[:300]}"}
|
||
|
||
return _write_temp_wav(resp.content)
|
||
|
||
|
||
def synthesize(text: str, voice: str = DEFAULT_VOICE, lang: str = DEFAULT_LANG) -> dict:
|
||
"""Rutează sinteza pe engine-ul rezolvat din catalog (fallback la tts.default_engine).
|
||
|
||
Returns:
|
||
{"ok": True, "path": "/tmp/echo-tts-xxx.wav", "size_bytes": N, "engine_used": "pockettts"}
|
||
{"ok": False, "error": "mesaj eroare", "engine_used": "..."}
|
||
"""
|
||
if not text or not text.strip():
|
||
return {"ok": False, "error": "Text gol."}
|
||
|
||
entry = _catalog_entry(voice) or {}
|
||
engine = entry.get("engine") or _config_get("tts.default_engine", DEFAULT_ENGINE)
|
||
|
||
engine_used = "supertonic"
|
||
if engine == "pockettts":
|
||
try:
|
||
result = _synthesize_pockettts(text, entry)
|
||
engine_used = "pockettts"
|
||
except _PocketTTSUnavailable:
|
||
result = _synthesize_supertonic(text, voice=DEFAULT_VOICE, lang=DEFAULT_LANG)
|
||
else:
|
||
result = _synthesize_supertonic(text, voice=voice, lang=lang)
|
||
|
||
result["engine_used"] = engine_used
|
||
return result
|
||
|
||
|
||
if __name__ == "__main__":
|
||
parser = argparse.ArgumentParser(description="Echo Core TTS CLI")
|
||
parser.add_argument("--text", required=True, help="Text de convertit în audio")
|
||
parser.add_argument(
|
||
"--voice", default=DEFAULT_VOICE,
|
||
help="Voce din tts_voices.json (ex: M1-M5, F1-F5, 'Marius 1', 'alba'). Default: M2"
|
||
)
|
||
parser.add_argument(
|
||
"--lang", default=DEFAULT_LANG,
|
||
help="Limbă (ro, en, na) — folosită doar pe calea Supertonic. Default: ro."
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
result = synthesize(args.text, voice=args.voice, lang=args.lang)
|
||
print(json.dumps(result, ensure_ascii=False))
|
||
sys.exit(0 if result.get("ok") else 1)
|