pamoja.mavlink

Idiomatic MAVLink wire-protocol facade.

MAVLink is the language drones speak: PX4 and ArduPilot autopilots and MAVSDK ground stations all exchange MAVLink frames, so talking to a vehicle means putting exactly the right bytes on the wire and trusting the bytes that come back. This is that byte layer: v1 and v2 frames, the CRC-16/MCRF4XX checksum every frame carries, the per-message CRC_EXTRA seed that catches a frame whose shape does not match, and MAVLink 2 signing.

Nothing here is limited to the messages this build happens to know. The common dialect's seeds are built in, and Dialect carries any others, derived from a message definition the way the specification does.

Above the bytes sits the shape: MessageSchema names a message's fields, so a MavlinkMessage is filled in and read back by name rather than by byte offset, and MessageSchemaBuilder describes a message this build has never heard of.

Above the messages sit the exchanges: MissionSender and MissionReceiver carry a plan between a station and a vehicle, CommandProtocol matches a command to its acknowledgement and counts retries, and local_position(), local_velocity(), and global_position() build setpoints. Each takes a frame off the link and hands back the frame to send, with no IO or timers of its own.

  1"""Idiomatic MAVLink wire-protocol facade.
  2
  3MAVLink is the language drones speak: PX4 and ArduPilot autopilots and MAVSDK
  4ground stations all exchange MAVLink frames, so talking to a vehicle means
  5putting exactly the right bytes on the wire and trusting the bytes that come
  6back. This is that byte layer: v1 and v2 frames, the CRC-16/MCRF4XX checksum
  7every frame carries, the per-message ``CRC_EXTRA`` seed that catches a frame
  8whose shape does not match, and MAVLink 2 signing.
  9
 10Nothing here is limited to the messages this build happens to know. The common
 11dialect's seeds are built in, and :class:`Dialect` carries any others, derived
 12from a message definition the way the specification does.
 13
 14Above the bytes sits the shape: :class:`MessageSchema` names a message's fields,
 15so a :class:`MavlinkMessage` is filled in and read back by name rather than by
 16byte offset, and :class:`MessageSchemaBuilder` describes a message this build has
 17never heard of.
 18
 19Above the messages sit the exchanges: :class:`MissionSender` and
 20:class:`MissionReceiver` carry a plan between a station and a vehicle,
 21:class:`CommandProtocol` matches a command to its acknowledgement and counts
 22retries, and :func:`local_position`, :func:`local_velocity`, and
 23:func:`global_position` build setpoints. Each takes a frame off the link and
 24hands back the frame to send, with no IO or timers of its own.
 25"""
 26
 27from __future__ import annotations
 28
 29import time
 30from enum import IntEnum, IntFlag
 31
 32from pamoja._native import (
 33    AckOutcome,
 34    CommandProtocol,
 35    Dialect,
 36    MavlinkFieldInfo,
 37    MavlinkFrame,
 38    MavlinkHeader,
 39    MavlinkMessage,
 40    MavlinkParser,
 41    MavlinkSigner,
 42    MavlinkVerifier,
 43    MessageSchema,
 44    MessageSchemaBuilder,
 45    MissionReceiver,
 46    MissionSender,
 47    ReceiverStep,
 48    SenderStep,
 49    mavlink_crc16_mcrf4xx,
 50    mavlink_known_crc_extra,
 51    mavlink_known_messages,
 52    mavlink_message_crc_extra,
 53    mavlink_offboard_global_position,
 54    mavlink_offboard_local_position,
 55    mavlink_offboard_local_velocity,
 56    mavlink_offboard_type_mask,
 57    mavlink_timestamp_from_unix_micros,
 58)
 59
 60__all__ = [
 61    "AckOutcome",
 62    "CommandProtocol",
 63    "DEFAULT_TIMESTAMP_WINDOW",
 64    "Dialect",
 65    "FieldType",
 66    "KEY_LEN",
 67    "MAX_FRAME",
 68    "MAX_PAYLOAD",
 69    "MAX_RETRIES",
 70    "MavlinkFieldInfo",
 71    "MavlinkFrame",
 72    "MavlinkHeader",
 73    "MavlinkMessage",
 74    "MavlinkParser",
 75    "MavlinkSigner",
 76    "MavlinkVerifier",
 77    "MessageSchema",
 78    "MessageSchemaBuilder",
 79    "MissionReceiver",
 80    "MissionSender",
 81    "ReceiverStep",
 82    "SIGNATURE_LEN",
 83    "SenderStep",
 84    "TypeMask",
 85    "crc16",
 86    "frame",
 87    "from_dict",
 88    "global_position",
 89    "known_crc_extra",
 90    "known_messages",
 91    "local_position",
 92    "local_velocity",
 93    "message",
 94    "message_crc_extra",
 95    "schema_for",
 96    "timestamp_from_unix_micros",
 97    "timestamp_now",
 98    "to_dict",
 99    "type_mask",
100]
101
102#: The largest payload a frame can carry, in bytes.
103MAX_PAYLOAD = 255
104
105#: The largest frame, in bytes, header, checksum and signature included.
106MAX_FRAME = 280
107
108#: The length of a v2 signature block, in bytes.
109SIGNATURE_LEN = 13
110
111#: The length of a signing key, in bytes.
112KEY_LEN = 32
113
114#: The default window a verifier accepts a timestamp within.
115DEFAULT_TIMESTAMP_WINDOW = 6_000_000
116
117#: The number of times a request is retransmitted before a transfer is abandoned, as
118#: the mission protocol recommends.
119MAX_RETRIES = 5
120
121
122def crc16(data: bytes) -> int:
123    """Return the CRC-16/MCRF4XX checksum of a byte string.
124
125    This is the checksum every MAVLink frame carries, exposed because a host
126    that implements part of the protocol itself needs the same arithmetic.
127
128    :param data: The data to checksum.
129    :returns: The checksum.
130    """
131    return mavlink_crc16_mcrf4xx(data)
132
133
134def message_crc_extra(name: str, fields: list[tuple[str, str, int]]) -> int:
135    """Derive the ``CRC_EXTRA`` seed of a message from its definition.
136
137    This is what makes a dialect this build has never seen usable: given a
138    message's name and its base fields in wire order, the seed comes out the
139    same as the one the dialect publishes, and a frame carrying that message
140    then checks like any other.
141
142    Extension fields are excluded from the seed and must not be listed, which is
143    what lets a peer that predates them still check the frame.
144
145    :param name: The message name, such as ``HEARTBEAT``.
146    :param fields: The base fields in wire order, as ``(type, name, array_len)``
147        triples; ``array_len`` is ``0`` for a scalar.
148    :returns: The seed.
149
150    >>> message_crc_extra("PRIVATE_STATUS", [("uint32_t", "uptime", 0)]) >= 0
151    True
152    """
153    return mavlink_message_crc_extra(name, fields)
154
155
156def known_crc_extra(msgid: int) -> int | None:
157    """Return the ``CRC_EXTRA`` the common dialect publishes for a message id.
158
159    :param msgid: The message id to look up.
160    :returns: The seed, or ``None`` for an id outside the common dialect, which
161        is what a :class:`Dialect` is for.
162
163    >>> known_crc_extra(0)
164    50
165    >>> known_crc_extra(9999) is None
166    True
167    """
168    return mavlink_known_crc_extra(msgid)
169
170
171def timestamp_from_unix_micros(unix_micros: int) -> int:
172    """Convert Unix time into the timestamp MAVLink signing counts in.
173
174    :param unix_micros: The time in microseconds since the Unix epoch.
175    :returns: The signing timestamp, in units of ten microseconds since 2015.
176    """
177    return mavlink_timestamp_from_unix_micros(unix_micros)
178
179
180def timestamp_now() -> int:
181    """Return a signing timestamp for now.
182
183    :returns: The signing timestamp matching the current clock.
184    """
185    return mavlink_timestamp_from_unix_micros(int(time.time() * 1_000_000))
186
187
188def frame(header: MavlinkHeader, msgid: int, payload: bytes) -> MavlinkFrame:
189    """Build a v2 frame carrying a message the common dialect defines.
190
191    The seed is looked up rather than passed, which is the usual case: a sender
192    emitting a standard message should not have to know its checksum constant.
193
194    :param header: The addressing fields to stamp on the frame.
195    :param msgid: The message id.
196    :param payload: The message payload.
197    :returns: The frame ready to send.
198    :raises ValueError: If the id is outside the common dialect, in which case
199        build the frame with :meth:`MavlinkFrame.raw` and a seed of your own.
200
201    >>> heartbeat = bytes([0, 0, 0, 0, 18, 0, 0, 4, 3])
202    >>> sent = frame(MavlinkHeader(1, 1), 0, heartbeat)
203    >>> sent.message_id
204    0
205    >>> MavlinkFrame.parse_known(sent.bytes).payload == heartbeat
206    True
207    """
208    crc_extra = mavlink_known_crc_extra(msgid)
209    if crc_extra is None:
210        raise ValueError(
211            f"message {msgid} is not in the common dialect; "
212            "supply its CRC_EXTRA with MavlinkFrame.raw"
213        )
214    return MavlinkFrame.encode_v2(header, msgid, payload, crc_extra)
215
216
217class FieldType(IntEnum):
218    """The field types a message definition uses.
219
220    A builder accepts either one of these or the name a dialect writes, so
221    ``FieldType.UINT32`` and ``"uint32_t"`` mean the same thing.
222    """
223
224    UINT8 = 1
225    INT8 = 2
226    CHAR = 3
227    UINT16 = 4
228    INT16 = 5
229    UINT32 = 6
230    INT32 = 7
231    UINT64 = 8
232    INT64 = 9
233    FLOAT = 10
234    DOUBLE = 11
235
236
237def schema_for(message: int | str) -> MessageSchema:
238    """Return the shape of a message the engine types.
239
240    :param message: The message id or name, such as ``33`` or
241        ``"GLOBAL_POSITION_INT"``.
242    :returns: The shape.
243    :raises ValueError: If this build does not type that message, in which case
244        describe it with :class:`MessageSchemaBuilder`.
245
246    >>> schema_for("GLOBAL_POSITION_INT").id
247    33
248    >>> schema_for(0).name
249    'HEARTBEAT'
250    """
251    if isinstance(message, int):
252        return MessageSchema.for_id(message)
253    return MessageSchema.for_name(message)
254
255
256def known_messages() -> list[str]:
257    """Return the names of every message this build types, in message-id order.
258
259    :returns: The message names, each usable with :func:`schema_for`.
260
261    >>> "HEARTBEAT" in known_messages()
262    True
263    """
264    return mavlink_known_messages()
265
266
267def message(shape: MessageSchema | int | str) -> MavlinkMessage:
268    """Create a message with every field zero.
269
270    :param shape: The shape to build, or the id or name of a message the engine
271        types.
272    :returns: The zeroed message, ready for its fields to be set.
273
274    >>> heartbeat = message("HEARTBEAT")
275    >>> heartbeat.set("type", 18)  # MAV_TYPE_ONBOARD_CONTROLLER
276    >>> heartbeat.set("system_status", 4)  # MAV_STATE_ACTIVE
277    >>> frame = heartbeat.to_frame(MavlinkHeader(1, 1))
278    >>> frame.message_id
279    0
280    """
281    if not isinstance(shape, MessageSchema):
282        shape = schema_for(shape)
283    return MavlinkMessage.empty(shape)
284
285
286def to_dict(built: MavlinkMessage, shape: MessageSchema) -> dict[str, object]:
287    """Read a whole message as plain values, keyed by field name.
288
289    A scalar field comes back as a number, an array field as a list, and a
290    ``char`` array as the text it carries, so a received message reads like an
291    ordinary mapping.
292
293    :param built: The message to read.
294    :param shape: The shape it was built from, which names its fields.
295    :returns: The fields as plain values.
296
297    >>> status = schema_for("STATUSTEXT")
298    >>> written = from_dict(status, {"severity": 4, "text": "battery low"})
299    >>> to_dict(written, status)["text"]
300    'battery low'
301    """
302    values: dict[str, object] = {}
303    for field in shape.fields:
304        if field.array_len == 0:
305            values[field.name] = built.get(field.name)
306        elif field.field_type == FieldType.CHAR:
307            values[field.name] = built.get_text(field.name)
308        else:
309            values[field.name] = [
310                built.get(field.name, index) for index in range(field.array_len)
311            ]
312    return values
313
314
315def from_dict(shape: MessageSchema, values: dict[str, object]) -> MavlinkMessage:
316    """Build a message from plain values, keyed by field name.
317
318    A field left out stays zero, which is what a sender filling in part of a
319    message wants.
320
321    :param shape: The shape to build.
322    :param values: The fields to set.
323    :returns: The message.
324    :raises ValueError: If a name is not a field of the message, or a value does
325        not fit its field.
326
327    >>> position = schema_for("GLOBAL_POSITION_INT")
328    >>> report = from_dict(position, {"lat": -33856780, "lon": 151215300})
329    >>> report.get_int("lat")
330    -33856780
331    """
332    built = MavlinkMessage.empty(shape)
333    for name, value in values.items():
334        if isinstance(value, str):
335            built.set_text(name, value)
336        elif isinstance(value, (bytes, bytearray)):
337            built.set_bytes(name, bytes(value))
338        elif isinstance(value, (list, tuple)):
339            for index, element in enumerate(value):
340                built.set(name, float(element), index)
341        else:
342            built.set(name, float(value))
343    return built
344
345
346class TypeMask(IntFlag):
347    """The fields of a setpoint the autopilot should act on.
348
349    Combine members with ``|``; the fields left out are ignored.
350    """
351
352    POSITION = 1
353    VELOCITY = 2
354    ACCELERATION = 4
355    YAW = 8
356    YAW_RATE = 16
357    FORCE = 32
358
359
360def type_mask(fields: TypeMask | int) -> int:
361    """Build a setpoint ``type_mask`` from the fields to use.
362
363    :param fields: The fields the autopilot should act on.
364    :returns: The mask, as the ``type_mask`` field of a setpoint carries it.
365
366    >>> type_mask(TypeMask.VELOCITY | TypeMask.YAW_RATE) == mavlink_offboard_type_mask(2 | 16)
367    True
368    """
369    return mavlink_offboard_type_mask(int(fields))
370
371
372def local_position(
373    header: MavlinkHeader,
374    time_boot_ms: int,
375    coordinate_frame: int,
376    target_system: int,
377    target_component: int,
378    x: float,
379    y: float,
380    z: float,
381) -> MavlinkFrame:
382    """Build a local-frame position setpoint, ready to send.
383
384    :param header: The addressing fields to stamp on the frame.
385    :param time_boot_ms: The sender's boot timestamp, in milliseconds.
386    :param coordinate_frame: The ``MAV_FRAME`` of the setpoint.
387    :param target_system: The target system id.
388    :param target_component: The target component id.
389    :param x: The position along x, in metres in the chosen frame.
390    :param y: The position along y.
391    :param z: The position along z.
392    :returns: The ``SET_POSITION_TARGET_LOCAL_NED`` frame.
393
394    >>> local_position(MavlinkHeader(255, 190), 1000, 1, 1, 1, 10.0, 0.0, -5.0).message_id
395    84
396    """
397    return mavlink_offboard_local_position(
398        header, time_boot_ms, coordinate_frame, target_system, target_component, x, y, z
399    )
400
401
402def local_velocity(
403    header: MavlinkHeader,
404    time_boot_ms: int,
405    coordinate_frame: int,
406    target_system: int,
407    target_component: int,
408    vx: float,
409    vy: float,
410    vz: float,
411) -> MavlinkFrame:
412    """Build a local-frame velocity setpoint, ready to send.
413
414    :param header: The addressing fields to stamp on the frame.
415    :param time_boot_ms: The sender's boot timestamp, in milliseconds.
416    :param coordinate_frame: The ``MAV_FRAME`` of the setpoint.
417    :param target_system: The target system id.
418    :param target_component: The target component id.
419    :param vx: The velocity along x, in metres per second in the chosen frame.
420    :param vy: The velocity along y.
421    :param vz: The velocity along z.
422    :returns: The ``SET_POSITION_TARGET_LOCAL_NED`` frame.
423    """
424    return mavlink_offboard_local_velocity(
425        header, time_boot_ms, coordinate_frame, target_system, target_component, vx, vy, vz
426    )
427
428
429def global_position(
430    header: MavlinkHeader,
431    time_boot_ms: int,
432    coordinate_frame: int,
433    target_system: int,
434    target_component: int,
435    lat_int: int,
436    lon_int: int,
437    alt: float,
438) -> MavlinkFrame:
439    """Build a global-frame position setpoint, ready to send.
440
441    :param header: The addressing fields to stamp on the frame.
442    :param time_boot_ms: The sender's boot timestamp, in milliseconds.
443    :param coordinate_frame: The ``MAV_FRAME`` of the setpoint.
444    :param target_system: The target system id.
445    :param target_component: The target component id.
446    :param lat_int: The latitude, in degrees times ten million.
447    :param lon_int: The longitude, in degrees times ten million.
448    :param alt: The altitude, in metres.
449    :returns: The ``SET_POSITION_TARGET_GLOBAL_INT`` frame.
450
451    >>> global_position(MavlinkHeader(255, 190), 1000, 6, 1, 1, -338567800, 1512153000, 50.0).message_id
452    86
453    """
454    return mavlink_offboard_global_position(
455        header,
456        time_boot_ms,
457        coordinate_frame,
458        target_system,
459        target_component,
460        lat_int,
461        lon_int,
462        alt,
463    )
class AckOutcome:

