Sail examples

Appearance

advancedView Python source

Curate humanoid robot episodes with every camera frame

Prepare real-robot demonstrations for training. This example reads all 301 episodes of a Unitree G1 humanoid stacking three blocks: 28 joints (both arms and both dexterous hands) recorded at 30 Hz, with four cameras. It checks every episode, selects whole episodes for training and validation, then decodes every frame of all four cameras for one episode (episode 279) and matches each frame to the state record taken at the same moment.

Run it

Run all commands below from the repository root (sail-examples/):

uv run multimodal/curate-humanoid-episodes/main.py

The first run downloads about 63 MB into multimodal/curate-humanoid-episodes/.data/: the state and action records of every episode, plus their metadata. It then reads episode 279's 22 seconds from each of the four camera videos. Each video file holds many episodes, so the run seeks to that episode over HTTP instead of downloading the whole file. No key, model, GPU or robot is needed.

Output

The newest capture appears first. Older output stays here when a result changes.

Current output · Sail 0.7.3

Captured on Sail 0.7.3. Latest capture: 2026-11-07, Python 3.14.10.

+----------+-----+
|     split|count|
+----------+-----+
|     train|  240|
|validation|   61|
+----------+-----+

+-------------+----+---------------+------+----------------+--------------+---------------+------+
|episode_index|task|expected_frames|frames|duplicate_frames|bad_timestamps|invalid_vectors|usable|
+-------------+----+---------------+------+----------------+--------------+---------------+------+
+-------------+----+---------------+------+----------------+--------------+---------------+------+

+---------------+--------------+--------------+----------------+----------------------+
|camera         |decoded_frames|matched_frames|distinct_records|max_time_error_seconds|
+---------------+--------------+--------------+----------------+----------------------+
|cam_left_high  |666           |666           |666             |8.900960324353946E-7  |
|cam_left_wrist |666           |666           |666             |8.900960324353946E-7  |
|cam_right_high |666           |666           |666             |8.900965440261643E-7  |
|cam_right_wrist|666           |666           |666             |8.900960892788135E-7  |
+---------------+--------------+--------------+----------------+----------------------+

+-------------+-------------+------------+-----------------------+-----------+----------------------+-------+
|camera       |episode_index|frame_number|episode_timestamp      |frame_index|time_error_seconds    |matched|
+-------------+-------------+------------+-----------------------+-----------+----------------------+-------+
|cam_left_high|279          |0           |-5.6843418860808015E-14|0          |5.6843418860808015E-14|true   |
|cam_left_high|279          |1           |0.03333333333330302    |1          |1.7384991224389523E-9 |true   |
|cam_left_high|279          |2           |0.06666666666666288    |2          |3.476941401459044E-9  |true   |
|cam_left_high|279          |3           |0.0999999999999659     |3          |1.4901502254360821E-9 |true   |
|cam_left_high|279          |4           |0.13333333333332575    |4          |6.953882802918088E-9  |true   |
+-------------+-------------+------------+-----------------------+-----------+----------------------+-------+
only showing top 5 rows

Python code

main.py323 lines
View main.py on GitHub 
# /// script
# requires-python = ">=3.12"
# dependencies = [
#   "pysail==0.7.3",
#   "pyspark-client==4.1.3",
#   "pyarrow==26.0.0",
#   "pandas==3.0.6",
#   "av==18.0.0",
#   "numpy==2.5.4",
# ]
# ///
"""Curate humanoid robot episodes and align every camera frame of one episode."""

import hashlib
import json
import os
import tempfile
import urllib.request
from pathlib import Path

import pyarrow as pa
from pyspark.sql import SparkSession, Window
from pyspark.sql import functions as F

