embodichain.data_pipeline.depth_video

embodichain.data_pipeline.depth_video#

Compressed depth sidecar storage for LeRobot datasets on Python 3.10–3.12.

This package implements issue #424 Path A: an EmbodiChain-owned depth writer that stores camera depth as gray12le/HEVC sidecar videos alongside a LeRobot dataset without modifying the installed LeRobot package.

Depth quantization math is vendored from lerobot v0.6.0 so that sidecar videos remain binary-compatible with the official reader.

Overview#

Compressed depth-sidecar storage for LeRobot datasets on Python 3.10–3.12. The package stores camera depth as gray12le/HEVC sidecar videos alongside a LeRobot dataset without modifying the installed LeRobot package. Depth quantization math is vendored from lerobot v0.6.0 so that sidecar videos stay binary-compatible with the official reader.

The main entry points are DepthVideoWriter and DepthSidecarManager for writing, DepthVideoReader for reading, and load_depth_dataset() / load_depth_meta() for loading depth data alongside a LeRobot dataset. Behavior is configured through DepthVideoCfg; codec selection goes through detect_depth_encoder() and resolve_depth_vcodec().

Classes

`DepthVideoCfg`	Configuration for the compressed depth sidecar writer.
`DepthVideoWriter`	Streaming writer that encodes one depth episode to a single MP4.
`DepthSidecarManager`	Owns the depth sidecar videos and metadata for one dataset recording.
`DepthVideoReader`	Decode and dequantize a single depth sidecar MP4.
`DepthVideoLibrary`	Index of all depth sidecar videos for a dataset.
`DepthCodecError`	Raised when no suitable depth video codec is available.

Functions

`detect_depth_encoder`([vcodec])	Return an available depth video encoder name, or `None`.
`resolve_depth_vcodec`([vcodec])	Return an available depth video encoder name, raising on failure.
`quantize_depth`(depth[, depth_min, ...])	Quantize depth to 12-bit codes (`uint16`, values `0…DEPTH_QMAX`).
`dequantize_depth`(quantized[, depth_min, ...])	Inverse of `quantize_depth()`.
`load_depth_dataset`(dataset_root, ...)	Load a LeRobot dataset together with its depth sidecar library.
`load_depth_meta`(dataset_root)	Load and return the `depth_meta.json` for a dataset.

Configuration#

Configuration for compressed depth sidecar storage.

Classes:

DepthVideoCfg

Configuration for the compressed depth sidecar writer.

class embodichain.data_pipeline.depth_video.cfg.DepthVideoCfg[source]#

Bases: object

Configuration for the compressed depth sidecar writer.

Depth is quantized to 12-bit codes and encoded as a single-channel gray12le video. With lossless=True (default) the 12-bit codes are preserved bit-exactly by HEVC; the only error is the configurable float32 -> 12-bit quantization step.

enable#: If False, depth is stored as numeric LeRobot features (PR #422).

vcodec#: Video codec for depth. Defaults to "libx265" (HEVC), which supports 12-bit grayscale losslessly on typical FFmpeg builds.

lossless#: If True, encode with HEVC lossless mode so 12-bit codes are bit-exact. If False, use crf for lossy encoding.

crf#: Constant rate factor for lossy mode (ignored when lossless=True). Lower is higher quality; 0 is lossless for libx265.

depth_min#: Depth (metres) mapped to quantum 0.

depth_max#: Depth (metres) mapped to quantum DEPTH_QMAX.

shift#: Pre-log offset (metres) for numerical stability near zero.

use_log#: Logarithmic (True) or linear (False) quantization.

pix_fmt#: Pixel format for the depth video. gray12le carries the 12-bit codes in a single channel.

input_unit#: Unit of the incoming depth arrays ("auto" infers from dtype: float -> metres, int -> millimetres).

output_unit#: Unit returned by the reader ("m" or "mm").

keep_numeric_fallback#: If True, also keep depth as a numeric LeRobot feature (exact raw values, ~2x depth storage). If False, depth lives only in the sidecar videos.

Methods:

`__init__`([enable, vcodec, lossless, crf, ...])
`copy`(**kwargs)	Return a new object replacing specified fields with new values.
`replace`(**kwargs)	Return a new object replacing specified fields with new values.
`to_dict`()	Convert an object into dictionary recursively.
`validate`([prefix])	Check the validity of configclass object.

Attributes:

`crf`
`depth_max`
`depth_min`
`enable`
`input_unit`
`keep_numeric_fallback`
`lossless`
`output_unit`
`pix_fmt`
`shift`
`use_log`
`vcodec`

__init__(enable=<factory>, vcodec=<factory>, lossless=<factory>, crf=<factory>, depth_min=<factory>, depth_max=<factory>, shift=<factory>, use_log=<factory>, pix_fmt=<factory>, input_unit=<factory>, output_unit=<factory>, keep_numeric_fallback=<factory>)#

copy(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

crf: int#

depth_max: float#

depth_min: float#

enable: bool#

input_unit: Literal['auto', 'm', 'mm']#

keep_numeric_fallback: bool#

lossless: bool#

output_unit: Literal['m', 'mm']#

pix_fmt: str#

replace(**kwargs)#

Return a new object replacing specified fields with new values.

This is especially useful for frozen classes. Example usage:

@configclass(frozen=True)
class C:
    x: int
    y: int

c = C(1, 2)
c1 = c.replace(x=3)
assert c1.x == 3 and c1.y == 2

Parameters:

obj (object) – The object to replace.
**kwargs – The fields to replace and their new values.

Return type:

object

Returns:

The new object.

shift: float#

to_dict()#

Convert an object into dictionary recursively.

Note

Ignores all names starting with “__” (i.e. built-in methods).

Parameters:: obj (object) – An instance of a class to convert.
Raises:: ValueError – When input argument is not an object.
Return type:: dict[str, Any]
Returns:: Converted dictionary mapping.

use_log: bool#

validate(prefix='')#

Check the validity of configclass object.

This function checks if the object is a valid configclass object. A valid configclass object contains no MISSING entries.

Parameters:

obj (object) – The object to check.
prefix (str) – The prefix to add to the missing fields. Defaults to ‘’.

Return type:

list[str]

Returns:

A list of missing fields.

Raises:

TypeError – When the object is not a valid configuration object.

vcodec: str#

Codec & Quantization#

Codec availability detection for depth video encoding.

Adapted from lerobot v0.6.0 pyav_utils.py (Apache-2.0). Probes the bundled FFmpeg build through PyAV so the writer can pick a working encoder and degrade gracefully (to numeric Parquet depth, PR #422) when HEVC is unavailable.

Exceptions:

DepthCodecError

Raised when no suitable depth video codec is available.

Functions:

`detect_depth_encoder`([vcodec])	Return an available depth video encoder name, or `None`.
`resolve_depth_vcodec`([vcodec])	Return an available depth video encoder name, raising on failure.

exception embodichain.data_pipeline.depth_video.codec.DepthCodecError[source]#

Bases: RuntimeError

Raised when no suitable depth video codec is available.

embodichain.data_pipeline.depth_video.codec.detect_depth_encoder(vcodec=None)[source]#

Return an available depth video encoder name, or None.

Parameters:: vcodec (str | None) – Preferred codec name. If None or unavailable, fall back to the default preference list (libx265 then hevc).
Return type:: str | None
Returns:: The first available encoder name, or None if none are available.

embodichain.data_pipeline.depth_video.codec.resolve_depth_vcodec(vcodec=None)[source]#

Return an available depth video encoder name, raising on failure.

Parameters:: vcodec (str | None) – Preferred codec name.
Return type:: str
Returns:: An available encoder name.
Raises:: DepthCodecError – If no depth video encoder is available in the bundled FFmpeg build.

Depth quantization/dequantization for compressed depth sidecar videos.

Depth maps are packed into 12-bit integer codes (uint16, values 0…DEPTH_QMAX) so they fit the gray12le pixel format and can be encoded losslessly by HEVC (libx265). Logarithmic quantization is the default because it allocates more quanta to near-range depth, matching the 1/depth error profile of typical depth sensors.

The math is ported from lerobot 0.6.0 (itself adapted from BEHAVIOR-1K’s obs_utils.py) and depends only on av, numpy and torch – all of which run on Python 3.10–3.12. See issue #424.

Functions:

`dequantize_depth`(quantized[, depth_min, ...])	Inverse of `quantize_depth()`.
`infer_depth_unit`(dtype)	Infer the depth unit from the array dtype.
`quantize_depth`(depth[, depth_min, ...])	Quantize depth to 12-bit codes (`uint16`, values `0…DEPTH_QMAX`).
`squeeze_single_channel`(array)	Drop a leading or trailing singleton channel dim: `(1, H, W)` / `(H, W, 1)` -> `(H, W)`.
`write_u16_plane`(plane, src[, fill_value])	Copy a 2D `uint16` image into the plane's memory buffer, row by row.

embodichain.data_pipeline.depth_video.depth_utils.dequantize_depth(quantized, depth_min=0.01, depth_max=10.0, shift=3.5, use_log=True, pix_fmt='gray12le', output_unit='mm', output_tensor=True, output_channel_last=False)[source]#

Inverse of quantize_depth().

Decoding inverts the same normalized code mapping as quantize_depth() using depth_min / depth_max / shift (in metres), then returns the requested output unit. Tuning arguments must match quantize_depth().

Accepted input layouts :

(H, W, 1) or (H, W) - single frame with channel-last.
(..., 1, H, W) - batched frames with channel-first.
(..., H, W, 1) - batched frames with channel-last.

Output layout is determined by output_channel_last.

Parameters:

quantized (ndarray[tuple[Any, ...], dtype[uint16]] | VideoFrame | Tensor) – 12-bit codes in [0, DEPTH_QMAX]. np.ndarray, av.VideoFrame, or torch.Tensor (any integer or float dtype).
depth_min (float) – Same as quantize_depth() (metres).
depth_max (float) – Same as quantize_depth() (metres).
shift (float) – Same as quantize_depth() (metres).
use_log (bool) – Same as quantize_depth() (metres).
pix_fmt (str) – Pixel format used to extract the plane from an av.VideoFrame.
output_unit (Literal['m', 'mm']) – "mm" returns uint16 millimetres (rint, clip [0, 65535]) when returning a numpy array, or float32 mm when output_tensor=True. "m" returns float32 metres in [depth_min, depth_max].
output_tensor (bool) – If True, return a torch.Tensor instead of a numpy array.
output_channel_last (bool) – Channel layout of the output ((H, W, 1) vs (1, H, W)).

Return type:

ndarray[tuple[Any, ...], dtype[uint16]] | ndarray[tuple[Any, ...], dtype[float32]] | Tensor

Returns:

Depth map in the requested unit and dtype.

Raises:

ValueError – If output_unit is not "m" or "mm".
ValueError – If use_log=True and depth_min + shift <= 0.

embodichain.data_pipeline.depth_video.depth_utils.infer_depth_unit(dtype)[source]#

Infer the depth unit from the array dtype.

Floating-point arrays are interpreted as metres, integer arrays as millimetres – the lerobot convention.

Parameters:: dtype (dtype | type) – NumPy dtype (or anything coercible to one).
Return type:: str
Returns:: "m" or "mm".

embodichain.data_pipeline.depth_video.depth_utils.quantize_depth(depth, depth_min=0.01, depth_max=10.0, shift=3.5, use_log=True, pix_fmt='gray12le', video_backend='pyav', input_unit='auto')[source]#

Quantize depth to 12-bit codes (uint16, values 0…DEPTH_QMAX).

Depth maps are packed into 12-bit integer frames so they fit in standard high-bit-depth pixel formats (e.g. yuv420p12le / gray12le) and can be encoded by widely supported video codecs (e.g. HEVC Main 12). Logarithmic quantization is the default because it allocates more quanta to near-range depth, which matches the (1/depth) error profile of typical depth sensors. Math is ported from BEHAVIOR-1K’s obs_utils.py.

Input units:

input_unit="auto" (default): infer from dtype (floating = m, non-floating = mm).
input_unit="mm": interpret input values as millimetres.
input_unit="m": interpret input values as metres.

Quantization math runs in the resolved input unit.

depth_min, depth_max, and shift are always in metres.

Parameters:

depth (ndarray[tuple[Any, ...], dtype[uint16]] | ndarray[tuple[Any, ...], dtype[float32]] | Tensor) – Depth map; torch.Tensor is moved to CPU for conversion.
depth_min (float) – Depth (metres) at quantum 0.
depth_max (float) – Depth (metres) at quantum DEPTH_QMAX.
shift (float) – Depth shift (metres); used in log mode. Must satisfy depth_min + shift > 0.
use_log (bool) – If True (default), quantize in log space.
video_backend (str | None) – If "pyav" (default), return an av.VideoFrame ready for encoding; otherwise return the raw uint16 code array.
input_unit (Literal['auto', 'm', 'mm']) – Input unit policy ("auto", "mm", "m").

Return type:

ndarray[tuple[Any, ...], dtype[uint16]] | VideoFrame

Returns:

av.VideoFrame (when video_backend="pyav") or numpy.ndarray of dtype=uint16, same spatial shape as depth, values in [0, DEPTH_QMAX].

Raises:

ValueError – If input_unit is not "auto", "mm", or "m".
ValueError – If use_log=True and depth_min + shift <= 0.

embodichain.data_pipeline.depth_video.depth_utils.squeeze_single_channel(array)[source]#

Drop a leading or trailing singleton channel dim: (1, H, W) / (H, W, 1) -> (H, W).

Unlike array.squeeze(), this only removes the channel axis, never an H or W of size 1.

Return type:: ndarray

embodichain.data_pipeline.depth_video.depth_utils.write_u16_plane(plane, src, fill_value=None)[source]#

Copy a 2D uint16 image into the plane’s memory buffer, row by row.

For speed, each row is padded to a wider size than width, so the true row width in memory is plane.line_size (bytes), not width. Copying as one straight stream would skew the image, so we write only the first width columns of each row and leave the padding untouched.

Parameters:

plane (VideoPlane) – Destination 16-bit plane.
src (ndarray) – Source image, shape (height, width), dtype uint16.
fill_value (int | None) – If given, every pixel (padding included) is set to this first, so the padding holds clean data instead of garbage.

Return type:

None

Writer#

Streaming depth video writer and per-episode sidecar manager.

Writes camera depth maps as gray12le/HEVC sidecar videos that live alongside a LeRobot dataset (issue #424, Path A). Depth never enters LeRobot’s own image/video pipeline (which is RGB-only in 0.4.4); instead each episode is encoded to a standalone MP4 and indexed by depth_meta.json.

Classes:

`DepthSidecarManager`	Owns the depth sidecar videos and metadata for one dataset recording.
`DepthVideoWriter`	Streaming writer that encodes one depth episode to a single MP4.

class embodichain.data_pipeline.depth_video.writer.DepthSidecarManager[source]#

Bases: object

Owns the depth sidecar videos and metadata for one dataset recording.

For each episode, opens one DepthVideoWriter per depth sensor (e.g. camera and camera_right), feeds frames, and on episode close moves the MP4s into the dataset’s depth_videos/ tree and updates depth_meta.json.

Methods:

`__init__`(dataset_root, fps, cfg[, vcodec])	Initialize the sidecar manager.
`abort_episode`()	Abort the current episode: discard all partial depth videos.
`add_frame`(sensor_key, depth)	Feed one depth frame for one sensor in the current episode.
`end_episode`(episode_index)	Close all writers for the current episode and update metadata.
`finalize`()	Flush metadata after the last episode.
`has_sensor`(sensor_key)	Return True if sensor_key has been registered.
`register_sensor`(sensor_key, shape)	Register a depth sensor and its static metadata.
`start_episode`(episode_index, sensor_keys)	Open a writer per sensor for a new episode.

__init__(dataset_root, fps, cfg, vcodec=None)[source]#

Initialize the sidecar manager.

Parameters:

dataset_root (Path) – Root directory of the LeRobot dataset. Sidecar videos are written under <root>/depth_videos/ and metadata under <root>/depth_meta.json.
fps (int) – Dataset frame rate.
cfg (DepthVideoCfg) – Depth video configuration.
vcodec (Optional[str]) – Resolved codec name. If None, resolved from cfg.

abort_episode()[source]#

Abort the current episode: discard all partial depth videos.

Called when the surrounding LeRobot save_episode fails, so the dataset never references a depth video for an episode that was not committed.

Return type:: None

add_frame(sensor_key, depth)[source]#

Feed one depth frame for one sensor in the current episode.

Parameters:

sensor_key (str) – Sensor identifier.
depth (Any) – Depth map.

Return type:

None

end_episode(episode_index)[source]#

Close all writers for the current episode and update metadata.

Every writer is attempted. Partial files from failed writers are discarded, metadata for successful writers is flushed, and any error is then raised so a committed LeRobot episode cannot silently lose its depth sidecar.

Parameters:: episode_index (int) – Global episode index.
Raises:: RuntimeError – If a depth video or its metadata cannot be finalized.
Return type:: None

finalize()[source]#

Flush metadata after the last episode.

Return type:: Path
Returns:: Path to depth_meta.json.

has_sensor(sensor_key)[source]#

Return True if sensor_key has been registered.

Return type:: bool

register_sensor(sensor_key, shape)[source]#

Register a depth sensor and its static metadata.

Parameters:

sensor_key (str) – Sensor identifier including any side suffix, e.g. "camera" or "camera_right".
shape (tuple[int, ...]) – Spatial shape of the depth frames, e.g. (480, 640).

Return type:

None

start_episode(episode_index, sensor_keys)[source]#

Open a writer per sensor for a new episode.

Parameters:

episode_index (int) – Global episode index.
sensor_keys (list[str]) – Depth sensor keys present in this episode.

Return type:

None

class embodichain.data_pipeline.depth_video.writer.DepthVideoWriter[source]#

Bases: object

Streaming writer that encodes one depth episode to a single MP4.

Depth frames are quantized to 12-bit codes and encoded as gray12le video. With lossless=True the 12-bit codes are bit-exact after a decode round-trip (verified on pyav 15.x + libx265).

The video is written to a temporary path and atomically moved to its final location on close(); on failure the partial file is removed so the dataset never references a corrupt sidecar.

Methods:

`__init__`(final_path, fps, cfg[, vcodec])	Initialize the writer.
`abort`()	Abort writing and remove any partial output.
`add_frame`(depth)	Quantize and encode a single depth frame.
`close`()	Flush the encoder and finalize the MP4.

Attributes:

`final_path`	Destination MP4 path.
`frame_count`	Number of frames written so far.

__init__(final_path, fps, cfg, vcodec=None)[source]#

Initialize the writer.

Parameters:

final_path (Path) – Destination MP4 path. Written via a temp file in the same directory and atomically renamed on success.
fps (int) – Episode frame rate (frames per second).
cfg (DepthVideoCfg) – Depth video configuration.
vcodec (Optional[str]) – Resolved codec name. If None, resolved from cfg.

abort()[source]#

Abort writing and remove any partial output.

Return type:: None

add_frame(depth)[source]#

Quantize and encode a single depth frame.

Parameters:: depth (Any) – Depth map (torch.Tensor or np.ndarray), shape (H, W) / (H, W, 1) / (1, H, W).
Return type:: None

close()[source]#

Flush the encoder and finalize the MP4.

Return type:: Path
Returns:: The final MP4 path on success.
Raises:: RuntimeError – If no frames were written.

property final_path: Path#: Destination MP4 path.

property frame_count: int#: Number of frames written so far.

Reader#

Readers for compressed depth sidecar videos.

Decodes the gray12le/HEVC MP4s written by DepthSidecarManager and dequantizes them back to metres or millimetres. load_depth_dataset composes a LeRobot dataset with its depth sidecar library so callers can read RGB, state, action, mask and depth together.

Classes:

`DepthVideoLibrary`	Index of all depth sidecar videos for a dataset.
`DepthVideoReader`	Decode and dequantize a single depth sidecar MP4.

Functions:

load_depth_dataset(dataset_root, ...)

Load a LeRobot dataset together with its depth sidecar library.

class embodichain.data_pipeline.depth_video.reader.DepthVideoLibrary[source]#

Bases: object

Index of all depth sidecar videos for a dataset.

Maps (episode_index, sensor_key) to a DepthVideoReader and provides random access by within-episode frame index.

Methods:

`__init__`(dataset_root, meta)	Initialize the library from parsed metadata.
`frame_count`(episode_index, sensor_key)	Return the number of depth frames for an episode/sensor.
`get`(episode_index, sensor_key, ...)	Read one depth frame.
`sensor_meta`(sensor_key)	Return the static metadata block for a sensor.

Attributes:

sensors

List of depth sensor keys recorded in the dataset.

__init__(dataset_root, meta)[source]#

Initialize the library from parsed metadata.

Parameters:

dataset_root (Path) – Root directory of the LeRobot dataset.
meta (Dict[str, Any]) – Parsed depth_meta.json contents.

frame_count(episode_index, sensor_key)[source]#

Return the number of depth frames for an episode/sensor.

Return type:: int

get(episode_index, sensor_key, frame_index_in_episode)[source]#

Read one depth frame.

Parameters:

episode_index (int) – Global episode index.
sensor_key (str) – Sensor identifier (e.g. "camera", "camera_right").
frame_index_in_episode (int) – Frame index within the episode.

Return type:

ndarray

Returns:

Depth map in the unit recorded in metadata, shape (1, H, W).

sensor_meta(sensor_key)[source]#

Return the static metadata block for a sensor.

Return type:: Dict[str, Any]

property sensors: list[str]#: List of depth sensor keys recorded in the dataset.

class embodichain.data_pipeline.depth_video.reader.DepthVideoReader[source]#

Bases: object

Decode and dequantize a single depth sidecar MP4.

Frames are decoded lazily on first access and cached, since one MP4 holds a single (typically short) episode.

Methods:

`__init__`(path, depth_min, depth_max, shift, ...)	Initialize the reader.
`read`(frame_index)	Decode and dequantize one frame.

Attributes:

frame_count

Number of frames in the sidecar video.

__init__(path, depth_min, depth_max, shift, use_log, pix_fmt='gray12le', output_unit='m')[source]#

Initialize the reader.

Parameters:

path (Path) – Path to the sidecar MP4.
depth_min (float) – Quantization depth_min (metres) used at write time.
depth_max (float) – Quantization depth_max (metres) used at write time.
shift (float) – Quantization shift (metres) used at write time.
use_log (bool) – Quantization use_log used at write time.
pix_fmt (str) – Pixel format of the stored video.
output_unit (str) – Unit to return ("m" or "mm").

property frame_count: int#: Number of frames in the sidecar video.

read(frame_index)[source]#

Decode and dequantize one frame.

Parameters:: frame_index (int) – Within-episode frame index.
Return type:: ndarray
Returns:: Depth map in the configured output unit, shape (1, H, W).

embodichain.data_pipeline.depth_video.reader.load_depth_dataset(dataset_root, **lerobot_kwargs)[source]#

Load a LeRobot dataset together with its depth sidecar library.

Parameters:

dataset_root (Path) – Root directory of the LeRobot dataset (the directory containing data/, videos/ and depth_meta.json).
**lerobot_kwargs (Any) – Forwarded to LeRobotDataset (e.g. episodes).

Return type:

tuple[Any, DepthVideoLibrary]

Returns:

A (LeRobotDataset, DepthVideoLibrary) tuple.

Raises:

ImportError – If LeRobot is not installed.
FileNotFoundError – If the dataset has no depth sidecar metadata.

embodichain.data_pipeline.depth_video

Contents

embodichain.data_pipeline.depth_video#

Overview#

Configuration#

Codec & Quantization#

Writer#

Reader#