#!/usr/bin/env python3 """Private XTTS-first TTS gateway with a Piper fallback. The gateway implements the narrow /status and /tts protocol already consumed by the profile router. Request text is never logged or persisted. """ from __future__ import annotations import io import json import os import re import subprocess import threading import time import urllib.error import urllib.request import wave from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer HOST = os.getenv("TTS_GATEWAY_HOST", "0.0.0.0") PORT = int(os.getenv("TTS_GATEWAY_PORT", "8085")) XTTS_URL = os.getenv("XTTS_URL", "http://xtts:80").rstrip("/") PIPER_URL = os.getenv("PIPER_URL", "http://piper:8085").rstrip("/") VOICE_ALIAS = os.getenv("TTS_VOICE_ALIAS", "alloy") XTTS_SPEAKER = os.getenv("XTTS_SPEAKER", "Annmarie Nele") DEFAULT_LANGUAGE = os.getenv("TTS_DEFAULT_LANGUAGE", "de") MAX_TEXT_CHARS = int(os.getenv("TTS_MAX_TEXT_CHARS", "8000")) MAX_REQUEST_BYTES = int(os.getenv("TTS_MAX_REQUEST_BYTES", "65536")) MAX_AUDIO_BYTES = int(os.getenv("TTS_MAX_AUDIO_BYTES", str(64 * 1024 * 1024))) XTTS_TIMEOUT = float(os.getenv("XTTS_TIMEOUT", "120")) PIPER_TIMEOUT = float(os.getenv("PIPER_TIMEOUT", "120")) QUEUE_TIMEOUT = float(os.getenv("XTTS_QUEUE_TIMEOUT", "15")) SILENCE_MS = int(os.getenv("XTTS_SEGMENT_SILENCE_MS", "20")) SYNTHESIS_LOCK = threading.Lock() STATE_LOCK = threading.Lock() SPEAKER_LOCK = threading.Lock() SPEAKER_CONDITIONING: dict | None = None STATE = { "last_backend": None, "xtts_failures": 0, "piper_fallbacks": 0, "last_error": None, } # Prefer full compounds to isolated terms. This keeps switches infrequent and # avoids making mixed-language speech sound like a sequence of separate clips. ENGLISH_TERMS = ( "Home Assistant", "Open WebUI", "OpenWebUI", "Unraid Dashboard", "Unraid-Dashboard", "Docker Container", "Docker-Container", "Server Log", "Server-Log", "GitHub Repository", "GitHub Repo", "WireGuard Tunnel", "Cron Job", "Cronjob", "Home Server", "API Key", "Tool Calling", "Context Window", "Prompt Injection", "Unraid", "Docker", "Container", "Dashboard", "Server", "Log", "OpenAI", "GitHub", "WireGuard", "Linux", "Debian", "Frontend", "Backend", "Router", "Browser", "Web", "Token", "Prompt", "Context", "Model", "Image", "Tool", "Workflow", "Benchmark", "Streaming", "SSH", "MCP", "API", "CPU", "GPU", "VRAM", "RAM", "HTTP", "HTTPS", ) TERM_PATTERN = re.compile( r"(? tuple[bytes, str]: data = None headers = {} method = "GET" if payload is not None: data = json.dumps(payload, separators=(",", ":")).encode() headers["Content-Type"] = "application/json" method = "POST" request = urllib.request.Request( url, data=data, headers=headers, method=method) with urllib.request.urlopen(request, timeout=timeout) as response: body = response.read(MAX_AUDIO_BYTES + 1) if len(body) > MAX_AUDIO_BYTES: raise RuntimeError("upstream audio response is too large") return body, response.headers.get_content_type() def _json(url: str, timeout: float = 10) -> dict | list: body, _ = _request(url, timeout=timeout) return json.loads(body) def _reachable(url: str, path: str, timeout: float = 2) -> bool: try: _request(f"{url}{path}", timeout=timeout) return True except (OSError, ValueError, RuntimeError, urllib.error.URLError): return False def _looks_english(text: str) -> bool: words = re.findall(r"[A-Za-zÀ-ÿ]+", text.lower()) if not words: return False german = sum(word in GERMAN_MARKERS for word in words) english = sum(word in ENGLISH_MARKERS for word in words) return english >= 2 and english > german * 1.5 and not re.search(r"[äöüß]", text.lower()) def segment_languages(text: str) -> list[tuple[str, str]]: """Return a compact German/English segment sequence.""" if _looks_english(text): return [("en", text)] if DEFAULT_LANGUAGE != "de": return [(DEFAULT_LANGUAGE, text)] segments: list[tuple[str, str]] = [] cursor = 0 for match in TERM_PATTERN.finditer(text): if match.start() > cursor: segments.append(("de", text[cursor:match.start()])) segments.append(("en", match.group(0))) cursor = match.end() if cursor < len(text): segments.append(("de", text[cursor:])) if not segments: return [("de", text)] merged: list[tuple[str, str]] = [] for language, part in segments: if not part: continue if merged and merged[-1][0] == language: previous_language, previous_text = merged[-1] merged[-1] = (previous_language, previous_text + part) else: merged.append((language, part)) return merged def _speaker_conditioning() -> dict: global SPEAKER_CONDITIONING with SPEAKER_LOCK: if SPEAKER_CONDITIONING is not None: return SPEAKER_CONDITIONING speakers = _json(f"{XTTS_URL}/studio_speakers", XTTS_TIMEOUT) if not isinstance(speakers, dict) or XTTS_SPEAKER not in speakers: raise RuntimeError("configured XTTS speaker is unavailable") selected = speakers[XTTS_SPEAKER] SPEAKER_CONDITIONING = { "speaker_embedding": selected["speaker_embedding"], "gpt_cond_latent": selected["gpt_cond_latent"], } return SPEAKER_CONDITIONING def _xtts_pcm(text: str, language: str, conditioning: dict) -> bytes: payload = { **conditioning, "text": text, "language": language, "add_wav_header": True, "stream_chunk_size": "20", } audio, _ = _request(f"{XTTS_URL}/tts_stream", payload=payload, timeout=XTTS_TIMEOUT) if len(audio) < 44 or audio[:4] != b"RIFF" or audio[8:12] != b"WAVE": raise RuntimeError("XTTS returned invalid WAV data") return audio[44:] def _wav(pcm: bytes) -> bytes: output = io.BytesIO() with wave.open(output, "wb") as wav_file: wav_file.setnchannels(1) wav_file.setsampwidth(2) wav_file.setframerate(24000) wav_file.writeframes(pcm) return output.getvalue() def _convert(wav_bytes: bytes, output_format: str, speed: float) -> tuple[bytes, str]: if output_format == "wav" and speed == 1.0: return wav_bytes, "audio/wav" codec = ["-codec:a", "libmp3lame", "-b:a", "96k", "-f", "mp3"] \ if output_format == "mp3" else ["-codec:a", "pcm_s16le", "-f", "wav"] command = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-f", "wav", "-i", "pipe:0"] if speed != 1.0: command.extend(["-filter:a", f"atempo={speed:.4f}"]) command.extend([*codec, "pipe:1"]) result = subprocess.run( command, input=wav_bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, timeout=120) if result.returncode != 0 or not result.stdout: raise RuntimeError("audio conversion failed") return result.stdout, "audio/mpeg" if output_format == "mp3" else "audio/wav" def synthesize_xtts(text: str, output_format: str, speed: float) -> tuple[bytes, str]: conditioning = _speaker_conditioning() pcm_parts: list[bytes] = [] silence = b"\x00\x00" * int(24000 * max(0, SILENCE_MS) / 1000) for language, segment in segment_languages(text): if not segment.strip(): continue pcm_parts.append(_xtts_pcm(segment, language, conditioning)) if silence: pcm_parts.append(silence) if pcm_parts and silence: pcm_parts.pop() if not pcm_parts: raise RuntimeError("no speech segments generated") return _convert(_wav(b"".join(pcm_parts)), output_format, speed) def synthesize_piper(text: str, output_format: str, speed: float) -> tuple[bytes, str]: return _request( f"{PIPER_URL}/tts", payload={"text": text, "voice": "alloy", "speed": speed, "format": output_format}, timeout=PIPER_TIMEOUT, ) def synthesize(text: str, output_format: str, speed: float) -> tuple[bytes, str]: acquired = SYNTHESIS_LOCK.acquire(timeout=QUEUE_TIMEOUT) if acquired: try: audio = synthesize_xtts(text, output_format, speed) with STATE_LOCK: STATE["last_backend"] = "xtts-v2" STATE["last_error"] = None return audio except Exception as exc: # fallback must cover all XTTS failures with STATE_LOCK: STATE["xtts_failures"] += 1 STATE["last_error"] = type(exc).__name__ finally: SYNTHESIS_LOCK.release() else: with STATE_LOCK: STATE["xtts_failures"] += 1 STATE["last_error"] = "queue-timeout" audio = synthesize_piper(text, output_format, speed) with STATE_LOCK: STATE["last_backend"] = "piper" STATE["piper_fallbacks"] += 1 return audio class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, fmt: str, *args: object) -> None: # Never log request URLs, bodies, synthesized text or speaker vectors. print(f"tts-gateway: {self.command} -> {args[1] if len(args) > 1 else '-'}") def send_bytes(self, status: int, body: bytes, content_type: str) -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def send_json(self, status: int, payload: dict) -> None: self.send_bytes(status, json.dumps(payload, separators=(",", ":")).encode(), "application/json") def do_GET(self) -> None: # noqa: N802 if self.path != "/status": self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"}) return primary_ready = _reachable(XTTS_URL, "/languages") fallback_ready = _reachable(PIPER_URL, "/status") with STATE_LOCK: state = dict(STATE) self.send_json( HTTPStatus.OK if fallback_ready else HTTPStatus.SERVICE_UNAVAILABLE, { "ready": fallback_ready, "engine": "xtts-v2-with-piper-fallback", "model": "xtts-v2", "voices": [VOICE_ALIAS], "speaker": XTTS_SPEAKER, "primary_ready": primary_ready, "fallback_ready": fallback_ready, **state, }, ) def do_POST(self) -> None: # noqa: N802 if self.path != "/tts": self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"}) return try: length = int(self.headers.get("Content-Length", "0")) except ValueError: length = 0 if length <= 0 or length > MAX_REQUEST_BYTES: self.send_json(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": "invalid request size"}) return try: request = json.loads(self.rfile.read(length)) text = request.get("text", "") voice = request.get("voice", VOICE_ALIAS) output_format = request.get("format", "mp3") speed = float(request.get("speed", 1.0)) except (json.JSONDecodeError, TypeError, ValueError): self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid JSON request"}) return if not isinstance(text, str) or not text.strip() or len(text) > MAX_TEXT_CHARS: self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid text"}) return if voice != VOICE_ALIAS: self.send_json(HTTPStatus.BAD_REQUEST, {"error": "unknown voice"}) return if output_format not in {"wav", "mp3"}: self.send_json(HTTPStatus.BAD_REQUEST, {"error": "unsupported format"}) return if not 0.5 <= speed <= 2.0: self.send_json(HTTPStatus.BAD_REQUEST, {"error": "invalid speed"}) return started = time.monotonic() try: audio, content_type = synthesize(text.strip(), output_format, speed) except Exception as exc: with STATE_LOCK: STATE["last_error"] = type(exc).__name__ self.send_json(HTTPStatus.SERVICE_UNAVAILABLE, {"error": "all local speech backends failed"}) return print(f"tts-gateway: synthesized via {STATE['last_backend']} in " f"{time.monotonic() - started:.2f}s") self.send_bytes(HTTPStatus.OK, audio, content_type) if __name__ == "__main__": print(f"TTS gateway ready on {HOST}:{PORT}; primary={XTTS_SPEAKER}; fallback=Piper") ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()