DEV Community: pikoTutorial

From AUTOSAR to S-Core: the first C++ pub/sub implementation

pikoTutorial — Wed, 14 Jan 2026 22:53:00 +0000

Looking at S-Core today feels a bit like walking into a German factory that has just announced a “major transformation initiative.” Before anything actually starts to change, there are:

discussions about processes
and documents about these discussions
and meetings about these documents
and calls to align on how these documents about those processes should be structured
and how such document structure will impact work in 20 years from now

The intention is good, but as my colleague from work pointed out some time ago: "bureaucracy expands to meet the needs of the expanding bureaucracy". It may sound funny, but this is how European automotive software has been built since years.

For a long time, this approach worked surprisingly well. German - and more broadly European - OEMs optimized for correctness, documentation and process, not for speed or innovation. That slowness was not a bug, it was really a feature and part of the mindset. The market was stable, competition was predictable and even strong players like Japanese or Korean manufacturers did not disrupt that golden setup . Nobody was releasing cars like software updates because nobody really had to. Building slowly, reviewing it thoroughly and certifying it carefully was acceptable when everyone played the same game.

However, when Chinese manufacturers entered the market at full speed, they started producing new models at a pace that European processes simply cannot match. Not because the correctness does no longer matter, but because speed and innovation suddenly do.

This is the tension S-Core seems to be trying to address. Unlike many open-source projects that start out fast and experimental and only later attempt to “add automotive”, S-Core approaches the topic how we like it in Europe.

We don't like when the things break, especially when these things are 2 tons SUVs running 140km/h on a highway and especially in an undocumented way, so it starts strict. Process, structure and ASIL-certifiability are part of the baseline, not something to care about in the future. After all, automotive software engineers know that writing code is only part of the job, the other part is to make that code safe, compliant and legally sellable in a heavily regulated market.

At the same time, S-Core wants to be open source, so that the code evolves in a dynamic way, powered by multiple contributors. If this combination works (and not just combines the disadvantages of both approaches), it could be a revolution for the entire automotive industry.

So let's give it a chance and try to implement a minimal S-Core publisher and subscriber using the latest release v0.5.

S-Core development environment setup

Before writing any code, let's setup the development environment. The prerequisites are:

install Docker: https://docs.docker.com/engine/install/ubuntu/
install Bazel: https://bazel.build/install/ubuntu

S-Core provides its development environment as a container in this GitHub repo. To use it, create .devcontainer/devcontainer.json file in the root of your project with the following content:

{
    "name": "eclipse-s-core",
    "image": "ghcr.io/eclipse-score/devcontainer:v1.1.0"
}

Now, open the project with VS Code and click "Reopen in Container" button on the pop-up window.

Read also on pikotutorial.com: How to write Arduino Uno code with Python?

Minimal S-Core publisher/subscriber implementation

With the environment in place, we can finally write some code. The goal here is intentionally simple: a minimal publisher and subscriber where publisher broadcasts a vehicle position as an event and subscriber reads that data. No optimizations, no abstractions - just the smallest example that shows how the pieces fit together in S-Core and how it is different (or similar) to Adaptive AUTOSAR.

Note: although the namespaces coming from S-Core are long, I don't use any using namespace for shortening them and keep all the S-Core types prefixed with their namespaces explicitly. This is on purpose, to make it clear in the example which S-Core type comes from which namespace.

Communication modelling

Before writing publisher or subscriber code, communication must be modeled. The S-Core's documentation says explicitly that S-Core relies only on the source code and does not utilize any domain specific modelling language for modelling of the communication. So why do I call this section "modelling"?

It's just because LoLa (the implementation of S-Core's IPC communication), although it uses pure C++, it requires a very specific way of interface definition, so if you come from Adaptive AUTOSAR, it will be easier for you to understand the purpose of a bit bizarre C++ code that I will write in the next step.

Let's then define our interface between publisher and subscriber. Create model/position_interface.h:

#ifndef MODEL_POSITION_INTERFACE_H
#define MODEL_POSITION_INTERFACE_H

#include "score/mw/com/types.h"

struct Point
{
  std::uint32_t x;
  std::uint32_t y;
};

template <typename Trait>
class PositionInterface : public Trait::Base
{
  public:
    using Trait::Base::Base;

    typename Trait::template Event<Point> center{*this, "center"};
};

using PositionInterfaceProxy = score::mw::com::AsProxy<PositionInterface>;
using PositionInterfaceSkeleton = score::mw::com::AsSkeleton<PositionInterface>;

#endif // MODEL_POSITION_INTERFACE_H

Let's now break it down:

Point represents the structure of the data being exchanged between the publisher and subscriber - in this case it's just a 2D point representing the coordinates.
PositionInterface represents the interface which connects publisher and subscriber - both of them will have access to its elements.
Event<Point> center defines an element of the interface. S-Core promises supports for events, fields and methods, but at the time of writing this article, methods are not supported yet. In our example, we define center as an Event with an underlying data structure being Point. Single interface may have multiple events.
PositionInterfaceProxy - type of the object via which the subscriber will access the center event to read the data.
PositionInterfaceSkeleton - type of the object via which the publisher will access the center event to write the data.

Publisher implementation

Create src/publisher/main.cpp in your project. First add the necessary S-Core headers:

#include "score/mw/com/impl/instance_specifier.h"
#include "score/mw/com/runtime.h"

And the header to our interface definition:

#include "model/position_interface.h"

Define constant values to use. Don't worry about the kPathToComConfig for now - I will explain it later.

static constexpr const char* kPathToComConfig {"config/communication_config.json"};
static constexpr const char* kInstanceSpecifier {"SomeInstanceSpecifier"};

Now let's implement the main function. First, set the path to the communication configuration defined above:

score::StringLiteral runtime_args[2U] = {"--service_instance_manifest", kPathToComConfig};
score::mw::com::runtime::InitializeRuntime(2, runtime_args);

Next create the service instance specifier, unique for each service:

const score::Result<score::mw::com::InstanceSpecifier> instance_specifier_result = score::mw::com::InstanceSpecifier::Create(std::string{kInstanceSpecifier});

if (!instance_specifier_result.has_value())
{
    std::cerr << "Error: failed to create instance specifier!" << std::endl;
    return 1;
}

const score::mw::com::InstanceSpecifier instance_specifier = instance_specifier_result.value();

Use the created instance_specifier to build the skeleton via which we will send the data from the publisher. Notice that PositionInterfaceSkeleton is the type alias declared in model/position_interface.h:

score::Result<PositionInterfaceSkeleton> skeleton_result = PositionInterfaceSkeleton::Create(instance_specifier);

if (!skeleton_result.has_value())
{
    std::cerr << "Failed to create skeleton for instance specifier "
              << instance_specifier.ToString() << std::endl;
    return 1;
}

auto& position_skeleton = skeleton_result.value();

Service offering must be done explicitly with OfferService():

const auto offer_result = position_skeleton.OfferService();

if (!offer_result.has_value())
{
    std::cerr << "Failed to offer service "
              << instance_specifier.ToString() << std::endl;
    return 1;
}

Now it's time for the actual work - create the data and send it to the world. I add a 1 second delay between each send to make it easier to see what's going on in the terminal later.

for (std::uint32_t i=0U; i<50U; i++)
{
    Point data {
        .x = i,
        .y = i * 2U
    };

    std::cout << "Sending data ("
              << data.x << ", " << data.y << ")" << std::endl;

    position_skeleton.center.Send(data);
    std::this_thread::sleep_for(std::chrono::seconds(1U));
}

After sending the data 50 times, stop offering the service and exit:

position_skeleton.StopOfferService();

return 0;

Subscriber implementation

The subscriber begins similarly to the publisher, so first we add the necessary headers in src/subscriber/main.cpp:

#include "score/mw/com/impl/instance_specifier.h"
#include "score/mw/com/runtime.h"

#include "model/position_interface.h"

