Contact sensors

63 · Solver and runtime

Three 4-box towers. newton.sensors.SensorContact reads each base box's contact force from the solver's contact_f observable, split by counterpart so column 0 is the ground. With sleeping on, settled islands skip their contact rows, so their forces read zero; sleep_skip_constraints=False keeps solving them.

drag to orbit · scroll to zoom

0.00 s

Cost

Time per step, CUDA graph replay, median of 7 interleaved rounds on one NVIDIA RTX PRO 6000 Blackwell Server Edition; ratios are against the first row. Solver allocation is the GPU memory the solver allocates at construction, before warm-up and graph capture. Scene: the three 4-box towers in each world, timed after 4 s of settling.

Option	1 world	8 worlds	solver allocation, 8 worlds	result in this tutorial
no sleeping	1.48 ms	1.75 ms	3 MB	sensor reads 39.2 N
enable_sleeping=True	0.28 ms 0.19×	0.36 ms 0.21×	3 MB	sensor reads 0 while asleep
sleeping, sleep_skip_constraints=False	1.39 ms 0.94×	1.94 ms 1.10×	3 MB	sensor reads 39.3 N

What to look at

Without sleeping the ground carries each 4-box tower: 39.2 N against 39.24 N of weight.
With sleeping the reading drops to 0 at 0.53 s, when the towers fall asleep.
sleep_skip_constraints=False keeps 39.3 N while asleep.

Key parameter

SensorContact(model, sensing_bodies=..., counterpart_shapes=...)

Request its solver_observable_kinds from solver.observables(...), pass them to step, then call sensor.update(state, contacts, observables=...).

Try changing

Add counterpart_bodies to split the force between the ground and the box above.
Read total_force_friction on a pushed tower.

Concept

Contact sensing and sleeping

What it is

SensorContact sums the solver's per-contact forces onto chosen bodies or shapes, optionally per counterpart. Sleeping islands normally skip their contact rows to save work, so contact_f, and the sensor, read zero under them; the contact is still there.

In this scene

Sleeping drops the base reading from 39.2 N to 0; keeping the rows restores it. In the timing below the sleeping towers step 5.3x faster, and keeping their rows gives that back.

Considerations

Cost: One world: 1.48 ms per step awake, 0.28 ms asleep, 1.39 ms asleep with rows kept; 8 worlds: 1.75, 0.36 and 1.94 ms.
Failure modes: Force thresholds (grasp detection, contact rewards) see a false release when an island sleeps.
Interactions: Any wake event restores the rows; contact_f also needs the contacts passed to observables().
Use it when: Keep sleep_skip_constraints=True when nothing reads forces on resting objects; set it False, or leave sleeping off, when sensors drive decisions.

The script

cd tutorials && python 63_contact_sensors.py prints each variant's final metric; --record out.json also writes the replay. Every tutorial imports replay.py from the same folder.

63_contact_sensors.py
replay.py

# SPDX-FileCopyrightText: Copyright (c) 2026 Skild AI and the Steel contributors
# SPDX-License-Identifier: Apache-2.0

"""Contact sensors on Steel contacts, and what sleeping does to their readings."""

import newton
import warp as wp
from newton.sensors import SensorContact
from newton.solvers import SolverSteel

from replay import Recorder

DT = 1.0 / 240.0
SUBSTEPS = 8


def build():
    builder = newton.ModelBuilder()
    cfg = newton.ModelBuilder.ShapeConfig(mu=0.7)
    builder.add_ground_plane(cfg=cfg)
    for k in range(3):
        for i in range(4):
            label = f"base{k}" if i == 0 else f"box{k}_{i}"
            box = builder.add_body(
                xform=wp.transform((0.3 * k - 0.3, 0.0, 0.05 + 0.1005 * i), wp.quat_identity()),
                label=label,
            )
            builder.add_shape_box(box, hx=0.05, hy=0.05, hz=0.05, cfg=cfg)
    return builder.finalize()


