相机管理

相机曝光模式

EXPOSURE_MODE

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：E_DjiCameraManagerExposureMode

示例

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

输入对应参数“EXPOSURE_MODE”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“EXPOSURE_MODE”

输入要设置的数值

选择参数类型“1”

成功返回设置值

相机感光度

ISO

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：E_DjiCameraManagerISO

注：该功能仅在相机曝光模式设置为“手动（Exposure_Manual）时可用。

示例

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

输入对应参数“ISO”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“ISO”

输入要设置的数值

选择参数类型“1”

成功返回设置值

快门速度

SHUTTER_SPEED

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

mavlink消息

接收PARAM_REQUEST_READ
发送PARAM_VALUE

参数：

注意：该功能仅在相机曝光模式设置为“手动（Exposure_Manual）”或“快门优先（SHUTTER_PRIORITY）”时可用。

示例

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

输入对应参数“SHUTTER_SPEED”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“SHUTTER_SPEED”

输入要设置的数值

选择参数类型“1”

成功返回设置值

曝光补偿值

EXPOSURE_COMP

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

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

输入对应参数“EXPOSURE_COMP”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“EXPOSURE_COMP”

输入要设置的数值

选择参数类型“1”

成功返回设置值

曝光锁定是否使能

AE_LOCK

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

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

输入对应参数“AE_LOCK”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“AE_LOCK”

输入要设置的参数

选择参数类型“1”

成功返回设置值

开始录像

MAV_CMD_VIDEO_START_CAPTURE 2500
返回command = MAV_CMD_REQUEST_MESSAGE
返回值 param1 2500 执行命令id
返回值 param2 0或1 是否执行成功
返回值 param3 失败代码
0 无异常
1 无法初始化相机控制
2 设置相机为录像模式
3 获取相机版本失败
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 videoStart():
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_VIDEO_START_CAPTURE,
        0,
        0, 0, 0, 0, 0, 0, 0
    )

videoStart()

执行示例

结果

具体录像行为相关信息需用遥控器查看

MAV_CMD_VIDEO_STOP_CAPTURE 2501

返回command = MAV_CMD_REQUEST_MESSAGE
返回值 param1 2501 执行命令id
返回值 param2 0或1 是否执行成功
返回值 param3 失败代码
0 无异常
1 无法初始化相机控制
2 设置相机为录像模式
3 获取相机版本失败
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 videoStop():
    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        mavutil.mavlink.MAV_CMD_VIDEO_STOP_CAPTURE,
        0,
        0, 0, 0, 0, 0, 0, 0
    )

videoStop()

执行示例

结果

具体录像行为相关信息需用遥控器查看

获取相机流媒体数据源范围

STRM_SRC_RNG

参数：

value（联合体说明）

该字段为联合体，同一时间仅有一个数组有效：

示例

import time
import sys
import struct
import ctypes
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 pack_uint8_xy(x, y):
    """打包 {uint8_t x; uint8_t y} 结构体 (2字节)"""
    return struct.pack('<BB', x, y)

def pack_float32_xy(x, y):
    """打包 {float32 x; float32 y} 结构体 (8字节)"""
    return struct.pack('<ff', x, y)

def pack_float32_lxlyrxry(LX, LY, RX, RY):
    """打包 {float32 LX; float32 LY; float32 RX; float32 RY;} 结构体 (16字节)"""
    return struct.pack('<ffff', LX, LY, RX, RY)

def pack_ering_mode(size, eringMode, minValue, maxValue):
    """打包 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;} 结构体 (25字节)

    Args:
        size: uint8_t, eringMode数组的有效长度
        eringMode: list of int (0-255), 16个uint8_t值
        minValue: uint32_t, 最小值
        maxValue: uint32_t, 最大值

    Returns:
        bytes: 打包后的二进制数据
    """
    # 确保eringMode是16字节
    eringMode_bytes = bytearray(16)
    for i in range(min(len(eringMode), 16)):
        eringMode_bytes[i] = eringMode[i] & 0xFF

    return struct.pack('<B16sII', size, eringMode_bytes, minValue, maxValue)

def format_param_value(param_value):
    """格式化128字节参数值（接受二进制输入）"""
    if isinstance(param_value, str):
        # 如果是十六进制字符串（如 "48656c6c6f"），先转换为字节
        if param_value.startswith('0x') or all(c in '0123456789abcdefABCDEF' for c in param_value.replace(' ', '')):
            hex_str = param_value.replace('0x', '').replace(' ', '')
            if len(hex_str) % 2 == 0:
                param_value = bytes.fromhex(hex_str)
            else:
                raise ValueError("Invalid hex string: must have even number of characters")
        else:
            # 普通字符串转字节
            param_value = param_value.encode('utf-8')

    if len(param_value) > 128:
        raise ValueError("param_value cannot exceed 128 bytes")

    param_value_bytes = bytearray(128)
    param_value_bytes[:len(param_value)] = param_value

    return param_value_bytes

def hex_dump(data, bytes_per_line=16):
    """格式化二进制数据为 hex + ASCII 展示（类 hexdump 风格）

    Returns:
        str: 格式化后的多行字符串
    """
    lines = []
    for i in range(0, len(data), bytes_per_line):
        chunk = data[i:i + bytes_per_line]
        hex_part = ' '.join(f'{b:02X}' for b in chunk)
        ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
        lines.append(f"  {i:04X}  {hex_part:<{bytes_per_line*3}}  {ascii_part}")
    return '\n'.join(lines)

