From f66b289371a39ddc4a104e30892b186d2c5264db Mon Sep 17 00:00:00 2001 From: Yera All Date: Fri, 17 Apr 2026 01:06:45 +0500 Subject: [PATCH] feat(voice): add Yandex streaming ASR --- .env.example | 3 + deployment/docker-compose.parallel.server.yml | 1 + deployment/docker-compose.server.registry.yml | 1 + deployment/docker-compose.server.yml | 1 + deployment/docker-compose.yml | 1 + requirements.txt | 4 + scripts/local_stack.py | 1 + .../ai_voice_runtime_service/providers/asr.py | 381 +++++++++++++++++- tests/test_ai_voice_asr_provider.py | 125 ++++++ 9 files changed, 516 insertions(+), 2 deletions(-) diff --git a/.env.example b/.env.example index a8a43b2..66076ed 100644 --- a/.env.example +++ b/.env.example @@ -78,6 +78,7 @@ AI_VOICE_AUDIOSOCKET_PORT=9019 AI_VOICE_ASR_MODEL=gpt-4o-mini-transcribe AI_VOICE_ASR_YANDEX_API_BASE=https://stt.api.ml.yandexcloud.kz AI_VOICE_ASR_YANDEX_API_VERSION=auto +AI_VOICE_ASR_YANDEX_GRPC_TARGET=stt.api.ml.yandexcloud.kz:443 AI_VOICE_ASR_YANDEX_API_KEY= AI_VOICE_ASR_YANDEX_IAM_TOKEN= AI_VOICE_ASR_YANDEX_FOLDER_ID= @@ -86,6 +87,8 @@ AI_VOICE_ASR_YANDEX_TOPIC=general AI_VOICE_ASR_YANDEX_SAMPLE_RATE_HZ=8000 AI_VOICE_ASR_YANDEX_TIMEOUT_SECONDS=8 AI_VOICE_ASR_YANDEX_POLL_INTERVAL_SECONDS=0.25 +AI_VOICE_ASR_YANDEX_EOU_MAX_PAUSE_MS=450 +AI_VOICE_ASR_YANDEX_STREAM_QUEUE_MAX_CHUNKS=300 AI_VOICE_TTS_MODEL=gpt-4o-mini-tts AI_VOICE_TTS_VOICE=alloy AI_VOICE_TTS_YANDEX_API_BASE=https://tts.api.ml.yandexcloud.kz diff --git a/deployment/docker-compose.parallel.server.yml b/deployment/docker-compose.parallel.server.yml index bdec4f1..503ea4a 100644 --- a/deployment/docker-compose.parallel.server.yml +++ b/deployment/docker-compose.parallel.server.yml @@ -27,6 +27,7 @@ x-app-env: &app_env AI_VOICE_V2_STREAMING_ASR_BACKEND: yandex_speechkit AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021 AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4" + AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED: "1" x-service-defaults: &service_defaults build: diff --git a/deployment/docker-compose.server.registry.yml b/deployment/docker-compose.server.registry.yml index 7d06b06..bb302f2 100644 --- a/deployment/docker-compose.server.registry.yml +++ b/deployment/docker-compose.server.registry.yml @@ -27,6 +27,7 @@ x-app-env: &app_env AI_VOICE_V2_STREAMING_ASR_BACKEND: yandex_speechkit AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021 AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4" + AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED: "1" x-service-defaults: &service_defaults image: ${APP_IMAGE:?Set APP_IMAGE in deployment/.env.images or shell env} diff --git a/deployment/docker-compose.server.yml b/deployment/docker-compose.server.yml index b21d236..4254423 100644 --- a/deployment/docker-compose.server.yml +++ b/deployment/docker-compose.server.yml @@ -27,6 +27,7 @@ x-app-env: &app_env AI_VOICE_V2_STREAMING_ASR_BACKEND: yandex_speechkit AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021 AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4" + AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED: "1" x-service-defaults: &service_defaults build: diff --git a/deployment/docker-compose.yml b/deployment/docker-compose.yml index f5e5aba..76e29c6 100644 --- a/deployment/docker-compose.yml +++ b/deployment/docker-compose.yml @@ -38,6 +38,7 @@ x-app-env: &app_env AI_VOICE_V2_STREAMING_ASR_BACKEND: yandex_speechkit AI_VOICE_V2_STREAMING_ASR_BASE_URL: http://streaming-asr-sidecar:8021 AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS: "4" + AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED: "1" AI_VOICE_V2_PREBAKED_ACK_ENABLED: "1" AI_VOICE_V2_PREBAKED_ACK_DIR: /app/.data/voice_v2_ack_bank AI_VOICE_V2_STREAMING_TTS: "1" diff --git a/requirements.txt b/requirements.txt index 7287c34..fbe544f 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,4 +9,8 @@ pika==1.3.2 paramiko==3.5.0 python-multipart==0.0.20 faster-whisper==1.1.1 +grpcio==1.80.0 +grpcio-tools==1.80.0 +protobuf==6.33.6 requests==2.32.5 +yandexcloud==0.386.0 diff --git a/scripts/local_stack.py b/scripts/local_stack.py index 49c3b30..881a09c 100644 --- a/scripts/local_stack.py +++ b/scripts/local_stack.py @@ -319,6 +319,7 @@ def spawn_service(spec: dict[str, Any], runtime_dir: Path, data_dir: Path, base_ env["AI_VOICE_V2_STREAMING_ASR_BACKEND"] = env.get("AI_VOICE_V2_STREAMING_ASR_BACKEND", "yandex_speechkit") env["AI_VOICE_V2_STREAMING_ASR_BASE_URL"] = env.get("AI_VOICE_V2_STREAMING_ASR_BASE_URL", "http://127.0.0.1:8021") env["AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS"] = env.get("AI_VOICE_V2_STREAMING_ASR_TIMEOUT_SECONDS", "4") + env["AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED"] = env.get("AI_VOICE_V2_STREAMING_ASR_PARTIAL_POLL_ENABLED", "1") env["AI_VOICE_V2_STREAMING_ASR_MODEL"] = env.get("AI_VOICE_V2_STREAMING_ASR_MODEL", "base") env["AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE"] = env.get("AI_VOICE_V2_STREAMING_ASR_COMPUTE_TYPE", "int8") env["AI_VOICE_V2_STREAMING_ASR_DEVICE"] = env.get("AI_VOICE_V2_STREAMING_ASR_DEVICE", "cpu") diff --git a/services/ai_voice_runtime_service/providers/asr.py b/services/ai_voice_runtime_service/providers/asr.py index 66b30f8..c98cf98 100644 --- a/services/ai_voice_runtime_service/providers/asr.py +++ b/services/ai_voice_runtime_service/providers/asr.py @@ -3,11 +3,12 @@ from __future__ import annotations import base64 import io import os +import queue import threading import time import uuid import wave -from dataclasses import dataclass +from dataclasses import dataclass, field import httpx @@ -112,6 +113,33 @@ def _yandex_asr_poll_interval_seconds() -> float: return max(min(value, 2.0), 0.05) +def _yandex_asr_grpc_target(api_base: str) -> str: + explicit = os.getenv("AI_VOICE_ASR_YANDEX_GRPC_TARGET", "").strip() + if explicit: + return explicit + if "yandexcloud.kz" in str(api_base or "").lower(): + return "stt.api.ml.yandexcloud.kz:443" + return "stt.api.cloud.yandex.net:443" + + +def _yandex_asr_stream_queue_max_chunks() -> int: + raw = os.getenv("AI_VOICE_ASR_YANDEX_STREAM_QUEUE_MAX_CHUNKS", "300").strip() + try: + value = int(raw) + except ValueError: + value = 300 + return max(min(value, 2000), 10) + + +def _yandex_asr_eou_max_pause_ms() -> int: + raw = os.getenv("AI_VOICE_ASR_YANDEX_EOU_MAX_PAUSE_MS", "450").strip() + try: + value = int(raw) + except ValueError: + value = 450 + return max(min(value, 3000), 100) + + def _yandex_asr_default_language() -> str: return os.getenv("AI_VOICE_ASR_YANDEX_LANGUAGE", "ru-RU").strip() or "ru-RU" @@ -551,6 +579,353 @@ class LocalSidecarStreamingASRProvider(StreamingASRProvider): return +@dataclass +class _YandexGrpcStreamState: + stream_id: str + session_id: str + language: str + input_queue: queue.Queue[bytes | None] + updates_queue: queue.Queue[StreamingASRPartial] + thread: threading.Thread | None = None + latest_partial: StreamingASRPartial | None = None + final_transcription: ASRTranscription | None = None + error: BaseException | None = None + finish_requested: bool = False + lock: threading.Lock = field(default_factory=threading.Lock) + + +class YandexSpeechKitGrpcStreamingASRProvider(StreamingASRProvider): + name = "yandex-speechkit-grpc" + supports_streaming = True + + def __init__( + self, + *, + api_base: str | None = None, + api_key: str | None = None, + iam_token: str | None = None, + folder_id: str | None = None, + timeout_seconds: float | None = None, + sample_rate_hz: int | None = None, + topic: str | None = None, + grpc_target: str | None = None, + queue_max_chunks: int | None = None, + eou_max_pause_ms: int | None = None, + grpc_module: object | None = None, + stt_pb2_module: object | None = None, + stt_service_pb2_grpc_module: object | None = None, + ) -> None: + self._api_base = str(api_base or _yandex_asr_api_base()).strip().rstrip("/") + self._api_key = str(api_key if api_key is not None else _yandex_asr_api_key()).strip() + self._iam_token = str(iam_token if iam_token is not None else _yandex_asr_iam_token()).strip() + self._folder_id = str(folder_id if folder_id is not None else _yandex_asr_folder_id()).strip() + self._timeout_seconds = max( + float(timeout_seconds if timeout_seconds is not None else _yandex_asr_timeout_seconds()), + 3.0, + ) + self._sample_rate_hz = int(sample_rate_hz if sample_rate_hz is not None else _yandex_asr_sample_rate_hz()) + self._topic = str(topic if topic is not None else _yandex_asr_topic()).strip() or "general" + self._grpc_target = str(grpc_target or _yandex_asr_grpc_target(self._api_base)).strip() + self._queue_max_chunks = int(queue_max_chunks or _yandex_asr_stream_queue_max_chunks()) + self._eou_max_pause_ms = int(eou_max_pause_ms or _yandex_asr_eou_max_pause_ms()) + self._grpc_module = grpc_module + self._stt_pb2_module = stt_pb2_module + self._stt_service_pb2_grpc_module = stt_service_pb2_grpc_module + self._streams: dict[str, _YandexGrpcStreamState] = {} + self._lock = threading.Lock() + + def _load_grpc_modules(self) -> tuple[object, object, object]: + if ( + self._grpc_module is not None + and self._stt_pb2_module is not None + and self._stt_service_pb2_grpc_module is not None + ): + return self._grpc_module, self._stt_pb2_module, self._stt_service_pb2_grpc_module + try: + import grpc # type: ignore[import-not-found] + from yandex.cloud.ai.stt.v3 import stt_pb2 # type: ignore[import-not-found] + from yandex.cloud.ai.stt.v3 import stt_service_pb2_grpc # type: ignore[import-not-found] + except ImportError as exc: + raise StreamingASRUnavailable("Yandex SpeechKit gRPC dependencies are not installed") from exc + return grpc, stt_pb2, stt_service_pb2_grpc + + def _metadata(self) -> tuple[tuple[str, str], ...]: + if not self._api_key and not self._iam_token: + raise StreamingASRUnavailable( + "AI_VOICE_ASR_YANDEX_API_KEY or AI_VOICE_ASR_YANDEX_IAM_TOKEN is required for Yandex streaming ASR" + ) + metadata: list[tuple[str, str]] = [] + if self._api_key: + metadata.append(("authorization", f"Api-Key {self._api_key}")) + else: + metadata.append(("authorization", f"Bearer {self._iam_token}")) + if self._folder_id: + metadata.append(("x-folder-id", self._folder_id)) + return tuple(metadata) + + @staticmethod + def _enum_value(owner: object, name: str, default: int) -> int: + return int(getattr(owner, name, default)) + + def _build_session_options(self, stt_pb2: object, *, language: str) -> object: + raw_audio = stt_pb2.RawAudio( # type: ignore[attr-defined] + audio_encoding=self._enum_value(stt_pb2.RawAudio, "LINEAR16_PCM", 1), # type: ignore[attr-defined] + sample_rate_hertz=self._sample_rate_hz, + audio_channel_count=1, + ) + recognition_model = stt_pb2.RecognitionModelOptions( # type: ignore[attr-defined] + model=self._topic, + audio_format=stt_pb2.AudioFormatOptions(raw_audio=raw_audio), # type: ignore[attr-defined] + text_normalization=stt_pb2.TextNormalizationOptions( # type: ignore[attr-defined] + text_normalization=self._enum_value( + stt_pb2.TextNormalizationOptions, # type: ignore[attr-defined] + "TEXT_NORMALIZATION_ENABLED", + 1, + ), + profanity_filter=False, + literature_text=False, + phone_formatting_mode=self._enum_value( + stt_pb2.TextNormalizationOptions, # type: ignore[attr-defined] + "PHONE_FORMATTING_MODE_DISABLED", + 1, + ), + ), + language_restriction=stt_pb2.LanguageRestrictionOptions( # type: ignore[attr-defined] + restriction_type=self._enum_value(stt_pb2.LanguageRestrictionOptions, "WHITELIST", 1), # type: ignore[attr-defined] + language_code=[language], + ), + audio_processing_type=self._enum_value(stt_pb2.RecognitionModelOptions, "REAL_TIME", 1), # type: ignore[attr-defined] + ) + eou_classifier = stt_pb2.EouClassifierOptions( # type: ignore[attr-defined] + default_classifier=stt_pb2.DefaultEouClassifier( # type: ignore[attr-defined] + type=self._enum_value(stt_pb2.DefaultEouClassifier, "HIGH", 2), # type: ignore[attr-defined] + max_pause_between_words_hint_ms=self._eou_max_pause_ms, + ) + ) + return stt_pb2.StreamingOptions( # type: ignore[attr-defined] + recognition_model=recognition_model, + eou_classifier=eou_classifier, + ) + + def _request_iterator(self, state: _YandexGrpcStreamState, stt_pb2: object): + yield stt_pb2.StreamingRequest( # type: ignore[attr-defined] + session_options=self._build_session_options(stt_pb2, language=state.language) + ) + while True: + item = state.input_queue.get() + try: + if item is None: + yield stt_pb2.StreamingRequest(eou=stt_pb2.Eou()) # type: ignore[attr-defined] + return + if item: + yield stt_pb2.StreamingRequest(chunk=stt_pb2.AudioChunk(data=item)) # type: ignore[attr-defined] + finally: + state.input_queue.task_done() + + @staticmethod + def _has_field(message: object, field_name: str) -> bool: + try: + return bool(message.HasField(field_name)) # type: ignore[attr-defined] + except (AttributeError, ValueError): + return getattr(message, field_name, None) is not None + + @classmethod + def _partial_from_response(cls, response: object) -> StreamingASRPartial | None: + for field_name, is_final in ( + ("final_refinement", True), + ("final", True), + ("partial", False), + ): + if not cls._has_field(response, field_name): + continue + payload = getattr(response, field_name, None) + if field_name == "final_refinement" and payload is not None: + payload = getattr(payload, "normalized_text", None) + text, confidence, language = cls._extract_alternatives(payload) + if text: + return StreamingASRPartial( + text=text, + language=language, + confidence=confidence, + is_final=is_final, + is_stable=is_final, + ) + return None + + @staticmethod + def _extract_alternatives(payload: object) -> tuple[str, float | None, str | None]: + alternatives = getattr(payload, "alternatives", None) + if not alternatives: + return "", None, None + texts: list[str] = [] + confidence: float | None = None + language: str | None = None + for alternative in alternatives: + text = str(getattr(alternative, "text", "") or "").strip() + if text: + texts.append(text) + if confidence is None: + raw_confidence = getattr(alternative, "confidence", None) + if isinstance(raw_confidence, (int, float)): + confidence = float(raw_confidence) + if language is None: + languages = getattr(alternative, "languages", None) + if languages: + first_language = languages[0] + language_code = str(getattr(first_language, "language_code", "") or "").strip() + if language_code: + language = language_code + return " ".join(texts).strip(), confidence, language + + def _drain_updates(self, state: _YandexGrpcStreamState) -> None: + while True: + try: + update = state.updates_queue.get_nowait() + except queue.Empty: + return + with state.lock: + state.latest_partial = update + if update.is_final: + state.final_transcription = ASRTranscription( + text=update.text, + language=update.language or state.language, + confidence=update.confidence, + ) + + def _run_stream(self, state: _YandexGrpcStreamState) -> None: + channel = None + try: + grpc, stt_pb2, stt_service_pb2_grpc = self._load_grpc_modules() + channel = grpc.secure_channel( # type: ignore[attr-defined] + self._grpc_target, + grpc.ssl_channel_credentials(), # type: ignore[attr-defined] + ) + stub = stt_service_pb2_grpc.RecognizerStub(channel) # type: ignore[attr-defined] + responses = stub.RecognizeStreaming( + self._request_iterator(state, stt_pb2), + metadata=self._metadata(), + timeout=self._timeout_seconds, + ) + for response in responses: + partial = self._partial_from_response(response) + if partial is not None: + state.updates_queue.put(partial) + except BaseException as exc: + with state.lock: + state.error = exc + finally: + if channel is not None: + close = getattr(channel, "close", None) + if callable(close): + close() + + def _state(self, stream_id: str) -> _YandexGrpcStreamState: + with self._lock: + state = self._streams.get(stream_id) + if state is None: + raise StreamingASRUnavailable("Yandex SpeechKit gRPC stream is not active") + return state + + def _signal_finish(self, state: _YandexGrpcStreamState) -> None: + with state.lock: + if state.finish_requested: + return + state.finish_requested = True + try: + state.input_queue.put_nowait(None) + except queue.Full: + try: + state.input_queue.get_nowait() + state.input_queue.task_done() + except queue.Empty: + pass + state.input_queue.put_nowait(None) + + def open_stream(self, session_id: str, *, language_hint: str | None = None) -> str: + self._metadata() + self._load_grpc_modules() + stream_id = f"yasr_grpc_{uuid.uuid4().hex}" + language = _normalize_yandex_asr_language(language_hint) + state = _YandexGrpcStreamState( + stream_id=stream_id, + session_id=str(session_id or "").strip(), + language=language, + input_queue=queue.Queue(maxsize=self._queue_max_chunks), + updates_queue=queue.Queue(), + ) + thread = threading.Thread( + target=self._run_stream, + args=(state,), + name=f"yandex-speechkit-asr-{stream_id[:16]}", + daemon=True, + ) + state.thread = thread + with self._lock: + self._streams[stream_id] = state + thread.start() + return stream_id + + def push_pcm(self, stream_id: str, pcm_8k_chunk: bytes) -> None: + if not pcm_8k_chunk: + return + state = self._state(stream_id) + with state.lock: + if state.error is not None: + raise StreamingASRUnavailable(str(state.error)[:500]) + if state.finish_requested: + raise StreamingASRUnavailable("Yandex SpeechKit gRPC stream is already finalizing") + try: + state.input_queue.put(pcm_8k_chunk, timeout=0.1) + except queue.Full as exc: + raise StreamingASRUnavailable("Yandex SpeechKit gRPC input queue is full") from exc + + def poll_partial(self, stream_id: str) -> StreamingASRPartial | None: + state = self._state(stream_id) + self._drain_updates(state) + with state.lock: + if state.error is not None and state.latest_partial is None: + raise StreamingASRUnavailable(str(state.error)[:500]) + return state.latest_partial + + def finalize(self, stream_id: str) -> ASRTranscription: + state = self._state(stream_id) + self._signal_finish(state) + deadline = time.monotonic() + self._timeout_seconds + while True: + self._drain_updates(state) + thread = state.thread + if thread is None or not thread.is_alive(): + break + if time.monotonic() >= deadline: + raise StreamingASRUnavailable("Yandex SpeechKit gRPC stream finalize timed out") + thread.join(timeout=0.05) + self._drain_updates(state) + with state.lock: + if state.final_transcription is not None: + return state.final_transcription + if state.latest_partial is not None: + return ASRTranscription( + text=state.latest_partial.text, + language=state.latest_partial.language or state.language, + confidence=state.latest_partial.confidence, + ) + if state.error is not None: + raise StreamingASRUnavailable(str(state.error)[:500]) + return ASRTranscription(text="", language=state.language, confidence=None) + + def close_stream(self, stream_id: str) -> None: + try: + state = self._state(stream_id) + except StreamingASRUnavailable: + return + self._signal_finish(state) + thread = state.thread + if thread is not None and thread.is_alive(): + thread.join(timeout=0.5) + with self._lock: + self._streams.pop(stream_id, None) + + class YandexSpeechKitBufferedStreamingASRProvider(StreamingASRProvider): name = "yandex-speechkit-buffered" supports_streaming = True @@ -612,6 +987,8 @@ def build_streaming_asr_provider(name: str) -> StreamingASRProvider: normalized = str(name or "disabled").strip().lower() if normalized in {"local_sidecar", "local-sidecar", "sidecar"}: return LocalSidecarStreamingASRProvider() - if normalized in {"yandex", "yandex_speechkit", "yandex-speechkit", "speechkit"}: + if normalized in {"yandex", "yandex_speechkit", "yandex-speechkit", "speechkit", "yandex_grpc"}: + return YandexSpeechKitGrpcStreamingASRProvider() + if normalized in {"yandex_buffered", "yandex_speechkit_buffered", "yandex-speechkit-buffered"}: return YandexSpeechKitBufferedStreamingASRProvider() return StreamingASRProvider() diff --git a/tests/test_ai_voice_asr_provider.py b/tests/test_ai_voice_asr_provider.py index b03d350..2c5be41 100644 --- a/tests/test_ai_voice_asr_provider.py +++ b/tests/test_ai_voice_asr_provider.py @@ -1,5 +1,7 @@ from __future__ import annotations +import time + from services.ai_voice_runtime_service.audiosocket import pcm16le_to_wav_bytes from services.ai_voice_runtime_service.providers import asr as asr_module @@ -198,10 +200,133 @@ def test_yandex_buffered_streaming_provider_buffers_pcm_until_finalize(): assert calls == [{"audio_bytes": b"\x10\x00\x20\x00", "language_hint": "ru"}] +def test_yandex_grpc_streaming_provider_returns_partial_and_final(): + calls: list[dict] = [] + + class _Message: + def __init__(self, **kwargs) -> None: + self.__dict__.update(kwargs) + + def HasField(self, name: str) -> bool: + return getattr(self, name, None) is not None + + class _RawAudio(_Message): + LINEAR16_PCM = 1 + + class _TextNormalizationOptions(_Message): + TEXT_NORMALIZATION_ENABLED = 1 + PHONE_FORMATTING_MODE_DISABLED = 1 + + class _LanguageRestrictionOptions(_Message): + WHITELIST = 1 + + class _RecognitionModelOptions(_Message): + REAL_TIME = 1 + + class _DefaultEouClassifier(_Message): + HIGH = 2 + + class _FakeSttPb2: + RawAudio = _RawAudio + AudioFormatOptions = _Message + TextNormalizationOptions = _TextNormalizationOptions + LanguageRestrictionOptions = _LanguageRestrictionOptions + RecognitionModelOptions = _RecognitionModelOptions + DefaultEouClassifier = _DefaultEouClassifier + EouClassifierOptions = _Message + StreamingOptions = _Message + StreamingRequest = _Message + AudioChunk = _Message + Eou = _Message + + class _FakeChannel: + def close(self) -> None: + calls.append({"method": "close"}) + + class _FakeGrpc: + @staticmethod + def ssl_channel_credentials(): + return "ssl" + + @staticmethod + def secure_channel(target: str, credentials): + calls.append({"method": "secure_channel", "target": target, "credentials": credentials}) + return _FakeChannel() + + class _RecognizerStub: + def __init__(self, channel) -> None: + calls.append({"method": "stub_init", "channel": channel}) + + def RecognizeStreaming(self, request_iterator, *, metadata, timeout): + calls.append({"method": "recognize", "metadata": metadata, "timeout": timeout}) + for request in request_iterator: + calls.append({"method": "request", "request": request}) + if getattr(request, "chunk", None) is not None: + yield _Message( + partial=_Message( + alternatives=[ + _Message(text="need schedule", confidence=0.7, languages=[]), + ] + ) + ) + continue + if getattr(request, "eou", None) is not None: + yield _Message( + final_refinement=_Message( + normalized_text=_Message( + alternatives=[ + _Message(text="need schedule in almaty", confidence=0.91, languages=[]), + ] + ) + ) + ) + return + + class _FakeGrpcPb2: + RecognizerStub = _RecognizerStub + + provider = asr_module.YandexSpeechKitGrpcStreamingASRProvider( + api_key="asr-key", + grpc_target="stt.test:443", + timeout_seconds=3, + grpc_module=_FakeGrpc, + stt_pb2_module=_FakeSttPb2, + stt_service_pb2_grpc_module=_FakeGrpcPb2, + ) + stream_id = provider.open_stream("session-1", language_hint="ru") + provider.push_pcm(stream_id, b"\x10\x00\x20\x00") + + partial = None + for _ in range(20): + partial = provider.poll_partial(stream_id) + if partial is not None: + break + time.sleep(0.02) + + assert partial is not None + assert partial.text == "need schedule" + assert partial.is_final is False + + final = provider.finalize(stream_id) + provider.close_stream(stream_id) + + assert final.text == "need schedule in almaty" + recognize_call = next(call for call in calls if call.get("method") == "recognize") + assert ("authorization", "Api-Key asr-key") in recognize_call["metadata"] + requests = [call["request"] for call in calls if call.get("method") == "request"] + assert getattr(requests[0], "session_options", None) is not None + assert getattr(requests[1], "chunk", None).data == b"\x10\x00\x20\x00" + assert getattr(requests[-1], "eou", None) is not None + + def test_yandex_asr_builders(): assert isinstance(asr_module.build_asr_provider("yandex"), asr_module.YandexSpeechKitASRProvider) assert isinstance(asr_module.build_asr_provider("speechkit"), asr_module.YandexSpeechKitASRProvider) assert isinstance( asr_module.build_streaming_asr_provider("yandex_speechkit"), + asr_module.YandexSpeechKitGrpcStreamingASRProvider, + ) + assert isinstance( + asr_module.build_streaming_asr_provider("yandex_speechkit_buffered"), asr_module.YandexSpeechKitBufferedStreamingASRProvider, )