embodichain.data_pipeline

embodichain.data_pipeline#

Online streaming, recording, and offline dataset tools.

Subpackages load on first access so offline inspection does not initialize the simulation, PyTorch, or video-encoding stacks.

Overview#

Online data streaming, recording, and offline expert-dataset inspection. The online components are datasets (online datasets and samplers that stream trajectories from a running simulation), engine (a process-safe shared buffer that decouples simulation producers from training consumers), and depth_video (compressed depth-sidecar storage for LeRobot datasets on Python 3.10–3.12).

Submodules

datasets

Online datasets and offline inspection with lazy simulation dependencies.

depth_video

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

engine

Process-safe shared buffer (OnlineDataEngine) that decouples simulation producers from training consumers.

recording

Recording identity, durable commit diagnostics, and conservative repair.

runtime

Owner-side flow control for an explicitly started online data engine.

Datasets#

Online datasets and offline inspection with lazy simulation dependencies.

`online_data`
`sampler`

Classes:

`BatchSource`	A source of owned training batches with explicit data semantics.
`ChunkSizeSampler`	Abstract base class for chunk-size samplers.
`GMMChunkSampler`	Gaussian Mixture Model chunk-size sampler.
`LeRobotDataSource`	Load numeric episodes once and serve owned resident context batches.
`LeRobotFieldMappingCfg`	Declare SDK columns and optional EmbodiChain provenance explicitly.
`OfflineDataset`	alias of `SequenceDataset`
`OnlineDataSource`	Adapt a started ODS engine without taking ownership of its lifecycle.
`OnlineDataset`	Infinite IterableDataset backed by a live OnlineDataEngine shared buffer.
`SequenceDataset`	Iterate owned causal sequences from a live or resident data source.
`SequenceSpecCfg`	Specify causal history and current/future supervision.
`TensorDataSource`	Sample causal sequences from resident, explicitly described CPU data.
`UniformChunkSampler`	Discrete-uniform chunk-size sampler over `[low, high]`.

Exceptions:

`DataBudgetExhaustedError`	A complete requested batch exceeds eligible source delivery allowances.
`DataUnavailableError`	No eligible unlocked window exists under the current source/filter state.

Functions:

`data_schema_fingerprint`(schema)	Hash a validated finite JSON schema without changing feature order.
`validate_data_schema`(schema)	Validate declarations common to live and resident causal sources.

class embodichain.data_pipeline.datasets.BatchSource[source]#

Bases: Protocol

A source of owned training batches with explicit data semantics.

schema declares ordered observations/actions, their timing and units. source_id identifies the source dataset/run independently of cache slots. Production/refill and loading lifecycles belong to concrete sources and the caller, not to this sampling protocol. Sampling never performs hidden waits.

Methods:

`__init__`(args, *kwargs)
`sample_context`(batch_size, spec, *[, generator])	Return copied `obs/actions/valid/loss_mask` and source metadata.

Attributes:

`schema`
`source_id`

__init__(*args, **kwargs)#

sample_context(batch_size, spec, *, generator=None)[source]#

Return copied obs/actions/valid/loss_mask and source metadata.

Parameters:

batch_size (int) – Number of episode-uniform contexts, with replacement.
spec (SequenceSpecCfg) – Window, boundary and optional generation requirements.
generator (Generator | None) – Sampling RNG; None uses the current Torch RNG.

Return type:

TensorDict

Returns:

TensorDict with batch axes [B, H + F - 1]. Returned storage remains valid after subsequent sampling or source replacement.

Raises:

DataBudgetExhaustedError – Eligible delivery credits cannot provide the entire batch. No partial debit is performed.
DataUnavailableError – No eligible anchor exists in the current data.
RuntimeError – A producer or source read failed.
ValueError – Schema or requested capabilities are incompatible.

schema: Mapping[str, Any]#

source_id: str#

class embodichain.data_pipeline.datasets.ChunkSizeSampler[source]#

Bases: ABC

Abstract base class for chunk-size samplers.

Subclasses implement __call__() to return an integer chunk size on demand. A sampler is called once per OnlineDataset.__iter__() step, so consecutive samples / batches may have different time dimensions.

When used in batch mode the same chunk size is drawn once and applied to every trajectory in the batch so that the resulting TensorDict has a consistent shape [batch_size, chunk_size].

exception embodichain.data_pipeline.datasets.DataBudgetExhaustedError[source]#

Bases: RuntimeError

A complete requested batch exceeds eligible source delivery allowances.

exception embodichain.data_pipeline.datasets.DataUnavailableError[source]#

Bases: RuntimeError

No eligible unlocked window exists under the current source/filter state.

This is distinct from quota depletion and a terminal producer/read failure. The caller decides whether new publication can help and whether to wait with a deadline; a fixed offline source may never satisfy the request. Legacy consumers catching RuntimeError remain compatible.

class embodichain.data_pipeline.datasets.GMMChunkSampler[source]#

Bases: ChunkSizeSampler

Gaussian Mixture Model chunk-size sampler.

Selects a mixture component according to weights, samples a value from the corresponding Normal(mean, std) distribution, rounds to the nearest integer, and optionally clamps the result to [low, high].

Parameters:

means (List[float]) – Mean of each Gaussian component (number of elements = K).
stds (List[float]) – Standard deviation of each component (must be > 0, same length as means).
weights (Optional[List[float]]) – Unnormalised mixture weights (same length as means). Defaults to a uniform distribution over all components.
low (Optional[int]) – Optional lower bound for clamping the sampled value (inclusive, must be ≥ 1 if provided).
high (Optional[int]) – Optional upper bound for clamping the sampled value (inclusive, must be ≥ low if both are provided).

Raises:

ValueError – If means, stds, or weights have mismatched lengths, if any std ≤ 0, or if the bounds are inconsistent.

Example — two-component mixture favouring short and long chunks:

sampler = GMMChunkSampler(
    means=[16.0, 64.0],
    stds=[4.0, 8.0],
    weights=[0.6, 0.4],
    low=8,
    high=96,
)
chunk_size = sampler()  # e.g. 18

Methods:

__init__(means, stds[, weights, low, high])

__init__(means, stds, weights=None, low=None, high=None)[source]#

class embodichain.data_pipeline.datasets.LeRobotDataSource[source]#

Bases: TensorDataSource

Load numeric episodes once and serve owned resident context batches.

Parameters:

root (Path | str) – Existing local LeRobot dataset directory.
schema (Mapping[str, Any]) – Explicit ordered observation/action schema. Actions require a nonempty semantics or representation and positive step_dt.
memory_limit_bytes (int) – Positive hard limit for owned resident tensors. SDK caches and bounded read/validation staging are separate costs.
source_id (str) – Persisted dataset/origin identity; used for generated frame IDs when the dataset does not already record provenance.
mapping (LeRobotFieldMappingCfg | None) – Explicit numeric SDK columns and optional source metadata.
repo_id (str) – Dataset identifier required by the installed SDK.
max_episode_steps (int | None) – Optional fixed padded capacity. Inferred from source frame indices otherwise; specify it for exact ODS RNG pairing.
episode_frames (Sequence[Sequence[int]] | None) – Optional original valid frame indices, including gaps and empty padded rows. SDK episode order must match nonempty rows.
read_batch_size (int) – Maximum numeric SDK rows per read.
language (Mapping[str, Mapping[int | str, str]] | None) – Optional source-owned task/subtask ID-to-text dictionaries.

Uses the official SDK lazily and never imports simulation modules. Missing backing-column access selects scalar SDK fallback; integrity failures do not. External datasets need explicit semantic mapping, not a showcase sidecar. Timestamps must match the declared fixed-rate action grid.

Methods:

__init__(root, *, schema, ...[, mapping, ...])

__init__(root, *, schema, memory_limit_bytes, source_id, mapping=None, repo_id='local/dataset', max_episode_steps=None, episode_frames=None, read_batch_size=1024, language=None)[source]#

class embodichain.data_pipeline.datasets.LeRobotFieldMappingCfg[source]#

Bases: object

Declare SDK columns and optional EmbodiChain provenance explicitly.

Parameters:

observation_key (str) – Numeric vector column mapped to obs.
action_key (str) – Numeric vector column mapped to actions.
provenance_key (str | None) – Optional int64 vector [origin, episode, frame, generation].
metadata_keys (dict[str, str]) – Optional canonical scalar fields mapped to SDK columns; supports continuity/segment IDs, segment acceptance and language IDs.

Column mapping is a declaration of semantics by the caller, not automatic inference from a matching width. Image/video sequence decoding is not part of this resident numeric backend.

Methods:

`__init__`([observation_key, action_key, ...])
`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:

`action_key`
`metadata_keys`
`observation_key`
`provenance_key`

__init__(observation_key=<factory>, action_key=<factory>, provenance_key=<factory>, metadata_keys=<factory>)#

action_key: str#

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.

metadata_keys: dict[str, str]#

observation_key: str#

provenance_key: str | None#

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.

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.

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.

embodichain.data_pipeline.datasets.OfflineDataset#: alias of SequenceDataset

class embodichain.data_pipeline.datasets.OnlineDataSource[source]#

Bases: object

Adapt a started ODS engine without taking ownership of its lifecycle.

Parameters:

engine (OnlineDataEngine) – Caller-owned engine. Enable provenance for portable frame IDs.
schema (Mapping[str, Any]) – Explicit schema of the returned observations and actions.
observation_transform (Callable[[Any], Tensor | TensorDictBase] | None) – Optional row-local feature function. It runs on the owned copy, after the producer lock has been released.

Refill, waits, start and stop remain caller operations. Sampling propagates empty-window, quota and worker errors immediately. The adapter preserves additional continuity/language/segment fields when present.

Methods:

`__init__`(engine, *, schema[, ...])
`resolve_language`(batch)	Resolve copied task/subtask IDs through the engine's text registry.
`sample_context`(batch_size, spec, *[, generator])	Sample an owned canonical batch under the engine's normal checks.

__init__(engine, *, schema, observation_transform=None)[source]#

resolve_language(batch)[source]#

Resolve copied task/subtask IDs through the engine’s text registry.

Parameters:: batch (TensorDict) – Copied batch containing task/subtask index tensors.
Return type:: dict[str, object]
Returns:: Overall and segment texts with the batch’s nested shape.

sample_context(batch_size, spec, *, generator=None)[source]#

Sample an owned canonical batch under the engine’s normal checks.

Parameters:

batch_size (int) – Number of contexts to deliver atomically.
spec (SequenceSpecCfg) – History/future lengths, segment mode and generation filter.
generator (Generator | None) – Optional sampling RNG independent of model randomness.

Return type:

TensorDict

Returns:

Canonical TensorDict with obs/actions/valid/loss_mask and any available frame_ids/generations and segment/language fields. Optional acceptance Unix nanoseconds and publication host-monotonic nanoseconds remain provenance, not inputs to feature inference.

class embodichain.data_pipeline.datasets.OnlineDataset[source]#

Bases: IterableDataset

Infinite IterableDataset backed by a live OnlineDataEngine shared buffer.

Two sampling modes are supported depending on the batch_size argument:

Item mode (batch_size=None, default)

__iter__ yields one TensorDict of shape [chunk_size] per step. Use with a standard DataLoader(dataset, batch_size=B) so the DataLoader handles collation and worker sharding.

Batch mode (batch_size=N)

__iter__ yields one pre-batched TensorDict of shape [N, chunk_size] per step by calling engine.sample_batch(N, chunk_size) directly. Use with DataLoader(dataset, batch_size=None) to skip DataLoader collation and leverage the engine’s bulk-sampling efficiency.

Dynamic chunk sizes

Pass a ChunkSizeSampler as chunk_size to draw a fresh chunk length on every iteration step. In batch mode the size is sampled once per step and applied uniformly to all trajectories in the batch, ensuring a consistent [batch_size, chunk_size] shape. Two built-in samplers are provided:

UniformChunkSampler — uniform discrete distribution over [low, high].
GMMChunkSampler — Gaussian Mixture Model, useful for multi-modal chunk-length curricula.

Note

__len__ is intentionally absent — IterableDataset does not require it and the stream is infinite.

Note

Multi-worker DataLoader: each worker gets its own iterator; since sampling is independent random draws from shared memory, this is safe.

Language-conditioned training uses the numeric task_index and subtask_index fields. Call resolve_language() on the copied item or collated batch to get the overall task and current segment text, then tokenize them in the consumer or a transform. The text dictionaries append new entries without changing existing indices, so previously sampled items retain their language when the engine refills trajectory slots.

New context consumers may use SequenceDataset(OnlineDataSource(...), sequence, ...) directly. This wrapper preserves legacy chunk APIs and delegates its opt-in context iteration to that same Dataset implementation.

Parameters:

engine (OnlineDataEngine) – A started OnlineDataEngine whose shared buffer is used for sampling.
chunk_size (int | ChunkSizeSampler | None) – Fixed number of consecutive timesteps per chunk (int), or a ChunkSizeSampler that returns a fresh size on every iteration step.
batch_size (Optional[int]) – If None, yield single chunks of shape [chunk_size] (item mode). If an int, yield pre-batched TensorDicts of shape [batch_size, chunk_size] (batch mode).
transform (Optional[Callable[[TensorDict], TensorDict]]) – Optional (TensorDict) -> TensorDict applied to each yielded item/batch before returning.
sampling_mode (Literal['episode', 'segment', 'boundary']) – "episode" allows chunks to cross segment boundaries, "segment" keeps chunks within one segment, and "boundary" samples chunks that cross a boundary.
sequence (SequenceSpecCfg | None) – Opt-in causal context specification. Omit chunk_size and configure segment mode through this specification when supplied.
schema (Mapping[str, Any] | None) – Explicit observation/action semantics required in context mode.
observation_transform (Optional[Callable[[Any], Tensor]]) – Optional row-local observation conversion for context mode; canonical padding is restored after conversion.
generator (Generator | None) – Optional context-sampling RNG; explicit generators require owner-process iteration to avoid duplicated worker streams.

Example — fixed chunk size, item mode:

dataset = OnlineDataset(engine, chunk_size=64)
loader  = DataLoader(dataset, batch_size=32, num_workers=4,
                     collate_fn=OnlineDataset.collate_fn)
for batch in loader:
    # batch has shape [32, 64, ...]
    train_step(batch)

Example — fixed chunk size, batch mode:

dataset = OnlineDataset(engine, chunk_size=64, batch_size=32)
loader  = DataLoader(dataset, batch_size=None,
                     collate_fn=OnlineDataset.passthrough_collate_fn)
for batch in loader:
    # batch has shape [32, 64, ...]
    train_step(batch)

Example — dynamic chunk size with uniform sampler:

sampler = UniformChunkSampler(low=16, high=64)
dataset = OnlineDataset(engine, chunk_size=sampler)
loader  = DataLoader(dataset, batch_size=32)
for batch in loader:
    # chunk dimension varies each batch
    train_step(batch)

Example — dynamic chunk size with GMM sampler:

sampler = GMMChunkSampler(
    means=[16.0, 64.0], stds=[4.0, 8.0], weights=[0.6, 0.4],
    low=8, high=96,
)
dataset = OnlineDataset(engine, chunk_size=sampler, batch_size=32)
loader  = DataLoader(dataset, batch_size=None)
for batch in loader:
    train_step(batch)

Methods:

`__init__`(engine[, chunk_size, batch_size, ...])
`collate_fn`(batch)	Collate a list of TensorDicts into a single batched TensorDict.
`passthrough_collate_fn`(batch)	Collate function for batch-mode DataLoaders.
`resolve_language`(batch)	Decode the overall and segment instructions for an item or batch.

__init__(engine, chunk_size=None, batch_size=None, transform=None, sampling_mode='episode', *, sequence=None, schema=None, observation_transform=None, generator=None)[source]#

static collate_fn(batch)[source]#

Collate a list of TensorDicts into a single batched TensorDict.

Pass this as collate_fn to DataLoader when using item mode (batch_size not None on the DataLoader side) to avoid the default collation failure with TensorDict objects.

Parameters:: batch (List[TensorDict]) – List of TensorDicts, each of shape [chunk_size, ...].
Return type:: TensorDict
Returns:: Stacked TensorDict of shape [len(batch), chunk_size, ...].

static passthrough_collate_fn(batch)[source]#

Collate function for batch-mode DataLoaders.

When the dataset is in batch mode it already yields pre-batched TensorDicts. With batch_size=None, PyTorch’s DataLoader skips auto-batching and passes each item directly to collate_fn as-is (not wrapped in a list). This function returns the TensorDict unchanged.

Pass this as collate_fn to DataLoader when using batch mode (batch_size=None on the DataLoader side) to avoid the default collation failure with TensorDict objects.

Parameters:: batch (TensorDict) – A pre-batched TensorDict of shape [batch_size, chunk_size, ...] passed directly by the DataLoader.
Return type:: TensorDict
Returns:: The pre-batched TensorDict unchanged.

resolve_language(batch)[source]#

Decode the overall and segment instructions for an item or batch.

Parameters:: batch (TensorDict) – Copied numeric sample with task/subtask index fields.
Return type:: dict[str, object]
Returns:: task and subtask strings nested to match batch dimensions. Lookup remains valid after trajectory refills and engine shutdown.

class embodichain.data_pipeline.datasets.SequenceDataset[source]#

Bases: IterableDataset

Iterate owned causal sequences from a live or resident data source.

Parameters:

source (BatchSource) – BatchSource, such as OnlineDataSource, LeRobotDataSource or TensorDataSource. The caller owns its loading/production lifecycle.
sequence (SequenceSpecCfg) – Causal window and boundary requirements.
batch_size (int | None) – None yields one context for DataLoader collation; a positive integer yields complete batches (use DataLoader batch_size=None).
transform (Callable[[TensorDict], TensorDict] | None) – Optional batch/item transform preserving axes and masks.
generator (Generator | None) – Optional CPU sampling RNG. Explicit generators require num_workers=0 to prevent copied workers repeating the same stream.
max_batches (int | None) – Optional positive draws per iterator. Stored cardinality does not imply iteration length: random sampling otherwise repeats. Finite iteration requires num_workers=0.

There is no inferred epoch length or automatic refill. max_batches is a bound on each iterator, not a source quota or a checkpointed update count. Sampling charges delivery before any transform or optimizer update; a failed consumer does not refund it. Prefetched batches also count.

Quota exhaustion and source errors propagate immediately. Budgeted sources reject unsupported worker copies; this wrapper never clones allowances, retries permanent failures or guesses an epoch length.

Methods:

`__init__`(source, sequence, *[, batch_size, ...])
`collate_fn`(batch)	Stack contexts for item-mode DataLoader use.
`passthrough_collate_fn`(batch)	Pass through an already batched item.
`resolve_language`(batch)	Resolve source-local language IDs when the source supplies dictionaries.

__init__(source, sequence, *, batch_size=None, transform=None, generator=None, max_batches=None)[source]#

static collate_fn(batch)[source]#

Stack contexts for item-mode DataLoader use.

Parameters:: batch (list[TensorDict]) – Individual owned contexts.
Return type:: TensorDict
Returns:: TensorDict with a new leading batch axis.

static passthrough_collate_fn(batch)[source]#

Pass through an already batched item.

Parameters:: batch (TensorDict) – Source-owned copied minibatch.
Return type:: TensorDict
Returns:: The supplied TensorDict unchanged.

resolve_language(batch)[source]#

Resolve source-local language IDs when the source supplies dictionaries.

Parameters:: batch (TensorDict) – Copied numeric batch containing task/subtask indices.
Return type:: dict[str, object]
Returns:: Overall and segment text with the batch’s nested shape.

class embodichain.data_pipeline.datasets.SequenceSpecCfg[source]#

Bases: object

Specify causal history and current/future supervision.

Parameters:

history_length (int) – Number of observation frames, including the current one.
prediction_horizon (int) – Number of target frames, including the current one.
sampling_mode (Literal['episode', 'segment']) – Allow segment crossings within an episode, or restrict context to one accepted segment. Continuity changes always pad.
generations (tuple[int, ...] | None) – Optional source-generation filter. Missing generation metadata raises rather than silently ignoring this filter.

The current index is history_length - 1 in a sequence of length history_length + prediction_horizon - 1. Numeric padding is zero, source/language padding is -1, and only real current/future frames enter loss_mask. Boundary-crossing chunk sampling remains an engine chunk API.

Methods:

`__init__`([history_length, ...])
`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:

`generations`
`history_length`
`prediction_horizon`
`sampling_mode`

__init__(history_length=<factory>, prediction_horizon=<factory>, sampling_mode=<factory>, generations=<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.

generations: tuple[int, ...] | None#

history_length: int#

prediction_horizon: int#

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.

sampling_mode: Literal['episode', 'segment']#

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.

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.

class embodichain.data_pipeline.datasets.TensorDataSource[source]#

Bases: object

Sample causal sequences from resident, explicitly described CPU data.

Parameters:

data (TensorDictBase) – TensorDict with [episode, time] batch axes, obs, floating actions and boolean valid. Optional continuity, segment, language and provenance fields are preserved.
schema (Mapping[str, Any]) – Ordered observation/action semantics. No feature conversion is inferred from vector dimensions.
source_id (str) – Stable source dataset/run identity. Persist this for datasets without recorded frame_ids; it seeds their numeric source ID.
copy_data (bool) – Clone the supplied data by default. False transfers storage responsibility to the caller, which must not mutate it afterwards.
metadata (Mapping[str, Any] | None) – Optional detached storage/provenance diagnostics.

Each returned batch owns its storage. Optional source quotas are owner-only and cannot be reset. data is exposed for inspection and fitting frozen training statistics; treat its tensors as immutable after construction.

Methods:

`__init__`(data, *, schema, source_id[, ...])
`configure_sampling_budget`(budget, hard_limit)	Assign a permanent per-source chunk allowance before first sampling.
`get_sampling_budget_snapshot`()	Return detached delivery counts, using one context as one chunk.
`resolve_language`(batch)	Resolve IDs using source-owned dictionaries in metadata['language'].
`sample_context`(batch_size, spec, *[, generator])	Draw complete owned contexts, charging quotas only after copying.

__init__(data, *, schema, source_id, copy_data=True, metadata=None)[source]#

configure_sampling_budget(budget, hard_limit)[source]#

Assign a permanent per-source chunk allowance before first sampling.

The stateful sampler belongs to its creating process and thread. Copying it into DataLoader workers cannot duplicate allowances. Valid rows must each identify one distinct immutable (run, episode) source. Empty padded rows receive no allowance. There is no reset or refill operation.

Parameters:

budget (int) – Maximum delivered contexts for each retained source episode.
hard_limit (int) – Explicit protection ceiling; budget must not exceed it.

Return type:

None

get_sampling_budget_snapshot()[source]#

Return detached delivery counts, using one context as one chunk.

Return type:: dict[str, Any]
Returns:: Per-episode allowances/deliveries, total remaining credits and continuity-qualified eligible remaining credits. A sequence’s generation or segment filter can restrict eligibility further.

resolve_language(batch)[source]#

Resolve IDs using source-owned dictionaries in metadata[‘language’].

Parameters:: batch (TensorDict) – Copied task/subtask indices; select real frames before resolving since padding and unknown IDs are rejected.
Return type:: dict[str, object]
Returns:: Nested task/subtask texts with the corresponding index shapes.

sample_context(batch_size, spec, *, generator=None)[source]#

Draw complete owned contexts, charging quotas only after copying.

Parameters:

batch_size (int) – Positive number of contexts with replacement.
spec (SequenceSpecCfg) – Window lengths, boundary behavior and generation filter.
generator (Generator | None) – Optional CPU sampling RNG.

Return type:

TensorDict

Returns:

Canonical TensorDict of shape [B, H + F - 1] with validity and supervision masks. Gaps and continuity/segment boundaries pad.

class embodichain.data_pipeline.datasets.UniformChunkSampler[source]#

Bases: ChunkSizeSampler

Discrete-uniform chunk-size sampler over [low, high].

Draws an integer uniformly at random from the closed interval [low, high] on every call.

Parameters:

low (int) – Minimum chunk size (inclusive, must be ≥ 1).
high (int) – Maximum chunk size (inclusive, must be ≥ low).

Raises:

ValueError – If low < 1 or high < low.

Example:

sampler = UniformChunkSampler(low=16, high=64)
chunk_size = sampler()  # e.g. 37

Methods:

__init__(low, high)

__init__(low, high)[source]#

embodichain.data_pipeline.datasets.data_schema_fingerprint(schema)[source]#

Hash a validated finite JSON schema without changing feature order.

Parameters:: schema (Mapping[str, Any]) – Persistable declaration, including any feature-transform, normalization, coordinate-frame and action-contract versions.
Return type:: str
Returns:: SHA-256 of canonical UTF-8 JSON. Mapping key order is ignored; ordered lists and caller-declared versions remain significant. This is a schema identity, not a hash of trajectory content or callable implementation.
Raises:: ValueError – The schema is invalid or not finite JSON data.

embodichain.data_pipeline.datasets.validate_data_schema(schema)[source]#

Validate declarations common to live and resident causal sources.

Parameters:: schema (Mapping[str, Any]) – Observation/action declarations. Actions require explicit semantics or representation and finite positive step_dt. Optional names are ordered and unique; dtype names use Torch’s spelling without its prefix. Optional observation timing must be pre_action. Root and feature version values identify schemas and feature transformations when supplied.
Return type:: None

This validates declarations, not their truth or compatibility with a live robot. Source adapters also validate stored features and action contracts. Units, frames, rotations and normalization remain caller declarations; no conversion or missing semantic information is inferred from vector widths.

Raises:: ValueError – A required semantic declaration is absent or malformed.

Depth Video#

embodichain.data_pipeline.depth_video

Online Data Engine#

Process-safe shared buffer (OnlineDataEngine) that decouples simulation producers from training consumers.

data

Exceptions:

`OnlineDataBudgetExhaustedError`	alias of `DataBudgetExhaustedError`
`OnlineDataWorkerError`	Fallback error for a worker exception that cannot be reconstructed.

Classes:

`OnlineDataEngine`	Engine for managing Online Data Streaming (ODS) and environment rollouts.
`OnlineDataEngineCfg`	OnlineDataEngineCfg(buffer_size: 'int' = <factory>, max_episode_steps: 'int' = <factory>, state_dim: 'int' = <factory>, buffer_device: 'str' = <factory>, gym_config: 'dict' = <factory>, action_config: 'dict' = <factory>, refill_threshold: 'int' = <factory>, max_generation_attempts: 'int' = <factory>, initialization_timeout: 'float' = <factory>, language_buffer_bytes: 'int' = <factory>, action_dim: 'int \| None' = <factory>, refill_policy: "Literal['sample_threshold', 'manual']" = <factory>, include_provenance: 'bool' = <factory>, episode_sample_budget: 'int \| None' = <factory>, episode_sample_hard_limit: 'int \| None' = <factory>, lifecycle_event_capacity: 'int' = <factory>, action_semantics: "Literal['expert_controller_command', 'policy_request']" = <factory>, action_contracts: 'tuple[str, ...] \| None' = <factory>, shutdown_timeout: 'float' = <factory>)
`OnlineDataEngineState`	Lifecycle states for `OnlineDataEngine`.

embodichain.data_pipeline.engine.OnlineDataBudgetExhaustedError: alias of DataBudgetExhaustedError

class embodichain.data_pipeline.engine.OnlineDataEngine[source]

Bases: object

Engine for managing Online Data Streaming (ODS) and environment rollouts.

Creates a shared rollout buffer in CPU shared memory, spawns a dedicated simulation subprocess that fills the buffer with demonstration trajectories, and exposes a sample_batch() method for the training process to draw batches of trajectory chunks.

Subprocess lifecycle

The simulation subprocess is started in start() and immediately receives a fill signal so the buffer is populated before the first call to sample_batch(). The subprocess loops indefinitely: it waits for fill_signal, runs buffer_size // num_envs rollouts to overwrite every buffer slot, then goes back to waiting.

Concurrency and lock protection

_lock_index [write_start, write_end) is updated by the subprocess after each rollout so that sample_batch() can skip the slot currently being written to, preventing partial reads.

Refill criterion

sample_batch() accumulates the total number of individual trajectory samples drawn into _sample_count. When this counter exceeds refill_threshold the fill signal is raised and the counter resets to zero. This amortises the cost of GPU-accelerated simulation across many training iterations.

Lifecycle state

Every instance starts in OnlineDataEngineState.CREATED, passes through STARTING while the first fill is running, and only serves data in READY. Worker failures transition to FAILED and explicit cleanup transitions to terminal STOPPED; failed or stopped instances cannot be restarted.

Parameters:: cfg (OnlineDataEngineCfg) – Engine configuration.

shared_buffer: Shared-memory TensorDict of shape [buffer_size, max_episode_steps, ...].

buffer_size: Total number of trajectory slots in the shared buffer.

device: Device of the shared buffer.

state: Current OnlineDataEngineState.

is_init: True only while the engine is ready to sample.

Methods:

`__init__`(cfg)
`check_health`()	Require a ready engine and surface existing worker failures.
`drain_lifecycle_events`()	Return and remove pending publication/eviction records in order.
`get_flow_status`()	Return detached counters for READY-only host flow control.
`get_sampling_budget_snapshot`()	Return a detached quota audit under the producer/sampling copy lock.
`get_stats`()	Return a detached snapshot of production, consumption and valid rows.
`request_refill`()	Request a fill, coalescing an outstanding manual request.
`resolve_language`(batch)	Decode overall tasks and current subtasks for a copied sample batch.
`sample_batch`(batch_size, chunk_size[, ...])	Sample a batch of trajectory chunks from the shared rollout buffer.
`sample_batch_with_context`(batch_size, ...[, ...])	Sample causal observations and future supervision around valid anchors.
`start`()	Start the worker and block until its first fill completes.
`stop`()	Terminate the simulation subprocess and release resources.
`wait_for_generation`(generation[, timeout])	Wait for a requested generation while propagating worker failures.

Attributes:

`is_init`	Whether the engine is ready to serve initialized data.
`state`	Return the engine's current lifecycle state.

__init__(cfg)[source]

check_health()[source]

Require a ready engine and surface existing worker failures.

This check neither samples data nor requests a refill. It checks worker liveness through the existing owner/consumer error channel.

Raises:

RuntimeError – If the engine is not READY or its worker has exited.
BaseException – The original reported worker or startup exception.

Return type:

None

drain_lifecycle_events()[source]

Return and remove pending publication/eviction records in order.

Owner-only, including after stop or failure, so diagnostics are not lost. Disabled logging returns an empty list. Each record contains kind, sequence, run_id, episode_id, generation, slot, monotonic_ns, valid_frame_indices, eligible_anchor_indices, valid_frames and eligible_anchors. Eligible anchors refer to episode-mode context sampling (valid frames with nonnegative continuity). This is a host-local monotonic clock, not a wall-clock timestamp.

Drain regularly while producing, including during owner-side waits. wait_for_generation does not consume these records and may time out if an enabled log is full. Records are detached from later row reuse.

Return type:: list[dict[str, object]]
Returns:: Detached lifecycle records in publication order, or an empty list when logging is disabled or no records are pending.
Raises:: RuntimeError – If called outside the engine’s owner process.

get_flow_status()[source]

Return detached counters for READY-only host flow control.

Producer counters and quota totals use separate lock snapshots. The quota sum includes only published, unlocked rows with a valid anchor and nonnegative continuity when continuity metadata is present. It is a coarse bound; sequence lengths and filters may reduce eligibility. This method creates no per-row audit records or payload copies.

Return type:

dict[str, int | None]

Returns:

Requested/completed generations and successful episodes, plus unlocked_remaining_chunks and lifetime total_delivered_chunks. Quota counters are None when protection is disabled.

Raises:

RuntimeError – If the engine is not READY or its worker has exited.
BaseException – The original reported worker or startup exception.

get_sampling_budget_snapshot()[source]

Return a detached quota audit under the producer/sampling copy lock.

rows describes currently retained source identities and includes run_id, episode_id, generation, slot, assigned, delivered, remaining and locked. Lifetime total_delivered_chunks includes retired sources. The unlocked sum covers sources with at least one valid continuity-qualified anchor; a requested chunk length or sampling mode may further restrict eligibility. Known worker failures are propagated; clean STOPPED engines remain auditable. Disabled quotas return empty rows and unavailable counters.

Return type:: dict[str, object]

get_stats()[source]

Return a detached snapshot of production, consumption and valid rows.

Return type:: dict[str, int | float]
Returns:: Engine identity, requested/completed generations, episode/frame counts, cumulative production seconds and trajectory sample count.

property is_init: bool: Whether the engine is ready to serve initialized data.

request_refill()[source]

Request a fill, coalescing an outstanding manual request.

Return type:: int
Returns:: Generation that will contain the requested fresh data.
Raises:: RuntimeError – If called outside READY or from a reader process.

resolve_language(batch)[source]

Decode overall tasks and current subtasks for a copied sample batch.

The lookup remains valid after a ring-buffer refill or engine shutdown. Returned strings are separate from the shared numeric TensorDict; callers can tokenize them in a training transform without changing sampling locks or collation behavior.

Parameters:: batch (TensorDict) – Sampled TensorDict containing task_index/subtask_index.
Return type:: dict[str, object]
Returns:: task and subtask strings nested to match the batch dimensions (for example [batch_size][chunk_size]).
Raises:: ValueError – If language indices are absent, unknown, or padding.

sample_batch(batch_size, chunk_size, sampling_mode='episode')[source]

Sample a batch of trajectory chunks from the shared rollout buffer.

Only fully valid windows are candidates, so padding or stale tail frames are never returned. episode mode allows a window to cross segment boundaries within one causal-continuity region, segment keeps every window inside one accepted segment, and boundary deliberately samples windows crossing a boundary between accepted segments. No mode crosses a discontinuous state-restore boundary.

After sampling the internal _sample_count is incremented by batch_size; if the count exceeds refill_threshold a buffer refill is triggered automatically.

Parameters:

batch_size (int) – Number of trajectory chunks to include in the batch.
chunk_size (int) – Number of consecutive timesteps in each chunk.
sampling_mode (Literal['episode', 'segment', 'boundary']) – Segment-boundary policy for candidate windows.

Return type:

TensorDict

Returns:

TensorDict with batch size [batch_size, chunk_size]. Integer task_index and subtask_index identify immutable language snapshots; resolve_language() decodes their descriptions.

Raises:

ValueError – If an argument is invalid.
DataUnavailableError – If no unlocked valid window satisfies the policy. Subclasses RuntimeError for existing callers.
RuntimeError – If a producer fails or the engine is not ready.
OnlineDataBudgetExhaustedError – If eligible source quotas cannot supply the complete batch. No partial batch is charged.

sample_batch_with_context(batch_size, history_length, prediction_horizon, *, generator=None, sampling_mode='episode', generations=None)[source]

Sample causal observations and future supervision around valid anchors.

The current frame is at history_length - 1. Every valid frame, including episode starts and ends, can be an anchor. Context never crosses invalid frames or continuity boundaries. Out-of-range numeric payloads are zero; language/provenance IDs are -1. valid marks real frames and loss_mask selects only valid current/future targets. Selection, provenance and payload copies share the producer lock.

Parameters:

batch_size (int) – Number of contexts, with replacement.
history_length (int) – Number of causal observation frames including now.
prediction_horizon (int) – Number of target frames including now.
generator (Generator | None) – Optional RNG isolated from augmentation/model randomness.
sampling_mode (Literal['episode', 'segment']) – Episode context or one accepted segment.
generations (tuple[int, ...] | None) – Optional positive generation IDs; requires provenance.

Return type:

TensorDict

Returns:

TensorDict of shape [batch_size, history_length + prediction_horizon - 1]. Consumers must mask padded supervision and explicitly handle missing initial observation history.

Raises:

ValueError – If context lengths are not positive integers.
DataUnavailableError – If no unlocked valid anchor exists.
RuntimeError – If the worker fails or the engine is not ready.
OnlineDataBudgetExhaustedError – If eligible source quotas cannot supply the complete batch. Each context costs one chunk.

start()[source]

Start the worker and block until its first fill completes.

Raises:

RuntimeError – If the engine was already started or stopped.
TimeoutError – If the first fill exceeds initialization_timeout.
BaseException – The original exception raised by the worker.

Return type:

None

property state: OnlineDataEngineState: Return the engine’s current lifecycle state.

stop()[source]

Terminate the simulation subprocess and release resources.

Sets the close signal and waits up to cfg.shutdown_timeout for the subprocess to exit gracefully (it checks the signal between rollout steps). If it remains alive after that wait, it is force-terminated.

Safe to call multiple times — subsequent calls are no-ops if the subprocess has already been terminated.

Return type:: None

wait_for_generation(generation, timeout=300.0)[source]

Wait for a requested generation while propagating worker failures.

Parameters:

generation (int) – Positive generation returned by request_refill.
timeout (float) – Finite positive timeout in seconds.

Raises:

ValueError – If the generation was not requested or timeout is invalid.
TimeoutError – If publication does not finish before the deadline.

Return type:

None

class embodichain.data_pipeline.engine.OnlineDataEngineCfg[source]

Bases: object

OnlineDataEngineCfg(buffer_size: ‘int’ = <factory>, max_episode_steps: ‘int’ = <factory>, state_dim: ‘int’ = <factory>, buffer_device: ‘str’ = <factory>, gym_config: ‘dict’ = <factory>, action_config: ‘dict’ = <factory>, refill_threshold: ‘int’ = <factory>, max_generation_attempts: ‘int’ = <factory>, initialization_timeout: ‘float’ = <factory>, language_buffer_bytes: ‘int’ = <factory>, action_dim: ‘int | None’ = <factory>, refill_policy: “Literal[‘sample_threshold’, ‘manual’]” = <factory>, include_provenance: ‘bool’ = <factory>, episode_sample_budget: ‘int | None’ = <factory>, episode_sample_hard_limit: ‘int | None’ = <factory>, lifecycle_event_capacity: ‘int’ = <factory>, action_semantics: “Literal[‘expert_controller_command’, ‘policy_request’]” = <factory>, action_contracts: ‘tuple[str, …] | None’ = <factory>, shutdown_timeout: ‘float’ = <factory>)

Methods:

`__init__`([buffer_size, max_episode_steps, ...])
`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:

`action_config`	Action configuration dictionary.
`action_contracts`	Expected ActionManager contract IDs in flat policy order.
`action_dim`	Stored action width; None preserves the historical state_dim width.
`action_semantics`	Stored action meaning; the default preserves expert-controller commands.
`buffer_device`	Device on which the shared buffer is allocated.
`buffer_size`	Number of episodes (environment trajectories) that can be stored in the shared buffer at once.
`episode_sample_budget`	Maximum delivered chunks assigned to each fresh source episode.
`episode_sample_hard_limit`	Positive signed-int64 ceiling for `episode_sample_budget`.
`gym_config`	Gym environment configuration dictionary (already loaded, not a file path).
`include_provenance`	Include numeric run/episode/generation IDs, episode frame indices, and successful-publication Unix and host-monotonic timestamps in nanoseconds.
`initialization_timeout`	Maximum seconds to wait for the worker's initial buffer fill.
`language_buffer_bytes`	Append-only task/subtask dictionary capacity in UTF-8 bytes.
`lifecycle_event_capacity`	Optional bounded publication/eviction log; zero disables logging.
`max_episode_steps`	Maximum number of timesteps per episode.
`max_generation_attempts`	Maximum planning/execution attempts for each buffer write transaction.
`refill_policy`	Automatic sample-count refill or owner-requested refill.
`refill_threshold`	Total number of samples (refill_threshold * buffer_size) drawn from the shared buffer before a refill is triggered.
`shutdown_timeout`	Maximum seconds to wait for graceful worker close before forcing shutdown.
`state_dim`	Dimensionality of the state space.

__init__(buffer_size=<factory>, max_episode_steps=<factory>, state_dim=<factory>, buffer_device=<factory>, gym_config=<factory>, action_config=<factory>, refill_threshold=<factory>, max_generation_attempts=<factory>, initialization_timeout=<factory>, language_buffer_bytes=<factory>, action_dim=<factory>, refill_policy=<factory>, include_provenance=<factory>, episode_sample_budget=<factory>, episode_sample_hard_limit=<factory>, lifecycle_event_capacity=<factory>, action_semantics=<factory>, action_contracts=<factory>, shutdown_timeout=<factory>)

action_config: dict: Action configuration dictionary. The contents depend on the specific environment and robot being used.

action_contracts: tuple[str, ...] | None

Expected ActionManager contract IDs in flat policy order.

Required with an explicit action_dim for policy_request recording. The worker rejects a different live contract sequence, width, or joint order.

action_dim: int | None: Stored action width; None preserves the historical state_dim width.

action_semantics: Literal['expert_controller_command', 'policy_request']: Stored action meaning; the default preserves expert-controller commands.

buffer_device: str: Device on which the shared buffer is allocated.

buffer_size: int: Number of episodes (environment trajectories) that can be stored in the shared buffer at once. Must be ≥ num_envs and ideally a multiple of num_envs.

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.

episode_sample_budget: int | None

Maximum delivered chunks assigned to each fresh source episode.

Each returned batch row costs one, regardless of context length or repeated selection of the same episode. Requires episode_sample_hard_limit and a CPU shared buffer. None disables per-source quotas without changing sampling. Exhaustion does not itself request a refill; choose manual replenishment or an automatic threshold that can be reached within the assigned quotas.

episode_sample_hard_limit: int | None

Positive signed-int64 ceiling for episode_sample_budget.

Both quota fields must be supplied together and budget must not exceed this ceiling. The assigned budget is fixed at engine construction; no source may borrow another source’s unused quota or receive more on generation changes.

gym_config: dict: Gym environment configuration dictionary (already loaded, not a file path). The contents depend on the specific environment being used. Default is None.

include_provenance: bool: Include numeric run/episode/generation IDs, episode frame indices, and successful-publication Unix and host-monotonic timestamps in nanoseconds. Invalid frames use -1 for every provenance field.

initialization_timeout: float: Maximum seconds to wait for the worker’s initial buffer fill.

language_buffer_bytes: int: Append-only task/subtask dictionary capacity in UTF-8 bytes. Descriptions remain available for copied samples after trajectory slots are reused. Exhaustion fails the worker explicitly; increase this for many unique tasks.

lifecycle_event_capacity: int

Optional bounded publication/eviction log; zero disables logging.

When enabled, capacity must fit twice buffer_size and all initial-fill records (including overlapping final-batch evictions). The owner must drain events while refilling; a full log backpressures the producer outside the trajectory-copy lock. Stop interrupts this wait.

max_episode_steps: int: Maximum number of timesteps per episode. Must be ≥ chunk_size used by OnlineDataset.

max_generation_attempts: int: Maximum planning/execution attempts for each buffer write transaction.

refill_policy: Literal['sample_threshold', 'manual']: Automatic sample-count refill or owner-requested refill.

refill_threshold: int: Total number of samples (refill_threshold * buffer_size) drawn from the shared buffer before a refill is triggered. Accumulates across all calls to OnlineDataEngine.sample_batch(). When this threshold is exceeded the engine signals the simulation subprocess to regenerate the entire buffer, amortising the cost of environment simulation over many training steps.

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.

shutdown_timeout: float

Maximum seconds to wait for graceful worker close before forcing shutdown.

Must be finite and positive. Forced termination still fails the engine’s recorder-durability contract, regardless of this timeout.

state_dim: int: Dimensionality of the state space.

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.

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.

class embodichain.data_pipeline.engine.OnlineDataEngineState[source]

Bases: str, Enum

Lifecycle states for OnlineDataEngine.

Attributes:

`CREATED`
`FAILED`
`READY`
`STARTING`
`STOPPED`

Methods:

__new__(value)

CREATED = 'CREATED'

FAILED = 'FAILED'

READY = 'READY'

STARTING = 'STARTING'

STOPPED = 'STOPPED'

__new__(value)

exception embodichain.data_pipeline.engine.OnlineDataWorkerError[source]

Bases: RuntimeError

Fallback error for a worker exception that cannot be reconstructed.

Owner-side flow control#

The application starts and stops its manual-refill engine. The optional controller applies a measured plan, drains publication events, checks health, and bounds generation/data/quota waits. A plan or production limit requires an explicit producer count dividing cache size. Sources and Datasets retain immediate sampling/error behavior. See Configure ODS from measured throughput for matching engine settings, callback clocks and an integration example.

Owner-side flow control for an explicitly started online data engine.

`embodichain.data_pipeline.runtime.ODSFlowControllerCfg`	Bound owner-side waits and optional whole-cache production.
`embodichain.data_pipeline.runtime.ODSFlowController`	Apply a measured plan without changing sampling or engine ownership.

class embodichain.data_pipeline.runtime.ODSFlowControllerCfg[source]#

Bases: object

Bound owner-side waits and optional whole-cache production.

Parameters:

data_wait_timeout (float) – Maximum seconds blocked on missing windows or quota.
generation_timeout (float) – Maximum seconds waiting for a complete refill.
poll_interval (float) – Maximum sleep/probe duration between health/event checks.
max_generated_episodes (int | None) – Optional accepted-episode ceiling. A refill must fit a complete cache; unused partial-cache capacity is not produced.

Methods:

`__init__`([data_wait_timeout, ...])
`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.

__init__(data_wait_timeout=<factory>, generation_timeout=<factory>, poll_interval=<factory>, max_generated_episodes=<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.

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.

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.

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.

class embodichain.data_pipeline.runtime.ODSFlowController[source]#

Bases: object

Apply a measured plan without changing sampling or engine ownership.

The creating process/thread owns this controller. The application starts and stops the engine and owns optimizer/evaluation deadlines. Sources and Datasets retain their immediate sampling/error contracts.

Parameters:

engine (OnlineDataEngine) – Creator-owned engine configured for manual refill.
cfg (ODSFlowControllerCfg | None) – Wait bounds and optional production limit.
plan (ODSConfiguration | None) – Optional configuration to enforce before each batch. Cache, producer count, source allowances and batch size must match the plan.
progress (Callable[[], object] | None) – Called during polling; may raise for cancellation/deadlines.
on_events (Callable[[list[dict[str, object]]], object] | None) – Receives drained lifecycle records, including after stop.
on_refill (Callable[[int, float], object] | None) – Receives a new requested generation and its absolute host monotonic request time. Coalesced requests are reported once.
on_interval (Callable[[str, float, float], object] | None) – Receives wait kind and absolute host monotonic start/end.

Methods:

`__init__`(engine[, cfg, plan, progress, ...])
`drain_events`()	Drain accounting records, also during failure/shutdown diagnostics.
`poll`()	Check caller deadlines, drain lifecycle events and verify worker health.
`prepare_batch`(batch_size)	Refill/pace a planned batch; False means final supply is exhausted.
`request_refill`()	Request or coalesce a fill within the accepted-episode ceiling.
`sample_context`(source, batch_size, sequence, *)	Retry only transient live-sampling errors within the data deadline.
`wait_for_generation`(generation, *[, deadline])	Wait with a total bound, optionally shortened by an absolute deadline.

__init__(engine, cfg=None, *, plan=None, progress=None, on_events=None, on_refill=None, on_interval=None)[source]#

drain_events()[source]#

Drain accounting records, also during failure/shutdown diagnostics.

Return type:: list[dict[str, object]]
Returns:: Detached lifecycle records delivered to the optional event observer.

poll()[source]#

Check caller deadlines, drain lifecycle events and verify worker health.

Return type:: None

prepare_batch(batch_size)[source]#

Refill/pace a planned batch; False means final supply is exhausted.

Unplanned engines are only polled; their refill schedule stays explicit. Quota readiness is a coarse bound. Sampling still validates the exact context/filter request and can require a bounded data wait.

Parameters:: batch_size (int) – Positive chunk count, matching the plan when supplied.
Return type:: bool
Returns:: False only when capped final supply cannot fund a complete batch.
Raises:: TimeoutError – If missing unlocked quota exceeds the data wait bound.

request_refill()[source]#

Request or coalesce a fill within the accepted-episode ceiling.

Return type:: int | None
Returns:: Pending or newly requested generation, or None if no whole cache fits.

sample_context(source, batch_size, sequence, *, generator=None)[source]#

Retry only transient live-sampling errors within the data deadline.

Call prepare_batch() first to apply a plan’s pacing/supply limit. This helper accepts only a live source wrapping the controlled engine. Sample resident sources through their own sample_context() method; they are not accepted by this controller.

Parameters:

source (OnlineDataSource) – Live source wrapping this controller’s engine.
batch_size (int) – Number of chunks to sample.
sequence (SequenceSpecCfg) – Requested context and prediction window.
generator (Generator | None) – Optional sampling random generator.

Return type:

TensorDict

Returns:

The source’s batch, including masks and provenance.

Raises:

TimeoutError – If transient unavailability exceeds the data wait bound.

wait_for_generation(generation, *, deadline=None)[source]#

Wait with a total bound, optionally shortened by an absolute deadline.

Parameters:

generation (int) – Target generation returned by a refill request.
deadline (float | None) – Optional absolute host monotonic time limiting this wait.

Raises:

TimeoutError – If the generation does not complete within the bound.

Return type:

None

Recording integrity#

The offline recording utilities import without the simulator or training stack. They maintain stable episode identities, configuration fingerprints, and durable evidence for conservative metadata repair.

The usage guide Validate, split, and recover expert datasets covers validation, quality splits and interrupted commits. Export expert data for LeRobot 0.6.1 describes the independent Python 3.12 converter for official LeRobot 0.6.1 language/depth.

embodichain.data_pipeline.recording

`datasets`	Online datasets and offline inspection with lazy simulation dependencies.
`depth_video`	Compressed depth sidecar storage for LeRobot datasets on Python 3.10--3.12.
`engine`	Process-safe shared buffer (`OnlineDataEngine`) that decouples simulation producers from training consumers.
`recording`	Recording identity, durable commit diagnostics, and conservative repair.
`runtime`	Owner-side flow control for an explicitly started online data engine.

embodichain.data_pipeline

Contents

embodichain.data_pipeline#

Overview#

Datasets#

Depth Video#

Online Data Engine#

Owner-side flow control#

Recording integrity#