Demo

This section demonstrates how to build Cyphal applications using PyCyphal. It has been tested against GNU/Linux and Windows; it is also expected to work with any other major OS. The document is arranged as follows:

  • In the first section we introduce a couple of custom data types to illustrate how they can be dealt with.

  • The second section shows a simple demo node that implements a temperature controller and provides a custom RPC-service.

  • The third section provides a hands-on illustration of the data distribution functionality of Cyphal with the help of Yakut — a command-line utility for diagnostics and debugging of Cyphal networks.

  • The fourth section adds a second node that simulates the plant whose temperature is controlled by the first one.

  • The last section explains how to perform orchestration and configuration management of Cyphal networks.

You are expected to be familiar with terms like Cyphal node, DSDL, subject-ID, RPC-service. If not, skim through the Cyphal Guide first.

If you want to follow along, install PyCyphal and switch to a new directory before continuing.

DSDL definitions

Every Cyphal application depends on the standard DSDL definitions located in the namespace uavcan. The standard namespace is part of the regulated namespaces maintained by the OpenCyphal project. Grab your copy from git:

git clone https://github.com/OpenCyphal/public_regulated_data_types

The demo relies on two vendor-specific data types located in the root namespace sirius_cyber_corp. The root namespace directory layout is as follows:

sirius_cyber_corp/                              # root namespace directory
    PerformLinearLeastSquaresFit.1.0.dsdl       # service type definition
    PointXY.1.0.dsdl                            # nested message type definition

Type sirius_cyber_corp.PerformLinearLeastSquaresFit.1.0, file sirius_cyber_corp/PerformLinearLeastSquaresFit.1.0.dsdl:

 1# This service accepts a list of 2D point coordinates and returns the best-fit linear function coefficients.
 2# If no solution exists, the returned coefficients are NaN.
 3
 4PointXY.1.0[<64] points
 5
 6@extent 1024 * 8
 7
 8---
 9
10float64 slope
11float64 y_intercept
12
13@extent 64 * 8

Type sirius_cyber_corp.PointXY.1.0, file sirius_cyber_corp/PointXY.1.0.dsdl:

1float16 x
2float16 y
3@sealed

First node

Copy-paste the source code given below into a file named demo_app.py. For the sake of clarity, move the custom DSDL root namespace directory sirius_cyber_corp/ that we created above into custom_data_types/. You should end up with the following directory structure:

custom_data_types/
    sirius_cyber_corp/                          # Created in the previous section
        PerformLinearLeastSquaresFit.1.0.dsdl
        PointXY.1.0.dsdl
public_regulated_data_types/                    # Clone from git
    uavcan/                                     # The standard DSDL namespace
        ...
    ...
demo_app.py                                     # The thermostat node script

