Node

Node#

class pyaseba.client.Node(cached: bool = False)#

A class to interact with a single remote Aseba node that 1) has a simpler interface compared to pyaseba.client.Client, 2) uses a cache for Aseba variables, 3) exposes Aseba native functions and events, 4) access Aseba variables, events and functions by name through generic methods or through specific attributes.

Under the hood, it uses pyaseba.client.Client to interact with a single Aseba node, either reusing an existent client or instantiating one when required.

Compared to pyaseba.client.Client, which manages a collections remote Aseba nodes, it simplify the interface by keeping track of the network and id of a single remote node.

Nodes can be configured to execute an Aseba script that exposes remote local events and native functions and keeps variables up-to-date. When local events are emitted by the remote node, a custom event, together with updated Aseba variables, is broadcasted; once it is received, the value local variables are automatically updated. The script also reacts to custom events that in turns call native functions.

Nodes also provide a interface to access variables, events and functions by specific attribute instead of by name, where dots in the variables name are replaced by underscores in the attribute names. For example, Aseba variable leds.top would be linked to Python property leds_top.

Examples

>>> node = Node()
>>> node.connect("tcp:port=33333")
True
>>> node.description.variables
{"value": ...}

Setting and getting variables does not require passing node_id

>>> node.set("value", [1, 2, 3])
>>> node.get("value")
[1, 2, 3]

Assuming that the remote node defines local Aseba event e

