# This file was auto-generated by Fern from our API Definition. import datetime as dt import typing from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ..core.request_options import RequestOptions from ..types.config_stream import ConfigStream from ..types.create_stream_out import CreateStreamOut from ..types.get_probe_stream_out import GetProbeStreamOut from ..types.source_stream import SourceStream from ..types.target_stream import TargetStream from .raw_client import AsyncRawDeprecatedStreamingClient, RawDeprecatedStreamingClient # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class DeprecatedStreamingClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._raw_client = RawDeprecatedStreamingClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> RawDeprecatedStreamingClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- RawDeprecatedStreamingClient """ return self._raw_client def create_stream( self, *, voices: typing.Sequence[int], dictionaries: typing.Sequence[int], config: ConfigStream, source_stream: SourceStream, target_streams: typing.Sequence[TargetStream], name: typing.Optional[str] = OMIT, description: typing.Optional[str] = OMIT, initial_delay: typing.Optional[int] = OMIT, timeout_in_mins: typing.Optional[int] = OMIT, start_time: typing.Optional[dt.datetime] = OMIT, end_time: typing.Optional[dt.datetime] = OMIT, timezone: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> CreateStreamOut: """ Parameters ---------- voices : typing.Sequence[int] List of voice identifiers to be used in the stream. dictionaries : typing.Sequence[int] List of dictionary identifiers to be used in the stream. config : ConfigStream The shared configuration for the streaming pipeline. source_stream : SourceStream The source stream configuration details. target_streams : typing.Sequence[TargetStream] List of target stream configurations. name : typing.Optional[str] The name of the stream. description : typing.Optional[str] The description of the stream. initial_delay : typing.Optional[int] The initial delay in seconds before starting the stream creation process. timeout_in_mins : typing.Optional[int] The maximum duration in minutes for the stream creation process before timing out. start_time : typing.Optional[dt.datetime] end_time : typing.Optional[dt.datetime] timezone : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- CreateStreamOut Successful Response Examples -------- from camb import CambApi, ConfigStream, SourceStream, TargetStream client = CambApi( api_key="YOUR_API_KEY", ) client.deprecated_streaming.create_stream( voices=[1], dictionaries=[1], config=ConfigStream(), source_stream=SourceStream( language=1, url="url", ), target_streams=[ TargetStream( languages=[1], url="url", type=1, ) ], ) """ _response = self._raw_client.create_stream( voices=voices, dictionaries=dictionaries, config=config, source_stream=source_stream, target_streams=target_streams, name=name, description=description, initial_delay=initial_delay, timeout_in_mins=timeout_in_mins, start_time=start_time, end_time=end_time, timezone=timezone, request_options=request_options, ) return _response.data def get_stream_result( self, stream_id: int, *, run_id: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Dict[str, typing.Any]: """ Parameters ---------- stream_id : int run_id : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Dict[str, typing.Any] Successful Response Examples -------- from camb import CambApi client = CambApi( api_key="YOUR_API_KEY", ) client.deprecated_streaming.get_stream_result( stream_id=1, ) """ _response = self._raw_client.get_stream_result(stream_id, run_id=run_id, request_options=request_options) return _response.data def stop_stream( self, stream_id: int, *, run_id: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Dict[str, typing.Any]: """ Parameters ---------- stream_id : int run_id : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Dict[str, typing.Any] Successful Response Examples -------- from camb import CambApi client = CambApi( api_key="YOUR_API_KEY", ) client.deprecated_streaming.stop_stream( stream_id=1, ) """ _response = self._raw_client.stop_stream(stream_id, run_id=run_id, request_options=request_options) return _response.data def get_probe_stream( self, *, url: str, run_id: typing.Optional[int] = None, traceparent: typing.Optional[str] = None, passphrase: typing.Optional[str] = OMIT, stream_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> GetProbeStreamOut: """ Parameters ---------- url : str run_id : typing.Optional[int] traceparent : typing.Optional[str] passphrase : typing.Optional[str] stream_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- GetProbeStreamOut Successful Response Examples -------- from camb import CambApi client = CambApi( api_key="YOUR_API_KEY", ) client.deprecated_streaming.get_probe_stream( url="url", ) """ _response = self._raw_client.get_probe_stream( url=url, run_id=run_id, traceparent=traceparent, passphrase=passphrase, stream_id=stream_id, request_options=request_options, ) return _response.data class AsyncDeprecatedStreamingClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._raw_client = AsyncRawDeprecatedStreamingClient(client_wrapper=client_wrapper) @property def with_raw_response(self) -> AsyncRawDeprecatedStreamingClient: """ Retrieves a raw implementation of this client that returns raw responses. Returns ------- AsyncRawDeprecatedStreamingClient """ return self._raw_client async def create_stream( self, *, voices: typing.Sequence[int], dictionaries: typing.Sequence[int], config: ConfigStream, source_stream: SourceStream, target_streams: typing.Sequence[TargetStream], name: typing.Optional[str] = OMIT, description: typing.Optional[str] = OMIT, initial_delay: typing.Optional[int] = OMIT, timeout_in_mins: typing.Optional[int] = OMIT, start_time: typing.Optional[dt.datetime] = OMIT, end_time: typing.Optional[dt.datetime] = OMIT, timezone: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> CreateStreamOut: """ Parameters ---------- voices : typing.Sequence[int] List of voice identifiers to be used in the stream. dictionaries : typing.Sequence[int] List of dictionary identifiers to be used in the stream. config : ConfigStream The shared configuration for the streaming pipeline. source_stream : SourceStream The source stream configuration details. target_streams : typing.Sequence[TargetStream] List of target stream configurations. name : typing.Optional[str] The name of the stream. description : typing.Optional[str] The description of the stream. initial_delay : typing.Optional[int] The initial delay in seconds before starting the stream creation process. timeout_in_mins : typing.Optional[int] The maximum duration in minutes for the stream creation process before timing out. start_time : typing.Optional[dt.datetime] end_time : typing.Optional[dt.datetime] timezone : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- CreateStreamOut Successful Response Examples -------- import asyncio from camb import AsyncCambApi, ConfigStream, SourceStream, TargetStream client = AsyncCambApi( api_key="YOUR_API_KEY", ) async def main() -> None: await client.deprecated_streaming.create_stream( voices=[1], dictionaries=[1], config=ConfigStream(), source_stream=SourceStream( language=1, url="url", ), target_streams=[ TargetStream( languages=[1], url="url", type=1, ) ], ) asyncio.run(main()) """ _response = await self._raw_client.create_stream( voices=voices, dictionaries=dictionaries, config=config, source_stream=source_stream, target_streams=target_streams, name=name, description=description, initial_delay=initial_delay, timeout_in_mins=timeout_in_mins, start_time=start_time, end_time=end_time, timezone=timezone, request_options=request_options, ) return _response.data async def get_stream_result( self, stream_id: int, *, run_id: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Dict[str, typing.Any]: """ Parameters ---------- stream_id : int run_id : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Dict[str, typing.Any] Successful Response Examples -------- import asyncio from camb import AsyncCambApi client = AsyncCambApi( api_key="YOUR_API_KEY", ) async def main() -> None: await client.deprecated_streaming.get_stream_result( stream_id=1, ) asyncio.run(main()) """ _response = await self._raw_client.get_stream_result(stream_id, run_id=run_id, request_options=request_options) return _response.data async def stop_stream( self, stream_id: int, *, run_id: typing.Optional[int] = None, request_options: typing.Optional[RequestOptions] = None, ) -> typing.Dict[str, typing.Any]: """ Parameters ---------- stream_id : int run_id : typing.Optional[int] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- typing.Dict[str, typing.Any] Successful Response Examples -------- import asyncio from camb import AsyncCambApi client = AsyncCambApi( api_key="YOUR_API_KEY", ) async def main() -> None: await client.deprecated_streaming.stop_stream( stream_id=1, ) asyncio.run(main()) """ _response = await self._raw_client.stop_stream(stream_id, run_id=run_id, request_options=request_options) return _response.data async def get_probe_stream( self, *, url: str, run_id: typing.Optional[int] = None, traceparent: typing.Optional[str] = None, passphrase: typing.Optional[str] = OMIT, stream_id: typing.Optional[str] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> GetProbeStreamOut: """ Parameters ---------- url : str run_id : typing.Optional[int] traceparent : typing.Optional[str] passphrase : typing.Optional[str] stream_id : typing.Optional[str] request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- GetProbeStreamOut Successful Response Examples -------- import asyncio from camb import AsyncCambApi client = AsyncCambApi( api_key="YOUR_API_KEY", ) async def main() -> None: await client.deprecated_streaming.get_probe_stream( url="url", ) asyncio.run(main()) """ _response = await self._raw_client.get_probe_stream( url=url, run_id=run_id, traceparent=traceparent, passphrase=passphrase, stream_id=stream_id, request_options=request_options, ) return _response.data