Hitonabi
aa62c98247
Sidecar: /tts + /voices + /health um Engine `elevenlabs` erweitert (Flash v2.5, Key zur Laufzeit aus env ODER ~/.hermes/.env → Nachtragen ohne Deploy). Default bleibt Piper. Frontend: ElevenLabs im Stimm-Picker (fest sichtbar) inkl. „Key fehlt"-Hinweis + Quota-Note. Chatterbox-DE war zu akzentig. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
298 lines
12 KiB
Python
298 lines
12 KiB
Python
#!/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"
|
|
|
|
|
|
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 []
|
|
|
|
|
|
# =================================================================================
|
|
# 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", "elevenlabs"],
|
|
"elevenlabs_key": bool(_eleven_key()),
|
|
"piper_voices": [v["id"] for v in piper_list()]}
|
|
|
|
|
|
@app.get("/voices")
|
|
def voices() -> dict:
|
|
return {"voices": 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 == "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)
|