Here comes demo_app.py:

  1#!/usr/bin/env python3
  2# Distributed under CC0 1.0 Universal (CC0 1.0) Public Domain Dedication.
  3# pylint: disable=ungrouped-imports,wrong-import-position
  4
  5import os
  6import sys
  7import pathlib
  8import asyncio
  9import logging
 10import importlib
 11import pycyphal
 12
 13# Production applications are recommended to compile their DSDL namespaces as part of the build process. The enclosed
 14# file "setup.py" provides an example of how to do that. The output path we specify here shall match that of "setup.py".
 15# Here we compile DSDL just-in-time to demonstrate an alternative.
 16compiled_dsdl_dir = pathlib.Path(__file__).resolve().parent / ".demo_dsdl_compiled"
 17
 18# Make the compilation outputs importable. Let your IDE index this directory as sources to enable code completion.
 19sys.path.insert(0, str(compiled_dsdl_dir))
 20
 21try:
 22    import sirius_cyber_corp  # This is our vendor-specific root namespace. Custom data types.
 23    import pycyphal.application  # This module requires the root namespace "uavcan" to be transcompiled.
 24except (ImportError, AttributeError):  # Redistributable applications typically don't need this section.
 25    logging.warning("Transcompiling DSDL, this may take a while")
 26    src_dir = pathlib.Path(__file__).resolve().parent
 27    pycyphal.dsdl.compile_all(
 28        [
 29            src_dir / "custom_data_types/sirius_cyber_corp",
 30            src_dir / "public_regulated_data_types/uavcan/",
 31        ],
 32        output_directory=compiled_dsdl_dir,
 33    )
 34    importlib.invalidate_caches()  # Python runtime requires this.
 35    import sirius_cyber_corp
 36    import pycyphal.application
 37
 38# Import other namespaces we're planning to use. Nested namespaces are not auto-imported, so in order to reach,
 39# say, "uavcan.node.Heartbeat", you have to "import uavcan.node".
 40import uavcan.node  # noqa
 41import uavcan.si.sample.temperature  # noqa
 42import uavcan.si.unit.temperature  # noqa
 43import uavcan.si.unit.voltage  # noqa
 44
 45
 46class DemoApp:
 47    REGISTER_FILE = "demo_app.db"
 48    """
 49    The register file stores configuration parameters of the local application/node. The registers can be modified
 50    at launch via environment variables and at runtime via RPC-service "uavcan.register.Access".
 51    The file will be created automatically if it doesn't exist.
 52    """
 53
 54    def __init__(self) -> None:
 55        node_info = uavcan.node.GetInfo_1.Response(
 56            software_version=uavcan.node.Version_1(major=1, minor=0),
 57            name="org.opencyphal.pycyphal.demo.demo_app",
 58        )
 59        # The Node class is basically the central part of the library -- it is the bridge between the application and
 60        # the UAVCAN network. Also, it implements certain standard application-layer functions, such as publishing
 61        # heartbeats and port introspection messages, responding to GetInfo, serving the register API, etc.
 62        # The register file stores the configuration parameters of our node (you can inspect it using SQLite Browser).
 63        self._node = pycyphal.application.make_node(node_info, DemoApp.REGISTER_FILE)
 64
 65        # Published heartbeat fields can be configured as follows.
 66        self._node.heartbeat_publisher.mode = uavcan.node.Mode_1.OPERATIONAL  # type: ignore
 67        self._node.heartbeat_publisher.vendor_specific_status_code = os.getpid() % 100
 68
 69        # Now we can create ports to interact with the network.
 70        # They can also be created or destroyed later at any point after initialization.
 71        # A port is created by specifying its data type and its name (similar to topic names in ROS or DDS).
 72        # The subject-ID is obtained from the standard register named "uavcan.sub.temperature_setpoint.id".
 73        # It can also be modified via environment variable "UAVCAN__SUB__TEMPERATURE_SETPOINT__ID".
 74        self._sub_t_sp = self._node.make_subscriber(uavcan.si.unit.temperature.Scalar_1, "temperature_setpoint")
 75
 76        # As you may probably guess by looking at the port names, we are building a basic thermostat here.
 77        # We subscribe to the temperature setpoint, temperature measurement (process variable), and publish voltage.
 78        # The corresponding registers are "uavcan.sub.temperature_measurement.id" and "uavcan.pub.heater_voltage.id".
 79        self._sub_t_pv = self._node.make_subscriber(uavcan.si.sample.temperature.Scalar_1, "temperature_measurement")
 80        self._pub_v_cmd = self._node.make_publisher(uavcan.si.unit.voltage.Scalar_1, "heater_voltage")
 81
 82        # Create an RPC-server. The service-ID is read from standard register "uavcan.srv.least_squares.id".
 83        # This service is optional: if the service-ID is not specified, we simply don't provide it.
 84        try:
 85            srv_least_sq = self._node.get_server(sirius_cyber_corp.PerformLinearLeastSquaresFit_1, "least_squares")
 86            srv_least_sq.serve_in_background(self._serve_linear_least_squares)
 87        except pycyphal.application.register.MissingRegisterError:
 88            logging.info("The least squares service is disabled by configuration")
 89
 90        # Create another RPC-server using a standard service type for which a fixed service-ID is defined.
 91        # We don't specify the port name so the service-ID defaults to the fixed port-ID.
 92        # We could, of course, use it with a different service-ID as well, if needed.
 93        self._node.get_server(uavcan.node.ExecuteCommand_1).serve_in_background(self._serve_execute_command)
 94
 95        self._node.start()  # Don't forget to start the node!
 96
 97    @staticmethod
 98    async def _serve_linear_least_squares(
 99        request: sirius_cyber_corp.PerformLinearLeastSquaresFit_1.Request,
100        metadata: pycyphal.presentation.ServiceRequestMetadata,
101    ) -> sirius_cyber_corp.PerformLinearLeastSquaresFit_1.Response:
102        logging.info("Least squares request %s from node %d", request, metadata.client_node_id)
103        sum_x = sum(map(lambda p: p.x, request.points))  # type: ignore
104        sum_y = sum(map(lambda p: p.y, request.points))  # type: ignore
105        a = sum_x * sum_y - len(request.points) * sum(map(lambda p: p.x * p.y, request.points))  # type: ignore
106        b = sum_x * sum_x - len(request.points) * sum(map(lambda p: p.x**2, request.points))  # type: ignore
107        try:
108            slope = a / b
109            y_intercept = (sum_y - slope * sum_x) / len(request.points)
110        except ZeroDivisionError:
111            slope = float("nan")
112            y_intercept = float("nan")
113        return sirius_cyber_corp.PerformLinearLeastSquaresFit_1.Response(slope=slope, y_intercept=y_intercept)
114
115    @staticmethod
116    async def _serve_execute_command(
117        request: uavcan.node.ExecuteCommand_1.Request,
118        metadata: pycyphal.presentation.ServiceRequestMetadata,
119    ) -> uavcan.node.ExecuteCommand_1.Response:
120        logging.info("Execute command request %s from node %d", request, metadata.client_node_id)
121        if request.command == uavcan.node.ExecuteCommand_1.Request.COMMAND_FACTORY_RESET:
122            try:
123                os.unlink(DemoApp.REGISTER_FILE)  # Reset to defaults by removing the register file.
124            except OSError:  # Do nothing if already removed.
125                pass
126            return uavcan.node.ExecuteCommand_1.Response(uavcan.node.ExecuteCommand_1.Response.STATUS_SUCCESS)
127        return uavcan.node.ExecuteCommand_1.Response(uavcan.node.ExecuteCommand_1.Response.STATUS_BAD_COMMAND)
128
129    async def run(self) -> None:
130        """
131        The main method that runs the business logic. It is also possible to use the library in an IoC-style
132        by using receive_in_background() for all subscriptions if desired.
133        """
134        temperature_setpoint = 0.0
135        temperature_error = 0.0
136
137        def on_setpoint(msg: uavcan.si.unit.temperature.Scalar_1, _: pycyphal.transport.TransferFrom) -> None:
138            nonlocal temperature_setpoint
139            temperature_setpoint = msg.kelvin
140
141        self._sub_t_sp.receive_in_background(on_setpoint)  # IoC-style handler.
142
143        # Expose internal states to external observers for diagnostic purposes. Here, we define read-only registers.
144        # Since they are computed at every invocation, they are never stored in the register file.
145        self._node.registry["thermostat.error"] = lambda: temperature_error
146        self._node.registry["thermostat.setpoint"] = lambda: temperature_setpoint
147
148        # Read application settings from the registry. The defaults will be used only if a new register file is created.
149        gain_p, gain_i, gain_d = self._node.registry.setdefault("thermostat.pid.gains", [0.12, 0.18, 0.01]).floats
150
151        logging.info("Application started with PID gains: %.3f %.3f %.3f", gain_p, gain_i, gain_d)
152        print("Running. Press Ctrl+C to stop.", file=sys.stderr)
153
154        # This loop will exit automatically when the node is close()d. It is also possible to use receive() instead.
155        async for m, _metadata in self._sub_t_pv:
156            assert isinstance(m, uavcan.si.sample.temperature.Scalar_1)
157            temperature_error = temperature_setpoint - m.kelvin
158            voltage_output = temperature_error * gain_p  # Suppose this is a basic P-controller.
159            await self._pub_v_cmd.publish(uavcan.si.unit.voltage.Scalar_1(voltage_output))
160
161    def close(self) -> None:
162        """
163        This will close all the underlying resources down to the transport interface and all publishers/servers/etc.
164        All pending tasks such as serve_in_background()/receive_in_background() will notice this and exit automatically.
165        """
166        self._node.close()
167
168
169async def main() -> None:
170    logging.root.setLevel(logging.INFO)
171    app = DemoApp()
172    try:
173        await app.run()
174    except KeyboardInterrupt:
175        pass
176    finally:
177        app.close()
178
179
180if __name__ == "__main__":
181    asyncio.run(main())

