先不要从模型侧动手。机械臂操作数据合成老对不齐，常见原因不是合成模型那一层理解错，而是原始数据的时间戳来源、通道定义、角度与弧度单位在三处以上不一致，接入层只是把已有的错位露了出来。判断顺序建议是：先在只读加载里确认字段能对上，再看单条 episode 内时间轴能否单调对齐，最后才决定改数据层、表征层还是训练层。Xiaomi-Robotics-U0 这一侧按统一合成模型对待，接入点选在哪层，取决于你数据的规范程度，而不是模型名气。

接入点不是选“最好”的那层，而是选改动最少、能回退的那层：落盘格式不规范就先补数据层字段；多模态单位、坐标系不一致就在表征层做一次显式转换；只有 batch 组装与对齐策略确实是瓶颈才动训练层。任何一层都先用只读脚本验证字段映射与时间轴单调性，通过后再串小样本端到端计时。失败时按类型分治，不要一边改格式一边调模型。以上判断需结合你自己的采集链路确认。

统计现有数据的模态数量、帧率与单条时长分布

把目录下每条 episode 的元数据读一遍，统计每个模态的通道数、时间戳来源、帧率与单条时长分布。规格不明就直接接，报错会集中在维度不符和时间轴断裂两类，事后很难分清是数据问题还是接入层问题。这一步只读不写，产出一张统计表再决定下一步。

脚本骨架如下，字段名要按你自己的元数据键替换，先跑通再谈统计结论。

import json, pathlib, statistics

root = pathlib.Path('data/robot_arm')
rows = []
for ep in sorted(p for p in root.iterdir() if p.is_dir()):
    meta_path = ep / 'meta.json'
    if not meta_path.exists():
        rows.append({'episode': ep.name, 'err': 'MISSING_META'})
        continue
    meta = json.loads(meta_path.read_text(encoding='utf-8'))
    for mod in meta.get('modalities', []):
        ts = mod.get('timestamps', [])
        dt = [b - a for a, b in zip(ts, ts[1:])]
        rows.append({
            'episode': ep.name,
            'modality': mod.get('name'),
            'channels': mod.get('channels'),
            'frames': len(ts),
            'monotonic': all(d > 0 for d in dt),
            'fps_est': round(1 / statistics.median(dt), 3) if dt else None,
            'duration_s': round(ts[-1] - ts[0], 3) if len(ts) > 1 else None,
        })
print(json.dumps(rows[:20], ensure_ascii=False, indent=2))

异常值处理口径建议统一成下面几条，避免边清洗边接入导致问题无法定位：时间戳非单调的 episode 标记隔离，原始文件不删；反推帧率为空或与声明值差一个量级的，按声明值归档并单独复核；帧数少于 2 或时长为 0 的片段不参与首轮验证；单条时长偏离同模态中位数数倍以上的单独成组，不混进小样本；缺模态的 episode 记录缺失项，不要用空数组补齐后再跑。

把三种接入点写成可对比的表格

三种接入点的差别不在概念，而在改哪些文件、补哪些字段、出问题能不能退回去。先按自己数据的实际缺陷对号入座，再动手。

判断依据可以按这个顺序走：如果统计阶段大量出现 MISSING_META、路径缺失，说明数据层还没规范，先在数据层补字段最省事；如果路径都对、字段也齐，但单位、坐标系、采样率在不同模态间不一致，就在表征层做一次转换，并把换算关系写进配置而不是硬编码在脚本里；只有当字段和单位都对齐、耗时仍集中在对齐与组 batch 上时，才值得动训练层。任何一层的改动都要能用一个开关退回上一版，否则后续排错会失去参照。

写只读加载骨架验证字段映射

这一步的目标是在不修改原始目录的前提下，确认字段名、路径、时间戳长度能对上。先写映射配置，再写只读加载脚本，脚本只读文件、只打日志，不发写操作。

