pycyphal.application package

Subpackages

• pycyphal.application.register package

Submodules

• pycyphal.application.diagnostic module
• pycyphal.application.file module
• pycyphal.application.heartbeat_publisher module
• pycyphal.application.node_tracker module
• pycyphal.application.plug_and_play module

Module contents

Application layer overview

The application module contains the application-layer API. This module is not imported automatically because it depends on the transpiled DSDL namespace uavcan. The DSDL namespace can be either transpiled manually or lazily ad-hoc; see pycyphal.dsdl for related docs.

Node class

The abstract class pycyphal.application.Node models a Cyphal node — it is one of the main entities of the library, along with its factory make_node(). The application uses its Node instance to interact with the network: create publications/subscriptions, invoke and serve RPC-services.

Constructing a node

Create a node using the factory make_node() and start it:

>>> import pycyphal.application
>>> import uavcan.node                                  # Transcompiled DSDL namespace (see pycyphal.dsdl).
>>> node_info = pycyphal.application.NodeInfo(          # This is an alias for uavcan.node.GetInfo.Response.
...     software_version=uavcan.node.Version_1(major=1, minor=0),
...     name="org.uavcan.pycyphal.docs",
... )
>>> node = pycyphal.application.make_node(node_info)    # Some of the fields in node_info are set automatically.
>>> node.start()

The node instance we just started will periodically publish uavcan.node.Heartbeat and uavcan.node.port.List, respond to uavcan.node.GetInfo and uavcan.register.Access/uavcan.register.List, and do some other standard things – read the docs for Node for details.

Now we can create ports — that is, instances of pycyphal.presentation.Publisher, pycyphal.presentation.Subscriber, pycyphal.presentation.Client, pycyphal.presentation.Server — to interact with the network. To create a new port you need to specify its type and name (the name can be omitted if a fixed port-ID is defined for the data type).

Publishers and subscribers

Create a publisher and publish a message (here and below, doctest_await substitutes for the await statement):

>>> import uavcan.si.unit.voltage
>>> pub_voltage = node.make_publisher(uavcan.si.unit.voltage.Scalar_1, "measured_voltage")
>>> pub_voltage.publish_soon(uavcan.si.unit.voltage.Scalar_1(402.15))            # Publish message asynchronously.
>>> doctest_await(pub_voltage.publish(uavcan.si.unit.voltage.Scalar_1(402.15)))  # Or synchronously.
True

Create a subscription and receive a message from it:

>>> import uavcan.si.unit.length
>>> sub_position = node.make_subscriber(uavcan.si.unit.length.Vector3_1, "position_setpoint")
>>> msg = doctest_await(sub_position.get(timeout=0.5))              # None if timed out.
>>> round(msg.meter[0]), round(msg.meter[1]), round(msg.meter[2])   # Some payload in the message we received.
(42, 15, -9)

RPC-service clients and servers

Define an RPC-service of an application-specific type:

>>> from sirius_cyber_corp import PerformLinearLeastSquaresFit_1    # An application-specific DSDL definition.
>>> async def solve_linear_least_squares(                           # Refer to the Demo chapter for the DSDL sources.
...     request: PerformLinearLeastSquaresFit_1.Request,
...     metadata: pycyphal.presentation.ServiceRequestMetadata,
... ) -> PerformLinearLeastSquaresFit_1.Response:                   # Business logic.
...     import numpy as np
...     x = np.array([p.x for p in request.points])
...     y = np.array([p.y for p in request.points])
...     s, *_ = np.linalg.lstsq(np.vstack([x, np.ones(len(x))]).T, y, rcond=None)
...     return PerformLinearLeastSquaresFit_1.Response(slope=s[0], y_intercept=s[1])
>>> srv_least_squares = node.get_server(PerformLinearLeastSquaresFit_1, "least_squares")
>>> srv_least_squares.serve_in_background(solve_linear_least_squares)  # Run the server in a background task.

Invoke the service we defined above assuming that it is served by node 42:

>>> from sirius_cyber_corp import PointXY_1
>>> cln_least_sq = node.make_client(PerformLinearLeastSquaresFit_1, 42, "least_squares")
>>> req = PerformLinearLeastSquaresFit_1.Request([PointXY_1(10, 1), PointXY_1(20, 2)])
>>> response = doctest_await(cln_least_sq(req))                         # None if timed out.
>>> round(response.slope, 1), round(response.y_intercept, 1)
(0.1, 0.0)

Here is another example showcasing the use of a standard service with a fixed port-ID:

>>> client_node_info = node.make_client(uavcan.node.GetInfo_1, 42)    # Port name is not required.
>>> response = doctest_await(client_node_info(uavcan.node.GetInfo_1.Request()))
>>> response.software_version
uavcan.node.Version.1.0(major=1, minor=0)

Registers and application settings

You are probably wondering, how come we just created a node without specifying which transport it should use, its node-ID, or even the subject-IDs and service-IDs? Where did these values come from?

They were read from from the registry — a key-value configuration parameter storage [1] defined in the Cyphal Specification, chapter Application layer, section Register interface. The factory make_node() we used above just reads the registers and figures out how to construct the node from that: which transport to use, the node-ID, the subject-IDs, and so on. Any Cyphal application is also expected to keep its own configuration parameters in the registers so that it can be reconfigured and controlled at runtime via Cyphal.

The registry of the local node can be accessed via Node.registry which is an instance of class pycyphal.application.register.Registry:

>>> int(node.registry["uavcan.node.id"])        # Standard registers defined by Cyphal are named like "uavcan.*"
42
>>> node.id                                     # Yup, indeed, the node-ID is picked up from the register.
42
>>> int(node.registry["uavcan.pub.measured_voltage.id"])    # This is where we got the subject-ID from.
6543
>>> pub_voltage.port_id
6543
>>> int(node.registry["uavcan.sub.position_setpoint.id"])   # And so on.
6544
>>> str(node.registry["uavcan.sub.position_setpoint.type"]) # Port types are automatically exposed via registry, too.
'uavcan.si.unit.length.Vector3.1.0'

Every port created by the application (publisher, subscriber, etc.) is automatically exposed via the register interface as prescribed by the Specification [2].

New registers (application-specific registers in particular) can be created using pycyphal.application.register.Registry.setdefault():

>>> from pycyphal.application.register import Value, Real64  # Convenience aliases for uavcan.register.Value, etc.
>>> gains = node.registry.setdefault("my_app.controller.pid_gains", Real64([1.3, 0.8, 0.05]))   # Explicit real64 here.
>>> gains.floats
[1.3, 0.8, 0.05]
>>> import numpy as np
>>> node.registry.setdefault("my_app.estimator.state_vector",       # Not stored, but computed at every invocation.
...                          lambda: np.random.random(4)).floats    # Deduced type: real64.
[..., ..., ..., ...]

But the above does not explain where did the example get the register values from. There are two places:

• The register file which contains a simple key-value database table. If the file does not exist (like at the first run), it is automatically created. If no file location is provided when invoking make_node(), the registry is stored in memory so that all state is lost when the node is closed.

• The environment variables. A register like m.motor.inductance_dq can be assigned via environment variable M__MOTOR__INDUCTANCE_DQ (the mapping is documented in the standard RPC-service uavcan.register.Access). The value of an environment variable is a space-separated list of values (in case of arrays), or a plain string. The environment variables are checked once when the node is constructed, and also whenever a new register is created using pycyphal.application.register.Registry.setdefault().

