Compare commits

..

5 Commits

Author SHA1 Message Date
Pepijn 4045589246 docs(rtc): use 15-step training delay 2026-08-04 21:09:12 +02:00
Pepijn 80761e8a7a feat(pi05): add training-time RTC to SE3 branch 2026-08-04 21:07:26 +02:00
Pepijn b0cceb2a5f feat(pi05): support continuous 6D SE(3) actions (#4132)
* feat(pi05): add continuous 6D SE3 actions

* fix(processors): align rotation 6D with UMI rows
2026-07-27 12:16:33 +02:00
Pepijn 8e12a5351a feat(pi05): compose relative poses in SE3 2026-07-21 20:43:04 +02:00
Pepijn 7e1077f19a feat(pi05): synthesize relative proprioceptive history 2026-07-18 23:56:19 +02:00
65 changed files with 3132 additions and 1089 deletions
+21 -30
View File
@@ -81,12 +81,6 @@ merged. Both prompts also carry a causal **event-boundary** definition (a
new event starts when an object becomes held / is released / reaches a new
location / a lid changes state / contents move) to sharpen where cuts land.
Optionally, a third **seeded-relabel** pass (`--plan.subtask_seeded_relabel`)
revisits each span with its previous/current/next segment contact sheets and
minimally corrects the label, using the first label as a prior — it keeps the
boundaries fixed and only sharpens wording, at the cost of one extra call per
subtask.
The resulting spans are then stitched into a gap-free, full-episode
cover, so **every frame has exactly one active subtask**. See
[`run_hf_job.py`](https://github.com/huggingface/lerobot/blob/main/examples/annotations/run_hf_job.py)
@@ -163,33 +157,30 @@ Every module is on by default and can be toggled independently (set to
### The VLM (`--vlm.*`)
| Flag | Default | What it does |
| -------------------------- | ------------------ | ------------------------------------------------------------------------------------ |
| `--vlm.model_id` | `Qwen/Qwen3.6-27B` | The model to serve and prompt. |
| `--vlm.camera_key` | first `images.*` | Which camera every prompt is grounded on. |
| `--vlm.serve_command` | auto | The exact `vllm serve …` command (set TP size, GPU memory, `--max-model-len` here). |
| `--vlm.parallel_servers` | `1` | Independent servers for round-robin routing (one per GPU). |
| `--vlm.num_gpus` | `0` | GPUs per server (`0` = one each). |
| `--vlm.client_concurrency` | `16` | In-flight requests across all servers. |
| `--vlm.max_new_tokens` | `512` | Generation cap per call. |
| `--vlm.temperature` | `0.2` | Sampling temperature. |
| `--vlm.reasoning_effort` | `null` | Thinking-budget hint (`low`/`medium`/`high`) forwarded to OpenAI-compatible servers. |
| Flag | Default | What it does |
| -------------------------- | ------------------ | ----------------------------------------------------------------------------------- |
| `--vlm.model_id` | `Qwen/Qwen3.6-27B` | The model to serve and prompt. |
| `--vlm.camera_key` | first `images.*` | Which camera every prompt is grounded on. |
| `--vlm.serve_command` | auto | The exact `vllm serve …` command (set TP size, GPU memory, `--max-model-len` here). |
| `--vlm.parallel_servers` | `1` | Independent servers for round-robin routing (one per GPU). |
| `--vlm.num_gpus` | `0` | GPUs per server (`0` = one each). |
| `--vlm.client_concurrency` | `16` | In-flight requests across all servers. |
| `--vlm.max_new_tokens` | `512` | Generation cap per call. |
| `--vlm.temperature` | `0.2` | Sampling temperature. |
### Subtasks / plan / memory (`--plan.*`)
| Flag | Default | What it does |
| ------------------------------- | ---------- | ---------------------------------------------------------------------------------------------------------------------------- |
| `--plan.frames_per_second` | `2.0` | Frame sampling rate for the contact sheets (`2.0` = one frame every 0.5s). |
| `--plan.max_frames_per_prompt` | `60` | Frame budget per VLM call. Episodes whose sampling exceeds this are auto-windowed at the same density, then stitched. |
| `--plan.contact_sheet_columns` | `5` | Columns per contact-sheet grid (`contact_sheet_frames_per_sheet` tiles, time row-major). |
| `--plan.plan_max_steps` | `8` | Upper bound on subtasks per episode. |
| `--plan.subtask_describe_first` | `true` | Run the describe→segment grounding pass (best subtask quality; +1 call/episode). |
| `--plan.subtask_seeded_relabel` | `false` | Second pass: re-label each subtask from its prev/current/next contact sheets, seeded with the first label (+1 call/subtask). |
| `--plan.subtask_relabel_frames` | `5` | Frames sampled uniformly per segment sheet in the relabel pass (only used when `subtask_seeded_relabel=true`). |
| `--plan.emit_plan` | `true` | Emit the numbered `plan` rows (`false` = subtasks + memory only). |
| `--plan.emit_memory` | `true` | Emit the `memory` rows (`false` = subtasks + plan only); symmetric to `emit_plan`. |
| `--plan.n_task_rephrasings` | `10` | How many `task_aug` rephrasings to emit (`0` disables). |
| `--plan.derive_task_from_video` | `if_short` | Use the dataset task as-is (`off`), only when it's missing/short (`if_short`), or always re-derive from video (`always`). |
| Flag | Default | What it does |
| ------------------------------- | ---------- | ------------------------------------------------------------------------------------------------------------------------- |
| `--plan.frames_per_second` | `2.0` | Frame sampling rate for the contact sheets (`2.0` = one frame every 0.5s). |
| `--plan.max_frames_per_prompt` | `60` | Frame budget per VLM call. Episodes whose sampling exceeds this are auto-windowed at the same density, then stitched. |
| `--plan.contact_sheet_columns` | `5` | Columns per contact-sheet grid (`contact_sheet_frames_per_sheet` tiles, time row-major). |
| `--plan.plan_max_steps` | `8` | Upper bound on subtasks per episode. |
| `--plan.subtask_describe_first` | `true` | Run the describe→segment grounding pass (best subtask quality; +1 call/episode). |
| `--plan.emit_plan` | `true` | Emit the numbered `plan` rows (`false` = subtasks + memory only). |
| `--plan.emit_memory` | `true` | Emit the `memory` rows (`false` = subtasks + plan only); symmetric to `emit_plan`. |
| `--plan.n_task_rephrasings` | `10` | How many `task_aug` rephrasings to emit (`0` disables). |
| `--plan.derive_task_from_video` | `if_short` | Use the dataset task as-is (`off`), only when it's missing/short (`if_short`), or always re-derive from video (`always`). |
### Interjections + VQA
+14 -16
View File
@@ -150,14 +150,14 @@ class MyPolicy(PreTrainedPolicy):
The methods called by the train/eval loops:
| Method | Used by | What it does |
| ----------------------------------------------------------------- | ----------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `reset() -> None` | `lerobot-eval` | Clear per-episode state at the start of each episode. |
| `select_action(batch, **kwargs) -> Tensor` | `lerobot-eval` | Return the next action `(B, action_dim)`. Called every step. |
| `predict_action_chunk(batch, **kwargs) -> Tensor` | the policy itself | Return an action chunk `(B, chunk_size, action_dim)`. Currently abstract on the base class — raise `NotImplementedError` if your policy doesn't chunk. |
| `forward(batch, reduction="mean") -> tuple[Tensor, dict \| None]` | `lerobot-train` | Return `(loss, output_dict)`. Accept `reduction="none"` if you want to support per-sample weighting. |
| `get_optim_params() -> dict` | the optimizer | Return `self.parameters()` for simple policies; return a named parameter dict for multi-optimizer policies (see `get_optim_params` in [`modeling_act.py`](https://github.com/huggingface/lerobot/blob/main/src/lerobot/policies/act/modeling_act.py) for a per-group learning-rate example). |
| `update() -> None` _(optional)_ | `lerobot-train` | Called after each optimizer step _if defined_. Use for EMA, target nets, replay buffers (TDMPC uses this). |
| Method | Used by | What it does |
| ----------------------------------------------------------------- | ----------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `reset() -> None` | `lerobot-eval` | Clear per-episode state at the start of each episode. |
| `select_action(batch, **kwargs) -> Tensor` | `lerobot-eval` | Return the next action `(B, action_dim)`. Called every step. |
| `predict_action_chunk(batch, **kwargs) -> Tensor` | the policy itself | Return an action chunk `(B, chunk_size, action_dim)`. Currently abstract on the base class — raise `NotImplementedError` if your policy doesn't chunk. |
| `forward(batch, reduction="mean") -> tuple[Tensor, dict \| None]` | `lerobot-train` | Return `(loss, output_dict)`. Accept `reduction="none"` if you want to support per-sample weighting. |
| `get_optim_params() -> dict` | the optimizer | Return `self.parameters()` for simple policies; return a named parameter dict for [multi-optimizer policies](https://github.com/huggingface/lerobot/blob/ecd38c50d7d15b4184cf42649ff1185ee2e11eeb/src/lerobot/policies/sac/modeling_sac.py#L61-L73). |
| `update() -> None` _(optional)_ | `lerobot-train` | Called after each optimizer step _if defined_. Use for EMA, target nets, replay buffers (TDMPC uses this). |
Batches are flat dictionaries keyed by the constants in [`lerobot.utils.constants`](https://github.com/huggingface/lerobot/blob/main/src/lerobot/utils/constants.py): `OBS_STATE` (`observation.state.<motor>`), `OBS_IMAGES` (`observation.images.<camera>`), `OBS_LANGUAGE`, `ACTION`, etc. Reuse the constants — don't invent new prefixes.
@@ -295,10 +295,12 @@ The file names are load-bearing: the factory does lazy imports by name, and the
### Wiring
Two places need to know about your policy. All by name.
Four places need to know about your policy. All by name.
1. **`policies/__init__.py`** — re-export `MyPolicyConfig` and add it to `__all__`. This import is what registers your policy: `@PreTrainedConfig.register_subclass("my_policy")` runs, and from then on the factory resolves everything by convention. **Don't** re-export the modeling class; it loads lazily through the factory (so `import lerobot` stays fast).
2. **`templates/lerobot_modelcard_template.md` and the root `README.md`** — the template is what `push_model_to_hub` renders into the model card of every checkpoint trained with your policy: add a one-line description of your policy in the `model_name` branches, map it in `policy_docs` so cards link to your MDX guide, and optionally add an architecture image to `diagrams`. Then add your policy to the models table in the root `README.md`, under the right category, linking to your doc page.
1. **`policies/__init__.py`** — re-export `MyPolicyConfig` and add it to `__all__`. **Don't** re-export the modeling class; it loads lazily through the factory (so `import lerobot` stays fast).
2. **`factory.py:get_policy_class`** — add a branch returning `MyPolicy` from a lazy import.
3. **`factory.py:make_policy_config`** and **`factory.py:make_pre_post_processors`** — same idea, two more branches.
4. **`templates/lerobot_modelcard_template.md` and the root `README.md`** — the template is what `push_model_to_hub` renders into the model card of every checkpoint trained with your policy: add a one-line description of your policy in the `model_name` branches, map it in `policy_docs` so cards link to your MDX guide, and optionally add an architecture image to `diagrams`. Then add your policy to the models table in the root `README.md`, under the right category, linking to your doc page.
Mirror an existing policy that's structurally similar to yours; the diff is small.
@@ -330,10 +332,6 @@ This way:
Add a matching extra to [`pyproject.toml`](https://github.com/huggingface/lerobot/blob/main/pyproject.toml) `[project.optional-dependencies]` and include it in the `all` extra so `pip install 'lerobot[all]'` keeps installing everything.
### Avoid copying a modeling file — subclass it
If your policy needs to modify a backbone that already exists in `transformers` (custom conditioning, extra inputs, a swapped sub-module), **do not vendor a copy of its `modeling_*.py`**. Instead, subclass the smallest upstream unit and override only what changes. [`pi_gemma.py`](https://github.com/huggingface/lerobot/blob/main/src/lerobot/policies/pi_gemma.py) is the canonical reference: it injects AdaRMS conditioning into PaliGemma/Gemma in ~370 lines by subclassing `GemmaModel`/`PaliGemmaModel` and overriding the decoder-layer forward, instead of forking the ~2,000-line modeling file. Model surgery on a _loaded_ native model is also fine (layer truncation, tokenizer expansion, hidden-state capture — see `evo1/internvl3_embedder.py`, `eo1/modeling_eo1.py`, `groot/groot_n1_7.py` for working examples). Reviewers will ask for this pattern when a PR arrives with a copied modeling file; the only accepted exception is a model that does not exist in `transformers` at all.
### Benchmarks and a published checkpoint
A new policy is much easier to review — and far more useful — when it ships with a working checkpoint and at least one number you can reproduce.
@@ -369,7 +367,7 @@ If your policy is real-robot-only and no sim benchmark applies, swap the sim eva
The general expectations are in [`CONTRIBUTING.md`](https://github.com/huggingface/lerobot/blob/main/CONTRIBUTING.md) and the [PR template](https://github.com/huggingface/lerobot/blob/main/.github/PULL_REQUEST_TEMPLATE.md). On top of those, reviewers will look for:
- [ ] `MyPolicy` and `MyPolicyConfig` cover the surface above; `__init_subclass__` accepts the class.
- [ ] `policies/__init__.py` re-exports the config (this registers the policy; the factory resolves modeling/processor by naming convention).
- [ ] `factory.py` and `policies/__init__.py` are wired (lazy imports for modeling).
- [ ] `make_my_policy_pre_post_processors` follows the naming convention.
- [ ] Optional deps live behind a `[project.optional-dependencies]` extra and the `TYPE_CHECKING + require_package` guard.
- [ ] `tests/policies/` updated; backward-compat artifact committed & policy-specific tests.
+7 -6
View File
@@ -228,12 +228,13 @@ lerobot-rollout \
--device=cuda
```
| Flag | Description |
| ------------------------------------------- | -------------------------------------------------------------- |
| `--inference.rtc.execution_horizon` | Steps to blend with previous chunk (default: varies by policy) |
| `--inference.rtc.max_guidance_weight` | Consistency enforcement strength (default: varies by policy) |
| `--inference.rtc.prefix_attention_schedule` | Blend schedule: `LINEAR`, `EXP`, `ONES`, `ZEROS` |
| `--inference.queue_threshold` | Max queue size before backpressure (default: 30) |
| Flag | Description |
| ------------------------------------------- | ------------------------------------------------------------------------------- |
| `--inference.rtc.execution_horizon` | Steps to blend with previous chunk (default: varies by policy) |
| `--inference.rtc.mode` | `guided` (default) or trained-prefix `trained` for compatible Pi0.5 checkpoints |
| `--inference.rtc.max_guidance_weight` | Consistency enforcement strength (default: varies by policy) |
| `--inference.rtc.prefix_attention_schedule` | Blend schedule: `LINEAR`, `EXP`, `ONES`, `ZEROS` |
| `--inference.queue_threshold` | Backpressure threshold; trained RTC requires at least its maximum delay |
See the [Real-Time Chunking](./rtc) guide for details on tuning RTC parameters.
+57 -1
View File
@@ -1,6 +1,6 @@
# Real-Time Chunking (RTC)
Real-Time Chunking (RTC) is an inference-time method that allows large, flow-matching based robotic policies, such as [Pi0](./pi0), [Pi0.5](./pi05), and [SmolVLA](./smolvla), to produce smooth, continuous, and reactive motion despite having high inference latency.
Real-Time Chunking (RTC) allows large, flow-matching based robotic policies, such as [Pi0](./pi0), [Pi0.5](./pi05), and [SmolVLA](./smolvla), to produce smooth, continuous, and reactive motion despite having high inference latency. LeRobot provides the original inference-time guided mode and, for compatible Pi0.5 checkpoints, training-time action conditioning with cheap hard-prefix inference.
These policies generate chunks of future actions (e.g., 50 steps at a time) instead of single actions.
Because the models are large, producing each chunk takes longer than the time it takes the robot to execute it.
@@ -92,6 +92,15 @@ for step in range(num_steps):
`RTCConfig` has the following parameters to tune:
**`mode`** selects the action-prefix conditioning method:
- `guided` (default) applies the original Jacobian guidance during denoising and works with ordinary flow-matching checkpoints.
- `trained` hard-inpaints the previous chunk's prefix with per-action flow timesteps. It currently requires a Pi0.5 checkpoint trained with `policy.rtc_training_max_delay > 0` and avoids the guidance backward pass.
For trained mode, both `execution_horizon` and the rollout backend's
`inference.queue_threshold` must be at least the checkpoint's
`rtc_training_max_delay`; rollout validates this before connecting the robot.
**`execution_horizon`**: How many timesteps from the previous chunk to maintain consistency with. Higher values mean smoother transitions but potentially less reactivity.
Typical values: 8-12 steps
@@ -111,6 +120,27 @@ RTCConfig(execution_horizon=10)
**`inference_delay`**: How many timesteps of inference latency your system has. This is passed to `predict_action_chunk()` rather than the config, since it may vary at runtime.
## Training Pi0.5 for Trained RTC
Set the maximum prefix delay when fine-tuning Pi0.5:
```bash
lerobot-train \
--policy.type=pi05 \
--policy.pretrained_path=lerobot/pi05_base \
--policy.rtc_training_max_delay=15 \
--dataset.repo_id=${HF_USERNAME}/dataset_repo_id \
--output_dir=outputs/pi05_training_rtc
```
Each example samples a clean prefix from zero through the configured maximum;
the flow loss is computed only on the remaining postfix. Choose the maximum as
approximately `ceil(p95 end-to-end inference latency * control frequency)` and
keep it smaller than `policy.chunk_size`. Setting the value to zero preserves
ordinary Pi0.5 training and existing checkpoints remain compatible with guided
RTC. At a 50 Hz control rate, 15 steps cover up to 300 ms of end-to-end
inference latency.
## Testing RTC Offline
Before running on a real robot, test RTC with dataset samples to visualize how it works:
@@ -124,6 +154,10 @@ python examples/rtc/eval_dataset.py \
--device=cuda
```
Add `--rtc.mode=trained` when evaluating a compatible training-time RTC Pi0.5
checkpoint. Unsupported policies reject trained mode instead of falling back to
guided RTC.
The script generates a visualization of the denoising process, comparing standard generation (left) with RTC (right). In the RTC plots, you can see how the first few steps (blue/purple lines) are guided to match the red ground truth trajectory (previous chunk's tail), ensuring a smooth transition between chunks.
<p align="center">
@@ -141,6 +175,7 @@ lerobot-rollout \
--strategy.type=base \
--policy.path=${HF_USERNAME}/policy_repo_id \
--inference.type=rtc \
--inference.rtc.mode=guided \
--inference.rtc.execution_horizon=10 \
--inference.rtc.max_guidance_weight=10.0 \
--robot.type=so100_follower \
@@ -151,6 +186,25 @@ lerobot-rollout \
--device=cuda
```
For a training-time RTC Pi0.5 checkpoint, change the mode to `trained`. The
checkpoint records its maximum supported delay, and rollout validates measured
latency against it:
```bash
lerobot-rollout \
--strategy.type=base \
--policy.path=${HF_USERNAME}/pi05_training_rtc \
--inference.type=rtc \
--inference.rtc.mode=trained \
--inference.rtc.execution_horizon=15 \
--inference.queue_threshold=15 \
--robot.type=so100_follower \
--robot.port=/dev/tty.usbmodem58FA0834591 \
--task="Move green small object into the purple platform" \
--duration=120 \
--device=cuda
```
## How It Differs from the Async Inference in LeRobot
Both RTC and [async inference](./async) improve real-time robot control, but they solve different problems.
@@ -189,3 +243,5 @@ See `examples/rtc/eval_dataset.py` for a complete example of offline RTC visuali
- [Smooth-As-Butter Robot Policies](https://alexander-soare.github.io/robotics/2025/08/05/smooth-as-butter-robot-policies.html) - Excellent technical explanation with real robot results
- [Physical Intelligence - Real-Time Chunking](https://www.physicalintelligence.company/research/real_time_chunking) - Original paper and research
- [Kinetix RTC Implementation](https://github.com/Physical-Intelligence/real-time-chunking-kinetix) - Reference implementation from Physical Intelligence
- [Training-Time Action Conditioning](https://arxiv.org/abs/2512.05964) - Efficient RTC with clean-prefix conditioning during training
- [RLDX-1](https://github.com/RLWRLD/RLDX-1) - PyTorch reference used for the training-time RTC integration
+1 -4
View File
@@ -46,11 +46,8 @@ CMD = (
"apt-get update -qq && apt-get install -y -qq git ffmpeg && "
"pip install --no-deps "
"'lerobot @ git+https://github.com/huggingface/lerobot.git@main' && "
# Pins mirror pyproject.toml — unpinned installs pull av 18 / datasets 5 /
# draccus 0.11, which break lerobot at import time.
"pip install --upgrade-strategy only-if-needed "
"'datasets>=4.7.0,<5.0.0' 'pyarrow>=21.0.0,<30.0.0' 'av>=15.0.0,<16.0.0' 'draccus==0.10.0' "
"'pandas>=2.0.0,<3.0.0' jsonlines gymnasium torchcodec mergedeep pyyaml-include toml typing-inspect "
"datasets pyarrow av jsonlines draccus gymnasium torchcodec mergedeep pyyaml-include toml typing-inspect "
"openai && "
"export VLLM_MEMORY_PROFILER_ESTIMATE_CUDAGRAPHS=0 && "
"export VLLM_VIDEO_BACKEND=pyav && "
+1
View File
@@ -306,6 +306,7 @@ class RTCEvaluator:
# Configure RTC
rtc_config = RTCConfig(
enabled=rtc_enabled,
mode=self.cfg.rtc.mode,
execution_horizon=self.cfg.rtc.execution_horizon,
max_guidance_weight=self.cfg.rtc.max_guidance_weight,
prefix_attention_schedule=self.cfg.rtc.prefix_attention_schedule,
@@ -65,14 +65,6 @@ class PlanConfig:
# invented from the task text (+1 VLM call/episode).
subtask_describe_first: bool = True
# Seeded relabeling: after segmentation, re-label each span with a focused
# pass that sees the previous / current / next segment contact sheets and
# minimally corrects the seed label (macrodata's best end-to-end labeling
# step). Costs +1 VLM call per subtask; off by default.
subtask_seeded_relabel: bool = False
# Frames sampled uniformly per segment sheet in the relabel pass.
subtask_relabel_frames: int = 5
# Emit ``style="plan"`` rows at each boundary; False = subtasks + memory only.
emit_plan: bool = True
@@ -168,11 +160,6 @@ class VlmConfig:
# Forwarded as extra_body.chat_template_kwargs (e.g. {"enable_thinking": false}).
chat_template_kwargs: dict[str, Any] | None = None
# OpenAI-style thinking budget hint ("low"/"medium"/"high"); forwarded to
# the server when set. Used to cap a thinking model's reasoning so it
# leaves tokens for the actual JSON answer on OpenAI-compatible endpoints.
reasoning_effort: str | None = None
@dataclass
class ExecutorConfig:
@@ -413,16 +413,7 @@ def _draw_timestamp_badge(image: PIL.Image.Image, timestamp: float) -> PIL.Image
result = image.copy()
draw = ImageDraw.Draw(result)
# Scale the timestamp to the tile so it stays legible after the model
# downsamples the full sheet into 768px tiles — a tiny bitmap font blurs
# at contact-sheet resolution and the VLM can no longer read the exact
# source time, which is what the boundary score depends on. ``size=`` is
# supported by Pillow's bitmap default since 10.1; fall back otherwise.
badge_px = max(14, round(image.height * 0.12))
try:
font = ImageFont.load_default(size=badge_px)
except TypeError:
font = ImageFont.load_default()
font = ImageFont.load_default()
label = f"{timestamp:06.2f}s"
left, top, right, bottom = draw.textbbox((0, 0), label, font=font)
text_w, text_h = right - left, bottom - top
@@ -116,8 +116,6 @@ class PlanSubtasksMemoryModule:
rows.extend(self._task_aug_rows([effective_task, *variants], t0))
subtask_spans = self._generate_subtasks(record, task=effective_task)
if self.config.subtask_seeded_relabel and subtask_spans:
subtask_spans = self._seeded_relabel(record, subtask_spans, effective_task)
# subtask rows
for span in subtask_spans:
@@ -511,51 +509,6 @@ class PlanSubtasksMemoryModule:
return cleaned
def _seeded_relabel(
self, record: EpisodeRecord, spans: list[dict[str, Any]], task: str
) -> list[dict[str, Any]]:
"""Re-label each span using prev/current/next segment contact sheets.
Boundaries are kept fixed; only ``text`` is refined. The original
("seed") label is passed as a strong prior so the model verifies and
minimally corrects it rather than re-describing from scratch — the
macrodata seeded-relabeling step. One VLM call per span.
"""
n = len(spans)
out: list[dict[str, Any]] = []
for i, span in enumerate(spans):
content: list[dict[str, Any]] = []
if i > 0:
content += self._segment_sheet(record, spans[i - 1])
content += self._segment_sheet(record, span)
if i < n - 1:
content += self._segment_sheet(record, spans[i + 1])
prompt = load_prompt("plan_subtask_relabel").format(
episode_task=task,
seed_label=span["text"],
segment_index=i + 1,
segment_count=n,
start=float(span["start"]),
end=float(span["end"]),
)
content.append({"type": "text", "text": prompt})
label = self._vlm_field([{"role": "user", "content": content}], "label")
text = label.strip() if isinstance(label, str) and label.strip() else span["text"]
out.append({**span, "text": text})
return out
def _segment_sheet(self, record: EpisodeRecord, span: dict[str, Any]) -> list[dict[str, Any]]:
"""Contact-sheet block(s) for one span: up to N frames sampled uniformly."""
s, e = float(span["start"]), float(span["end"])
n = max(1, int(self.config.subtask_relabel_frames))
if e <= s or n == 1:
timestamps = [s]
else:
step = (e - s) / (n - 1)
timestamps = [s + i * step for i in range(n)]
frames = self.frame_provider.frames_at(record, timestamps)
return self._contact_sheet_blocks(frames, timestamps[: len(frames)])
def _generate_subtasks_windowed(
self, record: EpisodeRecord, task: str, window_s: float
) -> list[dict[str, Any]]:
@@ -22,23 +22,12 @@ plain editors and roundtrip cleanly through ``ruff format``.
from __future__ import annotations
import os
from pathlib import Path
_DIR = Path(__file__).parent
def load(name: str) -> str:
"""Read prompt template ``name.txt`` from the ``prompts/`` directory.
A ``LEROBOT_PROMPT_OVERRIDE_<name>`` environment variable, when set to a
non-empty value, takes precedence over the packaged file. This lets prompt
search (e.g. GEPA) inject candidate templates into a remote job without
rebuilding the package; the override must keep the same ``{placeholder}``
fields the call site formats in.
"""
override = os.environ.get(f"LEROBOT_PROMPT_OVERRIDE_{name}")
if override and override.strip():
return override
"""Read prompt template ``name.txt`` from the ``prompts/`` directory."""
path = _DIR / f"{name}.txt"
return path.read_text(encoding="utf-8")
@@ -1,35 +0,0 @@
Annotate one fixed segment from a longer robot demonstration.
Return only JSON:
{{"label": "<short descriptive subtask label>"}}
You are shown up to three timestamped contact sheets, in order:
- The FIRST sheet is the PREVIOUS segment (context only); it may be absent.
- The SECOND sheet is the CURRENT target segment.
- The THIRD sheet is the NEXT segment (context only); it may be absent.
Each tile has its timestamp (seconds, absolute video time) burned into its
top-left corner.
Episode instruction: "{episode_task}"
Target segment: {segment_index} of {segment_count}
Target time: {start:.2f}s to {end:.2f}s
Original predicted label for this exact segment: "{seed_label}"
Rules:
- Label ONLY the current target segment (the second sheet). Use the
previous/next sheets only to disambiguate what changed.
- Treat the original predicted label as a STRONG PRIOR, not ground truth:
verify it against the current segment and correct it minimally.
- If it already names the right action and main object, keep it; only fix
grammar or add a clearly visible essential detail.
- If it is vague but directionally correct, make it more specific.
- If it describes the previous/next segment, the wrong action, wrong
object, wrong destination, or a wrong state change, replace it.
- Do not describe the previous or next segment, and do not split, merge,
or move the fixed segment.
- Do not introduce an action that is not clearly visible in the current
target segment.
- Use one concise imperative phrase. Name the manipulated object and the
action / state change. Include source, destination, side, direction,
final placement, or opened/closed state when visible and central.
- Do not mention timestamps, frame numbers, uncertainty, or intent.
@@ -1,68 +1,112 @@
You are annotating a teleoperated robot demonstration shown as
timestamped contact sheets (each tile has its time in seconds burned
into the top-left corner). The operator's goal was: "{episode_task}"
You are labeling a teleoperated robot demonstration.
{observation_block}Reconstruct the sequence of COMPLETED manipulation events the robot
performs, in chronological order. Output one segment per event with a
[start, end] time in seconds and a short action label.
The user originally asked: "{episode_task}"
GROUNDING — read first, it overrides everything below:
- Label ONLY events you can SEE in the frames. The instruction is the
goal; the VIDEO is the ground truth for what actually happened.
- Do NOT invent, anticipate, or pad steps that are not shown.
You are shown the entire demonstration as a single video. Watch the
whole clip, then segment it into a list of consecutive atomic subtasks
the robot performs.
Granularity — segment by completed events, not by motion:
- Start a NEW segment whenever the world state changes: an object is
grasped, lifted, transported, placed, or released; a held object
changes; a drawer/door/lid/container opens or closes; contents move
between containers (poured); a tool starts or stops acting on a
surface. Watch the gripper open/close transitions — they usually mark
boundaries.
- Do NOT split approach, reach, grasp adjustment, small repositioning,
hesitation, or retreat into their own segments. Fold each into the
event it belongs to (the approach is part of the pick; the retreat is
part of the place).
- Do NOT merge separate completed events. Each distinct pick, place,
open, close, pour, push, wipe, or insert is its own segment, even when
they repeat on different objects or locations.
- Most segments last 2-10 seconds. Shorter segments are okay ONLY for
fast pick / place / open / close / release events. Never emit a
segment shorter than {min_subtask_seconds} seconds; merge a too-short
candidate into its neighbour instead.
- Skip idle time, pure camera motion, and tiny hand jitter.
{observation_block}GROUNDING — read this first, it overrides everything below:
- Label ONLY what the robot actually does in the video. Every subtask
you emit must correspond to motion you can SEE in specific frames.
- Do NOT invent, anticipate, or pad. If the robot only does one thing
(e.g. it just navigates to a location and the clip ends), emit
EXACTLY ONE subtask. Many demonstrations are a single atomic skill.
- ``max_steps`` below is a hard CEILING, not a target. Emitting fewer
subtasks than the ceiling is not just allowed, it is expected for
short / atomic demonstrations. One correct subtask is far better
than several invented ones.
- If the video does not clearly show the action implied by the task,
describe what you actually see — do NOT fabricate the task's steps
from the instruction text. The instruction tells you the goal; the
VIDEO is the ground truth for what happened.
Labels — short imperative phrases:
- One concise command naming the action and the manipulated object, e.g.
"pick up the red cup", "put the cup on the shelf", "open the top
drawer", "pour water into the glass", "insert the plug into the
socket".
- Include source, destination, side, direction, or the final
open/closed state when it is visible and central to the event.
- Prefer these verbs (extend only when none fits): pick up, put, place,
push, pull, turn, press, open, close, pour, insert, wipe, stack.
Disambiguate by what you SEE:
* STACK vs PUT: object placed ON TOP OF another object -> "stack".
* INSERT vs PUT: object pushed INTO a fitted slot/hole/socket -> "insert".
* PICK UP vs PUT (direction): gripper CLOSES and object moves WITH
the hand -> "pick up"; gripper OPENS and object stays -> "put".
* POUR vs PUT: source is tilted and contents flow -> "pour".
- Use the exact object nouns implied by the task; stay consistent across
the episode (don't switch "cube" to "block").
- Write imperative commands, never third person ("the robot ..."), and
drop articles/adverbs.
Authoring rules — Hi Robot atom granularity, pi0.7-style short prompts:
Timing:
- Use the burned-in timestamps to set start and end. Boundaries should
land on or near a printed time, and every [start, end] must lie within
[0.0, {episode_duration}] seconds, be non-overlapping, and cover the
episode in order.
- Emit at most {max_steps} segments.
- Each subtask = one COMPOSITE atomic skill the low-level policy can
execute end-to-end. A "skill" bundles its own approach motion with
its terminal action — do NOT split the approach off as its own
subtask. The whole-arm policy already learns to reach as part of
every manipulation primitive.
- Write each subtask as an IMPERATIVE COMMAND, starting with one of
these verbs (extend only when none fits):
pick up <obj> — approach + grasp + lift in one subtask
put <obj> on/in <loc> — transport + release in one subtask
place <obj> on/in <loc> — synonym of "put"; pick one and stay consistent
push <obj> — contact + linear shove
pull <obj> — contact + linear retract
turn <knob/dial/handle> — rotary actuation
press <button> — single-press contact
open <drawer/door/lid> — full open motion
close <drawer/door/lid> — full close motion
pour <src> into <dst> — tilt + flow
insert <obj> into <slot>— alignment + push-fit
go to <loc> — ONLY when no grasp / actuation follows
(e.g. a pure relocation between phases).
If the next subtask grasps something at
that location, drop "go to ..." and just
write "pick up ..." instead.
- Forbidden ultra-fine splits — the VLM is NOT allowed to emit these
as standalone subtasks; fold them into the parent composite:
"move to X" → fold into "pick up X" (or whatever follows)
"reach for X" → fold into "pick up X"
"grasp X" → fold into "pick up X"
"lift X" → fold into "pick up X" (or "put X on Y" if it's
the transport phase of a place)
"release X" → fold into "put X on Y" (or "place X in Y")
- Keep it SHORT — a verb phrase, not a sentence. Drop articles
("the", "a") and adverbs ("carefully", "slowly"). Add a "how"
detail (which hand, which grasp point) ONLY when it is needed to
disambiguate. Every subtask must begin with one of the verbs
above (no leading nouns, no "then", no "first").
- NEVER use third person. Never write "the robot", "the arm", "the
gripper moves", "it picks up" — the robot is implied. Command it,
do not describe it.
- Use the exact object nouns from the task above. If the task says
"cube", every subtask says "cube" — never switch to "block". If it
says "box", never switch to "bin"/"container". Keep vocabulary
consistent across the whole episode.
- Good: "pick up blue cube", "put blue cube in box", "open drawer",
"turn red knob", "press start button", "go to sink".
- Bad: "move to blue cube" (approach as its own subtask — forbidden,
must be folded into "pick up blue cube"); "the robot arm moves
towards the blue cube" (third person, too long); "carefully pick
up the cube" (adverb, article); "release the yellow block"
("block" when the task said "cube", and "release" must be folded
into a "put"/"place" subtask).
- Subtasks are non-overlapping and cover the full episode in order.
Choose the cut points yourself based on what you see in the video
(gripper open/close events, contact, regrasps, transitions).
- Each subtask spans at least {min_subtask_seconds} seconds. If a
candidate span would be shorter, merge it into its neighbour
rather than emitting it.
- Do not exceed {max_steps} subtasks total. Fewer, larger composites
are preferred over many micro-steps.
- Every subtask's [start_time, end_time] must lie within
[0.0, {episode_duration}] seconds.
SPECIAL CASES — verb disambiguation (each rule is narrowly visual and
fires ONLY on the spatial situation it names; it must not change how you
label any other situation):
- STACK vs PUT: if an object is placed ON TOP OF another specific object
(not on a flat table / shelf / counter), use "stack ... on ...", not
"put". "stack blue book on green book", NOT "put blue book on table".
- INSERT vs PUT: if an object goes INTO a fitted slot / hole / socket /
receptacle (push-fit), use "insert ... into ...", not "put".
- RETRIEVE/PICK-UP vs PUT (direction): watch the gripper. If it CLOSES
on the object and the object moves WITH the hand, it is "pick up" /
"retrieve" (object leaves its location). If the gripper OPENS and the
object stays where the hand left it, it is "put" / "place" (object
arrives at a location). Decide by which way the object moves, not by
where the hand ends up.
- POUR vs PUT: only use "pour" when the source is tilted and contents
flow out; moving a full container without tilting is "put"/"place".
Output strictly valid JSON of shape:
{{
"subtasks": [
{{"text": "<short imperative action label>", "start": <float>, "end": <float>}},
{{"text": "<short imperative verb phrase>", "start": <float>, "end": <float>}},
...
]
}}
@@ -285,8 +285,6 @@ def _make_openai_client(config: VlmConfig) -> VlmClient:
"max_tokens": max_tok,
"temperature": temp,
}
if config.reasoning_effort:
kwargs["reasoning_effort"] = config.reasoning_effort
extra_body: dict[str, Any] = {}
if send_mm_kwargs and mm_kwargs:
extra_body["mm_processor_kwargs"] = {**mm_kwargs, "do_sample_frames": True}
@@ -298,13 +296,7 @@ def _make_openai_client(config: VlmConfig) -> VlmClient:
chosen = clients[rr_counter["i"] % len(clients)]
rr_counter["i"] += 1
response = chosen.chat.completions.create(**kwargs)
# Some OpenAI-compatible servers can return a choice with no message
# (safety filter, or a "thinking" model that spends the whole budget
# before emitting content). Treat that as an empty reply so the
# JSON-retry path handles it instead of crashing the run.
choice = response.choices[0] if response.choices else None
message = choice.message if choice is not None else None
return (message.content if message is not None else None) or ""
return response.choices[0].message.content or ""
def _gen(batch: Sequence[Sequence[dict[str, Any]]], max_tok: int, temp: float) -> list[str]:
if len(batch) <= 1 or config.client_concurrency <= 1:
+9 -15
View File
@@ -205,30 +205,24 @@ class PreTrainedConfig(draccus.ChoiceRegistry, HubMixin, abc.ABC): # type: igno
f"{CONFIG_NAME} not found on the HuggingFace Hub in {model_id}"
) from e
# HACK: Parse the original config to get the config subclass, so that we can
# apply cli overrides.
# This is very ugly, ideally we'd like to be able to do that natively with draccus
# something like --policy.path (in addition to --policy.type)
with draccus.config_type("json"):
orig_config = draccus.parse(cls, config_file, args=[])
if config_file is None:
raise FileNotFoundError(f"{CONFIG_NAME} not found in {model_id}")
with open(config_file) as f:
config = json.load(f)
# Resolve the concrete config subclass from the serialized "type" tag, then parse
# the config (with CLI overrides) directly for that class. The "type" key is
# stripped because draccus only consumes it when parsing the registry base class.
policy_type = config.pop("type", None)
if policy_type is None:
raise ValueError(f"Missing 'type' field in {CONFIG_NAME} of {model_id}")
try:
config_cls = cls.get_choice_class(policy_type)
except Exception as e:
raise ValueError(
f"Policy type '{policy_type}' (from {CONFIG_NAME} of {model_id}) is not registered. "
f"Available policy types: {cls.get_known_choices()}"
) from e
config.pop("type")
with tempfile.NamedTemporaryFile("w+", delete=False, suffix=".json") as f:
json.dump(config, f)
config_file = f.name
cli_overrides = policy_kwargs.pop("cli_overrides", [])
with draccus.config_type("json"):
return draccus.parse(config_cls, config_file, args=cli_overrides)
return draccus.parse(orig_config.__class__, config_file, args=cli_overrides)
+93 -5
View File
@@ -18,8 +18,13 @@ from __future__ import annotations
import logging
import numpy as np
import torch
from lerobot.processor import RelativeActionsProcessorStep
from lerobot.processor import (
RelativeActionsProcessorStep,
relative_action_output_dim,
to_relative_actions,
)
from lerobot.utils.constants import ACTION, OBS_STATE
from .io_utils import load_image_as_numpy
@@ -660,17 +665,29 @@ def _compute_relative_chunk_batch(
all_states: np.ndarray,
chunk_size: int,
relative_mask: np.ndarray,
pose_representation: str = "componentwise",
se3_pose_groups: list[list[int]] | None = None,
) -> np.ndarray:
"""Vectorised relative-action computation for a batch of start indices.
Returns an ``(N * chunk_size, action_dim)`` float32 array.
Returns an ``(N * chunk_size, model_action_dim)`` float32 array.
"""
if len(start_indices) == 0:
return np.empty((0, all_actions.shape[1]), dtype=np.float32)
output_dim = relative_action_output_dim(all_actions.shape[1], pose_representation, se3_pose_groups)
return np.empty((0, output_dim), dtype=np.float32)
offsets = np.arange(chunk_size)
frame_idx = start_indices[:, None] + offsets[None, :]
chunks = all_actions[frame_idx].copy()
states = all_states[start_indices]
if pose_representation in {"se3", "se3_6d"}:
converted = to_relative_actions(
torch.from_numpy(chunks),
torch.from_numpy(states),
relative_mask.astype(bool).tolist(),
pose_representation=pose_representation,
se3_pose_groups=se3_pose_groups,
)
return converted.numpy().reshape(-1, converted.shape[-1])
mask_dim = len(relative_mask)
chunks[:, :, :mask_dim] -= states[:, None, :mask_dim] * relative_mask[None, None, :]
return chunks.reshape(-1, all_actions.shape[1])
@@ -682,6 +699,9 @@ def compute_relative_action_stats(
chunk_size: int,
exclude_joints: list[str] | None = None,
num_workers: int = 0,
state_from_action: bool = False,
pose_representation: str = "componentwise",
se3_pose_groups: list[list[int]] | None = None,
) -> dict[str, np.ndarray]:
"""Compute normalization statistics for relative actions over the full dataset.
@@ -700,6 +720,9 @@ def compute_relative_action_stats(
num_workers: Number of parallel threads for computation. Values ≤1
mean single-threaded. Numpy releases the GIL so threads give
real parallelism here.
state_from_action: Use the current absolute action as state. This is
intended for state-less pose datasets where each action row is the
synchronized measured robot pose.
Returns:
Statistics dict with keys "mean", "std", "min", "max", "q01", …, "q99".
@@ -722,7 +745,7 @@ def compute_relative_action_stats(
logging.info("Loading action/state data for relative action stats...")
all_actions = np.array(hf_dataset[ACTION], dtype=np.float32)
all_states = np.array(hf_dataset[OBS_STATE], dtype=np.float32)
all_states = all_actions if state_from_action else np.array(hf_dataset[OBS_STATE], dtype=np.float32)
episode_indices = np.array(hf_dataset["episode_index"])
valid_starts = _get_valid_chunk_starts(episode_indices, chunk_size)
@@ -754,6 +777,8 @@ def compute_relative_action_stats(
all_states,
chunk_size,
relative_mask,
pose_representation,
se3_pose_groups,
)
for batch in batches
]
@@ -762,7 +787,15 @@ def compute_relative_action_stats(
else:
for batch in batches:
running_stats.update(
_compute_relative_chunk_batch(batch, all_actions, all_states, chunk_size, relative_mask)
_compute_relative_chunk_batch(
batch,
all_actions,
all_states,
chunk_size,
relative_mask,
pose_representation,
se3_pose_groups,
)
)
stats = running_stats.get_statistics()
@@ -777,3 +810,58 @@ def compute_relative_action_stats(
)
return stats
def compute_state_history_stats(
hf_dataset,
features: dict,
history_steps: int,
exclude_joints: list[str] | None = None,
relative: bool = False,
pose_representation: str = "componentwise",
se3_pose_groups: list[list[int]] | None = None,
) -> dict[str, np.ndarray]:
"""Compute stats for flattened state history synthesized from absolute actions.
History is left-padded with the first action of each episode, matching dataset
boundary padding. When ``relative`` is enabled, every history pose is expressed
relative to its newest pose while excluded dimensions remain absolute.
"""
if history_steps < 1:
raise ValueError("history_steps must be at least 1")
if exclude_joints is None:
exclude_joints = []
actions = np.asarray(hf_dataset[ACTION], dtype=np.float32)
episode_indices = np.asarray(hf_dataset["episode_index"])
sample_indices = np.arange(len(actions))
episode_starts = np.maximum.accumulate(
np.where(
np.concatenate(([True], episode_indices[1:] != episode_indices[:-1])),
sample_indices,
0,
)
)
offsets = np.arange(-(history_steps - 1), 1)
history_indices = np.maximum(sample_indices[:, None] + offsets[None, :], episode_starts[:, None])
history = actions[history_indices].copy()
if relative:
state_dim = actions.shape[-1]
names = features.get(ACTION, {}).get("names")
mask_step = RelativeActionsProcessorStep(
enabled=True,
exclude_joints=exclude_joints,
action_names=names,
)
mask = mask_step._build_mask(state_dim)
history = to_relative_actions(
torch.from_numpy(history),
torch.from_numpy(history[:, -1].copy()),
mask,
pose_representation=pose_representation,
se3_pose_groups=se3_pose_groups,
).numpy()
flattened = history.reshape(len(history), -1)
return get_feature_stats(flattened, axis=0, keepdims=False)
+37 -1
View File
@@ -54,6 +54,7 @@ from .compute_stats import (
aggregate_stats,
compute_episode_stats,
compute_relative_action_stats,
compute_state_history_stats,
)
from .dataset_metadata import LeRobotDatasetMetadata
from .image_writer import write_image
@@ -1566,6 +1567,12 @@ def recompute_stats(
relative_exclude_joints: list[str] | None = None,
chunk_size: int = 50,
num_workers: int = 0,
state_from_action: bool = False,
state_history_steps: int = 1,
relative_state_history: bool = False,
relative_state_exclude_joints: list[str] | None = None,
relative_pose_representation: str = "componentwise",
relative_se3_pose_groups: list[list[int]] | None = None,
) -> LeRobotDataset:
"""Recompute stats.json from scratch by iterating all episodes.
@@ -1583,6 +1590,16 @@ def recompute_stats(
``policy.chunk_size``. Only used when ``relative_action=True``.
num_workers: Number of parallel threads for relative action stats computation.
Values ≤1 mean single-threaded. Only used when ``relative_action=True``.
state_from_action: Use absolute action rows as synthetic state while
computing relative-action stats, and write their absolute statistics
under ``observation.state``.
state_history_steps: Number of consecutive synthesized state samples.
relative_state_history: Express state history relative to its newest pose.
relative_state_exclude_joints: State dimensions to retain as absolute.
relative_pose_representation: ``componentwise`` for legacy subtraction,
``se3`` for composition with an axis-angle output, or ``se3_6d`` for
composition with a continuous two-column rotation output.
relative_se3_pose_groups: Six-index xyz+rotation-vector pose groups.
Returns:
The same dataset with updated stats.
@@ -1606,7 +1623,21 @@ def recompute_stats(
# (matching what the model sees during training) and skip action in the
# per-episode pass below.
relative_action_stats = None
if relative_action and ACTION in features and OBS_STATE in features:
synthetic_state_stats = None
if state_from_action:
if ACTION not in features:
raise ValueError("state_from_action requires an action feature")
synthetic_state_stats = compute_state_history_stats(
dataset.hf_dataset,
features,
history_steps=state_history_steps,
exclude_joints=relative_state_exclude_joints,
relative=relative_state_history,
pose_representation=relative_pose_representation,
se3_pose_groups=relative_se3_pose_groups,
)
if relative_action and ACTION in features and (OBS_STATE in features or state_from_action):
if relative_exclude_joints is None:
relative_exclude_joints = ["gripper"]
relative_action_stats = compute_relative_action_stats(
@@ -1615,6 +1646,9 @@ def recompute_stats(
chunk_size=chunk_size,
exclude_joints=relative_exclude_joints,
num_workers=num_workers,
state_from_action=state_from_action,
pose_representation=relative_pose_representation,
se3_pose_groups=relative_se3_pose_groups,
)
features_to_compute.pop(ACTION, None)
@@ -1654,6 +1688,8 @@ def recompute_stats(
if relative_action_stats is not None:
new_stats[ACTION] = relative_action_stats
if synthetic_state_stats is not None:
new_stats[OBS_STATE] = synthetic_state_stats
# Merge: keep existing stats for features we didn't recompute
if dataset.meta.stats:
-2
View File
@@ -32,7 +32,6 @@ from .pretrained import PreTrainedPolicy as PreTrainedPolicy
from .smolvla.configuration_smolvla import SmolVLAConfig as SmolVLAConfig
from .tdmpc.configuration_tdmpc import TDMPCConfig as TDMPCConfig
from .utils import make_robot_action, prepare_observation_for_inference
from .vla_jepa.configuration_vla_jepa import VLAJEPAConfig as VLAJEPAConfig
from .vqbet.configuration_vqbet import VQBeTConfig as VQBeTConfig
from .wall_x.configuration_wall_x import WallXConfig as WallXConfig
from .xvla.configuration_xvla import XVLAConfig as XVLAConfig
@@ -58,7 +57,6 @@ __all__ = [
"PI05Config",
"SmolVLAConfig",
"TDMPCConfig",
"VLAJEPAConfig",
"VQBeTConfig",
"WallXConfig",
"XVLAConfig",
+39 -2
View File
@@ -18,10 +18,17 @@ from typing import Any
import torch
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
make_default_pre_post_processors,
RenameObservationsProcessorStep,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
from .configuration_act import ACTConfig
@@ -47,4 +54,34 @@ def make_act_pre_post_processors(
tuple[PolicyProcessorPipeline[dict[str, Any], dict[str, Any]], PolicyProcessorPipeline[PolicyAction, PolicyAction]]: A tuple containing the
pre-processor pipeline and the post-processor pipeline.
"""
return make_default_pre_post_processors(config, dataset_stats, normalizer_device=config.device)
input_steps = [
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
DeviceProcessorStep(device=config.device),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
device=config.device,
),
]
output_steps = [
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
DeviceProcessorStep(device="cpu"),
]
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
@@ -19,10 +19,17 @@ from typing import Any
import torch
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
make_default_pre_post_processors,
RenameObservationsProcessorStep,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
from .configuration_diffusion import DiffusionConfig
@@ -56,4 +63,32 @@ def make_diffusion_pre_post_processors(
Returns:
A tuple containing the configured pre-processor and post-processor pipelines.
"""
return make_default_pre_post_processors(config, dataset_stats)
input_steps = [
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
DeviceProcessorStep(device=config.device),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
]
output_steps = [
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
DeviceProcessorStep(device="cpu"),
]
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
+37 -12
View File
@@ -23,16 +23,24 @@ import torch
from lerobot.configs.types import FeatureType, PipelineFeatureType, PolicyFeature
from lerobot.processor import (
AddBatchDimensionProcessorStep,
ComplementaryDataProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStep,
ProcessorStepRegistry,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
RenameObservationsProcessorStep,
UnnormalizerProcessorStep,
)
from lerobot.processor.converters import policy_action_to_transition, transition_to_policy_action
from lerobot.types import TransitionKey
from lerobot.utils.constants import OBS_STATE
from lerobot.utils.constants import (
OBS_STATE,
POLICY_POSTPROCESSOR_DEFAULT_NAME,
POLICY_PREPROCESSOR_DEFAULT_NAME,
)
from lerobot.utils.import_utils import _transformers_available, require_package
from .configuration_eo1 import EO1Config
@@ -234,12 +242,14 @@ def make_eo1_pre_post_processors(
]:
"""Build pre/post processor pipelines for EO1."""
steps = make_default_policy_processor_steps(config, dataset_stats)
input_steps: list[ProcessorStep] = [
steps.rename_observations,
steps.add_batch_dim,
steps.normalize,
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
EO1ConversationTemplateStep(input_features=config.input_features, chunk_size=config.chunk_size),
EO1QwenProcessorStep(
processor_name=config.vlm_base,
@@ -247,12 +257,27 @@ def make_eo1_pre_post_processors(
image_max_pixels=config.image_max_pixels,
use_fast_processor=config.use_fast_processor,
),
steps.to_device,
DeviceProcessorStep(device=config.device),
]
output_steps: list[ProcessorStep] = [
steps.unnormalize,
steps.to_cpu,
UnnormalizerProcessorStep(
features=config.output_features,
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
DeviceProcessorStep(device="cpu"),
]
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
@@ -27,11 +27,9 @@ from lerobot.utils.import_utils import _transformers_available, require_package
if TYPE_CHECKING or _transformers_available:
from transformers import AutoModel, AutoTokenizer
from transformers.utils import is_flash_attn_2_available
else:
AutoModel = None
AutoTokenizer = None
is_flash_attn_2_available = None
IMAGENET_MEAN = (0.485, 0.456, 0.406)
IMAGENET_STD = (0.229, 0.224, 0.225)
@@ -137,13 +135,9 @@ class InternVL3Embedder(nn.Module):
raise ValueError(f"Unsupported EVO1 vlm_dtype '{model_dtype}'") from exc
self.model_dtype = model_dtype
attn_implementation = (
"flash_attention_2" if (use_flash_attn and is_flash_attn_2_available()) else "eager"
)
attn_implementation = "flash_attention_2" if (use_flash_attn and _flash_attn_available()) else "eager"
if use_flash_attn and attn_implementation == "eager":
logger.warning(
"Flash Attention 2 is unavailable on this runtime. Falling back to eager attention."
)
logger.warning("flash_attn is not installed. Falling back to eager attention.")
self.model = AutoModel.from_pretrained(
model_name,
@@ -365,3 +359,11 @@ class InternVL3Embedder(nn.Module):
@property
def device(self) -> torch.device:
return next(self.model.parameters()).device
def _flash_attn_available() -> bool:
try:
import flash_attn # noqa: F401
except ModuleNotFoundError:
return False
return True
+318 -66
View File
@@ -17,7 +17,6 @@
from __future__ import annotations
import importlib
import inspect
import logging
from typing import TYPE_CHECKING, Any, TypedDict, Unpack
@@ -45,10 +44,26 @@ from lerobot.utils.constants import (
)
from lerobot.utils.feature_utils import dataset_to_policy_features
from .act.configuration_act import ACTConfig
from .diffusion.configuration_diffusion import DiffusionConfig
from .eo1.configuration_eo1 import EO1Config
from .evo1.configuration_evo1 import Evo1Config
from .fastwam.configuration_fastwam import FastWAMConfig
from .gaussian_actor.configuration_gaussian_actor import GaussianActorConfig
from .groot.configuration_groot import GrootConfig
from .lingbot_va.configuration_lingbot_va import LingBotVAConfig
from .molmoact2.configuration_molmoact2 import MolmoAct2Config
from .multi_task_dit.configuration_multi_task_dit import MultiTaskDiTConfig
from .pi0.configuration_pi0 import PI0Config
from .pi05.configuration_pi05 import PI05Config
from .pretrained import PreTrainedPolicy
from .smolvla.configuration_smolvla import SmolVLAConfig
from .tdmpc.configuration_tdmpc import TDMPCConfig
from .utils import validate_visual_features_consistency
from .vla_jepa.configuration_vla_jepa import VLAJEPAConfig
from .vqbet.configuration_vqbet import VQBeTConfig
from .wall_x.configuration_wall_x import WallXConfig
from .xvla.configuration_xvla import XVLAConfig
def _reconnect_relative_absolute_steps(
@@ -73,23 +88,100 @@ def get_policy_class(name: str) -> type[PreTrainedPolicy]:
"""
Retrieves a policy class by its registered name.
Resolution is convention-based: the draccus-registered config class of ``name`` is
looked up, its ``configuration_*`` module path is rewritten to ``modeling_*``, and
the ``<X>Policy`` class is imported from there. The modeling module is only imported
at call time, keeping heavy optional dependencies lazy. This works for both built-in
policies and third-party lerobot plugins (anything registered via
``@PreTrainedConfig.register_subclass``).
This function uses dynamic imports to avoid loading all policy classes into memory
at once, improving startup time and reducing dependencies.
Args:
name: The registered name of the policy (e.g. "act", "diffusion", "pi0").
name: The name of the policy. Supported names are "tdmpc", "diffusion", "act",
"multi_task_dit", "vqbet", "pi0", "pi05", "gaussian_actor", "smolvla", "wall_x",
"molmoact2", "eo1", "evo1".
Returns:
The policy class corresponding to the given name.
Raises:
ValueError: If the policy name is not registered.
ImportError: If the policy's optional dependencies are not installed.
NotImplementedError: If the policy name is not recognized.
"""
return _get_policy_cls_from_policy_name(name=name)
if name == "tdmpc":
from .tdmpc.modeling_tdmpc import TDMPCPolicy
return TDMPCPolicy
elif name == "diffusion":
from .diffusion.modeling_diffusion import DiffusionPolicy
return DiffusionPolicy
elif name == "act":
from .act.modeling_act import ACTPolicy
return ACTPolicy
elif name == "multi_task_dit":
from .multi_task_dit.modeling_multi_task_dit import MultiTaskDiTPolicy
return MultiTaskDiTPolicy
elif name == "vqbet":
from .vqbet.modeling_vqbet import VQBeTPolicy
return VQBeTPolicy
elif name == "pi0":
from .pi0.modeling_pi0 import PI0Policy
return PI0Policy
elif name == "pi0_fast":
from .pi0_fast.modeling_pi0_fast import PI0FastPolicy
return PI0FastPolicy
elif name == "pi05":
from .pi05.modeling_pi05 import PI05Policy
return PI05Policy
elif name == "gaussian_actor":
from .gaussian_actor.modeling_gaussian_actor import GaussianActorPolicy
return GaussianActorPolicy
elif name == "smolvla":
from .smolvla.modeling_smolvla import SmolVLAPolicy
return SmolVLAPolicy
elif name == "groot":
from .groot.modeling_groot import GrootPolicy
return GrootPolicy
elif name == "xvla":
from .xvla.modeling_xvla import XVLAPolicy
return XVLAPolicy
elif name == "wall_x":
from .wall_x.modeling_wall_x import WallXPolicy
return WallXPolicy
elif name == "eo1":
from .eo1.modeling_eo1 import EO1Policy
return EO1Policy
elif name == "molmoact2":
from .molmoact2.modeling_molmoact2 import MolmoAct2Policy
return MolmoAct2Policy
elif name == "vla_jepa":
from .vla_jepa.modeling_vla_jepa import VLAJEPAPolicy
return VLAJEPAPolicy
elif name == "lingbot_va":
from .lingbot_va.modeling_lingbot_va import LingBotVAPolicy
return LingBotVAPolicy
elif name == "fastwam":
from .fastwam.modeling_fastwam import FastWAMPolicy
return FastWAMPolicy
elif name == "evo1":
from .evo1.modeling_evo1 import Evo1Policy
return Evo1Policy
else:
try:
return _get_policy_cls_from_policy_name(name=name)
except Exception as e:
raise ValueError(f"Policy type '{name}' is not available.") from e
def make_policy_config(policy_type: str, **kwargs) -> PreTrainedConfig:
@@ -100,8 +192,9 @@ def make_policy_config(policy_type: str, **kwargs) -> PreTrainedConfig:
mapping a string identifier to the corresponding config class.
Args:
policy_type: The registered type of the policy (any name registered via
``@PreTrainedConfig.register_subclass``, e.g. "act", "diffusion", "pi0").
policy_type: The type of the policy. Supported types include "tdmpc",
"multi_task_dit", "diffusion", "act", "vqbet", "pi0", "pi05", "gaussian_actor",
"smolvla", "wall_x", "molmoact2", "eo1", "evo1".
**kwargs: Keyword arguments to be passed to the configuration class constructor.
Returns:
@@ -110,11 +203,48 @@ def make_policy_config(policy_type: str, **kwargs) -> PreTrainedConfig:
Raises:
ValueError: If the `policy_type` is not recognized.
"""
try:
config_cls = PreTrainedConfig.get_choice_class(policy_type)
except Exception as e:
raise ValueError(f"Policy type '{policy_type}' is not available.") from e
return config_cls(**kwargs)
if policy_type == "tdmpc":
return TDMPCConfig(**kwargs)
elif policy_type == "diffusion":
return DiffusionConfig(**kwargs)
elif policy_type == "act":
return ACTConfig(**kwargs)
elif policy_type == "multi_task_dit":
return MultiTaskDiTConfig(**kwargs)
elif policy_type == "vqbet":
return VQBeTConfig(**kwargs)
elif policy_type == "pi0":
return PI0Config(**kwargs)
elif policy_type == "pi05":
return PI05Config(**kwargs)
elif policy_type == "gaussian_actor":
return GaussianActorConfig(**kwargs)
elif policy_type == "smolvla":
return SmolVLAConfig(**kwargs)
elif policy_type == "groot":
return GrootConfig(**kwargs)
elif policy_type == "xvla":
return XVLAConfig(**kwargs)
elif policy_type == "wall_x":
return WallXConfig(**kwargs)
elif policy_type == "eo1":
return EO1Config(**kwargs)
elif policy_type == "molmoact2":
return MolmoAct2Config(**kwargs)
elif policy_type == "vla_jepa":
return VLAJEPAConfig(**kwargs)
elif policy_type == "lingbot_va":
return LingBotVAConfig(**kwargs)
elif policy_type == "fastwam":
return FastWAMConfig(**kwargs)
elif policy_type == "evo1":
return Evo1Config(**kwargs)
else:
try:
config_cls = PreTrainedConfig.get_choice_class(policy_type)
return config_cls(**kwargs)
except Exception as e:
raise ValueError(f"Policy type '{policy_type}' is not available.") from e
class ProcessorConfigKwargs(TypedDict, total=False):
@@ -168,7 +298,8 @@ def make_pre_post_processors(
A tuple containing the input (pre-processor) and output (post-processor) pipelines.
Raises:
ValueError: If no processor factory exists for the given policy configuration type.
NotImplementedError: If a processor factory is not implemented for the given
policy configuration type.
"""
if pretrained_path:
if isinstance(policy_cfg, GrootConfig):
@@ -220,13 +351,166 @@ def make_pre_post_processors(
)
return preprocessor, postprocessor
# Create new processors from the policy config, resolving the per-policy factory
# function by naming convention (lazy import keeps optional dependencies optional).
return _make_processors_from_policy_config(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
dataset_meta=kwargs.get("dataset_meta"),
)
# Create a new processor based on policy type
if isinstance(policy_cfg, TDMPCConfig):
from .tdmpc.processor_tdmpc import make_tdmpc_pre_post_processors
processors = make_tdmpc_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, DiffusionConfig):
from .diffusion.processor_diffusion import make_diffusion_pre_post_processors
processors = make_diffusion_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, ACTConfig):
from .act.processor_act import make_act_pre_post_processors
processors = make_act_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, MultiTaskDiTConfig):
from .multi_task_dit.processor_multi_task_dit import (
make_multi_task_dit_pre_post_processors,
)
processors = make_multi_task_dit_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, VQBeTConfig):
from .vqbet.processor_vqbet import make_vqbet_pre_post_processors
processors = make_vqbet_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, PI0Config):
from .pi0.processor_pi0 import make_pi0_pre_post_processors
processors = make_pi0_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, PI05Config):
from .pi05.processor_pi05 import make_pi05_pre_post_processors
processors = make_pi05_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, GaussianActorConfig):
from .gaussian_actor.processor_gaussian_actor import make_gaussian_actor_pre_post_processors
processors = make_gaussian_actor_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, SmolVLAConfig):
from .smolvla.processor_smolvla import make_smolvla_pre_post_processors
processors = make_smolvla_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, GrootConfig):
from .groot.processor_groot import make_groot_pre_post_processors
processors = make_groot_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
dataset_meta=kwargs.get("dataset_meta"),
)
elif isinstance(policy_cfg, XVLAConfig):
from .xvla.processor_xvla import (
make_xvla_pre_post_processors,
)
processors = make_xvla_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, WallXConfig):
from .wall_x.processor_wall_x import make_wall_x_pre_post_processors
processors = make_wall_x_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, EO1Config):
from .eo1.processor_eo1 import make_eo1_pre_post_processors
processors = make_eo1_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, Evo1Config):
from .evo1.processor_evo1 import make_evo1_pre_post_processors
processors = make_evo1_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, MolmoAct2Config):
from .molmoact2.processor_molmoact2 import make_molmoact2_pre_post_processors
processors = make_molmoact2_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
dataset_meta=kwargs.get("dataset_meta"),
)
elif isinstance(policy_cfg, VLAJEPAConfig):
from .vla_jepa.processor_vla_jepa import make_vla_jepa_pre_post_processors
processors = make_vla_jepa_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, LingBotVAConfig):
from .lingbot_va.processor_lingbot_va import make_lingbot_va_pre_post_processors
processors = make_lingbot_va_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
elif isinstance(policy_cfg, FastWAMConfig):
from .fastwam.processor_fastwam import make_fastwam_pre_post_processors
processors = make_fastwam_pre_post_processors(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
else:
try:
processors = _make_processors_from_policy_config(
config=policy_cfg,
dataset_stats=kwargs.get("dataset_stats"),
)
except Exception as e:
raise ValueError(f"Processor for policy type '{policy_cfg.type}' is not implemented.") from e
return processors
def make_policy(
@@ -370,12 +654,10 @@ def make_policy(
return policy
def _get_policy_cls_from_policy_name(name: str) -> type[PreTrainedPolicy]:
def _get_policy_cls_from_policy_name(name: str) -> type[PreTrainedConfig]:
"""Get policy class from its registered name using dynamic imports.
Works for built-in policies and 3rd party lerobot plugins alike: the config class
registered under ``name`` is resolved via the draccus ChoiceRegistry, and the policy
class is imported from the sibling ``modeling_*`` module by naming convention.
This is used as a helper function to import policies from 3rd party lerobot plugins.
Args:
name: The name of the policy.
@@ -401,39 +683,22 @@ def _get_policy_cls_from_policy_name(name: str) -> type[PreTrainedPolicy]:
"configuration_", "modeling_"
) # e.g., configuration_diffusion -> modeling_diffusion
try:
module = importlib.import_module(module_path)
except ModuleNotFoundError as e:
if e.name == module_path:
# The modeling_* module itself does not exist for this policy type. A missing
# optional dependency inside an existing module propagates unchanged instead,
# so its actionable install hint stays visible.
raise ValueError(f"Policy class for '{name}' is not implemented.") from e
raise
policy_cls = getattr(module, cls_name, None)
if policy_cls is None:
raise ValueError(
f"Policy class '{cls_name}' not found in '{module_path}'. "
f"Policies must expose '<Name>Policy' in the sibling 'modeling_*' module by naming convention."
)
module = importlib.import_module(module_path)
policy_cls = getattr(module, cls_name)
return policy_cls
def _make_processors_from_policy_config(
config: PreTrainedConfig,
dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None,
dataset_meta: Any | None = None,
) -> tuple[Any, Any]:
"""Create pre- and post-processors from a policy configuration using dynamic imports.
Resolves ``make_{type}_pre_post_processors`` from the policy's ``processor_*`` module
by naming convention. Works for built-in policies and 3rd party lerobot plugins.
This is used as a helper function to import processor factories from 3rd party lerobot plugins.
Args:
config: The policy configuration object.
dataset_stats: Dataset statistics for normalization.
dataset_meta: Dataset metadata, forwarded only to factories that declare a
``dataset_meta`` parameter (e.g. groot, molmoact2).
Returns:
A tuple containing the input (pre-processor) and output (post-processor) pipelines.
"""
@@ -446,19 +711,6 @@ def _make_processors_from_policy_config(
logging.debug(
f"Instantiating pre/post processors using function '{function_name}' from module '{module_path}'"
)
try:
module = importlib.import_module(module_path)
except ModuleNotFoundError as e:
if e.name == module_path:
# The processor_* module itself does not exist for this policy type. A missing
# optional dependency inside an existing module propagates unchanged instead,
# so its actionable install hint stays visible.
raise ValueError(f"Processor for policy type '{policy_type}' is not implemented.") from e
raise
function = getattr(module, function_name, None)
if function is None:
raise ValueError(f"Processor for policy type '{policy_type}' is not implemented.")
call_kwargs: dict[str, Any] = {"dataset_stats": dataset_stats}
if "dataset_meta" in inspect.signature(function).parameters:
call_kwargs["dataset_meta"] = dataset_meta
return function(config, **call_kwargs)
module = importlib.import_module(module_path)
function = getattr(module, function_name)
return function(config, dataset_stats=dataset_stats)
@@ -22,11 +22,20 @@ import torch
from lerobot.configs import PipelineFeatureType, PolicyFeature
from lerobot.processor import (
ActionProcessorStep,
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStepRegistry,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
RenameObservationsProcessorStep,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.utils.constants import (
POLICY_POSTPROCESSOR_DEFAULT_NAME,
POLICY_PREPROCESSOR_DEFAULT_NAME,
)
from .configuration_fastwam import FastWAMConfig
@@ -96,20 +105,38 @@ def make_fastwam_pre_post_processors(
# anyway) and unsafe across fine-tuning: its `resize_size` would be inherited from the base
# checkpoint's camera geometry, not this dataset's, making the concatenation N_cameras x too wide.
steps = make_default_policy_processor_steps(config, normalization_stats, normalizer_device=config.device)
input_steps = [
steps.rename_observations,
steps.add_batch_dim,
steps.to_device,
steps.normalize,
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
DeviceProcessorStep(device=config.device),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=normalization_stats,
device=config.device,
),
]
output_steps = [
steps.unnormalize,
UnnormalizerProcessorStep(
features=config.output_features,
norm_map=config.normalization_mapping,
stats=normalization_stats,
),
]
if config.toggle_action_dimensions:
output_steps.append(
FastWAMActionToggleProcessorStep(toggle_dimensions=config.toggle_action_dimensions)
)
output_steps.append(steps.to_cpu)
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
output_steps.append(DeviceProcessorStep(device="cpu"))
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
@@ -20,10 +20,17 @@ from typing import Any
import torch
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
make_default_pre_post_processors,
RenameObservationsProcessorStep,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
from .configuration_gaussian_actor import GaussianActorConfig
@@ -55,4 +62,33 @@ def make_gaussian_actor_pre_post_processors(
Returns:
A tuple containing the configured pre-processor and post-processor pipelines.
"""
return make_default_pre_post_processors(config, dataset_stats)
# Add remaining processors
input_steps = [
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
DeviceProcessorStep(device=config.device),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
]
output_steps = [
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
DeviceProcessorStep(device="cpu"),
]
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
@@ -25,12 +25,19 @@ import torch
from lerobot.configs.types import FeatureType, NormalizationMode
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStep,
RenameObservationsProcessorStep,
UnnormalizerProcessorStep,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
)
from lerobot.processor.converters import policy_action_to_transition, transition_to_policy_action
from lerobot.utils.constants import (
POLICY_POSTPROCESSOR_DEFAULT_NAME,
POLICY_PREPROCESSOR_DEFAULT_NAME,
)
from .configuration_lingbot_va import LingBotVAConfig
@@ -45,13 +52,15 @@ def make_lingbot_va_pre_post_processors(
]:
"""Build the pre/post processor pipelines for LingBot-VA."""
steps = make_default_policy_processor_steps(config, dataset_stats)
input_steps: list[ProcessorStep] = [
steps.rename_observations,
steps.add_batch_dim,
steps.normalize,
steps.to_device,
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
DeviceProcessorStep(device=config.device),
]
# Unnormalize actions from [-1, 1] to physical units (QUANTILES) using q01/q99 restored from the checkpoint.
@@ -61,7 +70,18 @@ def make_lingbot_va_pre_post_processors(
norm_map={FeatureType.ACTION: NormalizationMode.QUANTILES},
stats=dataset_stats,
),
steps.to_cpu,
DeviceProcessorStep(device="cpu"),
]
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
@@ -19,12 +19,18 @@ from typing import Any
import torch
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
RenameObservationsProcessorStep,
TokenizerProcessorStep,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
from .configuration_multi_task_dit import MultiTaskDiTConfig
@@ -60,11 +66,9 @@ def make_multi_task_dit_pre_post_processors(
A tuple containing the configured pre-processor and post-processor pipelines.
"""
steps = make_default_policy_processor_steps(config, dataset_stats, normalizer_device=config.device)
input_steps = [
steps.rename_observations,
steps.add_batch_dim,
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
TokenizerProcessorStep(
tokenizer_name=config.text_encoder_name,
padding=config.tokenizer_padding,
@@ -72,12 +76,32 @@ def make_multi_task_dit_pre_post_processors(
max_length=config.tokenizer_max_length,
truncation=config.tokenizer_truncation,
),
steps.to_device,
steps.normalize,
DeviceProcessorStep(device=config.device),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
device=config.device,
),
]
output_steps = [
steps.unnormalize,
steps.to_cpu,
UnnormalizerProcessorStep(
features=config.output_features,
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
DeviceProcessorStep(device="cpu"),
]
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
+32 -11
View File
@@ -21,16 +21,22 @@ import torch
from lerobot.configs import PipelineFeatureType, PolicyFeature
from lerobot.processor import (
AbsoluteActionsProcessorStep,
AddBatchDimensionProcessorStep,
ComplementaryDataProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStep,
ProcessorStepRegistry,
RelativeActionsProcessorStep,
RenameObservationsProcessorStep,
TokenizerProcessorStep,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
from .configuration_pi0 import PI0Config
@@ -130,12 +136,10 @@ def make_pi0_pre_post_processors(
action_names=getattr(config, "action_feature_names", None),
)
steps = make_default_policy_processor_steps(config, dataset_stats)
# OpenPI order: raw → relative → normalize → model → unnormalize → absolute
input_steps: list[ProcessorStep] = [
steps.rename_observations, # To mimic the same processor as pretrained one
steps.add_batch_dim,
RenameObservationsProcessorStep(rename_map={}), # To mimic the same processor as pretrained one
AddBatchDimensionProcessorStep(),
Pi0NewLineProcessor(), # Add newlines before tokenization for PaliGemma
TokenizerProcessorStep(
tokenizer_name="google/paligemma-3b-pt-224",
@@ -143,15 +147,32 @@ def make_pi0_pre_post_processors(
padding_side="right",
padding="max_length",
),
steps.to_device,
DeviceProcessorStep(device=config.device),
relative_step,
steps.normalize,
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
]
output_steps: list[ProcessorStep] = [
steps.unnormalize,
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
AbsoluteActionsProcessorStep(enabled=config.use_relative_actions, relative_step=relative_step),
steps.to_cpu,
DeviceProcessorStep(device="cpu"),
]
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
@@ -55,9 +55,25 @@ class PI05Config(PreTrainedConfig):
relative_exclude_joints: list[str] = field(default_factory=lambda: ["gripper"])
# Populated at runtime from dataset metadata by make_policy.
action_feature_names: list[str] | None = None
# ``se3`` uses inv(T_current) @ T_target for each xyz+rotation-vector pose group.
# ``se3_6d`` uses the same composition and expands each relative rotation
# vector to the continuous first-two-row 6-D rotation representation.
# ``componentwise`` preserves the legacy action - state behavior.
relative_pose_representation: str = "componentwise"
relative_se3_pose_groups: list[list[int]] = field(default_factory=lambda: [list(range(6))])
# Build proprioception from absolute action samples when the dataset has no
# observation.state. With history_steps=2, training samples request t-1 as
# well as the normal t..t+chunk_size-1 action targets.
state_from_action: bool = False
proprioception_history_steps: int = 1
use_relative_state_history: bool = False
relative_state_exclude_joints: list[str] = field(default_factory=lambda: ["gripper"])
# Real-Time Chunking (RTC) configuration
rtc_config: RTCConfig | None = None
# Maximum clean action-prefix length sampled during training. Zero disables trained RTC.
rtc_training_max_delay: int = 0
image_resolution: tuple[int, int] = (
DEFAULT_IMAGE_SIZE,
@@ -111,6 +127,11 @@ class PI05Config(PreTrainedConfig):
raise ValueError(
f"n_action_steps ({self.n_action_steps}) cannot be greater than chunk_size ({self.chunk_size})"
)
if not 0 <= self.rtc_training_max_delay < self.chunk_size:
raise ValueError(
"rtc_training_max_delay must satisfy "
f"0 <= delay < chunk_size ({self.chunk_size}), got {self.rtc_training_max_delay}"
)
if self.paligemma_variant not in ["gemma_300m", "gemma_2b"]:
raise ValueError(f"Invalid paligemma_variant: {self.paligemma_variant}")
@@ -121,6 +142,25 @@ class PI05Config(PreTrainedConfig):
if self.dtype not in ["bfloat16", "float32"]:
raise ValueError(f"Invalid dtype: {self.dtype}")
if self.proprioception_history_steps < 1:
raise ValueError("proprioception_history_steps must be at least 1")
if self.relative_pose_representation not in {"componentwise", "se3", "se3_6d"}:
raise ValueError(
"relative_pose_representation must be 'componentwise', 'se3', or 'se3_6d', got "
f"{self.relative_pose_representation!r}"
)
for group in self.relative_se3_pose_groups:
if len(group) != 6 or len(set(group)) != 6 or any(index < 0 for index in group):
raise ValueError(f"Invalid six-index SE(3) pose group: {group}")
if self.relative_pose_representation == "se3_6d" and group != list(range(group[0], group[0] + 6)):
raise ValueError("se3_6d pose groups must contain six contiguous ascending indices")
if self.relative_pose_representation in {"se3", "se3_6d"} and not self.relative_se3_pose_groups:
raise ValueError(
f"relative_pose_representation={self.relative_pose_representation!r} "
"requires relative_se3_pose_groups"
)
def validate_features(self) -> None:
"""Validate and set up input/output features."""
for i in range(self.empty_cameras):
@@ -131,19 +171,54 @@ class PI05Config(PreTrainedConfig):
)
self.input_features[key] = empty_camera
if OBS_STATE not in self.input_features:
state_feature = PolicyFeature(
type=FeatureType.STATE,
shape=(self.max_state_dim,), # Padded to max_state_dim
)
self.input_features[OBS_STATE] = state_feature
if ACTION not in self.output_features:
action_feature = PolicyFeature(
type=FeatureType.ACTION,
shape=(self.max_action_dim,), # Padded to max_action_dim
)
self.output_features[ACTION] = action_feature
elif self.relative_pose_representation == "se3_6d":
action_feature = self.output_features[ACTION]
source_dim = (
len(self.action_feature_names)
if self.action_feature_names is not None
else action_feature.shape[-1]
)
model_dim = source_dim + 3 * len(self.relative_se3_pose_groups)
if action_feature.shape[-1] == source_dim:
self.output_features[ACTION] = PolicyFeature(
type=action_feature.type,
shape=(model_dim,),
)
elif action_feature.shape[-1] != model_dim:
raise ValueError(
"se3_6d action feature has incompatible width: "
f"source={source_dim}, expected model width={model_dim}, "
f"got={action_feature.shape[-1]}"
)
if model_dim > self.max_action_dim:
raise ValueError(
f"se3_6d action width {model_dim} exceeds max_action_dim={self.max_action_dim}"
)
if OBS_STATE not in self.input_features:
state_shape = (self.max_state_dim,)
if self.state_from_action and ACTION in self.output_features:
state_shape = self.output_features[ACTION].shape
state_feature = PolicyFeature(
type=FeatureType.STATE,
shape=state_shape,
)
self.input_features[OBS_STATE] = state_feature
state_dim = self.input_features[OBS_STATE].shape[-1]
history_state_dim = state_dim * self.proprioception_history_steps
if history_state_dim > self.max_state_dim:
raise ValueError(
"Flattened proprioception history exceeds max_state_dim: "
f"{state_dim} * {self.proprioception_history_steps} = {history_state_dim} > "
f"{self.max_state_dim}"
)
def get_optimizer_preset(self) -> AdamWConfig:
return AdamWConfig(
@@ -168,7 +243,8 @@ class PI05Config(PreTrainedConfig):
@property
def action_delta_indices(self) -> list:
return list(range(self.chunk_size))
history_prefix = self.proprioception_history_steps - 1 if self.state_from_action else 0
return list(range(-history_prefix, self.chunk_size))
@property
def reward_delta_indices(self) -> None:
+169 -24
View File
@@ -66,6 +66,107 @@ class ActionSelectKwargs(TypedDict, total=False):
execution_horizon: int | None
def _prepare_trained_rtc_prefix(
x_t: Tensor,
prev_chunk_left_over: Tensor | None,
inference_delay: int,
training_max_delay: int,
) -> tuple[Tensor | None, Tensor | None]:
"""Pad and validate a hard prefix for training-time RTC inference."""
if prev_chunk_left_over is None or inference_delay <= 0:
return None, None
if training_max_delay <= 0:
raise ValueError(
"RTC mode='trained' requires a checkpoint trained with policy.rtc_training_max_delay > 0."
)
if inference_delay > training_max_delay:
raise ValueError(
f"Measured RTC inference delay ({inference_delay}) exceeds the checkpoint's "
f"rtc_training_max_delay ({training_max_delay})."
)
if inference_delay >= x_t.shape[1]:
raise ValueError(
f"RTC inference delay ({inference_delay}) must be smaller than chunk_size ({x_t.shape[1]})."
)
previous = prev_chunk_left_over.to(device=x_t.device, dtype=x_t.dtype)
if not torch.isfinite(previous).all():
raise ValueError("RTC prefix contains NaN or Inf values.")
if previous.ndim == 2:
previous = previous.unsqueeze(0)
if previous.ndim != 3:
raise ValueError(f"Expected RTC prefix shape (B, T, A), got {tuple(previous.shape)}")
if previous.shape[0] == 1 and x_t.shape[0] > 1:
previous = previous.expand(x_t.shape[0], -1, -1)
if previous.shape[0] != x_t.shape[0]:
raise ValueError(
f"RTC prefix batch size ({previous.shape[0]}) does not match policy batch ({x_t.shape[0]})."
)
if previous.shape[1] < inference_delay:
raise ValueError(f"RTC prefix has {previous.shape[1]} steps, but inference_delay={inference_delay}.")
if previous.shape[2] > x_t.shape[2]:
raise ValueError(
f"RTC prefix action dimension ({previous.shape[2]}) exceeds model dimension ({x_t.shape[2]})."
)
padded_prefix = torch.zeros_like(x_t)
padded_prefix[:, :inference_delay, : previous.shape[2]] = previous[:, :inference_delay]
prefix_mask = torch.arange(x_t.shape[1], device=x_t.device) < inference_delay
prefix_mask = prefix_mask[None, :, None].expand(x_t.shape[0], -1, x_t.shape[2])
return padded_prefix, prefix_mask
def _sample_training_rtc_prefix_mask(
batch_size: int,
action_horizon: int,
max_delay: int,
device: torch.device,
) -> Tensor | None:
"""Sample a clean action-prefix length independently for each training example."""
if max_delay <= 0:
return None
delays = torch.randint(0, max_delay + 1, (batch_size,), device=device)
positions = torch.arange(action_horizon, device=device)
return positions.unsqueeze(0) < delays.unsqueeze(1)
def _build_flow_matching_inputs(
actions: Tensor,
noise: Tensor,
time: Tensor,
prefix_mask: Tensor | None,
) -> tuple[Tensor, Tensor]:
"""Keep the sampled RTC prefix clean while noising the remaining action chunk."""
if prefix_mask is None:
model_time = time
expanded_time = time[:, None, None]
else:
model_time = time[:, None].expand_as(prefix_mask)
model_time = torch.where(prefix_mask, torch.zeros_like(model_time), model_time)
expanded_time = model_time.unsqueeze(-1)
x_t = expanded_time * noise + (1 - expanded_time) * actions
return x_t, model_time
def _reduce_training_rtc_loss(
losses: Tensor,
prefix_mask: Tensor | None,
reduction: str,
) -> Tensor:
"""Average flow loss over predicted postfix actions, excluding the clean RTC prefix."""
if reduction not in {"mean", "none"}:
raise ValueError(f"Unsupported loss reduction: {reduction!r}")
if prefix_mask is None:
return losses.mean() if reduction == "mean" else losses.mean(dim=(1, 2))
postfix_mask = (~prefix_mask).unsqueeze(-1).expand_as(losses)
if reduction == "none":
numerator = (losses * postfix_mask).sum(dim=(1, 2))
denominator = postfix_mask.sum(dim=(1, 2))
return numerator / denominator.clamp(min=1)
return (losses * postfix_mask).sum() / postfix_mask.sum().clamp(min=1)
def get_safe_dtype(target_dtype, device_type):
"""Get a safe dtype for the given device type."""
if device_type == "mps" and target_dtype == torch.float64:
@@ -82,21 +183,20 @@ def get_safe_dtype(target_dtype, device_type):
def create_sinusoidal_pos_embedding( # see openpi `create_sinusoidal_pos_embedding` (exact copy)
time: torch.Tensor, dimension: int, min_period: float, max_period: float, device="cpu"
) -> Tensor:
"""Computes sine-cosine positional embedding vectors for scalar positions."""
"""Compute sine-cosine embeddings for scalar or per-action positions."""
if dimension % 2 != 0:
raise ValueError(f"dimension ({dimension}) must be divisible by 2")
if time.ndim != 1:
raise ValueError("The time tensor is expected to be of shape `(batch_size, )`.")
if time.ndim not in (1, 2):
raise ValueError("The time tensor must have shape (batch_size,) or (batch_size, action_horizon).")
dtype = get_safe_dtype(torch.float64, device.type)
fraction = torch.linspace(0.0, 1.0, dimension // 2, dtype=dtype, device=device)
period = min_period * (max_period / min_period) ** fraction
# Compute the outer product
scaling_factor = 1.0 / period * 2 * math.pi
sin_input = scaling_factor[None, :] * time[:, None]
return torch.cat([torch.sin(sin_input), torch.cos(sin_input)], dim=1)
sin_input = time[..., None] * scaling_factor
return torch.cat([torch.sin(sin_input), torch.cos(sin_input)], dim=-1)
def sample_beta(alpha, beta, bsize, device): # see openpi `sample_beta` (exact copy)
@@ -739,14 +839,23 @@ class PI05Pytorch(nn.Module): # see openpi `PI0Pytorch`
return embs, pad_masks, att_masks, adarms_cond
def forward(self, images, img_masks, tokens, masks, actions, noise, time) -> Tensor:
def forward(
self,
images,
img_masks,
tokens,
masks,
actions,
noise,
time,
prefix_mask: Tensor | None = None,
) -> Tensor:
"""Do a full training forward pass and compute the loss."""
time_expanded = time[:, None, None]
x_t = time_expanded * noise + (1 - time_expanded) * actions
x_t, model_time = _build_flow_matching_inputs(actions, noise, time, prefix_mask)
u_t = noise - actions
prefix_embs, prefix_pad_masks, prefix_att_masks = self.embed_prefix(images, img_masks, tokens, masks)
suffix_embs, suffix_pad_masks, suffix_att_masks, adarms_cond = self.embed_suffix(x_t, time)
suffix_embs, suffix_pad_masks, suffix_att_masks, adarms_cond = self.embed_suffix(x_t, model_time)
if (
self.paligemma_with_expert.paligemma.model.language_model.layers[0].self_attn.q_proj.weight.dtype
@@ -833,11 +942,35 @@ class PI05Pytorch(nn.Module): # see openpi `PI0Pytorch`
dt = -1.0 / num_steps
x_t = noise
rtc_mode = "guided"
trained_prefix = trained_prefix_mask = None
if self._rtc_enabled():
rtc_mode = self.rtc_processor.rtc_config.mode
if rtc_mode == "trained":
training_max_delay = int(getattr(self.config, "rtc_training_max_delay", 0))
if training_max_delay <= 0:
raise ValueError(
"RTC mode='trained' requires a checkpoint trained with "
"policy.rtc_training_max_delay > 0."
)
trained_prefix, trained_prefix_mask = _prepare_trained_rtc_prefix(
x_t,
kwargs.get("prev_chunk_left_over"),
int(kwargs.get("inference_delay") or 0),
training_max_delay,
)
for step in range(num_steps):
time = 1.0 + step * dt
time_tensor = torch.tensor(time, dtype=torch.float32, device=device).expand(bsize)
def denoise_step_partial_call(input_x_t, current_timestep=time_tensor):
denoise_timestep = time_tensor
if trained_prefix is not None:
x_t = torch.where(trained_prefix_mask, trained_prefix, x_t)
denoise_timestep = time_tensor[:, None].expand(bsize, x_t.shape[1]).clone()
denoise_timestep[trained_prefix_mask[..., 0]] = 0.0
def denoise_step_partial_call(input_x_t, current_timestep=denoise_timestep):
return self.denoise_step(
prefix_pad_masks=prefix_pad_masks,
past_key_values=past_key_values,
@@ -845,7 +978,7 @@ class PI05Pytorch(nn.Module): # see openpi `PI0Pytorch`
timestep=current_timestep,
)
if self._rtc_enabled():
if self._rtc_enabled() and rtc_mode == "guided":
inference_delay = kwargs.get("inference_delay")
prev_chunk_left_over = kwargs.get("prev_chunk_left_over")
execution_horizon = kwargs.get("execution_horizon")
@@ -862,6 +995,8 @@ class PI05Pytorch(nn.Module): # see openpi `PI0Pytorch`
v_t = denoise_step_partial_call(x_t)
x_t = x_t + dt * v_t
if trained_prefix is not None:
x_t = torch.where(trained_prefix_mask, trained_prefix, x_t)
if self.rtc_processor is not None and self.rtc_processor.is_debug_enabled():
self.rtc_processor.track(time=time, x_t=x_t, v_t=v_t)
@@ -1137,7 +1272,10 @@ class PI05Policy(PreTrainedPolicy):
# Create processor if config provided
# If RTC is not enabled - we can still track the denoising data
if self.config.rtc_config is not None:
self.rtc_processor = RTCProcessor(self.config.rtc_config)
self.rtc_processor = RTCProcessor(
self.config.rtc_config,
trained_mode_supported=int(getattr(self.config, "rtc_training_max_delay", 0)) > 0,
)
model_value = getattr(self, "model", None)
if model_value is not None:
@@ -1269,28 +1407,35 @@ class PI05Policy(PreTrainedPolicy):
noise = self.model.sample_noise(actions.shape, actions.device)
time = self.model.sample_time(actions.shape[0], actions.device)
prefix_mask = _sample_training_rtc_prefix_mask(
actions.shape[0],
actions.shape[1],
self.config.rtc_training_max_delay,
actions.device,
)
# Compute loss (no separate state needed for PI05)
losses = self.model.forward(images, img_masks, tokens, masks, actions, noise, time)
losses = self.model.forward(images, img_masks, tokens, masks, actions, noise, time, prefix_mask)
# Truncate losses to actual action dimensions
original_action_dim = self.config.output_features[ACTION].shape[0]
losses = losses[:, :, :original_action_dim]
loss_dict = {
"loss_per_dim": losses.mean(dim=[0, 1]).detach().cpu().numpy().tolist(),
}
if prefix_mask is None:
loss_per_dim = losses.mean(dim=(0, 1))
else:
postfix_mask = (~prefix_mask).unsqueeze(-1).expand_as(losses)
loss_per_dim = (losses * postfix_mask).sum(dim=(0, 1)) / postfix_mask.sum(dim=(0, 1)).clamp(min=1)
loss_dict = {"loss_per_dim": loss_per_dim.detach().cpu().numpy().tolist()}
if reduction == "none":
# Return per-sample losses (B,) by averaging over time and action dims
per_sample_loss = losses.mean(dim=(1, 2))
per_sample_loss = _reduce_training_rtc_loss(losses, prefix_mask, reduction="none")
loss_dict["loss"] = per_sample_loss.mean().item()
return per_sample_loss, loss_dict
else:
# Default: return scalar mean loss
loss = losses.mean()
loss_dict["loss"] = loss.item()
return loss, loss_dict
loss = _reduce_training_rtc_loss(losses, prefix_mask, reduction="mean")
loss_dict["loss"] = loss.item()
return loss, loss_dict
def _get_default_peft_targets(self) -> dict[str, any]:
"""Return default PEFT target modules for PI0.5 fine-tuning."""
+212 -13
View File
@@ -15,7 +15,7 @@
# limitations under the License.
from copy import deepcopy
from dataclasses import dataclass
from dataclasses import dataclass, field
from typing import Any
import numpy as np
@@ -24,21 +24,190 @@ import torch
from lerobot.configs import PipelineFeatureType, PolicyFeature
from lerobot.processor import (
AbsoluteActionsProcessorStep,
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStep,
ProcessorStepRegistry,
RelativeActionsProcessorStep,
RenameObservationsProcessorStep,
TokenizerProcessorStep,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
UnnormalizerProcessorStep,
policy_action_to_transition,
relative_action_output_dim,
to_relative_actions,
transition_to_policy_action,
)
from lerobot.types import EnvTransition, TransitionKey
from lerobot.utils.constants import OBS_STATE
from lerobot.utils.constants import (
OBS_STATE,
POLICY_POSTPROCESSOR_DEFAULT_NAME,
POLICY_PREPROCESSOR_DEFAULT_NAME,
)
from .configuration_pi05 import PI05Config
@ProcessorStepRegistry.register(name="pi05_state_from_action_processor_step")
@dataclass
class Pi05StateFromActionProcessorStep(ProcessorStep):
"""Synthesize proprioception from absolute actions in state-less datasets.
The dataset loader supplies ``history_steps - 1`` actions before the normal
target chunk. Those leading samples and action(t) become state history; only
the leading samples are then removed from the action targets.
"""
enabled: bool = False
history_steps: int = 1
_inference_history: torch.Tensor | None = field(default=None, init=False, repr=False)
def __call__(self, transition: EnvTransition) -> EnvTransition:
if not self.enabled:
return transition
observation = transition.get(TransitionKey.OBSERVATION, {})
observed_state = observation.get(OBS_STATE)
if observed_state is not None:
# At inference the robot normally provides only the current state and
# there is no action target. Build a rolling history in the processor.
if transition.get(TransitionKey.ACTION) is None and observed_state.ndim == 2:
if self._inference_history is None:
self._inference_history = observed_state.unsqueeze(1).repeat(1, self.history_steps, 1)
else:
self._inference_history = torch.cat(
[self._inference_history[:, 1:], observed_state.unsqueeze(1)], dim=1
)
new_transition = transition.copy()
new_observation = dict(observation)
new_observation[OBS_STATE] = self._inference_history.clone()
new_transition[TransitionKey.OBSERVATION] = new_observation
return new_transition
return transition
action = transition.get(TransitionKey.ACTION)
if action is None:
raise ValueError("Cannot synthesize PI0.5 state without action")
if action.ndim != 3:
raise ValueError(f"Expected batched action chunks with shape (B, T, D), got {action.shape}")
if action.shape[1] < self.history_steps:
raise ValueError(
f"Action chunk has {action.shape[1]} steps, fewer than history_steps={self.history_steps}"
)
new_transition = transition.copy()
new_observation = dict(observation)
state = action[:, : self.history_steps].clone()
if self.history_steps == 1:
state = state[:, 0]
new_observation[OBS_STATE] = state
new_transition[TransitionKey.OBSERVATION] = new_observation
new_transition[TransitionKey.ACTION] = action[:, self.history_steps - 1 :]
return new_transition
def get_config(self) -> dict[str, Any]:
return {"enabled": self.enabled, "history_steps": self.history_steps}
def reset(self) -> None:
self._inference_history = None
def transform_features(
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
return features
@ProcessorStepRegistry.register(name="pi05_flatten_state_history_processor_step")
@dataclass
class Pi05FlattenStateHistoryProcessorStep(ProcessorStep):
"""Optionally relativize raw state history, then flatten it for PI0.5."""
history_steps: int = 1
max_state_dim: int = 32
relative: bool = False
exclude_joints: list[str] = field(default_factory=list)
state_names: list[str] | None = None
pose_representation: str = "componentwise"
se3_pose_groups: list[list[int]] = field(default_factory=list)
def __call__(self, transition: EnvTransition) -> EnvTransition:
observation = transition.get(TransitionKey.OBSERVATION, {})
state = observation.get(OBS_STATE)
if state is None:
raise ValueError("State is required for PI05")
if self.history_steps == 1 and state.ndim == 2:
state = state.unsqueeze(1)
if state.ndim != 3 or state.shape[1] != self.history_steps:
raise ValueError(
f"Expected state history with shape (B, {self.history_steps}, D), got {state.shape}"
)
processed_state = state.clone()
if self.relative:
mask_step = RelativeActionsProcessorStep(
enabled=True,
exclude_joints=self.exclude_joints,
action_names=self.state_names,
)
processed_state = to_relative_actions(
state,
state[:, -1],
mask_step._build_mask(state.shape[-1]),
pose_representation=self.pose_representation,
se3_pose_groups=self.se3_pose_groups,
)
flattened_dim = processed_state.shape[1] * processed_state.shape[2]
if flattened_dim > self.max_state_dim:
raise ValueError(
f"Flattened state history has {flattened_dim} dimensions, above max_state_dim={self.max_state_dim}"
)
new_transition = transition.copy()
new_observation = dict(observation)
new_observation[OBS_STATE] = processed_state.flatten(start_dim=1)
new_transition[TransitionKey.OBSERVATION] = new_observation
return new_transition
def get_config(self) -> dict[str, Any]:
return {
"history_steps": self.history_steps,
"max_state_dim": self.max_state_dim,
"relative": self.relative,
"exclude_joints": self.exclude_joints,
"state_names": self.state_names,
"pose_representation": self.pose_representation,
"se3_pose_groups": self.se3_pose_groups,
}
def transform_features(
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
transformed = deepcopy(features)
for feature_group in transformed.values():
state_feature = feature_group.get(OBS_STATE)
if state_feature is not None:
state_dim = state_feature.shape[-1]
if self.relative:
source_dim = len(self.state_names) if self.state_names is not None else state_dim
model_dim = relative_action_output_dim(
source_dim,
self.pose_representation,
self.se3_pose_groups,
)
if state_dim == source_dim:
state_dim = model_dim
elif state_dim != model_dim:
raise ValueError(
f"Expected source/model state width {source_dim}/{model_dim}, got {state_dim}"
)
state_dim *= self.history_steps
feature_group[OBS_STATE] = PolicyFeature(type=state_feature.type, shape=(state_dim,))
return transformed
@ProcessorStepRegistry.register(name="pi05_prepare_state_tokenizer_processor_step")
@dataclass
class Pi05PrepareStateTokenizerProcessorStep(ProcessorStep):
@@ -124,18 +293,35 @@ def make_pi05_pre_post_processors(
enabled=config.use_relative_actions,
exclude_joints=getattr(config, "relative_exclude_joints", []),
action_names=getattr(config, "action_feature_names", None),
pose_representation=config.relative_pose_representation,
se3_pose_groups=config.relative_se3_pose_groups,
)
steps = make_default_policy_processor_steps(config, dataset_stats)
# OpenPI order: raw → relative → normalize → model → unnormalize → absolute
input_steps: list[ProcessorStep] = [
steps.rename_observations, # To mimic the same processor as pretrained one
steps.add_batch_dim,
RenameObservationsProcessorStep(rename_map={}), # To mimic the same processor as pretrained one
AddBatchDimensionProcessorStep(),
Pi05StateFromActionProcessorStep(
enabled=config.state_from_action,
history_steps=config.proprioception_history_steps,
),
relative_step,
Pi05FlattenStateHistoryProcessorStep(
history_steps=config.proprioception_history_steps,
max_state_dim=config.max_state_dim,
relative=config.use_relative_state_history,
exclude_joints=config.relative_state_exclude_joints,
state_names=config.action_feature_names,
pose_representation=config.relative_pose_representation,
se3_pose_groups=config.relative_se3_pose_groups,
),
# NOTE: NormalizerProcessorStep MUST come before Pi05PrepareStateTokenizerProcessorStep
# because the tokenizer step expects normalized state in [-1, 1] range for discretization
steps.normalize,
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
Pi05PrepareStateTokenizerProcessorStep(max_state_dim=config.max_state_dim),
TokenizerProcessorStep(
tokenizer_name="google/paligemma-3b-pt-224",
@@ -143,13 +329,26 @@ def make_pi05_pre_post_processors(
padding_side="right",
padding="max_length",
),
steps.to_device,
DeviceProcessorStep(device=config.device),
]
output_steps: list[ProcessorStep] = [
steps.unnormalize,
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
AbsoluteActionsProcessorStep(enabled=config.use_relative_actions, relative_step=relative_step),
steps.to_cpu,
DeviceProcessorStep(device="cpu"),
]
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
@@ -25,17 +25,26 @@ from lerobot.configs import PipelineFeatureType, PolicyFeature
from lerobot.processor import (
AbsoluteActionsProcessorStep,
ActionTokenizerProcessorStep,
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStep,
ProcessorStepRegistry,
RelativeActionsProcessorStep,
RenameObservationsProcessorStep,
TokenizerProcessorStep,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.types import EnvTransition, TransitionKey
from lerobot.utils.constants import OBS_STATE
from lerobot.utils.constants import (
OBS_STATE,
POLICY_POSTPROCESSOR_DEFAULT_NAME,
POLICY_PREPROCESSOR_DEFAULT_NAME,
)
from .configuration_pi0_fast import PI0FastConfig
@@ -126,8 +135,6 @@ def make_pi0_fast_pre_post_processors(
action_names=getattr(config, "action_feature_names", None),
)
steps = make_default_policy_processor_steps(config, dataset_stats)
# Pi0Fast order: relative → normalize → tokenize → model → unnormalize → absolute
# This matches pi0/pi0.5: RelativeActionsProcessorStep runs first on raw absolute actions,
# caching the raw state. NormalizerProcessorStep then normalizes the raw relative actions,
@@ -137,10 +144,14 @@ def make_pi0_fast_pre_post_processors(
# before Pi0FastPrepareStateAndLanguageTokenizerProcessorStep, so the state tokenizer
# continues to receive normalized state in [-1, 1] as expected.
input_steps: list[ProcessorStep] = [
steps.rename_observations, # To mimic the same processor as pretrained one
steps.add_batch_dim,
RenameObservationsProcessorStep(rename_map={}), # To mimic the same processor as pretrained one
AddBatchDimensionProcessorStep(),
relative_step,
steps.normalize,
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
Pi0FastPrepareStateAndLanguageTokenizerProcessorStep(max_state_dim=config.max_state_dim),
TokenizerProcessorStep(
tokenizer_name=config.text_tokenizer_name,
@@ -154,13 +165,26 @@ def make_pi0_fast_pre_post_processors(
fast_skip_tokens=config.fast_skip_tokens,
paligemma_tokenizer_name=config.text_tokenizer_name,
),
steps.to_device,
DeviceProcessorStep(device=config.device),
]
output_steps: list[ProcessorStep] = [
steps.unnormalize,
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
AbsoluteActionsProcessorStep(enabled=config.use_relative_actions, relative_step=relative_step),
steps.to_cpu,
DeviceProcessorStep(device="cpu"),
]
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
+4 -1
View File
@@ -121,7 +121,10 @@ class PiGemmaRMSNorm(nn.Module):
if cond.shape[-1] != self.cond_dim:
raise ValueError(f"Expected cond dim {self.cond_dim}, got {cond.shape[-1]}")
modulation = self.dense(cond)
if len(x.shape) == 3:
# Per-sample conditioning (B, D) is broadcast across the sequence.
# Training-time RTC supplies per-action conditioning (B, T, D), which
# is already aligned with x and must keep its token dimension intact.
if len(x.shape) == 3 and modulation.dim() == 2:
modulation = modulation.unsqueeze(1)
scale, shift, gate = modulation.chunk(3, dim=-1)
normed = normed * (1 + scale.float()) + shift.float()
+21 -4
View File
@@ -23,6 +23,8 @@ from pathlib import Path
from tempfile import TemporaryDirectory
from typing import TYPE_CHECKING, TypedDict, TypeVar, Unpack
import packaging
import safetensors
from huggingface_hub import HfApi, ModelCard, ModelCardData, hf_hub_download, save_torch_state_dict
from huggingface_hub.constants import SAFETENSORS_SINGLE_FILE
from huggingface_hub.errors import HfHubHTTPError
@@ -32,7 +34,6 @@ from torch import Tensor, nn
from lerobot.__version__ import __version__
from lerobot.configs import PreTrainedConfig
from lerobot.configs.train import TrainPipelineConfig
from lerobot.utils.device_utils import resolve_safetensors_device
from lerobot.utils.hub import HubMixin
from .utils import log_model_loading_keys
@@ -220,10 +221,26 @@ class PreTrainedPolicy(nn.Module, HubMixin, abc.ABC):
@classmethod
def _load_as_safetensor(cls, model: T, model_file: str, map_location: str, strict: bool) -> T:
missing_keys, unexpected_keys = load_model_as_safetensor(
model, model_file, strict=strict, device=resolve_safetensors_device(map_location)
)
# Create base kwargs
kwargs = {"strict": strict}
# Add device parameter for newer versions that support it
if packaging.version.parse(safetensors.__version__) >= packaging.version.parse("0.4.3"):
kwargs["device"] = map_location
# Load the model with appropriate kwargs
missing_keys, unexpected_keys = load_model_as_safetensor(model, model_file, **kwargs)
log_model_loading_keys(missing_keys, unexpected_keys)
# For older versions, manually move to device if needed
if "device" not in kwargs and map_location != "cpu":
logging.warning(
"Loading model weights on other devices than 'cpu' is not supported natively in your version of safetensors."
" This means that the model is loaded on 'cpu' first and then copied to the device."
" This leads to a slower loading time."
" Please update safetensors to version 0.4.3 or above for improved performance."
)
model.to(map_location)
return model
@abc.abstractmethod
@@ -37,6 +37,10 @@ class RTCConfig:
# Infrastructure
enabled: bool = True
# ``guided`` is the original inference-time Jacobian guidance. ``trained``
# hard-inpaints a prefix and requires a compatible training-time RTC checkpoint.
mode: str = "guided"
# Core RTC settings
# Todo change to exp
prefix_attention_schedule: RTCAttentionSchedule = RTCAttentionSchedule.LINEAR
@@ -49,6 +53,8 @@ class RTCConfig:
def __post_init__(self):
"""Validate RTC configuration parameters."""
if self.mode not in {"guided", "trained"}:
raise ValueError(f"mode must be 'guided' or 'trained', got {self.mode!r}")
if self.max_guidance_weight <= 0:
raise ValueError(f"max_guidance_weight must be positive, got {self.max_guidance_weight}")
if self.debug_maxlen <= 0:
+6 -1
View File
@@ -42,7 +42,12 @@ class RTCProcessor:
prefix attention, and adaptive chunk processing.
"""
def __init__(self, rtc_config: RTCConfig):
def __init__(self, rtc_config: RTCConfig, *, trained_mode_supported: bool = False):
if rtc_config.enabled and rtc_config.mode == "trained" and not trained_mode_supported:
raise ValueError(
"RTC mode='trained' requires a PI05-compatible checkpoint trained with "
"rtc_training_max_delay > 0."
)
self.rtc_config = rtc_config
self.tracker = None
+7 -1
View File
@@ -49,7 +49,13 @@ def reanchor_relative_rtc_prefix(
action_cpu = prev_actions_absolute.detach().cpu()
mask = relative_step._build_mask(action_cpu.shape[-1])
relative_actions = to_relative_actions(action_cpu, state, mask)
relative_actions = to_relative_actions(
action_cpu,
state,
mask,
pose_representation=relative_step.pose_representation,
se3_pose_groups=relative_step.se3_pose_groups,
)
transition = create_transition(action=relative_actions)
if normalizer_step is not None:
@@ -19,13 +19,19 @@ from typing import Any
import torch
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NewLineTaskProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
RenameObservationsProcessorStep,
TokenizerProcessorStep,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
from .configuration_smolvla import SmolVLAConfig
@@ -60,11 +66,9 @@ def make_smolvla_pre_post_processors(
A tuple containing the configured pre-processor and post-processor pipelines.
"""
steps = make_default_policy_processor_steps(config, dataset_stats)
input_steps = [
steps.rename_observations, # To mimic the same processor as pretrained one
steps.add_batch_dim,
RenameObservationsProcessorStep(rename_map={}), # To mimic the same processor as pretrained one
AddBatchDimensionProcessorStep(),
NewLineTaskProcessorStep(),
TokenizerProcessorStep(
tokenizer_name=config.vlm_model_name,
@@ -72,11 +76,28 @@ def make_smolvla_pre_post_processors(
padding_side="right",
max_length=config.tokenizer_max_length,
),
steps.to_device,
steps.normalize,
DeviceProcessorStep(device=config.device),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
]
output_steps = [
steps.unnormalize,
steps.to_cpu,
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
DeviceProcessorStep(device="cpu"),
]
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
+37 -2
View File
@@ -19,10 +19,17 @@ from typing import Any
import torch
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
make_default_pre_post_processors,
RenameObservationsProcessorStep,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
from .configuration_tdmpc import TDMPCConfig
@@ -54,4 +61,32 @@ def make_tdmpc_pre_post_processors(
Returns:
A tuple containing the configured pre-processor and post-processor pipelines.
"""
return make_default_pre_post_processors(config, dataset_stats)
input_steps = [
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
DeviceProcessorStep(device=config.device),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
]
output_steps = [
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
DeviceProcessorStep(device="cpu"),
]
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
@@ -20,16 +20,20 @@ import torch
from lerobot.policies.vla_jepa.configuration_vla_jepa import VLAJEPAConfig
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
EnvTransition,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStep,
ProcessorStepRegistry,
RenameObservationsProcessorStep,
TransitionKey,
UnnormalizerProcessorStep,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
)
from lerobot.processor.converters import policy_action_to_transition, transition_to_policy_action
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
@ProcessorStepRegistry.register(name="vla_jepa_clip_actions")
@@ -108,12 +112,15 @@ def make_vla_jepa_pre_post_processors(
PolicyProcessorPipeline[PolicyAction, PolicyAction],
]:
features = {**config.input_features, **config.output_features}
steps = make_default_policy_processor_steps(config, dataset_stats)
input_steps = [
steps.rename_observations,
steps.add_batch_dim,
steps.to_device,
steps.normalize,
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
DeviceProcessorStep(device=config.device),
NormalizerProcessorStep(
features=features,
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
]
output_steps: list[ProcessorStep] = []
if config.clip_normalized_actions:
@@ -122,8 +129,6 @@ def make_vla_jepa_pre_post_processors(
output_steps.append(
PreSnapGripperProcessorStep(gripper_dim=config.gripper_dim, threshold=config.gripper_threshold)
)
# NOTE: unlike the default policy unnormalizer (output features only), VLA-JEPA
# unnormalizes over BOTH input and output features.
output_steps.append(
UnnormalizerProcessorStep(
features=features,
@@ -135,5 +140,16 @@ def make_vla_jepa_pre_post_processors(
output_steps.append(
BinarizeGripperProcessorStep(gripper_dim=config.gripper_dim, threshold=config.gripper_threshold)
)
output_steps.append(steps.to_cpu)
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
output_steps.append(DeviceProcessorStep(device="cpu"))
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
+37 -2
View File
@@ -20,10 +20,17 @@ from typing import Any
import torch
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
make_default_pre_post_processors,
RenameObservationsProcessorStep,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
from .configuration_vqbet import VQBeTConfig
@@ -55,4 +62,32 @@ def make_vqbet_pre_post_processors(
Returns:
A tuple containing the configured pre-processor and post-processor pipelines.
"""
return make_default_pre_post_processors(config, dataset_stats)
input_steps = [
RenameObservationsProcessorStep(rename_map={}), # Let the possibility to the user to rename the keys
AddBatchDimensionProcessorStep(),
DeviceProcessorStep(device=config.device),
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
]
output_steps = [
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
DeviceProcessorStep(device="cpu"),
]
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
+32 -11
View File
@@ -20,13 +20,19 @@ import torch
from lerobot.configs import PipelineFeatureType, PolicyFeature
from lerobot.processor import (
AddBatchDimensionProcessorStep,
ComplementaryDataProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStepRegistry,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
RenameObservationsProcessorStep,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
from .configuration_wall_x import WallXConfig
@@ -59,22 +65,37 @@ def make_wall_x_pre_post_processors(
A tuple containing the configured pre-processor and post-processor pipelines
"""
steps = make_default_policy_processor_steps(config, dataset_stats)
input_steps = [
steps.rename_observations,
steps.add_batch_dim,
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
WallXTaskProcessor(), # Process task description
steps.normalize,
steps.to_device,
NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
DeviceProcessorStep(device=config.device),
]
output_steps = [
steps.unnormalize,
steps.to_cpu,
UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
DeviceProcessorStep(device="cpu"),
]
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
@ProcessorStepRegistry.register(name="wall_x_task_processor")
+34 -11
View File
@@ -22,14 +22,19 @@ import torch
from lerobot.configs import PipelineFeatureType, PolicyFeature
from lerobot.processor import (
AddBatchDimensionProcessorStep,
DeviceProcessorStep,
NormalizerProcessorStep,
ObservationProcessorStep,
PolicyAction,
PolicyProcessorPipeline,
ProcessorStep,
ProcessorStepRegistry,
RenameObservationsProcessorStep,
TokenizerProcessorStep,
make_default_policy_processor_steps,
make_policy_processor_pipelines,
UnnormalizerProcessorStep,
policy_action_to_transition,
transition_to_policy_action,
)
from lerobot.types import EnvTransition, TransitionKey
from lerobot.utils.constants import (
@@ -37,6 +42,8 @@ from lerobot.utils.constants import (
OBS_IMAGES,
OBS_PREFIX,
OBS_STATE,
POLICY_POSTPROCESSOR_DEFAULT_NAME,
POLICY_PREPROCESSOR_DEFAULT_NAME,
)
from .configuration_xvla import XVLAConfig
@@ -54,11 +61,10 @@ def make_xvla_pre_post_processors(
Build the LeRobot processor pipelines for XVLA.
"""
steps = make_default_policy_processor_steps(config, dataset_stats)
features = {**config.input_features, **config.output_features}
input_steps = [
steps.rename_observations,
steps.add_batch_dim,
RenameObservationsProcessorStep(rename_map={}),
AddBatchDimensionProcessorStep(),
TokenizerProcessorStep(
tokenizer_name=config.tokenizer_name,
max_length=config.tokenizer_max_length,
@@ -68,15 +74,32 @@ def make_xvla_pre_post_processors(
XVLAImageToFloatProcessorStep(),
XVLAImageNetNormalizeProcessorStep(),
XVLAAddDomainIdProcessorStep(),
steps.to_device,
steps.normalize,
DeviceProcessorStep(device=config.device),
NormalizerProcessorStep(
features=features, norm_map=config.normalization_mapping, stats=dataset_stats
),
]
output_steps = [
steps.unnormalize,
steps.to_cpu,
UnnormalizerProcessorStep(
features=config.output_features,
norm_map=config.normalization_mapping,
stats=dataset_stats,
),
DeviceProcessorStep(device="cpu"),
]
return make_policy_processor_pipelines(input_steps=input_steps, output_steps=output_steps)
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
# Custom XVLA processor steps
+16 -10
View File
@@ -42,14 +42,10 @@ from .delta_action_processor import MapDeltaActionToRobotActionStep, MapTensorTo
from .device_processor import DeviceProcessorStep
from .env_processor import IsaaclabArenaProcessorStep, LiberoProcessorStep
from .factory import (
DefaultPolicyProcessorSteps,
make_default_policy_processor_steps,
make_default_pre_post_processors,
make_default_processors,
make_default_robot_action_processor,
make_default_robot_observation_processor,
make_default_teleop_action_processor,
make_policy_processor_pipelines,
)
from .gym_action_processor import (
Numpy2TorchActionProcessorStep,
@@ -93,8 +89,15 @@ from .policy_robot_bridge import (
from .relative_action_processor import (
AbsoluteActionsProcessorStep,
RelativeActionsProcessorStep,
relative_action_output_dim,
rotation_6d_to_rotvec,
rotvec_to_rotation_6d,
to_absolute_actions,
to_absolute_se3_pose,
to_absolute_se3_pose_6d,
to_relative_actions,
to_relative_se3_pose,
to_relative_se3_pose_6d,
)
from .rename_processor import RenameObservationsProcessorStep, rename_stats
from .tokenizer_processor import ActionTokenizerProcessorStep, TokenizerProcessorStep
@@ -133,16 +136,21 @@ __all__ = [
"ImageCropResizeProcessorStep",
"InfoProcessorStep",
"InterventionActionProcessorStep",
"DefaultPolicyProcessorSteps",
"make_default_policy_processor_steps",
"make_default_pre_post_processors",
"make_default_processors",
"make_default_teleop_action_processor",
"make_default_robot_action_processor",
"make_default_robot_observation_processor",
"make_policy_processor_pipelines",
"AbsoluteActionsProcessorStep",
"RelativeActionsProcessorStep",
"relative_action_output_dim",
"rotation_6d_to_rotvec",
"rotvec_to_rotation_6d",
"to_absolute_actions",
"to_absolute_se3_pose",
"to_absolute_se3_pose_6d",
"to_relative_actions",
"to_relative_se3_pose",
"to_relative_se3_pose_6d",
"MapDeltaActionToRobotActionStep",
"MapTensorToDeltaActionDictStep",
"NewLineTaskProcessorStep",
@@ -176,8 +184,6 @@ __all__ = [
"transition_to_batch",
"TransitionKey",
"TruncatedProcessorStep",
"to_absolute_actions",
"to_relative_actions",
"UnnormalizerProcessorStep",
"VanillaObservationProcessorStep",
]
+2 -114
View File
@@ -14,33 +14,15 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from dataclasses import dataclass
from typing import Any
from lerobot.types import RobotAction, RobotObservation
import torch
from lerobot.configs.policies import PreTrainedConfig
from lerobot.types import PolicyAction, RobotAction, RobotObservation
from lerobot.utils.constants import POLICY_POSTPROCESSOR_DEFAULT_NAME, POLICY_PREPROCESSOR_DEFAULT_NAME
from .batch_processor import AddBatchDimensionProcessorStep
from .converters import (
observation_to_transition,
policy_action_to_transition,
robot_action_observation_to_transition,
transition_to_observation,
transition_to_policy_action,
transition_to_robot_action,
)
from .device_processor import DeviceProcessorStep
from .normalize_processor import NormalizerProcessorStep, UnnormalizerProcessorStep
from .pipeline import (
IdentityProcessorStep,
PolicyProcessorPipeline,
ProcessorStep,
RobotProcessorPipeline,
)
from .rename_processor import RenameObservationsProcessorStep
from .pipeline import IdentityProcessorStep, RobotProcessorPipeline
def make_default_teleop_action_processor() -> RobotProcessorPipeline[
@@ -79,97 +61,3 @@ def make_default_processors():
robot_action_processor = make_default_robot_action_processor()
robot_observation_processor = make_default_robot_observation_processor()
return (teleop_action_processor, robot_action_processor, robot_observation_processor)
@dataclass
class DefaultPolicyProcessorSteps:
"""The canonical processor steps shared by most policies' pre/post pipelines.
Policies compose these in their own order (step ORDER is a Hub-serialized contract
and intentionally stays explicit per policy) and interleave their custom steps.
"""
rename_observations: RenameObservationsProcessorStep
add_batch_dim: AddBatchDimensionProcessorStep
to_device: DeviceProcessorStep
normalize: NormalizerProcessorStep
unnormalize: UnnormalizerProcessorStep
to_cpu: DeviceProcessorStep
def make_default_policy_processor_steps(
config: PreTrainedConfig,
dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None,
*,
normalizer_device: torch.device | str | None = None,
) -> DefaultPolicyProcessorSteps:
"""Construct the canonical policy processor steps from a policy config.
Args:
config: A `PreTrainedConfig` providing `device`, `input_features`,
`output_features` and `normalization_mapping`.
dataset_stats: Dataset statistics used for (un)normalization.
normalizer_device: Device passed to `NormalizerProcessorStep` (some policies pin
their normalization stats to the policy device; most leave it unset).
"""
return DefaultPolicyProcessorSteps(
rename_observations=RenameObservationsProcessorStep(rename_map={}),
add_batch_dim=AddBatchDimensionProcessorStep(),
to_device=DeviceProcessorStep(device=config.device),
normalize=NormalizerProcessorStep(
features={**config.input_features, **config.output_features},
norm_map=config.normalization_mapping,
stats=dataset_stats,
device=normalizer_device,
),
unnormalize=UnnormalizerProcessorStep(
features=config.output_features, norm_map=config.normalization_mapping, stats=dataset_stats
),
to_cpu=DeviceProcessorStep(device="cpu"),
)
def make_policy_processor_pipelines(
input_steps: list[ProcessorStep],
output_steps: list[ProcessorStep],
) -> tuple[
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]],
PolicyProcessorPipeline[PolicyAction, PolicyAction],
]:
"""Wrap pre/post step lists into the canonical policy pipeline pair.
Uses the standard pipeline names (which determine the serialized JSON filenames on
the Hub) and the standard policy-action converters on the postprocessor.
"""
return (
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
steps=input_steps,
name=POLICY_PREPROCESSOR_DEFAULT_NAME,
),
PolicyProcessorPipeline[PolicyAction, PolicyAction](
steps=output_steps,
name=POLICY_POSTPROCESSOR_DEFAULT_NAME,
to_transition=policy_action_to_transition,
to_output=transition_to_policy_action,
),
)
def make_default_pre_post_processors(
config: PreTrainedConfig,
dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None,
*,
normalizer_device: torch.device | str | None = None,
) -> tuple[
PolicyProcessorPipeline[dict[str, Any], dict[str, Any]],
PolicyProcessorPipeline[PolicyAction, PolicyAction],
]:
"""The pure-scaffold policy pipeline pair: Rename -> Batch -> Device -> Normalize,
and Unnormalize -> Device(cpu). Policies with custom steps or a different step order
compose `make_default_policy_processor_steps` themselves instead.
"""
s = make_default_policy_processor_steps(config, dataset_stats, normalizer_device=normalizer_device)
return make_policy_processor_pipelines(
input_steps=[s.rename_observations, s.add_batch_dim, s.to_device, s.normalize],
output_steps=[s.unnormalize, s.to_cpu],
)
@@ -21,7 +21,7 @@ from torch import Tensor
from lerobot.configs import PipelineFeatureType, PolicyFeature
from lerobot.types import EnvTransition, TransitionKey
from lerobot.utils.constants import OBS_STATE
from lerobot.utils.constants import ACTION, OBS_STATE
from .delta_action_processor import MapDeltaActionToRobotActionStep, MapTensorToDeltaActionDictStep
from .pipeline import ProcessorStep, ProcessorStepRegistry
@@ -34,57 +34,399 @@ __all__ = [
"AbsoluteActionsProcessorStep",
"to_relative_actions",
"to_absolute_actions",
"to_relative_se3_pose",
"to_absolute_se3_pose",
"to_relative_se3_pose_6d",
"to_absolute_se3_pose_6d",
"rotation_6d_to_rotvec",
"rotvec_to_rotation_6d",
"relative_action_output_dim",
]
def to_relative_actions(actions: Tensor, state: Tensor, mask: Sequence[bool]) -> Tensor:
"""Convert absolute actions to relative: relative = action - state (for masked dims).
def _rotvec_to_quaternion(rotvec: Tensor) -> Tensor:
angle = torch.linalg.vector_norm(rotvec, dim=-1, keepdim=True)
angle_sq = angle.square()
small_scale = 0.5 - angle_sq / 48.0 + angle_sq.square() / 3840.0
scale = torch.where(angle > 1e-6, torch.sin(angle / 2.0) / angle.clamp_min(1e-12), small_scale)
return torch.cat((torch.cos(angle / 2.0), rotvec * scale), dim=-1)
def _quaternion_to_rotvec(quaternion: Tensor) -> Tensor:
quaternion = quaternion / torch.linalg.vector_norm(quaternion, dim=-1, keepdim=True).clamp_min(1e-12)
quaternion = quaternion * torch.where(quaternion[..., :1] < 0, -1.0, 1.0)
vector = quaternion[..., 1:]
sin_half_angle = torch.linalg.vector_norm(vector, dim=-1, keepdim=True)
angle = 2.0 * torch.atan2(sin_half_angle, quaternion[..., :1].clamp_min(0.0))
small_scale = 2.0 + sin_half_angle.square() / 3.0
scale = torch.where(
sin_half_angle > 1e-6,
angle / sin_half_angle.clamp_min(1e-12),
small_scale,
)
return vector * scale
def _quaternion_multiply(left: Tensor, right: Tensor) -> Tensor:
left_w, left_xyz = left[..., :1], left[..., 1:]
right_w, right_xyz = right[..., :1], right[..., 1:]
return torch.cat(
(
left_w * right_w - (left_xyz * right_xyz).sum(dim=-1, keepdim=True),
left_w * right_xyz + right_w * left_xyz + torch.linalg.cross(left_xyz, right_xyz, dim=-1),
),
dim=-1,
)
def _quaternion_conjugate(quaternion: Tensor) -> Tensor:
return torch.cat((quaternion[..., :1], -quaternion[..., 1:]), dim=-1)
def _quaternion_rotate(quaternion: Tensor, vector: Tensor) -> Tensor:
quaternion_xyz = quaternion[..., 1:]
uv = torch.linalg.cross(quaternion_xyz, vector, dim=-1)
uuv = torch.linalg.cross(quaternion_xyz, uv, dim=-1)
return vector + 2.0 * (quaternion[..., :1] * uv + uuv)
def _quaternion_to_matrix(quaternion: Tensor) -> Tensor:
quaternion = quaternion / torch.linalg.vector_norm(quaternion, dim=-1, keepdim=True).clamp_min(1e-12)
w, x, y, z = quaternion.unbind(-1)
two_s = 2.0
return torch.stack(
(
1.0 - two_s * (y * y + z * z),
two_s * (x * y - z * w),
two_s * (x * z + y * w),
two_s * (x * y + z * w),
1.0 - two_s * (x * x + z * z),
two_s * (y * z - x * w),
two_s * (x * z - y * w),
two_s * (y * z + x * w),
1.0 - two_s * (x * x + y * y),
),
dim=-1,
).reshape(quaternion.shape[:-1] + (3, 3))
def _matrix_to_quaternion(matrix: Tensor) -> Tensor:
"""Convert proper rotation matrices to normalized ``[w, x, y, z]`` quaternions."""
if matrix.shape[-2:] != (3, 3):
raise ValueError(f"Rotation matrices must have shape (..., 3, 3), got {matrix.shape}")
m00 = matrix[..., 0, 0]
m01 = matrix[..., 0, 1]
m02 = matrix[..., 0, 2]
m10 = matrix[..., 1, 0]
m11 = matrix[..., 1, 1]
m12 = matrix[..., 1, 2]
m20 = matrix[..., 2, 0]
m21 = matrix[..., 2, 1]
m22 = matrix[..., 2, 2]
# Each row is a quaternion candidate scaled by the magnitude of its
# best-conditioned component. Selecting the largest component avoids the
# trace singularity at rotations close to pi.
q_abs = torch.sqrt(
torch.clamp(
torch.stack(
(
1.0 + m00 + m11 + m22,
1.0 + m00 - m11 - m22,
1.0 - m00 + m11 - m22,
1.0 - m00 - m11 + m22,
),
dim=-1,
),
min=0.0,
)
)
quat_by_rijk = torch.stack(
(
torch.stack((q_abs[..., 0].square(), m21 - m12, m02 - m20, m10 - m01), dim=-1),
torch.stack((m21 - m12, q_abs[..., 1].square(), m10 + m01, m02 + m20), dim=-1),
torch.stack((m02 - m20, m10 + m01, q_abs[..., 2].square(), m12 + m21), dim=-1),
torch.stack((m10 - m01, m02 + m20, m12 + m21, q_abs[..., 3].square()), dim=-1),
),
dim=-2,
)
candidates = quat_by_rijk / (2.0 * q_abs[..., :, None].clamp_min(0.1))
best = torch.nn.functional.one_hot(q_abs.argmax(dim=-1), num_classes=4).to(dtype=matrix.dtype)
quaternion = (candidates * best[..., :, None]).sum(dim=-2)
return quaternion / torch.linalg.vector_norm(quaternion, dim=-1, keepdim=True).clamp_min(1e-12)
def rotvec_to_rotation_6d(rotvec: Tensor) -> Tensor:
"""Encode an axis-angle rotation as the first two rotation-matrix rows."""
matrix = _quaternion_to_matrix(_rotvec_to_quaternion(rotvec))
return matrix[..., :2, :].reshape(matrix.shape[:-2] + (6,))
def rotation_6d_to_rotvec(rotation_6d: Tensor) -> Tensor:
"""Decode two predicted 3-D vectors into an axis-angle rotation.
Gram-Schmidt orthonormalization follows the continuous 6-D rotation
representation. Degenerate predictions fail closed instead of producing an
invalid physical rotation.
"""
if rotation_6d.shape[-1] != 6:
raise ValueError(f"6-D rotations must have six values, got {rotation_6d.shape}")
first = rotation_6d[..., :3]
second = rotation_6d[..., 3:]
first_norm = torch.linalg.vector_norm(first, dim=-1, keepdim=True)
first_unit = first / first_norm.clamp_min(1e-12)
second_orthogonal = second - (first_unit * second).sum(dim=-1, keepdim=True) * first_unit
second_norm = torch.linalg.vector_norm(second_orthogonal, dim=-1, keepdim=True)
if bool(torch.any(first_norm <= 1e-8)) or bool(torch.any(second_norm <= 1e-8)):
raise ValueError("Cannot decode a degenerate 6-D rotation prediction")
second_unit = second_orthogonal / second_norm
third_unit = torch.linalg.cross(first_unit, second_unit, dim=-1)
matrix = torch.stack((first_unit, second_unit, third_unit), dim=-2)
return _quaternion_to_rotvec(_matrix_to_quaternion(matrix))
def to_relative_se3_pose(target_pose: Tensor, reference_pose: Tensor) -> Tensor:
"""Encode a pose as ``inv(T_reference) @ T_target``.
Poses use ``[x, y, z, rx, ry, rz]`` with an axis-angle rotation vector.
The relative translation is therefore expressed in the reference EE frame.
"""
if target_pose.shape[-1] != 6 or reference_pose.shape[-1] != 6:
raise ValueError("SE(3) poses must have six values: xyz followed by a rotation vector")
reference_quaternion = _rotvec_to_quaternion(reference_pose[..., 3:])
target_quaternion = _rotvec_to_quaternion(target_pose[..., 3:])
inverse_reference_quaternion = _quaternion_conjugate(reference_quaternion)
relative_translation = _quaternion_rotate(
inverse_reference_quaternion, target_pose[..., :3] - reference_pose[..., :3]
)
relative_quaternion = _quaternion_multiply(inverse_reference_quaternion, target_quaternion)
return torch.cat((relative_translation, _quaternion_to_rotvec(relative_quaternion)), dim=-1)
def to_absolute_se3_pose(relative_pose: Tensor, reference_pose: Tensor) -> Tensor:
"""Decode a pose with ``T_target = T_reference @ T_relative``."""
if relative_pose.shape[-1] != 6 or reference_pose.shape[-1] != 6:
raise ValueError("SE(3) poses must have six values: xyz followed by a rotation vector")
reference_quaternion = _rotvec_to_quaternion(reference_pose[..., 3:])
relative_quaternion = _rotvec_to_quaternion(relative_pose[..., 3:])
target_translation = reference_pose[..., :3] + _quaternion_rotate(
reference_quaternion, relative_pose[..., :3]
)
target_quaternion = _quaternion_multiply(reference_quaternion, relative_quaternion)
return torch.cat((target_translation, _quaternion_to_rotvec(target_quaternion)), dim=-1)
def to_relative_se3_pose_6d(target_pose: Tensor, reference_pose: Tensor) -> Tensor:
"""Encode ``inv(T_reference) @ T_target`` as xyz plus continuous 6-D rotation."""
relative_pose = to_relative_se3_pose(target_pose, reference_pose)
return torch.cat((relative_pose[..., :3], rotvec_to_rotation_6d(relative_pose[..., 3:])), dim=-1)
def to_absolute_se3_pose_6d(relative_pose: Tensor, reference_pose: Tensor) -> Tensor:
"""Decode xyz plus continuous 6-D rotation with ``T_target = T_reference @ T_relative``."""
if relative_pose.shape[-1] != 9:
raise ValueError("6-D encoded SE(3) poses must have nine values: xyz plus rotation-6D")
relative_rotvec_pose = torch.cat(
(relative_pose[..., :3], rotation_6d_to_rotvec(relative_pose[..., 3:])), dim=-1
)
return to_absolute_se3_pose(relative_rotvec_pose, reference_pose)
def _broadcast_reference(actions: Tensor, state: Tensor) -> Tensor:
if state.device != actions.device or state.dtype != actions.dtype:
state = state.to(device=actions.device, dtype=actions.dtype)
if actions.ndim == state.ndim + 1:
state = state.unsqueeze(-2)
return state
def _validate_se3_pose_groups(
pose_representation: str,
se3_pose_groups: Sequence[Sequence[int]] | None,
mask: Sequence[bool],
action_dim: int,
) -> list[list[int]]:
if pose_representation not in {"componentwise", "se3", "se3_6d"}:
raise ValueError(
f"Unsupported pose_representation={pose_representation!r}; expected "
"'componentwise', 'se3', or 'se3_6d'"
)
if pose_representation == "componentwise":
return []
if not se3_pose_groups:
raise ValueError(
f"pose_representation={pose_representation!r} requires at least one six-index se3_pose_group"
)
normalized_groups: list[list[int]] = []
used_indices: set[int] = set()
for raw_group in se3_pose_groups:
group = [int(index) for index in raw_group]
if len(group) != 6:
raise ValueError(f"Each SE(3) pose group must contain six indices, got {group}")
if len(set(group)) != 6 or any(index < 0 or index >= action_dim for index in group):
raise ValueError(f"Invalid SE(3) pose group for action_dim={action_dim}: {group}")
if pose_representation == "se3_6d" and group != list(range(group[0], group[0] + 6)):
raise ValueError("se3_6d pose groups must contain six contiguous ascending indices")
if any(index >= len(mask) for index in group):
raise ValueError(f"SE(3) pose group lies outside the relative mask: {group}")
if used_indices.intersection(group):
raise ValueError(f"SE(3) pose groups must not overlap: {group}")
group_mask = [bool(mask[index]) for index in group]
if any(group_mask) and not all(group_mask):
raise ValueError(f"An SE(3) pose group must be wholly relative or wholly absolute: {group}")
used_indices.update(group)
if all(group_mask):
normalized_groups.append(group)
return normalized_groups
def relative_action_output_dim(
source_dim: int,
pose_representation: str,
se3_pose_groups: Sequence[Sequence[int]] | None,
) -> int:
"""Return the model-space action width for a source action width."""
if pose_representation != "se3_6d":
return source_dim
groups = se3_pose_groups or []
return source_dim + 3 * len(groups)
def _expand_se3_6d_actions(
actions: Tensor,
state: Tensor,
groups: Sequence[Sequence[int]],
) -> Tensor:
group_by_start = {group[0]: list(group) for group in groups}
grouped_indices = {index for group in groups for index in group}
parts: list[Tensor] = []
for index in range(actions.shape[-1]):
group = group_by_start.get(index)
if group is not None:
parts.append(to_relative_se3_pose_6d(actions[..., group], state[..., group]))
elif index not in grouped_indices:
parts.append(actions[..., index : index + 1])
return torch.cat(parts, dim=-1)
def _collapse_se3_6d_actions(
actions: Tensor,
state: Tensor,
mask: Sequence[bool],
groups: Sequence[Sequence[int]],
) -> Tensor:
source_dim = len(mask)
expected_dim = relative_action_output_dim(source_dim, "se3_6d", groups)
if actions.shape[-1] != expected_dim:
raise ValueError(
f"Expected se3_6d action width {expected_dim} for source width {source_dim}, "
f"got {actions.shape[-1]}"
)
group_by_start = {group[0]: list(group) for group in groups}
grouped_indices = {index for group in groups for index in group}
parts: list[Tensor] = []
cursor = 0
for index in range(source_dim):
group = group_by_start.get(index)
if group is not None:
parts.append(to_absolute_se3_pose_6d(actions[..., cursor : cursor + 9], state[..., group]))
cursor += 9
elif index not in grouped_indices:
value = actions[..., cursor : cursor + 1]
if mask[index]:
value = value + state[..., index : index + 1]
parts.append(value)
cursor += 1
if cursor != actions.shape[-1]:
raise RuntimeError(f"Consumed {cursor} action values from width {actions.shape[-1]}")
return torch.cat(parts, dim=-1)
def to_relative_actions(
actions: Tensor,
state: Tensor,
mask: Sequence[bool],
*,
pose_representation: str = "componentwise",
se3_pose_groups: Sequence[Sequence[int]] | None = None,
) -> Tensor:
"""Convert absolute actions to a configured relative representation.
Component-wise mode computes ``action - state``. SE(3) modes compute
``inv(T_state) @ T_action`` for each configured pose group. ``se3_6d``
replaces each three-value relative rotation vector with its continuous
six-value encoding, increasing the output width by three per pose group.
Args:
actions: (B, T, action_dim) or (B, action_dim).
state: (B, state_dim). Broadcast across time dimension.
mask: Which dims to convert. Can be shorter than action_dim.
"""
groups = _validate_se3_pose_groups(pose_representation, se3_pose_groups, mask, actions.shape[-1])
mask_t = torch.tensor(mask, dtype=actions.dtype, device=actions.device)
dims = mask_t.shape[0]
# Align state to the same device/dtype as actions. _last_state is cached before
# DeviceProcessorStep moves the transition, so it can be on CPU while actions are on CUDA.
if state.device != actions.device or state.dtype != actions.dtype:
state = state.to(device=actions.device, dtype=actions.dtype)
state_offset = state[..., :dims] * mask_t
if actions.ndim == 3:
state_offset = state_offset.unsqueeze(-2)
state = _broadcast_reference(actions, state)
component_mask = mask_t.clone()
for group in groups:
component_mask[group] = 0
state_offset = state[..., :dims] * component_mask
actions = actions.clone()
actions[..., :dims] -= state_offset
if pose_representation == "se3_6d":
return _expand_se3_6d_actions(actions, state, groups)
for group in groups:
actions[..., group] = to_relative_se3_pose(actions[..., group], state[..., group])
return actions
def to_absolute_actions(actions: Tensor, state: Tensor, mask: Sequence[bool]) -> Tensor:
"""Convert relative actions back to absolute: absolute = relative + state (for masked dims).
def to_absolute_actions(
actions: Tensor,
state: Tensor,
mask: Sequence[bool],
*,
pose_representation: str = "componentwise",
se3_pose_groups: Sequence[Sequence[int]] | None = None,
) -> Tensor:
"""Convert relative actions back to absolute actions.
Component-wise mode computes ``relative + state``. SE(3) mode computes
``T_state @ T_relative`` for each configured pose group.
Args:
actions: (B, T, action_dim) or (B, action_dim).
state: (B, state_dim). Broadcast across time dimension.
mask: Which dims to convert. Can be shorter than action_dim.
"""
source_dim = len(mask)
groups = _validate_se3_pose_groups(pose_representation, se3_pose_groups, mask, source_dim)
state = _broadcast_reference(actions, state)
if pose_representation == "se3_6d":
return _collapse_se3_6d_actions(actions, state, mask, groups)
mask_t = torch.tensor(mask, dtype=actions.dtype, device=actions.device)
dims = mask_t.shape[0]
# Align state to the same device/dtype as actions. _last_state is cached before
# DeviceProcessorStep moves the transition, so it can be on CPU while actions are on CUDA.
if state.device != actions.device or state.dtype != actions.dtype:
state = state.to(device=actions.device, dtype=actions.dtype)
state_offset = state[..., :dims] * mask_t
if actions.ndim == 3:
state_offset = state_offset.unsqueeze(-2)
state = _broadcast_reference(actions, state)
component_mask = mask_t.clone()
for group in groups:
component_mask[group] = 0
state_offset = state[..., :dims] * component_mask
actions = actions.clone()
actions[..., :dims] += state_offset
for group in groups:
actions[..., group] = to_absolute_se3_pose(actions[..., group], state[..., group])
return actions
@ProcessorStepRegistry.register("relative_actions_processor")
@dataclass
class RelativeActionsProcessorStep(ProcessorStep):
"""Converts absolute actions to relative actions (action -= state) for masked dimensions.
"""Converts absolute actions to the configured relative representation.
Mirrors OpenPI's DeltaActions transform. Applied during preprocessing so the model
trains on relative offsets instead of absolute positions.
@@ -101,7 +443,10 @@ class RelativeActionsProcessorStep(ProcessorStep):
enabled: bool = False
exclude_joints: list[str] = field(default_factory=list)
action_names: list[str] | None = None
pose_representation: str = "componentwise"
se3_pose_groups: list[list[int]] = field(default_factory=list)
_last_state: torch.Tensor | None = field(default=None, init=False, repr=False)
_last_mask: list[bool] | None = field(default=None, init=False, repr=False)
def _build_mask(self, action_dim: int) -> list[bool]:
if not self.exclude_joints or self.action_names is None:
@@ -126,42 +471,78 @@ class RelativeActionsProcessorStep(ProcessorStep):
observation = transition.get(TransitionKey.OBSERVATION, {})
state = observation.get(OBS_STATE) if observation else None
# Always cache state for the paired AbsoluteActionsProcessorStep.
if state is not None:
self._last_state = state
# State history has shape (B, H, D). Relative actions are referenced to
# the newest proprioceptive state, not the whole history tensor.
reference_state = state[:, -1] if state is not None and state.ndim == 3 else state
# Always cache state for the paired AbsoluteActionsProcessorStep
if reference_state is not None:
self._last_state = reference_state
self._last_mask = self._build_mask(reference_state.shape[-1])
if not self.enabled:
return transition
new_transition = transition.copy()
action = new_transition.get(TransitionKey.ACTION)
if action is None or state is None:
if action is None or reference_state is None:
return new_transition
mask = self._build_mask(action.shape[-1])
new_transition[TransitionKey.ACTION] = to_relative_actions(action, state, mask)
mask = self._last_mask or self._build_mask(action.shape[-1])
new_transition[TransitionKey.ACTION] = to_relative_actions(
action,
reference_state,
mask,
pose_representation=self.pose_representation,
se3_pose_groups=self.se3_pose_groups,
)
return new_transition
def get_cached_state(self) -> torch.Tensor | None:
"""Return the cached ``observation.state`` used as the reference point for relative/absolute action conversions."""
return self._last_state
def set_cached_state(self, state: torch.Tensor | None) -> None:
"""Override the cached anchor state, e.g. to re-pin a chunk's anchor after the
per-tick pipeline overwrote it (see ``SyncInferenceEngine``)."""
self._last_state = state
def get_cached_mask(self) -> list[bool] | None:
"""Return the source-space mask cached with the latest state."""
return self._last_mask
def reset(self) -> None:
"""Drop the inference reference so it cannot leak between sessions."""
self._last_state = None
self._last_mask = None
def get_config(self) -> dict[str, Any]:
return {
"enabled": self.enabled,
"exclude_joints": self.exclude_joints,
"action_names": self.action_names,
"pose_representation": self.pose_representation,
"se3_pose_groups": self.se3_pose_groups,
}
def transform_features(
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
return features
if not self.enabled or self.pose_representation != "se3_6d":
return features
transformed = {feature_type: dict(feature_group) for feature_type, feature_group in features.items()}
for feature_group in transformed.values():
action_feature = feature_group.get(ACTION)
if action_feature is None:
continue
source_dim = len(self.action_names) if self.action_names is not None else action_feature.shape[-1]
model_dim = relative_action_output_dim(source_dim, self.pose_representation, self.se3_pose_groups)
if action_feature.shape[-1] == source_dim:
feature_group[ACTION] = PolicyFeature(
type=action_feature.type,
shape=(model_dim,),
)
elif action_feature.shape[-1] != model_dim:
raise ValueError(
f"Expected source/model action width {source_dim}/{model_dim}, "
f"got {action_feature.shape[-1]}"
)
return transformed
@ProcessorStepRegistry.register("absolute_actions_processor")
@@ -203,8 +584,16 @@ class AbsoluteActionsProcessorStep(ProcessorStep):
if action is None:
return new_transition
mask = self.relative_step._build_mask(action.shape[-1])
new_transition[TransitionKey.ACTION] = to_absolute_actions(action, cached_state, mask)
mask = self.relative_step.get_cached_mask()
if mask is None:
mask = self.relative_step._build_mask(cached_state.shape[-1])
new_transition[TransitionKey.ACTION] = to_absolute_actions(
action,
cached_state,
mask,
pose_representation=self.relative_step.pose_representation,
se3_pose_groups=self.relative_step.se3_pose_groups,
)
return new_transition
def get_config(self) -> dict[str, Any]:
+21 -4
View File
@@ -21,6 +21,8 @@ from pathlib import Path
from tempfile import TemporaryDirectory
from typing import TYPE_CHECKING, Any, TypeVar
import packaging
import safetensors
from huggingface_hub import HfApi, ModelCard, ModelCardData, hf_hub_download
from huggingface_hub.constants import SAFETENSORS_SINGLE_FILE
from huggingface_hub.errors import HfHubHTTPError
@@ -28,7 +30,6 @@ from safetensors.torch import load_model as load_model_as_safetensor, save_model
from torch import Tensor, nn
from lerobot.configs.rewards import RewardModelConfig
from lerobot.utils.device_utils import resolve_safetensors_device
from lerobot.utils.hub import HubMixin
if TYPE_CHECKING:
@@ -128,13 +129,29 @@ class PreTrainedRewardModel(nn.Module, HubMixin, abc.ABC):
@classmethod
def _load_as_safetensor(cls, model: T, model_file: str, map_location: str, strict: bool) -> T:
missing_keys, unexpected_keys = load_model_as_safetensor(
model, model_file, strict=strict, device=resolve_safetensors_device(map_location)
)
# Create base kwargs
kwargs = {"strict": strict}
# Add device parameter for newer versions that support it
if packaging.version.parse(safetensors.__version__) >= packaging.version.parse("0.4.3"):
kwargs["device"] = map_location
# Load the model with appropriate kwargs
missing_keys, unexpected_keys = load_model_as_safetensor(model, model_file, **kwargs)
if missing_keys:
logging.warning(f"Missing key(s) when loading model: {missing_keys}")
if unexpected_keys:
logging.warning(f"Unexpected key(s) when loading model: {unexpected_keys}")
# For older versions, manually move to device if needed
if "device" not in kwargs and map_location != "cpu":
logging.warning(
"Loading model weights on other devices than 'cpu' is not supported natively in your version of safetensors."
" This means that the model is loaded on 'cpu' first and then copied to the device."
" This leads to a slower loading time."
" Please update safetensors to version 0.4.3 or above for improved performance."
)
model.to(map_location)
return model
def get_optim_params(self):
+76
View File
@@ -43,14 +43,17 @@ from lerobot.processor import (
make_default_processors,
rename_stats,
)
from lerobot.processor.relative_action_processor import RelativeActionsProcessorStep
from lerobot.robots import make_robot_from_config
from lerobot.teleoperators import Teleoperator, make_teleoperator_from_config
from lerobot.utils.constants import OBS_STATE
from lerobot.utils.feature_utils import combine_feature_dicts, hw_to_dataset_features
from .configs import BaseStrategyConfig, DAggerStrategyConfig, RolloutConfig
from .inference import (
InferenceEngine,
RTCInferenceConfig,
SyncInferenceConfig,
create_inference_engine,
)
from .robot_wrapper import ThreadSafeRobot
@@ -58,6 +61,35 @@ from .robot_wrapper import ThreadSafeRobot
logger = logging.getLogger(__name__)
def _validate_trained_rtc_rollout_config(policy_config, inference_config: RTCInferenceConfig) -> None:
"""Fail fast when rollout cannot retain every trained RTC prefix."""
rtc = inference_config.rtc
if not rtc.enabled or rtc.mode != "trained":
return
if policy_config.type not in {"pi05", "pi052"}:
raise ValueError(
"--inference.rtc.mode=trained currently requires a PI05-compatible checkpoint; "
f"got policy type {policy_config.type!r}."
)
training_max_delay = int(getattr(policy_config, "rtc_training_max_delay", 0))
if training_max_delay <= 0:
raise ValueError(
"--inference.rtc.mode=trained requires a checkpoint trained with "
"--policy.rtc_training_max_delay > 0."
)
if rtc.execution_horizon < training_max_delay:
raise ValueError(
f"--inference.rtc.execution_horizon ({rtc.execution_horizon}) must be at least the "
f"checkpoint's rtc_training_max_delay ({training_max_delay})."
)
if inference_config.queue_threshold < training_max_delay:
raise ValueError(
f"--inference.queue_threshold ({inference_config.queue_threshold}) must be at least the "
f"checkpoint's rtc_training_max_delay ({training_max_delay})."
)
def _resolve_action_key_order(
policy_action_names: list[str] | None, dataset_action_names: list[str]
) -> list[str]:
@@ -78,6 +110,26 @@ def _resolve_action_key_order(
return policy_action_names
def _align_relative_state_feature_order(
hw_features: dict[str, dict], policy_action_names: list[str] | None
) -> dict[str, dict]:
"""Align policy-facing state with named relative-action dimensions."""
if not policy_action_names or OBS_STATE not in hw_features:
return hw_features
state_feature = hw_features[OBS_STATE]
state_names = state_feature.get("names")
if not state_names or len(state_names) != len(policy_action_names):
return hw_features
if set(state_names) != set(policy_action_names) or state_names == policy_action_names:
return hw_features
aligned = dict(hw_features)
aligned[OBS_STATE] = {**state_feature, "names": list(policy_action_names)}
logger.info("Aligned relative-action state order with checkpoint action names")
return aligned
# ---------------------------------------------------------------------------
# Sub-contexts
# ---------------------------------------------------------------------------
@@ -176,6 +228,9 @@ def build_rollout_context(
policy_config = cfg.policy
policy_class = get_policy_class(policy_config.type)
if is_rtc:
_validate_trained_rtc_rollout_config(policy_config, cfg.inference)
if hasattr(policy_config, "compile_model"):
policy_config.compile_model = cfg.use_torch_compile
@@ -397,6 +452,27 @@ def build_rollout_context(
},
)
relative_action_step = next(
(
step
for step in getattr(preprocessor, "steps", ())
if isinstance(step, RelativeActionsProcessorStep) and step.enabled
),
None,
)
if relative_action_step is not None:
relative_action_names = relative_action_step.action_names or policy_action_names
hw_features = _align_relative_state_feature_order(
hw_features,
list(relative_action_names) if relative_action_names else None,
)
if isinstance(cfg.inference, SyncInferenceConfig) and relative_action_step is not None:
raise NotImplementedError(
"SyncInferenceEngine does not support policies with relative actions for now."
"Use --inference.type=rtc or remove relative action processor steps from the policy pipeline."
)
# --- 7. Inference strategy (needs policy + pre/post + hardware) --
logger.info(
"Creating inference engine (type=%s)...",
+95 -2
View File
@@ -57,6 +57,18 @@ _RTC_MAX_CONSECUTIVE_ERRORS: int = 10
_RTC_JOIN_TIMEOUT_S: float = 3.0
class _FatalRTCInferenceError(RuntimeError):
"""Base class for RTC errors that cannot become valid after a retry."""
class _TrainedRTCDelayExceededError(_FatalRTCInferenceError):
"""Raised when measured latency exceeds a trained RTC checkpoint's support."""
class _TrainedRTCPrefixUnavailableError(_FatalRTCInferenceError):
"""Raised when the queue cannot provide the prefix used for conditioning."""
# ---------------------------------------------------------------------------
# RTC helpers
# ---------------------------------------------------------------------------
@@ -76,6 +88,50 @@ def _normalize_prev_actions_length(prev_actions: torch.Tensor, target_steps: int
return padded
def _trained_rtc_chunk_can_merge(
*,
conditioned_delay: int,
measured_delay: int,
training_max_delay: int,
has_previous_actions: bool,
) -> bool:
"""Check that a trained RTC chunk covers the overlap observed during inference."""
if not has_previous_actions:
return True
if measured_delay > training_max_delay:
raise _TrainedRTCDelayExceededError(
f"Measured RTC inference delay ({measured_delay}) exceeds the checkpoint's "
f"rtc_training_max_delay ({training_max_delay})."
)
return measured_delay <= conditioned_delay
def _estimate_rtc_delay(
*,
latency: float,
time_per_step: float,
mode: str,
training_max_delay: int,
has_previous_actions: bool,
) -> int:
"""Estimate overlap, using the trained capacity to bootstrap the first transition."""
if latency:
return math.ceil(latency / time_per_step)
if mode == "trained" and has_previous_actions:
return training_max_delay
return 0
def _validate_trained_rtc_prefix_available(*, conditioned_delay: int, available_steps: int) -> None:
"""Reject hard-prefix inference when the real queue is shorter than its delay."""
if conditioned_delay > available_steps:
raise _TrainedRTCPrefixUnavailableError(
f"Trained RTC needs {conditioned_delay} committed prefix actions, but the queue has "
f"only {available_steps}. Increase --inference.queue_threshold and "
"--inference.rtc.execution_horizon."
)
# ---------------------------------------------------------------------------
# RTCInferenceEngine
# ---------------------------------------------------------------------------
@@ -272,9 +328,23 @@ class RTCInferenceEngine(InferenceEngine):
current_time = time.perf_counter()
idx_before = queue.get_action_index()
prev_actions = queue.get_left_over()
has_previous_actions = prev_actions is not None and prev_actions.numel() > 0
training_max_delay = int(getattr(self._policy.config, "rtc_training_max_delay", 0))
latency = latency_tracker.max()
delay = math.ceil(latency / time_per_chunk) if latency else 0
delay = _estimate_rtc_delay(
latency=latency,
time_per_step=time_per_chunk,
mode=self._rtc_config.mode,
training_max_delay=training_max_delay,
has_previous_actions=has_previous_actions,
)
if self._rtc_config.mode == "trained" and delay > 0:
available_steps = 0 if prev_actions is None else prev_actions.shape[0]
_validate_trained_rtc_prefix_available(
conditioned_delay=delay,
available_steps=available_steps,
)
obs_batch = build_dataset_frame(self._hw_features, obs, prefix="observation")
obs_batch = prepare_observation_for_inference(
@@ -316,11 +386,32 @@ class RTCInferenceEngine(InferenceEngine):
inference_count += 1
consecutive_errors = 0
is_warmup = self._use_torch_compile and inference_count <= warmup_required
if is_warmup:
is_initial_trained_chunk = (
self._rtc_config.mode == "trained" and not has_previous_actions
)
if is_warmup or is_initial_trained_chunk:
latency_tracker.reset()
else:
latency_tracker.add(new_latency)
if (
not is_warmup
and self._rtc_config.mode == "trained"
and not _trained_rtc_chunk_can_merge(
conditioned_delay=delay,
measured_delay=new_delay,
training_max_delay=training_max_delay,
has_previous_actions=has_previous_actions,
)
):
logger.warning(
"Discarding trained RTC chunk: measured delay %d exceeded "
"conditioned delay %d; retrying with updated latency",
new_delay,
delay,
)
continue
queue.merge(original, processed, new_delay, idx_before)
if (
@@ -333,6 +424,8 @@ class RTCInferenceEngine(InferenceEngine):
logger.debug("RTC inference latency=%.2fs, queue=%d", new_latency, queue.qsize())
except _FatalRTCInferenceError:
raise
except Exception as e:
consecutive_errors += 1
logger.error(
+14 -73
View File
@@ -24,21 +24,26 @@ import torch
from lerobot.policies.pretrained import PreTrainedPolicy
from lerobot.policies.utils import make_robot_action, prepare_observation_for_inference
from lerobot.processor import PolicyProcessorPipeline, RelativeActionsProcessorStep
from lerobot.processor import PolicyProcessorPipeline
from .base import InferenceEngine
logger = logging.getLogger(__name__)
# Relative-action support: a predicted chunk of offsets is anchored to the robot
# state at prediction time, but the sync engine reruns the pre/post pipeline every
# tick, so ``RelativeActionsProcessorStep`` would re-anchor cached actions to the
# current (moved) state and drift through the chunk. We pin the anchor per chunk:
# a probe on the policy's public ``predict_action_chunk`` flags the ticks that
# predict a fresh chunk; on the others the engine restores the anchor the relative
# step overwrote. ``select_action`` stays on the hot path, so per-tick side effects
# (e.g. LingBot-VA keyframe feedback) are preserved.
# TODO(Steven): support relative-action policies. The per-tick flow refreshes
# ``RelativeActionsProcessorStep._last_state`` every call, so cached chunk
# actions popped on later ticks get reanchored to the *current* robot state and
# absolute targets drift through the chunk. Relative-action policies are
# rejected at context-build time today; RTC postprocesses the whole chunk and
# is unaffected.
#
# Candidate fix: drive the policy via ``predict_action_chunk`` and serve a
# local FIFO of postprocessed actions. Eliminates drift by construction and
# saves per-tick pre/post work, but bypasses ``select_action`` — needs
# fallbacks for SAC (raises), ACT temporal ensembling (ensembler lives in
# ``select_action``), and Diffusion-family (obs-history queues populated as a
# side effect of ``select_action``).
class SyncInferenceEngine(InferenceEngine):
@@ -68,31 +73,6 @@ class SyncInferenceEngine(InferenceEngine):
self._task = task
self._device = torch.device(device or "cpu")
self._robot_type = robot_type
# Find an enabled RelativeActionsProcessorStep to pin its anchor per chunk
# (see module comment), mirroring the RTC engine.
self._relative_step = next(
(
s
for s in getattr(preprocessor, "steps", ())
if isinstance(s, RelativeActionsProcessorStep) and s.enabled
),
None,
)
# Set by the probe for the current tick / ever, respectively.
self._chunk_predicted = False
self._ever_predicted_chunk = False
self._original_predict_action_chunk = None # set while the probe is installed
if self._relative_step is not None:
# ``action_names`` is optional on the step; fill it lazily from the
# policy/dataset so the relative<->absolute mask is built correctly. This is
# a deliberate engine->step side effect (the step is configured by its consumer).
if self._relative_step.action_names is None:
cfg_names = getattr(policy.config, "action_feature_names", None)
self._relative_step.action_names = list(cfg_names) if cfg_names else list(ordered_action_keys)
self._install_chunk_probe()
logger.info("Relative actions enabled: chunk anchor pinned per predicted chunk")
logger.info(
"SyncInferenceEngine initialized (device=%s, action_keys=%d)",
self._device,
@@ -105,11 +85,6 @@ class SyncInferenceEngine(InferenceEngine):
def stop(self) -> None:
"""No background resources to stop."""
# Undo the probe so the policy object isn't left permanently patched
# (it may outlive this engine or be reused by another).
if self._original_predict_action_chunk is not None:
self._policy.predict_action_chunk = self._original_predict_action_chunk
self._original_predict_action_chunk = None
logger.info("SyncInferenceEngine stopped")
def reset(self) -> None:
@@ -118,27 +93,6 @@ class SyncInferenceEngine(InferenceEngine):
self._policy.reset()
self._preprocessor.reset()
self._postprocessor.reset()
# New episode: the next tick predicts a fresh chunk and re-anchors.
self._chunk_predicted = False
self._ever_predicted_chunk = False
def _install_chunk_probe(self) -> None:
"""Wrap the policy's public ``predict_action_chunk`` so we learn which ticks
predict a fresh chunk (when the anchor must advance) without introspecting any
private action queue. Chunking policies call it from ``select_action``.
Wraps whatever callable is currently bound (e.g. an already-``torch.compile``d
one, since ``build_rollout_context`` compiles before building the engine); undone
in ``stop()``."""
self._original_predict_action_chunk = self._policy.predict_action_chunk
inner = self._original_predict_action_chunk
def probe(*args, **kwargs):
self._chunk_predicted = True
self._ever_predicted_chunk = True
return inner(*args, **kwargs)
self._policy.predict_action_chunk = probe
def get_action(self, obs_frame: dict | None) -> torch.Tensor | None:
"""Run the full inference pipeline on ``obs_frame`` and return an action tensor."""
@@ -153,25 +107,12 @@ class SyncInferenceEngine(InferenceEngine):
if self._device.type == "cuda" and self._policy.config.use_amp
else nullcontext()
)
# Snapshot the chunk anchor before the preprocessor overwrites it with this
# tick's state; restore it below if this tick only served a cached action.
# ``clone`` so the snapshot survives even if the cached tensor is ever mutated
# in place (today it is only rebound, but the copy is cheap for a state vector).
anchor_before = None
if self._relative_step is not None:
cached = self._relative_step.get_cached_state()
anchor_before = cached.clone() if cached is not None else None
self._chunk_predicted = False
with torch.inference_mode(), autocast_ctx:
observation = prepare_observation_for_inference(
observation, self._device, self._task, self._robot_type
)
observation = self._preprocessor(observation)
action = self._policy.select_action(observation)
# Hold the anchor only for a chunking policy serving a cached action this
# tick; policies that never chunk or that recomputed keep refreshing.
if self._relative_step is not None and self._ever_predicted_chunk and not self._chunk_predicted:
self._relative_step.set_cached_state(anchor_before)
action = self._postprocessor(action)
action_tensor = action.squeeze(0).cpu()
+25 -23
View File
@@ -28,12 +28,7 @@ For distributed runs, see ``examples/annotations/run_hf_job.py``.
"""
import logging
from contextlib import suppress
from pathlib import Path
from typing import TYPE_CHECKING
from huggingface_hub import HfApi, snapshot_download
from huggingface_hub.errors import RevisionNotFoundError
from lerobot.annotations.steerable_pipeline.config import AnnotationPipelineConfig
from lerobot.annotations.steerable_pipeline.executor import Executor
@@ -47,12 +42,6 @@ from lerobot.annotations.steerable_pipeline.validator import StagingValidator
from lerobot.annotations.steerable_pipeline.vlm_client import make_vlm_client
from lerobot.annotations.steerable_pipeline.writer import LanguageColumnsWriter
from lerobot.configs import parser
from lerobot.utils.import_utils import _datasets_available, require_package
if TYPE_CHECKING or _datasets_available:
from lerobot.datasets.dataset_metadata import CODEBASE_VERSION
from lerobot.datasets.io_utils import load_info
from lerobot.datasets.utils import create_lerobot_dataset_card
logger = logging.getLogger(__name__)
@@ -61,6 +50,8 @@ def _resolve_root(cfg: AnnotationPipelineConfig) -> Path:
if cfg.root is not None:
return Path(cfg.root)
if cfg.repo_id is not None:
from huggingface_hub import snapshot_download
return Path(snapshot_download(repo_id=cfg.repo_id, repo_type="dataset"))
raise ValueError("Either --root or --repo_id must be provided.")
@@ -134,7 +125,7 @@ def _push_to_hub(root: Path, cfg: AnnotationPipelineConfig) -> None:
Pushes to ``cfg.new_repo_id`` when set, otherwise back to ``cfg.repo_id``.
"""
require_package("datasets", "dataset")
from huggingface_hub import HfApi # noqa: PLC0415
repo_id = cfg.new_repo_id or cfg.repo_id
commit_message = cfg.push_commit_message or "Add steerable annotations (lerobot-annotate)"
@@ -152,26 +143,33 @@ def _push_to_hub(root: Path, cfg: AnnotationPipelineConfig) -> None:
repo_id=repo_id,
repo_type="dataset",
commit_message=commit_message,
# README.md is excluded because when pushing to ``new_repo_id`` the
# source card's links (e.g. the visualize badge) would keep pointing
# at the source dataset; a fresh card is generated below instead.
ignore_patterns=[".annotate_staging/**", "**/.DS_Store", "README.md"],
ignore_patterns=[".annotate_staging/**", "**/.DS_Store"],
)
print(f"[lerobot-annotate] uploaded to https://huggingface.co/datasets/{repo_id}", flush=True)
dataset_info = load_info(root)
card = create_lerobot_dataset_card(dataset_info=dataset_info, license="apache-2.0", repo_id=repo_id)
card.push_to_hub(repo_id=repo_id, repo_type="dataset")
# Tag the upload with the codebase version. ``LeRobotDatasetMetadata``
# resolves the dataset revision via ``get_safe_version`` which scans
# for tags like ``v3.0``; without a tag it raises
# ``RevisionNotFoundError``. Read the version straight from the
# dataset's own ``meta/info.json`` so we tag whatever the writer
# actually wrote (no accidental drift if the codebase floor moves).
version_tag = (
dataset_info.codebase_version if dataset_info.codebase_version.startswith("v") else CODEBASE_VERSION
)
from lerobot.datasets.dataset_metadata import CODEBASE_VERSION # noqa: PLC0415
info_path = root / "meta" / "info.json"
version_tag = CODEBASE_VERSION
if info_path.exists():
try:
from lerobot.utils.io_utils import load_json # noqa: PLC0415
info = load_json(info_path)
ds_version = info.get("codebase_version")
if isinstance(ds_version, str) and ds_version.startswith("v"):
version_tag = ds_version
except Exception as exc: # noqa: BLE001
print(
f"[lerobot-annotate] could not read codebase_version from info.json ({exc}); falling back to {version_tag}",
flush=True,
)
revision = getattr(commit_info, "oid", None)
tag_kwargs = {
"repo_id": repo_id,
@@ -182,6 +180,10 @@ def _push_to_hub(root: Path, cfg: AnnotationPipelineConfig) -> None:
tag_kwargs["revision"] = revision
try:
from contextlib import suppress # noqa: PLC0415
from huggingface_hub.errors import RevisionNotFoundError # noqa: PLC0415
with suppress(RevisionNotFoundError):
api.delete_tag(repo_id, tag=version_tag, repo_type="dataset")
api.create_tag(**tag_kwargs)
@@ -325,6 +325,8 @@ class RecomputeStatsConfig(OperationConfig):
relative_exclude_joints: list[str] | None = None
chunk_size: int = 50
num_workers: int = 0
relative_pose_representation: str = "componentwise"
relative_se3_pose_groups: list[list[int]] | None = None
overwrite: bool = False
@@ -698,6 +700,8 @@ def handle_recompute_stats(cfg: EditDatasetConfig) -> None:
relative_exclude_joints=cfg.operation.relative_exclude_joints,
chunk_size=cfg.operation.chunk_size,
num_workers=cfg.operation.num_workers,
relative_pose_representation=cfg.operation.relative_pose_representation,
relative_se3_pose_groups=cfg.operation.relative_se3_pose_groups,
)
logging.info(f"Stats written to {dataset.root}")
+5 -7
View File
@@ -171,9 +171,6 @@ def update_policy(
train_metrics.update_s = time.perf_counter() - start_time
if torch.cuda.is_available():
train_metrics.gpu_mem_gb = torch.cuda.max_memory_allocated() / (1024**3)
# Aggregate the policy's scalar outputs for logging and rank-reduction across the log window.
if output_dict:
train_metrics.update_metrics(output_dict)
return train_metrics, output_dict
@@ -346,6 +343,8 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
"enabled": True,
"exclude_joints": getattr(active_cfg, "relative_exclude_joints", []),
"action_names": getattr(active_cfg, "action_feature_names", None),
"pose_representation": getattr(active_cfg, "relative_pose_representation", "componentwise"),
"se3_pose_groups": getattr(active_cfg, "relative_se3_pose_groups", []),
}
postprocessor_overrides["absolute_actions_processor"] = {"enabled": True}
processor_kwargs["preprocessor_overrides"] = preprocessor_overrides
@@ -575,7 +574,7 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
batch = preprocessor(batch)
train_tracker.dataloading_s = time.perf_counter() - start_time
train_tracker, _ = update_policy(
train_tracker, output_dict = update_policy(
train_tracker,
policy,
batch,
@@ -608,10 +607,9 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
train_tracker.samples_per_s = effective_batch_size / step_time
logging.info(train_tracker)
if wandb_logger:
# Policy sub-losses (latent_loss, action_loss, ...) are aggregated into the
# tracker by update_policy, so to_dict() already carries their windowed,
# rank-reduced averages — no per-step output_dict passthrough needed.
wandb_log_dict = train_tracker.to_dict()
if output_dict:
wandb_log_dict.update(output_dict)
# Log sample weighting statistics if enabled
if sample_weighter is not None:
weighter_stats = sample_weighter.get_stats()
-14
View File
@@ -59,20 +59,6 @@ def get_safe_torch_device(try_device: str, log: bool = False) -> torch.device:
return device
def resolve_safetensors_device(map_location: str | torch.device) -> str:
"""Resolve a device string for a safetensors load, working around a device-mapping quirk.
safetensors' load maps the bare string "cuda" to cuda:0 regardless of the current device
(unlike torch's .to("cuda"), which honors torch.cuda.current_device()). Under multi-GPU
accelerate/FSDP every rank would then load its weights onto GPU 0, OOMing it before sharding.
Resolve "cuda" to the concrete current-device index so each rank loads onto its own GPU.
"""
map_location = str(map_location)
if map_location == "cuda" and torch.cuda.is_available():
return f"cuda:{torch.cuda.current_device()}"
return map_location
def get_safe_dtype(dtype: torch.dtype, device: str | torch.device):
"""
mps is currently not compatible with float64
-19
View File
@@ -104,7 +104,6 @@ class MetricsTracker:
"episodes",
"epochs",
"accelerator",
"_caller_metrics",
]
def __init__(
@@ -130,9 +129,6 @@ class MetricsTracker:
self.episodes = self.samples / self._avg_samples_per_ep
self.epochs = self.samples / self._num_frames
self.accelerator = accelerator
# Meter names the caller registered up front. update_metrics() leaves these untouched, so a
# policy that echoes e.g. "loss" in its output dict can't clobber the aggregated meter.
self._caller_metrics: set[str] = set(self.metrics)
def __getattr__(self, name: str) -> int | dict[str, AverageMeter] | AverageMeter | Any:
if name in self.__dict__:
@@ -160,21 +156,6 @@ class MetricsTracker:
self.episodes = self.samples / self._avg_samples_per_ep
self.epochs = self.samples / self._num_frames
def update_metrics(self, values: dict[str, Any]) -> None:
"""Accumulate a dict of scalar metrics, auto-registering a meter for each new key.
Non-numeric values and bools are ignored.
Caller-registered metrics (those passed to the constructor) are never overridden.
"""
for name, value in values.items():
if isinstance(value, bool) or not isinstance(value, (int, float)):
continue
if name in self._caller_metrics:
continue
if name not in self.metrics:
self.metrics[name] = AverageMeter(name, ":.3f", reduction="mean")
self.metrics[name].update(float(value))
def reduce_across_ranks(self) -> None:
"""
Synchronises the running averages of every metric whose ``reduction`` is not ``"none"``
+3 -3
View File
@@ -85,7 +85,7 @@ def _spy_responder(captured: list[list[dict[str, Any]]], reply: Any):
def test_module1_plan_memory_subtask_smoke(fixture_dataset_root: Path, tmp_path: Path) -> None:
vlm = make_canned_responder(
{
"COMPLETED manipulation events": {
"atomic subtasks": {
"subtasks": [
{"text": "grasp the handle of the sponge", "start": 0.0, "end": 0.4},
{"text": "wipe the counter from left to right", "start": 0.4, "end": 0.8},
@@ -126,7 +126,7 @@ def test_module1_emit_memory_false_skips_memory_keeps_subtasks_and_plan(
leaving subtask + plan generation intact symmetric to ``emit_plan``."""
vlm = make_canned_responder(
{
"COMPLETED manipulation events": {
"atomic subtasks": {
"subtasks": [
{"text": "grasp the handle of the sponge", "start": 0.0, "end": 0.4},
{"text": "wipe the counter from left to right", "start": 0.4, "end": 0.8},
@@ -318,7 +318,7 @@ def test_module1_attaches_contact_sheets_to_subtask_prompt(
return block.get("text", "")
return ""
subtask_calls = [m for m in captured if "COMPLETED manipulation events" in _prompt_text(m)]
subtask_calls = [m for m in captured if "atomic subtasks" in _prompt_text(m)]
assert len(subtask_calls) == 1, "expected exactly one subtask-prompt VLM call"
content = subtask_calls[0][0]["content"]
video_blocks = [b for b in content if isinstance(b, dict) and b.get("type") == "video"]
@@ -0,0 +1,94 @@
#!/usr/bin/env python
# Copyright 2026 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.
import pytest
import torch
pytest.importorskip("transformers")
from lerobot.policies.pi05.configuration_pi05 import PI05Config # noqa: E402
from lerobot.policies.pi05.modeling_pi05 import ( # noqa: E402
_build_flow_matching_inputs,
_prepare_trained_rtc_prefix,
_reduce_training_rtc_loss,
create_sinusoidal_pos_embedding,
)
from lerobot.policies.pi_gemma import PiGemmaRMSNorm # noqa: E402
def test_pi05_training_rtc_uses_clean_prefix_and_per_token_time():
actions = torch.tensor([[[1.0], [2.0], [3.0], [4.0]]])
noise = torch.tensor([[[10.0], [20.0], [30.0], [40.0]]])
time = torch.tensor([0.25])
prefix_mask = torch.tensor([[True, True, False, False]])
x_t, model_time = _build_flow_matching_inputs(actions, noise, time, prefix_mask)
assert model_time.tolist() == [[0.0, 0.0, 0.25, 0.25]]
assert torch.equal(x_t[:, :2], actions[:, :2])
assert torch.equal(x_t[:, 2:], 0.25 * noise[:, 2:] + 0.75 * actions[:, 2:])
def test_pi05_training_rtc_loss_excludes_clean_prefix():
losses = torch.tensor([[[100.0], [100.0], [2.0], [4.0]]])
prefix_mask = torch.tensor([[True, True, False, False]])
loss = _reduce_training_rtc_loss(losses, prefix_mask, reduction="mean")
assert loss.item() == pytest.approx(3.0)
def test_pi05_training_rtc_adaptive_norm_accepts_per_action_time_conditioning():
norm = PiGemmaRMSNorm(dim=4, cond_dim=3)
hidden = torch.randn(2, 5, 4)
per_action_condition = torch.randn(2, 5, 3)
output, gate = norm(hidden, per_action_condition)
assert output.shape == hidden.shape
assert gate.shape == hidden.shape
def test_pi05_training_rtc_embeds_per_action_timesteps():
per_action_time = torch.tensor([[0.0, 0.0, 0.25, 0.25]])
embedding = create_sinusoidal_pos_embedding(
per_action_time,
dimension=8,
min_period=4e-3,
max_period=4.0,
device=per_action_time.device,
)
assert embedding.shape == (1, 4, 8)
torch.testing.assert_close(embedding[:, 0], embedding[:, 1])
torch.testing.assert_close(embedding[:, 2], embedding[:, 3])
def test_pi05_trained_rtc_prefix_is_padded_to_model_width():
latent = torch.zeros(1, 5, 32)
previous = torch.arange(30, dtype=torch.float32).reshape(1, 3, 10)
prefix, mask = _prepare_trained_rtc_prefix(
latent,
previous,
inference_delay=2,
training_max_delay=4,
)
assert prefix.shape == latent.shape
assert mask.shape == latent.shape
torch.testing.assert_close(prefix[:, :2, :10], previous[:, :2])
assert torch.count_nonzero(prefix[:, :2, 10:]) == 0
assert mask[:, :2].all()
assert not mask[:, 2:].any()
@pytest.mark.parametrize("max_delay", [-1, 5])
def test_pi05_config_rejects_invalid_training_rtc_delay(max_delay):
with pytest.raises(ValueError, match="rtc_training_max_delay"):
PI05Config(chunk_size=5, n_action_steps=5, rtc_training_max_delay=max_delay)
@@ -16,6 +16,8 @@
"""Tests for RTC configuration module."""
import pytest
from lerobot.configs.types import RTCAttentionSchedule
from lerobot.policies.rtc.configuration_rtc import RTCConfig
@@ -27,6 +29,7 @@ def test_rtc_config_default_initialization():
config = RTCConfig()
assert config.enabled is True
assert config.mode == "guided"
assert config.prefix_attention_schedule == RTCAttentionSchedule.LINEAR
assert config.max_guidance_weight == 10.0
assert config.execution_horizon == 10
@@ -34,10 +37,16 @@ def test_rtc_config_default_initialization():
assert config.debug_maxlen == 100
def test_rtc_config_rejects_unknown_mode():
with pytest.raises(ValueError, match="mode must be"):
RTCConfig(mode="unknown")
def test_rtc_config_custom_initialization():
"""Test RTCConfig initializes with custom values."""
config = RTCConfig(
enabled=True,
mode="trained",
prefix_attention_schedule=RTCAttentionSchedule.EXP,
max_guidance_weight=5.0,
execution_horizon=20,
@@ -46,6 +55,7 @@ def test_rtc_config_custom_initialization():
)
assert config.enabled is True
assert config.mode == "trained"
assert config.prefix_attention_schedule == RTCAttentionSchedule.EXP
assert config.max_guidance_weight == 5.0
assert config.execution_horizon == 20
+13
View File
@@ -93,6 +93,19 @@ def test_rtc_processor_initialization_without_debug(rtc_config_debug_disabled):
assert processor.tracker is None
def test_rtc_processor_rejects_trained_mode_when_policy_does_not_support_it():
config = RTCConfig(mode="trained")
with pytest.raises(ValueError, match="requires a PI05-compatible checkpoint"):
RTCProcessor(config)
processor = RTCProcessor(config, trained_mode_supported=True)
assert processor.rtc_config.mode == "trained"
disabled = RTCProcessor(RTCConfig(enabled=False, mode="trained"))
assert disabled.rtc_config.enabled is False
# ====================== Tracker Proxy Methods Tests ======================
@@ -505,6 +505,44 @@ class TestRTCReanchoringWithStateNormalizer:
assert not torch.allclose(cached, post_normalize_state, atol=1e-3)
def test_reanchor_se3_6d_prefix_uses_current_camera_frame_and_model_width():
"""A leftover absolute EE chunk is recomposed relative to the latest camera-frame EE pose."""
names = ["pos_x", "pos_y", "pos_z", "rot_x", "rot_y", "rot_z", "gripper"]
relative_step = RelativeActionsProcessorStep(
enabled=True,
exclude_joints=["gripper"],
action_names=names,
pose_representation="se3_6d",
se3_pose_groups=[list(range(6))],
)
current_state = torch.tensor([[0.20, -0.10, 0.40, 0.31, -0.22, 0.17, 0.03]])
previous_absolute = torch.tensor(
[
[0.28, -0.03, 0.46, -0.18, 0.27, 0.41, 0.02],
[0.31, 0.02, 0.50, -0.11, 0.35, 0.52, 0.01],
]
)
result = reanchor_relative_rtc_prefix(
prev_actions_absolute=previous_absolute,
current_state=current_state,
relative_step=relative_step,
normalizer_step=None,
policy_device="cpu",
)
expected = to_relative_actions(
previous_absolute,
current_state,
relative_step._build_mask(previous_absolute.shape[-1]),
pose_representation="se3_6d",
se3_pose_groups=[list(range(6))],
)
assert result.shape == (2, 10)
torch.testing.assert_close(result, expected, atol=1e-6, rtol=1e-6)
torch.testing.assert_close(result[:, -1], previous_absolute[:, -1])
def _detect_relative_actions(preprocessor) -> bool:
"""Mirror of the helper in lerobot-rollout for testing without importing it."""
return any(isinstance(step, RelativeActionsProcessorStep) and step.enabled for step in preprocessor.steps)
-7
View File
@@ -346,10 +346,3 @@ def test_state_not_modified_by_relative_processor(dataset, action_dim):
result_state = result[TransitionKey.OBSERVATION][OBS_STATE]
torch.testing.assert_close(result_state, original_state)
def test_cached_anchor_not_in_config():
"""The cached anchor is ephemeral runtime state and must not leak into the config."""
step = RelativeActionsProcessorStep(enabled=True)
step.set_cached_state(torch.tensor([[1.0, 2.0, 3.0, 4.0]]))
assert set(step.get_config()) == {"enabled", "exclude_joints", "action_names"}
@@ -0,0 +1,321 @@
#!/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.
from math import pi
import numpy as np
import pytest
import torch
pytest.importorskip("transformers")
from lerobot.configs import FeatureType, PolicyFeature # noqa: E402
from lerobot.datasets.compute_stats import ( # noqa: E402
compute_relative_action_stats,
compute_state_history_stats,
)
from lerobot.policies.pi05.configuration_pi05 import PI05Config # noqa: E402
from lerobot.policies.pi05.processor_pi05 import ( # noqa: E402
Pi05FlattenStateHistoryProcessorStep,
Pi05StateFromActionProcessorStep,
)
from lerobot.processor.relative_action_processor import ( # noqa: E402
AbsoluteActionsProcessorStep,
RelativeActionsProcessorStep,
)
from lerobot.types import TransitionKey # noqa: E402
from lerobot.utils.constants import ACTION, OBS_STATE # noqa: E402
def _transition(action: torch.Tensor | None, state: torch.Tensor | None = None) -> dict:
observation = {} if state is None else {OBS_STATE: state}
return {
TransitionKey.OBSERVATION: observation,
TransitionKey.ACTION: action,
TransitionKey.REWARD: None,
TransitionKey.DONE: None,
TransitionKey.TRUNCATED: None,
TransitionKey.COMPLEMENTARY_DATA: {},
}
def test_pi05_config_requests_action_history_prefix():
config = PI05Config(
device="cpu",
chunk_size=4,
n_action_steps=4,
state_from_action=True,
proprioception_history_steps=2,
)
assert config.action_delta_indices == [-1, 0, 1, 2, 3]
def test_pi05_config_accepts_se3_6d_action_and_state_with_two_step_history():
names = ["x", "y", "z", "rx", "ry", "rz", "gripper_width"]
config = PI05Config(
device="cpu",
use_relative_actions=True,
state_from_action=True,
proprioception_history_steps=2,
use_relative_state_history=True,
relative_pose_representation="se3_6d",
action_feature_names=names,
output_features={
ACTION: PolicyFeature(type=FeatureType.ACTION, shape=(7,)),
},
)
config.validate_features()
assert config.output_features[ACTION].shape == (10,)
assert config.input_features[OBS_STATE].shape == (10,)
def test_state_from_action_extracts_history_and_preserves_target_horizon():
action = torch.arange(2 * 5 * 3, dtype=torch.float32).reshape(2, 5, 3)
step = Pi05StateFromActionProcessorStep(enabled=True, history_steps=2)
result = step(_transition(action))
torch.testing.assert_close(result[TransitionKey.OBSERVATION][OBS_STATE], action[:, :2])
torch.testing.assert_close(result[TransitionKey.ACTION], action[:, 1:])
def test_relative_actions_use_newest_state_in_history_and_roundtrip():
state_history = torch.tensor([[[1.0, 10.0], [2.0, 20.0]]])
absolute = torch.tensor([[[3.0, 30.0], [4.0, 40.0]]])
relative_step = RelativeActionsProcessorStep(enabled=True)
absolute_step = AbsoluteActionsProcessorStep(enabled=True, relative_step=relative_step)
relative = relative_step(_transition(absolute, state_history))
expected = torch.tensor([[[1.0, 10.0], [2.0, 20.0]]])
torch.testing.assert_close(relative[TransitionKey.ACTION], expected)
recovered = absolute_step(_transition(relative[TransitionKey.ACTION]))
torch.testing.assert_close(recovered[TransitionKey.ACTION], absolute)
def test_relative_action_reference_is_reset_between_inference_sessions():
step = RelativeActionsProcessorStep(enabled=True)
step(_transition(None, torch.tensor([[1.0, 2.0]])))
step.reset()
assert step.get_cached_state() is None
assert step.get_cached_mask() is None
def test_flatten_state_history_preserves_chronological_order():
state_history = torch.tensor([[[1.0, 2.0], [3.0, 4.0]]])
step = Pi05FlattenStateHistoryProcessorStep(history_steps=2, max_state_dim=4)
result = step(_transition(torch.zeros(1, 2, 2), state_history))
torch.testing.assert_close(
result[TransitionKey.OBSERVATION][OBS_STATE], torch.tensor([[1.0, 2.0, 3.0, 4.0]])
)
def test_state_history_can_be_relative_with_absolute_gripper():
state_history = torch.tensor([[[1.0, 10.0, 0.2], [3.0, 20.0, 0.4]]])
step = Pi05FlattenStateHistoryProcessorStep(
history_steps=2,
max_state_dim=6,
relative=True,
exclude_joints=["gripper"],
state_names=["x", "y", "gripper_width"],
)
result = step(_transition(torch.zeros(1, 2, 3), state_history))
torch.testing.assert_close(
result[TransitionKey.OBSERVATION][OBS_STATE],
torch.tensor([[-2.0, -10.0, 0.2, 0.0, 0.0, 0.4]]),
)
def test_state_history_can_use_se3_composition_with_absolute_gripper():
state_history = torch.tensor(
[[[0.0, 1.0, 0.0, 0.0, 0.0, pi / 2, 0.2], [0.0, 0.0, 0.0, 0.0, 0.0, pi / 2, 0.4]]]
)
step = Pi05FlattenStateHistoryProcessorStep(
history_steps=2,
max_state_dim=14,
relative=True,
exclude_joints=["gripper"],
state_names=["x", "y", "z", "rx", "ry", "rz", "gripper_width"],
pose_representation="se3",
se3_pose_groups=[list(range(6))],
)
result = step(_transition(torch.zeros(1, 2, 7), state_history))
expected = torch.tensor([[1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.2, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.4]])
torch.testing.assert_close(result[TransitionKey.OBSERVATION][OBS_STATE], expected, atol=1e-6, rtol=1e-6)
def test_state_history_can_use_se3_6d_rotation_with_absolute_gripper():
state_history = torch.tensor(
[[[0.0, 1.0, 0.0, 0.0, 0.0, pi / 2, 0.2], [0.0, 0.0, 0.0, 0.0, 0.0, pi / 2, 0.4]]]
)
step = Pi05FlattenStateHistoryProcessorStep(
history_steps=2,
max_state_dim=20,
relative=True,
exclude_joints=["gripper"],
state_names=["x", "y", "z", "rx", "ry", "rz", "gripper_width"],
pose_representation="se3_6d",
se3_pose_groups=[list(range(6))],
)
result = step(_transition(torch.zeros(1, 2, 7), state_history))
identity_6d = [1.0, 0.0, 0.0, 0.0, 1.0, 0.0]
expected = torch.tensor([[1.0, 0.0, 0.0, *identity_6d, 0.2, 0.0, 0.0, 0.0, *identity_6d, 0.4]])
torch.testing.assert_close(result[TransitionKey.OBSERVATION][OBS_STATE], expected, atol=1e-6, rtol=1e-6)
def test_inference_state_history_is_rolled_and_reset():
step = Pi05StateFromActionProcessorStep(enabled=True, history_steps=2)
first = step(_transition(None, torch.tensor([[1.0, 2.0]])))
second = step(_transition(None, torch.tensor([[3.0, 4.0]])))
torch.testing.assert_close(
first[TransitionKey.OBSERVATION][OBS_STATE], torch.tensor([[[1.0, 2.0], [1.0, 2.0]]])
)
torch.testing.assert_close(
second[TransitionKey.OBSERVATION][OBS_STATE], torch.tensor([[[1.0, 2.0], [3.0, 4.0]]])
)
step.reset()
reset = step(_transition(None, torch.tensor([[5.0, 6.0]])))
torch.testing.assert_close(
reset[TransitionKey.OBSERVATION][OBS_STATE], torch.tensor([[[5.0, 6.0], [5.0, 6.0]]])
)
def test_flatten_state_history_checks_max_state_dim():
step = Pi05FlattenStateHistoryProcessorStep(history_steps=2, max_state_dim=3)
with pytest.raises(ValueError, match="above max_state_dim"):
step(_transition(torch.zeros(1, 2, 2), torch.zeros(1, 2, 2)))
def test_relative_stats_can_use_absolute_action_as_state():
actions = np.asarray([[0.0, 0.0], [1.0, 2.0], [2.0, 4.0], [3.0, 6.0]], dtype=np.float32)
dataset = {"action": actions, "episode_index": np.zeros(4, dtype=np.int64)}
features = {"action": {"shape": [2], "names": ["x", "y"]}}
stats = compute_relative_action_stats(
dataset,
features,
chunk_size=2,
state_from_action=True,
)
np.testing.assert_allclose(stats["mean"], [0.5, 1.0])
def test_relative_state_history_stats_match_processor_representation():
actions = np.asarray(
[[0.0, 0.1], [1.0, 0.2], [3.0, 0.3]],
dtype=np.float32,
)
dataset = {"action": actions, "episode_index": np.zeros(3, dtype=np.int64)}
features = {"action": {"shape": [2], "names": ["x", "gripper_width"]}}
stats = compute_state_history_stats(
dataset,
features,
history_steps=2,
exclude_joints=["gripper"],
relative=True,
)
expected = np.asarray([[0.0, 0.1, 0.0, 0.1], [-1.0, 0.1, 0.0, 0.2], [-2.0, 0.2, 0.0, 0.3]])
np.testing.assert_allclose(stats["mean"], expected.mean(axis=0))
def test_se3_relative_action_stats_use_reference_frame():
actions = np.asarray(
[
[0.0, 0.0, 0.0, 0.0, 0.0, pi / 2, 0.2],
[0.0, 1.0, 0.0, 0.0, 0.0, pi / 2, 0.3],
],
dtype=np.float32,
)
dataset = {"action": actions, "episode_index": np.zeros(2, dtype=np.int64)}
features = {
"action": {
"shape": [7],
"names": ["x", "y", "z", "rx", "ry", "rz", "gripper_width"],
}
}
stats = compute_relative_action_stats(
dataset,
features,
chunk_size=2,
exclude_joints=["gripper"],
state_from_action=True,
pose_representation="se3",
se3_pose_groups=[list(range(6))],
)
np.testing.assert_allclose(stats["mean"][:3], [0.5, 0.0, 0.0], atol=1e-6)
np.testing.assert_allclose(stats["mean"][6], 0.25, atol=1e-6)
def test_se3_6d_stats_expand_action_and_state_history():
actions = np.asarray(
[
[0.0, 0.0, 0.0, 0.0, 0.0, pi / 2, 0.2],
[0.0, 1.0, 0.0, 0.0, 0.0, pi / 2, 0.3],
],
dtype=np.float32,
)
dataset = {"action": actions, "episode_index": np.zeros(2, dtype=np.int64)}
features = {
"action": {
"shape": [7],
"names": ["x", "y", "z", "rx", "ry", "rz", "gripper_width"],
}
}
action_stats = compute_relative_action_stats(
dataset,
features,
chunk_size=2,
exclude_joints=["gripper"],
state_from_action=True,
pose_representation="se3_6d",
se3_pose_groups=[list(range(6))],
)
state_stats = compute_state_history_stats(
dataset,
features,
history_steps=2,
exclude_joints=["gripper"],
relative=True,
pose_representation="se3_6d",
se3_pose_groups=[list(range(6))],
)
assert action_stats["mean"].shape == (10,)
assert state_stats["mean"].shape == (20,)
np.testing.assert_allclose(action_stats["mean"][:3], [0.5, 0.0, 0.0], atol=1e-6)
np.testing.assert_allclose(action_stats["mean"][9], 0.25, atol=1e-6)
@@ -0,0 +1,144 @@
import math
import pytest
import torch
from lerobot.processor.relative_action_processor import (
rotation_6d_to_rotvec,
rotvec_to_rotation_6d,
to_absolute_actions,
to_absolute_se3_pose,
to_absolute_se3_pose_6d,
to_relative_actions,
to_relative_se3_pose,
to_relative_se3_pose_6d,
)
POSE_GROUP = [list(range(6))]
def test_se3_translation_is_expressed_in_reference_frame():
reference = torch.tensor([[1.0, 2.0, 3.0, 0.0, 0.0, math.pi / 2]])
target = torch.tensor([[1.0, 3.0, 3.0, 0.0, 0.0, math.pi / 2]])
relative = to_relative_se3_pose(target, reference)
torch.testing.assert_close(relative, torch.tensor([[1.0, 0.0, 0.0, 0.0, 0.0, 0.0]]), atol=1e-6, rtol=1e-6)
def test_se3_pose_roundtrip_for_batched_chunks():
torch.manual_seed(0)
reference = torch.randn(4, 6)
reference[:, 3:] *= 0.8
target = torch.randn(4, 11, 6)
target[..., 3:] *= 0.8
relative = to_relative_se3_pose(target, reference.unsqueeze(1))
recovered = to_absolute_se3_pose(relative, reference.unsqueeze(1))
torch.testing.assert_close(recovered, target, atol=2e-5, rtol=2e-5)
def test_mixed_se3_pose_and_absolute_gripper_roundtrip():
reference = torch.tensor([[0.2, -0.1, 0.4, 0.1, 0.2, -0.3, 0.06]])
target = torch.tensor([[[0.3, 0.2, 0.5, -0.2, 0.1, 0.4, 0.03], [0.1, -0.3, 0.2, 0.5, -0.1, 0.2, 0.05]]])
mask = [True, True, True, True, True, True, False]
relative = to_relative_actions(
target,
reference,
mask,
pose_representation="se3",
se3_pose_groups=POSE_GROUP,
)
recovered = to_absolute_actions(
relative,
reference,
mask,
pose_representation="se3",
se3_pose_groups=POSE_GROUP,
)
torch.testing.assert_close(relative[..., 6], target[..., 6])
torch.testing.assert_close(recovered, target, atol=2e-5, rtol=2e-5)
def test_se3_pose_group_cannot_be_partially_relative():
with pytest.raises(ValueError, match="wholly relative or wholly absolute"):
to_relative_actions(
torch.zeros(1, 7),
torch.zeros(1, 7),
[True, True, True, False, False, False, False],
pose_representation="se3",
se3_pose_groups=POSE_GROUP,
)
@pytest.mark.parametrize(
"rotvec",
[
[0.0, 0.0, 0.0],
[0.2, -0.5, 0.8],
[math.pi - 1e-4, 0.0, 0.0],
],
)
def test_rotation_6d_roundtrip(rotvec):
source = torch.tensor([rotvec], dtype=torch.float64)
recovered = rotation_6d_to_rotvec(rotvec_to_rotation_6d(source))
torch.testing.assert_close(recovered, source, atol=2e-6, rtol=2e-6)
def test_rotation_6d_uses_umi_first_two_rows():
source = torch.tensor([[0.0, 0.0, math.pi / 2]], dtype=torch.float64)
encoded = rotvec_to_rotation_6d(source)
expected = torch.tensor([[0.0, -1.0, 0.0, 1.0, 0.0, 0.0]], dtype=torch.float64)
torch.testing.assert_close(encoded, expected, atol=1e-7, rtol=1e-7)
torch.testing.assert_close(rotation_6d_to_rotvec(expected), source, atol=1e-7, rtol=1e-7)
def test_se3_6d_pose_roundtrip_for_batched_chunks():
torch.manual_seed(1)
reference = torch.randn(4, 6)
reference[:, 3:] *= 0.8
target = torch.randn(4, 11, 6)
target[..., 3:] *= 0.8
relative = to_relative_se3_pose_6d(target, reference.unsqueeze(1))
recovered = to_absolute_se3_pose_6d(relative, reference.unsqueeze(1))
assert relative.shape == (4, 11, 9)
torch.testing.assert_close(recovered, target, atol=2e-5, rtol=2e-5)
def test_mixed_se3_6d_pose_and_absolute_gripper_roundtrip():
reference = torch.tensor([[0.2, -0.1, 0.4, 0.1, 0.2, -0.3, 0.06]])
target = torch.tensor([[[0.3, 0.2, 0.5, -0.2, 0.1, 0.4, 0.03], [0.1, -0.3, 0.2, 0.5, -0.1, 0.2, 0.05]]])
mask = [True, True, True, True, True, True, False]
relative = to_relative_actions(
target,
reference,
mask,
pose_representation="se3_6d",
se3_pose_groups=POSE_GROUP,
)
recovered = to_absolute_actions(
relative,
reference,
mask,
pose_representation="se3_6d",
se3_pose_groups=POSE_GROUP,
)
assert relative.shape == (1, 2, 10)
torch.testing.assert_close(relative[..., 9], target[..., 6])
torch.testing.assert_close(recovered, target, atol=2e-5, rtol=2e-5)
def test_rotation_6d_rejects_degenerate_prediction():
with pytest.raises(ValueError, match="degenerate"):
rotation_6d_to_rotvec(torch.zeros(1, 6))
+7 -20
View File
@@ -18,8 +18,6 @@ import json
from types import SimpleNamespace
import pytest
import requests
from huggingface_hub.errors import RevisionNotFoundError
# ``lerobot.scripts.lerobot_annotate`` (and the ``_push_to_hub`` path it
# exercises) imports ``lerobot.datasets``, which only ships under the
@@ -28,13 +26,11 @@ pytest.importorskip("datasets", reason="datasets is required (install lerobot[da
def test_push_to_hub_tags_uploaded_dataset_revision(tmp_path, monkeypatch):
from lerobot.scripts import lerobot_annotate
from lerobot.scripts.lerobot_annotate import _push_to_hub
root = tmp_path / "dataset"
(root / "meta").mkdir(parents=True)
(root / "meta" / "info.json").write_text(
json.dumps({"codebase_version": "v3.0", "fps": 30, "features": {}})
)
(root / "meta" / "info.json").write_text(json.dumps({"codebase_version": "v3.0"}))
calls = {}
@@ -47,6 +43,9 @@ def test_push_to_hub_tags_uploaded_dataset_revision(tmp_path, monkeypatch):
return SimpleNamespace(oid="abc123")
def delete_tag(self, repo_id, **kwargs):
import requests
from huggingface_hub.errors import RevisionNotFoundError
calls["delete_tag"] = {"repo_id": repo_id, **kwargs}
# Simulate the common case: no stale tag to delete.
raise RevisionNotFoundError("no such tag", response=requests.Response())
@@ -54,12 +53,7 @@ def test_push_to_hub_tags_uploaded_dataset_revision(tmp_path, monkeypatch):
def create_tag(self, **kwargs):
calls["create_tag"] = kwargs
monkeypatch.setattr(lerobot_annotate, "HfApi", FakeHfApi)
def fake_card_push(self, **kwargs):
calls["card_push"] = {"content": str(self), **kwargs}
monkeypatch.setattr("huggingface_hub.DatasetCard.push_to_hub", fake_card_push)
monkeypatch.setattr("huggingface_hub.HfApi", FakeHfApi)
cfg = SimpleNamespace(
repo_id="source/dataset",
@@ -68,7 +62,7 @@ def test_push_to_hub_tags_uploaded_dataset_revision(tmp_path, monkeypatch):
push_commit_message=None,
)
lerobot_annotate._push_to_hub(root, cfg)
_push_to_hub(root, cfg)
assert calls["create_repo"] == {
"repo_id": "annotated/dataset",
@@ -77,13 +71,6 @@ def test_push_to_hub_tags_uploaded_dataset_revision(tmp_path, monkeypatch):
"exist_ok": True,
}
assert calls["upload_folder"]["repo_id"] == "annotated/dataset"
# The source README must not be copied over: its links (e.g. the
# visualize badge) point at the source dataset. A card regenerated for
# the target repo is pushed instead.
assert "README.md" in calls["upload_folder"]["ignore_patterns"]
assert calls["card_push"]["repo_id"] == "annotated/dataset"
assert "visualize_dataset?path=annotated/dataset" in calls["card_push"]["content"]
assert "source/dataset" not in calls["card_push"]["content"]
# A stale tag (e.g. from a previous annotation run) is deleted first so
# the new tag always points at the upload we just made.
assert calls["delete_tag"] == {
+126 -222
View File
@@ -17,6 +17,7 @@
from __future__ import annotations
import dataclasses
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest
@@ -98,6 +99,131 @@ def test_inference_config_types():
assert rtc.rtc is not None
def test_trained_rtc_retries_chunk_when_measured_delay_exceeds_conditioning():
from lerobot.rollout.inference.rtc import _trained_rtc_chunk_can_merge
assert not _trained_rtc_chunk_can_merge(
conditioned_delay=2,
measured_delay=3,
training_max_delay=4,
has_previous_actions=True,
)
assert _trained_rtc_chunk_can_merge(
conditioned_delay=2,
measured_delay=5,
training_max_delay=4,
has_previous_actions=False,
)
def test_trained_rtc_bootstraps_first_overlap_with_checkpoint_capacity():
from lerobot.rollout.inference.rtc import _estimate_rtc_delay
assert (
_estimate_rtc_delay(
latency=0,
time_per_step=1 / 30,
mode="trained",
training_max_delay=10,
has_previous_actions=False,
)
== 0
)
assert (
_estimate_rtc_delay(
latency=0,
time_per_step=1 / 30,
mode="trained",
training_max_delay=10,
has_previous_actions=True,
)
== 10
)
def test_trained_rtc_rejects_measured_delay_above_checkpoint_support():
from lerobot.rollout.inference.rtc import (
_trained_rtc_chunk_can_merge,
_TrainedRTCDelayExceededError,
)
with pytest.raises(_TrainedRTCDelayExceededError, match="rtc_training_max_delay"):
_trained_rtc_chunk_can_merge(
conditioned_delay=3,
measured_delay=5,
training_max_delay=4,
has_previous_actions=True,
)
def test_trained_rtc_rejects_prefix_shorter_than_conditioned_delay():
from lerobot.rollout.inference.rtc import (
_TrainedRTCPrefixUnavailableError,
_validate_trained_rtc_prefix_available,
)
with pytest.raises(_TrainedRTCPrefixUnavailableError, match="only 2"):
_validate_trained_rtc_prefix_available(conditioned_delay=4, available_steps=2)
@pytest.mark.parametrize(
("execution_horizon", "queue_threshold", "match"),
[
(3, 4, "execution_horizon"),
(4, 3, "queue_threshold"),
],
)
def test_trained_rtc_rollout_requires_capacity_for_max_delay(execution_horizon, queue_threshold, match):
from lerobot.policies.rtc.configuration_rtc import RTCConfig
from lerobot.rollout.context import _validate_trained_rtc_rollout_config
from lerobot.rollout.inference import RTCInferenceConfig
policy_config = SimpleNamespace(type="pi05", rtc_training_max_delay=4)
inference_config = RTCInferenceConfig(
rtc=RTCConfig(mode="trained", execution_horizon=execution_horizon),
queue_threshold=queue_threshold,
)
with pytest.raises(ValueError, match=match):
_validate_trained_rtc_rollout_config(policy_config, inference_config)
def test_relative_state_order_follows_checkpoint_action_names():
from lerobot.rollout.context import _align_relative_state_feature_order
from lerobot.utils.constants import OBS_STATE
from lerobot.utils.feature_utils import build_dataset_frame
hw_features = {
OBS_STATE: {
"dtype": "float32",
"shape": (4,),
"names": ["left_joint.pos", "left_gripper.pos", "right_joint.pos", "right_gripper.pos"],
}
}
checkpoint_order = [
"right_joint.pos",
"right_gripper.pos",
"left_joint.pos",
"left_gripper.pos",
]
aligned = _align_relative_state_feature_order(hw_features, checkpoint_order)
frame = build_dataset_frame(
aligned,
{
"left_joint.pos": 1.0,
"left_gripper.pos": 2.0,
"right_joint.pos": 3.0,
"right_gripper.pos": 4.0,
},
prefix="observation",
)
assert aligned[OBS_STATE]["names"] == checkpoint_order
assert frame[OBS_STATE].tolist() == [3.0, 4.0, 1.0, 2.0]
assert hw_features[OBS_STATE]["names"][0] == "left_joint.pos"
def test_sentry_config_defaults():
from lerobot.rollout import SentryStrategyConfig
@@ -348,225 +474,3 @@ def test_rollout_context_fields():
field_names = {f.name for f in dataclasses.fields(RolloutContext)}
assert field_names == {"runtime", "hardware", "policy", "processors", "data"}
# ---------------------------------------------------------------------------
# Sync engine: relative-action anchoring (drift-free chunk execution)
# ---------------------------------------------------------------------------
_REL_ACTION_NAMES = ["j0.pos", "j1.pos", "j2.pos", "gripper.pos"]
_REL_ACTION_DIM = len(_REL_ACTION_NAMES)
def _relative_pre_post(exclude_joints=None):
"""Pre/post processors wrapping the real relative (caches anchor) and absolute
(relative + cached state) steps, mirroring what the sync engine feeds them."""
from lerobot.processor import (
AbsoluteActionsProcessorStep,
RelativeActionsProcessorStep,
TransitionKey,
create_transition,
)
from lerobot.utils.constants import OBS_STATE
relative_step = RelativeActionsProcessorStep(
enabled=True, exclude_joints=exclude_joints or [], action_names=list(_REL_ACTION_NAMES)
)
absolute_step = AbsoluteActionsProcessorStep(enabled=True, relative_step=relative_step)
class _Pre:
steps = [relative_step]
def __call__(self, observation):
# Run the relative step so it caches the anchor, then pass the batch through.
transition = create_transition(observation={OBS_STATE: observation[OBS_STATE]})
relative_step(transition)
return observation
def reset(self):
pass
class _Post:
def __call__(self, action):
transition = create_transition(action=action)
return absolute_step(transition)[TransitionKey.ACTION]
def reset(self):
pass
return _Pre(), _Post(), relative_step
def _fake_relative_policy(chunk_rel, n_action_steps, chunking=True):
"""Fake relative-action policy for the sync engine.
``chunking=True`` buffers a chunk and serves it one action per tick, calling the
public ``predict_action_chunk`` only on refill (pi0/fastwam/lingbot). ``False``
returns an action directly and never calls it. The engine's anchor probe keys off
that public call, so the fake routes through it rather than any private queue.
"""
from collections import deque
policy = MagicMock()
policy.config.use_amp = False
policy.config.action_feature_names = list(_REL_ACTION_NAMES)
state = {"predict_calls": 0}
queue = deque(maxlen=n_action_steps)
def predict_action_chunk(_batch=None, **_kwargs):
state["predict_calls"] += 1
return chunk_rel.unsqueeze(0) # [B=1, n, dim]
def select_action(_observation):
if not chunking:
return chunk_rel[0].unsqueeze(0)
if len(queue) == 0:
actions = policy.predict_action_chunk(_observation)
queue.extend(actions.transpose(0, 1)) # [n, 1, dim]
return queue.popleft()
policy.predict_action_chunk.side_effect = predict_action_chunk
policy.select_action.side_effect = select_action
policy.reset.side_effect = queue.clear
policy._predict_state = state
return policy
def _build_sync_engine(policy, pre, post):
from lerobot.rollout import SyncInferenceEngine
return SyncInferenceEngine(
policy=policy,
preprocessor=pre,
postprocessor=post,
dataset_features={"action": {"names": list(_REL_ACTION_NAMES)}},
ordered_action_keys=list(_REL_ACTION_NAMES),
task="test",
device="cpu",
robot_type="mock",
)
def _obs_frame(state_values):
import numpy as np
return {"observation.state": np.asarray(state_values, dtype=np.float32)}
def test_sync_relative_holds_anchor_across_chunk():
"""Every action popped within a chunk must anchor to the tick-0 state (no drift)."""
n = 4
# A distinct relative offset per chunk step so a wrong anchor would be visible.
chunk_rel = torch.stack([torch.full((_REL_ACTION_DIM,), 0.1 * (i + 1)) for i in range(n)])
pre, post, relative_step = _relative_pre_post()
policy = _fake_relative_policy(chunk_rel, n_action_steps=n)
engine = _build_sync_engine(policy, pre, post)
assert engine._relative_step is relative_step # introspection wired the step
s0 = [1.0, 2.0, 3.0, 4.0]
outputs = []
for tick in range(n):
# Feed a *different* state each tick; a drifting anchor would use it.
state = [v + tick for v in s0]
outputs.append(engine.get_action(_obs_frame(state)))
# Exactly one chunk was predicted across the n ticks.
assert policy._predict_state["predict_calls"] == 1
for tick in range(n):
expected = torch.tensor(s0) + chunk_rel[tick]
torch.testing.assert_close(outputs[tick], expected)
# Next tick empties the queue -> fresh chunk -> anchor advances to the new state.
s_next = [10.0, 20.0, 30.0, 40.0]
out = engine.get_action(_obs_frame(s_next))
assert policy._predict_state["predict_calls"] == 2
torch.testing.assert_close(out, torch.tensor(s_next) + chunk_rel[0])
# The anchor now reflects the fresh-chunk state, not the held one.
torch.testing.assert_close(relative_step.get_cached_state(), torch.tensor([s_next]))
def test_sync_relative_reset_reanchors_new_episode():
"""After ``reset()`` the first tick of the next episode anchors to the new state."""
n = 3
chunk_rel = torch.stack([torch.full((_REL_ACTION_DIM,), 0.2) for _ in range(n)])
pre, post, relative_step = _relative_pre_post()
policy = _fake_relative_policy(chunk_rel, n_action_steps=n)
engine = _build_sync_engine(policy, pre, post)
# Episode 1: one tick anchors to s0 and leaves cached actions in the queue.
engine.get_action(_obs_frame([1.0, 1.0, 1.0, 1.0]))
assert policy._predict_state["predict_calls"] == 1
engine.reset() # clears the queue and the per-episode chunk flags
# Episode 2: a fresh state must produce a fresh chunk anchored to that state,
# not carry over the previous episode's anchor.
s_new = [7.0, 8.0, 9.0, 10.0]
out = engine.get_action(_obs_frame(s_new))
assert policy._predict_state["predict_calls"] == 2
torch.testing.assert_close(out, torch.tensor(s_new) + chunk_rel[0])
torch.testing.assert_close(relative_step.get_cached_state(), torch.tensor([s_new]))
def test_sync_relative_non_chunking_policy_refreshes_every_tick():
"""A policy that never calls ``predict_action_chunk`` must not freeze the anchor."""
n = 3
chunk_rel = torch.stack([torch.full((_REL_ACTION_DIM,), 0.5) for _ in range(n)])
pre, post, _ = _relative_pre_post()
policy = _fake_relative_policy(chunk_rel, n_action_steps=n, chunking=False)
engine = _build_sync_engine(policy, pre, post)
s0 = [1.0, 1.0, 1.0, 1.0]
for tick in range(3):
state = [v + tick for v in s0]
out = engine.get_action(_obs_frame(state))
# Anchor must track the current state every tick (no chunk => no hold).
torch.testing.assert_close(out, torch.tensor(state) + chunk_rel[0])
assert policy._predict_state["predict_calls"] == 0
def test_sync_engine_no_relative_step_is_none():
"""Without an enabled relative step, the engine takes the plain select_action path."""
policy = MagicMock()
policy.config.use_amp = False
engine = _build_sync_engine(policy, MagicMock(steps=[]), MagicMock())
assert engine._relative_step is None
def test_sync_relative_exclude_joints_stay_absolute():
"""With ``exclude_joints``, excluded dims pass through absolute while the relative
dims still hold the tick-0 anchor across the chunk."""
n = 4
# Distinct offset per step *and* per dim so a wrong anchor or a wrong mask shows up.
chunk_rel = torch.stack([torch.full((_REL_ACTION_DIM,), 0.1 * (i + 1)) for i in range(n)])
pre, post, relative_step = _relative_pre_post(exclude_joints=["gripper"])
policy = _fake_relative_policy(chunk_rel, n_action_steps=n)
engine = _build_sync_engine(policy, pre, post)
# gripper (last dim) is kept absolute; j0..j2 are relative.
mask = torch.tensor([1.0, 1.0, 1.0, 0.0])
s0 = [1.0, 2.0, 3.0, 4.0]
outputs = []
for tick in range(n):
state = [v + tick for v in s0] # moving state; a drifting anchor would use it
outputs.append(engine.get_action(_obs_frame(state)))
assert policy._predict_state["predict_calls"] == 1 # one chunk held across n ticks
for tick in range(n):
# relative dims: anchor held at s0; excluded gripper dim: raw predicted value.
expected = chunk_rel[tick] + torch.tensor(s0) * mask
torch.testing.assert_close(outputs[tick], expected)
def test_sync_relative_stop_restores_policy_method():
"""``stop()`` un-patches the probe so the policy object isn't permanently modified."""
n = 3
chunk_rel = torch.stack([torch.full((_REL_ACTION_DIM,), 0.2) for _ in range(n)])
pre, post, _ = _relative_pre_post()
policy = _fake_relative_policy(chunk_rel, n_action_steps=n)
original = policy.predict_action_chunk
engine = _build_sync_engine(policy, pre, post)
assert policy.predict_action_chunk is not original # probe installed
engine.stop()
assert policy.predict_action_chunk is original # restored
-34
View File
@@ -233,37 +233,3 @@ def test_metrics_tracker_reduce_across_ranks_invokes_reduce():
# accumulate against the cluster view rather than the stale per-rank sum.
meter = tracker.update_s
assert meter.sum / meter.count == pytest.approx(meter.avg)
def test_metrics_tracker_update_metrics_registers_and_averages():
tracker = MetricsTracker(batch_size=32, num_frames=1000, num_episodes=50, metrics={})
tracker.update_metrics({"latent_loss": 0.2, "action_loss": 0.4})
tracker.update_metrics({"latent_loss": 0.4, "action_loss": 0.6})
# New keys are auto-registered as mean-reduced meters and averaged over the window.
assert tracker.metrics["latent_loss"].reduction == "mean"
assert tracker.metrics["latent_loss"].avg == pytest.approx(0.3)
assert tracker.metrics["action_loss"].avg == pytest.approx(0.5)
assert tracker.to_dict()["latent_loss"] == pytest.approx(0.3)
def test_metrics_tracker_update_metrics_skips_non_numeric():
tracker = MetricsTracker(batch_size=32, num_frames=1000, num_episodes=50, metrics={})
tracker.update_metrics({"loss": 0.5, "head_mode": "sparse", "enabled": True})
# strings and bools ignored
assert "loss" in tracker.metrics
assert "head_mode" not in tracker.metrics
assert "enabled" not in tracker.metrics
def test_metrics_tracker_update_metrics_does_not_override_caller_meter():
# A policy that echoes "loss" in its output dict must not overwrite the caller-owned,
# already-aggregated loss meter.
metrics = {"loss": AverageMeter("loss", ":.3f", reduction="mean")}
tracker = MetricsTracker(batch_size=32, num_frames=1000, num_episodes=50, metrics=metrics)
tracker.loss = 1.0 # caller-set optimized loss
tracker.update_metrics({"loss": 99.0, "latent_loss": 0.2})
assert tracker.metrics["loss"].avg == pytest.approx(1.0) # snapshot ignored
assert tracker.metrics["latent_loss"].avg == pytest.approx(0.2)