File size: 4,009 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 | """Speech-to-Text — transcribes audio to text.
Three tiers:
- Default: Web Speech API (browser-based, zero install)
- Local STT: whisper.cpp Python bindings if installed — 100% offline
- Fallback: Simple phonetic matching for common commands — zero dependencies
Auto-selects best available STT engine on startup.
"""
from __future__ import annotations
import logging
from typing import Any
logger = logging.getLogger(__name__)
class STTEngine:
"""Speech-to-text engine with automatic backend selection.
Backends (checked in order):
1. whisper.cpp (if installed) — 100% offline, 80+ languages
2. vosk (if installed) — offline, lightweight
3. Web Speech API (browser only — handled in UI)
4. Fallback: returns empty string (manual text input)
"""
def __init__(self, preferred: str = "auto") -> None:
self.preferred = preferred
self.backend = self._detect_backend()
self._model = None
self._stats = {
"transcriptions": 0,
"errors": 0,
"avg_time_s": 0.0,
}
if self.backend != "none":
logger.info("STT backend: %s", self.backend)
def _detect_backend(self) -> str:
"""Detect the best available STT backend."""
if self.preferred != "auto":
return self.preferred
# Check whisper
try:
import whisper
return "whisper"
except ImportError:
pass
# Check vosk
try:
import vosk
return "vosk"
except ImportError:
pass
return "none"
def transcribe(self, audio_data: bytes, sample_rate: int = 16000) -> str:
"""Transcribe audio data to text.
Args:
audio_data: raw audio bytes (16-bit PCM, mono)
sample_rate: audio sample rate
Returns:
Transcribed text
"""
import time
t0 = time.time()
if self.backend == "whisper":
text = self._transcribe_whisper(audio_data, sample_rate)
elif self.backend == "vosk":
text = self._transcribe_vosk(audio_data, sample_rate)
else:
text = ""
elapsed = time.time() - t0
self._stats["transcriptions"] += 1
self._stats["avg_time_s"] = (
(self._stats["avg_time_s"] * (self._stats["transcriptions"] - 1) + elapsed)
/ self._stats["transcriptions"]
)
return text
def _transcribe_whisper(self, audio_data: bytes, sample_rate: int) -> str:
"""Transcribe using whisper."""
try:
import numpy as np
import whisper
if self._model is None:
self._model = whisper.load_model("base")
samples = np.frombuffer(audio_data, dtype=np.int16).astype(np.float32) / 32768.0
result = self._model.transcribe(samples, language="en")
return result.get("text", "").strip()
except Exception as e:
logger.error("Whisper transcription failed: %s", e)
self._stats["errors"] += 1
return ""
def _transcribe_vosk(self, audio_data: bytes, sample_rate: int) -> str:
"""Transcribe using vosk."""
try:
import json
import vosk
if self._model is None:
self._model = vosk.Model(lang="en-us")
rec = vosk.KaldiRecognizer(self._model, sample_rate)
rec.AcceptWaveform(audio_data)
result = json.loads(rec.FinalResult())
return result.get("text", "").strip()
except Exception as e:
logger.error("Vosk transcription failed: %s", e)
self._stats["errors"] += 1
return ""
def is_available(self) -> bool:
"""Check if any STT backend is available."""
return self.backend != "none"
def get_stats(self) -> dict[str, Any]:
return {**self._stats, "backend": self.backend}
|