>>> import os
>>> for k in os.environ:  # Suppose that the following environment variables were passed to our process:
...     if "__" in k:
...         print(k.ljust(40), os.environ[k])
UAVCAN__NODE__ID                         42
UAVCAN__PUB__MEASURED_VOLTAGE__ID        6543
UAVCAN__SUB__OPTIONAL_PORT__ID           65535
UAVCAN__UDP__IFACE                       127.0.0.1
UAVCAN__SERIAL__IFACE                    socket://127.0.0.1:50905
UAVCAN__DIAGNOSTIC__SEVERITY             3.1
M__MOTOR__INDUCTANCE_DQ                  0.12 0.13
>>> node = pycyphal.application.make_node(node_info, "registers.db")  # The file will be created if doesn't exist.
>>> node.id
42
>>> node.presentation.transport     # Heterogeneously redundant transport: UDP+Serial, as specified in env vars.
RedundantTransport(UDPTransport('127.0.0.1', local_node_id=42, ...), SerialTransport('socket://127.0.0.1:50905', ...))
>>> pub_voltage = node.make_publisher(uavcan.si.unit.voltage.Scalar_1, "measured_voltage")
>>> pub_voltage.port_id
6543
>>> int(node.registry["uavcan.diagnostic.severity"])                            # This is a standard register.
3
>>> node.registry.setdefault("m.motor.inductance_dq", [1.23, -8.15]).floats     # The value is taken from environment!
[0.12, 0.13]
>>> node.registry.setdefault("m.motor.flux_linkage_dq", [1.23, -8.15]).floats   # No environment variable for this one.
[1.23, -8.15]
>>> node.registry["m.motor.inductance_dq"] = [1.9, 6]                           # Assign new value.
>>> node.registry["m.motor.inductance_dq"].floats
[1.9, 6.0]
>>> node.make_subscriber(uavcan.si.unit.voltage.Scalar_1, "optional_port")      
Traceback (most recent call last):
...
PortNotConfiguredError: 'uavcan.sub.optional_port.id'
>>> node.close()

Per the Specification, a port-ID of 65535 (0xFFFF) represents an unconfigured port, as illustrated in the above snippet.

Application-layer function implementations

As mentioned in the description of the Node class, it provides certain bare-minumum standard application-layer functionality like publishing heartbeats, responding to GetInfo, serving the register API, etc. More complex capabilities are to be set up by the user as needed; some of them are:

pycyphal.application.diagnostic.DiagnosticSubscriber(node)	Subscribes to uavcan.diagnostic.Record and forwards every received message into Python's logging.
pycyphal.application.node_tracker.NodeTracker(node)	This class is designed for tracking the list of online nodes in real time.
pycyphal.application.plug_and_play.Allocatee(...)	Plug-and-play node-ID protocol client.
pycyphal.application.plug_and_play.Allocator()	An abstract PnP allocator interface.
pycyphal.application.file.FileServer(node, roots)	Exposes local filesystems via the standard RPC-services defined in uavcan.file.
pycyphal.application.file.FileClient(...[, ...])	This class is deprecated and should not be used in new applications; instead, consider using FileClient2.

[1]

Those familiar with ROS may find similarities with the ROS Parameter Server, except that each node keeps its own registers locally instead of relying on a remote centralized provider.

[2]

The application therefore should not attempt to create new ports using the presentation-layer API because that would circumvent the introspection services.

class pycyphal.application.Node

Bases: ABC

This is the top-level abstraction representing a Cyphal node on the bus. This is an abstract class; instantiate it using the factory pycyphal.application.make_node() or (in special cases) create custom implementations.

This class automatically instantiates the following application-layer function implementations:

• heartbeat_publisher.HeartbeatPublisher

• Register API server (uavcan.register.*)

• Node info server (uavcan.node.GetInfo)

• Port introspection publisher (uavcan.port.List)

Attention

If the underlying transport is anonymous, some of these functions may not be available.

Start the instance when initialization is finished by invoking start(). This will also automatically start all function implementation instances.

__init__() → None

abstract property presentation: Presentation

Provides access to the underlying instance of pycyphal.presentation.Presentation.

abstract property info: Response

Provides access to the local node info structure. See pycyphal.application.NodeInfo.

abstract property registry: Registry

Provides access to the local registry instance (see pycyphal.application.register.Registry). The registry manages Cyphal registers as defined by the standard network service uavcan.register.

The registers store the configuration parameters of the current application, both standard (like subject-IDs, service-IDs, transport configuration, the local node-ID, etc.) and application-specific ones.

See also make_publisher(), make_subscriber(), make_client(), get_server().

property loop: AbstractEventLoop

