| """ |
| 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 |
|
|
|
|
| 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) |
|
|
