File size: 2,820 Bytes
a695f69 cc3764e a695f69 cc3764e a695f69 cc3764e a695f69 cc3764e a695f69 cc3764e a695f69 cc3764e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """
AnalysisController — processes a recorded audio file.
Loads Whisper as a lazy pipeline and returns a transcript.
If the model dependencies are unavailable, it returns a helpful error.
"""
from __future__ import annotations
from pathlib import Path
from models.session_models import AnalysisResult
MIN_SESSION_SEC = 5 # mirrors JS constant
class AnalysisController:
def __init__(self) -> None:
self._pipeline = None
self._pipeline_error: str | None = None
def analyse(self, audio_path: str | None) -> AnalysisResult:
if not audio_path:
return AnalysisResult(error="No recording provided.")
audio_file = Path(audio_path)
if not audio_file.exists():
return AnalysisResult(error="Recorded audio file not found.")
pipe = self._get_pipeline()
if pipe is None:
return AnalysisResult(error=self._pipeline_error or "Speech analysis pipeline unavailable.")
try:
transcript = self.transcribe(audio_path)
except Exception as exc:
return AnalysisResult(error=f"Audio analysis failed: {exc}")
transcript = transcript.strip()
if not transcript:
return AnalysisResult(error="No transcript produced from the audio.")
return AnalysisResult(
transcript=transcript,
pronunciation=None,
fluency=None,
)
def _get_pipeline(self):
if self._pipeline is not None:
return self._pipeline
try:
import torch
from transformers import pipeline
except Exception as exc:
self._pipeline_error = (
"AI analysis dependencies are missing. "
"Install `transformers` and `torch` to enable speech recognition. "
f"({exc})"
)
return None
device = 0 if torch.cuda.is_available() else -1
try:
self._pipeline = pipeline(
"automatic-speech-recognition",
model="openai/whisper-tiny",
chunk_length_s=30,
device=device,
)
return self._pipeline
except Exception as exc:
self._pipeline_error = f"Failed to load ASR model: {exc}"
return None
def transcribe(self, audio_path: str) -> str:
pipe = self._get_pipeline()
if pipe is None:
raise RuntimeError(self._pipeline_error or "Speech recognition pipeline is unavailable.")
result = pipe(audio_path)
if isinstance(result, dict):
return result.get("text", "")
if isinstance(result, list):
return " ".join(str(item.get("text", "")) for item in result if isinstance(item, dict))
return str(result)
|