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
# Engine: parakeet (Default, ~10× schneller + bessere Deutsch-WER als whisper-medium,
# Review 2026-07-02) | whisper (Fallback). Pro Request via Form-Feld `engine` überschreibbar.
STT_ENGINE = os.environ.get("VOICE_STT_ENGINE", "parakeet")
STT_MODEL = os.environ.get("VOICE_STT_MODEL", "medium")       # whisper: 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")
# Parakeet-TDT 0.6B v3 (25 EU-Sprachen inkl. DE, Auto-Language) via onnx-asr.
PARAKEET_MODEL = os.environ.get("VOICE_PARAKEET_MODEL", "nemo-parakeet-tdt-0.6b-v3")
PARAKEET_QUANT = os.environ.get("VOICE_PARAKEET_QUANT", "int8")  # ""=fp32
# 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", "")
# Vom Nutzer hochgeladene Klon-Referenz (persistiert über Restarts). active.txt zeigt auf die Datei.
REFS_DIR = Path(os.environ.get("VOICE_REFS_DIR", str(Path.home() / ".voice" / "refs")))
ACTIVE_REF = REFS_DIR / "active.txt"


def active_ref() -> str:
    """Pfad der aktuell aktiven Klon-Referenz (Upload bevorzugt, sonst Env-Default)."""
    try:
        p = ACTIVE_REF.read_text(encoding="utf-8").strip()
        if p and os.path.exists(p):
            return p
    except OSError:
        pass
    return 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).
