# This file was auto-generated by Fern from our API Definition. import typing from contextlib import asynccontextmanager, contextmanager import httpx import websockets.exceptions import websockets.sync.client as websockets_sync_client from ..core.api_error import ApiError from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.request_options import RequestOptions from .raw_client import AsyncRawSpeechToTextTranslateStreamingClient, RawSpeechToTextTranslateStreamingClient from .socket_client import AsyncSpeechToTextTranslateStreamingSocketClient, SpeechToTextTranslateStreamingSocketClient from .types.speech_to_text_translate_streaming_flush_signal import SpeechToTextTranslateStreamingFlushSignal from .types.speech_to_text_translate_streaming_high_vad_sensitivity import ( SpeechToTextTranslateStreamingHighVadSensitivity, ) from .types.speech_to_text_translate_streaming_input_audio_codec import SpeechToTextTranslateStreamingInputAudioCodec from .types.speech_to_text_translate_streaming_vad_signals import SpeechToTextTranslateStreamingVadSignals try: from websockets.legacy.client import connect as websockets_client_connect # type: ignore except ImportError: from websockets import connect as websockets_client_connect # type: ignore class SpeechToTextTranslateStreamingClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawSpeechToTextTranslateStreamingClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawSpeechToTextTranslateStreamingClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawSpeechToTextTranslateStreamingClient """ return self._raw_client @contextmanager def connect( self, *, model: typing.Optional[typing.Literal["saaras:v2.5"]] = None, sample_rate: typing.Optional[str] = None, high_vad_sensitivity: typing.Optional[SpeechToTextTranslateStreamingHighVadSensitivity] = None, vad_signals: typing.Optional[SpeechToTextTranslateStreamingVadSignals] = None, flush_signal: typing.Optional[SpeechToTextTranslateStreamingFlushSignal] = None, input_audio_codec: typing.Optional[SpeechToTextTranslateStreamingInputAudioCodec] = None, api_subscription_key: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[SpeechToTextTranslateStreamingSocketClient]: """ WebSocket channel for real-time speech to text streaming with English translation. **Note:** This API Reference page is provided for informational purposes only. The Try It playground may not provide the best experience for streaming audio. For optimal streaming performance, please use the SDK or implement your own WebSocket client. Parameters ---------- model : typing.Optional[typing.Literal["saaras:v2.5"]] Model to be used for speech to text translation. - **saaras:v2.5** (default): Translation model that translates audio from any spoken Indic language to English. - Example: Hindi audio → English text output sample_rate : typing.Optional[str] Audio sample rate for the WebSocket connection. When specified as a connection parameter, only 16kHz and 8kHz are supported. 8kHz is only available via this connection parameter. If not specified, defaults to 16kHz. high_vad_sensitivity : typing.Optional[SpeechToTextTranslateStreamingHighVadSensitivity] Enable high VAD (Voice Activity Detection) sensitivity vad_signals : typing.Optional[SpeechToTextTranslateStreamingVadSignals] Enable VAD signals in response flush_signal : typing.Optional[SpeechToTextTranslateStreamingFlushSignal] Signal to flush the audio buffer and finalize transcription and translation input_audio_codec : typing.Optional[SpeechToTextTranslateStreamingInputAudioCodec] Audio codec/format of the input stream. Use this when sending raw PCM audio. Supported values: wav, pcm_s16le, pcm_l16, pcm_raw. api_subscription_key : typing.Optional[str] API subscription key for authentication request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- SpeechToTextTranslateStreamingSocketClient """ ws_url = self._raw_client._client_wrapper.get_environment().production + "/speech-to-text-translate/ws" query_params = httpx.QueryParams() if model is not None: query_params = query_params.add("model", model) if sample_rate is not None: query_params = query_params.add("sample_rate", sample_rate) if high_vad_sensitivity is not None: query_params = query_params.add("high_vad_sensitivity", high_vad_sensitivity) if vad_signals is not None: query_params = query_params.add("vad_signals", vad_signals) if flush_signal is not None: query_params = query_params.add("flush_signal", flush_signal) if input_audio_codec is not None: query_params = query_params.add("input_audio_codec", input_audio_codec) ws_url = ws_url + f"?{query_params}" headers = self._raw_client._client_wrapper.get_headers() if api_subscription_key is not None: headers["Api-Subscription-Key"] = str(api_subscription_key) if request_options and "additional_headers" in request_options: headers.update(request_options["additional_headers"]) try: with websockets_sync_client.connect(ws_url, additional_headers=headers) as protocol: yield SpeechToTextTranslateStreamingSocketClient(websocket=protocol) except websockets.exceptions.InvalidStatusCode as exc: status_code: int = exc.status_code if status_code == 401: raise ApiError( status_code=status_code, headers=dict(headers), body="Websocket initialized with invalid credentials.", ) raise ApiError( status_code=status_code, headers=dict(headers), body="Unexpected error when initializing websocket connection.", ) class AsyncSpeechToTextTranslateStreamingClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawSpeechToTextTranslateStreamingClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> AsyncRawSpeechToTextTranslateStreamingClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawSpeechToTextTranslateStreamingClient """ return self._raw_client @asynccontextmanager async def connect( self, *, model: typing.Optional[typing.Literal["saaras:v2.5"]] = None, sample_rate: typing.Optional[str] = None, high_vad_sensitivity: typing.Optional[SpeechToTextTranslateStreamingHighVadSensitivity] = None, vad_signals: typing.Optional[SpeechToTextTranslateStreamingVadSignals] = None, flush_signal: typing.Optional[SpeechToTextTranslateStreamingFlushSignal] = None, input_audio_codec: typing.Optional[SpeechToTextTranslateStreamingInputAudioCodec] = None, api_subscription_key: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[AsyncSpeechToTextTranslateStreamingSocketClient]: """ WebSocket channel for real-time speech to text streaming with English translation. **Note:** This API Reference page is provided for informational purposes only. The Try It playground may not provide the best experience for streaming audio. For optimal streaming performance, please use the SDK or implement your own WebSocket client. Parameters ---------- model : typing.Optional[typing.Literal["saaras:v2.5"]] Model to be used for speech to text translation. - **saaras:v2.5** (default): Translation model that translates audio from any spoken Indic language to English. - Example: Hindi audio → English text output sample_rate : typing.Optional[str] Audio sample rate for the WebSocket connection. When specified as a connection parameter, only 16kHz and 8kHz are supported. 8kHz is only available via this connection parameter. If not specified, defaults to 16kHz. high_vad_sensitivity : typing.Optional[SpeechToTextTranslateStreamingHighVadSensitivity] Enable high VAD (Voice Activity Detection) sensitivity vad_signals : typing.Optional[SpeechToTextTranslateStreamingVadSignals] Enable VAD signals in response flush_signal : typing.Optional[SpeechToTextTranslateStreamingFlushSignal] Signal to flush the audio buffer and finalize transcription and translation input_audio_codec : typing.Optional[SpeechToTextTranslateStreamingInputAudioCodec] Audio codec/format of the input stream. Use this when sending raw PCM audio. Supported values: wav, pcm_s16le, pcm_l16, pcm_raw. api_subscription_key : typing.Optional[str] API subscription key for authentication request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncSpeechToTextTranslateStreamingSocketClient """ ws_url = self._raw_client._client_wrapper.get_environment().production + "/speech-to-text-translate/ws" query_params = httpx.QueryParams() if model is not None: query_params = query_params.add("model", model) if sample_rate is not None: query_params = query_params.add("sample_rate", sample_rate) if high_vad_sensitivity is not None: query_params = query_params.add("high_vad_sensitivity", high_vad_sensitivity) if vad_signals is not None: query_params = query_params.add("vad_signals", vad_signals) if flush_signal is not None: query_params = query_params.add("flush_signal", flush_signal) if input_audio_codec is not None: query_params = query_params.add("input_audio_codec", input_audio_codec) ws_url = ws_url + f"?{query_params}" headers = self._raw_client._client_wrapper.get_headers() if api_subscription_key is not None: headers["Api-Subscription-Key"] = str(api_subscription_key) if request_options and "additional_headers" in request_options: headers.update(request_options["additional_headers"]) try: async with websockets_client_connect(ws_url, extra_headers=headers) as protocol: yield AsyncSpeechToTextTranslateStreamingSocketClient(websocket=protocol) except websockets.exceptions.InvalidStatusCode as exc: status_code: int = exc.status_code if status_code == 401: raise ApiError( status_code=status_code, headers=dict(headers), body="Websocket initialized with invalid credentials.", ) raise ApiError( status_code=status_code, headers=dict(headers), body="Unexpected error when initializing websocket connection.", )