embodichain.data_pipeline.recording

embodichain.data_pipeline.recording#

Overview#

Recording provenance and commit evidence link training episodes, natural segment fragments, and replay trajectories through stable identities. The journal records progress around the LeRobot commit boundary; it does not make Parquet, RGB/depth video, and JSONL one atomic transaction.

Use inspect_recording after stopping a writer to diagnose incomplete commits. recover_recording(..., repair=True) can restore a missing sidecar from its journal only when committed frame and depth artifacts verify. It never replays actions or rewrites an unfinished SDK write.

See Validate, split, and recover expert datasets for journal phases, inspection statuses, UUID/replay relationships and the explicit repair procedure.

`RecordingJournal`	One atomic, fsynced commit record per stable episode UUID.
`inspect_recording`(root)	Diagnose persisted phases against frame/depth/episode-sidecar evidence.
`recover_recording`(root, *[, repair])	Inspect or repair missing episode sidecars using committed disk evidence.
`build_recording_provenance`(env)	Snapshot configuration, runtime versions, and control timing.
`stable_config_hash`(config)	Fingerprint a config by values and stable callable names.

Recording identity, durable commit diagnostics, and conservative repair.

Classes:

RecordingJournal

One atomic, fsynced commit record per stable episode UUID.

Functions:

`build_recording_provenance`(env)	Snapshot configuration, runtime versions, and control timing.
`inspect_recording`(root)	Diagnose persisted phases against frame/depth/episode-sidecar evidence.
`recover_recording`(root, *[, repair])	Inspect or repair missing episode sidecars using committed disk evidence.
`stable_config_hash`(config)	Fingerprint a config by values and stable callable names.

class embodichain.data_pipeline.recording.RecordingJournal[source]#

One atomic, fsynced commit record per stable episode UUID.

Parameters:: root (str | Path) – Dataset directory. A recorder’s single writer owns mutations; offline inspection/repair must run after that writer has stopped.

Methods:

`__init__`(root)
`advance`(episode_uuid, phase)	Durably advance a commit phase without moving backwards.
`get`(episode_uuid)	Read a persisted commit record.
`prepare`(sidecar[, depth_sensors])	Persist intended sidecar before submitting any episode frames.
`record_error`(episode_uuid, error)	Keep the last confirmed phase and persist the error.

__init__(root)[source]#

advance(episode_uuid, phase)[source]#

Durably advance a commit phase without moving backwards.

Parameters:

episode_uuid (str) – Prepared episode identity.
phase (str) – Next phase in the commit sequence.

Return type:

None

get(episode_uuid)[source]#

Read a persisted commit record.

Parameters:: episode_uuid (str) – Canonical episode identity, never a filesystem path.
Return type:: dict[str, Any] | None
Returns:: Commit evidence, or None if no write has been prepared.

prepare(sidecar, depth_sensors=None)[source]#

Persist intended sidecar before submitting any episode frames.

Parameters:

sidecar (Mapping[str, Any]) – Complete target episode metadata with episode_uuid.
depth_sensors (list[str] | None) – Depth streams that must exist before repair succeeds.

Raises:

RuntimeError – If this identity already crossed the SDK commit boundary.

Return type:

None

record_error(episode_uuid, error)[source]#

Keep the last confirmed phase and persist the error.

Parameters:

episode_uuid (str) – Prepared episode identity.
error (BaseException) – Exception raised while committing this episode.

Return type:

None

embodichain.data_pipeline.recording.build_recording_provenance(env)[source]#

Snapshot configuration, runtime versions, and control timing.

Parameters:: env (Any) – Environment exposing cfg and optional simulation timing fields.
Return type:: dict[str, Any]
Returns:: JSON-compatible provenance. This records fingerprints rather than a replacement for the full simulation initial state in a replay artifact.

embodichain.data_pipeline.recording.inspect_recording(root)[source]#

Diagnose persisted phases against frame/depth/episode-sidecar evidence.

Parameters:: root (str | Path) – Stopped recording’s dataset directory.
Return type:: dict[str, Any]
Returns:: ok, global errors, and per-episode commits. prepared without frames is safely retryable; lerobot_committing without matching frames remains unknown. Even complete records must have matching frame identities and contiguous SDK ranges on disk: SDK buffering is not crash durability. Referenced videos are decoded with PyAV; unreadable or incomplete media stays unresolved.

embodichain.data_pipeline.recording.recover_recording(root, *, repair=False)[source]#

Inspect or repair missing episode sidecars using committed disk evidence.

Parameters:

root (str | Path) – Stopped recording’s dataset directory.
repair (bool) – Explicitly permit sidecar append and journal completion. The default is read-only. Stop all recorder processes before repair.

Return type:

dict[str, Any]

Returns:

Inspection report with repaired UUIDs. Incomplete or conflicting frames/media, malformed JSONL and unknown SDK writes stay unresolved. No Parquet, video, replay artifact, or SDK metadata is rewritten.

embodichain.data_pipeline.recording.stable_config_hash(config)[source]#

Fingerprint a config by values and stable callable names.

Parameters:: config (Any) – Dataclass, mapping, sequence, or JSON-compatible configuration.
Return type:: str
Returns:: SHA-256 hexadecimal digest of the canonical configuration. Runtime object addresses are deliberately excluded; unknown object types are represented by their qualified type name.

Commit journal#

Durable recording commit evidence and conservative offline sidecar repair.

