Skip to content

Communications Reference

kevinbotlib.comm.redis

kevinbotlib.comm.redis.T = TypeVar('T', bound=BaseSendable) module-attribute

kevinbotlib.comm.redis.RedisCommClient

Bases: AbstractSetGetNetworkClient, AbstractPubSubNetworkClient

SENDABLE_TYPES = DEFAULT_SENDABLES class-attribute

redis = None instance-attribute

on_connect = on_connect instance-attribute

on_disconnect = on_disconnect instance-attribute

running = False instance-attribute

sub_thread = None instance-attribute

hooks = [] instance-attribute

pubsub = None instance-attribute

sub_callbacks = {} instance-attribute

host property writable

Get the currently connected server host.

Returns:

Type Description
str

Server host.

port property writable

Get the currently connected server port.

Returns:

Type Description
int

Server port.

unix_socket property writable

timeout property

Get the current server timeout.

Returns:

Type Description
float

Server timeout in seconds.

__init__(host='localhost', port=6379, unix_socket=None, db=0, timeout=2, on_connect=None, on_disconnect=None)

Initialize a Redis Communication Client.

Parameters:

Name Type Description Default
host str

Host of the Redis server.

 'localhost'
port int

Port of the Redis server.

 6379
unix_socket str | None

Optional UNIX socket path. A UNIX socket path is preferred over TCP.

 None
db int

Database number to use.

 0
timeout float

Socket timeout in seconds.

 2
on_connect Callable[[], None] | None

Connection callback.

 None
on_disconnect Callable[[], None] | None

Disconnection callback.

 None

register_type(data_type)

Register a custom sendable type.

Parameters:

Name Type Description Default
data_type type[BaseSendable]

Sendable type to register.

 required

add_hook(key, data_type, callback)

Add a callback to be triggered when sendable of data_type is set for a key.

Parameters:

Name Type Description Default
key CommPath | str

Key to listen to.

 required
data_type type[T]

Sendable type to listen for.

 required
callback Callable[[str, T | None], None]

Callback to trigger.

 required

get(key, data_type)

Retrieve and deserialize sendable by key.

Parameters:

Name Type Description Default
key CommPath | str

Key to retrieve.

 required
data_type type[T]

Sendable type to deserialize to.

 required

Returns:

Type Description
T | None

Sendable or None if not found.

multi_get(requests)

Retrieve and deserialize multiple sendables by a list of GetRequest objects.

Parameters:

Name Type Description Default
requests list[GetRequest]

List of GetRequest objects.

 required

Returns:

Type Description

List of sendables or None for each request if not found.

get_keys()

Gets all keys in the Redis database.

Returns:

Type Description
list[str]

List of keys.

get_raw(key)

Retrieve the raw JSON for a key, ignoring the sendable deserialization.

Parameters:

Name Type Description Default
key CommPath | str

Key to retrieve.

 required

Returns:

Type Description
dict | None

Raw JSON value or None if not found.

get_all_raw()

Retrieve all raw JSON values as a dictionary of a key to raw value. May have slow performance.

Returns:

Type Description
dict[str, dict] | None

Dictionary of a key to raw value or None if not found.

set(key, sendable)

Set sendable in the Redis database.

Parameters:

Name Type Description Default
key CommPath | str

Key to set

 required
sendable BaseSendable | SendableGenerator

Sendable to set

 required

multi_set(requests)

Set multiple sendables in the Redis database.

Parameters:

Name Type Description Default
requests list[SetRequest]

Sequence of SetRequest objects.

 required

publish(key, sendable)

Publish sendable in the Redis Pub/Sub client.

Parameters:

Name Type Description Default
key CommPath | str

Key to publish to

 required
sendable BaseSendable | SendableGenerator

Sendable to publish

 required

subscribe(key, data_type, callback)

Subscribe to a Pub/Sub key.

Parameters:

Name Type Description Default
key CommPath | str

Key to subscribe to.

 required
data_type type[T]

Sendable type to deserialize to.

 required
callback Callable[[str, T], None]

Callback when data is received.

 required

wipeall()

Delete all keys in the Redis database.

delete(key)

Delete a key from the Redis database.

Parameters:

Name Type Description Default
key CommPath | str

Key to delete.

 required

connect()

Connect to the Redis server.

is_connected()

Check if the Redis connection is established.

Returns:

Type Description
bool

Is the connection established?

get_latency()

Measure the round-trip latency to the Redis server in milliseconds.

Returns:

Type Description
float | None

Latency in milliseconds or None if not connected.

wait_until_connected(timeout=5.0)

Wait until the Redis connection is established.

Parameters:

Name Type Description Default
timeout float

Timeout in seconds. Defaults to 5.0 seconds.

 5.0

close()

Close the Redis connection and stop the pubsub thread.

reset_connection()

Reset the connection to the Redis server

kevinbotlib.comm.sendables

kevinbotlib.comm.sendables.DEFAULT_SENDABLES = {'kevinbotlib.dtype.int': IntegerSendable, 'kevinbotlib.dtype.bool': BooleanSendable, 'kevinbotlib.dtype.str': StringSendable, 'kevinbotlib.dtype.float': FloatSendable, 'kevinbotlib.dtype.list.any': AnyListSendable, 'kevinbotlib.dtype.dict': DictSendable, 'kevinbotlib.dtype.bin': BinarySendable, 'kevinbotlib.dtype.coord2d': Coord2dSendable, 'kevinbotlib.dtype.coord3d': Coord3dSendable, 'kevinbotlib.dtype.angle2d': Angle2dSendable, 'kevinbotlib.dtype.angle3d': Angle3dSendable, 'kevinbotlib.dtype.pose2d': Pose2dSendable, 'kevinbotlib.dtype.pose3d': Pose3dSendable, 'kevinbotlib.dtype.list.coord2d': Coord2dListSendable, 'kevinbotlib.dtype.list.coord3d': Coord3dListSendable, 'kevinbotlib.dtype.list.angle2d': Angle2dListSendable, 'kevinbotlib.dtype.list.angle3d': Angle3dListSendable, 'kevinbotlib.dtype.list.pose2d': Pose2dListSendable, 'kevinbotlib.dtype.list.pose3d': Pose3dListSendable} module-attribute

kevinbotlib.comm.sendables.BaseSendable

Bases: BaseModel, ABC

The base for all of KevinbotLib's sendables.

What is a sendable?

A sendable is a basic unit of data that can be transported through the RedisCommClient and server

timeout = None class-attribute instance-attribute

data_id = 'kevinbotlib.dtype.null' class-attribute instance-attribute

Internally used to differentiate sendable types

flags = [] class-attribute instance-attribute

struct = {} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.SendableGenerator

Bases: ABC

Abstract class for a function capable of being sent over RedisCommClient

generate_sendable() abstractmethod

Abstract method to generate a sendable

Returns:

Name Type Description
BaseSendable BaseSendable

The returned sendable

kevinbotlib.comm.sendables.IntegerSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.int' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.BooleanSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.bool' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.StringSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.str' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.FloatSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.float' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.AnyListSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.list.any' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.DictSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.dict' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.BinarySendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.bin' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'limit:1024'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Coord2dSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.coord2d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Coord3dSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.coord3d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Angle2dSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.angle2d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Angle3dSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.angle3d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Pose2dSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.pose2d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Pose3dSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.pose3d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Coord2dListSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.list.coord2d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Coord3dListSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.list.coord3d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Angle2dListSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.list.angle2d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Angle3dListSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.list.angle3d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Pose2dListSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.list.pose2d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.sendables.Pose3dListSendable

Bases: BaseSendable

value instance-attribute

Value to send

data_id = 'kevinbotlib.dtype.list.pose3d' class-attribute instance-attribute

