Knonik Ingest

This guide explains how to use Knonik ingest from the wheel-based Knonik package.

You can use ingest in two ways:

• ▸Terminal: knonik ingest ... and knonik pack ...
• ▸Python: knonik.ingest.run(...) and knonik.ingest.pack(...)

1. What Ingest Does

Knonik ingest converts raw robot datasets into Knonik compressed episode directories. Those episode directories can then be packed into KHLP shards for the Knonik multidataloader.

Supported source families include:

• ▸HDF5
• ▸NPY / NPZ
• ▸Zarr (per-episode or consolidated)
• ▸LeRobot v2
• ▸LeRobot v3
• ▸ROS bags — ROS1 .bag and ROS2 MCAP / SQLite3 (.mcap, .db3, or a ROS2 bag directory)
• ▸RLDS / TFRecord, when the required optional dependencies are available
• ▸D3IL

Unsupported format? You do not have to wait for a built-in reader. You can ingest any format into Knonik using one of two generic adapters — an offline one (dir_type: "npy_separate") and a streaming one (live ingest). See section 7.

The normal workflow is:

1. ▸Install the Knonik wheel.
2. ▸Log in once with an account entitled for ingest.
3. ▸Run offline ingest.
4. ▸Pack the ingested episodes into KHLP shards.
5. ▸Train with the Knonik multidataloader.

Need to go the other way? The Export guide turns Knonik data back into HDF5, Zarr, LeRobot, ROS bag / MCAP, and more.

2. Install

Create or activate the Python environment you want to use:

python -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip

Install the wheel provided by Knonik:

python -m pip install /path/to/knonik-0.1.0-cp311-cp311-manylinux*.whl

3. Log In

Log in once before using ingest:

knonik login --product ingest

On headless Linux machines without an OS keyring, use the explicit file-key fallback:

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

Check status:

knonik status --product ingest

Log out:

knonik logout --product ingest

4. Offline Ingest From Terminal

Run offline ingest with:

knonik ingest \
  --config /path/to/ingest_config.json \
  --input  /path/to/raw_dataset \
  --output /path/to/output_dir

The --config file describes your dataset — its on-disk layout, which keys hold each camera/state/action stream, and how to compress them. See section 6 for how to write one.

config files usually have data_dir in them, the --input flag overrides this. This helps in using the same config across multiple machines without any local dir and path issues.

Example:

knonik ingest \
  --config /home/me/configs/config_lerobot_v3.json \
  --input  /home/me/data/thread_velcro \
  --output /home/me/data/out/knonik_thread_velcro

The dataset config ingest leaves behind

Every run finishes by writing a dataset config to <output>/config.json — a different file from the ingest config you passed in, describing the episodes that came out rather than the raw data that went in. It is what the quality agent, the visualizer, and auto-segmentation read. Ingest fills in everything it can prove (stream names, widths, codecs, frame rate, timestamp semantics, and per-channel joint names where the source declares them) and marks the semantic parts — which stream is the action, where the arms are — with <FILL:...> placeholders for you to complete. See Processing §6.

An existing config.json in the output directory is never overwritten. Add --force-config to regenerate it:

knonik ingest --config ingest.json --input /raw --output /out --force-config

5. Pack Into KHLP From Terminal

Default packing: After ingest writes episode directories and you have processed it (quality check, visualize and annotation), pack them for the dataloader:

knonik pack \
  --input-dir    /path/to/output_dir/knonik_dataset \
  --output-dir   /path/to/output_dir \
  --dataset-name my_dataset \
  --shard-size   45 \
  --window-size  64 \
  --fps          30

Example:

knonik pack \
  --input-dir    /home/me/data/out/knonik_thread_velcro \
  --output-dir   /home/me/data/out \
  --dataset-name thread_velcro \
  --shard-size   45 \
  --window-size  64 \
  --fps          50

This writes KHLP files and hlp_manifest.json into a packed output directory. The manifest path is the main input to ShardedDataLoader.

FOR ADVANCED PACKERS (OPTIMIZED FOR SPECIFIC USE CASES, SEE THE CONFIGS BELOW)

Storage Profiles

The packer can store RGB differently depending on the training workload. This is a pack-time transcode: you do not need to re-ingest the raw data to try a different storage profile.

Use compact_video (the default) for traditional video-shard behavior. Use training_fast for the fastest random-access training reads. Use training_compressed when frames are redundant and you want a smaller artifact with good sparse-access throughput.

