Blame for processor/app/hdf5_export.py

"""HDF5 dataset export — 与 LeRobot 导出同源数据,写成单一 .h5 文件。

组织(与 LeRobot observation 语义对齐,一集一个 group):

    dataset.h5
    ├─ /meta/info                      attrs: format/version/fps/features
    ├─ /episode_000000
    │  ├─ /observation
    │  │  ├─ /images/<source_key>      (T, H, W, 3) uint8   视频抽帧(RGB)
    │  │  ├─ /state/hand_left_world    (T, 21, 3) float32   米制 3D 骨骼
    │  │  ├─ /state/hand_right_world   (T, 21, 3) float32
    │  │  ├─ /tactile/left             (T, 16, 16) float32  SenseGlove 压力阵列
    │  │  ├─ /tactile/right            (T, 16, 16) float32
    │  │  ├─ /hand_pose/left           (T, 63) float32      采集端姿态参数
    │  │  ├─ /hand_pose/right          (T, 63) float32
    │  │  ├─ /imu                      (T, 6) float32
    │  │  ├─ /hand_0_gesture ...       (T,) 变长字符串      手指伸展标签(角度判定)
    │  │  └─ /annotation ...           (T,) 变长字符串列
    │  ├─ /action                      (T, D) float32
    │  ├─ /videos/hand_skeleton        (nbytes,) uint8    手部骨骼渲染 MP4 原始字节
    │  └─ /meta                        attrs: episode_id/task/split/annotations

数值列统一 float32(缺帧 NaN 填充),字符串列变长 utf8;
视频逐帧流式写入(内存安全),gzip 压缩 + 每帧一个 chunk。
"""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

from app.localstore import get_episode
from app.lerobot_export import (
    _annotation_maps,
    _calibration_docs,
    _depth_video_paths,
    _hand_columns,
    _infer_hand_3d_unit,
    _key_matches_any,
    _prepare_hand_export_inputs,
    _probe_video,
    _read_hand_rows,
    _read_hand_3d_rows_by_source,
    _device_hand_features,
    _device_metadata,
    _read_sensor_rows,
    _source_path,
)

# 采集端姿态参数(63)与压力阵列(256)的语义路径
_TACTILE_KEYS = ("observation.tactile.left", "observation.tactile.right")
_POSE_KEYS = ("observation.left_hand_pose", "observation.right_hand_pose")


def _read_timestamps(session_dir: Path):
    """读取采集端 timestamps.json(帧级时间戳,每帧 2 条 = 左右目各 1 条)。

    返回 (M, 4) float64 数组 [frame_index, timestamp(秒), wall_time(Unix 秒),
    hardware_ns], 保持原始顺序; 文件缺失/解析失败 → None(不影响导出)。
    """
    import numpy as np
    path = Path(session_dir) / "timestamps.json"
    if not path.exists():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        rows = data.get("timestamps") or []
        if not rows:
            return None
        return np.asarray(
            [[float(r.get("frame_index", 0)), float(r.get("timestamp", 0)),
              float(r.get("wall_time", 0)), float(r.get("hardware_ns", 0))]
             for r in rows], dtype=np.float64)
    except Exception:
        return None


def _find_hand_render_video(hand_3d_paths: list[str], session_dir: Path,
                            hand_keypoints_paths: list[str] | None = None) -> Path | None:
    """No rendered hand video is exported; previews are browser overlays."""
    return None


def _collect(columns: dict[str, dict[int, Any]], str_columns: set[str],
             key: str, value: Any, frame_index: int) -> None:
    """按 key + 帧号收集(字典天然对齐,缺帧在写入时 NaN/空串填充)。"""
    if isinstance(value, (str, bytes)):
        str_columns.add(key)
    columns.setdefault(key, {})[frame_index] = value