# field_map.yaml
version: 1
episode_root: data/robot_arm
required_fields: [episode_id, modality, timestamp, unit]
modalities:
  rgb:
    source: raw/front_rgb
    timestamp_from: meta.cameras.front.ts
    unit: s
  action:
    source: raw/joint_target.npy
    timestamp_from: meta.ctrl.ts
    unit: rad   # 原始若为角度，在表征层声明换算，不在数据层改

import json, pathlib, argparse, logging, yaml

logging.basicConfig(level=logging.INFO, format='%(levelname)s %(message)s')
log = logging.getLogger('dryrun')

def main(map_path, limit):
    fm = yaml.safe_load(pathlib.Path(map_path).read_text(encoding='utf-8'))
    root = pathlib.Path(fm['episode_root'])
    eps = sorted(p for p in root.iterdir() if p.is_dir())[:limit]
    for ep in eps:
        log.info('open episode=%s mode=readonly', ep.name)
        meta_file = ep / 'meta.json'
        if not meta_file.exists():
            log.error('missing episode=%s path=meta.json', ep.name)
            continue
        meta = json.loads(meta_file.read_text(encoding='utf-8'))
        for name, spec in fm['modalities'].items():
            src = ep / spec['source']
            if not src.exists():
                log.error('missing episode=%s modality=%s path=%s', ep.name, name, spec['source'])
                continue
            ts_key = spec['timestamp_from'].split('.')
            node = meta
            for k in ts_key:
                node = node.get(k, {}) if isinstance(node, dict) else {}
            n = len(node) if isinstance(node, list) else 0
            log.info('map episode=%s modality=%s unit=%s n=%s', ep.name, name, spec['unit'], n)

if __name__ == '__main__':
    ap = argparse.ArgumentParser()
    ap.add_argument('`--map`', default='field_map.yaml')
    ap.add_argument('`--limit`', type=int, default=5)
    a = ap.parse_args()
    main(a.map, a.limit)

预期日志与判读方式：出现 INFO map ... n=0 说明时间戳键路径写错了或该模态没有时间戳，先改映射配置；出现 ERROR missing ... path=raw/front_rgb 说明目录命名与配置不一致，属于数据层问题；如果各模态都能读到但日志里帧数差异明显，说明重采样或对齐要放到表征层，而不是在加载脚本里补零。只读脚本跑通再谈写入，能省掉大量“改了格式后原始数据对不上”的返工。

在小样本上串一次端到端并计时

小样本规模建议控制在 3 到 5 条，覆盖每种模态组合，并刻意包含最短、最长和中间时长各一条。样本再大就会把环境问题和数据问题混在一起，样本只有一条又验证不出长短样本混批时的对齐差异。

计时点按段拆开，每段用 time.perf_counter() 记录，写入一张表后再比较各层接入方式的差异。

import time
marks = {}
t0 = time.perf_counter()
marks['enumerate'] = time.perf_counter() - t0
# 依次在解析元数据、字段映射、构建对齐索引、组 batch、
# 调用一次合成或前向之后插入同样的打点
print(marks)

耗时表只用于定位瓶颈在哪一段，不建议拿单次数字当结论。若耗时集中在前两项，先修数据层；集中在构建对齐索引，说明表征层的转换与索引策略要调整；集中在组 batch，再考虑训练层接入。

给接入失败建立错误分类与定位顺序

接入失败时最容易犯的错是把类型不匹配、缺帧、路径错误混在一起改。先在日志里给每类错误打固定前缀，再按类别顺序排查，回退动作也按类别定义。

定位顺序建议固定为：先看 PATH_NOT_FOUND 与 MISSING_META，排除取数问题；再看 SCHEMA_MISMATCH 与 UNIT_MISMATCH，确认字段映射与单位声明；最后看 GAP 与 SHAPE_MISMATCH，处理对齐与维度。每一类改动都保留一个开关或分支，回退时只回退当前类别，不要把数据层和训练层的改动一起撤回，否则下一次排错又要从头开始。