NexusOS/synapse/stt.py

"""Local speech-to-text via faster-whisper.

Deliberately torch-free: faster-whisper runs on CTranslate2 (CPU, int8), so it
fits the "no ML stack in the venv" rule that keeps Promethean small. Replaces
the browser's Web Speech API, which in Chrome ships audio to Google — the whole
point is to keep dictation on-device.

Model size via NEXUS_STT_MODEL (default "base"); downloaded and cached on first
use. If faster-whisper isn't installed, `available()` is False and the backend
reports it so the UI can fall back.
"""
from __future__ import annotations

import base64
import logging
import os
import tempfile

_log = logging.getLogger("nexus.stt")
_MODEL = None
_MODEL_SIZE = os.getenv("NEXUS_STT_MODEL", "base")


def available() -> bool:
    try:
        import faster_whisper  # noqa: F401
        return True
    except Exception:
        return False


def _get_model():
    global _MODEL
    if _MODEL is None:
        from faster_whisper import WhisperModel
        _log.info("loading whisper model %s (cpu/int8)", _MODEL_SIZE)
        _MODEL = WhisperModel(_MODEL_SIZE, device="cpu", compute_type="int8")
    return _MODEL


def transcribe_b64(audio_b64: str) -> str:
    """Transcribe a base64 audio blob (any container PyAV can decode — the
    browser's MediaRecorder produces webm/opus)."""
    raw = base64.b64decode(audio_b64)
    with tempfile.NamedTemporaryFile(suffix=".webm", delete=False) as f:
        f.write(raw)
        path = f.name
    try:
        segments, _info = _get_model().transcribe(path, vad_filter=True)
        return " ".join(s.text for s in segments).strip()
    finally:
        try:
            os.unlink(path)
        except OSError:
            pass