mission-control-v2/voice_service/app.py

#!/usr/bin/env python3
"""
Voice-Sidecar für Mission Control 2.0 — lokales STT + gestuftes TTS für „Mit Hermes reden".

WARUM ein eigener Dienst? Die ML-Stacks (faster-whisper, piper, chatterbox + torch) brauchen
ihr eigenes Python-3.12-venv — das MC2-Backend läuft auf Python 3.14 und kann sie nicht
importieren. Genau wie der Mem0-Sidecar (mem0_service/) kapselt dieser schlanke FastAPI-Dienst
die schwere Voice-Logik und exponiert sie auf localhost. MC2 (backend/routers/voice.py) proxyt
ihn nach außen; der Browser-Voice-Client (Frontend „Sprechen"-Tab) redet nie direkt mit ihm.

Pipeline-Rolle:
- STT : faster-whisper (Default `medium`, int8, CPU, Sprache=de) — Mikro-Audio → Text.
- TTS : GESTUFT, Engine im Request wählbar (kein Lock-in):
        * `piper`      — schneller Standard, CPU, quasi-sofort, robustes Deutsch (thorsten).
        * `chatterbox` — Premium/Wunschstimme (MIT), Voice-Cloning, dt. über Multilingual.
                         Lazy-Load (Modell erst beim ersten Aufruf) → Dienststart bleibt schnell.
                         Device via VOICE_CHATTERBOX_DEVICE (cpu | cuda); ROCm/iGPU (Strix Halo)
                         per HSA_OVERRIDE_GFX_VERSION=11.0.0 als späterer Umschalter.

Läuft als systemd-User-Dienst (deploy/voice-service.service) im ~/.voice/venv (Python 3.12).
Bind: 127.0.0.1 (nur lokal; MC2 proxyt nach außen).
"""

import io
import json
import logging
import os
import subprocess
import tempfile
from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from fastapi.responses import Response
from pydantic import BaseModel

log = logging.getLogger("voice_service")

# --- Konfiguration (alles über Env überschreibbar; Defaults = Box-Stand) ----------
PORT = int(os.environ.get("VOICE_PORT", "8650"))
# STT
STT_MODEL = os.environ.get("VOICE_STT_MODEL", "medium")       # base|small|medium|large-v3
STT_DEVICE = os.environ.get("VOICE_STT_DEVICE", "cpu")
STT_COMPUTE = os.environ.get("VOICE_STT_COMPUTE", "int8")     # int8=CPU-schonend
STT_LANG = os.environ.get("VOICE_STT_LANG", "de")
# Piper (Default-TTS): Verzeichnis mit *.onnx (+ *.onnx.json) Stimmen + das Piper-Binary.
# Wir nutzen das offizielle Binary (statt des piper-tts-PyPI-Pakets), weil dessen Abhängigkeit
# piper-phonemize auf neueren Linux-Systemen keine Wheels hat. Das Binary bringt espeak-ng mit.
VOICES_DIR = Path(os.environ.get("VOICE_PIPER_DIR", str(Path.home() / ".voice" / "voices")))
PIPER_DEFAULT = os.environ.get("VOICE_PIPER_DEFAULT", "de_DE-thorsten-medium")
PIPER_BIN = os.environ.get("VOICE_PIPER_BIN", str(Path.home() / ".voice" / "piper" / "piper"))
# Chatterbox (Premium-TTS)
CHATTERBOX_DEVICE = os.environ.get("VOICE_CHATTERBOX_DEVICE", "cpu")
CHATTERBOX_LANG = os.environ.get("VOICE_CHATTERBOX_LANG", "de")
# Optionaler Referenz-WAV für Voice-Cloning (10 s Sprachprobe). Leer = Chatterbox-Standardstimme.
CHATTERBOX_REF = os.environ.get("VOICE_CHATTERBOX_REF", "")
# ElevenLabs (Premium-Cloud-TTS): saubere, native deutsche Stimmen (Voice Library). Key wird zur
# Laufzeit gelesen (env ODER ~/.hermes/.env), damit Nachtragen ohne Code-Deploy reicht. Default-
# Modell = Flash v2.5 (multilingual, 0,5 Credit/Zeichen → schont das Free-Tier-Kontingent).
ELEVEN_MODEL = os.environ.get("VOICE_ELEVENLABS_MODEL", "eleven_flash_v2_5")
ELEVEN_DEFAULT_VOICE = os.environ.get("VOICE_ELEVENLABS_VOICE", "21m00Tcm4TlvDq8ikWAM")
HERMES_ENV = Path.home() / ".hermes" / ".env"
# Edge-TTS: native deutsche Azure-Neural-Stimmen, gratis & ohne Key (kein Cloning → kein Akzent).
EDGE_DEFAULT = os.environ.get("VOICE_EDGE_VOICE", "de-DE-KatjaNeural")


