PD drives

30 · Joints and drives

Four arms with gravity. Three get a position target of 45° at 0.5 s with different gains: soft and underdamped, soft and damped, stiff and damped. The fourth has only target_kd and follows a velocity target of 2 rad/s from 1 s. By default Steel integrates drives implicitly in the mass matrix, so stiff gains stay stable at large steps.

drag to orbit · scroll to zoom

0.00 s

What to look at

k_e 50, k_d 1 overshoots to about 61° and rings.
k_e 400 settles closest to 45°; the soft drives sag under gravity.
The velocity-driven arm spins continuously.

Key parameter

target_ke / target_kd

Joint arguments; Control.joint_target_q and joint_target_qd set the targets each step.

Try changing

Set target_ke=5000 at DT = 1/60.
Add effort_limit=1.0 to the stiff arm.

Concept

Implicit drives

What it is

A PD drive pulls a joint toward a target position with stiffness kₑ and toward a target velocity with damping k_d. By default Steel folds h·k_d + h²·kₑ into the mass matrix, an implicit integration that stays stable for stiff gains at large steps.

H̃ = H + h·k_d + h²·kₑ

In this scene

kₑ 50, k_d 1 overshoots to 61°; kₑ 400, k_d 20 settles at 44.7°. Gravity makes soft drives sag.

Considerations

Cost: None extra: the drive is part of the factorization.
Accuracy: Only the explicit drive force is clamped to effort_limit; the implicit part is not.
Interactions: Driven articulations never fall asleep.
Use it when: Default for robot joints; use drive rows when effort limits must hold exactly.

The script

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

30_pd_drives.py
replay.py

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

"""Joint PD drives: position targets (target_ke, target_kd) and velocity targets (target_kd)."""

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
GAINS = [(50.0, 1.0), (50.0, 10.0), (400.0, 20.0)]  # (ke, kd) of the three position-driven arms


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


def arm(builder, y, **drive):
    link = builder.add_link(xform=tf(0.2, y, 0.6))
    builder.add_shape_box(
        link, hx=0.2, hy=0.02, hz=0.02, cfg=newton.ModelBuilder.ShapeConfig(density=400.0)
    )
    builder.add_shape_sphere(
        link,
        xform=tf(0.2, 0.0, 0.0),
        radius=0.04,
        cfg=newton.ModelBuilder.ShapeConfig(density=2000.0),
    )
    j = builder.add_joint_revolute(
        -1,
        link,
        axis=(0.0, -1.0, 0.0),
        parent_xform=tf(0.0, y, 0.6),
        child_xform=tf(-0.2, 0.0, 0.0),
        **drive,
    )
    builder.add_articulation([j])


def build():
    builder = newton.ModelBuilder()
    for i, (ke, kd) in enumerate(GAINS):
        arm(builder, 0.3 * i - 0.45, target_ke=ke, target_kd=kd)
    arm(builder, 0.45, target_kd=5.0)  # velocity drive: no stiffness
    return builder.finalize()


def simulate(rec, label):
    model = build()
    solver = SolverSteel(model)
    pipeline = newton.CollisionPipeline(model)
    contacts = pipeline.contacts()
    state_0, state_1, control = model.state(), model.state(), model.control()

    run = rec.variant(label, model)
    for frame in range(rec.frames(4.0)):
        t = frame / rec.fps
        q = np.degrees(state_0.joint_q.numpy())
        run.frame(
            state_0,
            **{f"ke{ke:.0f}_kd{kd:.0f}": q[i] for i, (ke, kd) in enumerate(GAINS)},
            spin_rate=state_0.joint_qd.numpy()[3],
        )
        # step the position target to 45 degrees at 0.5 s; spin the last arm at 2 rad/s after 1 s
        control.joint_target_q.assign(
            np.array([math.radians(45.0) if t >= 0.5 else 0.0] * 3 + [0.0], dtype=np.float32)
        )
        control.joint_target_qd.assign(
            np.array([0.0, 0.0, 0.0, 2.0 if t >= 1.0 else 0.0], dtype=np.float32)
        )
        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__":
    rec = Recorder()
    simulate(rec, "three PD gains and one velocity drive")
    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.