Examples

Two copy-pasteable walkthroughs that take a real robot dataset all the way to a packed KHLP artifact you can train on:

▸HDF5 — LIBERO Object, nested per-task HDF5 files.
▸LeRobot — a bimanual four-camera LeRobot v3 dataset.

Both follow the same shape. Ingest once, then pack the same ingested episodes twice to get both storage profiles:

raw dataset
  -> knonik ingest
  -> ingested episode directories
  -> pack once as SHDR      (storage_profile="compact_video")
  -> pack once as KDELTA    (storage_profile="training_compressed")
  -> hlp_manifest.json for each packed dataset

Ingest is the expensive step; packing is a pack-time transcode. Two artifacts built from the same INGEST_DIR are built from the same frames.

Desired packed format	`storage_profile`
SHDR	`compact_video`
KDELTA	`training_compressed`

Each example can be run from the terminal (knonik ingest + python -m knonik_ingest pack) or from one Python script (knonik.ingest.run(...) + knonik.ingest.pack(...)).

Prerequisites

Create a uv environment and install the Knonik wheel:

uv venv --python 3.11 .venv
source .venv/bin/activate
uv pip install /path/to/knonik-0.1.0-cp311-cp311-manylinux*.whl

knonik login --product ingest

On a headless machine without an OS keyring, use:

knonik login --product ingest --allow-file-key-store

Check the install:

knonik --version
python -c "import knonik.ingest, knonik_ingest; print('Knonik import OK')"

Why python -m knonik_ingest pack instead of knonik pack? The friendly knonik pack command creates the default compact_video artifact. The lower-level packer module exposes --storage-profile, which is what you need to create KDELTA from the same ingested episodes.

1. HDF5 (LIBERO Object)

LIBERO Object ships as HDF5 files, one file per task:

libero_object/
  pick_up_the_alphabet_soup_and_place_it_in_the_basket_demo.hdf5
  pick_up_the_black_bowl_and_place_it_on_the_plate_demo.hdf5
  ...

Each task file contains nested demos:

data/demo_0/obs/agentview_rgb
data/demo_0/obs/eye_in_hand_rgb
data/demo_0/robot_states
data/demo_0/actions

Knonik's episodic HDF5 reader expects one file per episode, so the first step stages those nested demos into flat per-episode HDF5 files. After that, Knonik reads the staged directory with dir_type: "episodic" and data_type: "hdf5".

1.1 Choose paths

Change only these two values:

export LIBERO_RAW=/absolute/path/to/libero_object
export LIBERO_WORK=/absolute/path/to/knonik_libero_example

Create the working paths:

mkdir -p "$LIBERO_WORK"

export LIBERO_STAGE_SCRIPT="$LIBERO_WORK/stage_libero_hdf5.py"
export LIBERO_EPISODE_DIR="$LIBERO_WORK/libero_object_episode_hdf5"
export LIBERO_INGEST_CONFIG="$LIBERO_WORK/libero_object_hdf5_ingest.json"
export LIBERO_INGEST_DIR="$LIBERO_WORK/libero_object_ingested"
export LIBERO_SHDR_DIR="$LIBERO_WORK/libero_object_khlp_shdr"
export LIBERO_KDELTA_DIR="$LIBERO_WORK/libero_object_khlp_kdelta"

1.2 Stage the nested demos

Write the staging script:

cat > "$LIBERO_STAGE_SCRIPT" <<'PY'
#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import re
import shutil
from pathlib import Path
from typing import Any

import h5py
import numpy as np


def demo_index(name: str) -> int:
    match = re.fullmatch(r"demo_(\d+)", name)
    if match is None:
        raise ValueError(f"unexpected demo key: {name!r}")
    return int(match.group(1))


def nested_get(group: h5py.Group, path: str) -> h5py.Dataset:
    obj: Any = group
    for part in path.split("/"):
        obj = obj[part]
    if not isinstance(obj, h5py.Dataset):
        raise TypeError(f"{path!r} did not resolve to an HDF5 dataset")
    return obj