If you just run the script as-is, you will notice that it fails with an error referring to some missing registers.

As explained in the comments (and — in great detail — in the Cyphal Specification), registers are basically named values that keep various configuration parameters of the local Cyphal node (application). Some of these parameters are used by the business logic of the application (e.g., PID gains); others are used by the Cyphal stack (e.g., port-IDs, node-ID, transport configuration, logging, and so on). Registers of the latter category are all named with the same prefix uavcan., and their names and semantics are regulated by the Specification to ensure consistency across the ecosystem.

So the application fails with an error that says that it doesn’t know how to reach the Cyphal network it is supposed to be part of because there are no registers to read that information from. We can resolve this by passing the correct register values via environment variables:

export UAVCAN__NODE__ID=42                           # Set the local node-ID 42 (anonymous by default)
export UAVCAN__UDP__IFACE=127.9.0.0                  # Use Cyphal/UDP transport via 127.9.0.42 (sic!)
export UAVCAN__SUB__TEMPERATURE_SETPOINT__ID=2345    # Subject "temperature_setpoint"    on ID 2345
export UAVCAN__SUB__TEMPERATURE_MEASUREMENT__ID=2346 # Subject "temperature_measurement" on ID 2346
export UAVCAN__PUB__HEATER_VOLTAGE__ID=2347          # Subject "heater_voltage"          on ID 2347
export UAVCAN__SRV__LEAST_SQUARES__ID=123            # Service "least_squares"           on ID 123
export UAVCAN__DIAGNOSTIC__SEVERITY=2                # This is optional to enable logging via Cyphal