REVISION = "57faa2cf516e008f96d91fe3b67ad53a74f012e6"
SOURCE = "https://huggingface.co/datasets/unitreerobotics/G1_Dex3_BlockStacking_Dataset"
DATA = "data/chunk-000/file-000.parquet"
EPISODES = "meta/episodes/chunk-000/file-000.parquet"
TASKS = "meta/tasks.parquet"
HASHES = {
    "meta/info.json": "3466132be194fb9487db582dd37afbb2585f924d1b722a8fc6ac30920a471927",
    DATA: "d49fd8e3e8d5d7bd682897644017c0a2542275af80f2fe0a936734d8abbd1d7b",
    EPISODES: "127bf236532cde778b197774f94e1869de6e5db7ab08aa7296baad737e1ba658",
    TASKS: "7bbcf6b738f9b336dade138404208e9ada671c50be38eb5e460ba443668330aa",
}
SIZE_LIMIT = 70_000_000  # bytes per downloaded file; the state/action Parquet is about 63 MB
CAMERAS = ("cam_left_high", "cam_right_high", "cam_left_wrist", "cam_right_wrist")
EPISODE = 279  # the episode whose camera frames are decoded
JOINTS = 28  # 14 arm joints and 14 hand joints, in state and in action
TIME_TOLERANCE = 1e-4  # seconds, accommodates this dataset's float32 timestamps
# SHA-256 over the concatenated hex SHA-256 of every decoded RGB frame, in frame order.
# The videos are read in byte ranges, so the decoded pixels are verified instead of whole files.
FRAME_DIGESTS = {
    "cam_left_high": "a844c9ceecdac821347a992562b70e36905e8b9c8594ae7607c77f4915b8f87d",
    "cam_right_high": "72f604941280f8962e623296c3bec444f55ad20e8baea0a864cd09a628af7d84",
    "cam_left_wrist": "dbf0767a9b396c1ea90a497e7b37eea224bd8b9bb586834ce077abc0083fd62a",
    "cam_right_wrist": "0f4c9b535211444b9a5df038f7070e8335ded98d19c5c7c9d9102fdd8c8154fa",
}


def dataset_file(name):
    """Fetch only allowlisted files; verify downloads and cached copies."""
    expected = HASHES[name]
    path = Path(__file__).resolve().parent / ".data" / REVISION / name
    path.parent.mkdir(parents=True, exist_ok=True)
    if not path.exists():
        temporary = None
        try:
            with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as target:
                temporary = Path(target.name)
                with urllib.request.urlopen(f"{SOURCE}/resolve/{REVISION}/{name}", timeout=60) as response:
                    size = 0
                    while chunk := response.read(1024 * 1024):
                        size += len(chunk)
                        if size > SIZE_LIMIT:
                            raise ValueError(f"download exceeds size limit: {name}")
                        target.write(chunk)
            if hashlib.sha256(temporary.read_bytes()).hexdigest() != expected:
                raise ValueError(f"SHA-256 mismatch: {name}")
            os.replace(temporary, path)
        finally:
            if temporary is not None:
                temporary.unlink(missing_ok=True)
    if hashlib.sha256(path.read_bytes()).hexdigest() != expected:
        raise ValueError(f"SHA-256 mismatch: {name}")
    return path


FRAME_COLUMNS = (
    "camera string, episode_index long, frame_number long, video_timestamp double, "
    "episode_timestamp double, width int, height int, rgb_sha256 string"
)
FRAME_SCHEMA = pa.schema(
    [
        ("camera", pa.string()),
        ("episode_index", pa.int64()),
        ("frame_number", pa.int64()),
        ("video_timestamp", pa.float64()),
        ("episode_timestamp", pa.float64()),
        ("width", pa.int32()),
        ("height", pa.int32()),
        ("rgb_sha256", pa.string()),
    ]
)


