mavlink id 11 SET_MODE

设置系统模式。由于该模式是针对整架飞机的，而非仅针对某个组件，因此不存在目标组件ID。

字段名
类型
单位
描述
target_system
uint8_t

要设置模式的系统id
custom_mode
uint32_t
见px4 custom_mode
要切换到的飞行模式

PX4 custom_mode

飞行模式

值

POSCTL

0x00030000

TAKEOFF

0x02040000

HOLD

0x03040000

MISSION

0x04040000

RTL

0x05040000

LAND

0x06040000

执行任务功能：确保已上传任务内容后，将此消息将custom_mode设为任务模式并发送给无人机。

MAV_CMD_NAV_TAKEOFF 22

返回消息：COMMAND_ACK
返回值 command 22 执行命令id
返回值 result 0或4 0：执行成功，4：执行失败

示例

import time
from pymavlink import mavutil

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
master.wait_heartbeat()
boot_time = time.time()

def ready():
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_NAV_TAKEOFF,
        0,
        0, 0, 0, 0, 0, 0, 0
    )

ready()

执行

结果

示例

import time
import sys
from pymavlink import mavutil
mode_map = {
    "MANUAL": 0x00010000,
    "POSCTL": 0x00030000,
    "AUTO_RTL": 0x05040000,
    "AUTO_MISSION": 0x04040000,
}# 仅列举部分模式
connection = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
connection .wait_heartbeat()
target_system = 1
target_component = 1
custom_mode = mode_map["MANUAL"]  # 任务模式
# 发送SET_MODE消息
connection.mav.set_mode_send(
    target_system,
    target_component,
    custom_mode,
    0
)

执行结果

降落命令

方法1：发送以下降落命令

MAV_CMD_NAV_LAND 21

返回消息：COMMAND_ACK
返回值 command 21 执行命令id
返回值 result 0或4 0：执行成功，4：执行失败

方法2：使用切换飞行模式将飞行模式切换至LAND

详见切换飞行模式页

取消降落命令

方法1：使用切换飞行模式将飞行模式切换至POSCTL
详见切换飞行模式页

方法2：发送以下COMMAND_LONG命令

MAV_CMD_DO_REPOSITION 192

command = MAV_CMD_DO_REPOSITION
param4，param5，param6 = NaN
param7 = 0

返回消息：COMMAND_ACK
返回值 command 192 执行命令id
返回值 result 0或4 0：执行成功，4：执行失败

示例

发送降落命令（方法1）

import time
from pymavlink import mavutil

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
master.wait_heartbeat()
boot_time = time.time()

def land():
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_NAV_LAND,
        0,
        0, 0, 0, 0, 0, 0, 0
    )

land()

发送取消降落命令（方法2）

import time
from pymavlink import mavutil
import math


def get_current_millis():
    """获取当前时间（毫秒级）"""
    return int(time.time() * 1000)


# MAVLink 命令常量
MAV_CMD_DO_REPOSITION = 192
DEFAULT_TIMEOUT = 5.0


def send_reposition_command(master, ground_speed, yaw, flags=0):
    """发送重新定位命令到无人机

    Args:
        master: MAVLink 连接对象
        ground_speed: 地速 (单位: m/s)
        yaw: 偏航角 (单位: 度, 0-360)
        flags: 标志位 (保留, 默认为0)
    """
    nan_value = math.nan
    send_time = get_current_millis()
    print(f"[{send_time}] 发送重新定位命令")
    print(f"  地速 = {ground_speed} m/s, 偏航角 = {yaw}°")

    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_DO_REPOSITION,
        0,  # 确认标志
        ground_speed,   # param1: 地速 (单位: m/s)
        yaw,            # param2: 偏航角 (单位: 度)
        0,              # param3: 保留
        nan_value,      # param4: NaN
        nan_value,      # param5: NaN
        nan_value,      # param6: NaN
        0               # param7: 0
    )

    return send_time


def wait_for_ack(master, send_time, timeout=DEFAULT_TIMEOUT):
    """等待命令确认响应"""
    print("等待重新定位命令响应...")

    ack = master.recv_match(
        type='COMMAND_ACK',
        blocking=True,
        timeout=timeout
    )

    if ack:
        ack_time = get_current_millis()
        delay = ack_time - send_time
        print(f"[{ack_time}] 收到命令应答 (MAV_CMD_DO_REPOSITION)")
        print(f" 命令ID: {ack.command}, 结果: {ack.result}, 延迟: {delay}ms")
    else:
        print(f"[{get_current_millis()}] 超时：未收到重新定位命令应答")


def main():
    """主函数：连接无人机并发送重新定位命令"""
    # 连接无人机（UDP端口14550）
    master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
    print("等待无人机心跳...")
    master.wait_heartbeat()
    print("无人机已连接，开始发送重新定位命令")

    # 发送重新定位命令（示例参数，请根据实际情况修改）
    ground_speed = 5.0   # 地速 5 m/s
    yaw = 90.0           # 偏航角 90° (东向)
    send_time = send_reposition_command(master, ground_speed, yaw)

    # 等待响应
    wait_for_ack(master, send_time)

    print("重新定位命令发送完成")


if __name__ == "__main__":
    main()

执行

结果

返航命令

方法1：发送以下COMMAND_LONG命令

MAV_CMD_NAV_RETURN_TO_LAUNCH 20

param1 = 0 (0为返航) 非0为取消返航
返回消息 COMMAND_LONG
返回值 command = MAV_CMD_REQUEST_MESSAGE
返回值 param1 20 执行命令id
返回值 param2 0或1 是否执行成功
返回值 param3 失败代码
失败代码列表
0 无异常
1 无法初始化无人机飞行控制
2 返航失败
3 取消返航失败

方法2：使用切换飞行模式将飞行模式切换至RTL
详见切换飞行模式页

取消返航命令

方法1：发送以下COMMAND_LONG命令

MAV_CMD_NAV_RETURN_TO_LAUNCH 20

param1 = 1 (非0为取消返航)
返回消息同上

方法2：使用切换飞行模式将飞行模式切换至POSCTL
详见切换飞行模式页

方法3：

示例

import time
from pymavlink import mavutil

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
master.wait_heartbeat()
boot_time = time.time()

def return_to_base(cancleReturn):
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_NAV_RETURN_TO_LAUNCH,
        0,
        cancleReturn, 0, 0, 0, 0, 0, 0
    )

return_to_base(0)
time.sleep(10)
return_to_base(1)

执行

开始返航

中途停止返航

GLOBAL_POSITION_INT 33

Field NameTypeUnitsDescription
latint32_tdegE7纬度int*10^7
lonint32_tdegE7经度int*10^7
altint32_tmm高度

返回值 param1 33 执行命令id
返回值 param2 0或1 是否执行成功
返回值 param3 失败代码

失败代码列表
0 无异常
1 无法初始化无人机飞行控制
2 获取无人机控制权失败
3 创建飞行线程失败

当到达目标点后，或通过手动控制传入非摇杆中立值的参数，导致无人机自动切换到POSCTL模式后，将自动退出经纬度位置控制。

示例

import time
from pymavlink import mavutil

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
master.wait_heartbeat()

boot_time = time.time()
def fly():
    master.mav.global_position_int_send(
        int(1e3 * (time.time() - boot_time)),  # ms since boot
        lat=400006100, lon=1100006100, alt=700,
        relative_alt=0, vx=0, vy=0, vz=0, hdg=0
    )

fly()

执行

结果

MANUAL_CONTROL (69)

Field Name Type Description
target uint8_t 控制模式选择，0、1: x,y,z轴为速度控制，r轴为偏航角速率；2：x，y轴为姿态角控制，z轴为上升速度，r轴为偏航角速率。
x int16_t X轴，范围已归一化为[-1000,1000]。INT16_MAX的值表示该轴无效。对应于操纵杆的前进（1000），后退（-1000）移动以及飞行器的俯仰。
y int16_t Y轴，范围已归一化为[-1000,1000]。INT16_MAX的值表示该轴无效。对应于操纵杆的左移（-1000），右移（1000）以及飞行器的滚转。
z int16_t Z轴，范围已归一化为[0,1000]。INT16_MAX的值表示该轴无效。对应于飞行器的推力。摇杆中立时值为500，大于500表示上升，小于于500表示下降。
r int16_t R轴，范围已归一化为[-1000,1000]。INT16_MAX的值表示该轴无效。对应于飞行器的偏航，逆时针旋转（-1000），顺时针旋转（1000）。

