回放下载

接口总览

示例

#!/usr/bin/env python3
"""
通过 MAVLink TUNNEL 消息获取相机媒体文件列表（分块下载）

协议说明:
  1. 发送 REQUEST_DOWNLOAD_FILE_LIST (32784) → 请求文件列表
     负载: uint16 sliceStartIndex + uint16 countPerSlice (4 字节, 小端序)
  2. 接收 DOWNLOAD_FILE_LIST (32785) → 文件列表中每个文件以分块方式返回
     分块协议: id(1) + chunk_seq(1) + data(126)
     相同 id 的块属于同一个文件, 按 chunk_seq 拼接后即为 T_DjiCameraManagerFileListInfo
  3. 发送 REQUEST_DOWNLOAD_FILE_TO_LOCAL (32786) → 请求下载文件到设备本地
     负载: uint32 fileIndex + uint8 fileType (5 字节, 小端序)
     注意: 文件内容存储在设备本地磁盘, 不通过 TUNNEL 回传

T_DjiCameraManagerFileListInfo 内存布局 (64-bit, 小端序, 无 pack):
  偏移      大小    字段
  0         256     fileName[256]
  256       4       fileSize (uint32)
  260       4       fileIndex (uint32)
  264       1       createTime.second
  265       1       createTime.minute
  266       1       createTime.hour
  267       1       createTime.day
  268       1       createTime.month
  269       1       (padding)
  270       2       createTime.year (uint16)
  272       4       type (E_DjiCameraMediaFileType, uint32)
  276       16      attributeData
  292       1       subFileListTotalNum
  293       3       (padding)
  296       8       subFileListInfo (指针, 不可用)
  总计: 304 字节

Usage:
    python tunnel_file_download.py [--list] [--download INDEX] [--file-type TYPE]
"""

import argparse
import struct
import time
import os
from collections import defaultdict
from pymavlink import mavutil

# ============================================================================
# MAVLink 组件 ID
# ============================================================================
MAV_COMP_ID_DEFAULT = 12  # DJI 指定组件 ID

# ============================================================================
# TUNNEL 协议 ID (与 mavlink_tool.h 中 TUNNEL_PROTOCOL_ID 枚举一致)
# ============================================================================
TUNNEL_PROTOCOL_ID_REQUEST_DOWNLOAD_FILE_LIST   = 32784  # 请求文件列表
TUNNEL_PROTOCOL_ID_DOWNLOAD_FILE_LIST           = 32785  # 文件列表响应 (分块)
TUNNEL_PROTOCOL_ID_REQUEST_DOWNLOAD_FILE_TO_LOCAL = 32786  # 请求下载文件到本地

# ============================================================================
# 分片参数 (与 E_DjiCameraManagerFileListCountPerSlice 枚举一致)
# ============================================================================
FILE_LIST_COUNT_60_PER_SLICE  = 60
FILE_LIST_COUNT_120_PER_SLICE = 120
FILE_LIST_COUNT_ALL_PER_SLICE = 0xFFFF

# ============================================================================
# T_DjiCameraManagerFileListInfo 结构体常量
# ============================================================================
DJI_FILE_NAME_SIZE_MAX = 256
# 结构体在 64-bit ARM/Linux 上的 sizeof (通过内存布局计算)
FILE_LIST_INFO_SIZE = 304  # bytes

# ============================================================================
# 媒体文件类型 (E_DjiCameraMediaFileType)
# ============================================================================
MEDIA_FILE_TYPE_NAMES = {
    0:  "JPEG",
    1:  "DNG",
    2:  "MOV",
    3:  "MP4",
    4:  "PANORAMA_JPEG",
    5:  "TIFF_SEQ",
    6:  "MP4_IR",
    7:  "JPEG_IR",
    8:  "M4A",       # M4D 可见光 JPEG
    9:  "IR",        # M4D 红外
    10: "WIDE",      # M4D 广角
    14: "PHOTO_DNG",
    15: "PHOTO_TIF",
    16: "PHOTO_TIF_SEQ",
    17: "PHOTO_SSD_RAW",
    18: "PHOTO_SSD_RAW_MULTI",
    19: "VIDEO_ORG",
    20: "PHOTO_ORG",
    21: "AUDIO_WAV",
    22: "VIDEO_DNG",
    23: "VIDEO_SSD_RAW",
    24: "PHOTO_TIFF_8K",
    25: "PHOTO_MP4",
    26: "PHOTO_JPEG_8K",
    27: "PHOTO_DNG_8K",
    32: "ORG_FILE",  # >= 32 表示原始文件
}