python demo_app.py                                   # Run the application!

The snippet is valid for sh/bash/zsh; if you are using PowerShell on Windows, replace export with $env: and take values into double quotes. Further snippets will not include this remark.

Tip

macOS loopback addresses

macOS does not properly adhere to RFC3330 such that you will need to manually create aliases for each iface + node-ID used. In our example, with iface being 127.9.0.0 and node-ID being 42 this would be ifconfig lo0 alias 127.9.0.42 up (see this superuser article for more details).

An environment variable UAVCAN__SUB__TEMPERATURE_SETPOINT__ID sets register uavcan.sub.temperature_setpoint.id, and so on.

Tip

Specifying the environment variables manually is inconvenient. A better option is to store the configuration you use often into a shell file, and then source that when necessary into your active shell session like source my_env.sh (this is similar to Python virtualenv). See Yakut user manual for practical examples.

In PyCyphal, registers are normally stored in the register file, in our case it’s my_registers.db (the Cyphal Specification does not regulate how the registers are to be stored, this is an implementation detail). Once you started the application with a specific configuration, it will store the values in the register file, so the next time you can run it without passing any environment variables at all.

The registers of any Cyphal node are exposed to other network participants via the standard RPC-services defined in the standard DSDL namespace uavcan.register. This means that other nodes on the network can reconfigure our demo application via Cyphal directly, without the need to resort to any secondary management interfaces. This is equally true for software nodes like our demo application and deeply embedded hardware nodes.

When you execute the commands above, you should see the script running. Leave it running and move on to the next section.

Tip

Just-in-time vs. ahead-of-time DSDL compilation

The script will transpile the required DSDL namespaces just-in-time at launch. While this approach works for some applications, those that are built for redistribution at large (e.g., via PyPI) may benefit from compiling DSDL ahead-of-time (at build time) and including the compilation outputs into the redistributable package. Ahead-of-time DSDL compilation can be trivially implemented in setup.py:

 1#!/usr/bin/env python
 2# Distributed under CC0 1.0 Universal (CC0 1.0) Public Domain Dedication.
 3# type: ignore
 4"""
 5A simplified setup.py demo that shows how to distribute compiled DSDL definitions with Python packages.
 6"""
 7
 8import setuptools
 9import logging
