feat: initial voice assistant implementation
5-state machine (LAUSCH/AUFNAHME/GENERATOR/VORLESE/DIALOG) with: - Wake-word detection via openwakeword (hey_jarvis) - Speech-to-text via faster-whisper large-v3 (GPU) - LLM streaming via OpenAI-compatible backends (llama.cpp/ollama/openai) - TTS via piper with automatic language detection and voice selection - Selectable mic input (ReSpeaker, MOTU M2, camera, bluetooth, index) - Selectable audio output sink - Control words during playback (Stopp/Pause/Weiter/Noch einmal/…) - "Stopp" abort in all active states → LAUSCH Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
commit
53b0dd4055
5 changed files with 1442 additions and 0 deletions
289
speak.py
Executable file
289
speak.py
Executable file
|
|
@ -0,0 +1,289 @@
|
|||
#!/usr/bin/env python3
|
||||
"""
|
||||
TTS-Ausgabe via Piper — Text aus Argument oder stdin, Ausgabe auf wählbaren Kanal.
|
||||
|
||||
Verwendung:
|
||||
echo "Hallo Welt" | python3 speak.py
|
||||
python3 speak.py --text "Hallo Welt"
|
||||
python3 speak.py --text "Hallo" --out respeaker
|
||||
python3 speak.py --text "Hello" --voice en_US-ryan-high --out hdmi
|
||||
python3 speak.py --list
|
||||
|
||||
Ausgabe-Kurznamen (--out):
|
||||
default PipeWire-Default-Sink (aktuell: MOTU M2)
|
||||
respeaker ReSpeaker XVF3800 3,5mm-Klinke
|
||||
hdmi HDA NVidia HDMI
|
||||
bluetooth erstes verbundenes Bluetooth-Gerät (dynamisch)
|
||||
<sink-name> vollständiger PipeWire-Sink-Name
|
||||
|
||||
Sprechbar auch als Bibliothek:
|
||||
from speak import speak
|
||||
speak("Hallo Welt", out="respeaker")
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from piper.voice import PiperVoice
|
||||
|
||||
try:
|
||||
from langdetect import detect_langs as _ld_detect_langs, LangDetectException
|
||||
from langdetect.detector_factory import DetectorFactory as _DetectorFactory
|
||||
_DetectorFactory.seed = 42 # deterministisch
|
||||
_LANGDETECT_OK = True
|
||||
except ImportError:
|
||||
_LANGDETECT_OK = False
|
||||
|
||||
VOICES_DIR = Path.home() / ".local/share/piper/voices"
|
||||
DEFAULT_VOICE = "de_DE-thorsten-high"
|
||||
|
||||
# Beste verfügbare Piper-Stimme pro Sprache (ISO 639-1 → voice_name)
|
||||
LANG_VOICES: dict[str, str] = {
|
||||
"de": "de_DE-thorsten-high",
|
||||
"en": "en_US-ryan-high", # überschreibbar via set_en_variant()
|
||||
"fr": "fr_FR-upmc-medium",
|
||||
"es": "es_ES-davefx-medium",
|
||||
"it": "it_IT-paola-medium",
|
||||
"nl": "nl_NL-mls-medium",
|
||||
"pt": "pt_BR-faber-medium",
|
||||
"ru": "ru_RU-ruslan-medium",
|
||||
"pl": "pl_PL-bass-high",
|
||||
"uk": "uk_UA-lada-x_low",
|
||||
"cs": "cs_CZ-jirka-medium",
|
||||
}
|
||||
|
||||
EN_VARIANT_VOICES: dict[str, str] = {
|
||||
"us": "en_US-ryan-high",
|
||||
"gb": "en_GB-alan-medium",
|
||||
}
|
||||
|
||||
_voice_cache: dict[str, "PiperVoice"] = {}
|
||||
|
||||
# Feste Sink-Kurznamen
|
||||
SINK_ALIASES: dict[str, str] = {
|
||||
"respeaker": "alsa_output.usb-Seeed_Studio_reSpeaker_XVF3800_4-Mic_Array_114993700261100055-00.analog-stereo",
|
||||
"hdmi": "alsa_output.pci-0000_0b_00.1.hdmi-stereo",
|
||||
"motu": "alsa_output.usb-MOTU_M2_M20000046918-00.analog-stereo",
|
||||
}
|
||||
|
||||
|
||||
def list_sinks() -> dict[str, str]:
|
||||
"""Gibt {sink_name: beschreibung} für alle aktiven PipeWire-Sinks zurück."""
|
||||
result = subprocess.run(
|
||||
["pactl", "list", "sinks"],
|
||||
capture_output=True, text=True
|
||||
)
|
||||
sinks: dict[str, str] = {}
|
||||
name = ""
|
||||
for line in result.stdout.splitlines():
|
||||
line = line.strip()
|
||||
if line.startswith("Name:"):
|
||||
name = line.split(None, 1)[1]
|
||||
elif line.startswith("Beschreibung:") or line.startswith("Description:"):
|
||||
desc = line.split(None, 1)[1]
|
||||
if name:
|
||||
sinks[name] = desc
|
||||
return sinks
|
||||
|
||||
|
||||
def resolve_sink(out: str) -> str | None:
|
||||
"""Löst einen Kurznamen oder 'bluetooth' in einen echten Sink-Namen auf.
|
||||
Gibt None zurück → PipeWire-Default wird verwendet."""
|
||||
if out == "default":
|
||||
return None
|
||||
if out in SINK_ALIASES:
|
||||
return SINK_ALIASES[out]
|
||||
if out == "bluetooth":
|
||||
for name in list_sinks():
|
||||
if "bluez_output" in name:
|
||||
return name
|
||||
print("Kein Bluetooth-Gerät verbunden.", file=sys.stderr)
|
||||
return None
|
||||
# Vollständiger Sink-Name oder unbekannter Alias → direkt weitergeben
|
||||
return out
|
||||
|
||||
|
||||
def load_voice(voice_name: str) -> PiperVoice:
|
||||
onnx = VOICES_DIR / f"{voice_name}.onnx"
|
||||
if not onnx.exists():
|
||||
from piper.download_voices import download_voice
|
||||
VOICES_DIR.mkdir(parents=True, exist_ok=True)
|
||||
last_exc: Exception | None = None
|
||||
for attempt in range(1, 4):
|
||||
print(f"Stimme '{voice_name}' nicht lokal — lade herunter (Versuch {attempt}/3) …",
|
||||
file=sys.stderr, flush=True)
|
||||
try:
|
||||
download_voice(voice_name, VOICES_DIR)
|
||||
last_exc = None
|
||||
break
|
||||
except Exception as exc:
|
||||
last_exc = exc
|
||||
if attempt < 3:
|
||||
print(f"[Download-Fehler: {exc}] — wiederhole in {2 ** attempt}s …",
|
||||
file=sys.stderr, flush=True)
|
||||
time.sleep(2 ** attempt) # 2s, 4s
|
||||
if last_exc is not None or not onnx.exists():
|
||||
raise RuntimeError(
|
||||
f"Download von '{voice_name}' nach 3 Versuchen fehlgeschlagen"
|
||||
+ (f": {last_exc}" if last_exc else "")
|
||||
)
|
||||
print(f"Stimme '{voice_name}' gespeichert in {VOICES_DIR}.", file=sys.stderr)
|
||||
return PiperVoice.load(str(onnx))
|
||||
|
||||
|
||||
def get_voice(voice_name: str = DEFAULT_VOICE) -> PiperVoice:
|
||||
"""Gibt eine gecachte PiperVoice zurück — lädt nur beim ersten Aufruf."""
|
||||
if voice_name not in _voice_cache:
|
||||
_voice_cache[voice_name] = load_voice(voice_name)
|
||||
return _voice_cache[voice_name]
|
||||
|
||||
|
||||
def detect_lang(text: str) -> str | None:
|
||||
"""Erkennt die dominante Sprache. Gibt ISO-639-1-Code zurück oder None.
|
||||
|
||||
Akronyme (ALL-CAPS-Wörter wie GPT, KI, URL) und URLs werden vor der
|
||||
Erkennung entfernt, damit sie die Spracherkennung nicht verfälschen.
|
||||
Mindestkonfidenzschwelle: 0.85.
|
||||
"""
|
||||
if not _LANGDETECT_OK:
|
||||
return None
|
||||
# Akronyme (≥2 Großbuchstaben) und URLs entfernen
|
||||
cleaned = re.sub(r'\b[A-Z]{2,}\b', ' ', text)
|
||||
cleaned = re.sub(r'https?://\S+', ' ', cleaned)
|
||||
cleaned = re.sub(r'\s+', ' ', cleaned).strip()
|
||||
if len(cleaned) < 20:
|
||||
return None
|
||||
try:
|
||||
langs = _ld_detect_langs(cleaned)
|
||||
if langs and langs[0].prob >= 0.85:
|
||||
return langs[0].lang
|
||||
except LangDetectException:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def voice_for_lang(lang_code: str | None, fallback: str = DEFAULT_VOICE) -> PiperVoice:
|
||||
"""Gibt die passende (gecachte) PiperVoice für einen Sprachcode zurück.
|
||||
Lädt und cached die Stimme beim ersten Aufruf (inkl. Auto-Download)."""
|
||||
voice_name = LANG_VOICES.get(lang_code or "", fallback) if lang_code else fallback
|
||||
return get_voice(voice_name)
|
||||
|
||||
|
||||
def set_en_variant(variant: str) -> None:
|
||||
"""Wählt US- oder GB-Englisch für die automatische Spracherkennung.
|
||||
Muss vor dem ersten get_voice()-Aufruf gesetzt werden."""
|
||||
LANG_VOICES["en"] = EN_VARIANT_VOICES.get(variant, EN_VARIANT_VOICES["us"])
|
||||
|
||||
|
||||
def synth_bytes(text: str, voice: PiperVoice) -> bytes:
|
||||
"""Synthetisiert text vollständig in einen Byte-Puffer (ohne Subprocess)."""
|
||||
return b"".join(c.audio_int16_bytes for c in voice.synthesize(text))
|
||||
|
||||
|
||||
def speak(text: str, voice_name: str = DEFAULT_VOICE, out: str = "default") -> None:
|
||||
"""Synthetisiert text und gibt ihn auf dem gewünschten Ausgang aus."""
|
||||
voice = load_voice(voice_name)
|
||||
sink = resolve_sink(out)
|
||||
|
||||
# paplay liest raw-PCM von stdin
|
||||
cmd = ["paplay", "--raw",
|
||||
f"--rate={voice.config.sample_rate}",
|
||||
"--channels=1",
|
||||
"--format=s16le"]
|
||||
if sink:
|
||||
cmd += [f"--device={sink}"]
|
||||
|
||||
proc = subprocess.Popen(cmd, stdin=subprocess.PIPE)
|
||||
for chunk in voice.synthesize(text):
|
||||
proc.stdin.write(chunk.audio_int16_bytes)
|
||||
proc.stdin.close()
|
||||
proc.wait()
|
||||
|
||||
|
||||
def print_sink_list() -> None:
|
||||
sinks = list_sinks()
|
||||
default = subprocess.run(
|
||||
["pactl", "get-default-sink"], capture_output=True, text=True
|
||||
).stdout.strip()
|
||||
|
||||
print("Verfügbare Ausgänge:\n")
|
||||
print(f" {'Kurzname':<12} {'Sink-Name':<65} Beschreibung")
|
||||
print(f" {'-'*12} {'-'*65} {'-'*30}")
|
||||
|
||||
# Kurznamen zuerst
|
||||
for alias, sink_name in sorted(SINK_ALIASES.items()):
|
||||
desc = sinks.get(sink_name, "–")
|
||||
marker = " *" if sink_name == default else ""
|
||||
print(f" {alias:<12} {sink_name:<65} {desc}{marker}")
|
||||
|
||||
print(f" {'default':<12} {'(PipeWire-Default-Sink)':<65} → aktuell: {sinks.get(default, default)}")
|
||||
print(f" {'bluetooth':<12} {'(dynamisch, bluez_output.*)':<65} erstes verbundenes BT-Gerät")
|
||||
|
||||
# Weitere Sinks ohne Alias
|
||||
aliased = set(SINK_ALIASES.values())
|
||||
others = [(n, d) for n, d in sinks.items() if n not in aliased]
|
||||
if others:
|
||||
print("\n Weitere (kein Kurzname, mit vollem Namen ansprechen):")
|
||||
for name, desc in others:
|
||||
marker = " *" if name == default else ""
|
||||
print(f" {'':12} {name:<65} {desc}{marker}")
|
||||
print("\n * = aktueller Default")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(description="Piper TTS mit wählbarem Ausgang")
|
||||
parser.add_argument("--text", "-t", help="Auszugebender Text (sonst: stdin)")
|
||||
parser.add_argument("--voice", "-v", default=DEFAULT_VOICE,
|
||||
help=f"Piper-Stimme (default: {DEFAULT_VOICE})")
|
||||
parser.add_argument("--voice-lang", default="de",
|
||||
help="Sprachfilter für --list-voices, z.B. 'de', 'en', '' für alle")
|
||||
parser.add_argument("--out", "-o", default="default",
|
||||
help="Ausgabe: default | respeaker | hdmi | motu | bluetooth | <sink-name>")
|
||||
parser.add_argument("--list", "-l", action="store_true",
|
||||
help="Verfügbare Ausgänge anzeigen und beenden")
|
||||
parser.add_argument("--list-voices", "-L", action="store_true",
|
||||
help="Alle herunterladbaren Stimmen anzeigen (gefiltert nach --voice-lang)")
|
||||
args = parser.parse_args()
|
||||
|
||||
if args.list:
|
||||
print_sink_list()
|
||||
return
|
||||
|
||||
if args.list_voices:
|
||||
import json
|
||||
from urllib.request import urlopen
|
||||
from piper.download_voices import VOICES_JSON
|
||||
with urlopen(VOICES_JSON) as resp:
|
||||
voices = json.load(resp)
|
||||
lang = args.voice_lang
|
||||
local = {f.stem for f in VOICES_DIR.glob("*.onnx")}
|
||||
matches = {k: v for k, v in sorted(voices.items())
|
||||
if not lang or k.startswith(lang)}
|
||||
print(f"{'Stimme':<40} {'Qualität':<8} Lokal")
|
||||
print(f"{'-'*40} {'-'*8} -----")
|
||||
for name, info in matches.items():
|
||||
q = info.get("quality", "?")
|
||||
marker = "✓" if name in local else ""
|
||||
print(f"{name:<40} {q:<8} {marker}")
|
||||
print(f"\n{len(matches)} Stimmen. Lokal vorhanden: {len([n for n in matches if n in local])}")
|
||||
return
|
||||
|
||||
if args.text:
|
||||
text = args.text
|
||||
elif not sys.stdin.isatty():
|
||||
text = sys.stdin.read().strip()
|
||||
else:
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
if not text:
|
||||
sys.exit(0)
|
||||
|
||||
speak(text, voice_name=args.voice, out=args.out)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue