mirror of
https://github.com/huggingface/lerobot.git
synced 2026-07-28 12:15:59 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 dependabot[bot]
|
34bd3198a1 |
@@ -72,19 +72,19 @@ jobs:
|
||||
HF_USER_TOKEN: ${{ secrets.LEROBOT_HF_USER }}
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
lfs: true
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
|
||||
- name: Login to Docker Hub
|
||||
if: ${{ env.DOCKERHUB_USERNAME != '' }}
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
@@ -95,7 +95,7 @@ jobs:
|
||||
# from source-copy, so code-only changes skip the slow uv-sync layer
|
||||
# when the runner has a warm Docker daemon cache.
|
||||
- name: Build Libero benchmark image
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: docker/Dockerfile.benchmark.libero
|
||||
@@ -151,7 +151,7 @@ jobs:
|
||||
|
||||
- name: Upload Libero rollout video
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: libero-rollout-video
|
||||
path: /tmp/libero-artifacts/videos/
|
||||
@@ -159,7 +159,7 @@ jobs:
|
||||
|
||||
- name: Upload Libero eval metrics
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: libero-metrics
|
||||
path: /tmp/libero-artifacts/metrics.json
|
||||
@@ -214,7 +214,7 @@ jobs:
|
||||
|
||||
- name: Upload Libero train-smoke eval video
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: libero-train-smoke-video
|
||||
path: /tmp/libero-train-smoke-artifacts/eval/
|
||||
@@ -230,19 +230,19 @@ jobs:
|
||||
HF_USER_TOKEN: ${{ secrets.LEROBOT_HF_USER }}
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
lfs: true
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
|
||||
- name: Login to Docker Hub
|
||||
if: ${{ env.DOCKERHUB_USERNAME != '' }}
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
@@ -250,7 +250,7 @@ jobs:
|
||||
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
|
||||
- name: Build MetaWorld benchmark image
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: docker/Dockerfile.benchmark.metaworld
|
||||
@@ -303,7 +303,7 @@ jobs:
|
||||
|
||||
- name: Upload MetaWorld rollout video
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: metaworld-rollout-video
|
||||
path: /tmp/metaworld-artifacts/videos/
|
||||
@@ -311,7 +311,7 @@ jobs:
|
||||
|
||||
- name: Upload MetaWorld eval metrics
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: metaworld-metrics
|
||||
path: /tmp/metaworld-artifacts/metrics.json
|
||||
@@ -332,19 +332,19 @@ jobs:
|
||||
ROBOTWIN_TASKS: beat_block_hammer,click_bell,handover_block,stack_blocks_two,click_alarmclock,open_microwave,adjust_bottle,lift_pot,stamp_seal,turn_switch
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
lfs: true
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
|
||||
- name: Login to Docker Hub
|
||||
if: ${{ env.DOCKERHUB_USERNAME != '' }}
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
@@ -355,7 +355,7 @@ jobs:
|
||||
# simulation assets (~4 GB). Layer cache lives in the runner's local
|
||||
# Docker daemon — reused across re-runs on the same machine.
|
||||
- name: Build RoboTwin 2.0 benchmark image
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: docker/Dockerfile.benchmark.robotwin
|
||||
@@ -413,7 +413,7 @@ jobs:
|
||||
|
||||
- name: Upload RoboTwin rollout video
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4
|
||||
uses: actions/upload-artifact@v7
|
||||
with:
|
||||
name: robotwin-rollout-video
|
||||
path: /tmp/robotwin-artifacts/videos/
|
||||
@@ -421,7 +421,7 @@ jobs:
|
||||
|
||||
- name: Upload RoboTwin eval metrics
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4
|
||||
uses: actions/upload-artifact@v7
|
||||
with:
|
||||
name: robotwin-metrics
|
||||
path: /tmp/robotwin-artifacts/metrics.json
|
||||
@@ -439,19 +439,19 @@ jobs:
|
||||
HF_USER_TOKEN: ${{ secrets.LEROBOT_HF_USER }}
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
lfs: true
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
|
||||
- name: Login to Docker Hub
|
||||
if: ${{ env.DOCKERHUB_USERNAME != '' }}
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
@@ -459,7 +459,7 @@ jobs:
|
||||
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
|
||||
- name: Build RoboCasa365 benchmark image
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: docker/Dockerfile.benchmark.robocasa
|
||||
@@ -514,7 +514,7 @@ jobs:
|
||||
|
||||
- name: Upload RoboCasa365 rollout video
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: robocasa-rollout-video
|
||||
path: /tmp/robocasa-artifacts/videos/
|
||||
@@ -522,7 +522,7 @@ jobs:
|
||||
|
||||
- name: Upload RoboCasa365 eval metrics
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: robocasa-metrics
|
||||
path: /tmp/robocasa-artifacts/metrics.json
|
||||
@@ -540,19 +540,19 @@ jobs:
|
||||
HF_USER_TOKEN: ${{ secrets.LEROBOT_HF_USER }}
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
lfs: true
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
|
||||
- name: Login to Docker Hub
|
||||
if: ${{ env.DOCKERHUB_USERNAME != '' }}
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
@@ -560,7 +560,7 @@ jobs:
|
||||
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
|
||||
- name: Build RoboCerebra benchmark image
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: docker/Dockerfile.benchmark.robocerebra
|
||||
@@ -621,7 +621,7 @@ jobs:
|
||||
|
||||
- name: Upload RoboCerebra rollout video
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: robocerebra-rollout-video
|
||||
path: /tmp/robocerebra-artifacts/videos/
|
||||
@@ -629,7 +629,7 @@ jobs:
|
||||
|
||||
- name: Upload RoboCerebra eval metrics
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: robocerebra-metrics
|
||||
path: /tmp/robocerebra-artifacts/metrics.json
|
||||
@@ -648,19 +648,19 @@ jobs:
|
||||
ROBOMME_TASKS: PickXtimes,BinFill,StopCube,MoveCube,InsertPeg,SwingXtimes,VideoUnmask,ButtonUnmask,PickHighlight,PatternLock
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
lfs: true
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
|
||||
- name: Login to Docker Hub
|
||||
if: ${{ env.DOCKERHUB_USERNAME != '' }}
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
@@ -668,7 +668,7 @@ jobs:
|
||||
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
|
||||
- name: Build RoboMME benchmark image
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: docker/Dockerfile.benchmark.robomme
|
||||
@@ -726,7 +726,7 @@ jobs:
|
||||
|
||||
- name: Upload RoboMME rollout video
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: robomme-rollout-video
|
||||
path: /tmp/robomme-artifacts/videos/
|
||||
@@ -734,7 +734,7 @@ jobs:
|
||||
|
||||
- name: Upload RoboMME eval metrics
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: robomme-metrics
|
||||
path: /tmp/robomme-artifacts/metrics.json
|
||||
@@ -754,19 +754,19 @@ jobs:
|
||||
LIBERO_PLUS_TASK_IDS: "[0,100,260,500,1000,1500,2000,2400]"
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
lfs: true
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
|
||||
- name: Login to Docker Hub
|
||||
if: ${{ env.DOCKERHUB_USERNAME != '' }}
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
@@ -774,7 +774,7 @@ jobs:
|
||||
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
|
||||
- name: Build LIBERO-plus benchmark image
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: docker/Dockerfile.benchmark.libero_plus
|
||||
@@ -834,7 +834,7 @@ jobs:
|
||||
|
||||
- name: Upload LIBERO-plus rollout video
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: libero-plus-rollout-video
|
||||
path: /tmp/libero-plus-artifacts/videos/
|
||||
@@ -842,7 +842,7 @@ jobs:
|
||||
|
||||
- name: Upload LIBERO-plus eval metrics
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: libero-plus-metrics
|
||||
path: /tmp/libero-plus-artifacts/metrics.json
|
||||
@@ -858,19 +858,19 @@ jobs:
|
||||
HF_USER_TOKEN: ${{ secrets.LEROBOT_HF_USER }}
|
||||
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
lfs: true
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
|
||||
- name: Login to Docker Hub
|
||||
if: ${{ env.DOCKERHUB_USERNAME != '' }}
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
@@ -878,7 +878,7 @@ jobs:
|
||||
DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
|
||||
- name: Build VLABench benchmark image
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: docker/Dockerfile.benchmark.vlabench
|
||||
@@ -936,7 +936,7 @@ jobs:
|
||||
|
||||
- name: Upload VLABench rollout video
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: vlabench-rollout-video
|
||||
path: /tmp/vlabench-artifacts/videos/
|
||||
@@ -944,7 +944,7 @@ jobs:
|
||||
|
||||
- name: Upload VLABench eval metrics
|
||||
if: always()
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: vlabench-metrics
|
||||
path: /tmp/vlabench-artifacts/metrics.json
|
||||
|
||||
@@ -47,7 +47,7 @@ jobs:
|
||||
timeout-minutes: 30
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
|
||||
|
||||
@@ -52,21 +52,21 @@ jobs:
|
||||
sudo apt-get update
|
||||
sudo apt-get install git-lfs
|
||||
git lfs install
|
||||
- uses: actions/checkout@v6
|
||||
- uses: actions/checkout@v7
|
||||
with:
|
||||
lfs: true
|
||||
persist-credentials: false
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
- name: Login to Docker Hub
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
- name: Build and push Docker image CPU
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: ./docker/Dockerfile.user
|
||||
@@ -87,21 +87,21 @@ jobs:
|
||||
sudo apt-get update
|
||||
sudo apt-get install git-lfs
|
||||
git lfs install
|
||||
- uses: actions/checkout@v6
|
||||
- uses: actions/checkout@v7
|
||||
with:
|
||||
lfs: true
|
||||
persist-credentials: false
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
- name: Login to Docker Hub
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
- name: Build and push Docker image GPU
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: ./docker/Dockerfile.internal
|
||||
|
||||
@@ -33,7 +33,7 @@ jobs:
|
||||
github.event.workflow_run.event == 'pull_request' &&
|
||||
github.event.workflow_run.conclusion == 'success' &&
|
||||
github.repository == 'huggingface/lerobot'
|
||||
uses: huggingface/doc-builder/.github/workflows/upload_pr_documentation.yml@2430c1ec91d04667414e2fa31ecfc36c153ea391 # main
|
||||
uses: huggingface/doc-builder/.github/workflows/upload_pr_documentation.yml@6108e850ae1cf2f71bb0815a600bcd50c39abfa7 # main
|
||||
with:
|
||||
package_name: lerobot
|
||||
secrets:
|
||||
|
||||
@@ -55,7 +55,7 @@ jobs:
|
||||
github.repository == 'huggingface/lerobot'
|
||||
permissions:
|
||||
contents: read
|
||||
uses: huggingface/doc-builder/.github/workflows/build_main_documentation.yml@e60a538eea9817ab312196d0d233604b01697265 # main
|
||||
uses: huggingface/doc-builder/.github/workflows/build_main_documentation.yml@6108e850ae1cf2f71bb0815a600bcd50c39abfa7 # main
|
||||
with:
|
||||
commit_sha: ${{ github.sha }}
|
||||
package: lerobot
|
||||
@@ -78,7 +78,7 @@ jobs:
|
||||
permissions:
|
||||
contents: read
|
||||
pull-requests: write
|
||||
uses: huggingface/doc-builder/.github/workflows/build_pr_documentation.yml@e60a538eea9817ab312196d0d233604b01697265 # main
|
||||
uses: huggingface/doc-builder/.github/workflows/build_pr_documentation.yml@6108e850ae1cf2f71bb0815a600bcd50c39abfa7 # main
|
||||
with:
|
||||
commit_sha: ${{ github.event.pull_request.head.sha }}
|
||||
pr_number: ${{ github.event.number }}
|
||||
|
||||
@@ -69,7 +69,7 @@ jobs:
|
||||
HF_LEROBOT_HOME: /mnt/cache/.cache/huggingface/lerobot
|
||||
HF_USER_TOKEN: ${{ secrets.LEROBOT_HF_USER }}
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
lfs: true
|
||||
@@ -87,7 +87,7 @@ jobs:
|
||||
libusb-1.0-0-dev speech-dispatcher libgeos-dev portaudio19-dev
|
||||
|
||||
- name: Setup uv and Python
|
||||
uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e # v6
|
||||
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
|
||||
with:
|
||||
enable-cache: true
|
||||
version: ${{ env.UV_VERSION }}
|
||||
|
||||
@@ -63,7 +63,7 @@ jobs:
|
||||
HF_LEROBOT_HOME: /mnt/cache/.cache/huggingface/lerobot
|
||||
HF_USER_TOKEN: ${{ secrets.LEROBOT_HF_USER }}
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
lfs: true
|
||||
persist-credentials: false
|
||||
@@ -80,7 +80,7 @@ jobs:
|
||||
speech-dispatcher libgeos-dev portaudio19-dev
|
||||
|
||||
- name: Setup uv and Python
|
||||
uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e # v6
|
||||
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
|
||||
with:
|
||||
enable-cache: true
|
||||
version: ${{ env.UV_VERSION }}
|
||||
@@ -137,21 +137,21 @@ jobs:
|
||||
sudo apt-get update
|
||||
sudo apt-get install git-lfs
|
||||
git lfs install
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
lfs: true
|
||||
persist-credentials: false
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@8d2750c68a42422c14e847fe6c8ac0403b4cbd6f # v3
|
||||
uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
|
||||
with:
|
||||
cache-binary: false
|
||||
- name: Login to Docker Hub
|
||||
uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3
|
||||
uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
- name: Build and push Docker image
|
||||
uses: docker/build-push-action@10e90e3645eae34f1e60eeb005ba3a3d33f178e8 # v6
|
||||
uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
|
||||
with:
|
||||
context: .
|
||||
file: ./docker/Dockerfile.internal
|
||||
|
||||
@@ -29,7 +29,7 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
if: github.repository == 'huggingface/lerobot'
|
||||
steps:
|
||||
- uses: actions/github-script@v8
|
||||
- uses: actions/github-script@v9
|
||||
with:
|
||||
script: |
|
||||
// Setup Input Text
|
||||
|
||||
@@ -48,12 +48,12 @@ jobs:
|
||||
outputs:
|
||||
changed: ${{ steps.diff.outputs.changed }}
|
||||
steps:
|
||||
- uses: actions/checkout@v6
|
||||
- uses: actions/checkout@v7
|
||||
with:
|
||||
persist-credentials: false
|
||||
|
||||
- name: Setup uv and Python
|
||||
uses: astral-sh/setup-uv@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: astral-sh/setup-uv@v8.3.2 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
version: ${{ env.UV_VERSION }}
|
||||
python-version: ${{ env.PYTHON_VERSION }}
|
||||
@@ -74,7 +74,7 @@ jobs:
|
||||
|
||||
- name: Upload updated lockfile
|
||||
if: steps.diff.outputs.changed == 'true'
|
||||
uses: actions/upload-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/upload-artifact@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: uv-lock
|
||||
path: uv.lock
|
||||
@@ -93,13 +93,13 @@ jobs:
|
||||
HF_LEROBOT_HOME: /mnt/cache/.cache/huggingface/lerobot
|
||||
HF_USER_TOKEN: ${{ secrets.LEROBOT_HF_USER }}
|
||||
steps:
|
||||
- uses: actions/checkout@v6
|
||||
- uses: actions/checkout@v7
|
||||
with:
|
||||
lfs: true
|
||||
persist-credentials: false
|
||||
|
||||
- name: Download updated lockfile
|
||||
uses: actions/download-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/download-artifact@v8 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: uv-lock
|
||||
|
||||
@@ -115,7 +115,7 @@ jobs:
|
||||
speech-dispatcher libgeos-dev portaudio19-dev
|
||||
|
||||
- name: Setup uv and Python
|
||||
uses: astral-sh/setup-uv@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: astral-sh/setup-uv@v8.3.2 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
enable-cache: true
|
||||
version: ${{ env.UV_VERSION }}
|
||||
@@ -153,27 +153,27 @@ jobs:
|
||||
sudo apt-get update
|
||||
sudo apt-get install git-lfs
|
||||
git lfs install
|
||||
- uses: actions/checkout@v6
|
||||
- uses: actions/checkout@v7
|
||||
with:
|
||||
lfs: true
|
||||
persist-credentials: false
|
||||
|
||||
- name: Download updated lockfile
|
||||
uses: actions/download-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/download-artifact@v8 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: uv-lock
|
||||
|
||||
- name: Set up Docker Buildx
|
||||
uses: docker/setup-buildx-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/setup-buildx-action@v4 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
cache-binary: false
|
||||
- name: Login to Docker Hub
|
||||
uses: docker/login-action@v3 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/login-action@v4.4.0 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
username: ${{ secrets.DOCKERHUB_LEROBOT_USERNAME }}
|
||||
password: ${{ secrets.DOCKERHUB_LEROBOT_PASSWORD }}
|
||||
- name: Build and push Docker image
|
||||
uses: docker/build-push-action@v6 # zizmor: ignore[unpinned-uses]
|
||||
uses: docker/build-push-action@v7 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
context: .
|
||||
file: ./docker/Dockerfile.internal
|
||||
@@ -247,12 +247,12 @@ jobs:
|
||||
env:
|
||||
GH_TOKEN: ${{ secrets.UPDATE_LOCK_TOKEN }}
|
||||
steps:
|
||||
- uses: actions/checkout@v6
|
||||
- uses: actions/checkout@v7
|
||||
with:
|
||||
persist-credentials: false
|
||||
|
||||
- name: Download updated lockfile
|
||||
uses: actions/download-artifact@v4 # zizmor: ignore[unpinned-uses]
|
||||
uses: actions/download-artifact@v8 # zizmor: ignore[unpinned-uses]
|
||||
with:
|
||||
name: uv-lock
|
||||
|
||||
|
||||
@@ -33,7 +33,7 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
if: github.repository == 'huggingface/lerobot' && !github.event.pull_request.draft
|
||||
steps:
|
||||
- uses: actions/labeler@v6
|
||||
- uses: actions/labeler@v7
|
||||
with:
|
||||
repo-token: ${{ secrets.GITHUB_TOKEN }}
|
||||
sync-labels: true # Removes labels if files are removed from the PR
|
||||
|
||||
@@ -43,12 +43,12 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6
|
||||
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v6
|
||||
with:
|
||||
python-version: '3.12'
|
||||
|
||||
|
||||
@@ -38,12 +38,12 @@ jobs:
|
||||
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
persist-credentials: false
|
||||
|
||||
- name: Set up Python
|
||||
uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6
|
||||
uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v6
|
||||
with:
|
||||
python-version: '3.12'
|
||||
|
||||
@@ -104,7 +104,7 @@ jobs:
|
||||
- name: Publish to TestPyPI for pre-releases
|
||||
# True for tags like 'v0.2.0-rc1'
|
||||
if: startsWith(github.ref, 'refs/tags/v') && contains(github.ref, '-')
|
||||
uses: pypa/gh-action-pypi-publish@ed0c53931b1dc9bd32cbe73a98c7f6766f8a527e # v1.13.0
|
||||
uses: pypa/gh-action-pypi-publish@ba38be9e461d3875417946c167d0b5f3d385a247 # v1.14.1
|
||||
with:
|
||||
repository-url: https://test.pypi.org/legacy/
|
||||
verbose: true
|
||||
@@ -112,7 +112,7 @@ jobs:
|
||||
|
||||
- name: Publish to PyPI
|
||||
if: startsWith(github.ref, 'refs/tags/v') && !contains(github.ref, '-')
|
||||
uses: pypa/gh-action-pypi-publish@ed0c53931b1dc9bd32cbe73a98c7f6766f8a527e # v1.13.0
|
||||
uses: pypa/gh-action-pypi-publish@ba38be9e461d3875417946c167d0b5f3d385a247 # v1.14.1
|
||||
with:
|
||||
verbose: true
|
||||
print-hash: true
|
||||
@@ -127,7 +127,7 @@ jobs:
|
||||
env:
|
||||
MUJOCO_GL: egl
|
||||
steps:
|
||||
- uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
- uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
lfs: true
|
||||
persist-credentials: false
|
||||
@@ -137,7 +137,7 @@ jobs:
|
||||
git curl libglib2.0-0 libegl1-mesa-dev ffmpeg libusb-1.0-0-dev \
|
||||
speech-dispatcher libgeos-dev portaudio19-dev
|
||||
- name: Setup uv and Python
|
||||
uses: astral-sh/setup-uv@d0cc045d04ccac9d8b7881df0226f9e82c39688e # v6
|
||||
uses: astral-sh/setup-uv@11f9893b081a58869d3b5fccaea48c9e9e46f990 # v8.3.2
|
||||
with:
|
||||
enable-cache: true # zizmor: ignore[cache-poisoning]
|
||||
version: ${{ env.UV_VERSION }}
|
||||
|
||||
@@ -43,12 +43,12 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Checkout code
|
||||
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
|
||||
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
|
||||
with:
|
||||
fetch-depth: 0
|
||||
persist-credentials: false
|
||||
|
||||
- name: Secret Scanning
|
||||
uses: trufflesecurity/trufflehog@eafb8c5f6a06175141c27f17bcc17941853d0047 # v3.90.0
|
||||
uses: trufflesecurity/trufflehog@27b0417c16317ca9a472a9a8092acce143b49c55 # v3.95.9
|
||||
with:
|
||||
extra_args: --only-verified
|
||||
|
||||
@@ -191,162 +191,6 @@ def make_my_policy_pre_post_processors(
|
||||
|
||||
---
|
||||
|
||||
## Adding high- and low-level language control
|
||||
|
||||
The policy API above is sufficient for training and standard evaluation. To use a language-conditioned policy with interactive `lerobot-rollout`, also register a runtime adapter. The adapter keeps policy-specific prompting and tokenization out of the generic control loop.
|
||||
|
||||
The runtime supports two policy shapes:
|
||||
|
||||
| Policy shape | Behavior | Adapter |
|
||||
| ---------------- | ----------------------------------------------------------------------- | ---------------------------------------------- |
|
||||
| Low-level / flat | The operator's task or subtask directly conditions action prediction. | Reuse `DirectTaskPolicyAdapter`. |
|
||||
| High + low level | The policy generates subtasks or memory, then conditions actions on it. | Subclass `BaseLanguageAdapter`, as PI052 does. |
|
||||
|
||||
During a rollout, `RuntimeState` stores the high-level task and the active language context:
|
||||
|
||||
```text
|
||||
task ──> adapter.generate_text("subtask", ...) ──> state.language_context["subtask"]
|
||||
│
|
||||
observation ──> processors ──> adapter.select_action() ─┴─> action chunk ──> robot
|
||||
```
|
||||
|
||||
The generic runtime handles generation frequency, pause/resume, prompt replacement, action queues, and dispatch. The adapter only translates between that runtime contract and your policy.
|
||||
|
||||
### Low-level policies
|
||||
|
||||
If your policy already consumes the live task through its normal preprocessor and implements `predict_action_chunk`, register the shared direct adapter. PI0.5 and MolmoAct2 use this path:
|
||||
|
||||
```python
|
||||
# src/lerobot/runtime/registry.py
|
||||
_ADAPTERS = {
|
||||
# ...
|
||||
"my_policy": "lerobot.runtime.adapter:DirectTaskPolicyAdapter",
|
||||
}
|
||||
```
|
||||
|
||||
Run it with direct-subtask mode so the operator supplies the instruction used by the action policy:
|
||||
|
||||
```bash
|
||||
lerobot-rollout \
|
||||
--language \
|
||||
--policy.path=user/my_policy_checkpoint \
|
||||
--robot.type=so101_follower \
|
||||
--robot.port=/dev/ttyACM0 \
|
||||
--direct_subtask
|
||||
```
|
||||
|
||||
The rollout context builds the observation batch with the current instruction before `DirectTaskPolicyAdapter` calls `policy.predict_action_chunk(observation)`. No text-generation method is required.
|
||||
|
||||
### Hierarchical policies
|
||||
|
||||
For a policy that generates language and actions, subclass [`BaseLanguageAdapter`](https://github.com/huggingface/lerobot/blob/main/src/lerobot/runtime/adapter.py) and implement two methods:
|
||||
|
||||
- `generate_text(kind, observation, state, user_text=None) -> str` generates a `subtask`, `memory`, or interjection response.
|
||||
- `select_action(observation, state)` builds the low-level prompt from the active context and returns an action chunk.
|
||||
|
||||
This abbreviated adapter follows [`PI052PolicyAdapter`](https://github.com/huggingface/lerobot/blob/main/src/lerobot/policies/pi052/inference/pi052_adapter.py):
|
||||
|
||||
```python
|
||||
# inference/my_policy_adapter.py
|
||||
from typing import Any
|
||||
|
||||
from lerobot.runtime import RuntimeState
|
||||
from lerobot.runtime.adapter import BaseLanguageAdapter
|
||||
from lerobot.utils.constants import (
|
||||
OBS_LANGUAGE_ATTENTION_MASK,
|
||||
OBS_LANGUAGE_TOKENS,
|
||||
)
|
||||
|
||||
|
||||
class MyPolicyAdapter(BaseLanguageAdapter):
|
||||
def select_action(self, observation: dict[str, Any], state: RuntimeState):
|
||||
instruction = state.language_context.get("subtask") or state.task or ""
|
||||
tokens, attention_mask = tokenize_instruction(instruction)
|
||||
|
||||
batch = dict(observation)
|
||||
batch[OBS_LANGUAGE_TOKENS] = tokens
|
||||
batch[OBS_LANGUAGE_ATTENTION_MASK] = attention_mask
|
||||
return self.policy.predict_action_chunk(batch)
|
||||
|
||||
def generate_text(
|
||||
self,
|
||||
kind: str,
|
||||
observation: dict[str, Any] | None,
|
||||
state: RuntimeState,
|
||||
user_text: str | None = None,
|
||||
) -> str:
|
||||
messages = self.build_messages(kind, state, user_text)
|
||||
batch, tokenizer = tokenize_messages(messages, observation)
|
||||
return self.policy.select_message(
|
||||
batch,
|
||||
tokenizer=tokenizer,
|
||||
min_new_tokens=self.gen.min_new_tokens,
|
||||
temperature=self.gen.temperature,
|
||||
top_p=self.gen.top_p,
|
||||
)
|
||||
|
||||
def build_messages(
|
||||
self, kind: str, state: RuntimeState, user_text: str | None
|
||||
) -> list[dict[str, str]]:
|
||||
if kind == "subtask":
|
||||
return [{"role": "user", "content": state.task or ""}]
|
||||
if kind == "memory":
|
||||
return [
|
||||
{"role": "user", "content": state.task or ""},
|
||||
{
|
||||
"role": "user",
|
||||
"content": f"Completed subtask: {state.extra.get('prior_subtask', '')}",
|
||||
},
|
||||
]
|
||||
if kind == "interjection":
|
||||
return [
|
||||
{"role": "user", "content": state.task or ""},
|
||||
{"role": "user", "content": user_text or ""},
|
||||
]
|
||||
raise ValueError(f"Unsupported text kind: {kind}")
|
||||
```
|
||||
|
||||
`tokenize_instruction` and `tokenize_messages` are policy-specific helpers. They must reproduce the prompt format used during training; PI052, for example, adds the discretized robot state to its low-level subtask prompt and uses the same PaliGemma formatting for `select_message`.
|
||||
|
||||
`BaseLanguageAdapter` provides the default hierarchy: regenerate a subtask at action-chunk boundaries, update memory when the subtask changes, and handle user interjections. Override `_regenerate_context` only if your policy uses a different hierarchy.
|
||||
|
||||
Register the adapter with a lazy import so importing LeRobot does not load the model or its optional dependencies:
|
||||
|
||||
```python
|
||||
# src/lerobot/runtime/registry.py
|
||||
_ADAPTERS = {
|
||||
# ...
|
||||
"my_policy": "lerobot.policies.my_policy.inference.my_policy_adapter:MyPolicyAdapter",
|
||||
}
|
||||
```
|
||||
|
||||
The key must match the policy's registered type. Once registered, the same checkpoint works through the shared entry point:
|
||||
|
||||
```bash
|
||||
lerobot-rollout \
|
||||
--language \
|
||||
--policy.path=user/my_hierarchical_checkpoint \
|
||||
--robot.type=so101_follower \
|
||||
--robot.port=/dev/ttyACM0 \
|
||||
--task="put the cup in the sink"
|
||||
```
|
||||
|
||||
For RoboCasa-compatible policies, replace the robot arguments with `--sim --sim.task=<task>`. Without `--direct_subtask`, the adapter generates the low-level subtask; with it, the operator bypasses high-level generation and supplies each subtask.
|
||||
|
||||
### Keep training and deployment aligned
|
||||
|
||||
The adapter is intentionally small, but its prompts are part of the model contract:
|
||||
|
||||
- Use the same tokenizer, role formatting, special tokens, image ordering, and state encoding as training.
|
||||
- Condition `select_action` on `state.language_context["subtask"]`, falling back to `state.task` for direct or not-yet-generated prompts.
|
||||
- Return a full action chunk from `select_action`; the runtime handles control-rate dispatch.
|
||||
- Keep optional model dependencies inside lazy imports.
|
||||
- Test adapter selection, generated-message routing, action-batch construction, and direct-subtask behavior with a lightweight fake policy.
|
||||
|
||||
PI052 is the complete in-tree reference: its [processor](https://github.com/huggingface/lerobot/blob/main/src/lerobot/policies/pi052/processor_pi052.py) renders the training recipe, its [policy](https://github.com/huggingface/lerobot/blob/main/src/lerobot/policies/pi052/modeling_pi052.py) exposes text and action generation, and its [adapter](https://github.com/huggingface/lerobot/blob/main/src/lerobot/policies/pi052/inference/pi052_adapter.py) reconstructs those same prompts at deployment.
|
||||
|
||||
---
|
||||
|
||||
## Path A: Out-of-tree plugin
|
||||
|
||||
The fastest way to ship a policy: package it as a standalone Python distribution and install it alongside LeRobot. No PR required, you own the release cycle, and you can publish to PyPI under your own namespace.
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# Policy Deployment (lerobot-rollout)
|
||||
|
||||
`lerobot-rollout` is the single CLI for deploying trained policies on real robots or in an interactive simulator. It supports multiple execution strategies and inference backends, from quick evaluation to continuous recording, language-driven control, and human-in-the-loop data collection.
|
||||
`lerobot-rollout` is the single CLI for deploying trained policies on real robots. It supports multiple execution strategies and inference backends, from quick evaluation to continuous recording and human-in-the-loop data collection.
|
||||
|
||||
## Quick Start
|
||||
|
||||
@@ -197,52 +197,6 @@ Teleop is optional — if omitted the robot holds its position during the reset
|
||||
|
||||
---
|
||||
|
||||
## Interactive language control
|
||||
|
||||
Language-conditioned policies can expose a high-level text head in addition to
|
||||
their action head. Add `--language` to open-prompt one of these policies on a
|
||||
real robot. Language-only flags such as `--direct_subtask` select this mode
|
||||
automatically.
|
||||
|
||||
MolmoAct2 has no high-level planner, so use direct-subtask mode and type each
|
||||
next low-level instruction yourself:
|
||||
|
||||
```bash
|
||||
lerobot-rollout \
|
||||
--policy.path=lerobot/MolmoAct2-SO100_101-LeRobot \
|
||||
--policy.device=cuda \
|
||||
--robot.type=so101_follower \
|
||||
--robot.port=/dev/ttyACM1 \
|
||||
--robot.cameras='{"cam0":{"type":"opencv","index_or_path":"/dev/video0","width":640,"height":480,"fps":30,"fourcc":"MJPG","backend":200},"cam1":{"type":"opencv","index_or_path":"/dev/video2","width":640,"height":480,"fps":30,"fourcc":"MJPG","backend":200}}' \
|
||||
--direct_subtask \
|
||||
--robot.max_relative_target='{"shoulder_pan":5,"shoulder_lift":5,"elbow_flex":5,"wrist_flex":5,"wrist_roll":5,"gripper":5}'
|
||||
```
|
||||
|
||||
The robot starts paused. Type a subtask, then use `/resume` and `/pause` to
|
||||
control action dispatch. Check the workspace and motion limits before resuming.
|
||||
Without `--direct_subtask`, a policy such as PI052 generates its active subtask
|
||||
from the high-level `--task` itself.
|
||||
|
||||
RoboCasa uses the same runtime and processor path. `--sim` selects it
|
||||
automatically, so no robot configuration is needed:
|
||||
|
||||
```bash
|
||||
MUJOCO_GL=egl lerobot-rollout \
|
||||
--policy.path=lerobot/pi052_robocasa \
|
||||
--sim --sim.task=CloseFridge --sim.split=pretrain \
|
||||
--task="close the fridge" \
|
||||
--disable_memory \
|
||||
--sim.render_size=384 \
|
||||
--sim.views=robot0_agentview_left,robot0_eye_in_hand,robot0_agentview_right \
|
||||
--mode=action --ctrl_hz=20
|
||||
```
|
||||
|
||||
Open `http://localhost:8010` for the live simulator view. Add
|
||||
`--sim.direct_subtask` to bypass the language planner and make each typed prompt
|
||||
the action policy's current subtask.
|
||||
|
||||
---
|
||||
|
||||
## Inference Backends
|
||||
|
||||
Select a backend with `--inference.type=<name>`. All strategies work with both backends.
|
||||
|
||||
@@ -141,17 +141,6 @@ sample["target_message_indices"]
|
||||
|
||||
The renderer does not apply a tokenizer chat template. Policy processors decide how to serialize the messages for their backbone, which keeps the same dataset usable across SmolVLA, Pi0.5, and any future VLM that expects OpenAI-style chat messages.
|
||||
|
||||
## Blends
|
||||
|
||||
Blend recipes select one weighted sub-recipe deterministically from the sample index.
|
||||
`recipes/subtask_mem.yaml` trains the compact core blend — high-level subtask prediction, low-level execution, and memory. `recipes/subtask_mem_vqa_speech.yaml` is the fuller variant that also adds VQA and spoken interjection responses.
|
||||
|
||||
A message recipe with a supervised assistant turn on the `low_level` stream trains
|
||||
the π0.5 paper's joint sequence instead of a blend: the target span gets text CE
|
||||
while also conditioning the action losses in the same forward.
|
||||
`recipes/subtask_joint.yaml` is the provided example; pair it with
|
||||
`--policy.joint_subtask_conditioning=true` at inference.
|
||||
|
||||
## Graceful absence
|
||||
|
||||
If both language columns are missing, `None`, or empty, `RenderMessagesStep` is a no-op.
|
||||
|
||||
@@ -33,8 +33,6 @@ class DatasetConfig:
|
||||
# looked up under $HF_LEROBOT_HOME/repo_id and Hub downloads use a revision-safe cache under $HF_LEROBOT_HOME/hub.
|
||||
root: str | None = None
|
||||
episodes: list[int] | None = None
|
||||
# Episode indices to drop (e.g. corrupt or heterogeneous ones). Applied on top of `episodes`.
|
||||
exclude_episodes: list[int] | None = None
|
||||
image_transforms: ImageTransformsConfig = field(default_factory=ImageTransformsConfig)
|
||||
revision: str | None = None
|
||||
use_imagenet_stats: bool = True
|
||||
@@ -64,10 +62,6 @@ class DatasetConfig:
|
||||
if len(self.episodes) != len(set(self.episodes)):
|
||||
duplicates = sorted({ep for ep in self.episodes if self.episodes.count(ep) > 1})
|
||||
raise ValueError(f"Episode indices contain duplicates: {duplicates}")
|
||||
if self.exclude_episodes is not None and any(ep < 0 for ep in self.exclude_episodes):
|
||||
raise ValueError(
|
||||
f"exclude_episodes must be non-negative, got: {[ep for ep in self.exclude_episodes if ep < 0]}"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
|
||||
@@ -78,7 +78,7 @@ class MessageTurn:
|
||||
raise ValueError(f"Unsupported message stream: {self.stream!r}")
|
||||
if self.content is None and self.tool_calls_from is None:
|
||||
raise ValueError("MessageTurn.content is required unless tool_calls_from is set.")
|
||||
if self.content is not None and not isinstance(self.content, str | list):
|
||||
if self.content is not None and not isinstance(self.content, (str, list)):
|
||||
raise TypeError("MessageTurn.content must be a string, a list of HF-style blocks, or None.")
|
||||
if isinstance(self.content, list):
|
||||
for block in self.content:
|
||||
@@ -147,7 +147,7 @@ class TrainingRecipe:
|
||||
return cls.from_dict(data)
|
||||
|
||||
def _validate_message_recipe(self) -> None:
|
||||
"""Validate bindings and require text or low-level action supervision."""
|
||||
"""Ensure every templated binding is known and at least one turn is a target."""
|
||||
assert self.messages is not None
|
||||
known_bindings = set(DEFAULT_BINDINGS) | set(self.bindings or {}) | {"task"}
|
||||
|
||||
@@ -156,14 +156,8 @@ class TrainingRecipe:
|
||||
if missing:
|
||||
raise ValueError(f"MessageTurn references unknown binding(s): {sorted(missing)}")
|
||||
|
||||
has_target = any(turn.target for turn in self.messages)
|
||||
has_low_level = any(turn.stream == "low_level" for turn in self.messages)
|
||||
if not (has_target or has_low_level):
|
||||
raise ValueError(
|
||||
"Message recipes must contain at least one supervised turn — "
|
||||
"either ``target: true`` (text CE) or ``stream: low_level`` "
|
||||
"(flow/action loss)."
|
||||
)
|
||||
if not any(turn.target for turn in self.messages):
|
||||
raise ValueError("Message recipes must contain at least one target turn.")
|
||||
|
||||
def _validate_blend_recipe(self) -> None:
|
||||
"""Ensure each blend component is a non-empty, weighted message recipe."""
|
||||
|
||||
@@ -1,16 +0,0 @@
|
||||
# Predicts subtasks from tasks and trains subtask-conditioned action flow without memory or plans.
|
||||
# Requires `subtask` annotations; samples with missing `if_present` bindings do not render.
|
||||
|
||||
blend:
|
||||
|
||||
high_level_subtask:
|
||||
weight: 0.30
|
||||
messages:
|
||||
- {role: user, content: "${task}", stream: high_level}
|
||||
- {role: assistant, content: "${subtask}", stream: high_level, target: true, if_present: subtask}
|
||||
|
||||
low_level_execution:
|
||||
weight: 0.70
|
||||
messages:
|
||||
# The low-level stream trains action flow on the generated or annotated subtask.
|
||||
- {role: user, content: "${subtask}", stream: low_level, if_present: subtask}
|
||||
@@ -1,13 +0,0 @@
|
||||
# Paper-style joint sequence (pi0.5 §IV-B): one sample supervises the subtask
|
||||
# text with CE and, because the assistant turn is part of the prefix, conditions
|
||||
# the FAST and flow action losses on the same annotated subtask in one forward.
|
||||
# The supervised span is attended causally; the action losses see task + subtask.
|
||||
#
|
||||
# Pair with `--policy.joint_subtask_conditioning=true` at inference so the flow
|
||||
# prefix reproduces this layout (task turn with state + causal generated subtask).
|
||||
# Samples without a `subtask` annotation fall back to a plain task-prompt
|
||||
# low-level sample via `if_present`.
|
||||
|
||||
messages:
|
||||
- {role: user, content: "${task}", stream: low_level}
|
||||
- {role: assistant, content: "${subtask}", stream: low_level, target: true, if_present: subtask}
|
||||
@@ -1,30 +0,0 @@
|
||||
# Trains subtask prediction, subtask-conditioned action flow, and memory updates without plans.
|
||||
# Requires `subtask` and `memory`; missing `if_present` bindings skip the affected sub-recipe.
|
||||
|
||||
blend:
|
||||
|
||||
high_level_subtask:
|
||||
weight: 0.25
|
||||
messages:
|
||||
- {role: user, content: "${task}", stream: high_level}
|
||||
- {role: assistant, content: "${subtask}", stream: high_level, target: true, if_present: subtask}
|
||||
|
||||
low_level_execution:
|
||||
weight: 0.60
|
||||
messages:
|
||||
# The low-level stream trains action flow on the generated or annotated subtask.
|
||||
- {role: user, content: "${subtask}", stream: low_level, if_present: subtask}
|
||||
|
||||
memory_update:
|
||||
# `active_at` densifies sparse boundaries while preserving the prior-memory/subtask mapping.
|
||||
# Inference controls update timing through `subtask_change` events.
|
||||
weight: 0.15
|
||||
bindings:
|
||||
prior_memory: "nth_prev(style=memory, offset=1)"
|
||||
current_memory: "active_at(t, style=memory)"
|
||||
completed_subtask: "nth_prev(style=subtask, offset=1)"
|
||||
messages:
|
||||
- {role: user, content: "${task}", stream: high_level}
|
||||
- {role: assistant, content: "Previous memory: ${prior_memory}", stream: high_level, if_present: prior_memory}
|
||||
- {role: user, content: "Completed subtask: ${completed_subtask}", stream: high_level, if_present: completed_subtask}
|
||||
- {role: assistant, content: "${current_memory}", stream: high_level, target: true, if_present: current_memory}
|
||||
@@ -1,70 +0,0 @@
|
||||
# Adds memory, spoken interjection responses, and camera-grounded VQA to subtask/action training.
|
||||
# Missing optional annotations skip only their sub-recipe; `say` tool calls tokenize as `<say>...</say>`.
|
||||
|
||||
blend:
|
||||
|
||||
high_level_subtask:
|
||||
weight: 0.25
|
||||
messages:
|
||||
- {role: user, content: "${task}", stream: high_level}
|
||||
- {role: assistant, content: "${subtask}", stream: high_level, target: true, if_present: subtask}
|
||||
|
||||
low_level_execution:
|
||||
weight: 0.40
|
||||
messages:
|
||||
# The low-level stream trains action flow on the generated or annotated subtask.
|
||||
- {role: user, content: "${subtask}", stream: low_level, if_present: subtask}
|
||||
|
||||
memory_update:
|
||||
# `active_at` densifies sparse boundaries while preserving the prior-memory/subtask mapping.
|
||||
# Inference controls update timing through `subtask_change` events.
|
||||
weight: 0.10
|
||||
bindings:
|
||||
prior_memory: "nth_prev(style=memory, offset=1)"
|
||||
current_memory: "active_at(t, style=memory)"
|
||||
completed_subtask: "nth_prev(style=subtask, offset=1)"
|
||||
messages:
|
||||
- {role: user, content: "${task}", stream: high_level}
|
||||
- {role: assistant, content: "Previous memory: ${prior_memory}", stream: high_level, if_present: prior_memory}
|
||||
- {role: user, content: "Completed subtask: ${completed_subtask}", stream: high_level, if_present: completed_subtask}
|
||||
- {role: assistant, content: "${current_memory}", stream: high_level, target: true, if_present: current_memory}
|
||||
|
||||
user_interjection_response:
|
||||
weight: 0.10
|
||||
bindings:
|
||||
interjection: "emitted_at(t, style=interjection)"
|
||||
speech: "emitted_at(t, role=assistant, tool_name=say)"
|
||||
messages:
|
||||
- {role: user, content: "${task}", stream: high_level}
|
||||
- {role: user, content: "${interjection}", stream: high_level, if_present: interjection}
|
||||
# The assistant target is a `say` tool call flattened to a `<say>...</say>` marker.
|
||||
- {role: assistant, stream: high_level, target: true, if_present: speech, tool_calls_from: speech}
|
||||
|
||||
# Each camera uses a separate VQA sub-recipe for view-specific binding.
|
||||
ask_vqa_top:
|
||||
weight: 0.075
|
||||
bindings:
|
||||
vqa_query: "emitted_at(t, style=vqa, role=user, camera=observation.images.front)"
|
||||
vqa: "emitted_at(t, style=vqa, role=assistant, camera=observation.images.front)"
|
||||
messages:
|
||||
- role: user
|
||||
stream: high_level
|
||||
if_present: vqa_query
|
||||
content:
|
||||
- {type: image, feature: observation.images.front}
|
||||
- {type: text, text: "${vqa_query}"}
|
||||
- {role: assistant, content: "${vqa}", stream: high_level, target: true, if_present: vqa}
|
||||
|
||||
ask_vqa_wrist:
|
||||
weight: 0.075
|
||||
bindings:
|
||||
vqa_query: "emitted_at(t, style=vqa, role=user, camera=observation.images.wrist)"
|
||||
vqa: "emitted_at(t, style=vqa, role=assistant, camera=observation.images.wrist)"
|
||||
messages:
|
||||
- role: user
|
||||
stream: high_level
|
||||
if_present: vqa_query
|
||||
content:
|
||||
- {type: image, feature: observation.images.wrist}
|
||||
- {type: text, text: "${vqa_query}"}
|
||||
- {role: assistant, content: "${vqa}", stream: high_level, target: true, if_present: vqa}
|
||||
@@ -14,7 +14,6 @@
|
||||
import builtins
|
||||
import datetime as dt
|
||||
import json
|
||||
import multiprocessing
|
||||
import os
|
||||
import tempfile
|
||||
from dataclasses import dataclass, field
|
||||
@@ -102,12 +101,6 @@ class TrainPipelineConfig(HubMixin):
|
||||
batch_size: int = 8
|
||||
prefetch_factor: int = 4
|
||||
persistent_workers: bool = True
|
||||
# DataLoader worker start method. "spawn" is safer than "fork" with
|
||||
# non-fork-safe libs (PyAV / torchcodec / ffmpeg), but adds some
|
||||
# worker-startup time per run since workers re-import modules instead
|
||||
# of inheriting parent state. Override with `--dataloader_multiprocessing_context=fork`
|
||||
# when appropriate, or set it to `null` to use Python's platform default.
|
||||
dataloader_multiprocessing_context: str | None = "spawn"
|
||||
steps: int = 100_000
|
||||
# Run policy in the simulation environment every N steps to measure reward/success (0 = disabled).
|
||||
env_eval_freq: int = 20_000
|
||||
@@ -219,17 +212,6 @@ class TrainPipelineConfig(HubMixin):
|
||||
self.reward_model.pretrained_path = str(policy_dir)
|
||||
|
||||
def validate(self) -> None:
|
||||
available_contexts = multiprocessing.get_all_start_methods()
|
||||
if (
|
||||
self.dataloader_multiprocessing_context is not None
|
||||
and self.dataloader_multiprocessing_context not in available_contexts
|
||||
):
|
||||
raise ValueError(
|
||||
"`dataloader_multiprocessing_context` must be None or one of "
|
||||
f"{available_contexts} on this platform, got "
|
||||
f"{self.dataloader_multiprocessing_context!r}."
|
||||
)
|
||||
|
||||
self._resolve_pretrained_from_cli()
|
||||
|
||||
if self.policy is None and self.reward_model is None:
|
||||
|
||||
@@ -163,40 +163,10 @@ class DatasetReader:
|
||||
def _load_hf_dataset(self) -> datasets.Dataset:
|
||||
"""hf_dataset contains all the observations, states, actions, rewards, etc."""
|
||||
features = get_hf_features_from_features(self._meta.features)
|
||||
# Annotated datasets may have language columns absent from metadata.
|
||||
# Extend the schema before the strict Parquet cast.
|
||||
features = self._extend_features_with_language_columns(features)
|
||||
hf_dataset = load_nested_dataset(self.root / "data", features=features, episodes=self.episodes)
|
||||
hf_dataset.set_transform(hf_transform_to_torch)
|
||||
return hf_dataset
|
||||
|
||||
def _extend_features_with_language_columns(self, features: datasets.Features) -> datasets.Features:
|
||||
"""Register language columns found in Parquet but missing from metadata."""
|
||||
# Leave empty datasets to fail through the normal loading path.
|
||||
try:
|
||||
sample = next((self.root / "data").glob("*/*.parquet"))
|
||||
except StopIteration:
|
||||
return features
|
||||
|
||||
from pyarrow import parquet as _pq # noqa: PLC0415
|
||||
|
||||
schema_names = set(_pq.read_schema(sample).names)
|
||||
from .language import ( # noqa: PLC0415
|
||||
LANGUAGE_EVENTS,
|
||||
LANGUAGE_PERSISTENT,
|
||||
language_events_column_feature,
|
||||
language_persistent_column_feature,
|
||||
)
|
||||
|
||||
extra: dict[str, object] = {}
|
||||
if LANGUAGE_PERSISTENT in schema_names and LANGUAGE_PERSISTENT not in features:
|
||||
extra[LANGUAGE_PERSISTENT] = language_persistent_column_feature()
|
||||
if LANGUAGE_EVENTS in schema_names and LANGUAGE_EVENTS not in features:
|
||||
extra[LANGUAGE_EVENTS] = language_events_column_feature()
|
||||
if not extra:
|
||||
return features
|
||||
return datasets.Features({**features, **extra})
|
||||
|
||||
def _check_cached_episodes_sufficient(self) -> bool:
|
||||
"""Check if the cached dataset contains all requested episodes and their video files."""
|
||||
if self.hf_dataset is None or len(self.hf_dataset) == 0:
|
||||
|
||||
@@ -172,23 +172,6 @@ class DatasetWriter:
|
||||
def _get_image_file_dir(self, episode_index: int, image_key: str) -> Path:
|
||||
return self._get_image_file_path(episode_index, image_key, frame_index=0).parent
|
||||
|
||||
def _get_episode_buffer_index(self) -> int:
|
||||
episode_index = self.episode_buffer["episode_index"]
|
||||
# episode_index is `int` when freshly created, but becomes `np.ndarray` after
|
||||
# save_episode() mutates the buffer. Handle both types here.
|
||||
if isinstance(episode_index, np.ndarray):
|
||||
episode_index = episode_index.item() if episode_index.size == 1 else episode_index[0]
|
||||
return int(episode_index)
|
||||
|
||||
def _delete_camera_frame_dirs(self, camera_keys: list[str]) -> None:
|
||||
if self.image_writer is not None:
|
||||
self._wait_image_writer()
|
||||
episode_index = self._get_episode_buffer_index()
|
||||
for camera_key in camera_keys:
|
||||
img_dir = self._get_image_file_dir(episode_index, camera_key)
|
||||
if img_dir.is_dir():
|
||||
shutil.rmtree(img_dir)
|
||||
|
||||
def _save_image(
|
||||
self, image: torch.Tensor | np.ndarray | PIL.Image.Image, fpath: Path, compress_level: int = 1
|
||||
) -> None:
|
||||
@@ -386,9 +369,7 @@ class DatasetWriter:
|
||||
self._episodes_since_last_encoding = 0
|
||||
|
||||
if episode_data is None:
|
||||
if len(self._meta.image_keys) > 0:
|
||||
self._delete_camera_frame_dirs(self._meta.image_keys)
|
||||
self.episode_buffer = self._create_episode_buffer()
|
||||
self.clear_episode_buffer(delete_images=len(self._meta.image_keys) > 0)
|
||||
|
||||
def _batch_save_episode_video(self, start_episode: int, end_episode: int | None = None) -> None:
|
||||
"""Batch save videos for multiple episodes."""
|
||||
@@ -580,10 +561,10 @@ class DatasetWriter:
|
||||
return metadata
|
||||
|
||||
def clear_episode_buffer(self, delete_images: bool = True) -> None:
|
||||
"""Discard the current episode buffer and optionally delete temp camera frames.
|
||||
"""Discard the current episode buffer and optionally delete temp images.
|
||||
|
||||
Args:
|
||||
delete_images: If ``True``, remove temporary camera frame directories
|
||||
delete_images: If ``True``, remove temporary image directories
|
||||
written for the current episode.
|
||||
"""
|
||||
# Cancel streaming encoder if active
|
||||
@@ -591,7 +572,17 @@ class DatasetWriter:
|
||||
self._streaming_encoder.cancel_episode()
|
||||
|
||||
if delete_images:
|
||||
self._delete_camera_frame_dirs(self._meta.camera_keys)
|
||||
if self.image_writer is not None:
|
||||
self._wait_image_writer()
|
||||
episode_index = self.episode_buffer["episode_index"]
|
||||
# episode_index is `int` when freshly created, but becomes `np.ndarray` after
|
||||
# save_episode() mutates the buffer. Handle both types here.
|
||||
if isinstance(episode_index, np.ndarray):
|
||||
episode_index = episode_index.item() if episode_index.size == 1 else episode_index[0]
|
||||
for cam_key in self._meta.image_keys:
|
||||
img_dir = self._get_image_file_dir(episode_index, cam_key)
|
||||
if img_dir.is_dir():
|
||||
shutil.rmtree(img_dir)
|
||||
|
||||
self.episode_buffer = self._create_episode_buffer()
|
||||
|
||||
|
||||
@@ -66,17 +66,6 @@ def resolve_delta_timestamps(
|
||||
return delta_timestamps
|
||||
|
||||
|
||||
def _resolve_episodes(
|
||||
episodes: list[int] | None, exclude_episodes: list[int] | None, total_episodes: int
|
||||
) -> list[int] | None:
|
||||
"""Apply an episode exclusion list on top of an optional allowlist."""
|
||||
if not exclude_episodes:
|
||||
return episodes
|
||||
base = episodes if episodes is not None else list(range(total_episodes))
|
||||
excluded = set(exclude_episodes)
|
||||
return [episode for episode in base if episode not in excluded]
|
||||
|
||||
|
||||
def make_dataset(cfg: TrainPipelineConfig) -> LeRobotDataset | MultiLeRobotDataset:
|
||||
"""Handles the logic of setting up delta timestamps and image transforms before creating a dataset.
|
||||
|
||||
@@ -98,14 +87,11 @@ def make_dataset(cfg: TrainPipelineConfig) -> LeRobotDataset | MultiLeRobotDatas
|
||||
cfg.dataset.repo_id, root=cfg.dataset.root, revision=cfg.dataset.revision
|
||||
)
|
||||
delta_timestamps = resolve_delta_timestamps(cfg.trainable_config, ds_meta)
|
||||
episodes = _resolve_episodes(
|
||||
cfg.dataset.episodes, cfg.dataset.exclude_episodes, ds_meta.total_episodes
|
||||
)
|
||||
if not cfg.dataset.streaming:
|
||||
dataset = LeRobotDataset(
|
||||
cfg.dataset.repo_id,
|
||||
root=cfg.dataset.root,
|
||||
episodes=episodes,
|
||||
episodes=cfg.dataset.episodes,
|
||||
delta_timestamps=delta_timestamps,
|
||||
image_transforms=image_transforms,
|
||||
revision=cfg.dataset.revision,
|
||||
@@ -118,7 +104,7 @@ def make_dataset(cfg: TrainPipelineConfig) -> LeRobotDataset | MultiLeRobotDatas
|
||||
dataset = StreamingLeRobotDataset(
|
||||
cfg.dataset.repo_id,
|
||||
root=cfg.dataset.root,
|
||||
episodes=episodes,
|
||||
episodes=cfg.dataset.episodes,
|
||||
delta_timestamps=delta_timestamps,
|
||||
image_transforms=image_transforms,
|
||||
revision=cfg.dataset.revision,
|
||||
|
||||
@@ -162,28 +162,14 @@ def render_sample(
|
||||
task: str | None = None,
|
||||
dataset_ctx: Any | None = None,
|
||||
) -> RenderedMessages | None:
|
||||
"""Resolve one sample's bindings and render its message recipe.
|
||||
"""Render the chat-style messages for a single dataset sample.
|
||||
|
||||
Returns ``None`` when no text or low-level action supervision applies.
|
||||
Resolves the recipe's bindings against ``persistent`` and ``events`` rows
|
||||
at frame timestamp ``t``, then expands the recipe's message templates.
|
||||
Returns ``None`` if the resolved sample contains no target message.
|
||||
"""
|
||||
persistent_rows = _normalize_rows(persistent or [])
|
||||
event_rows = _normalize_rows(events or [])
|
||||
|
||||
# Route sparse VQA frames to a matching view-specific component before weighted selection.
|
||||
# This avoids dropping annotated frames or selecting VQA without annotations.
|
||||
if recipe.blend is not None:
|
||||
vqa_rendered = _render_vqa_if_present(
|
||||
recipe,
|
||||
persistent=persistent_rows,
|
||||
events=event_rows,
|
||||
t=t,
|
||||
sample_idx=sample_idx,
|
||||
task=task,
|
||||
dataset_ctx=dataset_ctx,
|
||||
)
|
||||
if vqa_rendered is not None:
|
||||
return vqa_rendered
|
||||
|
||||
selected_recipe = _select_recipe(recipe, sample_idx)
|
||||
bindings = _resolve_bindings(
|
||||
selected_recipe,
|
||||
@@ -197,55 +183,6 @@ def render_sample(
|
||||
return _render_message_recipe(selected_recipe, bindings)
|
||||
|
||||
|
||||
def _render_vqa_if_present(
|
||||
recipe: TrainingRecipe,
|
||||
*,
|
||||
persistent: Sequence[LanguageRow],
|
||||
events: Sequence[LanguageRow],
|
||||
t: float,
|
||||
sample_idx: int,
|
||||
task: str | None,
|
||||
dataset_ctx: Any | None,
|
||||
) -> RenderedMessages | None:
|
||||
"""Render a matching VQA component, or return ``None`` for normal selection.
|
||||
|
||||
Multiple matching views are selected deterministically by relative weight.
|
||||
"""
|
||||
assert recipe.blend is not None
|
||||
renderable: list[tuple[float, RenderedMessages]] = []
|
||||
for name, component in recipe.blend.items():
|
||||
if not name.startswith("ask_vqa"):
|
||||
continue
|
||||
bindings = _resolve_bindings(
|
||||
component,
|
||||
persistent=persistent,
|
||||
events=events,
|
||||
t=t,
|
||||
sample_idx=sample_idx,
|
||||
task=task,
|
||||
dataset_ctx=dataset_ctx,
|
||||
)
|
||||
rendered = _render_message_recipe(component, bindings)
|
||||
if rendered is not None:
|
||||
renderable.append((float(component.weight or 0.0), rendered))
|
||||
|
||||
if not renderable:
|
||||
return None
|
||||
if len(renderable) == 1:
|
||||
return renderable[0][1]
|
||||
|
||||
# Choose among matching cameras by relative weight, or uniformly when all weights are zero.
|
||||
total = sum(w for w, _ in renderable) or float(len(renderable))
|
||||
digest = hashlib.blake2b(f"vqa:{sample_idx}".encode(), digest_size=8).digest()
|
||||
draw = int.from_bytes(digest, "big") / 2**64 * total
|
||||
cumulative = 0.0
|
||||
for w, rendered in renderable:
|
||||
cumulative += w or (total / len(renderable))
|
||||
if draw < cumulative:
|
||||
return rendered
|
||||
return renderable[-1][1]
|
||||
|
||||
|
||||
def _select_recipe(recipe: TrainingRecipe, sample_idx: int) -> TrainingRecipe:
|
||||
"""Pick a deterministic blend component for ``sample_idx`` (or return ``recipe``)."""
|
||||
if recipe.blend is None:
|
||||
@@ -409,9 +346,7 @@ def _render_message_recipe(
|
||||
if turn.target:
|
||||
target_indices.append(message_idx)
|
||||
|
||||
# Keep samples with either text targets or low-level action supervision.
|
||||
has_low_level = any(stream == "low_level" for stream in streams)
|
||||
if not target_indices and not has_low_level:
|
||||
if not target_indices:
|
||||
return None
|
||||
|
||||
rendered = {
|
||||
@@ -468,12 +403,14 @@ def _validate_rendered(rendered: RenderedMessages) -> None:
|
||||
|
||||
if len(streams) != len(messages):
|
||||
raise ValueError("message_streams must be aligned with messages.")
|
||||
# Require text or low-level action supervision.
|
||||
if not target_indices and not any(s == "low_level" for s in streams):
|
||||
raise ValueError("Rendered samples must contain a target message or a low_level-stream message.")
|
||||
if not target_indices:
|
||||
raise ValueError("Rendered samples must contain at least one target message.")
|
||||
for idx in target_indices:
|
||||
if idx < 0 or idx >= len(messages):
|
||||
raise ValueError(f"Target message index {idx} is out of bounds.")
|
||||
# ``stream`` is enforced non-None at MessageTurn construction time
|
||||
# (see ``MessageTurn.__post_init__``), so a missing stream here would
|
||||
# mean the dataclass invariant was bypassed; no need to re-check.
|
||||
|
||||
|
||||
def _nth_relative(
|
||||
|
||||
@@ -560,13 +560,7 @@ class RoboCasaEnv(EnvConfig):
|
||||
kwargs["split"] = self.split
|
||||
return kwargs
|
||||
|
||||
def create_envs(
|
||||
self,
|
||||
n_envs: int,
|
||||
use_async_envs: bool = False,
|
||||
terminate_on_success: bool = True,
|
||||
horizon: int | None = None,
|
||||
):
|
||||
def create_envs(self, n_envs: int, use_async_envs: bool = False):
|
||||
from .robocasa import create_robocasa_envs
|
||||
|
||||
if self.task is None:
|
||||
@@ -580,8 +574,6 @@ class RoboCasaEnv(EnvConfig):
|
||||
env_cls=env_cls,
|
||||
episode_length=self.episode_length,
|
||||
obj_registries=tuple(self.obj_registries),
|
||||
terminate_on_success=terminate_on_success,
|
||||
horizon=horizon,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -155,7 +155,6 @@ class MetaworldEnv(gym.Env):
|
||||
env.model.cam_pos[2] = [0.75, 0.075, 0.7]
|
||||
env.reset()
|
||||
env._freeze_rand_vec = False # otherwise no randomization
|
||||
env.seeded_rand_vec = True # use seeded RNG so reset(seed=X) controls object positions
|
||||
self._env = env
|
||||
|
||||
def render(self) -> np.ndarray:
|
||||
@@ -221,8 +220,6 @@ class MetaworldEnv(gym.Env):
|
||||
self._ensure_env()
|
||||
super().reset(seed=seed)
|
||||
|
||||
if seed is not None:
|
||||
self._env.seed(seed)
|
||||
raw_obs, info = self._env.reset(seed=seed)
|
||||
|
||||
observation = self._format_raw_obs(raw_obs)
|
||||
|
||||
@@ -33,8 +33,8 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
# Dimensions for the flat action/state vectors used by the LeRobot wrapper.
|
||||
# These correspond to the PandaOmron robot in RoboCasa365.
|
||||
OBS_STATE_DIM = 16 # ee_pos_rel(3) + ee_quat_rel(4) + base_pos(3) + base_quat(4) + gripper_qpos(2)
|
||||
ACTION_DIM = 12 # ee_pos(3) + ee_rot(3) + gripper(1) + base_motion(4) + control_mode(1)
|
||||
OBS_STATE_DIM = 16 # base_pos(3) + base_quat(4) + ee_pos_rel(3) + ee_quat_rel(4) + gripper_qpos(2)
|
||||
ACTION_DIM = 12 # base_motion(4) + control_mode(1) + ee_pos(3) + ee_rot(3) + gripper(1)
|
||||
ACTION_LOW = -1.0
|
||||
ACTION_HIGH = 1.0
|
||||
|
||||
@@ -101,15 +101,14 @@ def _resolve_tasks(task: str) -> tuple[list[str], str | None]:
|
||||
def convert_action(flat_action: np.ndarray) -> dict[str, Any]:
|
||||
"""Split a flat (12,) action vector into a RoboCasa action dict.
|
||||
|
||||
Layout (openpi / robocasa.utils.env_utils.convert_action order):
|
||||
ee_pos(3) + ee_rot(3) + gripper(1) + base_motion(4) + control_mode(1)
|
||||
Layout: base_motion(4) + control_mode(1) + ee_pos(3) + ee_rot(3) + gripper(1)
|
||||
"""
|
||||
return {
|
||||
"action.end_effector_position": flat_action[0:3],
|
||||
"action.end_effector_rotation": flat_action[3:6],
|
||||
"action.gripper_close": flat_action[6:7],
|
||||
"action.base_motion": flat_action[7:11],
|
||||
"action.control_mode": flat_action[11:12],
|
||||
"action.base_motion": flat_action[0:4],
|
||||
"action.control_mode": flat_action[4:5],
|
||||
"action.end_effector_position": flat_action[5:8],
|
||||
"action.end_effector_rotation": flat_action[8:11],
|
||||
"action.gripper_close": flat_action[11:12],
|
||||
}
|
||||
|
||||
|
||||
@@ -137,16 +136,9 @@ class RoboCasaEnv(gym.Env):
|
||||
episode_length: int | None = None,
|
||||
obj_registries: Sequence[str] = DEFAULT_OBJ_REGISTRIES,
|
||||
episode_index: int = 0,
|
||||
terminate_on_success: bool = True,
|
||||
horizon: int | None = None,
|
||||
):
|
||||
super().__init__()
|
||||
self.task = task
|
||||
# When False, a task-success does NOT end/reset the episode — used by the
|
||||
# interactive sim so one kitchen persists across sequential prompts.
|
||||
self.terminate_on_success = terminate_on_success
|
||||
# Underlying robosuite horizon (steps before truncation). None -> default.
|
||||
self.horizon = horizon
|
||||
self.obs_type = obs_type
|
||||
self.render_mode = render_mode
|
||||
self.observation_width = observation_width
|
||||
@@ -218,16 +210,12 @@ class RoboCasaEnv(gym.Env):
|
||||
# (only None/"all"/"pretrain"/"target" are valid). Always pass a
|
||||
# valid value so we don't hit that default. Extra kwargs are
|
||||
# forwarded to the underlying kitchen env via create_env/robosuite.make.
|
||||
extra_kwargs: dict[str, Any] = {}
|
||||
if self.horizon is not None:
|
||||
extra_kwargs["horizon"] = int(self.horizon)
|
||||
self._env = RoboCasaGymEnv(
|
||||
env_name=self.task,
|
||||
camera_widths=self.observation_width,
|
||||
camera_heights=self.observation_height,
|
||||
split=self.split if self.split is not None else "all",
|
||||
obj_registries=self.obj_registries,
|
||||
**extra_kwargs,
|
||||
)
|
||||
|
||||
ep_meta = self._env.env.get_ep_meta()
|
||||
@@ -242,14 +230,12 @@ class RoboCasaEnv(gym.Env):
|
||||
return {"pixels": images}
|
||||
|
||||
# `state.*` keys come from PandaOmronKeyConverter inside the wrapper.
|
||||
# openpi state order: ee first, then base, then gripper (matches the
|
||||
# openpi robocasa pipeline / examples/robocasa/main.py state layout).
|
||||
agent_pos = np.concatenate(
|
||||
[
|
||||
raw_obs.get("state.end_effector_position_relative", np.zeros(3)),
|
||||
raw_obs.get("state.end_effector_rotation_relative", np.zeros(4)),
|
||||
raw_obs.get("state.base_position", np.zeros(3)),
|
||||
raw_obs.get("state.base_rotation", np.zeros(4)),
|
||||
raw_obs.get("state.end_effector_position_relative", np.zeros(3)),
|
||||
raw_obs.get("state.end_effector_rotation_relative", np.zeros(4)),
|
||||
raw_obs.get("state.gripper_qpos", np.zeros(2)),
|
||||
],
|
||||
axis=-1,
|
||||
@@ -294,7 +280,7 @@ class RoboCasaEnv(gym.Env):
|
||||
raw_obs, reward, done, truncated, info = self._env.step(action_dict)
|
||||
|
||||
is_success = bool(info.get("success", False))
|
||||
terminated = done or (is_success and self.terminate_on_success)
|
||||
terminated = done or is_success
|
||||
info.update({"task": self.task, "done": done, "is_success": is_success})
|
||||
|
||||
observation = self._format_raw_obs(raw_obs)
|
||||
@@ -327,8 +313,6 @@ def _make_env_fns(
|
||||
split: str | None,
|
||||
episode_length: int | None,
|
||||
obj_registries: Sequence[str],
|
||||
terminate_on_success: bool = True,
|
||||
horizon: int | None = None,
|
||||
) -> list[Callable[[], RoboCasaEnv]]:
|
||||
"""Build n_envs factory callables for a single task.
|
||||
|
||||
@@ -351,8 +335,6 @@ def _make_env_fns(
|
||||
episode_length=episode_length,
|
||||
obj_registries=obj_registries,
|
||||
episode_index=episode_index,
|
||||
terminate_on_success=terminate_on_success,
|
||||
horizon=horizon,
|
||||
)
|
||||
|
||||
return [partial(_make_env, i) for i in range(n_envs)]
|
||||
@@ -366,8 +348,6 @@ def create_robocasa_envs(
|
||||
env_cls: Callable[[Sequence[Callable[[], Any]]], Any] | None = None,
|
||||
episode_length: int | None = None,
|
||||
obj_registries: Sequence[str] = DEFAULT_OBJ_REGISTRIES,
|
||||
terminate_on_success: bool = True,
|
||||
horizon: int | None = None,
|
||||
) -> dict[str, dict[int, Any]]:
|
||||
"""Create vectorized RoboCasa365 environments with a consistent return shape.
|
||||
|
||||
@@ -429,8 +409,6 @@ def create_robocasa_envs(
|
||||
split=split,
|
||||
episode_length=episode_length,
|
||||
obj_registries=obj_registries,
|
||||
terminate_on_success=terminate_on_success,
|
||||
horizon=horizon,
|
||||
)
|
||||
|
||||
if is_async:
|
||||
|
||||
@@ -177,7 +177,6 @@ def make_pre_post_processors(
|
||||
return make_groot_pre_post_processors_from_pretrained(
|
||||
config=policy_cfg,
|
||||
pretrained_path=pretrained_path,
|
||||
revision=pretrained_revision,
|
||||
dataset_stats=kwargs.get("dataset_stats"),
|
||||
dataset_meta=kwargs.get("dataset_meta"),
|
||||
preprocessor_overrides=kwargs.get("preprocessor_overrides"),
|
||||
|
||||
@@ -475,7 +475,6 @@ def make_groot_pre_post_processors_from_pretrained(
|
||||
config: GrootConfig,
|
||||
pretrained_path: str,
|
||||
*,
|
||||
revision: str | None = None,
|
||||
dataset_stats: dict[str, dict[str, torch.Tensor]] | None = None,
|
||||
dataset_meta: Any | None = None,
|
||||
preprocessor_overrides: dict[str, Any] | None = None,
|
||||
@@ -512,7 +511,6 @@ def make_groot_pre_post_processors_from_pretrained(
|
||||
|
||||
preprocessor, postprocessor = _load_groot_processor_pipelines(
|
||||
pretrained_path,
|
||||
revision=revision,
|
||||
preprocessor_overrides=preprocessor_overrides,
|
||||
postprocessor_overrides=postprocessor_overrides,
|
||||
preprocessor_config_filename=preprocessor_config_filename,
|
||||
@@ -528,7 +526,6 @@ def make_groot_pre_post_processors_from_pretrained(
|
||||
def _load_groot_processor_pipelines(
|
||||
pretrained_path: str,
|
||||
*,
|
||||
revision: str | None,
|
||||
preprocessor_overrides: dict[str, Any],
|
||||
postprocessor_overrides: dict[str, Any],
|
||||
preprocessor_config_filename: str,
|
||||
@@ -543,7 +540,6 @@ def _load_groot_processor_pipelines(
|
||||
preprocessor = PolicyProcessorPipeline.from_pretrained(
|
||||
pretrained_model_name_or_path=pretrained_path,
|
||||
config_filename=preprocessor_config_filename,
|
||||
revision=revision,
|
||||
overrides=preprocessor_overrides,
|
||||
to_transition=batch_to_transition,
|
||||
to_output=transition_to_batch,
|
||||
@@ -551,7 +547,6 @@ def _load_groot_processor_pipelines(
|
||||
postprocessor = PolicyProcessorPipeline.from_pretrained(
|
||||
pretrained_model_name_or_path=pretrained_path,
|
||||
config_filename=postprocessor_config_filename,
|
||||
revision=revision,
|
||||
overrides=postprocessor_overrides,
|
||||
to_transition=policy_action_to_transition,
|
||||
to_output=transition_to_policy_action,
|
||||
|
||||
@@ -175,6 +175,9 @@ class AddBatchDimensionComplementaryDataStep(ComplementaryDataProcessorStep):
|
||||
if isinstance(task_index_value, Tensor) and task_index_value.dim() == 0:
|
||||
complementary_data["task_index"] = task_index_value.unsqueeze(0)
|
||||
|
||||
complementary_data.pop("language_persistent", None)
|
||||
complementary_data.pop("language_events", None)
|
||||
|
||||
if "messages" in complementary_data:
|
||||
messages = complementary_data["messages"]
|
||||
if isinstance(messages, list) and (not messages or isinstance(messages[0], dict)):
|
||||
|
||||
@@ -41,7 +41,7 @@ from pathlib import Path
|
||||
from typing import Any, TypedDict, TypeVar, cast
|
||||
|
||||
import torch
|
||||
from huggingface_hub import hf_hub_download, snapshot_download
|
||||
from huggingface_hub import hf_hub_download
|
||||
from safetensors.torch import load_file, save_file
|
||||
|
||||
from lerobot.configs import PipelineFeatureType, PolicyFeature
|
||||
@@ -205,10 +205,6 @@ class ProcessorStep(ABC):
|
||||
"""
|
||||
return None
|
||||
|
||||
def save_artifacts(self, save_directory: Path) -> dict[str, str]:
|
||||
"""Save non-tensor assets and map constructor arguments to relative paths."""
|
||||
return {}
|
||||
|
||||
def reset(self) -> None:
|
||||
"""Resets the internal state of the processor step, if any."""
|
||||
return None
|
||||
@@ -553,22 +549,6 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
pipeline_config = self.get_config()
|
||||
pipeline_state_dict = self.state_dict()
|
||||
|
||||
for processor_step, step_entry in zip(self.steps, pipeline_config["steps"], strict=True):
|
||||
artifacts = processor_step.save_artifacts(save_directory)
|
||||
if artifacts:
|
||||
for config_key, relative_path in artifacts.items():
|
||||
artifact_path = Path(relative_path)
|
||||
if artifact_path.is_absolute() or ".." in artifact_path.parts:
|
||||
raise ValueError(
|
||||
f"Processor artifact path must be relative to the checkpoint: {relative_path!r}"
|
||||
)
|
||||
if not (save_directory / artifact_path).exists():
|
||||
raise FileNotFoundError(
|
||||
f"Processor step did not save declared artifact '{relative_path}'"
|
||||
)
|
||||
step_entry["config"][config_key] = artifact_path.as_posix()
|
||||
step_entry["artifacts"] = artifacts
|
||||
|
||||
for state_key, step_state_dict in pipeline_state_dict.items():
|
||||
state_filename = f"{state_key}.safetensors"
|
||||
save_file(step_state_dict, save_directory / state_filename)
|
||||
@@ -733,8 +713,6 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
ProcessorMigrationError: If the model requires migration to processor format.
|
||||
"""
|
||||
model_id = str(pretrained_model_name_or_path)
|
||||
model_path = Path(model_id)
|
||||
is_local_source = model_path.is_dir() or model_path.is_file()
|
||||
hub_download_kwargs = {
|
||||
"force_download": force_download,
|
||||
"resume_download": resume_download,
|
||||
@@ -753,13 +731,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
|
||||
# 3. Build steps with overrides
|
||||
steps, validated_overrides = cls._build_steps_with_overrides(
|
||||
loaded_config,
|
||||
overrides or {},
|
||||
model_id,
|
||||
base_path,
|
||||
config_filename,
|
||||
hub_download_kwargs,
|
||||
is_local_source,
|
||||
loaded_config, overrides or {}, model_id, base_path, hub_download_kwargs
|
||||
)
|
||||
|
||||
# 4. Validate that all overrides were used
|
||||
@@ -948,9 +920,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
overrides: dict[str, Any],
|
||||
model_id: str,
|
||||
base_path: Path | None,
|
||||
config_filename: str,
|
||||
hub_download_kwargs: dict[str, Any],
|
||||
is_local_source: bool = False,
|
||||
) -> tuple[list[ProcessorStep], set[str]]:
|
||||
"""Build all processor steps with overrides and state loading.
|
||||
|
||||
@@ -974,7 +944,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
3. **State Loading** (via _load_step_state):
|
||||
- **If step has "state_file"**: Load tensor state from .safetensors
|
||||
- **Local first**: Check base_path/state_file.safetensors
|
||||
- **Hub fallback**: Download state file if the pipeline was loaded from the Hub
|
||||
- **Hub fallback**: Download state file if not found locally
|
||||
- **Optional**: Only load if step has load_state_dict method
|
||||
|
||||
4. **Override Tracking**:
|
||||
@@ -992,7 +962,6 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
model_id: The model identifier (needed for Hub state file downloads)
|
||||
base_path: Local directory path for finding state files
|
||||
hub_download_kwargs: Parameters for hf_hub_download (tokens, cache, etc.)
|
||||
is_local_source: Whether model_id resolved to a local directory or config file.
|
||||
|
||||
Returns:
|
||||
Tuple of (instantiated_steps_list, unused_override_keys)
|
||||
@@ -1003,68 +972,13 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
ImportError: If a step class cannot be imported or found in registry
|
||||
ValueError: If a step cannot be instantiated with its configuration
|
||||
"""
|
||||
loaded_config = deepcopy(loaded_config)
|
||||
cls._resolve_artifact_paths(
|
||||
loaded_config,
|
||||
model_id,
|
||||
base_path,
|
||||
config_filename,
|
||||
hub_download_kwargs,
|
||||
)
|
||||
steps, remaining_override_keys = cls._build_steps_from_config(loaded_config, overrides)
|
||||
|
||||
for step_instance, step_entry in zip(steps, loaded_config["steps"], strict=True):
|
||||
cls._load_step_state(
|
||||
step_instance,
|
||||
step_entry,
|
||||
model_id,
|
||||
base_path,
|
||||
config_filename,
|
||||
hub_download_kwargs,
|
||||
is_local_source,
|
||||
)
|
||||
cls._load_step_state(step_instance, step_entry, model_id, base_path, hub_download_kwargs)
|
||||
|
||||
return steps, remaining_override_keys
|
||||
|
||||
@classmethod
|
||||
def _resolve_artifact_paths(
|
||||
cls,
|
||||
loaded_config: dict[str, Any],
|
||||
model_id: str,
|
||||
base_path: Path | None,
|
||||
config_filename: str,
|
||||
hub_download_kwargs: dict[str, Any],
|
||||
) -> None:
|
||||
"""Resolve declared relative processor artifacts before step construction."""
|
||||
is_local = Path(model_id).is_dir() or Path(model_id).is_file()
|
||||
|
||||
for step_entry in loaded_config["steps"]:
|
||||
artifacts = step_entry.get("artifacts", {})
|
||||
for config_key, relative_path in artifacts.items():
|
||||
artifact_path = Path(relative_path)
|
||||
if artifact_path.is_absolute() or ".." in artifact_path.parts:
|
||||
raise ValueError(
|
||||
f"Processor artifact path must be relative to the checkpoint: {relative_path!r}"
|
||||
)
|
||||
|
||||
resolved_path = base_path / artifact_path if base_path is not None else artifact_path
|
||||
if not resolved_path.exists() and not is_local:
|
||||
repository_path = Path(config_filename).parent / artifact_path
|
||||
snapshot_download(
|
||||
repo_id=model_id,
|
||||
repo_type="model",
|
||||
allow_patterns=f"{repository_path.as_posix()}/**",
|
||||
**hub_download_kwargs,
|
||||
)
|
||||
|
||||
if not resolved_path.exists():
|
||||
step_name = step_entry.get("registry_name", step_entry.get("class", "unknown"))
|
||||
raise FileNotFoundError(
|
||||
f"Missing processor artifact '{relative_path}' for step '{step_name}' "
|
||||
f"next to '{config_filename}'. Checkpoint artifacts are incomplete."
|
||||
)
|
||||
step_entry["config"][config_key] = str(resolved_path)
|
||||
|
||||
@classmethod
|
||||
def _build_steps_from_config(
|
||||
cls,
|
||||
@@ -1224,9 +1138,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
step_entry: dict[str, Any],
|
||||
model_id: str,
|
||||
base_path: Path | None,
|
||||
config_filename: str,
|
||||
hub_download_kwargs: dict[str, Any],
|
||||
is_local_source: bool = False,
|
||||
) -> None:
|
||||
"""Load state dictionary for a processor step if available.
|
||||
|
||||
@@ -1245,7 +1157,7 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
- **Use case**: Loading from local saved model directory
|
||||
|
||||
2. **Hub download fallback**: Download state file from repository
|
||||
- **When triggered**: Local file not found and the pipeline source is a Hub repo
|
||||
- **When triggered**: Local file not found or base_path is None
|
||||
- **Process**: Use hf_hub_download with same parameters as config
|
||||
- **Example**: Download "normalize_step_0.safetensors" from "user/repo"
|
||||
- **Result**: Downloaded to local cache, path returned
|
||||
@@ -1266,7 +1178,6 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
model_id: The model identifier (used for Hub downloads if needed)
|
||||
base_path: Local directory path for finding state files (None for Hub-only)
|
||||
hub_download_kwargs: Parameters for hf_hub_download (tokens, cache, etc.)
|
||||
is_local_source: Whether model_id resolved to a local directory or config file.
|
||||
|
||||
Note:
|
||||
This method modifies step_instance in-place and returns None.
|
||||
@@ -1280,17 +1191,11 @@ class DataProcessorPipeline[TInput, TOutput](HubMixin):
|
||||
# Try local file first
|
||||
if base_path and (base_path / state_filename).exists():
|
||||
state_path = str(base_path / state_filename)
|
||||
elif is_local_source:
|
||||
state_path = base_path / state_filename if base_path else Path(state_filename)
|
||||
raise FileNotFoundError(
|
||||
f"State file '{state_filename}' was not found for local processor pipeline "
|
||||
f"'{model_id}' at '{state_path}'."
|
||||
)
|
||||
else:
|
||||
# Download from Hub
|
||||
state_path = hf_hub_download(
|
||||
repo_id=model_id,
|
||||
filename=(Path(config_filename).parent / state_filename).as_posix(),
|
||||
filename=state_filename,
|
||||
repo_type="model",
|
||||
**hub_download_kwargs,
|
||||
)
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import asdict, dataclass
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
from lerobot.configs import PipelineFeatureType, PolicyFeature
|
||||
@@ -32,18 +32,17 @@ from .pipeline import ProcessorStep, ProcessorStepRegistry
|
||||
@dataclass
|
||||
@ProcessorStepRegistry.register(name="render_messages_processor")
|
||||
class RenderMessagesStep(ProcessorStep):
|
||||
"""Render language columns into recipe-defined messages and supervision metadata."""
|
||||
"""Processor step that turns raw language columns into rendered chat messages.
|
||||
|
||||
Reads ``language_persistent`` and ``language_events`` from the transition's
|
||||
complementary data, renders them through ``recipe`` at the sample timestamp,
|
||||
and replaces the raw columns with the resulting ``messages`` /
|
||||
``message_streams`` / ``target_message_indices`` keys.
|
||||
"""
|
||||
|
||||
recipe: TrainingRecipe
|
||||
dataset_ctx: Any | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if isinstance(self.recipe, dict):
|
||||
self.recipe = TrainingRecipe.from_dict(self.recipe)
|
||||
|
||||
def get_config(self) -> dict[str, Any]:
|
||||
return {"recipe": asdict(self.recipe)}
|
||||
|
||||
def __call__(self, transition: EnvTransition) -> EnvTransition | None:
|
||||
"""Render messages for a single transition; return ``None`` to drop it."""
|
||||
complementary_data = transition.get(TransitionKey.COMPLEMENTARY_DATA) or {}
|
||||
@@ -51,17 +50,7 @@ class RenderMessagesStep(ProcessorStep):
|
||||
events = complementary_data.get(LANGUAGE_EVENTS) or []
|
||||
|
||||
if not persistent and not events:
|
||||
rendered = _fallback_low_level_render(complementary_data.get("task"))
|
||||
if rendered is None:
|
||||
return transition
|
||||
new_transition = transition.copy()
|
||||
new_complementary_data = dict(new_transition.get(TransitionKey.COMPLEMENTARY_DATA) or {})
|
||||
new_complementary_data.update(rendered)
|
||||
new_transition[TransitionKey.COMPLEMENTARY_DATA] = new_complementary_data
|
||||
return new_transition
|
||||
|
||||
if _is_batched_language(persistent) or _is_batched_language(events):
|
||||
return self._call_batch(transition, complementary_data, persistent, events)
|
||||
return transition
|
||||
|
||||
timestamp = complementary_data.get("timestamp")
|
||||
if timestamp is None:
|
||||
@@ -78,147 +67,18 @@ class RenderMessagesStep(ProcessorStep):
|
||||
dataset_ctx=self.dataset_ctx,
|
||||
)
|
||||
if rendered is None:
|
||||
rendered = _fallback_low_level_render(complementary_data.get("task"))
|
||||
if rendered is None:
|
||||
return None
|
||||
return None
|
||||
|
||||
new_transition = transition.copy()
|
||||
new_complementary_data = dict(new_transition.get(TransitionKey.COMPLEMENTARY_DATA) or {})
|
||||
new_complementary_data = dict(complementary_data)
|
||||
new_complementary_data.pop(LANGUAGE_PERSISTENT, None)
|
||||
new_complementary_data.pop(LANGUAGE_EVENTS, None)
|
||||
new_complementary_data.update(rendered)
|
||||
new_transition[TransitionKey.COMPLEMENTARY_DATA] = new_complementary_data
|
||||
return new_transition
|
||||
|
||||
def _call_batch(
|
||||
self,
|
||||
transition: EnvTransition,
|
||||
complementary_data: dict[str, Any],
|
||||
persistent_batch: list,
|
||||
events_batch: list,
|
||||
) -> EnvTransition | None:
|
||||
timestamp = complementary_data.get("timestamp")
|
||||
if timestamp is None:
|
||||
raise KeyError("RenderMessagesStep requires sample timestamp in complementary data.")
|
||||
|
||||
batch_size = max(len(persistent_batch), len(events_batch))
|
||||
messages: list[list[dict[str, Any]]] = []
|
||||
message_streams: list[list[str | None]] = []
|
||||
target_message_indices: list[list[int]] = []
|
||||
keep_indices: list[int] = []
|
||||
|
||||
for i in range(batch_size):
|
||||
rendered = render_sample(
|
||||
recipe=self.recipe,
|
||||
persistent=persistent_batch[i] if i < len(persistent_batch) else [],
|
||||
events=events_batch[i] if i < len(events_batch) else [],
|
||||
t=_batch_value(timestamp, i),
|
||||
sample_idx=int(_batch_value(complementary_data.get("index", 0), i)),
|
||||
task=_batch_value(complementary_data.get("task"), i),
|
||||
dataset_ctx=self.dataset_ctx,
|
||||
)
|
||||
if rendered is None:
|
||||
rendered = _fallback_low_level_render(_batch_value(complementary_data.get("task"), i))
|
||||
if rendered is None:
|
||||
continue
|
||||
keep_indices.append(i)
|
||||
messages.append(rendered["messages"])
|
||||
message_streams.append(rendered["message_streams"])
|
||||
target_message_indices.append(rendered["target_message_indices"])
|
||||
|
||||
if not messages:
|
||||
return None
|
||||
|
||||
new_transition = (
|
||||
_select_batch_indices(transition, keep_indices)
|
||||
if len(keep_indices) != batch_size
|
||||
else transition.copy()
|
||||
)
|
||||
new_complementary_data = dict(new_transition.get(TransitionKey.COMPLEMENTARY_DATA) or {})
|
||||
new_complementary_data.pop(LANGUAGE_PERSISTENT, None)
|
||||
new_complementary_data.pop(LANGUAGE_EVENTS, None)
|
||||
new_complementary_data["messages"] = messages
|
||||
new_complementary_data["message_streams"] = message_streams
|
||||
new_complementary_data["target_message_indices"] = target_message_indices
|
||||
new_transition[TransitionKey.COMPLEMENTARY_DATA] = new_complementary_data
|
||||
return new_transition
|
||||
|
||||
def transform_features(
|
||||
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
|
||||
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
|
||||
"""Pass features through unchanged; rendering only touches complementary data."""
|
||||
return features
|
||||
|
||||
|
||||
def _scalar(value: Any) -> float | int:
|
||||
"""Unwrap a tensor/array/single-element list into a Python scalar."""
|
||||
if hasattr(value, "item"):
|
||||
return value.item()
|
||||
if isinstance(value, list):
|
||||
if len(value) != 1:
|
||||
raise ValueError(f"Expected a scalar, got list of length {len(value)}: {value!r}")
|
||||
return _scalar(value[0])
|
||||
return value
|
||||
|
||||
|
||||
def _is_batched_language(value: Any) -> bool:
|
||||
return isinstance(value, list) and bool(value) and isinstance(value[0], list)
|
||||
|
||||
|
||||
def _batch_value(value: Any, index: int) -> Any:
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, list):
|
||||
return value[index]
|
||||
if hasattr(value, "ndim") and value.ndim > 0:
|
||||
return _scalar(value[index])
|
||||
return _scalar(value)
|
||||
|
||||
|
||||
def _select_batch_indices(transition: EnvTransition, indices: list[int]) -> EnvTransition:
|
||||
selected = transition.copy()
|
||||
for key in (TransitionKey.OBSERVATION, TransitionKey.COMPLEMENTARY_DATA):
|
||||
data = selected.get(key)
|
||||
if isinstance(data, dict):
|
||||
selected[key] = {k: _select_value(v, indices) for k, v in data.items()}
|
||||
action = selected.get(TransitionKey.ACTION)
|
||||
if action is not None:
|
||||
selected[TransitionKey.ACTION] = _select_value(action, indices)
|
||||
return selected
|
||||
|
||||
|
||||
def _select_value(value: Any, indices: list[int]) -> Any:
|
||||
if isinstance(value, list) and len(value) >= len(indices):
|
||||
return [value[i] for i in indices]
|
||||
if hasattr(value, "index_select") and hasattr(value, "new_tensor") and getattr(value, "ndim", 0) > 0:
|
||||
return value.index_select(0, value.new_tensor(indices).long())
|
||||
return value
|
||||
|
||||
|
||||
def _fallback_low_level_render(task: Any) -> dict[str, Any] | None:
|
||||
"""Keep action-only samples trainable when no recipe branch matches."""
|
||||
if hasattr(task, "item"):
|
||||
task = task.item()
|
||||
if isinstance(task, list):
|
||||
messages = []
|
||||
message_streams = []
|
||||
target_message_indices = []
|
||||
for t in task:
|
||||
rendered = _fallback_low_level_render(t)
|
||||
if rendered is None:
|
||||
return None
|
||||
messages.append(rendered["messages"])
|
||||
message_streams.append(rendered["message_streams"])
|
||||
target_message_indices.append(rendered["target_message_indices"])
|
||||
return {
|
||||
"messages": messages,
|
||||
"message_streams": message_streams,
|
||||
"target_message_indices": target_message_indices,
|
||||
}
|
||||
if not isinstance(task, str) or not task:
|
||||
return None
|
||||
return {
|
||||
"messages": [{"role": "user", "content": task}],
|
||||
"message_streams": ["low_level"],
|
||||
"target_message_indices": [],
|
||||
}
|
||||
|
||||
@@ -25,7 +25,6 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import torch
|
||||
@@ -33,7 +32,6 @@ import torch
|
||||
from lerobot.configs import FeatureType, PipelineFeatureType, PolicyFeature
|
||||
from lerobot.types import EnvTransition, RobotObservation, TransitionKey
|
||||
from lerobot.utils.constants import (
|
||||
ACTION_CODE_TOKEN_MASK,
|
||||
ACTION_TOKEN_MASK,
|
||||
ACTION_TOKENS,
|
||||
OBS_LANGUAGE_ATTENTION_MASK,
|
||||
@@ -138,7 +136,7 @@ class TokenizerProcessorStep(ObservationProcessorStep):
|
||||
# Standardize to a list of strings for the tokenizer
|
||||
if isinstance(task, str):
|
||||
return [task]
|
||||
elif isinstance(task, list | tuple) and all(isinstance(t, str) for t in task):
|
||||
elif isinstance(task, (list, tuple)) and all(isinstance(t, str) for t in task):
|
||||
return list(task)
|
||||
|
||||
return None
|
||||
@@ -351,7 +349,6 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
|
||||
max_action_tokens: int = 256
|
||||
fast_skip_tokens: int = 128
|
||||
paligemma_tokenizer_name: str = "google/paligemma-3b-pt-224"
|
||||
allow_truncation: bool = True
|
||||
# Internal tokenizer instance (not part of the config)
|
||||
action_tokenizer: Any = field(default=None, init=False, repr=False)
|
||||
_paligemma_tokenizer: Any = field(default=None, init=False, repr=False)
|
||||
@@ -415,15 +412,14 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
|
||||
# During inference, no action is available, skip tokenization
|
||||
return new_transition
|
||||
|
||||
# Tokenize and get masks for the full formatted sequence and the discrete action codes.
|
||||
tokens, mask, code_mask = self._tokenize_action(action)
|
||||
# Tokenize and get both tokens and mask
|
||||
tokens, mask = self._tokenize_action(action)
|
||||
|
||||
# Store mask in complementary data
|
||||
complementary_data = new_transition.get(TransitionKey.COMPLEMENTARY_DATA, {})
|
||||
if complementary_data is None:
|
||||
complementary_data = {}
|
||||
complementary_data[ACTION_TOKEN_MASK] = mask
|
||||
complementary_data[ACTION_CODE_TOKEN_MASK] = code_mask
|
||||
complementary_data[ACTION_TOKENS] = tokens
|
||||
new_transition[TransitionKey.COMPLEMENTARY_DATA] = complementary_data
|
||||
return new_transition
|
||||
@@ -434,7 +430,7 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
|
||||
"""
|
||||
return self._paligemma_tokenizer.vocab_size - 1 - self.fast_skip_tokens - tokens
|
||||
|
||||
def _tokenize_action(self, action: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
|
||||
def _tokenize_action(self, action: torch.Tensor) -> tuple[torch.Tensor, torch.Tensor]:
|
||||
"""
|
||||
Tokenizes the action tensor and creates a mask.
|
||||
|
||||
@@ -463,7 +459,6 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
|
||||
# The fast tokenizer expects action data and returns token IDs
|
||||
tokens_list = []
|
||||
masks_list = []
|
||||
code_masks_list = []
|
||||
|
||||
for i in range(batch_size):
|
||||
# Tokenize single action (move to CPU first as tokenizer uses scipy which requires numpy)
|
||||
@@ -481,82 +476,65 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
|
||||
if tokens.dim() > 1:
|
||||
tokens = tokens.flatten()
|
||||
|
||||
action_code_tokens = self._act_tokens_to_paligemma_tokens(tokens)
|
||||
bos_id = self._paligemma_tokenizer.bos_token_id
|
||||
prompt_tokens = torch.tensor(
|
||||
self._paligemma_tokenizer.encode("Action: ", add_special_tokens=False),
|
||||
device=action.device,
|
||||
)
|
||||
end_tokens = torch.tensor(self._paligemma_tokenizer.encode("|"), device=action.device)
|
||||
|
||||
code_start = 1 + len(prompt_tokens)
|
||||
code_end = code_start + len(action_code_tokens)
|
||||
# add bos
|
||||
tokens = torch.cat(
|
||||
[
|
||||
torch.tensor([bos_id], device=action.device),
|
||||
prompt_tokens,
|
||||
action_code_tokens,
|
||||
end_tokens,
|
||||
torch.tensor(
|
||||
self._paligemma_tokenizer.encode("Action: ", add_special_tokens=False),
|
||||
device=action.device,
|
||||
),
|
||||
self._act_tokens_to_paligemma_tokens(tokens),
|
||||
torch.tensor(self._paligemma_tokenizer.encode("|"), device=action.device),
|
||||
]
|
||||
)
|
||||
code_mask = torch.zeros(len(tokens), dtype=torch.bool, device=action.device)
|
||||
code_mask[code_start:code_end] = True
|
||||
|
||||
# Truncate or pad to max_action_tokens
|
||||
if len(tokens) > self.max_action_tokens:
|
||||
if not self.allow_truncation:
|
||||
raise ValueError(
|
||||
f"FAST action sequence has {len(tokens)} tokens, exceeding "
|
||||
f"max_action_tokens={self.max_action_tokens}."
|
||||
)
|
||||
logging.warning(
|
||||
f"Token length ({len(tokens)}) exceeds max length ({self.max_action_tokens}), truncating. "
|
||||
"Consider increasing the `max_action_tokens` in your model config if this happens frequently."
|
||||
)
|
||||
tokens = tokens[: self.max_action_tokens]
|
||||
code_mask = code_mask[: self.max_action_tokens]
|
||||
mask = torch.ones(self.max_action_tokens, dtype=torch.bool, device=action.device)
|
||||
else:
|
||||
pad_len = self.max_action_tokens - len(tokens)
|
||||
mask = torch.cat(
|
||||
[
|
||||
torch.ones(len(tokens), dtype=torch.bool, device=action.device),
|
||||
torch.zeros(pad_len, dtype=torch.bool, device=action.device),
|
||||
torch.zeros(
|
||||
self.max_action_tokens - len(tokens), dtype=torch.bool, device=action.device
|
||||
),
|
||||
]
|
||||
)
|
||||
code_mask = torch.nn.functional.pad(code_mask, (0, pad_len), value=False)
|
||||
# Pad tokens with zeros
|
||||
tokens = torch.nn.functional.pad(tokens, (0, pad_len), value=0)
|
||||
tokens = torch.nn.functional.pad(tokens, (0, self.max_action_tokens - len(tokens)), value=0)
|
||||
|
||||
tokens_list.append(tokens)
|
||||
masks_list.append(mask)
|
||||
code_masks_list.append(code_mask)
|
||||
|
||||
# Stack into batched tensors
|
||||
tokens_batch = torch.stack(tokens_list, dim=0) # (B, max_action_tokens)
|
||||
masks_batch = torch.stack(masks_list, dim=0) # (B, max_action_tokens)
|
||||
code_masks_batch = torch.stack(code_masks_list, dim=0) # (B, max_action_tokens)
|
||||
|
||||
# Remove batch dimension if input was single sample
|
||||
if single_sample:
|
||||
tokens_batch = tokens_batch.squeeze(0)
|
||||
masks_batch = masks_batch.squeeze(0)
|
||||
code_masks_batch = code_masks_batch.squeeze(0)
|
||||
|
||||
# Move to the same device as the input
|
||||
if device is not None:
|
||||
tokens_batch = tokens_batch.to(device)
|
||||
masks_batch = masks_batch.to(device)
|
||||
code_masks_batch = code_masks_batch.to(device)
|
||||
|
||||
return tokens_batch, masks_batch, code_masks_batch
|
||||
return tokens_batch, masks_batch
|
||||
|
||||
def action(self, action: torch.Tensor) -> torch.Tensor:
|
||||
"""
|
||||
This method is not used since we override __call__.
|
||||
Required by ActionProcessorStep ABC.
|
||||
"""
|
||||
tokens, _, _ = self._tokenize_action(action)
|
||||
tokens, _ = self._tokenize_action(action)
|
||||
return tokens
|
||||
|
||||
def get_config(self) -> dict[str, Any]:
|
||||
@@ -572,9 +550,6 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
|
||||
config = {
|
||||
"trust_remote_code": self.trust_remote_code,
|
||||
"max_action_tokens": self.max_action_tokens,
|
||||
"fast_skip_tokens": self.fast_skip_tokens,
|
||||
"paligemma_tokenizer_name": self.paligemma_tokenizer_name,
|
||||
"allow_truncation": self.allow_truncation,
|
||||
}
|
||||
|
||||
# Only save tokenizer_name if it was used to create the tokenizer
|
||||
@@ -583,14 +558,6 @@ class ActionTokenizerProcessorStep(ActionProcessorStep):
|
||||
|
||||
return config
|
||||
|
||||
def save_artifacts(self, save_directory: Path) -> dict[str, str]:
|
||||
artifact_path = Path("action_tokenizer")
|
||||
save_pretrained = getattr(self.action_tokenizer, "save_pretrained", None)
|
||||
if save_pretrained is None:
|
||||
raise TypeError("Action tokenizer must implement save_pretrained() to save a portable pipeline.")
|
||||
save_pretrained(save_directory / artifact_path)
|
||||
return {"action_tokenizer_name": artifact_path.as_posix()}
|
||||
|
||||
def transform_features(
|
||||
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
|
||||
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
|
||||
|
||||
@@ -323,10 +323,6 @@ class LeKiwiClient(Robot):
|
||||
np.ndarray: the action sent to the motors, potentially clipped.
|
||||
"""
|
||||
|
||||
# Action values may be torch tensors (e.g. replayed from a dataset) or numpy
|
||||
# scalars; json.dumps only serializes Python primitives, so coerce each value to a
|
||||
# plain float before sending.
|
||||
action = {key: float(value) for key, value in action.items()}
|
||||
self.zmq_cmd_socket.send_string(json.dumps(action)) # action is in motor space
|
||||
|
||||
# TODO(Steven): Remove the np conversion when it is possible to record a non-numpy array value
|
||||
|
||||
@@ -21,8 +21,6 @@ from lerobot.utils.import_utils import make_device_from_device_class
|
||||
from .config import RobotConfig
|
||||
from .robot import Robot
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def make_robot_from_config(config: RobotConfig) -> Robot:
|
||||
# TODO(Steven): Consider just using the make_device_from_device_class for all types
|
||||
@@ -120,7 +118,7 @@ def ensure_safe_goal_position(
|
||||
}
|
||||
|
||||
if warnings_dict:
|
||||
logger.warning(
|
||||
logging.warning(
|
||||
"Relative goal position magnitude had to be clamped to be safe.\n"
|
||||
f"{pformat(warnings_dict, indent=4)}"
|
||||
)
|
||||
|
||||
@@ -326,17 +326,8 @@ class RolloutConfig:
|
||||
|
||||
policy_path = parser.get_path_arg("policy")
|
||||
if policy_path:
|
||||
yaml_overrides = parser.get_yaml_overrides("policy")
|
||||
cli_overrides = parser.get_cli_overrides("policy") or []
|
||||
policy_overrides = yaml_overrides + cli_overrides
|
||||
pretrained_revision = parser.parse_arg("pretrained_revision", cli_overrides)
|
||||
if pretrained_revision is None:
|
||||
pretrained_revision = parser.parse_arg("pretrained_revision", yaml_overrides)
|
||||
self.policy = PreTrainedConfig.from_pretrained(
|
||||
policy_path,
|
||||
revision=pretrained_revision,
|
||||
cli_overrides=policy_overrides,
|
||||
)
|
||||
cli_overrides = parser.get_cli_overrides("policy")
|
||||
self.policy = PreTrainedConfig.from_pretrained(policy_path, cli_overrides=cli_overrides)
|
||||
self.policy.pretrained_path = policy_path
|
||||
if self.policy is None:
|
||||
raise ValueError("--policy.path is required for rollout")
|
||||
|
||||
@@ -27,7 +27,7 @@ from threading import Event
|
||||
|
||||
import torch
|
||||
|
||||
from lerobot.configs import FeatureType, PreTrainedConfig
|
||||
from lerobot.configs import FeatureType
|
||||
from lerobot.datasets import (
|
||||
LeRobotDataset,
|
||||
aggregate_pipeline_dataset_features,
|
||||
@@ -159,35 +159,6 @@ class RolloutContext:
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _load_pretrained_policy(policy_config: PreTrainedConfig) -> PreTrainedPolicy:
|
||||
"""Load policy weights, keeping adapter and base-model revisions independent."""
|
||||
pretrained_revision = policy_config.pretrained_revision
|
||||
policy_class = get_policy_class(policy_config.type)
|
||||
|
||||
if not policy_config.use_peft:
|
||||
return policy_class.from_pretrained(
|
||||
policy_config.pretrained_path,
|
||||
config=policy_config,
|
||||
revision=pretrained_revision,
|
||||
)
|
||||
|
||||
from peft import PeftConfig, PeftModel
|
||||
|
||||
peft_path = policy_config.pretrained_path
|
||||
peft_config = PeftConfig.from_pretrained(peft_path, revision=pretrained_revision)
|
||||
policy = policy_class.from_pretrained(
|
||||
pretrained_name_or_path=peft_config.base_model_name_or_path,
|
||||
config=policy_config,
|
||||
revision=peft_config.revision,
|
||||
)
|
||||
return PeftModel.from_pretrained(
|
||||
policy,
|
||||
peft_path,
|
||||
config=peft_config,
|
||||
revision=pretrained_revision,
|
||||
)
|
||||
|
||||
|
||||
def build_rollout_context(
|
||||
cfg: RolloutConfig,
|
||||
shutdown_event: Event,
|
||||
@@ -205,6 +176,7 @@ def build_rollout_context(
|
||||
# --- 1. Policy (heavy I/O, but no hardware yet) -------------------
|
||||
logger.info("Loading policy from '%s'...", cfg.policy.pretrained_path)
|
||||
policy_config = cfg.policy
|
||||
policy_class = get_policy_class(policy_config.type)
|
||||
|
||||
if hasattr(policy_config, "compile_model"):
|
||||
policy_config.compile_model = cfg.use_torch_compile
|
||||
@@ -215,7 +187,17 @@ def build_rollout_context(
|
||||
"Please use `cpu` or `cuda` backend."
|
||||
)
|
||||
|
||||
policy = _load_pretrained_policy(policy_config)
|
||||
if policy_config.use_peft:
|
||||
from peft import PeftConfig, PeftModel
|
||||
|
||||
peft_path = policy_config.pretrained_path
|
||||
peft_config = PeftConfig.from_pretrained(peft_path)
|
||||
policy = policy_class.from_pretrained(
|
||||
pretrained_name_or_path=peft_config.base_model_name_or_path, config=policy_config
|
||||
)
|
||||
policy = PeftModel.from_pretrained(policy, peft_path, config=peft_config)
|
||||
else:
|
||||
policy = policy_class.from_pretrained(policy_config.pretrained_path, config=policy_config)
|
||||
|
||||
if is_rtc:
|
||||
policy.config.rtc_config = cfg.inference.rtc
|
||||
@@ -410,7 +392,6 @@ def build_rollout_context(
|
||||
preprocessor, postprocessor = make_pre_post_processors(
|
||||
policy_cfg=policy_config,
|
||||
pretrained_path=cfg.policy.pretrained_path,
|
||||
pretrained_revision=policy_config.pretrained_revision,
|
||||
dataset_stats=dataset_stats,
|
||||
preprocessor_overrides={
|
||||
"device_processor": {"device": cfg.device},
|
||||
|
||||
@@ -1,38 +0,0 @@
|
||||
# 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.
|
||||
# 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.
|
||||
|
||||
"""Policy-agnostic runtime for language-conditioned policies.
|
||||
|
||||
Adapters registered in :mod:`lerobot.runtime.registry` are served by ``lerobot-rollout --language``.
|
||||
"""
|
||||
|
||||
from .adapter import BaseLanguageAdapter, GenerationConfig, LanguageDiagnostics
|
||||
from .language_runtime import (
|
||||
LanguageConditionedPolicyAdapter,
|
||||
LanguageConditionedRuntime,
|
||||
RuntimeState,
|
||||
Tick,
|
||||
TickClock,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"BaseLanguageAdapter",
|
||||
"GenerationConfig",
|
||||
"LanguageConditionedPolicyAdapter",
|
||||
"LanguageConditionedRuntime",
|
||||
"LanguageDiagnostics",
|
||||
"RuntimeState",
|
||||
"Tick",
|
||||
"TickClock",
|
||||
]
|
||||
@@ -1,165 +0,0 @@
|
||||
# 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.
|
||||
# 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.
|
||||
|
||||
"""Policy adapters for the language runtime.
|
||||
|
||||
The base adapter owns generation control and diagnostics while subclasses provide policy-specific actions and text.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from abc import ABC, abstractmethod
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from .language_runtime import RuntimeState
|
||||
|
||||
_SAY_RE = re.compile(r"<\s*say\s*>(.*?)<\s*/\s*say\s*>", re.IGNORECASE | re.DOTALL)
|
||||
|
||||
|
||||
@dataclass
|
||||
class GenerationConfig:
|
||||
"""Text-generation settings fixed for the adapter's lifetime."""
|
||||
|
||||
min_new_tokens: int = 0
|
||||
temperature: float = 0.0
|
||||
top_p: float = 1.0
|
||||
chunks_per_regen: int = 1 # regenerate the language context every N action chunks
|
||||
enable_memory: bool = True # generate a running memory note on subtask change
|
||||
enable_subtask: bool = True # generate the low-level subtask (off => use the given text directly)
|
||||
|
||||
|
||||
@dataclass
|
||||
class LanguageDiagnostics:
|
||||
"""Runtime-panel generation counters keyed by text kind."""
|
||||
|
||||
last_raw: dict[str, str] = field(default_factory=dict)
|
||||
empty: dict[str, int] = field(default_factory=dict)
|
||||
repeat: int = 0
|
||||
|
||||
def _bump(self, table: dict[str, int], kind: str) -> int:
|
||||
table[kind] = table.get(kind, 0) + 1
|
||||
return table[kind]
|
||||
|
||||
|
||||
class BaseLanguageAdapter(ABC):
|
||||
"""Batteries-included adapter: generic high-level control, policy primitives abstract."""
|
||||
|
||||
def __init__(self, policy: Any, gen: GenerationConfig | None = None) -> None:
|
||||
self.policy = policy
|
||||
self.gen = gen or GenerationConfig()
|
||||
self.diag = LanguageDiagnostics()
|
||||
self._chunks_until_regen = 0
|
||||
|
||||
@abstractmethod
|
||||
def select_action(self, observation: dict[str, Any], state: RuntimeState) -> Any:
|
||||
"""Produce an action chunk from the observation + current language context."""
|
||||
|
||||
@abstractmethod
|
||||
def generate_text(
|
||||
self,
|
||||
kind: str,
|
||||
observation: dict[str, Any] | None,
|
||||
state: RuntimeState,
|
||||
user_text: str | None = None,
|
||||
) -> str:
|
||||
"""Generate one text stream (``kind``) and return the decoded string."""
|
||||
|
||||
def update_language_state(self, observation: dict[str, Any] | None, state: RuntimeState) -> None:
|
||||
"""Throttled regeneration of the language context (subtask / memory / ...)."""
|
||||
if self._chunks_until_regen > 0:
|
||||
self._chunks_until_regen -= 1
|
||||
return
|
||||
self._chunks_until_regen = max(1, self.gen.chunks_per_regen) - 1
|
||||
self._regenerate_context(observation, state)
|
||||
|
||||
def handle_interjection(
|
||||
self, user_text: str, observation: dict[str, Any] | None, state: RuntimeState
|
||||
) -> None:
|
||||
"""React to a mid-run user message by regenerating the plan."""
|
||||
out = self.generate_text("interjection", observation, state, user_text=user_text)
|
||||
plan = self.plan_from_text(out)
|
||||
if plan:
|
||||
state.set_context("plan", plan, label="plan")
|
||||
|
||||
def plan_from_text(self, text: str) -> str:
|
||||
"""Strip ``<say>`` speech markers from a generated plan."""
|
||||
plan, _speech = split_plan_and_say(text)
|
||||
return plan
|
||||
|
||||
def _regenerate_context(self, observation: dict[str, Any] | None, state: RuntimeState) -> None:
|
||||
"""Default hierarchy: regenerate the subtask, then memory when it changes.
|
||||
|
||||
Override for a policy with a different language hierarchy.
|
||||
"""
|
||||
if not self.gen.enable_subtask:
|
||||
# Preserve operator-provided subtasks in direct mode.
|
||||
return
|
||||
subtask = self._generate_filtered("subtask", observation, state)
|
||||
if subtask is None:
|
||||
return
|
||||
previous = state.language_context.get("subtask")
|
||||
if not state.set_context("subtask", subtask, label="subtask"):
|
||||
self.diag.repeat += 1
|
||||
return
|
||||
self.diag.repeat = 0
|
||||
if previous:
|
||||
state.extra["prior_subtask"] = previous
|
||||
if not self.gen.enable_memory:
|
||||
return
|
||||
memory = self._generate_filtered("memory", observation, state)
|
||||
if memory is not None:
|
||||
state.set_context("memory", memory, label="memory")
|
||||
|
||||
def _generate_filtered(
|
||||
self, kind: str, observation: dict[str, Any] | None, state: RuntimeState
|
||||
) -> str | None:
|
||||
"""Generate one ``kind``, record diagnostics, and drop empty output."""
|
||||
text = self.generate_text(kind, observation, state)
|
||||
self.diag.last_raw[kind] = text or ""
|
||||
if not text:
|
||||
count = self.diag._bump(self.diag.empty, kind)
|
||||
if count == 1 or count % 5 == 0:
|
||||
state.log(f" [info] {kind} gen returned empty (x{count})")
|
||||
return None
|
||||
return text
|
||||
|
||||
|
||||
class DirectTaskPolicyAdapter(BaseLanguageAdapter):
|
||||
"""Adapter for flat policies whose preprocessors condition actions on the operator's task."""
|
||||
|
||||
def select_action(self, observation: dict[str, Any], state: RuntimeState) -> Any:
|
||||
return self.policy.predict_action_chunk(observation)
|
||||
|
||||
def generate_text(
|
||||
self,
|
||||
kind: str,
|
||||
observation: dict[str, Any] | None,
|
||||
state: RuntimeState,
|
||||
user_text: str | None = None,
|
||||
) -> str:
|
||||
return ""
|
||||
|
||||
|
||||
def split_plan_and_say(text: str) -> tuple[str, str]:
|
||||
"""Split ``plan <say>speech</say>`` into ``(plan, speech)``."""
|
||||
if not text:
|
||||
return "", ""
|
||||
match = _SAY_RE.search(text)
|
||||
if not match:
|
||||
return text.strip(), ""
|
||||
speech = match.group(1).strip().strip('"').strip("'")
|
||||
plan = (text[: match.start()] + text[match.end() :]).strip()
|
||||
return plan, speech
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,349 +0,0 @@
|
||||
# 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.
|
||||
# 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.
|
||||
|
||||
"""Small reusable runtime for language-conditioned robot policies."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from collections import deque
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Protocol
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class RuntimeState:
|
||||
"""Explicit state shared by the runtime and policy adapter."""
|
||||
|
||||
task: str = ""
|
||||
language_context: dict[str, str] = field(default_factory=dict)
|
||||
action_queue: deque[Any] = field(default_factory=deque)
|
||||
events: set[str] = field(default_factory=set)
|
||||
log_lines: list[str] = field(default_factory=list)
|
||||
mode: str = "action"
|
||||
stop: bool = False
|
||||
tick: Tick | None = None
|
||||
actions_dispatched: int = 0
|
||||
action_deadline: float | None = None
|
||||
extra: dict[str, Any] = field(default_factory=dict)
|
||||
revision: int = 0
|
||||
lock: Any = field(default_factory=threading.RLock, repr=False)
|
||||
|
||||
def emit(self, event_name: str) -> None:
|
||||
self.events.add(event_name)
|
||||
|
||||
def take_event(self, event_name: str) -> bool:
|
||||
if event_name not in self.events:
|
||||
return False
|
||||
self.events.remove(event_name)
|
||||
return True
|
||||
|
||||
def log(self, line: str) -> None:
|
||||
self.log_lines.append(line)
|
||||
|
||||
def set_context(self, key: str, value: str | None, *, label: str | None = None) -> bool:
|
||||
with self.lock:
|
||||
previous = self.language_context.get(key)
|
||||
if previous == value:
|
||||
return False
|
||||
if value is None:
|
||||
self.language_context.pop(key, None)
|
||||
else:
|
||||
self.language_context[key] = value
|
||||
self.revision += 1
|
||||
if label is not None and value:
|
||||
self.log(f" {label}: {value}")
|
||||
return True
|
||||
|
||||
def get(self, key: str, default: Any = None) -> Any:
|
||||
try:
|
||||
return self[key]
|
||||
except KeyError:
|
||||
return default
|
||||
|
||||
def setdefault(self, key: str, default: Any = None) -> Any:
|
||||
current = self.get(key, None)
|
||||
if current is not None:
|
||||
return current
|
||||
self[key] = default
|
||||
return default
|
||||
|
||||
def __getitem__(self, key: str) -> Any:
|
||||
if hasattr(self, key):
|
||||
return getattr(self, key)
|
||||
if key in self.extra:
|
||||
return self.extra[key]
|
||||
raise KeyError(key)
|
||||
|
||||
def __setitem__(self, key: str, value: Any) -> None:
|
||||
with self.lock:
|
||||
if hasattr(self, key):
|
||||
if key == "mode" and self.mode != value:
|
||||
self.revision += 1
|
||||
setattr(self, key, value)
|
||||
else:
|
||||
self.extra[key] = value
|
||||
|
||||
|
||||
class LanguageConditionedPolicyAdapter(Protocol):
|
||||
"""Runtime policy contract, implemented directly or through ``BaseLanguageAdapter``."""
|
||||
|
||||
def select_action(self, observation: dict[str, Any], state: RuntimeState) -> Any: ...
|
||||
|
||||
def update_language_state(self, observation: dict[str, Any] | None, state: RuntimeState) -> None: ...
|
||||
|
||||
def handle_interjection(
|
||||
self, user_text: str, observation: dict[str, Any] | None, state: RuntimeState
|
||||
) -> None: ...
|
||||
|
||||
|
||||
@dataclass
|
||||
class Tick:
|
||||
index: int
|
||||
monotonic_seconds: float
|
||||
|
||||
|
||||
@dataclass
|
||||
class TickClock:
|
||||
max_rate_hz: float = 50.0
|
||||
_index: int = field(default=0, init=False)
|
||||
_last_seconds: float | None = field(default=None, init=False)
|
||||
|
||||
def advance(self) -> Tick:
|
||||
period = 1.0 / max(self.max_rate_hz, 0.1)
|
||||
now = time.monotonic()
|
||||
if self._last_seconds is not None:
|
||||
sleep_for = (self._last_seconds + period) - now
|
||||
if sleep_for > 0:
|
||||
time.sleep(sleep_for)
|
||||
now = time.monotonic()
|
||||
self._last_seconds = now
|
||||
self._index += 1
|
||||
return Tick(index=self._index, monotonic_seconds=now)
|
||||
|
||||
|
||||
@dataclass
|
||||
class _RateGate:
|
||||
hz: float
|
||||
_last_seconds: float | None = None
|
||||
|
||||
def due(self, tick: Tick, *, force: bool = False) -> bool:
|
||||
if force:
|
||||
self._last_seconds = tick.monotonic_seconds
|
||||
return True
|
||||
period = 1.0 / max(self.hz, 1e-6)
|
||||
if self._last_seconds is None or tick.monotonic_seconds - self._last_seconds >= period:
|
||||
self._last_seconds = tick.monotonic_seconds
|
||||
return True
|
||||
return False
|
||||
|
||||
def rearm(self) -> None:
|
||||
self._last_seconds = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class LanguageConditionedRuntime:
|
||||
"""Generic tick loop for language-conditioned robot policies."""
|
||||
|
||||
policy_adapter: LanguageConditionedPolicyAdapter
|
||||
observation_provider: Callable[[], dict[str, Any] | None] | None = None
|
||||
action_executor: Callable[[Any], None] | None = None
|
||||
event_collector: Callable[[RuntimeState], None] | None = None
|
||||
chunk_hz: float = 4.0
|
||||
ctrl_hz: float = 50.0
|
||||
high_level_hz: float = 1.0
|
||||
max_rate_hz: float = 50.0
|
||||
|
||||
state: RuntimeState = field(default_factory=RuntimeState)
|
||||
_chunk_gate: _RateGate = field(init=False)
|
||||
_ctrl_gate: _RateGate = field(init=False)
|
||||
_language_gate: _RateGate = field(init=False)
|
||||
_stop: bool = field(default=False, init=False)
|
||||
_last_dispatch_seconds: float | None = field(default=None, init=False)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
self._chunk_gate = _RateGate(self.chunk_hz)
|
||||
self._ctrl_gate = _RateGate(self.ctrl_hz)
|
||||
self._language_gate = _RateGate(self.high_level_hz)
|
||||
|
||||
@property
|
||||
def policy(self) -> Any:
|
||||
return getattr(self.policy_adapter, "policy", self.policy_adapter)
|
||||
|
||||
def set_task(self, task: str) -> None:
|
||||
with self.state.lock:
|
||||
if self.state.task != task:
|
||||
self.state.revision += 1
|
||||
self.state.task = task
|
||||
self.state.log(f"Task: {task}")
|
||||
|
||||
def stop(self) -> None:
|
||||
self._stop = True
|
||||
self.state.stop = True
|
||||
|
||||
def run(self, *, max_ticks: int | None = None) -> None:
|
||||
clock = TickClock(max_rate_hz=self.max_rate_hz)
|
||||
while not self._stop:
|
||||
tick = clock.advance()
|
||||
self._run_tick(tick)
|
||||
self._flush_logs()
|
||||
if self.state.stop:
|
||||
self._stop = True
|
||||
if max_ticks is not None and tick.index >= max_ticks:
|
||||
break
|
||||
self._on_shutdown()
|
||||
|
||||
def step_once(self) -> list[str]:
|
||||
previous = self.state.tick.index if self.state.tick is not None else 0
|
||||
tick = Tick(index=previous + 1, monotonic_seconds=time.monotonic())
|
||||
self._run_tick(tick, force_rates=True)
|
||||
return list(self.state.log_lines)
|
||||
|
||||
def _run_tick(self, tick: Tick, *, force_rates: bool = False) -> None:
|
||||
self.state.tick = tick
|
||||
self.state.log_lines = []
|
||||
if self.event_collector is not None:
|
||||
self.event_collector(self.state)
|
||||
self._handle_action_deadline()
|
||||
if self.state.stop:
|
||||
return
|
||||
self.maybe_update_language_state(force=force_rates)
|
||||
self.maybe_handle_user_events()
|
||||
self.maybe_enqueue_action_chunk(force=force_rates)
|
||||
self.dispatch_action(force=force_rates)
|
||||
self.state.events.clear()
|
||||
|
||||
def _current_observation(self) -> dict[str, Any] | None:
|
||||
if self.observation_provider is None:
|
||||
return None
|
||||
try:
|
||||
return self.observation_provider()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.debug("observation_provider failed: %s", exc)
|
||||
return None
|
||||
|
||||
def maybe_update_language_state(self, *, force: bool = False) -> None:
|
||||
if self.state.mode != "action" or not self.state.task:
|
||||
return
|
||||
if self.state.action_queue:
|
||||
self._language_gate.rearm()
|
||||
return
|
||||
if self.state.tick is None or not self._language_gate.due(self.state.tick, force=force):
|
||||
return
|
||||
observation = self._current_observation()
|
||||
try:
|
||||
self.policy_adapter.update_language_state(observation, self.state)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("language update failed: %s", exc, exc_info=logger.isEnabledFor(logging.DEBUG))
|
||||
self.state.log(f" [warn] language update failed: {type(exc).__name__}: {exc}")
|
||||
|
||||
def maybe_handle_user_events(self) -> None:
|
||||
if self.state.take_event("user_interjection"):
|
||||
self._handle_user_interjection()
|
||||
|
||||
def _handle_user_interjection(self) -> None:
|
||||
text = str(self.state.extra.get("recent_interjection") or "")
|
||||
if not text:
|
||||
return
|
||||
observation = self._current_observation()
|
||||
self.policy_adapter.handle_interjection(text, observation, self.state)
|
||||
self.state.extra["recent_interjection"] = None
|
||||
|
||||
def maybe_enqueue_action_chunk(self, *, force: bool = False) -> None:
|
||||
with self.state.lock:
|
||||
if self.state.mode != "action" or not self.state.task:
|
||||
return
|
||||
if self.state.action_queue:
|
||||
return
|
||||
if self.state.tick is None or not self._chunk_gate.due(self.state.tick, force=force):
|
||||
return
|
||||
revision = self.state.revision
|
||||
observation = self._current_observation()
|
||||
if observation is None:
|
||||
return
|
||||
try:
|
||||
chunk = self.policy_adapter.select_action(observation, self.state)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("select_action failed: %s", exc, exc_info=logger.isEnabledFor(logging.DEBUG))
|
||||
self.state.log(f" [warn] select_action failed: {type(exc).__name__}: {exc}")
|
||||
return
|
||||
with self.state.lock:
|
||||
if (
|
||||
self.state.revision != revision
|
||||
or self.state.mode != "action"
|
||||
or self.state.stop
|
||||
or self._stop
|
||||
):
|
||||
logger.info("Discarded an action chunk invalidated during inference.")
|
||||
return
|
||||
self._enqueue_chunk(chunk)
|
||||
|
||||
def _enqueue_chunk(self, chunk: Any) -> None:
|
||||
if chunk is None:
|
||||
return
|
||||
chunk_iter = chunk[0] if getattr(chunk, "ndim", None) == 3 else chunk
|
||||
if getattr(chunk_iter, "ndim", None) == 1:
|
||||
chunk_iter = chunk_iter.unsqueeze(0)
|
||||
for step in chunk_iter:
|
||||
self.state.action_queue.append(step.unsqueeze(0) if hasattr(step, "unsqueeze") else step)
|
||||
try:
|
||||
self.state.extra["last_chunk_size"] = int(chunk_iter.shape[0])
|
||||
except Exception: # noqa: BLE001
|
||||
self.state.extra["last_chunk_size"] = len(self.state.action_queue)
|
||||
|
||||
def dispatch_action(self, *, force: bool = False) -> None:
|
||||
if self.state.mode != "action":
|
||||
self._last_dispatch_seconds = None
|
||||
return
|
||||
if self.state.tick is None or not self._ctrl_gate.due(self.state.tick, force=force):
|
||||
return
|
||||
queue = self.state.action_queue
|
||||
if not queue:
|
||||
self._last_dispatch_seconds = None
|
||||
return
|
||||
now = time.monotonic()
|
||||
if self._last_dispatch_seconds is None or self.ctrl_hz <= 0:
|
||||
n_to_pop = 1
|
||||
else:
|
||||
n_to_pop = max(1, min(len(queue), int(round((now - self._last_dispatch_seconds) * self.ctrl_hz))))
|
||||
self._last_dispatch_seconds = now
|
||||
latest = None
|
||||
for _ in range(n_to_pop):
|
||||
if not queue:
|
||||
break
|
||||
latest = queue.popleft()
|
||||
self.state.actions_dispatched += 1
|
||||
if latest is not None and self.action_executor is not None:
|
||||
self.action_executor(latest)
|
||||
|
||||
def _handle_action_deadline(self) -> None:
|
||||
deadline = self.state.action_deadline
|
||||
if self.state.mode == "action" and deadline is not None and time.monotonic() >= deadline:
|
||||
self.state.mode = "paused"
|
||||
self.state.action_deadline = None
|
||||
self.state.action_queue.clear()
|
||||
self.state.log("timed action elapsed — paused")
|
||||
|
||||
def _flush_logs(self) -> None:
|
||||
for line in self.state.log_lines:
|
||||
print(f"[runtime] {line}", flush=True)
|
||||
|
||||
def _on_shutdown(self) -> None:
|
||||
self.state.action_queue.clear()
|
||||
print("[runtime] stopped", flush=True)
|
||||
@@ -1,39 +0,0 @@
|
||||
# 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.
|
||||
# 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.
|
||||
|
||||
"""Lazy mapping from policy types to language-runtime adapters."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
|
||||
_ADAPTERS: dict[str, str] = {
|
||||
"pi052": "lerobot.policies.pi052.inference.pi052_adapter:PI052PolicyAdapter",
|
||||
"pi05": "lerobot.runtime.adapter:DirectTaskPolicyAdapter",
|
||||
"molmoact2": "lerobot.runtime.adapter:DirectTaskPolicyAdapter",
|
||||
}
|
||||
|
||||
|
||||
def get_language_adapter_factory(policy_type: str) -> Callable[..., Any]:
|
||||
"""Return the adapter class registered for ``policy_type``."""
|
||||
spec = _ADAPTERS.get(policy_type)
|
||||
if spec is None:
|
||||
raise ValueError(
|
||||
f"No language-runtime adapter registered for policy type {policy_type!r}. "
|
||||
f"Registered: {sorted(_ADAPTERS)}. Add an entry to lerobot.runtime.registry."
|
||||
)
|
||||
module_path, class_name = spec.split(":")
|
||||
return getattr(importlib.import_module(module_path), class_name)
|
||||
@@ -1,406 +0,0 @@
|
||||
# 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.
|
||||
# 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.
|
||||
|
||||
"""RoboCasa backend for interactive language-conditioned rollouts.
|
||||
|
||||
It reuses the eval observation/action pipeline while prompts control a persistent selected scene.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
from lerobot.utils.io_utils import StreamingVideoWriter
|
||||
from lerobot.utils.video_annotation import annotate_frame
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _short_cam_name(cam: str) -> str:
|
||||
"""Human-friendly view label for a RoboCasa camera name."""
|
||||
c = cam.replace("robot0_", "")
|
||||
return {
|
||||
"agentview_left": "left",
|
||||
"agentview_right": "right",
|
||||
"eye_in_hand": "wrist",
|
||||
}.get(c, c)
|
||||
|
||||
|
||||
def _label_panel(img: np.ndarray, label: str) -> np.ndarray:
|
||||
"""Draw a small camera-view label in the bottom-left corner of a panel."""
|
||||
try:
|
||||
import cv2 # noqa: PLC0415
|
||||
except ImportError:
|
||||
return img
|
||||
y = img.shape[0] - 6
|
||||
cv2.putText(img, label, (5, y), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (0, 0, 0), 3, cv2.LINE_AA)
|
||||
cv2.putText(img, label, (5, y), cv2.FONT_HERSHEY_SIMPLEX, 0.45, (255, 255, 0), 1, cv2.LINE_AA)
|
||||
return img
|
||||
|
||||
|
||||
# Two workers avoid broken single-worker EGL rendering; only env 0 is displayed.
|
||||
_SIM_N_ENVS = 2
|
||||
|
||||
|
||||
def create_sim_env(
|
||||
*,
|
||||
task: str,
|
||||
split: str | None,
|
||||
obj_registries: list[str],
|
||||
seed: int | None,
|
||||
render_size: int = 384,
|
||||
) -> tuple[Any, dict]:
|
||||
"""Create and reset the vectorized RoboCasa environment before CUDA initializes.
|
||||
|
||||
Two workers keep EGL stable, while only env 0 is driven and displayed.
|
||||
"""
|
||||
from lerobot.envs.configs import RoboCasaEnv as RoboCasaEnvConfig # noqa: PLC0415
|
||||
|
||||
# The policy resizes inputs, so render_size only affects display quality and cost.
|
||||
env_cfg = RoboCasaEnvConfig(
|
||||
task=task,
|
||||
split=split,
|
||||
obj_registries=list(obj_registries),
|
||||
observation_height=render_size,
|
||||
observation_width=render_size,
|
||||
)
|
||||
# Keep one kitchen alive across sequential prompts.
|
||||
envs = env_cfg.create_envs(
|
||||
n_envs=_SIM_N_ENVS,
|
||||
use_async_envs=True,
|
||||
terminate_on_success=False,
|
||||
horizon=100_000,
|
||||
)
|
||||
env = envs[next(iter(envs))][0]
|
||||
logger.info("[sim] resetting RoboCasa scene task=%r split=%r (n_envs=%d)", task, split, _SIM_N_ENVS)
|
||||
seeds = None if seed is None else [seed + i for i in range(_SIM_N_ENVS)]
|
||||
obs, _ = env.reset(seed=seeds)
|
||||
return env, obs
|
||||
|
||||
|
||||
def start_mjpeg_server(port: int, get_frame: Callable[[], np.ndarray | None]) -> Any:
|
||||
"""Start an MJPEG server that shows a placeholder until ``get_frame`` returns frames."""
|
||||
import io # noqa: PLC0415
|
||||
import threading # noqa: PLC0415
|
||||
import time # noqa: PLC0415
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # noqa: PLC0415
|
||||
|
||||
from PIL import Image # noqa: PLC0415
|
||||
|
||||
_placeholder = Image.new("RGB", (256, 256), (17, 17, 17))
|
||||
|
||||
class _Handler(BaseHTTPRequestHandler):
|
||||
def log_message(self, *args): # silence per-request logging
|
||||
pass
|
||||
|
||||
def do_GET(self): # noqa: N802
|
||||
if self.path in ("/", "/index.html"):
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/html")
|
||||
self.end_headers()
|
||||
self.wfile.write(
|
||||
b"<html><body style='margin:0;background:#111;text-align:center'>"
|
||||
b"<img src='/stream' style='max-width:100vw;max-height:100vh;"
|
||||
b"image-rendering:pixelated'></body></html>"
|
||||
)
|
||||
return
|
||||
if self.path != "/stream":
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "multipart/x-mixed-replace; boundary=frame")
|
||||
self.end_headers()
|
||||
try:
|
||||
while True:
|
||||
frame = get_frame()
|
||||
buf = io.BytesIO()
|
||||
img = Image.fromarray(frame) if frame is not None else _placeholder
|
||||
img.save(buf, format="JPEG", quality=80)
|
||||
data = buf.getvalue()
|
||||
self.wfile.write(
|
||||
b"--frame\r\nContent-Type: image/jpeg\r\nContent-Length: "
|
||||
+ str(len(data)).encode()
|
||||
+ b"\r\n\r\n"
|
||||
+ data
|
||||
+ b"\r\n"
|
||||
)
|
||||
time.sleep(0.05)
|
||||
except (BrokenPipeError, ConnectionResetError):
|
||||
pass
|
||||
|
||||
try:
|
||||
# Bind all interfaces intentionally so the viewer remains reachable
|
||||
# through the documented SSH port-forwarding workflow.
|
||||
server = ThreadingHTTPServer(("0.0.0.0", port), _Handler) # nosec B104
|
||||
except OSError as exc:
|
||||
logger.warning("[sim] could not start live stream on port %d: %s", port, exc)
|
||||
print(f"[runtime] WARNING: live stream port {port} unavailable ({exc})", flush=True)
|
||||
return None
|
||||
threading.Thread(target=server.serve_forever, daemon=True, name="sim-mjpeg").start()
|
||||
print(
|
||||
f"[runtime] live view: http://localhost:{port} "
|
||||
f"(over SSH: ssh -L {port}:localhost:{port} <host>) — loading until scene is ready",
|
||||
flush=True,
|
||||
)
|
||||
return server
|
||||
|
||||
|
||||
class RoboCasaSimBackend:
|
||||
"""Expose a RoboCasa environment through the runtime observation/action contract.
|
||||
|
||||
The environment must be created before the policy initializes CUDA.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
env: Any,
|
||||
last_obs: dict,
|
||||
task: str,
|
||||
seed: int | None,
|
||||
device: str,
|
||||
preprocessor: Any,
|
||||
postprocessor: Any,
|
||||
record: bool = True,
|
||||
output_dir: str | None = None,
|
||||
view_cams: list[str] | None = None,
|
||||
) -> None:
|
||||
self.env = env
|
||||
self._last_obs = last_obs
|
||||
self._scene_task = task
|
||||
self._view_cams = view_cams or [
|
||||
"robot0_agentview_left",
|
||||
"robot0_eye_in_hand",
|
||||
"robot0_agentview_right",
|
||||
]
|
||||
self.device = torch.device(device) if isinstance(device, str) else device
|
||||
self.preprocessor = preprocessor
|
||||
self.postprocessor = postprocessor
|
||||
self.seed = seed
|
||||
self.record = record
|
||||
self.output_dir = Path(output_dir) if output_dir else Path("outputs/runtime_sim")
|
||||
|
||||
self._video_writer: StreamingVideoWriter | None = None
|
||||
self._video_path: Path | None = None
|
||||
self._live_counter = 0
|
||||
self._latest_frame: np.ndarray | None = None
|
||||
self._stream_server: Any = None
|
||||
self._reset_count = 0
|
||||
# Bind these after runtime construction for live annotations.
|
||||
self._task_getter: Callable[[], str | None] | None = None
|
||||
self._subtask_getter: Callable[[], str | None] | None = None
|
||||
self._memory_getter: Callable[[], str | None] | None = None
|
||||
logger.info("[sim] scene ready — task_description=%r", self._scene_description())
|
||||
|
||||
def bind_runtime(self, runtime: Any) -> None:
|
||||
"""Wire live task/subtask/memory getters from the runtime state."""
|
||||
self._task_getter = lambda: runtime.state.get("task")
|
||||
self._subtask_getter = lambda: runtime.state.language_context.get("subtask")
|
||||
self._memory_getter = lambda: (runtime.state.get("language_context") or {}).get("memory")
|
||||
|
||||
def _scene_description(self) -> str:
|
||||
try:
|
||||
return str(self.env.get_attr("task_description")[0]) or self._scene_task
|
||||
except Exception: # noqa: BLE001
|
||||
return self._scene_task
|
||||
|
||||
def _current_task(self) -> str:
|
||||
task = self._task_getter() if self._task_getter else None
|
||||
return task or self._scene_description() or self._scene_task
|
||||
|
||||
def reset_scene(self) -> None:
|
||||
"""Re-roll the kitchen: reset the env to a fresh scene (new layout/style).
|
||||
|
||||
Uses a new seed each call so ``/reset`` explores different kitchens.
|
||||
"""
|
||||
self._reset_count += 1
|
||||
n = self.env.num_envs
|
||||
if self.seed is None:
|
||||
seeds = None
|
||||
else:
|
||||
base = self.seed + self._reset_count * 1000
|
||||
seeds = [base + i for i in range(n)]
|
||||
obs, _ = self.env.reset(seed=seeds)
|
||||
self._last_obs = obs
|
||||
logger.info("[sim] scene reset (#%d)", self._reset_count)
|
||||
|
||||
def _env0_obs(self) -> dict:
|
||||
"""Slice env 0 out of the batched vec-env observation (batch of 1)."""
|
||||
raw = self._last_obs or {}
|
||||
pixels = raw.get("pixels")
|
||||
out: dict[str, Any] = {}
|
||||
if isinstance(pixels, dict):
|
||||
out["pixels"] = {k: np.asarray(v)[0:1] for k, v in pixels.items()}
|
||||
agent_pos = raw.get("agent_pos")
|
||||
if agent_pos is not None:
|
||||
out["agent_pos"] = np.asarray(agent_pos)[0:1]
|
||||
return out
|
||||
|
||||
def observation_provider(self) -> dict | None:
|
||||
from lerobot.envs.utils import preprocess_observation # noqa: PLC0415
|
||||
|
||||
try:
|
||||
obs = preprocess_observation(self._env0_obs())
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("[sim] preprocess_observation failed: %s", exc)
|
||||
return None
|
||||
# The adapter later replaces this recipe input with its generated subtask.
|
||||
obs["task"] = [self._current_task()]
|
||||
if self.preprocessor is not None:
|
||||
try:
|
||||
obs = self.preprocessor(obs)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("[sim] preprocessor failed: %s", exc)
|
||||
return None
|
||||
return {
|
||||
k: (v.to(self.device) if isinstance(v, torch.Tensor) else v)
|
||||
for k, v in obs.items()
|
||||
if isinstance(k, str) and k.startswith("observation.")
|
||||
}
|
||||
|
||||
def action_executor(self, action: Any) -> None:
|
||||
try:
|
||||
if self.postprocessor is not None:
|
||||
action = self.postprocessor(action)
|
||||
if isinstance(action, torch.Tensor):
|
||||
if action.ndim > 1 and action.shape[0] == 1:
|
||||
action = action.squeeze(0)
|
||||
action = action.detach().to("cpu").numpy()
|
||||
# Tile env 0's action because the extra workers exist only for EGL stability.
|
||||
action_row = np.asarray(action, dtype=np.float32).reshape(-1)
|
||||
action_np = np.tile(action_row, (self.env.num_envs, 1))
|
||||
obs, _reward, terminated, truncated, _info = self.env.step(action_np)
|
||||
self._last_obs = obs
|
||||
self._capture_frame()
|
||||
# AsyncVectorEnv resets terminated sub-environments automatically.
|
||||
if bool(np.any(terminated)) or bool(np.any(truncated)):
|
||||
logger.info("[sim] episode ended — scene auto-reset")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.error("[sim] env.step failed: %s", exc, exc_info=True)
|
||||
|
||||
def _multiview_frame(self) -> np.ndarray | None:
|
||||
"""Label and compose env 0's existing observation views without extra rendering."""
|
||||
pixels = (self._last_obs or {}).get("pixels")
|
||||
if not isinstance(pixels, dict) or not pixels:
|
||||
return None
|
||||
panels: list[np.ndarray] = []
|
||||
for cam in self._view_cams:
|
||||
v = pixels.get(cam)
|
||||
if v is None:
|
||||
continue
|
||||
img = np.asarray(v)
|
||||
if img.ndim == 4: # (n_envs, H, W, C) -> env 0
|
||||
img = img[0]
|
||||
if img.ndim != 3 or img.shape[-1] != 3:
|
||||
continue
|
||||
panels.append(_label_panel(np.ascontiguousarray(img.astype(np.uint8)), _short_cam_name(cam)))
|
||||
if not panels:
|
||||
return None
|
||||
h = min(p.shape[0] for p in panels)
|
||||
panels = [p[:h] for p in panels]
|
||||
return np.concatenate(panels, axis=1)
|
||||
|
||||
def _capture_frame(self) -> None:
|
||||
frame = self._multiview_frame()
|
||||
if frame is None: # fallback to single env.render()
|
||||
try:
|
||||
rendered = self.env.call("render")[0]
|
||||
if isinstance(rendered, np.ndarray) and rendered.ndim == 3:
|
||||
frame = rendered
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.debug("[sim] render failed: %s", exc)
|
||||
if frame is None:
|
||||
return
|
||||
subtask = self._subtask_getter() if self._subtask_getter else None
|
||||
memory = self._memory_getter() if self._memory_getter else None
|
||||
annotated = annotate_frame(
|
||||
frame,
|
||||
(("Task", self._current_task()), ("Subtask", subtask), ("Memory", memory)),
|
||||
)
|
||||
self._latest_frame = annotated # served by the live MJPEG stream
|
||||
self._write_live_frame(annotated)
|
||||
if self.record:
|
||||
self._write_recording_frame(annotated)
|
||||
|
||||
def _write_recording_frame(self, frame: np.ndarray) -> None:
|
||||
try:
|
||||
if self._video_writer is None:
|
||||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
||||
stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
self._video_path = self.output_dir / f"sim_{stamp}.mp4"
|
||||
fps = int((getattr(self.env, "metadata", None) or {}).get("render_fps", 20))
|
||||
self._video_writer = StreamingVideoWriter(self._video_path, fps)
|
||||
self._video_writer.add_frame(frame)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("[sim] video encoding failed: %s", exc)
|
||||
self.record = False
|
||||
|
||||
def _write_live_frame(self, frame: np.ndarray) -> None:
|
||||
"""Write a rolling latest.png every few frames for live viewing over SSH.
|
||||
|
||||
Open ``{output_dir}/latest.png`` in an editor/viewer and refresh to watch
|
||||
the rollout in near-real-time without a GUI window. Written atomically
|
||||
(temp + replace) so a reader never sees a half-written file.
|
||||
"""
|
||||
self._live_counter += 1
|
||||
if self._live_counter % 3 != 0:
|
||||
return
|
||||
try:
|
||||
import os # noqa: PLC0415
|
||||
|
||||
from PIL import Image # noqa: PLC0415
|
||||
|
||||
self.output_dir.mkdir(parents=True, exist_ok=True)
|
||||
tmp = self.output_dir / ".latest.tmp.png"
|
||||
Image.fromarray(frame).save(tmp)
|
||||
os.replace(tmp, self.output_dir / "latest.png")
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.debug("[sim] live frame write failed: %s", exc)
|
||||
|
||||
def _flush_video(self) -> None:
|
||||
if self._video_writer is None:
|
||||
return
|
||||
writer = self._video_writer
|
||||
self._video_writer = None
|
||||
try:
|
||||
writer.close()
|
||||
logger.info("[sim] wrote video (%d frames) to %s", writer.frames_written, self._video_path)
|
||||
print(f"[runtime] sim video saved to {self._video_path}", flush=True)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.warning("[sim] video close failed: %s", exc)
|
||||
|
||||
def attach_stream_server(self, server: Any) -> None:
|
||||
"""Attach an already-running MJPEG server so disconnect() can stop it."""
|
||||
self._stream_server = server
|
||||
|
||||
def disconnect(self) -> None:
|
||||
"""Match the robot backend's cleanup contract."""
|
||||
if self._stream_server is not None:
|
||||
try:
|
||||
self._stream_server.shutdown()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.debug("[sim] stream server shutdown raised %s", exc)
|
||||
self._flush_video()
|
||||
try:
|
||||
self.env.close()
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.debug("[sim] env.close raised %s", exc)
|
||||
@@ -94,19 +94,6 @@ from lerobot.utils.utils import (
|
||||
init_logging,
|
||||
inside_slurm,
|
||||
)
|
||||
from lerobot.utils.video_annotation import annotate_frame
|
||||
|
||||
|
||||
def _annotate_eval_frames(frames: np.ndarray, task: str | None, subtask: str | None) -> np.ndarray:
|
||||
"""Overlay the high-level task and predicted subtask onto rendered frames.
|
||||
|
||||
``frames`` is ``(n_envs, H, W, C)`` uint8. Best-effort: if OpenCV isn't
|
||||
available the frames are returned unchanged so eval never fails over a
|
||||
visualization concern.
|
||||
"""
|
||||
if frames.ndim != 4 or frames.shape[-1] != 3:
|
||||
return frames
|
||||
return np.stack([annotate_frame(frame, (("Task", task), ("Subtask", subtask))) for frame in frames])
|
||||
|
||||
|
||||
def _env_features_to_dataset_features(env_features: dict) -> dict:
|
||||
@@ -466,9 +453,6 @@ def eval_policy(
|
||||
raise exc from None
|
||||
|
||||
start = time.time()
|
||||
# Preserve the mode for direct callers. eval_policy_all scopes the mode
|
||||
# around all tasks so parallel evaluations cannot race with each other.
|
||||
was_training = policy.training
|
||||
policy.eval()
|
||||
|
||||
# Determine how many batched rollouts we need to get n_episodes. Note that if n_episodes is not evenly
|
||||
@@ -490,36 +474,11 @@ def eval_policy(
|
||||
return
|
||||
n_to_render_now = min(max_episodes_rendered - n_episodes_rendered, env.num_envs)
|
||||
if isinstance(env, gym.vector.SyncVectorEnv):
|
||||
frames = np.stack([env.envs[i].render() for i in range(n_to_render_now)]) # noqa: B023
|
||||
ep_frames.append(np.stack([env.envs[i].render() for i in range(n_to_render_now)])) # noqa: B023
|
||||
elif hasattr(env, "call"):
|
||||
# Here we must render all frames and discard any we don't need.
|
||||
# Covers AsyncVectorEnv and _LazyAsyncVectorEnv (which wraps one).
|
||||
frames = np.stack(env.call("render")[:n_to_render_now])
|
||||
else:
|
||||
return
|
||||
|
||||
# Overlay the high-level task and (for hierarchical policies like
|
||||
# pi052) the predicted low-level subtask onto each frame. Both are
|
||||
# best-effort: missing values just skip that line.
|
||||
try:
|
||||
tasks = list(env.call("task_description"))
|
||||
except (AttributeError, NotImplementedError):
|
||||
try:
|
||||
tasks = list(env.call("task"))
|
||||
except (AttributeError, NotImplementedError):
|
||||
tasks = None
|
||||
subtasks = getattr(policy, "last_subtasks", None)
|
||||
annotated = []
|
||||
for i in range(frames.shape[0]):
|
||||
subtask_i = subtasks[i] if subtasks is not None and i < len(subtasks) else None
|
||||
annotated.append(
|
||||
_annotate_eval_frames(
|
||||
frames[i : i + 1],
|
||||
tasks[i] if tasks is not None and i < len(tasks) else None,
|
||||
subtask_i,
|
||||
)[0]
|
||||
)
|
||||
ep_frames.append(np.stack(annotated))
|
||||
ep_frames.append(np.stack(env.call("render")[:n_to_render_now]))
|
||||
|
||||
if max_episodes_rendered > 0:
|
||||
video_paths: list[str] = []
|
||||
@@ -715,8 +674,6 @@ def eval_policy(
|
||||
if save_predicted_video:
|
||||
info["predicted_video_paths"] = predicted_video_paths
|
||||
|
||||
policy.train(was_training)
|
||||
|
||||
return info
|
||||
|
||||
|
||||
@@ -1053,48 +1010,40 @@ def eval_policy_all(
|
||||
recording_private=recording_private,
|
||||
)
|
||||
|
||||
# Set the shared policy's mode before launching any workers. Restoring it
|
||||
# inside individual tasks would let one task enable training mode while
|
||||
# another task is still evaluating.
|
||||
was_training = policy.training
|
||||
policy.eval()
|
||||
try:
|
||||
if max_parallel_tasks <= 1:
|
||||
prefetch_thread: threading.Thread | None = None
|
||||
for i, (task_group, task_id, env) in enumerate(tasks):
|
||||
if prefetch_thread is not None:
|
||||
prefetch_thread.join()
|
||||
prefetch_thread = None
|
||||
if max_parallel_tasks <= 1:
|
||||
prefetch_thread: threading.Thread | None = None
|
||||
for i, (task_group, task_id, env) in enumerate(tasks):
|
||||
if prefetch_thread is not None:
|
||||
prefetch_thread.join()
|
||||
prefetch_thread = None
|
||||
|
||||
try:
|
||||
tg, tid, metrics = task_runner(task_group, task_id, env)
|
||||
_accumulate_to(tg, metrics)
|
||||
per_task_infos.append({"task_group": tg, "task_id": tid, "metrics": metrics})
|
||||
finally:
|
||||
env.close()
|
||||
# Prefetch next task's workers *after* closing current env to prevent
|
||||
# GPU memory overlap between consecutive tasks.
|
||||
if i + 1 < len(tasks):
|
||||
next_env = tasks[i + 1][2]
|
||||
if hasattr(next_env, "_ensure"):
|
||||
prefetch_thread = threading.Thread(target=next_env._ensure, daemon=True)
|
||||
prefetch_thread.start()
|
||||
else:
|
||||
with cf.ThreadPoolExecutor(max_workers=max_parallel_tasks) as executor:
|
||||
fut2meta = {}
|
||||
for task_group, task_id, env in tasks:
|
||||
fut = executor.submit(task_runner, task_group, task_id, env)
|
||||
fut2meta[fut] = (task_group, task_id, env)
|
||||
for fut in cf.as_completed(fut2meta):
|
||||
tg, tid, env = fut2meta[fut]
|
||||
try:
|
||||
tg, tid, metrics = task_runner(task_group, task_id, env)
|
||||
tg, tid, metrics = fut.result()
|
||||
_accumulate_to(tg, metrics)
|
||||
per_task_infos.append({"task_group": tg, "task_id": tid, "metrics": metrics})
|
||||
finally:
|
||||
env.close()
|
||||
# Prefetch next task's workers *after* closing current env to prevent
|
||||
# GPU memory overlap between consecutive tasks.
|
||||
if i + 1 < len(tasks):
|
||||
next_env = tasks[i + 1][2]
|
||||
if hasattr(next_env, "_ensure"):
|
||||
prefetch_thread = threading.Thread(target=next_env._ensure, daemon=True)
|
||||
prefetch_thread.start()
|
||||
else:
|
||||
with cf.ThreadPoolExecutor(max_workers=max_parallel_tasks) as executor:
|
||||
fut2meta = {}
|
||||
for task_group, task_id, env in tasks:
|
||||
fut = executor.submit(task_runner, task_group, task_id, env)
|
||||
fut2meta[fut] = (task_group, task_id, env)
|
||||
for fut in cf.as_completed(fut2meta):
|
||||
tg, tid, env = fut2meta[fut]
|
||||
try:
|
||||
tg, tid, metrics = fut.result()
|
||||
_accumulate_to(tg, metrics)
|
||||
per_task_infos.append({"task_group": tg, "task_id": tid, "metrics": metrics})
|
||||
finally:
|
||||
env.close()
|
||||
finally:
|
||||
policy.train(was_training)
|
||||
|
||||
# compute aggregated metrics helper (robust to lists/scalars)
|
||||
def _agg_from_list(xs):
|
||||
|
||||
@@ -453,11 +453,9 @@ def record(
|
||||
encoder_queue_maxsize=cfg.dataset.encoder_queue_maxsize,
|
||||
)
|
||||
|
||||
# Connect the teleoperator before the robot so the robot isn't left idle (and possibly
|
||||
# tripping a firmware watchdog) during teleop init. Matches lerobot_teleoperate.py.
|
||||
robot.connect()
|
||||
if teleop is not None:
|
||||
teleop.connect()
|
||||
robot.connect()
|
||||
|
||||
listener, events = init_keyboard_listener()
|
||||
|
||||
|
||||
@@ -61,7 +61,6 @@ from lerobot.robots import ( # noqa: F401
|
||||
earthrover_mini_plus,
|
||||
hope_jr,
|
||||
koch_follower,
|
||||
lekiwi,
|
||||
make_robot_from_config,
|
||||
omx_follower,
|
||||
openarm_follower,
|
||||
|
||||
@@ -151,7 +151,6 @@ Usage examples
|
||||
"""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
|
||||
from lerobot.cameras.opencv import OpenCVCameraConfig # noqa: F401
|
||||
from lerobot.cameras.realsense import RealSenseCameraConfig # noqa: F401
|
||||
@@ -242,69 +241,10 @@ def rollout(cfg: RolloutConfig):
|
||||
logger.info("Rollout finished")
|
||||
|
||||
|
||||
_LANGUAGE_RUNTIME_FLAGS = {
|
||||
"--language",
|
||||
"--no_robot",
|
||||
"--sim",
|
||||
"--direct_subtask",
|
||||
"--sim.direct_subtask",
|
||||
"--disable_memory",
|
||||
"--fp8",
|
||||
}
|
||||
_LANGUAGE_RUNTIME_PREFIXES = (
|
||||
"--sim.",
|
||||
"--chunk_hz",
|
||||
"--ctrl_hz",
|
||||
"--high_level_hz",
|
||||
"--subtask_chunks_per_gen",
|
||||
"--text_min_new_tokens",
|
||||
"--text_temperature",
|
||||
"--text_top_p",
|
||||
)
|
||||
|
||||
|
||||
def _uses_language_runtime(argv: list[str]) -> bool:
|
||||
"""Return whether *argv* selects the interactive language runtime.
|
||||
|
||||
``--language`` is the explicit selector for real-robot runs whose other
|
||||
options overlap with the standard rollout CLI. Language-only options also
|
||||
select it automatically, which keeps the former language-runtime examples
|
||||
working after replacing their command name with ``lerobot-rollout``.
|
||||
"""
|
||||
return any(
|
||||
arg.split("=", 1)[0] in _LANGUAGE_RUNTIME_FLAGS or arg.startswith(_LANGUAGE_RUNTIME_PREFIXES)
|
||||
for arg in argv
|
||||
)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None):
|
||||
"""CLI entry point for ``lerobot-rollout``.
|
||||
|
||||
Standard policy deployment continues through :class:`RolloutConfig`.
|
||||
Interactive language-conditioned and RoboCasa runs share this entry point
|
||||
and are selected with ``--language`` or any language-runtime-only option.
|
||||
"""
|
||||
def main():
|
||||
"""CLI entry point for ``lerobot-rollout``."""
|
||||
register_third_party_plugins()
|
||||
cli_args = list(sys.argv[1:] if argv is None else argv)
|
||||
if _uses_language_runtime(cli_args):
|
||||
from lerobot.runtime.cli import run as run_language_runtime
|
||||
|
||||
# ``--language`` is a dispatcher flag, not part of the runtime's own
|
||||
# argparse surface. All other arguments pass through unchanged.
|
||||
runtime_args = [arg for arg in cli_args if arg != "--language"]
|
||||
return run_language_runtime(runtime_args, prog="lerobot-rollout")
|
||||
|
||||
if argv is None:
|
||||
return rollout()
|
||||
|
||||
# draccus reads sys.argv. Supporting an explicit argv keeps this entry
|
||||
# point easy to smoke-test and mirrors the language-runtime branch above.
|
||||
previous_argv = sys.argv
|
||||
try:
|
||||
sys.argv = [previous_argv[0], *cli_args]
|
||||
return rollout()
|
||||
finally:
|
||||
sys.argv = previous_argv
|
||||
rollout()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -71,16 +71,6 @@ from lerobot.utils.utils import (
|
||||
from .lerobot_eval import eval_policy_all
|
||||
|
||||
|
||||
def _dataloader_worker_kwargs(cfg: TrainPipelineConfig) -> dict[str, Any]:
|
||||
"""Return worker-only DataLoader options, disabling them for single-process loading."""
|
||||
workers_enabled = cfg.num_workers > 0
|
||||
return {
|
||||
"prefetch_factor": cfg.prefetch_factor if workers_enabled else None,
|
||||
"persistent_workers": cfg.persistent_workers and workers_enabled,
|
||||
"multiprocessing_context": cfg.dataloader_multiprocessing_context if workers_enabled else None,
|
||||
}
|
||||
|
||||
|
||||
def update_policy(
|
||||
train_metrics: MetricsTracker,
|
||||
policy: PreTrainedPolicy,
|
||||
@@ -483,7 +473,8 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
|
||||
pin_memory=device.type == "cuda",
|
||||
drop_last=False,
|
||||
collate_fn=collate_fn,
|
||||
**_dataloader_worker_kwargs(cfg),
|
||||
prefetch_factor=cfg.prefetch_factor if cfg.num_workers > 0 else None,
|
||||
persistent_workers=cfg.persistent_workers and cfg.num_workers > 0,
|
||||
)
|
||||
|
||||
# Build eval dataloader if a held-out split exists
|
||||
@@ -509,7 +500,8 @@ def train(cfg: TrainPipelineConfig, accelerator: "Accelerator | None" = None):
|
||||
pin_memory=device.type == "cuda",
|
||||
drop_last=False,
|
||||
collate_fn=eval_collate_fn,
|
||||
**_dataloader_worker_kwargs(cfg),
|
||||
prefetch_factor=cfg.prefetch_factor if cfg.num_workers > 0 else None,
|
||||
persistent_workers=cfg.persistent_workers and cfg.num_workers > 0,
|
||||
)
|
||||
|
||||
# Prepare everything with accelerator
|
||||
|
||||
@@ -22,7 +22,7 @@ from torch.utils.data._utils.collate import default_collate
|
||||
|
||||
from lerobot.datasets.language import LANGUAGE_COLUMNS
|
||||
|
||||
_PYTHON_LIST_KEYS = {"messages", "message_streams", "target_message_indices", *LANGUAGE_COLUMNS}
|
||||
_PYTHON_LIST_KEYS = {"messages", "message_streams", "target_message_indices"}
|
||||
|
||||
|
||||
def lerobot_collate_fn(batch: list[dict[str, Any] | None]) -> dict[str, Any] | None:
|
||||
|
||||
@@ -26,7 +26,6 @@ OBS_IMAGES = OBS_IMAGE + "s"
|
||||
OBS_LANGUAGE = OBS_STR + ".language"
|
||||
OBS_LANGUAGE_TOKENS = OBS_LANGUAGE + ".tokens"
|
||||
OBS_LANGUAGE_ATTENTION_MASK = OBS_LANGUAGE + ".attention_mask"
|
||||
OBS_LANGUAGE_CAUSAL_MARKS = OBS_LANGUAGE + ".causal_marks"
|
||||
OBS_LANGUAGE_SUBTASK = OBS_STR + ".subtask"
|
||||
OBS_LANGUAGE_SUBTASK_TOKENS = OBS_LANGUAGE_SUBTASK + ".tokens"
|
||||
OBS_LANGUAGE_SUBTASK_ATTENTION_MASK = OBS_LANGUAGE_SUBTASK + ".attention_mask"
|
||||
@@ -35,7 +34,6 @@ ACTION = "action"
|
||||
ACTION_PREFIX = ACTION + "."
|
||||
ACTION_TOKENS = ACTION + ".tokens"
|
||||
ACTION_TOKEN_MASK = ACTION + ".token_mask"
|
||||
ACTION_CODE_TOKEN_MASK = ACTION + ".code_token_mask"
|
||||
REWARD = "next.reward"
|
||||
TRUNCATED = "next.truncated"
|
||||
DONE = "next.done"
|
||||
|
||||
@@ -23,46 +23,6 @@ logger = logging.getLogger(__name__)
|
||||
JsonLike = str | int | float | bool | None | list["JsonLike"] | dict[str, "JsonLike"] | tuple["JsonLike", ...]
|
||||
|
||||
|
||||
class StreamingVideoWriter:
|
||||
"""Incrementally encode RGB frames to an MP4 without retaining them in memory."""
|
||||
|
||||
def __init__(self, video_path: str | Path, fps: int) -> None:
|
||||
from .import_utils import require_package
|
||||
|
||||
require_package("av", extra="av-dep")
|
||||
import av
|
||||
|
||||
self._av = av
|
||||
self._container = av.open(str(video_path), mode="w")
|
||||
self._stream = self._container.add_stream("libx264", rate=fps)
|
||||
self._shape: tuple[int, int] | None = None
|
||||
self.frames_written = 0
|
||||
|
||||
def add_frame(self, frame_array) -> None:
|
||||
orig_height, orig_width = frame_array.shape[:2]
|
||||
height = orig_height - orig_height % 2
|
||||
width = orig_width - orig_width % 2
|
||||
if self._shape is None:
|
||||
self._shape = (height, width)
|
||||
self._stream.width = width
|
||||
self._stream.height = height
|
||||
self._stream.pix_fmt = "yuv420p"
|
||||
elif self._shape != (height, width):
|
||||
raise ValueError(f"Video frame shape changed from {self._shape} to {(height, width)}")
|
||||
frame = self._av.VideoFrame.from_ndarray(frame_array[:height, :width], format="rgb24")
|
||||
for packet in self._stream.encode(frame):
|
||||
self._container.mux(packet)
|
||||
self.frames_written += 1
|
||||
|
||||
def close(self) -> None:
|
||||
if self._container is None:
|
||||
return
|
||||
for packet in self._stream.encode():
|
||||
self._container.mux(packet)
|
||||
self._container.close()
|
||||
self._container = None
|
||||
|
||||
|
||||
def load_json(fpath: Path) -> Any:
|
||||
"""Load data from a JSON file.
|
||||
|
||||
@@ -98,12 +58,36 @@ def write_video(video_path: str | Path, stacked_frames: list, fps: int) -> None:
|
||||
stacked_frames: List of HWC uint8 numpy arrays (RGB).
|
||||
fps: Frames per second for the output video.
|
||||
"""
|
||||
writer = StreamingVideoWriter(video_path, fps)
|
||||
try:
|
||||
from .import_utils import require_package
|
||||
|
||||
require_package("av", extra="av-dep")
|
||||
import av
|
||||
|
||||
with av.open(str(video_path), mode="w") as container:
|
||||
orig_height, orig_width = stacked_frames[0].shape[:2]
|
||||
# yuv420p requires even dimensions; crop by one pixel if needed
|
||||
height = orig_height if orig_height % 2 == 0 else orig_height - 1
|
||||
width = orig_width if orig_width % 2 == 0 else orig_width - 1
|
||||
if height != orig_height or width != orig_width:
|
||||
logger.warning(
|
||||
"Frame dimensions %dx%d are not even; cropping to %dx%d for yuv420p compatibility.",
|
||||
orig_width,
|
||||
orig_height,
|
||||
width,
|
||||
height,
|
||||
)
|
||||
stream = container.add_stream("libx264", rate=fps)
|
||||
stream.width = width
|
||||
stream.height = height
|
||||
stream.pix_fmt = "yuv420p"
|
||||
for frame_array in stacked_frames:
|
||||
writer.add_frame(frame_array)
|
||||
finally:
|
||||
writer.close()
|
||||
if height != orig_height or width != orig_width:
|
||||
frame_array = frame_array[:height, :width]
|
||||
frame = av.VideoFrame.from_ndarray(frame_array, format="rgb24")
|
||||
for packet in stream.encode(frame):
|
||||
container.mux(packet)
|
||||
for packet in stream.encode():
|
||||
container.mux(packet)
|
||||
|
||||
|
||||
def deserialize_json_into_object[T: JsonLike](fpath: Path, obj: T) -> T:
|
||||
|
||||
@@ -38,10 +38,7 @@ def _is_scalar(x):
|
||||
|
||||
|
||||
def init_rerun(
|
||||
session_name: str = "lerobot_control_loop",
|
||||
ip: str | None = None,
|
||||
port: int | None = None,
|
||||
web_port: int | None = None,
|
||||
session_name: str = "lerobot_control_loop", ip: str | None = None, port: int | None = None
|
||||
) -> None:
|
||||
"""
|
||||
Initializes the Rerun SDK for visualizing the control loop.
|
||||
@@ -50,7 +47,6 @@ def init_rerun(
|
||||
session_name: Name of the Rerun session.
|
||||
ip: Optional IP for connecting to a Rerun server.
|
||||
port: Optional port for connecting to a Rerun server.
|
||||
web_port: Serve a headless web viewer on this port, using ``port`` for gRPC.
|
||||
"""
|
||||
|
||||
require_package("rerun-sdk", extra="viz", import_name="rerun")
|
||||
@@ -64,10 +60,6 @@ def init_rerun(
|
||||
memory_limit = os.getenv("LEROBOT_RERUN_MEMORY_LIMIT", "10%")
|
||||
if ip and port:
|
||||
rr.connect_grpc(url=f"rerun+http://{ip}:{port}/proxy")
|
||||
elif web_port is not None:
|
||||
grpc_port = port or 9876
|
||||
url = rr.serve_grpc(grpc_port=grpc_port)
|
||||
rr.serve_web_viewer(web_port=web_port, open_browser=False, connect_to=url)
|
||||
else:
|
||||
rr.spawn(memory_limit=memory_limit)
|
||||
|
||||
|
||||
@@ -1,71 +0,0 @@
|
||||
# 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.
|
||||
# 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.
|
||||
|
||||
"""Best-effort text overlays shared by evaluation and interactive rollouts."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Iterable
|
||||
|
||||
import numpy as np
|
||||
|
||||
|
||||
def annotate_frame(frame: np.ndarray, fields: Iterable[tuple[str, str | None]]) -> np.ndarray:
|
||||
"""Return an RGB frame annotated with the non-empty labeled ``fields``."""
|
||||
if frame.ndim != 3 or frame.shape[-1] != 3:
|
||||
return frame
|
||||
try:
|
||||
import cv2 # noqa: PLC0415
|
||||
except ImportError:
|
||||
return frame
|
||||
|
||||
text_rows = [f"{label}: {value}" for label, value in fields if value]
|
||||
if not text_rows:
|
||||
return frame
|
||||
|
||||
image = np.ascontiguousarray(frame).copy()
|
||||
font, scale, thickness, margin = cv2.FONT_HERSHEY_SIMPLEX, 0.5, 1, 6
|
||||
max_width = image.shape[1] - 2 * margin
|
||||
lines: list[str] = []
|
||||
for text in text_rows:
|
||||
current = ""
|
||||
for word in text.split():
|
||||
candidate = f"{current} {word}".strip()
|
||||
width = cv2.getTextSize(candidate, font, scale, thickness)[0][0]
|
||||
if width > max_width and current:
|
||||
lines.append(current)
|
||||
current = word
|
||||
else:
|
||||
current = candidate
|
||||
if current:
|
||||
lines.append(current)
|
||||
|
||||
line_height = 20
|
||||
header_height = min(image.shape[0], len(lines) * line_height + 6)
|
||||
backdrop = image.copy()
|
||||
cv2.rectangle(backdrop, (0, 0), (image.shape[1], header_height), (0, 0, 0), -1)
|
||||
cv2.addWeighted(backdrop, 0.55, image, 0.45, 0, dst=image)
|
||||
|
||||
for index, line in enumerate(lines):
|
||||
cv2.putText(
|
||||
image,
|
||||
line,
|
||||
(margin, 18 + index * line_height),
|
||||
font,
|
||||
scale,
|
||||
(255, 255, 255),
|
||||
thickness,
|
||||
cv2.LINE_AA,
|
||||
)
|
||||
return image
|
||||
@@ -29,13 +29,6 @@ def test_message_recipe_validates_unknown_binding():
|
||||
)
|
||||
|
||||
|
||||
def test_canonical_recipe_loads():
|
||||
"""The canonical PI052 blend YAML loads + validates."""
|
||||
recipe = TrainingRecipe.from_yaml(Path("src/lerobot/configs/recipes/subtask_mem_vqa_speech.yaml"))
|
||||
assert recipe.blend is not None
|
||||
assert sum(c.weight for c in recipe.blend.values()) == pytest.approx(1.0)
|
||||
|
||||
|
||||
def test_message_turn_requires_a_stream():
|
||||
"""Every turn must declare a stream — None is rejected at construction.
|
||||
|
||||
|
||||
@@ -204,38 +204,6 @@ def test_clear_resets_buffer(tmp_path):
|
||||
assert dataset.writer.episode_buffer["size"] == 0
|
||||
|
||||
|
||||
def test_clear_removes_video_frame_staging_dir(tmp_path):
|
||||
"""clear_episode_buffer() removes PNG staging dirs for video features."""
|
||||
video_key = "observation.images.cam"
|
||||
features = {
|
||||
video_key: {
|
||||
"dtype": "video",
|
||||
"shape": (64, 96, 3),
|
||||
"names": ["height", "width", "channels"],
|
||||
},
|
||||
"action": {"dtype": "float32", "shape": (2,), "names": None},
|
||||
}
|
||||
dataset = LeRobotDataset.create(
|
||||
repo_id=DUMMY_REPO_ID,
|
||||
fps=DEFAULT_FPS,
|
||||
features=features,
|
||||
root=tmp_path / "ds",
|
||||
use_videos=True,
|
||||
)
|
||||
|
||||
dataset.add_frame(_make_frame(features))
|
||||
video_staging_dir = (
|
||||
dataset.root
|
||||
/ Path(DEFAULT_IMAGE_PATH.format(image_key=video_key, episode_index=0, frame_index=0)).parent
|
||||
)
|
||||
assert video_staging_dir.is_dir()
|
||||
|
||||
dataset.clear_episode_buffer()
|
||||
|
||||
assert dataset.writer.episode_buffer["size"] == 0
|
||||
assert not video_staging_dir.exists()
|
||||
|
||||
|
||||
def test_finalize_is_idempotent(tmp_path):
|
||||
"""Calling finalize() twice does not raise."""
|
||||
dataset = LeRobotDataset.create(
|
||||
|
||||
@@ -343,84 +343,6 @@ def test_resolve_task_explicit_override_beats_rephrasings():
|
||||
assert rendered["messages"][0]["content"] == "explicit override wins"
|
||||
|
||||
|
||||
def test_flow_only_low_level_recipe_renders_without_target():
|
||||
"""Regression: a flow-only ``low_level`` recipe has no ``target`` turn —
|
||||
its supervision is the action-expert flow loss, not text-CE. It must
|
||||
still render (not ``None``), otherwise every blend draw of it is dropped
|
||||
and the action expert never receives a flow loss."""
|
||||
recipe = TrainingRecipe(
|
||||
messages=[
|
||||
MessageTurn(
|
||||
role="user",
|
||||
content="${subtask}",
|
||||
stream="low_level",
|
||||
if_present="subtask",
|
||||
),
|
||||
],
|
||||
bindings={"subtask": "active_at(t, style=subtask)"},
|
||||
)
|
||||
|
||||
rendered = render_sample(
|
||||
recipe=recipe,
|
||||
persistent=PERSISTENT,
|
||||
events=[],
|
||||
t=0.5,
|
||||
sample_idx=0,
|
||||
task="clean kitchen",
|
||||
)
|
||||
|
||||
assert rendered is not None
|
||||
assert rendered["messages"] == [{"role": "user", "content": "subtask 0"}]
|
||||
assert rendered["message_streams"] == ["low_level"]
|
||||
assert rendered["target_message_indices"] == []
|
||||
|
||||
|
||||
def test_vqa_frame_is_consumed_over_the_weighted_blend():
|
||||
"""A frame carrying a VQA annotation renders the ``ask_vqa*`` sub-recipe
|
||||
even when its blend weight is tiny — VQA annotations are sparse and must
|
||||
never be wasted on a subtask/action draw."""
|
||||
recipe = TrainingRecipe(
|
||||
blend={
|
||||
"high_level_subtask": TrainingRecipe(
|
||||
weight=0.99,
|
||||
messages=[
|
||||
MessageTurn(role="user", content="${task}", stream="high_level"),
|
||||
MessageTurn(role="assistant", content="a subtask", stream="high_level", target=True),
|
||||
],
|
||||
),
|
||||
"ask_vqa_top": TrainingRecipe(
|
||||
weight=0.01,
|
||||
bindings={
|
||||
"vqa_query": "emitted_at(t, style=vqa, role=user, camera=observation.images.top)",
|
||||
"vqa": "emitted_at(t, style=vqa, role=assistant, camera=observation.images.top)",
|
||||
},
|
||||
messages=[
|
||||
MessageTurn(
|
||||
role="user", content="${vqa_query}", stream="high_level", if_present="vqa_query"
|
||||
),
|
||||
MessageTurn(
|
||||
role="assistant",
|
||||
content="${vqa}",
|
||||
stream="high_level",
|
||||
target=True,
|
||||
if_present="vqa",
|
||||
),
|
||||
],
|
||||
),
|
||||
}
|
||||
)
|
||||
# A frame WITH a vqa event renders VQA on every sample_idx, despite the
|
||||
# ask_vqa weight being only 0.01.
|
||||
for sample_idx in range(20):
|
||||
rendered = render_sample(
|
||||
recipe=recipe, persistent=PERSISTENT, events=EVENTS_AT_1, t=1.0, sample_idx=sample_idx, task="x"
|
||||
)
|
||||
assert rendered["messages"][-1]["content"] == '{"count": 2}', sample_idx
|
||||
# A frame WITHOUT a vqa event falls back to the normal weighted blend.
|
||||
rendered = render_sample(recipe=recipe, persistent=PERSISTENT, events=[], t=1.0, sample_idx=0, task="x")
|
||||
assert rendered["messages"][-1]["content"] == "a subtask"
|
||||
|
||||
|
||||
def test_emitted_at_persistent_tolerates_small_timestamp_drift():
|
||||
"""Persistent ``emitted_at`` should match within EMITTED_AT_TOLERANCE_S
|
||||
so callers that derive ``t`` arithmetically (``frame_idx / fps``) still
|
||||
|
||||
@@ -25,7 +25,7 @@ from datasets import Dataset # noqa: E402
|
||||
from lerobot.datasets.io_utils import (
|
||||
hf_transform_to_torch,
|
||||
)
|
||||
from lerobot.datasets.sampler import EpisodeAwareSampler, compute_sampler_state
|
||||
from lerobot.datasets.sampler import EpisodeAwareSampler
|
||||
|
||||
|
||||
def calculate_episode_data_index(hf_dataset: Dataset) -> dict[str, torch.Tensor]:
|
||||
@@ -154,6 +154,8 @@ def test_partial_episode_drop_warns(caplog):
|
||||
|
||||
# --- seeded (seed, epoch) shuffling, resume, and state ---
|
||||
|
||||
from lerobot.datasets.sampler import compute_sampler_state # noqa: E402
|
||||
|
||||
EPISODE_BOUNDS = ([0, 2, 3], [2, 3, 6]) # episodes of 2, 1 and 3 frames
|
||||
|
||||
|
||||
|
||||
@@ -1,48 +0,0 @@
|
||||
#!/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.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import numpy as np
|
||||
|
||||
from lerobot.envs.robocasa import RoboCasaEnv, convert_action
|
||||
|
||||
|
||||
def test_robocasa_action_uses_openpi_checkpoint_order():
|
||||
action = np.arange(12, dtype=np.float32)
|
||||
|
||||
converted = convert_action(action)
|
||||
|
||||
np.testing.assert_array_equal(converted["action.end_effector_position"], [0, 1, 2])
|
||||
np.testing.assert_array_equal(converted["action.end_effector_rotation"], [3, 4, 5])
|
||||
np.testing.assert_array_equal(converted["action.gripper_close"], [6])
|
||||
np.testing.assert_array_equal(converted["action.base_motion"], [7, 8, 9, 10])
|
||||
np.testing.assert_array_equal(converted["action.control_mode"], [11])
|
||||
|
||||
|
||||
def test_robocasa_state_uses_openpi_checkpoint_order():
|
||||
env = object.__new__(RoboCasaEnv)
|
||||
env.obs_type = "pixels_agent_pos"
|
||||
env.camera_name = []
|
||||
raw_observation = {
|
||||
"state.end_effector_position_relative": np.arange(0, 3),
|
||||
"state.end_effector_rotation_relative": np.arange(3, 7),
|
||||
"state.base_position": np.arange(7, 10),
|
||||
"state.base_rotation": np.arange(10, 14),
|
||||
"state.gripper_qpos": np.arange(14, 16),
|
||||
}
|
||||
|
||||
observation = env._format_raw_obs(raw_observation)
|
||||
|
||||
np.testing.assert_array_equal(observation["agent_pos"], np.arange(16, dtype=np.float32))
|
||||
@@ -26,17 +26,8 @@ import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import torch
|
||||
from safetensors.torch import save_file
|
||||
|
||||
from lerobot.configs import PipelineFeatureType, PolicyFeature
|
||||
from lerobot.processor.pipeline import (
|
||||
DataProcessorPipeline,
|
||||
ProcessorMigrationError,
|
||||
ProcessorStep,
|
||||
ProcessorStepRegistry,
|
||||
)
|
||||
from lerobot.types import EnvTransition
|
||||
from lerobot.processor.pipeline import DataProcessorPipeline, ProcessorMigrationError
|
||||
|
||||
# Simplified Config Loading Tests
|
||||
|
||||
@@ -107,140 +98,6 @@ def test_load_config_nonexistent_path_tries_hub():
|
||||
DataProcessorPipeline._load_config("nonexistent/path", "processor.json", {})
|
||||
|
||||
|
||||
def test_from_pretrained_local_directory_missing_state_does_not_call_hub(monkeypatch):
|
||||
"""Local processor dirs must fail locally when a state file is missing."""
|
||||
|
||||
@ProcessorStepRegistry.register("local_missing_state_step")
|
||||
class LocalMissingStateStep(ProcessorStep):
|
||||
def __call__(self, transition: EnvTransition) -> EnvTransition:
|
||||
return transition
|
||||
|
||||
def transform_features(
|
||||
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
|
||||
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
|
||||
return features
|
||||
|
||||
def load_state_dict(self, state: dict[str, torch.Tensor]) -> None:
|
||||
pass
|
||||
|
||||
try:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
tmp_path = Path(tmp_dir)
|
||||
config = {
|
||||
"name": "LocalMissingStatePipeline",
|
||||
"steps": [{"registry_name": "local_missing_state_step", "state_file": "missing.safetensors"}],
|
||||
}
|
||||
(tmp_path / "processor.json").write_text(json.dumps(config))
|
||||
|
||||
def fail_hub_download(*args, **kwargs):
|
||||
pytest.fail("local missing processor state should not call hf_hub_download")
|
||||
|
||||
monkeypatch.setattr("lerobot.processor.pipeline.hf_hub_download", fail_hub_download)
|
||||
|
||||
with pytest.raises(FileNotFoundError, match="missing.safetensors.*local processor pipeline"):
|
||||
DataProcessorPipeline.from_pretrained(tmp_path, config_filename="processor.json")
|
||||
finally:
|
||||
ProcessorStepRegistry.unregister("local_missing_state_step")
|
||||
|
||||
|
||||
def test_from_pretrained_local_config_file_missing_state_does_not_call_hub(monkeypatch):
|
||||
"""Local single-file processor configs must also keep missing state resolution local."""
|
||||
|
||||
@ProcessorStepRegistry.register("local_file_missing_state_step")
|
||||
class LocalFileMissingStateStep(ProcessorStep):
|
||||
def __call__(self, transition: EnvTransition) -> EnvTransition:
|
||||
return transition
|
||||
|
||||
def transform_features(
|
||||
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
|
||||
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
|
||||
return features
|
||||
|
||||
def load_state_dict(self, state: dict[str, torch.Tensor]) -> None:
|
||||
pass
|
||||
|
||||
try:
|
||||
with tempfile.TemporaryDirectory() as tmp_dir:
|
||||
tmp_path = Path(tmp_dir)
|
||||
config_path = tmp_path / "processor.json"
|
||||
config = {
|
||||
"name": "LocalFileMissingStatePipeline",
|
||||
"steps": [
|
||||
{"registry_name": "local_file_missing_state_step", "state_file": "missing.safetensors"}
|
||||
],
|
||||
}
|
||||
config_path.write_text(json.dumps(config))
|
||||
|
||||
def fail_hub_download(*args, **kwargs):
|
||||
pytest.fail("local missing processor state should not call hf_hub_download")
|
||||
|
||||
monkeypatch.setattr("lerobot.processor.pipeline.hf_hub_download", fail_hub_download)
|
||||
|
||||
with pytest.raises(FileNotFoundError, match="missing.safetensors.*local processor pipeline"):
|
||||
DataProcessorPipeline.from_pretrained(config_path, config_filename="ignored.json")
|
||||
finally:
|
||||
ProcessorStepRegistry.unregister("local_file_missing_state_step")
|
||||
|
||||
|
||||
def test_from_pretrained_hub_source_missing_local_state_still_calls_hub(monkeypatch, tmp_path):
|
||||
"""Hub sources still fall back to hf_hub_download for state files."""
|
||||
|
||||
@ProcessorStepRegistry.register("hub_state_step")
|
||||
class HubStateStep(ProcessorStep):
|
||||
def __init__(self):
|
||||
self.value = torch.tensor(0)
|
||||
|
||||
def __call__(self, transition: EnvTransition) -> EnvTransition:
|
||||
return transition
|
||||
|
||||
def transform_features(
|
||||
self, features: dict[PipelineFeatureType, dict[str, PolicyFeature]]
|
||||
) -> dict[PipelineFeatureType, dict[str, PolicyFeature]]:
|
||||
return features
|
||||
|
||||
def load_state_dict(self, state: dict[str, torch.Tensor]) -> None:
|
||||
self.value = state["value"]
|
||||
|
||||
try:
|
||||
state_path = tmp_path / "downloaded.safetensors"
|
||||
save_file({"value": torch.tensor(7)}, state_path)
|
||||
loaded_config = {
|
||||
"name": "HubStatePipeline",
|
||||
"steps": [{"registry_name": "hub_state_step", "state_file": "hub_state.safetensors"}],
|
||||
}
|
||||
calls = []
|
||||
|
||||
def fake_load_config(cls, model_id, config_filename, hub_download_kwargs):
|
||||
return loaded_config, tmp_path / "hub_cache"
|
||||
|
||||
def fake_hub_download(**kwargs):
|
||||
calls.append(kwargs)
|
||||
return str(state_path)
|
||||
|
||||
monkeypatch.setattr(DataProcessorPipeline, "_load_config", classmethod(fake_load_config))
|
||||
monkeypatch.setattr("lerobot.processor.pipeline.hf_hub_download", fake_hub_download)
|
||||
|
||||
pipeline = DataProcessorPipeline.from_pretrained("user/repo", config_filename="processor.json")
|
||||
|
||||
assert calls == [
|
||||
{
|
||||
"repo_id": "user/repo",
|
||||
"filename": "hub_state.safetensors",
|
||||
"repo_type": "model",
|
||||
"force_download": False,
|
||||
"resume_download": None,
|
||||
"proxies": None,
|
||||
"token": None,
|
||||
"cache_dir": None,
|
||||
"local_files_only": False,
|
||||
"revision": None,
|
||||
}
|
||||
]
|
||||
assert pipeline.steps[0].value.item() == 7
|
||||
finally:
|
||||
ProcessorStepRegistry.unregister("hub_state_step")
|
||||
|
||||
|
||||
# Config Validation Tests
|
||||
|
||||
|
||||
|
||||
@@ -12,9 +12,7 @@ from lerobot.processor.render_messages_processor import RenderMessagesStep # no
|
||||
from lerobot.types import TransitionKey # noqa: E402
|
||||
|
||||
|
||||
def test_render_messages_step_renders_task_fallback_without_language_columns():
|
||||
"""No language columns + a task string → low-level task fallback render,
|
||||
matching what the policy sees at eval time on unannotated observations."""
|
||||
def test_render_messages_step_noops_without_language_columns():
|
||||
recipe = TrainingRecipe(
|
||||
messages=[
|
||||
MessageTurn(role="user", content="${task}", stream="high_level"),
|
||||
@@ -23,24 +21,6 @@ def test_render_messages_step_renders_task_fallback_without_language_columns():
|
||||
)
|
||||
transition = create_transition(complementary_data={"task": "do it"})
|
||||
|
||||
out = RenderMessagesStep(recipe)(transition)
|
||||
data = out[TransitionKey.COMPLEMENTARY_DATA]
|
||||
|
||||
assert data["messages"] == [{"role": "user", "content": "do it"}]
|
||||
assert data["message_streams"] == ["low_level"]
|
||||
assert data["target_message_indices"] == []
|
||||
assert data["task"] == "do it"
|
||||
|
||||
|
||||
def test_render_messages_step_noops_without_language_columns_or_task():
|
||||
recipe = TrainingRecipe(
|
||||
messages=[
|
||||
MessageTurn(role="user", content="${task}", stream="high_level"),
|
||||
MessageTurn(role="assistant", content="${subtask}", stream="low_level", target=True),
|
||||
]
|
||||
)
|
||||
transition = create_transition(complementary_data={})
|
||||
|
||||
assert RenderMessagesStep(recipe)(transition) == transition
|
||||
|
||||
|
||||
@@ -78,70 +58,3 @@ def test_render_messages_step_renders_and_drops_raw_language():
|
||||
assert data["messages"][-1]["content"] == "reach carefully"
|
||||
assert data["message_streams"] == ["high_level", "low_level"]
|
||||
assert data["target_message_indices"] == [1]
|
||||
|
||||
|
||||
def test_render_messages_step_falls_back_to_low_level_task_when_recipe_misses():
|
||||
recipe = TrainingRecipe(
|
||||
messages=[
|
||||
MessageTurn(
|
||||
role="assistant",
|
||||
content="${subtask}",
|
||||
stream="high_level",
|
||||
target=True,
|
||||
if_present="subtask",
|
||||
),
|
||||
]
|
||||
)
|
||||
transition = create_transition(
|
||||
complementary_data={
|
||||
"task": "pick the cube",
|
||||
"timestamp": torch.tensor(0.0),
|
||||
"index": torch.tensor(7),
|
||||
"language_persistent": [],
|
||||
"language_events": [{"style": "unmatched", "timestamp": 0.0}],
|
||||
}
|
||||
)
|
||||
|
||||
out = RenderMessagesStep(recipe)(transition)
|
||||
data = out[TransitionKey.COMPLEMENTARY_DATA]
|
||||
|
||||
assert data["messages"] == [{"role": "user", "content": "pick the cube"}]
|
||||
assert data["message_streams"] == ["low_level"]
|
||||
assert data["target_message_indices"] == []
|
||||
|
||||
|
||||
def test_render_messages_step_falls_back_per_sample_in_batched_language():
|
||||
recipe = TrainingRecipe(
|
||||
messages=[
|
||||
MessageTurn(
|
||||
role="assistant",
|
||||
content="${subtask}",
|
||||
stream="high_level",
|
||||
target=True,
|
||||
if_present="subtask",
|
||||
),
|
||||
]
|
||||
)
|
||||
transition = create_transition(
|
||||
action=torch.arange(4).reshape(2, 2),
|
||||
complementary_data={
|
||||
"task": ["pick the cube", "open the drawer"],
|
||||
"timestamp": torch.tensor([0.0, 1.0]),
|
||||
"index": torch.tensor([7, 8]),
|
||||
"language_persistent": [[], []],
|
||||
"language_events": [
|
||||
[{"style": "unmatched", "timestamp": 0.0}],
|
||||
[{"style": "unmatched", "timestamp": 1.0}],
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
out = RenderMessagesStep(recipe)(transition)
|
||||
data = out[TransitionKey.COMPLEMENTARY_DATA]
|
||||
|
||||
assert data["messages"] == [
|
||||
[{"role": "user", "content": "pick the cube"}],
|
||||
[{"role": "user", "content": "open the drawer"}],
|
||||
]
|
||||
assert data["message_streams"] == [["low_level"], ["low_level"]]
|
||||
assert data["target_message_indices"] == [[], []]
|
||||
|
||||
@@ -25,7 +25,7 @@ import pytest
|
||||
import torch
|
||||
|
||||
from lerobot.configs.types import FeatureType, PipelineFeatureType, PolicyFeature
|
||||
from lerobot.processor import ActionTokenizerProcessorStep, DataProcessorPipeline, TokenizerProcessorStep
|
||||
from lerobot.processor import DataProcessorPipeline, TokenizerProcessorStep
|
||||
from lerobot.processor.converters import create_transition, identity_transition
|
||||
from lerobot.types import TransitionKey
|
||||
from lerobot.utils.constants import (
|
||||
@@ -88,46 +88,6 @@ class MockTokenizer:
|
||||
return result
|
||||
|
||||
|
||||
def test_action_tokenizer_config_preserves_token_mapping():
|
||||
processor = object.__new__(ActionTokenizerProcessorStep)
|
||||
processor.trust_remote_code = True
|
||||
processor.max_action_tokens = 384
|
||||
processor.fast_skip_tokens = 64
|
||||
processor.paligemma_tokenizer_name = "custom/paligemma"
|
||||
processor.allow_truncation = False
|
||||
processor.action_tokenizer_name = "custom/fast"
|
||||
processor.action_tokenizer_input_object = None
|
||||
|
||||
assert processor.get_config() == {
|
||||
"trust_remote_code": True,
|
||||
"max_action_tokens": 384,
|
||||
"fast_skip_tokens": 64,
|
||||
"paligemma_tokenizer_name": "custom/paligemma",
|
||||
"allow_truncation": False,
|
||||
"action_tokenizer_name": "custom/fast",
|
||||
}
|
||||
|
||||
|
||||
def test_action_tokenizer_can_reject_truncated_sequences():
|
||||
processor = object.__new__(ActionTokenizerProcessorStep)
|
||||
processor.max_action_tokens = 4
|
||||
processor.fast_skip_tokens = 128
|
||||
processor.allow_truncation = False
|
||||
processor.action_tokenizer = lambda _actions: [1, 2, 3]
|
||||
processor._paligemma_tokenizer = type(
|
||||
"Tokenizer",
|
||||
(),
|
||||
{
|
||||
"vocab_size": 1000,
|
||||
"bos_token_id": 2,
|
||||
"encode": lambda _self, text, **_kwargs: [10, 11] if text == "Action: " else [12, 1],
|
||||
},
|
||||
)()
|
||||
|
||||
with pytest.raises(ValueError, match="max_action_tokens=4"):
|
||||
processor._tokenize_action(torch.zeros(1, 2, 1))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_tokenizer():
|
||||
"""Provide a mock tokenizer for testing."""
|
||||
|
||||
@@ -1,105 +0,0 @@
|
||||
# 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.
|
||||
# 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 lerobot.runtime import RuntimeState
|
||||
from lerobot.runtime.adapter import (
|
||||
BaseLanguageAdapter,
|
||||
DirectTaskPolicyAdapter,
|
||||
GenerationConfig,
|
||||
)
|
||||
|
||||
|
||||
class ScriptedAdapter(BaseLanguageAdapter):
|
||||
"""Base adapter whose text generation returns queued strings per kind."""
|
||||
|
||||
def __init__(self, scripts, gen=None):
|
||||
super().__init__(policy=object(), gen=gen)
|
||||
self.scripts = {k: list(v) for k, v in scripts.items()}
|
||||
self.calls = []
|
||||
|
||||
def select_action(self, observation, state):
|
||||
return None
|
||||
|
||||
def generate_text(self, kind, observation, state, user_text=None):
|
||||
self.calls.append(kind)
|
||||
queue = self.scripts.get(kind, [])
|
||||
return queue.pop(0) if queue else ""
|
||||
|
||||
|
||||
def test_cascade_sets_subtask_then_memory():
|
||||
adapter = ScriptedAdapter({"subtask": ["pick the red cup"], "memory": ["the cup is grasped"]})
|
||||
state = RuntimeState(task="clean")
|
||||
|
||||
adapter.update_language_state(None, state)
|
||||
|
||||
assert state.language_context["subtask"] == "pick the red cup"
|
||||
assert state.language_context["memory"] == "the cup is grasped"
|
||||
assert adapter.calls == ["subtask", "memory"]
|
||||
|
||||
|
||||
def test_nonempty_generation_is_used_verbatim():
|
||||
adapter = ScriptedAdapter({"subtask": [":::: ::"], "memory": ["memory"]})
|
||||
state = RuntimeState(task="clean")
|
||||
|
||||
adapter.update_language_state(None, state)
|
||||
|
||||
assert state.language_context["subtask"] == ":::: ::"
|
||||
assert state.language_context["memory"] == "memory"
|
||||
assert adapter.calls == ["subtask", "memory"]
|
||||
|
||||
|
||||
def test_throttle_regenerates_every_n_chunks():
|
||||
adapter = ScriptedAdapter(
|
||||
{
|
||||
"subtask": ["pick the first cup", "pick the second cup"],
|
||||
"memory": ["memory one two three", "memory four five six"],
|
||||
},
|
||||
gen=GenerationConfig(chunks_per_regen=2),
|
||||
)
|
||||
state = RuntimeState(task="clean")
|
||||
|
||||
adapter.update_language_state(None, state) # generates
|
||||
assert state.language_context["subtask"] == "pick the first cup"
|
||||
adapter.update_language_state(None, state) # throttled — no generation
|
||||
assert state.language_context["subtask"] == "pick the first cup"
|
||||
adapter.update_language_state(None, state) # generates again
|
||||
assert state.language_context["subtask"] == "pick the second cup"
|
||||
|
||||
|
||||
def test_handle_interjection_sets_plan_and_strips_say():
|
||||
adapter = ScriptedAdapter({"interjection": ["turn to the left now <say>heading left</say>"]})
|
||||
state = RuntimeState(task="clean")
|
||||
|
||||
adapter.handle_interjection("turn", None, state)
|
||||
|
||||
assert state.language_context["plan"] == "turn to the left now"
|
||||
|
||||
|
||||
def test_direct_task_adapter_delegates_action_chunk():
|
||||
class Policy:
|
||||
def predict_action_chunk(self, observation):
|
||||
return ("chunk", observation)
|
||||
|
||||
observation = {"task": "pick up the cube"}
|
||||
adapter = DirectTaskPolicyAdapter(Policy())
|
||||
|
||||
assert adapter.select_action(observation, RuntimeState()) == ("chunk", observation)
|
||||
assert adapter.generate_text("subtask", observation, RuntimeState()) == ""
|
||||
|
||||
|
||||
def test_flat_policy_registry_reuses_direct_task_adapter():
|
||||
from lerobot.runtime.registry import get_language_adapter_factory
|
||||
|
||||
assert get_language_adapter_factory("pi05") is DirectTaskPolicyAdapter
|
||||
assert get_language_adapter_factory("molmoact2") is DirectTaskPolicyAdapter
|
||||
@@ -1,75 +0,0 @@
|
||||
# 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.
|
||||
# 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 types import SimpleNamespace
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
import torch
|
||||
|
||||
from lerobot.runtime.cli import _build_rollout_runtime_io, _parse_args
|
||||
|
||||
|
||||
def test_parse_args_preserves_rollout_robot_overrides():
|
||||
args = _parse_args(
|
||||
[
|
||||
"--policy.path=checkpoint",
|
||||
"--robot.type=so101_follower",
|
||||
"--robot.calibration_dir=/tmp/calibration",
|
||||
]
|
||||
)
|
||||
|
||||
assert args.robot_type == "so101_follower"
|
||||
assert "--robot.calibration_dir=/tmp/calibration" in args.raw_argv
|
||||
|
||||
|
||||
def test_parse_args_rejects_removed_dataset_replay_flags():
|
||||
with pytest.raises(SystemExit):
|
||||
_parse_args(["--policy.path=checkpoint", "--dataset.repo_id=dataset"])
|
||||
|
||||
|
||||
def test_rollout_runtime_io_uses_context_processors():
|
||||
robot = MagicMock()
|
||||
robot.robot_type = "mock_robot"
|
||||
robot.cameras = {}
|
||||
robot.get_observation.return_value = {"joint.pos": 1.5}
|
||||
ctx = SimpleNamespace(
|
||||
hardware=SimpleNamespace(robot_wrapper=robot),
|
||||
runtime=SimpleNamespace(cfg=SimpleNamespace(device="cpu")),
|
||||
processors=SimpleNamespace(
|
||||
robot_observation_processor=lambda observation: observation,
|
||||
robot_action_processor=lambda pair: pair[0],
|
||||
),
|
||||
policy=SimpleNamespace(
|
||||
preprocessor=lambda observation: observation,
|
||||
postprocessor=lambda action: action,
|
||||
),
|
||||
data=SimpleNamespace(
|
||||
dataset_features={
|
||||
"observation.state": {
|
||||
"dtype": "float32",
|
||||
"shape": (1,),
|
||||
"names": ["joint.pos"],
|
||||
},
|
||||
"action": {"dtype": "float32", "shape": (1,), "names": ["joint.pos"]},
|
||||
}
|
||||
),
|
||||
)
|
||||
provider, executor = _build_rollout_runtime_io(ctx, rerun_log=False, get_task=lambda: "move")
|
||||
|
||||
observation = provider()
|
||||
executor(torch.tensor([[2.0]]))
|
||||
|
||||
assert observation["observation.state"].shape == (1, 1)
|
||||
robot.send_action.assert_called_once_with({"joint.pos": 2.0})
|
||||
@@ -1,100 +0,0 @@
|
||||
# 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.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import threading
|
||||
import time
|
||||
|
||||
from lerobot.runtime import LanguageConditionedRuntime, Tick
|
||||
|
||||
|
||||
class FakeAdapter:
|
||||
def __init__(self):
|
||||
self.updated = False
|
||||
self.interjections = []
|
||||
|
||||
def select_action(self, observation, state):
|
||||
assert observation == {"observation.state": 1}
|
||||
assert state.task == "clean"
|
||||
return ["a0", "a1"]
|
||||
|
||||
def update_language_state(self, observation, state):
|
||||
self.updated = True
|
||||
state.set_context("subtask", "pick cup", label="subtask")
|
||||
|
||||
def handle_interjection(self, user_text, observation, state):
|
||||
self.interjections.append(user_text)
|
||||
state.set_context("plan", "new plan", label="plan")
|
||||
|
||||
|
||||
def test_runtime_tick_updates_language_enqueues_and_dispatches_action():
|
||||
adapter = FakeAdapter()
|
||||
executed = []
|
||||
runtime = LanguageConditionedRuntime(
|
||||
policy_adapter=adapter,
|
||||
observation_provider=lambda: {"observation.state": 1},
|
||||
action_executor=executed.append,
|
||||
)
|
||||
runtime.set_task("clean")
|
||||
|
||||
logs = runtime.step_once()
|
||||
|
||||
assert adapter.updated
|
||||
assert runtime.state.language_context["subtask"] == "pick cup"
|
||||
assert executed == ["a0"]
|
||||
assert list(runtime.state.action_queue) == ["a1"]
|
||||
assert " subtask: pick cup" in logs
|
||||
|
||||
|
||||
def test_runtime_handles_user_interjection():
|
||||
adapter = FakeAdapter()
|
||||
runtime = LanguageConditionedRuntime(
|
||||
policy_adapter=adapter,
|
||||
observation_provider=lambda: {"observation.state": 1},
|
||||
)
|
||||
runtime.set_task("clean")
|
||||
runtime.state.extra["recent_interjection"] = "please say ok"
|
||||
runtime.state.emit("user_interjection")
|
||||
|
||||
runtime.step_once()
|
||||
|
||||
assert "please say ok" in adapter.interjections
|
||||
assert runtime.state.language_context["plan"] == "new plan"
|
||||
|
||||
|
||||
def test_prompt_change_discards_in_flight_action_chunk():
|
||||
started = threading.Event()
|
||||
release = threading.Event()
|
||||
|
||||
class BlockingAdapter(FakeAdapter):
|
||||
def select_action(self, observation, state):
|
||||
started.set()
|
||||
assert release.wait(timeout=2)
|
||||
return ["stale"]
|
||||
|
||||
runtime = LanguageConditionedRuntime(
|
||||
policy_adapter=BlockingAdapter(),
|
||||
observation_provider=lambda: {"observation.state": 1},
|
||||
)
|
||||
runtime.set_task("old task")
|
||||
runtime.state.tick = Tick(index=1, monotonic_seconds=time.monotonic())
|
||||
inference = threading.Thread(target=runtime.maybe_enqueue_action_chunk, kwargs={"force": True})
|
||||
inference.start()
|
||||
assert started.wait(timeout=2)
|
||||
|
||||
runtime.set_task("new task")
|
||||
release.set()
|
||||
inference.join(timeout=2)
|
||||
|
||||
assert not inference.is_alive()
|
||||
assert list(runtime.state.action_queue) == []
|
||||
@@ -1,82 +0,0 @@
|
||||
# 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.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import sys
|
||||
from types import SimpleNamespace
|
||||
|
||||
import numpy as np
|
||||
|
||||
from lerobot.runtime.sim_robocasa import RoboCasaSimBackend
|
||||
from lerobot.utils.video_annotation import annotate_frame
|
||||
|
||||
|
||||
def test_overlay_draws_each_label_once(monkeypatch):
|
||||
put_text_calls = []
|
||||
rectangle_calls = []
|
||||
|
||||
def put_text(image, text, origin, font, scale, color, thickness, line_type):
|
||||
put_text_calls.append((text, color, thickness))
|
||||
return image
|
||||
|
||||
def rectangle(image, start, end, color, thickness):
|
||||
rectangle_calls.append((start, end, color, thickness))
|
||||
return image
|
||||
|
||||
def add_weighted(src1, alpha, src2, beta, gamma, *, dst):
|
||||
dst[:] = src1 * alpha + src2 * beta + gamma
|
||||
return dst
|
||||
|
||||
fake_cv2 = SimpleNamespace(
|
||||
FONT_HERSHEY_SIMPLEX=0,
|
||||
LINE_AA=16,
|
||||
getTextSize=lambda text, font, scale, thickness: ((len(text) * 7, 10), 0),
|
||||
putText=put_text,
|
||||
rectangle=rectangle,
|
||||
addWeighted=add_weighted,
|
||||
)
|
||||
monkeypatch.setitem(sys.modules, "cv2", fake_cv2)
|
||||
|
||||
frame = np.full((120, 480, 3), 200, dtype=np.uint8)
|
||||
annotated = annotate_frame(
|
||||
frame,
|
||||
(("Task", "close the fridge"), ("Subtask", "reach for the handle"), ("Memory", None)),
|
||||
)
|
||||
|
||||
assert [call[0] for call in put_text_calls] == [
|
||||
"Task: close the fridge",
|
||||
"Subtask: reach for the handle",
|
||||
]
|
||||
assert all(color == (255, 255, 255) and thickness == 1 for _, color, thickness in put_text_calls)
|
||||
assert len(rectangle_calls) == 1
|
||||
assert not np.shares_memory(annotated, frame)
|
||||
|
||||
|
||||
def test_capture_updates_live_frame_when_recording_is_disabled(monkeypatch):
|
||||
backend = object.__new__(RoboCasaSimBackend)
|
||||
frame = np.full((8, 8, 3), 42, dtype=np.uint8)
|
||||
written = []
|
||||
backend.record = False
|
||||
backend.runtime_state = None
|
||||
backend._multiview_frame = lambda: frame
|
||||
backend._current_task = lambda: "task"
|
||||
backend._subtask_getter = None
|
||||
backend._memory_getter = None
|
||||
backend._latest_frame = None
|
||||
backend._write_live_frame = written.append
|
||||
monkeypatch.setattr("lerobot.runtime.sim_robocasa.annotate_frame", lambda image, labels: image)
|
||||
|
||||
backend._capture_frame()
|
||||
|
||||
assert backend._latest_frame is frame
|
||||
assert written == [frame]
|
||||
@@ -17,8 +17,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import sys
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
@@ -108,109 +106,6 @@ def test_sentry_config_defaults():
|
||||
assert cfg.target_video_file_size_mb is None
|
||||
|
||||
|
||||
def test_rollout_config_passes_policy_pretrained_revision(monkeypatch):
|
||||
from lerobot.configs import PreTrainedConfig, parser
|
||||
from lerobot.rollout import RolloutConfig
|
||||
from tests.mocks.mock_robot import MockRobotConfig
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_from_pretrained(cls, pretrained_name_or_path, **kwargs):
|
||||
captured["pretrained_name_or_path"] = pretrained_name_or_path
|
||||
captured.update(kwargs)
|
||||
return SimpleNamespace(device="cpu", pretrained_revision=kwargs["revision"])
|
||||
|
||||
monkeypatch.setattr(parser, "get_yaml_overrides", lambda _: ["--pretrained_revision=yaml-sha"])
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
["lerobot-rollout", "--policy.path=user/policy", "--policy.pretrained_revision=cli-sha"],
|
||||
)
|
||||
monkeypatch.setattr(PreTrainedConfig, "from_pretrained", classmethod(fake_from_pretrained))
|
||||
|
||||
cfg = RolloutConfig(robot=MockRobotConfig())
|
||||
|
||||
assert captured["pretrained_name_or_path"] == "user/policy"
|
||||
assert captured["revision"] == "cli-sha"
|
||||
assert captured["cli_overrides"] == [
|
||||
"--pretrained_revision=yaml-sha",
|
||||
"--pretrained_revision=cli-sha",
|
||||
]
|
||||
assert cfg.policy.pretrained_path == "user/policy"
|
||||
assert cfg.policy.pretrained_revision == "cli-sha"
|
||||
|
||||
|
||||
def test_load_pretrained_policy_passes_revision(monkeypatch):
|
||||
import lerobot.rollout.context as rollout_context
|
||||
|
||||
policy_config = SimpleNamespace(
|
||||
type="mock",
|
||||
use_peft=False,
|
||||
pretrained_path="user/policy",
|
||||
pretrained_revision="policy-sha",
|
||||
)
|
||||
policy_class = MagicMock()
|
||||
loaded_policy = MagicMock()
|
||||
policy_class.from_pretrained.return_value = loaded_policy
|
||||
monkeypatch.setattr(rollout_context, "get_policy_class", lambda _: policy_class)
|
||||
|
||||
policy = rollout_context._load_pretrained_policy(policy_config)
|
||||
|
||||
assert policy is loaded_policy
|
||||
policy_class.from_pretrained.assert_called_once_with(
|
||||
"user/policy",
|
||||
config=policy_config,
|
||||
revision="policy-sha",
|
||||
)
|
||||
|
||||
|
||||
def test_load_pretrained_peft_policy_keeps_adapter_and_base_revisions_separate(monkeypatch):
|
||||
import lerobot.rollout.context as rollout_context
|
||||
|
||||
policy_config = SimpleNamespace(
|
||||
type="mock",
|
||||
use_peft=True,
|
||||
pretrained_path="user/adapter",
|
||||
pretrained_revision="adapter-sha",
|
||||
)
|
||||
policy_class = MagicMock()
|
||||
base_policy = MagicMock()
|
||||
policy_class.from_pretrained.return_value = base_policy
|
||||
monkeypatch.setattr(rollout_context, "get_policy_class", lambda _: policy_class)
|
||||
|
||||
peft_config = SimpleNamespace(
|
||||
base_model_name_or_path="user/base-policy",
|
||||
revision="base-sha",
|
||||
)
|
||||
peft_config_from_pretrained = MagicMock(return_value=peft_config)
|
||||
adapted_policy = MagicMock()
|
||||
peft_model_from_pretrained = MagicMock(return_value=adapted_policy)
|
||||
monkeypatch.setitem(
|
||||
sys.modules,
|
||||
"peft",
|
||||
SimpleNamespace(
|
||||
PeftConfig=SimpleNamespace(from_pretrained=peft_config_from_pretrained),
|
||||
PeftModel=SimpleNamespace(from_pretrained=peft_model_from_pretrained),
|
||||
),
|
||||
)
|
||||
|
||||
policy = rollout_context._load_pretrained_policy(policy_config)
|
||||
|
||||
assert policy is adapted_policy
|
||||
peft_config_from_pretrained.assert_called_once_with("user/adapter", revision="adapter-sha")
|
||||
policy_class.from_pretrained.assert_called_once_with(
|
||||
pretrained_name_or_path="user/base-policy",
|
||||
config=policy_config,
|
||||
revision="base-sha",
|
||||
)
|
||||
peft_model_from_pretrained.assert_called_once_with(
|
||||
base_policy,
|
||||
"user/adapter",
|
||||
config=peft_config,
|
||||
revision="adapter-sha",
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# RolloutRingBuffer
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user