And define the constants, but here there is one more than in the publisher - the size of the buffer of the incoming data (once again - don't worry about the config path for now):

static constexpr const char* kPathToComConfig {"config/communication_config.json"};
static constexpr const char* kInstanceSpecifier {"SomeInstanceSpecifier"};
static constexpr std::size_t kBufferSize {3U};

In the main function we start with setting the config file path and the instance specifier creation:

score::StringLiteral runtime_args[2U] = {"--service_instance_manifest", kPathToComConfig};
score::mw::com::runtime::InitializeRuntime(2, runtime_args);

const score::Result<score::mw::com::InstanceSpecifier> instance_specifier_result = score::mw::com::InstanceSpecifier::Create(std::string{kInstanceSpecifier});

if (!instance_specifier_result.has_value())
{
    std::cerr << "Error: failed to create instance specifier!" << std::endl;
    return 1;
}

const score::mw::com::InstanceSpecifier instance_specifier = instance_specifier_result.value();

Now comes the important part - actually finding the service that our publisher offers. In general, there are 2 ways of finding the service:

synchronous with FindService function
asynchronous (non-blocking) with StartFindService function

In this example I will use FindService to make the example as explicit as possible by iterating and waiting 1 second after each failed iteration:

score::mw::com::ServiceHandleContainer<score::mw::com::impl::HandleType> services{};

do
{
    const auto services_result = PositionInterfaceProxy::FindService(instance_specifier);

    if (!services_result.has_value())
    {
        std::cerr << "Error: failed to find services for specifier "
                  << instance_specifier.ToString() << ": "
                  << services_result.error() << std::endl;
        return 1;
    }

    services = services_result.value();

    if (services.size() == 0U)
    {
        std::this_thread::sleep_for(std::chrono::seconds(1U));
    }
} while (services.size() == 0);

const auto service = services.front();

When the service is found, create the proxy via which subscriber will receive the data sent by the publisher:

auto proxy_result = PositionInterfaceProxy::Create(service);

if (!proxy_result.has_value())
{
    std::cerr << "Failed to create proxy for the found service!" << std::endl;
    return 1;
}

auto& position_proxy = proxy_result.value();

Now the most important part - the implementation of receiving the data. You do that by setting a receive handler on the center event. When the receive handler is called, it means that the new data is ready to be read. Inside of that receive handler implementation, you call GetNewSampes which let's you to actually access the received data:

position_proxy.center.SetReceiveHandler([&position_proxy](){
    score::Result<std::size_t> num_samples_received = position_proxy.center.GetNewSamples(
    [](score::mw::com::SamplePtr<Point> point) noexcept {
        if (!point) {
            std::cerr << "Received data is invalid" << std::endl;
            return;
        }

        std::cout << "Received data ("
                  << point->x << ", " << point->y << ")" << std::endl;
    },
    kBufferSize);

    if (!num_samples_received.has_value()) {
        std::cerr << "num_samples_received does not have a value!" << std::endl;
        return;
    }

    std::cout << "Received " << num_samples_received.value()
              << " new samples" << std::endl;
});

After that, S-Core knows how you want to read the data, but to actually start receiving it, you must subscribe to the center event:

position_proxy.center.Subscribe(kBufferSize);

Because calling the receive handler is done asynchronously in the background, add the loop preventing application from exiting. You can base it on the signal handler, but for purpose of this simple example, I'll just go with:

while (true) {}

If you however go with the proper signal handling, you may want to add a cleanup after the loop which unsubscribes from the center event:

position_proxy.center.Unsubscribe();
return 0;

Communication configuration

Ok, now it's the time to start bother about the line which both publisher and subscriber share and which I earlier told you not to worry about:

static constexpr const char* kPathToComConfig {"config/communication_config.json"};

The implementation of the publisher and subscriber is not enough - the communication must still be configured in the JSON file by providing at least information presented below. Create the config/communication_config.json file with the following content:

Note: I keep instance specifier as SomeInstanceSpecifier and service type name as SomeServiceTypeName instead of something like PositionService or Point to show that the names in the configuration file does not need to be the same or even related to the names in the interface model. They need to be however in sync inside the configuration file, so if you set serviceTypeName to SomeServiceTypeName in the serviceTypes list, you must use the same serviceTypeName in the serviceInstances list.

{
  "serviceTypes": [
    {
      "serviceTypeName": "SomeServiceTypeName",
      "version": {
        "major": 1,
        "minor": 0
      },
      "bindings": [
        {
          "binding": "SHM",
          "serviceId": 1,
          "events": [
            {
              "eventName": "center",
              "eventId": 1
            }
          ]
        }
      ]
    }
  ],
  "serviceInstances": [
    {
      "instanceSpecifier": "SomeInstanceSpecifier",
      "serviceTypeName": "SomeServiceTypeName",
      "version": {
        "major": 1,
        "minor": 0
      },
      "instances": [
        {
          "instanceId": 1,
          "asil-level": "QM",
          "binding": "SHM",
          "events": [
            {
              "eventName": "center",
              "numberOfSampleSlots": 3,
              "maxSubscribers": 10
            }
          ]
        }
      ]
    }
  ]
}

Read also on Medium: Combining Bazel with Docker

Building the project

General project setup

Create empty WORKSPACE file in the root of your project. Then create the Bazel runtime configuration .bazelrc file with the following content:

common --@score_baselibs//score/mw/log/detail/flags:KUse_Stub_Implementation_Only=False
common --@score_baselibs//score/mw/log/flags:KRemote_Logging=False
common --@score_baselibs//score/json:base_library=nlohmann
common --@score_communication//score/mw/com/flags:tracing_library=stub

common --registry=https://raw.githubusercontent.com/eclipse-score/bazel_registry/v0.5.0-beta/

common --registry=https://bcr.bazel.build

After that, configure thirdparty modules in MODULE.bazel file:

bazel_dep(name = "score_toolchains_gcc", version = "0.5", dev_dependency=True)

gcc = use_extension("@score_toolchains_gcc//extensions:gcc.bzl", "gcc", dev_dependency=True)
gcc.toolchain(
    url = "https://github.com/eclipse-score/toolchains_gcc_packages/releases/download/0.0.1/x86_64-unknown-linux-gnu_gcc12.tar.gz",
    sha256 = "457f5f20f57528033cb840d708b507050d711ae93e009388847e113b11bf3600",
    strip_prefix = "x86_64-unknown-linux-gnu",
)

use_repo(gcc, "gcc_toolchain", "gcc_toolchain_gcc")

bazel_dep(name = "rules_boost", repo_name = "com_github_nelhage_rules_boost")
archive_override(
    module_name = "rules_boost",
    urls = ["https://github.com/nelhage/rules_boost/archive/refs/heads/master.tar.gz"],
    strip_prefix = "rules_boost-master",
)

bazel_dep(name = "boost.program_options", version = "1.87.0")
bazel_dep(name = "score_baselibs", version = "0.2.2")
bazel_dep(name = "score_communication", version = "0.1.2")

bazel_dep(name = "trlc", version = "0.0.0")
git_override(
    module_name = "trlc",
    commit = "ede35c4411d41abe42b8f19e78f8989ff79ad3d8",
    remote = "https://github.com/bmw-software-engineering/trlc.git",
)

Interface model setup

Create position_interface target in model/BUILD file:

load("@score_baselibs//score/language/safecpp:toolchain_features.bzl", "COMPILER_WARNING_FEATURES")

cc_library(
    name = "position_interface",
    hdrs = [
        "position_interface.h",
    ],
    features = COMPILER_WARNING_FEATURES,
    deps = [
        "@score_communication//score/mw/com",
        "@score_baselibs//score/language/futurecpp",
    ],
    visibility = [
        "//src/publisher:__pkg__",
        "//src/subscriber:__pkg__"
    ]
)

Configuration setup

Our configuration is a single JSON file, so there's nothing to build per se, but we need to export the configuration file, so that it can be later used by publisher and subscriber targets as a target in data attribute. In config/BUILD add:

exports_files([
    "communication_config.json",
])

Publisher setup

In src/publisher/BUILD create publisher binary target:

load("@score_baselibs//score/language/safecpp:toolchain_features.bzl", "COMPILER_WARNING_FEATURES")

cc_binary(
    name = "publisher",
    srcs = ["main.cpp"],
    data = ["//config:communication_config.json"],
    features = COMPILER_WARNING_FEATURES,
    deps = [
        "//model:position_interface",
        "@score_communication//score/mw/com",
    ],
)

Subscriber setup

In src/subscriber/BUILD create subscriber binary target:

load("@score_baselibs//score/language/safecpp:toolchain_features.bzl", "COMPILER_WARNING_FEATURES")

cc_binary(
    name = "subscriber",
    srcs = ["main.cpp",],
    data = ["//config:communication_config.json"],
    features = COMPILER_WARNING_FEATURES,
    deps = [
        "//model:position_interface",
        "@score_communication//score/mw/com",
    ],
)

Building both applications

Now, to build the publisher, call:

bazel build //src/publisher

And to build the subscriber call:

bazel build //src/subscriber

Note: during my tests I was not able to successfully build the project without adding --copt=-Wno-error=deprecated-declarations to the above build commands.

Running

It's finally time to run both applications. Open 2 terminals. In the first one call:

bazel run //src/publisher

And in the second one call:

bazel run //src/subscriber

In the publisher's terminal you should see logs like:

Sending data (5, 10)
Sending data (6, 12)
Sending data (7, 14)

And in the subscriber's terminal you should see logs like:

Received data (5, 10)
Received 1 new samples
Received data (6, 12)
Received 1 new samples
Received data (7, 14)

How to write Arduino Uno code with Python?

pikoTutorial — Tue, 14 Oct 2025 07:15:00 +0000

Recently I came across a Reddit thread where someone asked:

"I was thinking about using an Arduino, but I have been learning Python and do not want to learn a whole other programming language to do basic things that an Arduino can. (Planning on making a robot, will be using raspberry pi also, but have seen motor controls and such are best done with Arduino).

What experience have you had programming an Arduino with python? How is it and is it difficult?"

There are more threads on the similar topic (like this and that) and when you scroll through the responses on all of these threads, you see people responding with messages like:

"you can’t program an arduino with python"

Or:

"Arduino does not have sufficient resources to run python."

Or:

"It's going to be extremely difficult to get any kind of Python script running directly on the Arduino."

Or:

"You gotta use pyfirmata"

What surprised me was that everyone seemed to be answering a question the original poster didn’t actually ask. The guy who started the thread didn’t say that he wanted to literally run Python on the Arduino and definitely he didn't want to just control the Arduino from the host (like the answer about pyfirmata suggests). To me, it sounds more like he just wanted to write his Arduino code in Python - and that he wouldn’t really care if the board ended up running an ordinary C++.

I thought that it shouldn't be that hard to write some reasonable MVP which does exactly that. This gave me the idea for an experimental project: PyBedded. It’s a tool that lets you write Arduino code in Python and then flash it to your board by just running your Python script. You still get the simplicity of Python, but the Arduino ends up running the same C++ firmware as it would normally do.

TL;DR

If you're here to just use the tool and not read about its details, download the GitHub repository and read the README to learn how to use it.

What this project is NOT

At the beginning, to avoid any disappointments after reading, I'd like to clearly outline what to not expect from this project:

as mentioned in the introduction, this is not another Firmata-like project. The original post on Reddit mentioned about the ability to program Arduino which could run as an independent device, so controlling the board with a Python script from host doesn't fulfill that requirement
I won't try to deploy the actual Python interpreter on the Arduino board or use Micro/CircuitPython which are supported only by some subset of all boards - my assumption is that someone has an old Arduino (like Uno, Nano, Mega2560 etc.) and wants to write a Python code
it's not (only) about the Python to C++ conversion - I want the tool to be "self-contained" meaning, that the user doesn't need to use tool A to convert the Python code, tool B to compile it, tool C to flash it etc. This tool should actually flash the board just by running the script with Arduino code written in Python

Read also on pikotutorial.com: UDP multicasting with Python

Challenges

Workflow

The main challenge was to define a workflow that would fulfill all the assumptions made in the previous paragraph . In the end, I wanted user to just write an ordinary Python script (with a reasonable amount of additional boilerplate code), run it and have the program running on the Arduino board.

The framework ended up looking as following - user uses an externally defined type ArduinoBoard which provides context manager functionality. The entire code contained within this block will be treated as code which is intended to run on Arduino.

with ArduinoBoard("/dev/ttyUSB0", Board.UNO):
    # Arduino code

The Arduino users are used to a specific main file structure which requires definition of setup and loop functions. Let's make them mandatory also in PyBedded. Python supports function definition inside the context manager's block, so the minimum required code looks like this:

with ArduinoBoard("/dev/ttyUSB0", Board.UNO):
    def setup() -> None:
        pass

    def loop() -> None:
        pass

Such structure gives also another benefit - the ease of finding where the Arduino code starts.

I mentioned above, that everything below the context manager block will end up flashed to Arduino, but how to actually find it in the file? I don't want to rely on hardcoding the line number because the user may add something above it. I also don't want to rely on searching of substring like "ArduinoBoard(" because if the name will be changed, I would have to change code extraction logic.

However, because I know that the user always uses context manager, I can use the inspect module to check from what file and which line in that file ArduinoBoard's context manager has been called:

import inspect

frame: FrameInfo = inspect.stack()[2]
file_path: str = frame.filename
start_line: int = frame.lineno

LOGGER.debug(f"Python code available in {file_path}, starting at line {start_line}")

Support for Arduino and external libraries

At the beginning one of my concerns was how to provide support for all the Arduino's core API (stuff like digitalWrite, digitalRead etc.) and all the libraries associated with it (like Servo, Ethernet etc.), but then I realized that actually the only thing that's needed to write a valid Python code, is the list of corresponding function's and classes' definitions. For example, to enable PyBedded user to set digital pin state, the only thing that needs to be provided is:

def digitalWrite(pin: int, state: int) -> None:
    pass

The call of this function will look exactly the same both in Python and C++ what makes the code conversion easier. Appropriate C++ header will be included in the generated code depending on the functions used in the code.

Toolchain

The main tool that will be used underneath will be Arduino CLI. The following commands will be especially useful in this project:

arduino-cli lib install - I will use it to in the installation script to install all the currently supported libraries which can be used in the sketch
arduino-cli core install - I will use it in the installation script to install all the currently supported boards
arduino-cli compile - compilation of the generated C++ code
arduino-cli upload - flashing Arduino with the compiled firmware

Design

The design relies almost entirely on the Python's context manager functionality - as soon as the with ArduinoBoard... block finishes, the following steps are executed:

read the script's path and the line where Arduino code starts
extract Python code from the script which is going to be converted to C++
create Arduino project (folder and .ino file)
compile the code
upload the code to the Arduino board
remove Arduino project folder and .ino file

The flow is nice and easy, so let's now look at the Python-to-C++ code conversion.

Python to C++ conversion

There are whole books about writing compilers, code transpilation, parsers, lexers etc., but because this proof of concept is rather focused on the Python framework for Arduino programming, I decided to implement a simple Python to C++ converter with the following structure:

Python code -> PythonParser -> code model -> CppGenerator -> C++ code

PythonParser takes as an input Python code from the user's script and converts it to a common, language-agnostic model. In practice, PythonParser will be responsible for recognizing certain language features and creating their models like this:

for python_line in python_code:
    if is_function_definition(python_line):  
        # parsing
        return FunctionDefinitionModel(...)  
    elif is_for_loop(python_line):  
        # parsing  
        return ForLoopModel(...) 
    elif is_if_statement(python_line):  
        # parsing
        return IfStatementModel(...)
    # etc

Then this model is taken as an input by the CppGenerator which runs over all the models and converts these models to the actual C++ code line depending on the type of the each model object:

cpp_code: str = ""

for model in models:
    if isinstance(model, FunctionDefinitionModel):
        cpp_code += generate_function_definition(model)
    elif isinstance(model, ForLoopModel):
        cpp_code += generate_for_loop(model)
    elif isinstance(model, IfStatement):
        cpp_code += generate_if_statement(model)
    # etc

Such logic is simple, but not perfect - it makes it ease to add support for the new features in simple scenarios (you just add a single elif in PythonParser and elif in CppGenerator), but can introduce some unexpected behavior because the parser actually relies on the order of the elifs.

To make it work, the least inclusive conditions must be placed earlier in the elif "ladder" and more inclusive ones must be placed later. For example, in the following code, the second condition will never be triggered because "if" is contained both in "if" and "elif", so it would never generate any "else if" in the C++ code.

if "if" in python_line:
    # process
elif "elif" in python_line:
    # process

So "elif" (as the less inclusive condition) must be checked before "if" (which is more general condition and matches more cases):

if "elif" in python_line:
    # process
elif "if" in python_line:
    # process

Ok, so let's now take a look on how to actually write the Arduino code with PyBedded and what C++ code it is being converted to.

Read also on Medium: Make C++ a batter place #2: CppFront as an alternative

Variable definition

Variable definition looks just as they do in Python with one addition - types annotations are no longer optional, but mandatory since C++ requires them. So the following code:

sensor_value: int = 0

Ends up as:

int sensor_value = 0;

In the resulting sketch.

Function definition

Python allows to define new functions almost everywhere, so user places custom functions definitions inside the Arduino context manager's block (in the same way as setup and loop functions are defined). The syntax is a typical Python's syntax, but again, type annotations are mandatory:

def do_something(a: int, b: unsigned_int) -> None:
    pass

The above code will be converted to:

void do_something(int a, unsigned int b)
{}

Notice 2 things:

None is an equivalent of void
types like unsigned int, unsigned longetc. are not supported by default in Python, so PyBedded adds such support by defining new aliases like:

unsigned_long = int  
unsigned_int = int

This way user can use such types and still be able to write a valid Python code.

Control statements

There are control statements whose syntax is very similar in Python and C++ (e.g. ifor while loop) and control statements whose syntax is completely different (e.g. for loops).

In case of the first category, generation boils down mainly to adding parenthesis and converting logic operators to the ones used in C++, so that this:

if not Serial and (a == 1 or b == 2):
    pass

becomes that:

if (!Serial && (a == 1 || b == 2))
{}

In case of the second category, you still write an ordinary Python syntax, but underneath Python-to-C++ converter tries to match the corresponding values to the C++'s syntax. Input:

# 1
for i in range(10):
    pass
# 2
for i in range(2, 10):
    pass
# 3
for i in range(2, 10, 3):
    pass
# 4
for i in range(10, 0, -1):
    pass

Output:

# 1
for (int i=0; i<10; i += 1)
{}
# 2
for (int i=2, i<10; i += 1)
{}
# 3
for (int i=2; i<10; i += 3)
{}
# 4
for (int i=10, i>0; i += -1)
{}

Preprocessor directives

Python is not a compiled programming language, so preprocessor directives like #define or ifdef must be supported by some kind of a convention.

In case of #define, I decided to rely on a naming convention - if a variable is declared using all uppercase letters, then it will end up as a compile-time constant. So the following Python code:

SENSOR_PIN: int = 4
sensor_value: int = 0

is converted to:

#define SENSOR_PIN 4
int sensor_value = 0;

When it comes to compile-time conditions, PyBedded just provides 3 special functions: IFDEF, IFNDEF and ENDIF:

IFDEF("SOME_FLAG")
sensor_value += 1
ENDIF()

C++ code equivalent:

#ifdef SOME_FLAG
sensor_value += 1;
#endif

Read also on pikotutorial.com: Hacking Python functions by changing their source code

Examples

Blink

Python code:

# create ArduinoBoard object providing the port it is connected to
# and the board's type 
with ArduinoBoard("/dev/ttyUSB0", Board.UNO):  
    def setup() -> None:
        # set pin mode  
        pinMode(LED_BUILTIN, OUTPUT)  

    def loop() -> None:
        # set pin state to high
        digitalWrite(LED_BUILTIN, HIGH)
        # wait 1000ms  
        delay(1000)  
        # set pin state to low
        digitalWrite(LED_BUILTIN, LOW)
        # wait 1000ms  
        delay(1000)

The generated C++ code:

void setup() {  
    pinMode(LED_BUILTIN, OUTPUT);  
}  
void loop() {  
    digitalWrite(LED_BUILTIN, HIGH);  
    delay(1000);  
    digitalWrite(LED_BUILTIN, LOW);  
    delay(1000);  
}

Blink without delay

Python code:

with ArduinoBoard("/dev/ttyUSB0", Board.UNO):  
    led_pin: int = LED_BUILTIN  
    led_state: int = LOW  
    previous_millis: unsigned_long = 0  
    interval: long = 1000  

    def setup() -> None:  
        pinMode(led_pin, OUTPUT)  

    def loop() -> None:
        global previous_millis, led_state  

        current_millis: unsigned_long = millis()  

        if current_millis - previous_millis >= interval:  
            previous_millis = current_millis  

            if led_state == LOW:  
                led_state = HIGH  
            else:  
                led_state = LOW  

            digitalWrite(led_pin, led_state)

The generated C++ code:

int led_pin = LED_BUILTIN;  
int led_state = LOW;  
unsigned long previous_millis = 0;  
long interval = 1000;  
void setup() {  
    pinMode(led_pin, OUTPUT);  
}  
void loop() {  
    unsigned long current_millis = millis();  
    if (current_millis - previous_millis >= interval) {  
        previous_millis = current_millis;  
        if (led_state == LOW) {  
            led_state = HIGH;  
        }  
        else {  
            led_state = LOW;  
        }  
        digitalWrite(led_pin, led_state);  
    }  
}

Servo sweep

Python code:

with ArduinoBoard("/dev/ttyUSB0", Board.UNO):  
    myservo: Servo = Servo()  
    pos: int = 0  

    def setup() -> None:  
        myservo.attach(9)  

    def loop() -> None:  
        for pos in range(180):  
            myservo.write(pos)  
            delay(15)  
        for pos in range(180, 0, -1):  
            myservo.write(pos)  
            delay(15)

The generated C++ code:

#include <Servo.h>  
Servo myservo = Servo();  
int pos = 0;  
void setup() {  
    myservo.attach(9);  
}  
void loop() {  
    for (int pos=0; pos<180; pos += 1) {  
        myservo.write(pos);  
        delay(15);  
    }  
    for (int pos=180; pos>=0; pos += -1) {  
        myservo.write(pos);  
        delay(15);  
    }  
}

Debugging options

While working with PyBedded I found it useful to e.g. compile the code, but not upload it to the board or compile and debug the compilation error by analysing the generated C++ code, so ArduinoBoard object accepts couple of additional parameters-flags:

compile - by default True, omits compilation step if set to False
upload - by default True, omits board flashing if set to False
clean_up - by default True, omits Arduino project folder removal if set to False

Maturity of the project

Currently the project supports all the features used in the examples available in Arduino IDE, except custom classes/structs definition.

Combining Bazel with Docker

pikoTutorial — Thu, 18 Sep 2025 07:09:00 +0000

Welcome to the next pikoTutorial!

In one of the recent articles, I showed how to build and run Docker containers using CMake. Today, we will see how to do a similar thing, but using Bazel. Today's project structure:

project/
├── app1/
│   ├── BUILD
│   ├── main.py
│   ├── run_container.sh
│   ├── some_config.json
├── app2/
│   ├── BUILD
│   ├── main.py
│   ├── run_container.sh
│   ├── some_config.json
├── MODULE.bazel
└── WORKSPACE

Note: on the contrary to the project structure used in the article about CMake, this file tree does not contain any Dockerfile. It's because Bazel supports building OCI-compliant images which are independent from a specific tool (like Docker). At the very last step, however, I will use Docker to run the container.

The MODULE.bazel file

First of all, we need to specify the dependencies of our build. They contain rules which I will use in the Bazel files:

# dependency necessary for creating a python binary and a python
# application layer in the image
bazel_dep(name = "aspect_rules_py", version = "1.4.0")
# dependency necessary for building OCI images
bazel_dep(name = "rules_oci", version = "2.2.6")
# dependency necessary for pkg_tar rule which we will use to create
# an additional image layer
bazel_dep(name = "rules_pkg", version = "1.1.0")

The WORKSPACE file

Next is the WORKSPACE file. Certain operations (like pulling a base image) can be executed only at the stage of loading the workspace, so they must be contained in the WORKSPACE file.

load("@rules_oci//oci:dependencies.bzl", "rules_oci_dependencies")
rules_oci_dependencies()

load("@rules_oci//oci:repositories.bzl", "oci_register_toolchains")
oci_register_toolchains(name = "oci")

load("@rules_oci//oci:pull.bzl", "oci_pull")
oci_pull(
    name = "python_base",
    image = "python",
    tag = "3.9-slim",
    platforms = ["linux/amd64"],
)

In this article, there are two Python applications, so I pull a Python-specific base image, but in case you use another language, this is the place to specify a proper base image.

Read also on pikotutorial.com: Combining CMake with Docker

app1/main.py and app2/main.py files

Nothing special here, just some Python code printing the content of the configuration file to test if the container works:

import json

with open("some_config.json", "r") as file:
    print(f"Configuration from app 1: {json.load(file)}")

The some_config.json file is a single-key dictionary:

{
    "Hello": "World"
}

Similarly, the content of app2/some_config.json is:

{
    "Bye": "World"
}

BUILD files

Both app1/BUILD and app2/BUILD files look similar with the only difference being the application name:

load("@aspect_rules_py//py:defs.bzl", "py_binary", "py_image_layer")
load("@rules_oci//oci:defs.bzl", "oci_image", "oci_load")
load("@rules_pkg//:pkg.bzl", "pkg_tar")
# create a Python application binary
py_binary(
    name = "main",
    srcs = ["main.py"],
)
# create a Python application layer in the image
py_image_layer(
    name = "application_layer",
    binary = ":main",
)
# create an example additional layer in the image
pkg_tar(
    name = "configuration_layer",
    srcs = ["some_config.json"],
)
# create OCI image
oci_image(
    name = "image_definition",
    base = "@python_base",
    entrypoint = ["/app1/main"],
    tars = [":configuration_layer", ":application_layer"],
)
# load the image into the local runtime
oci_load(
    name = "image",
    image = ":image_definition",
    repo_tags = ["image_app_1:latest"],
)

With such a BUILD file, I defined an example two-layer OCI image which can be then loaded to the local runtime (and accessed e.g. with Docker).

Building the images

Now when everything's ready, you can build the app1 image by calling:

bazel run //app1:image

This will produce the following output:

INFO: Analyzed target //app1:image (152 packages loaded, 3970 targets configured).
INFO: Found 1 target...
Target //app1:image up-to-date:
  bazel-bin/app1/image.sh
INFO: Elapsed time: 5.106s, Critical Path: 4.20s
INFO: 40 processes: 25 internal, 14 linux-sandbox, 1 local.
INFO: Build completed successfully, 40 total actions
INFO: Running command line: bazel-bin/app1/image.sh
6c4c763d22d0: Loading layer [==================================================>]  28.23MB/28.23MB
41757dc445c9: Loading layer [==================================================>]  3.512MB/3.512MB
529e75018436: Loading layer [==================================================>]  14.93MB/14.93MB
678221e973fe: Loading layer [==================================================>]     249B/249B
c025797afb0a: Loading layer [==================================================>]  10.24kB/10.24kB
7e7cd3ed4a69: Loading layer [==================================================>]  2.483MB/2.483MB
f465bed610cd: Loading layer [==================================================>]  23.19MB/23.19MB
7141b744acb3: Loading layer [==================================================>]  1.103MB/1.103MB
Loaded image: image_app_1:latest

And similarly for the app2 image:

bazel run //app2:image

Note for beginners: notice that oci_load rule that I used to define targets //app1:image and //app2:image must be called using bazel run. Calling bazel build on these targets will not load the images to the local runtime.

After the build is done, you can run:

docker images

to check that 2 new images appeared:

REPOSITORY    TAG       IMAGE ID       CREATED         SIZE
image_app_1   latest    95ef84a311e5   2 minutes ago   214MB
image_app_2   latest    8d757bc74a7e   2 minutes ago   214MB

You can run one of them by calling, e.g.:

docker run --rm --name container_app_1 image_app_1

And you see that not only does our application work inside the container, but it also has the access to the configuration file added to our image in the configuration layer:

Configuration from app 1: {'Hello': 'World'}

Read also on Medium: Trying ROS2: pub/sub within a single container

Running Docker container with Bazel

But why stop here? We have a nice abstraction for the image creation (in a form of Bazel //app1:image target), so let's add a similar abstraction for running the container from Bazel level like this:

bazel run //app1:container

In order to do that, there needs to be some kind of wrapper script which will be then used in the Bazel BUILD file, let's call it run_container.sh:

#!/bin/bash

docker run --rm --name container_app_1 image_app_1

Then, it can be used within the sh_binary rule in the app1/BUILD file:

sh_binary(
    name = "container",
    srcs = ["run_container.sh"],
    data = [":image"],
)

Now, when you call:

bazel run //app1:container

Bazel will handle running the specified container:

INFO: Analyzed target //app1:container (0 packages loaded, 24 targets configured).
INFO: Found 1 target...
Target //app1:container up-to-date:
  bazel-bin/app1/container
INFO: Elapsed time: 0.275s, Critical Path: 0.09s
INFO: 5 processes: 5 internal.
INFO: Build completed successfully, 5 total actions
INFO: Running command line: bazel-bin/app1/container
Configuration from app 1: {'Hello': 'World'}

Note for beginners: target //app1:container will not detect any changes done in the image (e.g. in the application code), so with such setup, if you want to run the newest version of the image, you must make sure that bazel run //app1:image has been called before running bazel run //app1:container.

Running commands with timeout on Linux

pikoTutorial — Tue, 16 Sep 2025 07:09:00 +0000

Welcome to the next pikoTutorial!

Bash comes with a bunch of useful, built-in commands. One of them is timeout which allows you to run another command with a time limit. The syntax is simple:

timeout [duration] [target_command]

To avoid providing large numbers for long timeouts, duration parameter accepts suffixes which mark the exact time frame:

s - seconds
m - minutes
h - hours
d - days

Simple example

For the first example, let's take a simple Python script which spins forever:

# some_job.py
import time

while True:
    time.sleep(1)

We can run it with a 3 seconds timeout using the following command:

timeout 3s python3 some_job.py

After 3 seconds of waiting, some_job.py is terminated.

Specifying termination signal

When the allowed time elapses, timeout sends SIGTERM signal to the given command. In your implementation you may have some custom signal handlers which do the cleanup before exiting. For example, if you expect that your script will be interrupted mainly by CTRL + C, you most probably have a SIGINT signal handler. In such case, you most likely want to keep this in case of timeout command as well to still be able to perform the cleanup when the timeout occurs:

# some_job.py
import sys
import signal
import time

def handle_sigint(signum, frame):
    print("Received SIGINT, cleaning up and exiting...")
    sys.exit(0)

signal.signal(signal.SIGINT, handle_sigint)

print("Starting script...")
time.sleep(10)

If you now call:

timeout 3s python3 some_job.py

You will see that your signal handler has not been called before exiting the script. To change this, specify the expected signal using --signal option:

timeout --signal=SIGINT 3s python3 some_job.py

After that, the log from handle_sigint handler is visible when the timeout occurred.

Read also on pikotutorial.com: How to search the internet from Linux terminal?

Giving time for graceful exit

In the previous example I assumed that the cleanup is quick and always succeeds, but in reality the application shutdown may be time consuming:

# some_job.py
import sys
import signal
import time

def handle_sigterm(signum, frame):
    print("Received SIGTERM, cleaning up and exiting...")
    time.sleep(5)

signal.signal(signal.SIGTERM, handle_sigterm)

print("Starting script...")
time.sleep(10)

In such case, it may require its own timeout for wrapping things up - upon violation of that final timeout, the application should be killed. This can be achieved by using --kill-after. The following command lets the script run for 5 seconds, sends SIGTERM (default signal) and then it gives the script 2 more seconds to exit. If the script doesn't exit within 2 seconds after receiving SIGTERM, SIGKILL is sent:

timeout --kill-after=2s 5s python3 some_job.py

Preserving return code

By default, if the command provided to timeout times out, its return code will be equal to 124. You can check it by running the example Python script and then checking its return code:

timeout 3s python3 some_job.py
echo $?

Note for beginners: ? in the above section is a special Shell variable which stores the exit code of the last executed command. $ sign, as usually in Shell environment, obtains the underlying value of that variable, so echo $? prints the exit status of the most recent command.

However, this may be something that your system does not expect, especially if your script has some logic which ends up returning specific exit codes. For example, the following script simulates a time consuming processing of 3 different phases. In case of interruption, the script returns the number of the phase being processed during the interruption:

# some_job.py
import sys
import signal
import time

current_phase: int = 0

def handle_sigterm(signum, frame):
    sys.exit(current_phase)

signal.signal(signal.SIGTERM, handle_sigterm)

for i in range(3):
    current_phase += 1
    time.sleep(2)

To preserve that return code after timeout, use --preserve-status flag:

timeout --preserve-status 3s python3 some_job.py

Now, when you check the exit status of such operation, you will see 2 because timeout terminated script while the phase 2 was being processed.

Running Python unit tests with CMake

pikoTutorial — Sun, 31 Aug 2025 07:20:00 +0000

Welcome to the next pikoTutorial !

If you haven't read the previous pikoTutorial on setting up a Python project with CMake, you may want to check it out first because in today's article I'll be using the setup we've done there:

project/
├── app1/
    ├── CMakeLists.txt
    ├── main.py
    ├── requirements.txt
├── app2/
    ├── CMakeLists.txt
    ├── main.py
    ├── requirements.txt
├── app3/
    ├── CMakeLists.txt
    ├── main.cpp
├── build/
├── CMakeLists.txt

However, I will extend it with the test directories for each Python application:

project/
├── app1/
    ├── CMakeLists.txt
    ├── main.py
    ├── requirements.txt
    ├── test
        ├── test_app.py
├── app2/
    ├── CMakeLists.txt
    ├── main.py
    ├── requirements.txt
    ├── test
        ├── test_app.py
├── app3/
    ├── CMakeLists.txt
    ├── main.cpp
├── build/
├── CMakeLists.txt

Adding test targets

As I was showing in the pikoTutorial about running C++ tests with CMake, to enable running unit tests with ctest, we must extend app1/CMakeLists.txt file with add_test call:

# specify app1 virtual environment directory
set(APP1_VENV ${CMAKE_BINARY_DIR}/app1_venv)
# create virtual environment for app1
create_venv(${APP1_VENV} ${CMAKE_SOURCE_DIR}/app1/requirements.txt)
# add custom target to run app1
add_custom_target(run_app1
    COMMAND ${APP1_VENV}/bin/python ${CMAKE_SOURCE_DIR}/app1/main.py
    DEPENDS ${APP1_VENV}
)
# add unit tests
add_test(
    NAME app1_unit_test
    COMMAND ${APP1_VENV}/bin/python -m unittest discover ${CMAKE_SOURCE_DIR}/app1/test
)

The same goes for app2/CMakeLists.txt file. To make it work, we also need to add enable_testing() to the main CMakeLists.txt file.

Now I can run unit tests for both Python applications just by calling:

ctest

What will produce the following output:

    Start 1: app1_unit_test
1/2 Test #1: app1_unit_test ...................   Passed    0.07 sec
    Start 2: app2_unit_test
2/2 Test #2: app2_unit_test ...................   Passed    0.06 sec

Read also on pikotutorial.com: How to test method call order with unittest in Python?

Mixing unit tests between the languages

It's time extend the configuration - let's add tests not only for Python applications, but also for app3 which is a C++ application:

├── app3/
    ├── CMakeLists.txt
    ├── main.cpp
    ├── test
        ├── test_app.cpp

I will add the simplest C++ unit test definition to app3/CMakeLists.txt file:

# create an executable out of C++ code
add_executable(main main.cpp)
# add custom target to run app3
add_custom_target(run_app3
    COMMAND ${CMAKE_BINARY_DIR}/app3/main
    DEPENDS main
)
# add unit tests executable
add_executable(app3_unit_test test/test_app.cpp)
# add unit tests
add_test(
    NAME app3_unit_test
    COMMAND app3_unit_test
)

Now I call ctest and voilà - both Python and C++ tests executed with a single command:

    Start 1: app1_unit_test
1/3 Test #1: app1_unit_test ...................   Passed    0.06 sec
    Start 2: app2_unit_test
2/3 Test #2: app2_unit_test ...................   Passed    0.06 sec
    Start 3: app3_unit_test
3/3 Test #3: app3_unit_test ...................   Passed    0.00 sec

Thirdparty dependencies with FetchContent

pikoTutorial — Tue, 26 Aug 2025 07:18:00 +0000

Welcome to the next pikoTutorial !

FetchContent module tries to be the solution to never-ending discussion about the package management in C++ (I wrote a bit about it in one of the articles of Make C++ a better place series). It's not a package manager by itself, but it integrates natively with CMake which the big part of C++ community uses anyway.

FetchContent promises also easy integration even with legacy project, especially given the fact that it doesn't require any propriety format of the dependency, but re-uses what is already widely used - GitHub repositories and tar.gz archives stored in project-specific artifactories.

Dependency as a GitHub repository

Let's say I want to add my command line parsing library as a dependency to a new project. I can use FetchContent to clone the repository:

cmake_minimum_required(VERSION 3.28)
project(FetchContentExample)
# include FetchContent module
include(FetchContent)
# fetch Comlint library
FetchContent_Declare(
    # specify library name
    ComlintCpp
    # specify GitHub repository
    GIT_REPOSITORY https://github.com/pasparsw/comlint_cpp
    # specify tag or commit hash
    GIT_TAG v0.3
)
# make the content available
FetchContent_MakeAvailable(ComlintCpp)
# add executable target
add_executable(${PROJECT_NAME} main.cpp)
# link the fetched library
target_link_libraries(${PROJECT_NAME} PRIVATE
    ComlintCpp
)

The repository is downloaded to build/_deps folder and sources are available in build/_deps/comlintcpp-src folder.

Dependency as a pre-built release package

Of course, in many projects building from source is out of question because of e.g. build time. Fortunately, FetchContent allows for downloading a tar.gz files from the given URL. Comlint comes in in a form of a release package containing public headers and .so dynamic library file, so let's use it:

minimum_required(VERSION 3.28)
project(FetchContentExample)
# include FetchContent module
include(FetchContent)
# fetch ComlintCpp library
FetchContent_Declare(
    ComlintCpp
    URL https://github.com/pasparsw/comlint_cpp/releases/download/v0.3/comlint_cpp_v0.3.tar.gz
)
FetchContent_MakeAvailable(ComlintCpp)
# add executable target
add_executable(${PROJECT_NAME} main.cpp)
# include headers
target_include_directories(${PROJECT_NAME} PRIVATE
    ${comlintcpp_SOURCE_DIR}/include
)
# link library
target_link_libraries(${PROJECT_NAME} PRIVATE
    ${comlintcpp_SOURCE_DIR}/libComlintCpp.so
)

Bug of the week #11

pikoTutorial — Tue, 19 Aug 2025 07:16:00 +0000

Welcome to the next pikoTutorial!

The error we're handling today is a C++ compilation error:

invalid use of non-static member function

What does it mean?

It often occurs when the compiler encounters a place in the code where it would normally expect a static member function, but the actual function is not static. Let's say for example, that we have a function which takes a function pointer to some other function and calls it:

void RunFunction(void (*func)())
{
    func();
}

If you create a SomeClass class with a non-static Run function and try to pass it to the RunFunction function using the range operator like below:

struct SomeClass
{
    void Run()
    {
        std::cout << "Running..." << std::endl;
    }
};

int main()
{
    RunFunction(SomeClass::Run);
}

the compiler will throw an error:

error: invalid use of non-static member function ‘void SomeClass::Run()’

How to fix it?

The easiest way is to just make Run function static, by adding static keyword to its definition:

static void Run()

However, if you can't make that function static, you would have to adjust the RunFunction function to accept some callable object instead of a raw function pointer:

void RunFunction(std::function<void()> func)
{
    func();
}

And bind SomeClass::Run function to a specific object instance using std::bind:

SomeClass instance;
RunFunction(std::bind(&SomeClass::Run, instance));

Or a simple lambda:

SomeClass instance;
RunFunction([&instance]{ instance.Run(); });

Combining CMake with Docker

pikoTutorial — Tue, 12 Aug 2025 07:13:00 +0000

Welcome to the next pikoTutorial!

In one of the recent articles, I showed how to use CMake for setting up a Python project. Today, we will see how to further extend it by adding building of Docker images to the CMakeLists.txt files. Today's project structure:

project/
├── app1/
│   ├── CMakeLists.txt
│   ├── Dockerfile
│   ├── main.py
│   ├── requirements.txt
├── app2/
│   ├── CMakeLists.txt
│   ├── Dockerfile
│   ├── main.py
│   ├── requirements.txt
└── build/
├── CMakeLists.txt

Top CMakeLists.txt file

Nothing special here, just assuring Python availability and adding sub-directories to the build:

# specify minimum CMake version
cmake_minimum_required(VERSION 3.28)
# specify project name
project(CMakeWithDocker)
# find Python
find_package(Python3 REQUIRED COMPONENTS Interpreter)
# include all subdirectoies into the build
add_subdirectory(app1)
add_subdirectory(app2)

Dockerfiles

Both app1/Dockerfile and app2/Dockerfile look the same:

FROM python:3.9-slim

WORKDIR /app
COPY . /app
RUN pip install --no-cache-dir -r requirements.txt

CMD ["python", "main.py"]

CMake files

Both app1/CMakeLists.txt and app2/CMakeLists.txt look similar with the only difference being the application name:

set(IMAGE_TARGET image_app_1)
# add custom target to build image for app1
add_custom_target(${IMAGE_TARGET} ALL
    WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}/app1
    COMMAND docker build -t image_app_1 .
    COMMENT "Building image for app1"
)

Pay attention to ALL placed after the target's name. CMake, by default, doesn't build custom targets, so you must make them explicitly dependent on all target (the default CMake target executed when calling cmake without --target flag).

Read also on pikotutorial.com: Combining Bazel with Docker

Building the project

Now when everything's ready, you can build the project by calling:

cd build
cmake ..
cmake --build . -j

Note for beginners: because both of our applications are independent from each other, I'm adding -j to the build command to parallelize build and speed up the whole process.

After the build is done, you can run:

docker images

to check that 2 new images appeared:

REPOSITORY    TAG       IMAGE ID       CREATED             SIZE
image_app_1   latest    150d015d8915   3 minutes ago       150MB
image_app_2   latest    d16b1453dbcc   3 minutes ago       147MB

You can run them calling, e.g.:

docker run --rm --name container_app_1 image_app_1

Adding a dependent target

It's often a good idea to initialize the project, build, run tests etc. before building the final image. Here, as an example, I'll add initialization of the virtual environment of each Python application as a mandatory step for building the images:

set(VENV_TARGET venv_app_1)
set(IMAGE_TARGET image_app_1)
# add custom target to create virtual environment
add_custom_target(${VENV_TARGET} ALL
    WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}/app1
    COMMAND ${Python3_EXECUTABLE} -m venv venv
    COMMENT "Creating virtual environment for app1"
)
# add custom target to build image for app1
add_custom_target(${IMAGE_TARGET} ALL
    WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}/app1
    COMMAND docker build -t image_app_1 .
    COMMENT "Building image for app1"
    DEPENDS ${VENV_TARGET}
)

