Client

Client#

class pyaseba.client.Client(port: int = -1, address: str = '0.0.0.0', ping_period_ms: int = 1000, automatic_query: bool = True, node_disconnection_timeout_ms: int = 3000, min_protocol_version: int = ..., max_protocol_version: int = ...)#

If port if positive and address is a valid IP4 address, it will listen for incoming connections. Other clients will be able to connect the target at “tcp:host=<address>;port=<port>”.

Parameters:

port – The port. Pass a negative number to disable listening for incoming connections.
address – If a valid IP address. Pass an empty to disable listening for incoming connections.
ping_period_ms – The period in milliseconds to broadcast node discovery messages (pyaseba.client.msgs.ListNodes). Set it to zero to disable node discovery.
automatic_query – Whether to automatically query discovered nodes for their description. If selected, it will effectively call query_description() when a presence message from a new node is received.
node_disconnection_timeout_ms – The maximal interval to consider a node as disconnected. Only relevant when pinging the network.
min_protocol_version – minimal Aseba protocol version that nodes must satisfy to interact with them.
max_protocol_version – maximal Aseba protocol version that nodes must satisfy to interact with them.

Examples

The typical life-cycle of a client starts by creating it

>>> client = Client()

and then connecting to one or more networks.

>>> connection = client.connect(...)
1

After interacting with the networks, we should close the connection

>>> client.close_connection(connection)
True

and/or close the client, which would also close all connections

>>> client.close()

Alternatively, we can use the client in a context that closes it automatically when it exits.

>>> with Client() as client:
        connection = client.connect(...)
        ...

add_connection_callback(self, callback: Callable[[int, str], None]) → None#

Adds a callback called when a connection is opened.

Parameters:: callback – The callback that receives the connection as argument.

add_disconnection_callback(self, callback: Callable[[int, str], None]) → None#

Adds a callback called when a connection is closed.

Parameters:: callback – The callback that receives the connection as argument.

add_event_callback(self, callback: Callable[[Event], None]) → None#

Adds a callback called when an event is received.

Parameters:: callback – A callback.

add_message_callback(self, callback: Callable[[Message, int], None]) → None#

Adds a callback called when a message is received.

Parameters:: callback – A callback that receives a two arguments (message, connection).

add_node_callback(self, callback: Callable[[int, int], None]) → None#

Adds a callback called called when a node is discovered.

Parameters:: callback – A callback that receives a two arguments (node_id, connection).

add_node_callback(self, callback: Callable[[int, int], None]) -> None

Adds a callback called called when a node is discovered.

Parameters:: callback – A callback that receives a two arguments (node_id, connection).

add_node_disconnection_callback(self, callback: Callable[[int, int], None]) → None#

Adds a callback called when a node is disconnected.

Parameters:: callback – A callback that receives a two arguments (node_id, connection).

advertise(self, name: str = '', nodes: Sequence[tuple[int, int, str]] = (), protocol_version: int = ...) → None#

Advertise with zeroconf.

Parameters:

name – The name of the zeroconf record. If empty, it defaults to “pyaseba <port>”
nodes – A list of (id, pid, name) node tuples.
protocol_version – The Aseba protocol version.

advertise_nodes()#

advertise(self, name: str = “”, specs: dict[str, tuple[str, int]] = {“thymio-II: (“Thymio II”, 8)}, query_product_id: bool = false) -> None

Advertise all nodes with zeroconf.

Parameters:

name – The name of the zeroconf record. If empty, it defaults to “pyaseba <port>”
specs – An optional mapping between node description names and a tuple with name and product id.
query_product_id – Whether to query the product id. If not set and no key is provided in specs, the product id will default to 0.
protocol_version – The Aseba protocol version.

property automatic_query#

Readonly

Whether to automatically query discovered nodes for their description. If set, it will effectively call query_description() when a presence message from a new node is received.

clear_incoming_messages(self) → None#: Deletes all incoming messages.

clear_message_callbacks(self) → None#: Clears all message callbacks.

close(self) → None#: Closes all connections.

close_connection(self, connection: int, wait_ms: int = 1000) → bool#

Closes a connection.

