Embedded Ros Bridge

---
name: embedded-ros-bridge
description: Use when writing firmware that bridges a microcontroller (STM32/micro-ROS) to a ROS 2 graph — bare-metal C/C++ driver patterns, UART/CAN transport layers, micro-ROS executor setup, and LVGL GUI initialization for embedded displays.
---

# Embedded / micro-ROS Bridge Firmware

Rules and templates for STM32 (or similar Cortex-M) firmware that participates in a ROS 2
graph via micro-ROS, plus patterns for raw CAN bus drivers and embedded LVGL displays that sit
alongside it.

## 1. Architecture overview

```
┌─────────────────────┐   serial / UDP   ┌───────────────────────────────┐
│ micro-ROS Agent      │◄────────────────►│ STM32 firmware                │
│ (ros2 run micro_ros_ │   XRCE-DDS       │  - rclc executor              │
│  agent ...)          │                  │  - publishers/subscribers     │
│ runs on the Linux    │                  │  - HAL-based transport layer  │
│ host / SBC           │                  │  - sensor/motor drivers       │
└─────────────────────┘                  └───────────────────────────────┘
```

- The **agent** runs on the Linux side (the robot's SBC or a dev machine) and bridges XRCE-DDS
  micro-ROS traffic into the full ROS 2 graph.
- The **client** (this firmware) links against `micro_ros_stm32cubemx_utils` /
  `micro_ros_platformio` and only ever depends on `rcl`, `rclc`, and generated message headers
  — never the full `rclcpp`, which doesn't fit on a microcontroller.
- **Version pinning matters**: the micro-ROS client library major.minor must match the agent's
  major.minor (both track a ROS 2 distro, e.g. both built against Humble). A mismatch fails
  the XRCE-DDS session handshake with an opaque timeout, not a clear version error.

## 2. Firmware project structure (STM32CubeIDE / CubeMX + micro-ROS)

```
firmware/
├── Core/
│   ├── Inc/                 # CubeMX-generated HAL headers
│   └── Src/
│       ├── main.c           # rclc_support_init, node/pub/sub creation, executor loop
│       ├── freertos.c        # (optional) if using a micro-ROS FreeRTOS task
│       └── uart_transport.c  # custom rmw transport bound to a HAL UART
├── microros_static_library/  # generated by the micro_ros_setup / Docker builder
└── Drivers/                  # CubeMX HAL + BSP drivers
```

## 3. `main.c` pattern (bare-metal, no RTOS)

```c
#include "rcl/rcl.h"
#include "rclc/rclc.h"
#include "rclc/executor.h"
#include <geometry_msgs/msg/twist.h>
#include <std_msgs/msg/int32.h>

#define RCCHECK(fn) do { \
  rcl_ret_t rc = fn; \
  if (rc != RCL_RET_OK) { on_rcl_error(rc, __LINE__); } \
} while (0)

static rclc_support_t support;
static rcl_allocator_t allocator;
static rcl_node_t node;
static rcl_subscription_t cmd_vel_sub;
static geometry_msgs__msg__Twist cmd_vel_msg;
static rclc_executor_t executor;

static void on_rcl_error(rcl_ret_t rc, int line)
{
  (void)rc; (void)line;
  /* Latch a fault LED / watchdog-safe halt instead of silently continuing with
   * an uninitialized node -- never ignore an RCL init failure on an embedded target. */
  while (1) { HAL_GPIO_TogglePin(FAULT_LED_GPIO_Port, FAULT_LED_Pin); HAL_Delay(100); }
}

static void cmd_vel_callback(const void * msgin)
{
  const geometry_msgs__msg__Twist * msg = (const geometry_msgs__msg__Twist *)msgin;
  apply_wheel_setpoints(msg->linear.x, msg->angular.z);
}

void micro_ros_main(void)
{
  allocator = rcl_get_default_allocator();

  rmw_uros_set_custom_transport(
    true, NULL,
    stm32_uart_open, stm32_uart_close,
    stm32_uart_write, stm32_uart_read);

  RCCHECK(rclc_support_init(&support, 0, NULL, &allocator));
  RCCHECK(rclc_node_init_default(&node, "stm32_motor_bridge", "", &support));

  RCCHECK(rclc_subscription_init_default(
    &cmd_vel_sub, &node,
    ROSIDL_GET_MSG_TYPE_SUPPORT(geometry_msgs, msg, Twist),
    "cmd_vel"));

  RCCHECK(rclc_executor_init(&executor, &support.context, 1, &allocator));
  RCCHECK(rclc_executor_add_subscription(
    &executor, &cmd_vel_sub, &cmd_vel_msg, &cmd_vel_callback, ON_NEW_DATA));

  while (1) {
    rclc_executor_spin_some(&executor, RCL_MS_TO_NS(10));
    run_control_loop_step();   /* bounded, non-blocking, never a HAL_Delay() here */
    HAL_Delay(10);
  }
}
```

Rules:
- Wrap **every** `rcl_ret_t`-returning call in an `RCCHECK`-style macro. A silently ignored
  `RCL_RET_ERROR` on an embedded target leaves the node in an undefined half-initialized state
  with no stack trace to debug later — fail loud and fail fast (a latched fault indicator),
  never fail silent.
- Use `rclc_executor_spin_some()` in the main loop with a small bounded timeout, never a
  blocking `rclc_executor_spin()` that never returns — the loop needs to return control so
  `run_control_loop_step()` (motor PID, watchdog feed, sensor polling) still executes even if
  no message arrived.
- Size the executor's handle count (`rclc_executor_init(&executor, &support.context, N, ...)`)
  to exactly the number of subscriptions/timers/services you register — `micro_ros_utilities`
  allocates fixed-size static memory for these, so this must be known at compile time; there is
  no dynamic growth.
- Never call `malloc`/`free` from an interrupt context or from the UART RX ISR that feeds the
  transport layer — use the static memory profile (`colcon.meta` / `rmw_microxrcedds` config)
  micro-ROS ships for constrained targets.
- Feed an independent hardware watchdog (`IWDG`) from the main loop, not from inside the
  executor callback — if XRCE-DDS communication stalls, the robot must still be able to detect
  a hung control loop and reset.

## 4. UART transport layer

```c
#include "usart.h"   /* CubeMX-generated huart2 etc. */

bool stm32_uart_open(struct uxrCustomTransport * transport)
{
  (void)transport;
  return true;   /* UART already initialized by MX_USART2_UART_Init() in main() */
}

bool stm32_uart_close(struct uxrCustomTransport * transport)
{
  (void)transport;
  return true;
}

size_t stm32_uart_write(struct uxrCustomTransport * transport,
                         const uint8_t * buf, size_t len, uint8_t * err)
{
  (void)transport;
  HAL_StatusTypeDef status = HAL_UART_Transmit(&huart2, (uint8_t *)buf, len, 100);
  *err = (status == HAL_OK) ? 0 : 1;
  return (status == HAL_OK) ? len : 0;
}

size_t stm32_uart_read(struct uxrCustomTransport * transport,
                        uint8_t * buf, size_t len, int timeout_ms, uint8_t * err)
{
  (void)transport;
  HAL_StatusTypeDef status = HAL_UART_Receive(&huart2, buf, len, timeout_ms);
  *err = (status == HAL_OK) ? 0 : 1;
  return (status == HAL_OK) ? len : 0;
}
```

Rules:
- The XRCE-DDS session layer already frames and checksums messages — the transport only needs
  reliable byte-in/byte-out semantics, never add extra framing here.
- Use bounded `HAL_UART_Transmit`/`Receive` timeouts, never `HAL_MAX_DELAY` — a stalled UART
  peripheral must not hang the whole executor loop indefinitely.
- Pick a baud rate that comfortably covers your message rate and size with margin (115200 is a
  reasonable default; raise it for high-rate sensor topics like IMU at >100 Hz).

## 5. CAN bus: bare-metal driver + Linux-side bridge

micro-ROS itself communicates over serial or UDP, not CAN — CAN is best used as the
low-latency bus between the MCU and motor controllers/sensors, bridged into ROS 2 on the Linux
side via SocketCAN, keeping the DDS session off the real-time bus entirely.

```c
/* Firmware side: fixed-format CAN frame for periodic telemetry (bxCAN/FDCAN HAL). */
typedef struct __attribute__((packed)) {
  int16_t wheel_velocity_mrad_s;   /* milli-rad/s, avoids float-in-CAN-payload ambiguity */
  uint16_t battery_mv;
  uint8_t fault_flags;
} MotorTelemetryFrame;

void send_motor_telemetry(int16_t velocity_mrad_s, uint16_t battery_mv, uint8_t faults)
{
  MotorTelemetryFrame frame = {velocity_mrad_s, battery_mv, faults};
  CAN_TxHeaderTypeDef header = {
    .StdId = CAN_ID_MOTOR_TELEMETRY,
    .IDE = CAN_ID_STD,
    .RTR = CAN_RTR_DATA,
    .DLC = sizeof(frame),
  };
  uint32_t mailbox;
  HAL_CAN_AddTxMessage(&hcan1, &header, (uint8_t *)&frame, &mailbox);
}
```

```python
# Linux side: a plain rclpy node bridging SocketCAN into a ROS 2 topic (python-can + rclpy).
# Keep this bridge node minimal and single-purpose -- it should only translate frames, not
# implement control logic, so the safety-critical loop stays entirely on the MCU.
import can
import rclpy
from rclpy.node import Node
from std_msgs.msg import Int32MultiArray

CAN_ID_MOTOR_TELEMETRY = 0x100


class CanBridgeNode(Node):
    def __init__(self) -> None:
        super().__init__('can_bridge_node')
        self._bus = can.interface.Bus(channel='can0', bustype='socketcan')
        self._pub = self.create_publisher(Int32MultiArray, 'motor_telemetry', 10)
        self._timer = self.create_timer(0.01, self._poll_can)

    def _poll_can(self) -> None:
        msg = self._bus.recv(timeout=0.0)
        if msg is not None and msg.arbitration_id == CAN_ID_MOTOR_TELEMETRY:
            out = Int32MultiArray()
            out.data = list(msg.data)
            self._pub.publish(out)
```

Rules:
- Use fixed-width packed structs for CAN payloads (max 8 bytes classic CAN, 64 bytes FDCAN) —
  never send floats without an explicit fixed-point convention documented next to the struct.
- Assign CAN IDs from a single header shared by firmware and any Linux-side decoder — never
  duplicate magic ID numbers in two places that can drift out of sync.
- Keep safety-critical control decisions (e-stop, current limiting) on the MCU side of the CAN
  bus; the Linux-side bridge should be a thin, replaceable translation layer.

## 6. LVGL GUI on an embedded display

```c
#include "lvgl.h"

static lv_disp_draw_buf_t draw_buf;
static lv_color_t buf1[MY_DISP_HOR_RES * 10];

static void disp_flush_cb(lv_disp_drv_t * drv, const lv_area_t * area, lv_color_t * color_p)
{
  /* Use DMA for the actual transfer; never block here with a busy-wait longer than one
   * refresh period, or LVGL's internal timing budget is violated. */
  st7789_draw_bitmap_dma(area->x1, area->y1, area->x2, area->y2, (uint16_t *)color_p);
}

static void dma_transfer_complete_callback(void)
{
  lv_disp_flush_ready(&disp_drv);   /* called from DMA ISR when the transfer finishes */
}

void lvgl_display_init(void)
{
  lv_init();
  lv_disp_draw_buf_init(&draw_buf, buf1, NULL, MY_DISP_HOR_RES * 10);

  static lv_disp_drv_t disp_drv;
  lv_disp_drv_init(&disp_drv);
  disp_drv.hor_res = MY_DISP_HOR_RES;
  disp_drv.ver_res = MY_DISP_VER_RES;
  disp_drv.flush_cb = disp_flush_cb;
  disp_drv.draw_buf = &draw_buf;
  lv_disp_drv_register(&disp_drv);
}

/* Called from the periodic loop (a 5-10 ms tick, e.g. a low-priority FreeRTOS task or a
 * SysTick-driven counter in the bare-metal main loop). */
void lvgl_tick_and_handle(void)
{
  lv_tick_inc(5);
  lv_timer_handler();
}
```

Bridging live robot state into the GUI safely:

```c
/* Shared state written by the micro-ROS subscription callback, read by the GUI loop.
 * Use a critical section or lightweight spinlock, not a full RTOS mutex, if the writer is an
 * ISR/executor callback and the reader is a bounded polling loop -- keep the critical
 * section as short as a struct copy, never hold it across an LVGL draw call. */
typedef struct { float battery_voltage; uint8_t fault_flags; } RobotStateSnapshot;
static volatile RobotStateSnapshot g_robot_state;

static void battery_topic_callback(const void * msgin)
{
  const std_msgs__msg__Float32 * msg = (const std_msgs__msg__Float32 *)msgin;
  __disable_irq();
  g_robot_state.battery_voltage = msg->data;
  __enable_irq();
}

void lvgl_update_battery_label(lv_obj_t * label)
{
  RobotStateSnapshot snapshot;
  __disable_irq();
  snapshot = g_robot_state;
  __enable_irq();
  lv_label_set_text_fmt(label, "%.2f V", (double)snapshot.battery_voltage);
}
```

Rules:
- Drive the physical panel refresh with DMA and signal `lv_disp_flush_ready()` from the DMA
  completion ISR — a blocking SPI/parallel write in `flush_cb` will visibly stall the whole UI
  and starve the micro-ROS executor if they share a core.
- Call `lv_timer_handler()` on a steady period (5-30 ms depending on `LV_DISP_DEF_REFR_PERIOD`
  in `lv_conf.h`) from a single consistent context — never from both the main loop and an ISR.
- Cross-thread/ISR state shared between a micro-ROS callback and the LVGL loop must go through
  a minimal critical section (disable IRQ / RTOS mutex) around a small struct copy, never a
  raw unguarded global read/write — a torn read of a multi-byte value is a real, if rare,
  correctness bug on Cortex-M.