feat: import_lerobot に LeRobotConverter を追加（取り込みのカスタマイズ） by rikunosuke · Pull Request #285 · fastlabel/fastlabel-python-sdk

rikunosuke · 2026-07-24T08:08:03Z

概要

Client.import_lerobot（LeRobot v3 → FastLabel 取り込み）に、カスタマイズ点をフックとして開放する LeRobotConverter を追加しました。全体フロー（episode ループ・タスク作成・アップロード・後処理）は Client が保持し、ユーザーは必要な hook だけを override します。

from fastlabel import Client
from fastlabel.lerobot import LeRobotConverter

class JointsOnlyConverter(LeRobotConverter):
    CAMERA_KEYS = ("cam_high", "cam_low", "cam_left_wrist", "cam_right_wrist")

    def select_observation_state_names(self, names):        # 関節値(.pos)だけ残す
        return [n for n in names if n.endswith(".pos")]

    def build_telemetry_frame(self, frame):                 # gripper を追加
        t = super().build_telemetry_frame(frame)
        t["gripper"] = self.build_action(frame)[-1]
        return t

    def build_task_kwargs(self, *, episode_index, episode_name, frames):
        return {"tags": ["lerobot"], "metadatas": [{"key": "n", "value": str(len(frames))}]}

client = Client()
client.import_lerobot(project="slug", lerobot_data_path="/path", converter=JointsOnlyConverter)

追加した hook（`LeRobotConverter`）

hook	役割
`OBSERVATION_STATE_NAMES` / `ACTION_NAMES` / `CAMERA_KEYS`（class var, tuple）	宣言的な選別
`select_observation_state_names` / `select_action_names`	残す feature 名を動的選別（name→index は SDK が `__init__` で1回解決）
`build_observation_state` / `build_action` / `build_telemetry_frame`	telemetry の値・項目を構築
`select_cameras`	登録する動画を選別（`Camera`: `path`/`key`/`content_name`）
`build_task_kwargs`	`create_robotics_task` へ渡す `tags`/`metadatas` 等
`select_episodes`	episode 選別（`index→frame数` のみ受け取り、dataset version 非依存）
`build_episode_name`	タスク / 成果物の命名

converter はクラス/ファクトリで渡し、SDK が meta/info.json 付きで生成します（デフォルトは LeRobotConverter＝全 feature・全カメラ・全 episode を取り込み）。

設計上のポイント

疎結合: converter(ポリシー, version 非依存) / v3(version 固有のデータアクセス) / common(version 非依存の共有カーネル) / Client(オーケストレーション) を分離。converter は v3 に依存しません。
名前ベースの選別: meta/info.json の feature 名で選別できるため、巨大な observation.state（例 12014 次元）を関節値のみへ、カメラを主要4つのみへ、といった絞り込みが可能。
一時ファイル: with tempfile.TemporaryDirectory で管理。ZIP はタスク作成前に生成し orphan task を回避。
エラー方針: 想定内 (FastLabelException) は per-episode 記録して継続、想定外（接続エラー等）は伝播。

互換性・破壊的変更

シグネチャ互換: import_lerobot(project, lerobot_data_path, episode_indices=...) は従来通り呼び出せます（converter 省略時は全 feature・全カメラ・全 episode）。
破壊的変更: meta/info.json が必須になりました。 telemetry の値は feature 名で選別するため、features["observation.state"]["names"] と features["action"]["names"] を持たないデータセットは、converter 構築時（episode ループ開始前）に FastLabelInvalidException で失敗します。デフォルトの LeRobotConverter でも同様です。旧実装は parquet の列のみからテレメトリを生成していたため、info.json の無いデータセットはこれまで通っていました。LeRobot v3 は通常 meta/info.json を持つため、必須化する方針としています（README / docstring も同様に記載）。
破壊的変更: fastlabel.lerobot.format_episode_name を削除（命名は LeRobotConverter.build_episode_name に統合）。

テスト

tests/test_lerobot_converter.py（新規, 純ロジック）＋ tests/test_lerobot_v3_parquet.py 更新。
Python 3.10 で pytest tests/ → 72 passed、flake8 / black / isort 通過。

案件を想定した実装例

import math
import os
import re
from collections import defaultdict
from pathlib import Path
from typing import Any

import dotenv
from fastlabel import Client
from fastlabel.lerobot import LeRobotConverter

dotenv.load_dotenv()

os.environ["FASTLABEL_API_URL"] = "https://api.dev.fastlabel.ai/v1/"

BASE_DIR = Path(__file__).resolve().parent

# LeRobot データセットがあるディレクトリ
LEROBOT_DIR = BASE_DIR / "input" / "07222026_5_6"

# right_sensor_right_near_fx_0 系の触覚センサーチャンネル名
# （<side>_sensor_<sensor>_<axis>_<taxel index>、例: right_sensor_right_near_fx_0）
SENSOR_NAME_PATTERN = re.compile(r"^(\w+?_sensor_\w+?)_(fx|fy|fz|mx|my|mz)_(\d+)$")

# センサー名（sensor_index のキー）-> タスクの metadata キー
SENSOR_METADATA_KEYS = {
    "left_near": "sensor_channel_count_left_near",
    "left_far": "sensor_channel_count_left_far",
    "right_near": "sensor_channel_count_right_near",
    "right_far": "sensor_channel_count_right_far",
}


def has_value(state: list, index: int) -> bool:
    """observation.state の index に実データが入っているか（欠損・NaN は除外）。"""
    if index >= len(state):
        return False
    value = state[index]
    return value is not None and not math.isnan(value)


class Converter(LeRobotConverter):
    OBSERVATION_STATE_NAMES = (
        "left_joint_0.pos",
        "left_joint_1.pos",
        "left_joint_2.pos",
        "left_joint_3.pos",
        "left_joint_4.pos",
        "left_joint_5.pos",
        "left_left_carriage_joint.pos",
        "right_joint_0.pos",
        "right_joint_1.pos",
        "right_joint_2.pos",
        "right_joint_3.pos",
        "right_joint_4.pos",
        "right_joint_5.pos",
        "right_left_carriage_joint.pos",
    )

    def __init__(self, meta: dict[str, Any] | None = None) -> None:
        super().__init__(meta)
        # 特徴量名は「どのインデックスがどのセンサーか」の対応表としてのみ使う
        # 例) left_sensor_left_near_fx_0 -> sensor_index["left_near"] に index を追加
        all_names: list[str] = self.meta["features"]["observation.state"]["names"]
        self.sensor_index: dict[str, list[int]] = defaultdict(list)
        for index, name in enumerate(all_names):
            if not (matched := SENSOR_NAME_PATTERN.match(name)):
                continue
            sensor = matched.group(1).split("_sensor_")[-1]
            self.sensor_index[sensor].append(index)

    def build_task_kwargs(
        self,
        *,
        frames: list,
        **kwargs,
    ) -> dict[str, Any]:
        """実データに値が入っているチャンネル数をセンサーごとに数えて metadata に載せる。"""
        counts: dict[str, int] = {}
        for sensor, indexes in self.sensor_index.items():
            counts[sensor] = sum(
                1
                for index in indexes
                if any(has_value(frame["observation.state"], index) for frame in frames)
            )

        return {
            "metadatas": [
                {
                    "key": SENSOR_METADATA_KEYS.get(sensor, sensor),
                    "value": int(count),
                }
                for sensor, count in counts.items()
            ]
        }


if __name__ == "__main__":
    c = Client()
    c.import_lerobot("sample-test", str(LEROBOT_DIR), converter=Converter)

🤖 Generated with Claude Code

Client.import_lerobot に LeRobot v3 → FastLabel 取り込みのカスタマイズ点をフックとして開放する LeRobotConverter を追加。 - LeRobotConverter（新規, pure strategy）。SDK が meta/info.json 付きで __init__ 生成（クラス/ファクトリを渡す）。全体フロー（ループ・タスク作成・アップロード・後処理）は Client が保持し、converter は hook のみ: - select_observation_state_names / select_action_names: 残す feature 名を選ぶ（name→index 変換は SDK が __init__ で1回だけ解決） - build_observation_state / build_action / build_telemetry_frame: telemetry の値・項目（gripper 等）を構築 - select_cameras: 登録動画を選別（Camera: path/key/content_name） - build_task_kwargs: create_robotics_task へ渡す tags/metadatas 等 - select_episodes: episode 選別（index→frame数のみ, v3 非依存） - build_episode_name: タスク/成果物の命名 - import_lerobot(converter=LeRobotConverter)。一時ディレクトリは with tempfile.TemporaryDirectory で管理。ZIP はタスク作成前に生成し orphan task を回避。想定内 (FastLabelException) は per-episode 記録、想定外は伝播。 - レイヤ疎結合化: converter は v3 非依存 / v3 は version 固有データアクセス / common は version 非依存の共有カーネル。info.json の feature 名で name ベース選別が可能（例: observation.state を ".pos" のみ、カメラを主要4つ）。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

rikunosuke · 2026-07-27T11:31:42Z

-def format_episode_name(episode_index: int) -> str:
-    return f"episode_{episode_index:06d}"


episode 名を変更できるように、converter に移動しました。

