SDK reference · EmbodiFlow docs

This page is generated from the same SDK index the browser editor uses for completion, so it always matches the SDK you run. Everything is imported from embodiflow:

from embodiflow import Context, Feature, FeatureWriter, Recording, algorithm, param

The SDK ships type hints (py.typed), so your own editor completes it too, and every class and method has a docstring that the browser editor shows on hover. The descriptions below are those docstrings.

Beta. Custom algorithms and workflows are in the product and ready to try, but they haven't been through a full acceptance on real data yet. Expect some rough edges, and check results before you rely on them.

Decorators

@algorithm.standardize

Declare a standardizer: it turns one raw recording into the canonical recording.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`description`	`str \| None`	no	`None`
`checks`	`list[Check] \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`
`services`	`list[str] \| None`	no	`None`

@algorithm.check

Declare a check: it judges data and never changes it.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`checks`	`list[Check]`	yes
`description`	`str \| None`	no	`None`
`level`	`str \| None`	no	`None`
`reads`	`str \| None`	no	`None`
`inputs`	`dict[str, Input] \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`
`services`	`list[str] \| None`	no	`None`

@algorithm.transform

Declare a transform: it returns a recording with some streams replaced or cut.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`description`	`str \| None`	no	`None`
`inputs`	`dict[str, Input] \| None`	no	`None`
`checks`	`list[Check] \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`
`services`	`list[str] \| None`	no	`None`
`privacy`	`str \| None`	no	`None`

@algorithm.enrich

Declare an enricher: it computes new information without changing the recording.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`features`	`list[Feature]`	yes
`description`	`str \| None`	no	`None`
`level`	`str \| None`	no	`None`
`inputs`	`dict[str, Input] \| None`	no	`None`
`checks`	`list[Check] \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`
`services`	`list[str] \| None`	no	`None`
`privacy`	`str \| None`	no	`None`

@algorithm.segment

Declare a segmenter: it proposes episodes, time ranges with a task text.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`description`	`str \| None`	no	`None`
`labels`	`dict[str, str] \| None`	no	`None`
`inputs`	`dict[str, Input] \| None`	no	`None`
`checks`	`list[Check] \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`
`services`	`list[str] \| None`	no	`None`
`segment_source`	`str \| None`	no	`None`

@algorithm.curate

Declare a curator: it decides per episode of a dataset version whether to keep it, its split and its task.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`description`	`str \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`

@algorithm.export

Declare an exporter: it turns a dataset version into downloadable files.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`description`	`str \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`

Functions

param(default, *, minimum=None, maximum=None, enum=None, title=None, description=None, widget=None)

Declare a field of the Params dataclass with its default and limits.

Classes

ArtifactWriter

Collects the files of an export (embodiflow-artifact/v1).

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).

Method	Meaning
`format(name: str) -> None`	Name the format of the files (`lerobot-v3`, `csv`, ...).
`path(rel: str) -> Path`	A path inside the artifact to write to (parent folders are created).
`add(src: str \| Path, rel: str) -> None`	Copy the file or folder `src` into the artifact at `rel`.
`summary(doc: dict[str, Any]) -> None`	Describe the export (counts, sizes); shown on the download page.

Camera

One camera of a recording.

Attribute	Meaning
`anonymized: bool`
`distortion: list[float] \| None`
`height: int`
`intrinsics: list[list[float]] \| None`
`key: str`
`model: str \| None`
`role: str`
`video_path: Path`	The camera's mp4 in the recording tree.
`width: int`

Method	Meaning
`valid() -> 'np.ndarray'`	A boolean per frame: `False` where the camera had a gap and the video holds a black placeholder.
`frames(start: int=0, stop: int \| None=None, step: int=1, gray: bool=False, size: tuple[int, int] \| None=None) -> Iterator[Frame]`	Decode frames `start..stop` every `step` (indices relative to this recording or episode window).
`frame(i: int) -> Frame`	Decode the single frame `i`.

Check

One check result id an algorithm may report through CheckWriter.result.

Attribute	Meaning
`description: str`
`id: str`
`title: str`

Method	Meaning
`to_dict() -> dict[str, Any]`	The declaration as the manifest input.

CheckWriter

Collects check results (embodiflow-checks/v1).

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).

Method	Meaning
`result(id: str, status: str, value: float \| None=None, threshold: float \| None=None, message: str='', evidence: Sequence[str \| Path] \| None=None, at: Mapping[str, Any] \| None=None) -> None`	Report one result.
`evidence_path(name: str) -> Path`	A path under `evidence/` to write a picture or file to; pass it to `result(evidence=[...])`.
`metrics(doc: dict[str, Any]) -> None`	Attach an `embodiflow-quality-metrics/v1` body (built-in checks only).

Column

One tabular column of a recording: name, dtype, shape, names and the conventions unit, frame, source.

Attribute	Meaning
`dtype: str`
`frame: str \| None`
`name: str`
`names: list[str] \| None`
`shape: list[int]`
`source: str \| None`
`type: str \| None`
`unit: str \| None`