Notice that here I had to add DEPENDS argument to image_app_1 custom target definition. It assures that venv_app_1 target will be completed before image starts to build.

Read also on Medium: Trying ROS2: client/server withing a single container

Running Docker container with CMake

But why stop here? Let's add the target for running the container from CMake level:

set(VENV_TARGET venv_app_1)
set(IMAGE_TARGET image_app_1)
# add custom target to create virtual environment
add_custom_target(${VENV_TARGET} ALL
    WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}/app1
    COMMAND ${Python3_EXECUTABLE} -m venv venv
    COMMENT "Creating virtual environment for app1"
)
# add custom target to build image for app1
add_custom_target(${IMAGE_TARGET} ALL
    WORKING_DIRECTORY ${CMAKE_SOURCE_DIR}/app1
    COMMAND docker build -t image_app_1 .
    COMMENT "Building image for app1"
    DEPENDS ${VENV_TARGET}
)
# add custom target to run container for app1
add_custom_target(container_app_1
    COMMAND docker run --rm --name container_app_1 image_app_1
    COMMENT "Running container for app1"
    DEPENDS ${IMAGE_TARGET}
)

When it comes to specifying dependencies for the container_app_1 target, you have 2 options:

you can add DEPENDS image_app_1 line, as I did above - this will make sure that the container being started, always bases on the newest version of the image, so it can be potentially re-built every single time when running the container

