Call the Python API

Before you start, complete the station setup.

Use this path when you already have a model you can call from Python. Wrap it in a Policy class and pass a JobRequest to se3labs.sdk.eval_client.serve_policy(). Your model runs in your Python process; the SDK exchanges observations and actions with the station.

Define the task and starting pose

Save this as my_eval.py. This example asks a bimanual YAM station to hold its starting pose for ten seconds. Set the task instructions, scoring question, and starting joint angles to match your own evaluation. Confirm the starting pose with the operator before running it.

my_eval.py

import numpy as np

from se3labs.interface.eval.arm import MotorizedArm_JointPose
from se3labs.interface.eval.station import Station_JointPose
from se3labs.interface.eval.task import (
    Task_ScoreItem,
    Task_ScoreKind,
    Task_ScoreSheet,
    Task_Spec,
)

TASK = Task_Spec(
    name="hold_position",
    description="Hold both arms at their starting pose for ten seconds.",
    reset_instruction="Clear the workspace and leave both grippers empty.",
    instruction="Hold both arms still.",
    max_rollout_time=10.0,
    scoring=Task_ScoreSheet(
        success=Task_ScoreItem(
            "still", "Did both arms remain still?", Task_ScoreKind.Binary,
        ),
    ),
)

TASKS = (TASK,) * 3  # Three episodes, in execution order.

# Six joint angles in radians and one gripper open fraction per YAM arm.
START = Station_JointPose({
    name: MotorizedArm_JointPose(
        np.zeros(6, dtype=np.float32),
        np.ones(1, dtype=np.float32),
    )
    for name in ("left", "right")
})

description and reset_instruction guide the operator. instruction goes to your policy. The operator answers the scoring question after each episode. TASKS may be a list or tuple; each entry defines one episode with its own instruction, layout, scoring, and time limit. Use the RoboDojo Real task registry to sample predefined tasks. The XPolicyLab integration uses this same file.

Wrap your model and run it

Save this as evaluate.py beside my_eval.py. HoldPosition is a runnable stand-in for your model: its callable returns joint and gripper target arrays for each arm. Replace it with your model and adapt the call in MyPolicy.infer() to your model's inputs and outputs.

evaluate.py

import numpy as np

from se3labs.interface.eval.arm import MotorizedArm_ActionChunk
from se3labs.interface.eval.job import JobRequest
from se3labs.interface.eval.policy import Policy
from se3labs.interface.eval.station import Station_ActionChunk
from se3labs.sdk import eval_client

from my_eval import START, TASKS


class HoldPosition:
    def __call__(self, observation, instruction):
        return {
            name: (
                np.repeat(arm.joint_angle_q[None, :], 4, axis=0),
                np.repeat(arm.gripper_pos[None, :], 4, axis=0),
            )
            for name, arm in observation.arms.items()
        }


class MyPolicy(Policy):
    def __init__(self, model):
        self.model = model

    def initialize(self, station, task):
        self.instruction = task.instruction
        return START

    def infer(self, observation):
        actions = self.model(observation, self.instruction)
        return Station_ActionChunk(
            timestamp=max(arm.timestamp for arm in observation.arms.values()),
            rate=10.0,
            arms={
                name: MotorizedArm_ActionChunk(
                    np.asarray(joints, dtype=np.float32),
                    np.asarray(gripper, dtype=np.float32),
                )
                for name, (joints, gripper) in actions.items()
            },
        )


if __name__ == "__main__":
    request = JobRequest(tasks=TASKS, policy=MyPolicy(HoldPosition()))
    report = eval_client.serve_policy(
        "example-station-id", "example-station-addr", request,
    )
    print(report.state, report.close_reason)
    print(report.results)
    print("Policy errors:", report.errors)

Update the station ID and relay address in evaluate.py, then run:

python evaluate.py

The SDK calls initialize() before each episode, infer() on observations, and reset() after each episode. This example has no model state to clear; override reset() when your model keeps history or cached actions.

The wrapper expects absolute joint angles in radians and gripper open fractions (0 closed, 1 open). Each arm returns arrays shaped (horizon, n_arm_dof) and (horizon, n_gripper_dof), with the same horizon for every arm. Here, four target rows are spaced at 10 Hz. Set the rate to match your model, and use the observation's station timestamp for each chunk. Camera frames are available in observation.cameras for your model's preprocessing.

Return None from infer() to continue the current chunk, or raise EndEpisode to finish an episode before its time budget expires. See the policy contract for the full lifecycle and action format.

Continue with reading the evaluation results.