target设置为0时摇杆轴取值的缩放：
x为control.x /200
y为control.y /200
z为(control.z /100 -5) / 2
r为control.r /16
target设置为1时摇杆轴取值的缩放：
x为control.x /50
y为control.y /50
z为(control.z - 500) /50
r为control.r /10
target设置为时摇杆轴取值的缩放：
x为control.x /10
y为control.y /10
z为control.z /10
r为control.r /10

当无人机为Orbit模式时，轴取值含义改为下表

X轴 更改盘旋半径，缩小半径（1000），增大半径（-1000）。最小半径是 1 米。 最大半径是 100 米。
Y轴 更改盘旋速度。与当前盘旋方向相同时则增加速度，反之则减少速度。速度小于0时将改变盘旋方向。最大速度为 10 m/s，进一步的限制是将向心加速度保持在 2 m/s^2 以下。
Z轴 更改盘旋时的高度。摇杆中立时值为500，大于500表示上升，小于于500表示下降。 
# Import mavutil
import time
from pymavlink import mavutil

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
master.wait_heartbeat()

def ready(x,y,z,r):
    master.mav.manual_control_send(
        1,
        x,
        y,
        z,
        r,
        0)


x = 0
while x < 1000:
    ready(10,10,10,10)
    x += 1
    time.sleep(0.05)

使用QGC虚拟摇杆发送操控指令遥控无人机运动

根据命令设置不同飞行状态以及目标

SET_POSITION_TARGET_LOCAL_NED 84

Field Name
Type
Units
Control Type
Description
time_boot_ms
uint32_t
ms
Timestamp (time since system boot).
target_system
uint8_t
System ID
target_component
uint8_t
Component ID
coordinate_frame
uint8_t
MAV_FRAME
坐标系配置 支持机体坐标系以及NED坐标系
type_mask
uint16_t
POSITION_TARGET_TYPEMASK
Bitmap决定使用什么方式控制飞行
x
float
m

X坐标(NED坐标系)
y
float
m

Y坐标(NED坐标系)
z
float
m

Z坐标(NED坐标系)
vx
float
m/s

X速度(NED坐标系)
vy
float
m/s

Y速度(NED坐标系)
vz
float
m/s

Z速度(NED坐标系)
afx
float
m/s/s
0
afy
float
m/s/s
0
afz
float
m/s/s
0
yaw
float
rad
角度
yaw_rate
float
rad/s
角速率

type_mask参数说明

Value
Name
Description
1
POSITION_TARGET_TYPEMASK_X_IGNORE
Ignore position x
2
POSITION_TARGET_TYPEMASK_Y_IGNORE
Ignore position y
4
POSITION_TARGET_TYPEMASK_Z_IGNORE
Ignore position z
8
POSITION_TARGET_TYPEMASK_VX_IGNORE
Ignore velocity x
16
POSITION_TARGET_TYPEMASK_VY_IGNORE
Ignore velocity y
32
POSITION_TARGET_TYPEMASK_VZ_IGNORE
Ignore velocity z
64
POSITION_TARGET_TYPEMASK_AX_IGNORE
Ignore acceleration x
128
POSITION_TARGET_TYPEMASK_AY_IGNORE
Ignore acceleration y
256
POSITION_TARGET_TYPEMASK_AZ_IGNORE
Ignore acceleration z
512
POSITION_TARGET_TYPEMASK_FORCE_SET
Use force instead of acceleration
1024
POSITION_TARGET_TYPEMASK_YAW_IGNORE
Ignore yaw
2048
POSITION_TARGET_TYPEMASK_YAW_RATE_IGNORE
Ignore yaw rate

当不忽略x,y,z时 此命令为位置飞行
当不忽略vx,vy,vz时 此命令为速度飞行
当不忽略yaw时 此命令为角度设置
当不忽略yaw_rate时 此命令为角速率命令

coordinate_frame 为机体坐标系参数
目前设置为两个参数
MAV_FRAME_LOCAL_NED 大地坐标系(NED坐标)
MAV_FRAME_BODY_FRD 机体坐标系(遵循右手定则)

返回command = MAV_CMD_REQUEST_MESSAGE
返回值 param1 84 执行命令id
返回值 param2 0或1 是否执行成功
返回值 param3 失败代码
失败代码列表
0 无异常
1 无法初始化无人机飞行控制
2 获取无人机控制权失败
3 未知坐标系
4 销毁FRU坐标系移动线程失败
5 销毁NED坐标系移动线程失败
6 创建FRU坐标系移动线程失败
7 创建NED坐标系移动线程失败

示例

import time
from pymavlink import mavutil

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
master.wait_heartbeat()
boot_time = time.time()


def ready():
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_NAV_TAKEOFF,
        0,
        0, 0, 0, 0, 0, 0, 0
    )


def fly():
    # test pos mod use ned frame
    print('test pos mod use ned frame')
    master.mav.set_position_target_local_ned_send(
        int(1e3 * (time.time() - boot_time)),  # ms since boot
        master.target_system,
        master.target_component,
        coordinate_frame=mavutil.mavlink.MAV_FRAME_LOCAL_NED,
        type_mask=(
            # mavutil.mavlink.POSITION_TARGET_TYPEMASK_X_IGNORE |
            # mavutil.mavlink.POSITION_TARGET_TYPEMASK_Y_IGNORE |
            # mavutil.mavlink.POSITION_TARGET_TYPEMASK_Z_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_VX_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_VY_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_VZ_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_AX_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_AY_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_AZ_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_FORCE_SET |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_YAW_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_YAW_RATE_IGNORE
        ),
        x=10, y=10, z=-10,
        vx=0, vy=0, vz=0,
        afx=0, afy=0, afz=0,
        yaw=0, yaw_rate=0
    )

    # test yaw rate mod
    time.sleep(10)
    print('test yaw rate mod')
    master.mav.set_position_target_local_ned_send(
        int(1e3 * (time.time() - boot_time)),  # ms since boot
        master.target_system,
        master.target_component,
        coordinate_frame=mavutil.mavlink.MAV_FRAME_LOCAL_NED,
        type_mask=(
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_X_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_Y_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_Z_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_VX_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_VY_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_VZ_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_AX_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_AY_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_AZ_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_FORCE_SET |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_YAW_IGNORE
            # mavutil.mavlink.POSITION_TARGET_TYPEMASK_YAW_RATE_IGNORE
        ),
        x=0, y=0, z=0,
        vx=0, vy=0, vz=0,
        afx=0, afy=0, afz=0,
        yaw=0, yaw_rate=90
    )

    # test velocity mod fru mod
    time.sleep(10)
    print('test velocity mod fru mod')
    x = 0
    while x < 200:
        master.mav.set_position_target_local_ned_send(
            int(1e3 * (time.time() - boot_time)),  # ms since boot
            master.target_system,
            master.target_component,
            coordinate_frame=mavutil.mavlink.MAV_FRAME_BODY_FRD,
            type_mask=(
                mavutil.mavlink.POSITION_TARGET_TYPEMASK_X_IGNORE |
                mavutil.mavlink.POSITION_TARGET_TYPEMASK_Y_IGNORE |
                mavutil.mavlink.POSITION_TARGET_TYPEMASK_Z_IGNORE |
                # mavutil.mavlink.POSITION_TARGET_TYPEMASK_VX_IGNORE |
                # mavutil.mavlink.POSITION_TARGET_TYPEMASK_VY_IGNORE |
                # mavutil.mavlink.POSITION_TARGET_TYPEMASK_VZ_IGNORE |
                mavutil.mavlink.POSITION_TARGET_TYPEMASK_AX_IGNORE |
                mavutil.mavlink.POSITION_TARGET_TYPEMASK_AY_IGNORE |
                mavutil.mavlink.POSITION_TARGET_TYPEMASK_AZ_IGNORE |
                mavutil.mavlink.POSITION_TARGET_TYPEMASK_FORCE_SET |
                mavutil.mavlink.POSITION_TARGET_TYPEMASK_YAW_IGNORE |
                mavutil.mavlink.POSITION_TARGET_TYPEMASK_YAW_RATE_IGNORE
            ),
            x=0, y=0, z=0,
            vx=5, vy=0, vz=-5,
            afx=0, afy=0, afz=0,
            yaw=0, yaw_rate=0)
        x += 1
        time.sleep(0.25)
        print('----------------' + str(x))

    # test yaw mod
    time.sleep(10)
    print('test yaw mod')
    master.mav.set_position_target_local_ned_send(
        int(1e3 * (time.time() - boot_time)),  # ms since boot
        master.target_system,
        master.target_component,
        coordinate_frame=mavutil.mavlink.MAV_FRAME_LOCAL_NED,
        type_mask=(
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_X_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_Y_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_Z_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_VX_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_VY_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_VZ_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_AX_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_AY_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_AZ_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_FORCE_SET |
            # mavutil.mavlink.POSITION_TARGET_TYPEMASK_YAW_IGNORE |
            mavutil.mavlink.POSITION_TARGET_TYPEMASK_YAW_RATE_IGNORE
        ),
        x=0, y=0, z=0,
        vx=0, vy=0, vz=0,
        afx=0, afy=0, afz=0,
        yaw=-110, yaw_rate=0
    )


