Sometimes the recording itself needs to change before anyone trains on it: the operator reaching for the keyboard at the start, a camera you never use, a stream at the wrong rate. A transform algorithm returns a new recording with just those parts replaced or cut. Streams you don't touch are carried over by reference, never copied, so a small change stays cheap.
A transform runs once per recording. It reads recording and writes recording through out, a RecordingEditor:
| Method | Use it to |
|---|---|
out.unchanged() | Pass the input through, with no copy, when there is nothing to do. |
out.replace_camera(key, frames=… or video=path, anonymized=None) | Replace one camera. |
out.drop_camera(key) | Remove one camera. |
out.replace_column(name, values) | Replace one column. |
out.trim(start_s, end_s) | Keep only a time range. |
Each transform adds itself to the lineage of recording.json and keeps every earlier entry, so you can always see what happened to a recording. A transform may also write checks through out.checks.
Example: trim the ends
This transform cuts a number of seconds off the start and the end. When both are zero it passes the recording through untouched, and it refuses a cut that would leave nothing. The file is packages/sdk-python/examples/v1/transform-trim/main.py.
"""Transform: cut seconds off the start and the end of the recording.
In: the canonical recording.
Out: a recording with the same cameras and columns, without the first and last seconds. Lineage records the step.
Why: the first seconds of a teleoperation recording often hold the operator reaching for the keyboard.
With nothing to cut the recording is handed through untouched (zero copy).
"""
from dataclasses import dataclass
from embodiflow import Context, Recording, RecordingEditor, algorithm, param
@dataclass
class Params:
head_s: float = param(0.0, minimum=0.0, title="Seconds to cut at the start")
tail_s: float = param(0.0, minimum=0.0, title="Seconds to cut at the end")
@algorithm.transform(name="trim-ends", title="Trim the ends", params=Params)
def run(recording: Recording, out: RecordingEditor, ctx: Context[Params]) -> None:
head, tail = ctx.params.head_s, ctx.params.tail_s
if head == 0 and tail == 0:
out.unchanged()
return
if head + tail >= recording.duration_s:
ctx.input_unusable("TRIM_EMPTIES_RECORDING", f"cutting {head + tail:.2f} s leaves nothing of {recording.duration_s:.2f} s")
out.trim(head, recording.duration_s - tail)Its test is packages/sdk-python/examples/v1/transform-trim/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 transform, 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
ctx.input_unusable(code, message) ends the step as failed and quarantines the recording. Use it for data that can't be fixed, as the example does when the cut would leave nothing.
Anonymization is the one change a team transform can't make count. Marking a camera as anonymized belongs to transforms that declare privacy: redacts, and only built-in algorithms may do that. So a team transform never makes a recording count as anonymized.
In a processing workflow, every step that reads the recording after standardizing reads the output of the built-in anonymize step. Place your transform after anonymize; placed before it, the workflow is refused with WORKFLOW_PRIVACY_BARRIER.