simulo.Task

The RL task contract — what a simulo.Task subclass provides.

A task owns the learning problem: it builds its scene, defines the observation and action spaces (observation_dim / action_dim), and supplies the per-step tensors — observations, rewards, dones — for every parallel environment instance. All tensor methods are vectorized: they operate on a batch of num_envs environments at once, never one at a time.

Structural mirror of simulo.core.task.Task, the abstract base your task subclasses on the worker.

Each environment step calls on_pre_physics_step(actions), advances physics, then calls on_post_physics_step() before collecting dones, rewards, observations, and a pre-reset recording snapshot. It resets environments whose episodes ended after that snapshot.

Real usage, trimmed from the shipped cartpole training app — every Task subclass fills in the same shape: build the scene once, then read observations, compute rewards, decide dones, and apply actions every step:

class CartpoleTask(simulo.Task):
    observation_dim = 4
    action_dim = 1

    def build(self, scene: simulo.Scene) -> None:
        self.robot = simulo.Robot(asset=cartpole, initial_pose=simulo.Pose.identity())
        scene.add(self.robot, at="/World/Robot")

    def apply_actions(self, actions: torch.Tensor) -> None:
        self.robot.set_joint_effort_target(
            self.action_scale * actions, joint_ids=self._cart_dof_idx
        )

get_observations, get_rewards, get_dones, and reset_idx fill in the rest. get_dones() returns named early end conditions and one truncation tensor. See the member list below for each method’s real shape. Wired into a run: env = simulo.LearningEnv(task=CartpoleTask(), num_envs=4096, device="cuda"), then simulo.RLTrainer(env=env, algorithm="PPO", device="cuda") — see Execution.

observation_dim

Task.observation_dim: int

Size of one environment’s observation vector (the policy’s input width).

action_dim

Task.action_dim: int

Size of one environment’s action vector (the policy’s output width).

physics_dt

Task.physics_dt: float

Physics timestep in seconds, default 1/120. LearningEnv uses it unless its own dt is passed.

physics_steps_per_action

Task.physics_steps_per_action: int

Physics steps per policy step, default 2. LearningEnv uses it unless its own argument is passed.

build()

Task.build(scene: SceneProtocol) -> None

Populate scene with this task’s world — robots, objects, lights, terrain.

Called once, before simulation starts. Everything the task’s tensors later refer to (a robot to observe, an object to reach) must be added here.

on_start()

Task.on_start(env: Any) -> None

One-time hook after the environment is live and physics handles exist.

Use it to cache device tensors, resolve robot/sensor handles, and size internal buffers to env.num_envs.

on_pre_physics_step()

Task.on_pre_physics_step(actions: TensorLike) -> None

Process one policy action batch before physics advances.

LearningEnv.step gives this hook a copy of the policy actions, so tasks may safely scale or otherwise modify actions in place. The environment keeps the original batch for recording what the policy chose.

on_post_physics_step()

Task.on_post_physics_step() -> None

Run once after physics advances and before dones, rewards, and observations.

Use this hook for work that must run exactly once per environment step, or for an expensive value shared by several later task methods. Values cached here must also be set for reset environments in reset_idx.

get_observations()

Task.get_observations() -> TensorLike

Return the observation batch, shape (num_envs, observation_dim).

get_rewards()

Task.get_rewards() -> TensorLike

Return the per-step reward for every environment, shape (num_envs,).

get_dones()

Task.get_dones() -> Tuple[Dict[str, TensorLike], TensorLike]

Return named termination batches and one truncated batch.

Each terminated value marks where the task ended the episode, whatever the reason. truncated marks episodes cut off by a time limit.

apply_actions()

Task.apply_actions(actions: TensorLike) -> None

Apply a copy of the policy’s action batch to the sim.

actions has shape (num_envs, action_dim). It may be modified in place without changing the policy action batch the environment records.

reset_idx()

Task.reset_idx(env_ids: TensorLike) -> None

Reset the environments named by env_ids (a 1-D index tensor) to start states.

Only the listed environments reset — the rest keep running. Randomize start states here for robust policies.

Declared as:

@runtime_checkable
class TaskProtocol(Protocol)