# multilingual_v2 = natürlichstes Modell (statt Flash, das auf Tempo optimiert/flacher ist).
ELEVEN_MODEL = os.environ.get("VOICE_ELEVENLABS_MODEL", "eleven_multilingual_v2")
ELEVEN_DEFAULT_VOICE = os.environ.get("VOICE_ELEVENLABS_VOICE", "21m00Tcm4TlvDq8ikWAM")
# Stimm-Settings: niedrige Stability = ausdrucksstärker/lebendiger (hohe = monoton/roboterhaft).
ELEVEN_STABILITY = float(os.environ.get("VOICE_ELEVENLABS_STABILITY", "0.4"))
ELEVEN_SIMILARITY = float(os.environ.get("VOICE_ELEVENLABS_SIMILARITY", "0.8"))
ELEVEN_STYLE = float(os.environ.get("VOICE_ELEVENLABS_STYLE", "0.35"))
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_whisper(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


# --- Parakeet-TDT (onnx-asr, lazy Singleton) --------------------------------------
_parakeet = None
_parakeet_failed = False


def parakeet_model():
    global _parakeet, _parakeet_failed
    if _parakeet is None and not _parakeet_failed:
        try:
            import onnx_asr
            kwargs = {"quantization": PARAKEET_QUANT} if PARAKEET_QUANT else {}
            log.info("Lade Parakeet '%s' (quant=%s) …", PARAKEET_MODEL, PARAKEET_QUANT or "fp32")
            _parakeet = onnx_asr.load_model(PARAKEET_MODEL, **kwargs)
        except Exception:
            # int8-Variante kann je nach Release fehlen → einmal fp32 probieren, dann aufgeben.
            try:
                import onnx_asr
                log.warning("Parakeet quant=%s nicht ladbar, versuche fp32", PARAKEET_QUANT, exc_info=True)
                _parakeet = onnx_asr.load_model(PARAKEET_MODEL)
            except Exception:
                _parakeet_failed = True
                log.exception("Parakeet nicht verfügbar — STT fällt auf whisper zurück.")
    return _parakeet


def transcribe_parakeet(audio_bytes: bytes, suffix: str) -> str | None:
    """None = Engine nicht verfügbar (Aufrufer fällt auf whisper zurück).
    Parakeet v3 erkennt die Sprache selbst; onnx-asr liest nur WAV → Browser-webm/opus
    erst via PyAV (steckt in faster-whisper) auf 16 kHz mono dekodieren."""
    model = parakeet_model()
    if model is None:
        return None
    import soundfile as sf
    from faster_whisper.audio import decode_audio
    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tf:
        tf.write(audio_bytes)
        src = tf.name
    wav = src + ".wav"
    try:
        arr = decode_audio(src, sampling_rate=16000)
        sf.write(wav, arr, 16000, subtype="PCM_16")
        return (model.recognize(wav) or "").strip()
    finally:
        for p in (src, wav):
            try:
                os.unlink(p)
            except OSError:
                pass


def transcribe(audio_bytes: bytes, suffix: str, language: str, engine: str = "") -> str:
    eng = (engine or STT_ENGINE).strip().lower()
    if eng == "parakeet":
        text = transcribe_parakeet(audio_bytes, suffix)
        if text is not None:
            return text
    return transcribe_whisper(audio_bytes, suffix, language)


# =================================================================================
# Turn-Detection — Smart Turn v3 (pipecat, BSD-2): semantisches Äußerungs-Ende.
# Whisper-Tiny-Encoder + Klassifikator (8 MB int8, ~12 ms CPU). Der Lucy-Client fragt
# nach dem akustischen VAD-Ende hier nach: "War das ein fertiger Satz?" — bei
# 'incomplete' wartet er kurz weiter, statt mitten im Gedanken loszuantworten.
# Inferenz-Pfad 1:1 aus pipecat-ai/smart-turn inference.py (Feature-Shape muss passen).
# =================================================================================
TURN_REPO = os.environ.get("VOICE_TURN_REPO", "pipecat-ai/smart-turn-v3")
TURN_FILE = os.environ.get("VOICE_TURN_FILE", "smart-turn-v3.2-cpu.onnx")
_turn = None          # (session, feature_extractor)
_turn_failed = False


def turn_model():
    global _turn, _turn_failed
    if _turn is None and not _turn_failed:
        try:
            import onnxruntime as ort
            from huggingface_hub import hf_hub_download
            from transformers import WhisperFeatureExtractor
            log.info("Lade Smart Turn '%s/%s' …", TURN_REPO, TURN_FILE)
            path = hf_hub_download(TURN_REPO, TURN_FILE)
            so = ort.SessionOptions()
            so.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
            sess = ort.InferenceSession(path, sess_options=so, providers=["CPUExecutionProvider"])
            _turn = (sess, WhisperFeatureExtractor(chunk_length=8))
        except Exception:
            _turn_failed = True
            log.exception("Smart Turn nicht verfügbar — /turn meldet complete=true (Fallback).")
    return _turn


def check_turn(audio_bytes: bytes, suffix: str) -> dict:
    model = turn_model()
    if model is None:
        return {"complete": True, "probability": 1.0, "engine": "none"}
    sess, fe = model
    from faster_whisper.audio import decode_audio
    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tf:
        tf.write(audio_bytes)
        src = tf.name
    try:
        import numpy as np
        arr = decode_audio(src, sampling_rate=16000)
        # Smart Turn erwartet das Audio AM ENDE des 8-s-Fensters (Zeros vorn) — die Turn-Ende-
        # Hinweise liegen in den letzten Frames. Der FeatureExtractor padded rechts (Zeros hinten),
        # damit sah das Modell immer nur Padding und meldete konstant 'complete' (live diagnostiziert).
        # Darum manuell links auffüllen und exakt 8 s übergeben.
        n = 8 * 16000
        arr = arr[-n:]
        if len(arr) < n:
            arr = np.concatenate([np.zeros(n - len(arr), dtype=np.float32), arr.astype(np.float32)])
        inputs = fe(arr, sampling_rate=16000, return_tensors="np", padding="do_not_pad",
                    truncation=True, do_normalize=True)
        out = sess.run(None, {"input_features": inputs.input_features})
        # Output-Tensor heißt 'logits' — je nach Export roher Logit ODER schon Sigmoid.
        # Robust: Werte außerhalb [0,1] durch Sigmoid schicken, sonst direkt nutzen.
        raw = float(out[0][0].item() if hasattr(out[0][0], "item") else out[0][0])
        p = raw if 0.0 <= raw <= 1.0 else 1.0 / (1.0 + float(np.exp(-raw)))
        # EMPIRISCH VERIFIZIERT (2026-07-02, Box): v3.2-cpu liefert P(UNFERTIG) —
        # fertiger Satz -> 0.26, mitten im Wort abgeschnitten -> 0.96, Stille -> 0.99.
        # (Die Upstream-Doku beschreibt es andersherum; Messung schlägt Doku.)
        return {"complete": p <= 0.5, "probability": round(1.0 - p, 3), "engine": "smart-turn-v3"}
    finally:
        try:
            os.unlink(src)
        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 active_ref()
    if ref and os.path.exists(ref):
        kwargs["audio_prompt_path"] = ref          # Zero-Shot Voice-Cloning aus Referenz
    # Chatterbox hat einen sporadischen Bug (alignment_stream_analyzer → NoneType), der einzelne
    # Generierungen fehlschlagen lässt → im Voice-Loop „Stimme kam nicht". Ein Retry klappt meist.
    wav = None
    last_err: Exception | None = None
    for attempt in range(3):
        try:
            wav = model.generate(text, **kwargs)
            break
        except Exception as exc:
            last_err = exc
            log.warning("Chatterbox-Generation fehlgeschlagen (Versuch %d/3): %s", attempt + 1, exc)
    if wav is None:
        raise HTTPException(502, f"Chatterbox-Generation fehlgeschlagen: {last_err}")
    # 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)
    # Peak-Normalisierung gegen Übersteuern/Clipping (Klon-Output ist oft zu laut → kratzig).
    peak = float(np.max(np.abs(arr))) if arr.size else 0.0
    if peak > 0:
        arr = arr / peak * float(os.environ.get("VOICE_CHATTERBOX_PEAK", "0.9"))
    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 active_ref():
        items.append({"engine": "chatterbox", "id": "clone", "label": "Chatterbox (deine Referenz)", "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,
        "voice_settings": {
            "stability": ELEVEN_STABILITY,
            "similarity_boost": ELEVEN_SIMILARITY,
            "style": ELEVEN_STYLE,
            "use_speaker_boost": True,
        },
    }).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"},
    )
    import time
    # Free-Tier hat ein Concurrency-Limit → parallele Sätze geben 429. Retry mit Backoff fängt das ab.
    last = ""
    for attempt in range(4):
        try:
            with urllib.request.urlopen(req, timeout=60) as r:
                return r.read()
        except urllib.error.HTTPError as e:
            last = f"{e.code}: {e.read()[:200].decode('utf-8', 'replace')}"
            if e.code in (429, 500, 502, 503, 529) and attempt < 3:
                time.sleep(0.5 * (attempt + 1))
                continue
            raise HTTPException(502, f"ElevenLabs-Fehler {last}")
        except urllib.error.URLError as e:
            last = str(e)
            if attempt < 3:
                time.sleep(0.5 * (attempt + 1))
                continue
            raise HTTPException(502, f"ElevenLabs nicht erreichbar: {last}")
    raise HTTPException(502, f"ElevenLabs-Fehler (nach Retries): {last}")


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())
        # NUR eigene Stimmen (generated/cloned/professional) — Library-/premade-Stimmen sperrt das
        # Free-Tier per API (402). So landet im Picker nie eine unbrauchbare Stimme.
        return [
            {"engine": "elevenlabs", "id": v["voice_id"], "label": v.get("name", v["voice_id"])}
            for v in data.get("voices", [])
            if v.get("category") != "premade"
        ]
    except Exception:
        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:
        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:
        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:
        # Nur die konfigurierte Engine vorwärmen — whisper bleibt als Fallback lazy
        # (spart ~1,5 GB RAM, solange Parakeet läuft).
        if STT_ENGINE == "parakeet" and parakeet_model() is not None:
            log.info("Voice-Sidecar bereit (STT parakeet '%s', Piper-Dir %s).", PARAKEET_MODEL, VOICES_DIR)
        else:
            stt_model()
            log.info("Voice-Sidecar bereit (STT whisper '%s', Piper-Dir %s).", STT_MODEL, VOICES_DIR)
        # Smart Turn vorwärmen (Modell + transformers-Import + 1 Probe-Inferenz): der erste
        # /turn-Aufruf kostete sonst ~3,3 s — das traf beim Realtest den ersten gesprochenen Satz.
        try:
            import io as _io

            import numpy as _np
            import soundfile as _sf
            buf = _io.BytesIO()
            _sf.write(buf, _np.zeros(16000, dtype="float32"), 16000, format="WAV", subtype="PCM_16")
            check_turn(buf.getvalue(), ".wav")
            log.info("Smart Turn vorgewärmt.")
        except Exception:
            log.warning("Smart-Turn-Vorwärmen fehlgeschlagen (Feature bleibt lazy).", exc_info=True)
    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:
    stt_active = "parakeet" if (STT_ENGINE == "parakeet" and not _parakeet_failed) else "whisper"
    return {"ok": True, "stt_engine": stt_active,
            "stt_model": PARAKEET_MODEL if stt_active == "parakeet" else STT_MODEL,
            "engines": ["elevenlabs", "edge"],
            "elevenlabs_key": bool(_eleven_key())}