Selecting a Storage Profile

knonik pack always uses the default compact_video profile. To pack with training_fast or training_compressed, use the packer module (which exposes --storage-profile) or the Python API (section 9):

python -m knonik_ingest pack \
  --input-dir       /datasets/my_dataset_ingested \
  --output-dir      /datasets/my_dataset_fast \
  --dataset-name    my_dataset \
  --shard-size      30 \
  --window-size     64 \
  --fps             50 \
  --pack-mode       shard_batch \
  --storage-profile training_fast        # or training_compressed

Quick Manifest Check

You can inspect the packed manifest without loading the dataset:

python - <<'PY'
import json
from pathlib import Path

data = json.loads(Path("/datasets/my_dataset_fast/hlp_manifest.json").read_text())
for key in ("dataset", "shard_size", "fps", "pack_mode", "storage_profile", "num_hlps"):
    if key in data:
        print(f"{key}: {data[key]}")
PY

A packed manifest looks like:

{
  "dataset": "my_dataset",
  "shard_size": 30,
  "fps": 50,
  "pack_mode": "shard_batch",
  "storage_profile": "training_fast",
  "num_hlps": 12,
  "hlps": [ "..." ]
}

Packer options:

6. Ingest Config

The config tells ingest what your raw dataset looks like: its on-disk layout, which keys hold each camera / state / action stream, and how aggressively to compress them. Pass it with --config (terminal) or as the config= argument (Python). Config files may be written as JSON or YAML.

6.1 Writing a Config For Your Own Dataset

A config is made of a few top-level blocks. At a glance:

¹ data_dir is optional if you always pass --input, which overrides it.

Step 1 — Pick dir_type

dir_type selects the reader for your dataset's on-disk layout:

For episodic, add data_type to say what is inside each episode directory:

episodic file naming. When reading a format directly, ingest globs data_dir for a fixed pattern — anything named differently is invisible:

data_type	Expects, directly inside data_dir
hdf5	episode_*.h5 or episode_*.hdf5
npy	episode_*.npz (one npz archive per episode, holding the arrays)
zarr	episode_<n>/ directories, with an integer <n> — sorted numerically

Episodes are ordered by that glob, so episode_0…episode_9 sort before episode_10 only for zarr; zero-pad your names (episode_000007) if the order matters for HDF5/NPZ.

Step 2 — Map your streams with layout

layout is the heart of the config: it tells ingest which keys in your source data become which output streams. The entry name is the output stream name you choose; the value is the source path in your raw data.

Source paths use dot notation to address nested keys:

LeRobot naming. For lerobot_v2 / lerobot_v3, image keys are the bare camera name (e.g. "cam_high", not "observation.images.cam_high"). These readers expose exactly two numeric streams — the primary proprioception column as "state" and the primary command column as "action" — so the values in layout.numeric must be "state" or "action" (the names on the left are yours to choose). Omit layout entirely to take every camera plus state / action as-is.

How much layout each reader uses

layout means slightly different things depending on which reader dir_type selects. This is the single most common source of confusion, so check your row before writing the block:

Unknown names are skipped, not errors. For the format readers (everything except episodic and npy_separate), a layout.images entry whose key is not present in the data is silently dropped — you get an episode without that camera rather than a failure. If a stream is missing from the output, check the spelling against episode_000000/metadata.json (see below) before anything else.

Check what you actually got

Every ingested episode records the stream names it wrote. After a short trial run (add "episode_limit": 1 if your dir_type is a format reader, or just point --input at a copy holding one episode), read them back:

python -c "import json;m=json.load(open('OUT/episode_000000/metadata.json'));print(m['rgb_streams'],m['depth_streams'],m['numeric_streams'])"

Those exact names are what you feed to the processing dataset config and what the dataloader requests, so it is worth doing before you ingest a thousand episodes.

Step 3 — Set compression with preset

Either name a built-in preset ("high", "medium", …) or give an explicit {rgb, depth, numeric} dict. It defaults to medium. See section 8 for every knob.

Putting it together

A minimal episodic-HDF5 config using all of the above:

{
  "dir_type": "episodic",
  "data_type": "hdf5",
  "data_dir": "/home/me/data/raw_bimanual",
  "fps": 50,
  "metadata": {
    "dataset": "bimanual_transfer",
    "robot_type": "bimanual",
    "layout": "two camera streams plus state/action"
  },
  "preset": {
    "rgb": {"codec": "vp9", "quality": 18, "gop": 3},
    "depth": {"quantization_mm": 1},
    "numeric": {"level": 1}
  },
  "layout": {
    "images": {
      "cam_high": {"key": "observations.images.cam_high", "fps": 50},
      "cam_left": {"key": "observations.images.cam_left", "fps": 50}
    },
    "numeric": {
      "state": "observations.qpos",
      "action": "action"
    },
    "timestamps": "timestamps"
  }
}

Full field reference

6.2 ROS Bags and MCAP

ROS recordings are the one family where you mostly do not write a layout: the reader walks the bag's topics and derives the streams itself. One reader handles all three container formats — ROS1 .bag, ROS2 .mcap, ROS2 .db3 — and auto-detects which it is. dir_type: "mcap" is an explicit alias for dir_type: "rosbag"; they behave identically. ROS does not need to be installed (the wheel bundles rosbags).

Step 1 — Point data_dir at the right level

data_dir may be either a single bag (one episode) or a parent directory holding many bags (one episode each). Both of these are single bags:

recording.bag                    ← ROS1 bag file
my_ros2_bag/                     ← ROS2 bag directory
├── metadata.yaml
└── my_ros2_bag_0.mcap

And both of these are multi-episode parents — point data_dir at the top:

airoa-moma-raw/                  ← one subdirectory per episode
├── episode_0001/
│   └── data.bag
├── episode_0002/
│   └── data.bag
└── …

loose_bags/                      ← or loose bag files side by side
├── run_01.mcap
├── run_02.mcap
└── run_03.mcap

The episode id is the subdirectory name (episode_0001) or the file stem (run_01). Discovery order is: single bag → one bag per subdirectory → loose .bag / .mcap / .db3 files in the directory. If none of those match, ingest raises No ROS bags found at ….

Step 2 — Know how topics become streams

Every connection in the bag is classified by message type first, topic name second:

Naming rules:

• ▸Cameras drop the boilerplate path parts (image_raw, image, image_rect, image_rect_color, rgb, color, rect, depth, depth_registered, compressed, msg) and join what is left with _: /camera/color/image_raw → camera, /left_cam/rgb/image_raw → left_cam, /head/zed/left/image_rect_color → head_zed_left.
• ▸Numeric streams keep the whole topic path, with non-alphanumeric runs collapsed to _: /joint_states → joint_states, /arm_controller/command → arm_controller_command, /robot/gripper/target → robot_gripper_target.

A stream whose first message cannot be decoded (RVL-compressed depth, for example) is skipped rather than failing the run, so a missing depth stream in the output usually means "not decodable", not "misconfigured".

Each numeric topic stays its own stream — nothing is concatenated into one opaque state vector, so you can pick out exactly the topic you want later. The vector width is fixed from the first message on that topic (later messages are padded or truncated to match), and the extracted fields are the obvious ones: JointState → position, Imu → linear acceleration + angular velocity + orientation quaternion, Odometry → position + orientation + twist, Pose/PoseStamped → position + quaternion, Twist → linear + angular, Wrench → force + torque, JointTrajectory → the first point's positions.

Step 3 — Understand the frame clock

ROS topics are asynchronous; Knonik episodes are frame-aligned. The reader picks the camera with the most messages as the clock, then samples every other stream at its nearest timestamp for each of those frames. Consequences worth knowing before you ingest:

• ▸A bag with no decodable camera produces no episode and is skipped. Numeric topics alone are not enough to anchor frames.
• ▸Frame timestamps are the bag's own nanosecond stamps, in seconds — you do not need layout.timestamps.
• ▸A slow topic (say 10 Hz odometry against a 30 Hz camera) is held at its last value between updates rather than interpolated.
• ▸fps in the config is metadata for the video encoder, not resampling. Set it near your real camera rate.

For a first pass over a huge bag, cap frames per episode with an environment variable:

KNONIK_ROSBAG_MAX_FRAMES=200 knonik ingest --config config_rosbag.json --input /bags --output /out

Step 4 — Write the config

A ROS config is short, because there is nothing to map:

{
  "dir_type": "rosbag",
  "data_dir": "/data/airoa-moma-raw",
  "preset": "medium",
  "fps": 15,
  "metadata": {
    "dataset": "moma_pick_place",
    "robot_type": "mobile_manipulator"
  }
}

