dschlueter
53b0dd4055
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>
289 lines
10 KiB
Python
Executable file
289 lines
10 KiB
Python
Executable file
#!/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()
|