云台功能

控制云台的俯仰和偏航角

MAV_CMD_DO_MOUNT_CONTROL（205）

MAV_MOUNT_MODE 为2时启动由MAVLink发送的 P/R/Y 角度控制。
当param1、param2、param3均为0时复位
返回消息：COMMAND_ACK
返回值 command 205 执行命令id
返回值 result 0或4 0：执行成功，4：执行失败

示例

import time
from pymavlink import mavutil
from datetime import datetime

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

# 连接无人机（UDP端口14550）
master = mavutil.mavlink_connection('udpin:0.0.0.0:14550')
print("等待无人机心跳...")
master.wait_heartbeat()
print("无人机已连接，开始发送云台控制命令")

# 云台控制参数设置
# 参数说明：
#   param1: 横滚角度 (0.0~1.0, 0.5=中立)
#   param2: 俯仰角度 (0.0~1.0, 0.5=中立)
#   param3: 偏航角度 (0.0~1.0, 0.5=中立)
#   param4: 模式 (0=无模式, 1=锁定, 2=跟踪)
#   param5: 方向 (0=无, 1=上, 2=下, 3=左, 4=右)
#   param6: 未使用
roll = 0   # 云台横滚中立位置
pitch = 20  # 云台俯仰中立位置
yaw = 0    # 云台偏航中立位置
mode = 0     # 无模式（直接控制角度）
direction = 0  # 无方向

# 发送云台控制命令
send_time = get_current_millis()
print(f"[{send_time}] 发送云台控制命令: roll={roll}, pitch={pitch}, yaw={yaw}, mode={mode}")

master.mav.command_long_send(
    master.target_system,
    master.target_component,  # 云台通常使用组件ID 2 (MAV_COMPONENT_MOUNT)
    mavutil.mavlink.MAV_CMD_DO_MOUNT_CONTROL,
    0,
    pitch,      # param1
    roll,     # param2
    yaw,       # param3
    0,      # param4
    0, # param5
    0,       # param6 (未使用)
    2
)

# 等待 COMMAND_ACK 响应（超时5秒）
print("等待云台控制命令响应...")
ack = master.recv_match(
    type='COMMAND_ACK',
    blocking=True,
    timeout=5.0
)

if ack:
    ack_time = get_current_millis()
    delay = ack_time - send_time
    print(f"[{ack_time}] 收到命令应答 (MAV_CMD_DO_MOUNT_CONTROL)")
    print(f" 命令ID: {ack.command}, 结果: {ack.result}, 延迟: {delay}ms")
else:
    print(f"[{get_current_millis()}] 超时：未收到云台控制命令应答")

print("云台控制命令发送完成")

执行结果

mavlink id 265 MOUNT_ORIENTATION

结果（使用QGC查看）

启用或禁用云台俯仰限位扩展

GIMBAL_EXT_ENAB

mavlink消息：

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：

示例

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”选择读取参数

输入对应参数“GIMBAL_EXT_ENAB”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“GIMBAL_EXT_ENAB”

输入要设置的数值

选择参数类型“1”

成功返回设置值

云台控制器的最大速度百分比

GIMBAL_MAXSPD_P
GIMBAL_MAXSPD_R
GIMBAL_MAXSPD_Y

mavlink消息：

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：

示例

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”选择读取参数

输入对应参数“GIMBAL_MAXSPD_P”/“GIMBAL_MAXSPD_R”/“GIMBAL_MAXSPD_Y”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“GIMBAL_MAXSPD_P”/“GIMBAL_MAXSPD_R”/“GIMBAL_MAXSPD_Y”

输入要设置的数值

选择参数类型“1”

成功返回设置值

设置云台控制器的平滑因子

GIMBAL_SMOOTH_P
GIMBAL_SMOOTH_R
GIMBAL_SMOOTH_Y

mavlink消息：

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：

平滑因子（Smooth Factor）。数值越大，云台的加速度越小，启动和停止动作会更加平缓。
取值范围：
0 ~ 30

示例

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”选择读取参数

输入对应参数“GIMBAL_SMOOTH_P”/“GIMBAL_SMOOTH_R”/“GIMBAL_SMOOTH_Y”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“GIMBAL_SMOOTH_P”/“GIMBAL_SMOOTH_R”/“GIMBAL_SMOOTH_Y”

输入要设置的数值

选择参数类型“1”

成功返回设置值

恢复云台的出厂设置

发送

COMMAND_LONG

示例

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  # M4TD不支持
    MAV_DJI_CMD_START_RECORD_POINT_CLOUD=41 # M4TD不支持
    MAV_DJI_CMD_STOP_RECORD_POINT_CLOUD=42  # M4TD不支持
    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_RESTORE_FACTORY_SETTINGS)

ready()

示例使用说明

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

mavlink id 265 MOUNT_ORIENTATION

结果（使用QGC查看）