Context

Everything a run knows about itself: params, facts, identifiers, scratch space, log, progress.

Attribute	Meaning
`facts: Facts`
`log: logging.Logger`
`params: P`
`platform_params: dict[str, Any]`	What the platform resolved for a standardize step (`rigProfiles`, `teamDataFormat`); empty otherwise.
`recording_id: str \| None`	The recording this step works on (`None` for a dataset-level step).
`run_id: str`	The workflow run id.
`step_id: str`	The id of the step in the workflow.
`team_slug: str \| None`	The storage slug of the team (names a team feature); `None` when the job does not state it.
`tmp: Path`

Method	Meaning
`progress(done: int, total: int, message: str='') -> None`	Report progress to the run view (progress bar and ETA); safe to call often.
`not_applicable(code: str, message: str) -> NoReturn`	End the step as not applicable (exit 85, grey in the run view, never a failure).
`input_unusable(code: str, message: str) -> NoReturn`	End the step because the data is broken (exit 87): the step fails and the recording is quarantined.

DatasetVersion

A frozen dataset version: id, features (declared names and types) and members.

Attribute	Meaning
`document: dict[str, Any]`
`features: dict[str, dict[str, Any]]`
`id: str`
`members: list[Member]`

Method	Meaning
`to_pandas() -> 'pd.DataFrame'`	One row per member: id, recording, start/end/duration, task, split, labels, quality status, session, operator.

Episode

An episode of a recording: index, start_s, end_s (recording timeline), task, labels, confidence.

Attribute	Meaning
`confidence: float \| None`
`duration_s: float`	Length in seconds.
`end_s: float`
`index: int`
`labels: dict[str, Any]`
`start_s: float`
`task: str \| None`

EpisodeWriter

Collects proposed episodes (embodiflow-episodes/v1).

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).

Method	Meaning
`episode(start_s: float, end_s: float, task: str, labels: Mapping[str, Any] \| None=None, confidence: float \| None=None) -> None`	Add one episode: `startS`/`endS` in seconds, the `task` text, optional `labels` and a `confidence` in 0..1.

Facts

Facts of the main input (contract 2.1) with attribute access: ctx.facts.kind, ctx.facts.stereo.

Attribute	Meaning
`action: bool \| None`
`cameras: int \| None`
`duration_s: float \| None`
`episodes: int \| None`
`format: str \| None`
`gripper_width: bool \| None`
`imu: bool \| None`
`kind: str \| None`
`privacy_required: bool \| None`
`recorded_hand_pose: bool \| None`
`stereo: bool \| None`

Method	Meaning
`get(name: str, default: Any=None) -> Any`	The value of a fact, or `default` when it is unknown.
`as_dict() -> dict[str, Any]`	All known facts as a plain dict.

Feature

One feature an enrich algorithm declares and later writes through FeatureWriter.

Attribute	Meaning
`describe: dict[str, Any] \| None`
`dtype: str \| None`
`frame: str \| None`
`name: str`
`names: list[str] \| None`
`per: str`
`shape: list[int] \| None`
`source: str \| None`
`type: str`
`unit: str \| None`

Method	Meaning
`to_dict() -> dict[str, Any]`	The declaration as the manifest input: only the arguments that were given.

FeatureWriter

Collects the features an enrich algorithm computes (embodiflow-features/v1).

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).

Method	Meaning
`per_frame(name: str, values: Any, confidence: str \| None=None) -> None`	Write one row per frame: `values` has shape `(num_frames, *shape)` (a 1-D array for a scalar).
`per_recording(name: str, value: Any) -> None`	Write one value for the whole recording (JSON, at most 1 MiB).
`per_episode(name: str, value: Any) -> None`	Write the value for the episode being processed (episode-level `enrich` only).
`file(name: str, path: str \| Path, describe: Mapping[str, Any] \| None=None) -> None`	Attach a file (such as a depth video) to a `per="file"` feature.

Features

A features input (features.json plus frames.parquet, episodes.parquet and files/).

Attribute	Meaning
`document: dict[str, Any]`
`names: list[str]`	The names of all features.
`num_frames: int`

Method	Meaning
`info(name: str) -> dict[str, Any]`	The `features.json` entry of `name` (type, per, dtype, shape, unit, frame, source, describe).
`of_type(semantic_type: str) -> list[str]`	Names of the features whose semantic type is `semantic_type`.
`per_frame(name: str) -> 'np.ndarray'`	The values of a per-frame feature, shape `(num_frames, *shape)`; rows outside an episode window are NaN/None.
`per_recording(name: str) -> Any`	The value of a per-recording feature.
`per_episode(name: str) -> dict[int, Any]`	`{episodeIndex: value}` of a per-episode feature.
`file(name: str) -> Path`	The path of a per-file feature (a depth video).

Frame

One decoded video frame: index (in the recording or episode window), time_s and image (RGB H x W x 3 uint8, or H x W when gray).

