A long human recording usually holds many tasks, and training wants them one by one. A segment algorithm proposes where each episode starts and ends, by fixed windows, by pauses in motion or with a model. The proposals are suggestions: they show up for review and never replace episodes a person made.
A segment algorithm runs once per recording. It reads recording and writes episodes through out, an EpisodeWriter. Its one method, out.episode(start_s, end_s, task, labels=None, confidence=None), adds one proposal. Times are seconds from the start of the recording. If your episodes carry labels, declare the keys in the decorator with labels={key: semantic type}.
Example: fixed windows
The file is packages/sdk-python/examples/v1/segment-fixed-windows/main.py.
"""Segment: cut the recording into windows of a fixed length.
In: the canonical recording.
Out: episodes: consecutive windows with a task text and a "window" label; the last one may be shorter.
Why: a long recording of continuous work (a human doing chores, one long teleoperation session) must become
episodes before it can be curated or trained on. Replace the loop with your own idea of a boundary.
"""
from dataclasses import dataclass
from embodiflow import EpisodeWriter, Recording, algorithm, param
@dataclass
class Params:
window_s: float = param(10.0, minimum=0.5, title="Window length (s)")
task: str = param("do the task", title="Task text of every window")
@algorithm.segment(name="fixed-windows", title="Fixed windows", labels={"window": "scalar"}, params=Params)
def run(recording: Recording, out: EpisodeWriter, ctx) -> None:
start, index = 0.0, 0
while start < recording.duration_s - 1e-6:
end = min(start + ctx.params.window_s, recording.duration_s)
out.episode(start, end, ctx.params.task, labels={"window": index}, confidence=1.0)
start, index = end, index + 1Its test is packages/sdk-python/examples/v1/segment-fixed-windows/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 segment, 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
Ranges must be non-empty, inside the recording and must not overlap; otherwise the result is refused with EPISODES_INVALID. A label key you didn't declare is refused too.
The position of an episode in your list is its episode index, which episode-level steps use to refer to it.
Proposals are saved as AI-proposed and carry the "AI-proposed, not reviewed" badge until someone reviews them.