mavlink-integration

name	mavlink-integration
description	Use this skill when a task involves MAVLink protocol communication via pymavlink, UART/UDP/TCP drone connections, ArduPilot SITL simulation, or telemetry message parsing in the edge service. Don't use it for raw UART hardware driver code without MAVLink framing (use embedded-systems), WebSocket transport to the web portal (use api-integration), or Docker builds (use docker-build).
version	1.0.0
owner	swarmery-core
allowed-tools	Read, Bash, Grep, Glob
docs	{"status":"reviewed","source_sha":"a12b075c166f","updated":"2026-08-06T00:00:00.000Z"}

Purpose

Produce MAVLink protocol code for drone communication in the edge service (project.json → device; Python 3.11+ on Raspberry Pi 5). Covers pymavlink connection types (UART, UDP, TCP), message parsing for common telemetry messages, async I/O patterns with asyncio.to_thread, ArduPilot SITL simulation setup, and MOCK_MODE for CI testing without hardware. For the WebSocket bridge from the edge service to the web portal, defer to api-integration. For Helm deployment of edge service pods, defer to the project's deployment workflow.

Success criteria: generated code connects (or simulates connection in MOCK_MODE), sends GCS heartbeats at 1Hz, reads telemetry via recv_match, handles asyncio.CancelledError for graceful shutdown, and releases the connection on stop.

When to use

• Connecting to a drone via pymavlink (UART serial, UDP simulator, TCP)
• Parsing MAVLink telemetry messages (HEARTBEAT, GLOBAL_POSITION_INT, ATTITUDE, GPS_RAW_INT)
• Implementing async MAVLink reader/writer patterns in the edge service
• Setting up ArduPilot SITL for local development or CI testing
• Configuring MOCK_MODE for testing without physical hardware

When NOT to use

