| """Speech-to-text via faster-whisper.""" |
| from __future__ import annotations |
|
|
| import logging |
| import os |
| from typing import Tuple |
|
|
| from src.translator import translate |
|
|
| logger = logging.getLogger(__name__) |
|
|
| |
| _whisper_model = None |
|
|
|
|
| def _get_model(): |
| global _whisper_model |
| if _whisper_model is None: |
| from faster_whisper import WhisperModel |
| model_size = os.environ.get("RYUGAKU_WHISPER_MODEL", "small") |
| device = "cpu" if os.environ.get("RYUGAKU_FORCE_CPU") == "1" else "cuda" |
| logger.info("Loading faster-whisper model=%s device=%s", model_size, device) |
| _whisper_model = WhisperModel(model_size, device=device, compute_type="float16" if device == "cuda" else "int8") |
| return _whisper_model |
|
|
|
|
| def transcribe_and_translate(audio_path: str) -> Tuple[str, str, str]: |
| """Transcribe an audio file and translate the Japanese text. |
| |
| Returns (transcript, translation, status). |
| """ |
| if not audio_path or not os.path.exists(audio_path): |
| return "", "", "No audio file provided" |
|
|
| try: |
| model = _get_model() |
| segments, info = model.transcribe(audio_path, language="ja", beam_size=5) |
| text_parts = [segment.text for segment in segments] |
| transcript = " ".join(text_parts).strip() |
| if not transcript: |
| return "", "", "No speech detected" |
| translation, _ = translate(transcript) |
| return transcript, translation, "" |
| except Exception as e: |
| logger.exception("STT failed") |
| return "", "", f"STT error: {e}" |
|
|
