Policy Backends

Overview

Four backends connect to an external VLA policy server you start yourself; EVA ships only the client half. Two need no server: mock emits synthetic motion and replay plays back a recorded trajectory — both for pipeline / 3D-view smoke tests.

The transport is implied by the backend; you do not configure it.

Prerequisites

For server-backed backends: the OpenPI / starVLA / GR00T server is already running on a reachable host, with its checkpoint loaded, and you know its IP and port. Start from a deploy preset under configs/01_deploy/<robot>/; shared defaults live in configs/00_base/defaults.py. mock and replay need neither.

The policy config block

Every backend reads the same four top-level keys. Per-backend tuning goes in backend_options.

Minimal block:

policy = dict(
    type='openpi',             # pick a backend: openpi / openpi_rtc / starvla / gr00t / mock / replay
    host='<policy server IP>',  # ← your policy server's address
    port=<port>,              # ← the port your server listens on
)

openpi and openpi_rtc

Both connect to an OpenPI-compatible server. Use openpi_rtc for smoother motion at higher control rates.

openpi

Stateless: each observation goes out, an action chunk comes back. Needs only type, host, port.

openpi_rtc — real-time chunking

Feeds the previous chunk back to the policy as a hint so consecutive chunks align. The one knob is latency_k: how many steps the feedback is shifted forward to compensate round-trip delay. Higher values absorb more lag but track the policy less tightly. Start at 4.

policy = dict(
    type='openpi_rtc',
    host='<policy server IP>',        # ← your policy server's address
    port=<port>,                    # ← the port your server listens on
    backend_options=dict(latency_k=4),  # start at 4, raise if motion lags
)

starvla

Connects to a starVLA server. starVLA accepts a single camera image; camera_key selects which (defaults to the first). unnorm_key picks the dataset the server uses to de-normalize its output — its value comes from the trained model; ask the server owner. Both keys are optional.

policy = dict(
    type='starvla',
    host='<policy server IP>',  # ← your policy server's address
    port=<port>,              # ← the port your server listens on
    backend_options=dict(
        camera_key='<camera name>',  # ← which EVA camera to send (e.g. cam_high)
        unnorm_key='<dataset key>',  # ← from your model; ask the server owner
    ),
)

gr00t

Connects to an Isaac GR00T server. GR00T tags every payload with a modality key identifying the field (camera view, joint positions, task text). The names vary per model, so you must map EVA's camera and state names to the keys the server expects. Get the values from the server owner.

Defaults cover the standard *.qpos naming; usually only video_keys needs to be set:

policy = dict(
    type='gr00t',
    host='<policy server IP>',  # ← your policy server's address
    port=<port>,              # ← the port your server listens on
    backend_options=dict(
        video_keys=dict(
            cam_high='<GR00T video key>',        # ← map your camera to the server's key
            cam_left_wrist='<GR00T video key>',  # ← one entry per camera
        ),
        # state_key / language_key / action_keys keep their defaults unless your server differs
    ),
)

mock and replay (no server)

Neither needs a policy server.

mock

Returns smooth synthetic motion that ignores observations. For pipeline / 3D-view smoke tests, not control. Optional chunk_size (default 50).

replay

replay requires a dataset transport: pair it with transport = dict(type='dataset', ...) or EVA refuses to start. See Transport for dataset paths and episode selection.

transport = dict(
    type='dataset',
    dataset_dir='<path to recorded dataset>',  # ← folder of a LeRobot v2.1 recording
    episode_id=0,                          # which recorded run to play (0 = first)
)
policy = dict(
    type='replay',
    backend_options=dict(chunk_size=50),  # steps played per chunk; default is fine
)

Run and verify

Launch the console against the edited config:

eva --config configs/01_deploy/<robot>/your_preset.py --web-port 8080

Open the DEBUG tab:

• Connected: server details render with no error banner; triggering inference returns a chunk and the 3D view animates.
• Connection failed: the console reports a policy error instead of crashing; reconnect from DEBUG after fixing host/port or the server.
• Timeout (gr00t): raise timeout_ms or confirm reachability.

How to add a new policy backend

A backend is a subclass of PolicyClient (src/policy_client/base.py) registered into POLICY_REGISTRY. The package __init__ auto-imports every sibling module under src/policy_client/, so dropping a file there is enough to make the backend selectable.

1. Subclass PolicyClient and implement infer, reset, and the metadata property.
2. Decorate with @POLICY_REGISTRY.register_client("<your_name>") and provide a from_config(cls, config, ctx) classmethod — the registry calls it with the resolved policy config and a PolicyBuildContext (carries action_dim, chunk_size, etc.).
3. Save the file as src/policy_client/<your_name>.py. The package walker in src/policy_client/__init__.py imports it on launch, triggering the registration side effect — no extra wiring.
4. Select it from any config: policy = dict(type='<your_name>', host=..., port=..., backend_options=dict(...)).

# src/policy_client/mybackend.py  ← new file; auto-imported by the package
from __future__ import annotations

import numpy as np

from core.config import ConfigDict
from core.registry import POLICY_REGISTRY
from policy_client.base import PolicyBuildContext, PolicyClient


@POLICY_REGISTRY.register_client("mybackend")   # ← the string used in policy.type
class MyBackend(PolicyClient):
    def __init__(self, action_dim: int, chunk_size: int = 50) -> None:
        self._action_dim = action_dim
        self._chunk_size = chunk_size
        # ← open the connection to your model server here

    @classmethod
    def from_config(cls, config: ConfigDict, ctx: PolicyBuildContext) -> MyBackend:
        opts = config.backend_options
        return cls(action_dim=ctx.action_dim, chunk_size=opts.get("chunk_size", 50))

    def infer(self, observation: dict, prev_action: np.ndarray | None = None,
              rtc_params: dict | None = None) -> dict:
        # ← send observation to your server, receive an [T, action_dim] chunk
        actions = np.zeros((self._chunk_size, self._action_dim), dtype=np.float32)
        return {"actions": actions}

    def reset(self) -> None:
        # ← clear any per-episode state
        ...

    @property
    def metadata(self) -> dict:
        return {"server_name": "mybackend", "chunk_size": self._chunk_size}

The contract is in src/policy_client/base.py: infer returns {"actions": ndarray[T, action_dim] float32}; reset clears per-episode state; metadata reports server_name / chunk_size / action_mode. For a no-server skeleton mirror RandomPolicyClient in src/policy_client/base.py; for a WebSocket-backed real server, see src/policy_client/openpi.py.