# This file was auto-generated by Fern from our API Definition. import typing from json.decoder import JSONDecodeError from ...core.api_error import ApiError from ...core.client_wrapper import AsyncClientWrapper, SyncClientWrapper from ...core.http_response import AsyncHttpResponse, HttpResponse from ...core.jsonable_encoder import jsonable_encoder from ...core.request_options import RequestOptions from ...core.unchecked_base_model import construct_type from ...errors.unprocessable_entity_error import UnprocessableEntityError from ...types.dubbing_render_response_model import DubbingRenderResponseModel from ...types.dubbing_resource import DubbingResource from ...types.http_validation_error import HttpValidationError from ...types.render_type import RenderType from ...types.segment_dub_response import SegmentDubResponse from ...types.segment_migration_response import SegmentMigrationResponse from ...types.segment_transcription_response import SegmentTranscriptionResponse from ...types.segment_translation_response import SegmentTranslationResponse from .types.resource_render_request_language import ResourceRenderRequestLanguage # this is used as the default value for optional parameters OMIT = typing.cast(typing.Any, ...) class RawResourceClient: def __init__(self, *, client_wrapper: SyncClientWrapper): self._client_wrapper = client_wrapper def get( self, dubbing_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[DubbingResource]: """ Given a dubbing ID generated from the '/v1/dubbing' endpoint with studio enabled, returns the dubbing resource. Parameters ---------- dubbing_id : str ID of the dubbing project. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DubbingResource] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DubbingResource, construct_type( type_=DubbingResource, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def migrate_segments( self, dubbing_id: str, *, segment_ids: typing.Sequence[str], speaker_id: str, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[SegmentMigrationResponse]: """ Change the attribution of one or more segments to a different speaker. Parameters ---------- dubbing_id : str ID of the dubbing project. segment_ids : typing.Sequence[str] speaker_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[SegmentMigrationResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}/migrate-segments", method="POST", json={ "segment_ids": segment_ids, "speaker_id": speaker_id, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SegmentMigrationResponse, construct_type( type_=SegmentMigrationResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def transcribe( self, dubbing_id: str, *, segments: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[SegmentTranscriptionResponse]: """ Regenerate the transcriptions for the specified segments. Does not automatically regenerate translations or dubs. Parameters ---------- dubbing_id : str ID of the dubbing project. segments : typing.Sequence[str] Transcribe this specific list of segments. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[SegmentTranscriptionResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}/transcribe", method="POST", json={ "segments": segments, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SegmentTranscriptionResponse, construct_type( type_=SegmentTranscriptionResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def translate( self, dubbing_id: str, *, segments: typing.Sequence[str], languages: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[SegmentTranslationResponse]: """ Regenerate the translations for either the entire resource or the specified segments/languages. Will automatically transcribe missing transcriptions. Will not automatically regenerate the dubs. Parameters ---------- dubbing_id : str ID of the dubbing project. segments : typing.Sequence[str] Translate only this list of segments. languages : typing.Optional[typing.Sequence[str]] Translate only these languages for each segment. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[SegmentTranslationResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}/translate", method="POST", json={ "segments": segments, "languages": languages, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SegmentTranslationResponse, construct_type( type_=SegmentTranslationResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def dub( self, dubbing_id: str, *, segments: typing.Sequence[str], languages: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[SegmentDubResponse]: """ Regenerate the dubs for either the entire resource or the specified segments/languages. Will automatically transcribe and translate any missing transcriptions and translations. Parameters ---------- dubbing_id : str ID of the dubbing project. segments : typing.Sequence[str] Dub only this list of segments. languages : typing.Optional[typing.Sequence[str]] Dub only these languages for each segment. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[SegmentDubResponse] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}/dub", method="POST", json={ "segments": segments, "languages": languages, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SegmentDubResponse, construct_type( type_=SegmentDubResponse, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) def render( self, dubbing_id: str, language: ResourceRenderRequestLanguage, *, render_type: RenderType, normalize_volume: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> HttpResponse[DubbingRenderResponseModel]: """ Regenerate the output media for a language using the latest Studio state. Please ensure all segments have been dubbed before rendering, otherwise they will be omitted. Renders are generated asynchronously, and to check the status of all renders please use the 'Get Dubbing Resource' endpoint. Parameters ---------- dubbing_id : str ID of the dubbing project. language : ResourceRenderRequestLanguage The target language code to render, eg. 'es'. To render the source track use 'original'. render_type : RenderType The type of the render. One of ['mp4', 'aac', 'mp3', 'wav', 'aaf', 'tracks_zip', 'clips_zip'] normalize_volume : typing.Optional[bool] Whether to normalize the volume of the rendered audio. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- HttpResponse[DubbingRenderResponseModel] Successful Response """ _response = self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}/render/{jsonable_encoder(language)}", method="POST", json={ "render_type": render_type, "normalize_volume": normalize_volume, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DubbingRenderResponseModel, construct_type( type_=DubbingRenderResponseModel, # type: ignore object_=_response.json(), ), ) return HttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) class AsyncRawResourceClient: def __init__(self, *, client_wrapper: AsyncClientWrapper): self._client_wrapper = client_wrapper async def get( self, dubbing_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[DubbingResource]: """ Given a dubbing ID generated from the '/v1/dubbing' endpoint with studio enabled, returns the dubbing resource. Parameters ---------- dubbing_id : str ID of the dubbing project. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DubbingResource] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}", method="GET", request_options=request_options, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DubbingResource, construct_type( type_=DubbingResource, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def migrate_segments( self, dubbing_id: str, *, segment_ids: typing.Sequence[str], speaker_id: str, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[SegmentMigrationResponse]: """ Change the attribution of one or more segments to a different speaker. Parameters ---------- dubbing_id : str ID of the dubbing project. segment_ids : typing.Sequence[str] speaker_id : str request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[SegmentMigrationResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}/migrate-segments", method="POST", json={ "segment_ids": segment_ids, "speaker_id": speaker_id, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SegmentMigrationResponse, construct_type( type_=SegmentMigrationResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def transcribe( self, dubbing_id: str, *, segments: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[SegmentTranscriptionResponse]: """ Regenerate the transcriptions for the specified segments. Does not automatically regenerate translations or dubs. Parameters ---------- dubbing_id : str ID of the dubbing project. segments : typing.Sequence[str] Transcribe this specific list of segments. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[SegmentTranscriptionResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}/transcribe", method="POST", json={ "segments": segments, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SegmentTranscriptionResponse, construct_type( type_=SegmentTranscriptionResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def translate( self, dubbing_id: str, *, segments: typing.Sequence[str], languages: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[SegmentTranslationResponse]: """ Regenerate the translations for either the entire resource or the specified segments/languages. Will automatically transcribe missing transcriptions. Will not automatically regenerate the dubs. Parameters ---------- dubbing_id : str ID of the dubbing project. segments : typing.Sequence[str] Translate only this list of segments. languages : typing.Optional[typing.Sequence[str]] Translate only these languages for each segment. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[SegmentTranslationResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}/translate", method="POST", json={ "segments": segments, "languages": languages, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SegmentTranslationResponse, construct_type( type_=SegmentTranslationResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def dub( self, dubbing_id: str, *, segments: typing.Sequence[str], languages: typing.Optional[typing.Sequence[str]] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[SegmentDubResponse]: """ Regenerate the dubs for either the entire resource or the specified segments/languages. Will automatically transcribe and translate any missing transcriptions and translations. Parameters ---------- dubbing_id : str ID of the dubbing project. segments : typing.Sequence[str] Dub only this list of segments. languages : typing.Optional[typing.Sequence[str]] Dub only these languages for each segment. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[SegmentDubResponse] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}/dub", method="POST", json={ "segments": segments, "languages": languages, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( SegmentDubResponse, construct_type( type_=SegmentDubResponse, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) async def render( self, dubbing_id: str, language: ResourceRenderRequestLanguage, *, render_type: RenderType, normalize_volume: typing.Optional[bool] = OMIT, request_options: typing.Optional[RequestOptions] = None, ) -> AsyncHttpResponse[DubbingRenderResponseModel]: """ Regenerate the output media for a language using the latest Studio state. Please ensure all segments have been dubbed before rendering, otherwise they will be omitted. Renders are generated asynchronously, and to check the status of all renders please use the 'Get Dubbing Resource' endpoint. Parameters ---------- dubbing_id : str ID of the dubbing project. language : ResourceRenderRequestLanguage The target language code to render, eg. 'es'. To render the source track use 'original'. render_type : RenderType The type of the render. One of ['mp4', 'aac', 'mp3', 'wav', 'aaf', 'tracks_zip', 'clips_zip'] normalize_volume : typing.Optional[bool] Whether to normalize the volume of the rendered audio. request_options : typing.Optional[RequestOptions] Request-specific configuration. Returns ------- AsyncHttpResponse[DubbingRenderResponseModel] Successful Response """ _response = await self._client_wrapper.httpx_client.request( f"v1/dubbing/resource/{jsonable_encoder(dubbing_id)}/render/{jsonable_encoder(language)}", method="POST", json={ "render_type": render_type, "normalize_volume": normalize_volume, }, headers={ "content-type": "application/json", }, request_options=request_options, omit=OMIT, ) try: if 200 <= _response.status_code < 300: _data = typing.cast( DubbingRenderResponseModel, construct_type( type_=DubbingRenderResponseModel, # type: ignore object_=_response.json(), ), ) return AsyncHttpResponse(response=_response, data=_data) if _response.status_code == 422: raise UnprocessableEntityError( headers=dict(_response.headers), body=typing.cast( HttpValidationError, construct_type( type_=HttpValidationError, # type: ignore object_=_response.json(), ), ), ) _response_json = _response.json() except JSONDecodeError: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json)