PX4

nexus_uav.px4 holds PX4's own names: what a script that commands PX4 imports. OffboardClient is the host end of the offboard link of PX4 Software In The Loop (SITL). The run owns that link's address and names it in its port map, sim.ports, so the script connects the client itself once the sim has started. OffboardClient.connect(sim, timeout) listens on that address and waits for PX4's heartbeat. PX4 runs on the sim's clock, so the wait steps the sim, and timeout counts sim seconds. When no heartbeat comes, it raises RunError, which names the port, and so does any wait once the run has lost PX4.

import nexus_uav as nx
from nexus_uav.px4 import OffboardClient

with nx.Sim("astro_max_base", scene="empty") as sim:
    sim.start()
    with OffboardClient.connect(sim, 10.0) as op:  # sim.ports["offboard"]: UDP 14540 on instance 0
        op.takeoff(2.0)
        sim.wait_until(op.at_target, sim_timeout=60.0)

A run against the fake PX4 lists no offboard link, so sim.ports["offboard"] raises and names the fake. The mission plan types describe what upload_mission hands PX4:

from nexus_uav.px4 import MissionItem, Plan, read_plan

OffboardClient

OffboardClient(
    conn: str = f"udpin:0.0.0.0:{OFFBOARD_PORT}",
    *,
    system_id: int = 1,
    arrive_m: float = 2.0,
    yaw_tol_rad: float = radians(5.0),
    alt_tol_m: float = 1.0,
    wait_clear_s: float = 5.0,
)

Configure the offboard link; nothing opens a connection until entered.

PARAMETER	DESCRIPTION
conn	pymavlink connection string for the PX4 offboard link. The default listens on OFFBOARD_PORT, the PX4 offboard and GCS API port of instance 0; the run passes OFFBOARD_PORT + instance. TYPE: str DEFAULT: f'udpin:0.0.0.0:{OFFBOARD_PORT}'
system_id	PX4's MAVLink system id, instance + 1: the heartbeat this client waits for and the system every verb addresses. TYPE: int DEFAULT: 1
arrive_m	3D arrival radius for :meth:at_target after a :meth:goto, in meters. TYPE: float DEFAULT: 2.0
yaw_tol_rad	Heading tolerance for :meth:at_target when the caller commanded a yaw, in radians. TYPE: float DEFAULT: radians(5.0)
alt_tol_m	Altitude tolerance for :meth:at_target after a :meth:takeoff, in meters. TYPE: float DEFAULT: 1.0
wait_clear_s	Seconds the vehicle must be failure-free, and the initial settle period, before the pump sends an arm command. TYPE: float DEFAULT: 5.0

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def __init__(
    self,
    conn: str = f"udpin:0.0.0.0:{OFFBOARD_PORT}",
    *,
    system_id: int = 1,
    arrive_m: float = 2.0,
    yaw_tol_rad: float = math.radians(5.0),
    alt_tol_m: float = 1.0,
    wait_clear_s: float = 5.0,
):
    """Configure the offboard link; nothing opens a connection until entered.

    Args:
        conn: pymavlink connection string for the PX4 offboard link. The default
            listens on ``OFFBOARD_PORT``, the PX4 offboard and GCS API port of instance 0; the
            run passes ``OFFBOARD_PORT + instance``.
        system_id: PX4's MAVLink system id, ``instance + 1``: the heartbeat this client waits
            for and the system every verb addresses.
        arrive_m: 3D arrival radius for :meth:`at_target` after a :meth:`goto`, in meters.
        yaw_tol_rad: Heading tolerance for :meth:`at_target` when the caller commanded a yaw, in radians.
        alt_tol_m: Altitude tolerance for :meth:`at_target` after a :meth:`takeoff`, in meters.
        wait_clear_s: Seconds the vehicle must be failure-free, and the initial settle
            period, before the pump sends an arm command.
    """
    self._conn_str = conn
    self._system_id = system_id
    self._arrive_m = arrive_m
    self._yaw_tol_rad = yaw_tol_rad
    self._alt_tol_m = alt_tol_m
    self._wait_clear_s = wait_clear_s
    self._mav = None
    self._lock = threading.Lock()  # guards every send on the shared connection
    self._stop = threading.Event()
    self._thread: threading.Thread | None = None
    self._sysid = None
    self._compid = None
    # cached telemetry: the pump thread writes it, callers read it
    self._armed = False
    self._mode = ""
    self._landed = "UNKNOWN"
    self._rel_alt: float | None = None
    self._lat: float | None = None  # deg, from GLOBAL_POSITION_INT
    self._lon: float | None = None  # deg
    self._amsl: float | None = None  # m over mean sea level, the frame DO_REPOSITION resolves in
    self._yaw: float | None = None  # rad, world heading from ATTITUDE
    self._last_fail = 0.0
    self._rc: tuple[int, int, int, int] | None = None  # streamed MANUAL_CONTROL setpoint
    # Outstanding requests, serviced by the pump; see _service_requests. Each clears the
    # moment telemetry confirms it, so the pump never re-commands a state PX4 has already
    # left of its own accord: Takeoff mode hands over to Hold on arrival, and PX4 disarms
    # itself after landing; a latched request would fight the first and re-arm after the second.
    self._want_mode: str | None = None
    self._want_arm = False
    self._arm_gesture = False  # the outstanding arm is a stick gesture, not a command
    self._arm_since = 0.0  # when the caller requested the arm; the settle window starts here
    # The local tangent-plane anchor, lat0 [deg], lon0 [deg] and home altitude over mean sea level
    # [m], captured lazily on the first goto, and the target the last verb commanded, which
    # at_target measures against.
    self._anchor: tuple[float, float, float] | None = None
    self._target: _ClimbTarget | _GotoTarget | None = None
    # The uploaded mission; see upload_mission. The upload is a *handshake*, not a send: this
    # client puts MISSION_COUNT on the wire and PX4 then asks for each item by seq, so the pump
    # drives it from _on_msg and the caller waits on mission_uploaded().
    self._mission_items: tuple[MissionItem, ...] = ()
    self._want_upload = False  # outstanding upload: keep re-sending MISSION_COUNT until acked
    self._mission_ack: int | None = None  # MISSION_ACK type; 0 = accepted, None = not settled
    self._reached = -1  # highest MISSION_ITEM_REACHED seq; -1 = none reached yet
    self._mission_current = 0  # most recent MISSION_CURRENT seq
    # Outstanding start from the first item: keep sending MISSION_SET_CURRENT 0 until PX4 reports
    # item 0 current, and hold the arm until then.
    self._want_restart = False

