| """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 |
|
|
| |
| try: |
| import whisper |
| return "whisper" |
| except ImportError: |
| pass |
|
|
| |
| 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} |
|
|