[ 33%] Creating virtual environment for app1
[ 33%] Built target venv_app_1
[ 66%] Building image for app1
[+] Building 1.1s (9/9) FINISHED
=> [internal] load build definition from Dockerfile
=> => transferring dockerfile: 162B
=> [internal] load metadata for docker.io/library/python:3.9-slim
=> [internal] load .dockerignore
=> => transferring context: 2B
=> [1/4] FROM docker.io/library/python:3.9-slim
=> [internal] load build context
=> => transferring context: 125.50kB
=> CACHED [2/4] WORKDIR /app
=> CACHED [3/4] COPY . /app
=> CACHED [4/4] RUN pip install --no-cache-dir -r requirements.txt
=> exporting to image
=> => exporting layers=> => writing image
=> => naming to docker.io/library/image_app_1
[ 66%] Built target image_app_1
[100%] Running container for app1
Hello from Python app1
[100%] Built target container_app_1

you can omit DEPENDS image_app_1 line - in such approach, the user running container_app_1 target is responsible for assuring that image_app_1 image already exists, but it will allow you to just run the container basing on whatever image version has been recently built

[100%] Running container for app1
Hello from Python app1
[100%] Built target container_app_1

I didn't add ALL to container_app_1 custom target because most likely you want to run the container on demand, not during every project build. To run the container, call:

cmake --build . --target container_app_1

How to search the internet from Linux terminal?

pikoTutorial — Tue, 05 Aug 2025 07:11:00 +0000

Welcome to the next pikoTutorial!

Linux terminal is a basic tool for many software engineers and the place where we often spend a lot of our time during the work day. Even if you're not doing anything in the terminal right now, it is for sure opened for the entire time on your second screen or at the bottom of your IDE.

This is the reason why it's a good idea to incorporate as many of the every day actions as possible directly into the terminal. We've already covered how to embed hex/dec conversions, so now it's time to see how search the internet faster, without manually going to the internet browser.

How to open any website from the terminal?

If you use Google Chrome, the syntax for opening any website is very simple:

google-chrome "www.somepage.com"

How to search in Google from the terminal?

The easiest way to search Google from the terminal is to add a bash function to you .bashrc file:

google() {
    local query="$*"
    if [ -z "$query" ]; then
        google-chrome "https://www.google.com"
    else
        query=$(echo "$query" | jq -sRr @uri)
        google-chrome "https://www.google.com/search?q=$query"
    fi
}

It defines a function google which:

if called without any additional arguments, it just opens "www.google.com"
if called with any argument, it builds a Google query and opens web browser displaying Google search results for that query

For example, if I call:

google c++ future

