Adolfo Reyna
199f762616
- PipeWire/pw-record backend (no PortAudio needed) - tested mic live - VAD: Energy (immediate) + Silero (torch CPU) switchable - Wakeword: openWakeWord TFLite hey_jarvis 0.4 - STT: faster-whisper base int8 CPU - model loads OK on i3 - SpeakerID: ECAPA embeddings + fallback stats, enrollment store - Vision: OpenCV Haar + face_recognition + webcam cam0 640x480 verified - LLM: ollama/qwen2.5:3b, gemini_cli, hermes_api, echo strategies - TTS: Piper + espeak + echo fallbacks - Core pipeline: IDLE -> WAKE -> RECORDING -> STT -> SpeakerID/FaceID fusion -> LLM -> TTS -> followup window - Config via jarvis.yaml, soul.md persona - CLI tools: run, devices, enroll-speaker, enroll-face, test-vad, test-mic - Systemd user service ready - Tested on iMac 2010 i3 550 15GB Ubuntu 26.04 CPU-only
96 lines
3.6 KiB
Python
96 lines
3.6 KiB
Python
import logging, os, subprocess, tempfile, pathlib
|
|
from .base import TTSEngine
|
|
log = logging.getLogger(__name__)
|
|
|
|
class PiperTTS(TTSEngine):
|
|
def __init__(self, voice="en_US-lessac-medium", models_dir="models", volume=0.9):
|
|
self.voice = voice
|
|
self.models_dir = pathlib.Path(models_dir)
|
|
self.models_dir.mkdir(parents=True, exist_ok=True)
|
|
self.volume = volume
|
|
self.voice_path = None
|
|
self.config_path = None
|
|
self._ensure_model()
|
|
|
|
def _ensure_model(self):
|
|
pattern = self.models_dir / f"{self.voice}.onnx"
|
|
if pattern.exists():
|
|
self.voice_path = pattern
|
|
self.config_path = pathlib.Path(str(pattern) + ".json")
|
|
log.info(f"Piper voice found: {self.voice_path}")
|
|
return
|
|
for p in self.models_dir.rglob("*.onnx"):
|
|
if self.voice in p.name:
|
|
self.voice_path = p
|
|
self.config_path = pathlib.Path(str(p) + ".json")
|
|
log.info(f"Piper voice found: {p}")
|
|
return
|
|
log.warning(f"Piper voice {self.voice} not in {self.models_dir}")
|
|
|
|
def _download_voice(self):
|
|
try:
|
|
log.info(f"Downloading Piper voice {self.voice}")
|
|
result = subprocess.run(["python3", "-m", "piper.download_voices", self.voice], cwd=str(self.models_dir), capture_output=True, text=True, timeout=60)
|
|
log.info(result.stdout[:500])
|
|
self._ensure_model()
|
|
except Exception as e:
|
|
log.warning(f"Piper download failed: {e}")
|
|
|
|
def speak(self, text: str):
|
|
if not text.strip():
|
|
return
|
|
clean = text.replace("*","").replace("#","").replace("`","")[:600]
|
|
if self.voice_path is None:
|
|
self._download_voice()
|
|
if self.voice_path and self.voice_path.exists():
|
|
try:
|
|
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tf:
|
|
wav_path = tf.name
|
|
cmd = ["piper", "--model", str(self.voice_path), "--output_file", wav_path]
|
|
proc = subprocess.run(cmd, input=clean.encode(), capture_output=True, timeout=15)
|
|
if proc.returncode==0 and os.path.exists(wav_path):
|
|
self._play_wav(wav_path)
|
|
try:
|
|
os.remove(wav_path)
|
|
except:
|
|
pass
|
|
return
|
|
else:
|
|
log.warning(f"Piper failed: {proc.stderr[:300]}")
|
|
except Exception as e:
|
|
log.warning(f"Piper speak error: {e}")
|
|
self._fallback_speak(clean)
|
|
|
|
def _play_wav(self, wav_path):
|
|
try:
|
|
import soundfile as sf
|
|
import sounddevice as sd
|
|
data, sr = sf.read(wav_path)
|
|
data = data * self.volume
|
|
sd.play(data, sr)
|
|
sd.wait()
|
|
except Exception as e:
|
|
log.debug(f"sd playback fail {e}, trying aplay")
|
|
try:
|
|
subprocess.run(["aplay", wav_path], capture_output=True, timeout=10)
|
|
except Exception as e2:
|
|
log.warning(f"aplay fail: {e2}")
|
|
|
|
def _fallback_speak(self, text):
|
|
try:
|
|
subprocess.run(["espeak", text], capture_output=True, timeout=10)
|
|
return
|
|
except:
|
|
pass
|
|
try:
|
|
subprocess.run(["spd-say", text], capture_output=True, timeout=10)
|
|
return
|
|
except:
|
|
pass
|
|
log.info(f"[TTS fallback] {text}")
|
|
print(f"Jarvis says: {text}")
|
|
|
|
class EchoTTS(TTSEngine):
|
|
def speak(self, text: str):
|
|
print(f"[Jarvis TTS]: {text}")
|