Skip to content

Training template

simulo create mybot writes the six files below into mybot/. They are shown exactly as the command writes them, with mybot as the app name. The app previews and trains a Cartpole task briefly, then plays the trained policy, so you can run it before changing anything:

Terminal window
simulo create mybot
cd mybot
simulo run preview.py
simulo run train.py
simulo run eval.py --policy policy_<job>:best
simulo run play.py --policy policy_<job>:best # after training
Show complete codeHide complete codemybot/task.py
mybot/task.py
"""The Cartpole task shared by this training starter."""
from __future__ import annotations
import math
from typing import Tuple
import simulo
# Replace this catalog asset and the task logic below for your own robot.
cartpole = simulo.Asset.from_registry("simulo/robot/cartpole:v1")
app = simulo.App("mybot")
with app.runtime.imports():
import torch # noqa: F401
@app.runtime.torch_jit
def _compute_rewards(
pole_pos: torch.Tensor,
pole_vel: torch.Tensor,
cart_vel: torch.Tensor,
reset_terminated: torch.Tensor,
) -> torch.Tensor:
reward = 1.0 - 2.0 * reset_terminated.float()
reward -= torch.square(pole_pos.squeeze())
reward -= 0.01 * torch.abs(cart_vel.squeeze())
reward -= 0.005 * torch.abs(pole_vel.squeeze())
return reward.view(-1)
class StarterTask(simulo.Task):
"""Balance a pole on a cart. Replace this class for your own task."""
observation_dim = 4
action_dim = 1
episode_length_s = 5.0
action_scale = 100.0
max_cart_pos = 3.0
initial_pole_angle_range = (-0.25, 0.25)
device: str
max_episode_length: int
episode_length_buf: torch.Tensor
reset_terminated: torch.Tensor
def __init__(self, with_camera: bool = False):
super().__init__()
self._with_camera = with_camera
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")
if self._with_camera:
scene.add(
simulo.Camera(
width=640,
height=480,
data_types=["rgb"],
update_period=1.0 / 30.0,
offset=simulo.SensorOffset.look_at(pos=(0.0, 0.0, 5.0), target=(0.0, 0.0, 0.0)),
),
at="/World/overhead_cam",
per_environment=False,
)
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
return torch.cat(
(
joint_pos[:, pole_idx].view(-1, 1),
joint_vel[:, pole_idx].view(-1, 1),
joint_pos[:, cart_idx].view(-1, 1),
joint_vel[:, cart_idx].view(-1, 1),
),
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(
joint_pos[:, self._pole_dof_idx[0]],
joint_vel[:, self._pole_dof_idx[0]],
joint_vel[:, self._cart_dof_idx[0]],
self.reset_terminated,
)
def get_dones(self) -> Tuple[dict[str, torch.Tensor], torch.Tensor]:
joint_pos = self.robot.state.joint_positions
cart_out = torch.abs(joint_pos[:, self._cart_dof_idx[0]]) > self.max_cart_pos
pole_fallen = torch.abs(joint_pos[:, self._pole_dof_idx[0]]) > math.pi / 2
terminated = {
"cart left the track": cart_out,
"pole fell": pole_fallen,
}
truncated = self.episode_length_buf >= self.max_episode_length - 1
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:
if len(env_ids) == 0:
return
self.robot.reset(env_ids)
pole_idx = self._pole_dof_idx[0]
joint_pos = self.robot.default_joint_positions[env_ids]
joint_pos[:, pole_idx] += torch.empty(len(env_ids), device=self.device).uniform_(
self.initial_pole_angle_range[0] * math.pi,
self.initial_pole_angle_range[1] * math.pi,
)
self.robot.set_joint_state(
joint_pos,
velocities=self.robot.default_joint_velocities[env_ids],
env_ids=env_ids,
)

task.py holds everything about the problem:

  • simulo.Asset.from_registry("simulo/robot/cartpole:v1") picks the Cartpole robot from the Simulo global catalog and pins version 1, so every run uses the same model.
  • app = simulo.App("mybot") names the app. train.py imports it from here.
  • with app.runtime.imports(): defers import torch until the job runs on the GPU, so simulo run works on a machine without torch installed.
  • @app.runtime.torch_jit compiles the reward function with TorchScript when the job runs. When you submit, it does nothing.
  • StarterTask is the task. It observes 4 values and emits 1 action. build adds the ground, a light, and the robot to the scene. The other methods read observations, score rewards, decide when episodes end, apply actions, and reset the environments that finished.
Show complete codeHide complete codemybot/preview.py
mybot/preview.py
"""Check the robot inside the task before training; this preview makes no policy."""
import simulo
from task import StarterTask, app
@app.job(type="preview", system=simulo.SystemType.TIER_1, timeout=15 * 60)
def preview(num_envs: int = 16, checks: str = "all"):
return simulo.preview(StarterTask(), num_envs=num_envs, checks=checks, video=True)

preview.py checks the robot inside StarterTask before training. It records video, accepts --num-envs and --checks, and makes no policy. Run it again after every robot or task change. Preview explains the report and recording.

Show complete codeHide complete codemybot/train.py
mybot/train.py
"""Run a Cartpole training job."""
from __future__ import annotations
import simulo
from task import StarterTask, app
@app.job(type="train", system=simulo.SystemType.TIER_1, timeout=10 * 60)
def train(num_envs: int = 16, max_iterations: int = 2) -> dict:
"""Train Cartpole briefly. Simulo saves best.pt and latest.pt in SIMULO_CHECKPOINT_DIR."""
env = simulo.LearningEnv(
task=StarterTask(), num_envs=num_envs, device="cuda", dt=1.0 / 120.0,
physics_steps_per_action=2, env_spacing=4.0, headless=True, seed=42,
)
trainer = simulo.RLTrainer(env=env, algorithm="PPO", device="cuda", seed=42)
try:
stats = trainer.train(max_iterations=max_iterations)
return {"num_envs": num_envs, **stats}
finally:
trainer.close()
env.close()

train.py holds the one job:

  • @app.job(type="train", ...) makes train a training job. Training is the job type you can submit. It asks for a Tier 1 GPU and stops after 10 minutes.
  • The parameters num_envs and max_iterations become the flags --num-envs and --max-iterations. The defaults, 16 environments and 2 iterations, keep the first run short.
  • The body builds a simulo.LearningEnv from the task, trains it with simulo.RLTrainer using PPO, closes both, and returns a dictionary. That dictionary is what simulo result prints.

You don’t write checkpoint code. A training job saves a latest checkpoint as it trains and a best checkpoint whenever the mean episode reward improves. When the job ends, those checkpoints belong to a policy named after the job:

Terminal window
simulo policy list
simulo policy get policy_brisk-heron-4f8k2m:best

Continue training with simulo run train.py --max-iterations <N> --from policy_brisk-heron-4f8k2m:best, where --max-iterations is the new total, including what the checkpoint already has (see Continue training), or turn the checkpoint into a portable ONNX bundle with simulo export policy_brisk-heron-4f8k2m:best.

Show complete codeHide complete codemybot/eval.py
mybot/eval.py
"""Evaluate a Cartpole policy against the task's success rule."""
from __future__ import annotations
import math
import simulo
from task import StarterTask, app
@app.job(type="eval", system=simulo.SystemType.TIER_1, timeout=60 * 60)
def evaluate(
policy: simulo.PolicyCheckpoint,
episodes: int = 100,
num_envs: int = 100,
actions: str = "best",
):
return simulo.evaluate(
StarterTask(),
policy,
episodes=episodes,
num_envs=num_envs,
actions=actions,
)
@app.success
def balanced(task):
"""Count an episode as successful when the pole stays upright to 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° of upright", pole_angle.abs() < math.radians(10))
return lasted & upright

eval.py runs a checkpoint through fixed episodes and judges each episode with the one function marked @app.success. Its named checks become the evaluation report. After training, pass the checkpoint on the command line:

Terminal window
simulo run eval.py --policy policy_brisk-heron-4f8k2m:best

Evaluate a trained policy explains the report, recorded failures, and pipeline exit codes.

Show complete codeHide complete codemybot/play.py
mybot/play.py
"""Play a trained policy and record it."""
from __future__ import annotations
import simulo
from task import StarterTask, app
@app.job(type="play", system=simulo.SystemType.TIER_1, timeout=15 * 60)
def play(policy: simulo.PolicyCheckpoint, episodes: int = 3) -> None:
"""Watch or record a policy: simulo run play.py --policy policy_x:best."""
env = simulo.LearningEnv(
task=StarterTask(with_camera=True),
num_envs=1,
device="cuda",
headless=True,
seed=42,
enable_cameras=True,
)
player = simulo.RLPlayer(env=env, policy=policy, device="cuda")
try:
player.play(num_episodes=episodes, record=simulo.RecordConfig(include_video=True))
finally:
player.close()
env.close()

play.py plays a saved policy and records the episodes. After training, run it with the policy checkpoint you want to play:

Terminal window
simulo run play.py --policy policy_brisk-heron-4f8k2m:best
mybot/.simuloignore
# Gitignore-style excludes for files outside the source package.
*.pt
*.pth
*.ckpt
.venv/
venv/
env/
.env
.env.*
.DS_Store
*.swp
*.npy
*.npz
data/

simulo run packages the folder that holds the job file and uploads it. Each line in .simuloignore is a gitignore-style pattern for files to leave out, such as local checkpoints, virtual environments, .env secrets, and data folders. Add your own patterns the same way.

simulo run already leaves out some files without being told: the folders assets, datasets, checkpoints, outputs, logs, .git, __pycache__, and .simulo; files ending in .usd, .usda, .usdc, .pt, .pth, .onnx, .mcap, .mp4, or .pyc; and symbolic links.

  1. Replace the catalog asset in task.py with your robot. See Choose and validate assets for finding one or publishing your own.

  2. Rewrite StarterTask for your problem. Set observation_dim and action_dim to the sizes your observations and actions really have.

  3. Change the job’s defaults in train.py, or pass --num-envs and --max-iterations when you run it.

  4. Check the app before you submit it:

    Terminal window
    simulo prepare train.py

    simulo prepare asks the platform whether the job would run, without uploading anything or creating a job.

Keep one job per file. simulo run refuses a file with more than one job, or a job without type=, before it uploads anything. To see a full app built on this shape, start with the Cartpole example.