def parse_image_specs(raw_specs: list[str]) -> list[tuple[str, str]]:
    specs = []
    for item in raw_specs:
        if "=" not in item:
            raise SystemExit(f"--image must look like stream_name=hdf5/path, got {item!r}")
        name, path = item.split("=", 1)
        name = name.strip()
        path = path.strip().strip("/")
        if not name or not path:
            raise SystemExit(f"invalid --image value: {item!r}")
        specs.append((name, path))
    return specs


def task_name(path: Path) -> str:
    stem = path.stem
    return stem[:-5] if stem.endswith("_demo") else stem


def task_instruction(task_file: Path, root: h5py.File) -> str:
    data = root.get("data")
    if isinstance(data, h5py.Group):
        info = data.attrs.get("problem_info")
        if isinstance(info, bytes):
            info = info.decode("utf-8")
        if isinstance(info, str):
            try:
                parsed = json.loads(info)
            except json.JSONDecodeError:
                parsed = {}
            instruction = parsed.get("language_instruction")
            if instruction:
                return str(instruction)
    return task_name(task_file).replace("_", " ")


def source_files(source_dir: Path, task_file: Path | None) -> list[Path]:
    if task_file is not None:
        files = [task_file.expanduser().resolve()]
    else:
        files = sorted(source_dir.glob("*.hdf5"))
    if not files:
        raise SystemExit(f"no .hdf5 files found in {source_dir}")
    return files


def remove_existing(path: Path, force: bool) -> None:
    if not path.exists():
        return
    if not force:
        raise SystemExit(f"output exists: {path} (rerun with --force to replace it)")
    if path.is_dir():
        shutil.rmtree(path)
    else:
        path.unlink()


def stage(args: argparse.Namespace) -> None:
    source_dir = args.source_dir.expanduser().resolve()
    out_dir = args.out_dir.expanduser().resolve()
    image_specs = args.image or [
        "agentview_rgb=obs/agentview_rgb",
        "eye_in_hand_rgb=obs/eye_in_hand_rgb",
    ]
    images = parse_image_specs(image_specs)
    files = source_files(source_dir, args.task_file)

    remove_existing(out_dir, args.force)
    out_dir.mkdir(parents=True, exist_ok=True)

    episode_index = 0
    summary = {
        "source_dir": str(source_dir),
        "fps": int(args.fps),
        "images": [{"name": name, "path": path} for name, path in images],
        "numeric": {
            "state": args.state_path,
            "proprio": args.state_path,
            "action": args.action_path,
        },
        "tasks": [],
    }

    for task_id, task_path in enumerate(files):
        with h5py.File(task_path, "r") as src:
            demos = sorted(src["data"].keys(), key=demo_index)
            if args.max_episodes_per_task is not None:
                demos = demos[: args.max_episodes_per_task]
            instruction = task_instruction(task_path, src)
            summary["tasks"].append(
                {
                    "task_id": task_id,
                    "task_name": task_name(task_path),
                    "source_file": str(task_path),
                    "instruction": instruction,
                    "episodes": len(demos),
                }
            )

            for demo in demos:
                group = src["data"][demo]
                action = np.asarray(nested_get(group, args.action_path), dtype=np.float32)
                state = np.asarray(nested_get(group, args.state_path), dtype=np.float32)
                length = int(min(len(action), len(state)))
                image_arrays = {}
                for stream_name, h5_path in images:
                    image = np.asarray(nested_get(group, h5_path), dtype=np.uint8)
                    length = min(length, int(len(image)))
                    image_arrays[stream_name] = image
                if length <= 0:
                    continue

                out_path = out_dir / f"episode_{episode_index:06d}.hdf5"
                with h5py.File(out_path, "w") as dst:
                    dst.attrs["source_file"] = str(task_path)
                    dst.attrs["source_demo"] = demo
                    dst.attrs["task_name"] = task_name(task_path)
                    dst.attrs["language_instruction"] = instruction
                    dst.attrs["metadata_json"] = json.dumps(
                        {
                            "source_file": str(task_path),
                            "source_demo": demo,
                            "task_id": task_id,
                            "task_name": task_name(task_path),
                            "language_instruction": instruction,
                        }
                    )
                    dst.create_dataset("timestamps", data=np.arange(length) / float(args.fps))
                    for stream_name, image in image_arrays.items():
                        dst.create_dataset(stream_name, data=image[:length], compression="lzf")
                    dst.create_dataset("state", data=state[:length], compression="lzf")
                    dst.create_dataset("proprio", data=state[:length], compression="lzf")
                    dst.create_dataset("action", data=action[:length], compression="lzf")
                    dst.create_dataset("task_id", data=np.full(length, task_id, dtype=np.int16))
                episode_index += 1

    summary["episodes"] = episode_index
    (out_dir / "staging_summary.json").write_text(json.dumps(summary, indent=2) + "\n")
    print(f"wrote {episode_index} staged episode HDF5 files -> {out_dir}")
    print(f"summary: {out_dir / 'staging_summary.json'}")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--source-dir", required=True, type=Path)
    parser.add_argument("--out-dir", required=True, type=Path)
    parser.add_argument("--task-file", default=None, type=Path)
    parser.add_argument("--fps", type=int, default=20)
    parser.add_argument(
        "--image",
        action="append",
        default=None,
        help=(
            "Repeat as stream_name=hdf5/path. If omitted, the two common LIBERO "
            "RGB streams are staged: agentview_rgb and eye_in_hand_rgb."
        ),
    )
    parser.add_argument("--state-path", default="robot_states")
    parser.add_argument("--action-path", default="actions")
    parser.add_argument("--max-episodes-per-task", type=int, default=None)
    parser.add_argument("--force", action="store_true")
    stage(parser.parse_args())


if __name__ == "__main__":
    main()
PY

Stage LIBERO into episodic HDF5 files:

python "$LIBERO_STAGE_SCRIPT" \
  --source-dir "$LIBERO_RAW" \
  --out-dir "$LIBERO_EPISODE_DIR" \
  --fps 20 \
  --force

If your LIBERO files do not contain obs/eye_in_hand_rgb, run the same command with only the agent-view camera:

python "$LIBERO_STAGE_SCRIPT" \
  --source-dir "$LIBERO_RAW" \
  --out-dir "$LIBERO_EPISODE_DIR" \
  --fps 20 \
  --image agentview_rgb=obs/agentview_rgb \
  --force

1.3 Write the ingest config

The data_dir value is a placeholder because the command below passes --input, which overrides it.

cat > "$LIBERO_INGEST_CONFIG" <<'JSON'
{
  "dir_type": "episodic",
  "data_type": "hdf5",
  "data_dir": "/overridden/by/the/--input/flag",
  "fps": 20,
  "preset": "medium",
  "preset_overrides": {
    "rgb": {
      "codec": "jpeg",
      "quality": 90,
      "gop": 5,
      "shard_size": 260
    }
  },
  "metadata": {
    "dataset": "libero_object_hdf5",
    "robot_type": "libero_single_arm",
    "layout": "agentview_rgb + eye_in_hand_rgb + state + proprio + action"
  },
  "layout": {
    "images": {
      "agentview_rgb": {
        "key": "agentview_rgb",
        "fps": 20
      },
      "eye_in_hand_rgb": {
        "key": "eye_in_hand_rgb",
        "fps": 20
      }
    },
    "numeric": {
      "state": "state",
      "proprio": "proprio",
      "action": "action"
    },
    "timestamps": "timestamps"
  }
}
JSON

If you staged only agentview_rgb, remove the eye_in_hand_rgb entry from layout.images before ingesting.

1.4 Ingest and pack

Ingest the staged HDF5 episodes:

knonik ingest \
  --config "$LIBERO_INGEST_CONFIG" \
  --input "$LIBERO_EPISODE_DIR" \
  --output "$LIBERO_INGEST_DIR"

Alongside the episodes this writes $LIBERO_INGEST_DIR/config.json, the dataset config the quality agent and dashboard read. Packing and training do not need it, so you can go straight on; if you plan to quality-check this dataset, fill in the <FILL:...> values it leaves for you first — see Processing §6.

Pack as SHDR:

python -m knonik_ingest pack \
  --input-dir "$LIBERO_INGEST_DIR" \
  --output-dir "$LIBERO_SHDR_DIR" \
  --dataset-name libero_object_hdf5_shdr \
  --shard-size 260 \
  --window-size 64 \
  --fps 20 \
  --pack-mode shard_batch \
  --storage-profile compact_video

Pack as KDELTA:

python -m knonik_ingest pack \
  --input-dir "$LIBERO_INGEST_DIR" \
  --output-dir "$LIBERO_KDELTA_DIR" \
  --dataset-name libero_object_hdf5_kdelta \
  --shard-size 260 \
  --window-size 64 \
  --fps 20 \
  --pack-mode shard_batch \
  --storage-profile training_compressed

The resulting manifests are:

$LIBERO_SHDR_DIR/hlp_manifest.json
$LIBERO_KDELTA_DIR/hlp_manifest.json

1.5 The same thing as one Python script

This script stages LIBERO, ingests the staged episodes, packs SHDR, and packs KDELTA. It assumes both obs/agentview_rgb and obs/eye_in_hand_rgb exist; if your files only contain one camera, edit IMAGE_SPECS and the layout.images block in ingest_config(...) before running it.

cat > "$LIBERO_WORK/convert_libero_hdf5_to_knonik.py" <<'PY'
#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import re
import shutil
from pathlib import Path
from typing import Any

import h5py
import numpy as np

from knonik.ingest import pack, run


IMAGE_SPECS = [
    ("agentview_rgb", "obs/agentview_rgb"),
    ("eye_in_hand_rgb", "obs/eye_in_hand_rgb"),
]


def demo_index(name: str) -> int:
    match = re.fullmatch(r"demo_(\d+)", name)
    if match is None:
        raise ValueError(f"unexpected demo key: {name!r}")
    return int(match.group(1))


def nested_get(group: h5py.Group, path: str) -> h5py.Dataset:
    obj: Any = group
    for part in path.split("/"):
        obj = obj[part]
    if not isinstance(obj, h5py.Dataset):
        raise TypeError(f"{path!r} did not resolve to an HDF5 dataset")
    return obj


def task_name(path: Path) -> str:
    stem = path.stem
    return stem[:-5] if stem.endswith("_demo") else stem


def remove_existing(path: Path, force: bool) -> None:
    if not path.exists():
        return
    if not force:
        raise SystemExit(f"output exists: {path} (rerun with --force to replace it)")
    if path.is_dir():
        shutil.rmtree(path)
    else:
        path.unlink()


def stage_libero(source_dir: Path, out_dir: Path, *, fps: int, force: bool) -> None:
    remove_existing(out_dir, force)
    out_dir.mkdir(parents=True, exist_ok=True)
    files = sorted(source_dir.glob("*.hdf5"))
    if not files:
        raise SystemExit(f"no .hdf5 files found in {source_dir}")

    episode_index = 0
    for task_id, task_path in enumerate(files):
        with h5py.File(task_path, "r") as src:
            demos = sorted(src["data"].keys(), key=demo_index)
            for demo in demos:
                group = src["data"][demo]
                action = np.asarray(nested_get(group, "actions"), dtype=np.float32)
                state = np.asarray(nested_get(group, "robot_states"), dtype=np.float32)
                images = {
                    name: np.asarray(nested_get(group, h5_path), dtype=np.uint8)
                    for name, h5_path in IMAGE_SPECS
                }
                length = min([len(action), len(state), *(len(v) for v in images.values())])
                if length <= 0:
                    continue

                out_path = out_dir / f"episode_{episode_index:06d}.hdf5"
                with h5py.File(out_path, "w") as dst:
                    dst.attrs["source_file"] = str(task_path)
                    dst.attrs["source_demo"] = demo
                    dst.attrs["task_name"] = task_name(task_path)
                    dst.create_dataset("timestamps", data=np.arange(length) / float(fps))
                    for name, image in images.items():
                        dst.create_dataset(name, data=image[:length], compression="lzf")
                    dst.create_dataset("state", data=state[:length], compression="lzf")
                    dst.create_dataset("proprio", data=state[:length], compression="lzf")
                    dst.create_dataset("action", data=action[:length], compression="lzf")
                    dst.create_dataset("task_id", data=np.full(length, task_id, dtype=np.int16))
                episode_index += 1
    print(f"staged {episode_index} LIBERO episodes -> {out_dir}")


def ingest_config(episode_dir: Path, fps: int) -> dict:
    return {
        "dir_type": "episodic",
        "data_type": "hdf5",
        "data_dir": str(episode_dir),
        "fps": fps,
        "preset": "medium",
        "preset_overrides": {
            "rgb": {
                "codec": "jpeg",
                "quality": 90,
                "gop": 5,
                "shard_size": 260,
            }
        },
        "metadata": {
            "dataset": "libero_object_hdf5",
            "robot_type": "libero_single_arm",
            "layout": "agentview_rgb + eye_in_hand_rgb + state + proprio + action",
        },
        "layout": {
            "images": {
                "agentview_rgb": {"key": "agentview_rgb", "fps": fps},
                "eye_in_hand_rgb": {"key": "eye_in_hand_rgb", "fps": fps},
            },
            "numeric": {
                "state": "state",
                "proprio": "proprio",
                "action": "action",
            },
            "timestamps": "timestamps",
        },
    }


def verify_manifest(root: Path, expected_profile: str) -> None:
    manifest_path = root / "hlp_manifest.json"
    if not manifest_path.exists():
        raise RuntimeError(f"missing manifest: {manifest_path}")
    manifest = json.loads(manifest_path.read_text())
    print(f"\n{root.name}")
    print(f"  manifest: {manifest_path}")
    print(f"  storage_profile: {manifest.get('storage_profile')}")
    print(f"  num_hlps: {manifest.get('num_hlps')}")
    if manifest.get("storage_profile") != expected_profile:
        raise RuntimeError(f"{root}: wrong storage_profile")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--raw", required=True, type=Path, help="Directory of LIBERO HDF5 files")
    parser.add_argument("--work", required=True, type=Path, help="Output working directory")
    parser.add_argument("--fps", type=int, default=20)
    parser.add_argument("--force", action="store_true")
    args = parser.parse_args()

    raw_dir = args.raw.expanduser().resolve()
    work_dir = args.work.expanduser().resolve()
    episode_dir = work_dir / "libero_object_episode_hdf5"
    ingest_dir = work_dir / "libero_object_ingested"
    shdr_dir = work_dir / "libero_object_khlp_shdr"
    kdelta_dir = work_dir / "libero_object_khlp_kdelta"

    work_dir.mkdir(parents=True, exist_ok=True)
    for path in (episode_dir, ingest_dir, shdr_dir, kdelta_dir):
        remove_existing(path, args.force)

    stage_libero(raw_dir, episode_dir, fps=args.fps, force=args.force)
    run(config=ingest_config(episode_dir, args.fps), output=ingest_dir)

    pack(
        input_dir=ingest_dir,
        output_dir=shdr_dir,
        dataset_name="libero_object_hdf5_shdr",
        shard_size=260,
        window_size=64,
        fps=args.fps,
        seed=0,
        pack_mode="shard_batch",
        storage_profile="compact_video",
    )
    pack(
        input_dir=ingest_dir,
        output_dir=kdelta_dir,
        dataset_name="libero_object_hdf5_kdelta",
        shard_size=260,
        window_size=64,
        fps=args.fps,
        seed=0,
        pack_mode="shard_batch",
        storage_profile="training_compressed",
    )

    verify_manifest(shdr_dir, "compact_video")
    verify_manifest(kdelta_dir, "training_compressed")
    print("\nOK: LIBERO conversion complete.")


if __name__ == "__main__":
    main()
PY

Run it:

python "$LIBERO_WORK/convert_libero_hdf5_to_knonik.py" \
  --raw "$LIBERO_RAW" \
  --work "$LIBERO_WORK" \
  --force

2. LeRobot v3 (Sample Dataset)

This example uses a sample bimanual LeRobot v3 dataset: a two-arm robot doing a tabletop manipulation task (for example, picking up an object and placing it in a bin), with four RGB cameras and 14-dimensional state and action vectors. It ingests all four cameras plus state and action:

cam_high
cam_low
cam_left_wrist
cam_right_wrist

No staging step is needed — Knonik reads LeRobot v3 layouts directly. The dataset should already be on disk in the standard layout:

bimanual_sample/
  data/
  meta/
    info.json
    episodes/
    tasks.parquet
  videos/

2.1 Choose paths

Change only these two values:

export SAMPLE_RAW=/absolute/path/to/bimanual_sample
export KNONIK_WORK=/absolute/path/to/knonik_bimanual_sample_example

Create the output directories:

mkdir -p "$KNONIK_WORK"

export INGEST_CONFIG="$KNONIK_WORK/bimanual_sample_lerobot_v3_all_cameras.json"
export INGEST_DIR="$KNONIK_WORK/bimanual_sample_ingested"
export SHDR_DIR="$KNONIK_WORK/bimanual_sample_khlp_shdr"
export KDELTA_DIR="$KNONIK_WORK/bimanual_sample_khlp_kdelta"

2.2 Write the ingest config

This config tells Knonik how to read the LeRobot v3 dataset. The data_dir value is a placeholder because the command below passes --input, which overrides it.

cat > "$INGEST_CONFIG" <<'JSON'
{
  "dir_type": "lerobot_v3",
  "data_dir": "/overridden/by/the/--input/flag",
  "fps": 50,
  "preset": "medium",
  "preset_overrides": {
    "rgb": {
      "codec": "jpeg",
      "quality": 90,
      "gop": 5,
      "shard_size": 260
    }
  },
  "metadata": {
    "dataset": "bimanual_sample_all_cameras",
    "robot_type": "bimanual",
    "layout": "cam_high + cam_low + cam_left_wrist + cam_right_wrist RGB + state(14) + action(14)"
  },
  "layout": {
    "images": {
      "cam_high": {
        "key": "cam_high",
        "fps": 50
      },
      "cam_low": {
        "key": "cam_low",
        "fps": 50
      },
      "cam_left_wrist": {
        "key": "cam_left_wrist",
        "fps": 50
      },
      "cam_right_wrist": {
        "key": "cam_right_wrist",
        "fps": 50
      }
    },
    "numeric": {
      "state": "state",
      "action": "action"
    }
  }
}
JSON

For LeRobot datasets, image keys are bare camera names such as cam_high, not observation.images.cam_high. Numeric streams use the logical names state and action.

To build a smaller artifact, remove any cameras you do not need from layout.images. Everything else stays the same — Knonik decodes and stores only the streams you name in the layout.

2.3 Ingest and pack

knonik ingest \
  --config "$INGEST_CONFIG" \
  --input "$SAMPLE_RAW" \
  --output "$INGEST_DIR"

After this finishes, $INGEST_DIR contains one Knonik episode directory per source episode, plus the generated config.json. This is the reusable intermediate; do not ingest again just to try another packed format — and note that re-running ingest into the same directory leaves an existing config.json alone unless you pass --force-config.

Because LeRobot declares its own action and observation.state columns, the generated config already binds both roles and carries the per-channel joint names from meta/info.json; what it leaves to you is the arm layout and the action semantics.

Quick check:

find "$INGEST_DIR" -maxdepth 1 -type d -name 'episode_*' | sort | head

Pack as SHDR:

python -m knonik_ingest pack \
  --input-dir "$INGEST_DIR" \
  --output-dir "$SHDR_DIR" \
  --dataset-name bimanual_sample_all_cameras_shdr \
  --shard-size 260 \
  --window-size 64 \
  --fps 50 \
  --pack-mode shard_batch \
  --storage-profile compact_video

Pack as KDELTA:

python -m knonik_ingest pack \
  --input-dir "$INGEST_DIR" \
  --output-dir "$KDELTA_DIR" \
  --dataset-name bimanual_sample_all_cameras_kdelta \
  --shard-size 260 \
  --window-size 64 \
  --fps 50 \
  --pack-mode shard_batch \
  --storage-profile training_compressed

The packed dataset entry points are:

$SHDR_DIR/hlp_manifest.json
$KDELTA_DIR/hlp_manifest.json

2.4 The same thing as one Python script

This script ingests the sample dataset and packs both formats using the public knonik.ingest API.

cat > "$KNONIK_WORK/convert_bimanual_sample_to_knonik.py" <<'PY'
#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import shutil
from pathlib import Path

from knonik.ingest import pack, run


def build_config(raw_dir: Path) -> dict:
    return {
        "dir_type": "lerobot_v3",
        "data_dir": str(raw_dir),
        "fps": 50,
        "preset": "medium",
        "preset_overrides": {
            "rgb": {
                "codec": "jpeg",
                "quality": 90,
                "gop": 5,
                "shard_size": 260,
            }
        },
        "metadata": {
            "dataset": "bimanual_sample_all_cameras",
            "robot_type": "bimanual",
            "layout": "cam_high + cam_low + cam_left_wrist + cam_right_wrist RGB + state(14) + action(14)",
        },
        "layout": {
            "images": {
                "cam_high": {"key": "cam_high", "fps": 50},
                "cam_low": {"key": "cam_low", "fps": 50},
                "cam_left_wrist": {"key": "cam_left_wrist", "fps": 50},
                "cam_right_wrist": {"key": "cam_right_wrist", "fps": 50},
            },
            "numeric": {
                "state": "state",
                "action": "action",
            },
        },
    }


def remove_existing(path: Path, force: bool) -> None:
    if not path.exists():
        return
    if not force:
        raise SystemExit(f"output exists: {path} (rerun with --force to replace it)")
    if path.is_dir():
        shutil.rmtree(path)
    else:
        path.unlink()


def verify_manifest(root: Path, expected_profile: str) -> None:
    manifest_path = root / "hlp_manifest.json"
    if not manifest_path.exists():
        raise RuntimeError(f"missing manifest: {manifest_path}")
    manifest = json.loads(manifest_path.read_text())

    print(f"\n{root.name}")
    print(f"  manifest: {manifest_path}")
    print(f"  storage_profile: {manifest.get('storage_profile')}")
    print(f"  num_hlps: {manifest.get('num_hlps')}")

    if manifest.get("storage_profile") != expected_profile:
        raise RuntimeError(
            f"{root}: expected storage_profile={expected_profile!r}, "
            f"got {manifest.get('storage_profile')!r}"
        )


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--raw", required=True, type=Path, help="LeRobot v3 dataset root")
    parser.add_argument("--work", required=True, type=Path, help="Output working directory")
    parser.add_argument("--force", action="store_true", help="Replace existing outputs")
    args = parser.parse_args()

    raw_dir = args.raw.expanduser().resolve()
    work_dir = args.work.expanduser().resolve()
    ingest_dir = work_dir / "bimanual_sample_ingested"
    shdr_dir = work_dir / "bimanual_sample_khlp_shdr"
    kdelta_dir = work_dir / "bimanual_sample_khlp_kdelta"

    if not raw_dir.exists():
        raise SystemExit(f"raw dataset does not exist: {raw_dir}")

    work_dir.mkdir(parents=True, exist_ok=True)
    for path in (ingest_dir, shdr_dir, kdelta_dir):
        remove_existing(path, args.force)

    print(f"ingesting {raw_dir} -> {ingest_dir}")
    run(config=build_config(raw_dir), output=ingest_dir)

    print(f"packing SHDR -> {shdr_dir}")
    pack(
        input_dir=ingest_dir,
        output_dir=shdr_dir,
        dataset_name="bimanual_sample_all_cameras_shdr",
        shard_size=260,
        window_size=64,
        fps=50,
        seed=0,
        pack_mode="shard_batch",
        storage_profile="compact_video",
    )

    print(f"packing KDELTA -> {kdelta_dir}")
    pack(
        input_dir=ingest_dir,
        output_dir=kdelta_dir,
        dataset_name="bimanual_sample_all_cameras_kdelta",
        shard_size=260,
        window_size=64,
        fps=50,
        seed=0,
        pack_mode="shard_batch",
        storage_profile="training_compressed",
    )

    verify_manifest(shdr_dir, expected_profile="compact_video")
    verify_manifest(kdelta_dir, expected_profile="training_compressed")
    print("\nOK: conversion complete.")


if __name__ == "__main__":
    main()
PY

Run it:

python "$KNONIK_WORK/convert_bimanual_sample_to_knonik.py" \
  --raw "$SAMPLE_RAW" \
  --work "$KNONIK_WORK"

Add --force to replace previous outputs:

python "$KNONIK_WORK/convert_bimanual_sample_to_knonik.py" \
  --raw "$SAMPLE_RAW" \
  --work "$KNONIK_WORK" \
  --force

3. Adapting This To Your Dataset

The recipe has two dataset-specific pieces:

▸dir_type, which selects the source reader.
▸layout, which maps source stream names to Knonik stream names.

For another LeRobot v3 dataset, keep "dir_type": "lerobot_v3" and change only the camera names and metadata. For another episodic HDF5 dataset, use the episodic shape:

{
  "dir_type": "episodic",
  "data_type": "hdf5",
  "data_dir": "/path/to/episode_h5_directory",
  "fps": 30,
  "preset": "medium",
  "layout": {
    "images": {
      "top": {
        "key": "observations.images.top",
        "fps": 30
      }
    },
    "numeric": {
      "state": "observations.qpos",
      "action": "action"
    },
    "timestamps": "timestamps"
  }
}

For formats without a built-in reader, stage one episode as .npy files and use dir_type: "npy_separate", or stream frames directly with knonik.ingest.live_session(...).

Choose shard_size at least as large as the largest temporal window your training loader will request. Both examples use gop=5 and shard_size=260, with window_size=64 for shard-batch packing.

4. Recommended Dataloader Params

Use the same shared dataloader settings for both examples, then switch only the payload-specific block depending on whether you train from the SHDR or KDELTA manifest.

4.1 Dataset-specific values

Set the manifest for the packed dataset you are loading:

Dataset	SHDR manifest	KDELTA manifest
LIBERO	`$LIBERO_SHDR_DIR/hlp_manifest.json`	`$LIBERO_KDELTA_DIR/hlp_manifest.json`
Sample dataset	`$SHDR_DIR/hlp_manifest.json`	`$KDELTA_DIR/hlp_manifest.json`

Set knonik.image_key to the image stream your model consumes:

Dataset	Common `knonik.image_key` values
LIBERO	`agentview_rgb`, `eye_in_hand_rgb`
Sample dataset	`cam_high`, `cam_low`, `cam_left_wrist`, `cam_right_wrist`

The action stream is action for both examples.

4.2 Shared loader params

Use these for both datasets and both storage profiles:

knonik:
  framework: torch
  action_key: action

  num_fetchers: 8
  num_decoders: 8
  decode_concurrency: 8
  prefetch_shards: 12
  precollate_batches: 4
  batch_prefetch_timeout_s: 0.1
  batch_prefetch_daemon: true
  io_loops: 1
  num_orch_threads: null

  delta_cache_block_planning: 'on'
  delta_cache_block_max_shards: 16

  partial_decode_coalesce_batches: 8
  partial_decode_window_prefetch: 2

  pad_missing: true
  delta_rounding: nearest

Keep max_steps tied to your training run length so the loader plans the whole run up front:

max_steps: 4000

knonik:
  max_steps: ${max_steps}

4.3 KDELTA loader params

Use this block when loading a dataset packed with storage_profile="training_compressed":

defaults:
  - knonik: kdelta

knonik:
  payload: kdelta
  shuffle_mode: global_triplet_fast
  planner_locality: hlp_bounded

  kdelta_direct_local_decode: true
  kdelta_direct_bundle_decode: true
  kdelta_direct_bundle_target_frames: 48
  kdelta_direct_single_consumer: false

This is the recommended profile for random-access training on local files. Start with planner_locality: hlp_bounded; try planner_locality: global only if you want more batch diversity and have enough RAM.

4.4 SHDR loader params

Use this block when loading a dataset packed with storage_profile="compact_video":

defaults:
  - knonik: shdr

knonik:
  payload: shdr
  shuffle_mode: global_triplet
  planner_locality: global

  kdelta_direct_local_decode: false

For SHDR, keep shuffle_mode: global_triplet. The global_triplet_fast path is meant for KDELTA-style training reads and is not the best default for SHDR.