diff --git a/README.md b/README.md index 1dd8eea..77aad11 100644 --- a/README.md +++ b/README.md @@ -20,14 +20,15 @@ A curated collection of utilities for [LeRobot Projects](https://github.com/hugg ## πŸ“£ What's New +- **\[2025.10.04\]** We have collected and updated all Dataset Version Conversion Scripts for LeRobot! πŸ”₯πŸ”₯πŸ”₯ - **\[2025.09.28\]** We have upgraded LeRobotDataset from v2.1 to v3.0! πŸ”₯πŸ”₯πŸ”₯ - **\[2025.06.27\]** We have supported Data Conversion from LIBERO to LeRobot! πŸ”₯πŸ”₯πŸ”₯ - **\[2025.05.16\]** We have supported Data Conversion from LeRobot to RLDS! πŸ”₯πŸ”₯πŸ”₯ - **\[2025.05.12\]** We have supported Data Conversion from RoboMIND to LeRobot! πŸ”₯πŸ”₯πŸ”₯ -- **\[2025.04.15\]** We add Dataset Merging Tool for merging multi-source lerobot datasets! πŸ”₯πŸ”₯πŸ”₯
More News +- **\[2025.04.15\]** We add Dataset Merging Tool for merging multi-source lerobot datasets! πŸ”₯πŸ”₯πŸ”₯ - **\[2025.04.14\]** We have supported Data Conversion from AgiBotWorld to LeRobot! πŸ”₯πŸ”₯πŸ”₯ - **\[2025.04.11\]** We change the repo from `openx2lerobot` to `any4lerobot`, making a ​​universal toolbox for LeRobot​​! πŸ”₯πŸ”₯πŸ”₯ - **\[2025.02.19\]** We have supported Data Conversion from Open X-Embodiment to LeRobot! πŸ”₯πŸ”₯πŸ”₯ @@ -53,10 +54,13 @@ A curated collection of utilities for [LeRobot Projects](https://github.com/hugg - [ ] Dataset Filtering - [ ] Dataset Sampling -- ​**Version Conversion​**​: +- ​[**Version Conversion​**​](./ds_version_convert/README.md): - - [x] [LeRobotv2.0 to LeRobotv2.1](./ds_version_convert/README.md) - - [ ] LeRobotv2.1 to LeRobotv2.0 + - [x] [LeRobotv1.6 to LeRobotv2.0](./ds_version_convert/v16_to_v20/README.md) + - [x] [LeRobotv2.0 to LeRobotv2.1](./ds_version_convert/v20_to_v21/README.md) + - [x] [LeRobotv2.1 to LeRobotv2.0](./ds_version_convert/v21_to_v20/README.md) + - [x] [LeRobotv2.1 to LeRobotv3.0](./ds_version_convert/v21_to_v30/README.md) + - [ ] LeRobotv3.0 to LeRobotv2.1 - [**Want more features?**](https://github.com/Tavish9/any4lerobot/issues/new?template=feature-request.yml) diff --git a/ds_version_convert/README.md b/ds_version_convert/README.md index 1e3f81f..fa814eb 100644 --- a/ds_version_convert/README.md +++ b/ds_version_convert/README.md @@ -1,117 +1,10 @@ -# What's New in This Version Converter Script +# LeRobot Dataset Version Convert -> [!IMPORTANT] -> -> This is not a universally applicable method, so we decided not to save it as an executable python script, but to write it in a tutorial for reference by those who need it. -> -> If you are using `libx264` encoding and want to use `decord` as the video backend to speed up the stats conversion, or want to use the process pool to speed up the conversion when converting the huge dataset like droid, you can use this script. -> -> However, please note that the droid dataset may get stuck at episode 5545 during the conversion process. - -Key improvements: - -- support loading the local dataset -- support use decord as video backend (NOTICE: decord is not supported to 'libsvtav1' encode method, we test it using 'libx264', ref: https://github.com/dmlc/decord/issues/319) -- support process pool for huge dataset like droid to accelerate conversation speed - -# 1. Convert LeRobot Dataset v20 to v21 Utils - -## Installation - -Install decord: https://github.com/dmlc/decord - - -## Default usage - -This equal to lerobot projects, it will use dataset from huggingface hub, delete `stats.json` and push to huggingface hub (multi-thread and `pyav` as video backend), you can: - -```bash -python ds_version_convert/convert_dataset_v20_to_v21.py \ - --repo-id=aliberts/koch_tutorial \ - --delete-old-stats \ - --push-to-hub \ - --num-workers=8 \ - --video-backend=pyav -``` - - - -## Using `decord` as video backend - -> [!IMPORTANT] -> -> 1.We recommend use default method to convert stats and use decord and process pool if you want to convert huge dataset like droid. -> -> 2.If you want to use decord as video backend, you should modify the `video_utils.py` source code from lerobot. - -```python -def decode_video_frames( - video_path: Path | str, - timestamps: list[float], - tolerance_s: float, - backend: str | None = None, -) -> torch.Tensor: - """ - Decodes video frames using the specified backend. - - Args: - video_path (Path): Path to the video file. - timestamps (list[float]): List of timestamps to extract frames. - tolerance_s (float): Allowed deviation in seconds for frame retrieval. - backend (str, optional): Backend to use for decoding. Defaults to "torchcodec" when available in the platform; otherwise, defaults to "pyav".. - - Returns: - torch.Tensor: Decoded frames. - - Currently supports torchcodec on cpu and pyav. - """ - if backend is None: - backend = get_safe_default_codec() - if backend == "torchcodec": - return decode_video_frames_torchcodec(video_path, timestamps, tolerance_s) - elif backend in ["pyav", "video_reader"]: - return decode_video_frames_torchvision(video_path, timestamps, tolerance_s, backend) - elif backend == "decord": - return decode_video_frames_decord(video_path, timestamps) - else: - raise ValueError(f"Unsupported video backend: {backend}") - - -def decode_video_frames_decord( - video_path: Path | str, - timestamps: list[float], -) -> torch.Tensor: - video_path = str(video_path) - vr = decord.VideoReader(video_path) - num_frames = len(vr) - frame_ts: np.ndarray = vr.get_frame_timestamp(range(num_frames)) - indices = np.abs(frame_ts[:, :1] - timestamps).argmin(axis=0) - frames = vr.get_batch(indices) - - frames_tensor = torch.tensor(frames.asnumpy()).type(torch.float32).permute(0, 3, 1, 2) / 255 - return frames_tensor -``` - -This will load local dataset, use `decord` as video backend and process pool, you can: - -```bash -python utils/version_convert/convert_dataset_v20_to_v21.py \ - --repo-id=aliberts/koch_tutorial \ - --root=/home/path/to/your/lerobot/dataset/path \ - --num-workers=8 \ - --video-backend=decord \ - --use-process-pool - -``` - -## Speed Test - -Table I. dataset conversation time use stats. - -| dataset | episodes | video_backend | method | workers | video_encode | Time | -| -------------------- | -------- | ------------- | ------- | ------- | ------------ | ----- | -| bekerley_autolab_ur5 | 896 | pyav | thread | 16 | libx264 | 10:56 | -| bekerley_autolab_ur5 | 896 | pyav | process | 16 | libx264 | -- | -| bekerley_autolab_ur5 | 896 | decord | thread | 16 | libx264 | 11:44 | -| bekerley_autolab_ur5 | 896 | decord | process | 16 | libx264 | 14:26 | +The LeRobot Dataset has undergone multiple versions over time, with significant improvements in data storage and reading performance in each iteration. The versions of the dataset are as follows: v1.0-v1.6, v2.0, v2.1, v3.0. +| **Version** | **Release Date** | **Version Conversion Link** | +| ------------------------------------------------------------ | ---------------- | ------------------------------------------------------------------------------ | +| [v1.0-v1.6](https://github.com/huggingface/lerobot/pull/302) | 2024-07-23 | --- | +| [v2.0](https://github.com/huggingface/lerobot/pull/461) | 2024-11-30 | [v1.6 to v2.0](./v16_to_v20/README.md), [v2.1 to v2.0](./v21_to_v20/README.md) | +| [v2.1](https://github.com/huggingface/lerobot/pull/711) | 2025-02-25 | [v2.0 to v2.1](./v20_to_v21/README.md) | +| [v3.0](https://github.com/huggingface/lerobot/pull/1412) | 2025-09-15 | [v2.1 to v3.0](./v21_to_v30/README.md) | diff --git a/ds_version_convert/convert_dataset_v20_to_v21.py b/ds_version_convert/convert_dataset_v20_to_v21.py deleted file mode 100644 index 497fe92..0000000 --- a/ds_version_convert/convert_dataset_v20_to_v21.py +++ /dev/null @@ -1,176 +0,0 @@ -# Copyright 2024 The HuggingFace Inc. team. All rights reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - - -import argparse -from concurrent.futures import ProcessPoolExecutor, as_completed -from multiprocessing import cpu_count - -import numpy as np -from huggingface_hub import HfApi -from lerobot.datasets.compute_stats import get_feature_stats -from lerobot.datasets.lerobot_dataset import LeRobotDataset -from lerobot.datasets.utils import EPISODES_STATS_PATH, STATS_PATH, load_stats, write_episode_stats, write_info -from lerobot.datasets.v21.convert_dataset_v20_to_v21 import V20, V21, SuppressWarnings -from lerobot.datasets.v21.convert_stats import check_aggregate_stats, convert_stats, sample_episode_video_frames -from tqdm import tqdm - - -def convert_episode_stats(dataset: LeRobotDataset, ep_idx: int, is_parallel: bool = False): - ep_start_idx = dataset.episode_data_index["from"][ep_idx] - ep_end_idx = dataset.episode_data_index["to"][ep_idx] - ep_data = dataset.hf_dataset.select(range(ep_start_idx, ep_end_idx)) - - ep_stats = {} - for key, ft in dataset.features.items(): - if ft["dtype"] == "video": - # We sample only for videos - ep_ft_data = sample_episode_video_frames(dataset, ep_idx, key) - ep_ft_data = ep_ft_data[None, ...] if ep_ft_data.ndim == 3 else ep_ft_data - else: - ep_ft_data = np.array(ep_data[key]) - - axes_to_reduce = (0, 2, 3) if ft["dtype"] in ["image", "video"] else 0 - keepdims = True if ft["dtype"] in ["image", "video"] else ep_ft_data.ndim == 1 - ep_stats[key] = get_feature_stats(ep_ft_data, axis=axes_to_reduce, keepdims=keepdims) - - if ft["dtype"] in ["image", "video"]: # remove batch dim - ep_stats[key] = {k: v if k == "count" else np.squeeze(v, axis=0) for k, v in ep_stats[key].items()} - - if not is_parallel: - dataset.meta.episodes_stats[ep_idx] = ep_stats - - return ep_stats, ep_idx - - -def convert_stats_by_process_pool(dataset: LeRobotDataset, num_workers: int = 0): - """Convert stats in parallel using multiple process.""" - assert dataset.episodes is None - - total_episodes = dataset.meta.total_episodes - futures = [] - - if num_workers > 0: - max_workers = min(cpu_count() - 1, num_workers) - with ProcessPoolExecutor(max_workers=max_workers) as executor: - for ep_idx in range(total_episodes): - futures.append(executor.submit(convert_episode_stats, dataset, ep_idx, True)) - for future in tqdm(as_completed(futures), total=total_episodes, desc="Converting episodes stats"): - ep_stats, ep_idx = future.result() - dataset.meta.episodes_stats[ep_idx] = ep_stats - else: - for ep_idx in tqdm(range(total_episodes)): - convert_episode_stats(dataset, ep_idx) - - for ep_idx in tqdm(range(total_episodes)): - write_episode_stats(ep_idx, dataset.meta.episodes_stats[ep_idx], dataset.root) - - -def convert_dataset( - repo_id: str, - root: str | None = None, - push_to_hub: bool = False, - delete_old_stats: bool = False, - branch: str | None = None, - num_workers: int = 4, - video_backend: str = "pyav", - use_process_pool: bool = True, -): - with SuppressWarnings(): - if root is not None: - dataset = LeRobotDataset(repo_id, root, revision=V20, video_backend=video_backend) - else: - dataset = LeRobotDataset(repo_id, revision=V20, force_cache_sync=True, video_backend=video_backend) - - if (dataset.root / EPISODES_STATS_PATH).is_file(): - (dataset.root / EPISODES_STATS_PATH).unlink() - - if use_process_pool: - convert_stats_by_process_pool(dataset, num_workers=num_workers) - else: - convert_stats(dataset, num_workers=num_workers) - ref_stats = load_stats(dataset.root) - check_aggregate_stats(dataset, ref_stats) - - dataset.meta.info["codebase_version"] = V21 - write_info(dataset.meta.info, dataset.root) - - if push_to_hub: - dataset.push_to_hub(branch=branch, tag_version=False, allow_patterns="meta/") - - # delete old stats.json file - if delete_old_stats and (dataset.root / STATS_PATH).is_file: - (dataset.root / STATS_PATH).unlink() - - hub_api = HfApi() - if delete_old_stats and hub_api.file_exists( - repo_id=dataset.repo_id, filename=STATS_PATH, revision=branch, repo_type="dataset" - ): - hub_api.delete_file(path_in_repo=STATS_PATH, repo_id=dataset.repo_id, revision=branch, repo_type="dataset") - if push_to_hub: - hub_api.create_tag(repo_id, tag=V21, revision=branch, repo_type="dataset") - - -if __name__ == "__main__": - parser = argparse.ArgumentParser() - parser.add_argument( - "--repo-id", - type=str, - required=True, - help="Repository identifier on Hugging Face: a community or a user name `/` the name of the dataset " - "(e.g. `lerobot/pusht`, `cadene/aloha_sim_insertion_human`).", - ) - parser.add_argument( - "--root", - type=str, - default=None, - help="Path to the local dataset root directory. If not provided, the script will use the dataset from local.", - ) - parser.add_argument( - "--push-to-hub", - action="store_true", - help="Push the dataset to the hub after conversion. Defaults to False.", - ) - parser.add_argument( - "--delete-old-stats", - action="store_true", - help="Delete the old stats.json file after conversion. Defaults to False.", - ) - parser.add_argument( - "--branch", - type=str, - default=None, - help="Repo branch to push your dataset. Defaults to the main branch.", - ) - parser.add_argument( - "--num-workers", - type=int, - default=4, - help="Number of workers for parallelizing stats compute. Defaults to 4.", - ) - parser.add_argument( - "--video-backend", - type=str, - default="pyav", - choices=["pyav", "decord"], - help="Video backend to use. Defaults to pyav.", - ) - parser.add_argument( - "--use-process-pool", - action="store_true", - help="Use process pool for parallelizing stats compute. Defaults to False.", - ) - - args = parser.parse_args() - convert_dataset(**vars(args)) \ No newline at end of file diff --git a/ds_version_convert/v16_to_v20/README.md b/ds_version_convert/v16_to_v20/README.md new file mode 100644 index 0000000..c91ec38 --- /dev/null +++ b/ds_version_convert/v16_to_v20/README.md @@ -0,0 +1,19 @@ +# LeRobot Dataset v16 to v20 + +## Get started + +1. Install v2.0 lerobot + ```bash + git clone https://github.com/huggingface/lerobot.git + git checkout c574eb49845d48f5aad532d823ef56aec1c0d0f2 + pip install -e . + ``` + +2. Run the converter: + ```bash + python convert_dataset_v16_to_v20.py \ + --repo-id=your_id \ + --single-task=task_desc \ + --tasks-col=task_column_name \ + --tasks-path=path_to_json \ + ``` \ No newline at end of file diff --git a/ds_version_convert/v16_to_v20/convert_dataset_v16_to_v20.py b/ds_version_convert/v16_to_v20/convert_dataset_v16_to_v20.py new file mode 100644 index 0000000..245d6b4 --- /dev/null +++ b/ds_version_convert/v16_to_v20/convert_dataset_v16_to_v20.py @@ -0,0 +1,547 @@ +import argparse +import contextlib +import filecmp +import json +import logging +import math +import shutil +import subprocess +import tempfile +from pathlib import Path + +import datasets +import pyarrow.compute as pc +import pyarrow.parquet as pq +import torch +from datasets import Dataset +from huggingface_hub import HfApi +from huggingface_hub.errors import EntryNotFoundError, HfHubHTTPError +from lerobot.common.datasets.utils import ( + DEFAULT_CHUNK_SIZE, + DEFAULT_PARQUET_PATH, + DEFAULT_VIDEO_PATH, + EPISODES_PATH, + INFO_PATH, + STATS_PATH, + TASKS_PATH, + create_branch, + create_lerobot_dataset_card, + flatten_dict, + get_hub_safe_version, + load_json, + unflatten_dict, + write_json, + write_jsonlines, +) +from lerobot.common.datasets.video_utils import ( + VideoFrame, # noqa: F401 + get_image_pixel_channels, + get_video_info, +) +from lerobot.common.robot_devices.robots.configs import RobotConfig +from lerobot.common.robot_devices.robots.utils import make_robot_config +from safetensors.torch import load_file + +V16 = "v1.6" +V20 = "v2.0" + +GITATTRIBUTES_REF = "aliberts/gitattributes_reference" +V1_VIDEO_FILE = "{video_key}_episode_{episode_index:06d}.mp4" +V1_INFO_PATH = "meta_data/info.json" +V1_STATS_PATH = "meta_data/stats.safetensors" + + +def parse_robot_config(robot_cfg: RobotConfig) -> tuple[str, dict]: + if robot_cfg.type in ["aloha", "koch"]: + state_names = [ + f"{arm}_{motor}" if len(robot_cfg.follower_arms) > 1 else motor + for arm in robot_cfg.follower_arms + for motor in robot_cfg.follower_arms[arm].motors + ] + action_names = [ + # f"{arm}_{motor}" for arm in ["left", "right"] for motor in robot_cfg["leader_arms"][arm]["motors"] + f"{arm}_{motor}" if len(robot_cfg.leader_arms) > 1 else motor + for arm in robot_cfg.leader_arms + for motor in robot_cfg.leader_arms[arm].motors + ] + # elif robot_cfg["robot_type"] == "stretch3": TODO + else: + raise NotImplementedError( + "Please provide robot_config={'robot_type': ..., 'names': ...} directly to convert_dataset()." + ) + + return { + "robot_type": robot_cfg.type, + "names": { + "observation.state": state_names, + "observation.effort": state_names, + "action": action_names, + }, + } + + +def convert_stats_to_json(v1_dir: Path, v2_dir: Path) -> None: + safetensor_path = v1_dir / V1_STATS_PATH + stats = load_file(safetensor_path) + serialized_stats = {key: value.tolist() for key, value in stats.items()} + serialized_stats = unflatten_dict(serialized_stats) + + json_path = v2_dir / STATS_PATH + json_path.parent.mkdir(exist_ok=True, parents=True) + with open(json_path, "w") as f: + json.dump(serialized_stats, f, indent=4) + + # Sanity check + with open(json_path) as f: + stats_json = json.load(f) + + stats_json = flatten_dict(stats_json) + stats_json = {key: torch.tensor(value) for key, value in stats_json.items()} + for key in stats: + torch.testing.assert_close(stats_json[key], stats[key]) + + +def get_features_from_hf_dataset(dataset: Dataset, robot_config: RobotConfig | None = None) -> dict[str, list]: + robot_config = parse_robot_config(robot_config) + features = {} + for key, ft in dataset.features.items(): + if isinstance(ft, datasets.Value): + dtype = ft.dtype + shape = (1,) + names = None + if isinstance(ft, datasets.Sequence): + assert isinstance(ft.feature, datasets.Value) + dtype = ft.feature.dtype + shape = (ft.length,) + motor_names = robot_config["names"][key] if robot_config else [f"motor_{i}" for i in range(ft.length)] + assert len(motor_names) == shape[0] + names = {"motors": motor_names} + elif isinstance(ft, datasets.Image): + dtype = "image" + image = dataset[0][key] # Assuming first row + channels = get_image_pixel_channels(image) + shape = (image.height, image.width, channels) + names = ["height", "width", "channels"] + elif ft._type == "VideoFrame": + dtype = "video" + shape = None # Add shape later + names = ["height", "width", "channels"] + + features[key] = { + "dtype": dtype, + "shape": shape, + "names": names, + } + + return features + + +def add_task_index_by_episodes(dataset: Dataset, tasks_by_episodes: dict) -> tuple[Dataset, list[str]]: + df = dataset.to_pandas() + tasks = list(set(tasks_by_episodes.values())) + tasks_to_task_index = {task: task_idx for task_idx, task in enumerate(tasks)} + episodes_to_task_index = {ep_idx: tasks_to_task_index[task] for ep_idx, task in tasks_by_episodes.items()} + df["task_index"] = df["episode_index"].map(episodes_to_task_index).astype(int) + + features = dataset.features + features["task_index"] = datasets.Value(dtype="int64") + dataset = Dataset.from_pandas(df, features=features, split="train") + return dataset, tasks + + +def add_task_index_from_tasks_col(dataset: Dataset, tasks_col: str) -> tuple[Dataset, dict[str, list[str]], list[str]]: + df = dataset.to_pandas() + + # HACK: This is to clean some of the instructions in our version of Open X datasets + prefix_to_clean = "tf.Tensor(b'" + suffix_to_clean = "', shape=(), dtype=string)" + df[tasks_col] = df[tasks_col].str.removeprefix(prefix_to_clean).str.removesuffix(suffix_to_clean) + + # Create task_index col + tasks_by_episode = df.groupby("episode_index")[tasks_col].unique().apply(lambda x: x.tolist()).to_dict() + tasks = df[tasks_col].unique().tolist() + tasks_to_task_index = {task: idx for idx, task in enumerate(tasks)} + df["task_index"] = df[tasks_col].map(tasks_to_task_index).astype(int) + + # Build the dataset back from df + features = dataset.features + features["task_index"] = datasets.Value(dtype="int64") + dataset = Dataset.from_pandas(df, features=features, split="train") + dataset = dataset.remove_columns(tasks_col) + + return dataset, tasks, tasks_by_episode + + +def split_parquet_by_episodes( + dataset: Dataset, + total_episodes: int, + total_chunks: int, + output_dir: Path, +) -> list: + table = dataset.data.table + episode_lengths = [] + for ep_chunk in range(total_chunks): + ep_chunk_start = DEFAULT_CHUNK_SIZE * ep_chunk + ep_chunk_end = min(DEFAULT_CHUNK_SIZE * (ep_chunk + 1), total_episodes) + chunk_dir = "/".join(DEFAULT_PARQUET_PATH.split("/")[:-1]).format(episode_chunk=ep_chunk) + (output_dir / chunk_dir).mkdir(parents=True, exist_ok=True) + for ep_idx in range(ep_chunk_start, ep_chunk_end): + ep_table = table.filter(pc.equal(table["episode_index"], ep_idx)) + episode_lengths.insert(ep_idx, len(ep_table)) + output_file = output_dir / DEFAULT_PARQUET_PATH.format(episode_chunk=ep_chunk, episode_index=ep_idx) + pq.write_table(ep_table, output_file) + + return episode_lengths + + +def move_videos( + repo_id: str, + video_keys: list[str], + total_episodes: int, + total_chunks: int, + work_dir: Path, + clean_gittatributes: Path, + branch: str = "main", +) -> None: + """ + HACK: Since HfApi() doesn't provide a way to move files directly in a repo, this function will run git + commands to fetch git lfs video files references to move them into subdirectories without having to + actually download them. + """ + _lfs_clone(repo_id, work_dir, branch) + + videos_moved = False + video_files = [str(f.relative_to(work_dir)) for f in work_dir.glob("videos*/*.mp4")] + if len(video_files) == 0: + video_files = [str(f.relative_to(work_dir)) for f in work_dir.glob("videos*/*/*/*.mp4")] + videos_moved = True # Videos have already been moved + + assert len(video_files) == total_episodes * len(video_keys) + + lfs_untracked_videos = _get_lfs_untracked_videos(work_dir, video_files) + + current_gittatributes = work_dir / ".gitattributes" + if not filecmp.cmp(current_gittatributes, clean_gittatributes, shallow=False): + fix_gitattributes(work_dir, current_gittatributes, clean_gittatributes) + + if lfs_untracked_videos: + fix_lfs_video_files_tracking(work_dir, video_files) + + if videos_moved: + return + + video_dirs = sorted(work_dir.glob("videos*/")) + for ep_chunk in range(total_chunks): + ep_chunk_start = DEFAULT_CHUNK_SIZE * ep_chunk + ep_chunk_end = min(DEFAULT_CHUNK_SIZE * (ep_chunk + 1), total_episodes) + for vid_key in video_keys: + chunk_dir = "/".join(DEFAULT_VIDEO_PATH.split("/")[:-1]).format(episode_chunk=ep_chunk, video_key=vid_key) + (work_dir / chunk_dir).mkdir(parents=True, exist_ok=True) + + for ep_idx in range(ep_chunk_start, ep_chunk_end): + target_path = DEFAULT_VIDEO_PATH.format(episode_chunk=ep_chunk, video_key=vid_key, episode_index=ep_idx) + video_file = V1_VIDEO_FILE.format(video_key=vid_key, episode_index=ep_idx) + if len(video_dirs) == 1: + video_path = video_dirs[0] / video_file + else: + for dir in video_dirs: + if (dir / video_file).is_file(): + video_path = dir / video_file + break + + video_path.rename(work_dir / target_path) + + commit_message = "Move video files into chunk subdirectories" + subprocess.run(["git", "add", "."], cwd=work_dir, check=True) + subprocess.run(["git", "commit", "-m", commit_message], cwd=work_dir, check=True) + subprocess.run(["git", "push"], cwd=work_dir, check=True) + + +def fix_lfs_video_files_tracking(work_dir: Path, lfs_untracked_videos: list[str]) -> None: + """ + HACK: This function fixes the tracking by git lfs which was not properly set on some repos. In that case, + there's no other option than to download the actual files and reupload them with lfs tracking. + """ + for i in range(0, len(lfs_untracked_videos), 100): + files = lfs_untracked_videos[i : i + 100] + try: + subprocess.run(["git", "rm", "--cached", *files], cwd=work_dir, capture_output=True, check=True) + except subprocess.CalledProcessError as e: + print("git rm --cached ERROR:") + print(e.stderr) + subprocess.run(["git", "add", *files], cwd=work_dir, check=True) + + commit_message = "Track video files with git lfs" + subprocess.run(["git", "commit", "-m", commit_message], cwd=work_dir, check=True) + subprocess.run(["git", "push"], cwd=work_dir, check=True) + + +def fix_gitattributes(work_dir: Path, current_gittatributes: Path, clean_gittatributes: Path) -> None: + shutil.copyfile(clean_gittatributes, current_gittatributes) + subprocess.run(["git", "add", ".gitattributes"], cwd=work_dir, check=True) + subprocess.run(["git", "commit", "-m", "Fix .gitattributes"], cwd=work_dir, check=True) + subprocess.run(["git", "push"], cwd=work_dir, check=True) + + +def _lfs_clone(repo_id: str, work_dir: Path, branch: str) -> None: + subprocess.run(["git", "lfs", "install"], cwd=work_dir, check=True) + repo_url = f"https://huggingface.co/datasets/{repo_id}" + env = {"GIT_LFS_SKIP_SMUDGE": "1"} # Prevent downloading LFS files + subprocess.run( + ["git", "clone", "--branch", branch, "--single-branch", "--depth", "1", repo_url, str(work_dir)], + check=True, + env=env, + ) + + +def _get_lfs_untracked_videos(work_dir: Path, video_files: list[str]) -> list[str]: + lfs_tracked_files = subprocess.run( + ["git", "lfs", "ls-files", "-n"], cwd=work_dir, capture_output=True, text=True, check=True + ) + lfs_tracked_files = set(lfs_tracked_files.stdout.splitlines()) + return [f for f in video_files if f not in lfs_tracked_files] + + +def get_videos_info(repo_id: str, local_dir: Path, video_keys: list[str], branch: str) -> dict: + # Assumes first episode + video_files = [ + DEFAULT_VIDEO_PATH.format(episode_chunk=0, video_key=vid_key, episode_index=0) for vid_key in video_keys + ] + hub_api = HfApi() + hub_api.snapshot_download( + repo_id=repo_id, repo_type="dataset", local_dir=local_dir, revision=branch, allow_patterns=video_files + ) + videos_info_dict = {} + for vid_key, vid_path in zip(video_keys, video_files, strict=True): + videos_info_dict[vid_key] = get_video_info(local_dir / vid_path) + + return videos_info_dict + + +def convert_dataset( + repo_id: str, + local_dir: Path, + single_task: str | None = None, + tasks_path: Path | None = None, + tasks_col: Path | None = None, + robot_config: RobotConfig | None = None, + test_branch: str | None = None, + **card_kwargs, +): + v1 = get_hub_safe_version(repo_id, V16) + v1x_dir = local_dir / V16 / repo_id + v20_dir = local_dir / V20 / repo_id + v1x_dir.mkdir(parents=True, exist_ok=True) + v20_dir.mkdir(parents=True, exist_ok=True) + + hub_api = HfApi() + hub_api.snapshot_download( + repo_id=repo_id, repo_type="dataset", revision=v1, local_dir=v1x_dir, ignore_patterns="videos*/" + ) + branch = "main" + if test_branch: + branch = test_branch + create_branch(repo_id=repo_id, branch=test_branch, repo_type="dataset") + + metadata_v1 = load_json(v1x_dir / V1_INFO_PATH) + dataset = datasets.load_dataset("parquet", data_dir=v1x_dir / "data", split="train") + features = get_features_from_hf_dataset(dataset, robot_config) + video_keys = [key for key, ft in features.items() if ft["dtype"] == "video"] + + if single_task and "language_instruction" in dataset.column_names: + logging.warning( + "'single_task' provided but 'language_instruction' tasks_col found. Using 'language_instruction'.", + ) + single_task = None + tasks_col = "language_instruction" + + # Episodes & chunks + episode_indices = sorted(dataset.unique("episode_index")) + total_episodes = len(episode_indices) + assert episode_indices == list(range(total_episodes)) + total_videos = total_episodes * len(video_keys) + total_chunks = total_episodes // DEFAULT_CHUNK_SIZE + if total_episodes % DEFAULT_CHUNK_SIZE != 0: + total_chunks += 1 + + # Tasks + if single_task: + tasks_by_episodes = {ep_idx: single_task for ep_idx in episode_indices} + dataset, tasks = add_task_index_by_episodes(dataset, tasks_by_episodes) + tasks_by_episodes = {ep_idx: [task] for ep_idx, task in tasks_by_episodes.items()} + elif tasks_path: + tasks_by_episodes = load_json(tasks_path) + tasks_by_episodes = {int(ep_idx): task for ep_idx, task in tasks_by_episodes.items()} + dataset, tasks = add_task_index_by_episodes(dataset, tasks_by_episodes) + tasks_by_episodes = {ep_idx: [task] for ep_idx, task in tasks_by_episodes.items()} + elif tasks_col: + dataset, tasks, tasks_by_episodes = add_task_index_from_tasks_col(dataset, tasks_col) + else: + raise ValueError + + assert set(tasks) == {task for ep_tasks in tasks_by_episodes.values() for task in ep_tasks} + tasks = [{"task_index": task_idx, "task": task} for task_idx, task in enumerate(tasks)] + write_jsonlines(tasks, v20_dir / TASKS_PATH) + features["task_index"] = { + "dtype": "int64", + "shape": (1,), + "names": None, + } + + # Videos + if video_keys: + assert metadata_v1.get("video", False) + dataset = dataset.remove_columns(video_keys) + clean_gitattr = Path( + hub_api.hf_hub_download( + repo_id=GITATTRIBUTES_REF, repo_type="dataset", local_dir=local_dir, filename=".gitattributes" + ) + ).absolute() + with tempfile.TemporaryDirectory() as tmp_video_dir: + move_videos(repo_id, video_keys, total_episodes, total_chunks, Path(tmp_video_dir), clean_gitattr, branch) + videos_info = get_videos_info(repo_id, v1x_dir, video_keys=video_keys, branch=branch) + for key in video_keys: + features[key]["shape"] = ( + videos_info[key].pop("video.height"), + videos_info[key].pop("video.width"), + videos_info[key].pop("video.channels"), + ) + features[key]["video_info"] = videos_info[key] + assert math.isclose(videos_info[key]["video.fps"], metadata_v1["fps"], rel_tol=1e-3) + if "encoding" in metadata_v1: + assert videos_info[key]["video.pix_fmt"] == metadata_v1["encoding"]["pix_fmt"] + else: + assert metadata_v1.get("video", 0) == 0 + videos_info = None + + # Split data into 1 parquet file by episode + episode_lengths = split_parquet_by_episodes(dataset, total_episodes, total_chunks, v20_dir) + + if robot_config is not None: + robot_type = robot_config.type + repo_tags = [robot_type] + else: + robot_type = "unknown" + repo_tags = None + + # Episodes + episodes = [ + {"episode_index": ep_idx, "tasks": tasks_by_episodes[ep_idx], "length": episode_lengths[ep_idx]} + for ep_idx in episode_indices + ] + write_jsonlines(episodes, v20_dir / EPISODES_PATH) + + # Assemble metadata v2.0 + metadata_v2_0 = { + "codebase_version": V20, + "robot_type": robot_type, + "total_episodes": total_episodes, + "total_frames": len(dataset), + "total_tasks": len(tasks), + "total_videos": total_videos, + "total_chunks": total_chunks, + "chunks_size": DEFAULT_CHUNK_SIZE, + "fps": metadata_v1["fps"], + "splits": {"train": f"0:{total_episodes}"}, + "data_path": DEFAULT_PARQUET_PATH, + "video_path": DEFAULT_VIDEO_PATH if video_keys else None, + "features": features, + } + write_json(metadata_v2_0, v20_dir / INFO_PATH) + convert_stats_to_json(v1x_dir, v20_dir) + card = create_lerobot_dataset_card(tags=repo_tags, dataset_info=metadata_v2_0, **card_kwargs) + + with contextlib.suppress(EntryNotFoundError, HfHubHTTPError): + hub_api.delete_folder(repo_id=repo_id, path_in_repo="data", repo_type="dataset", revision=branch) + + with contextlib.suppress(EntryNotFoundError, HfHubHTTPError): + hub_api.delete_folder(repo_id=repo_id, path_in_repo="meta_data", repo_type="dataset", revision=branch) + + with contextlib.suppress(EntryNotFoundError, HfHubHTTPError): + hub_api.delete_folder(repo_id=repo_id, path_in_repo="meta", repo_type="dataset", revision=branch) + + hub_api.upload_folder( + repo_id=repo_id, + path_in_repo="data", + folder_path=v20_dir / "data", + repo_type="dataset", + revision=branch, + ) + hub_api.upload_folder( + repo_id=repo_id, + path_in_repo="meta", + folder_path=v20_dir / "meta", + repo_type="dataset", + revision=branch, + ) + + card.push_to_hub(repo_id=repo_id, repo_type="dataset", revision=branch) + + if not test_branch: + create_branch(repo_id=repo_id, branch=V20, repo_type="dataset") + + +def main(): + parser = argparse.ArgumentParser() + task_args = parser.add_mutually_exclusive_group(required=True) + + parser.add_argument( + "--repo-id", + type=str, + required=True, + help="Repository identifier on Hugging Face: a community or a user name `/` the name of the dataset (e.g. `lerobot/pusht`, `cadene/aloha_sim_insertion_human`).", + ) + task_args.add_argument( + "--single-task", + type=str, + help="A short but accurate description of the single task performed in the dataset.", + ) + task_args.add_argument( + "--tasks-col", + type=str, + help="The name of the column containing language instructions", + ) + task_args.add_argument( + "--tasks-path", + type=Path, + help="The path to a .json file containing one language instruction for each episode_index", + ) + parser.add_argument( + "--robot", + type=str, + default=None, + help="Robot config used for the dataset during conversion (e.g. 'koch', 'aloha', 'so100', etc.)", + ) + parser.add_argument( + "--local-dir", + type=Path, + default=None, + help="Local directory to store the dataset during conversion. Defaults to /tmp/lerobot_dataset_v2", + ) + parser.add_argument( + "--license", + type=str, + default="apache-2.0", + help="Repo license. Must be one of https://huggingface.co/docs/hub/repositories-licenses. Defaults to mit.", + ) + parser.add_argument( + "--test-branch", + type=str, + default=None, + help="Repo branch to test your conversion first (e.g. 'v2.0.test')", + ) + + args = parser.parse_args() + if not args.local_dir: + args.local_dir = Path("/tmp/lerobot_dataset_v2") + + if args.robot is not None: + robot_config = make_robot_config(args.robot) + + del args.robot + + convert_dataset(**vars(args), robot_config=robot_config) + + +if __name__ == "__main__": + main() diff --git a/ds_version_convert/v20_to_v21/README.md b/ds_version_convert/v20_to_v21/README.md new file mode 100644 index 0000000..365db2c --- /dev/null +++ b/ds_version_convert/v20_to_v21/README.md @@ -0,0 +1,20 @@ +# LeRobot Dataset v20 to v21 + +## Get started + +1. Install v2.1 lerobot + ```bash + git clone https://github.com/huggingface/lerobot.git + git checkout d602e8169cbad9e93a4a3b3ee1dd8b332af7ebf8 + pip install -e . + ``` + +2. Run the converter: + ```bash + python convert_dataset_v20_to_v21.py \ + --repo-id=your_id \ + --root=your_local_dir \ + --delete-old-stats \ + --push-to-hub \ + --num-workers=8 + ``` \ No newline at end of file diff --git a/ds_version_convert/v20_to_v21/convert_dataset_v20_to_v21.py b/ds_version_convert/v20_to_v21/convert_dataset_v20_to_v21.py new file mode 100644 index 0000000..9745edc --- /dev/null +++ b/ds_version_convert/v20_to_v21/convert_dataset_v20_to_v21.py @@ -0,0 +1,88 @@ +import argparse + +from convert_stats import check_aggregate_stats, convert_stats +from huggingface_hub import HfApi +from lerobot.datasets.lerobot_dataset import LeRobotDataset +from lerobot.datasets.utils import EPISODES_STATS_PATH, STATS_PATH, load_stats, write_info +from lerobot.datasets.v21.convert_dataset_v20_to_v21 import V20, V21 + + +def convert_dataset( + repo_id: str, + root: str | None = None, + push_to_hub: bool = False, + delete_old_stats: bool = False, + branch: str | None = None, + num_workers: int = 4, +): + if root is not None: + dataset = LeRobotDataset(repo_id, root, revision=V20) + else: + dataset = LeRobotDataset(repo_id, revision=V20, force_cache_sync=True) + + if (dataset.root / EPISODES_STATS_PATH).is_file(): + (dataset.root / EPISODES_STATS_PATH).unlink() + + convert_stats(dataset, num_workers=num_workers) + ref_stats = load_stats(dataset.root) + check_aggregate_stats(dataset, ref_stats) + + dataset.meta.info["codebase_version"] = V21 + write_info(dataset.meta.info, dataset.root) + + if push_to_hub: + dataset.push_to_hub(branch=branch, tag_version=False, allow_patterns="meta/") + + # delete old stats.json file + if delete_old_stats and (dataset.root / STATS_PATH).is_file: + (dataset.root / STATS_PATH).unlink() + + hub_api = HfApi() + if delete_old_stats and hub_api.file_exists( + repo_id=dataset.repo_id, filename=STATS_PATH, revision=branch, repo_type="dataset" + ): + hub_api.delete_file(path_in_repo=STATS_PATH, repo_id=dataset.repo_id, revision=branch, repo_type="dataset") + if push_to_hub: + hub_api.create_tag(repo_id, tag=V21, revision=branch, repo_type="dataset") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument( + "--repo-id", + type=str, + required=True, + help="Repository identifier on Hugging Face: a community or a user name `/` the name of the dataset " + "(e.g. `lerobot/pusht`, `cadene/aloha_sim_insertion_human`).", + ) + parser.add_argument( + "--root", + type=str, + default=None, + help="Path to the local dataset root directory. If not provided, the script will use the dataset from local.", + ) + parser.add_argument( + "--push-to-hub", + action="store_true", + help="Push the dataset to the hub after conversion. Defaults to False.", + ) + parser.add_argument( + "--delete-old-stats", + action="store_true", + help="Delete the old stats.json file after conversion. Defaults to False.", + ) + parser.add_argument( + "--branch", + type=str, + default=None, + help="Repo branch to push your dataset. Defaults to the main branch.", + ) + parser.add_argument( + "--num-workers", + type=int, + default=4, + help="Number of workers for parallelizing stats compute. Defaults to 4.", + ) + + args = parser.parse_args() + convert_dataset(**vars(args)) diff --git a/ds_version_convert/v20_to_v21/convert_stats.py b/ds_version_convert/v20_to_v21/convert_stats.py new file mode 100644 index 0000000..947d86c --- /dev/null +++ b/ds_version_convert/v20_to_v21/convert_stats.py @@ -0,0 +1,81 @@ +from concurrent.futures import ProcessPoolExecutor, as_completed + +import numpy as np +from lerobot.datasets.compute_stats import aggregate_stats, get_feature_stats, sample_indices +from lerobot.datasets.lerobot_dataset import LeRobotDataset +from lerobot.datasets.utils import write_episode_stats +from tqdm import tqdm + + +def sample_episode_video_frames(dataset: LeRobotDataset, episode_index: int, ft_key: str) -> np.ndarray: + ep_len = dataset.meta.episodes[episode_index]["length"] + sampled_indices = sample_indices(ep_len) + query_timestamps = dataset._get_query_timestamps(0.0, {ft_key: sampled_indices}) + video_frames = dataset._query_videos(query_timestamps, episode_index) + return video_frames[ft_key].numpy() + + +def convert_episode_stats(dataset: LeRobotDataset, ep_idx: int): + ep_start_idx = dataset.episode_data_index["from"][ep_idx] + ep_end_idx = dataset.episode_data_index["to"][ep_idx] + ep_data = dataset.hf_dataset.select(range(ep_start_idx, ep_end_idx)) + + ep_stats = {} + for key, ft in dataset.features.items(): + if ft["dtype"] == "video": + # We sample only for videos + ep_ft_data = sample_episode_video_frames(dataset, ep_idx, key) + else: + ep_ft_data = np.array(ep_data[key]) + + axes_to_reduce = (0, 2, 3) if ft["dtype"] in ["image", "video"] else 0 + keepdims = True if ft["dtype"] in ["image", "video"] else ep_ft_data.ndim == 1 + ep_stats[key] = get_feature_stats(ep_ft_data, axis=axes_to_reduce, keepdims=keepdims) + + if ft["dtype"] in ["image", "video"]: # remove batch dim + ep_stats[key] = {k: v if k == "count" else np.squeeze(v, axis=0) for k, v in ep_stats[key].items()} + + return ep_stats, ep_idx + + +def convert_stats(dataset: LeRobotDataset, num_workers: int = 0): + assert dataset.episodes is None + print("Computing episodes stats") + total_episodes = dataset.meta.total_episodes + if num_workers > 0: + with ProcessPoolExecutor(max_workers=num_workers) as executor: + futures = { + executor.submit(convert_episode_stats, dataset, ep_idx): ep_idx for ep_idx in range(total_episodes) + } + for future in tqdm(as_completed(futures), total=total_episodes): + ep_stats, ep_idx = future.result() + dataset.meta.episodes_stats[ep_idx] = ep_stats + else: + for ep_idx in tqdm(range(total_episodes)): + ep_stats, _ = convert_episode_stats(dataset, ep_idx) + dataset.meta.episodes_stats[ep_idx] = ep_stats + + for ep_idx in tqdm(range(total_episodes)): + write_episode_stats(ep_idx, dataset.meta.episodes_stats[ep_idx], dataset.root) + + +def check_aggregate_stats( + dataset: LeRobotDataset, + reference_stats: dict[str, dict[str, np.ndarray]], + video_rtol_atol: tuple[float] = (1e-2, 1e-2), + default_rtol_atol: tuple[float] = (5e-6, 6e-5), +): + """Verifies that the aggregated stats from episodes_stats are close to reference stats.""" + agg_stats = aggregate_stats(list(dataset.meta.episodes_stats.values())) + for key, ft in dataset.features.items(): + # These values might need some fine-tuning + if ft["dtype"] == "video": + # to account for image sub-sampling + rtol, atol = video_rtol_atol + else: + rtol, atol = default_rtol_atol + + for stat, val in agg_stats[key].items(): + if key in reference_stats and stat in reference_stats[key]: + err_msg = f"feature='{key}' stats='{stat}'" + np.testing.assert_allclose(val, reference_stats[key][stat], rtol=rtol, atol=atol, err_msg=err_msg) diff --git a/ds_version_convert/v21_to_v20/README.md b/ds_version_convert/v21_to_v20/README.md new file mode 100644 index 0000000..e1eda4a --- /dev/null +++ b/ds_version_convert/v21_to_v20/README.md @@ -0,0 +1,19 @@ +# LeRobot Dataset v21 to v20 + +## Get started + +1. Install v2.1 lerobot + ```bash + git clone https://github.com/huggingface/lerobot.git + git checkout d602e8169cbad9e93a4a3b3ee1dd8b332af7ebf8 + pip install -e . + ``` + +2. Run the converter: + ```bash + python convert_dataset_v21_to_v20.py \ + --repo-id=your_id \ + --root=your_local_dir \ + --delete-old-stats \ + --push-to-hub + ``` \ No newline at end of file diff --git a/ds_version_convert/v21_to_v20/convert_dataset_v21_to_v20.py b/ds_version_convert/v21_to_v20/convert_dataset_v21_to_v20.py new file mode 100644 index 0000000..9fae74f --- /dev/null +++ b/ds_version_convert/v21_to_v20/convert_dataset_v21_to_v20.py @@ -0,0 +1,80 @@ +import argparse + +from huggingface_hub import HfApi +from lerobot.datasets.lerobot_dataset import LeRobotDataset +from lerobot.datasets.utils import EPISODES_STATS_PATH, STATS_PATH, write_info, write_stats +from lerobot.datasets.v21.convert_dataset_v20_to_v21 import V20, V21 + + +def convert_dataset( + repo_id: str, + root: str | None = None, + push_to_hub: bool = False, + delete_old_stats: bool = False, + branch: str | None = None, +): + if root is not None: + dataset = LeRobotDataset(repo_id, root, revision=V21) + else: + dataset = LeRobotDataset(repo_id, revision=V21, force_cache_sync=True) + + if (dataset.root / STATS_PATH).is_file(): + (dataset.root / STATS_PATH).unlink() + + write_stats(dataset.meta.stats, dataset.root) + + dataset.meta.info["codebase_version"] = V20 + write_info(dataset.meta.info, dataset.root) + + if push_to_hub: + dataset.push_to_hub(branch=branch, tag_version=False, allow_patterns="meta/") + + # delete old stats.json file + if delete_old_stats and (dataset.root / EPISODES_STATS_PATH).is_file: + (dataset.root / EPISODES_STATS_PATH).unlink() + + hub_api = HfApi() + if delete_old_stats and hub_api.file_exists( + repo_id=dataset.repo_id, filename=EPISODES_STATS_PATH, revision=branch, repo_type="dataset" + ): + hub_api.delete_file( + path_in_repo=EPISODES_STATS_PATH, repo_id=dataset.repo_id, revision=branch, repo_type="dataset" + ) + if push_to_hub: + hub_api.create_tag(repo_id, tag=V20, revision=branch, repo_type="dataset") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument( + "--repo-id", + type=str, + required=True, + help="Repository identifier on Hugging Face: a community or a user name `/` the name of the dataset " + "(e.g. `lerobot/pusht`, `cadene/aloha_sim_insertion_human`).", + ) + parser.add_argument( + "--root", + type=str, + default=None, + help="Path to the local dataset root directory. If not provided, the script will use the dataset from local.", + ) + parser.add_argument( + "--push-to-hub", + action="store_true", + help="Push the dataset to the hub after conversion. Defaults to False.", + ) + parser.add_argument( + "--delete-old-stats", + action="store_true", + help="Delete the old stats.json file after conversion. Defaults to False.", + ) + parser.add_argument( + "--branch", + type=str, + default=None, + help="Repo branch to push your dataset. Defaults to the main branch.", + ) + + args = parser.parse_args() + convert_dataset(**vars(args)) diff --git a/ds_version_convert/v21_to_v30/README.md b/ds_version_convert/v21_to_v30/README.md new file mode 100644 index 0000000..fd9802d --- /dev/null +++ b/ds_version_convert/v21_to_v30/README.md @@ -0,0 +1,15 @@ +# LeRobot Dataset v21 to v30 + +## Get started + +1. Install v3.0 lerobot + ```bash + git clone https://github.com/huggingface/lerobot.git + pip install -e . + ``` + +2. Run the converter: + ```bash + python convert_dataset_v21_to_v30.py \ + --repo-id=your_id + ``` \ No newline at end of file diff --git a/ds_version_convert/v21_to_v30/convert_dataset_v21_to_v30.py b/ds_version_convert/v21_to_v30/convert_dataset_v21_to_v30.py new file mode 100644 index 0000000..6f320fa --- /dev/null +++ b/ds_version_convert/v21_to_v30/convert_dataset_v21_to_v30.py @@ -0,0 +1,507 @@ +#!/usr/bin/env python + +# Copyright 2025 The HuggingFace Inc. team. All rights reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +This script will help you convert any LeRobot dataset already pushed to the hub from codebase version 2.1 to +3.0. It will: + +- Generate per-episodes stats and writes them in `episodes_stats.jsonl` +- Check consistency between these new stats and the old ones. +- Remove the deprecated `stats.json`. +- Update codebase_version in `info.json`. +- Push this new version to the hub on the 'main' branch and tags it with "v3.0". + +Usage: + +```bash +python src/lerobot/datasets/v30/convert_dataset_v21_to_v30.py \ + --repo-id=lerobot/pusht +``` + +""" + +import argparse +import logging +import shutil +from pathlib import Path +from typing import Any + +import jsonlines +import pandas as pd +import pyarrow as pa +import tqdm +from datasets import Dataset, Features, Image +from huggingface_hub import HfApi, snapshot_download +from lerobot.datasets.compute_stats import aggregate_stats +from lerobot.datasets.lerobot_dataset import CODEBASE_VERSION, LeRobotDataset +from lerobot.datasets.utils import ( + DEFAULT_CHUNK_SIZE, + DEFAULT_DATA_FILE_SIZE_IN_MB, + DEFAULT_DATA_PATH, + DEFAULT_VIDEO_FILE_SIZE_IN_MB, + DEFAULT_VIDEO_PATH, + LEGACY_EPISODES_PATH, + LEGACY_EPISODES_STATS_PATH, + LEGACY_TASKS_PATH, + cast_stats_to_numpy, + flatten_dict, + get_parquet_file_size_in_mb, + get_parquet_num_frames, + get_video_size_in_mb, + load_info, + update_chunk_file_indices, + write_episodes, + write_info, + write_stats, + write_tasks, +) +from lerobot.datasets.video_utils import concatenate_video_files, get_video_duration_in_s +from lerobot.utils.constants import HF_LEROBOT_HOME +from lerobot.utils.utils import init_logging +from requests import HTTPError + +V21 = "v2.1" + + +""" +------------------------- +OLD +data/chunk-000/episode_000000.parquet + +NEW +data/chunk-000/file_000.parquet +------------------------- +OLD +videos/chunk-000/CAMERA/episode_000000.mp4 + +NEW +videos/chunk-000/file_000.mp4 +------------------------- +OLD +episodes.jsonl +{"episode_index": 1, "tasks": ["Put the blue block in the green bowl"], "length": 266} + +NEW +meta/episodes/chunk-000/episodes_000.parquet +episode_index | video_chunk_index | video_file_index | data_chunk_index | data_file_index | tasks | length +------------------------- +OLD +tasks.jsonl +{"task_index": 1, "task": "Put the blue block in the green bowl"} + +NEW +meta/tasks/chunk-000/file_000.parquet +task_index | task +------------------------- +OLD +episodes_stats.jsonl + +NEW +meta/episodes_stats/chunk-000/file_000.parquet +episode_index | mean | std | min | max +------------------------- +UPDATE +meta/info.json +------------------------- +""" + + +def load_jsonlines(fpath: Path) -> list[Any]: + with jsonlines.open(fpath, "r") as reader: + return list(reader) + + +def legacy_load_episodes(local_dir: Path) -> dict: + episodes = load_jsonlines(local_dir / LEGACY_EPISODES_PATH) + return {item["episode_index"]: item for item in sorted(episodes, key=lambda x: x["episode_index"])} + + +def legacy_load_episodes_stats(local_dir: Path) -> dict: + episodes_stats = load_jsonlines(local_dir / LEGACY_EPISODES_STATS_PATH) + return { + item["episode_index"]: cast_stats_to_numpy(item["stats"]) + for item in sorted(episodes_stats, key=lambda x: x["episode_index"]) + } + + +def legacy_load_tasks(local_dir: Path) -> tuple[dict, dict]: + tasks = load_jsonlines(local_dir / LEGACY_TASKS_PATH) + tasks = {item["task_index"]: item["task"] for item in sorted(tasks, key=lambda x: x["task_index"])} + task_to_task_index = {task: task_index for task_index, task in tasks.items()} + return tasks, task_to_task_index + + +def convert_tasks(root, new_root): + logging.info(f"Converting tasks from {root} to {new_root}") + tasks, _ = legacy_load_tasks(root) + task_indices = tasks.keys() + task_strings = tasks.values() + df_tasks = pd.DataFrame({"task_index": task_indices}, index=task_strings) + write_tasks(df_tasks, new_root) + + +def concat_data_files(paths_to_cat, new_root, chunk_idx, file_idx, image_keys): + # TODO(rcadene): to save RAM use Dataset.from_parquet(file) and concatenate_datasets + dataframes = [pd.read_parquet(file) for file in paths_to_cat] + # Concatenate all DataFrames along rows + concatenated_df = pd.concat(dataframes, ignore_index=True) + + path = new_root / DEFAULT_DATA_PATH.format(chunk_index=chunk_idx, file_index=file_idx) + path.parent.mkdir(parents=True, exist_ok=True) + + if len(image_keys) > 0: + schema = pa.Schema.from_pandas(concatenated_df) + features = Features.from_arrow_schema(schema) + for key in image_keys: + features[key] = Image() + schema = features.arrow_schema + else: + schema = None + + concatenated_df.to_parquet(path, index=False, schema=schema) + + +def convert_data(root: Path, new_root: Path, data_file_size_in_mb: int): + data_dir = root / "data" + ep_paths = sorted(data_dir.glob("*/*.parquet")) + + image_keys = get_image_keys(root) + + ep_idx = 0 + chunk_idx = 0 + file_idx = 0 + size_in_mb = 0 + num_frames = 0 + paths_to_cat = [] + episodes_metadata = [] + + logging.info(f"Converting data files from {len(ep_paths)} episodes") + + for ep_path in tqdm.tqdm(ep_paths, desc="convert data files"): + ep_size_in_mb = get_parquet_file_size_in_mb(ep_path) + ep_num_frames = get_parquet_num_frames(ep_path) + ep_metadata = { + "episode_index": ep_idx, + "data/chunk_index": chunk_idx, + "data/file_index": file_idx, + "dataset_from_index": num_frames, + "dataset_to_index": num_frames + ep_num_frames, + } + size_in_mb += ep_size_in_mb + num_frames += ep_num_frames + episodes_metadata.append(ep_metadata) + ep_idx += 1 + + if size_in_mb < data_file_size_in_mb: + paths_to_cat.append(ep_path) + continue + + if paths_to_cat: + concat_data_files(paths_to_cat, new_root, chunk_idx, file_idx, image_keys) + + # Reset for the next file + size_in_mb = ep_size_in_mb + paths_to_cat = [ep_path] + + chunk_idx, file_idx = update_chunk_file_indices(chunk_idx, file_idx, DEFAULT_CHUNK_SIZE) + + # Write remaining data if any + if paths_to_cat: + concat_data_files(paths_to_cat, new_root, chunk_idx, file_idx, image_keys) + + return episodes_metadata + + +def get_video_keys(root): + info = load_info(root) + features = info["features"] + video_keys = [key for key, ft in features.items() if ft["dtype"] == "video"] + return video_keys + + +def get_image_keys(root): + info = load_info(root) + features = info["features"] + image_keys = [key for key, ft in features.items() if ft["dtype"] == "image"] + return image_keys + + +def convert_videos(root: Path, new_root: Path, video_file_size_in_mb: int): + logging.info(f"Converting videos from {root} to {new_root}") + + video_keys = get_video_keys(root) + if len(video_keys) == 0: + return None + + video_keys = sorted(video_keys) + + eps_metadata_per_cam = [] + for camera in video_keys: + eps_metadata = convert_videos_of_camera(root, new_root, camera, video_file_size_in_mb) + eps_metadata_per_cam.append(eps_metadata) + + num_eps_per_cam = [len(eps_cam_map) for eps_cam_map in eps_metadata_per_cam] + if len(set(num_eps_per_cam)) != 1: + raise ValueError(f"All cams dont have same number of episodes ({num_eps_per_cam}).") + + episods_metadata = [] + num_cameras = len(video_keys) + num_episodes = num_eps_per_cam[0] + for ep_idx in tqdm.tqdm(range(num_episodes), desc="convert videos"): + # Sanity check + ep_ids = [eps_metadata_per_cam[cam_idx][ep_idx]["episode_index"] for cam_idx in range(num_cameras)] + ep_ids += [ep_idx] + if len(set(ep_ids)) != 1: + raise ValueError(f"All episode indices need to match ({ep_ids}).") + + ep_dict = {} + for cam_idx in range(num_cameras): + ep_dict.update(eps_metadata_per_cam[cam_idx][ep_idx]) + episods_metadata.append(ep_dict) + + return episods_metadata + + +def convert_videos_of_camera(root: Path, new_root: Path, video_key: str, video_file_size_in_mb: int): + # Access old paths to mp4 + videos_dir = root / "videos" + ep_paths = sorted(videos_dir.glob(f"*/{video_key}/*.mp4")) + + ep_idx = 0 + chunk_idx = 0 + file_idx = 0 + size_in_mb = 0 + duration_in_s = 0.0 + paths_to_cat = [] + episodes_metadata = [] + + for ep_path in tqdm.tqdm(ep_paths, desc=f"convert videos of {video_key}"): + ep_size_in_mb = get_video_size_in_mb(ep_path) + ep_duration_in_s = get_video_duration_in_s(ep_path) + + # Check if adding this episode would exceed the limit + if size_in_mb + ep_size_in_mb >= video_file_size_in_mb and len(paths_to_cat) > 0: + # Size limit would be exceeded, save current accumulation WITHOUT this episode + concatenate_video_files( + paths_to_cat, + new_root / DEFAULT_VIDEO_PATH.format(video_key=video_key, chunk_index=chunk_idx, file_index=file_idx), + ) + + # Update episodes metadata for the file we just saved + for i, _ in enumerate(paths_to_cat): + past_ep_idx = ep_idx - len(paths_to_cat) + i + episodes_metadata[past_ep_idx][f"videos/{video_key}/chunk_index"] = chunk_idx + episodes_metadata[past_ep_idx][f"videos/{video_key}/file_index"] = file_idx + + # Move to next file and start fresh with current episode + chunk_idx, file_idx = update_chunk_file_indices(chunk_idx, file_idx, DEFAULT_CHUNK_SIZE) + size_in_mb = 0 + duration_in_s = 0.0 + paths_to_cat = [] + + # Add current episode metadata + ep_metadata = { + "episode_index": ep_idx, + f"videos/{video_key}/chunk_index": chunk_idx, # Will be updated when file is saved + f"videos/{video_key}/file_index": file_idx, # Will be updated when file is saved + f"videos/{video_key}/from_timestamp": duration_in_s, + f"videos/{video_key}/to_timestamp": duration_in_s + ep_duration_in_s, + } + episodes_metadata.append(ep_metadata) + + # Add current episode to accumulation + paths_to_cat.append(ep_path) + size_in_mb += ep_size_in_mb + duration_in_s += ep_duration_in_s + ep_idx += 1 + + # Write remaining videos if any + if paths_to_cat: + concatenate_video_files( + paths_to_cat, + new_root / DEFAULT_VIDEO_PATH.format(video_key=video_key, chunk_index=chunk_idx, file_index=file_idx), + ) + + # Update episodes metadata for the final file + for i, _ in enumerate(paths_to_cat): + past_ep_idx = ep_idx - len(paths_to_cat) + i + episodes_metadata[past_ep_idx][f"videos/{video_key}/chunk_index"] = chunk_idx + episodes_metadata[past_ep_idx][f"videos/{video_key}/file_index"] = file_idx + + return episodes_metadata + + +def generate_episode_metadata_dict(episodes_legacy_metadata, episodes_metadata, episodes_stats, episodes_videos=None): + num_episodes = len(episodes_metadata) + episodes_legacy_metadata_vals = list(episodes_legacy_metadata.values()) + episodes_stats_vals = list(episodes_stats.values()) + episodes_stats_keys = list(episodes_stats.keys()) + + for i in range(num_episodes): + ep_legacy_metadata = episodes_legacy_metadata_vals[i] + ep_metadata = episodes_metadata[i] + ep_stats = episodes_stats_vals[i] + + ep_ids_set = { + ep_legacy_metadata["episode_index"], + ep_metadata["episode_index"], + episodes_stats_keys[i], + } + + if episodes_videos is None: + ep_video = {} + else: + ep_video = episodes_videos[i] + ep_ids_set.add(ep_video["episode_index"]) + + if len(ep_ids_set) != 1: + raise ValueError(f"Number of episodes is not the same ({ep_ids_set}).") + + ep_dict = {**ep_metadata, **ep_video, **ep_legacy_metadata, **flatten_dict({"stats": ep_stats})} + ep_dict["meta/episodes/chunk_index"] = 0 + ep_dict["meta/episodes/file_index"] = 0 + yield ep_dict + + +def convert_episodes_metadata(root, new_root, episodes_metadata, episodes_video_metadata=None): + logging.info(f"Converting episodes metadata from {root} to {new_root}") + + episodes_legacy_metadata = legacy_load_episodes(root) + episodes_stats = legacy_load_episodes_stats(root) + + num_eps_set = {len(episodes_legacy_metadata), len(episodes_metadata)} + if episodes_video_metadata is not None: + num_eps_set.add(len(episodes_video_metadata)) + + if len(num_eps_set) != 1: + raise ValueError(f"Number of episodes is not the same ({num_eps_set}).") + + ds_episodes = Dataset.from_generator( + lambda: generate_episode_metadata_dict( + episodes_legacy_metadata, episodes_metadata, episodes_stats, episodes_video_metadata + ) + ) + write_episodes(ds_episodes, new_root) + + stats = aggregate_stats(list(episodes_stats.values())) + write_stats(stats, new_root) + + +def convert_info(root, new_root, data_file_size_in_mb, video_file_size_in_mb): + info = load_info(root) + info["codebase_version"] = "v3.0" + del info["total_chunks"] + del info["total_videos"] + info["data_files_size_in_mb"] = data_file_size_in_mb + info["video_files_size_in_mb"] = video_file_size_in_mb + info["data_path"] = DEFAULT_DATA_PATH + info["video_path"] = DEFAULT_VIDEO_PATH + info["fps"] = int(info["fps"]) + logging.info(f"Converting info from {root} to {new_root}") + for key in info["features"]: + if info["features"][key]["dtype"] == "video": + # already has fps in video_info + continue + info["features"][key]["fps"] = info["fps"] + write_info(info, new_root) + + +def convert_dataset( + repo_id: str, + branch: str | None = None, + data_file_size_in_mb: int | None = None, + video_file_size_in_mb: int | None = None, +): + root = HF_LEROBOT_HOME / repo_id + old_root = HF_LEROBOT_HOME / f"{repo_id}_old" + new_root = HF_LEROBOT_HOME / f"{repo_id}_v30" + + if data_file_size_in_mb is None: + data_file_size_in_mb = DEFAULT_DATA_FILE_SIZE_IN_MB + if video_file_size_in_mb is None: + video_file_size_in_mb = DEFAULT_VIDEO_FILE_SIZE_IN_MB + + if old_root.is_dir() and root.is_dir(): + shutil.rmtree(str(root)) + shutil.move(str(old_root), str(root)) + + if new_root.is_dir(): + shutil.rmtree(new_root) + + snapshot_download( + repo_id, + repo_type="dataset", + revision=V21, + local_dir=root, + ) + + convert_info(root, new_root, data_file_size_in_mb, video_file_size_in_mb) + convert_tasks(root, new_root) + episodes_metadata = convert_data(root, new_root, data_file_size_in_mb) + episodes_videos_metadata = convert_videos(root, new_root, video_file_size_in_mb) + convert_episodes_metadata(root, new_root, episodes_metadata, episodes_videos_metadata) + + shutil.move(str(root), str(old_root)) + shutil.move(str(new_root), str(root)) + + hub_api = HfApi() + try: + hub_api.delete_tag(repo_id, tag=CODEBASE_VERSION, repo_type="dataset") + except HTTPError as e: + print(f"tag={CODEBASE_VERSION} probably doesn't exist. Skipping exception ({e})") + pass + hub_api.delete_files( + delete_patterns=["data/chunk*/episode_*", "meta/*.jsonl", "videos/chunk*"], + repo_id=repo_id, + revision=branch, + repo_type="dataset", + ) + hub_api.create_tag(repo_id, tag=CODEBASE_VERSION, revision=branch, repo_type="dataset") + + LeRobotDataset(repo_id).push_to_hub() + + +if __name__ == "__main__": + init_logging() + parser = argparse.ArgumentParser() + parser.add_argument( + "--repo-id", + type=str, + required=True, + help="Repository identifier on Hugging Face: a community or a user name `/` the name of the dataset " + "(e.g. `lerobot/pusht`, `cadene/aloha_sim_insertion_human`).", + ) + parser.add_argument( + "--branch", + type=str, + default=None, + help="Repo branch to push your dataset. Defaults to the main branch.", + ) + parser.add_argument( + "--data-file-size-in-mb", + type=int, + default=None, + help="File size in MB. Defaults to 100 for data and 500 for videos.", + ) + parser.add_argument( + "--video-file-size-in-mb", + type=int, + default=None, + help="File size in MB. Defaults to 100 for data and 500 for videos.", + ) + + args = parser.parse_args() + convert_dataset(**vars(args))