def _write_columns(obs_group: Any, columns: dict[str, dict[int, Any]],
                   str_columns: set[str], T: int) -> dict[str, dict]:
    """把收集好的列写成 h5py dataset,返回 {feature key: 描述}。"""
    import h5py
    import numpy as np
    features: dict[str, dict] = {}

    for key, by_frame in columns.items():
        if not by_frame:
            continue
        path = key[len("observation."):].replace(".", "/") \
            if key.startswith("observation.") else key

        # 字符串列:变长 utf8
        if key in str_columns:
            ds = obs_group.create_dataset(
                path, shape=(T,), dtype=h5py.string_dtype(encoding="utf-8"),
                compression="gzip")
            for i in range(T):
                ds[i] = str(by_frame.get(i, ""))
            features[key if key.startswith("observation.") else f"observation.{key}"] = {
                "dtype": "string", "shape": [1]}
            continue

        # 数值列:首值定 shape,缺帧 NaN
        first = next(value for value in by_frame.values() if value is not None)
        base_shape = np.asarray(first).shape
        is_tactile = key in _TACTILE_KEYS
        if is_tactile:
            base_shape = (16, 16)
        # 手部骨骼列与 LeRobot 一致走 63 位扁平存储,这里恢复 (21,3)
        is_hand_coordinate = any(
            key == f"observation.state.hand_{side}_{coord}{suffix}"
            for side in ("left", "right")
            for coord in ("world", "3d", "2d")
            for suffix in ("", "_rcam")
        ) or ("observation.state.devices." in key and any(
            f"hand_{side}_{coord}" in key
            for side in ("left", "right")
            for coord in ("world", "3d", "2d")
        ))
        if is_hand_coordinate:
            base_shape = (21, 3)
        # 跨帧形状不一致 = 变长聚合(如 IMU 每帧样本数不定)→ 平铺 + 帧索引,
        # 不丢数据;固定形状走常规 (T, ...) NaN 填充
        shapes = {np.asarray(v).shape for v in by_frame.values() if v is not None}
        if len(shapes) > 1:
            # dtype 保持源语义:整型(如 imu_ts_ns 的 ns 时间戳)用 int64,
            # 否则 float32 —— float32 只有 7 位有效数字,会丢 ns 精度
            kinds = {np.asarray(v).dtype.kind for v in by_frame.values() if v is not None}
            out_dtype = np.int64 if kinds <= {"i", "u"} else np.float32
            arrays = [np.asarray(v, dtype=out_dtype)
                      for v in by_frame.values() if v is not None]
            # 1D 变长列(如 imu_ts_ns)→ 列向量 (N,1) 再按帧拼接;
            # 2D+ 列(如 imu (N,6))保持原样(axis=0 按帧堆叠)
            arrays = [a if a.ndim >= 2 else a.reshape(-1, 1) for a in arrays]
            dim = arrays[0].shape[-1]
            flat = np.concatenate(arrays, axis=0)
            frame_ids = np.concatenate([
                np.full(len(a), fi, dtype=np.int32)
                for fi, a in zip(by_frame.keys(), arrays)])
            ds = obs_group.create_dataset(path, data=flat, compression="gzip")
            ds.attrs["note"] = (
                "变长聚合:每帧样本数不定(如 IMU 按帧聚合,200Hz/25fps≈8样本/帧);"
                f"按 {path}_frame_index 切片还原每帧样本")
            obs_group.create_dataset(path + "_frame_index", data=frame_ids,
                                     compression="gzip")
            features[key if key.startswith("observation.") else f"observation.{key}"] = {
                "dtype": str(np.dtype(out_dtype)), "shape": [dim],
                "note": "变长聚合(每帧样本数不定),配 {path}_frame_index 列切片还原",
            }
            continue
        stacked = np.full((T,) + base_shape, np.nan, dtype=np.float64)
        for i, value in by_frame.items():
            arr = np.asarray(value)
            if is_tactile:
                arr = arr.reshape(16, 16)
            elif is_hand_coordinate:
                arr = arr.reshape(21, 3)
            stacked[i] = arr
        ds = obs_group.create_dataset(path, data=stacked.astype(np.float32),
                                      compression="gzip")
        features[key if key.startswith("observation.") else f"observation.{key}"] = {
            "dtype": "float32",
            "shape": list(stacked.shape[1:]) or [1],
        }
    return features


