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)[source]
Bases:
object
Subscribes to
uavcan.diagnostic.Record
and forwards every received message into Python’slogging
. The logger name is that of the current module. The log level mapping is defined bySEVERITY_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)[source]
- class pycyphal.application.diagnostic.DiagnosticPublisher(node: pycyphal.application._node.Node, level: int = 30)[source]
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 [source]
- property timestamping_enabled: bool[source]
If True, the publisher will be setting the field
timestamp
of the published log messages tologging.LogRecord.created
(with the appropriate unit conversion). If False (default), published messages will not be timestamped at all.
- emit(record: logging.LogRecord) None [source]
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 [source]