mcap2kusa 0.1.0

Convert MCAP files to KUSA record format

mcap_to_kusa_record 工具说明

功能概述

• 该工具负责将 Foxglove 生态的 MCAP 文件转换为 KUSA ADS 系统可识别的 rec_x 原始 underlay 日志格式。
• 支持将 MCAP 文件中的所有消息类型转换为 KUSA record 格式，保持原始数据的完整性。
• 输出的 record 文件可以直接在 KUSA ADS 系统中使用，用于数据回放和分析。

主要特性

单个文件处理

• 读取 MCAP 文件，解析所有 schema 和 channel 信息
• 将每条消息转换为 KUSA UnderlayMessage 格式
• 按每 1 万条消息分割为一个 record 文件

目录处理（优化版本）

• 步骤1：建立时间索引 - 扫描所有 MCAP 文件，建立每个文件的时间范围索引，包含消息数量统计
• 步骤2：按数据间隙分组 - 当 5 秒内没有任何数据时产生新分组
• 步骤3：按需转换 - 按照 10 秒间隔 + 最大消息数量限制（10,000 条）生成 record 文件
• 步骤4：内存优化 - 每次只打开需要的文件，避免一次性加载全部文件导致内存不足
• 步骤5：记录无效文件 - 记录无效/损坏的 MCAP 文件

内存监控

• 实时监控内存使用情况
• 15GB 内存阈值，超过自动退出
• 生成详细的内存使用报告

额外功能

• stats：统计 MCAP 文件各 topic 的消息数量
• analyze：分析时间跳变和帧率稳定性
• time_index.log：保存完整的时间索引信息
• invalid_files.log：记录无效文件

依赖环境

• Python 3.8+（注意：必须使用 Python 3，不支持 Python 2）
• click
• google.protobuf
• kusa_ads_protobuf（内部 Proto 定义）
• mcap
• psutil（用于内存监控）
• matplotlib（用于生成图表）

安装方法

该工具需要安装以下依赖：

pip3 install click protobuf mcap psutil matplotlib kusa_ads_protobuf

注意：kusa_ads_protobuf 是内部包，需要根据实际环境配置安装路径。

使用方法

1. 转换 MCAP 到 Record 格式

python3 -m mcap2kusa.cli convert <input_path> -o <output_dir> [-v]

• input_path：输入的 MCAP 文件路径或包含 MCAP 文件的目录路径。
• -o, --output：必填，输出目录路径，用于存放生成的 record 文件。
• -v/--verbose：打印更多提示（例如处理进度、schema 信息等）。

注意：对于目录处理，--time-diff 和 --common-time 参数已不再使用（已被新的 5 秒数据间隙分组逻辑取代）。

单个文件执行过程

1. 读取 MCAP 文件，解析所有 schema 和 channel 信息
2. 遍历所有消息，将每条消息转换为 KUSA UnderlayMessage 格式
3. 按每 1 万条消息分割为一个 record 文件（rec_0, rec_1, ...）
4. 在输出目录中生成对应的 info 文件，包含时间范围和消息统计信息
5. 结束时输出各 topic 的消息条数及总耗时

目录执行过程

1. 步骤1：建立时间索引

   • 收集目录中所有 MCAP 文件
   • 逐个读取文件，建立每个文件的时间范围索引
   • 统计每个文件的消息数量
   • 收集所有有效消息的时间戳
   • 保存时间索引到 time_index.log

2. 步骤2：按数据间隙分组

   • 分析所有时间戳，找出超过 5 秒的数据间隙
   • 在间隙处分割，形成连续的数据组
   • 显示分组信息

3. 步骤3-4：转换为 record 文件

   • 按 10 秒时间间隔 + 10,000 条消息限制划分 record 文件
   • 按需打开文件 - 处理每个 record 文件时只打开需要的 mcap 文件
   • 快速跳过时间范围不相关的文件
   • 实时监控内存，超过 15GB 自动退出
   • 在每个组的子目录中生成 record 文件

4. 步骤5：记录无效文件

   • 将无效/损坏的文件记录到 invalid_files.log

5. 结束

   • 生成各分组的 info 文件
   • 输出各 topic 的消息条数
   • 打印完整的内存使用报告

2. 查看统计信息

python3 -m mcap2kusa.cli stats <input.mcap> [-v]

• 默认模式：解析 MCAP 文件并统计每个 topic 的消息数量。

3. 分析时间跳变和帧率稳定性

python3 -m mcap2kusa.cli analyze <input_path> [--threshold THRESHOLD] [--generate-plots] [-o OUTPUT_DIR]

