4. Python MAVSDK入门：安装pymavlink、创建连接、心跳检测

好，咱们正式开始动手了。

前面几章讲了MAVLink协议的结构、消息格式，还有SITL怎么跑起来。但光看不练，那是纸上谈兵。这一章，我带你用Python把MAVLink的通信链路真正打通。

说白了，就是装个库、连上SITL、然后看看心跳包是不是在跳。这是所有后续分析的基础，你想想看，连心跳都收不到，后面那些复杂的消息流根本无从谈起。

4.1 为什么选pymavlink？

Python生态里操作MAVLink的库有好几个，比如dronekit、pymavlink、MAVSDK-Python。我个人习惯用pymavlink作为底层通信工具。

原因很简单：它最贴近协议本身。pymavlink是MAVLink协议的官方Python实现，你能直接操作每一个字节、每一个字段。我在项目中遇到过好几次，上层库封装得太好，出了问题根本不知道底层在传什么。用pymavlink，你能看到最原始的数据流。

核心定位：pymavlink = MAVLink协议的Python原生绑定，适合做底层数据流分析、协议调试、自定义消息处理。

4.2 安装pymavlink

安装过程其实没什么坑，但有几个细节我提醒一下。

4.2.1 基础安装

pip install pymavlink

嗯，就这么一行。但如果你是在Linux或者macOS上，可能会遇到编译问题。pymavlink依赖一些C扩展，比如mavnative，用来加速消息解析。

我曾经在Ubuntu 20.04上直接pip install，结果报错说缺少Python头文件。解决办法是先把开发包装上：

sudo apt-get install python3-dev
pip install pymavlink

4.2.2 验证安装

装完之后，打开Python交互环境，跑一下这段代码：

import pymavlink
print(pymavlink.__version__)

如果能正常输出版本号，比如2.4.37，那就说明装好了。如果报错ModuleNotFoundError，检查一下你是不是在正确的虚拟环境里。

我建议：用虚拟环境管理项目依赖。别一股脑装到系统Python里，后面项目多了容易打架。

python -m venv mavlink_env
source mavlink_env/bin/activate  # Linux/macOS
mavlink_env\Scripts\activate     # Windows
pip install pymavlink

4.3 创建MAVLink连接

装好库，接下来就是连上SITL。SITL默认在UDP端口14550上广播MAVLink数据。我们写个Python脚本，作为地面站去接收这些数据。

4.3.1 连接SITL的UDP端口

先确保SITL已经在运行。如果你忘了怎么启动，回顾一下第三章：

sim_vehicle.py -v ArduCopter --map --console

然后，新建一个Python文件，比如mavlink_connect.py：

#!/usr/bin/env python3
"""
MAVLink连接示例：连接SITL并接收心跳
"""

import time
from pymavlink import mavutil

# 创建连接，指向SITL的UDP端口
# SITL默认在14550端口广播
master = mavutil.mavlink_connection('udp:127.0.0.1:14550')

print("正在等待心跳...")

# 等待第一个心跳包，超时10秒
master.wait_heartbeat(timeout=10)

print(f"收到心跳！系统ID: {master.target_system}, 组件ID: {master.target_component}")

跑一下看看：

python mavlink_connect.py

如果一切正常，你会看到类似这样的输出：

正在等待心跳...
收到心跳！系统ID: 1, 组件ID: 1

这里有个关键点：mavutil.mavlink_connection是pymavlink提供的统一连接入口。它支持多种协议：

协议	连接字符串示例	典型场景
UDP	`udp:127.0.0.1:14550`	连接本地SITL
TCP	`tcp:127.0.0.1:5760`	连接真实飞控（TCP模式）
串口	`/dev/ttyUSB0:57600`	连接真实飞控（串口模式）

4.4 心跳检测机制

收到心跳，说明链路通了。但心跳到底是什么？为什么它这么重要？

4.4.1 心跳消息结构

MAVLink的HEARTBEAT消息（消息ID: 0）是所有通信的起点。它告诉地面站：

我还在（系统在线）
我是什么（系统类型、自动驾驶仪类型）
我当前的状态（飞行模式、系统状态）

它的频率通常是1Hz，也就是每秒发一次。如果超过一定时间没收到心跳，地面站就应该认为连接已断开。

4.4.2 解析心跳数据

光说“收到心跳”还不够，我们得看看里面到底有什么。修改一下上面的脚本：

#!/usr/bin/env python3
"""
解析心跳消息的各个字段
"""

import time
from pymavlink import mavutil

master = mavutil.mavlink_connection('udp:127.0.0.1:14550')

# 等待心跳
msg = master.wait_heartbeat(timeout=10)
if msg is None:
    print("超时，未收到心跳")
    exit(1)

