newton.actuators.InputProcessorDelay#

class newton.actuators.InputProcessorDelay(delay_steps, max_delay=None)[source]#

Bases: InputProcessorBase

Per-DOF command input delay for actuators.

Delays command inputs (control targets and feedforward terms) using a circular buffer of depth max_delay. Each DOF has its own lag stored in delay_steps (shape (N,)). The buffer is sized for the maximum lag across all DOFs so that DOFs with different delay steps can share the same actuator group.

The delay always produces output. When the buffer is empty (e.g. right after reset) or a DOF has delay_steps == 0, the current command inputs are used directly. When underfilled, the lag is clamped to the available history so the oldest available entry is returned.

classmethod resolve_arguments(args)#

Resolve user-provided arguments with defaults.

Parameters:

args (dict[str, Any]) – User-provided arguments.

Returns:

Complete arguments with defaults filled in.

Return type:

dict[str, Any]

__init__(delay_steps, max_delay=None)#

Initialize delay.

Parameters:

  • delay_steps (wp.array[wp.int32]) – Per-DOF delay values [actuator timesteps], shape (N,).

  • max_delay (int | None) – Maximum delay across all DOFs. Determines the circular-buffer depth. None uses the largest entry of delay_steps.

Raises:

ValueError – If max_delay < 1.

finalize(device, num_actuators, requires_grad=False)#

Called by Actuator after construction.

Parameters:

  • device (Device) – Warp device to use.

  • num_actuators (int) – Number of actuators (DOFs).

  • requires_grad (bool) – Allocate output arrays with gradient support.

get_delayed_targets(target_pos, target_vel, feedforward, pos_indices, vel_indices, current_state)#

Read per-DOF delayed command inputs from the circular buffer.

Each DOF reads from its own lag offset stored in delay_steps, clamped to available history (per-DOF num_pushes). When the buffer is empty, falls back to the current command inputs; when underfilled, the lag is clamped to the oldest available entry.

Parameters:

  • target_pos (wp.array[wp.float32]) – Current target positions [m or rad].

  • target_vel (wp.array[wp.float32]) – Current target velocities [m/s or rad/s].

  • feedforward (wp.array[wp.float32] | None) – Feedforward control input [N or N·m] (may be None).

  • pos_indices (wp.array[wp.uint32]) – Indices into target_pos for each DOF.

  • vel_indices (wp.array[wp.uint32]) – Indices into target_vel and feedforward for each DOF.

  • current_state (State) – InputProcessorDelay state to read from.

Returns:

(delayed_pos, delayed_vel, delayed_feedforward). When feedforward is None, delayed_feedforward is all zeros.

Return type:

tuple[wp.array[wp.float32], wp.array[wp.float32], wp.array[wp.float32]]

is_stateful()#
process(inputs, state, dt)#

Replace the command inputs with their delayed values.

Parameters:

  • inputs (Inputs) – Actuator inputs.

  • state (State) – InputProcessorDelay state to read from.

  • dt (float | None) – Timestep [s] (unused).

Returns:

inputs with delayed target_pos, target_vel and feedforward.

Return type:

Inputs

state(num_actuators, device)#

Create a new delay state with zeroed circular buffers.

Parameters:

  • num_actuators (int) – Number of actuators (buffer width N).

  • device (Device) – Warp device for buffer allocation.

Returns:

Freshly allocated InputProcessorDelay.State.

Return type:

State

update_state(*args, **kwargs)#

Push the current command inputs into the buffer.

Call as update_state(inputs, current_state, next_state):

  • inputs: the inputs this delay received in process().

  • current_state: delay state to read from.

  • next_state: delay state to write into.

Deprecated since version 1.9: The form update_state(target_pos, target_vel, feedforward, pos_indices, vel_indices, current_state, next_state).

buf_depth#

Circular-buffer depth (equals max_delay).

delay_steps#

Per-DOF delay values [actuator timesteps], shape (N,).