What an incoming acknowledgement means for the command in flight.

kind

"unrelated" if the ack was for another command, "in_progress" if the command is still running, or "final" if it finished.

value

The progress percent when in progress (255 when the autopilot does not report one), or the MAV_RESULT when final.

class CommandProtocol:

Tracks one command awaiting its acknowledgement.

def on_frame(self, /, frame):

Classifies an incoming frame against the command in flight.

Returns the outcome, or None if the frame is not a COMMAND_ACK.

def on_timeout(self, /):

Records a timeout and reports whether the command may be resent.

Returns the new confirmation count to stamp on the resend, or None once the retry budget is exhausted.

command

The command id being tracked.

confirmation

The confirmation count to stamp on the command being sent: zero for the first transmission, incremented on each retransmission.

DEFAULT_TIMESTAMP_WINDOW = 6000000
class Dialect:

The CRC_EXTRA seeds of a dialect beyond the common one.

Entries added here are consulted before the built-in common-dialect registry, so a private dialect may also override an id the common one defines.

def add(self, /, msgid, crc_extra):

Adds or replaces the seed for a message id.

def add_message(self, /, msgid, name, fields):

Adds a message by its definition, deriving the seed, and returns it.

This is the whole path for a vendor dialect: describe the message once, and every frame carrying it checks from then on.

def crc_extra(self, /, msgid):

Returns the seed this dialect resolves a message id to, or None if neither it nor the common dialect knows the id.

class FieldType(enum.IntEnum): 
218class FieldType(IntEnum):
219    """The field types a message definition uses.
220
221    A builder accepts either one of these or the name a dialect writes, so
222    ``FieldType.UINT32`` and ``"uint32_t"`` mean the same thing.
223    """
224
225    UINT8 = 1
226    INT8 = 2
227    CHAR = 3
228    UINT16 = 4
229    INT16 = 5
230    UINT32 = 6
231    INT32 = 7
232    UINT64 = 8
233    INT64 = 9
234    FLOAT = 10
235    DOUBLE = 11

The field types a message definition uses.

A builder accepts either one of these or the name a dialect writes, so FieldType.UINT32 and "uint32_t" mean the same thing.

UINT8 = <FieldType.UINT8: 1>
INT8 = <FieldType.INT8: 2>
CHAR = <FieldType.CHAR: 3>
UINT16 = <FieldType.UINT16: 4>
INT16 = <FieldType.INT16: 5>
UINT32 = <FieldType.UINT32: 6>
INT32 = <FieldType.INT32: 7>
UINT64 = <FieldType.UINT64: 8>
INT64 = <FieldType.INT64: 9>
FLOAT = <FieldType.FLOAT: 10>
DOUBLE = <FieldType.DOUBLE: 11>
KEY_LEN = 32
MAX_FRAME = 280
MAX_PAYLOAD = 255
MAX_RETRIES = 5
class MavlinkFieldInfo:

One field of a message shape.

name

The field name as the dialect writes it, such as custom_mode.

offset

The field's byte offset within the payload.

type_name

The field's type name as the dialect writes it, such as uint32_t.

array_len

The element count for an array field, or 0 for a scalar.

extension

Whether this is a MAVLink 2 extension field.

field_type

The field's type, one of the FieldType values.

class MavlinkFrame:

One MAVLink frame, assembled or received.

def encode_v2(header, msgid, payload, crc_extra):

Assembles a v2 frame carrying a message.

This is the current wire format and what a modern autopilot expects.

def encode_v1(header, msgid, payload, crc_extra):

Assembles a v1 frame, for a peer that predates MAVLink 2.

A v1 frame only carries message ids below 256.

def parse(data, crc_extra):

Parses one frame, checking it against a known CRC_EXTRA.

Raises ValueError if the bytes are not a whole frame or the checksum does not match, which is what rejects a frame mangled in transit.

def parse_known(data, dialect=None):

Parses one frame, looking its CRC_EXTRA up as it goes.

This is what a receiver holding many message types uses: the id comes out of the frame, and the seed comes from the dialect or the common registry.

def raw(header, msgid, crc_extra, payload):

Assembles a v2 frame carrying a message this build does not type.

The escape hatch a private dialect needs: supply the id, the payload, and the seed, and the frame is built and checked like any other.

header

The addressing fields the frame carries.

message_id

The id of the message the frame carries.

signature

The signature block, or None when the frame is not signed.

version

Which wire format this frame uses: 1 or 2.

incompat_flags

The incompatibility flags a v2 frame declares.

payload

The message payload.

A v2 frame drops trailing zero bytes, so a payload can arrive shorter than the message's full length; a decoder zero-extends it.

signed

Whether the frame carries a signature.

This says only that the frame was signed, not that the signature is good; a MavlinkVerifier decides that.

bytes

The whole frame, ready to put on the wire.

class MavlinkHeader:

The addressing fields a sender stamps on every frame.

component_id

The sending component's id.

sequence

The sender's sequence number, which wraps at 256.

system_id

The sending system's id.

class MavlinkMessage:

A message read and written by field name against a schema.

def empty(schema):

Creates a message with every field zero.

Raises ValueError if the shape does not fit a MAVLink payload.

def decode(schema, payload):

Reads a message out of a frame payload.

A payload shorter than the shape is zero-extended, as MAVLink 2 truncation requires, so a frame from a peer that trimmed trailing zeros or predates an extension field still decodes. Raises ValueError if the payload is longer than the shape describes.

def to_frame(self, /, header):

Builds a v2 frame carrying this message.

Raises ValueError if the message does not fit a frame.

def get(self, /, field, index=0):

Reads a field as a number.

Every field reads this way. An integer field wider than 53 bits can exceed what a float holds exactly; read those with get_int where the exact value matters.

def get_int(self, /, field, index=0):

Reads an integer field exactly, whatever its width or sign.

def set(self, /, field, value, index=0):

Writes a number into a field, converting it to the field's type.

A value bound for an integer field must be a whole number within that field's range, so a fractional or oversized value is refused rather than silently truncated.

def set_int(self, /, field, value, index=0):

Writes an integer into a field exactly, whatever its width or sign.

def get_bytes(self, /, field):

Copies the raw bytes of a byte-wide array field out, padding included.

def set_bytes(self, /, field, data):

Writes the raw bytes of a byte-wide array field, zero-padding the rest.

def get_text(self, /, field):

Reads a char array as text, stopping at the padding.

def set_text(self, /, field, text):

Writes text into a char array, padding the rest with zeros.

name

The name of the message this carries.

message_id

The id of the message this carries.

payload

The message's bytes as they go on the wire.

class MavlinkParser:

A streaming frame parser, and the frames it has completed.

def push(self, /, data, dialect=None):

Feeds bytes off a link and returns the frames that completed.

Whatever a serial port or socket delivers can be pushed as it arrives, however it is split. Noise between frames is skipped rather than reported, which is what lets a parser join a stream already in progress.

def feed(self, /, data, dialect=None):

Feeds bytes and queues what completed, for a caller that drains later.

def next_frame(self, /):

Takes the next queued frame, or None when the parser needs more bytes.

pending

How many completed frames are waiting to be taken.

class MavlinkSigner:

A signing key and the monotonic timestamp that goes with it.

def sign(self, /, header, msgid, payload, crc_extra):

Signs a message into a v2 frame.

Each call advances the timestamp, which is what makes a replayed frame detectable.

class MavlinkVerifier:

A signing key and the timestamps it has already accepted.

def set_window(self, /, window):

Sets how far a timestamp may run ahead of the last one accepted.

A wider window tolerates a noisier link; a narrower one narrows the chance of a replay landing inside it.

def verify(self, /, frame):

Checks a frame's signature and its place in the timestamp sequence.

Raises ValueError when the frame is unsigned, the signature does not match the key, or the timestamp has been seen before.

class MessageSchema:

The shape of one message: its id, name, seed, and fields.

def for_id(msgid):

Returns the shape of a message the engine types, by id.

Raises ValueError if this build does not type that id, which is what a builder is for.

def for_name(name):

Returns the shape of a message the engine types, by name.

Raises ValueError if this build does not type that name.

id

The id of the message this schema describes.

wire_len

The length of the message on the wire, in bytes, extensions included.

name

The name of the message this schema describes.

crc_extra

The CRC_EXTRA seed a frame carrying this message folds into its checksum.

fields

The fields in wire order: the base fields largest first, then extensions.

class MessageSchemaBuilder:

Describes a message this build does not type, one field at a time.

Fields are added in the order the message definition lists them; building puts them in wire order and derives the CRC_EXTRA seed from the result.

def field(self, /, name, field_type, array_len=0):

Adds a base field, in the order the definition declares it.

The type is either a FieldType value or the name a dialect writes, such as "uint32_t". Raises ValueError if it is neither, or if the shape has already been built.

def extension(self, /, name, field_type, array_len=0):

Adds a MAVLink 2 extension field, in the order the definition declares it.

Extensions keep their declared order, stay out of the CRC_EXTRA seed, and read as zero from a frame sent by a peer that predates them.

def build(self, /):

Puts the declared fields in wire order and finishes the shape.

Raises ValueError if two fields share a name, the fields do not fit a MAVLink payload, or the shape has already been built.

class MissionReceiver:

Requests a plan's items in order and collects them, ending with an acknowledgement.

def request_list(self, /, header):

Builds the MISSION_REQUEST_LIST frame that starts a download.

def on_frame(self, /, frame, header):

Handles an incoming frame, if it is one this transfer is waiting for.

A MISSION_COUNT opens the transfer and a MISSION_ITEM_INT advances it. Returns the step taken, or None if the frame carries a message this transfer does not handle.

expected

The next sequence number the receiver expects.

complete

Whether every item has been received and the acknowledgement produced.

class MissionSender:

Holds a plan and answers a receiver's requests for its items.

def add_item(self, /, item):

Appends an item to the plan, from a MISSION_ITEM_INT payload.

The sender stamps the sequence number, target ids, and mission type onto each item as it is handed out, so the item need only carry its content: command, frame, position, and parameters. Build one by field name with the MISSION_ITEM_INT schema and pass its payload.

def count(self, /, header):

Builds the MISSION_COUNT frame that opens an upload.

def on_frame(self, /, frame, header):

Handles an incoming frame, if it is one this transfer answers.

A MISSION_REQUEST_LIST is answered with the count, a MISSION_REQUEST_INT (or the older MISSION_REQUEST) with the item asked for, and a request past the end of the plan with a MISSION_ACK reporting an invalid sequence. A MISSION_ACK from the receiver ends the transfer. Returns None for a message this transfer does not handle.

class ReceiverStep:

What one incoming frame produced for a mission receiver.

accepted

The MISSION_ITEM_INT the frame carried, if it was the one expected next, as a message read by field name.

kind

What the receiver answered with: "request" for the next item, or "ack" to end the transfer.

reply

The frame to send back.

SIGNATURE_LEN = 13
class SenderStep:

What one incoming frame produced for a mission sender.

kind

What happened: "reply" when there is a frame to send, or "finished" when the receiver acknowledged the transfer.

result

The receiver's MAV_MISSION_RESULT, when the transfer finished.

reply

The frame to send back, when there is one.

class TypeMask(enum.IntFlag): 
347class TypeMask(IntFlag):
348    """The fields of a setpoint the autopilot should act on.
349
350    Combine members with ``|``; the fields left out are ignored.
351    """
352
353    POSITION = 1
354    VELOCITY = 2
355    ACCELERATION = 4
356    YAW = 8
357    YAW_RATE = 16
358    FORCE = 32

The fields of a setpoint the autopilot should act on.

Combine members with |; the fields left out are ignored.

POSITION = <TypeMask.POSITION: 1>
VELOCITY = <TypeMask.VELOCITY: 2>
ACCELERATION = <TypeMask.ACCELERATION: 4>
YAW = <TypeMask.YAW: 8>
YAW_RATE = <TypeMask.YAW_RATE: 16>
FORCE = <TypeMask.FORCE: 32>
def crc16(data: bytes) -> int: 
123def crc16(data: bytes) -> int:
124    """Return the CRC-16/MCRF4XX checksum of a byte string.
125
126    This is the checksum every MAVLink frame carries, exposed because a host
127    that implements part of the protocol itself needs the same arithmetic.
128
129    :param data: The data to checksum.
130    :returns: The checksum.
131    """
132    return mavlink_crc16_mcrf4xx(data)

Return the CRC-16/MCRF4XX checksum of a byte string.

This is the checksum every MAVLink frame carries, exposed because a host that implements part of the protocol itself needs the same arithmetic.

Parameters
  • data: The data to checksum. :returns: The checksum.
def frame(header: MavlinkHeader, msgid: int, payload: bytes) -> MavlinkFrame: 
189def frame(header: MavlinkHeader, msgid: int, payload: bytes) -> MavlinkFrame:
190    """Build a v2 frame carrying a message the common dialect defines.
191
192    The seed is looked up rather than passed, which is the usual case: a sender
193    emitting a standard message should not have to know its checksum constant.
194
195    :param header: The addressing fields to stamp on the frame.
196    :param msgid: The message id.
197    :param payload: The message payload.
198    :returns: The frame ready to send.
199    :raises ValueError: If the id is outside the common dialect, in which case
200        build the frame with :meth:`MavlinkFrame.raw` and a seed of your own.
201
202    >>> heartbeat = bytes([0, 0, 0, 0, 18, 0, 0, 4, 3])
203    >>> sent = frame(MavlinkHeader(1, 1), 0, heartbeat)
204    >>> sent.message_id
205    0
206    >>> MavlinkFrame.parse_known(sent.bytes).payload == heartbeat
207    True
208    """
209    crc_extra = mavlink_known_crc_extra(msgid)
210    if crc_extra is None:
211        raise ValueError(
212            f"message {msgid} is not in the common dialect; "
213            "supply its CRC_EXTRA with MavlinkFrame.raw"
214        )
215    return MavlinkFrame.encode_v2(header, msgid, payload, crc_extra)

Build a v2 frame carrying a message the common dialect defines.

The seed is looked up rather than passed, which is the usual case: a sender emitting a standard message should not have to know its checksum constant.

Parameters
  • header: The addressing fields to stamp on the frame.
  • msgid: The message id.
  • payload: The message payload. :returns: The frame ready to send.
Raises
  • ValueError: If the id is outside the common dialect, in which case build the frame with MavlinkFrame.raw() and a seed of your own.
>>> heartbeat = bytes([0, 0, 0, 0, 18, 0, 0, 4, 3])
>>> sent = frame(MavlinkHeader(1, 1), 0, heartbeat)
>>> sent.message_id
0
>>> MavlinkFrame.parse_known(sent.bytes).payload == heartbeat
True
def from_dict(shape: MessageSchema, values: dict[str, object]) -> MavlinkMessage: 
316def from_dict(shape: MessageSchema, values: dict[str, object]) -> MavlinkMessage:
317    """Build a message from plain values, keyed by field name.
318
319    A field left out stays zero, which is what a sender filling in part of a
320    message wants.
321
322    :param shape: The shape to build.
323    :param values: The fields to set.
324    :returns: The message.
325    :raises ValueError: If a name is not a field of the message, or a value does
326        not fit its field.
327
328    >>> position = schema_for("GLOBAL_POSITION_INT")
329    >>> report = from_dict(position, {"lat": -33856780, "lon": 151215300})
330    >>> report.get_int("lat")
331    -33856780
332    """
333    built = MavlinkMessage.empty(shape)
334    for name, value in values.items():
335        if isinstance(value, str):
336            built.set_text(name, value)
337        elif isinstance(value, (bytes, bytearray)):
338            built.set_bytes(name, bytes(value))
339        elif isinstance(value, (list, tuple)):
340            for index, element in enumerate(value):
341                built.set(name, float(element), index)
342        else:
343            built.set(name, float(value))
344    return built

Build a message from plain values, keyed by field name.

A field left out stays zero, which is what a sender filling in part of a message wants.

Parameters
  • shape: The shape to build.
  • values: The fields to set. :returns: The message.
Raises
  • ValueError: If a name is not a field of the message, or a value does not fit its field.
>>> position = schema_for("GLOBAL_POSITION_INT")
>>> report = from_dict(position, {"lat": -33856780, "lon": 151215300})
>>> report.get_int("lat")
-33856780
def global_position( header: MavlinkHeader, time_boot_ms: int, coordinate_frame: int, target_system: int, target_component: int, lat_int: int, lon_int: int, alt: float) -> MavlinkFrame: 
430def global_position(
431    header: MavlinkHeader,
432    time_boot_ms: int,
433    coordinate_frame: int,
434    target_system: int,
435    target_component: int,
436    lat_int: int,
437    lon_int: int,
438    alt: float,
439) -> MavlinkFrame:
440    """Build a global-frame position setpoint, ready to send.
441
442    :param header: The addressing fields to stamp on the frame.
443    :param time_boot_ms: The sender's boot timestamp, in milliseconds.
444    :param coordinate_frame: The ``MAV_FRAME`` of the setpoint.
445    :param target_system: The target system id.
446    :param target_component: The target component id.
447    :param lat_int: The latitude, in degrees times ten million.
448    :param lon_int: The longitude, in degrees times ten million.
449    :param alt: The altitude, in metres.
450    :returns: The ``SET_POSITION_TARGET_GLOBAL_INT`` frame.
451
452    >>> global_position(MavlinkHeader(255, 190), 1000, 6, 1, 1, -338567800, 1512153000, 50.0).message_id
453    86
454    """
455    return mavlink_offboard_global_position(
456        header,
457        time_boot_ms,
458        coordinate_frame,
459        target_system,
460        target_component,
461        lat_int,
462        lon_int,
463        alt,
464    )

Build a global-frame position setpoint, ready to send.

Parameters
  • header: The addressing fields to stamp on the frame.
  • time_boot_ms: The sender's boot timestamp, in milliseconds.
  • coordinate_frame: The MAV_FRAME of the setpoint.
  • target_system: The target system id.
  • target_component: The target component id.
  • lat_int: The latitude, in degrees times ten million.
  • lon_int: The longitude, in degrees times ten million.
  • alt: The altitude, in metres. :returns: The SET_POSITION_TARGET_GLOBAL_INT frame.
>>> global_position(MavlinkHeader(255, 190), 1000, 6, 1, 1, -338567800, 1512153000, 50.0).message_id
86
def known_crc_extra(msgid: int) -> int | None: 
157def known_crc_extra(msgid: int) -> int | None:
158    """Return the ``CRC_EXTRA`` the common dialect publishes for a message id.
159
160    :param msgid: The message id to look up.
161    :returns: The seed, or ``None`` for an id outside the common dialect, which
162        is what a :class:`Dialect` is for.
163
164    >>> known_crc_extra(0)
165    50
166    >>> known_crc_extra(9999) is None
167    True
168    """
169    return mavlink_known_crc_extra(msgid)

Return the CRC_EXTRA the common dialect publishes for a message id.

Parameters
  • msgid: The message id to look up. :returns: The seed, or None for an id outside the common dialect, which is what a Dialect is for.
>>> known_crc_extra(0)
50
>>> known_crc_extra(9999) is None
True
def known_messages() -> list[str]: 
257def known_messages() -> list[str]:
258    """Return the names of every message this build types, in message-id order.
259
260    :returns: The message names, each usable with :func:`schema_for`.
261
262    >>> "HEARTBEAT" in known_messages()
263    True
264    """
265    return mavlink_known_messages()

Return the names of every message this build types, in message-id order.

:returns: The message names, each usable with schema_for().

>>> "HEARTBEAT" in known_messages()
True
def local_position( header: MavlinkHeader, time_boot_ms: int, coordinate_frame: int, target_system: int, target_component: int, x: float, y: float, z: float) -> MavlinkFrame: 
373def local_position(
374    header: MavlinkHeader,
375    time_boot_ms: int,
376    coordinate_frame: int,
377    target_system: int,
378    target_component: int,
379    x: float,
380    y: float,
381    z: float,
382) -> MavlinkFrame:
383    """Build a local-frame position setpoint, ready to send.
384
385    :param header: The addressing fields to stamp on the frame.
386    :param time_boot_ms: The sender's boot timestamp, in milliseconds.
387    :param coordinate_frame: The ``MAV_FRAME`` of the setpoint.
388    :param target_system: The target system id.
389    :param target_component: The target component id.
390    :param x: The position along x, in metres in the chosen frame.
391    :param y: The position along y.
392    :param z: The position along z.
393    :returns: The ``SET_POSITION_TARGET_LOCAL_NED`` frame.
394
395    >>> local_position(MavlinkHeader(255, 190), 1000, 1, 1, 1, 10.0, 0.0, -5.0).message_id
396    84
397    """
398    return mavlink_offboard_local_position(
399        header, time_boot_ms, coordinate_frame, target_system, target_component, x, y, z
400    )

Build a local-frame position setpoint, ready to send.

Parameters
  • header: The addressing fields to stamp on the frame.
  • time_boot_ms: The sender's boot timestamp, in milliseconds.
  • coordinate_frame: The MAV_FRAME of the setpoint.
  • target_system: The target system id.
  • target_component: The target component id.
  • x: The position along x, in metres in the chosen frame.
  • y: The position along y.
  • z: The position along z. :returns: The SET_POSITION_TARGET_LOCAL_NED frame.
>>> local_position(MavlinkHeader(255, 190), 1000, 1, 1, 1, 10.0, 0.0, -5.0).message_id
84
def local_velocity( header: MavlinkHeader, time_boot_ms: int, coordinate_frame: int, target_system: int, target_component: int, vx: float, vy: float, vz: float) -> MavlinkFrame: 
403def local_velocity(
404    header: MavlinkHeader,
405    time_boot_ms: int,
406    coordinate_frame: int,
407    target_system: int,
408    target_component: int,
409    vx: float,
410    vy: float,
411    vz: float,
412) -> MavlinkFrame:
413    """Build a local-frame velocity setpoint, ready to send.
414
415    :param header: The addressing fields to stamp on the frame.
416    :param time_boot_ms: The sender's boot timestamp, in milliseconds.
417    :param coordinate_frame: The ``MAV_FRAME`` of the setpoint.
418    :param target_system: The target system id.
419    :param target_component: The target component id.
420    :param vx: The velocity along x, in metres per second in the chosen frame.
421    :param vy: The velocity along y.
422    :param vz: The velocity along z.
423    :returns: The ``SET_POSITION_TARGET_LOCAL_NED`` frame.
424    """
425    return mavlink_offboard_local_velocity(
426        header, time_boot_ms, coordinate_frame, target_system, target_component, vx, vy, vz
427    )

Build a local-frame velocity setpoint, ready to send.

Parameters
  • header: The addressing fields to stamp on the frame.
  • time_boot_ms: The sender's boot timestamp, in milliseconds.
  • coordinate_frame: The MAV_FRAME of the setpoint.
  • target_system: The target system id.
  • target_component: The target component id.
  • vx: The velocity along x, in metres per second in the chosen frame.
  • vy: The velocity along y.
  • vz: The velocity along z. :returns: The SET_POSITION_TARGET_LOCAL_NED frame.
def message(shape: MessageSchema | int | str) -> MavlinkMessage: 
268def message(shape: MessageSchema | int | str) -> MavlinkMessage:
269    """Create a message with every field zero.
270
271    :param shape: The shape to build, or the id or name of a message the engine
272        types.
273    :returns: The zeroed message, ready for its fields to be set.
274
275    >>> heartbeat = message("HEARTBEAT")
276    >>> heartbeat.set("type", 18)  # MAV_TYPE_ONBOARD_CONTROLLER
277    >>> heartbeat.set("system_status", 4)  # MAV_STATE_ACTIVE
278    >>> frame = heartbeat.to_frame(MavlinkHeader(1, 1))
279    >>> frame.message_id
280    0
281    """
282    if not isinstance(shape, MessageSchema):
283        shape = schema_for(shape)
284    return MavlinkMessage.empty(shape)

Create a message with every field zero.

Parameters
  • shape: The shape to build, or the id or name of a message the engine types. :returns: The zeroed message, ready for its fields to be set.
