From cb28dd2ba9624e4ab019ac26db8f80c7e33266a8 Mon Sep 17 00:00:00 2001 From: Jon Saad-Falcon Date: Tue, 3 Mar 2026 06:18:57 +0000 Subject: [PATCH] feat(speech): implement FasterWhisperBackend (local STT) Co-Authored-By: Claude Opus 4.6 --- src/openjarvis/speech/faster_whisper.py | 100 ++++++++++++++++++++++++ tests/speech/test_faster_whisper.py | 69 ++++++++++++++++ 2 files changed, 169 insertions(+) create mode 100644 src/openjarvis/speech/faster_whisper.py create mode 100644 tests/speech/test_faster_whisper.py diff --git a/src/openjarvis/speech/faster_whisper.py b/src/openjarvis/speech/faster_whisper.py new file mode 100644 index 00000000..e3d02843 --- /dev/null +++ b/src/openjarvis/speech/faster_whisper.py @@ -0,0 +1,100 @@ +"""Faster-Whisper speech-to-text backend (local, CTranslate2-based).""" + +from __future__ import annotations + +import tempfile +from typing import List, Optional + +from openjarvis.core.registry import SpeechRegistry +from openjarvis.speech._stubs import Segment, SpeechBackend, TranscriptionResult + +try: + from faster_whisper import WhisperModel +except ImportError: + WhisperModel = None # type: ignore[assignment, misc] + + +@SpeechRegistry.register("faster-whisper") +class FasterWhisperBackend(SpeechBackend): + """Local speech-to-text using Faster-Whisper (CTranslate2).""" + + backend_id = "faster-whisper" + + def __init__( + self, + model_size: str = "base", + device: str = "auto", + compute_type: str = "float16", + ) -> None: + self._model_size = model_size + self._device = device + self._compute_type = compute_type + self._model: Optional[WhisperModel] = None + + def _ensure_model(self) -> WhisperModel: + """Lazy-load the Whisper model on first use.""" + if self._model is None: + if WhisperModel is None: + raise ImportError( + "faster-whisper is not installed. " + "Install with: pip install 'openjarvis[speech]'" + ) + self._model = WhisperModel( + self._model_size, + device=self._device, + compute_type=self._compute_type, + ) + return self._model + + def transcribe( + self, + audio: bytes, + *, + format: str = "wav", + language: Optional[str] = None, + ) -> TranscriptionResult: + """Transcribe audio bytes using Faster-Whisper.""" + model = self._ensure_model() + + # Write audio to a temp file (faster-whisper needs a file path) + suffix = f".{format}" if not format.startswith(".") else format + with tempfile.NamedTemporaryFile(suffix=suffix, delete=True) as tmp: + tmp.write(audio) + tmp.flush() + + kwargs = {} + if language: + kwargs["language"] = language + + segments_iter, info = model.transcribe(tmp.name, **kwargs) + segments_list = list(segments_iter) + + # Build result + text = "".join(seg.text for seg in segments_list).strip() + segments = [ + Segment( + text=seg.text.strip(), + start=seg.start, + end=seg.end, + confidence=None, + ) + for seg in segments_list + ] + + return TranscriptionResult( + text=text, + language=getattr(info, "language", None), + confidence=getattr(info, "language_probability", None), + duration_seconds=getattr(info, "duration", 0.0), + segments=segments, + ) + + def health(self) -> bool: + """Check if model is loaded or loadable.""" + if self._model is not None: + return True + return WhisperModel is not None + + def supported_formats(self) -> List[str]: + """Supported audio formats (same as ffmpeg/Whisper).""" + return ["wav", "mp3", "m4a", "ogg", "flac", "webm"] diff --git a/tests/speech/test_faster_whisper.py b/tests/speech/test_faster_whisper.py new file mode 100644 index 00000000..7ad96fe8 --- /dev/null +++ b/tests/speech/test_faster_whisper.py @@ -0,0 +1,69 @@ +"""Tests for Faster-Whisper speech backend.""" + +from unittest.mock import MagicMock, patch + + +def test_faster_whisper_backend_registers(): + """Backend registers itself in SpeechRegistry.""" + import openjarvis.speech.faster_whisper # noqa: F401 + from openjarvis.core.registry import SpeechRegistry + + assert SpeechRegistry.contains("faster-whisper") + + +def test_faster_whisper_transcribe(): + """Transcribe returns a TranscriptionResult.""" + from openjarvis.speech._stubs import TranscriptionResult + + mock_model = MagicMock() + mock_segment = MagicMock() + mock_segment.text = " Hello world" + mock_segment.start = 0.0 + mock_segment.end = 1.2 + mock_segment.avg_logprob = -0.3 + + mock_info = MagicMock() + mock_info.language = "en" + mock_info.language_probability = 0.95 + mock_info.duration = 1.5 + + mock_model.transcribe.return_value = ([mock_segment], mock_info) + + with patch( + "openjarvis.speech.faster_whisper.WhisperModel", + return_value=mock_model, + ): + from openjarvis.speech.faster_whisper import FasterWhisperBackend + + backend = FasterWhisperBackend(model_size="base", device="cpu") + result = backend.transcribe(b"fake audio bytes") + + assert isinstance(result, TranscriptionResult) + assert result.text == "Hello world" + assert result.language == "en" + assert result.duration_seconds == 1.5 + + +def test_faster_whisper_health_no_model(): + """Health returns False before model is loaded.""" + with patch( + "openjarvis.speech.faster_whisper.WhisperModel", + new=None, + ): + from openjarvis.speech.faster_whisper import FasterWhisperBackend + + backend = FasterWhisperBackend.__new__(FasterWhisperBackend) + backend._model = None + assert backend.health() is False + + +def test_faster_whisper_supported_formats(): + """Backend supports standard audio formats.""" + with patch("openjarvis.speech.faster_whisper.WhisperModel"): + from openjarvis.speech.faster_whisper import FasterWhisperBackend + + backend = FasterWhisperBackend.__new__(FasterWhisperBackend) + formats = backend.supported_formats() + assert "wav" in formats + assert "mp3" in formats + assert "webm" in formats