feat: full-platform STT API (v2.3 consensus plan) #1

Open
lukehemmin wants to merge 21 commits from feat/full-platform into main
2 changed files with 61 additions and 3 deletions
Showing only changes of commit be5f505410 - Show all commits
@@ -133,7 +133,7 @@ class FasterWhisperEngine(TranscriptionEngine):
wrapped = _CancellableSegmentIterator( wrapped = _CancellableSegmentIterator(
self._to_dict_segments(segments_iter), should_cancel or (lambda: False) self._to_dict_segments(segments_iter), should_cancel or (lambda: False)
) )
return TranscriptionOutcome(wrapped, info=info) return TranscriptionOutcome(wrapped, info=self._to_dict_info(info))
@staticmethod @staticmethod
def _to_dict_segments(segments): def _to_dict_segments(segments):
@@ -169,6 +169,32 @@ class FasterWhisperEngine(TranscriptionEngine):
if (v := getattr(seg, k, None)) is not None if (v := getattr(seg, k, None)) is not None
} }
@staticmethod
def _to_dict_info(info):
"""faster-whisper TranscriptionInfo(namedtuple) → dict.
다운스트림(배치 파이프라인)은 info를 dict 계약으로 접근한다
(``info.get("language")``) — GPU 실전에서만 재현되는 버그:
'TranscriptionInfo' object has no attribute 'get'.
"""
if info is None or isinstance(info, dict):
return info
asdict = getattr(info, "_asdict", None)
if asdict is not None:
return dict(asdict())
try:
import dataclasses
if dataclasses.is_dataclass(info):
return dataclasses.asdict(info)
except Exception:
pass
return {
k: v
for k in ("language", "language_probability", "duration", "duration_after_vad")
if (v := getattr(info, k, None)) is not None
}
@staticmethod @staticmethod
def _split_device(device: str) -> tuple[str, int]: def _split_device(device: str) -> tuple[str, int]:
"""'cuda:N' → ('cuda', N). CTranslate2는 device='cuda'만 허용하므로 분리한다. """'cuda:N' → ('cuda', N). CTranslate2는 device='cuda'만 허용하므로 분리한다.
+34 -2
View File
@@ -16,17 +16,18 @@ from luke_scribe.engine.faster_whisper_engine import FasterWhisperEngine
class FakeWhisperModel: class FakeWhisperModel:
"""faster_whisper.WhisperModel 대체 — 생성 인자/세그먼트를 기록한다.""" """faster_whisper.WhisperModel 대체 — 생성 인자/세그먼트/info를 기록한다."""
calls: list[dict] = [] calls: list[dict] = []
segments: list = [] # transcribe()가 yield할 세그먼트 (기본: namedtuple) segments: list = [] # transcribe()가 yield할 세그먼트 (기본: namedtuple)
info: object = None # 기본: TranscriptionInfo(namedtuple) 흉내
def __init__(self, *args, **kwargs) -> None: def __init__(self, *args, **kwargs) -> None:
self.kwargs = kwargs self.kwargs = kwargs
self.__class__.calls.append(kwargs) self.__class__.calls.append(kwargs)
def transcribe(self, audio_path, **kwargs): def transcribe(self, audio_path, **kwargs):
return iter(list(self.__class__.segments)), {} return iter(list(self.__class__.segments)), self.__class__.info
def _install_fake(monkeypatch) -> None: def _install_fake(monkeypatch) -> None:
@@ -35,6 +36,7 @@ def _install_fake(monkeypatch) -> None:
monkeypatch.setitem(sys.modules, "faster_whisper", mod) monkeypatch.setitem(sys.modules, "faster_whisper", mod)
FakeWhisperModel.calls.clear() FakeWhisperModel.calls.clear()
FakeWhisperModel.segments = [] FakeWhisperModel.segments = []
FakeWhisperModel.info = None
def _opts(**kw) -> TranscriptionOptions: def _opts(**kw) -> TranscriptionOptions:
@@ -137,3 +139,33 @@ def test_dict_segments_passthrough(monkeypatch):
) )
segs = list(outcome.segments) segs = list(outcome.segments)
assert segs == [{"index": 0, "start": 0.0, "end": 1.0, "text": "x"}] assert segs == [{"index": 0, "start": 0.0, "end": 1.0, "text": "x"}]
def test_namedtuple_info_normalized_to_dict(monkeypatch):
"""faster-whisper TranscriptionInfo(namedtuple) → dict (GPU 실전 버그)."""
from collections import namedtuple
_install_fake(monkeypatch)
Info = namedtuple(
"TranscriptionInfo",
["language", "language_probability", "duration", "duration_after_vad"],
)
FakeWhisperModel.info = Info(
language="ko", language_probability=0.99, duration=10.464, duration_after_vad=9.088
)
outcome = FasterWhisperEngine().transcribe(
"/tmp/x.wav", _opts(device="cpu", compute_type="int8")
)
# dict 계약: .get() 사용 가능 (batch.py가 이걸로 접근)
assert outcome.info.get("language") == "ko"
assert outcome.info.get("duration") == 10.464
def test_dict_info_passthrough(monkeypatch):
"""이미 dict인 info는 그대로 (mock 계약과 호환)."""
_install_fake(monkeypatch)
FakeWhisperModel.info = {"language": "ko"}
outcome = FasterWhisperEngine().transcribe(
"/tmp/x.wav", _opts(device="cpu", compute_type="int8")
)
assert outcome.info == {"language": "ko"}