rorappandClaude Opus 5.5
f90b7df31c
ci / test (push) Successful in 24s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
89 lines
3.8 KiB
Python
89 lines
3.8 KiB
Python
"""Schlanker OpenAI-kompatibler TTS-Dienst (`/v1/audio/speech`) für CPU-Engines wie Piper.
|
|
|
|
Der Befehl ist als Vorlage konfigurierbar, weil sich die Piper-CLI zwischen Versionen unterscheidet:
|
|
KNASTER_TTS_CMD Standard: "piper -m {model} -f {out}" (Text kommt über stdin)
|
|
KNASTER_TTS_VOICES "thorsten=/voices/de_DE-thorsten-medium.onnx,kerstin=/voices/de_DE-kerstin-low.onnx"
|
|
KNASTER_TTS_DEFAULT_VOICE Name der Standardstimme (sonst die erste)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
import tempfile
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
from fastapi import FastAPI, Header, HTTPException, Response
|
|
from pydantic import BaseModel, Field
|
|
|
|
DEFAULT_CMD = "piper -m {model} -f {out}"
|
|
MAX_CHARS = 5000
|
|
|
|
|
|
class SpeechIn(BaseModel):
|
|
model: str = "tts-de"
|
|
input: str = Field(min_length=1, max_length=MAX_CHARS)
|
|
voice: str = "default"
|
|
response_format: str = "wav"
|
|
|
|
|
|
class CommandBackend:
|
|
def __init__(self, voices: dict[str, str], default_voice: str | None = None, cmd: str = DEFAULT_CMD,
|
|
timeout: float = 120.0, max_parallel: int = 2):
|
|
if not voices:
|
|
raise ValueError("Mindestens eine Stimme konfigurieren")
|
|
self.voices = voices
|
|
self.default_voice = default_voice or next(iter(voices))
|
|
self.cmd = cmd
|
|
self.timeout = timeout
|
|
self._slots = threading.Semaphore(max_parallel) # CPU nicht überbuchen
|
|
|
|
@classmethod
|
|
def from_env(cls) -> CommandBackend:
|
|
raw = os.environ.get("KNASTER_TTS_VOICES", "")
|
|
voices = dict(item.split("=", 1) for item in raw.split(",") if "=" in item)
|
|
return cls(voices, os.environ.get("KNASTER_TTS_DEFAULT_VOICE") or None,
|
|
os.environ.get("KNASTER_TTS_CMD", DEFAULT_CMD),
|
|
max_parallel=int(os.environ.get("KNASTER_TTS_PARALLEL", "2")))
|
|
|
|
def synthesize(self, text: str, voice: str = "default") -> bytes:
|
|
name = self.default_voice if voice in ("default", "", None) else voice
|
|
if name not in self.voices:
|
|
raise KeyError(f"Stimme {name} unbekannt (verfügbar: {', '.join(self.voices)})")
|
|
with self._slots, tempfile.TemporaryDirectory() as tmp:
|
|
out = Path(tmp) / "out.wav"
|
|
args = [part.format(model=self.voices[name], out=str(out)) for part in shlex.split(self.cmd)]
|
|
proc = subprocess.run(args, input=text.encode("utf-8"), capture_output=True, timeout=self.timeout)
|
|
if proc.returncode != 0 or not out.exists():
|
|
raise RuntimeError(f"TTS-Befehl fehlgeschlagen ({proc.returncode}): "
|
|
f"{proc.stderr.decode('utf-8', 'replace')[-300:]}")
|
|
return out.read_bytes()
|
|
|
|
|
|
def create_tts_app(backend: CommandBackend | None = None, token: str | None = None) -> FastAPI:
|
|
backend = backend or CommandBackend.from_env()
|
|
token = os.environ.get("KNASTER_TTS_TOKEN", "") if token is None else token
|
|
app = FastAPI(title="knaster-tts")
|
|
|
|
@app.get("/health")
|
|
def health() -> dict:
|
|
return {"status": "ok", "voices": list(backend.voices), "default": backend.default_voice}
|
|
|
|
@app.post("/v1/audio/speech")
|
|
def speech(body: SpeechIn, authorization: str | None = Header(default=None)) -> Response:
|
|
if token and authorization != f"Bearer {token}":
|
|
raise HTTPException(401, "unauthorized")
|
|
if body.response_format != "wav":
|
|
raise HTTPException(400, "nur response_format=wav")
|
|
try:
|
|
audio = backend.synthesize(body.input, body.voice)
|
|
except KeyError as exc:
|
|
raise HTTPException(400, str(exc)) from exc
|
|
except (RuntimeError, subprocess.TimeoutExpired) as exc:
|
|
raise HTTPException(502, str(exc)) from exc
|
|
return Response(audio, media_type="audio/wav")
|
|
|
|
return app
|