Deprecated; use asyncio.get_event_loop() instead.

property id: int | None

Shortcut for self.presentation.transport.local_node_id

property heartbeat_publisher: HeartbeatPublisher

Provides access to the heartbeat publisher instance of this node.

make_publisher(dtype: Type[pycyphal.application.T], port_name: str | int = '') → Publisher[pycyphal.application.T]

Wrapper over pycyphal.presentation.Presentation.make_publisher() that takes the subject-ID from the standard register uavcan.pub.PORT_NAME.id. If the register is missing or no name is given, the fixed subject-ID is used unless it is also missing. The type information is automatically exposed via uavcan.pub.PORT_NAME.type based on dtype. For details on the standard registers see Specification.

Experimental: the port_name may also be the integer port-ID. In this case, new port registers will be created with the names derived from the supplied port-ID (e.g., uavcan.pub.1234.id, uavcan.pub.1234.type). If ID registers created this way are overridden externally, the supplied ID will be ignored in favor of the override.

Raises:

PortNotConfiguredError if the register is not set and no fixed port-ID is defined. TypeError if no name is given and no fixed port-ID is defined.

make_subscriber(dtype: Type[pycyphal.application.T], port_name: str | int = '') → Subscriber[pycyphal.application.T]

Wrapper over pycyphal.presentation.Presentation.make_subscriber() that takes the subject-ID from the standard register uavcan.sub.PORT_NAME.id. If the register is missing or no name is given, the fixed subject-ID is used unless it is also missing. The type information is automatically exposed via uavcan.sub.PORT_NAME.type based on dtype. For details on the standard registers see Specification.

The port_name may also be the integer port-ID; see make_publisher() for details.

Raises:

PortNotConfiguredError if the register is not set and no fixed port-ID is defined. TypeError if no name is given and no fixed port-ID is defined.

make_client(dtype: Type[pycyphal.application.T], server_node_id: int, port_name: str | int = '') → Client[pycyphal.application.T]

Wrapper over pycyphal.presentation.Presentation.make_client() that takes the service-ID from the standard register uavcan.cln.PORT_NAME.id. If the register is missing or no name is given, the fixed service-ID is used unless it is also missing. The type information is automatically exposed via uavcan.cln.PORT_NAME.type based on dtype. For details on the standard registers see Specification.

The port_name may also be the integer port-ID; see make_publisher() for details.

Raises:

PortNotConfiguredError if the register is not set and no fixed port-ID is defined. TypeError if no name is given and no fixed port-ID is defined.

get_server(dtype: Type[pycyphal.application.T], port_name: str | int = '') → Server[pycyphal.application.T]

Wrapper over pycyphal.presentation.Presentation.get_server() that takes the service-ID from the standard register uavcan.srv.PORT_NAME.id. If the register is missing or no name is given, the fixed service-ID is used unless it is also missing. The type information is automatically exposed via uavcan.srv.PORT_NAME.type based on dtype. For details on the standard registers see Specification.

The port_name may also be the integer port-ID; see make_publisher() for details.

Raises:

PortNotConfiguredError if the register is not set and no fixed port-ID is defined. TypeError if no name is given and no fixed port-ID is defined.

start() → None

Starts all application-layer function implementations that are initialized on this node (like the heartbeat publisher, diagnostics, and basically anything that takes a node reference in its constructor). These will be automatically terminated when the node is closed. This method is idempotent.

close() → None

Closes the presentation (which includes the transport), the registry, the application-layer functions. The user does not have to close every port manually as it will be done automatically. This method is idempotent. Calling start() on a closed node may lead to unpredictable results.

add_lifetime_hooks(start: Callable[[], None] | None, close: Callable[[], None] | None) → None

The start hook will be invoked when this node is start()-ed. If the node is already started when this method is invoked, the start hook is called immediately.

The close hook is invoked when this node is close()-d. If the node is already closed, the close hook will never be invoked.

__enter__() → Node

Invokes start() upon entering the context. Does nothing if already started.

__exit__(*_: Any) → None

Invokes close() upon leaving the context. Does nothing if already closed.

