mirror of
https://github.com/open-jarvis/OpenJarvis.git
synced 2026-07-31 03:12:16 +00:00
feat(speech): implement FasterWhisperBackend (local STT)
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
002c63a205
commit
cb28dd2ba9
@@ -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"]
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user