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 (~/pycyphal-demo) 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:

# This service accepts a list of 2D point coordinates and returns the best-fit linear function coefficients.
# If no solution exists, the returned coefficients are NaN.

PointXY.1.0[<64] points

@extent 1024 * 8

---

float64 slope
float64 y_intercept

@extent 64 * 8

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

float16 x
float16 y
@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:

pycyphal-demo/
    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

The CYPHAL_PATH environment variable should contain the list of paths where the DSDL root namespace directories are to be found (be sure to modify the values to match your environment):

export CYPHAL_PATH="$HOME/pycyphal-demo/custom_data_types:$HOME/pycyphal-demo/public_regulated_data_types"

Here comes demo_app.py:

#!/usr/bin/env python3
# Distributed under CC0 1.0 Universal (CC0 1.0) Public Domain Dedication.
# pylint: disable=ungrouped-imports,wrong-import-position

import os
import sys
import asyncio
import logging
import pycyphal  # Importing PyCyphal will automatically install the import hook for DSDL compilation.

# DSDL files are automatically compiled by pycyphal import hook from sources pointed by CYPHAL_PATH env variable.
import sirius_cyber_corp  # This is our vendor-specific root namespace. Custom data types.
import pycyphal.application  # This module requires the root namespace "uavcan" to be transcompiled.

# Import other namespaces we're planning to use. Nested namespaces are not auto-imported, so in order to reach,
# say, "uavcan.node.Heartbeat", you have to "import uavcan.node".
import uavcan.node  # noqa
import uavcan.si.sample.temperature  # noqa
import uavcan.si.unit.temperature  # noqa
import uavcan.si.unit.voltage  # noqa


