Files
AI-Profile-Router/platform/docker/tts-gateway/tts_gateway.py
T

634 lines
23 KiB
Python

#!/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"(?<![\w])(" + "|".join(
re.escape(term) for term in sorted(ENGLISH_TERMS, key=len, reverse=True)
) + r")(?![\w])",
re.IGNORECASE,
)
GERMAN_MARKERS = {
"aber", "auch", "auf", "das", "der", "die", "ein", "eine", "für",
"ich", "ist", "kann", "mit", "nicht", "noch", "oder", "soll", "und",
"wenn", "wir", "wird", "zu",
}
ENGLISH_MARKERS = {
"a", "and", "are", "can", "for", "from", "if", "in", "is", "it",
"of", "on", "or", "please", "the", "this", "to", "with", "you",
}
ENGLISH_PRONUNCIATIONS = {
"containern": "containers",
"docker containern": "Docker containers",
"docker-containern": "Docker containers",
"docker-container": "Docker container",
"docker-containers": "Docker containers",
"health-checks": "health checks",
"restart-loops": "restart loops",
"deliveryerrors": "delivery errors",
"ack-problem": "ack problem",
"i/o timeout": "I O timeout",
"parity-check": "parity check",
"homeassistant": "Home Assistant",
"openwebui": "Open Web U I",
"rtsp": "R T S P",
"jpeg": "J peg",
"nvme": "N V M E",
"gib": "gigabytes",
}
GERMAN_MONTHS = {
1: "Januar", 2: "Februar", 3: "März", 4: "April",
5: "Mai", 6: "Juni", 7: "Juli", 8: "August",
9: "September", 10: "Oktober", 11: "November", 12: "Dezember",
}
GERMAN_DIGITS = {
"0": "null", "1": "eins", "2": "zwei", "3": "drei", "4": "vier",
"5": "fünf", "6": "sechs", "7": "sieben", "8": "acht", "9": "neun",
}
def _request(url: str, *, payload: dict | None = None,
timeout: float = 10) -> 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("←", ". ")
if DEFAULT_LANGUAGE == "de":
text = normalize_for_german_speech(text)
# XTTS occasionally hallucinates syllables when visual punctuation is
# submitted literally or isolated at a chunk boundary. Preserve its pause
# semantics, but never ask the model to pronounce the glyph itself.
text = re.sub(r"(?<=\w)[\-‐‑‒–—−](?=\w)", " ", text)
text = re.sub(r"\s+[\-‐‑‒–—−]\s+", ", ", text)
text = re.sub(r"\s*[:;]+\s*", ", ", text)
# XTTS can pronounce literal question/exclamation glyphs as short
# nonsense syllables (for example "?" as "nau"). Retain a sentence
# boundary for pacing, but never pass those glyphs to the model.
text = re.sub(r"\s*[!?]+\s*", ". ", 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*\.", ".", text)
text = re.sub(r"(?:,\s*){2,}", ", ", 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()