The MCAP alias, pointed at a single ROS2 bag directory, with a cheaper preset for a first look:

{
  "dir_type": "mcap",
  "data_dir": "/data/ros2_bags/session_04",
  "preset": "low",
  "fps": 30,
  "episode_limit": 2,
  "metadata": {
    "dataset": "session_04_smoke_test"
  }
}

Keeping only two of six cameras — the one thing layout is good for here. The key is the derived camera name from step 2, and the name on the left is what the stream is called in the output:

{
  "dir_type": "rosbag",
  "data_dir": "/data/bags",
  "preset": "high",
  "fps": 30,
  "metadata": {
    "dataset": "two_cam_subset",
    "robot_type": "ur5e"
  },
  "layout": {
    "images": {
      "cam_head":  {"key": "head_zed_left", "fps": 30},
      "cam_wrist": {"key": "wrist_cam",     "fps": 30}
    },
    "depths": {
      "cam_head_depth": {"key": "head_zed", "fps": 30}
    }
  }
}

Two traps in that example:

• ▸Depth is filtered under layout.depths (plural) for ROS and the other format readers — layout.depth (singular) is the episodic spelling.
• ▸layout.numeric does nothing for ROS bags. Every numeric topic is ingested under its sanitized topic name whether you list it or not. To use only some of them, select the streams you want downstream (in the processing dataset config and in the dataloader), or record a narrower bag.

Step 5 — Read back the names ROS gave you

Because the names are derived rather than declared, always check them after the first episode. ROS ingest additionally records the raw mapping in the episode metadata:

python -c "import json;m=json.load(open('/out/episode_000000/metadata.json'));print(m['ros_cameras'],m['ros_depth_streams'],m['ros_numeric_streams'],m['source_bag'])"

ros_numeric_streams maps each stream name to its width — exactly what you need to fill in arms.indices in the processing dataset config. camera_info holds the intrinsics captured from any CameraInfo topics.

6.3 Examples

Complete, working configs for the common source families.

Episodic HDF5 — ALOHA bimanual. A directory of per-episode HDF5 files: one top camera, plus qpos, qvel, and action numeric streams nested under observations. Note how each layout.numeric entry renames a source path to a clean output stream name:

{
  "dir_type": "episodic",
  "data_type": "hdf5",
  "preset": {
    "rgb":     {"codec": "vp9", "quality": 15, "gop": 3},
    "depth":   {"quantization_mm": 1},
    "numeric": {"level": 1}
  },
  "fps": 50,
  "data_dir": "/data/input",

  "metadata": {
    "dataset": "bimanual_transfer",
    "robot_type": "aloha_bimanual",
    "layout": "qpos(14) + qvel(14) + action(14) + cam_top"
  },

  "layout": {
    "images": {
      "top": {"key": "observations.images.top", "fps": 50}
    },

    "numeric": {
      "qpos":    "observations.qpos",
      "qvel":    "observations.qvel",
      "actions": "action"
    }
  }
}

LeRobot v3 — four cameras. A LeRobot v3 dataset with four cameras and state/action. Because this is a LeRobot reader, the image keys are the bare camera names and the primary proprioception/command columns use the logical state / action names; a timestamp stream is also mapped. A dataset carrying extra columns would name them directly — "qvel": "observation.qvel" — or drop layout.numeric to take every numeric column as-is:

{
  "dir_type": "lerobot_v3",
  "preset": {
    "rgb":     {"codec": "vp9", "quality": 15, "gop": 30},
    "depth":   {"quantization_mm": 1},
    "numeric": {"level": 1}
  },
  "fps": 50,
  "data_dir": "/data/input",

  "metadata": {
    "dataset": "thread_velcro",
    "robot_type": "unknown",
    "layout": "state(14) + action(14) + cam_high + cam_low + cam_left_wrist + cam_right_wrist"
  },

  "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"
    },

    "timestamps": "timestamp"
  }
}

Episodic Zarr — per-episode stores. A directory of per-episode Zarr stores. Source paths are dot-separated group paths inside the store, so the nested array camera_0/rgb is written "camera_0.rgb". Four numeric arrays are mapped one-to-one:

{
  "dir_type": "episodic",
  "data_type": "zarr",
  "data_dir": "/data/pickNplace",
  "preset": "high",

  "metadata": {
    "dataset": "pickNplace",
    "robot_type": "dexterous_hand",
    "layout": "hand_action + pose + proprioception + fsr + camera_0"
  },

  "layout": {
    "images": {
      "camera_0": {"key": "camera_0.rgb", "fps": 30}
    },
    "numeric": {
      "hand_action":    "hand_action",
      "pose":           "pose",
      "proprioception": "proprioception",
      "fsr":            "fsr"
    }
  }
}

Episodic NPY — CALVIN-style, with depth. Per-episode .npy files, two RGB cameras and two depth cameras. Note that the episodic reader spells depth layout.depth (singular), and that preset: "ultra_low" is a near-lossless choice for iterating on a small slice:

{
  "dir_type": "episodic",
  "data_type": "npy",
  "data_dir": "/data/calvin",
  "preset": "ultra_low",

  "metadata": {
    "dataset": "calvin_debug",
    "layout": "actions + rel_actions + robot_obs + scene_obs, 2 rgb + 2 depth"
  },

  "layout": {
    "images": {
      "rgb_static":  {"key": "rgb_static",  "fps": 30},
      "rgb_gripper": {"key": "rgb_gripper", "fps": 30}
    },
    "depth": {
      "depth_static":  {"key": "depth_static",  "fps": 30},
      "depth_gripper": {"key": "depth_gripper", "fps": 30}
    },
    "numeric": {
      "actions":     "actions",
      "rel_actions": "rel_actions",
      "robot_obs":   "robot_obs",
      "scene_obs":   "scene_obs"
    }
  }
}

ROS bag / MCAP. See section 6.2 — these need no layout at all.

6.4 Troubleshooting A Config

7. Converting Any Format Into Knonik

When no built-in reader recognizes your dataset, you convert it yourself with one of two generic adapters. Both feed the same ingest engine as the built-in readers, so the compressed episodes and KHLP shards they produce are identical in every way — you are only responsible for handing Knonik the raw arrays.

• ▸Offline adapter (dir_type: "npy_separate") — you first write each stream to disk as .npy files, then run a normal ingest over that staged directory. Best when the data already fits on disk and you want a simple, re-runnable step.
• ▸Streaming adapter (live ingest) — you push frames into ingest one at a time as plain numpy arrays, with no staging. Best when data arrives serially, when you have many episodes, or when staging the whole dataset first is impractical.

Whichever you use, the frame contract is the same:

The number of frames ingested for an episode is the minimum length across all of its streams, so make sure your streams are aligned before you ingest.

Offline adapter — stage streams as .npy

Stage each stream on disk, then point an npy_separate config at the staged files. RGB and depth streams are a directory of per-frame files named with a printf-style pattern (indices from 0); numeric streams are a single (T, D) array; timestamps are a single (T,) array.

The script below reads one episode from your own format, stages it, and ingests it. Replace read_my_episode with your loader — everything else is generic:

import numpy as np
from pathlib import Path
from knonik.ingest import run

STAGE = Path("/home/me/staged/episode_0")
OUT   = "/home/me/data/out/my_custom_dataset"
FPS   = 30

# 1. Read one episode from your own format. Return aligned arrays:
#      rgb:    uint8   (T, H, W, 3)
#      state:  float32 (T, state_dim)
#      action: float32 (T, action_dim)
#      ts:     float64 (T,)   epoch seconds   (optional)
def read_my_episode(src):
    ...  # your code here
    return {"rgb": rgb, "state": state, "action": action, "ts": ts}

ep = read_my_episode("/raw/episode_0")

# 2. Stage each stream as .npy.
(STAGE / "cam_top").mkdir(parents=True, exist_ok=True)
for i, frame in enumerate(ep["rgb"]):                     # RGB: one file per frame
    np.save(STAGE / "cam_top" / f"frame_{i:06d}.npy", frame.astype(np.uint8))
np.save(STAGE / "state.npy",      ep["state"].astype(np.float32))    # numeric: (T, D)
np.save(STAGE / "action.npy",     ep["action"].astype(np.float32))
np.save(STAGE / "timestamps.npy", ep["ts"].astype(np.float64))      # optional

# 3. Describe the staged layout with an npy_separate config.
config = {
    "dir_type": "npy_separate",
    "metadata": {"dataset": "my_custom_dataset"},
    "timestamps": str(STAGE / "timestamps.npy"),          # optional
    "streams": {
        "rgb": {
            "cam_top": {"dir": str(STAGE / "cam_top"),
                        "pattern": "frame_%06d.npy", "fps": FPS},
        },
        "numeric": {
            "state":  {"path": str(STAGE / "state.npy")},
            "action": {"path": str(STAGE / "action.npy")},
        },
        # "depth": {"cam_top_depth": {"dir": ..., "pattern": ..., "fps": FPS}},
    },
}