def _eleven_key() -> str:
    """ElevenLabs-Key aus Prozess-Env oder (un-auskommentiert) aus ~/.hermes/.env."""
    k = os.environ.get("ELEVENLABS_API_KEY")
    if k:
        return k.strip()
    try:
        for line in HERMES_ENV.read_text(encoding="utf-8").splitlines():
            s = line.strip()
            if s.startswith("ELEVENLABS_API_KEY=") and not s.startswith("#"):
                return s.split("=", 1)[1].strip().strip('"').strip("'")
    except OSError:
        pass
    return ""


# =================================================================================
# STT — faster-whisper (lazy Singleton)
# =================================================================================
_stt = None


def stt_model():
    global _stt
    if _stt is None:
        from faster_whisper import WhisperModel
        log.info("Lade faster-whisper '%s' (%s/%s) …", STT_MODEL, STT_DEVICE, STT_COMPUTE)
        _stt = WhisperModel(STT_MODEL, device=STT_DEVICE, compute_type=STT_COMPUTE)
    return _stt


def transcribe(audio_bytes: bytes, suffix: str, language: str) -> str:
    # PyAV (in faster-whisper) dekodiert webm/opus/ogg/wav am robustesten aus einer Datei.
    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tf:
        tf.write(audio_bytes)
        tmp = tf.name
    try:
        segments, _info = stt_model().transcribe(
            tmp, language=language or None, vad_filter=True, beam_size=5,
        )
        return "".join(s.text for s in segments).strip()
    finally:
        try:
            os.unlink(tmp)
        except OSError:
            pass


# =================================================================================
# TTS — Piper (offizielles Binary; Stimme = ONNX-Datei + .json daneben)
# =================================================================================
def piper_tts(text: str, voice: str) -> bytes:
    name = voice or PIPER_DEFAULT
    onnx = VOICES_DIR / f"{name}.onnx"
    if not onnx.exists():
        raise HTTPException(404, f"Piper-Stimme '{name}' nicht gefunden ({onnx}).")
    if not os.path.exists(PIPER_BIN):
        raise HTTPException(500, f"Piper-Binary fehlt ({PIPER_BIN}).")
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tf:
        out = tf.name
    try:
        proc = subprocess.run(
            [PIPER_BIN, "-m", str(onnx), "-f", out],
            input=text.encode("utf-8"), capture_output=True, timeout=60,
        )
        if proc.returncode != 0:
            raise HTTPException(500, f"Piper-Fehler: {proc.stderr.decode('utf-8', 'replace')[:300]}")
        return Path(out).read_bytes()
    finally:
        try:
            os.unlink(out)
        except OSError:
            pass


def piper_list() -> list[dict]:
    if not VOICES_DIR.exists():
        return []
    return [
        {"engine": "piper", "id": p.stem, "label": p.stem, "clonable": False}
        for p in sorted(VOICES_DIR.glob("*.onnx"))
    ]


# =================================================================================
# TTS — Chatterbox (lazy; Multilingual für Deutsch; optional Voice-Cloning)
# =================================================================================
_chatterbox = None


