#!/usr/bin/env python3 """frank-tts — the fleet's OpenAI-compatible speech façade (component: the shim). One local endpoint that answers ``voice=`` with audio in a cloned voice. Consumers (the omp harness, OpenWebUI, open-notebook) speak the OpenAI ``POST /v1/audio/speech`` shape and never learn a file path: a voice is a directory under ``$VOICES_DIR`` and ``reference.wav`` inside it is what makes the name real. A name that has no reference is not a voice — the shim answers 400 with the list that is, instead of silently narrating in a fallback voice (silent fallback is how a wrong-voice endpoint stays wrong for weeks). A request that names NO voice at all takes ``$DEFAULT_VOICE``: that is a spoken choice in the env file rather than a surprise in the code, and the journal marks it ``(default)`` so a name nobody asked for is never mistaken for one that was. Two backends sit behind the one surface, chosen by ``TTS_BACKEND``: ``omni-http`` Forward to an OpenAI-shaped upstream TTS server (``$TTS_UPSTREAM``) and carry the cloned voice with the request as ``reference_audio`` (base64 of ``reference.wav``). The upstream does the inference; this process only resolves the voice and, when asked, re-encodes to mp3 with ``ffmpeg``. ``llama-tts-cli`` Run a local binary per request (``$LLAMA_TTS_BIN``), hand it the reference with ``--tts-speaker-file``, read the WAV it writes. ``nice -n 19`` because this box also serves text and a voice request must never starve it. Everything comes from the environment, read at REQUEST time (no import-time-only globals): the same module can be imported with an injected ``Config`` or run with ``python3 app.py`` and the unit's env file. Runtime law: single process, standard library only, no build step. Audio bytes never reach the log — request lines and error text only. """ from __future__ import annotations import argparse import base64 import json import os import re import subprocess import sys import tempfile import urllib.error import urllib.parse import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path WAV_CONTENT_TYPE = "audio/wav" MP3_CONTENT_TYPE = "audio/mpeg" # A voice name is a plain directory name. Slashes and `..` are refused here so # a client-supplied name can never escape $VOICES_DIR. VOICE_NAME_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}") class ShimError(Exception): """A failure with the HTTP status the client should see.""" def __init__(self, status, message): super().__init__(message) self.status = status self.message = message class Config: """Every knob of the shim. Constructed from the environment per request by default, or injected into ``create_server`` for tests and embedding.""" def __init__(self, voices_dir, backend, upstream, upstream_model, llama_bin, llama_model, llama_mmproj, llama_lang, llama_threads, port, max_chars, timeout, default_voice=""): self.voices_dir = Path(voices_dir) self.backend = backend self.upstream = upstream self.upstream_model = upstream_model self.llama_bin = llama_bin self.llama_model = llama_model self.llama_mmproj = llama_mmproj self.llama_lang = llama_lang self.llama_threads = llama_threads self.port = port self.max_chars = max_chars self.timeout = timeout self.default_voice = default_voice @classmethod def from_env(cls, environ=None): env = os.environ if environ is None else environ return cls( voices_dir=env.get("VOICES_DIR", "/voices"), backend=(env.get("TTS_BACKEND") or "omni-http").strip().lower(), upstream=(env.get("TTS_UPSTREAM") or "http://host.containers.internal:8880").rstrip("/"), upstream_model=env.get("TTS_UPSTREAM_MODEL") or "voxcpm2", llama_bin=env.get("LLAMA_TTS_BIN") or "", llama_model=env.get("LLAMA_TTS_MODEL") or "", llama_mmproj=env.get("LLAMA_TTS_MMPROJ") or "", llama_lang=env.get("LLAMA_TTS_LANG") or "english", llama_threads=_int(env, "LLAMA_TTS_THREADS", 8), port=_int(env, "SHIM_PORT", 7772), max_chars=_int(env, "SHIM_MAX_CHARS", 15000), timeout=float(_int(env, "SHIM_TIMEOUT", 600)), default_voice=(env.get("DEFAULT_VOICE") or "").strip(), ) def _int(env, name, default): try: return int(str(env.get(name, "")).strip()) except (TypeError, ValueError): return default def list_voices(voices_dir): """Voice names in ``voices_dir``: directories that hold a ``reference.wav``. A missing or unreadable directory is an empty fleet, not an error — the health endpoint must still answer so the unit reports its state. """ try: entries = sorted(p.name for p in voices_dir.iterdir() if p.is_dir() and (p / "reference.wav").is_file()) except OSError: return [] return entries def resolve_voice(voices_dir, voice): """``$VOICES_DIR//reference.wav`` for a valid name, else ``None``.""" if not isinstance(voice, str) or not VOICE_NAME_RE.fullmatch(voice): return None reference = voices_dir / voice / "reference.wav" return reference if reference.is_file() else None def to_mp3(wav_bytes, timeout): """Re-encode WAV bytes to mp3 with the image's ``ffmpeg`` (pipes, no temp file: the audio never lands on disk to be left behind).""" argv = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-i", "pipe:0", "-f", "mp3", "pipe:1"] try: proc = subprocess.run(argv, input=wav_bytes, capture_output=True, timeout=timeout) except FileNotFoundError: raise ShimError(500, "response_format=mp3 needs the ffmpeg binary, which is not on PATH") except subprocess.TimeoutExpired: raise ShimError(502, "ffmpeg mp3 conversion timed out after %gs" % timeout) if proc.returncode != 0 or not proc.stdout: raise ShimError(502, "ffmpeg mp3 conversion failed (exit %s): %s" % (proc.returncode, _tail(proc.stderr))) return proc.stdout def _tail(raw): """Last line of a subprocess' stderr, for an error message. Text only — audio bytes are never echoed, and never logged.""" text = (raw or b"").decode("utf-8", "replace").strip() return text.splitlines()[-1][:300] if text else "no output" def _error(message, voices=None): body = {"error": {"message": message, "type": "invalid_request_error"}} if voices is not None: body["error"]["available_voices"] = voices return body class SpeechHandler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" server_version = "frank-tts" sys_version = "" # -- plumbing ---------------------------------------------------------- def config(self): return getattr(self.server, "config", None) or Config.from_env() def log_message(self, fmt, *args): # One prefix for the journal. Request lines and outcomes only: a # synthesized text, a base64 reference voice and the audio bytes # itself never reach stderr, in success or in failure. sys.stderr.write("frank-tts: %s - %s\n" % (self.address_string(), fmt % args)) def log_error(self, fmt, *args): # The base class routes refused requests through a separate sink; keep # them on the same line format so the journal reads uniformly. self.log_message(fmt, *args) def _json(self, status, obj, head=False): self._bytes(status, json.dumps(obj).encode("utf-8"), "application/json", head) def _bytes(self, status, data, content_type, head=False): self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.end_headers() if not head: self.wfile.write(data) def _drain(self): """Read and discard a body this route will not answer. HTTP/1.1 keep-alive makes an unread body the NEXT request line, so a client that reuses the connection would see a parse error it never sent.""" if getattr(self, "_body_read", False): return # already consumed; a second read would block the response try: length = int(self.headers.get("Content-Length") or 0) except ValueError: return if length > 0: self.rfile.read(length) def _body(self): """Parse the JSON request body. Never returned to a log or an error.""" try: length = int(self.headers.get("Content-Length") or 0) except ValueError: raise ValueError("Content-Length is not a number") raw = self.rfile.read(length) if length > 0 else b"" self._body_read = True if not raw: raise ValueError("expected a JSON body") parsed = json.loads(raw.decode("utf-8")) if not isinstance(parsed, dict): raise ValueError("expected a JSON object") return parsed # -- routes ------------------------------------------------------------ def do_GET(self, head=False): # HEAD shares the route table: `wget --spider` (the fleet's probe # shape) asks for HEAD, and a 501 there would read as a dead service. path = urllib.parse.urlparse(self.path).path cfg = self.config() if path == "/health": self._json(200, {"status": "ok", "backend": cfg.backend, "voices": list_voices(cfg.voices_dir)}, head) elif path == "/v1/audio/voices": self._json(200, list_voices(cfg.voices_dir), head) else: self._json(404, _error("no such endpoint: %s" % path), head) def do_HEAD(self): self.do_GET(head=True) def do_POST(self): path = urllib.parse.urlparse(self.path).path if path != "/v1/audio/speech": self._drain() self._json(404, _error("no such endpoint: %s" % path)) return cfg = self.config() try: body = self._body() except (ValueError, UnicodeDecodeError) as exc: self._drain() self._json(400, _error("malformed request body: %s" % exc)) return text = body.get("input") if not isinstance(text, str) or not text.strip(): self._json(400, _error("`input` must be a non-empty string")) return if len(text) > cfg.max_chars: self._json(414, _error("input is %d characters, over the %d limit " "(SHIM_MAX_CHARS); split it or raise the limit" % (len(text), cfg.max_chars))) return fmt = str(body.get("response_format") or "wav").strip().lower() if fmt not in ("wav", "mp3"): self._json(400, _error("unsupported response_format %r; this " "endpoint serves wav and mp3" % fmt)) return # `voice_id` is the alias the omp harness forwards for `voice`. A request # that names nothing takes the declared default — the harness has no # setting that feeds a voice, so voice-less calls are the normal shape # there, not a client bug. requested = body.get("voice") or body.get("voice_id") defaulted = not requested and bool(cfg.default_voice) voice = requested or cfg.default_voice reference = resolve_voice(cfg.voices_dir, voice) if reference is None: available = list_voices(cfg.voices_dir) listed = ", ".join(available) if available else "(none: no voice pack with a reference.wav in %s)" % cfg.voices_dir self._json(400, _error("unknown voice %r; available voices: %s" % (voice, listed), available)) return model = str(body.get("model") or cfg.upstream_model) try: if cfg.backend == "omni-http": audio = self._speech_http(cfg, model, text, str(voice), reference) elif cfg.backend == "llama-tts-cli": audio = self._speech_cli(cfg, text, reference) else: raise ShimError(500, "unknown TTS_BACKEND %r (expected omni-http or llama-tts-cli)" % cfg.backend) if fmt == "mp3": audio = to_mp3(audio, cfg.timeout) except ShimError as exc: self._json(exc.status, _error(exc.message)) return content_type = MP3_CONTENT_TYPE if fmt == "mp3" else WAV_CONTENT_TYPE self._bytes(200, audio, content_type) self.log_message("speech voice=%s%s format=%s chars=%d -> %d bytes %s", voice, " (default)" if defaulted else "", fmt, len(text), len(audio), content_type) # -- backends ---------------------------------------------------------- def _speech_http(self, cfg, model, text, voice, reference): """Forward to the upstream with the cloned voice riding along as base64 ``reference_audio``. The upstream is asked for wav always; mp3 is our own re-encode on the way out.""" try: payload = { "model": model, "input": text, "voice": voice, "response_format": "wav", "reference_audio": base64.b64encode(reference.read_bytes()).decode("ascii"), } except OSError as exc: raise ShimError(500, "cannot read %s: %s" % (reference, exc)) url = cfg.upstream + "/v1/audio/speech" request = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST") try: with urllib.request.urlopen(request, timeout=cfg.timeout) as response: audio = response.read() except urllib.error.HTTPError as exc: detail = exc.read(512).decode("utf-8", "replace").strip() raise ShimError(502, "upstream %s returned HTTP %d: %s" % (url, exc.code, detail or "no detail")) except urllib.error.URLError as exc: raise ShimError(502, "upstream %s unreachable: %s" % (url, exc.reason)) except (TimeoutError, OSError) as exc: raise ShimError(502, "upstream %s failed: %s" % (url, exc)) if not audio: raise ShimError(502, "upstream %s returned no audio" % url) return audio def _speech_cli(self, cfg, text, reference): """Run the local TTS binary once, in a temp dir that is always removed — also on failure, or a crash-looping binary fills /tmp.""" if not cfg.llama_bin or not cfg.llama_model: raise ShimError(500, "TTS_BACKEND=llama-tts-cli needs LLAMA_TTS_BIN and LLAMA_TTS_MODEL") work = tempfile.TemporaryDirectory(prefix="frank-tts-") try: out_path = Path(work.name) / "speech.wav" argv = ["nice", "-n", "19", "env", "OMP_NUM_THREADS=%d" % cfg.llama_threads, cfg.llama_bin, "-m", cfg.llama_model] if cfg.llama_mmproj: argv += ["-mm", cfg.llama_mmproj] argv += ["--tts-lang", cfg.llama_lang, "--tts-speaker-file", str(reference), "-p", text, "--output", str(out_path)] try: proc = subprocess.run(argv, cwd=work.name, capture_output=True, timeout=cfg.timeout) except FileNotFoundError as exc: raise ShimError(500, "cannot run the TTS command: %s" % exc) except subprocess.TimeoutExpired: raise ShimError(502, "TTS binary timed out after %gs" % cfg.timeout) if proc.returncode != 0: raise ShimError(502, "TTS binary failed (exit %s): %s" % (proc.returncode, _tail(proc.stderr))) if not out_path.is_file(): raise ShimError(502, "TTS binary exited 0 but wrote no %s" % out_path.name) return out_path.read_bytes() finally: work.cleanup() class SpeechServer(ThreadingHTTPServer): """Threads so a health probe answers while a long synthesis runs; daemon threads so SIGTERM is never held hostage by a stuck request.""" daemon_threads = True allow_reuse_address = True config = None def create_server(config=None, host="0.0.0.0", port=None): """Bind the shim. ``config``/``port`` are the injectable seams: tests pass a Config with a temp VOICES_DIR and ``port=0`` for an ephemeral port.""" cfg = Config.from_env() if config is None else config server = SpeechServer((host, cfg.port if port is None else port), SpeechHandler) server.config = cfg return server def main(argv=None): parser = argparse.ArgumentParser( prog="frank-tts", description="OpenAI-compatible speech endpoint for cloned voice packs") parser.add_argument("--host", default="0.0.0.0", help="bind address (default: all interfaces)") parser.add_argument("--port", type=int, default=None, help="bind port (default: $SHIM_PORT, 0 = ephemeral)") args = parser.parse_args(argv) server = create_server(host=args.host, port=args.port) cfg = server.config bound = server.server_address[1] print("frank-tts: backend=%s voices_dir=%s listening on http://%s:%d" % (cfg.backend, cfg.voices_dir, args.host, bound), flush=True) try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() return 0 if __name__ == "__main__": sys.exit(main())