rikunosuke · 2026-07-27T11:32:17Z

-def get_camera_dirs(lerobot_data_path: Path) -> list:
-    """Get camera directories and their content names.
-    Returns [(camera_dir, content_name), ...].
-    e.g. observation.images.top -> content_name = "images_top"
-    """
-    videos_dir = lerobot_data_path / "videos"
-    if not videos_dir.exists():
-        return []
-
-    results = []
-    for obs_dir in sorted(videos_dir.iterdir()):
-        if not obs_dir.is_dir():
-            continue
-        parts = obs_dir.name.split(".")
-        if parts[0] != "observation":
-            raise FastLabelInvalidException(
-                f"Unexpected camera dir name: {obs_dir.name}"
-            )
-
-        content_name = "_".join(parts[1:])
-        results.append((obs_dir, content_name))
-    return results


camera のディレクトリの置き場所は v3 に依存していたので、v3 側に移動させました。

rikunosuke · 2026-07-27T11:33:34Z

+def load_info(lerobot_data_path: Path) -> dict[str, Any]:
+    """Load meta/info.json for a LeRobot dataset.
+
+    Returns the parsed dict, or {} when the file is absent (older datasets),
+    so name-based selection is opt-in and default behavior is unaffected.
+    """
+    info_path = lerobot_data_path / "meta" / "info.json"
+    if not info_path.exists():
+        return {}
+    return json.loads(info_path.read_text())


meta/info.json は v2, v3 共通で同じ箇所なため、common においています。

rikunosuke · 2026-07-27T11:44:01Z

v2 は実装はしていませんが、v2, v3 が実装された時の、バージョンを意識しなくていいように内部で分岐させるためのインターフェース群、として定義しています。

rikunosuke · 2026-07-27T11:45:03Z

+    # ---- declarative selection (static configuration) ----
+    OBSERVATION_STATE_NAMES: ClassVar[tuple[str, ...] | None] = None
+    ACTION_NAMES: ClassVar[tuple[str, ...] | None] = None
+    CAMERA_KEYS: ClassVar[tuple[str, ...] | None] = None


基本的にはクラス変数で静的に設定できるようにしています。大体はこの設定で observation.state, action, cameras を指定する想定です。

rikunosuke · 2026-07-27T11:45:38Z

+    def select_observation_state_names(self, names: list[str]) -> list[str]:
+        """Return the ``observation.state`` feature names to keep (default: all).
+
+        ``names`` is the full ordered name list from ``meta/info.json``. Override
+        for dynamic selection, e.g. ``[n for n in names if n.endswith(".pos")]``.
+        """
+        return (
+            list(self.OBSERVATION_STATE_NAMES)
+            if self.OBSERVATION_STATE_NAMES
+            else names
+        )


observation.state をロジックで指定したい場合のフックです。
regext などの可能性があると思っています。

rikunosuke · 2026-07-27T11:46:13Z

+    def select_action_names(self, names: list[str]) -> list[str]:
+        """Return the ``action`` feature names to keep (default: all)."""
+        return list(self.ACTION_NAMES) if self.ACTION_NAMES else names


observation.state と同様に、action も動的にできるようにしています。

rikunosuke · 2026-07-27T11:47:35Z

+    def build_observation_state(self, frame: Frame) -> list[Any]:
+        values = frame["observation.state"]
+        return [values[i] for i in self._state_index]


センサー名の list[str] から、実際のデータを取得する箇所です。

rikunosuke · 2026-07-27T11:48:14Z

+    def build_telemetry_frame(self, frame: Frame) -> dict[str, Any]:
+        """Build one telemetry frame written to the episode JSON.
+
+        Override and call ``super()`` to add extra items (e.g. gripper).
+        """
+        return {
+            "observation.state": self.build_observation_state(frame),
+            "action": self.build_action(frame),
+            "frame_index": int(frame["frame_index"]),
+            "timestamp": float(frame["timestamp"]),
+        }


この4つを暗黙的に登録していましたが、拡張できるようにしています。

rikunosuke · 2026-07-27T11:48:53Z

+    def select_cameras(self, cameras: list[Camera]) -> list[Camera]:
+        """Return the cameras to include (default: all — the given list).
+
+        Each ``Camera`` has ``path`` / ``key`` / ``content_name``; ``key`` matches
+        ``self.meta["features"][key]`` for looking up resolution etc. When
+        ``CAMERA_KEYS`` is set, keep cameras whose key suffix (e.g. ``cam_high``
+        of ``observation.images.cam_high``) is in the set.
+        """
+        if self.CAMERA_KEYS is None:
+            return cameras
+        return [
+            camera
+            for camera in cameras
+            if camera.key.split(".")[-1] in self.CAMERA_KEYS
+        ]


