The platform reads MCAP, LeRobot and video uploads out of the box. If your recordings come as an HDF5 file, a rosbag or a folder of videos and CSV files, a standardize algorithm is the bridge: it turns one upload in your format into the canonical recording. From then on, checks, anonymization, poses and exports all work on your data as on any other.
A standardize algorithm runs once per recording. It reads raw, the uploaded files, and builds the recording through out, a RecordingBuilder:
| Method | Use it to |
|---|---|
out.fps(30) | Set the frame rate of the recording. |
out.task(text) | Set the task text. |
out.camera(key, role, frames=… or video=path, intrinsics=…, distortion=…, model=…) | Add a camera from decoded frames or from a video file. |
out.column(name, values, type, unit, frame, source="recorded") | Add a table column, one row per frame. |
out.timestamps(seconds) | Give the time of every frame when it isn't uniform. |
out.stereo_pair(left, right) | Mark two cameras as a stereo pair. |
out.dropped(what, reason) | Say what you left out and why. |
When your function returns, the platform checks the result against the canonical rules R1 to R7. A recording that breaks one is refused with CANONICAL_OUTPUT_INVALID, and the message names the rule, so you know exactly what to fix.
Example: one video and one CSV
This is the starter the editor gives you. It reads one .mp4 and one .csv with a header row and one numeric row per video frame. The file is packages/sdk-python/examples/v1/standardize-video-csv/main.py.
"""Standardize: one MP4 and a per-frame CSV become the canonical recording.
In: the uploaded files (raw): one .mp4 from the camera and one .csv with a header row of joint names and
one row of joint values per video frame, recorded at 30 Hz.
Out: the canonical recording: one camera, one "observation.state" column, a task text.
Why: every later step (checks, curation, export, training) reads the canonical recording, never the raw files.
Anything that does not fit (no video, two CSVs) is "not applicable", never guessed at.
"""
import csv
from dataclasses import dataclass
import numpy as np
from embodiflow import Context, RawFiles, RecordingBuilder, algorithm, param
@dataclass
class Params:
task: str = param("move the arm", title="Task text", description="Recorded as the task of the episode")
@algorithm.standardize(
name="video-csv",
title="Video and CSV episode",
description="Reads one .mp4 and one .csv with a header row and one numeric row per video frame, at 30 Hz.",
params=Params,
)
def run(raw: RawFiles, out: RecordingBuilder, ctx: Context[Params]) -> None:
videos = raw.find("**/*.mp4")
tables = raw.find("**/*.csv")
if len(videos) != 1 or len(tables) != 1:
ctx.not_applicable("NOT_ONE_VIDEO_AND_CSV", f"expected one .mp4 and one .csv, found {len(videos)} and {len(tables)}")
with open(tables[0].local_path, newline="", encoding="utf-8") as handle:
reader = csv.reader(handle)
names = next(reader)
states = np.asarray([[float(cell) for cell in row] for row in reader], dtype=np.float32)
out.fps(30)
out.task(ctx.params.task)
out.camera("observation.images.top", "top", video=videos[0].local_path)
out.column("observation.state", states, "vector", names=names)
ctx.progress(1, 1, "converted")Its test is packages/sdk-python/examples/v1/standardize-video-csv/test_main.py: it runs this code through the real runner on a small fixture and checks the values it writes. To start from it, run embodiflow algorithm init my-algorithm --category standardize, which writes main.py and test_main.py; then run embodiflow algorithm test my-algorithm (needs Python with pytest and the SDK).
Example: one HDF5 file
This one reads /frames and /joint_positions from a single .h5 file. It needs the h5py package, so it lists it in requirements. The file is packages/sdk-python/examples/v1/standardize-hdf5/main.py.
"""Standardize: one HDF5 file (frames, joint positions) becomes the canonical recording.
In: the uploaded files (raw): one .h5 file with /frames (N,H,W,3 uint8) and /joint_positions (N,7).
Out: the canonical recording: one camera, one "observation.state" column in radians, a task text.
Why: teleoperation tools often dump everything into one HDF5 file; this shows how to read one with h5py
and hand the arrays to the builder. Anything but exactly one .h5 file is "not applicable".
"""
from dataclasses import dataclass
import h5py
import numpy as np
from embodiflow import Context, RawFiles, RecordingBuilder, algorithm, param
@dataclass
class Params:
task: str = param("move the arm", title="Task text", description="Used when the file carries no task attribute")
@algorithm.standardize(
name="hdf5-arm",
title="HDF5 arm episode",
description="Reads /frames (N,H,W,3 uint8) and /joint_positions (N,7) from one .h5 file.",
params=Params,
requirements=["h5py>=3.10"],
)
def run(raw: RawFiles, out: RecordingBuilder, ctx: Context[Params]) -> None:
files = raw.find("**/*.h5")
if len(files) != 1:
ctx.not_applicable("NOT_ONE_HDF5", f"expected one .h5 file, found {len(files)}")
with h5py.File(files[0].local_path, "r") as h5:
frames = np.asarray(h5["frames"])
joints = np.asarray(h5["joint_positions"], dtype=np.float32)
task = h5.attrs.get("task", ctx.params.task)
out.fps(30)
out.task(str(task))
out.camera("observation.images.top", "top", frames=frames)
out.column("observation.state", joints, "action.joint", unit="rad", frame="base", names=[f"j{i}" for i in range(joints.shape[1])])
ctx.progress(1, 1, "converted")Its test is packages/sdk-python/examples/v1/standardize-hdf5/test_main.py: it runs this code through the real runner on a small fixture and checks the values it writes. To start from it, run embodiflow algorithm init my-algorithm --category standardize --example standardize-hdf5, which writes main.py and test_main.py; then run embodiflow algorithm test my-algorithm (needs Python with pytest and the SDK).
Good to know
raw.find("**/*.h5") lists the uploaded files that match a pattern, and each RawFile.local_path is a path you can open directly.
Say clearly when an upload isn't for you. ctx.not_applicable(code, message) ends the step as not applicable, shown grey and never as a failure. ctx.input_unusable(code, message) is for data that is broken: the step fails and the recording is quarantined.
A step that reads raw only knows what the upload told the platform. For a team format that's kind, format and privacy_required; everything else is unknown rather than guessed.
In a workflow, a standardize step can say uses: auto. The platform then picks the standardizer for the recording's data format and records that choice in the run. The auto standardizer takes a task text parameter, so uses: auto accepts with: { task: … }; your own standardizer can take one too, as both examples do.
Next
- Check algorithms
- Image algorithms, for a standardizer that isn't Python
- Use your algorithm in a workflow