4、MAVSDK与地面站通信：使用Python MAVSDK连接仿真飞控，实现心跳检测与参数读取

好，我们进入第四章。前面几章我们把仿真环境搭起来了，也跑通了PX4的软件在环仿真。现在，得让地面站和飞控真正“说上话”。

这一章的核心，就是用Python MAVSDK这个库，去连接我们跑在仿真里的飞控。说白了，就是写几行代码，让电脑上的地面站程序能收到飞控的“心跳”，还能把飞控的参数读出来。

我个人习惯，在开始写任何通信代码之前，先搞清楚数据是怎么流的。你想想看，如果连数据流向都没概念，出了问题你都不知道该查哪一头。

4.1 通信链路全景：MAVSDK站在哪？

我们先画一张图，把整个通信链路看清楚。飞控（不管是真机还是仿真）通过串口或UDP发出MAVLink数据包。MAVSDK就像个翻译官，把这些二进制数据包翻译成Python里的对象和方法。

嗯，这里要注意，MAVSDK不是直接操作串口的。它内部封装了MAVLink协议的解析和组包。你调用一个param.get_param_int()，它自动帮你组装好请求参数的命令包，发出去，再解析飞控返回的响应包。

从这张图你能看到，MAVSDK是中间层。它把底层的MAVLink协议细节藏起来了。你不需要手动去解析HEARTBEAT消息的字节流，MAVSDK直接给你一个system.heartbeat事件。

4.2 环境准备：安装Python MAVSDK

动手之前，先把工具装好。我建议你用虚拟环境，别把系统Python搞乱了。曾经有个同事，直接在系统环境里装了一堆库，结果版本冲突，折腾了一下午。

安装MAVSDK很简单，一行命令：

pip install mavsdk

它会自动把依赖的grpcio、protobuf这些库一起装上。如果你在安装过程中遇到grpcio编译报错，多半是C++编译环境的问题。Windows用户装个Visual Studio Build Tools就行，Mac用户装Xcode Command Line Tools。

4.3 连接仿真飞控：建立通信链路

仿真飞控默认在UDP端口14540上监听。MAVSDK连接时，需要指定这个地址。代码很简单，但有几个坑我得提前跟你说。

import asyncio
from mavsdk import System

async def connect_to_drone():
    drone = System()
    # 连接到仿真飞控
    await drone.connect(system_address="udp://:14540")
    
    print("正在等待飞控连接...")
    # 等待飞控上线
    async for state in drone.core.connection_state():
        if state.is_connected:
            print("飞控已连接！")
            break

if __name__ == "__main__":
    asyncio.run(connect_to_drone())

这段代码做了三件事：创建System对象、发起UDP连接、等待连接状态确认。注意那个async for，MAVSDK的API几乎全是异步的。你如果不熟悉asyncio，可能会觉得别扭。但习惯了就好，异步的好处是不会阻塞主线程。

4.4 心跳检测：验证链路是否活着

连接上了，怎么确认链路是通的？看心跳。飞控每隔1秒会发一个HEARTBEAT消息。MAVSDK把这个封装成了telemetry.health()和telemetry.armed()等事件。

我个人习惯，在正式干活之前，先写一个心跳检测循环。这就像医生给病人测脉搏，先确认人还活着，再谈其他检查。

async def check_heartbeat(drone):
    print("开始检测心跳...")
    async for health in drone.telemetry.health():
        print(f"陀螺仪: {health.is_gyrometer_calibration_ok}")
        print(f"加速度计: {health.is_accelerometer_calibration_ok}")
        print(f"磁力计: {health.is_magnetometer_calibration_ok}")
        print(f"气压计: {health.is_barometer_calibration_ok}")
        print(f"全局定位: {health.is_global_position_ok}")
        print(f"本地定位: {health.is_local_position_ok}")
        print("---")
        # 只检测一次就退出，避免刷屏
        break

这里我只检测了一次就break了。实际项目中，你可能需要持续监控心跳状态。比如，如果连续5秒没收到心跳，就触发报警。

