embedded-systems

name	embedded-systems
description	Use this skill when writing or reviewing Python code for Raspberry Pi 5 edge devices -- UART hardware drivers, camera streaming via picamera2, GPIO control, systemd services, or resource monitoring. Don't use it for MAVLink protocol parsing (use mavlink-integration), web portal server code, or Helm work.
version	1.0.0
owner	swarmery-core
allowed-tools	Read, Edit, Bash, Grep, Glob
docs	{"status":"reviewed","source_sha":"36398d0f597c","updated":"2026-08-06T00:00:00.000Z"}

Purpose

Produce and review Python code for Raspberry Pi 5 edge devices running the edge service. Covers UART serial communication, picamera2 camera streaming, GPIO control, resource monitoring, and systemd service configuration. All generated code passes mypy, handles MOCK_MODE for CI, and releases hardware resources on exit. For MAVLink protocol parsing (message IDs, dialects, command sequences), compose with mavlink-integration.

Success criteria: generated code imports without hardware attached (MOCK_MODE=true), passes mypy with zero errors, and releases every acquired resource in a finally block.

When to use

• Writing or reviewing Python code that interfaces with RPi5 hardware (UART, GPIO, camera)
• Configuring systemd services for the edge service on edge devices
• Debugging hardware communication issues (serial port, camera initialization)
• Implementing resource monitoring (CPU temperature, memory, throttling) on edge devices

When NOT to use