def parse_file_list_info(data: bytes) -> dict:
    """
    从原始字节解析 T_DjiCameraManagerFileListInfo 结构体

    Args:
        data: 文件列表信息的原始字节 (至少 304 字节)

    Returns:
        dict: 解析后的文件信息
    """
    if len(data) < FILE_LIST_INFO_SIZE:
        raise ValueError(f"数据不足: 需要 {FILE_LIST_INFO_SIZE} 字节, 实际 {len(data)} 字节")

    offset = 0
    # fileName: 256 bytes, 以 null 结尾
    raw_name = data[offset:offset + DJI_FILE_NAME_SIZE_MAX]
    offset += DJI_FILE_NAME_SIZE_MAX
    null_pos = raw_name.find(b'\x00')
    file_name = raw_name[:null_pos].decode('utf-8', errors='replace') if null_pos >= 0 else raw_name.decode('utf-8', errors='replace')

    # fileSize: uint32_t LE
    file_size = struct.unpack_from('<I', data, offset)[0]
    offset += 4

    # fileIndex: uint32_t LE
    file_index = struct.unpack_from('<I', data, offset)[0]
    offset += 4

    # createTime:
    #   second(1) + minute(1) + hour(1) + day(1) + month(1) + pad(1) + year(2) = 8 bytes
    second, minute, hour, day, month = struct.unpack_from('<BBBBB', data, offset)
    offset += 5
    # padding 1 byte (for uint16 alignment)
    offset += 1
    year = struct.unpack_from('<H', data, offset)[0]
    offset += 2

    # type: uint32_t LE (E_DjiCameraMediaFileType)
    file_type = struct.unpack_from('<I', data, offset)[0]
    offset += 4

    # attributeData: 16 bytes (union, raw)
    attribute_data = data[offset:offset + 16]
    offset += 16

    # subFileListTotalNum: uint8_t
    sub_file_count = data[offset]
    offset += 1

    # padding 3 bytes + subFileListInfo pointer 8 bytes (跳过)
    # offset += 3 + 8

    type_name = MEDIA_FILE_TYPE_NAMES.get(file_type, f"UNKNOWN({file_type})")

    return {
        'file_name': file_name,
        'file_size': file_size,
        'file_index': file_index,
        'create_time': f"{year:04d}-{month:02d}-{day:02d} {hour:02d}:{minute:02d}:{second:02d}",
        'year': year, 'month': month, 'day': day, 'hour': hour, 'minute': minute, 'second': second,
        'file_type': file_type,
        'type_name': type_name,
        'sub_file_count': sub_file_count,
        'raw_bytes': data[:FILE_LIST_INFO_SIZE],
    }


def send_file_list_request(mavlink_conn, slice_start_index=0, count_per_slice=FILE_LIST_COUNT_ALL_PER_SLICE):
    """
    发送请求文件列表命令

    Args:
        mavlink_conn: MAVLink 连接对象
        slice_start_index: 分片起始索引 (0 表示从头开始)
        count_per_slice: 每片文件数量 (60/120/0xFFFF)
    """
    # 构建负载: uint16 sliceStartIndex + uint16 countPerSlice (小端序)
    payload = struct.pack('<HH', slice_start_index, count_per_slice)

    print(f"\n[REQUEST] 请求文件列表: sliceStartIndex={slice_start_index}, countPerSlice={count_per_slice}")

    send_tunnel_message(mavlink_conn, payload, TUNNEL_PROTOCOL_ID_REQUEST_DOWNLOAD_FILE_LIST)


