Skip to content
Simulo

Hello

Hello is the shortest real training job: Cartpole with 16 environments for 2 training iterations. Run it first to check that you can sign in, submit a job, watch its logs, and get a policy back before you spend time on a longer run.

Save these files in a folder called hello:

Show complete codeHide complete codehello/task.py
hello/task.py
from __future__ import annotations


import math
from typing import Tuple


import simulo


# The cartpole robot - a validated, version-pinned global-catalog asset.
# Capturing it at module level lets submit resolve this exact version before the
# job runs; the worker mounts it and the Task below consumes the same handle.
cartpole = simulo.Asset.from_registry("simulo/robot/cartpole:v1")


# Advanced: pick a different Simulo runtime with
# App("name", runtime=simulo.Runtime.from_registry("simulo/gpu-rl:2026.06"));
# see the Runtimes docs.
app = simulo.App("hello")


# The ONE module-level heavy import - deferred under the runtime guard so discovery
# records it as a remote import instead of resolving it.
with app.runtime.imports():
    import torch  # noqa: F401  (resolved only in execution mode, on the worker)




@app.runtime.torch_jit
def _compute_rewards(
    rew_scale_alive: float,
    rew_scale_terminated: float,
    rew_scale_pole_pos: float,
    rew_scale_cart_vel: float,
    rew_scale_pole_vel: float,
    pole_pos: torch.Tensor,
    pole_vel: torch.Tensor,
    cart_pos: torch.Tensor,
    cart_vel: torch.Tensor,
    reset_terminated: torch.Tensor,
) -> torch.Tensor:
    """JIT-compiled reward kernel (the classic cartpole balance reward).


    ``@app.runtime.torch_jit`` is a no-op marker at submit (no ``torch`` locally) and
    real ``torch.jit.script`` on the worker - so this lives at module level and is
    still torch-free to *define* during discovery (its body never runs at submit).
    """
    pole_pos = pole_pos.squeeze()
    pole_vel = pole_vel.squeeze()
    cart_pos = cart_pos.squeeze()
    cart_vel = cart_vel.squeeze()
    reset_terminated = reset_terminated.squeeze()


    rew_alive = rew_scale_alive * (1.0 - reset_terminated.float())
    rew_termination = rew_scale_terminated * reset_terminated.float()
    rew_pole_pos = rew_scale_pole_pos * torch.square(pole_pos)
    rew_cart_vel = rew_scale_cart_vel * torch.abs(cart_vel)
    rew_pole_vel = rew_scale_pole_vel * torch.abs(pole_vel)


    reward: torch.Tensor = rew_alive + rew_termination + rew_pole_pos + rew_cart_vel + rew_pole_vel
    # Keep the (num_envs,) per-env reward contract even when num_envs == 1: the
    # squeezes above collapse a single-env batch to a 0-d scalar, which breaks
    # per-env consumers (e.g. the MCAP recorder's per-env /reward indexing).
    return reward.view(-1)




class CartpoleTask(simulo.Task):
    """Balance a pole on a cart by applying horizontal forces to the cart.


    Observation (4-dim): pole angle, pole angular velocity, cart position, cart
    velocity. Action (1-dim): scaled horizontal force on the cart.


    Defined at module level: ``simulo.Task`` is a torch-free contract stand-in at
    submit and the real ``simulo.core.Task`` on the worker, so the same class
    authors lean and trains heavy.
    """


    observation_dim = 4
    action_dim = 1


    episode_length_s = 5.0
    action_scale = 100.0  # [N]


    max_cart_pos = 3.0  # [m]
    initial_pole_angle_range = (-0.25, 0.25)  # fraction of pi [rad]


    rew_scale_alive = 1.0
    rew_scale_terminated = -2.0
    rew_scale_pole_pos = -1.0
    rew_scale_cart_vel = -0.01
    rew_scale_pole_vel = -0.005


    # Framework-injected at runtime by ``simulo.core.Task`` / ``LearningEnv``
    # (declared here only so the type checker sees the names the methods read; the
    # annotations are PEP 563 strings and never shadow the inherited values).
    device: str
    max_episode_length: int
    episode_length_buf: torch.Tensor
    reset_terminated: torch.Tensor


    def build(self, scene: simulo.Scene) -> None:
        scene.add(simulo.Terrain.plane(name="ground"), at="/", per_environment=False)
        scene.add(
            simulo.Light.dome(name="light", intensity=2000.0, color=(0.75, 0.75, 0.75)),
            at="/",
            per_environment=False,
        )
        self.robot = simulo.Robot(asset=cartpole, initial_pose=simulo.Pose.identity())
        scene.add(self.robot, at="/World/Robot")


    def on_start(self, env: simulo.LearningEnv) -> None:
        self._cart_dof_idx = self.robot.find_joints("slider_to_cart")
        self._pole_dof_idx = self.robot.find_joints("cart_to_pole")


    def get_observations(self) -> torch.Tensor:
        pole_idx = self._pole_dof_idx[0]
        cart_idx = self._cart_dof_idx[0]
        joint_pos = self.robot.state.joint_positions
        joint_vel = self.robot.state.joint_velocities
        pole_pos = joint_pos[:, pole_idx].view(-1, 1)
        pole_vel = joint_vel[:, pole_idx].view(-1, 1)
        cart_pos = joint_pos[:, cart_idx].view(-1, 1)
        cart_vel = joint_vel[:, cart_idx].view(-1, 1)
        return torch.cat((pole_pos, pole_vel, cart_pos, cart_vel), dim=-1)


    def get_rewards(self) -> torch.Tensor:
        joint_pos = self.robot.state.joint_positions
        joint_vel = self.robot.state.joint_velocities
        return _compute_rewards(
            self.rew_scale_alive,
            self.rew_scale_terminated,
            self.rew_scale_pole_pos,
            self.rew_scale_cart_vel,
            self.rew_scale_pole_vel,
            joint_pos[:, self._pole_dof_idx[0]],
            joint_vel[:, self._pole_dof_idx[0]],
            joint_pos[:, self._cart_dof_idx[0]],
            joint_vel[:, self._cart_dof_idx[0]],
            self.reset_terminated,
        )


    def get_dones(self) -> Tuple[dict[str, torch.Tensor], torch.Tensor]:
        pole_idx = self._pole_dof_idx[0]
        cart_idx = self._cart_dof_idx[0]
        joint_pos = self.robot.state.joint_positions
        truncated = self.episode_length_buf >= self.max_episode_length - 1
        cart_out = torch.abs(joint_pos[:, cart_idx]) > self.max_cart_pos
        pole_fallen = torch.abs(joint_pos[:, pole_idx]) > math.pi / 2
        terminated = {"cart left the track": cart_out, "pole fell": pole_fallen}
        return terminated, truncated


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


    def reset_idx(self, env_ids: torch.Tensor) -> None:
        num_resets = len(env_ids)
        if num_resets == 0:
            return
        self.robot.reset(env_ids)
        pole_idx = self._pole_dof_idx[0]
        # Start from the default joint positions for the selected environments.
        joint_pos = self.robot.default_joint_positions[env_ids]
        random_angles = torch.empty(num_resets, device=self.device).uniform_(
            self.initial_pole_angle_range[0] * math.pi,
            self.initial_pole_angle_range[1] * math.pi,
        )
        joint_pos[:, pole_idx] += random_angles
        # Apply the randomized starting positions and default velocities together.
        joint_vel = self.robot.default_joint_velocities[env_ids]
        self.robot.set_joint_state(joint_pos, velocities=joint_vel, env_ids=env_ids)
