Contact forces and regularization

22 · Compliance and adhesion

A table with an off-center load stands on four legs. Four leg forces with three equations of balance have many solutions, so rigid PGS returns whichever split its sweep order produces. contact_regularization g relaxes each impulse toward zero with weight g/(1+g), which makes the split unique at the cost of a tiny penetration. The forces come from the contact_f observable.

drag to orbit · scroll to zoom

Pause0.00 s

0.1×¼×½×1×2×

Reset view

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 8-box stack in each world, 8 steps of 1/240 s per frame.

Option	1 world	64 worlds	1024 worlds	solver allocation, 1024 worlds	result in this tutorial
contact_regularization=0	1.02 ms	1.21 ms	1.64 ms	215 MB	arbitrary split, one leg at 5.9 N
contact_regularization=0.02	1.04 ms 1.02×	1.25 ms 1.03×	1.73 ms 1.05×	217 MB	moves toward the unique split

What to look at

• All variants carry the same total load (about 105 N).
• Converged rigid PGS (48 iterations) stops at a split with one leg nearly unloaded (5.9 N); separate runs at 200 and 800 iterations give the same, so more iterations do not fix it.
• Regularization 0.2 reaches the unique split an elastic table would have (15.7, 23.1, 29.7, 37.3 N) by 48 iterations; 0.02 at 12 iterations is partway there.

Key parameter

solver.observables({"contact_f"}, contacts=...)

Allocate once and pass observables= to step. contact_f holds each contact's force [N] on shape 0's body, in world frame.

Try changing

• Move the load to the table's center.
• Try contact_regularization=0.2.

Concept

Contact regularization

What it is

Four legs and three equations leave the force split undetermined; rigid PGS returns whichever split its sweep order produces. contact_regularization g relaxes each impulse toward zero with weight g/(1+g), like an implicit spring, which makes the split unique. cfm adds a small value to every row's diagonal for the same reason in redundant rows.

In this scene

Rigid PGS converges to a split with one leg at 5.9 N. Regularization 0.2 converges by 48 iterations to 15.7, 23.1, 29.7 and 37.3 N, the split of a table with elastic legs, unchanged at 800 iterations.

Considerations

Cost

About 5%: 1.64 → 1.73 ms per step in 1024 worlds.

Accuracy

Resting penetration is about g·a·h²/β for a body under acceleration a.

Interactions

fixed_normal only; not with adhesion or contact compliance.

Convergence

Small g approaches the unique split slowly (g = 0.02 reached 14.6 N on the light leg only at 200 iterations); g = 0.2 got there in 48.

Use it when

You read contact forces, for example for force sensing or load estimates, and need a physically plausible, repeatable split.

The script

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

22_contact_forces.py

tutorials/22_contact_forces.py

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

"""Per-contact forces from the contact_f observable, and a unique indeterminate load split."""

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

from replay import Recorder

DT = 1.0 / 240.0
SUBSTEPS = 8


def build():
    builder = newton.ModelBuilder()
    builder.add_ground_plane()
    table = builder.add_body(xform=wp.transform((0.0, 0.0, 0.2501), wp.quat_identity()))
    cfg = newton.ModelBuilder.ShapeConfig(density=500.0, mu=0.6)
    builder.add_shape_box(
        table,
        xform=wp.transform((0.0, 0.0, 0.02), wp.quat_identity()),
        hx=0.3,
        hy=0.2,
        hz=0.02,
        cfg=cfg,
    )
    legs = []
    for sx in (-1, 1):
        for sy in (-1, 1):
            legs.append(
                builder.add_shape_box(
                    table,
                    xform=wp.transform((0.27 * sx, 0.17 * sy, -0.125), wp.quat_identity()),
                    hx=0.02,
                    hy=0.02,
                    hz=0.125,
                    cfg=cfg,
                )
            )
    load = builder.add_body(xform=wp.transform((0.15, 0.05, 0.36), wp.quat_identity()))
    builder.add_shape_box(
        load, hx=0.06, hy=0.06, hz=0.06, cfg=newton.ModelBuilder.ShapeConfig(density=3000.0)
    )
    return builder.finalize(), legs


def simulate(rec, label, **options):
    model, legs = build()
    solver = SolverSteel(model, **options)
    pipeline = newton.CollisionPipeline(model)
    contacts = pipeline.contacts()
    observed = solver.observables({"contact_f"}, contacts=contacts)
    state_0, state_1, control = model.state(), model.state(), model.control()

    run = rec.variant(label, model)
    for _ in range(rec.frames(3.0)):
        n = contacts.rigid_contact_count.numpy()[0]
        force = observed.contact_f.numpy()[:n]
        pairs = zip(
            contacts.rigid_contact_shape0.numpy()[:n], contacts.rigid_contact_shape1.numpy()[:n]
        )
        leg_force = [0.0] * 4
        for k, (a, b) in enumerate(pairs):
            for shape in (a, b):
                if shape in legs:
                    leg_force[legs.index(shape)] += abs(force[k][2])
        run.frame(state_0, **{f"leg{i}": f for i, f in enumerate(leg_force)})
        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
    run.check(solver)


if __name__ == "__main__":
    rec = Recorder()
    simulate(rec, "contact_regularization=0 (default)")
    simulate(rec, "contact_regularization=0.02", contact_regularization=0.02)
    # converged: rigid PGS stops at an arbitrary split, regularization at the unique one
    simulate(rec, "regularization 0, iterations=48", iterations=48)
    simulate(rec, "regularization 0.2, iterations=48", contact_regularization=0.2, iterations=48)
    rec.save()

replay.py

tutorials/replay.py

# 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