# 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 door service."""
from bosdyn.api.spot import door_service_pb2_grpc
from bosdyn.client.common import (BaseClient, handle_common_header_errors,
handle_lease_use_result_errors)
from .lease import add_lease_wallet_processors
[docs]class DoorClient(BaseClient):
"""Client for the door service."""
default_service_name = 'door'
service_type = 'bosdyn.api.spot.DoorService'
def __init__(self):
super(DoorClient, self).__init__(door_service_pb2_grpc.DoorServiceStub)
[docs] def open_door(self, request, **kwargs):
"""Issue an open door command to the robot.
Args:
request (door_pb2.OpenDoorCommandRequest): The door command.
Returns:
The full OpenDoorCommandResponse message, which includes a command id for feedback.
Raises:
RpcError: Problem communicating with the robot.
LeaseUseError: The lease for the request failed.
"""
return self.call(self._stub.OpenDoor, request, None, _open_door_error_handler, **kwargs)
[docs] def open_door_async(self, request, **kwargs):
"""Async version of open_door()."""
return self.call_async(self._stub.OpenDoor, request, None, _open_door_error_handler,
**kwargs)
[docs] def open_door_feedback(self, request, **kwargs):
"""Get feedback from the robot on a specific door command.
Args:
request (door_pb2.OpenDoorFeedbackRequest): The request for feedback of
the door command.
Returns:
The full OpenDoorFeedbackResponse message.
Raises:
RpcError: Problem communicating with the robot.
"""
return self.call(self._stub.OpenDoorFeedback, request, None,
_open_door_feedback_error_handler, **kwargs)
[docs] def open_door_feedback_async(self, request, **kwargs):
"""Async version of open_door_feedback()."""
return self.call_async(self._stub.OpenDoorFeedback, request, None,
_open_door_feedback_error_handler, **kwargs)
[docs] def update_from(self, other):
"""Update instance from another object.
Args:
other: The object where to copy from.
"""
# The door service requires a lease.
super(DoorClient, self).update_from(other)
if self.lease_wallet:
add_lease_wallet_processors(self, self.lease_wallet)
@handle_common_header_errors
@handle_lease_use_result_errors
def _open_door_error_handler(response):
"""Return a custom exception based on response, None if no error."""
return None
@handle_common_header_errors
def _open_door_feedback_error_handler(response):
"""Return a custom exception based on response, None if no error."""
return None