自定义HMS功能

向应用程序上报自定义 HMS 错误码

TUNNEL_PROTOCOL_ID_INJECT_HMS_ERROR

mavlink消息：

接收TUNNEL_PROTOCOL_ID_INJECT_HMS_ERROR

参数：

示例

#!/usr/bin/env python3
"""
通过 MAVLink TUNNEL 消息发送 DJI HMS 结构体数据
HmsErrCode 和 HmsAlarmEnhancedCtrl

Usage:
    python tunnel_hms_send.py
"""

import struct
import time
from pymavlink import mavutil

# MAVLink 组件 ID (DJI 指定)
MAV_COMP_ID_HMS = 12

# HMS 目标组件
TARGET_COMPONENT = MAV_COMP_ID_HMS

def create_hmserrcode_struct(error_code: int, error_level: int, enable: bool, error_id: int) -> bytes:
    """
    创建 HmsErrCode 结构体 (7字节, packed)

    typedef struct Hms_ErrorCode_Ctrl{
        uint32_t errorCode;       // 4 bytes
        uint32_t errorLevel;       // 4 byte (E_DjiHmsErrorLevel)
        bool enable;              // 1 byte
        uint8_t id;               // 1 byte
    } __attribute__((packed)) HmsErrCode;
    """
    return struct.pack('<IIbB', error_code, error_level, 1 if enable else 0, error_id)


def create_hmsalarminfo_struct(
    alarm_type: int, 
    times: int, 
    interval: int, 
    action: int, 
    alarm_id: int
) -> bytes:
    """
    创建 HmsAlarmEnhancedCtrl 结构体 (6字节, packed)

    typedef struct Hms_Alarm_Enhanced_Ctrl{
        uint32_t type;             // 4 byte (E_DjiHmsAlarmEnhancedType)
        int8_t times;              // 1 byte
        uint8_t reserved;          // 1 byte
        int16_t interval;          // 2 bytes
        uint32_t action;           // 4 byte (E_DjiHmsAlarmEnhancedAction)
        uint8_t id;               // 1 byte
    } __attribute__((packed)) HmsAlarmEnhancedCtrl;
    """
    return struct.pack('<IbbhIB', alarm_type, times, 0, interval, action, alarm_id)


def send_tunnel_message(mavlink_conn, payload: bytes, message_type: int = 0):
    """
    通过 MAVLink TUNNEL 消息发送数据

    Args:
        mavlink_conn: MAVLink 连接对象
        payload: 要发送的数据负载 (最多 228 字节)
        message_type: 消息类型标识 (可选)
    """
    if len(payload) > 228:
        raise ValueError(f"Payload too large: {len(payload)} bytes (max 228)")

    # 填充到固定长度 (确保小端序)
    payload_bytes = payload.ljust(228, b'\x00')

    # 发送 TUNNEL 消息
    # MAVLINK_MSG_ID_TUNNEL = 110
    mavlink_conn.mav.tunnel_send(
        target_system=0,           # 广播到所有系统
        target_component=TARGET_COMPONENT,
        payload_length=len(payload),
        payload_type=message_type,
        payload=payload_bytes
    )


def send_hms_error_code(mavlink_conn,message_type: int = 0):
    """
    发送 HmsErrCode 结构体
    示例数据:
        errorCode: 0x1E020001 (设备错误)
        errorLevel: DJI_HMS_ERROR_LEVEL_WARN (2)
        enable: True
        id: 1
    """
    payload = create_hmserrcode_struct(
        error_code=0x1E020001,    # 示例错误码
        error_level=2,             # WARN 级别
        enable=True,
        error_id=1
    )

    # message_type 用于标识 HMS 错误码消息类型
    if message_type==32771:
        payload = create_hmserrcode_struct(
            error_code=0x1E020001,    # 示例错误码
            error_level=2,             # WARN 级别
            enable=False,
            error_id=1
        )
    send_tunnel_message(mavlink_conn, payload, message_type)
    print(f"[HMS] 发送 HmsErrCode: errorCode=0x1E020001, level=2, enable=True, id=1")


