cb60172b51
lightning-whisper-mlx returns segments as [start, end, text] lists (not dicts) in some versions, which broke transcribe(). Normalize all three observed shapes: str output, list/tuple segments, dict segments. Also fall back to joining segment texts when top-level text is empty.
99 lines
4.0 KiB
Python
99 lines
4.0 KiB
Python
"""Abstraktion über mehrere Whisper-Implementierungen."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .config import settings
|
|
|
|
log = logging.getLogger("mac-worker.whisper")
|
|
|
|
|
|
class WhisperEngine:
|
|
def __init__(self) -> None:
|
|
self.engine = settings.whisper_engine.lower()
|
|
self._impl: Any = None
|
|
|
|
def _ensure_loaded(self) -> None:
|
|
if self._impl is not None:
|
|
return
|
|
if self.engine == "mlx":
|
|
from lightning_whisper_mlx import LightningWhisperMLX # type: ignore
|
|
|
|
log.info("Loading lightning-whisper-mlx model=%s batch=%d", settings.whisper_model, settings.whisper_batch_size)
|
|
self._impl = LightningWhisperMLX(
|
|
model=settings.whisper_model,
|
|
batch_size=settings.whisper_batch_size,
|
|
quant=None,
|
|
)
|
|
elif self.engine == "faster":
|
|
from faster_whisper import WhisperModel # type: ignore
|
|
|
|
log.info("Loading faster-whisper model=%s", settings.whisper_model)
|
|
self._impl = WhisperModel(settings.whisper_model, device="auto", compute_type="auto")
|
|
elif self.engine == "mock":
|
|
self._impl = "mock"
|
|
else:
|
|
raise ValueError(f"Unknown WHISPER_ENGINE: {self.engine}")
|
|
|
|
def transcribe(self, audio_path: Path, language: str | None = None) -> dict:
|
|
self._ensure_loaded()
|
|
lang = language or settings.whisper_language
|
|
|
|
if self.engine == "mock":
|
|
return {
|
|
"text": "[MOCK] Dies ist eine Beispiel-Transkription für Tests ohne Whisper-Modell.",
|
|
"language": lang,
|
|
"segments": [
|
|
{"start": 0.0, "end": 3.0, "text": "[MOCK] Erste Aussage."},
|
|
{"start": 3.0, "end": 6.0, "text": "[MOCK] Zweite Aussage."},
|
|
],
|
|
}
|
|
|
|
if self.engine == "mlx":
|
|
out = self._impl.transcribe(audio_path=str(audio_path), language=lang)
|
|
# MLX-Output schwankt zwischen Versionen:
|
|
# - String
|
|
# - {"text": "...", "segments": [...]}
|
|
# - {"text": "...", "segments": [[start, end, text], ...]}
|
|
# - {"text": "...", "segments": [{"start":..,"end":..,"text":..}, ...]}
|
|
if isinstance(out, str):
|
|
return {"text": out, "language": lang, "segments": []}
|
|
raw_segments = out.get("segments") or []
|
|
normalized = []
|
|
for s in raw_segments:
|
|
if isinstance(s, dict):
|
|
normalized.append({
|
|
"start": float(s.get("start", 0.0)),
|
|
"end": float(s.get("end", 0.0)),
|
|
"text": (s.get("text") or "").strip(),
|
|
})
|
|
elif isinstance(s, (list, tuple)) and len(s) >= 3:
|
|
normalized.append({
|
|
"start": float(s[0] or 0.0),
|
|
"end": float(s[1] or 0.0),
|
|
"text": str(s[2] or "").strip(),
|
|
})
|
|
else:
|
|
normalized.append({"start": 0.0, "end": 0.0, "text": str(s).strip()})
|
|
text = (out.get("text") or "").strip()
|
|
if not text and normalized:
|
|
text = " ".join(seg["text"] for seg in normalized).strip()
|
|
return {"text": text, "language": lang, "segments": normalized}
|
|
|
|
if self.engine == "faster":
|
|
segments_iter, info = self._impl.transcribe(str(audio_path), language=lang, vad_filter=True)
|
|
segments = []
|
|
full = []
|
|
for s in segments_iter:
|
|
t = s.text.strip()
|
|
full.append(t)
|
|
segments.append({"start": float(s.start), "end": float(s.end), "text": t})
|
|
return {"text": " ".join(full), "language": info.language or lang, "segments": segments}
|
|
|
|
raise RuntimeError(f"Engine {self.engine} not implemented")
|
|
|
|
|
|
engine = WhisperEngine()
|