Skip to content
mcardupilot

Use the library from a test harness

The MCP tools are built on mcardupilot.sitl.Sitl, and a test harness can use it directly. You get the same leases, process-group kills and log handling, with your own code in the loop on every message.

A flight in a few lines

Section titled “A flight in a few lines”
from pathlib import Path
from mcardupilot.sitl import Sitl


TREE = Path("~/ardupilot").expanduser()


with Sitl(
    frame="quadplane",
    defaults=[TREE / "Tools/autotest/default_params/quadplane.parm"],
    params={"BATT_MONITOR": 4},     # written to overlay.parm, loaded after the defaults
    speedup=10,
    owner="hover check",
) as s:
    s.arm_on_pad("QLOITER")         # throttle down, mode, arm; raises ArmRefused with the PreArm text
    s.rc[2] = 1800                  # rc is 0-based: index 2 is channel 3, throttle
    s.wait(6)                       # sim seconds, on the autopilot's clock
    print(s.height(), s.provenance["ardupilot_sha"])
    s.rc[2] = 1500
    s.m.set_mode("QLAND")
    s.wait_text("Disarm", 60)

start() (called by with) leases the first free pool instance whose ports are free, lays out <run_root>/<instance>, copies parameter files and Lua scripts in, launches SITL in its own process group, connects, requests streams and waits for home. Leaving the block kills the group, waits for the ports and releases the lease, exception or not.

Everything that waits does it through pump(), which sends the RC override and, with heartbeat=True, a GCS heartbeat paced on sim time. So wait, wait_text, get and mission uploads never starve the autopilot of RC, and a SITL that dies or PANICs mid-wait raises SitlDied after being killed.

Drive inputs from tick

Section titled “Drive inputs from tick”

tick(kind, msg) is called from pump for every message, after the bookkeeping, so the subclass sees this message’s armed state and text. Override it to make inputs evolve through every wait. Here, seeded Dryden gusts, the same model the server uses:

from pymavlink import mavutil
from mcardupilot.sitl import Sitl
from mcardupilot.turbulence import Dryden, sitl_wind


VFR_HUD = mavutil.mavlink.MAVLINK_MSG_ID_VFR_HUD




class GustySitl(Sitl):
    def __init__(self, mean_mps: float, from_deg: float, seed: int, **kw):
        # set your own state before start(): start() already pumps through tick()
        self.mean, self.from_deg = mean_mps, from_deg
        self.dryden = Dryden(seed)
        super().__init__(streams=[(VFR_HUD, 10)], **kw)


    def tick(self, kind, msg):
        if kind != "SYSTEM_TIME" or self.pad_alt is None:
            return
        hud = self.last.get("VFR_HUD")
        gust = self.dryden.step(msg.time_boot_ms / 1000, self.mean, self.height(),
                                hud.airspeed if hud else 0.0)
        spd, direction, tilt = sitl_wind(self.mean, self.from_deg, gust)
        self.set("SIM_WIND_SPD", round(spd, 3))
        self.set("SIM_WIND_DIR", round(direction, 1))
        self.set("SIM_WIND_DIR_Z", round(tilt, 2))

Rules for tick: it runs inside pump, so it must not pump or block. set() is fire and forget and sends nothing when the value repeats, which makes it safe to call every tick. get() pumps, so it does not belong here.

Plug in external physics: Bridge

Section titled “Plug in external physics: Bridge”

A Bridge is an external physics process SITL talks to over its JSON backend (9002 + 10n). Implement the protocol and pass bridge=:

from mcardupilot import procs




class MyBridge:
    def ports(self, instance):            # ports the bridge itself binds, probed before launch;
        return [("tcp", 7000 + instance)] # SITL's MAVLink and JSON ports are probed anyway


    def start(self, instance, run_dir):   # launch it; return SITL's --model argument
        self.proc = procs.launch_group(["uv", "run", "my-physics", "-I", str(instance)],
                                       run_dir, run_dir / "bridge.log")
        return "JSON:127.0.0.1"


    def set(self, name, value):           # True: the bridge models this one, so no PARAM_SET
        if name.startswith("SIM_WIND_"):  # goes to SITL; hand the value to your physics here
            return True
        return False


    def alive(self):
        return self.proc.poll() is None


    def kill(self):                       # the whole group: a process under uv outlives uv
        procs.kill_group(self.proc.pid, proc=self.proc)

launch_group and kill_group are in mcardupilot.procs. A bridge that dies raises SitlDied on the next liveness check, like SITL itself.

Keep each flight’s log under its own name

Section titled “Keep each flight’s log under its own name”

Sitl deletes logs at close by default (keep_logs=False), because one flight’s dataflash log is tens to hundreds of MB. Ask to keep the last one and move it out:

with Sitl(frame="quadplane", defaults=[...], keep_logs="last",
          log_dest=Path("results/logs"), log_stem="case-042") as s:
    ...
print(s.close())        # [PosixPath('results/logs/case-042.BIN')]; close() again is a no-op

The move matters: the next flight on the same instance treats whatever is left in its run directory as its own to delete. close(keep_logs=..., dest=..., stem=...) overrides the constructor for one flight, for example to keep only failures. mcardupilot.logs.prune holds a results directory to a budget the same way the server does.

When things go wrong

Section titled “When things go wrong”
Raised When
LeasePoolExhausted every pool instance is held or has busy ports; the message lists each with its holder or reason
InstanceBusy the instance you asked for is leased, or its ports are taken
ArmRefused arm_on_pad ran out of tries; the message is the autopilot’s last PreArm text
SitlDied SITL exited, PANICked, or its bridge died; SITL has already been killed

All four subclass McardupilotError. A script that exits without close() (an uncaught exception, sys.exit, Ctrl-C) still kills its SITL from an atexit hook. One that is SIGKILLed leaves <run_root>/<n>/sitl.pid; mcardupilot.sitl.kill_instance(n, run_root) reads it and kills the group if its leader is still the process that was launched.