码流AI识别

功能说明

LiveView AI 功能用于开启 DJI PSDK LiveView 图像流，使用示例中的 YOLOvFastest 目标检测模块处理图像，并通过 MAVLink TUNNEL 消息输出识别框信息。

该功能同时会将 AI meta 信息回传给 DJI Pilot，并将检测结果转换为 MAVLink 自定义二进制 payload，方便地面端或伴随计算机解析。

注意：

当前仅在 Matrice 400 无人机挂载 H30 相机和 DJI 妙算3 场景下支持该功能。

编解码回调函数中均不允许进行阻塞耗时操作，推荐创建独立线程进行目标识别等功能。

相关文件

启停命令

使用 MAVLink COMMAND_LONG 发送 MAV_CMD_USER_1。

命令执行后，程序返回 COMMAND_ACK。

MAVLink 输出

LiveView AI 识别结果通过 MAVLink TUNNEL 输出。

Payload 格式

每个 TUNNEL payload 由一个 header 加若干个 box 组成。单包最多携带 8 个目标框；超过 8 个目标时自动拆成多包。

Header

Box

使用流程

1. 确认工程编译时开启 OPEN_CV_INSTALLED。
2. 地面端发送 MAV_CMD_USER_1，param1=1。
3. 程序开启 LiveView 图像流并启动 AI 处理线程。
4. 地面端监听 MAVLink TUNNEL payload_type=32790。
5. 停止时发送 MAV_CMD_USER_1，param1=0。

示例

启动默认相机源：

COMMAND_LONG.command = MAV_CMD_USER_1
COMMAND_LONG.param1 = 1
COMMAND_LONG.param2 = 0
COMMAND_LONG.param3 = 0

停止：

COMMAND_LONG.command = MAV_CMD_USER_1
COMMAND_LONG.param1 = 0

Python 例程

以下例程使用 pymavlink 启动 LiveView AI，监听 TUNNEL payload_type=32790，解析识别框并打印目标信息，最后发送停止命令。

import struct
import time
from pymavlink import mavutil


LIVEVIEW_AI_PAYLOAD_TYPE = 32790
MAV_CMD_USER_1 = mavutil.mavlink.MAV_CMD_USER_1


def get_current_millis():
    return int(time.time() * 1000)


def send_liveview_ai_command(master, enable, camera_position=0, camera_source=0):
    send_time = get_current_millis()
    print(
        f"[{send_time}] 发送 LiveView AI 命令: "
        f"enable={enable}, position={camera_position}, source={camera_source}"
    )

    master.mav.command_long_send(
        master.target_system,
        master.target_component,
        MAV_CMD_USER_1,
        0,
        1 if enable else 0,   # param1: 1=start, 0=stop
        camera_position,      # param2: DJI LiveView camera position，0 使用默认值
        camera_source,        # param3: DJI LiveView camera source，0 使用默认值
        0, 0, 0, 0
    )

    ack = master.recv_match(type="COMMAND_ACK", blocking=True, timeout=5.0)
    if ack:
        ack_time = get_current_millis()
        print(
            f"[{ack_time}] 收到 COMMAND_ACK: "
            f"command={ack.command}, result={ack.result}, delay={ack_time - send_time}ms"
        )
        return ack.result == mavutil.mavlink.MAV_RESULT_ACCEPTED

    print(f"[{get_current_millis()}] 超时：未收到 LiveView AI 命令应答")
    return False


def parse_liveview_ai_payload(payload):
    # Header: version, packet_index, packet_count, box_count, frame_id, width, height
    header_fmt = "<BBBBIHH"
    box_fmt = "<HBBHHHHI"
    header_size = struct.calcsize(header_fmt)
    box_size = struct.calcsize(box_fmt)

    if len(payload) < header_size:
        return None, []

    version, packet_index, packet_count, box_count, frame_id, width, height = struct.unpack_from(
        header_fmt, payload, 0
    )

    header = {
        "version": version,
        "packet_index": packet_index,
        "packet_count": packet_count,
        "box_count": box_count,
        "frame_id": frame_id,
        "width": width,
        "height": height,
    }

    boxes = []
    offset = header_size
    for _ in range(box_count):
        if offset + box_size > len(payload):
            break
        target_id, target_type, state, cx, cy, w, h, distance_mm = struct.unpack_from(
            box_fmt, payload, offset
        )
        boxes.append({
            "id": target_id,
            "type": target_type,
            "state": state,
            "cx": cx,
            "cy": cy,
            "w": w,
            "h": h,
            "distance_mm": distance_mm,
        })
        offset += box_size

    return header, boxes


# 连接无人机（UDP 端口 14550）
master = mavutil.mavlink_connection("udpin:0.0.0.0:14550")
print("等待无人机心跳...")
master.wait_heartbeat()
print(f"无人机已连接: system={master.target_system}, component={master.target_component}")

if not send_liveview_ai_command(master, enable=True):
    raise SystemExit("LiveView AI 启动失败")

print("开始监听 LiveView AI 识别结果，按 Ctrl+C 停止")

try:
    while True:
        msg = master.recv_match(type="TUNNEL", blocking=True, timeout=5.0)
        now_ms = get_current_millis()

        if msg is None:
            print(f"[{now_ms}] 超时：未收到 LiveView AI TUNNEL 数据")
            continue

        if msg.payload_type != LIVEVIEW_AI_PAYLOAD_TYPE:
            continue

        payload = bytes(msg.payload[:msg.payload_length])
        header, boxes = parse_liveview_ai_payload(payload)
        if header is None:
            print(f"[{now_ms}] 收到无效 LiveView AI payload")
            continue

        print(
            f"[{now_ms}] frame={header['frame_id']} "
            f"packet={header['packet_index'] + 1}/{header['packet_count']} "
            f"size={header['width']}x{header['height']} boxes={len(boxes)}"
        )

        for box in boxes:
            print(
                f"  id={box['id']} type={box['type']} state={box['state']} "
                f"cx={box['cx']} cy={box['cy']} w={box['w']} h={box['h']} "
                f"distance={box['distance_mm']}mm"
            )

except KeyboardInterrupt:
    print("准备停止 LiveView AI")
finally:
    send_liveview_ai_command(master, enable=False)
    print("LiveView AI 例程结束")

注意事项

• 如果未定义 OPEN_CV_INSTALLED，启动接口会返回 DJI_ERROR_SYSTEM_MODULE_CODE_NONSUPPORT，MAVLink ACK 为 MAV_RESULT_FAILED。
• cx/cy/w/h 是 DJI 归一化坐标，不是像素坐标。地面端如需像素坐标，需要结合 width/height 自行换算。
• LiveView AI 使用 LiveView 图像流和编码回调。若系统同时使用普通视频流，需要联调确认 LiveView 初始化和回调不会互相抢占。
• 自定义 payload 为二进制结构，不是 JSON；地面端解析时需按小端结构体读取。