#!/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 logging import os import tempfile import wave 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 VOICES_DIR = Path(os.environ.get("VOICE_PIPER_DIR", str(Path(__file__).resolve().parent / "voices"))) PIPER_DEFAULT = os.environ.get("VOICE_PIPER_DEFAULT", "de_DE-thorsten-medium") # 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", "") # ================================================================================= # 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 (lazy, je Stimme gecacht) # ================================================================================= _piper: dict = {} def piper_voice(name: str): if name not in _piper: from piper import PiperVoice onnx = VOICES_DIR / f"{name}.onnx" if not onnx.exists(): raise HTTPException(404, f"Piper-Stimme '{name}' nicht gefunden ({onnx}).") log.info("Lade Piper-Stimme '%s' …", name) _piper[name] = PiperVoice.load(str(onnx)) return _piper[name] def piper_tts(text: str, voice: str) -> bytes: v = piper_voice(voice or PIPER_DEFAULT) buf = io.BytesIO() with wave.open(buf, "wb") as wav: # piper-tts 1.2.x: synthesize(text, wave_file) schreibt einen kompletten WAV-Stream. v.synthesize(text, wav) return buf.getvalue() 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 # ================================================================================= # 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": ["piper", "chatterbox"], "piper_voices": [v["id"] for v in piper_list()]} @app.get("/voices") def voices() -> dict: return {"voices": piper_list() + chatterbox_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 == "chatterbox": audio = chatterbox_tts(text, body.language, body.ref_path) else: audio = piper_tts(text, body.voice) return Response(content=audio, media_type="audio/wav") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=PORT)