• Raw UART hardware driver code without MAVLink framing (use embedded-systems)
• WebSocket streaming from the edge service to the web portal (use api-integration)
• Deploying the edge service via Helm/k3s (use the project's deployment workflow)
• Building Docker images for the edge service (use docker-build)
• Debugging Kubernetes pod health for the edge service (use troubleshooting)

Required environment

• Runtime: .claude/skills/mavlink-integration/SKILL.md
• Language: Python 3.11+
• Libraries: pymavlink, asyncio, serial (for serial.SerialException)
• Repo: the edge service repo (project.json → device)
• MAVLink dialect: ardupilotmega (superset of common)
• Hardware: Raspberry Pi 5 with UART (/dev/ttyAMA0 default, configurable via MAVLINK_CONNECTION env var)
• Simulator: ArduPilot SITL (UDP 127.0.0.1:14550)
• CI testing: MOCK_MODE=true env var disables UART connection

Inputs

• connection_type: enum -- one of: uart, udp, tcp
• connection_string: string -- pymavlink connection string (e.g., serial:/dev/ttyAMA0:57600, udp:127.0.0.1:14550)
• message_types: string[] -- MAVLink message types to read (e.g., HEARTBEAT, GLOBAL_POSITION_INT)

Outputs

• Format: Python code using pymavlink with async patterns, or SITL setup commands
• Length budget: max 100 lines for a connection manager class; max 30 lines for a telemetry parsing snippet
• All code follows edge service standards: type hints, asyncio.to_thread for blocking calls, specific exception handling

Procedure

1. Determine connection type -- UART for real hardware, UDP for SITL, TCP for remote. Checkpoint: Connection string format matches pymavlink expectations (e.g., serial:/dev/ttyAMA0:57600 for UART, udp:127.0.0.1:14550 for SITL).

2. Check MOCK_MODE -- If MOCK_MODE=true, skip real connection and return synthetic telemetry. Checkpoint: Env var checked before attempting hardware connection.

3. Establish connection -- Use mavutil.mavlink_connection() wrapped in asyncio.to_thread. Checkpoint: Connection object is not None and recv_match(type='HEARTBEAT') succeeds.

4. Start heartbeat sender -- Send GCS heartbeats at 1Hz minimum to maintain connection. Guard against double-start by checking if the heartbeat task already exists. Checkpoint: Heartbeat task running with asyncio.CancelledError handling.

5. Read telemetry -- Use recv_match with specific message types and timeouts. Checkpoint: Message data validated before use.

Self-check

- [ ] I used `GLOBAL_POSITION_INT` (message #33) or `GPS_RAW_INT` (message #24), not the non-existent `GPS_POSITION` - [ ] Async code wraps all pymavlink blocking calls in `asyncio.to_thread` - [ ] The heartbeat loop handles `asyncio.CancelledError` for graceful shutdown - [ ] Exception handling uses specific exceptions (`serial.SerialException`, `OSError`, `ConnectionRefusedError`), not bare `except Exception` - [ ] `import serial` is present whenever `serial.SerialException` is referenced -- pymavlink does not bring `serial` into scope - [ ] Connection string is read from `MAVLINK_CONNECTION` env var, not hardcoded - [ ] I mentioned `MOCK_MODE=true` for CI/testing contexts - [ ] I specified the MAVLink dialect (`ardupilotmega`) when referencing non-common messages - [ ] The `stop()` method cancels the heartbeat task AND closes the MAVLink connection

Common mistakes

• DO NOT use GPS_POSITION as a message name -- it does not exist in MAVLink; use GLOBAL_POSITION_INT (#33) for fused position or GPS_RAW_INT (#24) for raw GPS

  # WRONG -- GPS_POSITION does not exist, recv_match returns None forever
  msg = await reader.read_message("GPS_POSITION", timeout=1.0)
  
  # CORRECT -- use GLOBAL_POSITION_INT (#33) for fused lat/lon/alt
  msg = await reader.read_message("GLOBAL_POSITION_INT", timeout=1.0)
  

• DO NOT reference serial.SerialException without import serial -- pymavlink does not bring serial into scope; missing this causes NameError at runtime

  # WRONG -- NameError: name 'serial' is not defined
  from pymavlink import mavutil
  # ... later ...
  except (OSError, serial.SerialException) as err:
  
  # CORRECT -- import serial explicitly
  import serial
  from pymavlink import mavutil
  # ... later ...
  except (OSError, serial.SerialException) as err:
  

• DO NOT hardcode /dev/ttyAMA0 -- read from MAVLINK_CONNECTION env var; UART device paths vary across RPi models

• DO NOT use bare except Exception -- catch serial.SerialException and OSError for connection errors, ConnectionRefusedError for SITL UDP

• DO NOT run an infinite while True loop without asyncio.CancelledError handling -- the task will leak on shutdown and trigger Python 3.11+ warnings

• DO NOT test against real hardware in CI -- set MOCK_MODE=true to disable UART and return synthetic telemetry

Escalation

• Stop and ask when: UART device path is not /dev/ttyAMA0 and the user has not specified the correct path
• Stop and ask when: SITL is unreachable after setup and the ArduPilot firmware version is unknown
• Stop and ask when: The user needs to send flight commands (arm, takeoff, waypoint) to a real drone -- this is a safety-critical operation requiring explicit operator confirmation

What to surface

• Which connection type and string are being used
• Whether MOCK_MODE is active
• The MAVLink dialect assumed (common vs ardupilotmega)
• Any baud rate or device path configuration that may need adjustment
• SITL connection status and which messages are being received

Examples

## Async MAVLink reader with proper error handling

import asyncio
import logging
import os
import serial
from typing import Any, Optional

from pymavlink import mavutil

logger = logging.getLogger(__name__)

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


class MAVLinkReader:
    """Async MAVLink connection manager for the edge service."""

    def __init__(self, connection_string: Optional[str] = None) -> None:
        self.connection_string = connection_string or os.environ.get(
            "MAVLINK_CONNECTION", "serial:/dev/ttyAMA0:57600"
        )
        self.connection: Optional[Any] = None
        self._heartbeat_task: Optional[asyncio.Task[None]] = None

    async def connect(self) -> bool:
        """Establish MAVLink connection."""
        if MOCK_MODE:
            logger.info("MOCK_MODE active -- skipping real MAVLink connection")
            return True

        try:
            self.connection =  asyncio.to_thread(
                mavutil.mavlink_connection,
                .connection_string,
                source_system=,
                source_component=,
            )
            
            msg =  asyncio.to_thread(
                .connection.recv_match,
                =,
                blocking=,
                timeout=,
            )
             msg  :
                logger.error()
                 
            logger.info(, msg.get_srcSystem())
             
         serial.SerialException  err:
            logger.error(, err)
             
         OSError  err:
            logger.error(, err)
             

      () -> []:
        
         MOCK_MODE:
             
          .connection:
             
        :
              asyncio.to_thread(
                .connection.recv_match,
                =msg_type,
                blocking=,
                timeout=timeout,
            )
         OSError  err:
            logger.error(, msg_type, err)
             

      () -> :
        
        :
             :
                 .connection:
                     asyncio.to_thread(
                        .connection.mav.heartbeat_send,
                        mavutil.mavlink.MAV_TYPE_GCS,
                        mavutil.mavlink.MAV_AUTOPILOT_INVALID,
                        ,
                        ,
                        mavutil.mavlink.MAV_STATE_ACTIVE,
                    )
                 asyncio.sleep()
         asyncio.CancelledError:
            logger.info()
            

      () -> :
        
           .connect():
             
         ._heartbeat_task   :
            logger.warning()
             
        ._heartbeat_task = asyncio.create_task(.send_heartbeat_loop())
         

      () -> :
        
         ._heartbeat_task:
            ._heartbeat_task.cancel()
             ._heartbeat_task
            ._heartbeat_task = 
         .connection:
             asyncio.to_thread(.connection.close)
            .connection = 

## Reading GPS telemetry (GLOBAL_POSITION_INT #33)

msg = await reader.read_message("GLOBAL_POSITION_INT", timeout=1.0)
if msg:
    telemetry = {
        "lat": msg.lat / 1e7,           # degrees
        "lon": msg.lon / 1e7,           # degrees
        "alt": msg.alt / 1000,          # meters above MSL
        "relative_alt": msg.relative_alt / 1000,  # meters above ground
        "heading": msg.hdg / 100,       # degrees
        "vx": msg.vx / 100,            # m/s north
        "vy": msg.vy / 100,            # m/s east
        "vz": msg.vz / 100,            # m/s down
    }

For raw GPS data use GPS_RAW_INT (#24) instead -- it provides fix_type, satellites_visible, eph (HDOP), and epv (VDOP).

## ArduPilot SITL setup

# Clone and build (one-time)
git clone https://github.com/ArduPilot/ardupilot.git
cd ardupilot
git submodule update --init --recursive
Tools/environment_install/install-prereqs-ubuntu.sh -y

# Run SITL (listens on UDP:127.0.0.1:14550)
cd ArduCopter
sim_vehicle.py -v ArduCopter --console --map

# Connect from Python
# MAVLINK_CONNECTION=udp:127.0.0.1:14550 python your_script.py

## Testing without hardware (MOCK_MODE)

Set MOCK_MODE=true in CI environments to skip UART connections:

# Kubernetes values or CI env
env:
  - name: MOCK_MODE
    value: "true"
  - name: LOG_LEVEL
    value: "DEBUG"

In code, check MOCK_MODE before attempting hardware access (see connect() method above). Return synthetic telemetry data for testing.

Failure modes

Mode	Symptom	Detection	Fix
Wrong message name GPS_POSITION	recv_match returns None forever	Message name is not in MAVLink common/ardupilotmega dialect	Use GLOBAL_POSITION_INT (#33) or GPS_RAW_INT (#24)
UART baud rate mismatch	Connection succeeds but no messages received	Serial monitor shows garbled data	Verify baud rate matches ArduPilot SERIAL1_BAUD parameter (typically 57600 or 115200)
Heartbeat task leak on shutdown	RuntimeWarning: coroutine was never awaited or task pending at interpreter shutdown	No CancelledError handler in heartbeat loop	Add try/except asyncio.CancelledError: return and cancel the task in stop()
SITL connection refused	ConnectionRefusedError on UDP connect	ArduPilot SITL is not running	Start SITL with sim_vehicle.py, then retry connection
Missing import serial	NameError: name 'serial' is not defined at runtime	serial.SerialException referenced without import	Add import serial to import block

Related skills

• api-integration -- defer for the WebSocket bridge from the edge service to the web portal; compose when MAVLink data needs to reach the browser via SSE
• The project's deployment workflow -- defer for deploying edge service pods to k3s on Raspberry Pi; MAVLINK_CONNECTION and MOCK_MODE env vars are set in Helm values there
• embedded-systems -- defer for RPi hardware config (GPIO, UART enable, camera); compose when UART device path needs to be determined
• code-standards -- follow its Python section for type hints, specific exceptions, and MOCK_MODE patterns

How to use

What it does

This skill produces MAVLink code for talking to a drone from an edge service in Python. It covers pymavlink connection strings for serial, UDP, and TCP, async patterns that wrap blocking pymavlink calls in asyncio.to_thread, telemetry message parsing, ArduPilot SITL setup, and a mock mode so tests run without hardware. It also encodes the mistakes that bite here — wrong message names, missing imports, leaked heartbeat tasks.

When to use it

• You are opening a pymavlink connection over UART, UDP to a simulator, or TCP.
• You are parsing telemetry — HEARTBEAT, GLOBAL_POSITION_INT, ATTITUDE, GPS_RAW_INT.
• You are writing an async MAVLink reader or writer with a heartbeat loop and clean shutdown.
• You are setting up ArduPilot SITL locally, or need a hardware-free mode for CI.

When not to use it

• Raw serial driver code with no MAVLink framing — use embedded-systems.
• Streaming telemetry onward to a browser over WebSocket or SSE — use api-integration.
• Building or pushing container images for the service — use docker-build.
• Diagnosing a failing pod or deployment rather than the protocol — use troubleshooting.

How to invoke

Skill(skill: "uav-pack:mavlink-integration")

Invoke it before writing the MAVLink code, then state the connection type, the connection string, and whether mock mode is on.

Inputs

• connection_type — one of uart, udp, tcp — required.
• connection_string — a pymavlink string such as udp:127.0.0.1:14550 — required.
• message_types — the MAVLink message names you want to read — optional; defaults to a heartbeat-only connection.

What you get back

Python code with type hints and specific exception handling: a connection-manager class of up to about 100 lines, or a telemetry-parsing snippet of up to about 30, or SITL setup commands. The code reads its connection string from an environment variable rather than hard-coding a device path, sends heartbeats at 1 Hz, handles asyncio.CancelledError, and closes the connection on stop. You also get a short note on the dialect assumed and any baud rate or device path you may need to adjust.

Worked example

Skill(skill: "uav-pack:mavlink-integration")

You ask: "Read GPS position from SITL and push it into the telemetry queue."

You get back: a reader that connects to udp:127.0.0.1:14550, waits for the
first HEARTBEAT to confirm the link, then reads GLOBAL_POSITION_INT (#33) and
scales the fields — lat and lon by 1e7, alt by 1000 — into a plain dict.
The skill flags that GPS_POSITION is not a real message name, so recv_match
on it would return None forever.

Related

• embedded-systems — reach for it when the question is the hardware itself: enabling UART, GPIO, or finding the right device path.
• api-integration — reach for it when the MAVLink data has to leave the device and reach a browser.
• code-standards — its Python section is the style the generated code follows.