• Web portal server-side TypeScript code (use api-integration or code-standards)
• Infrastructure or Helm chart work (use the project's deployment workflow)
• MAVLink message parsing, dialect selection, or command sequencing (use mavlink-integration)
• Docker image building for the edge service (use docker-build)
• Python dependency auditing (use deps-check)
• Any Python code in the edge service that is purely MAVLink protocol without hardware driver concerns (use mavlink-integration)

Required environment

• Runtime: .claude/skills/embedded-systems/SKILL.md
• Tools/libraries: Python 3.11+, serial_asyncio, picamera2, psutil, asyncio
• Hardware: Raspberry Pi 5 (or MOCK_MODE=true for CI/dev without hardware)
• Code standards: mypy for type checking, black (line length 100), isort, flake8 (max complexity 10)

Inputs

• component: "uart" | "camera" | "gpio" | "systemd" | "monitoring" -- which hardware subsystem to work with
• mock_mode: boolean -- if true, generate code that works without physical hardware (for CI testing)

Outputs

• Format: Python code with type hints, specific exception handling, and MOCK_MODE support. All code passes mypy without errors.
• Length budget: max 120 lines per class; max 40 lines per standalone function; systemd unit files under 25 lines.

Procedure

1. Check mock mode -- Determine if the code should support running without hardware. If MOCK_MODE is set in the environment, all hardware interfaces return simulated data.

   import os
   MOCK_MODE = os.environ.get("MOCK_MODE", "false").lower() == "true"
   

   Checkpoint: MOCK_MODE check is present at module level. The check reads from the environment at runtime, never hardcoded to a literal "false".

2. Import required modules -- Always include explicit imports. Never assume modules are available in scope. If using serial.EIGHTBITS or serial.SerialException, you must import serial separately -- serial_asyncio does not bring serial constants into scope.

   import asyncio
   import logging
   import serial          # required for serial.EIGHTBITS, serial.SerialException
   import serial_asyncio
   from typing import Optional
   
   logger = logging.getLogger(__name__)
   

   Checkpoint: logger and all type imports are explicitly defined. import serial is present if any serial.* constant or exception is referenced.

3. Implement with specific exceptions -- Use specific exception types, not bare except Exception. Hardware code handles known failure modes:

   • UART: serial.SerialException, OSError, asyncio.TimeoutError
   • Camera: RuntimeError (picamera2 init failure), OSError (device not found)
   • GPIO: PermissionError, FileNotFoundError (sysfs access)

   Checkpoint: Every except block catches a specific exception type.

4. -- Hardware resources (serial ports, camera) are released on exit.

Self-check

- [ ] All Python code has type hints on parameters and return types - [ ] `logger` is explicitly created via `logging.getLogger(__name__)`, not used as an undefined name - [ ] `serial` module is imported before referencing `serial.EIGHTBITS`, `serial.SerialException`, etc. -- `serial_asyncio` does NOT bring these into scope - [ ] Every `except` block catches a specific exception, not bare `Exception` - [ ] Every streaming loop has an exit condition (e.g., `stop_event.is_set()`) - [ ] Hardware resources are released in `finally` blocks - [ ] `MOCK_MODE` support is present for CI compatibility -- the check reads `os.environ`, never hardcoded - [ ] No hardcoded paths like `/home/pi/` -- use environment variables or systemd specifiers - [ ] `mypy` would pass on the generated code - [ ] Systemd service templates use `MOCK_MODE=${MOCK_MODE:-false}`, not a hardcoded literal - [ ] Before/after diff was shown for every Edit applied to existing files

Common mistakes

• DO NOT use bare except Exception -- always catch specific exceptions (serial.SerialException, OSError, RuntimeError)
• DO NOT reference logger without defining it -- always add logger = logging.getLogger(__name__) at module level
• DO NOT reference serial.EIGHTBITS without import serial -- the serial_asyncio import does not bring serial constants into scope
• DO NOT write infinite loops without exit conditions -- always use while not stop_event.is_set() or a cancellation token
• DO NOT hardcode /home/pi/ in systemd service files -- use %h (home directory specifier) or environment variables; the deploy user may be ubuntu (as on the staging environment)
• DO NOT use raspi-config menu instructions for camera enable -- on Raspberry Pi OS Bookworm (2024+), use dtoverlay=camera_auto_detect in /boot/firmware/config.txt
• DO NOT call self.camera.capture_file() without checking self.camera is not None first
• DO NOT open UART without a corresponding finally block to release the port
• DO NOT hardcode MOCK_MODE=false in systemd service templates -- use MOCK_MODE=${MOCK_MODE:-false} so environment-specific overrides work via drop-in files

Escalation

• Stop and ask when: The target RPi model is not RPi5 (GPIO pinout and UART assignments differ)
• Stop and ask when: Multiple services need the same UART port (serial port contention)
• Stop and ask when: Camera resolution exceeds 1920x1080 at 30fps (may exceed RPi5 ISP throughput)
• Stop and ask when: The code must interact with MAVLink messages (hand off to mavlink-integration)

What to surface

• Whether MOCK_MODE is being used (affects test validity)
• Hardware-specific side effects: UART monopolizes the serial port, camera locks the CSI interface
• The systemd service user and working directory (must match the target device)
• Any asyncio.to_thread() calls wrapping blocking I/O (pymavlink, picamera2 capture)

Examples

**Scenario**: Implement a UART reader for serial communication on the edge service

import asyncio
import logging
import os
import serial
import serial_asyncio
from typing import Optional

logger = logging.getLogger(__name__)

MOCK_MODE = os.environ.get("MOCK_MODE", "false").lower() == "true"


class UARTReader:
    """Async UART reader for serial communication with flight controller."""

    def __init__(self, port: str = "/dev/ttyAMA0", baudrate: int = 57600) -> None:
        self.port = port
        self.baudrate = baudrate
        self.reader: Optional[asyncio.StreamReader] = None
        self.writer: Optional[asyncio.StreamWriter] = None

    async def connect(self) -> bool:
        """Establish UART connection. Returns True on success."""
        if MOCK_MODE:
            logger.info("MOCK_MODE: Simulating UART connection on %s", self.port)
            return True

        try:
            self.reader, self.writer = await serial_asyncio.open_serial_connection(
                url=self.port,
                baudrate=.baudrate,
                bytesize=serial.EIGHTBITS,
                parity=serial.PARITY_NONE,
                stopbits=serial.STOPBITS_ONE,
            )
            logger.info(, .port)
             
         serial.SerialException  err:
            logger.error(, .port, err)
             
         OSError  err:
            logger.error(, .port, err)
             

      () -> []:
        
         MOCK_MODE:
              * size

         .reader  :
             

        :
            data =  asyncio.wait_for(.reader.read(size), timeout=timeout)
             data
         asyncio.TimeoutError:
            logger.debug(, timeout)
             
         serial.SerialException  err:
            logger.error(, err)
             
         OSError  err:
            logger.error(, err)
             

      () -> :
        
         .writer   :
            :
                .writer.close()
                 .writer.wait_closed()
             OSError  err:
                logger.warning(, err)
            :
                .writer = 
                .reader = 
        logger.info(, .port)

**Scenario**: Camera streaming with exit condition and cleanup

import asyncio
import logging
import os
from io import BytesIO
from typing import Awaitable, Callable, Optional

logger = logging.getLogger(__name__)

MOCK_MODE = os.environ.get("MOCK_MODE", "false").lower() == "true"


class CameraStreamer:
    """Async camera streamer using picamera2."""

    def __init__(self, width: int = 640, height: int = 480, fps: int = 30) -> None:
        self.width = width
        self.height = height
        self.fps = fps
        self._camera: Optional[object] = None  # Picamera2 instance

    async def initialize(self) -> bool:
        """Initialize camera. Returns True on success."""
        if MOCK_MODE:
            logger.info("MOCK_MODE: Simulating camera initialization")
            return True

        try:
            from picamera2 import Picamera2

            self._camera = Picamera2()
            config = ._camera.create_still_configuration(
                main={: (.width, .height)}
            )
            ._camera.configure(config)
            ._camera.start()
            logger.info(, .width, .height, .fps)
             
         RuntimeError  err:
            logger.error(, err)
             
         OSError  err:
            logger.error(, err)
             

      () -> []:
        
         MOCK_MODE:
              +  *  + 

         ._camera  :
            logger.warning()
             

        :
            buffer = BytesIO()
             asyncio.to_thread(._camera.capture_file, buffer, =)
             buffer.getvalue()
         RuntimeError  err:
            logger.error(, err)
             

      () -> :
        
        logger.info()
        :
              stop_event.is_set():
                jpeg_data =  .capture_jpeg()
                 jpeg_data:
                     callback(jpeg_data)
                 asyncio.sleep( / .fps)
        :
            logger.info()

      () -> :
        
         ._camera   :
            :
                ._camera.stop()
                ._camera.close()
             RuntimeError  err:
                logger.warning(, err)
            :
                ._camera = 
        logger.info()

**Scenario**: Systemd service template (parameterized, 12-factor compliant)

# /etc/systemd/system/<device>@.service
[Unit]
Description=Edge Control Box Service
After=network.target

[Service]
Type=simple
User=%i
WorkingDirectory=%h/<device>
ExecStart=%h/<device>/venv/bin/python src/send_data.py
Restart=always
RestartSec=10
Environment="PYTHONUNBUFFERED=1"
Environment="LOG_LEVEL=INFO"
Environment="MOCK_MODE=${MOCK_MODE:-false}"

[Install]
WantedBy=multi-user.target

Usage: sudo systemctl enable <device>@ubuntu (where ubuntu is the deploy user).

To override MOCK_MODE for a specific device, create a drop-in:

sudo systemctl edit <device>@ubuntu
# Add: [Service]
# Add: Environment="MOCK_MODE=true"

Failure modes

Mode	Symptom	Detection	Fix
UART permission denied	PermissionError on /dev/ttyAMA0	Exception caught in connect()	Add user to dialout group (sudo usermod -a -G dialout $USER), then log out and back in
Camera not detected	RuntimeError during Picamera2() init	Exception caught in initialize()	Verify dtoverlay=camera_auto_detect in /boot/firmware/config.txt and reboot; check CSI cable connection
CPU thermal throttling	Performance degradation, CPU temp > 80C	get_cpu_temperature() returns > 80.0	Add heatsink, reduce camera resolution, or reduce telemetry polling rate
Serial port contention	serial.SerialException with "device busy"	Another process holds /dev/ttyAMA0	Check for other UART consumers (lsof /dev/ttyAMA0), stop conflicting service

Related skills

• mavlink-integration -- compose for MAVLink protocol handling; embedded-systems provides the UART transport layer, mavlink-integration provides message parsing and command sequences
• docker-build -- defer for creating the edge service Docker image (ARM64)
• code-standards -- code-standards defines Python style rules (black, isort, flake8, mypy) that apply to all edge service code
• api-integration -- the edge service WebSocket server (port 8081) that the web portal connects to is documented in api-integration

How to use

What it does

This skill writes and reviews Python that talks to real hardware on a single-board edge computer — serial ports, camera capture, GPIO lines, resource monitoring, and the service unit that keeps it running. It bakes in the habits that hardware code fails without: explicit imports, specific exception types, cleanup in finally, loops that can be stopped, and a mock path so the module still imports on a CI box with no hardware attached.

When to use it

• You are writing or reviewing Python that opens a serial port, drives GPIO pins, or captures frames from a CSI camera.
• You need a service unit file for the edge service, parameterized by user and home directory rather than hardcoded paths.
• A hardware call is failing — the port is busy, the camera never initializes, the process holds a device after exit.
• You are adding CPU temperature, memory, or throttling checks to a device agent.

When not to use it

• Parsing MAVLink messages, choosing dialects, or sequencing commands — use mavlink-integration; this skill only supplies the transport underneath it.
• Server-side TypeScript for a web portal — use api-integration or code-standards.
• Building the container image for the edge service — use docker-build.
• Auditing Python dependencies — use deps-check.

How to invoke

Skill(skill: "uav-pack:embedded-systems")

Invoke it before you write the code, not after. Tell it which subsystem you are on — uart, camera, gpio, systemd, or monitoring — and whether the result must run without hardware.

Inputs

• component — which subsystem: uart, camera, gpio, systemd, or monitoring — required.
• mock_mode — whether the generated code must run with no hardware attached, for CI — optional, defaults to supporting it.

What you get back

Typed Python with a module-level logger, a runtime MOCK_MODE check, specific except clauses, and every acquired resource released in a finally block. Classes stay under 120 lines, standalone functions under 40, service unit files under 25. Before any edit to an existing file you get a before/after diff and a note on which hardware subsystem it touches.

Worked example

Skill(skill: "uav-pack:embedded-systems")
"Add a camera streamer to the edge service that pushes JPEG frames to a callback."

You get an async class with initialize, capture_jpeg, a stream_loop that runs while not stop_event.is_set(), and a cleanup that stops and closes the device. Blocking capture is wrapped in asyncio.to_thread. With MOCK_MODE=true the module imports and returns a stub JPEG, so the test suite passes on a machine with no camera.

Related

• mavlink-integration — compose with it when the bytes on the wire are MAVLink frames rather than raw serial.
• code-standards — the underlying style rules (formatting, import order, lint, type checking) for all Python here.
• api-integration — for the socket server the portal connects to, rather than the device-side driver.