def chatterbox_model():
    global _chatterbox
    if _chatterbox is None:
        log.info("Lade Chatterbox (Multilingual, device=%s) — einmalig, dauert kurz …", CHATTERBOX_DEVICE)
        from chatterbox.mtl_tts import ChatterboxMultilingualTTS
        _chatterbox = ChatterboxMultilingualTTS.from_pretrained(device=CHATTERBOX_DEVICE)
    return _chatterbox


def chatterbox_tts(text: str, language: str, ref_path: str) -> bytes:
    import soundfile as sf
    model = chatterbox_model()
    kwargs = {"language_id": language or CHATTERBOX_LANG}
    ref = ref_path or CHATTERBOX_REF
    if ref and os.path.exists(ref):
        kwargs["audio_prompt_path"] = ref          # Zero-Shot Voice-Cloning aus Referenz
    wav = model.generate(text, **kwargs)
    # wav = torch.Tensor [1, N] @ model.sr → in WAV-Bytes serialisieren.
    import numpy as np
    arr = wav.squeeze(0).detach().cpu().numpy().astype(np.float32)
    buf = io.BytesIO()
    sf.write(buf, arr, int(model.sr), format="WAV", subtype="PCM_16")
    return buf.getvalue()


def chatterbox_list() -> list[dict]:
    # Chatterbox hat keine festen „Stimm-Dateien": Standardstimme + optionale Klon-Referenz.
    items = [{"engine": "chatterbox", "id": "default", "label": "Chatterbox (Standard, dt.)", "clonable": True}]
    if CHATTERBOX_REF and os.path.exists(CHATTERBOX_REF):
        items.append({"engine": "chatterbox", "id": "clone", "label": "Chatterbox (geklonte Stimme)", "clonable": True})
    return items


# =================================================================================
# TTS — ElevenLabs (Premium-Cloud; native dt. Stimmen, Voice Library). Nur HTTP (stdlib).
# =================================================================================
def elevenlabs_tts(text: str, voice: str) -> bytes:
    import urllib.error
    import urllib.request

    key = _eleven_key()
    if not key:
        raise HTTPException(503, "ELEVENLABS_API_KEY nicht gesetzt (in ~/.hermes/.env eintragen).")
    vid = voice or ELEVEN_DEFAULT_VOICE
    body = json.dumps({"text": text, "model_id": ELEVEN_MODEL}).encode("utf-8")
    req = urllib.request.Request(
        f"https://api.elevenlabs.io/v1/text-to-speech/{vid}",
        data=body,
        headers={"xi-api-key": key, "Content-Type": "application/json", "Accept": "audio/mpeg"},
    )
    try:
        with urllib.request.urlopen(req, timeout=60) as r:
            return r.read()
    except urllib.error.HTTPError as e:
        detail = e.read()[:200].decode("utf-8", "replace")
        raise HTTPException(502, f"ElevenLabs-Fehler {e.code}: {detail}")
    except urllib.error.URLError as e:
        raise HTTPException(502, f"ElevenLabs nicht erreichbar: {e}")


def elevenlabs_list() -> list[dict]:
    import urllib.request

    key = _eleven_key()
    if not key:
        return []
    try:
        req = urllib.request.Request("https://api.elevenlabs.io/v1/voices", headers={"xi-api-key": key})
        with urllib.request.urlopen(req, timeout=10) as r:
            data = json.loads(r.read())
        return [
            {"engine": "elevenlabs", "id": v["voice_id"], "label": v.get("name", v["voice_id"]), "clonable": False}
            for v in data.get("voices", [])
        ]
    except Exception:  # noqa: BLE001
        log.warning("ElevenLabs-Stimmenliste fehlgeschlagen", exc_info=True)
        return []


# =================================================================================
# TTS — Edge-TTS (native dt. Azure-Neural-Stimmen, gratis, kein Key, KEIN Cloning → kein Akzent)
# =================================================================================
def edge_tts_synth(text: str, voice: str) -> bytes:
    import asyncio
    import edge_tts

    v = voice or EDGE_DEFAULT

    async def _run() -> bytes:
        buf = bytearray()
        communicate = edge_tts.Communicate(text, v)
        async for chunk in communicate.stream():
            if chunk["type"] == "audio":
                buf.extend(chunk["data"])
        return bytes(buf)

    try:
        return asyncio.run(_run())
    except Exception as exc:  # noqa: BLE001
        raise HTTPException(502, f"Edge-TTS-Fehler: {exc}")