I get my browser immediately opened with all the Google results for "c++ future" search term. Line query=$(echo "$query" | jq -sRr @uri) is crucial because it allows you to use non-alphanumerical characters (like "+") in your queries:

jq is a lightweight JSON processor which we're using here to URL-encode the input string
-s flag treats the input as a single string instead of individual lines
-R flag tells jq to treat the input as raw text instead JSON data
-r flag makes jq output raw text instead of JSON-encoded
@uri encodes the string as a URL, so special characters are replaced with percent ASCII-encoded equivalents

What else is worth having in the .bashrc file?

Search StackOverflow from the terminal

so() {
    local query="$*"
    if [ -z "$query" ]; then
        google-chrome "https://stackoverflow.com"
    else
        query=$(echo "$query" | jq -sRr @uri)
        google-chrome "https://stackoverflow.com/search?q=$query"
    fi
}

If you get some unfamiliar error prompt, it's often useful to just call:

so c++ use of deleted function

Search YouTube from the terminal

yt() {
    local query="$*"
    if [ -z "$query" ]; then
        google-chrome "https://www.youtube.com"
    else
        query=$(echo "$query" | jq -sRr @uri)
        google-chrome "https://www.youtube.com/results?search_query=$query"
    fi
}

YouTube is priceless when it comes to every GUI-based tool, so by having yt function at hand, you can always call things like:

yt vs code configure snippets

Hard code aliases for frequently used pages

If you have any pages that you use on a daily bases, just hard code the in form of the aliases. For example, when I'm in the terminal, I often need to check some error code value of the UDS protocol (Unified Diagnostics Services). There's a section on the UDS Wikipedia page for that, so I just have an alias uds in my .bashrc file for this purpose:

alias uds='google-chrome https://en.wikipedia.org/wiki/Unified_Diagnostic_Services#Negative_response_codes'

Adding # after the Wikipedia page name in the URL allows you to link directly to a page section, instead of the beginning of the page.

Folding expressions in C++

pikoTutorial — Tue, 29 Jul 2025 07:09:00 +0000

Welcome to the next pikoTutorial!

What is a folding expression?

Imagine that you have a vector of integers and you need to add some predefined values to it. The obvious choice seems to be just using push_back() function applied a couple of times:

# include <vector>

int main()
{
    std::vector<int> vec;

    vec.push_back(-3);
    vec.push_back(5);
    vec.push_back(8);
    vec.push_back(12);
    vec.push_back(47);
}

This works, but look how much vertical space it took and how many times you had to call push_back() function. To make it more concise, we could define a function which uses a C++ folding expression. Such expression applies certain function (in this case push_back()) to all elements of the parameters pack:

# include <vector>

template <typename ... Args>
void PushBackAtOnce(std::vector<int> &vec, Args&& ... args)
{
    (vec.push_back(std::forward<Args>(args)), ...); // folding expression
}

int main()
{
    std::vector<int> vec;
    PushBackAtOnce(vec, -3, 5, 8, 12, 47);
}

Folding expressions and function overloading

Such syntax can be also combined with the function's overloading. In such case, you can apply a different implementation of the function to each of the parameters pack element, depending on the type of that element.

#include <iostream>
// implementation for integer type
void Process(const int input)
{
    std::cout << "Doing something with an integer..." << std::endl;
}
// implementation for float type
void Process(const float input)
{
    std::cout << "Doing something with a float..." << std::endl;
}

template <typename ... Args>
void DoSomething(Args&& ... args)
{
    (Process(std::forward<Args>(args)), ...);
}

int main()
{
    DoSomething(12, 24.0F);
}

The output of the above code is:

Doing something with an integer...
Doing something with a float...

Read also on pikotutorial.com: Destruction order vs thread safety in C++

Potential pitfalls

You must remember that when your code relies fully on function overloading within a folding expression, you are responsible for providing overloaded function implementations for all the types - otherwise, the compilation will fail. In the example above I can add just one more argument 24.0 at the end and the compilation fails:

DoSomething(12, 24.0F, 24.0);

Literals like 24.0 (without F suffix) are resolved to a double type and because there's no implementation of Process() function for double, compiler reports an error. In such case you must either provide such implementation or create a generic function which will perform some default behavior for all the types which don't have their dedicated implementations. For example:

// generic implementation
template <typename T>
void Process(const T input)
{
    std::cout << "Unknown type - skipping..." << std::endl;
}

Now, the program compiles again and the output is:

Doing something with an integer...
Doing something with a float...
Unknown type - skipping...

How to derive from an enum in Python?

pikoTutorial — Tue, 22 Jul 2025 07:07:00 +0000

Welcome to the next pikoTutorial!

Recently I'm experimenting a lot with different ways of error handling in Python and I stumbled upon a use case in which it would be very helpful to create an enum by deriving from a different enum.

Use case

Let's say we design an interface for a class with multiple member functions, each returning some error code:

class Client:
    def execute_transaction() -> ErrorCode:
        pass
    def withdraw_money() -> ErrorCode:
        pass
    ...

The common approach is to gather all the error codes in a single enum type like the following:

class ErrorCode(Enum):
    INVALID_TOKEN = 0
    CONNECTION_DROPPED = 1
    INVALID_PRICE = 2
    OUT_OF_FUNDS = 3

However, if your interface exposes many methods, such ErrorCode type quickly becomes a mess, embracing all the possible error codes that different domains of the system may report.

The use case I wanted to try out is to split the ErrorCode type into types dedicated for the individual domains and their specific errors, so that the interface looks more like this:

class Client:
    def execute_transaction() -> TransactionError:
        pass
    def withdraw_money() -> AccountError:
        pass
    ...

TransactionError and AccountError must be defined in separate enums:

class TransactionError(Enum):
    INVALID_TOKEN = 0
    CONNECTION_DROPPED = 1
    INVALID_PRICE = 2

class AccountError(Enum):
    INVALID_TOKEN = 0
    CONNECTION_DROPPED = 1
    OUT_OF_FUNDS = 2

The problem now is that the generic errors like INVALID_TOKEN and CONNECTION_DROPPED can happen when calling both methods, so they must be repeated in both of the enums. What if I need to change the name of the INVALID_TOKEN field? I had to do that in both of them. What if I need to change the value of CONNECTION_DROPPED field? I also had to do that in both of them. What if I forgot about one of the enums? I have a problem - problem which scales proportionally to the number of error domains and remember that here we're assuming the use case in which there is plenty of different domains.

One could say that the above problem is related only to the maintainability of the code, but unfortunately there's a deeper issue which impacts the usability of such approach. If execute_transaction() returns TransactionError.CONNECTION_DROPPED error and withdraw_money() returns AccountError.CONNECTION_DROPPED error, both of these values are intended to represent exactly the same error. However, if you compare them, Python will say that they are not equal:

conn_dropped_1 = TransactionError.CONNECTION_DROPPED
conn_dropped_2 = AccountError.CONNECTION_DROPPED
print(conn_dropped_1 == conn_dropped_2)  # prints False

And the worst part is that this is expected behavior - quick look at the underlying types of both objects reveal that they actually are not equal:

print(type(conn_dropped_1))  # prints "<enum 'TransactionError'>"
print(type(conn_dropped_2))  # prints "<enum 'AccountError'>"

In order for such objects to be equal, they would have to share some base type which would own the common error codes. The domain-specific types would derive from it adding they domain-specific error codes. This would resolve both maintainability problem and the comparison problem.

class GenericError(Enum):
    INVALID_TOKEN = 0
    CONNECTION_DROPPED = 1

class TransactionError(GenericError):   
    INVALID_PRICE = 2

class AccountError(GenericError):
    OUT_OF_FUNDS = 2

Here we however hit the wall because Python does not allow for the derivation from enumeration types. I needed some kind of an "expandable enum" which would allow for derivation and which would keep the type consistency across such hierarchy of enums. And because it sounded like a good exercise, I decided to implement such expandable enums for my use case.

Read also on pikotutorial.com: UDP multicasting with Python

Expandable enum definition

Let's first define what would be the properties of such expandable enum type, especially their:

types across the hierarchy
rules of enum field definitions
fields equality
per-enum field types

Types across the hierarchy

First of all, I want to be able to use such type as every other enumeration type, what means that I want every field of my expandable enum type to be of type of the expandable enum type itself. Quick inspection of the default Python enum shows the mentioned properties:

enum.EnumType is a metaclass of type T
field T.F is of type T

class T(Enum):  
    F = 0  

print(type(T))        # prints "<class 'enum.EnumType'>"
print(type(T.F))      # prints "<enum 'T'>"
print(type(T.F) == T) # prints "True"

Assuming that expandable enum type E has field F, the generalized requirement written in Python (will be useful for unit tests) looks like below:

assert type(E) == ExpandableEnumType
assert type(E.F) == E

print(type(E))   # prints "<class 'ExpandableEnumType'>"
print(type(E.F)) # prints "<enum 'E'>"

The exact type of the field will depend on the actual type where the field has been defined. It means that if there's an expandable enum type B which has field F1 and another expandable enum type E which derives from B and defines field F2, the type of both B.F1 and E.F1 will be B, but the type of E.F2 will be E:

class B(metaclass=ExpandableEnumType):
    F1 = 0

class E(B):
    F2 = 1

assert type(B) == ExpandableEnumType
assert type(E) == ExpandableEnumType
assert type(B.F1) == B
assert type(E.F1) == B
assert type(E.F2) == E