Attribute	Meaning
`image: 'np.ndarray'`
`index: int`
`time_s: float`

Input

An extra input of type features (a calibration, a hand pose).

Attribute	Meaning
`optional: bool`
`requires: list[str]`

Method	Meaning
`to_dict() -> dict[str, Any]`	The declaration as the manifest input.

InputUnavailableError

An input the algorithm asked for is not there (absent port, missing file or download).

ManifestDriftError

The code's interface differs from the frozen manifest of this version.

ManifestError

The algorithm interface is invalid (ALGORITHM_FILE_INVALID and its siblings).

McapMessage

A decoded ROS 2 message: topic, schema, log_time_s (seconds since the file's first message).

McapReader

An MCAP file read through its summary and only the chunks a query needs.

Attribute	Meaning
`duration_s: float`
`end_ns: int`
`start_ns: int`
`topics: list[McapTopic]`

Method	Meaning
`messages(topics: Sequence[str] \| None=None, start_s: float \| None=None, end_s: float \| None=None) -> Iterator[McapMessage]`	Decoded messages in log-time order; `topics` filters by name, `start_s`/`end_s` bound log time.

McapTopic

One topic of an MCAP file: name, message schema name, message count and average rate in Hz.

Attribute	Meaning
`count: int`
`name: str`
`rate: float`
`schema: str`

Member

One episode of a dataset version.

Attribute	Meaning
`duration_s: float`	Length of the episode in seconds.
`end_s: float`
`episode_id: str`
`features: dict[str, Any]`
`labels: dict[str, Any]`
`operator: str \| None`
`quality: dict[str, Any]`
`recording_id: str`
`root: str`
`session: str \| None`
`split: str \| None`
`start_s: float`
`task: str \| None`

Method	Meaning
`recording() -> Recording`	The episode's recording window, downloading its files on first use.

OutputError

A writer refused what the algorithm gave it.

RawFile

One uploaded file: path (relative to the upload root), size in bytes and local_path on disk.

Attribute	Meaning
`local_path: Path`
`path: str`
`size: int`

Method	Meaning
`open(mode: str='rb') -> IO[Any]`	Open the file (read-only; binary by default).

RawFiles

One recording's uploaded files, at their upload-relative paths.

Attribute	Meaning
`files: list[RawFile]`

Method	Meaning
`find(pattern: str) -> list[RawFile]`	The files whose path matches the glob `pattern` (`` within a folder, `*` across folders).
`read_json(path: str) -> Any`	Parse the JSON file at `path`.
`read_yaml(path: str) -> Any`	Parse the YAML file at `path`.
`mcap(path: str \| None=None) -> 'McapReader'`	Open an MCAP file; `path` may be omitted when the upload holds exactly one.

Recording

The canonical recording (LeRobot v3 tree plus meta/embodiflow/recording.json), or a time window of it.

Attribute	Meaning
`calibration: dict[str, Any]`	The conventions sidecar's calibration: `frames`, `stereoPairs` and `timeOffsets` as stored.
`cameras: list[Camera]`
`columns: list[Column]`
`conventions: dict[str, Any]`
`duration_s: float`	Length in seconds.
`facts: dict[str, Any]`
`fps: float`
`info: dict[str, Any]`
`kind: str \| None`
`meta: dict[str, Any]`
`num_frames: int`
`start_s: float`	Where this recording (or episode window) starts on the recording timeline, in seconds.
`stereo_pairs: list[dict[str, str]]`
`task: str \| None`
`total_frames: int`

Method	Meaning
`window(start_s: float, end_s: float) -> 'Recording'`	The part of this recording between `start_s` and `end_s` (seconds on this recording's own timeline).
`table(columns: Sequence[str] \| None=None) -> 'pa.Table'`	The frame table (one row per frame of this recording or window) as a `pyarrow.Table`.
`column(name: str) -> 'np.ndarray'`	One column as a numpy array of shape `(num_frames, *shape)`.

RecordingBuilder

Builds the canonical recording from a source format.

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).
`root: Path`	The output folder of the `recording` port (for :meth:`adopt_tree`).

Method	Meaning
`adopt_tree() -> None`	Say that the run wrote the complete canonical tree under `out.root` itself (a built-in standardizer with its own tree writer).
`fps(value: int) -> None`	Declare the frame rate of the recording: `30` (the platform grid; other rates are refused).
`task(text: str) -> None`	Set the task text of the recording (one episode, one task).
`camera(key: str, role: str, frames: Iterable[Any] \| Any=None, video: str \| Path \| None=None, intrinsics: Sequence[Sequence[float]] \| None=None, distortion: Sequence[float] \| None=None, model: str \| None=None) -> None`	Add a camera from RGB `frames` (an array `N x H x W x 3` uint8 or an iterator of `H x W x 3` frames) or a `video` file.
`column(name: str, values: Any, type: str='vector', unit: str='none', frame: str='none', source: str='recorded', names: Sequence[str] \| None=None) -> None`	Add a tabular column with one row per frame: `values` has shape `(N,)` or `(N, ...)`.
`timestamps(seconds: Any) -> None`	Record the source capture time of each row (seconds, non-decreasing); kept beside the recording.
`stereo_pair(left: str, right: str) -> None`	Name two added cameras as a stereo pair (keys or short names).
`dropped(what: str, reason: str) -> None`	Declare something of the source that did not make it into the recording, and why.

RecordingEditor

Edits the input recording: replace or drop a camera, replace a column, trim.

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).
`root: Path`	The output folder of the `recording` port (for :meth:`adopt_tree`).

