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.

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 — 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).

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.

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[TensorLike, TensorLike]

Return (terminated, truncated) boolean batches, each shape (num_envs,).

terminated marks episodes ended by the task itself (success/failure); truncated marks episodes cut off by a time limit.

apply_actions()

Task.apply_actions(actions: TensorLike) -> None

Apply the policy’s action batch, shape (num_envs, action_dim), to the sim.

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.

---

Note

simulo.Task is a mode-aware name: at submit time it resolves to a lean, torch-free stand-in so your app imports instantly; on the GPU worker it resolves to the real runtime class. The contract below is the definitive list of members every simulo.Task provides, and the contract your own subclass fulfils.

Declared as:

@runtime_checkable
class TaskProtocol(Protocol)