10import distutils.command.build_py
11from pathlib import Path
12
13NAME = "demo_app"
14
15
16# noinspection PyUnresolvedReferences
17class BuildPy(distutils.command.build_py.build_py):
18    def run(self):
19        import pycyphal
20
21        pycyphal.dsdl.compile_all(
22            [
23                "public_regulated_data_types/uavcan",  # All Cyphal applications need the standard namespace, always.
24                "custom_data_types/sirius_cyber_corp",
25                # "public_regulated_data_types/reg",  # Many applications also need the non-standard regulated DSDL.
26            ],
27            output_directory=Path(self.build_lib, NAME, ".demo_dsdl_compiled"),  # Store in the build output archive.
28        )
29        super().run()
30
31
32logging.basicConfig(level=logging.INFO, format="%(levelname)-3.3s %(name)s: %(message)s")
33
34setuptools.setup(
35    name=NAME,
36    py_modules=["demo_app"],
37    cmdclass={"build_py": BuildPy},
38)

Poking the node using Yakut

The demo is running now so we can interact with it and see how it responds. We could write another script for that using PyCyphal, but in this section we will instead use Yakut — a simple CLI tool for diagnostics and management of Cyphal networks. You will need to open a couple of new terminal sessions now.

If you don’t have Yakut installed on your system yet, install it now by following its documentation.

Yakut requires us to compile our DSDL namespaces beforehand using yakut compile:

yakut compile  custom_data_types/sirius_cyber_corp  public_regulated_data_types/uavcan

The outputs will be stored in the current working directory. If you decided to change the working directory or move the compilation outputs, make sure to export the YAKUT_PATH environment variable pointing to the correct location.

The commands shown later need to operate on the same network as the demo. Earlier we configured the demo to use Cyphal/UDP via 127.9.0.42. So, for Yakut, we can export this configuration to let it run on the same network anonymously:

export UAVCAN__UDP__IFACE=127.9.0.0  # We don't export the node-ID, so it will remain anonymous.

To listen to the demo’s heartbeat and diagnostics, launch the following in a new terminal and leave it running (y is a convenience shortcut for yakut):

export UAVCAN__UDP__IFACE=127.9.0.0
y sub --with-metadata uavcan.node.heartbeat uavcan.diagnostic.record    # You should see heartbeats

Now let’s see how the simple thermostat node is operating. Launch another subscriber to see the published voltage command (it is not going to print anything yet):

export UAVCAN__UDP__IFACE=127.9.0.0
y sub 2347:uavcan.si.unit.voltage.scalar --redraw       # Prints nothing.

And publish the setpoint along with the measurement (process variable):

export UAVCAN__UDP__IFACE=127.9.0.0
export UAVCAN__NODE__ID=111         # We need a node-ID to publish messages
y pub --count=10 2345:uavcan.si.unit.temperature.scalar   250 \
                 2346:uavcan.si.sample.temperature.scalar 'kelvin: 240'

You should see the voltage subscriber that we just started print something along these lines:

---
2347: {volt: 1.1999999284744263}
# And so on...

Okay, the thermostat is working. If you change the setpoint (via subject-ID 2345) or measurement (via subject-ID 2346), you will see the published command messages (subject-ID 2347) update accordingly.

One important feature of the register interface is that it allows one to monitor internal states of the application, which is critical for debugging. In some way it is similar to performance counters or tracing probes:

y r 42 thermostat.error     # Read register

We will see the current value of the temperature error registered by the thermostat. If you run the last command with -dd (d for detailed), you will see the register metadata:

real64:
  value: [10.0]
_meta_: {mutable: false, persistent: false}

mutable: false says that this register cannot be modified and persistent: false says that it is not committed to any persistent storage (like a register file). Together they mean that the value is computed at runtime dynamically.

We can use the very same interface to query or modify the configuration parameters. For example, we can change the PID gains of the thermostat:

y r thermostat.pid.gains 2 0 0

Which returns [2.0, 0.0, 0.0], meaning that the new value was assigned successfully. Observe that the register server does implicit type conversion to the type specified by the application (our script). The Cyphal Specification does not require this behavior, though, so some simpler nodes (embedded systems in particular) may just reject mis-typed requests.

If you restart the application now, you will see it use the updated PID gains.

Now let’s try the linear regression service:

# The following commands do the same thing but differ in verbosity/explicitness:
y call 42 123:sirius_cyber_corp.PerformLinearLeastSquaresFit 'points: [{x: 10, y: 3}, {x: 20, y: 4}]'
y q 42 least_squares '[[10, 3], [20, 4]]'

The response should look like:

123: {slope: 0.1, y_intercept: 2.0}

And the diagnostic subscriber we started in the beginning (type uavcan.diagnostic.Record) should print a message.

Second node

To make this tutorial more hands-on, we are going to add another node and make it interoperate with the first one. As the first node implements a basic thermostat, the second one simulates the plant whose temperature is controlled by the thermostat. Put the following into plant.py in the same directory:

 1#!/usr/bin/env python3
 2# Distributed under CC0 1.0 Universal (CC0 1.0) Public Domain Dedication.
 3"""
 4This application simulates the plant controlled by the thermostat node: it takes a voltage command,
 5runs a crude thermodynamics simulation, and publishes the temperature (i.e., one subscription, one publication).
 6"""
 7
 8import time
 9import asyncio
10import uavcan.si.unit.voltage
11import uavcan.si.sample.temperature
12import uavcan.time
13import pycyphal
14from pycyphal.application.heartbeat_publisher import Health
15from pycyphal.application import make_node, NodeInfo, register
16
17
18UPDATE_PERIOD = 0.5
19
20heater_voltage = 0.0
21saturation = False
22
23
24def handle_command(msg: uavcan.si.unit.voltage.Scalar_1, _metadata: pycyphal.transport.TransferFrom) -> None:
25    global heater_voltage, saturation
26    if msg.volt < 0.0:
27        heater_voltage = 0.0
28        saturation = True
29    elif msg.volt > 50.0:
30        heater_voltage = 50.0
31        saturation = True
32    else:
33        heater_voltage = msg.volt
34        saturation = False
35
36
37async def main() -> None:
38    with make_node(NodeInfo(name="org.opencyphal.pycyphal.demo.plant"), "plant.db") as node:
39        # Expose internal states for diagnostics.
40        node.registry["status.saturation"] = lambda: saturation  # The register type will be deduced as "bit[1]".
41
42        # Initialize values from the registry. The temperature is in kelvin because in UAVCAN everything follows SI.
43        # Here, we specify the type explicitly as "real32[1]". If we pass a native float, it would be "real64[1]".
44        temp_environment = float(node.registry.setdefault("model.environment.temperature", register.Real32([292.15])))
45        temp_plant = temp_environment
46
47        # Set up the ports.
48        pub_meas = node.make_publisher(uavcan.si.sample.temperature.Scalar_1, "temperature")
49        pub_meas.priority = pycyphal.transport.Priority.HIGH
50        sub_volt = node.make_subscriber(uavcan.si.unit.voltage.Scalar_1, "voltage")
51        sub_volt.receive_in_background(handle_command)
52
53        # Run the main loop forever.
54        next_update_at = asyncio.get_running_loop().time()
55        while True:
56            # Publish new measurement and update node health.
57            await pub_meas.publish(
58                uavcan.si.sample.temperature.Scalar_1(
59                    timestamp=uavcan.time.SynchronizedTimestamp_1(microsecond=int(time.time() * 1e6)),
60                    kelvin=temp_plant,
61                )
62            )
63            node.heartbeat_publisher.health = Health.ADVISORY if saturation else Health.NOMINAL
64
65            # Sleep until the next iteration.
66            next_update_at += UPDATE_PERIOD
67            await asyncio.sleep(next_update_at - asyncio.get_running_loop().time())
68
69            # Update the simulation.
70            temp_plant += heater_voltage * 0.1 * UPDATE_PERIOD  # Energy input from the heater.
71            temp_plant -= (temp_plant - temp_environment) * 0.05 * UPDATE_PERIOD  # Dissipation.
72
73
74if __name__ == "__main__":
75    try:
76        asyncio.run(main())
77    except KeyboardInterrupt:
78        pass

You may launch it if you want, but you will notice that tinkering with registers by way of manual configuration gets old fast. The next section introduces a better way.

Orchestration

Attention

Yakut Orchestrator is in the alpha stage. Breaking changes may be introduced between minor versions until Yakut v1.0 is released. Freeze the minor version number to avoid unexpected changes.

