# 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 AsyncRawTextToSpeechStreamingClient, RawTextToSpeechStreamingClient from .socket_client import AsyncTextToSpeechStreamingSocketClient, TextToSpeechStreamingSocketClient from .types.text_to_speech_streaming_model import TextToSpeechStreamingModel from .types.text_to_speech_streaming_send_completion_event import TextToSpeechStreamingSendCompletionEvent 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 TextToSpeechStreamingClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawTextToSpeechStreamingClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawTextToSpeechStreamingClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawTextToSpeechStreamingClient """ return self._raw_client @contextmanager def connect( self, *, model: typing.Optional[TextToSpeechStreamingModel] = None, send_completion_event: typing.Optional[TextToSpeechStreamingSendCompletionEvent] = None, api_subscription_key: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Iterator[TextToSpeechStreamingSocketClient]: """ WebSocket channel for real-time TTS synthesis. **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. **Model-Specific Notes:** - **bulbul:v2:** Supports pitch, loudness, pace (0.3-3.0). Default sample rate: 22050 Hz. - **bulbul:v3-beta:** Does NOT support pitch/loudness. Pace range: 0.5-2.0. Supports temperature parameter. Default sample rate: 24000 Hz. Preprocessing is always enabled. Parameters ---------- model : typing.Optional[TextToSpeechStreamingModel] Text to speech model to use. - **bulbul:v2** (default): Standard TTS model with pitch/loudness support - **bulbul:v3-beta**: Advanced model with temperature control (no pitch/loudness) send_completion_event : typing.Optional[TextToSpeechStreamingSendCompletionEvent] Enable completion event notifications when TTS generation finishes. When set to true, an event message will be sent when the final audio chunk has been generated. api_subscription_key : typing.Optional[str] API subscription key for authentication request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- TextToSpeechStreamingSocketClient """ ws_url = self._raw_client._client_wrapper.get_environment().production + "/text-to-speech/ws" query_params = httpx.QueryParams() if model is not None: query_params = query_params.add("model", model) if send_completion_event is not None: query_params = query_params.add("send_completion_event", send_completion_event) 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 TextToSpeechStreamingSocketClient(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 AsyncTextToSpeechStreamingClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawTextToSpeechStreamingClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> AsyncRawTextToSpeechStreamingClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawTextToSpeechStreamingClient """ return self._raw_client @asynccontextmanager async def connect( self, *, model: typing.Optional[TextToSpeechStreamingModel] = None, send_completion_event: typing.Optional[TextToSpeechStreamingSendCompletionEvent] = None, api_subscription_key: typing.Optional[str] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.AsyncIterator[AsyncTextToSpeechStreamingSocketClient]: """ WebSocket channel for real-time TTS synthesis. **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. **Model-Specific Notes:** - **bulbul:v2:** Supports pitch, loudness, pace (0.3-3.0). Default sample rate: 22050 Hz. - **bulbul:v3-beta:** Does NOT support pitch/loudness. Pace range: 0.5-2.0. Supports temperature parameter. Default sample rate: 24000 Hz. Preprocessing is always enabled. Parameters ---------- model : typing.Optional[TextToSpeechStreamingModel] Text to speech model to use. - **bulbul:v2** (default): Standard TTS model with pitch/loudness support - **bulbul:v3-beta**: Advanced model with temperature control (no pitch/loudness) send_completion_event : typing.Optional[TextToSpeechStreamingSendCompletionEvent] Enable completion event notifications when TTS generation finishes. When set to true, an event message will be sent when the final audio chunk has been generated. api_subscription_key : typing.Optional[str] API subscription key for authentication request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncTextToSpeechStreamingSocketClient """ ws_url = self._raw_client._client_wrapper.get_environment().production + "/text-to-speech/ws" query_params = httpx.QueryParams() if model is not None: query_params = query_params.add("model", model) if send_completion_event is not None: query_params = query_params.add("send_completion_event", send_completion_event) 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 AsyncTextToSpeechStreamingSocketClient(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.", )