connected property

connected: bool

Whether PX4 has answered on the offboard link: a heartbeat has arrived and the pump has learned the peer's system id, so every verb has somewhere to send.

RETURNS	DESCRIPTION
bool	True once the first PX4 heartbeat has arrived on :14540.

connect classmethod

connect(sim: Sim, timeout: float) -> OffboardClient

Open a client on the run's offboard link and wait for PX4's heartbeat, stepping the sim.

The client listens on the address the run's port map names, sim.ports["offboard"], and addresses the system it names. PX4 runs on the sim's clock, so the wait steps the sim, and timeout counts sim seconds. Call it once the sim has started; the client returned is open, and entering it as a context manager closes it on the way out::

sim.start()
with OffboardClient.connect(sim, 10.0) as op:
    op.takeoff(2.0)
    sim.wait_until(op.at_target, sim_timeout=60.0)

PARAMETER	DESCRIPTION
sim	The started run. TYPE: Sim
timeout	Sim seconds to wait for PX4's first heartbeat. TYPE: float

RETURNS	DESCRIPTION
OffboardClient	The open client, with PX4's heartbeat: :attr:connected is True.

RAISES	DESCRIPTION
KeyError	The run has no offboard link; the message says why, such as a run on the fake PX4.
RunError	No heartbeat came within timeout, naming the port, or the run lost a peer meanwhile; the client closes first.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

@classmethod
def connect(cls, sim: Sim, timeout: float) -> OffboardClient:
    """Open a client on the run's offboard link and wait for PX4's heartbeat, stepping the sim.

    The client listens on the address the run's port map names, ``sim.ports["offboard"]``, and
    addresses the system it names. PX4 runs on the sim's clock, so the wait steps the sim, and
    ``timeout`` counts sim seconds. Call it once the sim has started; the client returned is open,
    and entering it as a context manager closes it on the way out::

        sim.start()
        with OffboardClient.connect(sim, 10.0) as op:
            op.takeoff(2.0)
            sim.wait_until(op.at_target, sim_timeout=60.0)

    Args:
        sim: The started run.
        timeout: Sim seconds to wait for PX4's first heartbeat.

    Returns:
        The open client, with PX4's heartbeat: :attr:`connected` is ``True``.

    Raises:
        KeyError: The run has no offboard link; the message says why, such as a run on the fake PX4.
        RunError: No heartbeat came within ``timeout``, naming the port, or the run lost a peer meanwhile;
            the client closes first.
    """
    link = sim.ports["offboard"]
    client = cls(f"udpin:0.0.0.0:{link['port']}", system_id=link["system_id"]).open()
    try:
        sim.wait_until(lambda: client.connected, sim_timeout=timeout)
    except TimeoutError:
        client.close()
        raise RunError(
            f"no PX4 heartbeat on the offboard link, UDP port {link['port']}, within {timeout}s of sim time"
        ) from None
    except BaseException:
        client.close()
        raise
    return client

open

open() -> OffboardClient

Open the MAVLink link and start the pump, returning at once, with no wait for PX4.

The caller owns the waiting, because PX4's clock is the sim's under lockstep: the sim has to keep stepping or no heartbeat ever arrives, so the wait has to be a stepping loop rather than a sleep, sim.wait_until(lambda: op.connected, sim_timeout=…). Poll :attr:connected to know when PX4 has answered. Entering the client as a context manager calls this, and leaving it calls :meth:close.

RETURNS	DESCRIPTION
OffboardClient	self, so it chains.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def open(self) -> OffboardClient:
    """Open the MAVLink link and start the pump, returning at once, with no wait for PX4.

    The caller owns the waiting, because PX4's clock is the sim's under lockstep: the sim has
    to keep stepping or no heartbeat ever arrives, so the wait has to be a stepping loop rather
    than a sleep, ``sim.wait_until(lambda: op.connected, sim_timeout=…)``. Poll
    :attr:`connected` to know when PX4 has answered. Entering the client as a context manager
    calls this, and leaving it calls :meth:`close`.

    Returns:
        ``self``, so it chains.
    """
    self._stop.clear()
    self._mav = mavutil.mavlink_connection(self._conn_str, source_system=255, source_component=240)
    self._thread = threading.Thread(target=self._pump, name="px4-offboard", daemon=True)
    self._thread.start()
    return self

