Curate

Curation decides what a training run sees. A curate algorithm reads a dataset version and makes a decision per episode: drop the short or poor ones, balance the tasks, hold some out for validation. The platform turns those decisions into a new dataset version, and the source version is never touched.

A curate algorithm runs once per dataset version. It reads version and writes selection through out, a SelectionWriter:

version.members lists the episodes with their task, split, labels, quality, session, operator, duration and per-episode features, and version.to_pandas() gives you one row per member if you prefer a table. An episode you make no decision about is kept as it is.

Example: drop short episodes

This algorithm drops episodes shorter than a minimum, keeps the rest and gives every fifth kept one to validation. The file is packages/sdk-python/examples/v1/curate-min-duration/main.py.

"""Curate: drop episodes shorter than a minimum, keep the rest and give every 5th one to validation.

In:  the dataset version: its episodes, with their length and task.
Out: a decision for every episode: keep (train or val) or drop with a reason.
Why: curation decides what a training run sees; dropping short episodes and holding some out for
     validation are the two decisions almost every dataset needs.
"""

from dataclasses import dataclass

from embodiflow import DatasetVersion, SelectionWriter, algorithm, param


@dataclass
class Params:
    min_duration_s: float = param(2.0, minimum=0.0, title="Shortest episode to keep (s)")
    val_every: int = param(5, minimum=0, title="Every n-th kept episode goes to validation (0 = none)")


@algorithm.curate(name="min-duration", title="Drop short episodes", params=Params)
def run(version: DatasetVersion, out: SelectionWriter, ctx) -> None:
    kept = 0
    for member in version.members:
        if member.duration_s < ctx.params.min_duration_s:
            out.drop(member.episode_id, f"shorter than {ctx.params.min_duration_s} s ({member.duration_s:.2f} s)")
            continue
        kept += 1
        validation = ctx.params.val_every and kept % ctx.params.val_every == 0
        out.keep(member.episode_id, split="val" if validation else "train")

Its test is packages/sdk-python/examples/v1/curate-min-duration/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 curate, which writes main.py and test_main.py; then run embodiflow algorithm test my-algorithm (needs Python with pytest and the SDK).

Try it on a dataset version

In the editor's Test run, choose a Dataset version instead of a recording. The test reads the first 3 episodes by default, and up to 50, then shows each episode's decision and reason with the kept and dropped counts. As with any test run, nothing is saved.

Good to know

Decide each episode at most once and give a reason for every drop, or the result is refused with SELECTION_INVALID. Splits work as you'd expect, but setting a task text is refused with CURATE_TASK_CHANGE_UNSUPPORTED: a version keeps each episode as it was frozen.

If every episode is dropped, no version is created (CURATE_NOTHING_KEPT). If the kept episodes already form a version, that version is returned instead of a duplicate. The platform recomputes the normalization statistics of the new version itself, and the new version links to its source and to the run. A redacted version can't be curated (CURATE_SOURCE_VERSION_UNUSABLE).

Not every member field is filled in yet. session, operator and the per-episode features are empty for now, and a member's quality is the worst stored finding of its recording, or not evaluated when there is none. Don't base decisions on them yet.

A curate workflow can have several curate steps. An episode is kept only if every step keeps it, at most one step may set splits and at most one may set task texts; otherwise the workflow is refused with CURATE_SELECTION_CONFLICT.

Next

• Export algorithms
• Use your algorithm in a workflow