Parameters:

connection – a connection
wait_ms – time to wait for the connection to close.

Returns:

Whether the connection has been closed.

cmd_pause(self, node_id: int, include: set[int] = set(), exclude: set[int] = set()) → None#

Sends a command to a node to pause

Parameters:

node_id – The id of the node.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

cmd_reset(self, node_id: int, include: set[int] = set(), exclude: set[int] = set()) → None#

Sends a command to a node to reset

Parameters:

node_id – The id of the node.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

cmd_run(self, node_id: int, include: set[int] = set(), exclude: set[int] = set()) → None#

Sends a command to a node to start running

Parameters:

node_id – The id of the node.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

cmd_sleep(self, node_id: int, include: set[int] = set(), exclude: set[int] = set()) → None#

Sends a command to a node to sleep

Parameters:

node_id – The id of the node.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

cmd_stop(self, node_id: int, include: set[int] = set(), exclude: set[int] = set()) → None#

Sends a command to a node to stop running

Parameters:

node_id – The id of the node.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

connect(self, target: str, wait_ms: int = 1000, max_retries: int = 3, **kwargs: Any) → int#

Connects to an Aseba network.

Parameters:

target – a valid Dashel target
wait_ms – time to wait before retrying to connect in case of failure.
max_retries – maximal number of time to try to connect before returning a failure.
**kwargs – parameters that are appended to target as "<key>=<value>". For example, if target is "tcp", passing port=33333 will result in a target "tcp:port=33333".

Returns:

The positive index of the connected network in case of success, or 0 in case of failure.

property connections#

Readonly

A dictionary of connected dashel target indexed by connection.

deadvertise(self, name: str = '') → None#

De-advertise with zeroconf.

Parameters:: name – The name of the zeroconf record. If empty, it defaults to “pyaseba <port>”

property descriptions#

Readonly

A dictionary with all discovered node descriptions indexed by connection.

emit_event(self, node_id: int, name: str, *, payload: Sequence[int] = (), include: set[int] = set(), exclude: set[int] = set()) → Description | None#

Send a message.

Parameters:

node_id – The id of the node defining the event.
name – The name of the event.
payload – The payload of the event
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

