963 lines
37 KiB
Python
963 lines
37 KiB
Python
#!/usr/bin/env python3
|
||
"""Private Qwen3-TTS-first 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 http.client
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import threading
|
||
import time
|
||
import unicodedata
|
||
import urllib.error
|
||
import urllib.request
|
||
import urllib.parse
|
||
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"))
|
||
QWEN_TTS_URL = os.getenv("QWEN_TTS_URL", "http://qwen3-tts:8001").rstrip("/")
|
||
QWEN_TTS_MODEL = os.getenv("QWEN_TTS_MODEL", "tts-1")
|
||
QWEN_TTS_VOICE = os.getenv("QWEN_TTS_VOICE", "serena")
|
||
QWEN_TTS_LANGUAGE = os.getenv("QWEN_TTS_LANGUAGE", "German")
|
||
QWEN_TTS_TIMEOUT = float(os.getenv("QWEN_TTS_TIMEOUT", "120"))
|
||
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 loses natural prosody when a sentence is synthesized as many tiny
|
||
# requests: every request starts a fresh utterance. Keep complete sentences
|
||
# together and use this only as a safety ceiling for unusually long sentences.
|
||
CHUNK_CHARS = int(os.getenv("XTTS_CHUNK_CHARS", "420"))
|
||
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"))
|
||
INTERNAL_SILENCE_THRESHOLD = int(os.getenv("XTTS_INTERNAL_SILENCE_THRESHOLD", "512"))
|
||
INTERNAL_SILENCE_TRIGGER_MS = int(os.getenv("XTTS_INTERNAL_SILENCE_TRIGGER_MS", "650"))
|
||
INTERNAL_SILENCE_KEEP_MS = int(os.getenv("XTTS_INTERNAL_SILENCE_KEEP_MS", "220"))
|
||
|
||
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 _spoken_short_date_range(match: re.Match) -> str:
|
||
first_day = int(match.group(1))
|
||
last_day = int(match.group(2))
|
||
month = int(match.group(3))
|
||
year = match.group(4)
|
||
month_name = GERMAN_MONTHS.get(month)
|
||
if not month_name or not 1 <= first_day <= 31 or not 1 <= last_day <= 31:
|
||
return match.group(0)
|
||
result = f"{first_day}. bis {last_day}. {month_name}"
|
||
if year:
|
||
result += f" {year}"
|
||
return result
|
||
|
||
|
||
def _spoken_full_date_range(match: re.Match) -> str:
|
||
first_day = int(match.group(1))
|
||
first_month = int(match.group(2))
|
||
last_day = int(match.group(3))
|
||
last_month = int(match.group(4))
|
||
year = match.group(5)
|
||
first_name = GERMAN_MONTHS.get(first_month)
|
||
last_name = GERMAN_MONTHS.get(last_month)
|
||
if (not first_name or not last_name
|
||
or not 1 <= first_day <= 31 or not 1 <= last_day <= 31):
|
||
return match.group(0)
|
||
if first_month == last_month:
|
||
result = f"{first_day}. bis {last_day}. {last_name}"
|
||
else:
|
||
result = f"{first_day}. {first_name} bis {last_day}. {last_name}"
|
||
if year:
|
||
result += f" {year}"
|
||
return result
|
||
|
||
|
||
def _spell_digits(value: str) -> str:
|
||
return " ".join(GERMAN_DIGITS[digit] for digit in value)
|
||
|
||
|
||
def _spoken_ipv4(match: re.Match) -> str:
|
||
"""Keep IPv4 octets intact while making the separators pronounceable."""
|
||
return " Punkt ".join(str(int(part)) for part in match.group(0).split("."))
|
||
|
||
|
||
def normalize_for_german_speech(text: str) -> str:
|
||
"""Turn common visual notation into unambiguous spoken German."""
|
||
text = re.sub(r"\bv\.\s*a\.", "vor allem", text, flags=re.IGNORECASE)
|
||
text = re.sub(r"\bca\.", "circa", text, flags=re.IGNORECASE)
|
||
text = re.sub(r"\bmax\.", "maximal", text, flags=re.IGNORECASE)
|
||
# Run these before the date rule: otherwise 192.168.1.5 could be partly
|
||
# interpreted as a visual date.
|
||
text = re.sub(
|
||
r"\b(?:\d{1,3}\.){3}\d{1,3}\b",
|
||
_spoken_ipv4,
|
||
text,
|
||
)
|
||
text = re.sub(
|
||
r"\b([01]?\d|2[0-3]):([0-5]\d)\b",
|
||
lambda match: f"{int(match.group(1))} Uhr {int(match.group(2))}",
|
||
text,
|
||
)
|
||
text = re.sub(
|
||
r"\b([01]?\d|2[0-3])\s*[-‐‑‒–—−]\s*"
|
||
r"([01]?\d|2[0-3])\s*Uhr\b",
|
||
lambda match: f"{int(match.group(1))} bis {int(match.group(2))} Uhr",
|
||
text,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
text = re.sub(
|
||
r"(-?\d+(?:[,.]\d+)?)[ \t]*°?[ \t]*[-‐‑‒–—−][ \t]*"
|
||
r"(-?\d+(?:[,.]\d+)?)[ \t]*°[ \t]*(?:C)?",
|
||
r"\1 bis \2 Grad",
|
||
text,
|
||
flags=re.IGNORECASE,
|
||
)
|
||
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,
|
||
)
|
||
# Date ranges are deliberately strict: both variants require dotted
|
||
# calendar notation, so ordinary numeric ranges remain untouched.
|
||
text = re.sub(
|
||
r"\b([0-3]?\d)\.([01]?\d)\.\s*[-‐‑‒–—−]\s*"
|
||
r"([0-3]?\d)\.([01]?\d)\.(?:(\d{4})\b)?",
|
||
_spoken_full_date_range,
|
||
text,
|
||
)
|
||
text = re.sub(
|
||
r"\b([0-3]?\d)\.\s*[-‐‑‒–—−]\s*"
|
||
r"([0-3]?\d)\.([01]?\d)\.(?:(\d{4})\b)?",
|
||
_spoken_short_date_range,
|
||
text,
|
||
)
|
||
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)
|
||
text = re.sub(r"\b(\d+)\s*[×x]\s*", r"\1 mal ", text)
|
||
return text
|
||
|
||
|
||
def prepare_for_qwen_speech(text: str) -> str:
|
||
"""Normalize technical display text without destroying Qwen's prosody."""
|
||
text = re.sub(r"```.*?```", " Codeblock. ", 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("_", " ")
|
||
text = text.replace("→", ". ").replace("←", ". ")
|
||
if DEFAULT_LANGUAGE == "de":
|
||
text = normalize_for_german_speech(text)
|
||
# Preserve slashes until after unit normalization so km/h becomes
|
||
# "Kilometer pro Stunde" instead of the broken "Kilometer, h".
|
||
text = text.replace("/", ", ")
|
||
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)
|
||
return text.strip()
|
||
|
||
|
||
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]:
|
||
"""Return paragraph-sized XTTS requests with a conservative hard ceiling.
|
||
|
||
Complete neighbouring sentences are combined while they fit. This avoids
|
||
restarting the generative XTTS decoder after every short sentence, which
|
||
can create invented tail syllables between sentences. Overlong sentences
|
||
are split at clause boundaries first and at words only as a last resort.
|
||
"""
|
||
text = text.strip()
|
||
if not text:
|
||
return []
|
||
|
||
sentences = re.split(r"(?<=[.!?])\s+|\s+(?=\d+[.)]\s)", text)
|
||
chunks: list[str] = []
|
||
current = ""
|
||
for sentence in sentences:
|
||
sentence = sentence.strip()
|
||
if not sentence:
|
||
continue
|
||
if len(sentence) <= limit:
|
||
candidate = f"{current} {sentence}".strip()
|
||
if current and len(candidate) > limit:
|
||
chunks.append(current)
|
||
current = sentence
|
||
else:
|
||
current = candidate
|
||
continue
|
||
|
||
if current:
|
||
chunks.append(current)
|
||
current = ""
|
||
|
||
# Retain commas in the preceding clause so XTTS can reproduce the
|
||
# intended pause. Semicolons and colons were normalized earlier.
|
||
clauses = re.split(r"(?<=,)\s+", sentence)
|
||
long_current = ""
|
||
for clause in clauses:
|
||
clause = clause.strip()
|
||
candidate = f"{long_current} {clause}".strip()
|
||
if long_current and len(candidate) > limit:
|
||
chunks.append(long_current)
|
||
long_current = ""
|
||
if len(clause) <= limit:
|
||
long_current = f"{long_current} {clause}".strip()
|
||
continue
|
||
|
||
# A clause without a usable pause can still exceed the ceiling.
|
||
for word in clause.split():
|
||
candidate = f"{long_current} {word}".strip()
|
||
if long_current and len(candidate) > limit:
|
||
chunks.append(long_current)
|
||
long_current = word
|
||
else:
|
||
long_current = candidate
|
||
if long_current:
|
||
chunks.append(long_current)
|
||
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):
|
||
if not re.search(r"\w", chunk, flags=re.UNICODE):
|
||
if prepared:
|
||
previous_language, previous_text = prepared[-1]
|
||
prepared[-1] = (
|
||
previous_language,
|
||
previous_text.rstrip() + chunk.strip(),
|
||
)
|
||
continue
|
||
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 _stabilize_xtts_ending(text: str) -> str:
|
||
"""Give XTTS a reliable stop cue without changing the visible answer.
|
||
|
||
XTTS v2 can continue with invented syllables after a terminal full stop,
|
||
especially in German. A terminal semicolon is tokenized as a stronger,
|
||
more reliable boundary while retaining neutral sentence intonation.
|
||
"""
|
||
spoken = text.rstrip()
|
||
if spoken.endswith((".", "!", "?", ";", ":")):
|
||
spoken = spoken[:-1].rstrip()
|
||
return f"{spoken};"
|
||
|
||
|
||
def _xtts_pcm(text: str, language: str, conditioning: dict) -> bytes:
|
||
payload = {
|
||
**conditioning,
|
||
"text": _stabilize_xtts_ending(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 _compress_internal_silence(pcm: bytes) -> bytes:
|
||
"""Shorten XTTS silence hallucinations while preserving normal pauses."""
|
||
samples = array("h")
|
||
samples.frombytes(pcm)
|
||
if not samples:
|
||
return pcm
|
||
trigger = int(24000 * max(0, INTERNAL_SILENCE_TRIGGER_MS) / 1000)
|
||
keep = int(24000 * max(0, INTERNAL_SILENCE_KEEP_MS) / 1000)
|
||
if trigger <= 0 or keep >= trigger:
|
||
return pcm
|
||
output = array("h")
|
||
index = 0
|
||
while index < len(samples):
|
||
if abs(samples[index]) > INTERNAL_SILENCE_THRESHOLD:
|
||
output.append(samples[index])
|
||
index += 1
|
||
continue
|
||
end = index + 1
|
||
while end < len(samples) and abs(samples[end]) <= INTERNAL_SILENCE_THRESHOLD:
|
||
end += 1
|
||
run = end - index
|
||
if run >= trigger:
|
||
output.extend(samples[index:index + keep])
|
||
else:
|
||
output.extend(samples[index:end])
|
||
index = end
|
||
return output.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 _compress_internal_silence(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"
|
||
if output_format == "mp3":
|
||
codec = ["-codec:a", "libmp3lame", "-b:a", "96k", "-f", "mp3"]
|
||
content_type = "audio/mpeg"
|
||
elif output_format == "pcm":
|
||
# Hermes' OpenAI streaming TTS client expects headerless 24 kHz,
|
||
# mono, signed 16-bit little-endian PCM chunks.
|
||
codec = ["-ac", "1", "-ar", "24000", "-codec:a", "pcm_s16le", "-f", "s16le"]
|
||
content_type = "application/octet-stream"
|
||
else:
|
||
codec = ["-codec:a", "pcm_s16le", "-f", "wav"]
|
||
content_type = "audio/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, content_type
|
||
|
||
|
||
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]:
|
||
upstream_format = "wav" if output_format == "pcm" else output_format
|
||
audio, content_type = _request(
|
||
f"{PIPER_URL}/tts",
|
||
payload={"text": text, "voice": "alloy", "speed": speed,
|
||
"format": upstream_format},
|
||
timeout=PIPER_TIMEOUT,
|
||
)
|
||
return _convert(audio, "pcm", 1.0) if output_format == "pcm" else (audio, content_type)
|
||
|
||
|
||
def synthesize_qwen(text: str, output_format: str,
|
||
speed: float) -> tuple[bytes, str]:
|
||
text = prepare_for_qwen_speech(text)
|
||
upstream_format = "wav" if output_format == "pcm" else output_format
|
||
audio, content_type = _request(
|
||
f"{QWEN_TTS_URL}/v1/audio/speech",
|
||
payload={"model": QWEN_TTS_MODEL, "input": text,
|
||
"voice": QWEN_TTS_VOICE, "language": QWEN_TTS_LANGUAGE,
|
||
"response_format": upstream_format, "speed": speed},
|
||
timeout=QWEN_TTS_TIMEOUT,
|
||
)
|
||
return _convert(audio, "pcm", 1.0) if output_format == "pcm" else (audio, content_type)
|
||
|
||
|
||
def open_qwen_pcm_stream(text: str, chunk_size: int = 4) \
|
||
-> tuple[http.client.HTTPConnection, http.client.HTTPResponse]:
|
||
"""Open Qwen's native token-level PCM stream without buffering it.
|
||
|
||
The upstream emits headerless 24 kHz mono signed 16-bit little-endian
|
||
PCM. Keeping this response streaming is what lets playback begin while
|
||
the remainder of the sentence is still being synthesized.
|
||
"""
|
||
parsed = urllib.parse.urlparse(QWEN_TTS_URL)
|
||
if parsed.scheme != "http" or not parsed.hostname:
|
||
raise RuntimeError("QWEN_TTS_URL must be an http URL")
|
||
port = parsed.port or 80
|
||
prefix = parsed.path.rstrip("/")
|
||
payload = json.dumps({
|
||
"model": QWEN_TTS_MODEL,
|
||
"input": prepare_for_qwen_speech(text),
|
||
"voice": QWEN_TTS_VOICE,
|
||
"language": QWEN_TTS_LANGUAGE,
|
||
"chunk_size": chunk_size,
|
||
}, separators=(",", ":")).encode()
|
||
connection = http.client.HTTPConnection(
|
||
parsed.hostname, port, timeout=QWEN_TTS_TIMEOUT)
|
||
try:
|
||
connection.request(
|
||
"POST",
|
||
f"{prefix}/v1/audio/speech/pcm-stream",
|
||
body=payload,
|
||
headers={"Content-Type": "application/json",
|
||
"Accept": "application/octet-stream"},
|
||
)
|
||
response = connection.getresponse()
|
||
if response.status != HTTPStatus.OK:
|
||
message = response.read(512).decode(errors="replace")
|
||
raise RuntimeError(
|
||
f"Qwen PCM stream failed ({response.status}): {message}")
|
||
return connection, response
|
||
except Exception:
|
||
connection.close()
|
||
raise
|
||
|
||
|
||
def synthesize(text: str, output_format: str, speed: float) -> tuple[bytes, str]:
|
||
acquired = SYNTHESIS_LOCK.acquire(timeout=QUEUE_TIMEOUT)
|
||
if acquired:
|
||
try:
|
||
audio = synthesize_qwen(text, output_format, speed)
|
||
with STATE_LOCK:
|
||
STATE["last_backend"] = "qwen3-tts-1.7b"
|
||
STATE["last_error"] = None
|
||
return audio
|
||
except Exception as exc: # fallback must cover all Qwen 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(QWEN_TTS_URL, "/health")
|
||
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": "qwen3-tts-with-piper-fallback",
|
||
"model": "Qwen3-TTS-12Hz-1.7B-Base",
|
||
"voices": [VOICE_ALIAS],
|
||
"speaker": QWEN_TTS_VOICE,
|
||
"primary_ready": primary_ready,
|
||
"fallback_ready": fallback_ready,
|
||
**state,
|
||
},
|
||
)
|
||
|
||
def do_POST(self) -> None: # noqa: N802
|
||
if self.path not in {"/tts", "/tts/pcm-stream"}:
|
||
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("input", 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", "pcm"}:
|
||
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
|
||
if self.path == "/tts/pcm-stream":
|
||
self._stream_qwen_pcm(text.strip(), request)
|
||
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)
|
||
|
||
def _stream_qwen_pcm(self, text: str, request: dict) -> None:
|
||
"""Unframe Qwen's PCM frames and relay their audio immediately."""
|
||
try:
|
||
chunk_size = max(1, min(32, int(request.get("chunk_size", 4))))
|
||
except (TypeError, ValueError):
|
||
self.send_json(HTTPStatus.BAD_REQUEST,
|
||
{"error": "invalid chunk_size"})
|
||
return
|
||
acquired = SYNTHESIS_LOCK.acquire(timeout=QUEUE_TIMEOUT)
|
||
if not acquired:
|
||
self.send_json(HTTPStatus.SERVICE_UNAVAILABLE,
|
||
{"error": "speech queue timeout"})
|
||
return
|
||
connection = None
|
||
started = time.monotonic()
|
||
headers_sent = False
|
||
try:
|
||
connection, response = open_qwen_pcm_stream(text, chunk_size)
|
||
self.send_response(HTTPStatus.OK)
|
||
self.send_header("Content-Type", "application/octet-stream")
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.send_header("Connection", "close")
|
||
self.end_headers()
|
||
headers_sent = True
|
||
first = True
|
||
while True:
|
||
frame_header = response.read(4)
|
||
if not frame_header:
|
||
break
|
||
if len(frame_header) != 4:
|
||
raise RuntimeError("truncated Qwen PCM frame header")
|
||
frame_length = int.from_bytes(frame_header, "big")
|
||
if frame_length == 0:
|
||
break
|
||
if frame_length > MAX_AUDIO_BYTES:
|
||
raise RuntimeError("Qwen PCM frame is too large")
|
||
remaining = frame_length
|
||
while remaining:
|
||
chunk = response.read(min(16384, remaining))
|
||
if not chunk:
|
||
raise RuntimeError("truncated Qwen PCM frame")
|
||
if first:
|
||
print("tts-gateway: first Qwen PCM chunk in "
|
||
f"{time.monotonic() - started:.2f}s")
|
||
first = False
|
||
self.wfile.write(chunk)
|
||
self.wfile.flush()
|
||
remaining -= len(chunk)
|
||
with STATE_LOCK:
|
||
STATE["last_backend"] = "qwen3-tts-1.7b-stream"
|
||
STATE["last_error"] = None
|
||
except Exception as exc:
|
||
with STATE_LOCK:
|
||
STATE["last_error"] = type(exc).__name__
|
||
# Once PCM started, simply close the truncated response. Sending
|
||
# JSON into the audio stream would produce loud corrupt samples.
|
||
if not headers_sent and not self.wfile.closed:
|
||
try:
|
||
self.send_json(HTTPStatus.SERVICE_UNAVAILABLE,
|
||
{"error": "local PCM stream failed"})
|
||
except (OSError, BrokenPipeError):
|
||
pass
|
||
finally:
|
||
if connection is not None:
|
||
connection.close()
|
||
SYNTHESIS_LOCK.release()
|
||
self.close_connection = True
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print(f"TTS gateway ready on {HOST}:{PORT}; primary={QWEN_TTS_VOICE}; fallback=Piper")
|
||
ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()
|