健康状态字段	含义	仿真环境典型值
is_gyrometer_calibration_ok	陀螺仪校准状态	True
is_accelerometer_calibration_ok	加速度计校准状态	True
is_magnetometer_calibration_ok	磁力计校准状态	True
is_barometer_calibration_ok	气压计校准状态	True
is_global_position_ok	全球定位（GPS）状态	False（仿真默认无GPS）
is_local_position_ok	本地定位状态	True

你看，仿真环境下GPS是False，这很正常。因为SITL默认不模拟GPS信号。如果你需要GPS，得额外启动一个GPS仿真模块。

4.5 参数读取：从飞控获取配置信息

心跳确认链路通了，下一步就是读参数。飞控的参数成百上千，从SYS_AUTOSTART到MC_ROLL_P，都是键值对的形式。

MAVSDK读取参数有两种方式：按名字读单个参数，或者一次性拉取所有参数。我建议你按需读取，别一次性拉几千个参数，网络开销大，还容易超时。

async def read_parameters(drone):
    # 读取单个参数
    sys_autostart = await drone.param.get_param_int("SYS_AUTOSTART")
    print(f"SYS_AUTOSTART: {sys_autostart}")
    
    # 读取浮点参数
    mc_roll_p = await drone.param.get_param_float("MC_ROLL_P")
    print(f"MC_ROLL_P: {mc_roll_p}")
    
    # 读取所有参数（慎用）
    all_params = await drone.param.get_all_params()
    print(f"共读取到 {len(all_params.int_params) + len(all_params.float_params)} 个参数")

参数读取的返回值类型要搞清楚。get_param_int返回整数，get_param_float返回浮点数。如果你用错了类型，MAVSDK会抛异常。我曾经在读取BAT1_CAPACITY时用了get_param_int，结果它是个浮点参数，程序直接崩了。

4.6 完整示例：心跳检测+参数读取

把上面的代码串起来，写一个完整的脚本。这个脚本会连接飞控、检测心跳、读取几个关键参数，然后退出。

import asyncio
from mavsdk import System

async def main():
    drone = System()
    await drone.connect(system_address="udp://:14540")
    
    print("等待飞控连接...")
    async for state in drone.core.connection_state():
        if state.is_connected:
            print("飞控已连接")
            break
    
    # 心跳检测
    print("\n=== 心跳检测 ===")
    async for health in drone.telemetry.health():
        print(f"陀螺仪: {health.is_gyrometer_calibration_ok}")
        print(f"加速度计: {health.is_accelerometer_calibration_ok}")
        print(f"磁力计: {health.is_magnetometer_calibration_ok}")
        print(f"气压计: {health.is_barometer_calibration_ok}")
        print(f"全局定位: {health.is_global_position_ok}")
        print(f"本地定位: {health.is_local_position_ok}")
        break
    
    # 参数读取
    print("\n=== 参数读取 ===")
    sys_autostart = await drone.param.get_param_int("SYS_AUTOSTART")
    print(f"SYS_AUTOSTART: {sys_autostart}")
    
    mc_roll_p = await drone.param.get_param_float("MC_ROLL_P")
    print(f"MC_ROLL_P: {mc_roll_p}")
    
    bat_capacity = await drone.param.get_param_float("BAT1_CAPACITY")
    print(f"BAT1_CAPACITY: {bat_capacity} mAh")
    
    print("\n通信测试完成！")

if __name__ == "__main__":
    asyncio.run(main())

运行这个脚本之前，确保你的SITL仿真已经启动。如果一切正常，你会看到类似下面的输出：

等待飞控连接...
飞控已连接

=== 心跳检测 ===
陀螺仪: True
加速度计: True
磁力计: True
气压计: True
全局定位: False
本地定位: True

=== 参数读取 ===
SYS_AUTOSTART: 4001
MC_ROLL_P: 0.15
BAT1_CAPACITY: 5000.0 mAh

通信测试完成！

看到这个输出，恭喜你，地面站和仿真飞控的通信链路已经打通了。心跳检测证明了链路是活的，参数读取证明了数据交互是双向的。