_edge_cache: list | None = None


def edge_list() -> list[dict]:
    """Deutsche Edge/Azure-Stimmen (einmal vom Dienst geholt + gecacht). Weibliche zuerst."""
    global _edge_cache
    if _edge_cache is not None:
        return _edge_cache
    try:
        import asyncio
        import edge_tts

        voices = asyncio.run(edge_tts.list_voices())
        out = []
        for v in voices:
            if not v.get("Locale", "").startswith("de-"):
                continue
            short = v["ShortName"]                       # z.B. de-DE-KatjaNeural
            gender = "weiblich" if v.get("Gender") == "Female" else "männlich"
            name = short.split("-")[-1].replace("Neural", "").replace("Multilingual", " (multiling.)")
            out.append({"engine": "edge", "id": short,
                        "label": f"{name} · {v['Locale']} · {gender}",
                        "clonable": False, "_female": v.get("Gender") == "Female"})
        out.sort(key=lambda i: (not i.pop("_female"), i["id"]))
        _edge_cache = out
    except Exception:  # noqa: BLE001
        log.warning("Edge-Stimmenliste fehlgeschlagen", exc_info=True)
        _edge_cache = []
    return _edge_cache


# =================================================================================
# FastAPI
# =================================================================================
@asynccontextmanager
async def lifespan(_app: FastAPI):
    logging.basicConfig(level=logging.INFO)
    try:
        stt_model()  # STT beim Start vorwärmen (Modell aus HF-Cache laden)
        log.info("Voice-Sidecar bereit (STT '%s', Piper-Dir %s).", STT_MODEL, VOICES_DIR)
    except Exception:
        log.exception("STT-Vorwärmen fehlgeschlagen (Dienst läuft, /health meldet Detail).")
    yield


app = FastAPI(title="MC2 Voice Sidecar", lifespan=lifespan)


class TTSIn(BaseModel):
    text: str
    engine: str = "piper"          # piper | chatterbox
    voice: str = ""                # Piper-Stimmname; bei Chatterbox: "default" | "clone"
    language: str = ""             # überschreibt Default-Sprache
    ref_path: str = ""             # optionaler Klon-Referenz-WAV (Chatterbox)


@app.get("/health")
def health() -> dict:
    return {"ok": True, "stt_model": STT_MODEL,
            "engines": ["edge", "piper", "chatterbox", "elevenlabs"],
            "elevenlabs_key": bool(_eleven_key()),
            "piper_voices": [v["id"] for v in piper_list()]}


@app.get("/voices")
def voices() -> dict:
    return {"voices": edge_list() + piper_list() + chatterbox_list() + elevenlabs_list(),
            "default": {"engine": "piper", "voice": PIPER_DEFAULT}}


@app.post("/stt")
async def stt(audio: UploadFile = File(...), language: str = Form(default="")) -> dict:
    data = await audio.read()
    if not data:
        raise HTTPException(400, "Leeres Audio.")
    suffix = Path(audio.filename or "rec.webm").suffix or ".webm"
    text = transcribe(data, suffix, language or STT_LANG)
    return {"text": text}


@app.post("/tts")
def tts(body: TTSIn) -> Response:
    text = (body.text or "").strip()
    if not text:
        raise HTTPException(400, "Leerer Text.")
    if body.engine == "edge":
        return Response(content=edge_tts_synth(text, body.voice), media_type="audio/mpeg")
    if body.engine == "elevenlabs":
        return Response(content=elevenlabs_tts(text, body.voice), media_type="audio/mpeg")
    if body.engine == "chatterbox":
        return Response(content=chatterbox_tts(text, body.language, body.ref_path), media_type="audio/wav")
    return Response(content=piper_tts(text, body.voice), media_type="audio/wav")


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=PORT)