# 4. Ingest the staged episode.
run(config=config, output=OUT)

Notes:

• ▸Use a streams block here, not layout — npy_separate reads staged files directly rather than mapping source keys.
• ▸numeric entries take an optional chunk_len (e.g. {"path": ..., "chunk_len": 128}) to control on-disk chunking of the array.
• ▸Each run(...) ingests one episode. For a multi-episode dataset, prefer the streaming adapter below — it writes many episodes into a single output directory in one pass, with no staging.

Streaming adapter — push frames live (no staging)

When you have many episodes, or data that arrives serially, skip staging and push frames straight in with the in-process live_session API. You register each stream once, then, for every episode, open an episode(...) block and push frames in order. This uses the same engine as offline ingest.

import numpy as np
from pathlib import Path
from knonik.ingest import live_session

OUT = "/home/me/data/out/my_custom_dataset"
FPS = 30

# Your own reader: yield one episode at a time from your format.
def my_episodes():
    for path in sorted(Path("/raw").glob("*.myfmt")):
        yield load_my_episode(path)   # -> {"task", "rgb", "state", "action", "ts"}

with live_session(output=OUT, preset="high") as ses:
    # Register every stream once, before the first episode.
    ses.register_rgb("cam_top", fps=FPS)
    ses.register_numeric("state",  dim=14)
    ses.register_numeric("action", dim=14)

    for ep in my_episodes():
        with ses.episode(metadata={"task": ep["task"]}):
            for t in range(len(ep["action"])):
                ts = ep["ts"][t]                                      # epoch seconds
                ses.rgb("cam_top",  ep["rgb"][t],    timestamp=ts)    # uint8   (H, W, 3)
                ses.numeric("state",  ep["state"][t],  timestamp=ts)  # float32 (14,)
                ses.numeric("action", ep["action"][t], timestamp=ts)  # float32 (14,)

• ▸ses.episode(...) finalizes the episode on a clean exit and discards it on an exception, so a crashed episode is never half-written (and is not charged).
• ▸Depth works the same way: ses.register_depth("cam_top_depth", fps=FPS) then ses.depth("cam_top_depth", depth_m, timestamp=ts) (metres, float32).
• ▸Arrays are coerced to the expected dtype and contiguity for you.
• ▸preset takes the same forms as offline ingest (a named preset, an inline {rgb, depth, numeric} dict, or omitted → medium); see section 8.

If your producer is not Python (for example a C++/ROS controller), drive the same streaming path over the cross-process stdin protocol instead — see section 10.

8. Compression Presets

Recommended explicit preset:

{
  "rgb": {"codec": "vp9", "quality": 18, "gop": 3},
  "depth": {"quantization_mm": 1},
  "numeric": {"level": 1}
}

Common knobs:

Pack-time profiles are separate from ingest presets. The ingest preset controls the intermediate episode files. storage_profile controls how those ingested episodes are rewritten into KHLP shards for training.

Named presets

Instead of a dict you can name a built-in preset (preset defaults to medium if omitted):

"preset": "high"

Add preset_overrides to tweak a named preset, e.g.:

"preset": "high",
"preset_overrides": { "rgb": { "gop": 15 } }

Named presets, inline preset dicts, and preset_overrides all work for both offline and live ingest (see section 10).

9. Python API

Use the public wrapper API from knonik.ingest.

Offline ingest

from knonik.ingest import run

run(
    config="/home/me/configs/config_lerobot_v3.json",
    input="/home/me/data/thread_velcro",
    output="/home/me/data/out/knonik_thread_velcro",
    force_config=False,   # True to overwrite an existing dataset config.json
)

The run writes <output>/config.json — the dataset config described in Processing §6 — when it finishes. force_config is the API equivalent of --force-config; leave it at False and an existing config is kept.

You can also pass a config dictionary:

from knonik.ingest import run