get_all_variables(self, node_id: int, wait_ms: int = 1000, callback: Callable[[list[int]], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → dict[str, list[int]]#

Query a node for the value of all variables

Parameters:

node_id – The id of the node.
wait_ms – The maximal time (in ms) to wait for answers.
callback – An optional callback called when the value is received.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

get_changed_variables(self, node_id: int, wait_ms: int = 1000, callback: Callable[[list[int]], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → list[tuple[int, list[int]]]#

Query a node for the value of variables that have changed.

Only supported if built against Mobsya Aseba, see pyaseba.uses_mobsya_aseba().

Parameters:

node_id – The id of the node.
wait_ms – The maximal time (in ms) to wait for answers.
callback – An optional callback called when the value is received.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

get_description(self, node_id: int, include: set[int] = set(), exclude: set[int] = set(), connected: bool = True) → Description | None#

Gets a node description.

Parameters:

node_id – The id of the node to wait. If negative, it will match any node id.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.
connected – An optional criterion for the current connection state of the node.

Returns:

The node description or None is no suitable node was found.

get_descriptions(self, include: set[int] = set(), exclude: set[int] = set(), connected: bool = True) → dict[int, dict[int, Description]]#

Gets the discovered node descriptions.

Parameters:

include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.
connected – An optional criterion for the current connection state of the node.

Returns:

A dictionary with discovered node descriptions indexed by connection.

get_event(self, node_id: int, name: str, wait_ms: int = 1000, callback: Callable[[Event], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → Description | None#

Wait until an event is received.

Parameters:

node_id – The id of the node.
name – The name of the event.
wait_ms – The maximal time in milliseconds to wait.
callback – An optional callback called after the event is received.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

Returns:

A message or None if not received in time.

get_message(self, node_id: int = -1, types: set[int] = set(), wait_ms: int = 1000, callback: Callable[[Message, int], None] | None = None, include: set[int] = set(), exclude: set[int] = set(), pause: bool = false) → tuple[Message | None, int]#

Wait until a message is received.

Parameters:

node_id – The id of the node.
types – The type of messages. Leave empty to accept all messages.
wait_ms – The maximal time in milliseconds to wait.
callback – An optional callback called after the message is received.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.
pause – Whether to stop processing once a message is received.

Returns:

A message or None if not received in time.

get_node_ids(self, include: set[int] = set(), exclude: set[int] = set(), connected: bool = True) → dict[int, set[int]]#

Gets the discovered node ids.

Parameters:

include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.
connected – An optional criterion for the current connection state of the node.

Returns:

A dictionary with discovered node ids indexed by connection.

get_script(self, node_id: int, include: set[int] = set(), exclude: set[int] = set()) → str#

Gets a Aseba script loaded on the node (if any).

Parameters:

node_id – The id of the node to wait. If negative, it will match any node id.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

Returns:

The text of the Aseba script or an empty string, if none has been loaded.

get_variable(self, node_id: int, name: str, wait_ms: int = 1000, callback: Callable[[list[int]], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → list[int]#

Query a node for the value of a variable by name.

Parameters:

node_id – The id of the node.
name – The name of the variable.
wait_ms – The maximal time (in ms) to wait for answers.
callback – An optional callback called when the value is received.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

get_variable_by_index(self, node_id: int, index: int, length: int, wait_ms: int = 1000, callback: Callable[[list[int]], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → list[int]#

Query a node for the value of a variable by index and length

Parameters:

node_id – The id of the node.
index – The index of the variable.
length – The size of the variable.
wait_ms – The maximal time (in ms) to wait for answers.
callback – An optional callback called when the value is received.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

has_node(self, node_id: int, include: set[int] = set(), exclude: set[int] = set(), connected: bool = True) → bool#

Checks if a node is known.

Parameters:

node_id – The id of the node.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.
connected – An optional criterion for the current connection state of the node.

Returns:

True if a corresponding node is known.

property is_connected#

Readonly

Whether at least the client has at least one open connection.

load_script(self, node_id: int, script: str, events: dict[str, int] = {}, constants: dict[str, int] = {}, include: set[int] = set(), exclude: set[int] = set()) → None#

Set the value of a variable by index.

Parameters:

node_id – The id of the node.
script – The Aseba code.
events – A dictionary of {name -> payload size} defining each event in the script.
constants – A dictionary of {name -> value} defining each constant in the script.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

Raises:

RuntimeError – when it fails to compile the script.

property node_disconnection_timeout_ms#: The maximal interval to consider a node as disconnected. Only relevant when pinging the network.

property node_ids#

Readonly

A dictionary with all discovered node ids indexed by connection.

property pause_processing#: Whether incoming messages are processed or kept in the queue.

ping(self) → None#

Broadcast a pyaseba.client.msgs.ListNodes message on all connected networks.

Parameters:

include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

property ping_period_ms#: The period in milliseconds to broadcast node discovery messages (pyaseba.client.msgs.ListNodes). Set it to zero to disable node discovery.

property port#

Readonly

Whether at least the client has at least one open connection.

query_description(self, node_id: int, wait_ms: int = 1000, callback: Callable[[Description], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → Description | None#

Queries the description of a node.

Parameters:

node_id – The id of the node.
wait_ms – The maximal time in milliseconds to wait.
callback – An optional callback called after the description is received.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

Returns:

A description or None if not received in time.

query_description_fragment(self, node_id: int, fragment: int, wait_ms: int = 1000, callback: Callable[[DescriptionFragment], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → DescriptionFragment | None#

Queries a fragment of the description of a node.

Only supported if built against Mobsya Aseba, see pyaseba.uses_mobsya_aseba().

Parameters:

node_id – The id of the node.
fragment – -1 corresponds to the header, -2 to the whole description, while positive values to the index of the desired fragment, i.e. 0 for the first variable, 1 for the second and so on through all variables, events and functions (until the last functions).
wait_ms – The maximal time in milliseconds to wait.
callback – An optional callback called after the message is received.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

Returns:

An Aseba message with the description or None if not received in time.

query_device_info(self, node_id: int, type: DeviceInfoType, wait_ms: int = 1000, callback: Callable[[DeviceInfo], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → DeviceInfo | None#

Retrieves device information from a node.

Only supported if built against Mobsya Aseba, see pyaseba.uses_mobsya_aseba().

Parameters:

node_id – The id of the node.
type – The type of information.
wait_ms – The maximal time in milliseconds to wait.
callback – An optional callback called after the message is received.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

Returns:

An Aseba message with the information or None if not received in time.

query_device_name(self, node_id: int, wait_ms: int = 1000, callback: Callable[[str], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → str#

Retrieves the device name from a node.

Only supported if built against Mobsya Aseba, see pyaseba.uses_mobsya_aseba().

Parameters:

node_id – The id of the node.
wait_ms – The maximal time in milliseconds to wait.
callback – An optional callback called after the message is received.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

Returns:

The name. Will be empty if not received (or not set by the device)

query_device_uuid(self, node_id: int, wait_ms: int = 1000, callback: Callable[[list[int]], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → list[int]#

Retrieves the device uuid from a node.

Only supported if built against Mobsya Aseba, see pyaseba.uses_mobsya_aseba().

Parameters:

node_id – The id of the node.
wait_ms – The maximal time in milliseconds to wait.
callback – An optional callback called after the message is received.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

Returns:

A list of bytes. Will be empty if not received (or not set by the device).

query_thymio_rf_settings(self, node_id: int, wait_ms: int = 1000, callback: Callable[[ThymioRFSettings], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → ThymioRFSettings | None#

Retrieves the RF settings from a node.

Only supported if built against Mobsya Aseba, see pyaseba.uses_mobsya_aseba().

Warning

The Thymio does not read the RF channel, which is always set to 0xFFFF. The actual supported channels are 0, 1, and 2.

Parameters:

node_id – The id of the node.
wait_ms – The maximal time in milliseconds to wait.
callback – An optional callback called after the message is received.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

Returns:

The settings.

remove_message_callback(self, index: int) → None#

Removes a message callback.

Parameters:: index – The index of the callback.

scan(self, number: int = -1, wait_ms: int = 1000, callback: Callable[[int, int, bool], None] | None = None) → dict[int, set[int]]#

Scans for nodes on all connected networks.

Parameters:

number – The minimal number of nodes to find before returning.
wait_ms – The maximal time in ms to wait.
callback – An optional callback called each time a node is found. It receives a three arguments (node_id, connection, complete) where complete is True only if number nodes have been found.

Returns:

A dictionary of node ids indexed by connection.

send_event(self, type: int, *, payload: Sequence[int] = (), include: set[int] = set(), exclude: set[int] = set()) → Description | None#

Send an event of arbitrary type. Same as send_user_message().

Parameters:

type – The type of the event.
payload – The payload of the event
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

send_message(self, message: Message, include: set[int] = set(), exclude: set[int] = set()) → Description | None#

Send a message.

Parameters:

message – The message.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

send_user_message(self, type: int = -1, payload: Sequence[int] = (), include: set[int] = set(), exclude: set[int] = set()) → Description | None#

Send a message of arbitrary type.

Parameters:

type – The type of the Aseba message
payload – The payload of the message
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

set_device_info(self, node_id: int, type: DeviceInfoType, value: list[int], include: set[int] = set(), exclude: set[int] = set()) → None#

Send a request to set device information to a node.

Only supported if built against Mobsya Aseba, see pyaseba.uses_mobsya_aseba().

Parameters:

node_id – The id of the node.
type – The type of information.
data – A list of bytes.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

set_device_name(self, node_id: int, name: str, include: set[int] = set(), exclude: set[int] = set()) → None#

Send a request to set the device name to a node.

Only supported if built against Mobsya Aseba, see pyaseba.uses_mobsya_aseba().

Parameters:

node_id – The id of the node.
name – The name.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

set_device_uuid(self, node_id: int, uuid: list[int], include: set[int] = set(), exclude: set[int] = set()) → None#

Send a request to set the device uuid to a node.

Only supported if built against Mobsya Aseba, see pyaseba.uses_mobsya_aseba().

Parameters:

node_id – The id of the node.
type – The type of information.
uuid – A list of bytes.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

set_thymio_rf_settings(self, node_id: int, settings: ThymioRFSettings, include: set[int] = set(), exclude: set[int] = set()) → None#

Send a request to set RF settings to a node.

Only supported if built against Mobsya Aseba, see pyaseba.uses_mobsya_aseba().

Parameters:

node_id – The id of the node.
settings – The settings.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

set_variable(self, node_id: int, name: str, value: Sequence[int], wait_ms: int = 1000, include: set[int] = set(), exclude: set[int] = set()) → None#

Set the value of a variable by name

Parameters:

node_id – The id of the node.
name – The name of the variable.
value – The value.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

set_variable_by_index(self, node_id: int, index: int, value: Sequence[int], wait_ms: int = 1000, include: set[int] = set(), exclude: set[int] = set()) → None#

Set the value of a variable by index.

Parameters:

node_id – The id of the node.
index – The index of the variable.
value – The value.
include – If not empty, restricts to networks specified in this set.
exclude – Ignores networks specified in this set.

wait_connection(self, connection: int = 0, wait_ms: int = 1000, callback: Callable[[int], None] = None) → tuple[int, str]#

Waits for a new connection.

Parameters:

connection – the connection to wait. Set it to -1 to wait for the any connection.
wait_ms – the maximal time in ms to wait for a connection.
callback – An optional callback called when a connection is established. It receives the connection as argument.

Returns:

The positive index of the new connection or 0 in case of timeouts.

wait_disconnection(self, connection: int = 0, wait_ms: int = 1000, callback: Callable[[int], None] = None) → tuple[int, str]#

Waits for a disconnection.

Parameters:

connection – the connection to wait to disconnect. Set it to -1 to wait for the any disconnection.
wait_ms – the maximal time in ms to wait for a disconnection.
callback – An optional callback called when a connection is closed. It receives the connection as argument.

Returns:

The positive index of the connection closed or 0 in case of timeouts.

wait_node(self, node_id: int = -1, wait_ms: int = 1000, callback: Callable[[int, int], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → tuple[int, int]#

Waits until one node is discovered.

Nodes are considered as discovered when their complete description is received.

Parameters:

node_id – The id of the node to wait. If negative, it will match any node id.
wait_ms – The maximal time in milliseconds to wait.
callback – An optional callback, called each time a node is discovered. It receives a two arguments (node_id, connection).
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

Returns:

A tuple (node_id, connection). If it fails discovering a node, it returns (0, 0). Else connection will be strictly positive.

wait_node_disconnection(self, node_id: int = -1, wait_ms: int = 1000, callback: Callable[[int, int], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → tuple[int, int]#

Waits until one node is discovered.

Nodes are considered as discovered when their complete description is received.

Parameters:

node_id – The id of the node to wait. If negative, it will match any node id.
wait_ms – The maximal time in milliseconds to wait.
callback – An optional callback, called each time a node is discovered. It receives a two arguments (node_id, connection).
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

Returns:

A tuple (node_id, connection). If the node has not disconnected, it returns (0, 0). Else connection will be strictly positive.

wait_nodes(self, node_ids: set[int] = set(), number: int = -1, wait_ms: int = 1000, callback: Callable[[int, int, bool], None] | None = None, include: set[int] = set(), exclude: set[int] = set()) → dict[int, set[int]]#

Waits until nodes are discovered.

Nodes are considered as discovered when their complete description is received.

Parameters:

node_ids – A set of ids. If not empty, only nodes with ids in this set will be considered as discovered.
wait_ms – The maximal time in milliseconds to wait.
number – The number of nodes to discover.
callback – An optional callback, called each time a node is discovered. It receives a three arguments (node_id, connection, complete) where complete is True only if number nodes have been discovered.
include – If not empty, restricts the search to nodes on the networks specified in this set.
exclude – Ignore nodes on networks specified in this set.

Returns:

A dictionary with the discovered node ids indexed by connection.

Client

Contents

Client#