Method	Meaning
`adopt_tree() -> None`	Say that the run wrote the complete edited tree under `out.root` itself (a built-in transform with its own video path, such as anonymize with its hardware encoder).
`unchanged() -> None`	Hand the input recording through untouched (zero copy).
`replace_camera(key: str, frames: Iterable[Any] \| Any=None, video: str \| Path \| None=None, anonymized: bool \| None=None) -> None`	Replace the pictures of camera `key` with `frames` (an iterator of `H x W x 3` uint8 images, one per input frame) or a `video`.
`drop_camera(key: str) -> None`	Remove camera `key` from the recording.
`replace_column(name: str, values: Any) -> None`	Replace the values of column `name`; `values` has the shape of the existing column (one row per input frame).
`trim(start_s: float, end_s: float) -> None`	Keep only `start_s`..`end_s` (seconds) of the recording.

SelectionWriter

Collects per-episode decisions (embodiflow-selection/v1).

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).

Method	Meaning
`keep(episode_id: str, split: str \| None=None, task: str \| None=None, reason: str \| None=None) -> None`	Keep the episode, optionally assigning a `split` (`train`/`val`/`test`) and replacing its `task` text.
`count(name: str, value: int) -> None`	Report a named count in the selection summary (`summary.counts`), e.g. `count("grouped_by_recording", 3)`.
`drop(episode_id: str, reason: str) -> None`	Drop the episode from the new version; `reason` is shown to the team.

SetupContext

What setup(ctx) sees: the params, the facts, a scratch directory and a logger.

Attribute	Meaning
`facts: Facts`
`log: logging.Logger`
`params: Any`
`run_id: str`	The workflow run id.
`step_id: str`	The id of the step in the workflow.
`tmp: Path`

TransientError

Raise it when a retry may succeed (network, GPU busy): the step exits 88 and is retried.

Vocabulary

Name	Allowed values
`categories`	`standardize`, `check`, `transform`, `enrich`, `segment`, `curate`, `export`
`levels`	`recording`, `episode`, `dataset`
`statuses`	`pass`, `warn`, `fail`, `error`, `not_evaluated`, `not_applicable`
`per`	`frame`, `recording`, `episode`, `file`
`sources`	`recorded`, `imported`, `estimated`, `computed`, `retargeted`
`dtypes`	`float32`, `float64`, `int8`, `int16`, `int32`, `int64`, `uint8`, `uint16`
`splits`	`train`, `val`, `test`
`semanticTypes`	`scalar`, `vector`, `label.bool`, `label.category`, `label.text`, `spans`, `embedding`, `pose.se3`, `keypoints3d.hand21`, `skeleton.hand21`, `skeleton.body22`, `keypoints2d`, `confidence`, `calibration.stereo`, `calibration.cam_imu`, `depth.video`, `mask.video`, `video.preview`, `ros2.mcap`, `action.joint`, `action.ee_pose`, `json`
`facts`	`kind`, `format`, `privacy_required`, `duration_s`, `cameras`, `stereo`, `imu`, `action`, `recorded_hand_pose`, `gripper_width`, `episodes`
`privacy`	`redacts`, `numbers_only`
`services`	`vlm`
`runParameters`	`raw`, `recording`, `version`, `episode`, `out`, `ctx`, `state`

This page is generated from the same SDK index the browser editor uses for completion, so it always matches the SDK you run. Everything is imported from embodiflow:

from embodiflow import Context, Feature, FeatureWriter, Recording, algorithm, param

Decorators

@algorithm.standardize

Declare a standardizer: it turns one raw recording into the canonical recording.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`description`	`str \| None`	no	`None`
`checks`	`list[Check] \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`
`services`	`list[str] \| None`	no	`None`

@algorithm.check

Declare a check: it judges data and never changes it.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`checks`	`list[Check]`	yes
`description`	`str \| None`	no	`None`
`level`	`str \| None`	no	`None`
`reads`	`str \| None`	no	`None`
`inputs`	`dict[str, Input] \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`
`services`	`list[str] \| None`	no	`None`

@algorithm.transform

Declare a transform: it returns a recording with some streams replaced or cut.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`description`	`str \| None`	no	`None`
`inputs`	`dict[str, Input] \| None`	no	`None`
`checks`	`list[Check] \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`
`services`	`list[str] \| None`	no	`None`
`privacy`	`str \| None`	no	`None`