__repr__() → str

pycyphal.application.NodeInfo

alias of Response

exception pycyphal.application.PortNotConfiguredError

Bases: MissingRegisterError

Raised from Node.make_publisher(), Node.make_subscriber(), Node.make_client(), Node.get_server() if the application requested a port for which there is no configuration register and whose data type does not have a fixed port-ID.

Applications may catch this exception to implement optional ports, where the port is not enabled until explicitly configured while other components of the application are functional.

pycyphal.application.make_node(info: Response, registry: None | Registry | str | Path = None, *, transport: Transport | None = None, reconfigurable_transport: bool = False) → Node

Initialize a new node by parsing the configuration encoded in the Cyphal registers.

Aside from the registers that encode the transport configuration (which are documented in make_transport()), the following registers are considered (if they don’t exist, they are automatically created). They are split into groups by application-layer function they configure.

General

Register name	Register type	Register semantics
uavcan.node.unique_id	unstructured	The unique-ID of the local node. This register is only used if the caller did not set unique_id in info. If not defined, a new random value is generated and stored as immutable (therefore, if no persistent register file is used, a new unique-ID is generated at every launch, which may be undesirable in some applications, particularly those that require PnP node-ID allocation).
uavcan.node.description	string	As defined by the Cyphal Specification, this standard register is intended to store a human-friendly description of the node. Empty by default and never accessed by the library, since it is intended mostly for remote use.

pycyphal.application.diagnostic

Register name	Register type	Register semantics
uavcan.diagnostic.severity	natural8[1]	If the value is a valid severity level as defined in uavcan.diagnostic.Severity, the node will publish its application log records of matching severity level to the standard subject uavcan.diagnostic.Record using pycyphal.application.diagnostic.DiagnosticPublisher. This is done by installing a root handler in logging. Disabled by default.
uavcan.diagnostic.timestamp	bit[1]	If true, the published log messages will initialize the synchronized timestamp field from the log record timestamp provided by the logging library. This is only safe if the Cyphal network is known to be synchronized on the same time system as the wall clock of the local computer. Otherwise, the timestamp is left at zero (which means “unknown” per Specification). Disabled by default.

Additional application-layer functions and their respective registers may be added later.

Parameters:

• info –

  Response object to uavcan.node.GetInfo. The following fields will be populated automatically:

  • protocol_version from pycyphal.CYPHAL_SPECIFICATION_VERSION.

  • If not set by the caller: unique_id is read from register as specified above.

  • If not set by the caller: name is constructed from hex-encoded unique-ID like: anonymous.b0228a49c25ff23a3c39915f81294622.

• registry – If this is an instance of pycyphal.application.register.Registry, it is used as-is (ownership is taken). Otherwise, this is a register file path (or None) that is passed over to pycyphal.application.make_registry() to construct the registry instance for this node. This instance will be available under pycyphal.application.Node.registry.

• transport – If not provided (default), a new transport instance will be initialized based on the available registers using make_transport(). If provided, the node will be constructed with this transport instance and take its ownership. In the latter case, existence of transport-related registers will NOT be ensured.

• reconfigurable_transport – If True, the node will be constructed with pycyphal.transport.redundant, which permits runtime reconfiguration. If the transport argument is given and it is not a redundant transport, it will be wrapped into one. Also see make_transport().

Raises:

• pycyphal.application.register.MissingRegisterError if a register is expected but cannot be found, or if no transport is configured.

• pycyphal.application.register.ValueConversionError if a register is found but its value cannot be converted to the correct type, or if the value of an environment variable for a register is invalid or incompatible with the register’s type (e.g., an environment variable set to Hello world cannot initialize a register of type real64[3]).

• Also see make_transport().

Note

Consider extending this factory with a capability to automatically run the node-ID allocation client pycyphal.application.plug_and_play.Allocatee if uavcan.node.id is not set.

Until this is implemented, to run the allocator one needs to construct the transport manually using make_transport() and make_registry(), then run the allocation client, then invoke this factory again with the above-obtained Registry instance, having done registry["uavcan.node.id"] = allocated_node_id beforehand.

