# Copyright (c) 2023 Boston Dynamics, Inc. All rights reserved.
#
# Downloading, reproducing, distributing or otherwise using the SDK Software
# is subject to the terms and conditions of the Boston Dynamics Software
# Development Kit License (20191101-BDSDK-SL).
"""For clients to the Spot CAM Network service."""
import socket
import struct
from bosdyn.api.spot_cam import network_pb2, service_pb2_grpc
from bosdyn.client.common import BaseClient, handle_common_header_errors
[docs]class NetworkClient(BaseClient):
"""A client calling Spot CAM Network services such as ICE Candidates, SSL certs / Keys etc.
Note: Interactive Connectivity Establishment (ICE) is a protocol which lets two devices use
an intermediary to exchange offers and answers even if the two devices are separated
by Network Address Translation (NAT).
"""
default_service_name = 'spot-cam-network'
service_type = 'bosdyn.api.spot_cam.NetworkService'
def __init__(self):
super(NetworkClient, self).__init__(service_pb2_grpc.NetworkServiceStub)
[docs] def get_ice_configuration(self, **kwargs):
"""Set ICE configuration on Spot CAM. This overrides all existing configured servers"""
request = network_pb2.GetICEConfigurationRequest()
return self.call(self._stub.GetICEConfiguration, request, self._ice_servers_from_response,
self._ice_network_error_from_response, copy_request=False, **kwargs)
[docs] def get_ice_configuration_async(self, **kwargs):
"""Async version of get_ice_configuration()"""
request = network_pb2.GetICEConfigurationRequest()
return self.call_async(self._stub.GetICEConfiguration, request,
self._ice_servers_from_response,
self._ice_network_error_from_response, copy_request=False, **kwargs)
[docs] def set_ice_configuration(self, ice_servers, **kwargs):
"""Get ICE configuration from Spot CAM"""
request = self._set_ice_configuration_request(ice_servers)
return self.call(self._stub.SetICEConfiguration, request, None,
self._ice_network_error_from_response, copy_request=False, **kwargs)
[docs] def set_ice_configuration_async(self, ice_servers, **kwargs):
"""Async version of set_ice_configuration()"""
request = self._set_ice_configuration_request(ice_servers)
return self.call_async(self._stub.SetICEConfiguration, request, None,
self._ice_network_error_from_response, copy_request=False, **kwargs)
@staticmethod
def _set_ice_configuration_request(ice_servers):
request = network_pb2.SetICEConfigurationRequest()
request.servers.extend(ice_servers)
return request
@staticmethod
def _ice_servers_from_response(response):
return response.servers
@staticmethod
@handle_common_header_errors
def _ice_network_error_from_response(response): # pylint: disable=unused-argument
return None