>>> heartbeat = message("HEARTBEAT")
>>> heartbeat.set("type", 18)  # MAV_TYPE_ONBOARD_CONTROLLER
>>> heartbeat.set("system_status", 4)  # MAV_STATE_ACTIVE
>>> frame = heartbeat.to_frame(MavlinkHeader(1, 1))
>>> frame.message_id
0
def message_crc_extra(name: str, fields: list[tuple[str, str, int]]) -> int: 
135def message_crc_extra(name: str, fields: list[tuple[str, str, int]]) -> int:
136    """Derive the ``CRC_EXTRA`` seed of a message from its definition.
137
138    This is what makes a dialect this build has never seen usable: given a
139    message's name and its base fields in wire order, the seed comes out the
140    same as the one the dialect publishes, and a frame carrying that message
141    then checks like any other.
142
143    Extension fields are excluded from the seed and must not be listed, which is
144    what lets a peer that predates them still check the frame.
145
146    :param name: The message name, such as ``HEARTBEAT``.
147    :param fields: The base fields in wire order, as ``(type, name, array_len)``
148        triples; ``array_len`` is ``0`` for a scalar.
149    :returns: The seed.
150
151    >>> message_crc_extra("PRIVATE_STATUS", [("uint32_t", "uptime", 0)]) >= 0
152    True
153    """
154    return mavlink_message_crc_extra(name, fields)

Derive the CRC_EXTRA seed of a message from its definition.

This is what makes a dialect this build has never seen usable: given a message's name and its base fields in wire order, the seed comes out the same as the one the dialect publishes, and a frame carrying that message then checks like any other.

Extension fields are excluded from the seed and must not be listed, which is what lets a peer that predates them still check the frame.

Parameters
  • name: The message name, such as HEARTBEAT.
  • fields: The base fields in wire order, as (type, name, array_len) triples; array_len is 0 for a scalar. :returns: The seed.
>>> message_crc_extra("PRIVATE_STATUS", [("uint32_t", "uptime", 0)]) >= 0
True
def schema_for(message: int | str) -> MessageSchema: 
238def schema_for(message: int | str) -> MessageSchema:
239    """Return the shape of a message the engine types.
240
241    :param message: The message id or name, such as ``33`` or
242        ``"GLOBAL_POSITION_INT"``.
243    :returns: The shape.
244    :raises ValueError: If this build does not type that message, in which case
245        describe it with :class:`MessageSchemaBuilder`.
246
247    >>> schema_for("GLOBAL_POSITION_INT").id
248    33
249    >>> schema_for(0).name
250    'HEARTBEAT'
251    """
252    if isinstance(message, int):
253        return MessageSchema.for_id(message)
254    return MessageSchema.for_name(message)

Return the shape of a message the engine types.

Parameters
  • message: The message id or name, such as 33 or "GLOBAL_POSITION_INT". :returns: The shape.
Raises
  • ValueError: If this build does not type that message, in which case describe it with MessageSchemaBuilder.
>>> schema_for("GLOBAL_POSITION_INT").id
33
>>> schema_for(0).name
'HEARTBEAT'
def timestamp_from_unix_micros(unix_micros: int) -> int: 
172def timestamp_from_unix_micros(unix_micros: int) -> int:
173    """Convert Unix time into the timestamp MAVLink signing counts in.
174
175    :param unix_micros: The time in microseconds since the Unix epoch.
176    :returns: The signing timestamp, in units of ten microseconds since 2015.
177    """
178    return mavlink_timestamp_from_unix_micros(unix_micros)

Convert Unix time into the timestamp MAVLink signing counts in.

Parameters
  • unix_micros: The time in microseconds since the Unix epoch. :returns: The signing timestamp, in units of ten microseconds since 2015.
def timestamp_now() -> int: 
181def timestamp_now() -> int:
182    """Return a signing timestamp for now.
183
184    :returns: The signing timestamp matching the current clock.
185    """
186    return mavlink_timestamp_from_unix_micros(int(time.time() * 1_000_000))

Return a signing timestamp for now.

:returns: The signing timestamp matching the current clock.

def to_dict(built: MavlinkMessage, shape: MessageSchema) -> dict[str, object]: 
287def to_dict(built: MavlinkMessage, shape: MessageSchema) -> dict[str, object]:
288    """Read a whole message as plain values, keyed by field name.
289
290    A scalar field comes back as a number, an array field as a list, and a
291    ``char`` array as the text it carries, so a received message reads like an
292    ordinary mapping.
293
294    :param built: The message to read.
295    :param shape: The shape it was built from, which names its fields.
296    :returns: The fields as plain values.
297
298    >>> status = schema_for("STATUSTEXT")
299    >>> written = from_dict(status, {"severity": 4, "text": "battery low"})
300    >>> to_dict(written, status)["text"]
301    'battery low'
302    """
303    values: dict[str, object] = {}
304    for field in shape.fields:
305        if field.array_len == 0:
306            values[field.name] = built.get(field.name)
307        elif field.field_type == FieldType.CHAR:
308            values[field.name] = built.get_text(field.name)
309        else:
310            values[field.name] = [
311                built.get(field.name, index) for index in range(field.array_len)
312            ]
313    return values

Read a whole message as plain values, keyed by field name.

A scalar field comes back as a number, an array field as a list, and a char array as the text it carries, so a received message reads like an ordinary mapping.

Parameters
  • built: The message to read.
  • shape: The shape it was built from, which names its fields. :returns: The fields as plain values.
>>> status = schema_for("STATUSTEXT")
>>> written = from_dict(status, {"severity": 4, "text": "battery low"})
>>> to_dict(written, status)["text"]
'battery low'
def type_mask(fields: TypeMask | int) -> int: 
361def type_mask(fields: TypeMask | int) -> int:
362    """Build a setpoint ``type_mask`` from the fields to use.
363
364    :param fields: The fields the autopilot should act on.
365    :returns: The mask, as the ``type_mask`` field of a setpoint carries it.
366
367    >>> type_mask(TypeMask.VELOCITY | TypeMask.YAW_RATE) == mavlink_offboard_type_mask(2 | 16)
368    True
369    """
370    return mavlink_offboard_type_mask(int(fields))

Build a setpoint type_mask from the fields to use.

Parameters
  • fields: The fields the autopilot should act on. :returns: The mask, as the type_mask field of a setpoint carries it.
>>> type_mask(TypeMask.VELOCITY | TypeMask.YAW_RATE) == mavlink_offboard_type_mask(2 | 16)
True