4. 数据采集脚本编写：Python pymavlink库入门、连接SITL仿真、订阅并打印MAVLink消息

好，到了真正动手写代码的时候了。

前面几章我们把SITL跑起来了，也看到了MAVLink消息在终端里刷刷地过。但光看不够，咱们得把数据抓到自己手里，存下来，才能做后续的质量分析。这一章，我就带你从零开始，用Python和pymavlink库，把飞控的数据“接”过来。

4.1 为什么选pymavlink？

说白了，pymavlink就是MAVLink协议的Python实现。你想想看，飞控和地面站之间通信，靠的就是MAVLink协议。pymavlink能帮我们做两件事：解析和生成MAVLink消息。

我个人习惯用它来做数据采集，原因有三：

  • 纯Python：跨平台，装个pip就能用，不用折腾C++编译。
  • 协议全覆盖：从ArduPilot到PX4，常见的消息类型它都支持。
  • 自带解析器：你不需要自己手动去拆二进制包，它直接给你转成Python对象。
小提示： 如果你用的是ArduPilot的SITL，pymavlink几乎是标配。PX4那边虽然也有mavsdk，但pymavlink在底层数据采集上更灵活，适合做质量分析。

4.2 安装与第一个连接脚本

安装很简单，一行命令搞定：

pip install pymavlink

装好之后，我们来写第一个脚本。它的任务很简单：连上SITL，然后打印出飞控的心跳包（HEARTBEAT）。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
第一个pymavlink脚本：连接SITL并接收心跳
"""

import time
from pymavlink import mavutil

# 1. 创建一个连接
# SITL默认的UDP端口是14550
master = mavutil.mavlink_connection('udp:127.0.0.1:14550')

# 2. 等待心跳包
# 这一步很重要，它确认了飞控和脚本之间的通信是通的
print("等待飞控心跳...")
master.wait_heartbeat()
print("心跳已收到！")
print(f"飞控系统ID: {master.target_system}, 组件ID: {master.target_component}")

# 3. 持续接收并打印消息
print("开始接收消息，按Ctrl+C退出...")
try:
    while True:
        # 接收一个消息，超时时间1秒
        msg = master.recv_match(blocking=True, timeout=1)
        if msg:
            print(f"收到消息: {msg.get_type()}")
            # 如果是心跳包，打印详细信息
            if msg.get_type() == 'HEARTBEAT':
                print(f"  类型: {msg.type}, 状态: {msg.system_status}")
        else:
            print("等待消息...")
        time.sleep(0.1)
except KeyboardInterrupt:
    print("用户中断，退出。")

运行这个脚本之前，记得先启动SITL。我在项目中遇到过好几次，新手最容易犯的错就是忘了开SITL，然后脚本卡在wait_heartbeat()那里不动。

4.3 理解连接方式：UDP vs TCP vs 串口

刚才我们用udp:127.0.0.1:14550连接。但实际场景中，连接方式不止这一种。我整理了一个表格，方便你对照：

连接方式 连接字符串示例 适用场景 特点
UDP udp:127.0.0.1:14550 SITL仿真、本地地面站 速度快，无连接状态，丢包不重传
TCP tcp:127.0.0.1:5760 远程连接、需要可靠传输 有握手，保证数据顺序，延迟略高
串口 /dev/ttyUSB0:57600 真实飞控、数传电台 波特率需匹配，容易受干扰
注意： 在SITL仿真中，不要混用UDP和TCP端口。ArduPilot的SITL默认用UDP 14550输出MAVLink，如果你用TCP去连，是连不上的。我曾经调试一个脚本，花了半小时才发现是端口类型搞错了。

4.4 订阅特定消息：别什么都收

刚才的脚本是“来什么收什么”。但在实际数据采集中，我们只关心特定的消息，比如GPS位置、姿态角、电池电压等。如果什么都收，数据量太大，而且大部分是没用的。

pymavlink提供了recv_match方法，可以按条件过滤。看这个例子：

# 只接收GPS_RAW_INT消息
msg = master.recv_match(type='GPS_RAW_INT', blocking=True)
if msg:
    print(f"纬度: {msg.lat}, 经度: {msg.lon}, 海拔: {msg.alt}")

# 接收多个类型
msg = master.recv_match(type=['GPS_RAW_INT', 'ATTITUDE'], blocking=True)
if msg:
    if msg.get_type() == 'GPS_RAW_INT':
        print(f"GPS: {msg.lat}, {msg.lon}")
    elif msg.get_type() == 'ATTITUDE':
        print(f"姿态: 滚转{msg.roll:.2f}, 俯仰{msg.pitch:.2f}, 偏航{msg.yaw:.2f}")

这里有个小技巧：blocking=True表示如果没有收到消息，脚本会一直等。如果你不想等，可以设成False，然后自己加个循环和超时判断。

4.5 数据采集的核心逻辑：循环 + 过滤 + 存储

一个合格的数据采集脚本，结构应该是这样的：

  1. 初始化连接：建立与SITL的通信链路。
  2. 等待心跳：确认飞控在线。
  3. 进入主循环：持续接收消息。
  4. 消息过滤：只保留我们关心的消息类型。
  5. 数据提取：从消息对象中取出字段值。
  6. 数据存储：写入文件、数据库或打印到终端。

下面是一个更完整的示例，它订阅了GPS、姿态和电池消息，并打印到终端：

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
数据采集脚本：订阅GPS、姿态、电池消息
"""