def decode_every_frame(batches):
    """Decode every frame of each camera segment and emit one small record per frame."""
    import av
    from av.video.reformatter import Interpolation

    # Pixel integrity checks must not depend on the worker's CPU acceleration path.
    conversion = Interpolation.BILINEAR | Interpolation.BITEXACT | Interpolation.ACCURATE_RND

    for batch in batches:
        for segment in batch.to_pylist():
            start, end = segment["video_start"], segment["video_end"]
            digest = hashlib.sha256()
            rows = []
            number = 0
            # The shared video file holds many episodes; seek to this one over HTTP instead of downloading it.
            with av.open(segment["video_url"], timeout=60) as container:
                stream = container.streams.video[0]
                container.seek(int(start / stream.time_base), stream=stream, backward=True)
                for frame in container.decode(stream):
                    if frame.pts is None:
                        raise ValueError("video frame has no presentation timestamp")
                    video_time = float(frame.pts * stream.time_base)
                    if video_time + 1e-6 < start:
                        continue
                    if video_time >= end - 1e-6:
                        break
                    rgb = frame.to_ndarray(format="rgb24", interpolation=conversion)
                    rgb_sha256 = hashlib.sha256(rgb.tobytes()).hexdigest()
                    digest.update(rgb_sha256.encode())
                    rows.append(
                        {
                            "camera": segment["camera"],
                            "episode_index": segment["episode_index"],
                            "frame_number": number,
                            "video_timestamp": video_time,
                            "episode_timestamp": video_time - start,
                            "width": frame.width,
                            "height": frame.height,
                            "rgb_sha256": rgb_sha256,
                        }
                    )
                    number += 1
                    if len(rows) == 256:
                        yield pa.RecordBatch.from_pylist(rows, schema=FRAME_SCHEMA)
                        rows = []
            if digest.hexdigest() != FRAME_DIGESTS[segment["camera"]]:
                raise ValueError(f"decoded frames differ from the pinned revision: {segment['camera']}")
            if rows:
                yield pa.RecordBatch.from_pylist(rows, schema=FRAME_SCHEMA)


def curate(frames, episodes, tasks, fps):
    """Return a quality report and a whole-episode split for 28-joint state and action vectors."""
    order = Window.partitionBy("episode_index").orderBy("frame_index", "timestamp")
    previous = F.lag("timestamp").over(order)
    timestamp = F.col("timestamp").cast("double")
    bad_time = (
        timestamp.isNull()
        | F.isnan(timestamp)
        | (F.abs(timestamp - F.col("frame_index") / fps) > TIME_TOLERANCE)
        | (previous.isNotNull() & (F.abs(timestamp - previous - 1 / fps) > TIME_TOLERANCE))
    )
    invalid = F.lit(False)
    for name in ("observation.state", "action"):
        vector = F.col(f"`{name}`")
        bad_value = F.exists(vector, lambda v: v.isNull() | F.isnan(v) | (F.abs(v) == float("inf")))
        invalid = invalid | vector.isNull() | (F.size(vector) != JOINTS) | bad_value
    counted = (
        frames.withColumn("bad_time", F.coalesce(bad_time, F.lit(True)).cast("long"))
        .withColumn("invalid_vector", F.coalesce(invalid, F.lit(True)).cast("long"))
        .groupBy("episode_index")
        .agg(
            F.count("*").alias("frames"),
            F.countDistinct("frame_index").alias("distinct_frames"),
            F.min("frame_index").alias("first_frame"),
            F.max("frame_index").alias("last_frame"),
            F.sum("bad_time").alias("bad_timestamps"),
            F.sum("invalid_vector").alias("invalid_vectors"),
            F.min("task_index").alias("task_index"),
            F.countDistinct("task_index").alias("task_count"),
            F.count("task_index").alias("task_frames"),
        )
    )
    report = episodes.select("episode_index", F.col("length").alias("expected_frames")).join(
        counted, "episode_index", "full"
    )
    report = report.join(tasks, "task_index", "left").fillna(
        0, subset=["frames", "distinct_frames", "bad_timestamps", "invalid_vectors", "task_count"]
    )
    usable = (
        (F.col("frames") == F.col("expected_frames"))
        & (F.col("distinct_frames") == F.col("expected_frames"))
        & (F.col("first_frame") == 0)
        & (F.col("last_frame") == F.col("expected_frames") - 1)
        & (F.col("bad_timestamps") == 0)
        & (F.col("invalid_vectors") == 0)
        & (F.col("task_count") == 1)
        & (F.col("task_frames") == F.col("frames"))
        & F.col("task").isNotNull()
    )
    quality = report.select(
        "episode_index",
        "task",
        "expected_frames",
        "frames",
        (F.col("frames") - F.col("distinct_frames")).alias("duplicate_frames"),
        "bad_timestamps",
        "invalid_vectors",
        F.coalesce(usable, F.lit(False)).alias("usable"),
    )
    selection = quality.filter("usable").select(
        "episode_index",
        "task",
        "frames",
        F.when(F.col("episode_index") % 5 == 0, "validation").otherwise("train").alias("split"),
    )
    return {"quality": quality.orderBy("episode_index"), "selection": selection.orderBy("episode_index")}


