Drive rows in velocity iterations

39 · Joints and drives

With drive_mode="rows", drives are PGS rows. During velocity-only iterations, velocity_drive_mode="freeze" keeps the drive impulses from the position solve, and "active" keeps solving them. fuse_joint_velocity_limits replaces a driven joint's two velocity-limit rows with a clamp of its velocity at the end of each iteration. Scene one: a velocity-driven pusher (1 m/s target, 20 N limit) shoves a 2 kg box that starts 1.5 cm inside it, at 3 iterations with 2 velocity iterations. Scene two: a flywheel whose drive asks for 10 rad/s against a 6 rad/s limit.

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 driven flywheel with a velocity limit in each world.

Option	1 world	64 worlds	1024 worlds	solver allocation, 1024 worlds	result in this tutorial
fuse_joint_velocity_limits=False	0.16 ms	0.17 ms	0.24 ms	9 MB	holds 6.0 rad/s
fuse_joint_velocity_limits=True (default)	0.16 ms 0.99×	0.16 ms 0.97×	0.23 ms 0.97×	9 MB	holds 6.0 rad/s

What to look at

• Pusher: active reaches 0.866 m/s against 0.857 m/s with freeze, about 1% closer to the target.
• Flywheel: fused and unfused both hold 6.0 rad/s; the fused clamp is about 3% cheaper per step at 1024 worlds.

Key parameter

velocity_drive_mode / fuse_joint_velocity_limits

velocity_drive_mode is "freeze" (default) or "active" and is read every step. fuse_joint_velocity_limits defaults to True and applies only with drive_mode="rows" and enable_joint_velocity_limits; velocity_limits_fused reports whether it is active.

Try changing

• Raise velocity_iterations to 8.
• Set velocity_limit_activation_fraction=0.9 on the unfused flywheel.

Concept

Drive rows after the position solve

What it is

Velocity-only iterations re-solve velocities without the position bias. A frozen drive row keeps the impulse it found with the bias; an active one re-solves against the unbiased velocities. A fused velocity limit clamps the driven joint's velocity once per iteration instead of adding two rows.

In this scene

Active drives tracked 1% closer to the pusher's target (0.866 vs 0.857 m/s; mean 0.854 vs 0.850 at 8 velocity iterations, separate run). In other runs (a free arm, an arm pressed into a wall) the two modes gave identical results. The fused clamp gave the same 6 rad/s.

Considerations

Cost

Fused limits: 0.230 vs 0.238 ms per step at 1024 worlds, equal at 1 world.

Interactions

Both options only matter with drive_mode="rows"; velocity_drive_mode only with velocity_iterations > 0.

Use it when

Keep fused limits on. Choose active when drive rows push through contacts that the velocity iterations change.

The script

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

39_drive_rows_velocity.py

tutorials/39_drive_rows_velocity.py

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

"""Drive rows in velocity iterations (velocity_drive_mode) and fused velocity limits."""

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

from replay import Recorder

DT = 1.0 / 240.0
SUBSTEPS = 8


def tf(x, y, z):
    return wp.transform((x, y, z), wp.quat_identity())


def build_pusher():
    """A velocity-driven pusher shoving a heavy box that starts 1.5 cm inside it."""
    builder = newton.ModelBuilder()
    builder.add_ground_plane(cfg=newton.ModelBuilder.ShapeConfig(mu=0.5))
    pusher = builder.add_link(xform=tf(0.0, 0.0, 0.05))
    builder.add_shape_box(
        pusher,
        hx=0.05,
        hy=0.05,
        hz=0.05,
        cfg=newton.ModelBuilder.ShapeConfig(density=500.0, mu=0.0),
    )
    slide = builder.add_joint_prismatic(
        -1,
        pusher,
        axis=newton.Axis.X,
        parent_xform=tf(0.0, 0.0, 0.05),
        target_kd=80.0,
        target_qd=1.0,
        effort_limit=20.0,
    )
    builder.add_articulation([slide])
    box = builder.add_body(xform=tf(0.085, 0.0, 0.05))
    builder.add_shape_box(
        box, hx=0.05, hy=0.05, hz=0.05, cfg=newton.ModelBuilder.ShapeConfig(density=2000.0, mu=0.5)
    )
    return builder.finalize()


def build_flywheel():
    """A flywheel whose velocity drive asks for 10 rad/s against a 6 rad/s limit."""
    builder = newton.ModelBuilder(gravity=(0.0, 0.0, 0.0))
    hub = builder.add_link(xform=tf(0.0, 0.0, 0.5))
    builder.add_shape_cylinder(
        hub, radius=0.2, half_height=0.02, cfg=newton.ModelBuilder.ShapeConfig(density=500.0)
    )
    builder.add_shape_box(hub, xform=tf(0.15, 0.0, 0.03), hx=0.04, hy=0.02, hz=0.01)
    hinge = builder.add_joint_revolute(
        -1,
        hub,
        axis=newton.Axis.Z,
        parent_xform=tf(0.0, 0.0, 0.5),
        velocity_limit=6.0,
        target_kd=2.0,
        target_qd=10.0,
    )
    builder.add_articulation([hinge])
    return builder.finalize()


def simulate(rec, label, scene, **options):
    model = build_pusher() if scene == "pusher" else build_flywheel()
    solver = SolverSteel(model, drive_mode="rows", **options)
    pipeline = newton.CollisionPipeline(model)
    contacts = pipeline.contacts()
    state_0, state_1, control = model.state(), model.state(), model.control()

    run = rec.variant(label, model, scene=scene)
    fastest = 0.0  # the largest speed reached in any step of the frame
    for _ in range(rec.frames(1.0 if scene == "pusher" else 2.0)):
        if scene == "pusher":
            run.frame(state_0, pusher_speed=state_0.joint_qd.numpy()[0])
        else:
            run.frame(state_0, omega_max=fastest)
        fastest = 0.0
        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
            fastest = max(fastest, abs(state_0.joint_qd.numpy()[0]))
    run.check(solver)


if __name__ == "__main__":
    rec = Recorder()
    pusher = {"velocity_iterations": 2, "iterations": 3, "baumgarte": 0.8}
    simulate(rec, 'velocity_drive_mode="freeze" (default)', "pusher", **pusher)
    simulate(rec, 'velocity_drive_mode="active"', "pusher", velocity_drive_mode="active", **pusher)
    # rows are created only above 90% of the limit, so the unfused row engages a step late
    flywheel = {"enable_joint_velocity_limits": True, "velocity_limit_activation_fraction": 0.9}
    simulate(
        rec,
        "fuse_joint_velocity_limits=False",
        "flywheel",
        fuse_joint_velocity_limits=False,
        **flywheel,
    )
    simulate(rec, "fuse_joint_velocity_limits=True (default)", "flywheel", **flywheel)
    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