カメラも、不要なものは送らないようにすることが想定できるためフックを作っています。たとえば、センサー系のカメラは送らない、などを想定しています。

rikunosuke · 2026-07-27T11:49:19Z

+    def build_task_kwargs(
+        self,
+        *,
+        episode_index: int,
+        episode_name: str,
+        frames: list[Frame],
+    ) -> dict[str, Any]:
+        """Return keyword args forwarded to ``create_robotics_task``.
+
+        Keyword-only:
+        episode_index is the episode index.
+        episode_name is the task name (e.g. ``episode_000001``).
+        frames is the list of raw native frames for the episode.
+        Dataset-level metadata is available via ``self.meta``.
+        e.g. return ``{"tags": [...], "metadatas": [...]}``.
+        """
+        return {}


FastLabel に送る tags, metadatas を動的に設定できるフックです。

yo-tak · 2026-07-27T15:24:36Z

+        if parts[0] != "observation":
+            raise FastLabelInvalidException(
+                f"Unexpected camera dir name: {obs_dir.name}"
+            )


多分422の指定忘れ

yo-tak · 2026-07-27T16:17:07Z

+        self._state_index: list[int] = self._names_to_index(
+            "observation.state", self.select_observation_state_names
+        )
+        self._action_index: list[int] = self._names_to_index(
+            "action", self.select_action_names
+        )


[要確認] デフォルト converter が info.json（feature names）を必須化しており、「後方互換：従来通り動作」と食い違います。

__init__ は override の有無に関わらず observation.state と action の両方で _names_to_index を呼ぶため、デフォルトの LeRobotConverter（hook 無し）でも meta["features"]["observation.state"]["names"] / ["action"]["names"] が必須になります（無い場合は L82 で FastLabelInvalidException）。

Client.import_lerobot はこれを episode ループ・try の外で構築する（fastlabel/__init__.py:2286 の conv = converter(load_info(data_path))）ため、info.json が無い / names を欠くデータセットはループ開始前に import 全体が即失敗します。

旧 _convert_episode_frames は info.json を参照せず parquet 列だけからテレメトリを生成していたため、これまで info.json 無しで通っていたデータセットが通らなくなります（test_no_meta_raises、v3 fixture への info.json 追加、test_missing_required_columns_returns_empty 削除が裏付け）。

これは PR 本文「後方互換：従来通り動作（converter 省略時は全件）」および common.py の load_info docstring（"default behavior is unaffected"）と矛盾します。

対応案（どちらか）:

(a) 「info.json + feature names は今後必須」と PR 本文 / README / load_info docstring を修正

(b) meta が空のときはデフォルトで全列フォールバックし旧挙動を維持

LeRobot v3 は通常 info.json を持つので (a) で十分だと思いますが、意図の確認をお願いします。

yo-tak · 2026-07-27T16:17:15Z

+    def build_observation_state(self, frame: Frame) -> list[Any]:
+        values = frame["observation.state"]
+        return [values[i] for i in self._state_index]
+
+    def build_action(self, frame: Frame) -> list[Any]:
+        values = frame["action"]
+        return [values[i] for i in self._action_index]


[Medium] 想定外例外（非 FastLabelException）が1エピソードで import 全体を中断させ、旧のグレースフル劣化が失われています。

build_observation_state（L96）/ build_action（L100）はフレームの列を直接参照するため、フレームに当該列が無い、または宣言 names より配列が短い場合に KeyError / IndexError が発生します。

これらは FastLabelException のサブクラスではないため、Client.import_lerobot の per-episode ハンドラ except FastLabelException（fastlabel/__init__.py:2337）で捕捉されず、1エピソードの不整合で残りのエピソードごと import 全体が中断します。

旧 _convert_episode_frames は必須列が無ければ [] を返して per-episode で継続していたため、堅牢性が後退しています。

対応案: per-episode ループで想定外例外も広めに捕捉して記録・継続する、あるいは fail-fast を意図するなら docstring にその旨を明記してください。

rikunosuke commented Jul 27, 2026

View reviewed changes

rikunosuke requested a review from yo-tak July 27, 2026 11:56

yo-tak reviewed Jul 28, 2026

View reviewed changes

		def format_episode_name(episode_index: int) -> str:
		return f"episode_{episode_index:06d}"

Uh oh!

Conversation

rikunosuke commented Jul 24, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

概要

追加した hook（LeRobotConverter）

設計上のポイント

互換性・破壊的変更

テスト

案件を想定した実装例

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

rikunosuke commented Jul 24, 2026 •

edited

Loading

追加した hook（`LeRobotConverter`）