mavpilot

🇷🇺 Русская версия

Async PX4 drone controller for Python — sequential autonomous flight via MAVLink, with built-in live 3D visualization and a hardware-free mock mode.

Contents

• Features
• Installation
• Quick start — mock mode
• Library usage
• CLI reference
• API reference
• Coordinate system
• Visualization
• Architecture
• Connecting to real hardware
• Development
• License

Features

Installation

pip install mavpilot

Or from source:

git clone https://github.com/Onikore/mavpilot
cd mavpilot
pip install -e ".[dev]"

Runtime dependency: pymavlink (installed automatically).

Quick start — mock mode

No drone or SITL needed:

# Square flight pattern
python -m mavpilot --mock

# Star/pentagram flight pattern
python -m mavpilot --mock --pattern star

# Precision landing demo
python -m mavpilot --mock --precision-land

Open http://localhost:8765 in a browser to watch the live 3D visualization.

Library usage

import asyncio
from mavpilot import DroneController

async def mission():
    # `async with` connects on entry and tears down on exit (aclose());
    # if the block exits via an exception while still armed, it triggers
    # emergency_land() first.
    async with DroneController(
        connection_string="udp:127.0.0.1:14540",  # SITL default
        enable_viz=True,   # browser viz on :8765
    ) as drone:
        await drone.apply_safe_params()  # recommended PX4 safety params
        await drone.wait_until_ready()   # wait for EKF / LOCAL_POSITION_NED

        await drone.takeoff(altitude_m=5.0)

        # NED coordinates (x=North, y=East, z=Down)
        await drone.goto(x=10, y=0, z=-5)
        await drone.goto(x=10, y=10, z=-5, yaw_deg=90)
        await drone.goto_body_relative(forward_m=5, right_m=0, down_m=0)
        await drone.hover(duration_s=3.0)

        await drone.land()

asyncio.run(mission())

Precision landing

Supply a callback that returns the landing marker offset in body FRD frame:

from mavpilot import DroneController, MarkerObservation

def get_marker() -> MarkerObservation | None:
    # plug in your vision pipeline here
    # dx = forward offset (m), dy = right offset (m)
    return MarkerObservation(dx=0.3, dy=-0.1)

async def mission():
    async with DroneController(mock=True, enable_viz=False) as drone:
        await drone.takeoff(altitude_m=10.0)
        result = await drone.precision_land(
            get_marker_offset=get_marker,
            descent_rate_mps=0.3,
            final_altitude_m=0.5,
            horizontal_tolerance_m=0.15,
            min_altitude_floor_m=0.3,   # new in v0.2.0
        )
        if not result:
            # status ∈ {ABORTED_AT_FLOOR, MARKER_LOST, TIMEOUT}
            print(f"precision_land did not land cleanly: {result.status.value}")
            print(f"final position: {result.final_position}")

Converting camera pixels to body offset

from mavpilot.utils import pixel_to_body_offset

dx, dy = pixel_to_body_offset(
    px_norm_x=0.1,            # normalized [-1, 1]
    px_norm_y=-0.05,
    camera_hfov_deg=90.0,
    camera_vfov_deg=60.0,
    altitude_above_ground_m=drone.get_local_position().altitude,
    camera_mount_yaw_deg=0.0,
)

Live telemetry

Subscribe to ~10 Hz telemetry snapshots (works even with enable_viz=False):

async with DroneController(mock=True, enable_viz=False) as drone:
    # Callback style — returns an unsubscribe function:
    unsubscribe = drone.subscribe(lambda t: print(t["x"], t["y"], t["z"]))
    ...
    unsubscribe()

    # Or async-iterator style:
    async for t in drone.telemetry_stream():
        if t["landed"] == 1:
            break

CLI reference

python -m mavpilot [OPTIONS]

Options:
  --connection STR      MAVLink endpoint  [default: udp:127.0.0.1:14540]
  --mock                Hardware-free simulator mode
  --viz-port INT        Browser visualization port  [default: 8765]
  --viz-host STR        Interface the visualization server binds to  [default: 127.0.0.1]
                        Use 0.0.0.0 to expose on LAN (telemetry visible to everyone on the network)
  --no-viz              Disable browser visualization
  --precision-land      Use precision landing with a simulated marker
  --pattern {square,star}  Demo flight pattern  [default: square]
  --log-level {DEBUG,INFO,WARNING,ERROR}  Logging verbosity  [default: INFO]
  --loop-hz FLOAT       OFFBOARD setpoint publish rate in Hz  [default: 50.0]
  --watchdog-s FLOAT    Telemetry watchdog timeout in seconds  [default: 2.0]
  --version             Print the mavpilot version and exit

Error handling and Ctrl-C

• Ctrl-C at any point during a mission calls emergency_land(). This chains: AUTO_LAND mode switch, wait up to 10 s for touchdown, send MAV_CMD_NAV_LAND if mode switch is stuck, and as a last resort DO_FLIGHTTERMINATION (immediate motor cut — drone falls).
• RTL is not part of emergency_land(). Return-to-launch is a separate nominal operation (drone.return_to_launch()), not an emergency procedure.
• Any unhandled exception in the mission body (including KeyboardInterrupt) also triggers emergency_land().

Telemetry watchdog & protocol safety (v0.2.0)