The journal cannot make LeRobot, RGB/depth videos and JSONL one transaction. lerobot_committing is deliberately ambiguous after a crash. Recovery never replays buffered frames, removes files, or loads a pickle trajectory.

Classes:

RecordingJournal

One atomic, fsynced commit record per stable episode UUID.

Functions:

`inspect_recording`(root)	Diagnose persisted phases against frame/depth/episode-sidecar evidence.
`recover_recording`(root, *[, repair])	Inspect or repair missing episode sidecars using committed disk evidence.

class embodichain.data_pipeline.recording.journal.RecordingJournal[source]#

One atomic, fsynced commit record per stable episode UUID.

Parameters:: root (str | Path) – Dataset directory. A recorder’s single writer owns mutations; offline inspection/repair must run after that writer has stopped.

Methods:

`__init__`(root)
`advance`(episode_uuid, phase)	Durably advance a commit phase without moving backwards.
`get`(episode_uuid)	Read a persisted commit record.
`prepare`(sidecar[, depth_sensors])	Persist intended sidecar before submitting any episode frames.
`record_error`(episode_uuid, error)	Keep the last confirmed phase and persist the error.

__init__(root)[source]#

advance(episode_uuid, phase)[source]#

Durably advance a commit phase without moving backwards.

Parameters:

episode_uuid (str) – Prepared episode identity.
phase (str) – Next phase in the commit sequence.

Return type:

None

get(episode_uuid)[source]#

Read a persisted commit record.

Parameters:: episode_uuid (str) – Canonical episode identity, never a filesystem path.
Return type:: dict[str, Any] | None
Returns:: Commit evidence, or None if no write has been prepared.

prepare(sidecar, depth_sensors=None)[source]#

Persist intended sidecar before submitting any episode frames.

Parameters:

sidecar (Mapping[str, Any]) – Complete target episode metadata with episode_uuid.
depth_sensors (list[str] | None) – Depth streams that must exist before repair succeeds.

Raises:

RuntimeError – If this identity already crossed the SDK commit boundary.

Return type:

None

record_error(episode_uuid, error)[source]#

Keep the last confirmed phase and persist the error.

Parameters:

episode_uuid (str) – Prepared episode identity.
error (BaseException) – Exception raised while committing this episode.

Return type:

None

embodichain.data_pipeline.recording.journal.inspect_recording(root)[source]#

Diagnose persisted phases against frame/depth/episode-sidecar evidence.

Parameters:: root (str | Path) – Stopped recording’s dataset directory.
Return type:: dict[str, Any]
Returns:: ok, global errors, and per-episode commits. prepared without frames is safely retryable; lerobot_committing without matching frames remains unknown. Even complete records must have matching frame identities and contiguous SDK ranges on disk: SDK buffering is not crash durability. Referenced videos are decoded with PyAV; unreadable or incomplete media stays unresolved.

embodichain.data_pipeline.recording.journal.recover_recording(root, *, repair=False)[source]#

Inspect or repair missing episode sidecars using committed disk evidence.

Parameters:

root (str | Path) – Stopped recording’s dataset directory.
repair (bool) – Explicitly permit sidecar append and journal completion. The default is read-only. Stop all recorder processes before repair.

Return type:

dict[str, Any]

Returns:

Inspection report with repaired UUIDs. Incomplete or conflicting frames/media, malformed JSONL and unknown SDK writes stay unresolved. No Parquet, video, replay artifact, or SDK metadata is rewritten.

Configuration provenance#

Stable configuration fingerprints without importing the simulator.

Functions:

`build_recording_provenance`(env)	Snapshot configuration, runtime versions, and control timing.
`stable_config_hash`(config)	Fingerprint a config by values and stable callable names.

embodichain.data_pipeline.recording.provenance.build_recording_provenance(env)[source]#

Snapshot configuration, runtime versions, and control timing.

Parameters:: env (Any) – Environment exposing cfg and optional simulation timing fields.
Return type:: dict[str, Any]
Returns:: JSON-compatible provenance. This records fingerprints rather than a replacement for the full simulation initial state in a replay artifact.

embodichain.data_pipeline.recording.provenance.stable_config_hash(config)[source]#

Fingerprint a config by values and stable callable names.

Parameters:: config (Any) – Dataclass, mapping, sequence, or JSON-compatible configuration.
Return type:: str
Returns:: SHA-256 hexadecimal digest of the canonical configuration. Runtime object addresses are deliberately excluded; unknown object types are represented by their qualified type name.

Segment frame mapping#

Linear-time expansion of validated, half-open episode segment ranges.

Functions:

segments_for_frames(episode_metadata, length)

Resolve each frame's segment once, rejecting ambiguous ranges.

embodichain.data_pipeline.recording.segments.segments_for_frames(episode_metadata, length)[source]#

Resolve each frame’s segment once, rejecting ambiguous ranges.

Parameters:

episode_metadata (Mapping[str, Any] | None) – Episode sidecar containing half-open segments.
length (int) – Number of observation/action pairs in the saved episode.

Return type:

list[Mapping[str, Any] | None]

Returns:

One segment reference or None per frame. Gaps have no owner and use the recorder’s legacy annotation/overall-instruction fallback.

Raises:

ValueError – If a segment is out of bounds or overlaps another segment.
TypeError – If a segment is not a mapping or has non-integer boundaries.

embodichain.data_pipeline.recording

Contents

embodichain.data_pipeline.recording#

Overview#

Commit journal#

Configuration provenance#

Segment frame mapping#