close

close() -> None

Stop the background pump thread and close the MAVLink connection.

Idempotent and safe to call after a partial start: signals the pump to stop, joins it, for up to 5 s, then closes the connection if open made one.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def close(self) -> None:
    """Stop the background pump thread and close the MAVLink connection.

    Idempotent and safe to call after a partial start: signals the pump to stop,
    joins it, for up to 5 s, then closes the connection if ``open`` made one.
    """
    self._stop.set()
    if self._thread is not None:
        self._thread.join(timeout=5.0)
        self._thread = None
    if self._mav is not None:
        self._mav.close()
        self._mav = None

set_mode

set_mode(mode: str) -> None

Request a flight mode. Returns at once; the pump commands it until telemetry confirms.

Entering a manual mode requires a live manual-control source, so a neutral MANUAL_CONTROL stream is auto-started first. Observe the result with :meth:mode.

PARAMETER	DESCRIPTION
mode	Friendly mode name; one of the keys of the mode table, for example "Manual", "Altitude", "Position", "Offboard", "Hold", "Takeoff", "Mission", "RTL" or "Land". TYPE: str

RAISES	DESCRIPTION
ValueError	If mode isn't a known mode name.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def set_mode(self, mode: str) -> None:
    """Request a flight mode. Returns at once; the pump commands it until telemetry confirms.

    Entering a manual mode requires a live manual-control source, so a neutral MANUAL_CONTROL
    stream is auto-started first. Observe the result with :meth:`mode`.

    Args:
        mode: Friendly mode name; one of the keys of the mode table, for example ``"Manual"``,
            ``"Altitude"``, ``"Position"``, ``"Offboard"``, ``"Hold"``, ``"Takeoff"``,
            ``"Mission"``, ``"RTL"`` or ``"Land"``.

    Raises:
        ValueError: If ``mode`` isn't a known mode name.
    """
    if mode not in _MODES:
        raise ValueError(f"unknown mode {mode!r}; known: {sorted(_MODES)}")
    # PX4 rejects a switch *into* a manual mode unless a manual-control source is live, so
    # ensure the neutral MANUAL_CONTROL stream is running; the pump streams self._rc.
    if mode in _MANUAL_MODES and self._rc is None:
        self.set_rc(roll=0.0, pitch=0.0, throttle=0.5, yaw=0.0)
    self._want_mode = mode

arm

arm(*, gesture: bool = False) -> None

Request arming. Returns at once; the pump arms once the vehicle is armable.

Armable means telemetry has confirmed any pending mode change, PX4 is publishing a position estimate, so the EKF has converged, and it has been failure-free for the configured settle period. Observe the result with :meth:is_armed.

PARAMETER

DESCRIPTION

gesture

Arm with the pilot's throttle-down plus yaw-right stick gesture rather than MAV_CMD_COMPONENT_ARM_DISARM: the manual path a real pilot flies. The pump holds neutral sticks until the vehicle is armable, then presents the gesture, so PX4 only ever gets an edge it can act on. Needs a live manual-control source, which this starts if none is running. The default, False, is the command arm, which is what :meth:takeoff and every autonomous script use. TYPE: bool DEFAULT: False

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def arm(self, *, gesture: bool = False) -> None:
    """Request arming. Returns at once; the pump arms once the vehicle is armable.

    Armable means telemetry has confirmed any pending mode change, PX4 is publishing a position
    estimate, so the EKF has converged, and it has been failure-free for the configured settle
    period. Observe the result with :meth:`is_armed`.

    Args:
        gesture: Arm with the pilot's throttle-down plus yaw-right **stick gesture** rather than
            ``MAV_CMD_COMPONENT_ARM_DISARM``: the manual path a real pilot flies. The pump
            holds neutral sticks until the vehicle is armable, then presents the gesture, so
            PX4 only ever gets an edge it can act on. Needs a live manual-control source, which
            this starts if none is running. The default, ``False``, is the command arm, which is
            what :meth:`takeoff` and every autonomous script use.
    """
    self._arm_since = time.monotonic()  # the settle window starts now
    self._want_arm = True
    self._arm_gesture = gesture
    if gesture and self._rc is None:
        self.set_rc(**_NEUTRAL_RC)  # PX4 needs a live manual-control source before a gesture can arm

disarm

disarm() -> bool

Send a single disarm command, fire-and-forget and not verified.

RETURNS	DESCRIPTION
bool	True once MAV_CMD_COMPONENT_ARM_DISARM has gone out.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def disarm(self) -> bool:
    """Send a single disarm command, fire-and-forget and not verified.

    Returns:
        ``True`` once ``MAV_CMD_COMPONENT_ARM_DISARM`` has gone out.
    """
    self._want_arm = False  # drop any outstanding arm request, or the pump would undo this
    self._arm_gesture = False  # … and a dropped request must not leave the pump gesturing
    self._cmd(mavutil.mavlink.MAV_CMD_COMPONENT_ARM_DISARM, 0.0)
    return True

set_rc

set_rc(
    *,
    roll: float = 0.0,
    pitch: float = 0.0,
    throttle: float = 0.5,
    yaw: float = 0.0,
) -> None

Set the streamed MANUAL_CONTROL setpoint. roll/pitch/yaw in [-1,1]; throttle in [0,1].