def land():
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_NAV_LAND,
        0,
        0, 0, 0, 0, 0, 0, 0
    )


ready()
time.sleep(10)
fly()
time.sleep(10)
land()

执行结果

飞控全向避障行为

ALL_AVOID_ENABL

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
枚举值 值 说明
DJI_FLIGHT_CONTROLLER_DISABLE_ALL_AVIOD 0 禁用全部避障功能
DJI_FLIGHT_CONTROLLER_ENABLE_ALL_AVIOD 1 启用全部避障功能
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“ALL_AVOID_ENABL”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“ALL_AVOID_ENABL”

输入要设置的数值

选择参数类型“1”

成功返回设置值

无人机避障雷达的距离传感器信息

mavlink id 132 DISTANCE_SENSOR

返回字段名称
类型
单位
描述
time_boot_ms
uint32_t
ms
时间戳（系统启动时间）。
min_distance
uint16_t
cm
传感器能测量的最小距离。
max_distance
uint16_t
cm
传感器能测量的最大距离。
current_distance
uint16_t
cm
当前传感器测得距离。
type
uint8_t
MAV_DISTANCE_SENSOR
测距传感器的类型。
id
uint8_t

机载传感器的编号。
orientation
uint8_t
MAV_SENSOR_ORIENTATION
传感器面向的方向。
covariance
uint8_t
cm^2
测量方差。最大标准差为 6 厘米。未知：UINT8_MAX。

MAV_DISTANCE_SENSOR

距离传感器类型的枚举

Value

Field Name

Description

0

MAV_DISTANCE_SENSOR_LASER

激光测距仪，例如 LightWare SF02/F 或 PulsedLight 单元。

1

MAV_DISTANCE_SENSOR_ULTRASOUND

超声测距仪，例如 MaxBotix 单元。

2

MAV_DISTANCE_SENSOR_INFRARED

红外测距仪，例如 Sharp 单元。

3

MAV_DISTANCE_SENSOR_RADAR

雷达类型，例如 uLanding 单元

4

MAV_DISTANCE_SENSOR_UNKNOWN

故障或未知类型，例如模拟单元

MAV_SENSOR_ORIENTATION

根据旋转点枚举的传感器方向

Value

Field Name

Description

0

MAV_SENSOR_ROTATION_NONE

向前。滚转：0，俯仰：0，偏航：0

2

MAV_SENSOR_ROTATION_YAW_90

向左。滚转：0，俯仰：0，偏航：90

6

MAV_SENSOR_ROTATION_YAW_270

向右。滚转：0，俯仰：0，偏航：270

12

MAV_SENSOR_ROTATION_PITCH_180

向后。滚转：0，俯仰：180，偏航：0

24

MAV_SENSOR_ROTATION_PITCH_90

向上。滚转：0，俯仰：90，偏航：0

25

MAV_SENSOR_ROTATION_PITCH_270

向下。滚转：0，俯仰：270，偏航：0

结果（使用QGC查看）

最大速度

MAX_VELOCITY

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
类型 取值范围 单位 说明
uint8_t 1 ~ 15 m/s 最大飞行速度
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“MAX_VELOCITY”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“MAX_VELOCITY”

输入要设置的数值

选择参数类型“1”

成功返回设置值

设置最低高度

MAX_HEIGHT

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
参数名 类型 单位 描述
value float m 最小飞行高度。取值范围：1.0～3000.0
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“MAX_HEIGHT”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“MAX_HEIGHT”

输入要设置的数值

选择参数类型“8”

成功返回设置值

RTK位置功能

RTK_POS_ENABLE

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
枚举值 值 说明
DJI_FLIGHT_CONTROLLER_DISABLE_RTK_POSITION 0 禁用 RTK 定位，飞机使用 GPS 数据执行需要位置信息的功能（如航点飞行、返航等）
DJI_FLIGHT_CONTROLLER_ENABLE_RTK_POSITION 1 启用 RTK 定位，飞机使用 RTK 数据执行需要位置信息的功能（如航点飞行、返航等）
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“RTK_POS_ENABLE”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“RTK_POS_ENABLE”

输入要设置的数值

选择参数类型“1”

成功返回设置值

遥控器失联动作

LOST_ACTION

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
枚举值 值 说明
DJI_FLIGHT_CONTROLLER_RC_LOST_ACTION_HOVER 0 遥控器失联后，飞机执行悬停动作（Hover）
DJI_FLIGHT_CONTROLLER_RC_LOST_ACTION_LANDING 1 遥控器失联后，飞机执行降落动作（Landing）
DJI_FLIGHT_CONTROLLER_RC_LOST_ACTION_GOHOME 2 遥控器失联后，飞机执行返航动作（Go Home）
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“LOST_ACTION”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“LOST_ACTION”

输入要设置的数值

选择参数类型“1”

成功返回设置值

水平视觉障碍物避让的开关状态

HOR_VIS_AVO_ENA

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
枚举值 值 说明
DJI_FLIGHT_CONTROLLER_DISABLE_OBSTACLE_AVOIDANCE 0 禁用指定方向的避障功能
DJI_FLIGHT_CONTROLLER_ENABLE_OBSTACLE_AVOIDANCE 1 启用指定方向的避障功能
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“HOR_VIS_AVO_ENA”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“HOR_VIS_AVO_ENA”

输入要设置的数值

选择参数类型“1”

成功返回设置值

上视避障功能

UP_VIS_AVO_ENA

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
枚举值 值 说明
DJI_FLIGHT_CONTROLLER_DISABLE_OBSTACLE_AVOIDANCE 0 禁用指定方向的避障功能
DJI_FLIGHT_CONTROLLER_ENABLE_OBSTACLE_AVOIDANCE 1 启用指定方向的避障功能
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“UP_VIS_AVO_ENA”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“UP_VIS_AVO_ENA”

输入要设置的数值

选择参数类型“1”

成功返回设置值

向下视觉避障功能

DOW_VIS_AVO_ENA

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
枚举值 值 说明
DJI_FLIGHT_CONTROLLER_DISABLE_OBSTACLE_AVOIDANCE 0 禁用指定方向的避障功能
DJI_FLIGHT_CONTROLLER_ENABLE_OBSTACLE_AVOIDANCE 1 启用指定方向的避障功能
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“DOW_VIS_AVO_ENA”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“DOW_VIS_AVO_ENA”

输入要设置的数值

选择参数类型“1”

成功返回设置值

紧急制动飞行

MAV_DJI_CMD_ARREST_FLYING

示例
import time
from pymavlink import mavutil
from enum import IntEnum


# MAV_CMD 执行结果映射
MAV_RESULT_NAMES = {
    mavutil.mavlink.MAV_RESULT_ACCEPTED: "ACCEPTED",
    mavutil.mavlink.MAV_RESULT_TEMPORARILY_REJECTED: "TEMPORARILY_REJECTED",
    mavutil.mavlink.MAV_RESULT_DENIED: "DENIED",
    mavutil.mavlink.MAV_RESULT_UNSUPPORTED: "UNSUPPORTED",
    mavutil.mavlink.MAV_RESULT_FAILED: "FAILED",
    mavutil.mavlink.MAV_RESULT_IN_PROGRESS: "IN_PROGRESS",
}