def send_hms_alarm_enhanced(mavlink_conn):
    """
    发送 HmsAlarmEnhancedCtrl 结构体
    示例数据:
        type: SHAKE_MOTOR (1)
        times: 3
        interval: 500ms
        action: START (1)
        id: 1
    """
    payload = create_hmsalarminfo_struct(
        alarm_type=3,              # SHAKE_MOTOR
        times=3,                   # 振动3次
        interval=500,              # 间隔500ms
        action=1,                   # START
        alarm_id=1
    )

    # message_type 用于标识 HMS 告警控制消息类型
    HMS_ALARM_ENHANCED_TYPE = 32772
    send_tunnel_message(mavlink_conn, payload, HMS_ALARM_ENHANCED_TYPE)
    print(f"[HMS] 发送 HmsAlarmEnhancedCtrl: type=1, times=3, interval=500, action=1, id=1")


def main():
    print("=" * 60)
    print("MAVLink HMS TUNNEL 消息发送工具")
    print("=" * 60)

    # 创建 MAVLink UDP 连接 (监听 14550 端口)
    # udpin:0.0.0.0:14550 表示监听所有网卡地址的 14550 端口
    connection_string = "udpin:0.0.0.0:14550"
    print(f"\n正在连接到: {connection_string}")

    try:
        mavlink_conn = mavutil.mavlink_connection(connection_string)

        # 等待心跳包以确认连接
        print("等待心跳包...")
        heartbeat = mavlink_conn.wait_heartbeat(timeout=10)
        if heartbeat:
            print(f"已连接! ")
        else:
            print("警告: 未收到心跳包，继续发送...")

        print("\n" + "-" * 60)
        print("开始发送 HMS 消息...")
        print("-" * 60)

        # 发送 HmsErrCode 结构体
        send_hms_error_code(mavlink_conn,32770)
        time.sleep(3)  # 间隔1秒

        # 发送 HmsAlarmEnhancedCtrl 结构体
        send_hms_alarm_enhanced(mavlink_conn)

        print("\n" + "-" * 60)
        print("消息发送完成!")
        print("-" * 60)

        # 保持连接一段时间以便接收响应
        print("\n等待响应 (5秒)...")
        time.sleep(5)

        mavlink_conn.close()
        print("连接已关闭")

    except Exception as e:
        print(f"错误: {e}")
        raise


if __name__ == "__main__":
    main()

示例使用说明

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

清除应用程序上显示的自定义 HMS 错误码

TUNNEL_PROTOCOL_ID_ELIMINATE_HMS_ERROR

设置

mavlink消息：

接收TUNNEL_PROTOCOL_ID_ELIMINATE_HMS_ERROR

参数：

示例

#!/usr/bin/env python3
"""
通过 MAVLink TUNNEL 消息发送 DJI HMS 结构体数据
HmsErrCode 和 HmsAlarmEnhancedCtrl

Usage:
    python tunnel_hms_send.py
"""

import struct
import time
from pymavlink import mavutil

# MAVLink 组件 ID (DJI 指定)
MAV_COMP_ID_HMS = 12

# HMS 目标组件
TARGET_COMPONENT = MAV_COMP_ID_HMS

def create_hmserrcode_struct(error_code: int, error_level: int, enable: bool, error_id: int) -> bytes:
    """
    创建 HmsErrCode 结构体 (7字节, packed)

    typedef struct Hms_ErrorCode_Ctrl{
        uint32_t errorCode;       // 4 bytes
        uint32_t errorLevel;       // 4 byte (E_DjiHmsErrorLevel)
        bool enable;              // 1 byte
        uint8_t id;               // 1 byte
    } __attribute__((packed)) HmsErrCode;
    """
    return struct.pack('<IIbB', error_code, error_level, 1 if enable else 0, error_id)