@algorithm.enrich

Declare an enricher: it computes new information without changing the recording.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`features`	`list[Feature]`	yes
`description`	`str \| None`	no	`None`
`level`	`str \| None`	no	`None`
`inputs`	`dict[str, Input] \| None`	no	`None`
`checks`	`list[Check] \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`
`services`	`list[str] \| None`	no	`None`
`privacy`	`str \| None`	no	`None`

@algorithm.segment

Declare a segmenter: it proposes episodes, time ranges with a task text.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`description`	`str \| None`	no	`None`
`labels`	`dict[str, str] \| None`	no	`None`
`inputs`	`dict[str, Input] \| None`	no	`None`
`checks`	`list[Check] \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`
`services`	`list[str] \| None`	no	`None`
`segment_source`	`str \| None`	no	`None`

@algorithm.curate

Declare a curator: it decides per episode of a dataset version whether to keep it, its split and its task.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`description`	`str \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`

@algorithm.export

Declare an exporter: it turns a dataset version into downloadable files.

Argument	Type	Required	Default
`name`	`str`	yes
`title`	`str`	yes
`description`	`str \| None`	no	`None`
`requires`	`dict[str, Any] \| None`	no	`None`
`params`	`type \| None`	no	`None`
`resources`	`dict[str, Any] \| None`	no	`None`
`requirements`	`list[str] \| None`	no	`None`
`retries`	`int \| None`	no	`None`
`cache`	`bool \| None`	no	`None`

Functions

param(default, *, minimum=None, maximum=None, enum=None, title=None, description=None, widget=None)

Declare a field of the Params dataclass with its default and limits.

Classes

ArtifactWriter

Collects the files of an export (embodiflow-artifact/v1).

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).

Method	Meaning
`format(name: str) -> None`	Name the format of the files (`lerobot-v3`, `csv`, ...).
`path(rel: str) -> Path`	A path inside the artifact to write to (parent folders are created).
`add(src: str \| Path, rel: str) -> None`	Copy the file or folder `src` into the artifact at `rel`.
`summary(doc: dict[str, Any]) -> None`	Describe the export (counts, sizes); shown on the download page.

Camera

One camera of a recording.

Attribute	Meaning
`anonymized: bool`
`distortion: list[float] \| None`
`height: int`
`intrinsics: list[list[float]] \| None`
`key: str`
`model: str \| None`
`role: str`
`video_path: Path`	The camera's mp4 in the recording tree.
`width: int`

Method	Meaning
`valid() -> 'np.ndarray'`	A boolean per frame: `False` where the camera had a gap and the video holds a black placeholder.
`frames(start: int=0, stop: int \| None=None, step: int=1, gray: bool=False, size: tuple[int, int] \| None=None) -> Iterator[Frame]`	Decode frames `start..stop` every `step` (indices relative to this recording or episode window).
`frame(i: int) -> Frame`	Decode the single frame `i`.

Check

One check result id an algorithm may report through CheckWriter.result.

Attribute	Meaning
`description: str`
`id: str`
`title: str`

Method	Meaning
`to_dict() -> dict[str, Any]`	The declaration as the manifest input.

CheckWriter

Collects check results (embodiflow-checks/v1).

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).

Method	Meaning
`result(id: str, status: str, value: float \| None=None, threshold: float \| None=None, message: str='', evidence: Sequence[str \| Path] \| None=None, at: Mapping[str, Any] \| None=None) -> None`	Report one result.
`evidence_path(name: str) -> Path`	A path under `evidence/` to write a picture or file to; pass it to `result(evidence=[...])`.
`metrics(doc: dict[str, Any]) -> None`	Attach an `embodiflow-quality-metrics/v1` body (built-in checks only).

Column

One tabular column of a recording: name, dtype, shape, names and the conventions unit, frame, source.

Attribute	Meaning
`dtype: str`
`frame: str \| None`
`name: str`
`names: list[str] \| None`
`shape: list[int]`
`source: str \| None`
`type: str \| None`
`unit: str \| None`

Context

Everything a run knows about itself: params, facts, identifiers, scratch space, log, progress.

Attribute	Meaning
`facts: Facts`
`log: logging.Logger`
`params: P`
`platform_params: dict[str, Any]`	What the platform resolved for a standardize step (`rigProfiles`, `teamDataFormat`); empty otherwise.
`recording_id: str \| None`	The recording this step works on (`None` for a dataset-level step).
`run_id: str`	The workflow run id.
`step_id: str`	The id of the step in the workflow.
`team_slug: str \| None`	The storage slug of the team (names a team feature); `None` when the job does not state it.
`tmp: Path`

Method	Meaning
`progress(done: int, total: int, message: str='') -> None`	Report progress to the run view (progress bar and ETA); safe to call often.
`not_applicable(code: str, message: str) -> NoReturn`	End the step as not applicable (exit 85, grey in the run view, never a failure).
`input_unusable(code: str, message: str) -> NoReturn`	End the step because the data is broken (exit 87): the step fails and the recording is quarantined.

