#!/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 unicodedata import urllib.error import urllib.request import wave from array import array 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") CODE_SWITCH_ENABLED = os.getenv("TTS_CODE_SWITCH_ENABLED", "false").lower() \ in {"1", "true", "yes", "on"} 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")) # XTTS can insert multi-second silences or truncate the remainder when a # moderately long German paragraph is sent as one request. Keeping requests # close to one sentence proved substantially more stable with Annmarie Nele. CHUNK_CHARS = int(os.getenv("XTTS_CHUNK_CHARS", "60")) TRIM_THRESHOLD = int(os.getenv("XTTS_TRIM_THRESHOLD", "90")) TRIM_PADDING_MS = int(os.getenv("XTTS_TRIM_PADDING_MS", "18")) CROSSFADE_MS = int(os.getenv("XTTS_CROSSFADE_MS", "8")) SENTENCE_PAUSE_MS = int(os.getenv("XTTS_SENTENCE_PAUSE_MS", "65")) 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 = ( "unsupported image format", "premature end of JPEG", "incomplete scan", "Docker Containers", "Docker-Containers", "Docker Containern", "Docker-Containern", "Health Checks", "Health-Checks", "Restart Loops", "Restart-Loops", "False Positive", "Delivery Errors", "DeliveryErrors", "Ack Problem", "Ack-Problem", "I/O timeout", "Parity Check", "Parity-Check", "Disk disabled", "Disk invalid", "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", "Containern", "Containers", "Container", "Crashes", "healthy", "disabled", "invalid", "Dashboard", "Server", "Logs", "Log", "Matches", "up", "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", "HomeAssistant", "ESPHome", "iGotify", "Immich", "Vaultwarden", "UniFi", "UptimeKuma", "go2rtc", "Zigbee", "SONOFF", "eWeLink", "RTSP", "JPEG", "NVMe", "GiB", ) 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)] if not CODE_SWITCH_ENABLED: return [("de", 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 _spoken_date(match: re.Match) -> str: day = int(match.group(1)) month = int(match.group(2)) year = match.group(3) month_name = GERMAN_MONTHS.get(month) if not month_name or not 1 <= day <= 31: return match.group(0) result = f"{day}. {month_name}" if year: result += f" {year}" return result def _spell_digits(value: str) -> str: return " ".join(GERMAN_DIGITS[digit] for digit in value) def normalize_for_german_speech(text: str) -> str: """Turn common visual notation into unambiguous spoken German.""" text = re.sub( r"(-?\d+(?:[,.]\d+)?)[ \t]*°[ \t]*(?:C)?[ \t]*/[ \t]*" r"(-?\d+(?:[,.]\d+)?)[ \t]*°[ \t]*(?:C)?", r"Höchstwert \1 Grad, Tiefstwert \2 Grad", text, flags=re.IGNORECASE, ) text = re.sub( r"(-?\d+(?:[,.]\d+)?)[ \t]*°[ \t]*(?:C)?", r"\1 Grad", text, flags=re.IGNORECASE, ) text = re.sub( r"\b([0-3]?\d)\.([01]?\d)\.(?:(\d{4})\b)?", _spoken_date, text, ) text = re.sub( r"\bPLZ\s+(\d{5})\b", lambda match: "Postleitzahl " + _spell_digits(match.group(1)), text, flags=re.IGNORECASE, ) text = re.sub( r"\((\d{5})\)", lambda match: "(Postleitzahl " + _spell_digits(match.group(1)) + ")", text, ) text = re.sub( r"https?://(?:www\.)?([^/\s)]+)(?:/[^\s)]*)?", lambda match: match.group(1), text, flags=re.IGNORECASE, ) text = re.sub( r"\b([A-Za-z0-9][A-Za-z0-9-]*(?:\.[A-Za-z0-9-]+)*)" r"\.(de|com|org|net|io|ai)\b", lambda match: ( match.group(1).replace(".", " Punkt ") + " Punkt " + ( "de" if match.group(2).lower() == "de" else " ".join(match.group(2).upper()) ) ), text, flags=re.IGNORECASE, ) text = re.sub( r"(-?\d+(?:[,.]\d+)?)\s*km\s*(?:/\s*)?h\b", r"\1 Kilometer pro Stunde", text, flags=re.IGNORECASE, ) text = re.sub(r"(\d)\s*%", r"\1 Prozent", text) return text def clean_for_speech(text: str) -> str: """Remove visual markup that makes long TTS output unstable or noisy.""" text = re.sub(r"```.*?```", " Code block. ", text, flags=re.DOTALL) text = re.sub(r"`([^`]+)`", r"\1", text) text = re.sub(r"!\[([^]]*)\]\([^)]+\)", r"\1", text) text = re.sub(r"\[([^]]+)\]\([^)]+\)", r"\1", text) text = re.sub(r"(?m)^\s{0,3}#{1,6}\s*", "", text) text = re.sub(r"(?m)^\s*[-*+]\s+", "", text) text = text.replace("→", ". ").replace("←", ". ") text = text.replace("–", " - ").replace("—", " - ") if DEFAULT_LANGUAGE == "de": text = normalize_for_german_speech(text) text = "".join( char for char in text if unicodedata.category(char) not in {"So", "Cs"} ) text = re.sub(r"[ \t]+", " ", text) text = re.sub(r"\s*\n+\s*", ". ", text) text = re.sub(r"([:;])\s*\.", r"\1", text) text = re.sub(r"(?:\.\s*){2,}", ". ", text) return text.strip() def _split_chunk(text: str, limit: int = CHUNK_CHARS) -> list[str]: """Split at natural pauses and keep every XTTS request comfortably short.""" text = text.strip() if not text: return [] if len(text) <= limit: return [text] pieces = re.split(r"(?<=[.!?;:])\s+|\s+(?=\d+[.)]\s)", text) chunks: list[str] = [] current = "" for piece in pieces: piece = piece.strip() if not piece: continue if len(piece) > limit: words = piece.split() for word in words: candidate = f"{current} {word}".strip() if current and len(candidate) > limit: chunks.append(current) current = word else: current = candidate continue candidate = f"{current} {piece}".strip() if current and len(candidate) > limit: chunks.append(current) current = piece else: current = candidate if current: chunks.append(current) return chunks def prepare_segments(text: str) -> list[tuple[str, str]]: """Prepare short, deterministic German/English XTTS requests.""" prepared: list[tuple[str, str]] = [] for language, segment in segment_languages(clean_for_speech(text)): if not re.search(r"\w", segment, flags=re.UNICODE): if prepared: previous_language, previous_text = prepared[-1] prepared[-1] = (previous_language, previous_text + segment.strip()) continue spoken = segment if language == "en": spoken = ENGLISH_PRONUNCIATIONS.get(segment.strip().lower(), segment) for chunk in _split_chunk(spoken): prepared.append((language, chunk)) return prepared 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 _trim_pcm(pcm: bytes) -> bytes: """Trim generated edge silence while retaining a small safety padding.""" samples = array("h") samples.frombytes(pcm) if not samples: return pcm first = next((i for i, value in enumerate(samples) if abs(value) >= TRIM_THRESHOLD), 0) last = next((i for i in range(len(samples) - 1, -1, -1) if abs(samples[i]) >= TRIM_THRESHOLD), len(samples) - 1) padding = int(24000 * max(0, TRIM_PADDING_MS) / 1000) first = max(0, first - padding) last = min(len(samples) - 1, last + padding) return samples[first:last + 1].tobytes() def _fade_edge(pcm: bytes, *, fade_in: bool = False, fade_out: bool = False) -> bytes: samples = array("h") samples.frombytes(pcm) count = min(len(samples), int(24000 * max(0, CROSSFADE_MS) / 1000)) if count <= 1: return pcm if fade_in: for index in range(count): samples[index] = int(samples[index] * index / (count - 1)) if fade_out: start = len(samples) - count for index in range(count): samples[start + index] = int( samples[start + index] * (count - 1 - index) / (count - 1)) return samples.tobytes() def _join_pcm(parts: list[tuple[str, bytes]]) -> bytes: """Join clips without clicks; pause only at real sentence boundaries.""" if not parts: return b"" output = bytearray() sentence_silence = b"\x00\x00" * int( 24000 * max(0, SENTENCE_PAUSE_MS) / 1000) for index, (text, pcm) in enumerate(parts): pcm = _trim_pcm(pcm) previous_ends_sentence = index > 0 and bool( re.search(r"[.!?;:]\s*$", parts[index - 1][0])) if index == 0: output.extend(_fade_edge(pcm, fade_in=True)) elif previous_ends_sentence: if output: faded = _fade_edge(bytes(output), fade_out=True) output[:] = faded output.extend(sentence_silence) output.extend(_fade_edge(pcm, fade_in=True)) else: # Language switches inside a sentence get no artificial pause. # Small fades remove the discontinuity that otherwise sounds like # a high click or beep between independently generated clips. if output: faded = _fade_edge(bytes(output), fade_out=True) output[:] = faded output.extend(_fade_edge(pcm, fade_in=True)) return bytes(output) 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[tuple[str, bytes]] = [] for language, segment in prepare_segments(text): if not segment.strip(): continue pcm_parts.append((segment, _xtts_pcm(segment, language, conditioning))) if not pcm_parts: raise RuntimeError("no speech segments generated") return _convert(_wav(_join_pcm(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()