Enrich

Enrich is the category you'll probably use most. It adds information to a recording: hand keypoints, a head pose, a scene label, an embedding, a success label per episode. The recording stays exactly as it was, and the platform merges your features into the saved result, where the viewer, curation and export can use them.

An enrich algorithm runs once per recording by default, or once per episode with level="episode". It reads recording (plus episodes at episode level) and any extra features inputs you declare, and writes features through out, a FeatureWriter.

Declare the features

List every feature you write in the decorator: Feature(name, type, per, dtype=…, shape=…, unit=…, frame=…, names=…, source=…, describe=…). The type is a semantic type from the registry in the interface reference, and it's what the workflow editor checks connections against. The per value decides how you write it:

Example: motion energy per frame

The algorithm from Your first algorithm. The file is packages/sdk-python/examples/v1/enrich-motion-energy/main.py.

"""Enrich: the mean absolute difference between consecutive frames of the first camera, per frame.

In:  the canonical recording.
Out: one new per-frame feature "motion_energy"; the recording itself is not changed.
Why: idle stretches (the arm waits, the person stands still) show as low motion, so a later step can
     segment or drop them. A feature is how an algorithm adds knowledge without rewriting the recording.
"""

from dataclasses import dataclass

import numpy as np

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


@dataclass
class Params:
    step: int = param(1, minimum=1, title="Frame step")


@algorithm.enrich(
    name="motion-energy",
    title="Motion energy",
    description="How much the first camera's picture changes from one frame to the next.",
    features=[Feature("motion_energy", type="scalar", per="frame", dtype="float32", unit="none")],
    params=Params,
)
def run(recording: Recording, out: FeatureWriter, ctx: Context[Params]) -> None:
    cam = recording.cameras[0]
    energy = np.zeros((recording.num_frames, 1), np.float32)
    prev = None
    for frame in cam.frames(step=ctx.params.step, gray=True):
        current = frame.image.astype(np.float32)
        if prev is not None:
            energy[frame.index, 0] = np.abs(current - prev).mean()
        prev = current
    out.per_frame("motion_energy", energy)

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

Example: one value per episode

With level="episode" the function runs once per episode. Its recording argument is the window of that episode, so frame counts and times start at zero for each one. The file is packages/sdk-python/examples/v1/enrich-episode-length/main.py.

"""Enrich (episode level): the length of each episode and whether it is long enough to train on.

In:  the canonical recording and one episode of it, the step runs once per episode.
Out: two per-episode features: "length_s" and "is_long".
Why: shows the episode level: the step gets the episode window, not the whole recording.
"""

from dataclasses import dataclass

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


@dataclass
class Params:
    long_s: float = param(1.0, minimum=0.0, title="Seconds an episode needs to count as long")


@algorithm.enrich(
    name="episode-length",
    title="Episode length",
    level="episode",
    features=[
        Feature("length_s", type="scalar", per="episode", dtype="float64", unit="s"),
        Feature("is_long", type="label.bool", per="episode"),
    ],
    params=Params,
)
def run(recording: Recording, episode: Episode, out: FeatureWriter, ctx: Context[Params]) -> None:
    length = episode.duration_s  # seconds, on the recording's timeline
    out.per_episode("length_s", length)
    out.per_episode("is_long", length >= ctx.params.long_s)

Its test is packages/sdk-python/examples/v1/enrich-episode-length/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 enrich --level episode, 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

Names. You write a short name such as motion_energy, and the platform stores it under a name that belongs to your team. A name that collides with a built-in feature is refused with FEATURE_NAMESPACE_VIOLATION.

Geometry. Geometric types (pose.se3, keypoints3d.hand21, keypoints2d, depth) need unit and frame, except per file. A per-frame estimate should also write a confidence column. Mark a missing value with NaN or -1, and never interpolate silently.

Extra inputs. Declare them with inputs={"hands": Input(requires=["keypoints3d.hand21"])}. The workflow editor only connects an output that declares the required type. Add optional=True for an input that may be absent; the argument is then None.

Retargeted actions. A feature of type action.joint or action.ee_pose named action may be written with source="retargeted" and describe={"targetRig": …}. This is the only way a human recording carries an action, and the console labels it as retargeted.

Models. Load a model once by defining setup(ctx), and take its return value in run as state. Don't keep per-item state in globals.

Episode steps. An episode-level step runs once per episode of its episodes input, at most 4 at a time and for at most 100 episodes. One failed episode doesn't fail the others: the step succeeds when at least one episode succeeded, and the run is marked partially succeeded with the failed episodes listed. With no episodes the step is skipped, which counts like not applicable.

Merging. You never merge features yourself. When features are connected to the Output node, the workflow adds a step that writes them into the saved recording; the run view shows it as Merge features.

Next

• Segment algorithms
• Interface reference