mavtools

Still under construction

Storage

Messages are stored in a pickle file as:

MavMsg = namedtuple("MavMsg", "timestamp msg")
# timestamp : float  (time.time() when the frame was received)
# msg       : bytes  (complete raw MAVLink frame, no decoding)

The messages live in a deque (MavBuffer) and converted to a list when saved. The buffer can be written to / read from a pickle file (streaming supported). Playback respects the original inter-message timing.

This also makes it super simple to read these messages in Jupyter / Maxismo or other Python programs.

Installation

# local for development
pip install -e .

# pull from pypi
pip install -U mavtools

Tools

QUESTION: Can I combine some or all of these into one program? Or is that too complex?

Quick start

Record from UDP

from mavtools import MavBuffer, UDPTransport, Recorder

buf = MavBuffer()
with UDPTransport(local_addr=("0.0.0.0", 14550)) as transport:
    rec = Recorder(transport, buf)
    rec.run(duration=30.0)          # record for 30 seconds

buf.save("flight.pkl")
print(f"Recorded {len(buf)} messages")

Record from serial

from mavtools import MavBuffer, SerialTransport, Recorder

buf = MavBuffer()
with SerialTransport("/dev/ttyUSB0", baudrate=57600) as transport:
    rec = Recorder(transport, buf)
    rec.run(duration=60.0)

buf.save("serial_flight.pkl")

Play back

from mavtools import MavBuffer, UDPTransport, Player

buf = MavBuffer.from_file("flight.pkl")

with UDPTransport(remote_addr=("127.0.0.1", 14550)) as transport:
    player = Player(buf, transport)
    player.run(speed=1.0)           # real-time
    # player.run(speed=2.0)         # twice as fast
    # player.run(loop=True)         # continuous loop

Streaming write while recording

from mavtools import MavBuffer, UDPTransport, Recorder, MavMsg
import time

buf = MavBuffer()
path = "live.pkl"

with UDPTransport(local_addr=("0.0.0.0", 14550)) as t:
    rec = Recorder(t, buf)
    # In a real application you would run the recorder in a thread
    # and periodically call:
    #   for m in list(buf):
    #       buf.append_to_file(path, m)
    #       ...

API overview

Notes

• On UDP each datagram is usually one MAVLink frame; the framer is still applied for robustness.
• On serial the framer is essential because the link is a continuous byte stream.
• The framer does not verify the CRC or decode the payload – it only finds frame boundaries.
• Timestamps are wall-clock (time.time()). Playback computes deltas relative to the first message.
• MavBuffer.save(..., append=True) and append_to_file() allow true streaming writes.

ToDo

• Look at alternatives to pickle to work with other languages
• Include calibration tools

MIT License

Copyright (c) 2026 Kevin Walchko

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.