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 customsirius_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 (likecandump -decaxta any
) to inspect the network exchange.