LeRobot Dataset Format Explained: Episodes, Timestamps and Validation

"""Validate a small numeric LeRobot v3-style snapshot, not full compatibility.

No videos, task-text mapping, statistics, dtype/codec conformance, or policy
compatibility checks. Read-only; all input is loaded into memory.
Copyright 2026 RoboSkin.ai. SPDX-License-Identifier: Apache-2.0
"""
from __future__ import annotations
import argparse
import json
import math
from pathlib import Path
import sys
import pyarrow.parquet as pq

SCOPE = "numeric-v3-subset; not an official LeRobot compatibility validator"
SCALARS = ("index", "episode_index", "frame_index", "task_index", "timestamp")
REQUIRED = (*SCALARS, "observation.state", "action")
TACTILE = "observation.tactile"
VALID = "observation.tactile_valid"


def finite(value):
    return type(value) in (int, float) and math.isfinite(value)


def integer(value):
    return type(value) is int and value >= 0


def validate(root: Path, tolerance: float = 1e-4) -> dict:
    """Return all detected errors. A pass applies only to the documented subset."""
    errors = []
    report = {"scope": SCOPE, "ok": False, "frames": 0, "episodes": 0,
              "unknown_tactile_values": 0, "errors": errors}
    if not finite(tolerance) or tolerance < 0:
        errors.append("config: tolerance must be finite and non-negative")
        return report
    try:
        info = json.loads((root / "meta/info.json").read_text(encoding="utf-8"))
        if not isinstance(info, dict): raise ValueError("info.json must be an object")
        features = info.get("features")
        if not isinstance(features, dict): raise ValueError("features must be an object")
        if info.get("codebase_version") != "v3.0":
            errors.append("version: this exercise supports codebase_version v3.0 only")
        fps = info.get("fps")
        if not finite(fps) or fps <= 0: raise ValueError("fps must be finite and positive")
        for field in REQUIRED:
            if field not in features: errors.append(f"schema: missing feature {field}")
        for name, spec in features.items():
            if not isinstance(spec, dict): raise ValueError(f"invalid feature specification: {name}")
            if spec.get("dtype") not in ("float32", "float64", "int64", "bool"):
                errors.append(f"unsupported: {name} dtype {spec.get('dtype')}; numeric subset only")
            shape = spec.get("shape")
            if not isinstance(shape, list) or len(shape) != 1 or not integer(shape[0]) or shape[0] == 0:
                errors.append(f"unsupported: {name} needs a positive one-dimensional shape")
            if name in SCALARS and shape != [1]: errors.append(f"schema: {name} must have shape [1]")
        if (TACTILE in features) != (VALID in features):
            errors.append("schema: example tactile values and validity must be declared together")
        if TACTILE in features and VALID in features and features[TACTILE].get("shape") != features[VALID].get("shape"):
            errors.append("schema: tactile values and validity shapes differ")
        if info.get("video_path") is not None:
            errors.append("unsupported: video layout is outside this exercise")
        files = sorted((root / "data").glob("chunk-*/file-*.parquet"))
        ep_files = sorted((root / "meta/episodes").glob("chunk-*/file-*.parquet"))
        if not files or not ep_files: raise ValueError("data and episode Parquet shards are required")
        rows, locations = [], []
        for file in files:
            table = pq.read_table(file)
            for name in set(REQUIRED) | set(features):
                if name not in table.column_names: errors.append(f"fields: {file.relative_to(root).as_posix()} missing {name}")
            chunk_rows = table.to_pylist()
            rows.extend(chunk_rows)
            locations.extend([file.resolve()] * len(chunk_rows))
        episodes = [row for file in ep_files for row in pq.read_table(file).to_pylist()]
        report.update(frames=len(rows), episodes=len(episodes))
        if not rows or not episodes: errors.append("empty: frames and episodes must be non-empty")
        for key, count in (("total_frames", len(rows)), ("total_episodes", len(episodes))):
            if not integer(info.get(key)) or info[key] != count: errors.append(f"counts: {key} does not match files")
        for i, row in enumerate(rows):
            for key in SCALARS[:-1]:
                if not integer(row.get(key)): errors.append(f"values: row {i} {key} must be a non-negative integer")
            if row.get("index") != i: errors.append(f"index: row {i} must have global index {i}")
            if not finite(row.get("timestamp")): errors.append(f"time: row {i} timestamp must be finite")
            for name, spec in features.items():
                if name in SCALARS: continue
                values = row.get(name)
                shape = spec.get("shape", [])
                if not isinstance(values, list) or len(shape) != 1 or len(values) != shape[0]:
                    errors.append(f"dimension: row {i} {name} does not match {shape}")
                    continue
                flags = row.get(VALID)
                for j, value in enumerate(values):
                    if name == TACTILE and isinstance(flags, list) and j < len(flags) and flags[j] is False:
                        report["unknown_tactile_values"] += 1
                        continue  # Unknown stays unknown; never replace it with zero.
                    dtype = spec.get("dtype")
                    good = type(value) is bool if dtype == "bool" else (type(value) is int if dtype == "int64" else finite(value))
                    if not good: errors.append(f"values: row {i} {name}[{j}] invalid or missing")
        cursor = 0
        for number, ep in enumerate(episodes):
            keys = ("episode_index", "length", "dataset_from_index", "dataset_to_index", "data/chunk_index", "data/file_index")
            if not all(integer(ep.get(key)) for key in keys):
                errors.append(f"boundary: episode {number} missing or invalid metadata")
                continue
            start, stop = ep["dataset_from_index"], ep["dataset_to_index"]
            if ep["episode_index"] != number or start != cursor or stop <= start or stop > len(rows) or stop-start != ep["length"]:
                errors.append(f"boundary: episode {number} has a gap, overlap, count or index error")
            cursor = stop
            template = info.get("data_path")
            if not isinstance(template, str): raise ValueError("data_path must be a string")
            linked = (root / template.format(chunk_index=ep["data/chunk_index"], file_index=ep["data/file_index"])).resolve()
            if not linked.is_relative_to(root.resolve()): raise ValueError("data_path escapes dataset root")
            previous = None
            for frame, i in enumerate(range(start, min(stop, len(rows)))):
                row = rows[i]
                if row.get("episode_index") != number or row.get("frame_index") != frame:
                    errors.append(f"boundary: row {i} episode/frame index disagrees with metadata")
                if locations[i] != linked: errors.append(f"path: row {i} does not match episode data shard")
                stamp = row.get("timestamp")
                if finite(stamp):
                    if previous is not None and stamp <= previous: errors.append(f"time: episode {number} timestamp is not strictly increasing at frame {frame}")
                    if abs(stamp - frame/fps) > tolerance: errors.append(f"cadence: episode {number} frame {frame} differs from frame_index/fps")
                    previous = stamp
        if cursor != len(rows): errors.append("boundary: episode metadata does not cover all rows")
    except (OSError, ValueError, TypeError, KeyError, IndexError) as exc:
        errors.append(f"input: {exc}")
    report["ok"] = not errors
    return report


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("dataset", type=Path)
    parser.add_argument("--report", type=Path, required=True)
    parser.add_argument("--tolerance", type=float, default=1e-4, help="seconds, default 0.0001")
    args = parser.parse_args()
    report = validate(args.dataset, args.tolerance)
    # Reports are an explicit output; never overwrite a file inside the input.
    if args.report.resolve().is_relative_to(args.dataset.resolve()):
        parser.error("--report must be outside the dataset directory")
    args.report.parent.mkdir(parents=True, exist_ok=True)
    args.report.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n", encoding="utf-8")
    print(f"{'PASS' if report['ok'] else 'FAIL'} (numeric subset): {report['frames']} frames, {report['episodes']} episodes, {len(report['errors'])} errors, {report['unknown_tactile_values']} unknown tactile values")
    for error in report["errors"]: print(f"- {error}")
    return 0 if report["ok"] else 1


if __name__ == "__main__": sys.exit(main())