XPolicyLab Integration

Run an XPolicyLab deployment loop on an SE3 station through the relay.

from se3labs.sdk.bridge.xpolicylab import SE3Env, evaluate

evaluate(tasks=...) accepts a sequence of concrete task specs and supplies a SE3Env to the deployment function for each entry, in order. The deployment uses get_obs(), take_action(), and is_episode_end() while the SDK maintains the station connection. The caller owns the model client.

See the XPolicyLab tutorial for installation and command-line usage.

SE3Env

SE3Env(station: Station_Spec, task: Task_Spec, frequency: float)

One bimanual YAM rollout; used by an unchanged XPolicyLab deploy.py.

Joint actions are absolute radians; grippers are open fractions (0 closed, 1 open). take_action waits one action period on the station clock after submission. This is pacing, not an acknowledgement of physical execution.

Attributes

station

task

frequency

Methods

is_episode_end() -> bool

get_obs() -> dict[str, Any]

take_action(action: dict[str, Any]) -> None

exchange(observation: Station_Observation) -> Station_ActionChunk | None

Called by the SDK stream thread; never waits on model inference.

end(*, completed: bool = False) -> None

evaluate

evaluate(
    deploy: Callable[[SE3Env, Any], None],
    model_client: Any,
    *,
    station_id: str,
    address: str,
    tasks: Sequence[Task_Spec],
    start_pose: Station_JointPose,
    frequency: float,
) -> eval_client.SessionReport

Run an XPolicyLab eval_one_episode function through the SE3 relay.

The caller owns model_client (normally a WsModelClient). Each deployment invocation owns model reset, just as it does in XPolicyLab. A deployment error cancels the station stream and propagates to the caller. Each task in tasks defines one episode, in order, with its own instruction, layout, scoring, and time limit.