Scene

Build a simulation with no policy or trainer. Four Cartpoles move through a deterministic sine sweep, giving you a compact app for scripted control and live visual inspection.

Save this complete app as scene/app.py:

Show complete codeHide complete codescene/app.py

scene/app.py

from __future__ import annotations

import math
from typing import Any, List

import simulo

# The cartpole robot — a validated, version-pinned global-catalog asset.
cartpole = simulo.Asset.from_registry("simulo/robot/cartpole:v1")


class SceneScenario(simulo.Scenario):
    """A cartpole replicated across N environments, driven by a scripted sine sweep.

    The lifecycle is the Scenario contract: ``build`` declares the
    scene, ``on_start`` runs once the runtime is live, ``on_step`` runs every
    physics step, ``on_shutdown`` runs at teardown.

    The control law is deliberately trivial and deterministic — a phase-offset
    sinusoidal force on each cart, so the environments sweep as a visible wave
    and the pole swings. Every ``reset_every`` steps the robots snap back to a
    fanned set of pole tilts. There is nothing to learn here; the point is a
    simulation you can *watch*.
    """

    #: Peak horizontal force applied to the cart [N].
    effort_scale = 5.0
    #: Steps per full back-and-forth sweep of the cart.
    sweep_period_steps = 120.0
    #: Steps between scripted resets.
    reset_every = 500
    #: Peak pole tilt applied at reset [rad]; environments are fanned across ±this.
    initial_pole_tilt = 0.20

    # Framework-injected by ``simulo.scenario.Scenario.__init__(simulation=, scene=)``
    # — declared here only so a reader (and a type checker) sees the names the
    # methods below read. The annotations are PEP 563 strings and never shadow
    # the values the base class assigns. ``simulation`` is ``Any`` because the
    # object injected here is the backend's sim clock, whose concrete type is
    # not importable on the torch-free submit surface. Naming ``simulo.Simulation``
    # instead would not help: it IS on the torch-free client surface now — and it
    # is the same class — but it has no ``TYPE_CHECKING`` facade of its own, so it
    # types as ``object``, too restrictive for the
    # ``self.simulation.viewer.set_camera(...)`` call that ``on_start`` makes below.
    simulation: Any
    scene: Any

    def build(self, scene: simulo.Scene) -> None:
        """Declare the scene: ground, light, and one replicated cartpole."""
        # Shared environment items — one copy for the whole scene, not per-env.
        scene.add(
            simulo.Terrain.plane(name="ground"),
            at="/World/defaultGroundPlane",
            per_environment=False,
        )
        scene.add(
            simulo.Light.dome(name="light", intensity=3000.0, color=(0.75, 0.75, 0.75)),
            at="/World/Light",
            per_environment=False,
        )

        # The robot IS replicated: with num_envs > 1 the runner clones it across
        # every environment origin automatically.
        self.robot = simulo.Robot(
            asset=cartpole,
            initial_pose=simulo.Pose.identity(),
        )
        scene.add(self.robot, at="/World/Robot")

        self.step_count = 0

    def on_start(self) -> None:
        """Resolve joint indices, frame the camera, and place the robots."""
        self._cart_dof_idx = self.robot.find_joints("slider_to_cart")
        self._pole_dof_idx = self.robot.find_joints("cart_to_pole")
        self._num_joints = self.robot.num_joints

        print(f"[scene] environments:   {self.scene.num_envs}", flush=True)
        print(f"[scene] robot joints:   {self._num_joints}", flush=True)

        # Frame the whole grid of environments — this is the shot the viewstream
        # sends to the browser. Null-safe when there is no viewer (headless).
        self.simulation.viewer.set_camera(eye=(6.0, 0.0, 4.0), look_at=(0.0, 0.0, 2.0))

        self._reset_robots()

    def on_step(self) -> None:
        """Drive each cart with a phase-offset sine sweep; reset periodically."""
        self.step_count += 1

        if self.step_count % self.reset_every == 0:
            self._reset_robots()

        num_envs = self.scene.num_envs
        cart_idx = self._cart_dof_idx[0]
        phase = 2.0 * math.pi * self.step_count / self.sweep_period_steps

        # Shape (num_envs, num_joints) — a plain nested list, no torch. Every
        # joint gets an explicit entry (zero on the unactuated pole joint, which
        # is physically what we want) and we deliberately do NOT pass
        # ``joint_ids=``: ``Robot`` builds a CPU tensor from a list with no
        # ``device=``, and on a CUDA robot ``joint_ids=[...]`` triggers advanced
        # indexing (``index_put_``), which requires matching devices and raises.
        # ``joint_ids=None`` resolves to a plain slice (``copy_`` semantics),
        # which tolerates the CPU source. See the module docstring.
        efforts: List[List[float]] = [[0.0] * self._num_joints for _ in range(num_envs)]
        for env_index in range(num_envs):
            efforts[env_index][cart_idx] = self.effort_scale * math.sin(
                phase + 2.0 * math.pi * env_index / max(num_envs, 1)
            )
        self.robot.set_joint_effort_target(efforts)

    def _reset_robots(self) -> None:
        """Snap every robot back to a fanned set of pole tilts (velocities zeroed)."""
        print(f"[scene] reset at step {self.step_count}", flush=True)
        self.robot.reset()

        num_envs = self.scene.num_envs
        pole_idx = self._pole_dof_idx[0]

        # Shape (num_envs, num_joints). Fan the pole tilt across ±initial_pole_tilt so
        # the environments are visibly distinct from the first frame. Deterministic —
        # no RNG, so a rerun of the same job looks identical. num_envs == 1 is NOT the
        # num_envs > 1 formula's midpoint (spread = 0.5 there makes tilt cancel to
        # exactly 0.0 — the pole would start bolt upright, silently defeating the
        # "fanned tilt" this app is about) — it gets its own full-tilt branch instead.
        joint_pos: List[List[float]] = [[0.0] * self._num_joints for _ in range(num_envs)]
        for env_index in range(num_envs):
            if num_envs > 1:
                spread = env_index / (num_envs - 1)
                tilt = self.initial_pole_tilt * (2.0 * spread - 1.0)
            else:
                tilt = self.initial_pole_tilt
            joint_pos[env_index][pole_idx] = tilt

        # velocities default to zeros.
        self.robot.set_joint_state(joint_pos)

    def on_shutdown(self) -> None:
        """Report the step total at teardown."""
        print(f"[scene] shut down after {self.step_count} steps", flush=True)


