Hitonabi/mission-control-v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
dd4c4c8c0fdc108a23342b3c24ed70545b5b157a
mission-control-v2/voice_service/app.py
T
Hitonabi dd4c4c8c0f Fix: EL-Retry bei 429 (Free-Tier-Concurrency) + index.html no-cache (immer frisches Bundle) 
5 parallele EL-Calls -> 1x 502 (Concurrency-Limit). elevenlabs_tts retryt jetzt 429/5xx mit Backoff.
index.html wird nicht mehr gecacht -> nach Deploy kein altes Bundle mehr.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-28 02:02:50 +02:00

447 lines
18 KiB
Python
Raw Blame History

#!/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", "")
# 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).
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 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: # noqa: BLE001
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}).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())
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")
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)
Reference in New Issue View Git Blame Copy Permalink