def create_hmsalarminfo_struct(
    alarm_type: int, 
    times: int, 
    interval: int, 
    action: int, 
    alarm_id: int
) -> bytes:
    """
    创建 HmsAlarmEnhancedCtrl 结构体 (6字节, packed)

    typedef struct Hms_Alarm_Enhanced_Ctrl{
        uint32_t type;             // 4 byte (E_DjiHmsAlarmEnhancedType)
        int8_t times;              // 1 byte
        uint8_t reserved;          // 1 byte
        int16_t interval;          // 2 bytes
        uint32_t action;           // 4 byte (E_DjiHmsAlarmEnhancedAction)
        uint8_t id;               // 1 byte
    } __attribute__((packed)) HmsAlarmEnhancedCtrl;
    """
    return struct.pack('<IbbhIB', alarm_type, times, 0, interval, action, alarm_id)


def send_tunnel_message(mavlink_conn, payload: bytes, message_type: int = 0):
    """
    通过 MAVLink TUNNEL 消息发送数据

    Args:
        mavlink_conn: MAVLink 连接对象
        payload: 要发送的数据负载 (最多 228 字节)
        message_type: 消息类型标识 (可选)
    """
    if len(payload) > 228:
        raise ValueError(f"Payload too large: {len(payload)} bytes (max 228)")

    # 填充到固定长度 (确保小端序)
    payload_bytes = payload.ljust(228, b'\x00')

    # 发送 TUNNEL 消息
    # MAVLINK_MSG_ID_TUNNEL = 110
    mavlink_conn.mav.tunnel_send(
        target_system=0,           # 广播到所有系统
        target_component=TARGET_COMPONENT,
        payload_length=len(payload),
        payload_type=message_type,
        payload=payload_bytes
    )


def send_hms_error_code(mavlink_conn,message_type: int = 0):
    """
    发送 HmsErrCode 结构体
    示例数据:
        errorCode: 0x1E020001 (设备错误)
        errorLevel: DJI_HMS_ERROR_LEVEL_WARN (2)
        enable: True
        id: 1
    """
    payload = create_hmserrcode_struct(
        error_code=0x1E020001,    # 示例错误码
        error_level=2,             # WARN 级别
        enable=True,
        error_id=1
    )

    # message_type 用于标识 HMS 错误码消息类型
    if message_type==32771:
        payload = create_hmserrcode_struct(
            error_code=0x1E020001,    # 示例错误码
            error_level=2,             # WARN 级别
            enable=False,
            error_id=1
        )
    send_tunnel_message(mavlink_conn, payload, message_type)
    print(f"[HMS] 发送 HmsErrCode: errorCode=0x1E020001, level=2, enable=True, id=1")


def send_hms_alarm_enhanced(mavlink_conn):
    """
    发送 HmsAlarmEnhancedCtrl 结构体
    示例数据:
        type: SHAKE_MOTOR (1)
        times: 3
        interval: 500ms
        action: START (1)
        id: 1
    """
    payload = create_hmsalarminfo_struct(
        alarm_type=3,              # SHAKE_MOTOR
        times=3,                   # 振动3次
        interval=500,              # 间隔500ms
        action=1,                   # START
        alarm_id=1
    )

    # message_type 用于标识 HMS 告警控制消息类型
    HMS_ALARM_ENHANCED_TYPE = 32772
    send_tunnel_message(mavlink_conn, payload, HMS_ALARM_ENHANCED_TYPE)
    print(f"[HMS] 发送 HmsAlarmEnhancedCtrl: type=1, times=3, interval=500, action=1, id=1")