def send_param_ext_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_ext_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_EXT_REQUEST_READ for {target}")

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

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

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

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

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值（直接获取128字节原始数据）
    param_value_raw = msg.param_value
    # byte_pointer = (ctypes.c_ubyte * 128).from_buffer_copy(msg.param_value)
    if isinstance(param_value_raw, (bytes, bytearray)):
        param_value_bytes = bytes(param_value_raw)
    # param_value_bytes = bytes(byte_pointer)
    elif isinstance(param_value_raw, str):
        # 替换无法编码的字符（如 \ufffd）为 0x00，还原为128字节
        param_value_bytes = param_value_raw.encode('latin1', errors='replace')
    else:
        param_value_bytes = bytes(param_value_raw)

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

    print("\n📥 Received PARAM_EXT_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(msg.get_msgbuf().hex())
    print(f"  🧾 Type: {param_type}  |  Data: {len(param_value_bytes)} bytes")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")
    print(hex_dump(param_value_bytes))

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


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

    while True:
        msg = vehicle.recv_match(type='PARAM_EXT_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_raw = msg.param_value

        # 直接获取128字节原始数据
        if isinstance(param_value_raw, (bytes, bytearray)):
            param_value_bytes = bytes(param_value_raw)
        elif isinstance(param_value_raw, str):
            # 替换无法编码的字符（如 \ufffd）为 0x00
            param_value_bytes = param_value_raw.encode('latin1', errors='replace')
        else:
            param_value_bytes = bytes(param_value_raw)

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE",
            10: "CHAR", 11: "BOOL", 12: "UINT8_ENUM", 13: "STRING"
        }
        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_type}][{len(param_value_bytes)}B]")
        print(hex_dump(param_value_bytes))

        received_params.append({
            'id': param_id,
            'value': param_value_bytes,
            '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 receive_param_ext_ack(vehicle, timeout=5.0):
    """接收参数设置确认"""
    print(f"⏳ Waiting for PARAM_EXT_ACK ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_EXT_ACK', blocking=True, timeout=timeout)

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

    param_id = msg.param_id
    param_value = msg.param_value

    # 结果代码映射
    result_codes = {
        0: "✅ PARAM_ACK_ACCEPTED",
        1: "⚠️ PARAM_ACK_VALUE_UNSUPPORTED",
        2: "❌ PARAM_ACK_FAILED",
        3: "🔄 PARAM_ACK_IN_PROGRESS"
    }
    result = result_codes.get(msg.param_result, f"❓ UNKNOWN({msg.param_result})")

    print("\n📥 Received PARAM_EXT_ACK:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  📌 Result: {result}")

    return {
        'id': param_id,
        'value': param_value,
        'result': result,
        'result_code': msg.param_result
    }

def main():
    print("🚀 Starting MAVLink PARAM_EXT 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. Read parameter value (by ID or index)")
    print("2. Set parameter value (raw/hex/string)")
    print("3. Request all parameters (PARAM_EXT_REQUEST_LIST)")
    print("4. Set struct: uint8_xy {x, y}")
    print("5. Set struct: float32_xy {x, y}")
    print("6. Set struct: float32_lxlyrxry {LX, LY, RX, RY}")
    print("7. Set struct: ering_mode {size, eringMode[16], minValue, maxValue}")
    choice = input("\nSelect operation [1-7]: ").strip()

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

    elif choice == '2':
        param_id = input("Enter parameter ID to set: ").strip()

        print("\nEnter parameter value (supports):")
        print("  - Plain string (e.g., hello)")
        print("  - Hex string (e.g., 48656c6c6f or 0x48656c6c6f)")
        print("  - Raw bytes (e.g., b'\\x48\\x65\\x6c\\x6c\\x6f')")
        param_value_input = input("Enter value: ").strip()

        # 解析输入值
        if param_value_input.startswith("b'") or param_value_input.startswith('b"'):
            # Python字节字面量格式
            param_value = eval(param_value_input)
        elif param_value_input.startswith('0x') or all(c in '0123456789abcdefABCDEF ' for c in param_value_input):
            # 十六进制格式
            hex_str = param_value_input.replace('0x', '').replace(' ', '')
            try:
                param_value = bytes.fromhex(hex_str)
            except ValueError:
                param_value = param_value_input.encode('utf-8')
        else:
            # 默认作为字符串处理
            param_value = param_value_input.encode('utf-8')

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  13 = STRING")
        print("  10 = CHAR (raw bytes)")
        print("  Other types require specific binary formats")
        param_type = int(input("Enter type number [8]: ") or 8)

        send_param_ext_set(vehicle, param_id, param_value, param_type)
        receive_param_ext_ack(vehicle)

    elif choice == '4':
        """设置 uint8_xy 结构体 {uint8_t x; uint8_t y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = int(input("Enter x (0-255): "))
        y = int(input("Enter y (0-255): "))
        param_value = pack_uint8_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '5':
        """设置 float32_xy 结构体 {float32 x; float32 y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = float(input("Enter x: "))
        y = float(input("Enter y: "))
        param_value = pack_float32_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '6':
        """设置 float32_lxlyrxry 结构体 {float32 LX; float32 LY; float32 RX; float32 RY;}"""
        param_id = input("Enter parameter ID: ").strip()
        LX = float(input("Enter LX: "))
        LY = float(input("Enter LY: "))
        RX = float(input("Enter RX: "))
        RY = float(input("Enter RY: "))
        param_value = pack_float32_lxlyrxry(LX, LY, RX, RY)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '7':
        """设置 ering_mode 结构体 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;}"""
        param_id = input("Enter parameter ID: ").strip()
        size = int(input("Enter size (0-16): "))
        eringMode = []
        print("Enter 16 eringMode values (0-255), one per line:")
        for i in range(16):
            val = int(input(f"  eringMode[{i}]: "))
            eringMode.append(val)
        minValue = int(input("Enter minValue: "))
        maxValue = int(input("Enter maxValue: "))
        param_value = pack_ering_mode(size, eringMode, minValue, maxValue)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

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

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


if __name__ == "__main__":
    main()

示例使用说明

获取

运行后，成功建立连接

输入“1”选择读取参数

输入对应参数“STRM_SRC_RNG”

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

获取相机流媒体数据源范围

STRM_SRC_RNG

获取

mavlink消息：

接收PARAM_EXT_REQUEST_READ
发送PARAM_EXT_VALUE

该字段为联合体，同一时间仅有一个数组有效：

示例

import time
import sys
import struct
import ctypes
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 pack_uint8_xy(x, y):
    """打包 {uint8_t x; uint8_t y} 结构体 (2字节)"""
    return struct.pack('<BB', x, y)

def pack_float32_xy(x, y):
    """打包 {float32 x; float32 y} 结构体 (8字节)"""
    return struct.pack('<ff', x, y)

def pack_float32_lxlyrxry(LX, LY, RX, RY):
    """打包 {float32 LX; float32 LY; float32 RX; float32 RY;} 结构体 (16字节)"""
    return struct.pack('<ffff', LX, LY, RX, RY)

def pack_ering_mode(size, eringMode, minValue, maxValue):
    """打包 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;} 结构体 (25字节)

    Args:
        size: uint8_t, eringMode数组的有效长度
        eringMode: list of int (0-255), 16个uint8_t值
        minValue: uint32_t, 最小值
        maxValue: uint32_t, 最大值

    Returns:
        bytes: 打包后的二进制数据
    """
    # 确保eringMode是16字节
    eringMode_bytes = bytearray(16)
    for i in range(min(len(eringMode), 16)):
        eringMode_bytes[i] = eringMode[i] & 0xFF

    return struct.pack('<B16sII', size, eringMode_bytes, minValue, maxValue)

def format_param_value(param_value):
    """格式化128字节参数值（接受二进制输入）"""
    if isinstance(param_value, str):
        # 如果是十六进制字符串（如 "48656c6c6f"），先转换为字节
        if param_value.startswith('0x') or all(c in '0123456789abcdefABCDEF' for c in param_value.replace(' ', '')):
            hex_str = param_value.replace('0x', '').replace(' ', '')
            if len(hex_str) % 2 == 0:
                param_value = bytes.fromhex(hex_str)
            else:
                raise ValueError("Invalid hex string: must have even number of characters")
        else:
            # 普通字符串转字节
            param_value = param_value.encode('utf-8')

    if len(param_value) > 128:
        raise ValueError("param_value cannot exceed 128 bytes")

    param_value_bytes = bytearray(128)
    param_value_bytes[:len(param_value)] = param_value

    return param_value_bytes

def hex_dump(data, bytes_per_line=16):
    """格式化二进制数据为 hex + ASCII 展示（类 hexdump 风格）

    Returns:
        str: 格式化后的多行字符串
    """
    lines = []
    for i in range(0, len(data), bytes_per_line):
        chunk = data[i:i + bytes_per_line]
        hex_part = ' '.join(f'{b:02X}' for b in chunk)
        ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
        lines.append(f"  {i:04X}  {hex_part:<{bytes_per_line*3}}  {ascii_part}")
    return '\n'.join(lines)

def send_param_ext_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_ext_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_EXT_REQUEST_READ for {target}")

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

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

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

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

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值（直接获取128字节原始数据）
    param_value_raw = msg.param_value
    # byte_pointer = (ctypes.c_ubyte * 128).from_buffer_copy(msg.param_value)
    if isinstance(param_value_raw, (bytes, bytearray)):
        param_value_bytes = bytes(param_value_raw)
    # param_value_bytes = bytes(byte_pointer)
    elif isinstance(param_value_raw, str):
        # 替换无法编码的字符（如 \ufffd）为 0x00，还原为128字节
        param_value_bytes = param_value_raw.encode('latin1', errors='replace')
    else:
        param_value_bytes = bytes(param_value_raw)

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

    print("\n📥 Received PARAM_EXT_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🧾 Type: {param_type}  |  Data: {len(param_value_bytes)} bytes")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")
    print(hex_dump(param_value_bytes))

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


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

    while True:
        msg = vehicle.recv_match(type='PARAM_EXT_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_raw = msg.param_value

        # 直接获取128字节原始数据
        if isinstance(param_value_raw, (bytes, bytearray)):
            param_value_bytes = bytes(param_value_raw)
        elif isinstance(param_value_raw, str):
            # 替换无法编码的字符（如 \ufffd）为 0x00
            param_value_bytes = param_value_raw.encode('latin1', errors='replace')
        else:
            param_value_bytes = bytes(param_value_raw)

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE",
            10: "CHAR", 11: "BOOL", 12: "UINT8_ENUM", 13: "STRING"
        }
        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_type}][{len(param_value_bytes)}B]")
        print(hex_dump(param_value_bytes))

        received_params.append({
            'id': param_id,
            'value': param_value_bytes,
            '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 receive_param_ext_ack(vehicle, timeout=5.0):
    """接收参数设置确认"""
    print(f"⏳ Waiting for PARAM_EXT_ACK ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_EXT_ACK', blocking=True, timeout=timeout)

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

    param_id = msg.param_id
    param_value = msg.param_value

    # 结果代码映射
    result_codes = {
        0: "✅ PARAM_ACK_ACCEPTED",
        1: "⚠️ PARAM_ACK_VALUE_UNSUPPORTED",
        2: "❌ PARAM_ACK_FAILED",
        3: "🔄 PARAM_ACK_IN_PROGRESS"
    }
    result = result_codes.get(msg.param_result, f"❓ UNKNOWN({msg.param_result})")

    print("\n📥 Received PARAM_EXT_ACK:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  📌 Result: {result}")

    return {
        'id': param_id,
        'value': param_value,
        'result': result,
        'result_code': msg.param_result
    }

def main():
    print("🚀 Starting MAVLink PARAM_EXT 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. Read parameter value (by ID or index)")
    print("2. Set parameter value (raw/hex/string)")
    print("3. Request all parameters (PARAM_EXT_REQUEST_LIST)")
    print("4. Set struct: uint8_xy {x, y}")
    print("5. Set struct: float32_xy {x, y}")
    print("6. Set struct: float32_lxlyrxry {LX, LY, RX, RY}")
    print("7. Set struct: ering_mode {size, eringMode[16], minValue, maxValue}")
    choice = input("\nSelect operation [1-7]: ").strip()

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

    elif choice == '2':
        param_id = input("Enter parameter ID to set: ").strip()

        print("\nEnter parameter value (supports):")
        print("  - Plain string (e.g., hello)")
        print("  - Hex string (e.g., 48656c6c6f or 0x48656c6c6f)")
        print("  - Raw bytes (e.g., b'\\x48\\x65\\x6c\\x6c\\x6f')")
        param_value_input = input("Enter value: ").strip()

        # 解析输入值
        if param_value_input.startswith("b'") or param_value_input.startswith('b"'):
            # Python字节字面量格式
            param_value = eval(param_value_input)
        elif param_value_input.startswith('0x') or all(c in '0123456789abcdefABCDEF ' for c in param_value_input):
            # 十六进制格式
            hex_str = param_value_input.replace('0x', '').replace(' ', '')
            try:
                param_value = bytes.fromhex(hex_str)
            except ValueError:
                param_value = param_value_input.encode('utf-8')
        else:
            # 默认作为字符串处理
            param_value = param_value_input.encode('utf-8')

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  13 = STRING")
        print("  10 = CHAR (raw bytes)")
        print("  Other types require specific binary formats")
        param_type = int(input("Enter type number [8]: ") or 8)

        send_param_ext_set(vehicle, param_id, param_value, param_type)
        receive_param_ext_ack(vehicle)

    elif choice == '4':
        """设置 uint8_xy 结构体 {uint8_t x; uint8_t y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = int(input("Enter x (0-255): "))
        y = int(input("Enter y (0-255): "))
        param_value = pack_uint8_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '5':
        """设置 float32_xy 结构体 {float32 x; float32 y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = float(input("Enter x: "))
        y = float(input("Enter y: "))
        param_value = pack_float32_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '6':
        """设置 float32_lxlyrxry 结构体 {float32 LX; float32 LY; float32 RX; float32 RY;}"""
        param_id = input("Enter parameter ID: ").strip()
        LX = float(input("Enter LX: "))
        LY = float(input("Enter LY: "))
        RX = float(input("Enter RX: "))
        RY = float(input("Enter RY: "))
        param_value = pack_float32_lxlyrxry(LX, LY, RX, RY)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '7':
        """设置 ering_mode 结构体 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;}"""
        param_id = input("Enter parameter ID: ").strip()
        size = int(input("Enter size (0-16): "))
        eringMode = []
        print("Enter 16 eringMode values (0-255), one per line:")
        for i in range(16):
            val = int(input(f"  eringMode[{i}]: "))
            eringMode.append(val)
        minValue = int(input("Enter minValue: "))
        maxValue = int(input("Enter maxValue: "))
        param_value = pack_ering_mode(size, eringMode, minValue, maxValue)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

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

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


if __name__ == "__main__":
    main()

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“STRM_SRC_RNG”

成功返回完整消息

相机流媒体数据源

STREAM_SOURCE

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

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

输入对应参数“STREAM_SOURCE”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“STREAM_SOURCE”

输入要设置的数值

选择参数类型“1”

成功返回设置值

获取照片存储格式范围

PHT_STORE_FMTRNG

mavlink消息：

接收PARAM_EXT_REQUEST_READ
发送PARAM_EXT_VALUE

参数：

union（同一时刻仅生效一个字段）

示例

import time
import sys
import struct
import ctypes
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 pack_uint8_xy(x, y):
    """打包 {uint8_t x; uint8_t y} 结构体 (2字节)"""
    return struct.pack('<BB', x, y)

def pack_float32_xy(x, y):
    """打包 {float32 x; float32 y} 结构体 (8字节)"""
    return struct.pack('<ff', x, y)

def pack_float32_lxlyrxry(LX, LY, RX, RY):
    """打包 {float32 LX; float32 LY; float32 RX; float32 RY;} 结构体 (16字节)"""
    return struct.pack('<ffff', LX, LY, RX, RY)

def pack_ering_mode(size, eringMode, minValue, maxValue):
    """打包 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;} 结构体 (25字节)

    Args:
        size: uint8_t, eringMode数组的有效长度
        eringMode: list of int (0-255), 16个uint8_t值
        minValue: uint32_t, 最小值
        maxValue: uint32_t, 最大值

    Returns:
        bytes: 打包后的二进制数据
    """
    # 确保eringMode是16字节
    eringMode_bytes = bytearray(16)
    for i in range(min(len(eringMode), 16)):
        eringMode_bytes[i] = eringMode[i] & 0xFF

    return struct.pack('<B16sII', size, eringMode_bytes, minValue, maxValue)

def format_param_value(param_value):
    """格式化128字节参数值（接受二进制输入）"""
    if isinstance(param_value, str):
        # 如果是十六进制字符串（如 "48656c6c6f"），先转换为字节
        if param_value.startswith('0x') or all(c in '0123456789abcdefABCDEF' for c in param_value.replace(' ', '')):
            hex_str = param_value.replace('0x', '').replace(' ', '')
            if len(hex_str) % 2 == 0:
                param_value = bytes.fromhex(hex_str)
            else:
                raise ValueError("Invalid hex string: must have even number of characters")
        else:
            # 普通字符串转字节
            param_value = param_value.encode('utf-8')

    if len(param_value) > 128:
        raise ValueError("param_value cannot exceed 128 bytes")

    param_value_bytes = bytearray(128)
    param_value_bytes[:len(param_value)] = param_value

    return param_value_bytes

def hex_dump(data, bytes_per_line=16):
    """格式化二进制数据为 hex + ASCII 展示（类 hexdump 风格）

    Returns:
        str: 格式化后的多行字符串
    """
    lines = []
    for i in range(0, len(data), bytes_per_line):
        chunk = data[i:i + bytes_per_line]
        hex_part = ' '.join(f'{b:02X}' for b in chunk)
        ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
        lines.append(f"  {i:04X}  {hex_part:<{bytes_per_line*3}}  {ascii_part}")
    return '\n'.join(lines)

def send_param_ext_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_ext_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_EXT_REQUEST_READ for {target}")

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

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

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

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

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值（直接获取128字节原始数据）
    param_value_raw = msg.param_value
    # byte_pointer = (ctypes.c_ubyte * 128).from_buffer_copy(msg.param_value)
    if isinstance(param_value_raw, (bytes, bytearray)):
        param_value_bytes = bytes(param_value_raw)
    # param_value_bytes = bytes(byte_pointer)
    elif isinstance(param_value_raw, str):
        # 替换无法编码的字符（如 \ufffd）为 0x00，还原为128字节
        param_value_bytes = param_value_raw.encode('latin1', errors='replace')
    else:
        param_value_bytes = bytes(param_value_raw)

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

    print("\n📥 Received PARAM_EXT_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🧾 Type: {param_type}  |  Data: {len(param_value_bytes)} bytes")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")
    print(hex_dump(param_value_bytes))

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


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

    while True:
        msg = vehicle.recv_match(type='PARAM_EXT_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_raw = msg.param_value

        # 直接获取128字节原始数据
        if isinstance(param_value_raw, (bytes, bytearray)):
            param_value_bytes = bytes(param_value_raw)
        elif isinstance(param_value_raw, str):
            # 替换无法编码的字符（如 \ufffd）为 0x00
            param_value_bytes = param_value_raw.encode('latin1', errors='replace')
        else:
            param_value_bytes = bytes(param_value_raw)

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE",
            10: "CHAR", 11: "BOOL", 12: "UINT8_ENUM", 13: "STRING"
        }
        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_type}][{len(param_value_bytes)}B]")
        print(hex_dump(param_value_bytes))

        received_params.append({
            'id': param_id,
            'value': param_value_bytes,
            '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 receive_param_ext_ack(vehicle, timeout=5.0):
    """接收参数设置确认"""
    print(f"⏳ Waiting for PARAM_EXT_ACK ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_EXT_ACK', blocking=True, timeout=timeout)

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

    param_id = msg.param_id
    param_value = msg.param_value

    # 结果代码映射
    result_codes = {
        0: "✅ PARAM_ACK_ACCEPTED",
        1: "⚠️ PARAM_ACK_VALUE_UNSUPPORTED",
        2: "❌ PARAM_ACK_FAILED",
        3: "🔄 PARAM_ACK_IN_PROGRESS"
    }
    result = result_codes.get(msg.param_result, f"❓ UNKNOWN({msg.param_result})")

    print("\n📥 Received PARAM_EXT_ACK:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  📌 Result: {result}")

    return {
        'id': param_id,
        'value': param_value,
        'result': result,
        'result_code': msg.param_result
    }

def main():
    print("🚀 Starting MAVLink PARAM_EXT 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. Read parameter value (by ID or index)")
    print("2. Set parameter value (raw/hex/string)")
    print("3. Request all parameters (PARAM_EXT_REQUEST_LIST)")
    print("4. Set struct: uint8_xy {x, y}")
    print("5. Set struct: float32_xy {x, y}")
    print("6. Set struct: float32_lxlyrxry {LX, LY, RX, RY}")
    print("7. Set struct: ering_mode {size, eringMode[16], minValue, maxValue}")
    choice = input("\nSelect operation [1-7]: ").strip()

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

    elif choice == '2':
        param_id = input("Enter parameter ID to set: ").strip()

        print("\nEnter parameter value (supports):")
        print("  - Plain string (e.g., hello)")
        print("  - Hex string (e.g., 48656c6c6f or 0x48656c6c6f)")
        print("  - Raw bytes (e.g., b'\\x48\\x65\\x6c\\x6c\\x6f')")
        param_value_input = input("Enter value: ").strip()

        # 解析输入值
        if param_value_input.startswith("b'") or param_value_input.startswith('b"'):
            # Python字节字面量格式
            param_value = eval(param_value_input)
        elif param_value_input.startswith('0x') or all(c in '0123456789abcdefABCDEF ' for c in param_value_input):
            # 十六进制格式
            hex_str = param_value_input.replace('0x', '').replace(' ', '')
            try:
                param_value = bytes.fromhex(hex_str)
            except ValueError:
                param_value = param_value_input.encode('utf-8')
        else:
            # 默认作为字符串处理
            param_value = param_value_input.encode('utf-8')

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  13 = STRING")
        print("  10 = CHAR (raw bytes)")
        print("  Other types require specific binary formats")
        param_type = int(input("Enter type number [8]: ") or 8)

        send_param_ext_set(vehicle, param_id, param_value, param_type)
        receive_param_ext_ack(vehicle)

    elif choice == '4':
        """设置 uint8_xy 结构体 {uint8_t x; uint8_t y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = int(input("Enter x (0-255): "))
        y = int(input("Enter y (0-255): "))
        param_value = pack_uint8_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '5':
        """设置 float32_xy 结构体 {float32 x; float32 y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = float(input("Enter x: "))
        y = float(input("Enter y: "))
        param_value = pack_float32_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '6':
        """设置 float32_lxlyrxry 结构体 {float32 LX; float32 LY; float32 RX; float32 RY;}"""
        param_id = input("Enter parameter ID: ").strip()
        LX = float(input("Enter LX: "))
        LY = float(input("Enter LY: "))
        RX = float(input("Enter RX: "))
        RY = float(input("Enter RY: "))
        param_value = pack_float32_lxlyrxry(LX, LY, RX, RY)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '7':
        """设置 ering_mode 结构体 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;}"""
        param_id = input("Enter parameter ID: ").strip()
        size = int(input("Enter size (0-16): "))
        eringMode = []
        print("Enter 16 eringMode values (0-255), one per line:")
        for i in range(16):
            val = int(input(f"  eringMode[{i}]: "))
            eringMode.append(val)
        minValue = int(input("Enter minValue: "))
        maxValue = int(input("Enter maxValue: "))
        param_value = pack_ering_mode(size, eringMode, minValue, maxValue)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

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

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


if __name__ == "__main__":
    main()

示例使用说明

运行后，成功建立连接

输入“1”选择读取参数

输入对应参数“PHT_STORE_FMTRNG”

成功返回完整消息

照片存储格式

PHOTO_FORMAT

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

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

输入对应参数“PHOTO_FORMAT”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“PHOTO_FORMAT”

选择参数类型“1”

发送成功返回设置值

获取视频存储格式范围

VID_FMT_RNG

mavlink消息：

接收PARAM_EXT_REQUEST_READ
发送PARAM_EXT_VALUE

参数：

union（同一时刻仅生效一个字段）

示例

import time
import sys
import struct
import ctypes
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 pack_uint8_xy(x, y):
    """打包 {uint8_t x; uint8_t y} 结构体 (2字节)"""
    return struct.pack('<BB', x, y)

def pack_float32_xy(x, y):
    """打包 {float32 x; float32 y} 结构体 (8字节)"""
    return struct.pack('<ff', x, y)

def pack_float32_lxlyrxry(LX, LY, RX, RY):
    """打包 {float32 LX; float32 LY; float32 RX; float32 RY;} 结构体 (16字节)"""
    return struct.pack('<ffff', LX, LY, RX, RY)

def pack_ering_mode(size, eringMode, minValue, maxValue):
    """打包 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;} 结构体 (25字节)

    Args:
        size: uint8_t, eringMode数组的有效长度
        eringMode: list of int (0-255), 16个uint8_t值
        minValue: uint32_t, 最小值
        maxValue: uint32_t, 最大值

    Returns:
        bytes: 打包后的二进制数据
    """
    # 确保eringMode是16字节
    eringMode_bytes = bytearray(16)
    for i in range(min(len(eringMode), 16)):
        eringMode_bytes[i] = eringMode[i] & 0xFF

    return struct.pack('<B16sII', size, eringMode_bytes, minValue, maxValue)

def format_param_value(param_value):
    """格式化128字节参数值（接受二进制输入）"""
    if isinstance(param_value, str):
        # 如果是十六进制字符串（如 "48656c6c6f"），先转换为字节
        if param_value.startswith('0x') or all(c in '0123456789abcdefABCDEF' for c in param_value.replace(' ', '')):
            hex_str = param_value.replace('0x', '').replace(' ', '')
            if len(hex_str) % 2 == 0:
                param_value = bytes.fromhex(hex_str)
            else:
                raise ValueError("Invalid hex string: must have even number of characters")
        else:
            # 普通字符串转字节
            param_value = param_value.encode('utf-8')

    if len(param_value) > 128:
        raise ValueError("param_value cannot exceed 128 bytes")

    param_value_bytes = bytearray(128)
    param_value_bytes[:len(param_value)] = param_value

    return param_value_bytes

def hex_dump(data, bytes_per_line=16):
    """格式化二进制数据为 hex + ASCII 展示（类 hexdump 风格）

    Returns:
        str: 格式化后的多行字符串
    """
    lines = []
    for i in range(0, len(data), bytes_per_line):
        chunk = data[i:i + bytes_per_line]
        hex_part = ' '.join(f'{b:02X}' for b in chunk)
        ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
        lines.append(f"  {i:04X}  {hex_part:<{bytes_per_line*3}}  {ascii_part}")
    return '\n'.join(lines)

def send_param_ext_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_ext_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_EXT_REQUEST_READ for {target}")

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

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

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

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

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值（直接获取128字节原始数据）
    param_value_raw = msg.param_value
    # byte_pointer = (ctypes.c_ubyte * 128).from_buffer_copy(msg.param_value)
    if isinstance(param_value_raw, (bytes, bytearray)):
        param_value_bytes = bytes(param_value_raw)
    # param_value_bytes = bytes(byte_pointer)
    elif isinstance(param_value_raw, str):
        # 替换无法编码的字符（如 \ufffd）为 0x00，还原为128字节
        param_value_bytes = param_value_raw.encode('latin1', errors='replace')
    else:
        param_value_bytes = bytes(param_value_raw)

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

    print("\n📥 Received PARAM_EXT_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🧾 Type: {param_type}  |  Data: {len(param_value_bytes)} bytes")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")
    print(hex_dump(param_value_bytes))

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


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

    while True:
        msg = vehicle.recv_match(type='PARAM_EXT_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_raw = msg.param_value

        # 直接获取128字节原始数据
        if isinstance(param_value_raw, (bytes, bytearray)):
            param_value_bytes = bytes(param_value_raw)
        elif isinstance(param_value_raw, str):
            # 替换无法编码的字符（如 \ufffd）为 0x00
            param_value_bytes = param_value_raw.encode('latin1', errors='replace')
        else:
            param_value_bytes = bytes(param_value_raw)

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE",
            10: "CHAR", 11: "BOOL", 12: "UINT8_ENUM", 13: "STRING"
        }
        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_type}][{len(param_value_bytes)}B]")
        print(hex_dump(param_value_bytes))

        received_params.append({
            'id': param_id,
            'value': param_value_bytes,
            '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 receive_param_ext_ack(vehicle, timeout=5.0):
    """接收参数设置确认"""
    print(f"⏳ Waiting for PARAM_EXT_ACK ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_EXT_ACK', blocking=True, timeout=timeout)

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

    param_id = msg.param_id
    param_value = msg.param_value

    # 结果代码映射
    result_codes = {
        0: "✅ PARAM_ACK_ACCEPTED",
        1: "⚠️ PARAM_ACK_VALUE_UNSUPPORTED",
        2: "❌ PARAM_ACK_FAILED",
        3: "🔄 PARAM_ACK_IN_PROGRESS"
    }
    result = result_codes.get(msg.param_result, f"❓ UNKNOWN({msg.param_result})")

    print("\n📥 Received PARAM_EXT_ACK:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  📌 Result: {result}")

    return {
        'id': param_id,
        'value': param_value,
        'result': result,
        'result_code': msg.param_result
    }

def main():
    print("🚀 Starting MAVLink PARAM_EXT 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. Read parameter value (by ID or index)")
    print("2. Set parameter value (raw/hex/string)")
    print("3. Request all parameters (PARAM_EXT_REQUEST_LIST)")
    print("4. Set struct: uint8_xy {x, y}")
    print("5. Set struct: float32_xy {x, y}")
    print("6. Set struct: float32_lxlyrxry {LX, LY, RX, RY}")
    print("7. Set struct: ering_mode {size, eringMode[16], minValue, maxValue}")
    choice = input("\nSelect operation [1-7]: ").strip()

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

    elif choice == '2':
        param_id = input("Enter parameter ID to set: ").strip()

        print("\nEnter parameter value (supports):")
        print("  - Plain string (e.g., hello)")
        print("  - Hex string (e.g., 48656c6c6f or 0x48656c6c6f)")
        print("  - Raw bytes (e.g., b'\\x48\\x65\\x6c\\x6c\\x6f')")
        param_value_input = input("Enter value: ").strip()

        # 解析输入值
        if param_value_input.startswith("b'") or param_value_input.startswith('b"'):
            # Python字节字面量格式
            param_value = eval(param_value_input)
        elif param_value_input.startswith('0x') or all(c in '0123456789abcdefABCDEF ' for c in param_value_input):
            # 十六进制格式
            hex_str = param_value_input.replace('0x', '').replace(' ', '')
            try:
                param_value = bytes.fromhex(hex_str)
            except ValueError:
                param_value = param_value_input.encode('utf-8')
        else:
            # 默认作为字符串处理
            param_value = param_value_input.encode('utf-8')

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  13 = STRING")
        print("  10 = CHAR (raw bytes)")
        print("  Other types require specific binary formats")
        param_type = int(input("Enter type number [8]: ") or 8)

        send_param_ext_set(vehicle, param_id, param_value, param_type)
        receive_param_ext_ack(vehicle)

    elif choice == '4':
        """设置 uint8_xy 结构体 {uint8_t x; uint8_t y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = int(input("Enter x (0-255): "))
        y = int(input("Enter y (0-255): "))
        param_value = pack_uint8_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '5':
        """设置 float32_xy 结构体 {float32 x; float32 y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = float(input("Enter x: "))
        y = float(input("Enter y: "))
        param_value = pack_float32_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '6':
        """设置 float32_lxlyrxry 结构体 {float32 LX; float32 LY; float32 RX; float32 RY;}"""
        param_id = input("Enter parameter ID: ").strip()
        LX = float(input("Enter LX: "))
        LY = float(input("Enter LY: "))
        RX = float(input("Enter RX: "))
        RY = float(input("Enter RY: "))
        param_value = pack_float32_lxlyrxry(LX, LY, RX, RY)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '7':
        """设置 ering_mode 结构体 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;}"""
        param_id = input("Enter parameter ID: ").strip()
        size = int(input("Enter size (0-16): "))
        eringMode = []
        print("Enter 16 eringMode values (0-255), one per line:")
        for i in range(16):
            val = int(input(f"  eringMode[{i}]: "))
            eringMode.append(val)
        minValue = int(input("Enter minValue: "))
        maxValue = int(input("Enter maxValue: "))
        param_value = pack_ering_mode(size, eringMode, minValue, maxValue)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

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

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


if __name__ == "__main__":
    main()

示例使用说明

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“VID_FMT_RNG”

成功返回完整消息

视频存储格式

VID_STORE_FMT

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

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

输入对应参数“VID_STORE_FMT”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“VID_STORE_FMT”

输入要设置的数值

选择参数类型“1”

发送成功返回设置值

获取照片比例范围

PHT_RATIO_RNG

mavlink消息：

接收PARAM_EXT_REQUEST_READ
发送PARAM_EXT_VALUE

参数：

union（同一时刻仅生效一个字段）

示例

import time
import sys
import struct
import ctypes
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 pack_uint8_xy(x, y):
    """打包 {uint8_t x; uint8_t y} 结构体 (2字节)"""
    return struct.pack('<BB', x, y)

def pack_float32_xy(x, y):
    """打包 {float32 x; float32 y} 结构体 (8字节)"""
    return struct.pack('<ff', x, y)

def pack_float32_lxlyrxry(LX, LY, RX, RY):
    """打包 {float32 LX; float32 LY; float32 RX; float32 RY;} 结构体 (16字节)"""
    return struct.pack('<ffff', LX, LY, RX, RY)

def pack_ering_mode(size, eringMode, minValue, maxValue):
    """打包 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;} 结构体 (25字节)

    Args:
        size: uint8_t, eringMode数组的有效长度
        eringMode: list of int (0-255), 16个uint8_t值
        minValue: uint32_t, 最小值
        maxValue: uint32_t, 最大值

    Returns:
        bytes: 打包后的二进制数据
    """
    # 确保eringMode是16字节
    eringMode_bytes = bytearray(16)
    for i in range(min(len(eringMode), 16)):
        eringMode_bytes[i] = eringMode[i] & 0xFF

    return struct.pack('<B16sII', size, eringMode_bytes, minValue, maxValue)

def format_param_value(param_value):
    """格式化128字节参数值（接受二进制输入）"""
    if isinstance(param_value, str):
        # 如果是十六进制字符串（如 "48656c6c6f"），先转换为字节
        if param_value.startswith('0x') or all(c in '0123456789abcdefABCDEF' for c in param_value.replace(' ', '')):
            hex_str = param_value.replace('0x', '').replace(' ', '')
            if len(hex_str) % 2 == 0:
                param_value = bytes.fromhex(hex_str)
            else:
                raise ValueError("Invalid hex string: must have even number of characters")
        else:
            # 普通字符串转字节
            param_value = param_value.encode('utf-8')

    if len(param_value) > 128:
        raise ValueError("param_value cannot exceed 128 bytes")

    param_value_bytes = bytearray(128)
    param_value_bytes[:len(param_value)] = param_value

    return param_value_bytes

def hex_dump(data, bytes_per_line=16):
    """格式化二进制数据为 hex + ASCII 展示（类 hexdump 风格）

    Returns:
        str: 格式化后的多行字符串
    """
    lines = []
    for i in range(0, len(data), bytes_per_line):
        chunk = data[i:i + bytes_per_line]
        hex_part = ' '.join(f'{b:02X}' for b in chunk)
        ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
        lines.append(f"  {i:04X}  {hex_part:<{bytes_per_line*3}}  {ascii_part}")
    return '\n'.join(lines)

def send_param_ext_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_ext_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_EXT_REQUEST_READ for {target}")

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

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

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

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

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值（直接获取128字节原始数据）
    param_value_raw = msg.param_value
    # byte_pointer = (ctypes.c_ubyte * 128).from_buffer_copy(msg.param_value)
    if isinstance(param_value_raw, (bytes, bytearray)):
        param_value_bytes = bytes(param_value_raw)
    # param_value_bytes = bytes(byte_pointer)
    elif isinstance(param_value_raw, str):
        # 替换无法编码的字符（如 \ufffd）为 0x00，还原为128字节
        param_value_bytes = param_value_raw.encode('latin1', errors='replace')
    else:
        param_value_bytes = bytes(param_value_raw)

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

    print("\n📥 Received PARAM_EXT_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🧾 Type: {param_type}  |  Data: {len(param_value_bytes)} bytes")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")
    print(hex_dump(param_value_bytes))

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


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

    while True:
        msg = vehicle.recv_match(type='PARAM_EXT_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_raw = msg.param_value

        # 直接获取128字节原始数据
        if isinstance(param_value_raw, (bytes, bytearray)):
            param_value_bytes = bytes(param_value_raw)
        elif isinstance(param_value_raw, str):
            # 替换无法编码的字符（如 \ufffd）为 0x00
            param_value_bytes = param_value_raw.encode('latin1', errors='replace')
        else:
            param_value_bytes = bytes(param_value_raw)

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE",
            10: "CHAR", 11: "BOOL", 12: "UINT8_ENUM", 13: "STRING"
        }
        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_type}][{len(param_value_bytes)}B]")
        print(hex_dump(param_value_bytes))

        received_params.append({
            'id': param_id,
            'value': param_value_bytes,
            '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 receive_param_ext_ack(vehicle, timeout=5.0):
    """接收参数设置确认"""
    print(f"⏳ Waiting for PARAM_EXT_ACK ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_EXT_ACK', blocking=True, timeout=timeout)

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

    param_id = msg.param_id
    param_value = msg.param_value

    # 结果代码映射
    result_codes = {
        0: "✅ PARAM_ACK_ACCEPTED",
        1: "⚠️ PARAM_ACK_VALUE_UNSUPPORTED",
        2: "❌ PARAM_ACK_FAILED",
        3: "🔄 PARAM_ACK_IN_PROGRESS"
    }
    result = result_codes.get(msg.param_result, f"❓ UNKNOWN({msg.param_result})")

    print("\n📥 Received PARAM_EXT_ACK:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  📌 Result: {result}")

    return {
        'id': param_id,
        'value': param_value,
        'result': result,
        'result_code': msg.param_result
    }

def main():
    print("🚀 Starting MAVLink PARAM_EXT 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. Read parameter value (by ID or index)")
    print("2. Set parameter value (raw/hex/string)")
    print("3. Request all parameters (PARAM_EXT_REQUEST_LIST)")
    print("4. Set struct: uint8_xy {x, y}")
    print("5. Set struct: float32_xy {x, y}")
    print("6. Set struct: float32_lxlyrxry {LX, LY, RX, RY}")
    print("7. Set struct: ering_mode {size, eringMode[16], minValue, maxValue}")
    choice = input("\nSelect operation [1-7]: ").strip()

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

    elif choice == '2':
        param_id = input("Enter parameter ID to set: ").strip()

        print("\nEnter parameter value (supports):")
        print("  - Plain string (e.g., hello)")
        print("  - Hex string (e.g., 48656c6c6f or 0x48656c6c6f)")
        print("  - Raw bytes (e.g., b'\\x48\\x65\\x6c\\x6c\\x6f')")
        param_value_input = input("Enter value: ").strip()

        # 解析输入值
        if param_value_input.startswith("b'") or param_value_input.startswith('b"'):
            # Python字节字面量格式
            param_value = eval(param_value_input)
        elif param_value_input.startswith('0x') or all(c in '0123456789abcdefABCDEF ' for c in param_value_input):
            # 十六进制格式
            hex_str = param_value_input.replace('0x', '').replace(' ', '')
            try:
                param_value = bytes.fromhex(hex_str)
            except ValueError:
                param_value = param_value_input.encode('utf-8')
        else:
            # 默认作为字符串处理
            param_value = param_value_input.encode('utf-8')

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  13 = STRING")
        print("  10 = CHAR (raw bytes)")
        print("  Other types require specific binary formats")
        param_type = int(input("Enter type number [8]: ") or 8)

        send_param_ext_set(vehicle, param_id, param_value, param_type)
        receive_param_ext_ack(vehicle)

    elif choice == '4':
        """设置 uint8_xy 结构体 {uint8_t x; uint8_t y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = int(input("Enter x (0-255): "))
        y = int(input("Enter y (0-255): "))
        param_value = pack_uint8_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '5':
        """设置 float32_xy 结构体 {float32 x; float32 y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = float(input("Enter x: "))
        y = float(input("Enter y: "))
        param_value = pack_float32_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '6':
        """设置 float32_lxlyrxry 结构体 {float32 LX; float32 LY; float32 RX; float32 RY;}"""
        param_id = input("Enter parameter ID: ").strip()
        LX = float(input("Enter LX: "))
        LY = float(input("Enter LY: "))
        RX = float(input("Enter RX: "))
        RY = float(input("Enter RY: "))
        param_value = pack_float32_lxlyrxry(LX, LY, RX, RY)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '7':
        """设置 ering_mode 结构体 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;}"""
        param_id = input("Enter parameter ID: ").strip()
        size = int(input("Enter size (0-16): "))
        eringMode = []
        print("Enter 16 eringMode values (0-255), one per line:")
        for i in range(16):
            val = int(input(f"  eringMode[{i}]: "))
            eringMode.append(val)
        minValue = int(input("Enter minValue: "))
        maxValue = int(input("Enter maxValue: "))
        param_value = pack_ering_mode(size, eringMode, minValue, maxValue)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

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

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


if __name__ == "__main__":
    main()

示例使用说明

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“PHT_RATIO_RNG”

成功返回完整消息

照片比例

PHOTO_RATIO

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

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

输入对应参数“PHOTO_RATIO”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“PHOTO_RATIO”

输入要设置的数值

选择参数类型“1”

成功返回设置值

获取夜景模式范围

NGT_SCN_MD_RNG

mavlink消息：

接收PARAM_EXT_REQUEST_READ
发送PARAM_EXT_VALUE

参数：

union（同一时刻仅生效一个字段）

示例

import time
import sys
import struct
import ctypes
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 pack_uint8_xy(x, y):
    """打包 {uint8_t x; uint8_t y} 结构体 (2字节)"""
    return struct.pack('<BB', x, y)

def pack_float32_xy(x, y):
    """打包 {float32 x; float32 y} 结构体 (8字节)"""
    return struct.pack('<ff', x, y)

def pack_float32_lxlyrxry(LX, LY, RX, RY):
    """打包 {float32 LX; float32 LY; float32 RX; float32 RY;} 结构体 (16字节)"""
    return struct.pack('<ffff', LX, LY, RX, RY)

def pack_ering_mode(size, eringMode, minValue, maxValue):
    """打包 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;} 结构体 (25字节)

    Args:
        size: uint8_t, eringMode数组的有效长度
        eringMode: list of int (0-255), 16个uint8_t值
        minValue: uint32_t, 最小值
        maxValue: uint32_t, 最大值

    Returns:
        bytes: 打包后的二进制数据
    """
    # 确保eringMode是16字节
    eringMode_bytes = bytearray(16)
    for i in range(min(len(eringMode), 16)):
        eringMode_bytes[i] = eringMode[i] & 0xFF

    return struct.pack('<B16sII', size, eringMode_bytes, minValue, maxValue)

def format_param_value(param_value):
    """格式化128字节参数值（接受二进制输入）"""
    if isinstance(param_value, str):
        # 如果是十六进制字符串（如 "48656c6c6f"），先转换为字节
        if param_value.startswith('0x') or all(c in '0123456789abcdefABCDEF' for c in param_value.replace(' ', '')):
            hex_str = param_value.replace('0x', '').replace(' ', '')
            if len(hex_str) % 2 == 0:
                param_value = bytes.fromhex(hex_str)
            else:
                raise ValueError("Invalid hex string: must have even number of characters")
        else:
            # 普通字符串转字节
            param_value = param_value.encode('utf-8')

    if len(param_value) > 128:
        raise ValueError("param_value cannot exceed 128 bytes")

    param_value_bytes = bytearray(128)
    param_value_bytes[:len(param_value)] = param_value

    return param_value_bytes

def hex_dump(data, bytes_per_line=16):
    """格式化二进制数据为 hex + ASCII 展示（类 hexdump 风格）

    Returns:
        str: 格式化后的多行字符串
    """
    lines = []
    for i in range(0, len(data), bytes_per_line):
        chunk = data[i:i + bytes_per_line]
        hex_part = ' '.join(f'{b:02X}' for b in chunk)
        ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
        lines.append(f"  {i:04X}  {hex_part:<{bytes_per_line*3}}  {ascii_part}")
    return '\n'.join(lines)

def send_param_ext_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_ext_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_EXT_REQUEST_READ for {target}")

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

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

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

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

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值（直接获取128字节原始数据）
    param_value_raw = msg.param_value
    # byte_pointer = (ctypes.c_ubyte * 128).from_buffer_copy(msg.param_value)
    if isinstance(param_value_raw, (bytes, bytearray)):
        param_value_bytes = bytes(param_value_raw)
    # param_value_bytes = bytes(byte_pointer)
    elif isinstance(param_value_raw, str):
        # 替换无法编码的字符（如 \ufffd）为 0x00，还原为128字节
        param_value_bytes = param_value_raw.encode('latin1', errors='replace')
    else:
        param_value_bytes = bytes(param_value_raw)

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

    print("\n📥 Received PARAM_EXT_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🧾 Type: {param_type}  |  Data: {len(param_value_bytes)} bytes")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")
    print(hex_dump(param_value_bytes))

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


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

    while True:
        msg = vehicle.recv_match(type='PARAM_EXT_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_raw = msg.param_value

        # 直接获取128字节原始数据
        if isinstance(param_value_raw, (bytes, bytearray)):
            param_value_bytes = bytes(param_value_raw)
        elif isinstance(param_value_raw, str):
            # 替换无法编码的字符（如 \ufffd）为 0x00
            param_value_bytes = param_value_raw.encode('latin1', errors='replace')
        else:
            param_value_bytes = bytes(param_value_raw)

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE",
            10: "CHAR", 11: "BOOL", 12: "UINT8_ENUM", 13: "STRING"
        }
        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_type}][{len(param_value_bytes)}B]")
        print(hex_dump(param_value_bytes))

        received_params.append({
            'id': param_id,
            'value': param_value_bytes,
            '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 receive_param_ext_ack(vehicle, timeout=5.0):
    """接收参数设置确认"""
    print(f"⏳ Waiting for PARAM_EXT_ACK ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_EXT_ACK', blocking=True, timeout=timeout)

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

    param_id = msg.param_id
    param_value = msg.param_value

    # 结果代码映射
    result_codes = {
        0: "✅ PARAM_ACK_ACCEPTED",
        1: "⚠️ PARAM_ACK_VALUE_UNSUPPORTED",
        2: "❌ PARAM_ACK_FAILED",
        3: "🔄 PARAM_ACK_IN_PROGRESS"
    }
    result = result_codes.get(msg.param_result, f"❓ UNKNOWN({msg.param_result})")

    print("\n📥 Received PARAM_EXT_ACK:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  📌 Result: {result}")

    return {
        'id': param_id,
        'value': param_value,
        'result': result,
        'result_code': msg.param_result
    }

def main():
    print("🚀 Starting MAVLink PARAM_EXT 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. Read parameter value (by ID or index)")
    print("2. Set parameter value (raw/hex/string)")
    print("3. Request all parameters (PARAM_EXT_REQUEST_LIST)")
    print("4. Set struct: uint8_xy {x, y}")
    print("5. Set struct: float32_xy {x, y}")
    print("6. Set struct: float32_lxlyrxry {LX, LY, RX, RY}")
    print("7. Set struct: ering_mode {size, eringMode[16], minValue, maxValue}")
    choice = input("\nSelect operation [1-7]: ").strip()

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

    elif choice == '2':
        param_id = input("Enter parameter ID to set: ").strip()

        print("\nEnter parameter value (supports):")
        print("  - Plain string (e.g., hello)")
        print("  - Hex string (e.g., 48656c6c6f or 0x48656c6c6f)")
        print("  - Raw bytes (e.g., b'\\x48\\x65\\x6c\\x6c\\x6f')")
        param_value_input = input("Enter value: ").strip()

        # 解析输入值
        if param_value_input.startswith("b'") or param_value_input.startswith('b"'):
            # Python字节字面量格式
            param_value = eval(param_value_input)
        elif param_value_input.startswith('0x') or all(c in '0123456789abcdefABCDEF ' for c in param_value_input):
            # 十六进制格式
            hex_str = param_value_input.replace('0x', '').replace(' ', '')
            try:
                param_value = bytes.fromhex(hex_str)
            except ValueError:
                param_value = param_value_input.encode('utf-8')
        else:
            # 默认作为字符串处理
            param_value = param_value_input.encode('utf-8')

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  13 = STRING")
        print("  10 = CHAR (raw bytes)")
        print("  Other types require specific binary formats")
        param_type = int(input("Enter type number [8]: ") or 8)

        send_param_ext_set(vehicle, param_id, param_value, param_type)
        receive_param_ext_ack(vehicle)

    elif choice == '4':
        """设置 uint8_xy 结构体 {uint8_t x; uint8_t y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = int(input("Enter x (0-255): "))
        y = int(input("Enter y (0-255): "))
        param_value = pack_uint8_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '5':
        """设置 float32_xy 结构体 {float32 x; float32 y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = float(input("Enter x: "))
        y = float(input("Enter y: "))
        param_value = pack_float32_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '6':
        """设置 float32_lxlyrxry 结构体 {float32 LX; float32 LY; float32 RX; float32 RY;}"""
        param_id = input("Enter parameter ID: ").strip()
        LX = float(input("Enter LX: "))
        LY = float(input("Enter LY: "))
        RX = float(input("Enter RX: "))
        RY = float(input("Enter RY: "))
        param_value = pack_float32_lxlyrxry(LX, LY, RX, RY)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '7':
        """设置 ering_mode 结构体 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;}"""
        param_id = input("Enter parameter ID: ").strip()
        size = int(input("Enter size (0-16): "))
        eringMode = []
        print("Enter 16 eringMode values (0-255), one per line:")
        for i in range(16):
            val = int(input(f"  eringMode[{i}]: "))
            eringMode.append(val)
        minValue = int(input("Enter minValue: "))
        maxValue = int(input("Enter maxValue: "))
        param_value = pack_ering_mode(size, eringMode, minValue, maxValue)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

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

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


if __name__ == "__main__":
    main()

示例使用说明

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“NGT_SCN_MD_RNG”

成功返回完整消息

注：M3T不支持该功能。

夜景模式

NIGHT_SCENE_MODE

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

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

输入对应参数“NIGHT_SCENE_MODE”

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

设置

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“NIGHT_SCENE_MODE”

输入要设置的数值

选择参数类型“1”

成功返回设置值

注：M3T不支持该功能。

获取拍摄或录像时流媒体数据源的存储范围

STRM_STORE_RNG

mavlink消息：

接收PARAM_EXT_REQUEST_READ
发送PARAM_EXT_VALUE

参数：

union（同一时刻仅生效一个字段）

示例

import time
import sys
import struct
import ctypes
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 pack_uint8_xy(x, y):
    """打包 {uint8_t x; uint8_t y} 结构体 (2字节)"""
    return struct.pack('<BB', x, y)

def pack_float32_xy(x, y):
    """打包 {float32 x; float32 y} 结构体 (8字节)"""
    return struct.pack('<ff', x, y)

def pack_float32_lxlyrxry(LX, LY, RX, RY):
    """打包 {float32 LX; float32 LY; float32 RX; float32 RY;} 结构体 (16字节)"""
    return struct.pack('<ffff', LX, LY, RX, RY)

def pack_ering_mode(size, eringMode, minValue, maxValue):
    """打包 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;} 结构体 (25字节)

    Args:
        size: uint8_t, eringMode数组的有效长度
        eringMode: list of int (0-255), 16个uint8_t值
        minValue: uint32_t, 最小值
        maxValue: uint32_t, 最大值

    Returns:
        bytes: 打包后的二进制数据
    """
    # 确保eringMode是16字节
    eringMode_bytes = bytearray(16)
    for i in range(min(len(eringMode), 16)):
        eringMode_bytes[i] = eringMode[i] & 0xFF

    return struct.pack('<B16sII', size, eringMode_bytes, minValue, maxValue)

def format_param_value(param_value):
    """格式化128字节参数值（接受二进制输入）"""
    if isinstance(param_value, str):
        # 如果是十六进制字符串（如 "48656c6c6f"），先转换为字节
        if param_value.startswith('0x') or all(c in '0123456789abcdefABCDEF' for c in param_value.replace(' ', '')):
            hex_str = param_value.replace('0x', '').replace(' ', '')
            if len(hex_str) % 2 == 0:
                param_value = bytes.fromhex(hex_str)
            else:
                raise ValueError("Invalid hex string: must have even number of characters")
        else:
            # 普通字符串转字节
            param_value = param_value.encode('utf-8')

    if len(param_value) > 128:
        raise ValueError("param_value cannot exceed 128 bytes")

    param_value_bytes = bytearray(128)
    param_value_bytes[:len(param_value)] = param_value

    return param_value_bytes

def hex_dump(data, bytes_per_line=16):
    """格式化二进制数据为 hex + ASCII 展示（类 hexdump 风格）

    Returns:
        str: 格式化后的多行字符串
    """
    lines = []
    for i in range(0, len(data), bytes_per_line):
        chunk = data[i:i + bytes_per_line]
        hex_part = ' '.join(f'{b:02X}' for b in chunk)
        ascii_part = ''.join(chr(b) if 32 <= b < 127 else '.' for b in chunk)
        lines.append(f"  {i:04X}  {hex_part:<{bytes_per_line*3}}  {ascii_part}")
    return '\n'.join(lines)

def send_param_ext_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_ext_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_EXT_REQUEST_READ for {target}")

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

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

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

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

    # 解析参数ID
    param_id = msg.param_id

    # 解析参数值（直接获取128字节原始数据）
    param_value_raw = msg.param_value
    # byte_pointer = (ctypes.c_ubyte * 128).from_buffer_copy(msg.param_value)
    if isinstance(param_value_raw, (bytes, bytearray)):
        param_value_bytes = bytes(param_value_raw)
    # param_value_bytes = bytes(byte_pointer)
    elif isinstance(param_value_raw, str):
        # 替换无法编码的字符（如 \ufffd）为 0x00，还原为128字节
        param_value_bytes = param_value_raw.encode('latin1', errors='replace')
    else:
        param_value_bytes = bytes(param_value_raw)

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

    print("\n📥 Received PARAM_EXT_VALUE:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🧾 Type: {param_type}  |  Data: {len(param_value_bytes)} bytes")
    print(f"  📊 Index/Total: {msg.param_index}/{msg.param_count}")
    print(hex_dump(param_value_bytes))

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


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

    while True:
        msg = vehicle.recv_match(type='PARAM_EXT_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_raw = msg.param_value

        # 直接获取128字节原始数据
        if isinstance(param_value_raw, (bytes, bytearray)):
            param_value_bytes = bytes(param_value_raw)
        elif isinstance(param_value_raw, str):
            # 替换无法编码的字符（如 \ufffd）为 0x00
            param_value_bytes = param_value_raw.encode('latin1', errors='replace')
        else:
            param_value_bytes = bytes(param_value_raw)

        # 参数类型映射
        type_names = {
            0: "INT8", 1: "UINT8", 2: "INT16", 3: "UINT16", 4: "INT32",
            5: "UINT32", 6: "INT64", 7: "UINT64", 8: "FLOAT", 9: "DOUBLE",
            10: "CHAR", 11: "BOOL", 12: "UINT8_ENUM", 13: "STRING"
        }
        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_type}][{len(param_value_bytes)}B]")
        print(hex_dump(param_value_bytes))

        received_params.append({
            'id': param_id,
            'value': param_value_bytes,
            '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 receive_param_ext_ack(vehicle, timeout=5.0):
    """接收参数设置确认"""
    print(f"⏳ Waiting for PARAM_EXT_ACK ({timeout}s timeout)...")
    msg = vehicle.recv_match(type='PARAM_EXT_ACK', blocking=True, timeout=timeout)

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

    param_id = msg.param_id
    param_value = msg.param_value

    # 结果代码映射
    result_codes = {
        0: "✅ PARAM_ACK_ACCEPTED",
        1: "⚠️ PARAM_ACK_VALUE_UNSUPPORTED",
        2: "❌ PARAM_ACK_FAILED",
        3: "🔄 PARAM_ACK_IN_PROGRESS"
    }
    result = result_codes.get(msg.param_result, f"❓ UNKNOWN({msg.param_result})")

    print("\n📥 Received PARAM_EXT_ACK:")
    print(f"  🆔 ID: {param_id}")
    print(f"  🔢 Value: {param_value}")
    print(f"  📌 Result: {result}")

    return {
        'id': param_id,
        'value': param_value,
        'result': result,
        'result_code': msg.param_result
    }

def main():
    print("🚀 Starting MAVLink PARAM_EXT 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. Read parameter value (by ID or index)")
    print("2. Set parameter value (raw/hex/string)")
    print("3. Request all parameters (PARAM_EXT_REQUEST_LIST)")
    print("4. Set struct: uint8_xy {x, y}")
    print("5. Set struct: float32_xy {x, y}")
    print("6. Set struct: float32_lxlyrxry {LX, LY, RX, RY}")
    print("7. Set struct: ering_mode {size, eringMode[16], minValue, maxValue}")
    choice = input("\nSelect operation [1-7]: ").strip()

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

    elif choice == '2':
        param_id = input("Enter parameter ID to set: ").strip()

        print("\nEnter parameter value (supports):")
        print("  - Plain string (e.g., hello)")
        print("  - Hex string (e.g., 48656c6c6f or 0x48656c6c6f)")
        print("  - Raw bytes (e.g., b'\\x48\\x65\\x6c\\x6c\\x6f')")
        param_value_input = input("Enter value: ").strip()

        # 解析输入值
        if param_value_input.startswith("b'") or param_value_input.startswith('b"'):
            # Python字节字面量格式
            param_value = eval(param_value_input)
        elif param_value_input.startswith('0x') or all(c in '0123456789abcdefABCDEF ' for c in param_value_input):
            # 十六进制格式
            hex_str = param_value_input.replace('0x', '').replace(' ', '')
            try:
                param_value = bytes.fromhex(hex_str)
            except ValueError:
                param_value = param_value_input.encode('utf-8')
        else:
            # 默认作为字符串处理
            param_value = param_value_input.encode('utf-8')

        print("\nParameter types:")
        print("  8 = FLOAT (most common)")
        print("  13 = STRING")
        print("  10 = CHAR (raw bytes)")
        print("  Other types require specific binary formats")
        param_type = int(input("Enter type number [8]: ") or 8)

        send_param_ext_set(vehicle, param_id, param_value, param_type)
        receive_param_ext_ack(vehicle)

    elif choice == '4':
        """设置 uint8_xy 结构体 {uint8_t x; uint8_t y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = int(input("Enter x (0-255): "))
        y = int(input("Enter y (0-255): "))
        param_value = pack_uint8_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '5':
        """设置 float32_xy 结构体 {float32 x; float32 y}"""
        param_id = input("Enter parameter ID: ").strip()
        x = float(input("Enter x: "))
        y = float(input("Enter y: "))
        param_value = pack_float32_xy(x, y)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '6':
        """设置 float32_lxlyrxry 结构体 {float32 LX; float32 LY; float32 RX; float32 RY;}"""
        param_id = input("Enter parameter ID: ").strip()
        LX = float(input("Enter LX: "))
        LY = float(input("Enter LY: "))
        RX = float(input("Enter RX: "))
        RY = float(input("Enter RY: "))
        param_value = pack_float32_lxlyrxry(LX, LY, RX, RY)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

    elif choice == '7':
        """设置 ering_mode 结构体 {uint8_t size; uint8_t eringMode[16]; uint32_t minValue; uint32_t maxValue;}"""
        param_id = input("Enter parameter ID: ").strip()
        size = int(input("Enter size (0-16): "))
        eringMode = []
        print("Enter 16 eringMode values (0-255), one per line:")
        for i in range(16):
            val = int(input(f"  eringMode[{i}]: "))
            eringMode.append(val)
        minValue = int(input("Enter minValue: "))
        maxValue = int(input("Enter maxValue: "))
        param_value = pack_ering_mode(size, eringMode, minValue, maxValue)
        print(f"  Packed: {param_value.hex()} ({len(param_value)} bytes)")
        send_param_ext_set(vehicle, param_id, param_value, 10)  # CHAR type
        receive_param_ext_ack(vehicle)

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

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


if __name__ == "__main__":
    main()

运行后，成功建立连接

输入“1”选择设置参数

输入对应参数“模板”

成功返回完整消息

同步分屏缩放功能

SYNC_SPLT_ZM

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

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

输入对应参数“SYNC_SPLT_ZM”

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

相机测光模式

METER_MODE

设置

mavlink消息：

接收PARAM_SET
发送PARAM_VALUE

获取

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()