def recv_command_ack(timeout=5):
    """接收并显示 command_ack 消息

    Args:
        timeout: 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时
    """
    print(f"等待 command_ack 响应 (超时 {timeout}s)...")

    ack = master.recv_match(type='COMMAND_ACK', blocking=True, timeout=timeout)

    if ack is None:
        print("错误: 命令响应超时")
        return None

    # 获取命令名称
    cmd_name = getattr(ack, 'command_name', None)
    if not cmd_name:
        # 尝试通过枚举查找命令名称
        try:
            cmd_name = mavutil.mavlink.enums.get('MAV_CMD', {}).get(ack.command, {}).name
        except:
            cmd_name = str(ack.command)

    # 获取结果名称
    result_name = MAV_RESULT_NAMES.get(ack.result, f"UNKNOWN({ack.result})")

    print("=" * 50)
    print(f"命令: {cmd_name} ({ack.command})")
    print(f"结果: {result_name}")
    print(f"进度: {ack.progress * 100:.0f}%" if ack.progress > 0 else "")
    # print(f"参数2: {ack.result_param2}")
    print(f"系统ID: {ack.target_system}")
    print(f"组件ID: {ack.target_component}")
    print("=" * 50)

    return ack


class MavDjiCmd(IntEnum):
    MAV_CMD_NAV_LAND = 21
    MAV_DJI_CMD_TRIGGER_FFC=40
    MAV_DJI_CMD_START_RECORD_POINT_CLOUD=41
    MAV_DJI_CMD_STOP_RECORD_POINT_CLOUD=42
    MAV_DJI_CMD_RESTORE_FACTORY_SETTINGS=43
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNC=44
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNCV2=45
    MAV_DJI_CMD_START_CONFIRM_LANDING=46
    MAV_DJI_CMD_START_FORCE_LANDING=47
    MAV_DJI_CMD_EXECUTE_EMERGENCY_BRAKE_ACTION=48
    MAV_DJI_CMD_CANCEL_EMERGENCY_BRAKE_ACTION=49
    MAV_DJI_CMD_START_SLOW_ROTATE_MOTOR=50
    MAV_DJI_CMD_STOP_SLOW_ROTATE_MOTOR=51
    MAV_DJI_CMD_ARREST_FLYING=52
    MAV_DJI_CMD_CANCEL_ARREST_FLYING=53
    MAV_DJI_CMD_OUTPUT_HIGH_POWER=54
    TUNNEL_PROTOCOL_ID_INJECT_HMS_ERROR=32770

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
print("waiting for heartbeat")
master.wait_heartbeat()
print("heartbeat received")
boot_time = time.time()

def send_command(command, param1=0, param2=0, param3=0, param4=0, param5=0, param6=0, param7=0, wait_ack=True, timeout=5):
    """发送 command_long 并可选接收 command_ack

    Args:
        command: 命令ID (如 MavDjiCmd.TRIGGER_FFC)
        param1~param7: 命令参数
        wait_ack: 是否等待并显示 ACK
        timeout: ACK 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时或未等待
    """
    print(f"发送命令: {command} ({int(command)})")
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        int(command),
        0,  # confirmation
        param1, param2, param3, param4, param5, param6, param7
    )

    if wait_ack:
        return recv_command_ack(timeout=timeout)
    return None


def ready():
    send_command(MavDjiCmd.MAV_DJI_CMD_ARREST_FLYING)

ready()
示例使用说明

运行后，成功建立连接并发送指令

退出紧急制动状态

MAV_DJI_CMD_CANCEL_ARREST_FLYING

示例
import time
from pymavlink import mavutil
from enum import IntEnum


# MAV_CMD 执行结果映射
MAV_RESULT_NAMES = {
    mavutil.mavlink.MAV_RESULT_ACCEPTED: "ACCEPTED",
    mavutil.mavlink.MAV_RESULT_TEMPORARILY_REJECTED: "TEMPORARILY_REJECTED",
    mavutil.mavlink.MAV_RESULT_DENIED: "DENIED",
    mavutil.mavlink.MAV_RESULT_UNSUPPORTED: "UNSUPPORTED",
    mavutil.mavlink.MAV_RESULT_FAILED: "FAILED",
    mavutil.mavlink.MAV_RESULT_IN_PROGRESS: "IN_PROGRESS",
}


def recv_command_ack(timeout=5):
    """接收并显示 command_ack 消息

    Args:
        timeout: 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时
    """
    print(f"等待 command_ack 响应 (超时 {timeout}s)...")

    ack = master.recv_match(type='COMMAND_ACK', blocking=True, timeout=timeout)

    if ack is None:
        print("错误: 命令响应超时")
        return None

    # 获取命令名称
    cmd_name = getattr(ack, 'command_name', None)
    if not cmd_name:
        # 尝试通过枚举查找命令名称
        try:
            cmd_name = mavutil.mavlink.enums.get('MAV_CMD', {}).get(ack.command, {}).name
        except:
            cmd_name = str(ack.command)

    # 获取结果名称
    result_name = MAV_RESULT_NAMES.get(ack.result, f"UNKNOWN({ack.result})")

    print("=" * 50)
    print(f"命令: {cmd_name} ({ack.command})")
    print(f"结果: {result_name}")
    print(f"进度: {ack.progress * 100:.0f}%" if ack.progress > 0 else "")
    # print(f"参数2: {ack.result_param2}")
    print(f"系统ID: {ack.target_system}")
    print(f"组件ID: {ack.target_component}")
    print("=" * 50)

    return ack


class MavDjiCmd(IntEnum):
    MAV_CMD_NAV_LAND = 21
    MAV_DJI_CMD_TRIGGER_FFC=40
    MAV_DJI_CMD_START_RECORD_POINT_CLOUD=41
    MAV_DJI_CMD_STOP_RECORD_POINT_CLOUD=42
    MAV_DJI_CMD_RESTORE_FACTORY_SETTINGS=43
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNC=44
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNCV2=45
    MAV_DJI_CMD_START_CONFIRM_LANDING=46
    MAV_DJI_CMD_START_FORCE_LANDING=47
    MAV_DJI_CMD_EXECUTE_EMERGENCY_BRAKE_ACTION=48
    MAV_DJI_CMD_CANCEL_EMERGENCY_BRAKE_ACTION=49
    MAV_DJI_CMD_START_SLOW_ROTATE_MOTOR=50
    MAV_DJI_CMD_STOP_SLOW_ROTATE_MOTOR=51
    MAV_DJI_CMD_ARREST_FLYING=52
    MAV_DJI_CMD_CANCEL_ARREST_FLYING=53
    MAV_DJI_CMD_OUTPUT_HIGH_POWER=54
    TUNNEL_PROTOCOL_ID_INJECT_HMS_ERROR=32770

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
print("waiting for heartbeat")
master.wait_heartbeat()
print("heartbeat received")
boot_time = time.time()

def send_command(command, param1=0, param2=0, param3=0, param4=0, param5=0, param6=0, param7=0, wait_ack=True, timeout=5):
    """发送 command_long 并可选接收 command_ack

    Args:
        command: 命令ID (如 MavDjiCmd.TRIGGER_FFC)
        param1~param7: 命令参数
        wait_ack: 是否等待并显示 ACK
        timeout: ACK 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时或未等待
    """
    print(f"发送命令: {command} ({int(command)})")
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        int(command),
        0,  # confirmation
        param1, param2, param3, param4, param5, param6, param7
    )

    if wait_ack:
        return recv_command_ack(timeout=timeout)
    return None


def ready():
    send_command(MavDjiCmd.MAV_DJI_CMD_CANCEL_ARREST_FLYING)

ready()
示例使用说明
获取

运行后，成功建立连接并发送指令

在任何情况下紧急停止电机

MAV_CMD_COMPONENT_ARM_DISARM

示例
import time
from pymavlink import mavutil
from enum import IntEnum


# MAV_CMD 执行结果映射
MAV_RESULT_NAMES = {
    mavutil.mavlink.MAV_RESULT_ACCEPTED: "ACCEPTED",
    mavutil.mavlink.MAV_RESULT_TEMPORARILY_REJECTED: "TEMPORARILY_REJECTED",
    mavutil.mavlink.MAV_RESULT_DENIED: "DENIED",
    mavutil.mavlink.MAV_RESULT_UNSUPPORTED: "UNSUPPORTED",
    mavutil.mavlink.MAV_RESULT_FAILED: "FAILED",
    mavutil.mavlink.MAV_RESULT_IN_PROGRESS: "IN_PROGRESS",
}