Stores the setpoint; the background pump streams it at ~20 Hz until changed or cleared with stop_rc. This clamps the values to range, then scales them to the MANUAL_CONTROL integer fields.

PARAMETER	DESCRIPTION
roll	Roll stick, clamped to [-1, 1]. TYPE: float DEFAULT: 0.0
pitch	Pitch stick, clamped to [-1, 1]. TYPE: float DEFAULT: 0.0
throttle	Throttle stick, clamped to [0, 1]; 0.5 is mid-stick. TYPE: float DEFAULT: 0.5
yaw	Yaw stick, clamped to [-1, 1]. TYPE: float DEFAULT: 0.0

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def set_rc(self, *, roll: float = 0.0, pitch: float = 0.0, throttle: float = 0.5, yaw: float = 0.0) -> None:
    """Set the streamed MANUAL_CONTROL setpoint. roll/pitch/yaw in [-1,1]; throttle in [0,1].

    Stores the setpoint; the background pump streams it at ~20 Hz until changed or
    cleared with ``stop_rc``. This clamps the values to range, then scales them to the
    MANUAL_CONTROL integer fields.

    Args:
        roll: Roll stick, clamped to [-1, 1].
        pitch: Pitch stick, clamped to [-1, 1].
        throttle: Throttle stick, clamped to [0, 1]; 0.5 is mid-stick.
        yaw: Yaw stick, clamped to [-1, 1].
    """

    def clamp(v, lo, hi):
        return max(lo, min(hi, v))

    x = int(clamp(pitch, -1.0, 1.0) * 1000)
    y = int(clamp(roll, -1.0, 1.0) * 1000)
    z = int(clamp(throttle, 0.0, 1.0) * 1000)
    r = int(clamp(yaw, -1.0, 1.0) * 1000)
    self._rc = (x, y, z, r)

stop_rc

stop_rc() -> None

Stop streaming the MANUAL_CONTROL setpoint; the pump goes quiet until set_rc.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def stop_rc(self) -> None:
    """Stop streaming the MANUAL_CONTROL setpoint; the pump goes quiet until ``set_rc``."""
    self._rc = None

param_set

param_set(name: str, value: float) -> None

Set a REAL32 PX4 parameter: a fire-and-forget PARAM_SET that awaits no acknowledgement.

Float-only on purpose: an integer parameter doesn't ride the wire as a float, so it gets its own verb rather than a type argument this one can't honour; see :meth:param_set_int.

PARAMETER	DESCRIPTION
name	Parameter name, for example "MIS_TAKEOFF_ALT". TYPE: str
value	New value, sent as a float. TYPE: float

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def param_set(self, name: str, value: float) -> None:
    """Set a REAL32 PX4 parameter: a fire-and-forget ``PARAM_SET`` that awaits no acknowledgement.

    Float-only on purpose: an integer parameter doesn't ride the wire as a float, so it gets
    its own verb rather than a type argument this one can't honour; see :meth:`param_set_int`.

    Args:
        name: Parameter name, for example ``"MIS_TAKEOFF_ALT"``.
        value: New value, sent as a float.
    """
    with self._lock:
        self._mav.mav.param_set_send(
            self._sysid, self._compid, name.encode(), float(value), mavutil.mavlink.MAV_PARAM_TYPE_REAL32
        )

param_set_int

param_set_int(name: str, value: int) -> None

Set an INT32 PX4 parameter: a fire-and-forget PARAM_SET that awaits no acknowledgement.

PARAM_SET carries every value in one float field, and PX4 reads an integer parameter out of that field's bits rather than its numeric value. So this reinterprets the int rather than casting it: sending float(1) would have PX4 read 1065353216.

PARAMETER	DESCRIPTION
name	Parameter name, for example "COM_RC_IN_MODE". TYPE: str
value	New value, sent as the INT32 bit pattern of the float field. TYPE: int

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def param_set_int(self, name: str, value: int) -> None:
    """Set an INT32 PX4 parameter: a fire-and-forget ``PARAM_SET`` that awaits no acknowledgement.

    PARAM_SET carries every value in one float field, and PX4 reads an integer parameter out of
    that field's **bits** rather than its numeric value. So this reinterprets the int rather than
    casting it: sending ``float(1)`` would have PX4 read 1065353216.

    Args:
        name: Parameter name, for example ``"COM_RC_IN_MODE"``.
        value: New value, sent as the INT32 bit pattern of the float field.
    """
    bits = struct.unpack("<f", struct.pack("<i", int(value)))[0]
    with self._lock:
        self._mav.mav.param_set_send(
            self._sysid, self._compid, name.encode(), bits, mavutil.mavlink.MAV_PARAM_TYPE_INT32
        )

takeoff

takeoff(alt_m: float) -> None

Set the takeoff altitude, request Takeoff mode, then request arming. Returns at once.

The ordering is load-bearing and this method owns it: telemetry confirms the DO_SET_MODE for Takeoff mode before the arm command, which is why an autonomous script never calls :meth:arm itself. PX4 climbs on arming; wait for the climb with :meth:at_target.

PARAMETER	DESCRIPTION
alt_m	Target takeoff altitude in meters, written to MIS_TAKEOFF_ALT. TYPE: float

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def takeoff(self, alt_m: float) -> None:
    """Set the takeoff altitude, request Takeoff mode, then request arming. Returns at once.

    The ordering is load-bearing and this method owns it: telemetry confirms
    the ``DO_SET_MODE`` for Takeoff mode **before** the arm command, which is why an autonomous script never
    calls :meth:`arm` itself. PX4 climbs on arming; wait for the climb with :meth:`at_target`.

    Args:
        alt_m: Target takeoff altitude in meters, written to ``MIS_TAKEOFF_ALT``.
    """
    self.param_set("MIS_TAKEOFF_ALT", alt_m)
    self._want_restart = False  # a takeoff replaces any mission start still waiting on PX4
    self._target = _ClimbTarget(rel_alt=alt_m)
    self.set_mode("Takeoff")
    self.arm()

land

land() -> None

Request Land mode to land the vehicle in place. Returns at once.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def land(self) -> None:
    """Request Land mode to land the vehicle in place. Returns at once."""
    self._target = None  # nothing to arrive at; at_target goes False
    self._want_restart = False  # a landing replaces any mission start still waiting on PX4
    self.set_mode("Land")

upload_mission

upload_mission(
    plan: Plan | Sequence[MissionItem] | str | PathLike,
) -> None

Upload a mission to PX4. Returns at once; wait for it with :meth:mission_uploaded.

The items go on the wire verbatim, in the geodetic frame the .plan authored them in; nothing converts through the sim's world frame, so a plan flies where QGroundControl drew it.

Upload is a handshake rather than a send: this lodges the items, the pump puts MISSION_COUNT on the wire, PX4 then asks for each item by seq, and it ends with PX4's MISSION_ACK. Re-uploading replaces the mission and resets the progress counters, so a second mission never reports the first one's arrivals.

PARAMETER	DESCRIPTION
plan	A :class:~nexus_uav.px4.Plan, a bare sequence of :class:~nexus_uav.px4.MissionItem, or a path to a .plan file, read with :func:~nexus_uav.px4.read_plan. TYPE: Plan | Sequence[MissionItem] | str | PathLike

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def upload_mission(self, plan: Plan | Sequence[MissionItem] | str | os.PathLike) -> None:
    """Upload a mission to PX4. Returns at once; wait for it with :meth:`mission_uploaded`.

    The items go on the wire **verbatim**, in the geodetic frame the ``.plan`` authored them in;
    nothing converts through the sim's world frame, so a plan flies where QGroundControl drew it.

    Upload is a handshake rather than a send: this lodges the items, the pump puts
    ``MISSION_COUNT`` on the wire, PX4 then asks for each item by ``seq``, and it ends with
    PX4's ``MISSION_ACK``. Re-uploading replaces the mission and resets the progress counters,
    so a second mission never reports the first one's arrivals.

    Args:
        plan: A :class:`~nexus_uav.px4.Plan`, a bare sequence of
            :class:`~nexus_uav.px4.MissionItem`, or a path to a ``.plan``
            file, read with :func:`~nexus_uav.px4.read_plan`.
    """
    if isinstance(plan, (str, os.PathLike)):
        plan = read_plan(plan)
    self._mission_items = tuple(plan.items if isinstance(plan, Plan) else plan)
    self._mission_ack = None
    self._reached = -1
    self._mission_current = 0
    self._want_upload = True
    self._want_restart = False  # a new mission replaces any start still waiting on PX4

mission_uploaded

mission_uploaded() -> bool

Report whether PX4 has accepted the uploaded mission.

RETURNS	DESCRIPTION
bool	True once MISSION_ACK reported acceptance. Stays False while the handshake
bool	is in flight, and also when PX4 rejected the mission; check :meth:mission_ack
bool	to tell a "not yet" from a "refused" answer.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def mission_uploaded(self) -> bool:
    """Report whether PX4 has accepted the uploaded mission.

    Returns:
        ``True`` once ``MISSION_ACK`` reported acceptance. Stays ``False`` while the handshake
        is in flight, and also when PX4 **rejected** the mission; check :meth:`mission_ack`
        to tell a "not yet" from a "refused" answer.
    """
    return self._mission_ack == mavutil.mavlink.MAV_MISSION_ACCEPTED

mission_ack

mission_ack() -> int | None

Report PX4's MISSION_ACK result code.

RETURNS	DESCRIPTION
int | None	The MAV_MISSION_RESULT value, where 0 = accepted, or None if PX4 hasn't
int | None	acknowledged an upload yet.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def mission_ack(self) -> int | None:
    """Report PX4's ``MISSION_ACK`` result code.

    Returns:
        The MAV_MISSION_RESULT value, where ``0`` = accepted, or ``None`` if PX4 hasn't
        acknowledged an upload yet.
    """
    return self._mission_ack

mission_count

mission_count() -> int

Report how many items are in the uploaded mission.

RETURNS	DESCRIPTION
int	The item count, or 0 before any upload.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def mission_count(self) -> int:
    """Report how many items are in the uploaded mission.

    Returns:
        The item count, or ``0`` before any upload.
    """
    return len(self._mission_items)

mission_reached

mission_reached() -> int

Report the highest mission item PX4 says it has reached.

RETURNS	DESCRIPTION
int	The sequence number from the most recent MISSION_ITEM_REACHED, or -1 if none has
int	arrived yet. Never decreases within one mission.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def mission_reached(self) -> int:
    """Report the highest mission item PX4 says it has reached.

    Returns:
        The sequence number from the most recent ``MISSION_ITEM_REACHED``, or ``-1`` if none has
        arrived yet. Never decreases within one mission.
    """
    return self._reached

mission_current

mission_current() -> int

Report the item PX4 is flying to.

RETURNS	DESCRIPTION
int	The sequence number from the most recent MISSION_CURRENT, or 0 before the first.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def mission_current(self) -> int:
    """Report the item PX4 is flying to.

    Returns:
        The sequence number from the most recent ``MISSION_CURRENT``, or ``0`` before the first.
    """
    return self._mission_current

mission_complete

mission_complete() -> bool

Report whether the vehicle has reached the mission's last waypoint.

Measured on the last NAV_WAYPOINT, not the last item. A mission that ends in NAV_RETURN_TO_LAUNCH hands the flight to Return To Launch (RTL) instead of reporting an arrival there, so waiting on the final item could never return. Wait out the RTL tail with :meth:landed_state instead.

RETURNS	DESCRIPTION
bool	True once the vehicle has reached the final waypoint; False with no uploaded
bool	mission, or one that carries no waypoint at all.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def mission_complete(self) -> bool:
    """Report whether the vehicle has reached the mission's last **waypoint**.

    Measured on the last ``NAV_WAYPOINT``, not the last item. A mission that ends in
    ``NAV_RETURN_TO_LAUNCH`` hands the flight to Return To Launch (RTL) instead of reporting an
    arrival there, so waiting on the final item could never return. Wait out the RTL tail with
    :meth:`landed_state` instead.

    Returns:
        ``True`` once the vehicle has reached the final waypoint; ``False`` with no uploaded
        mission, or one that carries no waypoint at all.
    """
    last_wp = [it.seq for it in self._mission_items if it.command == NAV_WAYPOINT]
    return bool(last_wp) and self._reached >= last_wp[-1]

start_mission

start_mission() -> None

Request the mission's first item, Mission mode, then arming. Returns at once.

The ordering is load-bearing and this method owns it, exactly as :meth:takeoff does: telemetry confirms the DO_SET_MODE for Mission mode before the arm command, which is the proven headless recipe. PX4 flies the mission on arming, from the item it holds as current, so the arm also waits until PX4 reports item 0 current after MISSION_SET_CURRENT: PX4 skips an upload that matches the mission it already holds, which after a finished flight sits at its last item, and would end the new flight on the ground.

Upload the mission first and wait for :meth:mission_uploaded: PX4 refuses Mission mode while it has no valid mission, so engaging early just burns retries.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def start_mission(self) -> None:
    """Request the mission's first item, Mission mode, then arming. Returns at once.

    The ordering is load-bearing and this method owns it, exactly as :meth:`takeoff` does:
    telemetry confirms the ``DO_SET_MODE`` for Mission mode **before** the arm command, which is the proven
    headless recipe. PX4 flies the mission on arming, from the item it holds as current, so the
    arm also waits until PX4 reports item 0 current after ``MISSION_SET_CURRENT``: PX4 skips an
    upload that matches the mission it already holds, which after a finished flight sits at its
    last item, and would end the new flight on the ground.

    Upload the mission first and wait for :meth:`mission_uploaded`: PX4 refuses Mission mode
    while it has no valid mission, so engaging early just burns retries.
    """
    self._want_restart = True
    self.set_mode("Mission")
    self.arm()

goto

goto(
    setpoint: Setpoint, *, yaw: float | None = None
) -> None

Fly to a world-frame position. Returns at once; wait for it with :meth:at_target.

Commanded as DO_REPOSITION: the same guidance PX4 runs for a GCS "fly here" request, and what it latches until the next command, so this goes out once rather than streamed. The position is in world axes, Newton FLU and Z-up, the same tuples a mission in the loop takes, resolved against an anchor captured on the first call; see :meth:_anchor_now.

Yaw is in world axes too, and this reflects it on the way out for the same reason it reflects the position: a right-handed yaw about world +z turns +x toward +y, which is north toward west, while PX4's North-East-Down (NED) heading is positive north toward east. Leaving the position in one frame and the heading in the other is the trap #61 was.

PARAMETER	DESCRIPTION
setpoint	The target: a PositionGoal or a bare (x, y, z) world-frame position in meters, with z over the launch point. TYPE: Setpoint
yaw	Optional world heading in radians, right-handed about +z, so +90° faces +y = west; overrides a PositionGoal's own yaw. None leaves the heading to PX4. TYPE: float | None DEFAULT: None

