Hitonabi
403961da51
- onnx-asr (int8, CPU) als neue STT-Engine im Voice-Sidecar; whisper bleibt lazy Fallback - Engine per VOICE_STT_ENGINE + pro Request (Form-Feld engine) waehlbar - A/B auf der Box (10,8s DE-Audio): parakeet 0,45s vs whisper-medium 2,44s, Transkript identisch gut - Live deployt + verifiziert (auch ueber :9001-Proxy: 0,46s) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
537 lines
22 KiB
Python
537 lines
22 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
|
||
# 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)
|
||
|
||
|
||
# =================================================================================
|
||
# 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,
|
||
"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: # 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:
|
||
# 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)
|
||
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("/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)
|