Robots

uv add 'hud-python[robot]'

​

Overview

• RobotBridge - the one class you implement around your sim: reset / step / get_observation. The framework owns the WebSocket serve loop and the single-agent connection.
• RobotEndpoint - wraps the bridge - the environment server’s handle for the sim (even if the sim is running in another process)

• RobotAgent - the harness: connects to the env and bridge, owns adapter and model, drives model until env terminates.
• Model - the actual stateless checkpoint of the model (includes pre-/post-processing)
• Adapter - translates the env’s observation space to the model’s, and the model’s action space to the env’s

​

Environment side

from hud.environment.robot import RobotBridge

class MySimBridge(RobotBridge):
    async def reset(self, task_id: str, seed: int = 0) -> str:
        ...                              # build the episode
        await self._send_observation()   # push the first frame
        return self.task_description     # becomes the task prompt

    def step(self, action) -> None:
        ...  # advance one tick; set success / terminated

    def get_observation(self):
        return {"agentview_image": frame, "state": vec}, self.terminated

• reset starts a fresh episode for a task and returns its prompt (the text the agent is given).
• step applies one action and advances the sim a tick, setting success / terminated as the episode plays out.
• get_observation returns a strctured dict of the current observation plus whether the episode is done.

def get_observation(self):
    data = {
        "observation/image":       rgb,          # [256, 256, 3] uint8, RGB, HWC
        "observation/wrist_image": wrist_rgb,    # [256, 256, 3] uint8, RGB, HWC
        "observation/state": np.concatenate([    # [8] float32, in contract order
            eef_pos,         # xyz                 (3,)
            eef_axis_angle,  # orientation         (3,)
            gripper_qpos,    # gripper             (2,)
        ]).astype(np.float32),
    }
    return data, self.terminated   # terminated is a sibling key the framework adds

from hud import Environment
from hud.environment.robot import RobotEndpoint

env = Environment(name="my-sim")
endpoint = RobotEndpoint(MySimBridge())  # the env drives the bridge only through the endpoint

@env.initialize
async def _up():
    await endpoint.start()
    env.add_capability(await endpoint.capability(contract=CONTRACT))

@env.shutdown
async def _down():
    await endpoint.stop()

@env.template()
async def pick_and_place(task_id: str, seed: int = 0):
    prompt = yield {"prompt": await endpoint.reset(task_id=task_id, seed=seed)}
    yield await endpoint.result()  # {"score", "success", "total_reward"}

​

Agent side

• Model - something with an infer() function that returns action chunks (pre-/post-processing included)
• Adapter - translates env ↔ model spaces.

​

LeRobot integration

• LeRobotModel(policy, preprocess, postprocess) runs the policy through its own LeRobot pre/post-processors, so the checkpoint behaves exactly as it does upstream. Pass an Ensembler to reduce overlapping action chunks to one action per step.
• LeRobotAdapter(model_image_keys=...) maps the env’s cameras and state onto the policy’s inputs from the contract - HWC uint8 → CHW float, state and prompt passed through.

import torch
from lerobot.policies.factory import make_pre_post_processors
from lerobot.policies.pi05.modeling_pi05 import PI05Policy

from hud.agents.robot import RobotAgent, LeRobotModel, LeRobotAdapter

class PI05Agent(RobotAgent):
    def __init__(self):
        device = "cuda" if torch.cuda.is_available() else "cpu"
        policy = PI05Policy.from_pretrained("lerobot/pi05_libero_finetuned").to(device).eval()
        pre, post = make_pre_post_processors(policy.config, "lerobot/pi05_libero_finetuned",
                                             preprocessor_overrides={"device_processor": {"device": device}})
        self.model = LeRobotModel(policy, pre, post)
        self.adapter = LeRobotAdapter(model_image_keys=list(policy.config.image_features))

​

The Model

import numpy as np
from hud.agents.robot import Model

class MyModel(Model):
    def __init__(self, policy):
        self.policy = policy

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

    def infer(self, batch) -> np.ndarray:
        chunk = self.policy(batch)             # run your policy
        return np.asarray(chunk, np.float32)   # [T, A] chunk, in the env's action space

• Input (batch) - the policy-ready inputs your Adapter produced for this step (images, a state vector, the task prompt - whatever your policy consumes). Model and Adapter are a matched pair, so the batch is exactly what your adapter emits.
• Output - a [T, A] float32 numpy array: an action chunk of T timesteps × A action dims, already in the env’s action space. Single-action policies return T = 1.
• reset() - optional; clear per-episode state (an action queue, a chunk buffer) at the start of each episode.

​

The contract

{
  "features": {
    "observation/image": { "role": "observation", "type": "rgb" },
    "observation/state": { "role": "observation" },
    "action":            { "role": "action" }
  }
}

• role (observation / action) - spaces() splits the contract by it and the Adapter wires against that split. Required on every feature.
• type on image observations - rgb/bgr/gray/depth is how the bundled adapter spots a camera; the first observation without an image type becomes the state. Omit it and your image is mistaken for the state. (On the state and action, type is descriptive.)

​

Sim threading

• InlineSimRunner - runs on the event-loop thread. The default; for cheap/CPU sims and tests.
• ThreadSimRunner - sim on a dedicated worker thread, leaving the loop free during a blocking step. For render-heavy or thread-bound sims.
• MainThreadSimRunner - sim on the main thread, for runtimes that own both the main thread and the loop (Isaac/Omniverse); the owner’s pump loop drains queued sim touches between ticks.

​

Telemetry

​

Recording datasets

​

Running a sim in another process

• Env process - publish a remote handle with RobotEndpoint.remote(host, port). It dials the sim process and forwards every control call over JSON-RPC.
• Sim process - wrap the real bridge and expose it with RobotEndpoint(bridge).serve(host, port), using a MainThreadSimRunner so every sim touch runs on the main thread.

• Control plane (start / reset / result) - JSON-RPC between the remote endpoint and the serving process.
• Data plane (the agent’s observe → act loop) - tunnels straight to the bridge’s robot WebSocket; the contract stays env-side.

from hud import Environment
from hud.environment.robot import RobotEndpoint

env = Environment(name="isaac-sim")
endpoint = RobotEndpoint.remote("127.0.0.1", 9100)   # a handle on the bridge in the sim process

@env.initialize
async def _up():
    await endpoint.connect()    # retries until the sim process is serving
    await endpoint.start()
    env.add_capability(await endpoint.capability(contract=CONTRACT))

@env.shutdown
async def _down():
    await endpoint.close()      # drops the link; does not stop the sim

@env.template()
async def pick_and_place(task_id: str, seed: int = 0):
    prompt = yield {"prompt": await endpoint.reset(task_id=task_id, seed=seed)}
    yield await endpoint.result()

import asyncio
from hud.environment.robot import RobotEndpoint, MainThreadSimRunner

async def main():
    bridge = MySimBridge(sim_runner=MainThreadSimRunner())   # sim touches run on main
    server = await RobotEndpoint(bridge).serve("127.0.0.1", 9100)
    await server.wait_closed()

asyncio.run(main())   # launched on the main thread the sim owns

​

API summary

​

See also