Yakut Orchestrator does not support Windows at the moment.

Manual management of environment variables and node processes may work in simple setups, but it doesn’t really scale. Practical cyber-physical systems require a better way of managing Cyphal networks that may simultaneously include software nodes executed on the local or remote computers along with specialized bare-metal nodes running on dedicated hardware.

One solution to this is Yakut Orchestrator — an interpreter of a simple YAML-based domain-specific language that allows one to define process groups and conveniently manage them as a single entity. The language comes with a user-friendly syntax for managing Cyphal registers. Those familiar with ROS may find it somewhat similar to roslaunch.

The following orchestration file (orc-file) launch.orc.yaml does this:

  • Compiles two DSDL namespaces: the standard uavcan and the custom sirius_cyber_corp. If they are already compiled, this step is skipped.

  • When compilation is done, the two applications are launched. Be sure to stop the first script if it is still running!

  • Aside from the applications, a couple of diagnostic processes are started as well. A setpoint publisher will command the thermostat to drive the plant to the specified temperature.

The orchestrator runs everything concurrently, but join statements are used to enforce sequential execution as needed. The first process to fail (that is, exit with a non-zero code) will bring down the entire composition. Predicate scripts ?= are allowed to fail though — this is used to implement conditional execution.

The syntax allows the developer to define regular environment variables along with register names. The latter are translated into environment variables when starting a process.

 1#!/usr/bin/env -S yakut --verbose orchestrate
 2# Read the docs about the orc-file syntax: yakut orchestrate --help
 3
 4# Shared environment variables for all nodes/processes (can be overridden or selectively removed in local scopes).
 5YAKUT_COMPILE_OUTPUT: .yakut_compiled
 6YAKUT_PATH: .yakut_compiled
 7# Here we use Yakut for compiling DSDL. Normally one should use Nunavut though: https://github.com/OpenCyphal/nunavut
 8PYTHONPATH: .yakut_compiled
 9
10# Shared registers for all nodes/processes (can be overridden or selectively removed in local scopes).
11# See the docs for pycyphal.application.make_node() to see which registers can be used here.
12uavcan:
13  # Use Cyphal/UDP:
14  udp.iface: 127.9.0.0
15  # If you have Ncat or some other TCP broker, you can use Cyphal/serial tunneled over TCP (in a heterogeneous
16  # redundant configuration with UDP or standalone). Ncat launch example: ncat --broker --listen --source-port 50905
17  serial.iface: "" # socket://127.0.0.1:50905
18  # It is recommended to explicitly assign unused transports to ensure that previously stored transport
19  # configurations are not accidentally reused:
20  can.iface: ""
21  # Configure diagnostic publishing, too:
22  diagnostic:
23    severity: 2
24    timestamp: true
25
26# Keys with "=" define imperatives rather than registers or environment variables.
27$=:
28- ?=: '[ ! -d $YAKUT_COMPILE_OUTPUT ]'  # If the output directory does not exist, run the Yakut DSDL compiler.
29  $=:                                   # All script statements run concurrently.
30  - echo "Compiling DSDL, this may take a while"
31  - yakut compile custom_data_types/sirius_cyber_corp public_regulated_data_types/uavcan
32
33- # An empty statement is a join statement -- wait for the previously launched processes to exit before continuing.
34
35- $=:
36  # Wait a bit to let the diagnostic subscribers get ready (they are launched below).
37  - sleep 1
38  - # Remember, everything runs concurrently by default, but this join statement waits for the sleep to exit.
39
40  # Launch the demo app that implements the thermostat.
41  - $=: python demo_app.py
42    uavcan:
43      node.id: 42
44      sub.temperature_setpoint.id:    2345
45      sub.temperature_measurement.id: 2346
46      pub.heater_voltage.id:          2347
47      srv.least_squares.id:           0xFFFF    # We don't need this service. Disable by setting an invalid port-ID.
48    thermostat:
49      pid.gains: [0.1, 0, 0]
50
51  # Launch the controlled plant simulator.
52  - $=: python plant.py
53    uavcan:
54      node.id: 43
55      sub.voltage.id:     2347
56      pub.temperature.id: 2346
57    model.environment.temperature: 300.0    # In UAVCAN everything follows SI, so this temperature is in kelvin.
58
59  # Publish the setpoint a few times to show how the thermostat drives the plant to the correct temperature.
60  # You can publish a different setpoint by running this command in a separate terminal to see how the system responds:
61  #   yakut pub 2345 "kelvin: 200"
62  - $=: |
63      yakut pub 2345:uavcan.si.unit.temperature.scalar 450 -N3
64    uavcan.node.id: 100
65
66# Launch diagnostic subscribers to print messages in the terminal that runs the orchestrator.
67- yakut sub --with-metadata uavcan.diagnostic.record 2346:uavcan.si.sample.temperature.scalar
68
69# Exit automatically if STOP_AFTER is defined (frankly, this is just a testing aid, feel free to ignore).
70- ?=: test -n "$STOP_AFTER"
71  $=: sleep $STOP_AFTER && exit 111