def recv_command_ack(timeout=5):
    """接收并显示 command_ack 消息

    Args:
        timeout: 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时
    """
    print(f"等待 command_ack 响应 (超时 {timeout}s)...")

    ack = master.recv_match(type='COMMAND_ACK', blocking=True, timeout=timeout)

    if ack is None:
        print("错误: 命令响应超时")
        return None

    # 获取命令名称
    cmd_name = getattr(ack, 'command_name', None)
    if not cmd_name:
        # 尝试通过枚举查找命令名称
        try:
            cmd_name = mavutil.mavlink.enums.get('MAV_CMD', {}).get(ack.command, {}).name
        except:
            cmd_name = str(ack.command)

    # 获取结果名称
    result_name = MAV_RESULT_NAMES.get(ack.result, f"UNKNOWN({ack.result})")

    print("=" * 50)
    print(f"命令: {cmd_name} ({ack.command})")
    print(f"结果: {result_name}")
    print(f"进度: {ack.progress * 100:.0f}%" if ack.progress > 0 else "")
    # print(f"参数2: {ack.result_param2}")
    print(f"系统ID: {ack.target_system}")
    print(f"组件ID: {ack.target_component}")
    print("=" * 50)

    return ack


class MavDjiCmd(IntEnum):
    MAV_CMD_NAV_LAND = 21
    MAV_DJI_CMD_TRIGGER_FFC=40
    MAV_DJI_CMD_START_RECORD_POINT_CLOUD=41
    MAV_DJI_CMD_STOP_RECORD_POINT_CLOUD=42
    MAV_DJI_CMD_RESTORE_FACTORY_SETTINGS=43
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNC=44
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNCV2=45
    MAV_DJI_CMD_START_CONFIRM_LANDING=46
    MAV_DJI_CMD_START_FORCE_LANDING=47
    MAV_DJI_CMD_EXECUTE_EMERGENCY_BRAKE_ACTION=48
    MAV_DJI_CMD_CANCEL_EMERGENCY_BRAKE_ACTION=49
    MAV_DJI_CMD_START_SLOW_ROTATE_MOTOR=50
    MAV_DJI_CMD_STOP_SLOW_ROTATE_MOTOR=51
    MAV_DJI_CMD_ARREST_FLYING=52
    MAV_DJI_CMD_CANCEL_ARREST_FLYING=53
    MAV_DJI_CMD_OUTPUT_HIGH_POWER=54
    TUNNEL_PROTOCOL_ID_INJECT_HMS_ERROR=32770

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
print("waiting for heartbeat")
master.wait_heartbeat()
print("heartbeat received")
boot_time = time.time()

def send_command(command, param1=0, param2=0, param3=0, param4=0, param5=0, param6=0, param7=0, wait_ack=True, timeout=5):
    """发送 command_long 并可选接收 command_ack

    Args:
        command: 命令ID (如 MavDjiCmd.TRIGGER_FFC)
        param1~param7: 命令参数
        wait_ack: 是否等待并显示 ACK
        timeout: ACK 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时或未等待
    """
    print(f"发送命令: {command} ({int(command)})")
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        int(command),
        0,  # confirmation
        param1, param2, param3, param4, param5, param6, param7
    )

    if wait_ack:
        return recv_command_ack(timeout=timeout)
    return None


def ready():
    send_command(400,0,21196)

ready()
示例使用说明
获取

运行后，成功建立连接并发送指令

注意：紧急停机后电机锁定，必须重启无人机！

无人机在地面时请求起飞

MAV_CMD_NAV_TAKEOFF 22
返回消息：COMMAND_ACK
返回值 command 22 执行命令id
返回值 result 0或4 0：执行成功，4：执行失败

示例

import time
from pymavlink import mavutil

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
master.wait_heartbeat()
boot_time = time.time()

def ready():
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_NAV_TAKEOFF,
        0,
        0, 0, 0, 0, 0, 0, 0
    )

ready()

执行示例

结果

降落命令

方法1：发送以下降落命令

MAV_CMD_NAV_LAND 21

返回消息：COMMAND_ACK
返回值 command 21 执行命令id
返回值 result 0或4 0：执行成功，4：执行失败

方法2：使用切换飞行模式将飞行模式切换至LAND

详见切换飞行模式页

取消降落命令

方法1：使用切换飞行模式将飞行模式切换至POSCTL
详见切换飞行模式页

方法2：发送以下COMMAND_LONG命令

MAV_CMD_DO_REPOSITION 192

command = MAV_CMD_DO_REPOSITION
param4，param5，param6 = NaN
param7 = 0

返回消息：COMMAND_ACK
返回值 command 192 执行命令id
返回值 result 0或4 0：执行成功，4：执行失败

示例

发送降落命令（方法1）

import time
from pymavlink import mavutil

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
master.wait_heartbeat()
boot_time = time.time()

def land():
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_NAV_LAND,
        0,
        0, 0, 0, 0, 0, 0, 0
    )

land()

发送取消降落命令（方法2）

import time
from pymavlink import mavutil
import math


def get_current_millis():
    """获取当前时间（毫秒级）"""
    return int(time.time() * 1000)


# MAVLink 命令常量
MAV_CMD_DO_REPOSITION = 192
DEFAULT_TIMEOUT = 5.0


def send_reposition_command(master, ground_speed, yaw, flags=0):
    """发送重新定位命令到无人机

    Args:
        master: MAVLink 连接对象
        ground_speed: 地速 (单位: m/s)
        yaw: 偏航角 (单位: 度, 0-360)
        flags: 标志位 (保留, 默认为0)
    """
    nan_value = math.nan
    send_time = get_current_millis()
    print(f"[{send_time}] 发送重新定位命令")
    print(f"  地速 = {ground_speed} m/s, 偏航角 = {yaw}°")

    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_DO_REPOSITION,
        0,  # 确认标志
        ground_speed,   # param1: 地速 (单位: m/s)
        yaw,            # param2: 偏航角 (单位: 度)
        0,              # param3: 保留
        nan_value,      # param4: NaN
        nan_value,      # param5: NaN
        nan_value,      # param6: NaN
        0               # param7: 0
    )

    return send_time


def wait_for_ack(master, send_time, timeout=DEFAULT_TIMEOUT):
    """等待命令确认响应"""
    print("等待重新定位命令响应...")

    ack = master.recv_match(
        type='COMMAND_ACK',
        blocking=True,
        timeout=timeout
    )

    if ack:
        ack_time = get_current_millis()
        delay = ack_time - send_time
        print(f"[{ack_time}] 收到命令应答 (MAV_CMD_DO_REPOSITION)")
        print(f" 命令ID: {ack.command}, 结果: {ack.result}, 延迟: {delay}ms")
    else:
        print(f"[{get_current_millis()}] 超时：未收到重新定位命令应答")


def main():
    """主函数：连接无人机并发送重新定位命令"""
    # 连接无人机（UDP端口14550）
    master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
    print("等待无人机心跳...")
    master.wait_heartbeat()
    print("无人机已连接，开始发送重新定位命令")

    # 发送重新定位命令（示例参数，请根据实际情况修改）
    ground_speed = 5.0   # 地速 5 m/s
    yaw = 90.0           # 偏航角 90° (东向)
    send_time = send_reposition_command(master, ground_speed, yaw)

    # 等待响应
    wait_for_ack(master, send_time)

    print("重新定位命令发送完成")


if __name__ == "__main__":
    main()

执行降落示例

结果

当无人机距离地面0.7米时确认着陆

MAV_DJI_CMD_START_CONFIRM_LANDING

示例
import time
from pymavlink import mavutil
from enum import IntEnum


# MAV_CMD 执行结果映射
MAV_RESULT_NAMES = {
    mavutil.mavlink.MAV_RESULT_ACCEPTED: "ACCEPTED",
    mavutil.mavlink.MAV_RESULT_TEMPORARILY_REJECTED: "TEMPORARILY_REJECTED",
    mavutil.mavlink.MAV_RESULT_DENIED: "DENIED",
    mavutil.mavlink.MAV_RESULT_UNSUPPORTED: "UNSUPPORTED",
    mavutil.mavlink.MAV_RESULT_FAILED: "FAILED",
    mavutil.mavlink.MAV_RESULT_IN_PROGRESS: "IN_PROGRESS",
}


