Skip to content

Usage

How to build a world, write systems, run a simulation, and read the results. For the design rationale behind these pieces, see Architecture.

Building a world

A World owns every mutable array. Components are registered by name and shape (N, *dim):

from ssl_simulator import World

world = World(n=8)                       # 8 agents
world.add_state("p", dim=2, init=p0)     # integrated state  -> needs a derivative
world.add("u", dim=2)                    # a command signal
world.add("speed")                       # scalar per agent -> shape (8,)
  • add_state registers a component that the integration advances. It takes an optional manifold (default Flat).
  • add registers a plain component (commands, derivatives, diagnostics).
  • Access arrays with world["p"]. Write in place (world["u"][:] = ...) so other systems see the same buffer.

Static run metadata goes in world.params; it is written to the log's settings header.

Writing a system

A system declares which components it reads and writes, and mutates them in run:

from ssl_simulator import System

class Consensus(System):
    reads, writes = ("p",), ("u",)

    def __init__(self, laplacian, gain):
        self.L, self.gain = laplacian, gain

    def run(self, world, dt):
        world["u"][:] = -self.gain * (self.L @ world["p"])

Keep the body vectorized over all N agents - one numpy call, not a Python loop. Controllers, dynamics, and communication-graph builders are all just systems.

The declared reads/writes are what the scheduler uses to order systems, so they must be accurate. Registration order breaks ties, and before=/after= hints resolve ambiguity.

Operations that can't be vectorized

Most math you need is already array-shaped: numpy is, and so are the lieplusplus operators (lpp.so3_exp, lpp.se3_transform, ... take one element or a stack of a million through the same name, and accept any layout, so component arrays go straight in).

For an op that genuinely has no array form, map_entities loops over the leading agent axis while keeping the batched interface:

from ssl_simulator import map_entities

world["R"][:] = map_entities(lambda rot, w: step(rot, w), world["R"], world["omega"])

Vectorizing it later then touches only the op, not the system.

Integration and manifolds

IntegrationSystem advances each state by its derivative, retracting on that state's manifold:

from ssl_simulator import IntegrationSystem, SO3

world.add_state("R", dim=(3, 3), init=R0, manifold=SO3)   # requires the `lie` extra
world.add("omega", dim=3)

world.add_system(IntegrationSystem([("p", "u"), ("R", "omega")]))

Flat does x + dt·τ; SO3 does R · exp(dt·ξ) on (N,3,3) rotation matrices via lieplusplus. Flat and Lie-group states use the same integration path.

Running

from ssl_simulator import Engine

engine = Engine(time_step=0.01, log_filename="run.csv", log_time_step=0.05)
engine.run(world, duration=10.0)
  • time_step - integration step.
  • log_time_step - how often to log (must be ≥ time_step); omit to log every step.
  • run(..., eta=False) silences the progress bar.

The engine builds the schedule, runs a probe pass so control signals are populated before the first log entry, then ticks systems and logs snapshots.

Built-in systems

Ready-made pieces to compose or copy:

from ssl_simulator.robot_models import single_integrator, unicycle_2d
from ssl_simulator.controllers import ConstantSignal, Oscillator

robot = unicycle_2d(world, p=p0, theta=theta0, speed=speed0)  # -> [dynamics, integration]
world.add_system(Oscillator(A=1.0, omega=0.5, speed=1.0, cmd="omega"))
for system in robot:
    world.add_system(system)

The world-builder helpers (single_integrator, unicycle_2d) register the components a model needs and return its systems, which you then add to the world.

Observability

Three complementary mechanisms:

# 1. Component arrays are logged automatically.

# 2. Global scalars sampled at log time:
world.track("mean_speed", lambda: float(world["speed"].mean()))

# 3. Monitors: a running reduction with an optional one-shot warning.
from ssl_simulator import Monitor
Monitor("max_err", source=lambda: err, reduce="max", warn_if=lambda v: v > 1.0,
        warn_msg="tracking error diverged")

Declare monitors in a system's monitors attribute; MonitorSystem ticks them. Make the source return an already-reduced scalar - the lambda runs every step.

Reading results

from ssl_simulator import load_sim

data, settings = load_sim("run.csv")
data["p"]        # (T, N, 2)
data["time"]     # (T,)
settings["gain"] # whatever you put in world.params

Logged names are flat - p, u, time - with no robot./ctrl. prefixes. CSV, NPZ, and HDF5 are supported (HDF5 needs the hdf5 extra); save_sim writes a compatible file from raw arrays.

Visualization lives in the separate ssl_vista package, which reads these logs directly.

Development

just setup      # create the environment
just test       # run the test suite
just lint       # ruff format + check
just docs       # serve docs at http://localhost:8000
just docs-build # build the static site

Run just --list for everything else.