# This file was auto-generated by Fern from our API Definition. from __future__ import annotations import os import typing import httpx from .core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from .core.request_options import RequestOptions from .environment import CambApiEnvironment from .types.tts_provider import TtsProvider from .raw_client import AsyncRawCambApi, RawCambApi if typing.TYPE_CHECKING: from .audio_separation.client import AsyncAudioSeparationClient, AudioSeparationClient from .deprecated_streaming.client import AsyncDeprecatedStreamingClient, DeprecatedStreamingClient from .dictionaries.client import AsyncDictionariesClient, DictionariesClient from .dub.client import AsyncDubClient, DubClient from .folders.client import AsyncFoldersClient, FoldersClient from .languages.client import AsyncLanguagesClient, LanguagesClient from .project_setup.client import AsyncProjectSetupClient, ProjectSetupClient from .story.client import AsyncStoryClient, StoryClient from .streaming.client import AsyncStreamingClient, StreamingClient from .subtitles.client import AsyncSubtitlesClient, SubtitlesClient from .text_to_audio.client import AsyncTextToAudioClient, TextToAudioClient from .text_to_speech.client import AsyncTextToSpeechClient, TextToSpeechClient from .text_to_voice.client import AsyncTextToVoiceClient, TextToVoiceClient from .transcription.client import AsyncTranscriptionClient, TranscriptionClient from .translated_story.client import AsyncTranslatedStoryClient, TranslatedStoryClient from .translated_tts.client import AsyncTranslatedTtsClient, TranslatedTtsClient from .translation.client import AsyncTranslationClient, TranslationClient from .voice_cloning.client import AsyncVoiceCloningClient, VoiceCloningClient DEFAULT_TIMEOUT = 300 def save_stream_to_file(stream: typing.Iterable[bytes], filename: str) -> None: """Saves a byte stream to a file. Parameters ---------- stream : typing.Iterable[bytes] The byte stream to save. filename : str The name of the file to save the stream to. """ with open(filename, "wb") as f: for chunk in stream: f.write(chunk) async def save_async_stream_to_file(stream: typing.AsyncIterable[bytes], filename: str) -> None: """Saves an async byte stream to a file. Parameters ---------- stream : typing.AsyncIterable[bytes] The async byte stream to save. filename : str The name of the file to save the stream to. """ with open(filename, "wb") as f: async for chunk in stream: f.write(chunk) class CambAI: """ Use this class to access the different functions within the SDK. You can instantiate any number of clients with different configuration that will propagate to these functions. Parameters ---------- base_url : typing.Optional[str] The base url to use for requests from the client. environment : CambApiEnvironment The environment to use for requests from the client. from .environment import CambApiEnvironment Defaults to CambApiEnvironment.DEFAULT api_key : typing.Optional[str] Your Camb AI API key. If not provided, the ``CAMB_API_KEY`` environment variable is used. headers : typing.Optional[typing.Dict[str, str]] Additional headers to send with every request. timeout : typing.Optional[float] The timeout to be used, in seconds, for requests. By default the timeout is 300 seconds, unless a custom httpx client is used, in which case this default is not enforced. follow_redirects : typing.Optional[bool] Whether the default httpx client follows redirects or not, this is irrelevant if a custom httpx client is passed in. httpx_client : typing.Optional[httpx.Client] The httpx client to use for making requests, a preconfigured client is used by default, however this is useful should you want to pass in any custom httpx configuration. Examples -------- from camb import CambApi client = CambApi( api_key="YOUR_API_KEY", ) """ def __init__( self, *, base_url: typing.Optional[str] = None, environment: CambApiEnvironment = CambApiEnvironment.DEFAULT, api_key: typing.Optional[str] = None, headers: typing.Optional[typing.Dict[str, str]] = None, timeout: typing.Optional[float] = None, follow_redirects: typing.Optional[bool] = True, httpx_client: typing.Optional[httpx.Client] = None, tts_provider: typing.Optional[TtsProvider] = None, provider_params: typing.Optional[typing.Dict[str, typing.Any]] = None, ): if api_key is None: api_key = os.environ.get("CAMB_API_KEY") or None if api_key is None and (tts_provider is None or provider_params is None): raise ValueError( "Please provide 'api_key', set the CAMB_API_KEY environment variable, " "or provide both 'tts_provider' and 'provider_params'." ) _defaulted_timeout = ( timeout if timeout is not None else DEFAULT_TIMEOUT if httpx_client is None else httpx_client.timeout.read ) self._client_wrapper = SyncClientWrapper( base_url=_get_base_url(base_url=base_url, environment=environment), api_key=api_key, headers=headers, httpx_client=httpx_client if httpx_client is not None else httpx.Client(timeout=_defaulted_timeout, follow_redirects=follow_redirects) if follow_redirects is not None else httpx.Client(timeout=_defaulted_timeout), timeout=_defaulted_timeout, tts_provider=tts_provider, provider_params=provider_params, ) self._raw_client = RawCambApi(client_wrapper=self._client_wrapper) self._audio_separation: typing.Optional[AudioSeparationClient] = None self._dub: typing.Optional[DubClient] = None self._folders: typing.Optional[FoldersClient] = None self._languages: typing.Optional[LanguagesClient] = None self._story: typing.Optional[StoryClient] = None self._subtitles: typing.Optional[SubtitlesClient] = None self._translated_story: typing.Optional[TranslatedStoryClient] = None self._text_to_audio: typing.Optional[TextToAudioClient] = None self._text_to_voice: typing.Optional[TextToVoiceClient] = None self._text_to_speech: typing.Optional[TextToSpeechClient] = None self._translation: typing.Optional[TranslationClient] = None self._transcription: typing.Optional[TranscriptionClient] = None self._translated_tts: typing.Optional[TranslatedTtsClient] = None self._streaming: typing.Optional[StreamingClient] = None self._voice_cloning: typing.Optional[VoiceCloningClient] = None self._dictionaries: typing.Optional[DictionariesClient] = None self._project_setup: typing.Optional[ProjectSetupClient] = None self._deprecated_streaming: typing.Optional[DeprecatedStreamingClient] = None self._live_transcription: typing.Optional[typing.Any] = None self._realtime: typing.Optional[typing.Any] = None @property def with_raw_response(self) -> RawCambApi: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawCambApi """ return self._raw_client def get_swagger_docs_docs_get(self, *, request_options: typing.Optional[RequestOptions] = None) -> typing.Any: """ Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Any Successful Response Examples -------- from camb import CambApi client = CambApi( api_key="YOUR_API_KEY", ) client.get_swagger_docs_docs_get() """ _response = self._raw_client.get_swagger_docs_docs_get(request_options=request_options) return _response.data def get_redoc_docs_redocs_get(self, *, request_options: typing.Optional[RequestOptions] = None) -> typing.Any: """ Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Any Successful Response Examples -------- from camb import CambApi client = CambApi( api_key="YOUR_API_KEY", ) client.get_redoc_docs_redocs_get() """ _response = self._raw_client.get_redoc_docs_redocs_get(request_options=request_options) return _response.data def get_openapi_schema_openapi_json_get( self, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.Any: """ Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Any Successful Response Examples -------- from camb import CambApi client = CambApi( api_key="YOUR_API_KEY", ) client.get_openapi_schema_openapi_json_get() """ _response = self._raw_client.get_openapi_schema_openapi_json_get(request_options=request_options) return _response.data @property def audio_separation(self): if self._audio_separation is None: from .audio_separation.client import AudioSeparationClient # noqa: E402 self._audio_separation = AudioSeparationClient(client_wrapper=self._client_wrapper) return self._audio_separation @property def dub(self): if self._dub is None: from .dub.client import DubClient # noqa: E402 self._dub = DubClient(client_wrapper=self._client_wrapper) return self._dub @property def folders(self): if self._folders is None: from .folders.client import FoldersClient # noqa: E402 self._folders = FoldersClient(client_wrapper=self._client_wrapper) return self._folders @property def languages(self): if self._languages is None: from .languages.client import LanguagesClient # noqa: E402 self._languages = LanguagesClient(client_wrapper=self._client_wrapper) return self._languages @property def story(self): if self._story is None: from .story.client import StoryClient # noqa: E402 self._story = StoryClient(client_wrapper=self._client_wrapper) return self._story @property def subtitles(self): if self._subtitles is None: from .subtitles.client import SubtitlesClient # noqa: E402 self._subtitles = SubtitlesClient(client_wrapper=self._client_wrapper) return self._subtitles @property def translated_story(self): if self._translated_story is None: from .translated_story.client import TranslatedStoryClient # noqa: E402 self._translated_story = TranslatedStoryClient(client_wrapper=self._client_wrapper) return self._translated_story @property def text_to_audio(self): if self._text_to_audio is None: from .text_to_audio.client import TextToAudioClient # noqa: E402 self._text_to_audio = TextToAudioClient(client_wrapper=self._client_wrapper) return self._text_to_audio @property def text_to_voice(self): if self._text_to_voice is None: from .text_to_voice.client import TextToVoiceClient # noqa: E402 self._text_to_voice = TextToVoiceClient(client_wrapper=self._client_wrapper) return self._text_to_voice @property def text_to_speech(self): if self._text_to_speech is None: from .text_to_speech.client import TextToSpeechClient # noqa: E402 self._text_to_speech = TextToSpeechClient(client_wrapper=self._client_wrapper) return self._text_to_speech @property def translation(self): if self._translation is None: from .translation.client import TranslationClient # noqa: E402 self._translation = TranslationClient(client_wrapper=self._client_wrapper) return self._translation @property def transcription(self): if self._transcription is None: from .transcription.client import TranscriptionClient # noqa: E402 self._transcription = TranscriptionClient(client_wrapper=self._client_wrapper) return self._transcription @property def translated_tts(self): if self._translated_tts is None: from .translated_tts.client import TranslatedTtsClient # noqa: E402 self._translated_tts = TranslatedTtsClient(client_wrapper=self._client_wrapper) return self._translated_tts @property def streaming(self): if self._streaming is None: from .streaming.client import StreamingClient # noqa: E402 self._streaming = StreamingClient(client_wrapper=self._client_wrapper) return self._streaming @property def voice_cloning(self): if self._voice_cloning is None: from .voice_cloning.client import VoiceCloningClient # noqa: E402 self._voice_cloning = VoiceCloningClient(client_wrapper=self._client_wrapper) return self._voice_cloning @property def dictionaries(self): if self._dictionaries is None: from .dictionaries.client import DictionariesClient # noqa: E402 self._dictionaries = DictionariesClient(client_wrapper=self._client_wrapper) return self._dictionaries @property def project_setup(self): if self._project_setup is None: from .project_setup.client import ProjectSetupClient # noqa: E402 self._project_setup = ProjectSetupClient(client_wrapper=self._client_wrapper) return self._project_setup @property def deprecated_streaming(self): if self._deprecated_streaming is None: from .deprecated_streaming.client import DeprecatedStreamingClient # noqa: E402 self._deprecated_streaming = DeprecatedStreamingClient(client_wrapper=self._client_wrapper) return self._deprecated_streaming @property def live_transcription(self): if self._live_transcription is None: from .live_transcription.client import LiveTranscriptionClient # noqa: E402 self._live_transcription = LiveTranscriptionClient(client_wrapper=self._client_wrapper) return self._live_transcription @property def realtime(self): if self._realtime is None: from .realtime.client import RealtimeClient # noqa: E402 self._realtime = RealtimeClient(client_wrapper=self._client_wrapper) return self._realtime class AsyncCambAI: """ Use this class to access the different functions within the SDK. You can instantiate any number of clients with different configuration that will propagate to these functions. Parameters ---------- base_url : typing.Optional[str] The base url to use for requests from the client. environment : CambApiEnvironment The environment to use for requests from the client. from .environment import CambApiEnvironment Defaults to CambApiEnvironment.DEFAULT api_key : typing.Optional[str] Your Camb AI API key. If not provided, the ``CAMB_API_KEY`` environment variable is used. headers : typing.Optional[typing.Dict[str, str]] Additional headers to send with every request. timeout : typing.Optional[float] The timeout to be used, in seconds, for requests. By default the timeout is 300 seconds, unless a custom httpx client is used, in which case this default is not enforced. follow_redirects : typing.Optional[bool] Whether the default httpx client follows redirects or not, this is irrelevant if a custom httpx client is passed in. httpx_client : typing.Optional[httpx.AsyncClient] The httpx client to use for making requests, a preconfigured client is used by default, however this is useful should you want to pass in any custom httpx configuration. Examples -------- from camb import AsyncCambApi client = AsyncCambApi( api_key="YOUR_API_KEY", ) """ def __init__( self, *, base_url: typing.Optional[str] = None, environment: CambApiEnvironment = CambApiEnvironment.DEFAULT, api_key: typing.Optional[str] = None, headers: typing.Optional[typing.Dict[str, str]] = None, timeout: typing.Optional[float] = None, follow_redirects: typing.Optional[bool] = True, httpx_client: typing.Optional[httpx.AsyncClient] = None, tts_provider: typing.Optional[TtsProvider] = None, provider_params: typing.Optional[typing.Dict[str, typing.Any]] = None, ): if api_key is None: api_key = os.environ.get("CAMB_API_KEY") or None if api_key is None and (tts_provider is None or provider_params is None): raise ValueError( "Please provide 'api_key', set the CAMB_API_KEY environment variable, " "or provide both 'tts_provider' and 'provider_params'." ) _defaulted_timeout = ( timeout if timeout is not None else DEFAULT_TIMEOUT if httpx_client is None else httpx_client.timeout.read ) self._client_wrapper = AsyncClientWrapper( base_url=_get_base_url(base_url=base_url, environment=environment), api_key=api_key, headers=headers, httpx_client=httpx_client if httpx_client is not None else httpx.AsyncClient(timeout=_defaulted_timeout, follow_redirects=follow_redirects) if follow_redirects is not None else httpx.AsyncClient(timeout=_defaulted_timeout), timeout=_defaulted_timeout, tts_provider=tts_provider, provider_params=provider_params, ) self._raw_client = AsyncRawCambApi(client_wrapper=self._client_wrapper) self._audio_separation: typing.Optional[AsyncAudioSeparationClient] = None self._dub: typing.Optional[AsyncDubClient] = None self._folders: typing.Optional[AsyncFoldersClient] = None self._languages: typing.Optional[AsyncLanguagesClient] = None self._story: typing.Optional[AsyncStoryClient] = None self._subtitles: typing.Optional[AsyncSubtitlesClient] = None self._translated_story: typing.Optional[AsyncTranslatedStoryClient] = None self._text_to_audio: typing.Optional[AsyncTextToAudioClient] = None self._text_to_voice: typing.Optional[AsyncTextToVoiceClient] = None self._text_to_speech: typing.Optional[AsyncTextToSpeechClient] = None self._translation: typing.Optional[AsyncTranslationClient] = None self._transcription: typing.Optional[AsyncTranscriptionClient] = None self._translated_tts: typing.Optional[AsyncTranslatedTtsClient] = None self._streaming: typing.Optional[AsyncStreamingClient] = None self._voice_cloning: typing.Optional[AsyncVoiceCloningClient] = None self._dictionaries: typing.Optional[AsyncDictionariesClient] = None self._project_setup: typing.Optional[AsyncProjectSetupClient] = None self._deprecated_streaming: typing.Optional[AsyncDeprecatedStreamingClient] = None self._live_transcription: typing.Optional[typing.Any] = None self._realtime: typing.Optional[typing.Any] = None @property def with_raw_response(self) -> AsyncRawCambApi: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawCambApi """ return self._raw_client async def get_swagger_docs_docs_get(self, *, request_options: typing.Optional[RequestOptions] = None) -> typing.Any: """ Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Any Successful Response Examples -------- import asyncio from camb import AsyncCambApi client = AsyncCambApi( api_key="YOUR_API_KEY", ) async def main() -> None: await client.get_swagger_docs_docs_get() asyncio.run(main()) """ _response = await self._raw_client.get_swagger_docs_docs_get(request_options=request_options) return _response.data async def get_redoc_docs_redocs_get(self, *, request_options: typing.Optional[RequestOptions] = None) -> typing.Any: """ Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Any Successful Response Examples -------- import asyncio from camb import AsyncCambApi client = AsyncCambApi( api_key="YOUR_API_KEY", ) async def main() -> None: await client.get_redoc_docs_redocs_get() asyncio.run(main()) """ _response = await self._raw_client.get_redoc_docs_redocs_get(request_options=request_options) return _response.data async def get_openapi_schema_openapi_json_get( self, *, request_options: typing.Optional[RequestOptions] = None ) -> typing.Any: """ Parameters ---------- request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Any Successful Response Examples -------- import asyncio from camb import AsyncCambApi client = AsyncCambApi( api_key="YOUR_API_KEY", ) async def main() -> None: await client.get_openapi_schema_openapi_json_get() asyncio.run(main()) """ _response = await self._raw_client.get_openapi_schema_openapi_json_get(request_options=request_options) return _response.data @property def audio_separation(self): if self._audio_separation is None: from .audio_separation.client import AsyncAudioSeparationClient # noqa: E402 self._audio_separation = AsyncAudioSeparationClient(client_wrapper=self._client_wrapper) return self._audio_separation @property def dub(self): if self._dub is None: from .dub.client import AsyncDubClient # noqa: E402 self._dub = AsyncDubClient(client_wrapper=self._client_wrapper) return self._dub @property def folders(self): if self._folders is None: from .folders.client import AsyncFoldersClient # noqa: E402 self._folders = AsyncFoldersClient(client_wrapper=self._client_wrapper) return self._folders @property def languages(self): if self._languages is None: from .languages.client import AsyncLanguagesClient # noqa: E402 self._languages = AsyncLanguagesClient(client_wrapper=self._client_wrapper) return self._languages @property def story(self): if self._story is None: from .story.client import AsyncStoryClient # noqa: E402 self._story = AsyncStoryClient(client_wrapper=self._client_wrapper) return self._story @property def subtitles(self): if self._subtitles is None: from .subtitles.client import AsyncSubtitlesClient # noqa: E402 self._subtitles = AsyncSubtitlesClient(client_wrapper=self._client_wrapper) return self._subtitles @property def translated_story(self): if self._translated_story is None: from .translated_story.client import AsyncTranslatedStoryClient # noqa: E402 self._translated_story = AsyncTranslatedStoryClient(client_wrapper=self._client_wrapper) return self._translated_story @property def text_to_audio(self): if self._text_to_audio is None: from .text_to_audio.client import AsyncTextToAudioClient # noqa: E402 self._text_to_audio = AsyncTextToAudioClient(client_wrapper=self._client_wrapper) return self._text_to_audio @property def text_to_voice(self): if self._text_to_voice is None: from .text_to_voice.client import AsyncTextToVoiceClient # noqa: E402 self._text_to_voice = AsyncTextToVoiceClient(client_wrapper=self._client_wrapper) return self._text_to_voice @property def text_to_speech(self): if self._text_to_speech is None: from .text_to_speech.client import AsyncTextToSpeechClient # noqa: E402 self._text_to_speech = AsyncTextToSpeechClient(client_wrapper=self._client_wrapper) return self._text_to_speech @property def translation(self): if self._translation is None: from .translation.client import AsyncTranslationClient # noqa: E402 self._translation = AsyncTranslationClient(client_wrapper=self._client_wrapper) return self._translation @property def transcription(self): if self._transcription is None: from .transcription.client import AsyncTranscriptionClient # noqa: E402 self._transcription = AsyncTranscriptionClient(client_wrapper=self._client_wrapper) return self._transcription @property def translated_tts(self): if self._translated_tts is None: from .translated_tts.client import AsyncTranslatedTtsClient # noqa: E402 self._translated_tts = AsyncTranslatedTtsClient(client_wrapper=self._client_wrapper) return self._translated_tts @property def streaming(self): if self._streaming is None: from .streaming.client import AsyncStreamingClient # noqa: E402 self._streaming = AsyncStreamingClient(client_wrapper=self._client_wrapper) return self._streaming @property def voice_cloning(self): if self._voice_cloning is None: from .voice_cloning.client import AsyncVoiceCloningClient # noqa: E402 self._voice_cloning = AsyncVoiceCloningClient(client_wrapper=self._client_wrapper) return self._voice_cloning @property def dictionaries(self): if self._dictionaries is None: from .dictionaries.client import AsyncDictionariesClient # noqa: E402 self._dictionaries = AsyncDictionariesClient(client_wrapper=self._client_wrapper) return self._dictionaries @property def project_setup(self): if self._project_setup is None: from .project_setup.client import AsyncProjectSetupClient # noqa: E402 self._project_setup = AsyncProjectSetupClient(client_wrapper=self._client_wrapper) return self._project_setup @property def deprecated_streaming(self): if self._deprecated_streaming is None: from .deprecated_streaming.client import AsyncDeprecatedStreamingClient # noqa: E402 self._deprecated_streaming = AsyncDeprecatedStreamingClient(client_wrapper=self._client_wrapper) return self._deprecated_streaming @property def live_transcription(self): if self._live_transcription is None: from .live_transcription.client import LiveTranscriptionClient # noqa: E402 # The live transcription client is transport-agnostic; the sync # wrapper carries the same api_key/base_url state we need. self._live_transcription = LiveTranscriptionClient(client_wrapper=self._client_wrapper) return self._live_transcription @property def realtime(self): if self._realtime is None: from .realtime.client import RealtimeClient # noqa: E402 self._realtime = RealtimeClient(client_wrapper=self._client_wrapper) return self._realtime def _get_base_url(*, base_url: typing.Optional[str] = None, environment: CambApiEnvironment) -> str: if base_url is not None: return base_url elif environment is not None: return environment.value else: raise Exception("Please pass in either base_url or environment to construct the client")