While tedious, this is not that much of a problem because the PnP protocol is mostly intended for hardware nodes rather than software ones. A typical software node would normally receive its node-ID at startup (see also Yakut Orchestrator).

pycyphal.application.make_transport(registers: MutableMapping[str, ValueProxy], *, reconfigurable: bool = False) → Transport | None

Constructs a transport instance based on the configuration encoded in the supplied registers. If more than one transport is defined, a redundant instance will be constructed.

The register schema is documented below per transport class (refer to the transport class documentation to find the defaults for optional registers). All transports also accept the following standard registers:

Register name	Register type	Semantics
uavcan.node.id	natural16[1]	The node-ID to use. If the value exceeds the valid range, the constructed node will be anonymous.

pycyphal.transport.udp

Register name	Register type	Register semantics
uavcan.udp.iface	string	Whitespace-separated list of /16 IP subnet addresses. 16 least significant bits are replaced with the node-ID if configured, otherwise left unchanged. E.g.: 127.42.0.42: node-ID 257, result 127.42.1.1; 127.42.0.42: anonymous, result 127.42.0.42.
uavcan.udp.duplicate_service_transfers	bit[1]	Apply forward error correction to RPC-service transfers by setting multiplication factor = 2.
uavcan.udp.mtu	natural16[1]	The MTU for all constructed transport instances.

pycyphal.transport.serial

Register name	Register type	Register semantics
uavcan.serial.iface	string	Whitespace-separated list of serial port names. E.g.: /dev/ttyACM0, COM9, socket://127.0.0.1:50905.
uavcan.serial.duplicate_service_transfers	bit[1]	Apply forward error correction to RPC-service transfers by setting multiplication factor = 2.
uavcan.serial.baudrate	natural32[1]	The baudrate to set for all specified serial ports. Leave unchanged if zero.

pycyphal.transport.can

Register name	Register type	Register semantics
uavcan.can.iface	string	Whitespace-separated list of CAN iface names. Each iface name shall follow the format defined in pycyphal.transport.can.media.pythoncan. E.g.: socketcan:vcan0. On GNU/Linux, the socketcan: prefix selects pycyphal.transport.can.media.socketcan instead of PythonCAN. All platforms support the candump: prefix, which selects pycyphal.transport.can.media.candump; the text after colon is the path of the log file; e.g., candump:/home/pavel/candump-2022-07-14_150815.log.
uavcan.can.mtu	natural16[1]	The MTU value to use with all constructed CAN transports. Values other than 8 and 64 should not be used.
uavcan.can.bitrate	natural32[2]	The bitrates to use for all constructed CAN transports for arbitration (first value) and data (second value) segments. To use Classic CAN, set both to the same value and set MTU = 8.

pycyphal.transport.loopback

Register name	Register type	Register semantics
uavcan.loopback	bit[1]	If True, a loopback transport will be constructed. This is intended for testing only.

Parameters:

• registers – A mutable mapping of str to pycyphal.application.register.ValueProxy. Normally, it should be constructed by pycyphal.application.make_registry().

• reconfigurable –

  If False (default), the return value is:

  • None if the registers do not encode a valid transport configuration.

  • A single transport instance if a non-redundant configuration is defined.

  • An instance of pycyphal.transport.RedundantTransport if more than one transport configuration is defined.

  If True, then the returned instance is always of type pycyphal.transport.RedundantTransport, where the set of inferiors is empty if no transport configuration is defined. This case is intended for applications that may want to change the transport configuration afterwards.

Returns:

None if no transport is configured AND reconfigurable is False. Otherwise, a functional transport instance is returned.

Raises:

• pycyphal.application.register.MissingRegisterError if a register is expected but cannot be found.

• pycyphal.application.register.ValueConversionError if a register is found but its value cannot be converted to the correct type.