Show complete codeHide complete codehello/train.py
hello/train.py
from __future__ import annotations


from typing import Any


import simulo


from task import CartpoleTask, app




@app.job(type="train", system=simulo.SystemType.TIER_1, timeout=10 * 60)
def train(num_envs: int = 16, max_iterations: int = 2) -> dict[str, Any]:
    """Train Cartpole briefly."""
    env = simulo.LearningEnv(
        task=CartpoleTask(), num_envs=num_envs, device="cuda", env_spacing=4.0, headless=True, seed=42
    )
    trainer = simulo.RLTrainer(env=env, algorithm="PPO", device="cuda", seed=42)
    stats = trainer.train(max_iterations=max_iterations)
    trainer.close()
    env.close()
    return {"num_envs": num_envs, **stats}
Show complete codeHide complete codehello/eval.py
hello/eval.py
from __future__ import annotations


import math


import simulo


from task import CartpoleTask, app




@app.job(type="eval", system=simulo.SystemType.TIER_1, timeout=8 * 60 * 60)
def evaluate(
    policy: simulo.PolicyCheckpoint,
    episodes: int = 100,
    num_envs: int = 100,
    actions: str = "best",
):
    return simulo.evaluate(
        CartpoleTask(), policy, episodes=episodes, num_envs=num_envs, actions=actions
    )




@app.success
def balanced(task):
    """Count an episode when the pole remains upright until the time limit."""
    pole_angle = task.robot.state.joint_positions[
        :, task.robot.find_joints("cart_to_pole")[0]
    ]
    lasted = task.check("lasted to the time limit", task.survived())
    upright = task.check(
        "pole within 10 degrees of upright", pole_angle.abs() < math.radians(10)
    )
    return lasted & upright

task.py defines the app and the Cartpole task. train.py holds the one training job and imports the task from task.py. Keep them in the same folder, because simulo run packages the folder that holds the job file.

Run it

Section titled “Run it”
Terminal window
simulo run hello/train.py

The defaults are already small (--num-envs 16, --max-iterations 2), so you don’t need any flags. The job asks for a Tier 1 GPU and stops if it runs longer than 10 minutes.

simulo run follows the job’s logs while it trains. When it finishes, the job is completed and simulo run prints the new policy’s best and latest checkpoints with the next commands to try. Then read what the job returned:

Terminal window
simulo result

simulo result prints the job’s return value as JSON: the num_envs you ran with and the training statistics, such as iterations. For a run long enough to learn the task, move on to Cartpole.

Evaluate it

Section titled “Evaluate it”

The success rule counts an episode only when the pole lasts to the time limit and stays within 10 degrees of upright. Run simulo run hello/eval.py --policy <policy-id>:best; see Evaluate a trained policy.

What to reuse

Section titled “What to reuse”
  • @app.job(type="train") turns a typed Python function into a training job. Its parameters, num_envs and max_iterations, become the flags --num-envs and --max-iterations.
  • Printed output becomes the job’s live logs, and the returned dictionary becomes its result.
  • Put the task in task.py and the job in train.py. Each job file holds exactly one job.

See Apps and jobs for how jobs, parameters, and results fit together.