@app.get("/voices")
def voices() -> dict:
    # Nur die genutzten Engines: ElevenLabs (eigene Stimme) + Edge (gratis, nativ-deutsch).
    return {"voices": elevenlabs_list() + edge_list(),
            "default": {"engine": "edge", "voice": EDGE_DEFAULT}}


@app.post("/stt")
async def stt(audio: UploadFile = File(...), language: str = Form(default=""),
              engine: 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, engine)
    return {"text": text}


@app.post("/turn")
async def turn(audio: UploadFile = File(...)) -> dict:
    """Semantische Turn-Detection: War die Äußerung ein fertiger Gedanke? (Smart Turn v3)"""
    data = await audio.read()
    if not data:
        raise HTTPException(400, "Leeres Audio.")
    suffix = Path(audio.filename or "rec.wav").suffix or ".wav"
    return check_turn(data, suffix)


@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")


def _save_reference_as_wav(data: bytes, filename: str) -> str:
    """Upload (mp3/wav/…) → 24 kHz Mono-WAV für Chatterbox. Dekodierung via PyAV (steckt in
    faster-whisper → kein System-ffmpeg nötig); Fallback soundfile, sonst Rohdatei."""
    REFS_DIR.mkdir(parents=True, exist_ok=True)
    raw = REFS_DIR / ("upload" + (Path(filename or "ref").suffix or ".bin"))
    raw.write_bytes(data)
    dest = REFS_DIR / "ref.wav"
    import soundfile as sf
    try:
        from faster_whisper.audio import decode_audio  # PyAV-basiert, kann mp3/m4a/ogg/wav
        arr = decode_audio(str(raw), sampling_rate=24000)
        sf.write(str(dest), arr, 24000, subtype="PCM_16")
        return str(dest)
    except Exception:
        log.warning("PyAV-Dekodierung fehlgeschlagen, versuche soundfile", exc_info=True)
    try:
        arr, srr = sf.read(str(raw))
        sf.write(str(dest), arr, srr, subtype="PCM_16")
        return str(dest)
    except Exception:
        return str(raw)   # Fallback: Chatterbox versucht selbst zu laden


@app.post("/reference")
async def set_reference(audio: UploadFile = File(...)) -> dict:
    """Klon-Referenz hochladen (z.B. ElevenLabs-Erzeugnis) → Chatterbox nutzt sie als Stimme."""
    data = await audio.read()
    if not data:
        raise HTTPException(400, "Leeres Audio.")
    path = _save_reference_as_wav(data, audio.filename or "ref.wav")
    ACTIVE_REF.write_text(path, encoding="utf-8")
    return {"ok": True, "path": path, "bytes": len(data)}


@app.get("/reference")
def get_reference() -> dict:
    p = active_ref()
    return {"active": bool(p), "path": p}


@app.delete("/reference")
def clear_reference() -> dict:
    try:
        ACTIVE_REF.unlink(missing_ok=True)
    except OSError:
        pass
    return {"ok": True}


if __name__ == "__main__":
    import uvicorn

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