>>> from pycyphal.application.register import ValueProxy, Natural16, Natural32
>>> reg = {
...     "uavcan.udp.iface": ValueProxy("127.0.0.1"),
...     "uavcan.node.id": ValueProxy(Natural16([257])),
... }
>>> tr = make_transport(reg)
>>> tr
UDPTransport('127.0.0.1', local_node_id=257, ...)
>>> tr.close()
>>> tr = make_transport(reg, reconfigurable=True)                   # Same but reconfigurable.
>>> tr                                                              # Wrapped into RedundantTransport.
RedundantTransport(UDPTransport('127.0.0.1', local_node_id=257, ...))
>>> tr.close()

>>> int(reg["uavcan.udp.mtu"])      # Defaults created automatically to expose all configurables.
1200
>>> int(reg["uavcan.can.mtu"])
64
>>> reg["uavcan.can.bitrate"].ints
[1000000, 4000000]

>>> reg = {                                             # Triply-redundant heterogeneous transport:
...     "uavcan.udp.iface":    ValueProxy("127.99.0.15 127.111.0.15"),  # Double UDP transport
...     "uavcan.serial.iface": ValueProxy("socket://127.0.0.1:50905"),  # Serial transport
... }
>>> tr = make_transport(reg)                            # The node-ID was not set, so the transport is anonymous.
>>> tr                                          
RedundantTransport(UDPTransport('127.99.0.15',  local_node_id=None, ...),
                   UDPTransport('127.111.0.15', local_node_id=None, ...),
                   SerialTransport('socket://127.0.0.1:50905', local_node_id=None, ...))
>>> tr.close()

>>> reg = {
...     "uavcan.can.iface":   ValueProxy("virtual: virtual:"),    # Doubly-redundant CAN
...     "uavcan.can.mtu":     ValueProxy(Natural16([32])),
...     "uavcan.can.bitrate": ValueProxy(Natural32([500_000, 2_000_000])),
...     "uavcan.node.id":     ValueProxy(Natural16([123])),
... }
>>> tr = make_transport(reg)
>>> tr                                          
RedundantTransport(CANTransport(PythonCANMedia('virtual:', mtu=32), local_node_id=123),
                   CANTransport(PythonCANMedia('virtual:', mtu=32), local_node_id=123))
>>> tr.close()

>>> reg = {
...     "uavcan.udp.iface": ValueProxy("127.99.1.1"),       # Per the standard register specs,
...     "uavcan.node.id": ValueProxy(Natural16([0xFFFF])),  # 0xFFFF means unset/anonymous.
... }
>>> tr = make_transport(reg)
>>> tr
UDPTransport('127.99.1.1', local_node_id=None, ...)
>>> tr.close()

>>> tr = make_transport({})
>>> tr is None
True
>>> tr = make_transport({}, reconfigurable=True)
>>> tr                                                          # Redundant transport with no inferiors.
RedundantTransport()

pycyphal.application.make_registry(register_file: None | str | Path = None, environment_variables: Dict[str, bytes] | Dict[str, str] | Dict[bytes, bytes] | None = None) → Registry

Construct a new instance of pycyphal.application.register.Registry. Complex applications with uncommon requirements may choose to implement Registry manually instead of using this factory.

See also: standard RPC-service uavcan.register.Access.

Parameters:

• register_file – Path to the registry file; or, in other words, the configuration file of this application/node. If not provided (default), the registers of this instance will be stored in-memory (volatile configuration). If path is provided but the file does not exist, it will be created automatically. See Node.registry.

• environment_variables –

  During initialization, all registers will be updated based on the environment variables passed here. This dict is used to initialize pycyphal.application.register.Registry.environment_variables. Registers that are created later using pycyphal.application.register.Registry.setdefault() will use these values as well.

  If None (which is default), the value is initialized by copying os.environb. Pass an empty dict here to disable environment variable processing.

Raises:

• pycyphal.application.register.ValueConversionError if a register is found but its value cannot be converted to the correct type, or if the value of an environment variable for a register is invalid or incompatible with the register’s type (e.g., an environment variable set to Hello world cannot be assigned to register of type real64[3]).

exception pycyphal.application.NetworkTimeoutError

Bases: TimeoutError

API calls below the application layer return None on timeout. Some of the application-layer API calls raise this exception instead.