DatasetVersion

A frozen dataset version: id, features (declared names and types) and members.

Attribute	Meaning
`document: dict[str, Any]`
`features: dict[str, dict[str, Any]]`
`id: str`
`members: list[Member]`

Method	Meaning
`to_pandas() -> 'pd.DataFrame'`	One row per member: id, recording, start/end/duration, task, split, labels, quality status, session, operator.

Episode

An episode of a recording: index, start_s, end_s (recording timeline), task, labels, confidence.

Attribute	Meaning
`confidence: float \| None`
`duration_s: float`	Length in seconds.
`end_s: float`
`index: int`
`labels: dict[str, Any]`
`start_s: float`
`task: str \| None`

EpisodeWriter

Collects proposed episodes (embodiflow-episodes/v1).

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).

Method	Meaning
`episode(start_s: float, end_s: float, task: str, labels: Mapping[str, Any] \| None=None, confidence: float \| None=None) -> None`	Add one episode: `startS`/`endS` in seconds, the `task` text, optional `labels` and a `confidence` in 0..1.

Facts

Facts of the main input (contract 2.1) with attribute access: ctx.facts.kind, ctx.facts.stereo.

Attribute	Meaning
`action: bool \| None`
`cameras: int \| None`
`duration_s: float \| None`
`episodes: int \| None`
`format: str \| None`
`gripper_width: bool \| None`
`imu: bool \| None`
`kind: str \| None`
`privacy_required: bool \| None`
`recorded_hand_pose: bool \| None`
`stereo: bool \| None`

Method	Meaning
`get(name: str, default: Any=None) -> Any`	The value of a fact, or `default` when it is unknown.
`as_dict() -> dict[str, Any]`	All known facts as a plain dict.

Feature

One feature an enrich algorithm declares and later writes through FeatureWriter.

Attribute	Meaning
`describe: dict[str, Any] \| None`
`dtype: str \| None`
`frame: str \| None`
`name: str`
`names: list[str] \| None`
`per: str`
`shape: list[int] \| None`
`source: str \| None`
`type: str`
`unit: str \| None`

Method	Meaning
`to_dict() -> dict[str, Any]`	The declaration as the manifest input: only the arguments that were given.

FeatureWriter

Collects the features an enrich algorithm computes (embodiflow-features/v1).

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).

Method	Meaning
`per_frame(name: str, values: Any, confidence: str \| None=None) -> None`	Write one row per frame: `values` has shape `(num_frames, *shape)` (a 1-D array for a scalar).
`per_recording(name: str, value: Any) -> None`	Write one value for the whole recording (JSON, at most 1 MiB).
`per_episode(name: str, value: Any) -> None`	Write the value for the episode being processed (episode-level `enrich` only).
`file(name: str, path: str \| Path, describe: Mapping[str, Any] \| None=None) -> None`	Attach a file (such as a depth video) to a `per="file"` feature.

Features

A features input (features.json plus frames.parquet, episodes.parquet and files/).

Attribute	Meaning
`document: dict[str, Any]`
`names: list[str]`	The names of all features.
`num_frames: int`

Method	Meaning
`info(name: str) -> dict[str, Any]`	The `features.json` entry of `name` (type, per, dtype, shape, unit, frame, source, describe).
`of_type(semantic_type: str) -> list[str]`	Names of the features whose semantic type is `semantic_type`.
`per_frame(name: str) -> 'np.ndarray'`	The values of a per-frame feature, shape `(num_frames, *shape)`; rows outside an episode window are NaN/None.
`per_recording(name: str) -> Any`	The value of a per-recording feature.
`per_episode(name: str) -> dict[int, Any]`	`{episodeIndex: value}` of a per-episode feature.
`file(name: str) -> Path`	The path of a per-file feature (a depth video).

Frame

One decoded video frame: index (in the recording or episode window), time_s and image (RGB H x W x 3 uint8, or H x W when gray).

Attribute	Meaning
`image: 'np.ndarray'`
`index: int`
`time_s: float`

Input

An extra input of type features (a calibration, a hand pose).

Attribute	Meaning
`optional: bool`
`requires: list[str]`

Method	Meaning
`to_dict() -> dict[str, Any]`	The declaration as the manifest input.

InputUnavailableError

An input the algorithm asked for is not there (absent port, missing file or download).

ManifestDriftError

The code's interface differs from the frozen manifest of this version.

ManifestError

The algorithm interface is invalid (ALGORITHM_FILE_INVALID and its siblings).

McapMessage