def main():
    print("=" * 60)
    print("MAVLink HMS TUNNEL 消息发送工具")
    print("=" * 60)

    # 创建 MAVLink UDP 连接 (监听 14550 端口)
    # udpin:0.0.0.0:14550 表示监听所有网卡地址的 14550 端口
    connection_string = "udpin:0.0.0.0:14550"
    print(f"\n正在连接到: {connection_string}")

    try:
        mavlink_conn = mavutil.mavlink_connection(connection_string)

        # 等待心跳包以确认连接
        print("等待心跳包...")
        heartbeat = mavlink_conn.wait_heartbeat(timeout=10)
        if heartbeat:
            print(f"已连接! ")
        else:
            print("警告: 未收到心跳包，继续发送...")

        print("\n" + "-" * 60)
        print("开始发送 HMS 消息...")
        print("-" * 60)

        # 发送 HmsErrCode 结构体
        send_hms_error_code(mavlink_conn,32771)
        time.sleep(3)  # 间隔1秒

        # 发送 HmsAlarmEnhancedCtrl 结构体
        send_hms_alarm_enhanced(mavlink_conn)

        print("\n" + "-" * 60)
        print("消息发送完成!")
        print("-" * 60)

        # 保持连接一段时间以便接收响应
        print("\n等待响应 (5秒)...")
        time.sleep(5)

        mavlink_conn.close()
        print("连接已关闭")

    except Exception as e:
        print(f"错误: {e}")
        raise


if __name__ == "__main__":
    main()

示例使用说明

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

清除应用程序上显示的自定义 HMS 错误码

TUNNEL_PROTOCOL_ID_ELIMINATE_HMS_ERROR

设置

mavlink消息：

接收TUNNEL_PROTOCOL_ID_ELIMINATE_HMS_ERROR

参数：

示例

#!/usr/bin/env python3
"""
通过 MAVLink TUNNEL 消息发送 DJI HMS 结构体数据
HmsErrCode 和 HmsAlarmEnhancedCtrl

Usage:
    python tunnel_hms_send.py
"""

import struct
import time
from pymavlink import mavutil

# MAVLink 组件 ID (DJI 指定)
MAV_COMP_ID_HMS = 12

# HMS 目标组件
TARGET_COMPONENT = MAV_COMP_ID_HMS

def create_hmserrcode_struct(error_code: int, error_level: int, enable: bool, error_id: int) -> bytes:
    """
    创建 HmsErrCode 结构体 (7字节, packed)

    typedef struct Hms_ErrorCode_Ctrl{
        uint32_t errorCode;       // 4 bytes
        uint32_t errorLevel;       // 4 byte (E_DjiHmsErrorLevel)
        bool enable;              // 1 byte
        uint8_t id;               // 1 byte
    } __attribute__((packed)) HmsErrCode;
    """
    return struct.pack('<IIbB', error_code, error_level, 1 if enable else 0, error_id)


def create_hmsalarminfo_struct(
    alarm_type: int, 
    times: int, 
    interval: int, 
    action: int, 
    alarm_id: int
) -> bytes:
    """
    创建 HmsAlarmEnhancedCtrl 结构体 (6字节, packed)

    typedef struct Hms_Alarm_Enhanced_Ctrl{
        uint32_t type;             // 4 byte (E_DjiHmsAlarmEnhancedType)
        int8_t times;              // 1 byte
        uint8_t reserved;          // 1 byte
        int16_t interval;          // 2 bytes
        uint32_t action;           // 4 byte (E_DjiHmsAlarmEnhancedAction)
        uint8_t id;               // 1 byte
    } __attribute__((packed)) HmsAlarmEnhancedCtrl;
    """
    return struct.pack('<IbbhIB', alarm_type, times, 0, interval, action, alarm_id)


