From 08b1dedacee864c80fdbfce891bdc9f8c468931e Mon Sep 17 00:00:00 2001 From: RomirJ Date: Mon, 12 Oct 2026 02:58:39 -0700 Subject: [PATCH] Admit local LeRobot v3 datasets as training input (S-48) `tether finetune --dataset-root` now admits a raw local LeRobot v3 dataset for ACT (from scratch) and SmolVLA LoRA, not only a Studio export receipt. Admission checks info.json, features, stats, tasks, the episode index, every data row, and decoded video timestamps, then binds a SHA-256 over every retained file. Training runs with HF offline flags. - describe_lerobot_v3: metadata-only check for listings; reads meta/ only. - verify_lerobot_v3: full admission. Video frames are checked with LeRobot's own rule: query = from_timestamp + the row timestamp, nearest decoded PTS distance < tolerance_s (1e-4), compared in float32 as torch.cdist does. float64 disagreed with LeRobot on 8% of near-edge samples at long-shard timestamps. - Local configs refuse a tolerance_s override so admission and training use the same tolerance. - SmolVLA raw path binds only max_state/action_dim padding; LeRobot derives features from the dataset. - Width-1 vectors stored as scalars (LeRobot 0.5.1) are admitted. Co-Authored-By: Claude Opus 5.5 --- src/tether/finetune/cli.py | 14 +- src/tether/finetune/config.py | 8 +- src/tether/finetune/lerobot_v3.py | 494 ++++++++++++++++++ src/tether/finetune/local_dataset.py | 24 +- src/tether/finetune/preflight/dataset_size.py | 8 +- src/tether/finetune/preflight/runner.py | 11 +- src/tether/finetune/preflight/schema.py | 21 +- src/tether/finetune/run.py | 25 +- tests/test_finetune_lerobot_v3.py | 432 +++++++++++++++ tests/test_finetune_local_dataset.py | 1 + 10 files changed, 1017 insertions(+), 21 deletions(-) create mode 100644 src/tether/finetune/lerobot_v3.py create mode 100644 tests/test_finetune_lerobot_v3.py diff --git a/src/tether/finetune/cli.py b/src/tether/finetune/cli.py index e67a96c..542da64 100644 --- a/src/tether/finetune/cli.py +++ b/src/tether/finetune/cli.py @@ -8,6 +8,7 @@ """ from __future__ import annotations +import json import logging from pathlib import Path @@ -53,12 +54,14 @@ def finetune_command( ), dataset_root: str = typer.Option( "", "--dataset-root", - help="Actual EXPORT/dataset directory from a qualified Studio SmolVLA export.", + help="Local LeRobot v3 dataset directory; validated offline before training.", ), dataset_manifest_sha256: str = typer.Option( "", "--dataset-manifest-sha256", help="Canonical sibling export-receipt.json SHA-256 (excludes its sha256 field).", ), + dataset_content_sha256: str = typer.Option("", "--dataset-content-sha256", help="Expected local content manifest SHA-256."), + dataset_features: str = typer.Option("{}", "--dataset-features", help="Recipe expected LeRobot feature shapes as JSON."), output: str = typer.Option( ..., "--output", @@ -143,6 +146,13 @@ def finetune_command( from tether.finetune.config import FinetuneConfig from tether.finetune.run import run_finetune + try: + expected_features = json.loads(dataset_features) + except ValueError as exc: + raise typer.BadParameter("dataset_features must be a JSON feature object", param_hint="--dataset-features") from exc + if not isinstance(expected_features, dict): + raise typer.BadParameter("dataset_features must be a JSON feature object", param_hint="--dataset-features") + cfg = FinetuneConfig( base=base, base_revision=base_revision or None, @@ -150,6 +160,8 @@ def finetune_command( dataset_revision=dataset_revision or None, dataset_root=Path(dataset_root) if dataset_root else None, dataset_manifest_sha256=dataset_manifest_sha256 or None, + dataset_content_sha256=dataset_content_sha256 or None, + dataset_features=expected_features, output=Path(output), num_steps=num_steps, batch_size=batch_size, diff --git a/src/tether/finetune/config.py b/src/tether/finetune/config.py index c88f8ba..d7666e7 100644 --- a/src/tether/finetune/config.py +++ b/src/tether/finetune/config.py @@ -179,11 +179,17 @@ class FinetuneConfig: """ dataset_root: Path | None = None - """Actual EXPORT/dataset directory of a qualified local Studio export.""" + """Local LeRobot v3 directory, or a qualified Studio export dataset directory.""" dataset_manifest_sha256: str | None = None """Canonical sibling export-receipt.json digest, excluding its sha256 field.""" + dataset_content_sha256: str | None = None + """Optional expected SHA-256 of the local content manifest, checked before launch.""" + + dataset_features: dict[str, Any] = field(default_factory=dict) + """Recipe expected features, keyed by LeRobot names with shapes.""" + def __post_init__(self) -> None: self.output = Path(self.output) if self.dataset_root is not None: diff --git a/src/tether/finetune/lerobot_v3.py b/src/tether/finetune/lerobot_v3.py new file mode 100644 index 0000000..cc5c1d3 --- /dev/null +++ b/src/tether/finetune/lerobot_v3.py @@ -0,0 +1,494 @@ +"""Local LeRobot v3 schema, row integrity and content identity, without Hub access.""" + +from __future__ import annotations + +import math +from bisect import bisect_left +from pathlib import Path +import stat +from typing import Any + +from tether.finetune.local_dataset import ( + LocalDatasetError, + _file_hash, + _hash, + _integer, + _json, + _object, + _require, +) + + +def _check(condition: bool, field: str, detail: str) -> None: + _require(condition, "local-v3-invalid", f"{field}: {detail}") + + +def validate_features(info: dict, expected: dict | None = None) -> None: + _check( + expected is None or isinstance(expected, dict), + "dataset_features", + "expected a feature object", + ) + features = _object(info.get("features"), "meta/info.json.features") + for key, feature in features.items(): + feature = _object(feature, f"features.{key}") + shape = feature.get("shape") + _check( + isinstance(shape, list) and bool(shape) and all(_integer(n, 1) for n in shape), + f"features.{key}.shape", + "expected positive integer dimensions", + ) + _check( + feature.get("dtype") + in { + "float16", + "float32", + "float64", + "int8", + "int16", + "int32", + "int64", + "uint8", + "uint16", + "uint32", + "uint64", + "bool", + "string", + "image", + "video", + }, + f"features.{key}.dtype", + "unsupported LeRobot dtype", + ) + names = feature.get("names") + _check( + names is None or isinstance(names, (list, dict)), + f"features.{key}.names", + "expected dimension names or null", + ) + for key in ("action", "observation.state"): + feature = features.get(key, {}) + _check( + feature.get("dtype") == "float32" and len(feature.get("shape", [])) == 1, + f"features.{key}", + "expected a float32 vector", + ) + cameras = {k for k, f in features.items() if f["dtype"] in {"image", "video"}} + _check(bool(cameras), "features.observation.images", "the recipe requires at least one camera") + for key in cameras: + f = features[key] + _check( + key.startswith("observation.images.") and len(f["shape"]) == 3 and f["shape"][-1] == 3, + f"features.{key}.shape", + "expected HWC RGB camera", + ) + for key in ("index", "episode_index", "frame_index", "task_index", "timestamp"): + f = features.get(key, {}) + _check( + f.get("shape") == [1] + and f.get("dtype") == ("float32" if key == "timestamp" else "int64"), + f"features.{key}", + "invalid index/timestamp schema", + ) + for key, wanted in (expected or {}).items(): + _check(key in features, f"features.{key}", "required by the recipe/base checkpoint") + shape = wanted.get("shape") if isinstance(wanted, dict) else wanted + # Policy configs store visual shapes CHW, dataset info stores HWC. + if isinstance(wanted, dict) and wanted.get("type") == "VISUAL": + shape = [shape[1], shape[2], shape[0]] + _check( + features[key]["shape"] == shape, + f"features.{key}.shape", + f"expected {shape}, got {features[key]['shape']}", + ) + expected_cameras = {k for k in (expected or {}) if k.startswith("observation.images.")} + if expected_cameras: + _check( + cameras == expected_cameras, + "features.observation.images", + f"expected camera keys {sorted(expected_cameras)}, got {sorted(cameras)}", + ) + + +def _rows(paths: list[Path], label: str): + _check(bool(paths), label, "missing local parquet files") + try: + import pyarrow.parquet as pq + except ImportError as exc: + raise LocalDatasetError( + "local-v3-reader", + "episode_index: pyarrow from the LeRobot training runtime is required", + ) from exc + for path in paths: + try: + parquet = pq.ParquetFile(path) + for batch in parquet.iter_batches(batch_size=32): + yield from batch.to_pylist() + except Exception as exc: + raise LocalDatasetError("local-v3-invalid", f"{label} ({path.name}): {exc}") from exc + + +def _inspect_lerobot_v3( + root: Path, *, expected_features: dict | None = None, expected_sha256: str | None = None, + metadata_only: bool = False, +) -> dict: + """Validate local metadata and episode/frame indexes, hash every retained byte. + + Includes videos in identity, since camera training inputs must also be bound. + Callers must keep the admitted directory immutable until training completes. + """ + try: + root = Path(root).expanduser().absolute() + for path in [*root.parents, root]: + _check(not path.is_symlink(), "dataset_root", f"symlink component {path}") + _check(root.is_dir(), "dataset_root", "directory does not exist") + manifest = [] + for directory in (("meta",) if metadata_only else ("meta", "data", "videos", "images")): + base = root / directory + if not base.exists(): + continue + _check(not base.is_symlink(), directory, "symlinks are forbidden") + for path in sorted(base.rglob("*")): + mode = path.lstat().st_mode + _check( + stat.S_ISDIR(mode) or stat.S_ISREG(mode), + str(path.relative_to(root)), + "only regular files/directories are allowed", + ) + if stat.S_ISREG(mode) and not metadata_only: + manifest.append( + { + "path": path.relative_to(root).as_posix(), + "size_bytes": path.stat().st_size, + "sha256": _file_hash(path), + } + ) + manifest.sort(key=lambda row: row["path"]) + content = None if metadata_only else _hash(manifest) + _check( + expected_sha256 is None or content == expected_sha256, + "dataset_content_sha256", + "local bytes changed since review", + ) + info = _object(_json(root / "meta/info.json"), "meta/info.json") + _check(info.get("codebase_version") == "v3.0", "codebase_version", "expected LeRobot v3.0") + for key in ("total_frames", "total_episodes", "total_tasks", "chunks_size"): + _check(_integer(info.get(key), 1), key, "expected a positive integer") + _check( + info["total_episodes"] <= info["total_frames"], "total_episodes", "exceeds total_frames" + ) + fps = info.get("fps") + _check( + type(fps) in (float, int) and math.isfinite(fps) and fps > 0, + "fps", + "expected a finite positive rate", + ) + _check( + isinstance(info.get("robot_type"), str) or info.get("robot_type") is None, + "robot_type", + "expected string or null", + ) + _check( + info.get("data_path") == "data/chunk-{chunk_index:03d}/file-{file_index:03d}.parquet", + "data_path", + "expected the v3 shard layout", + ) + splits = info.get("splits") + _check(isinstance(splits, dict) and bool(splits), "splits", "expected named episode ranges") + for key, span in splits.items(): + try: + start, end = map(int, span.split(":")) + except (AttributeError, ValueError): + _check(False, f"splits.{key}", "expected start:end") + _check( + 0 <= start < end <= info["total_episodes"], + f"splits.{key}", + "episode range out of bounds", + ) + validate_features(info, expected_features) + stats = _object(_json(root / "meta/stats.json"), "meta/stats.json") + from tether.finetune.local_dataset import _shape + + for key in ("action", "observation.state"): + moments = _object(stats.get(key), f"meta/stats.json.{key}") + for moment in ("min", "max", "mean", "std"): + _shape( + moments.get(moment), + info["features"][key]["shape"], + f"meta/stats.json.{key}.{moment}", + ) + tasks = list(_rows([root / "meta/tasks.parquet"], "task_index")) + _check(len(tasks) == info["total_tasks"], "total_tasks", "differs from tasks.parquet") + episodes = list(_rows(sorted((root / "meta/episodes").rglob("*.parquet")), "episode_index")) + _check( + len(episodes) == info["total_episodes"], "total_episodes", "differs from episode index" + ) + offset = 0 + video_timestamps: dict[str, tuple[list[float], list[float]]] = {} + video_queries: list[tuple[int, str, float, list[float]]] = [] + for i, episode in enumerate(episodes): + _check( + type(episode.get("episode_index")) is int and episode["episode_index"] == i, + "episode_index", + f"expected contiguous episode {i}", + ) + length = episode.get("length") + _check( + _integer(length, 1), f"episode_index[{i}].length", "expected positive frame count" + ) + _check( + episode.get("dataset_from_index") == offset + and episode.get("dataset_to_index") == offset + length, + f"episode_index[{i}].dataset_from_index/dataset_to_index", + "gapped or inconsistent frame range", + ) + for field in ("data/chunk_index", "data/file_index"): + _check( + _integer(episode.get(field)), + f"episode_index[{i}].{field}", + "expected shard index", + ) + shard = info["data_path"].format( + chunk_index=episode["data/chunk_index"], file_index=episode["data/file_index"] + ) + _check( + metadata_only or (root / shard).is_file(), + f"episode_index[{i}].data/file_index", + "missing local shard", + ) + for camera, feature in info["features"].items(): + if feature["dtype"] != "video": + continue + _check( + info.get("video_path") + == "videos/{video_key}/chunk-{chunk_index:03d}/file-{file_index:03d}.mp4", + "video_path", + "expected the v3 video layout", + ) + values = { + suffix: episode.get(f"videos/{camera}/{suffix}") + for suffix in ("chunk_index", "file_index", "from_timestamp", "to_timestamp") + } + _check( + _integer(values["chunk_index"]) and _integer(values["file_index"]), + f"episode_index[{i}].videos/{camera}", + "missing video shard indexes", + ) + _check( + all( + type(values[k]) in (int, float) and math.isfinite(values[k]) + for k in ("from_timestamp", "to_timestamp") + ) + and 0 <= values["from_timestamp"] < values["to_timestamp"], + f"episode_index[{i}].videos/{camera}/from_timestamp", + "invalid video range", + ) + video = info["video_path"].format( + video_key=camera, + chunk_index=values["chunk_index"], + file_index=values["file_index"], + ) + _check( + metadata_only or (root / video).is_file(), + f"episode_index[{i}].videos/{camera}/file_index", + "missing local video", + ) + if not metadata_only: + field = f"episode_index[{i}].videos/{camera}/to_timestamp" + tolerance_s = LEROBOT_TOLERANCE_S + start, end = values["from_timestamp"], values["to_timestamp"] + _check(abs(end - start - length / fps) < tolerance_s, field, + "video duration differs from episode length/fps") + if video not in video_timestamps: + video_timestamps[video] = _video_timestamps(root / video, field) + timestamps, frame_ends = video_timestamps[video] + lo = bisect_left(timestamps, start - tolerance_s) + hi = bisect_left(timestamps, end - tolerance_s) + _check(hi - lo == length, field, + "decoded video frame count differs from episode length") + _check(abs(frame_ends[hi - 1] - end) < tolerance_s, field, + "decoded video duration differs from episode length/fps") + # LeRobot queries from_timestamp + the row's own timestamp, so the + # per-frame check runs after the row scan below. + video_queries.append((i, camera, start, timestamps)) + offset += length + _check(offset == info["total_frames"], "total_frames", "differs from episode lengths") + if metadata_only: + return {"kind": "local-lerobot-v3", "dataset_root": str(root), "info": info, + "metadata_checked": True} + retained_paths = {r["path"] for r in manifest} + count, episode_id, frame = 0, 0, 0 + row_timestamps: list[list[float]] = [[] for _ in episodes] + for path in sorted((root / "data").rglob("*.parquet")): + for row in _rows([path], "data"): + _check(episode_id < len(episodes), "index", "unexpected extra frame") + episode = episodes[episode_id] + shard = info["data_path"].format( + chunk_index=episode["data/chunk_index"], file_index=episode["data/file_index"] + ) + _check( + path.relative_to(root).as_posix() == shard, + "data/file_index", + "frame is in a different shard from its episode index", + ) + for key, value in ( + ("index", count), + ("episode_index", episode_id), + ("frame_index", frame), + ): + _check( + type(row.get(key)) is int and row[key] == value, + key, + f"expected {value} at frame {count}", + ) + _check( + _integer(row.get("task_index")) and row["task_index"] < info["total_tasks"], + "task_index", + f"out of bounds at frame {count}", + ) + timestamp = row.get("timestamp") + _check( + type(timestamp) in (int, float) + and math.isfinite(timestamp) + and abs(timestamp - frame / fps) <= 1e-4, + "timestamp", + f"inconsistent with fps at frame {count}", + ) + row_timestamps[episode_id].append(timestamp) + for key in ("action", "observation.state"): + value = row.get(key) + # LeRobot 0.5.1 stores shape [1] as a scalar Value, not a list + # (lerobot/datasets/feature_utils.py:54). + if info["features"][key]["shape"] == [1] and type(value) in (int, float): + value = [value] + _check( + isinstance(value, list) + and len(value) == info["features"][key]["shape"][0] + and all(type(n) in (int, float) and math.isfinite(n) for n in value), + f"features.{key}.shape", + f"invalid vector at frame {count}", + ) + for key, feature in info["features"].items(): + if feature["dtype"] == "image": + image = row.get(key) + _check( + isinstance(image, dict) + and bool(image.get("bytes") or image.get("path")), + f"features.{key}", + f"missing image at frame {count}", + ) + if image.get("path") and not image.get("bytes"): + image_path = Path(image["path"]) + _check( + not image_path.is_absolute() + and ".." not in image_path.parts + and image_path.as_posix() in retained_paths + and (root / image_path).is_file() + and not (root / image_path).is_symlink(), + f"features.{key}.path", + "missing/unsafe local image", + ) + count += 1 + frame += 1 + if frame == episode["length"]: + episode_id += 1 + frame = 0 + _check(count == info["total_frames"], "total_frames", "differs from local data rows") + for i, camera, start, timestamps in video_queries: + for frame, row_ts in enumerate(row_timestamps[i]): + query = start + row_ts + index = bisect_left(timestamps, query) + neighbors = timestamps[max(0, index - 1):index + 1] + _check( + any(_within_lerobot_tolerance(query, t) for t in neighbors), + f"episode_index[{i}].videos/{camera}/from_timestamp", + f"decoded frame {frame} violates LeRobot tolerance_s={LEROBOT_TOLERANCE_S}", + ) + return { + "kind": "local-lerobot-v3", + "dataset_root": str(root), + "info": info, + "content_sha256": content, + "artifacts": manifest, + "size_bytes": sum(r["size_bytes"] for r in manifest), + } + except LocalDatasetError: + raise + except (OSError, ValueError, TypeError, KeyError) as exc: + raise LocalDatasetError( + "local-v3-invalid", f"meta/info.json or episode_index: {exc}" + ) from exc + + +# LeRobot's TrainPipelineConfig.tolerance_s default (configs/train.py). Local +# configs refuse a tolerance_s override (local_dataset.validate_local_config). +LEROBOT_TOLERANCE_S = 1e-4 + + +def _within_lerobot_tolerance(query: float, pts: float) -> bool: + """LeRobot's decode rule: torch.cdist(p=1) on default float32 tensors, then + `min_ < tolerance_s` (video_utils.py). The float32 rounding matters for long + concatenated shards, where float64 gives the opposite answer near the edge.""" + import numpy as np + + return bool(np.abs(np.float32(query) - np.float32(pts)) < np.float32(LEROBOT_TOLERANCE_S)) + + +def _video_timestamps(path: Path, field: str) -> tuple[list[float], list[float]]: + """Decode each referenced shard once, including PTS, during full admission.""" + try: + import av + + with av.open(str(path)) as container: + timestamps, frame_ends = [], [] + for frame in container.decode(video=0): + _check(frame.pts is not None and frame.duration > 0, field, + "video frame has no timestamp/duration") + timestamps.append(float(frame.pts * frame.time_base)) + frame_ends.append(float((frame.pts + frame.duration) * frame.time_base)) + _check(bool(timestamps) and all(math.isfinite(t) for t in timestamps) + and all(a < b for a, b in zip(timestamps, timestamps[1:])), + field, "video has no frames or invalid decoded timestamps") + return timestamps, frame_ends + except LocalDatasetError: + raise + except Exception as exc: + raise LocalDatasetError("local-v3-invalid", f"{field}: cannot decode local video: {exc}") from exc + + +def describe_lerobot_v3(root: Path, *, expected_features: dict | None = None) -> dict: + """Check only meta/ for listing; this is not content admission or identity.""" + return _inspect_lerobot_v3(root, expected_features=expected_features, metadata_only=True) + + +def verify_lerobot_v3( + root: Path, *, expected_features: dict | None = None, expected_sha256: str | None = None +) -> dict: + """Validate metadata, all frames and video timestamps, and hash retained bytes.""" + return _inspect_lerobot_v3(root, expected_features=expected_features, expected_sha256=expected_sha256) + + +def verify_local_dataset(cfg: Any) -> dict: + """Explicit receipt flag retains the old export contract; raw v3 never downgrades it.""" + from tether.finetune.local_dataset import verify_local_export + + if cfg.dataset_manifest_sha256: + verified = verify_local_export(cfg.dataset_root, cfg.dataset_manifest_sha256) + manifest = [ + {**row, "path": row["path"].removeprefix("dataset/")} + for row in verified["receipt"]["artifacts"] + if row["path"].startswith("dataset/") + ] + manifest.sort(key=lambda row: row["path"]) + return { + **verified, + "kind": "local-lerobot-export", + "content_sha256": _hash(manifest), + "artifacts": manifest, + "size_bytes": sum(row["size_bytes"] for row in manifest), + } + return verify_lerobot_v3( + cfg.dataset_root, + expected_features=cfg.dataset_features, + expected_sha256=cfg.dataset_content_sha256, + ) diff --git a/src/tether/finetune/local_dataset.py b/src/tether/finetune/local_dataset.py index 1cb452f..35fc4bf 100644 --- a/src/tether/finetune/local_dataset.py +++ b/src/tether/finetune/local_dataset.py @@ -734,12 +734,14 @@ def verify_local_export( def validate_local_config(cfg: Any) -> None: """Reject local escape hatches before any preflight or command is built.""" root, digest = cfg.dataset_root, cfg.dataset_manifest_sha256 - if root is None and digest is None: + content = getattr(cfg, "dataset_content_sha256", None) + raw = root is not None and digest is None + if root is None and digest is None and content is None: return _require( - root is not None and digest is not None and _digest(digest), + root is not None and (digest is None or _digest(digest)) and (content is None or _digest(content)) and not (digest and content), "local-config", - "dataset_root and dataset_manifest_sha256 must be supplied together.", + "dataset_root is required with a valid optional dataset_manifest_sha256 or dataset_content_sha256.", ) _require( not cfg.skip_preflight and not cfg.resume, @@ -749,15 +751,17 @@ def validate_local_config(cfg: Any) -> None: _require( cfg.phase == "train" and cfg.backend == "lerobot" - and cfg.policy == "auto" - and cfg.mode == "lora" - and bool(cfg.base) + and ((cfg.policy == "act" and cfg.mode == "full") or (cfg.policy == "auto" and cfg.mode == "lora" and bool(cfg.base))) + and (raw or cfg.policy == "auto") and bool(cfg.dataset) and not cfg.base_dataset, "unsupported-profile", - "Local datasets require pretrained SmolVLA LoRA (policy=auto, phase=train).", + "Local datasets require ACT from scratch (policy=act, mode=full) or pretrained SmolVLA LoRA (policy=auto, mode=lora), phase=train.", ) - export = Path(root).expanduser().resolve().parent + _require(not raw or not (Path(root).expanduser().resolve().parent / "export-receipt.json").exists(), "local-config", "dataset_manifest_sha256: a Studio export receipt requires its explicit digest; it cannot be downgraded to raw admission.") + export = Path(root).expanduser().resolve() + if not raw: + export = export.parent _require( not Path(cfg.output).expanduser().resolve().is_relative_to(export), "local-config", @@ -788,6 +792,8 @@ def validate_local_config(cfg: Any) -> None: "rename_map", "peft", "peft.method_type", + # Raw v3 admission checks video frames at LeRobot's default. + "tolerance_s", } and ( not normalized.startswith("dataset.") or normalized == "dataset.use_imagenet_stats" @@ -800,7 +806,7 @@ def validate_local_config(cfg: Any) -> None: "peft.method_type.", ) ) - and "normaliz" not in normalized.lower() + and ((raw and normalized == "policy.normalization_mapping") or "normaliz" not in normalized.lower()) and "processor" not in normalized.lower() and normalized not in { diff --git a/src/tether/finetune/preflight/dataset_size.py b/src/tether/finetune/preflight/dataset_size.py index 29e38ea..01af41b 100644 --- a/src/tether/finetune/preflight/dataset_size.py +++ b/src/tether/finetune/preflight/dataset_size.py @@ -19,6 +19,8 @@ """ from __future__ import annotations +from tether.finetune.lerobot_v3 import verify_local_dataset + import logging from typing import Any @@ -86,10 +88,10 @@ def _fetch_dataset_info(dataset_repo_id: str) -> dict[str, Any] | None: def check_dataset_size(cfg: FinetuneConfig) -> PreflightCheck: """Warn if episode count is below the guidance for the selected policy.""" if cfg.dataset_root is not None: - from tether.finetune.local_dataset import validate_local_config, verify_local_export + from tether.finetune.local_dataset import validate_local_config validate_local_config(cfg) - info = verify_local_export(cfg.dataset_root, cfg.dataset_manifest_sha256)["info"] - policy_type = "smolvla" # Mandatory schema check validates the actual base config. + info = verify_local_dataset(cfg)["info"] + policy_type = cfg.policy if cfg.policy != "auto" else "smolvla" else: info = None policy_type = _infer_policy_type(cfg.base) diff --git a/src/tether/finetune/preflight/runner.py b/src/tether/finetune/preflight/runner.py index 2141970..bfd4f29 100644 --- a/src/tether/finetune/preflight/runner.py +++ b/src/tether/finetune/preflight/runner.py @@ -10,6 +10,8 @@ """ from __future__ import annotations +from tether.finetune.lerobot_v3 import verify_local_dataset + import logging from tether.finetune.config import FinetuneConfig @@ -36,15 +38,18 @@ def run_preflight(cfg: FinetuneConfig) -> PreflightReport: """ report = PreflightReport() from tether.finetune.local_dataset import ( - effective_imagenet_stats, validate_local_config, verify_local_export, + effective_imagenet_stats, validate_local_config, ) # Local integrity is mandatory. Even an unexpected verifier exception blocks # before any remote metadata lookup or later checks. if cfg.dataset_root is not None or cfg.dataset_manifest_sha256 is not None: try: validate_local_config(cfg) - verified = verify_local_export(cfg.dataset_root, cfg.dataset_manifest_sha256) - report.add(PreflightCheck( + verified = verify_local_dataset(cfg) + if not cfg.dataset_manifest_sha256: + report.add(PreflightCheck("local_dataset", "ok", "Local LeRobot v3 schema and indexes verified.", {"content_sha256": verified["content_sha256"], "info": verified["info"]})) + else: + report.add(PreflightCheck( "local_dataset", "ok", "Local export bytes and retained admission evidence verified.", {"dataset_root": verified["dataset_root"], "manifest_sha256": verified["manifest_sha256"], diff --git a/src/tether/finetune/preflight/schema.py b/src/tether/finetune/preflight/schema.py index a243101..3f9fe4d 100644 --- a/src/tether/finetune/preflight/schema.py +++ b/src/tether/finetune/preflight/schema.py @@ -13,6 +13,8 @@ """ from __future__ import annotations +from tether.finetune.lerobot_v3 import verify_local_dataset + import logging from pathlib import Path from typing import Any @@ -126,14 +128,29 @@ def check_schema(cfg: FinetuneConfig) -> PreflightCheck: gated models still need a path through. """ if cfg.dataset_root is not None: - from tether.finetune.local_dataset import validate_local_config, verify_local_export + from tether.finetune.local_dataset import validate_local_config validate_local_config(cfg) - features = verify_local_export(cfg.dataset_root, cfg.dataset_manifest_sha256)["info"]["features"] + features = verify_local_dataset(cfg)["info"]["features"] else: features = _fetch_dataset_features(cfg.dataset) base_config = (_fetch_base_config(cfg.base, cfg.base_revision) if cfg.base_revision else _fetch_base_config(cfg.base)) local_details = {} + if cfg.dataset_root is not None and not cfg.dataset_manifest_sha256: + # Recipe features (cfg.dataset_features) were already enforced by + # verify_local_dataset. The base checkpoint's input/output_features are + # its pretraining embodiment; the command clears them + # (--policy.input_features=null) so LeRobot derives them from this + # dataset. Only the padding bounds bind, as on the export path. + if cfg.policy == "act" and cfg.mode == "full": + return PreflightCheck("schema", "ok", "Local features match the ACT recipe.", {"features": features}) + if not isinstance(base_config, dict) or base_config.get("type") != "smolvla": + return PreflightCheck("schema", "fail", "base.config.type: requires a resolvable supported SmolVLA checkpoint") + for key, maximum in (("action", "max_action_dim"), ("observation.state", "max_state_dim")): + limit = base_config.get(maximum) + if type(limit) is not int or features[key]["shape"][0] > limit: + return PreflightCheck("schema", "fail", f"features.{key}.shape: exceeds or cannot resolve base {maximum}") + return PreflightCheck("schema", "ok", "Local features match the SmolVLA checkpoint.") if cfg.dataset_root is not None: expected_modes = {"VISUAL": "IDENTITY", "STATE": "MEAN_STD", "ACTION": "MEAN_STD"} if not isinstance(base_config, dict) or base_config.get("type") != "smolvla": diff --git a/src/tether/finetune/run.py b/src/tether/finetune/run.py index 0f35f36..81bfca9 100644 --- a/src/tether/finetune/run.py +++ b/src/tether/finetune/run.py @@ -13,6 +13,8 @@ """ from __future__ import annotations +from tether.finetune.lerobot_v3 import verify_local_dataset + import logging import json import subprocess @@ -143,13 +145,13 @@ def _build_lerobot_command(cfg: FinetuneConfig) -> list[str]: doesn't expose a top-level precision flag; it's baked into the policy config. v0.5 will add per-policy precision overrides. """ - from tether.finetune.local_dataset import validate_local_config, verify_local_export + from tether.finetune.local_dataset import validate_local_config validate_local_config(cfg) local_export = None if cfg.dataset_root is not None: # Recheck immediately before emission. Callers must still keep exports # immutable: a path-based trainer is not a filesystem snapshot boundary. - local_export = verify_local_export(cfg.dataset_root, cfg.dataset_manifest_sha256) + local_export = verify_local_dataset(cfg) from tether.finetune.preflight.schema import check_schema schema = check_schema(cfg) if schema.severity != "ok": @@ -245,6 +247,12 @@ def _run_lerobot_training( """ cmd = _build_lerobot_command(cfg) logger.info("[finetune] exec: %s", " ".join(cmd)) + if cfg.dataset_root is not None: + import os + env = dict(os.environ if env is None else env) + # LeRobot must never recover missing local data by downloading it. Base + # weights are resolved above; offline mode also refuses uncached weights. + env.update(HF_HUB_OFFLINE="1", HF_DATASETS_OFFLINE="1") log_path.parent.mkdir(parents=True, exist_ok=True) with log_path.open("w") as log: log.write("# tether finetune — lerobot-train invocation\n") @@ -425,6 +433,16 @@ def run_finetune(cfg: FinetuneConfig, *, hooks=None) -> FinetuneResult: ) cfg.output.mkdir(parents=True, exist_ok=True) + if cfg.dataset_root is not None: + try: + identity = verify_local_dataset(cfg) + except Exception as exc: + return FinetuneResult(status="aborted", output_dir=cfg.output, error=f"Local dataset verification failed: {exc}") + identity["dataset_repository"] = cfg.dataset + identity["expected_features"] = cfg.dataset_features + if not cfg.dataset_manifest_sha256: + cfg = replace(cfg, dataset_content_sha256=identity["content_sha256"]) + (cfg.output / "training-dataset.json").write_text(json.dumps(identity, indent=2) + "\n") training_log = cfg.output / "training_log.jsonl" # Pre-flight validation (v0.5) — catches top customer pains before @@ -504,6 +522,9 @@ def run_finetune(cfg: FinetuneConfig, *, hooks=None) -> FinetuneResult: error=ckpt_result.error, ) + if cfg.dataset_root is not None: + (ckpt_result.final_checkpoint_path / "training-dataset.json").write_text((cfg.output / "training-dataset.json").read_text()) + if cfg.skip_export: logger.info("[finetune] skip_export=True; done (no ONNX)") return FinetuneResult( diff --git a/tests/test_finetune_lerobot_v3.py b/tests/test_finetune_lerobot_v3.py new file mode 100644 index 0000000..ac8ce28 --- /dev/null +++ b/tests/test_finetune_lerobot_v3.py @@ -0,0 +1,432 @@ +"""Small real-parquet v3 inputs, with all dataset networking forbidden.""" + +import io +import json +import socket +from dataclasses import replace +from pathlib import Path + +import pyarrow as pa +import pyarrow.parquet as pq +from PIL import Image +import pytest + +from tether.finetune.config import FinetuneConfig +from tether.finetune.lerobot_v3 import describe_lerobot_v3, verify_lerobot_v3 +from tether.finetune.local_dataset import LocalDatasetError +from tether.finetune.preflight import run_preflight +from tether.finetune.run import _build_lerobot_command, run_finetune + + +def write_v3(root, action_dim=6): + root = Path(root) + features = { + "action": {"dtype": "float32", "shape": [action_dim], "names": None}, + "observation.state": {"dtype": "float32", "shape": [6], "names": None}, + "observation.images.wrist": { + "dtype": "image", + "shape": [32, 32, 3], + "names": ["height", "width", "channels"], + }, + } + for key in ("timestamp", "index", "frame_index", "episode_index", "task_index"): + features[key] = { + "dtype": "float32" if key == "timestamp" else "int64", + "shape": [1], + "names": None, + } + info = { + "codebase_version": "v3.0", + "robot_type": "so101", + "total_frames": 9, + "total_episodes": 3, + "total_tasks": 1, + "chunks_size": 1000, + "data_files_size_in_mb": 100, + "video_files_size_in_mb": 200, + "fps": 10, + "splits": {"train": "0:3"}, + "data_path": "data/chunk-{chunk_index:03d}/file-{file_index:03d}.parquet", + "video_path": None, + "features": features, + } + (root / "meta/episodes/chunk-000").mkdir(parents=True) + (root / "data/chunk-000").mkdir(parents=True) + (root / "meta/info.json").write_text(json.dumps(info)) + stats = {} + for key, feature in features.items(): + shape = [3, 1, 1] if feature["dtype"] == "image" else feature["shape"] + + def full(value, shape): + return [full(value, shape[1:]) for _ in range(shape[0])] if shape else value + + stats[key] = { + "min": full(0.0, shape), + "max": full(1.0, shape), + "mean": full(0.5, shape), + "std": full(0.25, shape), + "count": [9], + } + (root / "meta/stats.json").write_text(json.dumps(stats)) + pq.write_table( + pa.table({"task_index": [0], "task": ["move cube"]}), root / "meta/tasks.parquet" + ) + episodes = [ + { + "episode_index": i, + "length": 3, + "tasks": ["move cube"], + "dataset_from_index": i * 3, + "dataset_to_index": (i + 1) * 3, + "data/chunk_index": 0, + "data/file_index": 0, + "meta/episodes/chunk_index": 0, + "meta/episodes/file_index": 0, + } + for i in range(3) + ] + pq.write_table( + pa.Table.from_pylist(episodes), root / "meta/episodes/chunk-000/file-000.parquet" + ) + image = io.BytesIO() + Image.new("RGB", (32, 32), (20, 40, 60)).save(image, format="PNG") + data = [ + { + "index": i, + "episode_index": i // 3, + "frame_index": i % 3, + "task_index": 0, + "timestamp": (i % 3) / 10, + # LeRobot 0.5.1 stores shape [1] as a scalar (feature_utils.py:54). + "action": 0.1 if action_dim == 1 else [0.1] * action_dim, + "observation.state": [0.2] * 6, + "observation.images.wrist": {"bytes": image.getvalue(), "path": None}, + } + for i in range(9) + ] + table = pa.Table.from_pylist(data) + for key in ("action", "observation.state", "timestamp"): + table = table.set_column( + table.schema.get_field_index(key), + key, + table[key].cast( + pa.float32() + if key == "timestamp" or (key == "action" and action_dim == 1) + else pa.list_(pa.float32()) + ), + ) + pq.write_table(table, root / "data/chunk-000/file-000.parquet") + return root + + +@pytest.fixture +def cfg(tmp_path, monkeypatch): + monkeypatch.setenv("HF_HUB_OFFLINE", "1") + monkeypatch.setenv("HF_DATASETS_OFFLINE", "1") + monkeypatch.setattr( + socket, "create_connection", lambda *a, **kw: pytest.fail("dataset network access") + ) + return FinetuneConfig( + base="", + dataset="studio-local/fixture", + output=tmp_path / "run", + policy="act", + mode="full", + dataset_root=write_v3(tmp_path / "local"), + dataset_features={"action": {"shape": [6]}, "observation.state": {"shape": [6]}}, + num_steps=2, + skip_export=True, + ) + + +def test_offline_schema_command_and_identity(cfg): + verified = verify_lerobot_v3(cfg.dataset_root, expected_features=cfg.dataset_features) + assert len(verified["content_sha256"]) == 64 + assert not run_preflight(cfg).has_failures + assert f"--dataset.root={cfg.dataset_root}" in _build_lerobot_command(cfg) + result = run_finetune(replace(cfg, dry_run=True)) + assert result.status == "ok" + record = json.loads((cfg.output / "training-dataset.json").read_text()) + assert record["content_sha256"] == verified["content_sha256"] + assert {r["path"] for r in record["artifacts"]} >= { + "meta/info.json", + "data/chunk-000/file-000.parquet", + } + + +def test_seven_actions_refused(cfg): + root = write_v3(cfg.output.parent / "seven", 7) + cfg = replace(cfg, dataset_root=root) + report = run_preflight(cfg) + assert report.has_failures + assert "features.action.shape" in report.error_message() + assert run_finetune(cfg).status == "aborted" + + +@pytest.mark.parametrize( + "field", ["episode_index", "frame_index", "index", "task_index", "timestamp"] +) +def test_corrupt_row_index_refused(cfg, field): + path = cfg.dataset_root / "data/chunk-000/file-000.parquet" + rows = pq.read_table(path).to_pylist() + rows[1][field] = 999 + pq.write_table(pa.Table.from_pylist(rows), path) + with pytest.raises(LocalDatasetError, match=field): + verify_lerobot_v3(cfg.dataset_root) + + +def test_content_changes_refused(cfg): + identity = verify_lerobot_v3(cfg.dataset_root)["content_sha256"] + (cfg.dataset_root / "meta/stats.json").write_text("{}") + with pytest.raises(LocalDatasetError, match="dataset_content_sha256"): + verify_lerobot_v3(cfg.dataset_root, expected_sha256=identity) + + +def test_camera_contract_refused(cfg): + with pytest.raises(LocalDatasetError, match="features.observation.images.front"): + verify_lerobot_v3( + cfg.dataset_root, expected_features={"observation.images.front": {"shape": [32, 32, 3]}} + ) + + +def test_smolvla_local_features_and_padding(cfg): + base = cfg.output.parent / "smolvla-base" + base.mkdir() + (base / "config.json").write_text( + json.dumps( + { + "type": "smolvla", + "max_state_dim": 32, + "max_action_dim": 32, + "input_features": {}, + "output_features": {}, + } + ) + ) + cfg = replace(cfg, base=str(base), policy="auto", mode="lora", dataset_features={}) + assert not run_preflight(cfg).has_failures + assert f"--dataset.root={cfg.dataset_root}" in _build_lerobot_command(cfg) + + +@pytest.mark.parametrize( + "field,value", + [ + ("codebase_version", "v2.1"), + ("fps", 0), + ("total_episodes", 4), + ("total_frames", 10), + ("data_path", "../outside.parquet"), + ], +) +def test_info_fields_named_in_refusal(cfg, field, value): + path = cfg.dataset_root / "meta/info.json" + info = json.loads(path.read_text()) + info[field] = value + path.write_text(json.dumps(info)) + with pytest.raises(LocalDatasetError, match=field): + verify_lerobot_v3(cfg.dataset_root) + + +@pytest.mark.parametrize( + "field,value", + [("episode_index", 0), ("length", 4), ("dataset_from_index", 2), ("data/file_index", 9)], +) +def test_episode_index_integrity_refused(cfg, field, value): + path = cfg.dataset_root / "meta/episodes/chunk-000/file-000.parquet" + table = pq.read_table(path) + rows = table.to_pylist() + rows[1][field] = value + pq.write_table(pa.Table.from_pylist(rows, schema=table.schema), path) + with pytest.raises(LocalDatasetError, match=field if field != "length" else "dataset_to_index"): + verify_lerobot_v3(cfg.dataset_root) + + +def test_public_cli_admits_raw_v3_and_binds_reviewed_identity(cfg): + import typer + from typer.testing import CliRunner + from tether.finetune.cli import finetune_command + + app = typer.Typer() + app.command()(finetune_command) + identity = verify_lerobot_v3(cfg.dataset_root)["content_sha256"] + result = CliRunner().invoke(app, ["--policy", "act", "--mode", "full", "--dataset", cfg.dataset, + "--dataset-root", str(cfg.dataset_root), "--dataset-content-sha256", identity, + "--dataset-features", json.dumps(cfg.dataset_features), "--output", str(cfg.output), "--steps", "2", "--dry-run"]) + assert result.exit_code == 0, result.output + record = json.loads((cfg.output/"training-dataset.json").read_text()) + assert record["content_sha256"] == identity + + +def test_width_one_action_stored_as_scalar_admitted(cfg): + root = write_v3(cfg.output.parent / "width-one", 1) + verified = verify_lerobot_v3(root, expected_features={"action": {"shape": [1]}}) + assert verified["info"]["features"]["action"]["shape"] == [1] + + +def test_smolvla_base_embodiment_does_not_bind_local_features(cfg): + """Real lerobot/smolvla_base features (camera1-3 at 256, 6-D) are cleared by + --policy.input_features=null; only max_*_dim padding binds.""" + base = cfg.output.parent / "smolvla-real-base" + base.mkdir() + real = { + "type": "smolvla", + "max_state_dim": 32, + "max_action_dim": 32, + "input_features": { + "observation.state": {"type": "STATE", "shape": [6]}, + **{f"observation.images.camera{i}": {"type": "VISUAL", "shape": [3, 256, 256]} for i in (1, 2, 3)}, + }, + "output_features": {"action": {"type": "ACTION", "shape": [6]}}, + } + (base / "config.json").write_text(json.dumps(real)) + seven = write_v3(cfg.output.parent / "seven-smolvla", 7) + cfg = replace(cfg, base=str(base), policy="auto", mode="lora", dataset_features={}, dataset_root=seven) + assert not run_preflight(cfg).has_failures + cmd = _build_lerobot_command(cfg) + assert "--policy.input_features=null" in cmd and f"--dataset.root={seven}" in cmd + real["max_action_dim"] = 6 + (base / "config.json").write_text(json.dumps(real)) + report = run_preflight(cfg) + assert report.has_failures and "features.action.shape" in report.error_message() + + +def write_video_v3(root, *, frames=9, timestamp_shift=0.0, long_last_frame=False, rate=10): + """One shared, real mp4 shard; episodes refer to disjoint timestamp ranges.""" + av = pytest.importorskip("av") + import numpy as np + + root = write_v3(root) + info_path = root / "meta/info.json" + info = json.loads(info_path.read_text()) + info["features"]["observation.images.wrist"]["dtype"] = "video" + info["video_path"] = "videos/{video_key}/chunk-{chunk_index:03d}/file-{file_index:03d}.mp4" + info_path.write_text(json.dumps(info)) + episode_path = root / "meta/episodes/chunk-000/file-000.parquet" + episodes = pq.read_table(episode_path).to_pylist() + for i, episode in enumerate(episodes): + prefix = "videos/observation.images.wrist/" + episode.update({prefix + "chunk_index": 0, prefix + "file_index": 0, + prefix + "from_timestamp": i * .3 + timestamp_shift, + prefix + "to_timestamp": (i + 1) * .3 + timestamp_shift}) + pq.write_table(pa.Table.from_pylist(episodes), episode_path) + video = root / "videos/observation.images.wrist/chunk-000/file-000.mp4" + video.parent.mkdir(parents=True) + with av.open(str(video), "w") as container: + stream = container.add_stream("mpeg4", rate=rate) + stream.width = stream.height = 32 + stream.pix_fmt = "yuv420p" + packets = [] + for i in range(frames): + frame = av.VideoFrame.from_ndarray(np.full((32, 32, 3), i * 10, np.uint8), format="rgb24") + packets.extend(stream.encode(frame)) + packets.extend(stream.encode()) + if long_last_frame: + packets[-1].duration = 2 # 0.2s at the encoder's 1/10 time base. + for packet in packets: + container.mux(packet) + return root + + +def test_metadata_listing_never_reads_content(cfg, monkeypatch): + import tether.finetune.lerobot_v3 as v3 + + original = v3._rows + def meta_rows(paths, label): + assert all(path.is_relative_to(cfg.dataset_root / "meta") for path in paths) + return original(paths, label) + monkeypatch.setattr(v3, "_rows", meta_rows) + monkeypatch.setattr(v3, "_file_hash", lambda *a: pytest.fail("listing hashes a file")) + monkeypatch.setattr(v3, "_video_timestamps", lambda *a: pytest.fail("listing reads video")) + described = describe_lerobot_v3(cfg.dataset_root) + assert described["metadata_checked"] + assert described["info"]["total_episodes"] == 3 + assert "content_sha256" not in described + + +@pytest.mark.parametrize("frames", [1, 8]) +def test_short_video_refused_with_field(tmp_path, frames): + root = write_video_v3(tmp_path / "short", frames=frames) + with pytest.raises(LocalDatasetError, match=r"episode_index\[.*\].videos/observation.images.wrist/to_timestamp"): + verify_lerobot_v3(root) + + +@pytest.mark.parametrize("shift,valid", [(0.0, True), (0.00009, True), (0.0001, False), (0.00011, False)]) +def test_video_uses_lerobot_timestamp_tolerance(tmp_path, shift, valid): + root = write_video_v3(tmp_path / "video", timestamp_shift=shift) + if valid: + assert verify_lerobot_v3(root)["content_sha256"] + else: + with pytest.raises(LocalDatasetError, match="videos/observation.images.wrist"): + verify_lerobot_v3(root) + + +def test_video_metadata_duration_refused(tmp_path): + root = write_video_v3(tmp_path / "video") + path = root / "meta/episodes/chunk-000/file-000.parquet" + rows = pq.read_table(path).to_pylist() + rows[0]["videos/observation.images.wrist/to_timestamp"] = .2 + pq.write_table(pa.Table.from_pylist(rows), path) + with pytest.raises(LocalDatasetError, match="videos/observation.images.wrist/to_timestamp"): + verify_lerobot_v3(root) + + +def test_metadata_listing_refuses_symlink_into_data(cfg, monkeypatch): + import tether.finetune.lerobot_v3 as v3 + path = cfg.dataset_root / "meta/info.json" + path.unlink() + path.symlink_to(cfg.dataset_root / "data/chunk-000/file-000.parquet") + monkeypatch.setattr(v3, "_json", lambda *a: pytest.fail("listing read symlink target")) + with pytest.raises(LocalDatasetError, match="meta/info.json"): + describe_lerobot_v3(cfg.dataset_root) + + +def test_decoded_video_duration_refused(tmp_path): + root = write_video_v3(tmp_path / "video", long_last_frame=True) + with pytest.raises(LocalDatasetError, match="videos/observation.images.wrist/to_timestamp.*duration"): + verify_lerobot_v3(root) + + +def test_video_tolerance_uses_lerobot_float32_rule(): + """LeRobot compares torch float32 distances; float64 disagrees on long shards.""" + from tether.finetune.lerobot_v3 import _within_lerobot_tolerance + + # Pairs checked against torch.cdist in LeRobot's runtime. Near 2000 s one + # float32 step is ~1.2e-4, so float64 gives the opposite answer for both. + assert abs(2000.00001 - 2000.000105) < 1e-4 + assert not _within_lerobot_tolerance(2000.00001, 2000.000105) + assert abs(2000.00007 - 2000.000175) >= 1e-4 + assert _within_lerobot_tolerance(2000.00007, 2000.000175) + assert _within_lerobot_tolerance(0.4, 0.40009) + assert not _within_lerobot_tolerance(0.4, 0.40011) + + +def test_video_query_uses_row_timestamp(tmp_path): + """LeRobot queries from_timestamp + the row timestamp, not frame/fps.""" + root = write_video_v3(tmp_path / "video", timestamp_shift=0.00006) + assert verify_lerobot_v3(root)["content_sha256"] + path = root / "data/chunk-000/file-000.parquet" + table = pq.read_table(path) + # Each row stays within the row-vs-fps bound (1e-4), but the sum with the + # from_timestamp shift breaks LeRobot's video tolerance. + shifted = pa.array([t + 0.00006 for t in table["timestamp"].to_pylist()], pa.float32()) + pq.write_table(table.set_column(table.schema.get_field_index("timestamp"), "timestamp", shifted), path) + with pytest.raises(LocalDatasetError, match=r"episode_index\[0\].videos/observation.images.wrist/from_timestamp"): + verify_lerobot_v3(root) + + +def test_extra_decoded_frames_refused(tmp_path): + """Every query has a frame within tolerance, but the shard holds 2x the frames.""" + root = write_video_v3(tmp_path / "video", frames=18, rate=20) + with pytest.raises(LocalDatasetError, match="to_timestamp: decoded video frame count"): + verify_lerobot_v3(root) + + +def test_metadata_duration_refused_when_decode_agrees(tmp_path): + """A long last frame makes decode end match a stretched to_timestamp.""" + root = write_video_v3(tmp_path / "video", long_last_frame=True) + path = root / "meta/episodes/chunk-000/file-000.parquet" + rows = pq.read_table(path).to_pylist() + rows[2]["videos/observation.images.wrist/to_timestamp"] = 1.0 + pq.write_table(pa.Table.from_pylist(rows), path) + with pytest.raises(LocalDatasetError, match=r"episode_index\[2\].*to_timestamp: video duration differs"): + verify_lerobot_v3(root) diff --git a/tests/test_finetune_local_dataset.py b/tests/test_finetune_local_dataset.py index 5afbb96..0758d44 100644 --- a/tests/test_finetune_local_dataset.py +++ b/tests/test_finetune_local_dataset.py @@ -568,6 +568,7 @@ def test_local_escape_hatches_block_before_training(local_cfg, field, value): "dataset.image_transforms", "policy.max_action_dim", "policy.max_state_dim", + "tolerance_s", "--dataset.root", "dataset.root=evil", ],