Common

Contains elements common to all service clients.

bosdyn.client.common.common_header_errors(response)[source]

Return an exception based on common response header. None if no error.

bosdyn.client.common.streaming_common_header_errors(response_iterator)[source]

Return an exception based on common response header for a streaming response iterator. None if no error.

bosdyn.client.common.common_lease_errors(response)[source]

Return an exception based on lease use result. None if no error.

bosdyn.client.common.streaming_common_lease_errors(response_iterator)[source]

Return an exception based on lease use result for a streaming response iterator. None if no error.

bosdyn.client.common.custom_params_error(response, status_value=None, status_field_name='status', error_field_name='custom_param_error', total_response=None)[source]

Return an exception based on having a custom parameter status and message. None if no error.

bosdyn.client.common.error_pair(error_message)[source]

Creates a pair of an error class and the associated docstring as the error message which can be used by the error_factory.

Parameters:

error_message – A class that inherits from the python Error class.

Returns:

The tuple of the error class and it’s associated docstring.

bosdyn.client.common.error_factory(response, status, status_to_string, status_to_error)[source]

Return an error based on the status field of the given response.

Since most callers of this function are “response to error” callbacks, any exceptions raised by this function are a considered a serious problem. Strongly consider using collections.defaultdict for the status_to_error mapping, and/or wrapping calls to this function in try/except blocks.

Parameters:
  • response – Protobuf message to examine or an iterator of protobuf responses.

  • status – Status from the protobuf message.

  • status_to_string – Function that converts numeric status value to string. May raise ValueError, in which case just the numeric code is included in a default error message.

  • status_to_error – mapping of status -> (error_constructor, error_message) error_constructor must take arguments “response” and “error_message”. (and ideally will subclass from ResponseError.)

Returns:

None if status_to_error[status] maps to (None, _). Otherwise, an instance of an error determined by status_to_error.

bosdyn.client.common.handle_unset_status_error(unset, field='status', statustype=None)[source]

Decorate “error from response” functions to handle unset status field errors.

bosdyn.client.common.handle_common_header_errors(func)[source]

Decorate “error from response” functions to handle typical header errors.

bosdyn.client.common.handle_lease_use_result_errors(func)[source]

Decorate “error from response” functions to handle typical lease errors.

bosdyn.client.common.handle_custom_params_errors(*args, status_value=None, status_field_name='status', error_field_name='custom_param_error')[source]

Decorate “error from response” functions to handle custom param errors.

bosdyn.client.common.maybe_raise(exc)[source]

raise the provided exception if it is not None

bosdyn.client.common.print_response(func)[source]

Decorate “error from response” functions to print for debugging specific messages.

bosdyn.client.common.process_kwargs(func)[source]
class bosdyn.client.common.BaseClient(stub_creation_func, name=None)[source]

Bases: object

Helper base class for all clients to Boston Dynamics services.

static request_trim_for_log(req)[source]

Deprecated since version 3.3.0: Forces serialization even if the logging is not happening. Do not use.

static response_trim_for_log(resp)[source]

Deprecated since version 3.3.0: Forces serialization even if the logging is not happening. Do not use.

property channel
update_from(other)[source]

Adopt key objects like processors, logger, and wallet from other.

update_request_iterator(request_iterator, logger, rpc_method, is_blocking, copy_request=True)[source]
update_response_iterator(response_iterator, logger, rpc_method, is_blocking)[source]
call(rpc_method, request, value_from_response=None, error_from_response=None, assemble_type=None, copy_request=True, **kwargs)[source]

Returns result of calling rpc_method(request, kwargs) after running processors.

value_from_response and error_from_response should not raise their own exceptions! Additionally, value_from_response and error_from_response that are not common handlers must accept streaming responses if it is a grpc streaming response.

handle_response(response, error_from_response, value_from_response)[source]
handle_response_streaming(response, error_from_response, value_from_response)[source]
call_async(rpc_method, request, value_from_response=None, error_from_response=None, copy_request=True, **kwargs)[source]

Returns a Future for rpc_method(request, kwargs) after running processors.

value_from_response and error_from_response should not raise their own exceptions!

Asynchronous calls cannot be done with streaming rpcs right now.

chunk_message(data_chunk_byte_size: int)

Take a message, and split it into data chunks

Parameters:

data_chunk_byte_size – max size of each streamed message

Deprecated since version 3.3.0: This has been moved to bosdyn.client.data_chunk. Please use bosdyn.client.data_chunk.chunk_message instead.

class bosdyn.client.common.FutureWrapper(future, value_from_response, error_from_response)[source]

Bases: object

Wraps a Future to aid more complicated clients’ async calls.

cancel()[source]
cancelled()[source]
running()[source]
done()[source]
traceback(**kwargs)[source]
add_done_callback(cb)[source]

Add callback executed on FutureWrapper when future is done.

result(**kwargs)[source]

Get the result of the value_from_response(future.result()).

exception(**kwargs)[source]

Get exceptions from the Future, or from custom response processing.

bosdyn.client.common.get_self_ip(robot_hostname)[source]

Get the IP address of the ethernet or WiFi interface used to talk to the robot.