app = simulo.App("scene")


@app.job(system=simulo.SystemType.TIER_1, timeout=30 * 60)  # Tier 1: T4, 16 GB VRAM. See `simulo systems`.
def simulate(num_steps: int = 1000, num_envs: int = 4) -> dict[str, Any]:
    """Run the scripted scenario on the GPU for ``num_steps`` physics steps.

    All heavy work happens here, on the worker: ``simulo.run`` resolves through
    the mode-aware learning resolver — an inert stand-in at submit (this job
    body is read but never executed then) and the real ``simulo.scenario.run``
    (which lives in the ``simulo-backend`` distribution, absent from the
    torch-free client) only once the worker executes it. No second import is
    needed, which is also why it needs no ``app.runtime.imports()`` guard.
    Keep the call INSIDE a job body: the identical line at module scope would
    silently do nothing at submit (an inert stand-in, no error) yet launch a
    real simulation at import time on the worker.

    ``viewstream`` is intentionally not passed: its ``None`` default is what lets
    ``simulo run --viewstream`` turn live view on via the worker-set
    ``SIMULO_VIEWSTREAM_ENABLED`` env. See the module docstring.

    Args:
        num_steps: Number of physics steps to simulate.
        num_envs: Number of parallel environments to replicate the robot across.

    Returns:
        A JSON-serialisable dict recording what was simulated.

    Raises:
        ValueError: If ``num_steps`` or ``num_envs`` is not a positive integer.
    """
    # A scripted-scenario job body is exactly the kind of code new users copy as
    # a template (``simulo create --type scenario``) — it should model good
    # argument hygiene, not omit it. ``num_envs < 1`` in particular is not a
    # merely-slow edge case: it produces an EMPTY effort list in on_step/
    # _reset_robots and crashes deep in the worker, far from this message.
    if num_steps < 1:
        raise ValueError(f"num_steps must be a positive integer, got {num_steps}")
    if num_envs < 1:
        raise ValueError(f"num_envs must be a positive integer, got {num_envs}")

    simulo.run(
        SceneScenario,
        device="cuda",
        headless=True,
        max_steps=num_steps,
        num_envs=num_envs,
        env_spacing=2.0,
    )

    return {"steps": num_steps, "num_envs": num_envs}

Run the app

simulo run scene/app.py --num-steps 1000 --num-envs 4

The job returns {"steps": 1000, "num_envs": 4} after the scenario shuts down. To watch a longer run in the live viewport:

simulo run scene/app.py --viewstream --num-steps 3000 --num-envs 4
simulo view

What to reuse

• Scenario is the right abstraction when your control law is scripted instead of learned.
• build, on_start, on_step, and on_shutdown separate scene setup from live simulation work.
• num_envs replicates the scene while keeping one control loop.

This scenario supports live view but does not create an MCAP recording. See Scenarios for lifecycle details and Recordings & Live Viewstream for recorded policy rollouts.