# This file was auto-generated by Fern from our API Definition. import typing import urllib.parse from contextlib import asynccontextmanager, contextmanager import websockets.sync.client as websockets_sync_client from ...core.api_error import ApiError from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ...core.jsonable_encoder import jsonable_encoder from ...core.query_encoder import encode_query from ...core.remove_none_from_dict import remove_none_from_dict from ...core.request_options import RequestOptions from ...core.websocket_compat import InvalidWebSocketStatus, get_status_code from ...types.speak_v2encoding import SpeakV2Encoding from ...types.speak_v2expressivity import SpeakV2Expressivity from ...types.speak_v2mip_opt_out import SpeakV2MipOptOut from ...types.speak_v2model import SpeakV2Model from ...types.speak_v2sample_rate import SpeakV2SampleRate from ...types.speak_v2speed import SpeakV2Speed from ...types.speak_v2tag import SpeakV2Tag from .socket_client import AsyncV2SocketClient, V2SocketClient 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 RawV2Client: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper @contextmanager def connect( self, *, model: SpeakV2Model, encoding: typing.Optional[SpeakV2Encoding] = None, sample_rate: typing.Optional[SpeakV2SampleRate] = None, speed: typing.Optional[SpeakV2Speed] = None, expressivity: typing.Optional[SpeakV2Expressivity] = None, mip_opt_out: typing.Optional[SpeakV2MipOptOut] = None, tag: typing.Optional[SpeakV2Tag] = None, authorization: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[V2SocketClient]: """ Streaming, turn-based text-to-speech (Flux TTS) built for voice-agent pipelines. Stream LLM tokens in, speak them to the user, and report per-turn billing and timing. Parameters ---------- model : SpeakV2Model encoding : typing.Optional[SpeakV2Encoding] sample_rate : typing.Optional[SpeakV2SampleRate] speed : typing.Optional[SpeakV2Speed] expressivity : typing.Optional[SpeakV2Expressivity] mip_opt_out : typing.Optional[SpeakV2MipOptOut] tag : typing.Optional[SpeakV2Tag] authorization : typing.Optional[str] Use your API key or a [temporary token](/guides/fundamentals/token-based-authentication) for authentication via the `Authorization` header. In client-side environments where custom headers are not supported, use the [`Sec-WebSocket-Protocol`](/docs/using-the-sec-websocket-protocol) header instead. **Example:** `Authorization: Token %DEEPGRAM_API_KEY%` or `Authorization: Bearer %DEEPGRAM_TOKEN%` request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- V2SocketClient """ ws_url = self._client_wrapper.get_environment().production + "/v2/speak" _encoded_query_params = encode_query( jsonable_encoder( remove_none_from_dict( { "model": model, "encoding": encoding, "sample_rate": sample_rate, "speed": speed, "expressivity": expressivity, "mip_opt_out": mip_opt_out, "tag": tag, **( request_options.get("additional_query_parameters", {}) or {} if request_options is not None else {} ), } ) ) ) if _encoded_query_params: ws_url = ws_url + "?" + urllib.parse.urlencode(_encoded_query_params) headers = self._client_wrapper.get_headers() if authorization is not None: headers["Authorization"] = str(authorization) 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 V2SocketClient(websocket=protocol) except InvalidWebSocketStatus as exc: status_code: int = get_status_code(exc) 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 AsyncRawV2Client: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper @asynccontextmanager async def connect( self, *, model: SpeakV2Model, encoding: typing.Optional[SpeakV2Encoding] = None, sample_rate: typing.Optional[SpeakV2SampleRate] = None, speed: typing.Optional[SpeakV2Speed] = None, expressivity: typing.Optional[SpeakV2Expressivity] = None, mip_opt_out: typing.Optional[SpeakV2MipOptOut] = None, tag: typing.Optional[SpeakV2Tag] = None, authorization: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[AsyncV2SocketClient]: """ Streaming, turn-based text-to-speech (Flux TTS) built for voice-agent pipelines. Stream LLM tokens in, speak them to the user, and report per-turn billing and timing. Parameters ---------- model : SpeakV2Model encoding : typing.Optional[SpeakV2Encoding] sample_rate : typing.Optional[SpeakV2SampleRate] speed : typing.Optional[SpeakV2Speed] expressivity : typing.Optional[SpeakV2Expressivity] mip_opt_out : typing.Optional[SpeakV2MipOptOut] tag : typing.Optional[SpeakV2Tag] authorization : typing.Optional[str] Use your API key or a [temporary token](/guides/fundamentals/token-based-authentication) for authentication via the `Authorization` header. In client-side environments where custom headers are not supported, use the [`Sec-WebSocket-Protocol`](/docs/using-the-sec-websocket-protocol) header instead. **Example:** `Authorization: Token %DEEPGRAM_API_KEY%` or `Authorization: Bearer %DEEPGRAM_TOKEN%` request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncV2SocketClient """ ws_url = self._client_wrapper.get_environment().production + "/v2/speak" _encoded_query_params = encode_query( jsonable_encoder( remove_none_from_dict( { "model": model, "encoding": encoding, "sample_rate": sample_rate, "speed": speed, "expressivity": expressivity, "mip_opt_out": mip_opt_out, "tag": tag, **( request_options.get("additional_query_parameters", {}) or {} if request_options is not None else {} ), } ) ) ) if _encoded_query_params: ws_url = ws_url + "?" + urllib.parse.urlencode(_encoded_query_params) headers = self._client_wrapper.get_headers() if authorization is not None: headers["Authorization"] = str(authorization) 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 AsyncV2SocketClient(websocket=protocol) except InvalidWebSocketStatus as exc: status_code: int = get_status_code(exc) 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.", )