class DemoApp:
    REGISTER_FILE = "demo_app.db"
    """
    The register file stores configuration parameters of the local application/node. The registers can be modified
    at launch via environment variables and at runtime via RPC-service "uavcan.register.Access".
    The file will be created automatically if it doesn't exist.
    """

    def __init__(self) -> None:
        node_info = uavcan.node.GetInfo_1.Response(
            software_version=uavcan.node.Version_1(major=1, minor=0),
            name="org.opencyphal.pycyphal.demo.demo_app",
        )
        # The Node class is basically the central part of the library -- it is the bridge between the application and
        # the UAVCAN network. Also, it implements certain standard application-layer functions, such as publishing
        # heartbeats and port introspection messages, responding to GetInfo, serving the register API, etc.
        # The register file stores the configuration parameters of our node (you can inspect it using SQLite Browser).
        self._node = pycyphal.application.make_node(node_info, DemoApp.REGISTER_FILE)

        # Published heartbeat fields can be configured as follows.
        self._node.heartbeat_publisher.mode = uavcan.node.Mode_1.OPERATIONAL  # type: ignore
        self._node.heartbeat_publisher.vendor_specific_status_code = os.getpid() % 100

        # Now we can create ports to interact with the network.
        # They can also be created or destroyed later at any point after initialization.
        # A port is created by specifying its data type and its name (similar to topic names in ROS or DDS).
        # The subject-ID is obtained from the standard register named "uavcan.sub.temperature_setpoint.id".
        # It can also be modified via environment variable "UAVCAN__SUB__TEMPERATURE_SETPOINT__ID".
        self._sub_t_sp = self._node.make_subscriber(uavcan.si.unit.temperature.Scalar_1, "temperature_setpoint")

        # As you may probably guess by looking at the port names, we are building a basic thermostat here.
        # We subscribe to the temperature setpoint, temperature measurement (process variable), and publish voltage.
        # The corresponding registers are "uavcan.sub.temperature_measurement.id" and "uavcan.pub.heater_voltage.id".
        self._sub_t_pv = self._node.make_subscriber(uavcan.si.sample.temperature.Scalar_1, "temperature_measurement")
        self._pub_v_cmd = self._node.make_publisher(uavcan.si.unit.voltage.Scalar_1, "heater_voltage")

        # Create an RPC-server. The service-ID is read from standard register "uavcan.srv.least_squares.id".
        # This service is optional: if the service-ID is not specified, we simply don't provide it.
        try:
            srv_least_sq = self._node.get_server(sirius_cyber_corp.PerformLinearLeastSquaresFit_1, "least_squares")
            srv_least_sq.serve_in_background(self._serve_linear_least_squares)
        except pycyphal.application.register.MissingRegisterError:
            logging.info("The least squares service is disabled by configuration")

        # Create another RPC-server using a standard service type for which a fixed service-ID is defined.
        # We don't specify the port name so the service-ID defaults to the fixed port-ID.
        # We could, of course, use it with a different service-ID as well, if needed.
        self._node.get_server(uavcan.node.ExecuteCommand_1).serve_in_background(self._serve_execute_command)

        self._node.start()  # Don't forget to start the node!

    @staticmethod
    async def _serve_linear_least_squares(
        request: sirius_cyber_corp.PerformLinearLeastSquaresFit_1.Request,
        metadata: pycyphal.presentation.ServiceRequestMetadata,
    ) -> sirius_cyber_corp.PerformLinearLeastSquaresFit_1.Response:
        logging.info("Least squares request %s from node %d", request, metadata.client_node_id)
        sum_x = sum(map(lambda p: p.x, request.points))  # type: ignore
        sum_y = sum(map(lambda p: p.y, request.points))  # type: ignore
        a = sum_x * sum_y - len(request.points) * sum(map(lambda p: p.x * p.y, request.points))  # type: ignore
        b = sum_x * sum_x - len(request.points) * sum(map(lambda p: p.x**2, request.points))  # type: ignore
        try:
            slope = a / b
            y_intercept = (sum_y - slope * sum_x) / len(request.points)
        except ZeroDivisionError:
            slope = float("nan")
            y_intercept = float("nan")
        return sirius_cyber_corp.PerformLinearLeastSquaresFit_1.Response(slope=slope, y_intercept=y_intercept)

    @staticmethod
    async def _serve_execute_command(
        request: uavcan.node.ExecuteCommand_1.Request,
        metadata: pycyphal.presentation.ServiceRequestMetadata,
    ) -> uavcan.node.ExecuteCommand_1.Response:
        logging.info("Execute command request %s from node %d", request, metadata.client_node_id)
        if request.command == uavcan.node.ExecuteCommand_1.Request.COMMAND_FACTORY_RESET:
            try:
                os.unlink(DemoApp.REGISTER_FILE)  # Reset to defaults by removing the register file.
            except OSError:  # Do nothing if already removed.
                pass
            return uavcan.node.ExecuteCommand_1.Response(uavcan.node.ExecuteCommand_1.Response.STATUS_SUCCESS)
        return uavcan.node.ExecuteCommand_1.Response(uavcan.node.ExecuteCommand_1.Response.STATUS_BAD_COMMAND)

    async def run(self) -> None:
        """
        The main method that runs the business logic. It is also possible to use the library in an IoC-style
        by using receive_in_background() for all subscriptions if desired.
        """
        temperature_setpoint = 0.0
        temperature_error = 0.0

        def on_setpoint(msg: uavcan.si.unit.temperature.Scalar_1, _: pycyphal.transport.TransferFrom) -> None:
            nonlocal temperature_setpoint
            temperature_setpoint = msg.kelvin

        self._sub_t_sp.receive_in_background(on_setpoint)  # IoC-style handler.

        # Expose internal states to external observers for diagnostic purposes. Here, we define read-only registers.
        # Since they are computed at every invocation, they are never stored in the register file.
        self._node.registry["thermostat.error"] = lambda: temperature_error
        self._node.registry["thermostat.setpoint"] = lambda: temperature_setpoint

        # Read application settings from the registry. The defaults will be used only if a new register file is created.
        gain_p, gain_i, gain_d = self._node.registry.setdefault("thermostat.pid.gains", [0.12, 0.18, 0.01]).floats

        logging.info("Application started with PID gains: %.3f %.3f %.3f", gain_p, gain_i, gain_d)
        print("Running. Press Ctrl+C to stop.", file=sys.stderr)

        # This loop will exit automatically when the node is close()d. It is also possible to use receive() instead.
        async for m, _metadata in self._sub_t_pv:
            assert isinstance(m, uavcan.si.sample.temperature.Scalar_1)
            temperature_error = temperature_setpoint - m.kelvin
            voltage_output = temperature_error * gain_p  # Suppose this is a basic P-controller.
            await self._pub_v_cmd.publish(uavcan.si.unit.voltage.Scalar_1(voltage_output))

    def close(self) -> None:
        """
        This will close all the underlying resources down to the transport interface and all publishers/servers/etc.
        All pending tasks such as serve_in_background()/receive_in_background() will notice this and exit automatically.
        """
        self._node.close()


async def main() -> None:
    logging.root.setLevel(logging.INFO)
    app = DemoApp()
    try:
        await app.run()
    except KeyboardInterrupt:
        pass
    finally:
        app.close()


if __name__ == "__main__":
    asyncio.run(main())

The following graph should give a rough visual overview of how the applications within the demo_app node are structured:

Legend

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.0.0.1                  # Use Cyphal/UDP transport via localhost
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.

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 demo_app.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:

#!/usr/bin/env python
# Distributed under CC0 1.0 Universal (CC0 1.0) Public Domain Dedication.
# type: ignore
"""
A simplified setup.py demo that shows how to distribute compiled DSDL definitions with Python packages.

To use precompiled DSDL files in app, the compilation output directory must be included in path:
    compiled_dsdl_dir = pathlib.Path(__file__).resolve().parent / ".demo_dsdl_compiled"
    sys.path.insert(0, str(compiled_dsdl_dir))
"""

import setuptools
import logging
import distutils.command.build_py
from pathlib import Path

NAME = "demo_app"


# noinspection PyUnresolvedReferences
class BuildPy(distutils.command.build_py.build_py):
    def run(self):
        import pycyphal

        pycyphal.dsdl.compile_all(
            [
                "public_regulated_data_types/uavcan",  # All Cyphal applications need the standard namespace, always.
                "custom_data_types/sirius_cyber_corp",
                # "public_regulated_data_types/reg",  # Many applications also need the non-standard regulated DSDL.
            ],
            output_directory=Path(self.build_lib, NAME, ".demo_dsdl_compiled"),  # Store in the build output archive.
        )
        super().run()


logging.basicConfig(level=logging.INFO, format="%(levelname)-3.3s %(name)s: %(message)s")

setuptools.setup(
    name=NAME,
    py_modules=["demo_app"],
    cmdclass={"build_py": BuildPy},
)

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 also needs to know where the DSDL files are located, this is specified via the same CYPHAL_PATH environment variable (this is a standard variable that many Cyphal tools rely on):

export CYPHAL_PATH="$HOME/pycyphal-demo/custom_data_types:$HOME/pycyphal-demo/public_regulated_data_types"

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

export UAVCAN__UDP__IFACE=127.0.0.1  # 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 CYPHAL_PATH="$HOME/pycyphal-demo/custom_data_types:$HOME/pycyphal-demo/public_regulated_data_types"
export UAVCAN__UDP__IFACE=127.0.0.1
y sub --with-metadata uavcan.node.heartbeat uavcan.diagnostic.record    # You should see heartbeats

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

export CYPHAL_PATH="$HOME/pycyphal-demo/custom_data_types:$HOME/pycyphal-demo/public_regulated_data_types"
export UAVCAN__UDP__IFACE=127.0.0.1
y sub 2347:uavcan.si.unit.voltage.scalar --redraw       # Prints nothing.

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

export CYPHAL_PATH="$HOME/pycyphal-demo/custom_data_types:$HOME/pycyphal-demo/public_regulated_data_types"
export UAVCAN__UDP__IFACE=127.0.0.1
export UAVCAN__NODE__ID=111         # We need a node-ID to publish messages properly
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 42 thermostat.pid.gains       # read current values
y r 42 thermostat.pid.gains 2 0 0 # write new values

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:

#!/usr/bin/env python3
# Distributed under CC0 1.0 Universal (CC0 1.0) Public Domain Dedication.
"""
This application simulates the plant controlled by the thermostat node: it takes a voltage command,
runs a crude thermodynamics simulation, and publishes the temperature (i.e., one subscription, one publication).
"""

import time
import asyncio
import pycyphal  # Importing PyCyphal will automatically install the import hook for DSDL compilation.

# Import DSDLs after pycyphal import hook is installed.
import uavcan.si.unit.voltage
import uavcan.si.sample.temperature
import uavcan.time
from pycyphal.application.heartbeat_publisher import Health
from pycyphal.application import make_node, NodeInfo, register


UPDATE_PERIOD = 0.5

heater_voltage = 0.0
saturation = False


def handle_command(msg: uavcan.si.unit.voltage.Scalar_1, _metadata: pycyphal.transport.TransferFrom) -> None:
    global heater_voltage, saturation
    if msg.volt < 0.0:
        heater_voltage = 0.0
        saturation = True
    elif msg.volt > 50.0:
        heater_voltage = 50.0
        saturation = True
    else:
        heater_voltage = msg.volt
        saturation = False