def send_download_request(mavlink_conn, file_index, file_type=32):
    """
    发送下载文件到本地命令

    注意: 此命令仅触发设备端的本地下载, 文件内容不会通过 TUNNEL 回传。

    Args:
        mavlink_conn: MAVLink 连接对象
        file_index: 文件索引
        file_type: 文件类型 (>=32 表示原始文件, <32 表示子文件)
    """
    # 构建负载: uint32 fileIndex + uint8 fileType (小端序)
    payload = struct.pack('<IB', file_index, file_type)
    # 确保是 5 字节
    payload = payload.ljust(5, b'\x00')[:5]

    print(f"\n[REQUEST] 请求下载文件: fileIndex={file_index}, fileType={file_type}")

    send_tunnel_message(mavlink_conn, payload, TUNNEL_PROTOCOL_ID_REQUEST_DOWNLOAD_FILE_TO_LOCAL)


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

    Args:
        mavlink_conn: MAVLink 连接对象
        payload: 要发送的数据负载
        message_type: TUNNEL payload_type
    """
    if len(payload) > 128:
        raise ValueError(f"Payload 过大: {len(payload)} 字节 (最大 128)")

    # 填充到 128 字节
    payload_bytes = payload.ljust(128, b'\x00')

    mavlink_conn.mav.tunnel_send(
        target_system=0,
        target_component=MAV_COMP_ID_DEFAULT,
        payload_length=len(payload),
        payload_type=message_type,
        payload=payload_bytes
    )

    hex_str = ' '.join(f'{b:02X}' for b in payload[:len(payload)])
    print(f"  [TUNNEL SEND] type={message_type}, len={len(payload)}, data=[{hex_str}]")


def receive_file_list(mavlink_conn, timeout: float = 60.0):
    """
    接收并解析文件列表

    文件列表中的每个文件通过分块协议发送:
      - 每个块包含: id(1) + chunk_seq(1) + data(126)
      - 相同 id 的块属于同一个文件
      - 按 chunk_seq 拼接后得到 T_DjiCameraManagerFileListInfo 的原始字节

    Args:
        mavlink_conn: MAVLink 连接对象
        timeout: 接收超时时间(秒)

    Returns:
        list: 解析后的文件信息列表
    """
    print("\n" + "=" * 70)
    print("等待接收文件列表...")
    print("=" * 70)

    # 按消息 id 收集分块
    # { msg_id: { chunk_seq: bytes } }
    chunk_buffers = defaultdict(dict)

    start_time = time.time()
    received_count = 0
    file_count_estimate = None

    try:
        while True:
            elapsed = time.time() - start_time
            if elapsed >= timeout:
                print(f"\n接收超时 ({timeout}秒)")
                break

            msg = mavlink_conn.recv_match(type='TUNNEL', blocking=True, timeout=1.0)

            if msg is None:
                continue

            payload_type = msg.payload_type

            # 只处理文件列表响应 (32785)
            if payload_type != TUNNEL_PROTOCOL_ID_DOWNLOAD_FILE_LIST:
                # 打印其他 TUNNEL 消息以帮助调试
                if payload_type != 0:
                    payload_bytes = bytes(msg.payload)
                    payload_len = msg.payload_length
                    actual_data = payload_bytes[:payload_len]
                    hex_str = ' '.join(f'{b:02X}' for b in actual_data)
                    ascii_str = ''.join(chr(b) if 32 <= b < 127 else '.' for b in actual_data)
                    print(f"  [TUNNEL] 收到其他类型消息: type={payload_type}, len={payload_len}")
                    print(f"    Hex: [{hex_str}]")
                    print(f"    ASCII: {ascii_str}")
                continue

            received_count += 1

            # 解析分块数据
            payload_bytes = bytes(msg.payload)
            payload_len = msg.payload_length

            if payload_len < 2:
                print(f"  [WARN] payload 太短: {payload_len} 字节, 跳过")
                continue

            # 分块头: id (1 byte) + chunk_seq (1 byte)
            chunk_id = payload_bytes[0]
            chunk_seq = payload_bytes[1]
            # 分块数据: 剩余的 payload
            chunk_data_size = min(payload_len - 2, 126)
            chunk_data = payload_bytes[2:2 + chunk_data_size]

            chunk_buffers[chunk_id][chunk_seq] = chunk_data

            # 计算总块数估算 (每个文件 304 字节 / 126 = 3 块)
            print(f"  [CHUNK] #{received_count}: id={chunk_id}, seq={chunk_seq}, "
                  f"data_len={len(chunk_data)}")

    except KeyboardInterrupt:
        print("\n\n用户中断接收")
    except Exception as e:
        print(f"\n接收异常: {e}")

    print(f"\n共接收 {received_count} 个分块, {len(chunk_buffers)} 个消息组")

    # ========================================================================
    # 重组并解析每个文件的完整信息
    # ========================================================================
    file_list = []

    # 按 id 排序以保证文件顺序 (id 是递增的)
    for chunk_id in sorted(chunk_buffers.keys()):
        chunks = chunk_buffers[chunk_id]

        # 按 chunk_seq 排序并拼接
        sorted_seqs = sorted(chunks.keys())
        full_data = b''
        for seq in sorted_seqs:
            full_data += chunks[seq]

        print(f"\n  [ASSEMBLE] id={chunk_id}: {len(sorted_seqs)} 块 → {len(full_data)} 字节")

        if len(full_data) < FILE_LIST_INFO_SIZE:
            print(f"  [WARN] id={chunk_id}: 数据不足 ({len(full_data)} < {FILE_LIST_INFO_SIZE}), 跳过")
            continue

        try:
            info = parse_file_list_info(full_data[:FILE_LIST_INFO_SIZE])
            file_list.append(info)
        except Exception as e:
            print(f"  [ERROR] id={chunk_id}: 解析失败: {e}")

    return file_list


def format_file_size(size_bytes: int) -> str:
    """格式化文件大小"""
    if size_bytes >= 1024 * 1024 * 1024:
        return f"{size_bytes / (1024 ** 3):.2f} GB"
    elif size_bytes >= 1024 * 1024:
        return f"{size_bytes / (1024 ** 2):.2f} MB"
    elif size_bytes >= 1024:
        return f"{size_bytes / 1024:.2f} KB"
    else:
        return f"{size_bytes} B"


def print_file_list(file_list: list):
    """打印文件列表"""
    if not file_list:
        print("\n[结果] 未发现任何文件 (SD 卡可能为空)")
        return

    print("\n" + "=" * 90)
    print(f"相机媒体文件列表 (共 {len(file_list)} 个文件)")
    print("=" * 90)
    print(f"{'#':>4}  {'索引':>7}  {'文件名':<30} {'大小':>10}  {'时间':<20}  {'类型':>10}  {'子文件':>6}")
    print("-" * 90)

    for i, f in enumerate(file_list):
        size_str = format_file_size(f['file_size'])
        # 截断过长的文件名
        name = f['file_name'][:28] + '..' if len(f['file_name']) > 30 else f['file_name']
        print(f"{i:>4}  {f['file_index']:>7}  {name:<30} {size_str:>10}  {f['create_time']:<20}  "
              f"{f['type_name']:>10}  {f['sub_file_count']:>6}")

    print("-" * 90)
    total_size = sum(f['file_size'] for f in file_list)
    print(f"总计: {len(file_list)} 个文件, {format_file_size(total_size)}")


def interactive_mode(mavlink_conn, file_list: list):
    """
    交互模式: 允许用户选择下载文件

    Args:
        mavlink_conn: MAVLink 连接
        file_list: 文件列表
    """
    if not file_list:
        print("没有可下载的文件")
        return

    while True:
        print("\n" + "-" * 50)
        print("操作选项:")
        print("  d <编号>    - 下载指定文件 (例如: d 0)")
        print("  d <编号> <类型> - 指定文件类型下载 (>=32 原始文件)")
        print("  l           - 重新列出文件")
        print("  q           - 退出")
        print("-" * 50)

        try:
            cmd = input("> ").strip().lower()
        except (EOFError, KeyboardInterrupt):
            print("\n退出")
            break

        if cmd == 'q':
            break
        elif cmd == 'l':
            print_file_list(file_list)
            continue
        elif cmd.startswith('d'):
            parts = cmd.split()
            if len(parts) < 2:
                print("用法: d <编号> [类型]")
                continue

            try:
                idx = int(parts[1])
            except ValueError:
                print(f"无效的编号: {parts[1]}")
                continue

            if idx < 0 or idx >= len(file_list):
                print(f"编号超出范围 (0-{len(file_list) - 1})")
                continue

            file_type = 32  # 默认下载原始文件
            if len(parts) >= 3:
                try:
                    file_type = int(parts[2])
                except ValueError:
                    print(f"无效的类型: {parts[2]}")
                    continue

            f = file_list[idx]
            print(f"\n即将下载: [{idx}] {f['file_name']} (index={f['file_index']}, type={file_type})")
            confirm = input("确认下载? (y/N): ").strip().lower()
            if confirm == 'y':
                send_download_request(mavlink_conn, f['file_index'], file_type)
                print("下载命令已发送 (文件将存储在设备本地磁盘)")
                # 短暂等待接收确认/状态消息
                print("等待设备响应...")
                start = time.time()
                while time.time() - start < 5:
                    msg = mavlink_conn.recv_match(type='TUNNEL', blocking=True, timeout=0.5)
                    if msg and msg.payload_type != 0:
                        payload = bytes(msg.payload)
                        text = payload[:msg.payload_length].decode('utf-8', errors='replace').strip('\x00')
                        if text:
                            print(f"  [响应] type={msg.payload_type}: {text}")
        else:
            print(f"未知命令: {cmd}")


def main():
    parser = argparse.ArgumentParser(
        description='MAVLink TUNNEL 相机媒体文件列表获取工具',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例:
  %(prog)s                          # 列出所有文件并进入交互模式
  %(prog)s --list                   # 仅列出文件
  %(prog)s --list --slice 0 --count 60   # 每片 60 个文件
  %(prog)s --download 5 --file-type 32   # 下载第 5 个文件的原始文件
  %(prog)s --download 3 --file-type 0    # 下载第 3 个文件的 JPEG
  %(prog)s --connect udpin:0.0.0.0:14550  # 指定连接地址
        """
    )
    parser.add_argument('--connect', '-c', default='udpin:0.0.0.0:14550',
                        help='MAVLink 连接字符串 (默认: udpin:0.0.0.0:14550)')
    parser.add_argument('--list', '-l', action='store_true',
                        help='列出相机媒体文件列表')
    parser.add_argument('--download', '-d', type=int, default=None,
                        help='下载指定编号的文件 (与 --list 配合使用, 或单独使用时直接按 fileIndex 下载)')
    parser.add_argument('--file-type', '-t', type=int, default=32,
                        help='文件类型 (>=32 原始文件, <32 子文件, 默认: 32)')
    parser.add_argument('--slice', type=int, default=0,
                        help='文件列表分片起始索引 (默认: 0)')
    parser.add_argument('--count', type=int, default=FILE_LIST_COUNT_ALL_PER_SLICE,
                        help=f'每片文件数量 (60/120/{FILE_LIST_COUNT_ALL_PER_SLICE}=全部, 默认: {FILE_LIST_COUNT_ALL_PER_SLICE})')
    parser.add_argument('--timeout', type=float, default=15.0,
                        help='接收文件列表的超时时间(秒) (默认: 15)')
    parser.add_argument('--no-interactive', '-n', action='store_true',
                        help='禁用交互模式 (仅输出文件列表)')
    args = parser.parse_args()

    print("=" * 70)
    print("MAVLink TUNNEL 相机媒体文件列表获取工具")
    print("=" * 70)

    # ========================================================================
    # 建立 MAVLink 连接
    # ========================================================================
    print(f"\n连接: {args.connect}")
    try:
        mavlink_conn = mavutil.mavlink_connection(args.connect)
    except Exception as e:
        print(f"连接失败: {e}")
        return 1

    # 等待心跳包
    print("等待心跳包...")
    heartbeat = mavlink_conn.wait_heartbeat(timeout=10)
    if heartbeat:
        print(f"已连接! sysid={heartbeat.get_srcSystem()}, compid={heartbeat.get_srcComponent()}")
    else:
        print("警告: 未收到心跳包, 继续尝试...")

    file_list = None

    try:
        # ====================================================================
        # 模式 1: 仅下载文件 (直接使用 fileIndex)
        # ====================================================================
        if args.download is not None and not args.list:
            send_download_request(mavlink_conn, args.download, args.file_type)
            print("下载命令已发送 (文件将存储在设备本地磁盘)")
            print("等待设备响应...")
            start = time.time()
            while time.time() - start < 5:
                msg = mavlink_conn.recv_match(type='TUNNEL', blocking=True, timeout=0.5)
                if msg and msg.payload_type not in (0, TUNNEL_PROTOCOL_ID_DOWNLOAD_FILE_LIST):
                    payload = bytes(msg.payload)
                    text = payload[:msg.payload_length].decode('utf-8', errors='replace').strip('\x00')
                    if text:
                        print(f"  [响应] type={msg.payload_type}: {text}")
            return 0

        # ====================================================================
        # 模式 2: 列出文件列表
        # ====================================================================
        send_file_list_request(mavlink_conn, args.slice, args.count)

        # 接收文件列表
        file_list = receive_file_list(mavlink_conn, timeout=args.timeout)

        # 打印文件列表
        print_file_list(file_list)

        # ====================================================================
        # 模式 3: 列出后下载指定文件
        # ====================================================================
        if args.download is not None and file_list and args.download < len(file_list):
            f = file_list[args.download]
            print(f"\n下载文件: [{args.download}] {f['file_name']} (index={f['file_index']}, type={args.file_type})")
            send_download_request(mavlink_conn, f['file_index'], args.file_type)
            print("下载命令已发送 (文件将存储在设备本地磁盘)")

        # ====================================================================
        # 交互模式
        # ====================================================================
        if not args.no_interactive and file_list is not None:
            interactive_mode(mavlink_conn, file_list)

    except KeyboardInterrupt:
        print("\n\n用户中断")
    except Exception as e:
        print(f"\n错误: {e}")
        import traceback
        traceback.print_exc()
        return 1
    finally:
        mavlink_conn.close()
        print("连接已关闭")

    return 0


if __name__ == "__main__":
    exit(main())