Randomizing friction

68 · Domain randomization

Five boxes on 25 degree ramps, one per world. At 1 s the script writes a sampled mu into shape_material_mu for the ramp and the box of each world, then calls notify_model_changed(SHAPE_PROPERTIES). Boxes with mu below tan(25 deg) = 0.47 slide; the rest hold. Steel keeps its own copy of the material table, so without the notification the edit has no effect.

drag to orbit · scroll to zoom

0.00 s

What to look at

With notify: the boxes with mu < 0.47 slide off the ramp.
Without notify: every box holds.

Key parameter

model.shape_material_mu

One coefficient per shape; a contact combines the two shapes' values with friction_combine (average by default).

Try changing

Randomize only the box's mu and watch the average with the ramp's mu decide.
Pass world_mask to notify_model_changed to discard the warm start of only the changed worlds.

Concept

When Steel sees a randomized value

What it is

Steel keeps a solver-side copy of this input, such as its material table, damping buffer, cached body-to-COM transforms or effective armature. Writing the model array changes nothing until notify_model_changed with the matching flag refreshes that copy.

In this scene

Without notify: every box holds.

Considerations

Cost: Call only, default solver, six-link chain and four boxes per world, 1024 worlds (1.12 ms per collide + step): SHAPE_PROPERTIES 84 µs (15 µs captured). Flags measured one at a time; combined calls and feature validation can cost more. This notification does not request a mass-matrix refresh.
Graph capture: These flags can be captured in a reset graph. Kinematic-flag changes, sleeping, and shape flags with sparse factors plus restitution run eagerly; captured calls skip some value checks.
Interactions: world_mask limits which worlds lose their warm start; cached data is refreshed for every world.
Use it when: Grasping and pushing, where the friction coefficient is the usual sim-to-real gap.

The script

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

dr_friction.py
replay.py

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

"""Friction randomization: write per-world shape_material_mu, then notify SHAPE_PROPERTIES."""

import math

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

from replay import Recorder

DT = 1.0 / 240.0
SUBSTEPS = 8
WORLDS = 5
SLOPE = math.radians(25.0)  # boxes slide where mu < tan(25 deg) = 0.47


def build():
    ramp = newton.ModelBuilder()
    tilt = wp.quat_from_axis_angle(wp.vec3(0.0, 1.0, 0.0), SLOPE)
    ramp.add_shape_box(-1, xform=wp.transform((0.0, 0.0, 0.4), tilt), hx=0.9, hy=0.1, hz=0.03)
    downhill = np.array(wp.quat_rotate(tilt, wp.vec3(1.0, 0.0, 0.0)))
    normal = np.array(wp.quat_rotate(tilt, wp.vec3(0.0, 0.0, 1.0)))
    box = ramp.add_body(xform=wp.transform((0.0, 0.0, 0.4) - 0.7 * downhill + 0.081 * normal, tilt))
    ramp.add_shape_box(box, hx=0.05, hy=0.05, hz=0.05)
    builder = newton.ModelBuilder()
    builder.replicate(ramp, WORLDS, spacing=(0.0, 0.3, 0.0))
    builder.add_ground_plane()
    return builder.finalize()


def simulate(rec, label, mu, notify):
    model = build()
    solver = SolverSteel(model)
    pipeline = newton.CollisionPipeline(model)
    contacts = pipeline.contacts()
    state_0, state_1, control = model.state(), model.state(), model.control()
    start = state_0.body_q.numpy()[:, :3].copy()

    run = rec.variant(label, model)
    for frame in range(rec.frames(4.0)):
        if frame == rec.fps:  # randomize at 1 s, as an episode reset would
            # ramp and box of world w both get mu[w]; the ground plane (world -1) keeps its value
            world = model.shape_world.numpy()
            values = model.shape_material_mu.numpy()
            values[world >= 0] = mu[world[world >= 0]]
            model.shape_material_mu.assign(values)
            if notify:
                solver.notify_model_changed(newton.ModelFlags.SHAPE_PROPERTIES)
        slide = np.linalg.norm(state_0.body_q.numpy()[:, :3] - start, axis=1)
        run.frame(state_0, **{f"mu {m:.2f}": d for m, d in zip(mu, slide)})
        for _ in range(SUBSTEPS):
            pipeline.collide(state_0, contacts)
            solver.step(state_0, state_1, control, contacts, DT)
            state_0, state_1 = state_1, state_0
    run.check(solver)


if __name__ == "__main__":
    rng = np.random.default_rng(3)
    mu = np.sort(rng.uniform(0.2, 0.8, WORLDS))
    rec = Recorder()
    simulate(rec, "mu sampled per world + notify", mu, True)
    simulate(rec, "mu sampled per world, no notify", mu, 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.