Internally used to differentiate sendable types

struct = {'dashboard': [{'element': 'value', 'format': 'raw'}]} class-attribute instance-attribute

Data structure suggestion for use in dashboard applications

get_dict()

Return the sendable in dictionary form

Returns:

Name Type Description
dict dict

The sendable data

kevinbotlib.comm.path

kevinbotlib.comm.path.CommPath

Class representing a forward-slash-separated path for use in a comms client.

path property

Get the path as a string.

Returns:

Type Description
str

Path

__init__(path)

Create a new CommPath object.

Parameters:

Name Type Description Default
path str | CommPath

Optional path to extend from

 required

kevinbotlib.comm.pipeline

kevinbotlib.comm.pipeline.PipelinedCommSetter

Pipeline creator for optimized comm setting. Normally, a single set request may take as long as many pipelined requests, due to overhead. Using pipelines greatly optimizes robot cycle times.

client = client instance-attribute

set_queue = [] instance-attribute

__init__(client)

Create a new comm setter pipeline.

Parameters:

Name Type Description Default
client AbstractSetGetNetworkClient

Client to use for the pipeline.

 required

add(request)

Add a new set request to the pipeline.

Parameters:

Name Type Description Default
request SetRequest

Set Request

 required

set(key, sendable)

Add a new sendable to the pipeline.

Parameters:

Name Type Description Default
key CommPath | str

Key to set

 required
sendable BaseSendable

Sendable to set

 required

extend(requests)

Adds a sequence of set requests to the pipeline.

Parameters:

Name Type Description Default
requests Sequence[SetRequest]

Set requests

 required

send()

Apply the pipeline to the network client.

kevinbotlib.comm.request

kevinbotlib.comm.request.GetRequestTypeVar = TypeVar('GetRequestTypeVar', bound=BaseSendable) module-attribute

kevinbotlib.comm.request.SetRequestTypeVar = TypeVar('SetRequestTypeVar', bound=BaseSendable) module-attribute

kevinbotlib.comm.request.GetRequest dataclass

Bases: Generic[GetRequestTypeVar]

Dataclass for a Comm Get Request

key instance-attribute

Key to retrieve

data_type instance-attribute

Sendable type

__init__(key, data_type)

kevinbotlib.comm.request.SetRequest dataclass

Bases: Generic[SetRequestTypeVar]

Dataclass for a Comm Set Request

key instance-attribute

Key to set

data instance-attribute

Sendable to set

__init__(key, data)

kevinbotlib.comm.abstract

kevinbotlib.comm.abstract.Ts = TypeVar('Ts', bound=BaseSendable) module-attribute

kevinbotlib.comm.abstract.SetGetClientWithPubSub = type('SetGetClientWithPubSub', (AbstractSetGetNetworkClient, AbstractPubSubNetworkClient), {}) module-attribute

kevinbotlib.comm.abstract.AbstractSetGetNetworkClient

Bases: ABC

host abstractmethod property writable

port abstractmethod property writable

timeout abstractmethod property

register_type(data_type) abstractmethod

add_hook(key, data_type, callback) abstractmethod

get(key, data_type) abstractmethod

multi_get(requests) abstractmethod

get_keys() abstractmethod

get_raw(key) abstractmethod

get_all_raw() abstractmethod

set(key, sendable) abstractmethod

multi_set(requests) abstractmethod

delete(key) abstractmethod

connect() abstractmethod

is_connected() abstractmethod

wait_until_connected() abstractmethod

close() abstractmethod

kevinbotlib.comm.abstract.AbstractPubSubNetworkClient

Bases: ABC

host abstractmethod property writable

port abstractmethod property writable

timeout abstractmethod property

register_type(data_type) abstractmethod

subscribe(key, data_type, callback) abstractmethod

publish(key, sendable) abstractmethod

connect() abstractmethod

is_connected() abstractmethod

wait_until_connected() abstractmethod

close() abstractmethod