RAISES	DESCRIPTION
RuntimeError	No position has arrived on the offboard link yet.
TypeError	setpoint isn't a PositionGoal or an (x, y, z) position.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def goto(self, setpoint: Setpoint, *, yaw: float | None = None) -> None:
    """Fly to a world-frame position. Returns at once; wait for it with :meth:`at_target`.

    Commanded as ``DO_REPOSITION``: the same guidance PX4 runs for a GCS "fly here" request, and
    what it latches until the next command, so this goes out once rather than streamed. The
    position is in **world axes**, Newton FLU and Z-up, the same tuples a mission in the loop
    takes, resolved against an anchor captured on the first call; see :meth:`_anchor_now`.

    Yaw is in world axes too, and this reflects it on the way out for the same reason it
    reflects the position: a right-handed yaw about world +z turns +x toward +y, which is north
    toward **west**, while PX4's North-East-Down (NED) heading is positive north toward
    **east**. Leaving the position in one frame and the heading in the other is the trap #61 was.

    Args:
        setpoint: The target: a ``PositionGoal`` or a bare
            ``(x, y, z)`` world-frame position in meters, with ``z`` over the launch point.
        yaw: Optional world heading in radians, right-handed about +z, so +90° faces +y =
            west; overrides a ``PositionGoal``'s own ``yaw``. ``None`` leaves the heading
            to PX4.

    Raises:
        RuntimeError: No position has arrived on the offboard link yet.
        TypeError: ``setpoint`` isn't a ``PositionGoal`` or an ``(x, y, z)`` position.
    """
    goal = as_position_goal(setpoint)
    x, y, z = goal.pos
    yaw_world = yaw if yaw is not None else goal.yaw
    yaw_px4 = None if yaw_world is None else _wrap_pi(-float(yaw_world))
    _, _, home_amsl = self._anchor_now()
    lat, lon = self._to_global(x, y)
    self._cmd_int(
        mavutil.mavlink.MAV_CMD_DO_REPOSITION,
        -1.0,  # p1 ground speed: -1 = default
        1.0,  # p2 bitmask: change to Hold on arrival
        0.0,  # p3 reserved
        float("nan") if yaw_px4 is None else yaw_px4,  # p4 yaw [rad], PX4 frame
        round(lat * 1e7),
        round(lon * 1e7),
        home_amsl + z,
    )
    # Store what went on the wire: ATTITUDE reports in PX4's frame, so at_target compares
    # PX4 against PX4 and never needs the conversion a second time.
    self._target = _GotoTarget(lat=lat, lon=lon, rel_alt=z, yaw=yaw_px4)

at_target

at_target() -> bool

Report whether the vehicle has reached what the last verb commanded.

Measured against what PX4 reports, the cached GLOBAL_POSITION_INT and ATTITUDE, never the sim's ground truth. The client is a remote GCS; reading sim state here would let it see a position the autopilot doesn't have.

RETURNS	DESCRIPTION
bool	True once inside the arrival tolerances; False while still en route, with no
bool	target, because no verb commanded one or the last verb was :meth:land, or before the
bool	first position report.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def at_target(self) -> bool:
    """Report whether the vehicle has reached what the last verb commanded.

    Measured against **what PX4 reports**, the cached ``GLOBAL_POSITION_INT`` and ``ATTITUDE``,
    never the sim's ground truth. The client is a remote GCS; reading sim state here would
    let it see a position the autopilot doesn't have.

    Returns:
        ``True`` once inside the arrival tolerances; ``False`` while still en route, with no
        target, because no verb commanded one or the last verb was :meth:`land`, or before the
        first position report.
    """
    target = self._target
    if target is None or self._rel_alt is None:
        return False
    if isinstance(target, _ClimbTarget):
        return abs(self._rel_alt - target.rel_alt) <= self._alt_tol_m
    if self._lat is None or self._lon is None:
        return False
    lat0, _, _ = self._anchor_now()
    m_per_deg = math.pi / 180.0 * _R_EARTH
    d_n = (self._lat - target.lat) * m_per_deg
    d_e = (self._lon - target.lon) * m_per_deg * math.cos(math.radians(lat0))
    d_z = self._rel_alt - target.rel_alt
    if math.sqrt(d_n * d_n + d_e * d_e + d_z * d_z) >= self._arrive_m:
        return False
    if target.yaw is None:
        return True
    if self._yaw is None:
        return False
    return abs(_wrap_pi(self._yaw - target.yaw)) < self._yaw_tol_rad

is_armed

is_armed() -> bool

Report the cached arming state from the most recent heartbeat.

RETURNS	DESCRIPTION
bool	True if the vehicle's last heartbeat reported the safety-armed flag.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def is_armed(self) -> bool:
    """Report the cached arming state from the most recent heartbeat.

    Returns:
        ``True`` if the vehicle's last heartbeat reported the safety-armed flag.
    """
    return self._armed

mode

mode() -> str

Report the cached flight mode decoded from the most recent heartbeat.

RETURNS	DESCRIPTION
str	The friendly mode name, for example "Takeoff", or a "main=...,sub=..."
str	string if the mode isn't in the known table.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def mode(self) -> str:
    """Report the cached flight mode decoded from the most recent heartbeat.

    Returns:
        The friendly mode name, for example ``"Takeoff"``, or a ``"main=...,sub=..."``
        string if the mode isn't in the known table.
    """
    return self._mode

landed_state

landed_state() -> str

Report the cached landed state from EXTENDED_SYS_STATE.

RETURNS	DESCRIPTION
str	"ON_GROUND", "IN_AIR", "TAKING_OFF", "LANDING", or
str	"UNKNOWN" if not yet reported.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def landed_state(self) -> str:
    """Report the cached landed state from EXTENDED_SYS_STATE.

    Returns:
        ``"ON_GROUND"``, ``"IN_AIR"``, ``"TAKING_OFF"``, ``"LANDING"``, or
        ``"UNKNOWN"`` if not yet reported.
    """
    return self._landed

relative_altitude

relative_altitude() -> float | None

Report the cached altitude over the launch point, in meters.

RETURNS	DESCRIPTION
float | None	Relative altitude in meters from GLOBAL_POSITION_INT, or None if no
float | None	position has arrived yet.

Source code in nexus_uav/_src/peers/px4_sitl/offboard.py

def relative_altitude(self) -> float | None:
    """Report the cached altitude over the launch point, in meters.

    Returns:
        Relative altitude in meters from GLOBAL_POSITION_INT, or ``None`` if no
        position has arrived yet.
    """
    return self._rel_alt