print(type(B.F1)) # prints "<enum 'B'>"
print(type(E.F1)) # prints "<enum 'B'>"
print(type(E.F2)) # prints "<enum 'E'>"

This is probably the most important assumption because this trait will resolve the problem that we encountered when comparing TransactionError.CONNECTION_DROPPED to AccountError.CONNECTION_DROPPED. The fact of accessing F1 via derived type E should not change its underlying type. In other words, I want field CONNECTION_DROPPED to remain GenericError regardless of whether I access it through GenericError, TransactionError or AccountError type.

Rules of enum field definitions

As soon as we start deriving one enum from another, we stumble upon the problem on how to treat the situation when an enum subclass defines a field which is equal to one of the fields defined in its base class. Additionally, what does it mean for an enum field to be equal to another one? Is FIELD = 0 equal to FIELD = 1? Is FIELD = 0 equal to OTHER = 0?

First things first - enumeration field has a name and a value. In case of AccountError defined above and its only field, OUT_OF_FUNDS is the name and 2 is its value. To get rid of any confusion which could arise while using expandable enums, there must be a rule that none of the fields defined across the enum hierarchy may share common name or value. If it happens, the field will be considered as conflicting. According to this rule, the following type is invalid due to duplicated field name F1:

class B(metaclass=ExpandableEnumType):
    F1 = 0x23

class E(B):
    F1 = 0x24

And the following type is invalid due to duplicated value 0x23:

class B(metaclass=ExpandableEnumType):
    F1 = 0x23

class E(B):
    F2 = 0x23

Without introducing this rule, it would be possible to achieve very disturbing constructs. For example, it would be allowed to define the following hierarchy:

class GenericError(metaclass=ExpandableEnumType):
    CONNECTION_DROPPED = 0

class TransactionError(GenericError):   
    CONNECTION_DROPPED = 1

What would be the type of TransactionError.CONNECTION_DROPPED? GenericError or TransactionError? This is the reason why it is enough for one of the two (name or value) to be duplicated to consider the whole enum hierarchy as invalid.

Fields equality

At this point you may start asking yourself a question: why is there an alternative (or) in the above condition? Shouldn't the field be considered as a conflicting if it's equal to some other field? Shouldn't equality of two fields include name and value?

Remember that in the previous paragraph we were talking about the rules of defining fields across the enum hierarchy, not about the rules of equality of two fields. Two fields are considered to be equal if their names and values and types are the same. When dealing with inheritance, the following must be fulfilled:

class T:
    F1 = 0

class E(T):
    F2 = 1

assert E.F1 == T.F1

Per-enum field types

There's one more capability that expandable enums can bring - after allowing an enum type to have a hierarchy, it is technically possible for each of the hierarchy levels to have its own field type. Look at this theoretical example:

class Rectangle:
    def get_area(self) -> int:
        return self.value[0] * self.value[1]

class Point:
    def get_coordinates(self) -> str:
        return f"x = {self.value[0]}, y = {self.value[1]}"

class B(Rectangle, metaclass=ExpandableEnumType):
    F1 = (10, 12)

class E(B, Point):
    F2 = (5, 12)

assert B.F1.get_area() == 120
assert E.F1.get_area() == 120
assert E.F2.get_coordinates() == "x = 5, y = 12"

It is an interesting property, however I don't think it will be a good practice to do such things when using expandable enums.

Read also on Medium: Yield in Python - beyond the data generation

Implementation

Let's start with a class diagram embracing all the rules that I've describe above. For simplicity, to not repeat every time "expandable enum type", I called this project ExpanEn (Expandable Enum).

Quick description of the relations:

class ExpandableEnumType is the key element - it will be used as a metaclass of every user defined enum and will determine the behavior that I expect from the expandable enums implementing custom logic in __new__ and __setattr__ functions
class Expanen is almost empty because its only responsibility is to provide a convenient interface for the user - instead of writing metaclass=ExpandableEnumMeta in every user defined type, user will be able to just derive from Expanen and extend it with enum fields
class ExpanenField represents a single enum field (name + value). It will be a parent of every user defined enum
UserDefinedEnumType may optionally derive from some custom field type to extend the functionalities of an enum field
ConflictingEnum is just an exception to raise when conflicting fields are detected in the enum hierarchy

ExpanenField

As the first one, let's create ExpanenField because it will be necessary during ExpandableEnumType class implementation:

class ExpanenField:  
    def __init__(self, name: str, value: any):  
        self.name: str = name  
        self.value: any = value  

    def __str__(self) -> str:  
        return f"({self.name}: {self.value})"

ExpandableEnumType

Implementation of ExpandableEnumType is focused mainly on the __new__ function:

class ExpandableEnumType(type):  
    def __new__(cls, name, bases, dct):  
        new_class = super().__new__(cls, name, bases, dct)  
        # gather fields which already exist in the base classes of the type
        # which is currently being created
        existing_fields = {field for base in bases
                                 for field in base.__dict__.values()
                                 if isinstance(field, ExpanenField)}  
        # iterate over class members in the type which is currently being created
        # and check if the conflicting fields exists higher in the hierarchy - if
        # yes, raise the exception
        for name, value in dct.items():
            # omit special member functions
            if name.startswith("__"):  
                continue
            for field in existing_fields:
                # extract raw field's name from the full name being "Type.Field"
                raw_field_name: str = field.name.split(".")[-1]  

                if raw_field_name == name or field.value == value:  
                    raise ConflictingEnumField(f"Failed to create an expandable enum {new_class.__name__} due to "                                               f"duplicated enum field! {new_class.__name__}.{name}: {value} conflicts "                                               f"with {field.name}: {field.value}")  
            # if the field is unique, convert the value assigned to the class
            # member to ExpanenField type by calling its constructor
            cls.__setattr__(new_class, name, value)  
            existing_fields.add(new_class(name, value))  

        return new_class  

    def __setattr__(cls, name, value):
        # expand the raw field's name, so that it contains also the class name
        # (it will display value as Type.Field instead of just Field)
        super().__setattr__(name, cls(f"{cls.__name__}.{name}", value))  

    def __repr__(cls):  
        return f"<enum '{cls.__name__}'>"

Expanen

As mentioned earlier, Expanen class is responsible for delivering a convenient interface, so its implementation is very simple:

class Expanen(ExpanenField, metaclass=ExpandableEnumType):  
    pass

Example usage

Since the implementation is ready, let's now look at the example usage of such expandable enum construct. Let's try to recreate the situation from the beginning of this article. First I define a custom field type of my error enum:

class ErrorModel:  
    def code(self) -> int:  
        return self.value.split(": ")[0]  

    def description(self) -> str:  
        return self.value.split(": ")[1]

Now, let's re-define the errors so that they can derive from each other:

class GenericError(Expanen, ErrorModel):  
    SUCCESS = "0: Operation successful"  
    INVALID_TOKEN = "387: Token has expired"  
    CONNECTION_DROPPED = "172: Connection has been established, but \
                          then dropped due to API policy violation"  

class TransactionError(GenericError, ErrorModel):  
    INVALID_PRICE = "948: Failed to execute transaction due to invalid price"  

class AccountError(GenericError, ErrorModel):  
    OUT_OF_FUNDS = "257: Withdrawal requested, but there is no enough money"

The interface of the client class could look like this:

class Client:  
    def execute_transaction(self) -> TransactionError:  
        pass  

    def withdraw_money(self) -> AccountError:  
        pass

And finally the example usage:

try:  
    error = Client().execute_transaction()  

    if error == error.CONNECTION_DROPPED:  
        print(f"{error.description()} - reconnecting...")  
    if error in (TransactionError.INVALID_TOKEN,
                 TransactionError.INVALID_PRICE):  
        raise RuntimeError(error)  
except Exception as e:  
    print(f"Unable to recover - {e}")

For me, it looks like such expandable enumeration type could really benefit when combined with Result object, but that's a topic for another article.

GitHub repository

The project is available on GitHub, so feel free to clone and test it out.

Bug of the week #10

pikoTutorial — Tue, 15 Jul 2025 07:05:00 +0000

Welcome to the next pikoTutorial!

The error we're handling today is a C++ compilation error:

error: ‘class’ has incomplete type

Or a similar one:

error: invalid use of incomplete type ‘class’

What does it mean?

This error occurs in situations when you provide a declaration of a type, but don't provide definition of that type. For example, if I try to compile the code below:

class SomeType;

void SomeFunction(SomeType input) {}

I'll get the following error:

error: ‘input’ has incomplete type

Or if the argument is a pointer:

class SomeType;

void SomeFunction(SomeType* input)
{
    std::cout << input->GetName();
}

I'll get the compilation error saying:

error: invalid use of incomplete type ‘class SomeType’

How to fix it?

Provide the full type definition

If you can, just provide a full definition of the problematic type:

class SomeType
{
public:
    std::string GetName() { return "Albert"; }
};

void SomeFunction(SomeType input)
{
    std::cout << input.GetName();
}

Stick to the pointers

If you can't provide the full definition of the type, but at some place of your program you must declare something else with help of that type (e.g. a function), just use a pointer to that type instead of type directly:

class SomeType;

void SomeFunction(SomeType* input);

Such code compiles because in order to store an address to some type, compiler does not need to know anything about that type, so the full definition is not necessary. Remember however that as soon as you want to use that type (e.g. to call a member function on it), compiler will complain about the full type definition, so in the end, you must provide that definition somewhere in your code base.