• <input_path>：输入的 MCAP 文件路径或包含 MCAP 文件的目录路径。
• --threshold：可选，时间跳变阈值（单位：秒，默认：根据文件时间长度和帧数量自动计算 1.5 倍平均间隔）。
• --generate-plots, -p：可选，为每个 topic 生成帧间隔随时间变化的图表。
• -o, --output：可选，输出目录，将分析结果保存为文本文件。

阈值计算方式

• 默认情况下，阈值会根据每个 topic 的时间长度和消息数量自动计算：
  • avg_interval = (end_time - start_time) / (num_messages - 1)
  • threshold = avg_interval * 1.5
• 如果用户指定了 --threshold 参数，则使用用户指定的值

输出文件

• 如果输入是单个文件，输出文件命名为：{filename}.txt
• 如果输入是目录，输出文件命名为：directory_analysis.txt
• 如果使用了 --generate-plots，还会为每个 topic 生成单独的 PNG 图表文件，命名为：{prefix}_{topic}.png

图表说明

每个图表包含以下内容：

• 横轴：logtime（绝对时间戳，格式：YYYY-MM-DD HH:MM:SS）
• 纵轴：帧间隔时间（毫秒）
• 蓝色实线：每一帧的间隔时间
• 绿色虚线：平均帧间隔
• 红色虚线：时间跳变阈值
• 红色圆点：检测到的时间跳变点

Y轴范围调整策略：

• 如果有跳变点：显示到最大跳变值的 1.1 倍
• 如果没有跳变点：显示到阈值的 2 倍或最大间隔的 1.5 倍
• 单位为毫秒，更符合数据间隔的实际尺度

4. 按文件分组分析时间跳变和帧率稳定性

python3 -m mcap2kusa.cli analyze-group <input_dir> [--threshold THRESHOLD] [--generate-plots] [-o OUTPUT_DIR]

• <input_dir>：包含 MCAP 文件的目录路径。
• --threshold：可选，时间跳变阈值（单位：秒，默认：根据文件时间长度和帧数量自动计算 1.5 倍平均间隔）。
• --generate-plots, -p：可选，为每个 topic 生成帧间隔随时间变化的图表。
• -o, --output：可选，输出目录，将分析结果保存为文本文件。

分组规则

文件名中 KUSA 前面字段一样的文件会被合并到同一组中，例如：

• rt.openads.task.task_config-KUSA-147-20260417164258-1748963-3.mcap
• rt.openads.task.task_config-KUSA-147-20260417164258-1748963-5.mcap

这两个文件会被合并到 rt.openads.task.task_config-KUSA-147 分组中。

输出文件

• 分析报告：group_analysis.txt
• 如果使用了 --generate-plots，还会为每个分组的每个 topic 生成单独的 PNG 图表文件，命名为：{group_key}_{topic}.png

5. 按数据间隙分组分析时间跳变和帧率稳定性

python3 -m mcap2kusa.cli analyze-gap <input_dir> [--threshold THRESHOLD] [--gap-seconds SECONDS] [--generate-plots] [-o OUTPUT_DIR]

• <input_dir>：包含 MCAP 文件的目录路径。
• --threshold：可选，时间跳变阈值（单位：秒，默认：根据文件时间长度和帧数量自动计算 1.5 倍平均间隔）。
• --gap-seconds：可选，数据间隙分组阈值（单位：秒，默认：5）。
• --generate-plots, -p：可选，为每个 topic 生成帧间隔随时间变化的图表。
• -o, --output：可选，输出目录，将分析结果保存为文本文件。

分组规则

与 convert 命令相同的分组方式：

1. 首先建立时间索引（记录每个 MCAP 文件的时间范围和消息数量
2. 然后按照数据间隙分组（默认 5 秒内没有任何数据时产生新分组）

输出文件

• 分析报告：gap_analysis.txt
• 如果使用了 --generate-plots，还会为每个分组的每个 topic 生成单独的 PNG 图表文件，命名为：group_{id}_{topic}.png

输出格式

单个文件输入

• 输出文件命名：rec_0, rec_1, rec_2, ...
• 生成对应的 info 文件，包含时间范围和消息统计信息

目录输入

• 为每个数据组创建子目录，命名格式：20260415110340（使用组内最早的时间戳）
• 在每个子目录中输出文件：rec_0, rec_1, rec_2, ...
• 在每个子目录中生成对应的 info 文件，包含时间范围和消息统计信息
• 在根输出目录生成：
  • time_index.log：完整的时间索引
  • invalid_files.log：无效文件记录