A decoded ROS 2 message: topic, schema, log_time_s (seconds since the file's first message).

McapReader

An MCAP file read through its summary and only the chunks a query needs.

Attribute	Meaning
`duration_s: float`
`end_ns: int`
`start_ns: int`
`topics: list[McapTopic]`

Method	Meaning
`messages(topics: Sequence[str] \| None=None, start_s: float \| None=None, end_s: float \| None=None) -> Iterator[McapMessage]`	Decoded messages in log-time order; `topics` filters by name, `start_s`/`end_s` bound log time.

McapTopic

One topic of an MCAP file: name, message schema name, message count and average rate in Hz.

Attribute	Meaning
`count: int`
`name: str`
`rate: float`
`schema: str`

Member

One episode of a dataset version.

Attribute	Meaning
`duration_s: float`	Length of the episode in seconds.
`end_s: float`
`episode_id: str`
`features: dict[str, Any]`
`labels: dict[str, Any]`
`operator: str \| None`
`quality: dict[str, Any]`
`recording_id: str`
`root: str`
`session: str \| None`
`split: str \| None`
`start_s: float`
`task: str \| None`

Method	Meaning
`recording() -> Recording`	The episode's recording window, downloading its files on first use.

OutputError

A writer refused what the algorithm gave it.

RawFile

One uploaded file: path (relative to the upload root), size in bytes and local_path on disk.

Attribute	Meaning
`local_path: Path`
`path: str`
`size: int`

Method	Meaning
`open(mode: str='rb') -> IO[Any]`	Open the file (read-only; binary by default).

RawFiles

One recording's uploaded files, at their upload-relative paths.

Attribute	Meaning
`files: list[RawFile]`

Method	Meaning
`find(pattern: str) -> list[RawFile]`	The files whose path matches the glob `pattern` (`` within a folder, `*` across folders).
`read_json(path: str) -> Any`	Parse the JSON file at `path`.
`read_yaml(path: str) -> Any`	Parse the YAML file at `path`.
`mcap(path: str \| None=None) -> 'McapReader'`	Open an MCAP file; `path` may be omitted when the upload holds exactly one.

Recording

The canonical recording (LeRobot v3 tree plus meta/embodiflow/recording.json), or a time window of it.

Attribute	Meaning
`calibration: dict[str, Any]`	The conventions sidecar's calibration: `frames`, `stereoPairs` and `timeOffsets` as stored.
`cameras: list[Camera]`
`columns: list[Column]`
`conventions: dict[str, Any]`
`duration_s: float`	Length in seconds.
`facts: dict[str, Any]`
`fps: float`
`info: dict[str, Any]`
`kind: str \| None`
`meta: dict[str, Any]`
`num_frames: int`
`start_s: float`	Where this recording (or episode window) starts on the recording timeline, in seconds.
`stereo_pairs: list[dict[str, str]]`
`task: str \| None`
`total_frames: int`

Method	Meaning
`window(start_s: float, end_s: float) -> 'Recording'`	The part of this recording between `start_s` and `end_s` (seconds on this recording's own timeline).
`table(columns: Sequence[str] \| None=None) -> 'pa.Table'`	The frame table (one row per frame of this recording or window) as a `pyarrow.Table`.
`column(name: str) -> 'np.ndarray'`	One column as a numpy array of shape `(num_frames, *shape)`.

RecordingBuilder

Builds the canonical recording from a source format.

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).
`root: Path`	The output folder of the `recording` port (for :meth:`adopt_tree`).

Method	Meaning
`adopt_tree() -> None`	Say that the run wrote the complete canonical tree under `out.root` itself (a built-in standardizer with its own tree writer).
`fps(value: int) -> None`	Declare the frame rate of the recording: `30` (the platform grid; other rates are refused).
`task(text: str) -> None`	Set the task text of the recording (one episode, one task).
`camera(key: str, role: str, frames: Iterable[Any] \| Any=None, video: str \| Path \| None=None, intrinsics: Sequence[Sequence[float]] \| None=None, distortion: Sequence[float] \| None=None, model: str \| None=None) -> None`	Add a camera from RGB `frames` (an array `N x H x W x 3` uint8 or an iterator of `H x W x 3` frames) or a `video` file.
`column(name: str, values: Any, type: str='vector', unit: str='none', frame: str='none', source: str='recorded', names: Sequence[str] \| None=None) -> None`	Add a tabular column with one row per frame: `values` has shape `(N,)` or `(N, ...)`.
`timestamps(seconds: Any) -> None`	Record the source capture time of each row (seconds, non-decreasing); kept beside the recording.
`stereo_pair(left: str, right: str) -> None`	Name two added cameras as a stereo pair (keys or short names).
`dropped(what: str, reason: str) -> None`	Declare something of the source that did not make it into the recording, and why.

RecordingEditor

Edits the input recording: replace or drop a camera, replace a column, trim.

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).
`root: Path`	The output folder of the `recording` port (for :meth:`adopt_tree`).

Method	Meaning
`adopt_tree() -> None`	Say that the run wrote the complete edited tree under `out.root` itself (a built-in transform with its own video path, such as anonymize with its hardware encoder).
`unchanged() -> None`	Hand the input recording through untouched (zero copy).
`replace_camera(key: str, frames: Iterable[Any] \| Any=None, video: str \| Path \| None=None, anonymized: bool \| None=None) -> None`	Replace the pictures of camera `key` with `frames` (an iterator of `H x W x 3` uint8 images, one per input frame) or a `video`.
`drop_camera(key: str) -> None`	Remove camera `key` from the recording.
`replace_column(name: str, values: Any) -> None`	Replace the values of column `name`; `values` has the shape of the existing column (one row per input frame).
`trim(start_s: float, end_s: float) -> None`	Keep only `start_s`..`end_s` (seconds) of the recording.

