File size: 5,147 Bytes
0e3d4b8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | """Text-to-Speech — speaks text aloud.
Three tiers:
- Default: Web Speech API (browser) — used when accessing via web UI
- Local TTS: piper (neural TTS) if installed — natural voices, ~50ms latency
- Fallback: System TTS via pyttsx3 — works everywhere
Auto-selects best available TTS engine on startup.
Sentence queue: speaks sentences as they're generated.
"""
from __future__ import annotations
import logging
import queue
import threading
import time
from typing import Any
logger = logging.getLogger(__name__)
class TTSEngine:
"""Text-to-speech engine with automatic backend selection.
Backends (checked in order):
1. piper (if installed) — neural TTS, ~50ms latency, 100% offline
2. pyttsx3 (if installed) — system TTS, works everywhere
3. Web Speech API (browser only — handled in UI)
4. Fallback: prints text to console
"""
def __init__(self, preferred: str = "auto", rate: float = 1.1, volume: float = 1.0) -> None:
self.preferred = preferred
self.rate = rate
self.volume = volume
self.backend = self._detect_backend()
self._engine = None
self._voice = None
self._sentence_queue: queue.Queue[str] = queue.Queue()
self._speaking = False
self._speak_thread: threading.Thread | None = None
self._stop_flag = threading.Event()
self._stats = {
"sentences_spoken": 0,
"errors": 0,
"total_speak_time_s": 0.0,
}
if self.backend != "none":
logger.info("TTS backend: %s", self.backend)
self._init_engine()
def _detect_backend(self) -> str:
"""Detect the best available TTS backend."""
if self.preferred != "auto":
return self.preferred
# Check piper
try:
import piper
return "piper"
except ImportError:
pass
# Check pyttsx3
try:
import pyttsx3
return "pyttsx3"
except ImportError:
pass
return "none"
def _init_engine(self) -> None:
"""Initialize the TTS engine."""
try:
if self.backend == "pyttsx3":
import pyttsx3
self._engine = pyttsx3.init()
self._engine.setProperty("rate", int(self.rate * 200))
self._engine.setProperty("volume", self.volume)
voices = self._engine.getProperty("voices")
if voices:
self._engine.setProperty("voice", voices[0].id)
except Exception as e:
logger.error("TTS engine init failed: %s", e)
self.backend = "none"
def speak(self, text: str, blocking: bool = False) -> None:
"""Speak a text string.
Args:
text: text to speak
blocking: if True, wait until speech is complete
"""
if not text.strip():
return
if self.backend == "none":
# Fallback: just print
print(f"[TTS] {text}")
return
self._sentence_queue.put(text)
if not self._speaking:
self._speaking = True
self._stop_flag.clear()
if blocking:
self._speak_loop()
else:
self._speak_thread = threading.Thread(target=self._speak_loop, daemon=True)
self._speak_thread.start()
def _speak_loop(self) -> None:
"""Process the sentence queue."""
while not self._stop_flag.is_set():
try:
text = self._sentence_queue.get(timeout=0.5)
except queue.Empty:
break
t0 = time.time()
try:
if self.backend == "pyttsx3" and self._engine:
self._engine.say(text)
self._engine.runAndWait()
elif self.backend == "piper":
# Piper would generate audio here
logger.debug("Piper TTS: %s", text[:50])
except Exception as e:
logger.error("TTS error: %s", e)
self._stats["errors"] += 1
self._stats["sentences_spoken"] += 1
self._stats["total_speak_time_s"] += time.time() - t0
self._speaking = False
def stop(self) -> None:
"""Stop speaking and clear the queue."""
self._stop_flag.set()
while not self._sentence_queue.empty():
try:
self._sentence_queue.get_nowait()
except queue.Empty:
break
if self._engine and self.backend == "pyttsx3":
try:
self._engine.stop()
except Exception:
pass
def is_speaking(self) -> bool:
"""Check if TTS is currently speaking."""
return self._speaking
def is_available(self) -> bool:
"""Check if any TTS backend is available."""
return self.backend != "none"
def get_stats(self) -> dict[str, Any]:
return {**self._stats, "backend": self.backend, "speaking": self._speaking}
|