150 lines
4.6 KiB
Python
150 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
||
"""Mock-TTS-Worker für lokale Tests (gleiche HTTP-API wie tts_worker.py).
|
||
|
||
Erzeugt ein kurzes, leises WAV statt echten Audios.
|
||
|
||
API:
|
||
GET /status -> {"status":"ok","ready":true,"voices":[...],...}
|
||
POST /tts -> {"text":"...","voice":"...","speed":1.0,"format":"wav"}
|
||
-> binäres Audio (WAV/MP3/FLAC/PCM)
|
||
|
||
Optionen (Umgebungsvariablen):
|
||
MOCK_TTS_PORT Port (Default 8082)
|
||
MOCK_TTS_DELAY Sekunden pro Synthese (Default 0.2)
|
||
MOCK_TTS_LOG Datei, in die die Requests geloggt werden (JSON-Zeilen)
|
||
|
||
Sonder-Texte:
|
||
"FAIL" -> Worker antwortet mit 500 (simulierter Fehler)
|
||
"SLOW" -> Worker schläft 5 s
|
||
"""
|
||
|
||
import io
|
||
import json
|
||
import os
|
||
import struct
|
||
import sys
|
||
import time
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
|
||
PORT = int(os.environ.get("MOCK_TTS_PORT", "8082"))
|
||
DELAY = float(os.environ.get("MOCK_TTS_DELAY", "0.2"))
|
||
LOG_FILE = os.environ.get("MOCK_TTS_LOG", "")
|
||
SAMPLE_RATE = 24000
|
||
|
||
|
||
def _make_wav(seconds: float = 0.1) -> bytes:
|
||
"""Erzeugt ein kurzes WAV (16-bit, mono, 24 kHz) mit Sinus-Ton."""
|
||
import math
|
||
n = int(SAMPLE_RATE * seconds)
|
||
samples = b"".join(
|
||
struct.pack("<h", int(8000 * math.sin(2 * math.pi * 440 * i / SAMPLE_RATE)))
|
||
for i in range(n)
|
||
)
|
||
header = (
|
||
b"RIFF" + struct.pack("<I", 36 + len(samples)) + b"WAVE"
|
||
+ b"fmt " + struct.pack("<IHHIIHH", 16, 1, 1, SAMPLE_RATE,
|
||
SAMPLE_RATE * 2, 2, 16)
|
||
+ b"data" + struct.pack("<I", len(samples))
|
||
)
|
||
return header + samples
|
||
|
||
|
||
def _log_request(req: dict) -> None:
|
||
if not LOG_FILE:
|
||
return
|
||
try:
|
||
with open(LOG_FILE, "a", encoding="utf-8") as f:
|
||
f.write(json.dumps(req) + "\n")
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = "MockXTTSTTS/1.0"
|
||
timeout = 60
|
||
|
||
def log_message(self, fmt, *args): # noqa: N802
|
||
pass # leise
|
||
|
||
def do_GET(self): # noqa: N802
|
||
if self.path.split("?", 1)[0] == "/status":
|
||
self._send_json(200, {
|
||
"status": "ok",
|
||
"ready": True,
|
||
"voices": ["alloy"],
|
||
"default_voice": "alloy",
|
||
"load_errors": [],
|
||
"sample_rate": SAMPLE_RATE,
|
||
"uptime_seconds": 1.0,
|
||
"total_requests": 0,
|
||
"last_seconds": None,
|
||
"last_voice": None,
|
||
"last_error": None,
|
||
})
|
||
else:
|
||
self._send_json(404, {"error": "not found"})
|
||
|
||
def do_POST(self): # noqa: N802
|
||
if self.path.split("?", 1)[0] != "/tts":
|
||
self._send_json(404, {"error": "not found"})
|
||
return
|
||
length = int(self.headers.get("Content-Length") or 0)
|
||
try:
|
||
req = json.loads(self.rfile.read(length))
|
||
except ValueError:
|
||
self._send_json(400, {"error": "ungültiges JSON"})
|
||
return
|
||
_log_request(req)
|
||
text = req.get("text", "")
|
||
if text == "SLOW":
|
||
time.sleep(5.0)
|
||
else:
|
||
time.sleep(DELAY)
|
||
if text == "FAIL":
|
||
self._send_json(500, {"error": "simulierter TTS-Fehler"})
|
||
return
|
||
fmt = req.get("format", "wav")
|
||
if fmt == "wav":
|
||
data, ctype = _make_wav(), "audio/wav"
|
||
elif fmt == "mp3":
|
||
# Minimales MP3-Frame (silence) – reicht für Format-Tests.
|
||
data = b"\xff\xfb\x90\x00" + b"\x00" * 417
|
||
ctype = "audio/mpeg"
|
||
elif fmt == "flac":
|
||
data = b"fLaC" + b"\x00" * 100
|
||
ctype = "audio/flac"
|
||
else: # pcm
|
||
data = b"\x00" * 4800
|
||
ctype = "application/octet-stream"
|
||
self.send_response(200)
|
||
self.send_header("Content-Type", ctype)
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.send_header("Connection", "close")
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def _send_json(self, code: int, payload: dict) -> None:
|
||
body = json.dumps(payload).encode()
|
||
self.send_response(code)
|
||
self.send_header("Content-Type", "application/json")
|
||
self.send_header("Content-Length", str(len(body)))
|
||
self.send_header("Connection", "close")
|
||
self.end_headers()
|
||
self.wfile.write(body)
|
||
|
||
|
||
def main() -> None:
|
||
server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)
|
||
server.daemon_threads = True
|
||
print(f"Mock-TTS-Worker auf Port {PORT}", file=sys.stderr)
|
||
try:
|
||
server.serve_forever()
|
||
except KeyboardInterrupt:
|
||
pass
|
||
finally:
|
||
server.server_close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|