def simulate(rec, label, **options):
    model = build()
    solver = SolverSteel(model, **options)
    pipeline = newton.CollisionPipeline(model)
    contacts = pipeline.contacts()
    # one sensor per tower base, split by counterpart: column 0 is the ground plane (shape 0)
    sensor = SensorContact(
        model, sensing_bodies="base*", counterpart_shapes=[0], request_contact_attributes=False
    )
    kinds = set(sensor.solver_observable_kinds) | (
        {"body_awake"} if options.get("enable_sleeping") else set()
    )
    observed = solver.observables(kinds, contacts=contacts)
    state_0, state_1, control = model.state(), model.state(), model.control()
    weight = float(model.body_mass.numpy()[0]) * 9.81

    run = rec.variant(label, model)
    for _ in range(rec.frames(4.0)):
        # ground force on a base = the weight of its 4-box tower
        reading = sensor.force_matrix.numpy()[:, 0, 2].mean()
        awake = observed.body_awake if options.get("enable_sleeping") else None
        run.frame(state_0, awake=awake, base_force_n=reading, tower_weight_n=4 * weight)
        for _ in range(SUBSTEPS):
            pipeline.collide(state_0, contacts)
            solver.step(state_0, state_1, control, contacts, DT, observables=observed)
            state_0, state_1 = state_1, state_0
            sensor.update(state_0, contacts, observables=observed)
    run.check(solver)


if __name__ == "__main__":
    rec = Recorder()
    simulate(rec, "no sleeping")
    # sleeping islands skip their contact rows, so their contact forces read zero
    simulate(rec, "enable_sleeping=True", enable_sleeping=True)
    # keep the rows of sleeping islands: readings stay, the speed-up shrinks
    simulate(
        rec,
        "sleeping, sleep_skip_constraints=False",
        enable_sleeping=True,
        sleep_skip_constraints=False,
    )
    rec.save()

# SPDX-FileCopyrightText: Copyright (c) 2026 Skild AI and the Steel contributors
# SPDX-License-Identifier: Apache-2.0

"""Record Steel tutorial runs as replay JSON for the tutorials page, or print a summary.

Every tutorial imports this module. ``python NN_name.py`` runs the variants and prints
the final metric of each; ``--record out.json`` also writes body poses, shapes and metric
series that the page replays in 3D.
"""

import argparse
import base64
import json
import os
import subprocess
import sys
import time

import newton
import numpy as np
import warp as wp

from source_hash import source_sha256

_RENDERED_TYPES = {
    newton.GeoType.PLANE,
    newton.GeoType.SPHERE,
    newton.GeoType.CAPSULE,
    newton.GeoType.ELLIPSOID,
    newton.GeoType.CYLINDER,
    newton.GeoType.BOX,
    newton.GeoType.MESH,
    newton.GeoType.CONE,
    newton.GeoType.CONVEX_MESH,
}


class Recorder:
    """Collect the variants of one tutorial; ``fps`` is the replay frame rate."""

    def __init__(self, fps: int = 30):
        parser = argparse.ArgumentParser()
        parser.add_argument(
            "--record", metavar="OUT.json", help="write a replay file for the tutorials page"
        )
        parser.add_argument("--device", default=None, help="Warp device, for example cuda:0")
        self.args = parser.parse_args()
        if self.args.device:
            wp.set_device(self.args.device)
        self.fps = fps
        self.script = os.path.abspath(sys.argv[0])
        self.variants: list[Variant] = []

    def frames(self, seconds: float) -> int:
        """Number of replay frames in ``seconds``, including the first."""
        return int(round(seconds * self.fps)) + 1

    def variant(
        self,
        label: str,
        model: newton.Model,
        *,
        ghost: bool = False,
        slot: int | None = None,
        scene: str | None = None,
    ) -> "Variant":
        """Start one run; ``ghost`` overlays run ``slot``, ``scene`` groups runs for charts."""
        variant = Variant(label, model, self.fps, ghost, slot, scene)
        self.variants.append(variant)
        return variant

    def save(self) -> None:
        """Print a summary and, with ``--record``, write the replay file."""
        for variant in self.variants:
            last = variant.metrics[-1] if variant.metrics else {}
            flag = "  (rows dropped)" if variant.overflow else ""
            print(f"{variant.label:<48} {json.dumps(last)}  {variant.wall:.2f} s{flag}")
        if not self.args.record:
            return
        source_sha = source_sha256(self.script)
        doc = {
            "script": os.path.basename(self.script),
            "script_sha256": source_sha,
            "newton_commit": _newton_commit(),
            "newton_version": newton.__version__,
            "warp_version": wp.config.version,
            "gpu": wp.get_device().name,
            "fps": self.fps,
            "variants": [v.to_json() for v in self.variants],
        }
        with open(self.args.record, "w") as f:
            json.dump(doc, f, separators=(",", ":"))
        print(f"wrote {self.args.record}")


