AI-Profile-Router/platform/docker/piper/piper_worker.py

#!/usr/bin/env python3
"""Small, private Piper worker for the Mike AI profile router.

The public OpenAI-compatible endpoint remains in the router.  This worker only
accepts the narrow internal /status and /tts protocol and never logs input text.
"""

from __future__ import annotations

import io
import json
import os
import subprocess
import threading
import wave
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

from piper import PiperVoice, SynthesisConfig


DATA_DIR = Path(os.getenv("PIPER_DATA_DIR", "/data"))
VOICE_NAME = os.getenv("PIPER_VOICE", "de_DE-thorsten-high")
VOICE_ALIAS = os.getenv("PIPER_VOICE_ALIAS", "alloy")
HOST = os.getenv("PIPER_HOST", "0.0.0.0")
PORT = int(os.getenv("PIPER_PORT", "8085"))
MAX_TEXT_CHARS = int(os.getenv("PIPER_MAX_TEXT_CHARS", "8000"))
MAX_REQUEST_BYTES = int(os.getenv("PIPER_MAX_REQUEST_BYTES", "65536"))

VOICE_PATH = DATA_DIR / f"{VOICE_NAME}.onnx"
VOICE = PiperVoice.load(str(VOICE_PATH))
SYNTHESIS_LOCK = threading.Lock()


def synthesize_wav(text: str, speed: float) -> bytes:
    """Synthesize a complete WAV in memory without retaining the text."""
    output = io.BytesIO()
    config = SynthesisConfig(length_scale=1.0 / speed)
    with SYNTHESIS_LOCK, wave.open(output, "wb") as wav_file:
        VOICE.synthesize_wav(text, wav_file, syn_config=config)
    return output.getvalue()


def wav_to_mp3(wav_bytes: bytes) -> bytes:
    """Convert Piper's WAV to the MP3 format Open WebUI requests by default."""
    result = subprocess.run(
        [
            "ffmpeg", "-hide_banner", "-loglevel", "error",
            "-f", "wav", "-i", "pipe:0",
            "-codec:a", "libmp3lame", "-b:a", "96k",
            "-f", "mp3", "pipe:1",
        ],
        input=wav_bytes,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
        timeout=120,
    )
    if result.returncode != 0:
        raise RuntimeError("ffmpeg conversion failed")
    return result.stdout


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, fmt: str, *args: object) -> None:
        # Deliberately omit URLs and request bodies from the log.
        print(f"piper-worker: {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
        self.send_json(
            HTTPStatus.OK,
            {
                "ready": True,
                "engine": "piper",
                "model": VOICE_NAME,
                "voices": [VOICE_ALIAS],
            },
        )

    def do_POST(self) -> None:  # noqa: N802
        if self.path != "/tts":
            self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"})
            return

        try:
            content_length = int(self.headers.get("Content-Length", "0"))
        except ValueError:
            content_length = 0
        if content_length <= 0 or content_length > MAX_REQUEST_BYTES:
            self.send_json(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": "invalid request size"})
            return

        try:
            request = json.loads(self.rfile.read(content_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

        try:
            audio = synthesize_wav(text.strip(), speed)
            if output_format == "mp3":
                audio = wav_to_mp3(audio)
                content_type = "audio/mpeg"
            else:
                content_type = "audio/wav"
        except (OSError, RuntimeError, subprocess.SubprocessError):
            self.send_json(HTTPStatus.INTERNAL_SERVER_ERROR, {"error": "synthesis failed"})
            return

        self.send_bytes(HTTPStatus.OK, audio, content_type)


if __name__ == "__main__":
    print(f"Piper worker ready: {VOICE_NAME} as {VOICE_ALIAS} on {HOST}:{PORT}")
    ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()