def send_tunnel_message(mavlink_conn, payload: bytes, message_type: int = 0):
    """
    通过 MAVLink TUNNEL 消息发送数据

    Args:
        mavlink_conn: MAVLink 连接对象
        payload: 要发送的数据负载 (最多 228 字节)
        message_type: 消息类型标识 (可选)
    """
    if len(payload) > 228:
        raise ValueError(f"Payload too large: {len(payload)} bytes (max 228)")

    # 填充到固定长度 (确保小端序)
    payload_bytes = payload.ljust(228, b'\x00')

    # 发送 TUNNEL 消息
    # MAVLINK_MSG_ID_TUNNEL = 110
    mavlink_conn.mav.tunnel_send(
        target_system=0,           # 广播到所有系统
        target_component=TARGET_COMPONENT,
        payload_length=len(payload),
        payload_type=message_type,
        payload=payload_bytes
    )


def send_hms_error_code(mavlink_conn,message_type: int = 0):
    """
    发送 HmsErrCode 结构体
    示例数据:
        errorCode: 0x1E020001 (设备错误)
        errorLevel: DJI_HMS_ERROR_LEVEL_WARN (2)
        enable: True
        id: 1
    """
    payload = create_hmserrcode_struct(
        error_code=0x1E020001,    # 示例错误码
        error_level=2,             # WARN 级别
        enable=True,
        error_id=1
    )

    # message_type 用于标识 HMS 错误码消息类型
    if message_type==32771:
        payload = create_hmserrcode_struct(
            error_code=0x1E020001,    # 示例错误码
            error_level=2,             # WARN 级别
            enable=False,
            error_id=1
        )
    send_tunnel_message(mavlink_conn, payload, message_type)
    print(f"[HMS] 发送 HmsErrCode: errorCode=0x1E020001, level=2, enable=True, id=1")


def send_hms_alarm_enhanced(mavlink_conn):
    """
    发送 HmsAlarmEnhancedCtrl 结构体
    示例数据:
        type: SHAKE_MOTOR (1)
        times: 3
        interval: 500ms
        action: START (1)
        id: 1
    """
    payload = create_hmsalarminfo_struct(
        alarm_type=3,              # SHAKE_MOTOR
        times=3,                   # 振动3次
        interval=500,              # 间隔500ms
        action=1,                   # START
        alarm_id=1
    )

    # message_type 用于标识 HMS 告警控制消息类型
    HMS_ALARM_ENHANCED_TYPE = 32772
    send_tunnel_message(mavlink_conn, payload, HMS_ALARM_ENHANCED_TYPE)
    print(f"[HMS] 发送 HmsAlarmEnhancedCtrl: type=1, times=3, interval=500, action=1, id=1")


def main():
    print("=" * 60)
    print("MAVLink HMS TUNNEL 消息发送工具")
    print("=" * 60)

    # 创建 MAVLink UDP 连接 (监听 14550 端口)
    # udpin:0.0.0.0:14550 表示监听所有网卡地址的 14550 端口
    connection_string = "udpin:0.0.0.0:14550"
    print(f"\n正在连接到: {connection_string}")

    try:
        mavlink_conn = mavutil.mavlink_connection(connection_string)

        # 等待心跳包以确认连接
        print("等待心跳包...")
        heartbeat = mavlink_conn.wait_heartbeat(timeout=10)
        if heartbeat:
            print(f"已连接! ")
        else:
            print("警告: 未收到心跳包，继续发送...")

        print("\n" + "-" * 60)
        print("开始发送 HMS 消息...")
        print("-" * 60)

        # 发送 HmsErrCode 结构体
        send_hms_error_code(mavlink_conn,32772)
        time.sleep(3)  # 间隔1秒

        # send_hms_error_code(mavlink_conn,32771)
        # time.sleep(3)  # 间隔1秒

        # 发送 HmsAlarmEnhancedCtrl 结构体
        send_hms_alarm_enhanced(mavlink_conn)

        print("\n" + "-" * 60)
        print("消息发送完成!")
        print("-" * 60)

        # 保持连接一段时间以便接收响应
        print("\n等待响应 (5秒)...")
        time.sleep(5)

        mavlink_conn.close()
        print("连接已关闭")

    except Exception as e:
        print(f"错误: {e}")
        raise


if __name__ == "__main__":
    main()

示例使用说明

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