client.py

# This file was auto-generated by Fern from our API Definition.

import typing
from ..core.client_wrapper import SyncClientWrapper
from .requests.tts_request_voice_specifier import TtsRequestVoiceSpecifierParams
from .requests.output_format import OutputFormatParams
from .types.supported_language import SupportedLanguage
from .requests.generation_config import GenerationConfigParams
from .types.model_speed import ModelSpeed
from ..core.request_options import RequestOptions
from ..core.serialization import convert_and_respect_annotation_metadata
from json.decoder import JSONDecodeError
from ..core.api_error import ApiError
from .requests.sse_output_format import SseOutputFormatParams
from .types.context_id import ContextId
from .types.web_socket_response import WebSocketResponse
import httpx_sse
from ..core.pydantic_utilities import parse_obj_as
import json
from ..core.client_wrapper import AsyncClientWrapper

# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)


class TtsClient:
    def __init__(self, *, client_wrapper: SyncClientWrapper):
        self._client_wrapper = client_wrapper

    def bytes(
        self,
        *,
        model_id: str,
        transcript: str,
        voice: TtsRequestVoiceSpecifierParams,
        output_format: OutputFormatParams,
        language: typing.Optional[SupportedLanguage] = OMIT,
        generation_config: typing.Optional[GenerationConfigParams] = OMIT,
        duration: typing.Optional[float] = OMIT,
        speed: typing.Optional[ModelSpeed] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
        pronunciation_dict_id: typing.Optional[str] = None
    ) -> typing.Iterator[bytes]:
        """
        Parameters
        ----------
        model_id : str
            The ID of the model to use for the generation. See [Models](/build-with-cartesia/models) for available models.

        transcript : str

        voice : TtsRequestVoiceSpecifierParams

        output_format : OutputFormatParams

        language : typing.Optional[SupportedLanguage]

        generation_config : typing.Optional[GenerationConfigParams]

        duration : typing.Optional[float]
            The maximum duration of the audio in seconds. You do not usually need to specify this.
            If the duration is not appropriate for the length of the transcript, the output audio may be truncated.

        speed : typing.Optional[ModelSpeed]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response.

        Yields
        ------
        typing.Iterator[bytes]

        Examples
        --------
        from cartesia import Cartesia

        client = Cartesia(
            api_key="YOUR_API_KEY",
        )
        client.tts.bytes(
            model_id="sonic-2",
            transcript="Hello, world!",
            voice={"mode": "id", "id": "694f9389-aac1-45b6-b726-9d9369183238"},
            language="en",
            output_format={
                "sample_rate": 44100,
                "bit_rate": 128000,
                "container": "mp3",
            },
        )
        """
        with self._client_wrapper.httpx_client.stream(
            "tts/bytes",
            method="POST",
            json={
                "model_id": model_id,
                "transcript": transcript,
                "voice": convert_and_respect_annotation_metadata(
                    object_=voice, annotation=TtsRequestVoiceSpecifierParams, direction="write"
                ),
                "language": language,
                "output_format": convert_and_respect_annotation_metadata(
                    object_=output_format, annotation=OutputFormatParams, direction="write"
                ),
                "generation_config": convert_and_respect_annotation_metadata(
                    object_=generation_config, annotation=GenerationConfigParams, direction="write"
                ),
                "duration": duration,
                "speed": speed,
            },
            request_options=request_options,
            omit=OMIT,
        ) as _response:
            try:
                if 200 <= _response.status_code < 300:
                    _chunk_size = request_options.get("chunk_size", None) if request_options is not None else None
                    for _chunk in _response.iter_bytes(chunk_size=_chunk_size):
                        yield _chunk
                    return
                _response.read()
                _response_json = _response.json()
            except JSONDecodeError:
                raise ApiError(status_code=_response.status_code, body=_response.text)
            raise ApiError(status_code=_response.status_code, body=_response_json)

    def sse(
        self,
        *,
        model_id: str,
        transcript: str,
        voice: TtsRequestVoiceSpecifierParams,
        output_format: SseOutputFormatParams,
        language: typing.Optional[SupportedLanguage] = OMIT,
        generation_config: typing.Optional[GenerationConfigParams] = OMIT,
        duration: typing.Optional[float] = OMIT,
        speed: typing.Optional[ModelSpeed] = OMIT,
        add_timestamps: typing.Optional[bool] = OMIT,
        add_phoneme_timestamps: typing.Optional[bool] = OMIT,
        use_normalized_timestamps: typing.Optional[bool] = OMIT,
        context_id: typing.Optional[ContextId] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> typing.Iterator[WebSocketResponse]:
        """
        Parameters
        ----------
        model_id : str
            The ID of the model to use for the generation. See [Models](/build-with-cartesia/models) for available models.

        transcript : str

        voice : TtsRequestVoiceSpecifierParams

        output_format : SseOutputFormatParams

        language : typing.Optional[SupportedLanguage]

        generation_config : typing.Optional[GenerationConfigParams]

        duration : typing.Optional[float]
            The maximum duration of the audio in seconds. You do not usually need to specify this.
            If the duration is not appropriate for the length of the transcript, the output audio may be truncated.

        speed : typing.Optional[ModelSpeed]

        add_timestamps : typing.Optional[bool]
            Whether to return word-level timestamps. If `false` (default), no word timestamps will be produced at all. If `true`, the server will return timestamp events containing word-level timing information.

        add_phoneme_timestamps : typing.Optional[bool]
            Whether to return phoneme-level timestamps. If `false` (default), no phoneme timestamps will be produced - if `add_timestamps` is `true`, the produced timestamps will be word timestamps instead. If `true`, the server will return timestamp events containing phoneme-level timing information.

        use_normalized_timestamps : typing.Optional[bool]
            Whether to use normalized timestamps (True) or original timestamps (False).

        context_id : typing.Optional[ContextId]
            Optional context ID for this request.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Yields
        ------
        typing.Iterator[WebSocketResponse]

        Examples
        --------
        from cartesia import Cartesia

        client = Cartesia(
            api_key="YOUR_API_KEY",
        )
        response = client.tts.sse(
            model_id="sonic-2",
            transcript="Hello, world!",
            voice={"mode": "id", "id": "694f9389-aac1-45b6-b726-9d9369183238"},
            language="en",
            output_format={
                "container": "raw",
                "sample_rate": 44100,
                "encoding": "pcm_f32le",
            },
        )
        for chunk in response:
            yield chunk
        """
        with self._client_wrapper.httpx_client.stream(
            "tts/sse",
            method="POST",
            json={
                "model_id": model_id,
                "transcript": transcript,
                "voice": convert_and_respect_annotation_metadata(
                    object_=voice, annotation=TtsRequestVoiceSpecifierParams, direction="write"
                ),
                "language": language,
                "output_format": convert_and_respect_annotation_metadata(
                    object_=output_format, annotation=SseOutputFormatParams, direction="write"
                ),
                "generation_config": convert_and_respect_annotation_metadata(
                    object_=generation_config, annotation=GenerationConfigParams, direction="write"
                ),
                "duration": duration,
                "speed": speed,
                "add_timestamps": add_timestamps,
                "add_phoneme_timestamps": add_phoneme_timestamps,
                "use_normalized_timestamps": use_normalized_timestamps,
                "context_id": context_id,
            },
            request_options=request_options,
            omit=OMIT,
        ) as _response:
            try:
                if 200 <= _response.status_code < 300:
                    _event_source = httpx_sse.EventSource(_response)
                    for _sse in _event_source.iter_sse():
                        try:
                            yield typing.cast(
                                WebSocketResponse,
                                parse_obj_as(
                                    type_=WebSocketResponse,  # type: ignore
                                    object_=json.loads(_sse.data),
                                ),
                            )
                        except:
                            pass
                    return
                _response.read()
                _response_json = _response.json()
            except JSONDecodeError:
                raise ApiError(status_code=_response.status_code, body=_response.text)
            raise ApiError(status_code=_response.status_code, body=_response_json)


class AsyncTtsClient:
    def __init__(self, *, client_wrapper: AsyncClientWrapper):
        self._client_wrapper = client_wrapper

    async def bytes(
        self,
        *,
        model_id: str,
        transcript: str,
        voice: TtsRequestVoiceSpecifierParams,
        output_format: OutputFormatParams,
        language: typing.Optional[SupportedLanguage] = OMIT,
        generation_config: typing.Optional[GenerationConfigParams] = OMIT,
        duration: typing.Optional[float] = OMIT,
        speed: typing.Optional[ModelSpeed] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> typing.AsyncIterator[bytes]:
        """
        Parameters
        ----------
        model_id : str
            The ID of the model to use for the generation. See [Models](/build-with-cartesia/models) for available models.

        transcript : str

        voice : TtsRequestVoiceSpecifierParams

        output_format : OutputFormatParams

        language : typing.Optional[SupportedLanguage]

        generation_config : typing.Optional[GenerationConfigParams]

        duration : typing.Optional[float]
            The maximum duration of the audio in seconds. You do not usually need to specify this.
            If the duration is not appropriate for the length of the transcript, the output audio may be truncated.

        speed : typing.Optional[ModelSpeed]

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration. You can pass in configuration such as `chunk_size`, and more to customize the request and response.

        Yields
        ------
        typing.AsyncIterator[bytes]

        Examples
        --------
        import asyncio

        from cartesia import AsyncCartesia

        client = AsyncCartesia(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            await client.tts.bytes(
                model_id="sonic-2",
                transcript="Hello, world!",
                voice={"mode": "id", "id": "694f9389-aac1-45b6-b726-9d9369183238"},
                language="en",
                output_format={
                    "sample_rate": 44100,
                    "bit_rate": 128000,
                    "container": "mp3",
                },
            )


        asyncio.run(main())
        """
        async with self._client_wrapper.httpx_client.stream(
            "tts/bytes",
            method="POST",
            json={
                "model_id": model_id,
                "transcript": transcript,
                "voice": convert_and_respect_annotation_metadata(
                    object_=voice, annotation=TtsRequestVoiceSpecifierParams, direction="write"
                ),
                "language": language,
                "output_format": convert_and_respect_annotation_metadata(
                    object_=output_format, annotation=OutputFormatParams, direction="write"
                ),
                "generation_config": convert_and_respect_annotation_metadata(
                    object_=generation_config, annotation=GenerationConfigParams, direction="write"
                ),
                "duration": duration,
                "speed": speed,
            },
            request_options=request_options,
            omit=OMIT,
        ) as _response:
            try:
                if 200 <= _response.status_code < 300:
                    _chunk_size = request_options.get("chunk_size", None) if request_options is not None else None
                    async for _chunk in _response.aiter_bytes(chunk_size=_chunk_size):
                        yield _chunk
                    return
                await _response.aread()
                _response_json = _response.json()
            except JSONDecodeError:
                raise ApiError(status_code=_response.status_code, body=_response.text)
            raise ApiError(status_code=_response.status_code, body=_response_json)

    async def sse(
        self,
        *,
        model_id: str,
        transcript: str,
        voice: TtsRequestVoiceSpecifierParams,
        output_format: SseOutputFormatParams,
        language: typing.Optional[SupportedLanguage] = OMIT,
        generation_config: typing.Optional[GenerationConfigParams] = OMIT,
        duration: typing.Optional[float] = OMIT,
        speed: typing.Optional[ModelSpeed] = OMIT,
        add_timestamps: typing.Optional[bool] = OMIT,
        add_phoneme_timestamps: typing.Optional[bool] = OMIT,
        use_normalized_timestamps: typing.Optional[bool] = OMIT,
        context_id: typing.Optional[ContextId] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> typing.AsyncIterator[WebSocketResponse]:
        """
        Parameters
        ----------
        model_id : str
            The ID of the model to use for the generation. See [Models](/build-with-cartesia/models) for available models.

        transcript : str

        voice : TtsRequestVoiceSpecifierParams

        output_format : SseOutputFormatParams

        language : typing.Optional[SupportedLanguage]

        generation_config : typing.Optional[GenerationConfigParams]

        duration : typing.Optional[float]
            The maximum duration of the audio in seconds. You do not usually need to specify this.
            If the duration is not appropriate for the length of the transcript, the output audio may be truncated.

        speed : typing.Optional[ModelSpeed]

        add_timestamps : typing.Optional[bool]
            Whether to return word-level timestamps. If `false` (default), no word timestamps will be produced at all. If `true`, the server will return timestamp events containing word-level timing information.

        add_phoneme_timestamps : typing.Optional[bool]
            Whether to return phoneme-level timestamps. If `false` (default), no phoneme timestamps will be produced - if `add_timestamps` is `true`, the produced timestamps will be word timestamps instead. If `true`, the server will return timestamp events containing phoneme-level timing information.

        use_normalized_timestamps : typing.Optional[bool]
            Whether to use normalized timestamps (True) or original timestamps (False).

        context_id : typing.Optional[ContextId]
            Optional context ID for this request.

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Yields
        ------
        typing.AsyncIterator[WebSocketResponse]

        Examples
        --------
        import asyncio

        from cartesia import AsyncCartesia

        client = AsyncCartesia(
            api_key="YOUR_API_KEY",
        )


        async def main() -> None:
            response = await client.tts.sse(
                model_id="sonic-2",
                transcript="Hello, world!",
                voice={"mode": "id", "id": "694f9389-aac1-45b6-b726-9d9369183238"},
                language="en",
                output_format={
                    "container": "raw",
                    "sample_rate": 44100,
                    "encoding": "pcm_f32le",
                },
            )
            async for chunk in response:
                yield chunk


        asyncio.run(main())
        """
        async with self._client_wrapper.httpx_client.stream(
            "tts/sse",
            method="POST",
            json={
                "model_id": model_id,
                "transcript": transcript,
                "voice": convert_and_respect_annotation_metadata(
                    object_=voice, annotation=TtsRequestVoiceSpecifierParams, direction="write"
                ),
                "language": language,
                "output_format": convert_and_respect_annotation_metadata(
                    object_=output_format, annotation=SseOutputFormatParams, direction="write"
                ),
                "generation_config": convert_and_respect_annotation_metadata(
                    object_=generation_config, annotation=GenerationConfigParams, direction="write"
                ),
                "duration": duration,
                "speed": speed,
                "add_timestamps": add_timestamps,
                "add_phoneme_timestamps": add_phoneme_timestamps,
                "use_normalized_timestamps": use_normalized_timestamps,
                "context_id": context_id,
            },
            request_options=request_options,
            omit=OMIT,
        ) as _response:
            try:
                if 200 <= _response.status_code < 300:
                    _event_source = httpx_sse.EventSource(_response)
                    async for _sse in _event_source.aiter_sse():
                        try:
                            yield typing.cast(
                                WebSocketResponse,
                                parse_obj_as(
                                    type_=WebSocketResponse,  # type: ignore
                                    object_=json.loads(_sse.data),
                                ),
                            )
                        except:
                            pass
                    return
                await _response.aread()
                _response_json = _response.json()
            except JSONDecodeError:
                raise ApiError(status_code=_response.status_code, body=_response.text)
            raise ApiError(status_code=_response.status_code, body=_response_json)