def build_hdf5_dataset(dataset_name: str, episode_ids: list[str], output_dir: Path,
                       split_ratio: float = 0.9,
                       include_video_keys: list[str] | None = None,
                       hand_keypoints_paths: list[str] | None = None,
                       hand_3d_paths: list[str] | None = None,
                       hand_3d_unit: str | None = None,
                       compression: str = "gzip", level: int = 4) -> Path:
    """把批次构建为单一 .h5 数据集(与 LeRobot 导出同源数据)。

    ``include_video_keys``: 只抽帧这些 source_key 对应的视频(工作流连接
    驱动);None = 全量;空列表 = 不抽帧。

    ``hand_keypoints_paths`` / ``hand_3d_paths``: 手部骨骼/深度 3D 产物。
    RGB_TO_3D 的 hand_3d 仅用于前端预览,导出时自动读取其中的 2D 关键点;
    只有真实深度产物才写入 HDF5 的 world 3D 字段。
    """
    try:
        import h5py
    except ImportError as exc:
        raise RuntimeError("h5py is required for HDF5 export") from exc
    import cv2
    import numpy as np

    episodes = [get_episode(str(episode_id)) for episode_id in episode_ids]
    episodes = [episode for episode in episodes if episode]
    if not episodes:
        raise RuntimeError("No valid episodes selected for export")

    output_dir.mkdir(parents=True, exist_ok=True)
    out_path = output_dir / f"{dataset_name}.h5"

    (hand_keypoints_paths, hand_3d_paths, _hand_3d_right_paths,
     hand_3d_unit) = _prepare_hand_export_inputs(
         hand_keypoints_paths, hand_3d_paths, None, hand_3d_unit)
    hand_rows = _read_hand_rows(hand_keypoints_paths, hand_3d_paths)
    device_hand_rows = _read_hand_3d_rows_by_source(hand_3d_paths)
    hand_3d_unit = hand_3d_unit or _infer_hand_3d_unit(hand_3d_paths)
    coordinate_key = (
        "world" if hand_3d_unit in {"camera_meters", "meter", "meters"} else
        "3d" if hand_3d_paths else "2d"
    )
    is_3d = any(
        entry and entry.get("is_3d")
        for hand in hand_rows.values()
        for entry in hand.values()
    ) if hand_rows else False

    total_frames = 0
    features: dict[str, dict] = {}
    fps_first = float(episodes[0].get("fps") or 30)

    with h5py.File(str(out_path), "w") as f:
        meta = f.create_group("meta/info")

        for out_episode_index, episode in enumerate(episodes):
            episode_id = str(episode["id"])
            session_dir = Path(episode["path"])
            streams = episode.get("camera_streams") or {}
            sensor_rows = _read_sensor_rows(session_dir)
            ts_arr = _read_timestamps(session_dir)
            annotation_map, annotation_defs, _skipped_candidates = _annotation_maps(episode_id)
            episode_fps = float(episode.get("fps") or 30)

            # 连接驱动:只抽帧连到导出节点的视频源
            video_meta: list[tuple[str, Path, int, float, int, int]] = []
            for source_key, stream in sorted(streams.items()):
                if include_video_keys is not None and not _key_matches_any(source_key, include_video_keys):
                    continue
                source = _source_path(stream.get("path"))
                if source is None or not source.exists():
                    continue
                count, vfps, width, height = _probe_video(source)
                video_meta.append((source_key, source, count or 0, vfps or episode_fps,
                                   width, height))

            sensor_max = max(sensor_rows.keys(), default=-1) + 1
            hand_max = max(hand_rows.keys(), default=-1) + 1
            T = max([meta[2] for meta in video_meta] + [sensor_max, hand_max, 0])
            total_frames += T

            ep = f.create_group(f"episode_{out_episode_index:06d}")
            obs = ep.create_group("observation")

            # 1) 视频 → 逐帧抽帧写 uint8 数组(gzip + 每帧 chunk,流式省内存)
            for source_key, source, count, vfps, width, height in video_meta:
                if width <= 0 or height <= 0 or T <= 0:
                    continue
                ds = obs.create_dataset(
                    f"images/{source_key}", shape=(T, height, width, 3),
                    dtype="uint8", chunks=(1, height, width, 3),
                    compression=compression, compression_opts=level)
                ds.attrs["fps"] = vfps
                ds.attrs["source_key"] = source_key
                ds.attrs["source_episode"] = episode_id
                cap = cv2.VideoCapture(str(source))
                try:
                    i = 0
                    while i < T:
                        ok, frame = cap.read()
                        if not ok:
                            break
                        ds[i] = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
                        i += 1
                finally:
                    cap.release()
                features[f"observation.images.{source_key}"] = {
                    "dtype": "uint8", "shape": [T, height, width, 3],
                    "video_info": {"video.fps": vfps, "video.is_depth_map": False},
                }

            # 1b) 深度流 → 逐帧 uint16 毫米(单独一组,不和 RGB 混在 images/ 下)
            #
            # 深度**不在** camera_streams 里 —— 它单独放在 depth/<source>/,
            # 所以上面的 RGB 循环根本看不到它,必须用 lerobot_export 的发现函数。
            #
            # 解码必须走 DepthVideoReader(ffmpeg 解 gray12le):OpenCV 会把
            # 12-bit 深度当普通显示图转成 8-bit,真实毫米值全丢。
            #
            # **0 表示"没测到深度"的无效像素,不是 0 毫米** —— 与采集端语义一致,
            # 原样保留,不要当数值去算均值。
            for source_key, depth_path in _depth_video_paths(session_dir, episode_id):
                if include_video_keys is not None and not _key_matches_any(
                        source_key, include_video_keys):
                    continue
                reader = None
                try:
                    from app.lerobot_v21 import DEPTH_VIDEO_ENCODING, DepthVideoReader
                    reader = DepthVideoReader(depth_path)
                    dheight, dwidth = reader.height, reader.width
                    if dheight <= 0 or dwidth <= 0 or T <= 0:
                        continue
                    dfps = float(reader.fps or episode_fps)
                    ds = obs.create_dataset(
                        f"depth/{source_key}", shape=(T, dheight, dwidth),
                        dtype="uint16", chunks=(1, dheight, dwidth),
                        compression=compression, compression_opts=level)
                    ds.attrs["unit"] = "millimetre"
                    ds.attrs["invalid_value"] = 0
                    ds.attrs["fps"] = dfps
                    ds.attrs["source_key"] = source_key
                    ds.attrs["source_episode"] = episode_id
                    ds.attrs["encoding"] = DEPTH_VIDEO_ENCODING
                    i = 0
                    while i < T:
                        frame = reader.read_mm()
                        if frame is None:
                            break
                        ds[i] = frame
                        i += 1
                    features[f"observation.depth.{source_key}"] = {
                        "dtype": "uint16", "shape": [T, dheight, dwidth],
                        "unit": "millimetre",
                        "note": "0 = 无效像素(该点未测到深度),不是 0 毫米",
                        "video_info": {
                            "video.fps": dfps,
                            "video.is_depth_map": True,
                            "video.height": dheight,
                            "video.width": dwidth,
                        },
                    }
                except Exception as exc:
                    print(f"[hdf5_export] depth stream {source_key} "
                          f"skipped: {exc}")
                finally:
                    if reader is not None:
                        try:
                            reader.close()
                        except Exception:
                            pass

            # 2) 表列:传感器 + 手部骨骼 + 标注(与 LeRobot 列同构,按帧号对齐)
            columns: dict[str, dict[int, Any]] = {}
            str_columns: set[str] = set()
            for frame_index in range(T):
                row: dict[str, Any] = {}
                row.update(sensor_rows.get(frame_index, {}))
                hand = hand_rows.get(frame_index)
                if hand:
                    row.update(_hand_columns(hand, coordinate_key=coordinate_key))
                for device_source, rows_by_frame in device_hand_rows.items():
                    device_hand = rows_by_frame.get(frame_index)
                    if device_hand:
                        row.update(_hand_columns(
                            device_hand, coordinate_key=coordinate_key,
                            device_namespace=device_source))
                anno, anno_index, _scope = annotation_map.get(
                    frame_index, (None, -1, ["episode"]))
                row["annotation"] = anno or ""
                row["annotation_index"] = anno_index
                for key, value in row.items():
                    _collect(columns, str_columns, key, value, frame_index)

            # 3) 动作列(独立 dataset,非 observation 子项)
            if "action" in columns and "action" not in str_columns:
                import numpy as np
                by_frame = columns["action"]
                first = next(value for value in by_frame.values()
                             if value is not None)
                act = np.full((T,) + np.asarray(first).shape, np.nan)
                for i, value in by_frame.items():
                    act[i] = np.asarray(value)
                if act.ndim == 1:
                    act = act.reshape(-1, 1)
                ep.create_dataset("action", data=act.astype(np.float32),
                                  compression="gzip")
                features["action"] = {"dtype": "float32",
                                      "shape": list(act.shape[1:]) or [1]}
                del columns["action"]

            col_features = _write_columns(obs, columns, str_columns, T)
            features.update(col_features)
            if device_hand_rows:
                device_unit = hand_3d_unit or (
                    "mediapipe_world_relative" if coordinate_key == "3d"
                    else "normalized_image_coords")
                device_note = (
                    "深度图抬升相机系米制 3D(每个物理设备独立坐标系,米)"
                    if coordinate_key == "world" else
                    "MediaPipe world_landmarks 相对 3D(设备独立)"
                    if coordinate_key == "3d" else
                    "MediaPipe 21 关键点归一化图像坐标(设备独立)")
                for device_source in device_hand_rows:
                    features.update(_device_hand_features(
                        device_source, coordinate_key, device_unit, device_note))

            # 2b) 采集端帧级时间戳(timestamps.json 原样保留):
            # 每帧 2 条 = 左右目各 1 条; timestamp(秒) / wall_time(Unix 秒) /
            # hardware_ns 三路时钟,客户可做帧级时间对齐分析。
            if ts_arr is not None and len(ts_arr):
                tds = obs.create_dataset("timestamps", data=ts_arr,
                                         compression="gzip")
                tds.attrs["note"] = (
                    "采集端帧时间戳(每帧2条=左右目各1条),列:"
                    "[frame_index, timestamp(秒), wall_time(Unix秒), hardware_ns];"
                    "timestamp/wall_time 交叉验证文件帧率与真实录制时长一致")
                features["observation.timestamps"] = {
                    "dtype": "float64", "shape": list(ts_arr.shape),
                    "note": "采集端帧时间戳,每帧2条(左右目),"
                            "列=[frame_index, timestamp, wall_time, hardware_ns]",
                }

            # 4) 手部骨骼渲染视频(MP4 原始字节)→ /videos/hand_skeleton。
            # 与 hand_3d parquet 同源交付;attrs 记录格式信息,客户可提取为
            # 独立 MP4 直接用播放器打开(无需 h5py 逐帧还原)。只存原始字节,
            # 体积小(约 10 MB),不重复抽帧。
            hand_video = _find_hand_render_video(
                hand_3d_paths or [], session_dir, hand_keypoints_paths)
            if hand_video is not None:
                vcount, vfps, vwidth, vheight = _probe_video(hand_video)
                mp4_bytes = np.frombuffer(hand_video.read_bytes(), dtype="uint8")
                vds = ep.create_group("videos").create_dataset(
                    "hand_skeleton", data=mp4_bytes)
                vds.attrs["format"] = "mp4"
                vds.attrs["fps"] = vfps
                vds.attrs["width"] = vwidth
                vds.attrs["height"] = vheight
                vds.attrs["frames"] = vcount
                vds.attrs["source"] = hand_video.name
                # 单目(mediapipe_hand 的 skeleton/ 产物)与双目(三角化
                # hand_3d/ 产物)渲染视频语义不同, description 区分开。
                mono = hand_video.parent.name == "skeleton"
                vds.attrs["description"] = (
                    "MediaPipe hand-skeleton render video "
                    "(single camera, skeleton overlay)" if mono
                    else "Stereo-triangulated hand-skeleton render video "
                         "(left/right side by side, skeleton overlay)")
                features["videos.hand_skeleton"] = {
                    "dtype": "uint8", "format": "mp4", "fps": vfps,
                    "width": vwidth, "height": vheight, "frames": vcount,
                    "note": "MP4 原始字节,提取后为独立 .mp4 文件",
                }

            # 5) episode meta
            ep_meta = ep.create_group("meta")
            ep_meta.attrs["episode_id"] = episode_id
            ep_meta.attrs["task_description"] = str(episode.get("project") or "")
            ep_meta.attrs["fps"] = episode_fps
            ep_meta.attrs["length"] = T
            ep_meta.attrs["devices"] = json.dumps(
                _device_metadata(session_dir), ensure_ascii=False)
            ep_meta.attrs["split"] = (
                "train" if out_episode_index < max(1, int(len(episodes) * split_ratio))
                else "val")
            if annotation_defs:
                ep_meta.attrs["annotations"] = json.dumps(
                    annotation_defs, ensure_ascii=False)

            # 6) 标定 → /meta/calibration/<episode>/<name> attrs(JSON 字符串)
            # 每集一份,所以把 episode_id 一起传进去 —— 批次内各集标定可以不同,
            # 不传就只会看到批次级的 calibration/ 目录。
            # 缺失/全零时的默认标定回退统一在 lerobot_export._calibration_docs。
            cal_group = meta.create_group(f"calibration/{episode_id}")
            cal_docs = _calibration_docs(session_dir, episode_id)
            for name, value in cal_docs.items():
                cal_group.attrs[name] = json.dumps(value, ensure_ascii=False)

        # 手部骨骼 features 补充(unit 语义)
        if hand_rows:
            for side in ("left", "right"):
                key = f"observation.state.hand_{side}_{coordinate_key}"
                if key in features:
                    features[key].update({
                        "unit": hand_3d_unit or (
                            "mediapipe_world_relative" if coordinate_key == "3d"
                            else "normalized_image_coords"),
                        "note": (
                            "深度图抬升相机系 3D 世界坐标(米)"
                            if coordinate_key == "world" else
                            "MediaPipe 21 关键点相对 3D"
                            if coordinate_key == "3d" else
                            "MediaPipe 21 关键点归一化图像坐标"),
                    })

        meta.attrs["format"] = "egodata-hdf5"
        meta.attrs["version"] = "1.0"
        meta.attrs["total_episodes"] = len(episodes)
        meta.attrs["total_frames"] = total_frames
        meta.attrs["fps"] = fps_first
        meta.attrs["features"] = json.dumps(features, ensure_ascii=False)
        meta.attrs["calibration_root"] = "calibration/"

        # 内嵌说明文档(结构树/内容表/读取示例)—— 随 h5 交付,
        # 客户无需额外文档即可理解数据。在 f 关闭前生成(需要实际形状)。
        f.create_dataset("meta/README", data=_build_readme(dataset_name, f))

    return out_path


