pycyphal.application.diagnostic module

This module implements forwarding between the standard subject uavcan.diagnostic.Record and Python’s standard logging facilities (logging).

class pycyphal.application.diagnostic.DiagnosticSubscriber(node: pycyphal.application._node.Node)

Bases: object

Subscribes to uavcan.diagnostic.Record and forwards every received message into Python’s logging. The logger name is that of the current module. The log level mapping is defined by SEVERITY_CYPHAL_TO_PYTHON.

This class is convenient for various CLI tools and automation scripts where the user will not need to implement additional logic to see log messages from the network.

SEVERITY_CYPHAL_TO_PYTHON = {0: 20, 1: 20, 2: 20, 3: 20, 4: 30, 5: 40, 6: 50, 7: 50}

__init__(node: pycyphal.application._node.Node)

class pycyphal.application.diagnostic.DiagnosticPublisher(node: pycyphal.application._node.Node, level: int = 30)

Bases: logging.Handler

Implementation of logging.Handler that forwards all log messages via the standard diagnostics subject of Cyphal. Log messages that are too long to fit into a Cyphal Record object are truncated. Log messages emitted by PyCyphal itself may be dropped to avoid infinite recursion. No messages will be published if the local node is anonymous.

Here’s a usage example. Set up test rigging:

>>> from pycyphal.transport.loopback import LoopbackTransport
>>> from pycyphal.application import make_node, NodeInfo, make_registry
>>> node = make_node(NodeInfo(), transport=LoopbackTransport(1))
>>> node.start()

Instantiate publisher and install it with the logging system:

>>> diagnostic_pub = DiagnosticPublisher(node, level=logging.INFO)
>>> logging.root.addHandler(diagnostic_pub)
>>> diagnostic_pub.timestamping_enabled = True  # This is only allowed if the Cyphal network uses the wall clock.
>>> diagnostic_pub.timestamping_enabled
True

Test it:

>>> sub = node.make_subscriber(Record)
>>> logging.info('Test message')
>>> msg, _ = doctest_await(sub.receive_for(1.0))
>>> msg.text.tobytes().decode()
'root: Test message'
>>> msg.severity.value == Severity.INFO     # The log level is mapped automatically.
True

Don’t forget to remove it afterwards:

>>> logging.root.removeHandler(diagnostic_pub)
>>> node.close()

The node factory pycyphal.application.make_node() actually allows you to do this automatically, so that you don’t have to hard-code behaviors in the application sources:

>>> registry = make_registry(None, {"UAVCAN__DIAGNOSTIC__SEVERITY": "2", "UAVCAN__DIAGNOSTIC__TIMESTAMP": "1"})
>>> node = make_node(NodeInfo(), registry, transport=LoopbackTransport(1))
>>> node.start()
>>> sub = node.make_subscriber(Record)
>>> logging.info('Test message')
>>> msg, _ = doctest_await(sub.receive_for(1.0))
>>> msg.text.tobytes().decode()
'root: Test message'
>>> msg.severity.value == Severity.INFO
True
>>> node.close()

__init__(node: pycyphal.application._node.Node, level: int = 30) → None

property timestamping_enabled: bool

If True, the publisher will be setting the field timestamp of the published log messages to logging.LogRecord.created (with the appropriate unit conversion). If False (default), published messages will not be timestamped at all.

emit(record: logging.LogRecord) → None

This method intentionally drops all low-severity messages originating from within PyCyphal itself to prevent infinite recursion through the logging system.

static log_record_to_diagnostic_message(record: logging.LogRecord, use_timestamp: bool) → uavcan.diagnostic.Record_1_1.Record_1_1

__repr__() → str

pycyphal.application.diagnostic.Record

alias of uavcan.diagnostic.Record_1_1.Record_1_1

pycyphal.application.diagnostic.Severity

alias of uavcan.diagnostic.Severity_1_0.Severity_1_0