Keepalive

Client implementation of the Keepalive service.

exception bosdyn.client.keepalive.KeepaliveResponseError(response, error_message=None)

Bases: ResponseError

Error in Keepalive RPC

exception bosdyn.client.keepalive.InvalidLeaseError(response, error_message=None)

Bases: KeepaliveResponseError

A policy’s associated lease was not the same, super, or sub lease of the active lease.

exception bosdyn.client.keepalive.InvalidPolicyError(response, error_message=None)

Bases: KeepaliveResponseError

The specified policy ID was not valid.

class bosdyn.client.keepalive.Policy(proto: None | Policy = None)

Bases: object

Helper class for API Policy.

property name: str

Get or set the name of the Policy

add_associated_lease(lease: bosdyn.client.lease.Lease | lease_pb2.Lease)

add_controlled_motors_off_action(after: float)

Add a ‘controlled motors off’ action that triggers after specified time (seconds).

add_immediate_robot_off_action(after: float)

Add an ‘immediate robot off’ action that triggers after specified time (seconds).

add_record_event_action(events: List[bosdyn.api.Event], after: float)

Add a ‘record event’ action that triggers after specified time (seconds).

add_auto_return_action(leases: List[bosdyn.client.lease.Lease], params: bosdyn.api.auto_return.Params, after: float)

Add an ‘auto return’ action that triggers after specified time (seconds).

add_lease_stale_action(leases: List[Lease], after: float)

Add a ‘mark lease stale’ action that triggers after specified time (seconds).

shortest_action_delay() → None | float

Get the shortest delay on an action, or None if no actions are set.

For example: pol = Policy() pol.add_controlled_motors_off_action(2.5) pol.add_immediate_robot_off_action(1.2) assert pol.shortest_action_delay() == 1.2

class bosdyn.client.keepalive.KeepaliveClient

Bases: BaseClient

A client for the Keepalive service.

This client is in BETA and may undergo changes in future releases.

default_service_name = 'keepalive'

service_type = 'bosdyn.api.keepalive.KeepaliveService'

modify_policy(to_add: Policy | Policy | None = None, policy_ids_to_remove: List[int] | None = None, **kwargs)

Add given policy and remove policies with given ids.

modify_policy_async(to_add: Policy | Policy | None = None, policy_ids_to_remove: List[int] | None = None, **kwargs)

Async version of the modify_policy() RPC.

check_in(policy_id: int, **kwargs)

Check in for given policy_id, refreshing that policy’s timer.

check_in_async(policy_id: int, **kwargs)

Async version of the check_in() RPC.

get_status(**kwargs)

Get status on all policies.

get_status_async(**kwargs)

Async version of the get_status() RPC.

bosdyn.client.keepalive.modify_policy_error(response)

ModifyPolicy response to exception.

bosdyn.client.keepalive.check_in_error(response)

CheckIn response to exception.

class bosdyn.client.keepalive.PolicyKeepalive(client: KeepaliveClient, policy: Policy, rpc_timeout_seconds: float | None = None, rpc_interval_seconds: float | None = None, logger: Logger | None = None, remove_policy_on_exit: bool = False)

Bases: object

Specify a keepalive Policy that should be held to.

Meant to be used as a context manager. For example:

client = robot.ensure_client(KeepaliveClient.default_service_name) pol = Policy() # After 30 seconds of not hearing from this process, turn the robot motors off. pol.add_controlled_motors_off_action(30) with PolicyKeepalive(client, pol, rpc_interval_seconds=3) as policy_keepalive:

# A thread will attempt a CheckIn every 3 seconds. run_my_code()

remove_policy()

Remove this instance’s policy, if it did manage to add one.

start()

Start the checkin thread.

shutdown()

Stop the checkin thread and block until it ends.

bosdyn.client.keepalive.remove_all_policies(keepalive_client, attempts=1)

Remove all policies on the robot.

Optionally do this over a few attempts, in case other things are also removing policies.