def _tree_lines(entries: list[tuple[list[str], str]], prefix: str = "") -> list[str]:
    """把 (路径段, 描述文本) 渲染为 ASCII 树行(递归,按段分组)。"""
    groups: dict[str, list[tuple[list[str], str]]] = {}
    for parts, note in entries:
        groups.setdefault(parts[0], []).append((parts[1:], note))

    lines: list[str] = []
    keys = sorted(groups)
    for i, key in enumerate(keys):
        sub = groups[key]
        is_last = i == len(keys) - 1
        branch = "└─ " if is_last else "├─ "
        if sub and sub[0][0]:  # 有子项 → 目录
            lines.append(f"{prefix}{branch}{key}/")
            lines.extend(_tree_lines(sub, prefix + ("   " if is_last else "│  ")))
        else:
            lines.append(f"{prefix}{branch}{key:<34} {sub[0][1]}")
    return lines


def _build_readme(dataset_name: str, h5: Any) -> str:
    """从 h5 文件实际结构生成内嵌说明(中文):结构树/内容表/读取示例。

    直接遍历 dataset(真实 shape),不用 features 语义表(那里 shape 是
    LeRobot 语义,如 [1] 表示每帧 1 值,对客户不直观)。
    """
    import h5py as _h5py

    # —— 结构树(遍历实际 dataset,跳过 meta/ 与 README 自身)——
    entries: list[tuple[list[str], str]] = []

    def visit(name: str, obj: Any) -> None:
        if not isinstance(obj, _h5py.Dataset):
            return
        parts = name.split("/")
        if parts[0] == "meta":
            return
        shape = "(" + ", ".join(str(s) for s in obj.shape) + ")"
        note = f"{shape:<28} {str(obj.dtype)}"
        attrs = dict(obj.attrs)
        if attrs.get("format") == "mp4":
            note += f" · MP4({attrs.get('frames')}帧,可提取为视频)"
        entries.append((parts, note))

    h5.visititems(visit)
    root = "episode_000000"
    root_items = [e for e in entries if e[0][0] == root]
    others = [e for e in entries if e[0][0] != root]
    tree: list[str] = [f"{root}/"]
    tree.extend(_tree_lines([(parts[1:], note) for parts, note in root_items], ""))
    for parts, note in others:
        tree.append(f"{'/'.join(parts)}    {note}")
    tree_str = "\n".join(tree)

    # —— 内容表(按实际结构判断)——
    all_names = {"/".join(e[0]) for e in entries}
    rows: list[str] = []
    img_keys = sorted(n for n in all_names
                      if n.startswith(f"{root}/observation/images/"))
    if img_keys:
        # 不写"全量" —— 视频可能被工作流的 include_video_keys 过滤过,而且
        # 深度流走 depth_sources、不经 camera_streams,本文件**从来不导**。
        # 声称"全量"会让客户以为拿到的是全部视频。
        rows.append(f"| **原视频** | {len(img_keys)} 路原始帧(逐帧 uint8 RGB)"
                    " | ✅ 已导出,清单见上方结构树 |")
    depth_keys = sorted(n for n in all_names
                        if n.startswith(f"{root}/observation/depth/"))
    if depth_keys:
        rows.append(f"| **深度图** | {len(depth_keys)} 路,uint16 **毫米**"
                    " | ✅ 0 = 无效像素(未测到),不是 0 毫米 |")
    hand_coordinate = next(
        (coord for coord in ("world", "3d", "2d")
         if f"{root}/observation/state/hand_left_{coord}" in all_names),
        None,
    )
    if hand_coordinate:
        description = {
            "world": "21 个关键点 × X/Y/Z 深度相机米制坐标",
            "3d": "21 个关键点 × X/Y/Z MediaPipe 相对 3D 坐标",
            "2d": "21 个关键点 × X/Y/Z 归一化图像坐标",
        }[hand_coordinate]
        rows.append(
            f"| **手部 {hand_coordinate.upper()} 骨骼** | {description}"
            " | ✅ 见有效帧标记列 |")
    skel = f"{root}/videos/hand_skeleton"
    if skel in all_names:
        skel_attrs = h5[skel].attrs
        rows.append(
            f"| **手部骨骼渲染视频** | `videos/hand_skeleton`:MP4 原始字节"
            f"({skel_attrs['width']}×{skel_attrs['height']} 左右目并排,骨架叠加) | "
            f"✅ {skel_attrs['frames']} 帧,{skel_attrs['fps']} fps |")
    if any(n.startswith(f"{root}/observation/tactile/") for n in all_names):
        rows.append("| **手套压力** | 16×16 压力阵列(SenseGlove) | ✅ 全帧 |")
    if f"{root}/observation/imu" in all_names:
        rows.append("| **IMU** | 6 轴惯性数据,与视频帧对齐 | ✅ 变长聚合,"
                    "配 `imu_frame_index` 切片还原 |")
    if f"{root}/observation/timestamps" in all_names:
        rows.append("| **帧时间戳** | 采集端每帧 2 条(左右目):"
                    "`timestamp` / `wall_time` / `hardware_ns` 三路时钟 |"
                    "✅ 用于帧级时间对齐分析 |")
    content_table = "\n".join(rows)

    # 提取视频章节仅在文件确实含 videos/hand_skeleton 时输出
    video_extract = ""
    if skel in all_names:
        video_extract = """
### 提取手部骨骼渲染视频(MP4)

`videos/hand_skeleton` 存的是 MP4 文件的**原始字节**,提取后即为标准 MP4 文件:

```python
import h5py

with h5py.File("dataset.h5", "r") as h5:
    mp4_bytes = h5["episode_000000/videos/hand_skeleton"][...].tobytes()

with open("hand_skeleton.mp4", "wb") as f:
    f.write(mp4_bytes)
```
"""

    # 读取示例:只给文件里**确实存在**的路径。原来是左目写死 stereo_left、
    # 手部无数据时仍回落到 hand_right_3d —— 客户照抄都是 KeyError;而且
    # `import h5py` 在 0 列、`with` 却缩进了 4 格,示例本身就不是合法 Python。
    first_img = img_keys[0].split("/")[-1] if img_keys else None
    example = ["import h5py", "", 'with h5py.File("dataset.h5", "r") as h5:']
    if first_img:
        example.append(f"    # 读一帧图像({first_img})")
        example.append(f'    img = h5["{root}/observation/images/{first_img}"][0]')
    if depth_keys:
        first_depth = depth_keys[0].split("/")[-1]
        example.append(f"    # 读一帧深度({first_depth};uint16 毫米,0 = 无效像素)")
        example.append(f'    depth = h5["{root}/observation/depth/{first_depth}"][0]')
    if hand_coordinate:
        example.append("    # 读第一帧右手骨骼(字段可能是 world / 3d / 2d)")
        example.append(f'    hand3d = h5["{root}/observation/state/'
                       f'hand_right_{hand_coordinate}"][0]')
    if f"{root}/observation/annotation" in all_names:
        example.append("    # 读全部标注")
        example.append(f'    labels = h5["{root}/observation/annotation"][:]')
    read_example = "\n".join(example)

    readme = f"""# {dataset_name} 数据文件说明(HDF5 格式)

> 本文件由 EgoData 数据采集系统工作流自动生成
> 帧数/fps 与内容以 meta/info attrs(features)为准

---

## 一、文件是什么

本文件是 **HDF5 格式(Hierarchical Data Format 5)** 的单个数据容器——
一个文件内以"文件夹树"结构同时存放视频、传感器数据、手部 3D 骨骼、
骨骼渲染视频、标注等多种数据,便于传输、归档和程序化读取。

一个批次(一次数据采集片段)= 文件中的一个 episode。

## 二、文件内部结构

```
{tree_str}
```

## 三、核心内容一览

| 内容 | 说明 | 状态 |
|---|---|---|
{content_table}

## 四、如何读取

```python
{read_example}
```
{video_extract}
其他工具:HDFView、Panoply、MATLAB(h5read)等支持 HDF5 的软件均可打开。

---

*本说明随数据一同交付。*
"""
    return readme