pycyphal.application.node_tracker module

Keeps track of online nodes by subscribing to uavcan.node.Heartbeat and requesting uavcan.node.GetInfo when necessary; see pycyphal.application.node_tracker.NodeTracker.

class pycyphal.application.node_tracker.Entry(heartbeat, info)[source]

Bases: tuple

The data kept per online node. The heartbeat is the latest received one. The info is None until the node responds to the GetInfo request.

__match_args__ = ('heartbeat', 'info')
static __new__(_cls, heartbeat: Heartbeat_1_0, info: Response | None)

Create new instance of Entry(heartbeat, info)


Return a nicely formatted representation string

__slots__ = ()
heartbeat: Heartbeat_1_0

Alias for field number 0

info: Response | None

Alias for field number 1


Arguments: node-ID, old entry, new entry. See NodeTracker.add_update_handler() for details.

alias of Callable[[int, Optional[Entry], Optional[Entry]], None]

class pycyphal.application.node_tracker.NodeTracker(node: Node)[source]

Bases: object

This class is designed for tracking the list of online nodes in real time. It subscribes to uavcan.node.Heartbeat to keep a list of online nodes. Whenever a new node appears online or an existing node is restarted (restart is detected via the uptime counter), the tracker invokes uavcan.node.GetInfo on it and keeps the response until the node is restarted again or until it goes offline (offline nodes detected via heartbeat timeout). If the node did not reply to uavcan.node.GetInfo, the request will be retried later.

If the local node is anonymous, the info request functionality will be automatically disabled; it will be re-enabled automatically if the local node is assigned a node-ID later (nodes that are already known at this time may not be queried).

The tracked node registry does not include the local node. If the local node-ID is N, the registry will not contain an entry at key N unless there is a node-ID conflict in the network.

The class provides IoC events which are triggered on change. The collected data can also be accessed by direct polling synchronously.


The logic tolerates the loss of responses, hence the optional priority level. This way, we can retry without affecting high-priority communications.


The default request timeout is larger than the recommended default because the data is immutable (does not lose validity over time) and the priority level is low which may cause delays.


Abandon efforts if the remote node did not respond to GetInfo this many times. The counter will resume from scratch if the node is restarted or a new node under that node-ID is detected.

__init__(node: Node)[source]
property node: Node[source]
property get_info_priority: Priority[source]

Allows the user to override the default uavcan.node.GetInfo request priority.

property get_info_timeout: float[source]

Allows the user to override the default uavcan.node.GetInfo request timeout. The value shall be a finite positive number.

property get_info_attempts: int[source]

Allows the user to override the default uavcan.node.GetInfo request retry limit. The value shall be a non-negative integer number. The value of zero disables GetInfo requests completely.

property registry: Dict[int, Entry][source]

Access the live online node registry. Keys are node-ID, values are Entry. The returned value is a copy of the actual registry to prevent accidental mutation. Elements are ordered by node-ID.

add_update_handler(handler: Callable[[int, Entry | None, Entry | None], None]) None[source]

Register a callable that will be invoked whenever the node registry is changed. The arguments are: node-ID, old entry, new entry. The handler is invoked in the following cases with the specified arguments:

  • New node appeared online. The old-entry is None. The new-entry info is None.

  • A known node went offline. The new-entry is None.

  • A known node restarted. Neither entry is None. The new-entry info is None.

  • A node responds to a uavcan.node.GetInfo request. Neither entry is None. The new-entry info is not None.

Received Heartbeat messages change the registry as well, but they do not trigger the hook. Handlers can be added and removed at any moment regardless of whether the instance is started.

remove_update_handler(handler: Callable[[int, Entry | None, Entry | None], None]) None[source]

Remove a previously added hook identified by referential equivalence. Behaves like list.remove().