• Telemetry watchdog — telemetry_watchdog_s (default 2 s). If no fresh LOCAL_POSITION_NED arrives within this window, the streamer latches a watchdog flag and the next mission call (takeoff/goto/set_yaw/land/return_to_launch/precision_land) raises DroneError. emergency_land() deliberately ignores the flag — it is the recovery path the watchdog is meant to trigger.
• EKF health gate — wait_until_ready() now also validates EKF AHRS health (SYS_STATUS bit 5), not just position freshness.
• send_command_long() — exposes the COMMAND_ACK Future API: it awaits the terminal ACK keyed by (cmd_id, target_sys, target_comp). IN_PROGRESS extends the deadline; a duplicate in-flight command, a timeout, or a non-ACCEPTED result each raise DroneError.
• get_yaw_deg() is normalized to [-180, 180].

API reference

The full, always-current reference is generated from docstrings and published at the GitHub Pages docs site — treat that as the source of truth. The tables below are a quick map.

DroneController(…)

DroneController(
    connection_string    = "udp:127.0.0.1:14540",
    source_system        = 255,
    source_component     = MAV_COMP_ID_MISSIONPLANNER,
    loop_hz              = 50.0,       # setpoint publish rate
    enable_viz           = True,       # start browser viz server
    viz_port             = 8765,
    viz_host             = "127.0.0.1", # 0.0.0.0 to expose on LAN
    mock                 = False,      # no-hardware simulator
    yaw_slew_rate_deg    = 15.0,       # max yaw rate (deg/s)
    telemetry_watchdog_s = 2.0,        # LOCAL_POSITION_NED silence → watchdog
)

Flight methods

Telemetry

Data classes

from mavpilot import Position, MarkerObservation

# NED position (x=North, y=East, z=Down)
pos: Position       # pos.altitude == -pos.z

# Marker offset in body FRD frame
obs: MarkerObservation  # dx=forward, dy=right, dz=down (optional)

Coordinate system

mavpilot uses the PX4 NED convention from LOCAL_POSITION_NED:

Coordinate transform utilities:

from mavpilot.utils import body_to_ned, ned_to_body, pixel_to_body_offset

Visualization

A lightweight self-contained HTTP+SSE server serves a Three.js 3D view with no build step and no external package manager. Open http://localhost:8765 while the drone is running.

The right-hand panel displays:

• Armed status and flight mode
• Live position, velocity, heading, battery
• Active setpoint
• Command log (takeoff, goto, land, …)
• PX4 STATUSTEXT messages

The UI is composed of native ES modules served from mavpilot/viz/static/ (index.html + styles.css + main.js/scene.js/sse.js/telemetry.js/log.js) — no bundler, but a modern browser with ES-module support is required. The max_clients parameter (default 32) caps concurrent SSE connections; excess clients receive HTTP 503.

Architecture

asyncio event loop  <-- your mission code
        |
        v
 DroneController
        |
        +-- heartbeat_thread   (1 Hz MAVLink heartbeat)
        +-- receiver_thread    (parses incoming MAVLink → self._tel)
        +-- streamer_thread    (publishes SET_POSITION_TARGET_LOCAL_NED @ 50 Hz)
        +-- viz_server         (optional HTTP+SSE → browser)

All shared state is protected by _tel_lock and _setpoint_lock. No asyncio primitives are needed in user mission code — the asyncio loop and background threads only touch shared dicts through these locks.

Module layout (v0.2.0)

mavpilot/
├── controller.py          # DroneController facade (composition root)
├── errors.py              # DroneError
├── types.py               # Position, MarkerObservation, PrecisionLand{Status,Result}
├── utils.py               # coordinate transforms, pinhole, yaw normalization
├── constants.py           # PX4 mode bits, MAV_CMD ids, type_masks
├── cli.py                 # argparse entrypoint
├── core/                  # internal flight-stack collaborators
│   ├── connection.py      # MAVLinkConnection — pymavlink + I/O lock + heartbeat/receiver
│   ├── telemetry.py       # Telemetry — incoming-message parsing + state cache
│   ├── commands.py        # CommandSender — COMMAND_LONG with asyncio.Future ACK routing
│   ├── streamer.py        # OffboardStreamer — setpoint thread + telemetry watchdog
│   ├── mission.py         # MissionOps — takeoff/goto/hover/land/rtl/emergency_land
│   ├── precision_land.py  # PrecisionLand — vision descent with altitude floor
│   ├── safety.py          # SafetyOps — wait_until_ready
│   └── mock.py            # MockMavConnection + in-process simulator
└── viz/                   # browser UI server (HTTP + SSE) + static ES modules

Every MAVLink send and recv goes through MAVLinkConnection, which holds the single threading lock. Each subsystem receives its dependencies via constructor injection — easy to mock in tests.

Connecting to real hardware

# Serial (Raspberry Pi <-> Pixhawk via UART)
drone = DroneController(connection_string="/dev/ttyAMA0")

# UDP (SITL or companion computer bridge)
drone = DroneController(connection_string="udp:192.168.1.10:14540")

# TCP
drone = DroneController(connection_string="tcp:127.0.0.1:5760")

Recommended safety parameters (set by apply_safe_params() defaults):

Development

# Install in editable mode with dev extras
pip install -e ".[dev]"

# Run tests
pytest -q

# Lint
ruff check mavpilot/

# Type check
mypy mavpilot/

License

MIT