Terminate the first node before continuing since it is now managed by the orchestration script we just wrote. Ensure that the node script files are named demo_app.py and plant.py, otherwise the orchestrator won’t find them.

The orc-file can be executed as yakut orc launch.orc.yaml, or simply ./launch.orc.yaml (use --verbose to see which environment variables are passed to each launched process). Having started it, you should see roughly the following output appear in the terminal, indicating that the thermostat is driving the plant towards the setpoint:

---
2346:
  _meta_: {ts_system: 1651773332.157150, ts_monotonic: 3368.421244, source_node_id: 43, transfer_id: 0, priority: high, dtype: uavcan.si.sample.temperature.Scalar.1.0}
  timestamp: {microsecond: 1651773332156343}
  kelvin: 300.0
---
8184:
  _meta_: {ts_system: 1651773332.162746, ts_monotonic: 3368.426840, source_node_id: 42, transfer_id: 0, priority: optional, dtype: uavcan.diagnostic.Record.1.1}
  timestamp: {microsecond: 1651773332159267}
  severity: {value: 2}
  text: 'root: Application started with PID gains: 0.100 0.000 0.000'
---
2346:
  _meta_: {ts_system: 1651773332.157150, ts_monotonic: 3368.421244, source_node_id: 43, transfer_id: 1, priority: high, dtype: uavcan.si.sample.temperature.Scalar.1.0}
  timestamp: {microsecond: 1651773332657040}
  kelvin: 300.0
---
2346:
  _meta_: {ts_system: 1651773332.657383, ts_monotonic: 3368.921476, source_node_id: 43, transfer_id: 2, priority: high, dtype: uavcan.si.sample.temperature.Scalar.1.0}
  timestamp: {microsecond: 1651773333157512}
  kelvin: 300.0
---
2346:
  _meta_: {ts_system: 1651773333.158257, ts_monotonic: 3369.422350, source_node_id: 43, transfer_id: 3, priority: high, dtype: uavcan.si.sample.temperature.Scalar.1.0}
  timestamp: {microsecond: 1651773333657428}
  kelvin: 300.73126220703125
---
2346:
  _meta_: {ts_system: 1651773333.657797, ts_monotonic: 3369.921891, source_node_id: 43, transfer_id: 4, priority: high, dtype: uavcan.si.sample.temperature.Scalar.1.0}
  timestamp: {microsecond: 1651773334157381}
  kelvin: 301.4406433105469
---
2346:
  _meta_: {ts_system: 1651773334.158120, ts_monotonic: 3370.422213, source_node_id: 43, transfer_id: 5, priority: high, dtype: uavcan.si.sample.temperature.Scalar.1.0}
  timestamp: {microsecond: 1651773334657390}
  kelvin: 302.1288757324219
# And so on. Notice how the temperature is rising slowly towards the setpoint at 450 K!

You can run yakut monitor to see what is happening on the network. As an exercise, consider this:

  • Run the same composition over CAN by changing the transport configuration registers at the top of the orc-file. The full set of transport-related registers is documented at pycyphal.application.make_transport().

  • Implement saturation management by publishing the saturation flag over a dedicated subject and subscribing to it from the thermostat node.

  • Use Wireshark (capture filter expression: (udp or igmp) and src net 127.9.0.0/16) or candump (like candump -decaxta any) to inspect the network exchange.