def align(frames, decoded, fps):
    """Match each decoded frame to the nearest state record within half a record period."""
    states = frames.select(
        F.col("episode_index").alias("state_episode"),
        "frame_index",
        F.col("timestamp").cast("double").alias("state_timestamp"),
    )
    error = F.abs(F.col("episode_timestamp") - F.col("state_timestamp"))
    candidates = decoded.join(
        states,
        (F.col("episode_index") == F.col("state_episode")) & (error <= 0.5 / fps + TIME_TOLERANCE),
        "left",
    ).withColumn("time_error_seconds", error)
    nearest = Window.partitionBy("camera", "episode_index", "frame_number").orderBy(
        F.col("time_error_seconds").asc_nulls_last(), F.col("frame_index").asc_nulls_last()
    )
    alignment = (
        candidates.withColumn("rank", F.row_number().over(nearest))
        .filter("rank = 1")
        .select(
            "camera",
            "episode_index",
            "frame_number",
            "episode_timestamp",
            "frame_index",
            "time_error_seconds",
            F.col("frame_index").isNotNull().alias("matched"),
        )
    )
    coverage = alignment.groupBy("camera").agg(
        F.count("*").alias("decoded_frames"),
        F.sum(F.col("matched").cast("long")).alias("matched_frames"),
        F.countDistinct("frame_index").alias("distinct_records"),
        F.max("time_error_seconds").alias("max_time_error_seconds"),
    )
    return {
        "alignment": alignment.orderBy("camera", "frame_number"),
        "coverage": coverage.orderBy("camera"),
    }


def camera_segments(episodes):
    """One row per camera: the video file holding the episode and the episode's time span in it."""
    segments = None
    for camera in CAMERAS:
        prefix = f"videos/observation.images.{camera}"
        segment = episodes.filter(F.col("episode_index") == EPISODE).select(
            F.lit(camera).alias("camera"),
            "episode_index",
            F.concat(
                F.lit(f"{SOURCE}/resolve/{REVISION}/{prefix}/chunk-"),
                F.lpad(F.col(f"`{prefix}/chunk_index`").cast("string"), 3, "0"),
                F.lit("/file-"),
                F.lpad(F.col(f"`{prefix}/file_index`").cast("string"), 3, "0"),
                F.lit(".mp4"),
            ).alias("video_url"),
            F.col(f"`{prefix}/from_timestamp`").cast("double").alias("video_start"),
            F.col(f"`{prefix}/to_timestamp`").cast("double").alias("video_end"),
        )
        segments = segment if segments is None else segments.unionByName(segment)
    return segments


def run(spark):
    info = json.loads(dataset_file("meta/info.json").read_text())
    assert info["codebase_version"] == "v3.0" and float(info["fps"]) == 30.0
    assert all(info["features"][name]["shape"] == [JOINTS] for name in ("observation.state", "action"))
    fps = float(info["fps"])
    frames = spark.read.parquet(str(dataset_file(DATA)))
    episodes = spark.read.parquet(str(dataset_file(EPISODES)))
    tasks = spark.read.parquet(str(dataset_file(TASKS))).select(
        "task_index", F.col("__index_level_0__").alias("task")
    )
    results = curate(frames, episodes, tasks, fps)

    # Materialize decoded rows once, so the outputs reuse them without decoding again.
    import atexit
    import shutil

    temporary = tempfile.mkdtemp(prefix="humanoid-frames-")
    atexit.register(shutil.rmtree, temporary, ignore_errors=True)
    segments = camera_segments(episodes)
    if segments.count() != len(CAMERAS):
        raise ValueError(f"episode {EPISODE} is missing from the episode metadata")
    segments.repartition(len(CAMERAS)).mapInArrow(decode_every_frame, FRAME_COLUMNS).write.mode(
        "overwrite"
    ).parquet(temporary)
    episode_frames = frames.filter(F.col("episode_index") == EPISODE)
    results.update(align(episode_frames, spark.read.parquet(temporary), fps))
    return results