import time
from pymavlink import mavutil

# 连接SITL
master = mavutil.mavlink_connection('udp:127.0.0.1:14550')
master.wait_heartbeat()
print("连接成功，开始采集...")

# 定义我们关心的消息类型
TARGET_TYPES = ['GPS_RAW_INT', 'ATTITUDE', 'BATTERY_STATUS']

try:
    while True:
        # 只接收目标类型，阻塞等待
        msg = master.recv_match(type=TARGET_TYPES, blocking=True)
        if msg is None:
            continue

        msg_type = msg.get_type()
        timestamp = time.time()

        if msg_type == 'GPS_RAW_INT':
            print(f"[{timestamp:.2f}] GPS: lat={msg.lat/1e7:.6f}, "
                  f"lon={msg.lon/1e7:.6f}, alt={msg.alt/1e3:.1f}m, "
                  f"hdop={msg.eph:.1f}")

        elif msg_type == 'ATTITUDE':
            print(f"[{timestamp:.2f}] ATT: roll={msg.roll:.3f}, "
                  f"pitch={msg.pitch:.3f}, yaw={msg.yaw:.3f}")

        elif msg_type == 'BATTERY_STATUS':
            voltage = msg.voltages[0] / 1000.0  # 毫伏转伏
            current = msg.current_battery / 100.0  # 毫安转安
            print(f"[{timestamp:.2f}] BAT: voltage={voltage:.2f}V, "
                  f"current={current:.2f}A, "
                  f"remaining={msg.battery_remaining}%")

except KeyboardInterrupt:
    print("采集结束。")

你运行一下试试。如果SITL在跑，你会看到类似这样的输出：

[1678901234.56] GPS: lat=47.397742, lon=8.545594, alt=488.0m, hdop=0.8
[1678901234.57] ATT: roll=0.001, pitch=-0.002, yaw=1.234
[1678901234.58] BAT: voltage=12.50V, current=0.00A, remaining=100%

嗯，看到数据在刷新，心里就踏实了。

4.6 避坑指南：我踩过的三个坑

写采集脚本的时候，有几个地方特别容易出问题。我把自己踩过的坑列出来，你注意避开：

  • 坑一：端口被占用。如果你同时开了Mission Planner或者QGroundControl，它们会占用14550端口。解决办法是：关掉地面站，或者让SITL输出到另一个端口（比如加个-O 14551参数）。
  • 坑二：消息频率太高。SITL默认的消息频率很高，尤其是IMU数据。如果你不加过滤地接收所有消息，Python脚本可能处理不过来，导致数据积压。我建议你只订阅需要的消息，并且可以在循环里加个time.sleep(0.01)来降速。
  • 坑三：忘记处理异常。网络断开、飞控重启，都可能导致连接中断。脚本里最好加个try-except，捕获ConnectionError或OSError，然后自动重连。

4.7 本章知识体系

为了让你更直观地理解本章的内容结构，我画了一张流程图：

数据采集脚本编写：知识体系 安装pymavlink 连接SITL (UDP/TCP) 等待心跳 (wait_heartbeat) 主循环 (recv_match) 消息过滤 + 字段提取 打印 / 写入文件 关键点 • 只订阅需要的消息 • 注意端口冲突 • 加异常处理 • 控制采集频率

4.8 本章小结

这一章我们干了三件事：

  • 学会了用pymavlink连接SITL仿真。
  • 掌握了如何订阅并过滤MAVLink消息。
  • 写了一个能跑的数据采集脚本雏形。

你现在应该能自己动手，把飞控的GPS、姿态、电池数据抓到手了。别小看这一步，它是后面所有数据质量分析的基础。数据都拿不到，分析个啥？

← 上一章 📖 返回目录 下一章 →