config = {
    "dir_type": "lerobot_v3",
    "data_dir": "/home/me/data/thread_velcro",
    "preset": {
        "rgb": {"codec": "vp9", "quality": 18, "gop": 3},
        "depth": {"quantization_mm": 1},
        "numeric": {"level": 1}
    },
    "layout": {
        "images": {
            # LeRobot camera keys are the bare name; the reader strips the
            # "observation.images." prefix automatically.
            "cam_high": {"key": "cam_high", "fps": 30}
        },
        "numeric": {
            # The main proprioception/command columns are exposed as
            # "state" / "action"; any other numeric column keeps its own
            # name, e.g. "qvel": "observation.qvel".
            "state": "state",
            "action": "action"
        }
    }
}

run(config=config, output="/home/me/data/out/knonik_thread_velcro")

Pack

from knonik.ingest import pack

paths = pack(
    input_dir="/home/me/data/out/knonik_thread_velcro",
    output_dir="/home/me/data/out",
    dataset_name="thread_velcro",
    shard_size=45,
    window_size=64,
    fps=50,
    seed=0,
    pack_mode="shard_batch",
)

print(paths)

Pack with a storage profile

pack(...) takes an optional storage_profile ("compact_video" default, "training_fast", or "training_compressed"). The codec/quality/format internals for each profile are chosen automatically — there is nothing else to configure:

from knonik.ingest import pack

paths = pack(
    input_dir="/datasets/my_dataset_ingested",
    output_dir="/datasets/my_dataset_fast",
    dataset_name="my_dataset",
    shard_size=30,
    window_size=64,
    fps=50,
    seed=0,
    pack_mode="shard_batch",
    storage_profile="training_fast",   # or "training_compressed"
)

Live ingest

For a Python producer, drive ingest directly with live_session. This uses the same engine as offline ingest, fed frame-by-frame as numpy arrays — no stdin protocol and no .npy staging:

from knonik.ingest import live_session

with live_session(output="/home/me/data/live_out", preset="high") as ses:
    ses.register_rgb("cam_top", fps=30)        # register each stream once,
    ses.register_numeric("action", dim=7)      # before the first episode
    for _ in range(num_episodes):
        with ses.episode(metadata={"task": "pick_and_place"}):
            for t in range(episode_length):
                ses.rgb("cam_top", capture_rgb(), timestamp=t / 30)     # uint8 (H, W, 3)
                ses.numeric("action", get_action(), timestamp=t / 30)   # float32 (7,)

• ▸

  preset accepts the same forms as offline ingest: a named preset, an inline {rgb, depth, numeric} dict, or omitted (→ medium); pass preset_overrides=... to tweak a named preset.

• ▸

  register_numeric also takes role= ("action" or "state") and channel_names=[...]. You are the source of truth for a live capture, so anything you declare here is written straight into the dataset config at the end of the run instead of coming back to you as a question. Pass them only when they are true of the data you push:

  ses.register_numeric("action", dim=7, role="action",
                       channel_names=["j1","j2","j3","j4","j5","j6","gripper"])
  

• ▸

  live_session(..., force_config=True) overwrites an existing dataset config.json in the output directory; the default leaves it alone.

• ▸

  ses.episode(...) calls start on enter, end on success, and abort on error (a discarded episode is not written and charges nothing).

• ▸

  Depth works the same way: ses.register_depth("cam_top_depth", fps=30) then ses.depth("cam_top_depth", depth_m, timestamp=...) (metres, float32).

• ▸

  Arrays are coerced to the expected dtype/contiguity for you.

Live ingest (stdin protocol)

run_live runs the cross-process JSON protocol, reading commands from the current process's stdin. Use it when a non-Python program drives ingest; otherwise prefer live_session above.

from knonik.ingest import run_live

run_live(
    config="/home/me/configs/live_config.json",
    output="/home/me/data/live_out",
)

10. Live Ingest From Terminal (stdin protocol)

Live mode reads the Knonik live protocol from stdin:

knonik ingest live \
  --config /path/to/live_config.json \
  --output /path/to/live_output

This is the cross-process path: a separate producer (often non-Python, e.g. a C++/ROS controller) writes newline-delimited JSON commands to stdin, and frame payloads are staged as .npy files and referenced by path. If your producer is Python, prefer the in-process live_session API instead — it avoids the JSON protocol and the .npy staging entirely.

A minimal live config:

{
  "output_dir": "/path/to/live_output",
  "preset": "medium"
}

Important points:

• ▸Keep stdin open while streaming. Closing stdin ends the session (any open episode is finalized first).
• ▸Register streams after start, at the beginning of each episode (streams are bound to the episode being written), then push frames, then end. Repeat the start → register → push → end cycle for each episode.
• ▸Write each frame's .npy file to disk before sending the command that references it.
• ▸Use local filesystem paths, not Docker mount paths.
• ▸preset accepts the same forms as offline ingest: a named string ("high", "medium", …), an inline {rgb, depth, numeric} dict, or omitted (defaults to medium), plus optional preset_overrides.
• ▸output_dir is required in the config; --output overrides it.

Live ingest is also the streaming generic adapter for unrecognized formats: read your data one frame at a time in your own script and push each frame in. From Python, the in-process live_session API does this with plain numpy arrays; non-Python producers use the protocol below. See section 7.

Live Protocol Reference

The agent reads one JSON object per line from stdin and writes one JSON response per line to stdout. Every response is either {"ok": true} or {"ok": false, "error": "<message>"}.

Commands

If timestamp is omitted, the frame index is used as the timestamp.

Lifecycle

start  →  register_* (each stream)  →  rgb / depth / numeric × N  →  end
  ↓
start  →  register_*  →  …  →  end          (repeat for more episodes)
  ↓
exit

Driving live ingest from Python

import json, os, subprocess
import numpy as np

os.makedirs("/tmp/knonik_frames", exist_ok=True)

proc = subprocess.Popen(
    ["knonik", "ingest", "live",
     "--config", "/path/to/live_config.json",
     "--output", "/path/to/live_output"],
    stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True,
)

def send(cmd):
    proc.stdin.write(json.dumps(cmd) + "\n")
    proc.stdin.flush()
    return json.loads(proc.stdout.readline())

for ep in range(num_episodes):
    assert send({"cmd": "start", "metadata": {"task": "pick_and_place"}})["ok"]
    # Register streams after start, at the beginning of each episode.
    assert send({"cmd": "register_rgb",     "name": "cam_top", "fps": 30})["ok"]
    assert send({"cmd": "register_numeric", "name": "action",  "dim": 7})["ok"]
    for t in range(episode_length):
        ts = t / 30.0

        p = f"/tmp/knonik_frames/rgb_{t:04d}.npy"
        np.save(p, capture_rgb())            # uint8 (H, W, 3)
        send({"cmd": "rgb", "name": "cam_top", "path": p, "timestamp": ts})

        p = f"/tmp/knonik_frames/action_{t:04d}.npy"
        np.save(p, get_action())             # float32 (7,)
        send({"cmd": "numeric", "name": "action", "path": p, "timestamp": ts})
    assert send({"cmd": "end"})["ok"]

send({"cmd": "exit"})
proc.stdin.close()
proc.wait()

The in-process variant knonik.ingest.run_live(config, output) runs the same protocol but reads from the current process's stdin — handy when another program pipes commands into your Python process.

11. Output Layout

Offline ingest writes one directory per episode:

knonik_thread_velcro/
  config.json                   (dataset config — see Processing §6)
  episode_000000/
    metadata.json
    rgb_<stream>.webm           (one per RGB stream)
    depth_<stream>.hdf5         (one per depth stream, if any)
    <numeric_stream>/           (Zarr v2 array, shape [T, D], float32)
    <numeric_stream>_time/      (Zarr v2 array, shape [T], float64 epoch seconds)
    annotations/                (language annotations, if enabled — see below)
  episode_000001/
    ...

metadata.json records the dataset name, robot type, source format, episode id, the RGB/depth/numeric stream names, start/end time, and episode_length. It also carries what the dataset config generator needs: stream_params (the encoder and compressor settings each stream was written with), stream_roles when the source format declared which stream is the action, channel_names when the source declared per-channel labels, and urdf_joints when a ROS bag recorded /robot_description. Read a numeric stream with Zarr:

import zarr
action = zarr.open("knonik_thread_velcro/episode_000000/action", mode="r")
print(action.shape, action[:5])

Language annotations. Format-reader ingests also write per-episode files under episode_<n>/annotations/language/ (episode.yaml, rephrased.yaml, timestamps.yaml). The instruction comes from the dataset, the language config block, or metadata.language_instruction. Set language: false to skip, or pass a dict like {"instruction": "pick up the cube", "rephrases": [...]} to supply one when the dataset has none.

Packing writes KHLP shards and a manifest:

knonik_thread_velcro_hlp/
  hlp_000000.khlp
  hlp_000001.khlp
  hlp_manifest.json

Use hlp_manifest.json with the Knonik multidataloader.