Plan dataclass

Plan(
    home: tuple[float, float, float],
    items: tuple[MissionItem, ...],
)

A parsed .plan: where the vehicle starts, and the items it flies.

ATTRIBUTE	DESCRIPTION
home	plannedHomePosition as (lat [deg], lon [deg], alt [m]). The altitude is height over mean sea level, not the WGS84 ellipsoidal height a streamed globe's origin takes. TYPE: tuple[float, float, float]
items	The mission items in file order, with seq matching the index. TYPE: tuple[MissionItem, ...]

home instance-attribute

home: tuple[float, float, float]

items instance-attribute

items: tuple[MissionItem, ...]

MissionItem dataclass

MissionItem(
    seq: int,
    frame: int,
    command: int,
    autocontinue: bool,
    params: tuple[float, float, float, float],
    x: int,
    y: int,
    z: float,
)

One mission item, in the shape MISSION_ITEM_INT carries it.

ATTRIBUTE	DESCRIPTION
seq	Position in the mission, assigned from the item's index in the file. TYPE: int
frame	MAV_FRAME for this item; 3, FRAME_GLOBAL_RELATIVE_ALT, means z is the altitude in metres relative to the home position. TYPE: int
command	MAV_CMD for this item, for example 16 for NAV_WAYPOINT, 22 for NAV_TAKEOFF, 20 for NAV_RETURN_TO_LAUNCH. TYPE: int
autocontinue	Whether PX4 proceeds to the next item on arrival. TYPE: bool
params	param1 through param4. A null in the file arrives as NaN, which is how MAVLink spells "no value": param4 NaN leaves the heading to PX4. TYPE: tuple[float, float, float, float]
x	Latitude in degrees times 1e7: the integer field, which is why the MISSION_ITEM_INT variant exists. TYPE: int
y	Longitude in degrees times 1e7. TYPE: int
z	Altitude in metres, in this item's frame. TYPE: float

seq instance-attribute

seq: int

frame instance-attribute

frame: int

command instance-attribute

command: int

autocontinue instance-attribute

autocontinue: bool

params instance-attribute

params: tuple[float, float, float, float]

x instance-attribute

x: int

y instance-attribute

y: int

z instance-attribute

z: float

read_plan

read_plan(path: str | PathLike) -> Plan

Read a QGroundControl .plan file.

The items keep their file order and get a seq from their index, so the mission uploads as 0..N-1 with the first real item at seq 0. There is no implicit home item: PX4 takes the mission as sent, and this is what MAVSDK puts on the wire too.

This reads the mission and nothing else: it skips geoFence and rallyPoints.

PARAMETER	DESCRIPTION
path	Path to the .plan file. TYPE: str | PathLike

RETURNS	DESCRIPTION
Plan	The parsed :class:Plan.

RAISES	DESCRIPTION
ValueError	The file isn't a plan, meaning fileType isn't "Plan"; it carries no mission object; an item isn't a "SimpleItem", since this reader doesn't handle a QGroundControl survey ComplexItem; or an item doesn't carry exactly seven params.

Source code in nexus_uav/_src/peers/px4_sitl/qgc_plan.py

def read_plan(path: str | os.PathLike) -> Plan:
    """Read a QGroundControl ``.plan`` file.

    The items keep their file order and get a ``seq`` from their index, so the mission uploads as
    ``0..N-1`` with the first real item at ``seq`` 0. There is no implicit home item: PX4 takes the
    mission as sent, and this is what MAVSDK puts on the wire too.

    This reads the mission and nothing else: it skips ``geoFence`` and ``rallyPoints``.

    Args:
        path: Path to the ``.plan`` file.

    Returns:
        The parsed :class:`Plan`.

    Raises:
        ValueError: The file isn't a plan, meaning ``fileType`` isn't ``"Plan"``; it carries no
            ``mission`` object; an item isn't a ``"SimpleItem"``, since this reader doesn't handle a
            QGroundControl survey ``ComplexItem``; or an item doesn't carry exactly seven ``params``.
    """
    with open(path) as fh:
        doc = json.load(fh)

    if doc.get("fileType") != "Plan":
        raise ValueError(f"{path}: not a QGC .plan (fileType={doc.get('fileType')!r})")
    mission = doc.get("mission")
    if not isinstance(mission, dict):
        raise ValueError(f"{path}: no 'mission' object")

    items = []
    for seq, raw in enumerate(mission.get("items", [])):
        kind = raw.get("type")
        if kind != "SimpleItem":
            raise ValueError(f"{path}: item {seq} is a {kind!r}; only 'SimpleItem' is supported")
        params = raw.get("params", [])
        if len(params) != 7:
            raise ValueError(f"{path}: item {seq} has {len(params)} params, expected 7")
        items.append(
            MissionItem(
                seq=seq,
                frame=int(raw["frame"]),
                command=int(raw["command"]),
                autocontinue=bool(raw.get("autoContinue", True)),
                params=(_param(params[0]), _param(params[1]), _param(params[2]), _param(params[3])),
                x=_dege7(params[4]),
                y=_dege7(params[5]),
                z=0.0 if params[6] is None else float(params[6]),
            )
        )

    lat, lon, alt = mission.get("plannedHomePosition", [0.0, 0.0, 0.0])
    return Plan(home=(float(lat), float(lon), float(alt)), items=tuple(items))