mirror of
https://github.com/huggingface/lerobot.git
synced 2026-07-27 19:56:09 +00:00
Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 45243dcf7c | |||
| 034693f724 | |||
| bbeacfe57d | |||
| 801346e18c | |||
| ab87fd9764 | |||
| 6c57dfd2ee | |||
| d63e6e67a5 | |||
| bb3ef3537f |
@@ -0,0 +1,11 @@
|
||||
version: 2
|
||||
updates:
|
||||
- package-ecosystem: "github-actions"
|
||||
directory: "/"
|
||||
schedule:
|
||||
interval: "weekly"
|
||||
cooldown:
|
||||
default-days: 7
|
||||
groups:
|
||||
actions:
|
||||
patterns: ["*"]
|
||||
@@ -519,6 +519,13 @@ def compute_episode_stats(
|
||||
if features[key]["dtype"] in {"string", "language"}:
|
||||
continue
|
||||
|
||||
# Features with a zero-width dimension contain no statistics-bearing
|
||||
# values. Skip them like strings instead of letting
|
||||
# get_feature_stats -> RunningQuantileStats.update reshape a size-0 array,
|
||||
# which raises "ValueError: cannot reshape array of size 0".
|
||||
if any(dim == 0 for dim in features[key].get("shape", ())):
|
||||
continue
|
||||
|
||||
if features[key]["dtype"] in ["image", "video"]:
|
||||
ep_ft_array = sample_images(data)
|
||||
axes_to_reduce = (0, 2, 3)
|
||||
|
||||
@@ -172,6 +172,23 @@ class DatasetWriter:
|
||||
def _get_image_file_dir(self, episode_index: int, image_key: str) -> Path:
|
||||
return self._get_image_file_path(episode_index, image_key, frame_index=0).parent
|
||||
|
||||
def _get_episode_buffer_index(self) -> int:
|
||||
episode_index = self.episode_buffer["episode_index"]
|
||||
# episode_index is `int` when freshly created, but becomes `np.ndarray` after
|
||||
# save_episode() mutates the buffer. Handle both types here.
|
||||
if isinstance(episode_index, np.ndarray):
|
||||
episode_index = episode_index.item() if episode_index.size == 1 else episode_index[0]
|
||||
return int(episode_index)
|
||||
|
||||
def _delete_camera_frame_dirs(self, camera_keys: list[str]) -> None:
|
||||
if self.image_writer is not None:
|
||||
self._wait_image_writer()
|
||||
episode_index = self._get_episode_buffer_index()
|
||||
for camera_key in camera_keys:
|
||||
img_dir = self._get_image_file_dir(episode_index, camera_key)
|
||||
if img_dir.is_dir():
|
||||
shutil.rmtree(img_dir)
|
||||
|
||||
def _save_image(
|
||||
self, image: torch.Tensor | np.ndarray | PIL.Image.Image, fpath: Path, compress_level: int = 1
|
||||
) -> None:
|
||||
@@ -369,7 +386,9 @@ class DatasetWriter:
|
||||
self._episodes_since_last_encoding = 0
|
||||
|
||||
if episode_data is None:
|
||||
self.clear_episode_buffer(delete_images=len(self._meta.image_keys) > 0)
|
||||
if len(self._meta.image_keys) > 0:
|
||||
self._delete_camera_frame_dirs(self._meta.image_keys)
|
||||
self.episode_buffer = self._create_episode_buffer()
|
||||
|
||||
def _batch_save_episode_video(self, start_episode: int, end_episode: int | None = None) -> None:
|
||||
"""Batch save videos for multiple episodes."""
|
||||
@@ -561,10 +580,10 @@ class DatasetWriter:
|
||||
return metadata
|
||||
|
||||
def clear_episode_buffer(self, delete_images: bool = True) -> None:
|
||||
"""Discard the current episode buffer and optionally delete temp images.
|
||||
"""Discard the current episode buffer and optionally delete temp camera frames.
|
||||
|
||||
Args:
|
||||
delete_images: If ``True``, remove temporary image directories
|
||||
delete_images: If ``True``, remove temporary camera frame directories
|
||||
written for the current episode.
|
||||
"""
|
||||
# Cancel streaming encoder if active
|
||||
@@ -572,17 +591,7 @@ class DatasetWriter:
|
||||
self._streaming_encoder.cancel_episode()
|
||||
|
||||
if delete_images:
|
||||
if self.image_writer is not None:
|
||||
self._wait_image_writer()
|
||||
episode_index = self.episode_buffer["episode_index"]
|
||||
# episode_index is `int` when freshly created, but becomes `np.ndarray` after
|
||||
# save_episode() mutates the buffer. Handle both types here.
|
||||
if isinstance(episode_index, np.ndarray):
|
||||
episode_index = episode_index.item() if episode_index.size == 1 else episode_index[0]
|
||||
for cam_key in self._meta.image_keys:
|
||||
img_dir = self._get_image_file_dir(episode_index, cam_key)
|
||||
if img_dir.is_dir():
|
||||
shutil.rmtree(img_dir)
|
||||
self._delete_camera_frame_dirs(self._meta.camera_keys)
|
||||
|
||||
self.episode_buffer = self._create_episode_buffer()
|
||||
|
||||
|
||||
@@ -64,12 +64,20 @@ def get_hf_features_from_features(features: dict) -> datasets.Features:
|
||||
continue
|
||||
elif ft["dtype"] == "image":
|
||||
hf_features[key] = datasets.Image()
|
||||
elif len(ft["shape"]) > 1 and any(dim == 0 for dim in ft["shape"]):
|
||||
raise ValueError(
|
||||
f"Multidimensional features with a zero-width dimension are not supported: "
|
||||
f"'{key}' has shape {ft['shape']}. Only the one-dimensional shape (0,) is supported."
|
||||
)
|
||||
elif ft["shape"] == (1,):
|
||||
hf_features[key] = datasets.Value(dtype=ft["dtype"])
|
||||
elif len(ft["shape"]) == 1:
|
||||
hf_features[key] = datasets.Sequence(
|
||||
length=ft["shape"][0], feature=datasets.Value(dtype=ft["dtype"])
|
||||
)
|
||||
# A zero-width feature (shape=(0,)) has no fixed-size Arrow representation:
|
||||
# pyarrow rejects a fixed-size list of length 0 ("list_size needs to be a
|
||||
# strict positive integer"). Store it as a variable-length sequence
|
||||
# (length=-1) so each per-frame value is simply an empty list.
|
||||
seq_length = ft["shape"][0] if ft["shape"][0] > 0 else -1
|
||||
hf_features[key] = datasets.Sequence(length=seq_length, feature=datasets.Value(dtype=ft["dtype"]))
|
||||
elif len(ft["shape"]) == 2:
|
||||
hf_features[key] = datasets.Array2D(shape=ft["shape"], dtype=ft["dtype"])
|
||||
elif len(ft["shape"]) == 3:
|
||||
|
||||
@@ -61,6 +61,7 @@ import pyarrow as pa
|
||||
import tqdm
|
||||
from datasets import Dataset, Features, Image
|
||||
from huggingface_hub import HfApi, snapshot_download
|
||||
from huggingface_hub.errors import RevisionNotFoundError
|
||||
from requests import HTTPError
|
||||
|
||||
from lerobot.datasets import CODEBASE_VERSION, LeRobotDataset, aggregate_stats
|
||||
@@ -521,7 +522,7 @@ def convert_dataset(
|
||||
hub_api = HfApi()
|
||||
try:
|
||||
hub_api.delete_tag(repo_id, tag=CODEBASE_VERSION, repo_type="dataset")
|
||||
except HTTPError as e:
|
||||
except (HTTPError, RevisionNotFoundError) as e:
|
||||
print(f"tag={CODEBASE_VERSION} probably doesn't exist. Skipping exception ({e})")
|
||||
pass
|
||||
hub_api.delete_files(
|
||||
|
||||
@@ -453,6 +453,9 @@ def eval_policy(
|
||||
raise exc from None
|
||||
|
||||
start = time.time()
|
||||
# Preserve the mode for direct callers. eval_policy_all scopes the mode
|
||||
# around all tasks so parallel evaluations cannot race with each other.
|
||||
was_training = policy.training
|
||||
policy.eval()
|
||||
|
||||
# Determine how many batched rollouts we need to get n_episodes. Note that if n_episodes is not evenly
|
||||
@@ -674,6 +677,8 @@ def eval_policy(
|
||||
if save_predicted_video:
|
||||
info["predicted_video_paths"] = predicted_video_paths
|
||||
|
||||
policy.train(was_training)
|
||||
|
||||
return info
|
||||
|
||||
|
||||
@@ -1010,40 +1015,48 @@ def eval_policy_all(
|
||||
recording_private=recording_private,
|
||||
)
|
||||
|
||||
if max_parallel_tasks <= 1:
|
||||
prefetch_thread: threading.Thread | None = None
|
||||
for i, (task_group, task_id, env) in enumerate(tasks):
|
||||
if prefetch_thread is not None:
|
||||
prefetch_thread.join()
|
||||
prefetch_thread = None
|
||||
# Set the shared policy's mode before launching any workers. Restoring it
|
||||
# inside individual tasks would let one task enable training mode while
|
||||
# another task is still evaluating.
|
||||
was_training = policy.training
|
||||
policy.eval()
|
||||
try:
|
||||
if max_parallel_tasks <= 1:
|
||||
prefetch_thread: threading.Thread | None = None
|
||||
for i, (task_group, task_id, env) in enumerate(tasks):
|
||||
if prefetch_thread is not None:
|
||||
prefetch_thread.join()
|
||||
prefetch_thread = None
|
||||
|
||||
try:
|
||||
tg, tid, metrics = task_runner(task_group, task_id, env)
|
||||
_accumulate_to(tg, metrics)
|
||||
per_task_infos.append({"task_group": tg, "task_id": tid, "metrics": metrics})
|
||||
finally:
|
||||
env.close()
|
||||
# Prefetch next task's workers *after* closing current env to prevent
|
||||
# GPU memory overlap between consecutive tasks.
|
||||
if i + 1 < len(tasks):
|
||||
next_env = tasks[i + 1][2]
|
||||
if hasattr(next_env, "_ensure"):
|
||||
prefetch_thread = threading.Thread(target=next_env._ensure, daemon=True)
|
||||
prefetch_thread.start()
|
||||
else:
|
||||
with cf.ThreadPoolExecutor(max_workers=max_parallel_tasks) as executor:
|
||||
fut2meta = {}
|
||||
for task_group, task_id, env in tasks:
|
||||
fut = executor.submit(task_runner, task_group, task_id, env)
|
||||
fut2meta[fut] = (task_group, task_id, env)
|
||||
for fut in cf.as_completed(fut2meta):
|
||||
tg, tid, env = fut2meta[fut]
|
||||
try:
|
||||
tg, tid, metrics = fut.result()
|
||||
tg, tid, metrics = task_runner(task_group, task_id, env)
|
||||
_accumulate_to(tg, metrics)
|
||||
per_task_infos.append({"task_group": tg, "task_id": tid, "metrics": metrics})
|
||||
finally:
|
||||
env.close()
|
||||
# Prefetch next task's workers *after* closing current env to prevent
|
||||
# GPU memory overlap between consecutive tasks.
|
||||
if i + 1 < len(tasks):
|
||||
next_env = tasks[i + 1][2]
|
||||
if hasattr(next_env, "_ensure"):
|
||||
prefetch_thread = threading.Thread(target=next_env._ensure, daemon=True)
|
||||
prefetch_thread.start()
|
||||
else:
|
||||
with cf.ThreadPoolExecutor(max_workers=max_parallel_tasks) as executor:
|
||||
fut2meta = {}
|
||||
for task_group, task_id, env in tasks:
|
||||
fut = executor.submit(task_runner, task_group, task_id, env)
|
||||
fut2meta[fut] = (task_group, task_id, env)
|
||||
for fut in cf.as_completed(fut2meta):
|
||||
tg, tid, env = fut2meta[fut]
|
||||
try:
|
||||
tg, tid, metrics = fut.result()
|
||||
_accumulate_to(tg, metrics)
|
||||
per_task_infos.append({"task_group": tg, "task_id": tid, "metrics": metrics})
|
||||
finally:
|
||||
env.close()
|
||||
finally:
|
||||
policy.train(was_training)
|
||||
|
||||
# compute aggregated metrics helper (robust to lists/scalars)
|
||||
def _agg_from_list(xs):
|
||||
|
||||
@@ -453,9 +453,11 @@ def record(
|
||||
encoder_queue_maxsize=cfg.dataset.encoder_queue_maxsize,
|
||||
)
|
||||
|
||||
robot.connect()
|
||||
# Connect the teleoperator before the robot so the robot isn't left idle (and possibly
|
||||
# tripping a firmware watchdog) during teleop init. Matches lerobot_teleoperate.py.
|
||||
if teleop is not None:
|
||||
teleop.connect()
|
||||
robot.connect()
|
||||
|
||||
listener, events = init_keyboard_listener()
|
||||
|
||||
|
||||
@@ -687,6 +687,26 @@ def test_compute_episode_stats_string_features_skipped():
|
||||
assert "q01" in stats["action"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shape", [(0,), (0, 2), (2, 0), (1, 0, 2)])
|
||||
def test_compute_episode_stats_zero_width_feature_skipped(shape):
|
||||
"""Features with any zero-width dimension carry no values and are skipped."""
|
||||
episode_data = {
|
||||
"action": np.random.normal(0, 1, (100, 5)).astype(np.float32),
|
||||
"target": np.zeros((100, *shape), dtype=np.float32),
|
||||
}
|
||||
features = {
|
||||
"action": {"dtype": "float32", "shape": (5,)},
|
||||
"target": {"dtype": "float32", "shape": shape},
|
||||
}
|
||||
|
||||
stats = compute_episode_stats(episode_data, features)
|
||||
|
||||
# Zero-width features are skipped, just like strings; non-empty features are unaffected.
|
||||
assert "target" not in stats
|
||||
assert "action" in stats
|
||||
assert "q01" in stats["action"]
|
||||
|
||||
|
||||
def test_aggregate_feature_stats_with_quantiles():
|
||||
"""Test aggregating feature stats that include quantiles."""
|
||||
stats_ft_list = [
|
||||
|
||||
@@ -27,6 +27,7 @@ pytest.importorskip("datasets", reason="datasets is required (install lerobot[da
|
||||
|
||||
from lerobot.configs import VideoEncoderConfig
|
||||
from lerobot.datasets.dataset_writer import _encode_video_worker
|
||||
from lerobot.datasets.feature_utils import get_hf_features_from_features
|
||||
from lerobot.datasets.lerobot_dataset import LeRobotDataset
|
||||
from lerobot.datasets.utils import DEFAULT_IMAGE_PATH
|
||||
from tests.fixtures.constants import DEFAULT_FPS, DUMMY_REPO_ID
|
||||
@@ -189,6 +190,36 @@ def test_save_multiple_episodes(tmp_path):
|
||||
assert dataset.meta.total_frames == total_frames
|
||||
|
||||
|
||||
def test_save_episode_with_zero_width_feature(tmp_path):
|
||||
"""A one-dimensional empty numeric feature round-trips and has no statistics."""
|
||||
features = {
|
||||
**SIMPLE_FEATURES,
|
||||
"target": {"dtype": "float32", "shape": (0,), "names": None},
|
||||
}
|
||||
root = tmp_path / "ds"
|
||||
dataset = LeRobotDataset.create(repo_id=DUMMY_REPO_ID, fps=DEFAULT_FPS, features=features, root=root)
|
||||
for _ in range(4):
|
||||
dataset.add_frame(_make_frame(features))
|
||||
dataset.save_episode()
|
||||
dataset.finalize()
|
||||
|
||||
assert dataset.meta.total_episodes == 1
|
||||
assert dataset.meta.total_frames == 4
|
||||
|
||||
reloaded = LeRobotDataset(repo_id=DUMMY_REPO_ID, root=root)
|
||||
target = np.asarray(reloaded[0]["target"])
|
||||
assert target.shape == (0,)
|
||||
assert "target" not in (reloaded.meta.stats or {})
|
||||
|
||||
|
||||
@pytest.mark.parametrize("shape", [(0, 2), (2, 0), (1, 0, 2)])
|
||||
def test_multidimensional_zero_width_feature_rejected(shape):
|
||||
features = {"target": {"dtype": "float32", "shape": shape, "names": None}}
|
||||
|
||||
with pytest.raises(ValueError, match="Multidimensional features with a zero-width dimension"):
|
||||
get_hf_features_from_features(features)
|
||||
|
||||
|
||||
# ── clear / lifecycle ────────────────────────────────────────────────
|
||||
|
||||
|
||||
@@ -204,6 +235,38 @@ def test_clear_resets_buffer(tmp_path):
|
||||
assert dataset.writer.episode_buffer["size"] == 0
|
||||
|
||||
|
||||
def test_clear_removes_video_frame_staging_dir(tmp_path):
|
||||
"""clear_episode_buffer() removes PNG staging dirs for video features."""
|
||||
video_key = "observation.images.cam"
|
||||
features = {
|
||||
video_key: {
|
||||
"dtype": "video",
|
||||
"shape": (64, 96, 3),
|
||||
"names": ["height", "width", "channels"],
|
||||
},
|
||||
"action": {"dtype": "float32", "shape": (2,), "names": None},
|
||||
}
|
||||
dataset = LeRobotDataset.create(
|
||||
repo_id=DUMMY_REPO_ID,
|
||||
fps=DEFAULT_FPS,
|
||||
features=features,
|
||||
root=tmp_path / "ds",
|
||||
use_videos=True,
|
||||
)
|
||||
|
||||
dataset.add_frame(_make_frame(features))
|
||||
video_staging_dir = (
|
||||
dataset.root
|
||||
/ Path(DEFAULT_IMAGE_PATH.format(image_key=video_key, episode_index=0, frame_index=0)).parent
|
||||
)
|
||||
assert video_staging_dir.is_dir()
|
||||
|
||||
dataset.clear_episode_buffer()
|
||||
|
||||
assert dataset.writer.episode_buffer["size"] == 0
|
||||
assert not video_staging_dir.exists()
|
||||
|
||||
|
||||
def test_finalize_is_idempotent(tmp_path):
|
||||
"""Calling finalize() twice does not raise."""
|
||||
dataset = LeRobotDataset.create(
|
||||
|
||||
Reference in New Issue
Block a user