def recv_command_ack(timeout=5):
    """接收并显示 command_ack 消息

    Args:
        timeout: 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时
    """
    print(f"等待 command_ack 响应 (超时 {timeout}s)...")

    ack = master.recv_match(type='COMMAND_ACK', blocking=True, timeout=timeout)

    if ack is None:
        print("错误: 命令响应超时")
        return None

    # 获取命令名称
    cmd_name = getattr(ack, 'command_name', None)
    if not cmd_name:
        # 尝试通过枚举查找命令名称
        try:
            cmd_name = mavutil.mavlink.enums.get('MAV_CMD', {}).get(ack.command, {}).name
        except:
            cmd_name = str(ack.command)

    # 获取结果名称
    result_name = MAV_RESULT_NAMES.get(ack.result, f"UNKNOWN({ack.result})")

    print("=" * 50)
    print(f"命令: {cmd_name} ({ack.command})")
    print(f"结果: {result_name}")
    print(f"进度: {ack.progress * 100:.0f}%" if ack.progress > 0 else "")
    # print(f"参数2: {ack.result_param2}")
    print(f"系统ID: {ack.target_system}")
    print(f"组件ID: {ack.target_component}")
    print("=" * 50)

    return ack


class MavDjiCmd(IntEnum):
    MAV_CMD_NAV_LAND = 21
    MAV_DJI_CMD_TRIGGER_FFC=40
    MAV_DJI_CMD_START_RECORD_POINT_CLOUD=41
    MAV_DJI_CMD_STOP_RECORD_POINT_CLOUD=42
    MAV_DJI_CMD_RESTORE_FACTORY_SETTINGS=43
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNC=44
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNCV2=45
    MAV_DJI_CMD_START_CONFIRM_LANDING=46
    MAV_DJI_CMD_START_FORCE_LANDING=47
    MAV_DJI_CMD_EXECUTE_EMERGENCY_BRAKE_ACTION=48
    MAV_DJI_CMD_CANCEL_EMERGENCY_BRAKE_ACTION=49
    MAV_DJI_CMD_START_SLOW_ROTATE_MOTOR=50
    MAV_DJI_CMD_STOP_SLOW_ROTATE_MOTOR=51
    MAV_DJI_CMD_ARREST_FLYING=52
    MAV_DJI_CMD_CANCEL_ARREST_FLYING=53
    MAV_DJI_CMD_OUTPUT_HIGH_POWER=54
    TUNNEL_PROTOCOL_ID_INJECT_HMS_ERROR=32770

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
print("waiting for heartbeat")
master.wait_heartbeat()
print("heartbeat received")
boot_time = time.time()

def send_command(command, param1=0, param2=0, param3=0, param4=0, param5=0, param6=0, param7=0, wait_ack=True, timeout=5):
    """发送 command_long 并可选接收 command_ack

    Args:
        command: 命令ID (如 MavDjiCmd.TRIGGER_FFC)
        param1~param7: 命令参数
        wait_ack: 是否等待并显示 ACK
        timeout: ACK 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时或未等待
    """
    print(f"发送命令: {command} ({int(command)})")
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        int(command),
        0,  # confirmation
        param1, param2, param3, param4, param5, param6, param7
    )

    if wait_ack:
        return recv_command_ack(timeout=timeout)
    return None


def ready():
    send_command(MavDjiCmd.MAV_DJI_CMD_START_CONFIRM_LANDING)

ready()
示例使用说明
获取

运行后，成功建立连接并发送指令

注：仅在下避障开启时有效！

在任何情况下都强制着陆

MAV_DJI_CMD_START_FORCE_LANDING

示例
import time
from pymavlink import mavutil
from enum import IntEnum


# MAV_CMD 执行结果映射
MAV_RESULT_NAMES = {
    mavutil.mavlink.MAV_RESULT_ACCEPTED: "ACCEPTED",
    mavutil.mavlink.MAV_RESULT_TEMPORARILY_REJECTED: "TEMPORARILY_REJECTED",
    mavutil.mavlink.MAV_RESULT_DENIED: "DENIED",
    mavutil.mavlink.MAV_RESULT_UNSUPPORTED: "UNSUPPORTED",
    mavutil.mavlink.MAV_RESULT_FAILED: "FAILED",
    mavutil.mavlink.MAV_RESULT_IN_PROGRESS: "IN_PROGRESS",
}


def recv_command_ack(timeout=5):
    """接收并显示 command_ack 消息

    Args:
        timeout: 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时
    """
    print(f"等待 command_ack 响应 (超时 {timeout}s)...")

    ack = master.recv_match(type='COMMAND_ACK', blocking=True, timeout=timeout)

    if ack is None:
        print("错误: 命令响应超时")
        return None

    # 获取命令名称
    cmd_name = getattr(ack, 'command_name', None)
    if not cmd_name:
        # 尝试通过枚举查找命令名称
        try:
            cmd_name = mavutil.mavlink.enums.get('MAV_CMD', {}).get(ack.command, {}).name
        except:
            cmd_name = str(ack.command)

    # 获取结果名称
    result_name = MAV_RESULT_NAMES.get(ack.result, f"UNKNOWN({ack.result})")

    print("=" * 50)
    print(f"命令: {cmd_name} ({ack.command})")
    print(f"结果: {result_name}")
    print(f"进度: {ack.progress * 100:.0f}%" if ack.progress > 0 else "")
    # print(f"参数2: {ack.result_param2}")
    print(f"系统ID: {ack.target_system}")
    print(f"组件ID: {ack.target_component}")
    print("=" * 50)

    return ack


class MavDjiCmd(IntEnum):
    MAV_CMD_NAV_LAND = 21
    MAV_DJI_CMD_TRIGGER_FFC=40
    MAV_DJI_CMD_START_RECORD_POINT_CLOUD=41
    MAV_DJI_CMD_STOP_RECORD_POINT_CLOUD=42
    MAV_DJI_CMD_RESTORE_FACTORY_SETTINGS=43
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNC=44
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNCV2=45
    MAV_DJI_CMD_START_CONFIRM_LANDING=46
    MAV_DJI_CMD_START_FORCE_LANDING=47
    MAV_DJI_CMD_EXECUTE_EMERGENCY_BRAKE_ACTION=48
    MAV_DJI_CMD_CANCEL_EMERGENCY_BRAKE_ACTION=49
    MAV_DJI_CMD_START_SLOW_ROTATE_MOTOR=50
    MAV_DJI_CMD_STOP_SLOW_ROTATE_MOTOR=51
    MAV_DJI_CMD_ARREST_FLYING=52
    MAV_DJI_CMD_CANCEL_ARREST_FLYING=53
    MAV_DJI_CMD_OUTPUT_HIGH_POWER=54
    TUNNEL_PROTOCOL_ID_INJECT_HMS_ERROR=32770

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
print("waiting for heartbeat")
master.wait_heartbeat()
print("heartbeat received")
boot_time = time.time()

def send_command(command, param1=0, param2=0, param3=0, param4=0, param5=0, param6=0, param7=0, wait_ack=True, timeout=5):
    """发送 command_long 并可选接收 command_ack

    Args:
        command: 命令ID (如 MavDjiCmd.TRIGGER_FFC)
        param1~param7: 命令参数
        wait_ack: 是否等待并显示 ACK
        timeout: ACK 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时或未等待
    """
    print(f"发送命令: {command} ({int(command)})")
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        int(command),
        0,  # confirmation
        param1, param2, param3, param4, param5, param6, param7
    )

    if wait_ack:
        return recv_command_ack(timeout=timeout)
    return None


def ready():
    send_command(MavDjiCmd.MAV_DJI_CMD_START_FORCE_LANDING)

ready()
示例使用说明

运行后，成功建立连接并发送指令

注：该功能仅在无人机处于空中时可执行。

返航高度

RTL_RETURN_ALT

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
名称 类型 单位 描述
GoHomeAltitude uint16_t m 返航高度。取值范围：20～500 米。
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“RTL_RETURN_ALT”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“RTL_RETURN_ALT”

输入要设置的数值

选择参数类型“1”

