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:
simulo create mybotcd mybotsimulo run preview.pysimulo run train.pysimulo run eval.py --policy policy_<job>:bestsimulo run play.py --policy policy_<job>:best # after trainingtask.py
Section titled “task.py”Show complete codemybot/task.py
"""The Cartpole task shared by this training starter."""
from __future__ import annotations
import mathfrom 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_jitdef _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.pyimports it from here.with app.runtime.imports():defersimport torchuntil the job runs on the GPU, sosimulo runworks on a machine without torch installed.@app.runtime.torch_jitcompiles the reward function with TorchScript when the job runs. When you submit, it does nothing.StarterTaskis the task. It observes 4 values and emits 1 action.buildadds 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.
preview.py
Section titled “preview.py”Show complete codemybot/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.
train.py
Section titled “train.py”Show complete codemybot/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", ...)makestraina 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_envsandmax_iterationsbecome the flags--num-envsand--max-iterations. The defaults, 16 environments and 2 iterations, keep the first run short. - The body builds a
simulo.LearningEnvfrom the task, trains it withsimulo.RLTrainerusing PPO, closes both, and returns a dictionary. That dictionary is whatsimulo resultprints.
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:
simulo policy listsimulo policy get policy_brisk-heron-4f8k2m:bestContinue 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.
eval.py
Section titled “eval.py”Show complete codemybot/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.successdef 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 & uprighteval.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:
simulo run eval.py --policy policy_brisk-heron-4f8k2m:bestEvaluate a trained policy explains the report, recorded failures, and pipeline exit codes.
play.py
Section titled “play.py”Show complete codemybot/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:
simulo run play.py --policy policy_brisk-heron-4f8k2m:best.simuloignore
Section titled “.simuloignore”# Gitignore-style excludes for files outside the source package.*.pt*.pth*.ckpt.venv/venv/env/.env.env.*.DS_Store*.swp*.npy*.npzdata/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.
Make it yours
Section titled “Make it yours”-
Replace the catalog asset in
task.pywith your robot. See Choose and validate assets for finding one or publishing your own. -
Rewrite
StarterTaskfor your problem. Setobservation_dimandaction_dimto the sizes your observations and actions really have. -
Change the job’s defaults in
train.py, or pass--num-envsand--max-iterationswhen you run it. -
Check the app before you submit it:
Terminal window simulo prepare train.pysimulo prepareasks 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.