pycyphal.application.heartbeat_publisher module

Publishes uavcan.node.Heartbeat periodically and provides a couple of basic auxiliary services; see pycyphal.application.heartbeat_publisher.HeartbeatPublisher.


alias of uavcan.node.Heartbeat_1_0.Heartbeat_1_0

class pycyphal.application.heartbeat_publisher.Health(value)[source]

Bases: enum.IntEnum

Mirrors the health enumeration defined in uavcan.node.Heartbeat. When enumerations are natively supported in DSDL, this will be replaced with an alias.

class pycyphal.application.heartbeat_publisher.Mode(value)[source]

Bases: enum.IntEnum

Mirrors the mode enumeration defined in uavcan.node.Heartbeat. When enumerations are natively supported in DSDL, this will be replaced with an alias.

class pycyphal.application.heartbeat_publisher.HeartbeatPublisher(node: pycyphal.application._node.Node)[source]

Bases: object

This class manages periodic publication of the node heartbeat message. Also it subscribes to heartbeat messages from other nodes and logs cautionary messages if a node-ID conflict is detected on the bus.

The default states are as follows:

  • Health is NOMINAL.

  • Mode is OPERATIONAL.

  • Vendor-specific status code is zero.

  • Period is MAX_PUBLICATION_PERIOD (see the DSDL definition).

  • Priority is default (i.e., NOMINAL).

__init__(node: pycyphal.application._node.Node)[source]
property node: pycyphal.application._node.Node[source]
property uptime: float[source]

The current amount of time, in seconds, elapsed since the object was instantiated.

property health: pycyphal.application.heartbeat_publisher.Health[source]

The health value to report with Heartbeat; see Health.

property mode: pycyphal.application.heartbeat_publisher.Mode[source]

The mode value to report with Heartbeat; see Mode.

property vendor_specific_status_code: int[source]

The vendor-specific status code (VSSC) value to report with Heartbeat.

property period: float[source]

How often the Heartbeat messages should be published. The upper limit (i.e., the lowest frequency) is constrained by the Cyphal specification; please see the DSDL source of uavcan.node.Heartbeat.

property priority: pycyphal.transport._transfer.Priority[source]

The transfer priority level to use when publishing Heartbeat messages.

add_pre_heartbeat_handler(handler: Callable[[], None]) None[source]

Adds a new handler to be invoked immediately before a heartbeat message is published. The number of such handlers is unlimited. The handler invocation order follows the order of their registration. Handlers are invoked from a task running on the node’s event loop. Handlers are not invoked until the instance is started.

The handler can be used to synchronize the heartbeat message data (health, mode, vendor-specific status code) with external states. Observe that the handler will be invoked even if the heartbeat is not to be published, e.g., if the node is anonymous (does not have a node ID). If the handler throws an exception, it will be suppressed and logged. Note that the handler is to be not a coroutine but a regular function.

This is a good method of scheduling periodic status checks on the node.

make_message() uavcan.node.Heartbeat_1_0.Heartbeat_1_0[source]

Constructs a new heartbeat message from the object’s state.

__repr__() str[source]