通用设置

• 每个 record 文件最多包含 10,000 条消息（可通过修改代码中的 MAX_MESSAGES_PER_FILE 常量调整）
• 文件格式：KUSA UnderlayMessageStream protobuf 格式
• info 文件格式：文本格式，包含开始时间、结束时间、每个 topic 的消息数量和总消息数
• 内存限制：15 GB，超过自动退出

输出文件示例

time_index.log

Time Index - 2026-04-18 10:30:45
================================================================================

Global Time Range:
  Start: 2026-04-18 10:00:00.123
  End:   2026-04-18 10:15:30.456
  Total messages: 125890

File Time Index:
--------------------------------------------------------------------------------

File 1:
  Path: /input/record_00001.mcap
  Start: 2026-04-18 10:00:00.123
  End:   2026-04-18 10:05:23.456
  Duration: 323.333 seconds
  Message count: 45678

File 2:
  Path: /input/record_00002.mcap
  Start: 2026-04-18 10:05:15.789
  End:   2026-04-18 10:10:45.123
  Duration: 329.334 seconds
  Message count: 43210

...

invalid_files.log

Invalid files log - 2026-04-18 10:30:45
================================================================================

File 1:
  Path: /input/corrupted.mcap
  Reason: Not a valid MCAP file

File 2:
  Path: /input/empty.mcap
  Reason: No messages

内存使用报告

================================================================================
Memory Usage Report
================================================================================

  Initial: 45.23 MB
  Final:   67.89 MB
  Min:     45.23 MB
  Max:     78.12 MB
  Delta:   +22.66 MB
  Limit:   15 GB (15360 MB)

  Memory usage by stage:
    [10:30:15] Start                         -    45.23 MB
    [10:30:18] After build_time_index        -    52.15 MB
    [10:30:18] After save_time_index         -    52.18 MB
    [10:30:19] After group_time_index        -    52.45 MB
    [10:30:22] Group 0 - collected timestamps -    55.78 MB
    [10:30:25] Group 0 - rec_0 written      -    60.12 MB
    [10:30:28] Group 0 - rec_1 written      -    63.45 MB
    [10:30:32] After convert_groups_to_record -    67.89 MB

================================================================================

支持的消息类型

该工具支持所有 KUSA ADS 消息类型的转换，通过通用的 UnderlayMessage 格式进行处理，无需针对具体消息类型进行特殊处理。

转换过程中，工具会自动处理 MCAP 文件中的所有消息，将它们封装到 UnderlayMessage 中，并保留原始消息的 topic、数据类型和时间戳信息。

常见问题

• Python 版本问题：确保使用 python3 命令运行脚本，不要使用 python（可能是 Python 2）。
• 转换失败：确认 MCAP 文件格式正确，且所有 schema 都有对应的 protobuf 定义。
• 内存不足：
  • 工具已优化内存使用，按需打开文件
  • 如果仍有问题，可以考虑调整 MAX_MESSAGES_PER_FILE 的值来减少单个 record 文件的大小
  • 工具会在超过 15GB 时自动退出
• 依赖安装问题：如果遇到 protobuf 版本冲突，请确保使用 pip3 安装最新版本的 protobuf 和 psutil。

该工具的编译和发布

• 该工具采用 pyproject.toml 进行项目配置，使用 setuptools 和 pip 进行打包和发布。

打包命令：

python -m build

发布命令：

twine upload dist/* --repository pypi

• 可以通过 pip install mcap2kusa 命令安装该工具。

使用示例

示例 1：转换单个 MCAP 文件

python3 -m mcap2kusa.cli convert /path/to/input.mcap -o /path/to/output_dir -v

示例 2：处理目录中的 MCAP 文件

python3 -m mcap2kusa.cli convert /path/to/mcap_dir -o /path/to/output_dir -v

示例 3：查看 MCAP 文件统计信息

python3 -m mcap2kusa.cli stats /path/to/input.mcap -v

示例 4：分析 MCAP 文件的时间跳变和帧率稳定性

python3 -m mcap2kusa.cli analyze /path/to/input.mcap --threshold 0.03 -o ./analysis_output

与 kusa_record_to_mcap 的关系

• kusa_record_to_mcap：将 KUSA record 格式转换为 MCAP 格式，用于在 Foxglove Studio 中查看。
• mcap_to_kusa_record：将 MCAP 格式转换为 KUSA record 格式，用于在 KUSA ADS 系统中使用。

这两个工具互为逆操作，可以实现数据的双向转换。