# Copyright 2025 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # # Code generated by the Google Gen AI SDK generator DO NOT EDIT. from typing import Any, Optional, Union from . import _transformers as t from ._api_client import BaseApiClient from ._common import get_value_by_path as getv from ._common import set_value_by_path as setv def _Blob_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['data']) is not None: setv(to_object, ['data'], getv(from_object, ['data'])) if getv(from_object, ['display_name']) is not None: raise ValueError('display_name parameter is not supported in Gemini API.') if getv(from_object, ['mime_type']) is not None: setv(to_object, ['mimeType'], getv(from_object, ['mime_type'])) return to_object def _Content_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['parts']) is not None: setv( to_object, ['parts'], [ _Part_to_mldev(item, to_object) for item in getv(from_object, ['parts']) ], ) if getv(from_object, ['role']) is not None: setv(to_object, ['role'], getv(from_object, ['role'])) return to_object def _FileData_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['display_name']) is not None: raise ValueError('display_name parameter is not supported in Gemini API.') if getv(from_object, ['file_uri']) is not None: setv(to_object, ['fileUri'], getv(from_object, ['file_uri'])) if getv(from_object, ['mime_type']) is not None: setv(to_object, ['mimeType'], getv(from_object, ['mime_type'])) return to_object def _FunctionCall_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['id']) is not None: setv(to_object, ['id'], getv(from_object, ['id'])) if getv(from_object, ['args']) is not None: setv(to_object, ['args'], getv(from_object, ['args'])) if getv(from_object, ['name']) is not None: setv(to_object, ['name'], getv(from_object, ['name'])) if getv(from_object, ['partial_args']) is not None: raise ValueError('partial_args parameter is not supported in Gemini API.') if getv(from_object, ['will_continue']) is not None: raise ValueError('will_continue parameter is not supported in Gemini API.') return to_object def _FunctionDeclaration_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['description']) is not None: setv(to_object, ['description'], getv(from_object, ['description'])) if getv(from_object, ['name']) is not None: setv(to_object, ['name'], getv(from_object, ['name'])) if getv(from_object, ['parameters']) is not None: setv(to_object, ['parameters'], getv(from_object, ['parameters'])) if getv(from_object, ['parameters_json_schema']) is not None: setv( to_object, ['parametersJsonSchema'], getv(from_object, ['parameters_json_schema']), ) if getv(from_object, ['response']) is not None: setv(to_object, ['response'], getv(from_object, ['response'])) if getv(from_object, ['response_json_schema']) is not None: setv( to_object, ['responseJsonSchema'], getv(from_object, ['response_json_schema']), ) if getv(from_object, ['behavior']) is not None: raise ValueError('behavior parameter is not supported in Vertex AI.') return to_object def _GenerationConfig_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['model_selection_config']) is not None: setv( to_object, ['modelConfig'], getv(from_object, ['model_selection_config']), ) if getv(from_object, ['response_json_schema']) is not None: setv( to_object, ['responseJsonSchema'], getv(from_object, ['response_json_schema']), ) if getv(from_object, ['audio_timestamp']) is not None: setv(to_object, ['audioTimestamp'], getv(from_object, ['audio_timestamp'])) if getv(from_object, ['candidate_count']) is not None: setv(to_object, ['candidateCount'], getv(from_object, ['candidate_count'])) if getv(from_object, ['enable_affective_dialog']) is not None: setv( to_object, ['enableAffectiveDialog'], getv(from_object, ['enable_affective_dialog']), ) if getv(from_object, ['frequency_penalty']) is not None: setv( to_object, ['frequencyPenalty'], getv(from_object, ['frequency_penalty']), ) if getv(from_object, ['logprobs']) is not None: setv(to_object, ['logprobs'], getv(from_object, ['logprobs'])) if getv(from_object, ['max_output_tokens']) is not None: setv( to_object, ['maxOutputTokens'], getv(from_object, ['max_output_tokens']) ) if getv(from_object, ['media_resolution']) is not None: setv( to_object, ['mediaResolution'], getv(from_object, ['media_resolution']) ) if getv(from_object, ['presence_penalty']) is not None: setv( to_object, ['presencePenalty'], getv(from_object, ['presence_penalty']) ) if getv(from_object, ['response_logprobs']) is not None: setv( to_object, ['responseLogprobs'], getv(from_object, ['response_logprobs']), ) if getv(from_object, ['response_mime_type']) is not None: setv( to_object, ['responseMimeType'], getv(from_object, ['response_mime_type']), ) if getv(from_object, ['response_modalities']) is not None: setv( to_object, ['responseModalities'], getv(from_object, ['response_modalities']), ) if getv(from_object, ['response_schema']) is not None: setv(to_object, ['responseSchema'], getv(from_object, ['response_schema'])) if getv(from_object, ['routing_config']) is not None: setv(to_object, ['routingConfig'], getv(from_object, ['routing_config'])) if getv(from_object, ['seed']) is not None: setv(to_object, ['seed'], getv(from_object, ['seed'])) if getv(from_object, ['speech_config']) is not None: setv(to_object, ['speechConfig'], getv(from_object, ['speech_config'])) if getv(from_object, ['stop_sequences']) is not None: setv(to_object, ['stopSequences'], getv(from_object, ['stop_sequences'])) if getv(from_object, ['temperature']) is not None: setv(to_object, ['temperature'], getv(from_object, ['temperature'])) if getv(from_object, ['thinking_config']) is not None: setv(to_object, ['thinkingConfig'], getv(from_object, ['thinking_config'])) if getv(from_object, ['top_k']) is not None: setv(to_object, ['topK'], getv(from_object, ['top_k'])) if getv(from_object, ['top_p']) is not None: setv(to_object, ['topP'], getv(from_object, ['top_p'])) if getv(from_object, ['enable_enhanced_civic_answers']) is not None: raise ValueError( 'enable_enhanced_civic_answers parameter is not supported in Vertex AI.' ) return to_object def _GoogleMaps_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['auth_config']) is not None: raise ValueError('auth_config parameter is not supported in Gemini API.') if getv(from_object, ['enable_widget']) is not None: setv(to_object, ['enableWidget'], getv(from_object, ['enable_widget'])) return to_object def _GoogleSearch_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['exclude_domains']) is not None: raise ValueError( 'exclude_domains parameter is not supported in Gemini API.' ) if getv(from_object, ['blocking_confidence']) is not None: raise ValueError( 'blocking_confidence parameter is not supported in Gemini API.' ) if getv(from_object, ['time_range_filter']) is not None: setv( to_object, ['timeRangeFilter'], getv(from_object, ['time_range_filter']) ) return to_object def _LiveClientContent_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['turns']) is not None: setv( to_object, ['turns'], [ _Content_to_mldev(item, to_object) for item in getv(from_object, ['turns']) ], ) if getv(from_object, ['turn_complete']) is not None: setv(to_object, ['turnComplete'], getv(from_object, ['turn_complete'])) return to_object def _LiveClientMessage_to_mldev( api_client: BaseApiClient, from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['setup']) is not None: setv( to_object, ['setup'], _LiveClientSetup_to_mldev( api_client, getv(from_object, ['setup']), to_object ), ) if getv(from_object, ['client_content']) is not None: setv( to_object, ['clientContent'], _LiveClientContent_to_mldev( getv(from_object, ['client_content']), to_object ), ) if getv(from_object, ['realtime_input']) is not None: setv( to_object, ['realtimeInput'], _LiveClientRealtimeInput_to_mldev( getv(from_object, ['realtime_input']), to_object ), ) if getv(from_object, ['tool_response']) is not None: setv(to_object, ['toolResponse'], getv(from_object, ['tool_response'])) return to_object def _LiveClientMessage_to_vertex( api_client: BaseApiClient, from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['setup']) is not None: setv( to_object, ['setup'], _LiveClientSetup_to_vertex( api_client, getv(from_object, ['setup']), to_object ), ) if getv(from_object, ['client_content']) is not None: setv(to_object, ['clientContent'], getv(from_object, ['client_content'])) if getv(from_object, ['realtime_input']) is not None: setv( to_object, ['realtimeInput'], _LiveClientRealtimeInput_to_vertex( getv(from_object, ['realtime_input']), to_object ), ) if getv(from_object, ['tool_response']) is not None: setv(to_object, ['toolResponse'], getv(from_object, ['tool_response'])) return to_object def _LiveClientRealtimeInput_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['media_chunks']) is not None: setv( to_object, ['mediaChunks'], [ _Blob_to_mldev(item, to_object) for item in getv(from_object, ['media_chunks']) ], ) if getv(from_object, ['audio']) is not None: setv( to_object, ['audio'], _Blob_to_mldev(getv(from_object, ['audio']), to_object), ) if getv(from_object, ['audio_stream_end']) is not None: setv(to_object, ['audioStreamEnd'], getv(from_object, ['audio_stream_end'])) if getv(from_object, ['video']) is not None: setv( to_object, ['video'], _Blob_to_mldev(getv(from_object, ['video']), to_object), ) if getv(from_object, ['text']) is not None: setv(to_object, ['text'], getv(from_object, ['text'])) if getv(from_object, ['activity_start']) is not None: setv(to_object, ['activityStart'], getv(from_object, ['activity_start'])) if getv(from_object, ['activity_end']) is not None: setv(to_object, ['activityEnd'], getv(from_object, ['activity_end'])) return to_object def _LiveClientRealtimeInput_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['media_chunks']) is not None: setv( to_object, ['mediaChunks'], [item for item in getv(from_object, ['media_chunks'])], ) if getv(from_object, ['audio']) is not None: setv(to_object, ['audio'], getv(from_object, ['audio'])) if getv(from_object, ['audio_stream_end']) is not None: raise ValueError( 'audio_stream_end parameter is not supported in Vertex AI.' ) if getv(from_object, ['video']) is not None: setv(to_object, ['video'], getv(from_object, ['video'])) if getv(from_object, ['text']) is not None: setv(to_object, ['text'], getv(from_object, ['text'])) if getv(from_object, ['activity_start']) is not None: setv(to_object, ['activityStart'], getv(from_object, ['activity_start'])) if getv(from_object, ['activity_end']) is not None: setv(to_object, ['activityEnd'], getv(from_object, ['activity_end'])) return to_object def _LiveClientSetup_to_mldev( api_client: BaseApiClient, from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['model']) is not None: setv(to_object, ['model'], getv(from_object, ['model'])) if getv(from_object, ['generation_config']) is not None: setv( to_object, ['generationConfig'], getv(from_object, ['generation_config']), ) if getv(from_object, ['system_instruction']) is not None: setv( to_object, ['systemInstruction'], _Content_to_mldev( t.t_content(getv(from_object, ['system_instruction'])), to_object ), ) if getv(from_object, ['tools']) is not None: setv( to_object, ['tools'], [ _Tool_to_mldev(t.t_tool(api_client, item), to_object) for item in t.t_tools(api_client, getv(from_object, ['tools'])) ], ) if getv(from_object, ['session_resumption']) is not None: setv( to_object, ['sessionResumption'], _SessionResumptionConfig_to_mldev( getv(from_object, ['session_resumption']), to_object ), ) if getv(from_object, ['context_window_compression']) is not None: setv( to_object, ['contextWindowCompression'], getv(from_object, ['context_window_compression']), ) if getv(from_object, ['input_audio_transcription']) is not None: setv( to_object, ['inputAudioTranscription'], getv(from_object, ['input_audio_transcription']), ) if getv(from_object, ['output_audio_transcription']) is not None: setv( to_object, ['outputAudioTranscription'], getv(from_object, ['output_audio_transcription']), ) if getv(from_object, ['proactivity']) is not None: setv(to_object, ['proactivity'], getv(from_object, ['proactivity'])) if getv(from_object, ['explicit_vad_signal']) is not None: raise ValueError( 'explicit_vad_signal parameter is not supported in Gemini API.' ) return to_object def _LiveClientSetup_to_vertex( api_client: BaseApiClient, from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['model']) is not None: setv(to_object, ['model'], getv(from_object, ['model'])) if getv(from_object, ['generation_config']) is not None: setv( to_object, ['generationConfig'], _GenerationConfig_to_vertex( getv(from_object, ['generation_config']), to_object ), ) if getv(from_object, ['system_instruction']) is not None: setv( to_object, ['systemInstruction'], t.t_content(getv(from_object, ['system_instruction'])), ) if getv(from_object, ['tools']) is not None: setv( to_object, ['tools'], [ _Tool_to_vertex(t.t_tool(api_client, item), to_object) for item in t.t_tools(api_client, getv(from_object, ['tools'])) ], ) if getv(from_object, ['session_resumption']) is not None: setv( to_object, ['sessionResumption'], getv(from_object, ['session_resumption']), ) if getv(from_object, ['context_window_compression']) is not None: setv( to_object, ['contextWindowCompression'], getv(from_object, ['context_window_compression']), ) if getv(from_object, ['input_audio_transcription']) is not None: setv( to_object, ['inputAudioTranscription'], getv(from_object, ['input_audio_transcription']), ) if getv(from_object, ['output_audio_transcription']) is not None: setv( to_object, ['outputAudioTranscription'], getv(from_object, ['output_audio_transcription']), ) if getv(from_object, ['proactivity']) is not None: setv(to_object, ['proactivity'], getv(from_object, ['proactivity'])) if getv(from_object, ['explicit_vad_signal']) is not None: setv( to_object, ['explicitVadSignal'], getv(from_object, ['explicit_vad_signal']), ) return to_object def _LiveConnectConfig_to_mldev( api_client: BaseApiClient, from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['generation_config']) is not None: setv( parent_object, ['setup', 'generationConfig'], getv(from_object, ['generation_config']), ) if getv(from_object, ['response_modalities']) is not None: setv( parent_object, ['setup', 'generationConfig', 'responseModalities'], getv(from_object, ['response_modalities']), ) if getv(from_object, ['temperature']) is not None: setv( parent_object, ['setup', 'generationConfig', 'temperature'], getv(from_object, ['temperature']), ) if getv(from_object, ['top_p']) is not None: setv( parent_object, ['setup', 'generationConfig', 'topP'], getv(from_object, ['top_p']), ) if getv(from_object, ['top_k']) is not None: setv( parent_object, ['setup', 'generationConfig', 'topK'], getv(from_object, ['top_k']), ) if getv(from_object, ['max_output_tokens']) is not None: setv( parent_object, ['setup', 'generationConfig', 'maxOutputTokens'], getv(from_object, ['max_output_tokens']), ) if getv(from_object, ['media_resolution']) is not None: setv( parent_object, ['setup', 'generationConfig', 'mediaResolution'], getv(from_object, ['media_resolution']), ) if getv(from_object, ['seed']) is not None: setv( parent_object, ['setup', 'generationConfig', 'seed'], getv(from_object, ['seed']), ) if getv(from_object, ['speech_config']) is not None: setv( parent_object, ['setup', 'generationConfig', 'speechConfig'], t.t_live_speech_config(getv(from_object, ['speech_config'])), ) if getv(from_object, ['thinking_config']) is not None: setv( parent_object, ['setup', 'generationConfig', 'thinkingConfig'], getv(from_object, ['thinking_config']), ) if getv(from_object, ['enable_affective_dialog']) is not None: setv( parent_object, ['setup', 'generationConfig', 'enableAffectiveDialog'], getv(from_object, ['enable_affective_dialog']), ) if getv(from_object, ['system_instruction']) is not None: setv( parent_object, ['setup', 'systemInstruction'], _Content_to_mldev( t.t_content(getv(from_object, ['system_instruction'])), to_object ), ) if getv(from_object, ['tools']) is not None: setv( parent_object, ['setup', 'tools'], [ _Tool_to_mldev(t.t_tool(api_client, item), to_object) for item in t.t_tools(api_client, getv(from_object, ['tools'])) ], ) if getv(from_object, ['session_resumption']) is not None: setv( parent_object, ['setup', 'sessionResumption'], _SessionResumptionConfig_to_mldev( getv(from_object, ['session_resumption']), to_object ), ) if getv(from_object, ['input_audio_transcription']) is not None: setv( parent_object, ['setup', 'inputAudioTranscription'], getv(from_object, ['input_audio_transcription']), ) if getv(from_object, ['output_audio_transcription']) is not None: setv( parent_object, ['setup', 'outputAudioTranscription'], getv(from_object, ['output_audio_transcription']), ) if getv(from_object, ['realtime_input_config']) is not None: setv( parent_object, ['setup', 'realtimeInputConfig'], getv(from_object, ['realtime_input_config']), ) if getv(from_object, ['context_window_compression']) is not None: setv( parent_object, ['setup', 'contextWindowCompression'], getv(from_object, ['context_window_compression']), ) if getv(from_object, ['proactivity']) is not None: setv( parent_object, ['setup', 'proactivity'], getv(from_object, ['proactivity']), ) if getv(from_object, ['explicit_vad_signal']) is not None: raise ValueError( 'explicit_vad_signal parameter is not supported in Gemini API.' ) return to_object def _LiveConnectConfig_to_vertex( api_client: BaseApiClient, from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['generation_config']) is not None: setv( parent_object, ['setup', 'generationConfig'], _GenerationConfig_to_vertex( getv(from_object, ['generation_config']), to_object ), ) if getv(from_object, ['response_modalities']) is not None: setv( parent_object, ['setup', 'generationConfig', 'responseModalities'], getv(from_object, ['response_modalities']), ) if getv(from_object, ['temperature']) is not None: setv( parent_object, ['setup', 'generationConfig', 'temperature'], getv(from_object, ['temperature']), ) if getv(from_object, ['top_p']) is not None: setv( parent_object, ['setup', 'generationConfig', 'topP'], getv(from_object, ['top_p']), ) if getv(from_object, ['top_k']) is not None: setv( parent_object, ['setup', 'generationConfig', 'topK'], getv(from_object, ['top_k']), ) if getv(from_object, ['max_output_tokens']) is not None: setv( parent_object, ['setup', 'generationConfig', 'maxOutputTokens'], getv(from_object, ['max_output_tokens']), ) if getv(from_object, ['media_resolution']) is not None: setv( parent_object, ['setup', 'generationConfig', 'mediaResolution'], getv(from_object, ['media_resolution']), ) if getv(from_object, ['seed']) is not None: setv( parent_object, ['setup', 'generationConfig', 'seed'], getv(from_object, ['seed']), ) if getv(from_object, ['speech_config']) is not None: setv( parent_object, ['setup', 'generationConfig', 'speechConfig'], t.t_live_speech_config(getv(from_object, ['speech_config'])), ) if getv(from_object, ['thinking_config']) is not None: setv( parent_object, ['setup', 'generationConfig', 'thinkingConfig'], getv(from_object, ['thinking_config']), ) if getv(from_object, ['enable_affective_dialog']) is not None: setv( parent_object, ['setup', 'generationConfig', 'enableAffectiveDialog'], getv(from_object, ['enable_affective_dialog']), ) if getv(from_object, ['system_instruction']) is not None: setv( parent_object, ['setup', 'systemInstruction'], t.t_content(getv(from_object, ['system_instruction'])), ) if getv(from_object, ['tools']) is not None: setv( parent_object, ['setup', 'tools'], [ _Tool_to_vertex(t.t_tool(api_client, item), to_object) for item in t.t_tools(api_client, getv(from_object, ['tools'])) ], ) if getv(from_object, ['session_resumption']) is not None: setv( parent_object, ['setup', 'sessionResumption'], getv(from_object, ['session_resumption']), ) if getv(from_object, ['input_audio_transcription']) is not None: setv( parent_object, ['setup', 'inputAudioTranscription'], getv(from_object, ['input_audio_transcription']), ) if getv(from_object, ['output_audio_transcription']) is not None: setv( parent_object, ['setup', 'outputAudioTranscription'], getv(from_object, ['output_audio_transcription']), ) if getv(from_object, ['realtime_input_config']) is not None: setv( parent_object, ['setup', 'realtimeInputConfig'], getv(from_object, ['realtime_input_config']), ) if getv(from_object, ['context_window_compression']) is not None: setv( parent_object, ['setup', 'contextWindowCompression'], getv(from_object, ['context_window_compression']), ) if getv(from_object, ['proactivity']) is not None: setv( parent_object, ['setup', 'proactivity'], getv(from_object, ['proactivity']), ) if getv(from_object, ['explicit_vad_signal']) is not None: setv( parent_object, ['setup', 'explicitVadSignal'], getv(from_object, ['explicit_vad_signal']), ) return to_object def _LiveConnectParameters_to_mldev( api_client: BaseApiClient, from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['model']) is not None: setv( to_object, ['setup', 'model'], t.t_model(api_client, getv(from_object, ['model'])), ) if getv(from_object, ['config']) is not None: setv( to_object, ['config'], _LiveConnectConfig_to_mldev( api_client, getv(from_object, ['config']), to_object ), ) return to_object def _LiveConnectParameters_to_vertex( api_client: BaseApiClient, from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['model']) is not None: setv( to_object, ['setup', 'model'], t.t_model(api_client, getv(from_object, ['model'])), ) if getv(from_object, ['config']) is not None: setv( to_object, ['config'], _LiveConnectConfig_to_vertex( api_client, getv(from_object, ['config']), to_object ), ) return to_object def _LiveMusicClientMessage_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['setup']) is not None: raise ValueError('setup parameter is not supported in Vertex AI.') if getv(from_object, ['client_content']) is not None: raise ValueError('client_content parameter is not supported in Vertex AI.') if getv(from_object, ['music_generation_config']) is not None: raise ValueError( 'music_generation_config parameter is not supported in Vertex AI.' ) if getv(from_object, ['playback_control']) is not None: raise ValueError( 'playback_control parameter is not supported in Vertex AI.' ) return to_object def _LiveMusicConnectParameters_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['model']) is not None: setv(to_object, ['setup', 'model'], getv(from_object, ['model'])) return to_object def _LiveMusicConnectParameters_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['model']) is not None: raise ValueError('model parameter is not supported in Vertex AI.') return to_object def _LiveMusicSetConfigParameters_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['music_generation_config']) is not None: setv( to_object, ['musicGenerationConfig'], getv(from_object, ['music_generation_config']), ) return to_object def _LiveMusicSetConfigParameters_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['music_generation_config']) is not None: raise ValueError( 'music_generation_config parameter is not supported in Vertex AI.' ) return to_object def _LiveMusicSetWeightedPromptsParameters_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['weighted_prompts']) is not None: setv( to_object, ['weightedPrompts'], [item for item in getv(from_object, ['weighted_prompts'])], ) return to_object def _LiveMusicSetWeightedPromptsParameters_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['weighted_prompts']) is not None: raise ValueError( 'weighted_prompts parameter is not supported in Vertex AI.' ) return to_object def _LiveSendRealtimeInputParameters_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['media']) is not None: setv( to_object, ['mediaChunks'], [ _Blob_to_mldev(item, to_object) for item in t.t_blobs(getv(from_object, ['media'])) ], ) if getv(from_object, ['audio']) is not None: setv( to_object, ['audio'], _Blob_to_mldev(t.t_audio_blob(getv(from_object, ['audio'])), to_object), ) if getv(from_object, ['audio_stream_end']) is not None: setv(to_object, ['audioStreamEnd'], getv(from_object, ['audio_stream_end'])) if getv(from_object, ['video']) is not None: setv( to_object, ['video'], _Blob_to_mldev(t.t_image_blob(getv(from_object, ['video'])), to_object), ) if getv(from_object, ['text']) is not None: setv(to_object, ['text'], getv(from_object, ['text'])) if getv(from_object, ['activity_start']) is not None: setv(to_object, ['activityStart'], getv(from_object, ['activity_start'])) if getv(from_object, ['activity_end']) is not None: setv(to_object, ['activityEnd'], getv(from_object, ['activity_end'])) return to_object def _LiveSendRealtimeInputParameters_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['media']) is not None: setv( to_object, ['mediaChunks'], [item for item in t.t_blobs(getv(from_object, ['media']))], ) if getv(from_object, ['audio']) is not None: setv(to_object, ['audio'], t.t_audio_blob(getv(from_object, ['audio']))) if getv(from_object, ['audio_stream_end']) is not None: setv(to_object, ['audioStreamEnd'], getv(from_object, ['audio_stream_end'])) if getv(from_object, ['video']) is not None: setv(to_object, ['video'], t.t_image_blob(getv(from_object, ['video']))) if getv(from_object, ['text']) is not None: setv(to_object, ['text'], getv(from_object, ['text'])) if getv(from_object, ['activity_start']) is not None: setv(to_object, ['activityStart'], getv(from_object, ['activity_start'])) if getv(from_object, ['activity_end']) is not None: setv(to_object, ['activityEnd'], getv(from_object, ['activity_end'])) return to_object def _LiveServerMessage_from_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['setupComplete']) is not None: setv(to_object, ['setup_complete'], getv(from_object, ['setupComplete'])) if getv(from_object, ['serverContent']) is not None: setv(to_object, ['server_content'], getv(from_object, ['serverContent'])) if getv(from_object, ['toolCall']) is not None: setv(to_object, ['tool_call'], getv(from_object, ['toolCall'])) if getv(from_object, ['toolCallCancellation']) is not None: setv( to_object, ['tool_call_cancellation'], getv(from_object, ['toolCallCancellation']), ) if getv(from_object, ['usageMetadata']) is not None: setv(to_object, ['usage_metadata'], getv(from_object, ['usageMetadata'])) if getv(from_object, ['goAway']) is not None: setv(to_object, ['go_away'], getv(from_object, ['goAway'])) if getv(from_object, ['sessionResumptionUpdate']) is not None: setv( to_object, ['session_resumption_update'], getv(from_object, ['sessionResumptionUpdate']), ) if getv(from_object, ['voiceActivityDetectionSignal']) is not None: setv( to_object, ['voice_activity_detection_signal'], getv(from_object, ['voiceActivityDetectionSignal']), ) if getv(from_object, ['voiceActivity']) is not None: setv( to_object, ['voice_activity'], _VoiceActivity_from_mldev( getv(from_object, ['voiceActivity']), to_object ), ) return to_object def _LiveServerMessage_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['setupComplete']) is not None: setv(to_object, ['setup_complete'], getv(from_object, ['setupComplete'])) if getv(from_object, ['serverContent']) is not None: setv(to_object, ['server_content'], getv(from_object, ['serverContent'])) if getv(from_object, ['toolCall']) is not None: setv(to_object, ['tool_call'], getv(from_object, ['toolCall'])) if getv(from_object, ['toolCallCancellation']) is not None: setv( to_object, ['tool_call_cancellation'], getv(from_object, ['toolCallCancellation']), ) if getv(from_object, ['usageMetadata']) is not None: setv( to_object, ['usage_metadata'], _UsageMetadata_from_vertex( getv(from_object, ['usageMetadata']), to_object ), ) if getv(from_object, ['goAway']) is not None: setv(to_object, ['go_away'], getv(from_object, ['goAway'])) if getv(from_object, ['sessionResumptionUpdate']) is not None: setv( to_object, ['session_resumption_update'], getv(from_object, ['sessionResumptionUpdate']), ) if getv(from_object, ['voiceActivityDetectionSignal']) is not None: setv( to_object, ['voice_activity_detection_signal'], getv(from_object, ['voiceActivityDetectionSignal']), ) if getv(from_object, ['voiceActivity']) is not None: setv( to_object, ['voice_activity'], _VoiceActivity_from_vertex( getv(from_object, ['voiceActivity']), to_object ), ) return to_object def _Part_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['media_resolution']) is not None: setv( to_object, ['mediaResolution'], getv(from_object, ['media_resolution']) ) if getv(from_object, ['code_execution_result']) is not None: setv( to_object, ['codeExecutionResult'], getv(from_object, ['code_execution_result']), ) if getv(from_object, ['executable_code']) is not None: setv(to_object, ['executableCode'], getv(from_object, ['executable_code'])) if getv(from_object, ['file_data']) is not None: setv( to_object, ['fileData'], _FileData_to_mldev(getv(from_object, ['file_data']), to_object), ) if getv(from_object, ['function_call']) is not None: setv( to_object, ['functionCall'], _FunctionCall_to_mldev(getv(from_object, ['function_call']), to_object), ) if getv(from_object, ['function_response']) is not None: setv( to_object, ['functionResponse'], getv(from_object, ['function_response']), ) if getv(from_object, ['inline_data']) is not None: setv( to_object, ['inlineData'], _Blob_to_mldev(getv(from_object, ['inline_data']), to_object), ) if getv(from_object, ['text']) is not None: setv(to_object, ['text'], getv(from_object, ['text'])) if getv(from_object, ['thought']) is not None: setv(to_object, ['thought'], getv(from_object, ['thought'])) if getv(from_object, ['thought_signature']) is not None: setv( to_object, ['thoughtSignature'], getv(from_object, ['thought_signature']), ) if getv(from_object, ['video_metadata']) is not None: setv(to_object, ['videoMetadata'], getv(from_object, ['video_metadata'])) return to_object def _SessionResumptionConfig_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['handle']) is not None: setv(to_object, ['handle'], getv(from_object, ['handle'])) if getv(from_object, ['transparent']) is not None: raise ValueError('transparent parameter is not supported in Gemini API.') return to_object def _Tool_to_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['retrieval']) is not None: raise ValueError('retrieval parameter is not supported in Gemini API.') if getv(from_object, ['computer_use']) is not None: setv(to_object, ['computerUse'], getv(from_object, ['computer_use'])) if getv(from_object, ['file_search']) is not None: setv(to_object, ['fileSearch'], getv(from_object, ['file_search'])) if getv(from_object, ['code_execution']) is not None: setv(to_object, ['codeExecution'], getv(from_object, ['code_execution'])) if getv(from_object, ['enterprise_web_search']) is not None: raise ValueError( 'enterprise_web_search parameter is not supported in Gemini API.' ) if getv(from_object, ['function_declarations']) is not None: setv( to_object, ['functionDeclarations'], [item for item in getv(from_object, ['function_declarations'])], ) if getv(from_object, ['google_maps']) is not None: setv( to_object, ['googleMaps'], _GoogleMaps_to_mldev(getv(from_object, ['google_maps']), to_object), ) if getv(from_object, ['google_search']) is not None: setv( to_object, ['googleSearch'], _GoogleSearch_to_mldev(getv(from_object, ['google_search']), to_object), ) if getv(from_object, ['google_search_retrieval']) is not None: setv( to_object, ['googleSearchRetrieval'], getv(from_object, ['google_search_retrieval']), ) if getv(from_object, ['url_context']) is not None: setv(to_object, ['urlContext'], getv(from_object, ['url_context'])) return to_object def _Tool_to_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['retrieval']) is not None: setv(to_object, ['retrieval'], getv(from_object, ['retrieval'])) if getv(from_object, ['computer_use']) is not None: setv(to_object, ['computerUse'], getv(from_object, ['computer_use'])) if getv(from_object, ['file_search']) is not None: raise ValueError('file_search parameter is not supported in Vertex AI.') if getv(from_object, ['code_execution']) is not None: setv(to_object, ['codeExecution'], getv(from_object, ['code_execution'])) if getv(from_object, ['enterprise_web_search']) is not None: setv( to_object, ['enterpriseWebSearch'], getv(from_object, ['enterprise_web_search']), ) if getv(from_object, ['function_declarations']) is not None: setv( to_object, ['functionDeclarations'], [ _FunctionDeclaration_to_vertex(item, to_object) for item in getv(from_object, ['function_declarations']) ], ) if getv(from_object, ['google_maps']) is not None: setv(to_object, ['googleMaps'], getv(from_object, ['google_maps'])) if getv(from_object, ['google_search']) is not None: setv(to_object, ['googleSearch'], getv(from_object, ['google_search'])) if getv(from_object, ['google_search_retrieval']) is not None: setv( to_object, ['googleSearchRetrieval'], getv(from_object, ['google_search_retrieval']), ) if getv(from_object, ['url_context']) is not None: setv(to_object, ['urlContext'], getv(from_object, ['url_context'])) return to_object def _UsageMetadata_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['promptTokenCount']) is not None: setv( to_object, ['prompt_token_count'], getv(from_object, ['promptTokenCount']), ) if getv(from_object, ['cachedContentTokenCount']) is not None: setv( to_object, ['cached_content_token_count'], getv(from_object, ['cachedContentTokenCount']), ) if getv(from_object, ['candidatesTokenCount']) is not None: setv( to_object, ['response_token_count'], getv(from_object, ['candidatesTokenCount']), ) if getv(from_object, ['toolUsePromptTokenCount']) is not None: setv( to_object, ['tool_use_prompt_token_count'], getv(from_object, ['toolUsePromptTokenCount']), ) if getv(from_object, ['thoughtsTokenCount']) is not None: setv( to_object, ['thoughts_token_count'], getv(from_object, ['thoughtsTokenCount']), ) if getv(from_object, ['totalTokenCount']) is not None: setv( to_object, ['total_token_count'], getv(from_object, ['totalTokenCount']) ) if getv(from_object, ['promptTokensDetails']) is not None: setv( to_object, ['prompt_tokens_details'], [item for item in getv(from_object, ['promptTokensDetails'])], ) if getv(from_object, ['cacheTokensDetails']) is not None: setv( to_object, ['cache_tokens_details'], [item for item in getv(from_object, ['cacheTokensDetails'])], ) if getv(from_object, ['candidatesTokensDetails']) is not None: setv( to_object, ['response_tokens_details'], [item for item in getv(from_object, ['candidatesTokensDetails'])], ) if getv(from_object, ['toolUsePromptTokensDetails']) is not None: setv( to_object, ['tool_use_prompt_tokens_details'], [item for item in getv(from_object, ['toolUsePromptTokensDetails'])], ) if getv(from_object, ['trafficType']) is not None: setv(to_object, ['traffic_type'], getv(from_object, ['trafficType'])) return to_object def _VoiceActivity_from_mldev( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['type']) is not None: setv(to_object, ['voice_activity_type'], getv(from_object, ['type'])) return to_object def _VoiceActivity_from_vertex( from_object: Union[dict[str, Any], object], parent_object: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: to_object: dict[str, Any] = {} if getv(from_object, ['type']) is not None: setv(to_object, ['voice_activity_type'], getv(from_object, ['type'])) return to_object