class Variant:
    """One recorded run: a pose per frame for every body, plus named metric values."""

    def __init__(self, label, model, fps, ghost, slot, scene):
        self.label, self.model, self.fps, self.ghost, self.slot = label, model, fps, ghost, slot
        self.scene = scene
        self.poses: list[np.ndarray] = []
        self.metrics: list[dict] = []
        self.awake: list[list[int]] = []
        self.particles: list[np.ndarray] = []
        self.overflow = False
        self.start = time.perf_counter()
        self.wall = 0.0

    def frame(self, state: newton.State, awake=None, particles=None, **metrics) -> None:
        """Record ``state.body_q``, optional awake flags, particle positions and metric values."""
        self.poses.append(state.body_q.numpy().astype(np.float32).copy())
        if particles is not None:
            self.particles.append(particles.numpy().astype(np.float32).copy())
        self.metrics.append({k: _round(v) for k, v in metrics.items()})
        if awake is not None:
            self.awake.append([int(x) for x in awake.numpy()])
        self.wall = time.perf_counter() - self.start

    def check(self, solver) -> None:
        """Note whether the solver dropped rows; call once after the run."""
        try:
            solver.check_constraint_capacity()
        except RuntimeError:
            self.overflow = True

    def to_json(self) -> dict:
        poses = (
            np.stack(self.poses)
            if self.poses
            else np.zeros((0, self.model.body_count, 7), np.float32)
        )
        return {
            "label": self.label,
            "fps": self.fps,
            "ghost": self.ghost,
            "slot": self.slot,
            "scene": self.scene,
            "bodies": int(self.model.body_count),
            "frames": int(poses.shape[0]),
            "poses": base64.b64encode(poses.astype(np.float32).tobytes()).decode(),
            "shapes": _shape_table(self.model),
            "metric": self.metrics,
            "awake": self.awake or None,
            "particles": self._particles_json(),
            "wall_s": round(self.wall, 3),
            "overflow": self.overflow,
        }

    def _particles_json(self):
        if not self.particles:
            return None
        frames = np.stack(self.particles)
        tris = self.model.tri_indices
        radius = self.model.particle_radius.numpy()
        return {
            "count": int(frames.shape[1]),
            "frames": base64.b64encode(frames.astype(np.float32).tobytes()).decode(),
            "radius": round(float(radius.mean()), 5) if radius.size else 0.0,
            "tris": [int(v) for v in tris.numpy().reshape(-1)]
            if tris is not None and tris.shape[0]
            else None,
        }


def _shape_table(model):
    types = model.shape_type.numpy()
    scales = model.shape_scale.numpy()
    xforms = model.shape_transform.numpy()
    bodies = model.shape_body.numpy()
    flags = model.shape_flags.numpy()
    shapes = []
    for i, t in enumerate(types):
        if int(t) not in _RENDERED_TYPES or not flags[i] & int(newton.ShapeFlags.VISIBLE):
            continue
        shape = {
            "b": int(bodies[i]),
            "t": int(t),
            "s": [round(float(v), 5) for v in scales[i]],
            "x": [round(float(v), 6) for v in xforms[i]],
        }
        if int(t) in (newton.GeoType.MESH, newton.GeoType.CONVEX_MESH):
            mesh = model.shape_source[i]
            shape["v"] = [round(float(v), 5) for v in np.asarray(mesh.vertices).reshape(-1)]
            shape["i"] = [int(v) for v in np.asarray(mesh.indices).reshape(-1)]
        shapes.append(shape)
    return shapes


def _round(value):
    value = float(value)
    return round(value, 5) if np.isfinite(value) else None


def _newton_commit():
    if os.environ.get("NEWTON_COMMIT"):
        return os.environ["NEWTON_COMMIT"]
    try:
        root = os.path.dirname(os.path.dirname(newton.__file__))
        return subprocess.check_output(["git", "-C", root, "rev-parse", "HEAD"], text=True).strip()
    except (OSError, subprocess.CalledProcessError):
        return None

Recorded with SolverSteel on Newton 1.9.0.dev0, Warp 1.19.0, NVIDIA RTX PRO 6000 Blackwell Server Edition. Replays read the state back every frame, so they are not a speed benchmark. Script · page text

Steel docs and tutorials are licensed under Apache-2.0. Third-party notices. "Skild", "Skild AI" and the Skild logo are trademarks of Skild AI and are not licensed under Apache-2.0.