# 解析心跳字段
print("=== 心跳消息解析 ===")
print(f"系统类型: {msg.type} (MAV_TYPE)")
print(f"自动驾驶仪类型: {msg.autopilot} (MAV_AUTOPILOT)")
print(f"系统状态: {msg.system_status} (MAV_STATE)")
print(f"MAVLink协议版本: {msg.mavlink_version}")
print(f"自定义模式(飞行模式): {msg.custom_mode}")

# 用pymavlink的枚举类转成可读文本
from pymavlink.dialects.v20 import ardupilotmega as mavlink2

type_name = mavlink2.enums['MAV_TYPE'][msg.type].name
autopilot_name = mavlink2.enums['MAV_AUTOPILOT'][msg.autopilot].name
status_name = mavlink2.enums['MAV_STATE'][msg.system_status].name

print(f"\n可读信息:")
print(f"  系统类型: {type_name}")
print(f"  自动驾驶仪: {autopilot_name}")
print(f"  系统状态: {status_name}")

跑一下，输出大概是这样：

=== 心跳消息解析 ===
系统类型: 2 (MAV_TYPE)
自动驾驶仪类型: 3 (MAV_AUTOPILOT)
系统状态: 4 (MAV_STATE)
MAVLink协议版本: 2
自定义模式(飞行模式): 0

可读信息:
  系统类型: MAV_TYPE_QUADROTOR
  自动驾驶仪: MAV_AUTOPILOT_ARDUPILOTMEGA
  系统状态: MAV_STATE_ACTIVE

我建议：在实际项目中，不要只依赖wait_heartbeat一次。你应该在循环里持续监听心跳，并记录最后一次收到的时间。如果超过5秒没收到，就触发重连逻辑。我在做无人机集群项目时，就是靠这个机制判断节点是否掉线的。

4.5 持续监听心跳流

真实场景下，我们需要持续接收消息，而不是等一次就结束。下面这个脚本会一直监听，每秒打印一次心跳信息：

#!/usr/bin/env python3
"""
持续监听心跳流，每秒打印一次
"""

import time
from pymavlink import mavutil

master = mavutil.mavlink_connection('udp:127.0.0.1:14550')

# 先等第一个心跳建立连接
master.wait_heartbeat(timeout=10)
print("连接建立，开始持续监听...\n")

last_heartbeat_time = time.time()

while True:
    # 非阻塞接收消息
    msg = master.recv_match(type='HEARTBEAT', blocking=False)
    
    if msg is not None:
        now = time.time()
        interval = now - last_heartbeat_time
        last_heartbeat_time = now
        
        # 获取飞行模式（ArduPilot专用）
        mode = mavutil.mode_string_v10(msg)
        
        print(f"[{now:.1f}] 心跳间隔: {interval:.2f}s | 模式: {mode} | 状态: {msg.system_status}")
    
    # 检查是否超时
    if time.time() - last_heartbeat_time > 5:
        print("警告：超过5秒未收到心跳，可能连接已断开！")
        break
    
    time.sleep(0.1)  # 避免CPU跑满

注意：recv_match的blocking=False参数很关键。如果设为True（默认值），程序会卡在那里等消息，你就没法做超时检测了。我刚开始写的时候就被这个坑过，脚本跑起来像死机了一样。

4.6 本章知识结构图

下面这张图总结了从安装到心跳检测的完整流程：

4.7 常见问题与避坑

最后，我总结几个新手容易踩的坑：

SITL没启动就运行脚本：wait_heartbeat会一直等，直到超时。先确认SITL的终端窗口在正常运行。
端口被占用：如果之前有程序占用了14550端口，连接会失败。用lsof -i :14550（Linux/macOS）或netstat -ano | findstr :14550（Windows）检查。
防火墙拦截：UDP通信可能被防火墙拦截。本地调试一般没问题，但如果连远程飞控，记得开放端口。
消息类型拼写错误：recv_match(type='HEARTBEAT')里的类型名必须全大写，写成'Heartbeat'或'heartbeat'都收不到。

我个人习惯：写一个mavlink_utils.py工具模块，把连接、心跳检测、超时重连这些基础功能封装成函数。后面30章的分析脚本，都可以直接复用这些基础函数，不用每次都重写连接逻辑。

好，这一章就到这里。装好环境、连上SITL、看到心跳在跳，你的MAVLink数据流分析之旅，算是正式迈出了第一步。

📚 课程目录

4. Python MAVSDK入门：安装pymavlink、创建连接、心跳检测

4.1 为什么选pymavlink？

4.2 安装pymavlink

4.2.1 基础安装

4.2.2 验证安装

4.3 创建MAVLink连接

4.3.1 连接SITL的UDP端口

4.4 心跳检测机制

4.4.1 心跳消息结构

4.4.2 解析心跳数据

4.5 持续监听心跳流

4.6 本章知识结构图

4.7 常见问题与避坑