async def main() -> None:
    with make_node(NodeInfo(name="org.opencyphal.pycyphal.demo.plant"), "plant.db") as node:
        # Expose internal states for diagnostics.
        node.registry["status.saturation"] = lambda: saturation  # The register type will be deduced as "bit[1]".

        # Initialize values from the registry. The temperature is in kelvin because in UAVCAN everything follows SI.
        # Here, we specify the type explicitly as "real32[1]". If we pass a native float, it would be "real64[1]".
        temp_environment = float(node.registry.setdefault("model.environment.temperature", register.Real32([292.15])))
        temp_plant = temp_environment

        # Set up the ports.
        pub_meas = node.make_publisher(uavcan.si.sample.temperature.Scalar_1, "temperature")
        pub_meas.priority = pycyphal.transport.Priority.HIGH
        sub_volt = node.make_subscriber(uavcan.si.unit.voltage.Scalar_1, "voltage")
        sub_volt.receive_in_background(handle_command)

        # Run the main loop forever.
        next_update_at = asyncio.get_running_loop().time()
        while True:
            # Publish new measurement and update node health.
            await pub_meas.publish(
                uavcan.si.sample.temperature.Scalar_1(
                    timestamp=uavcan.time.SynchronizedTimestamp_1(microsecond=int(time.time() * 1e6)),
                    kelvin=temp_plant,
                )
            )
            node.heartbeat_publisher.health = Health.ADVISORY if saturation else Health.NOMINAL

            # Sleep until the next iteration.
            next_update_at += UPDATE_PERIOD
            await asyncio.sleep(next_update_at - asyncio.get_running_loop().time())

            # Update the simulation.
            temp_plant += heater_voltage * 0.1 * UPDATE_PERIOD  # Energy input from the heater.
            temp_plant -= (temp_plant - temp_environment) * 0.05 * UPDATE_PERIOD  # Dissipation.


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass

In graph form, the new node looks as follows:

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 launches the two applications (be sure to stop the first script if it is still running!) along with a couple of diagnostic processes that monitor the network. A setpoint publisher that will command the thermostat to drive the plant to the specified temperature is also started.

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.

#!/usr/bin/env -S yakut --verbose orchestrate
# Read the docs about the orc-file syntax: yakut orchestrate --help

# Shared environment variables for all nodes/processes (can be overridden or selectively removed in local scopes).
CYPHAL_PATH: "./public_regulated_data_types;./custom_data_types"
PYCYPHAL_PATH: ".pycyphal_generated"  # This one is optional; the default is "~/.pycyphal".

# Shared registers for all nodes/processes (can be overridden or selectively removed in local scopes).
# See the docs for pycyphal.application.make_node() to see which registers can be used here.
uavcan:
  # Use Cyphal/UDP via localhost:
  udp.iface: 127.0.0.1
  # If you have Ncat or some other TCP broker, you can use Cyphal/serial tunneled over TCP (in a heterogeneous
  # redundant configuration with UDP or standalone). Ncat launch example: ncat --broker --listen --source-port 50905
  serial.iface: "" # socket://127.0.0.1:50905
  # It is recommended to explicitly assign unused transports to ensure that previously stored transport
  # configurations are not accidentally reused:
  can.iface: ""
  # Configure diagnostic publishing, too:
  diagnostic:
    severity: 2
    timestamp: true

# Keys with "=" define imperatives rather than registers or environment variables.
$=:
- $=:
  # Wait a bit to let the diagnostic subscriber get ready (it is launched below).
  - sleep 5
  - # An empty statement is a join statement -- wait for the previously launched processes to exit before continuing.

  # Launch the demo app that implements the thermostat.
  - $=: python demo_app.py
    uavcan:
      node.id: 42
      sub.temperature_setpoint.id:    2345
      sub.temperature_measurement.id: 2346
      pub.heater_voltage.id:          2347
      srv.least_squares.id:           0xFFFF    # We don't need this service. Disable by setting an invalid port-ID.
    thermostat:
      pid.gains: [0.1, 0, 0]

  # Launch the controlled plant simulator.
  - $=: python plant.py
    uavcan:
      node.id: 43
      sub.voltage.id:     2347
      pub.temperature.id: 2346
    model.environment.temperature: 300.0    # In UAVCAN everything follows SI, so this temperature is in kelvin.

  # Publish the setpoint a few times to show how the thermostat drives the plant to the correct temperature.
  # You can publish a different setpoint by running this command in a separate terminal to see how the system responds:
  #   yakut pub 2345 "kelvin: 200"
  - $=: |
      yakut pub 2345:uavcan.si.unit.temperature.scalar 450 -N3
    uavcan.node.id: 100

# Launch diagnostic subscribers to print messages in the terminal that runs the orchestrator.
- yakut sub --with-metadata uavcan.diagnostic.record 2346:uavcan.si.sample.temperature.scalar

# Exit automatically if STOP_AFTER is defined (frankly, this is just a testing aid, feel free to ignore).
- ?=: test -n "$STOP_AFTER"
  $=: 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. (Don’t forget to set UAVCAN__UDP__IFACE or similar depending on your transport.)

Tip

macOS

Monitoring the network using yakut monitor, requires using root while preserving your environment variables:

sudo -E yakut monitor

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.