>>> class MyNode(Node):
>>>     mirroring_config = MirroringConfig(
...         events = {"e": EventSpec(variables=["a"])

after mirroring starts, the node will receive notifications when e is emitted

>>> node = MyNode()
>>> node.connect(..., start_mirroring=True)
>>> node.wait("e")
True
>>> node.set_callback("e", lambda node: ...)

and a will be updated without further explicit synchronization.

>>> node.get("a", cached=False)
...

Assuming that the Aseba node defines local function f that accepts two integers,

>>> class MyNode(Node):
>>>     mirroring_config = MirroringConfig(function_include=['f'])

after mirroring starts, the node will be able to call the function by name

>>> node = MyNode()
>>> node.connect(..., start_mirroring=True)
>>> node.call("f", 1, 2)

We can inspect the Aseba script loaded the remote node for mirroring

>>> print(node.script)
onevent call_f
call f args[1:3]
onevent e
emit event_e [a]

Assuming that the Aseba node has variables a and b

>>> class MyNode(Node):
>>>     properties = ["a", "b"]

exposes a and b as properties with cached values

>>> node = MyNode()
>>> node.connect(...)
>>> node.a = [1, 2]
>>> node.b
[3, 4, 5]

Callbacks for mirrored events are exposed as Python properties too. Following a previous example, instead of calling

>>> node.set_callback("e", cb)

we can set the attribute directly

>>> node.on_e = cb

The node class below expose Aseba functions as Python methods

>>> class MyNode(Node):
>>>     functions = ["f"]
...     ...
>>> ...

Instead of calling

>>> node.call("f", 1, 2)

we can call a specific method

>>> node.call_f(1, 2)

to perform the same work.

cached: bool#

The default value of cached used by set(), get(), and get_all()

call(name: str, *args: int) → None#

Calls a local function by emitting the corresponding user event.

Parameters:

  • name – The name of the local function

  • args – The arguments

close(reset: bool = False) → None#

Closes the node and the client, if it was created by the node.

Parameters:

reset – Whether to reset the remote node.

connect(client: Client | None = None, target: str = '', wait_ms: int = 5000, max_retries: int = 3, node_id: int = -1, start_mirroring: bool = False, **kwargs: Any) → bool#

Connect to a remote Aseba node through pyaseba.client.Client.connect()

Parameters:

  • client – The client. If not provided, it will instantiate a new client.

  • 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.

  • node_id – The node identifier. Negative value match any id.

  • start_mirroring – Whether to start mirroring after connecting. If not set, call start_mirroring() later to start mirroring.

  • 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:

Whether the connection was successful.

property connection: int#

The connected remote Aseba network

control_event = ''#

Which local event to use to trigger the control step

default_target = ''#

Default Dashel target

property description: Description | None#

The description

emit(name: str, *args: int) → None#

Emits an event

Parameters:

  • name – The name of the (user) event

  • args – The payload

property events: list[str]#

All Aseba user events are await-able through wait().

functions: Collection[str] = ()#

Which functions should be exposed as call_<name> methods

get(name: str, wait_ms: int = 1000, cached: bool | None = None) → int | list[int] | None#

Gets the value of an Aseba variable

Parameters:

  • name – The variable name

  • wait_ms – The maximal time to wait in milliseconds.

  • cached ({ type_description }) – Whether to return cached values if present. If false, it will query the remote object for an updated value. If not set, it will default to cached.

Returns:

None if no value is available. If the value has size 1, it returns an integer scalar, else a list of integer.

get_all(wait_ms: int = 1000, cached: bool | None = None, changed: bool = False) → dict[str, list[int]]#

Gets all variable.

Parameters:

  • wait_ms – The maximal time to wait in milliseconds.

  • cached – Whether to return cached values if present. If false, it will query the remote object for an updated value. If not set, it will default to cached.

  • changed – Whether to limit the query to changed variables. Only effective if pyaseba was uses Mobsya-Aseba, see pyaseba.uses_mobsya_aseba(). When changed=True, it ignores the value of cached.

Returns:

A dictionary of variable values keyed by variable names.

get_callback(name: str) → Callable[[Self], None] | None#

Gets the callback associated to a mirrored local event.

Parameters:

name – The name of the local event

Returns:

The callback, if any.

get_last_payload(event_name: str) → list[int]#

Gets the last event payload.

Parameters:

event_name – The event name

Returns:

The payload of the last received event, if any.

load_script(script: str, events: dict[str, int] = {}, constants: dict[str, int] = {}) → bool#

Loads an Aseba script.

Parameters:

  • script – The script

  • events – User defined events as dictionary of payload sizes keyed by name.

  • constants (User defined constants as dictionary of scalar sizes values keyed by name.) – The constants

Returns:

True if the script was successfully compiled and loaded.

property mirrored_events: list[str]#

Which local mirrored Aseba events are await-able through wait().

property mirrored_functions: list[str]#

Which Aseba functions are callable through call().

mirroring_config: MirroringConfig = MirroringConfig(events={}, function_include=[], function_exclude=[], script_inits=[])#

Mirroring config

property node_id: int#

The node id

pause() → None#

Sends a command to pause running the remote node.

properties: Collection[str] = ()#

Which variables should be exposed as Python properties

reset() → None#

Sends a command to pause reset the remote node.

run() → None#

Sends a command to start running the remote node. It is only effective if a script has been loaded.

property script: str#

The loaded Aseba code (if any)

set(name: str, value: Sequence[int] | int, cached: bool | None = None) → None#

Sets the value of an Aseba variable

Parameters:

  • name – The name of the variable

  • value – The value of the variable

  • cached – Whether to forward the change to the remote node immediately. If not, users need to call sync() to forward changes.

set_callback(name: str, callback: Callable[[Self], None] | None) → None#

Sets a callback for when a mirrored local event is emitted

Parameters:

  • name – The name of the local event

  • callback – The callback

set_control_period(time_step: float, event: str = '') → None#

A virtual method to setup the controller The base implementation is empty. Override to specialize the class.

Parameters:

  • time_step – The time step

  • event – The name of the local event that should trigger a control step.

set_controller(callback: Callable[[Self, float], None], time_step: float = 0.1, event: str = '') → None#

Setup a controller

Parameters:

  • callback – The callback called at each control step

  • time_step – The control time step

  • event – The event that should trigger a control step

setup() → None#

Virtual method called to finalize the object when a target is successfully connected. The base implementation is empty. Override to specialize the class.

start_mirroring() → None#

Starts a mirroring local events and native functions. through a custom Aseba script, as configured by mirroring_config.

stop() → None#

Sends a command to stop running the remote node.

sync() → None#

Forwards cached variables changes to the remote node.

Only variables selected by sync_include and not excluded by sync_exclude are forwarded.

sync_exclude: Collection[str] = ()#

Regular expressions for variables to be excluded from sync()

sync_include: Collection[str] = ('.*',)#

Regular expressions for variables to be included in sync()

property sync_variables: set[str]#

Which Aseba variables are included in synch().

property target: str#

The Dashel target of the remote Aseba network

update(wait_ms: int = 1000, changed: bool = False) → None#

Refresh the cache with the current (remote) value of all variables.

Parameters:

  • wait_ms – The maximal time to wait in milliseconds.

  • changed – Whether to limit the query to changed variables. Only effective if pyaseba was uses Mobsya-Aseba, see pyaseba.uses_mobsya_aseba().

wait(name: str, wait_ms: int = 1000) → bool#

Waits for an event.

If name is the name of a a mirrored local event, it waits for the corresponding user event.

Parameters:

  • name – The name of the event.

  • wait_ms – The maximal time to wait.

Returns:

Whether an event has been received before the deadline.