成功返回设置值

获取国家码

COUNTRY_CODE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
参数名 类型 描述
countryCode uint16_t 返回国家/地区代码。

表示飞行器当前所在的国家或地区，具体含义请参考 ISO 3166-1 国家代码标准。

示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“COUNTRY_CODE”

成功返回该参数对应的数值

请求紧急制动动作

MAV_DJI_CMD_EXECUTE_EMERGENCY_BRAKE_ACTION

示例
import time
from pymavlink import mavutil
from enum import IntEnum


# MAV_CMD 执行结果映射
MAV_RESULT_NAMES = {
    mavutil.mavlink.MAV_RESULT_ACCEPTED: "ACCEPTED",
    mavutil.mavlink.MAV_RESULT_TEMPORARILY_REJECTED: "TEMPORARILY_REJECTED",
    mavutil.mavlink.MAV_RESULT_DENIED: "DENIED",
    mavutil.mavlink.MAV_RESULT_UNSUPPORTED: "UNSUPPORTED",
    mavutil.mavlink.MAV_RESULT_FAILED: "FAILED",
    mavutil.mavlink.MAV_RESULT_IN_PROGRESS: "IN_PROGRESS",
}


def recv_command_ack(timeout=5):
    """接收并显示 command_ack 消息

    Args:
        timeout: 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时
    """
    print(f"等待 command_ack 响应 (超时 {timeout}s)...")

    ack = master.recv_match(type='COMMAND_ACK', blocking=True, timeout=timeout)

    if ack is None:
        print("错误: 命令响应超时")
        return None

    # 获取命令名称
    cmd_name = getattr(ack, 'command_name', None)
    if not cmd_name:
        # 尝试通过枚举查找命令名称
        try:
            cmd_name = mavutil.mavlink.enums.get('MAV_CMD', {}).get(ack.command, {}).name
        except:
            cmd_name = str(ack.command)

    # 获取结果名称
    result_name = MAV_RESULT_NAMES.get(ack.result, f"UNKNOWN({ack.result})")

    print("=" * 50)
    print(f"命令: {cmd_name} ({ack.command})")
    print(f"结果: {result_name}")
    print(f"进度: {ack.progress * 100:.0f}%" if ack.progress > 0 else "")
    # print(f"参数2: {ack.result_param2}")
    print(f"系统ID: {ack.target_system}")
    print(f"组件ID: {ack.target_component}")
    print("=" * 50)

    return ack


class MavDjiCmd(IntEnum):
    MAV_CMD_NAV_LAND = 21
    MAV_DJI_CMD_TRIGGER_FFC=40
    MAV_DJI_CMD_START_RECORD_POINT_CLOUD=41
    MAV_DJI_CMD_STOP_RECORD_POINT_CLOUD=42
    MAV_DJI_CMD_RESTORE_FACTORY_SETTINGS=43
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNC=44
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNCV2=45
    MAV_DJI_CMD_START_CONFIRM_LANDING=46
    MAV_DJI_CMD_START_FORCE_LANDING=47
    MAV_DJI_CMD_EXECUTE_EMERGENCY_BRAKE_ACTION=48
    MAV_DJI_CMD_CANCEL_EMERGENCY_BRAKE_ACTION=49
    MAV_DJI_CMD_START_SLOW_ROTATE_MOTOR=50
    MAV_DJI_CMD_STOP_SLOW_ROTATE_MOTOR=51
    MAV_DJI_CMD_ARREST_FLYING=52
    MAV_DJI_CMD_CANCEL_ARREST_FLYING=53
    MAV_DJI_CMD_OUTPUT_HIGH_POWER=54
    TUNNEL_PROTOCOL_ID_INJECT_HMS_ERROR=32770

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
print("waiting for heartbeat")
master.wait_heartbeat()
print("heartbeat received")
boot_time = time.time()

def send_command(command, param1=0, param2=0, param3=0, param4=0, param5=0, param6=0, param7=0, wait_ack=True, timeout=5):
    """发送 command_long 并可选接收 command_ack

    Args:
        command: 命令ID (如 MavDjiCmd.TRIGGER_FFC)
        param1~param7: 命令参数
        wait_ack: 是否等待并显示 ACK
        timeout: ACK 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时或未等待
    """
    print(f"发送命令: {command} ({int(command)})")
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        int(command),
        0,  # confirmation
        param1, param2, param3, param4, param5, param6, param7
    )

    if wait_ack:
        return recv_command_ack(timeout=timeout)
    return None


def ready():
    send_command(MavDjiCmd.MAV_DJI_CMD_EXECUTE_EMERGENCY_BRAKE_ACTION)

ready()
示例使用说明
获取

运行后，成功建立连接并发送指令

请求取消紧急制动动作

MAV_DJI_CMD_CANCEL_EMERGENCY_BRAKE_ACTION

示例
import time
from pymavlink import mavutil
from enum import IntEnum


# MAV_CMD 执行结果映射
MAV_RESULT_NAMES = {
    mavutil.mavlink.MAV_RESULT_ACCEPTED: "ACCEPTED",
    mavutil.mavlink.MAV_RESULT_TEMPORARILY_REJECTED: "TEMPORARILY_REJECTED",
    mavutil.mavlink.MAV_RESULT_DENIED: "DENIED",
    mavutil.mavlink.MAV_RESULT_UNSUPPORTED: "UNSUPPORTED",
    mavutil.mavlink.MAV_RESULT_FAILED: "FAILED",
    mavutil.mavlink.MAV_RESULT_IN_PROGRESS: "IN_PROGRESS",
}


def recv_command_ack(timeout=5):
    """接收并显示 command_ack 消息

    Args:
        timeout: 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时
    """
    print(f"等待 command_ack 响应 (超时 {timeout}s)...")

    ack = master.recv_match(type='COMMAND_ACK', blocking=True, timeout=timeout)

    if ack is None:
        print("错误: 命令响应超时")
        return None

    # 获取命令名称
    cmd_name = getattr(ack, 'command_name', None)
    if not cmd_name:
        # 尝试通过枚举查找命令名称
        try:
            cmd_name = mavutil.mavlink.enums.get('MAV_CMD', {}).get(ack.command, {}).name
        except:
            cmd_name = str(ack.command)

    # 获取结果名称
    result_name = MAV_RESULT_NAMES.get(ack.result, f"UNKNOWN({ack.result})")

    print("=" * 50)
    print(f"命令: {cmd_name} ({ack.command})")
    print(f"结果: {result_name}")
    print(f"进度: {ack.progress * 100:.0f}%" if ack.progress > 0 else "")
    # print(f"参数2: {ack.result_param2}")
    print(f"系统ID: {ack.target_system}")
    print(f"组件ID: {ack.target_component}")
    print("=" * 50)

    return ack


class MavDjiCmd(IntEnum):
    MAV_CMD_NAV_LAND = 21
    MAV_DJI_CMD_TRIGGER_FFC=40
    MAV_DJI_CMD_START_RECORD_POINT_CLOUD=41
    MAV_DJI_CMD_STOP_RECORD_POINT_CLOUD=42
    MAV_DJI_CMD_RESTORE_FACTORY_SETTINGS=43
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNC=44
    MAV_DJI_CMD_APPLY_HIGH_POWER_SYNCV2=45
    MAV_DJI_CMD_START_CONFIRM_LANDING=46
    MAV_DJI_CMD_START_FORCE_LANDING=47
    MAV_DJI_CMD_EXECUTE_EMERGENCY_BRAKE_ACTION=48
    MAV_DJI_CMD_CANCEL_EMERGENCY_BRAKE_ACTION=49
    MAV_DJI_CMD_START_SLOW_ROTATE_MOTOR=50
    MAV_DJI_CMD_STOP_SLOW_ROTATE_MOTOR=51
    MAV_DJI_CMD_ARREST_FLYING=52
    MAV_DJI_CMD_CANCEL_ARREST_FLYING=53
    MAV_DJI_CMD_OUTPUT_HIGH_POWER=54
    TUNNEL_PROTOCOL_ID_INJECT_HMS_ERROR=32770

timestamp = int(time.time() * 1000)
millis = int(time.time() * 1000)

master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
print("waiting for heartbeat")
master.wait_heartbeat()
print("heartbeat received")
boot_time = time.time()