if __name__ == "__main__":
    from pysail.spark import SparkConnectServer

    server = SparkConnectServer()
    server.start()
    _, port = server.listening_address
    spark = SparkSession.builder.remote(f"sc://localhost:{port}").getOrCreate()
    try:
        results = run(spark)
        results["selection"].groupBy("split").count().orderBy("split").show()
        results["quality"].filter(~F.col("usable")).show(truncate=False)
        results["coverage"].show(truncate=False)
        results["alignment"].show(5, truncate=False)
    finally:
        spark.stop()
        server.stop()

How it works

  1. Check every episode. Window functions compare each record's timestamp with its frame index and with the previous record, at the declared 30 Hz. Aggregations count records, distinct and missing frame indices, and state or action vectors that are missing, the wrong length or not finite. A full join with the episode metadata keeps episodes whose records are missing entirely.
  2. Select whole episodes. Only episodes that pass every check are selected. Episode IDs divisible by five go to validation and the others to training, so no episode has frames in both splits.
  3. Decode every frame. A four-row table lists, for each camera, the video file that holds the episode and the episode's start and end time inside it. mapInArrow opens each video where the task runs, seeks to the episode and decodes every frame with PyAV on CPU. It returns one small row per frame: presentation time, size and a SHA-256 of the decoded RGB pixels. Video bytes and pixel arrays never leave the task. The decoded rows are saved to temporary Parquet once, so the outputs reuse them without decoding again.
  4. Match frames to records. Each frame joins the episode's state records within half a record period (plus 0.0001 seconds for float32 timestamp rounding) and keeps the nearest one, breaking ties with the earlier record.

The outputs are:

  • quality: one row per episode with expected and actual record counts, duplicate records, bad timestamps, invalid vectors and a usable flag.
  • selection: the usable episodes with their train or validation split.
  • alignment: one row per decoded frame with the matched record and the time error. A frame with no record in range is kept, with matched false.
  • coverage: per camera, decoded frames, matched frames, distinct records matched and the largest time error.

At this revision all 301 episodes pass, the split is 240 training and 61 validation episodes, and all 666 frames of each camera match a different record of episode 279's 666.

These checks concern record structure and timing. They do not judge task success, demonstration quality or action safety, and the split is illustrative, not the dataset's own. The half-period rule suits this synchronized dataset; it does not estimate clock drift or calibrate cameras. A checksum identifies decoded pixels; it is not an image quality score. Adapt the joint count, cadence and episode choice for your own data.

For the same steps on a small simulated dataset, see curate robot episodes and align robot camera frames.

Data and verification

The G1 Dex3 block stacking dataset is published by Unitree Robotics under the Apache 2.0 license. This example pins revision 57faa2cf516e008f96d91fe3b67ad53a74f012e6. It uses the LeRobot v3 layout (format documentation), in which episode metadata locates each episode inside shared Parquet and video files.

Every downloaded file is checked against a pinned SHA-256, including cached copies. The four camera videos are 200 to 500 MB each and are read by seeking, not downloaded, so instead of hashing whole files the example pins a digest of every decoded frame of each camera and fails if the pixels differ. RGB conversion uses FFmpeg's bit-exact mode with accurate rounding so CPU acceleration does not change the checked pixels. Files are downloaded locally and the videos are read over HTTPS where each task runs; a multi-machine deployment needs the same network access, or storage reachable from its workers.