SelectionWriter

Collects per-episode decisions (embodiflow-selection/v1).

Attribute	Meaning
`checks: 'CheckWriter'`	The `CheckWriter` for the checks this algorithm declared (`checks=[Check(...)]`).

Method	Meaning
`keep(episode_id: str, split: str \| None=None, task: str \| None=None, reason: str \| None=None) -> None`	Keep the episode, optionally assigning a `split` (`train`/`val`/`test`) and replacing its `task` text.
`count(name: str, value: int) -> None`	Report a named count in the selection summary (`summary.counts`), e.g. `count("grouped_by_recording", 3)`.
`drop(episode_id: str, reason: str) -> None`	Drop the episode from the new version; `reason` is shown to the team.

SetupContext

What setup(ctx) sees: the params, the facts, a scratch directory and a logger.

Attribute	Meaning
`facts: Facts`
`log: logging.Logger`
`params: Any`
`run_id: str`	The workflow run id.
`step_id: str`	The id of the step in the workflow.
`tmp: Path`

TransientError

Raise it when a retry may succeed (network, GPU busy): the step exits 88 and is retried.

Vocabulary

Name	Allowed values
`categories`	`standardize`, `check`, `transform`, `enrich`, `segment`, `curate`, `export`
`levels`	`recording`, `episode`, `dataset`
`statuses`	`pass`, `warn`, `fail`, `error`, `not_evaluated`, `not_applicable`
`per`	`frame`, `recording`, `episode`, `file`
`sources`	`recorded`, `imported`, `estimated`, `computed`, `retargeted`
`dtypes`	`float32`, `float64`, `int8`, `int16`, `int32`, `int64`, `uint8`, `uint16`
`splits`	`train`, `val`, `test`
`semanticTypes`	`scalar`, `vector`, `label.bool`, `label.category`, `label.text`, `spans`, `embedding`, `pose.se3`, `keypoints3d.hand21`, `skeleton.hand21`, `skeleton.body22`, `keypoints2d`, `confidence`, `calibration.stereo`, `calibration.cam_imu`, `depth.video`, `mask.video`, `video.preview`, `ros2.mcap`, `action.joint`, `action.ee_pose`, `json`
`facts`	`kind`, `format`, `privacy_required`, `duration_s`, `cameras`, `stereo`, `imu`, `action`, `recorded_hand_pose`, `gripper_width`, `episodes`
`privacy`	`redacts`, `numbers_only`
`services`	`vlm`
`runParameters`	`raw`, `recording`, `version`, `episode`, `out`, `ctx`, `state`

Attribute	Meaning
`action: bool \| None`
`cameras: int \| None`
`duration_s: float \| None`
`episodes: int \| None`
`format: str \| None`
`gripper_width: bool \| None`
`imu: bool \| None`
`kind: str \| None`
`privacy_required: bool \| None`
`recorded_hand_pose: bool \| None`
`stereo: bool \| None`

Attribute	Meaning
`describe: dict[str, Any] \| None`
`dtype: str \| None`
`frame: str \| None`
`name: str`
`names: list[str] \| None`
`per: str`
`shape: list[int] \| None`
`source: str \| None`
`type: str`
`unit: str \| None`

Decorators

@algorithm.standardize

@algorithm.check

@algorithm.transform

@algorithm.enrich

@algorithm.segment

@algorithm.curate

@algorithm.export

Functions

param(default, *, minimum=None, maximum=None, enum=None, title=None, description=None, widget=None)

Classes

ArtifactWriter

Camera

Check

CheckWriter

Column

Context

DatasetVersion

Episode

EpisodeWriter

Facts

Feature

FeatureWriter

Features

Frame

Input

InputUnavailableError

ManifestDriftError

ManifestError

McapMessage

McapReader

McapTopic

Member

OutputError

RawFile

RawFiles

Recording

RecordingBuilder

RecordingEditor

SelectionWriter

SetupContext

TransientError

Vocabulary

Next

Decorators

@algorithm.standardize

@algorithm.check

@algorithm.transform

@algorithm.enrich

@algorithm.segment

@algorithm.curate

@algorithm.export

Functions

param(default, *, minimum=None, maximum=None, enum=None, title=None, description=None, widget=None)

Classes

ArtifactWriter

Camera

Check

CheckWriter

Column

Context

DatasetVersion

Episode

EpisodeWriter

Facts

Feature

FeatureWriter

Features

Frame

Input

InputUnavailableError

ManifestDriftError

ManifestError

McapMessage

McapReader

McapTopic

Member

OutputError

RawFile

RawFiles