def send_command(command, param1=0, param2=0, param3=0, param4=0, param5=0, param6=0, param7=0, wait_ack=True, timeout=5):
    """发送 command_long 并可选接收 command_ack

    Args:
        command: 命令ID (如 MavDjiCmd.TRIGGER_FFC)
        param1~param7: 命令参数
        wait_ack: 是否等待并显示 ACK
        timeout: ACK 超时时间（秒）

    Returns:
        ack 消息对象，None 表示超时或未等待
    """
    print(f"发送命令: {command} ({int(command)})")
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        int(command),
        0,  # confirmation
        param1, param2, param3, param4, param5, param6, param7
    )

    if wait_ack:
        return recv_command_ack(timeout=timeout)
    return None


def ready():
    send_command(MavDjiCmd.MAV_DJI_CMD_CANCEL_EMERGENCY_BRAKE_ACTION)

ready()
示例使用说明
获取

运行后，成功建立连接并发送指令

遥控器失联动作使能

RC_LOST_ENABLE

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
名称 枚举值 描述
DJI_FLIGHT_CONTROLLER_ENABLE_RC_LOST_ACTION 0 启用失控保护
DJI_FLIGHT_CONTROLLER_DISABLE_RC_LOST_ACTION 1 禁用失控保护
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“RC_LOST_ENABLE”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“RC_LOST_ENABLE”

输入要设置的数值

选择参数类型“1”

成功返回设置值

机臂灯状态

ARM_LIGHT

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：
名称 枚举值 描述
DJI_FLIGHT_CONTROLLER_LIGHT_OFF 0 关闭机臂灯。
DJI_FLIGHT_CONTROLLER_LIGHT_ON 1 开启机臂灯。
示例
import time
import sys
from pymavlink import mavutil

def connect_to_vehicle():
    """建立 UDP 监听连接（默认监听所有接口的 14550 端口）"""
    connection_string = "udpin:0.0.0.0:14550"
    print(f"Listening for MAVLink on: {connection_string}")
    vehicle = mavutil.mavlink_connection(connection_string)
    print("Waiting for HEARTBEAT (timeout: 10s)...")
    try:
        vehicle.wait_heartbeat(timeout=10)
        print(f"✅ Heartbeat received from system {vehicle.target_system}, component {vehicle.target_component}")
        return vehicle
    except Exception as e:
        print(f"❌ Connection failed: {str(e)}")
        print("Ensure your vehicle is sending MAVLink messages to 14550 port")
        print("Example QGroundControl setup: Settings > General > MAVLink > Listening Port = 14550")
        sys.exit(1)

def format_param_id(param_id):
    """格式化16字节参数ID（符合MAVLink协议要求）"""
    if len(param_id) > 16:
        raise ValueError("param_id cannot exceed 16 characters")

    param_id_bytes = bytearray(16)
    for i in range(min(len(param_id), 16)):
        param_id_bytes[i] = ord(param_id[i])

    # 长度<16时添加NULL终止符
    if len(param_id) < 16:
        param_id_bytes[len(param_id)] = 0

    return param_id_bytes

def send_param_set(vehicle, param_id, param_value, param_type):
    """发送参数设置请求"""
    vehicle.mav.param_set_send(
        vehicle.target_system,
        vehicle.target_component,
        format_param_id(param_id),
        param_value,
        param_type
    )
    print(f"📤 Sent PARAM_SET for '{param_id}' = {param_value} (type: {param_type})")

def send_param_request_read(vehicle, param_id=None, param_index=-1):
    """发送参数读取请求"""
    param_id_bytes = format_param_id(param_id) if param_id else bytearray(16)

    vehicle.mav.param_request_read_send(
        vehicle.target_system,
        vehicle.target_component,
        param_id_bytes,
        param_index
    )

    target = f"index {param_index}" if param_index >= 0 else f"ID '{param_id}'"
    print(f"📤 Sent PARAM_REQUEST_READ for {target}")

def send_param_request_list(vehicle):
    """发送请求所有参数列表"""
    vehicle.mav.param_request_list_send(
        vehicle.target_system,
        vehicle.target_component
    )
    print(f"📤 Sent PARAM_REQUEST_LIST to system {vehicle.target_system}, component {vehicle.target_component}")

def receive_param_value(vehicle, timeout=5.0):
    """接收参数值响应"""
    print(f"⏳ Waiting for PARAM_VALUE ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

    if not msg:
        print("❌ Timeout: No PARAM_VALUE received")
        return None

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值
    param_value = msg.param_value

    # 参数类型映射
    type_names = {
        0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
        5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
    }
    param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

    print("\n📥 Received PARAM_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  🧾 Type: {param_type}")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")

    return {
        'id': param_id,
        'value': param_value,
        'type': param_type,
        'index': msg.param_index,
        'count': msg.param_count
    }

def receive_all_param_values(vehicle, timeout=10.0):
    """接收所有参数值，直到收到 param_count 条消息后退出循环"""
    received_params = []
    expected_count = None
    print(f"⏳ Waiting for all PARAM_VALUE messages...")

    while True:
        msg = vehicle.recv_match(type='PARAM_VALUE', blocking=True, timeout=timeout)

        if not msg:
            print(f"❌ Timeout: Received {len(received_params)} parameters (expected {expected_count if expected_count else 'unknown'})")
            break

        param_id = msg.param_id
        param_value = msg.param_value

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE"
        }
        param_type = type_names.get(msg.param_type, f"UNKNOWN({msg.param_type})")

        # 首次收到消息时记录总参数数量
        if expected_count is None:
            expected_count = msg.param_count
            print(f"📊 Total parameters to receive: {expected_count}")

        print(f"  [{msg.param_index}/{expected_count}] {param_id} = {param_value} ({param_type})")

        received_params.append({
            'id': param_id,
            'value': param_value,
            'type': param_type,
            'index': msg.param_index,
            'count': msg.param_count
        })

        # 当已接收数量达到预期总数时退出循环
        if len(received_params) >= expected_count:
            print(f"\n✅ Successfully received all {expected_count} parameters")
            break

    return received_params

def main():
    print("🚀 Starting MAVLink PARAM_SET Tool (UDP Listening Mode)")
    print("-----------------------------------------------------")
    print("This script listens for MAVLink on ALL interfaces:14550")
    print("Ensure your vehicle is configured to send to this port")
    print("-----------------------------------------------------\n")

    vehicle = connect_to_vehicle()

    # 操作选择
    print("\nAvailable operations:")
    print("1. Set parameter value")
    print("2. Read parameter value (by ID or index)")
    print("3. Request all parameters (PARAM_REQUEST_LIST)")
    choice = input("\nSelect operation [1-3]: ").strip()

    if choice == '1':
        param_id = input("Enter parameter ID to set: ").strip()
        param_value = float(input("Enter new value: ").strip())

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  9 = DOUBLE")
        print("  4 = INT32")
        print("  1 = UINT8")
        param_type = int(input("Enter type number [8]: ").strip() or 8)

        send_param_set(vehicle, param_id, param_value, param_type)
        receive_param_value(vehicle)

    elif choice == '2':
        # mode = input("Read by [id/index]: ").lower()
        mode = 'id'
        if mode == 'id':
            param_id = input("Enter parameter ID: ").strip()
            send_param_request_read(vehicle, param_id=param_id)
            receive_param_value(vehicle)
        elif mode == 'index':
            param_index = int(input("Enter parameter index: "))
            send_param_request_read(vehicle, param_index=param_index)
            receive_param_value(vehicle)
        else:
            print("Invalid mode. Use 'id' or 'index'")

    elif choice == '3':
        print("\n📤 Requesting all parameters...")
        send_param_request_list(vehicle)
        all_params = receive_all_param_values(vehicle)
        print(f"\n📋 Total parameters received: {len(all_params)}")

    else:
        print("Invalid choice. Select 1, 2, or 3.")


if __name__ == "__main__":
    main()
示例使用说明
获取

运行后，成功建立连接

输入“2”选择读取参数

输入对应参数“ARM_LIGHT”

成功返回该参数对应的数值

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“ARM_LIGHT”

输入要设置的数值

选择参数类型“1”

成功返回设置值

作者：孙渝泓  创建时间：2026-06-15 18:30
最后编辑：孙渝泓  更新时间：2026-06-18 17:59
返回command = MAV_CMD_REQUEST_MESSAGE