# Copyright (c) 2022-2026, The Isaac Lab Project Developers (https://github.com/isaac-sim/IsaacLab/blob/main/CONTRIBUTORS.md).
# All rights reserved.
#
# SPDX-License-Identifier: BSD-3-Clause
# Flag for pyright to ignore type errors in this file.
# pyright: reportPrivateUsage=false
from __future__ import annotations
import logging
import re
import warnings
from collections.abc import Sequence
from typing import Any
import numpy as np
import torch
import warp as wp
from pxr import Usd, UsdPhysics
import isaaclab.sim as sim_utils
from isaaclab.assets.articulation import ordering_kernels
from isaaclab.assets.articulation.articulation_cfg import ArticulationCfg
from isaaclab.assets.articulation.base_articulation import BaseArticulation
from isaaclab.assets.articulation.ordering_resolvers import (
_BODY_KIND,
_JOINT_KIND,
_canonical_joint_dof_name,
)
from isaaclab.physics import PhysicsManager
from isaaclab.utils.buffers import TimestampedBufferWarp
from isaaclab.utils.string import resolve_matching_names
from isaaclab.utils.warp import ProxyArray
from isaaclab.utils.wrench_composer import WrenchComposer
from isaaclab_ov import tensor_types as TT
from isaaclab_ov.assets import kernels as shared_kernels
from isaaclab_ov.physics import OvPhysxManager
from isaaclab_ov.sim.views.ovphysx_view import OvPhysxView
from .articulation_data import ArticulationData
from .kernels import (
clamp_default_joint_pos_and_update_soft_limits_index_kernel,
clamp_default_joint_pos_and_update_soft_limits_mask,
update_soft_joint_pos_limits,
write_joint_friction_data_to_buffer_index_kernel,
write_joint_friction_data_to_buffer_mask,
write_joint_position_with_sim_ids_kernel,
write_joint_state_with_sim_ids_kernel,
write_joint_velocity_with_sim_ids_kernel,
)
# import logger
logger = logging.getLogger(__name__)
[docs]
class Articulation(BaseArticulation):
"""An articulation asset class.
An articulation is a collection of rigid bodies connected by joints. The joints can be either
fixed or actuated. The joints can be of different types, such as revolute, prismatic, D-6, etc.
However, the articulation class has currently been tested with revolute and prismatic joints.
The class supports both floating-base and fixed-base articulations. The type of articulation
is determined based on the root joint of the articulation. If the root joint is fixed, then
the articulation is considered a fixed-base system. Otherwise, it is considered a floating-base
system. This can be checked using the :attr:`Articulation.is_fixed_base` attribute.
For an asset to be considered an articulation, the root prim of the asset must have the
`USD ArticulationRootAPI`_. This API is used to define the sub-tree of the articulation using
the reduced coordinate formulation. On playing the simulation, the physics engine parses the
articulation root prim and creates the corresponding articulation in the physics engine. The
articulation root prim can be specified using the :attr:`AssetBaseCfg.prim_path` attribute.
OVPhysX exposes per-tensor-type :class:`ovphysx.TensorBinding` objects rather than a single
opaque view; binding handles are created eagerly in :meth:`_initialize_impl` and reused across
reads and writes. CPU-only bindings (mass, CoM, inertia, joint properties, tendon properties)
are routed through pinned-host staging buffers managed by :class:`ArticulationData`.
.. _`USD ArticulationRootAPI`: https://openusd.org/dev/api/class_usd_physics_articulation_root_a_p_i.html
"""
cfg: ArticulationCfg
"""Configuration instance for the articulation."""
__backend_name__: str = "ovphysx"
"""The name of the backend for the articulation."""
__backend_native_orderings__: tuple[str, ...] = ("physx",)
"""OVPhysX tensor-view order already matches the ``"physx"`` convention."""
[docs]
def __init__(self, cfg: ArticulationCfg):
"""Initialize the articulation.
Args:
cfg: A configuration instance.
"""
super().__init__(cfg)
# the binding manager is created in ``_initialize_impl``; it owns all
# TensorBinding creation, caching, and the CPU/GPU device policy.
self._root_view: OvPhysxView | None = None
"""
Properties
"""
@property
def data(self) -> ArticulationData:
return self._data
@property
def num_instances(self) -> int:
return self._num_instances
@property
def is_fixed_base(self) -> bool:
"""Whether the articulation is a fixed-base or floating-base system."""
return self._is_fixed_base
@property
def num_joints(self) -> int:
"""Number of joints in articulation."""
return self._num_joints
@property
def num_fixed_tendons(self) -> int:
"""Number of fixed tendons in articulation."""
return self._num_fixed_tendons
@property
def num_spatial_tendons(self) -> int:
"""Number of spatial tendons in articulation."""
return self._num_spatial_tendons
@property
def num_bodies(self) -> int:
"""Number of bodies in articulation."""
return self._num_bodies
@property
def fixed_tendon_names(self) -> list[str]:
"""Ordered names of fixed tendons in articulation."""
return self._fixed_tendon_names
@property
def spatial_tendon_names(self) -> list[str]:
"""Ordered names of spatial tendons in articulation."""
return self._spatial_tendon_names
@property
def backend_joint_names(self) -> list[str]:
"""Ordered names of joints as exposed by the active backend."""
return self._joint_names
@property
def backend_body_names(self) -> list[str]:
"""Ordered names of bodies as exposed by the active backend."""
return self._body_names
@property
def root_view(self) -> OvPhysxView:
"""Root view for the asset.
On OVPhysX this is an :class:`~isaaclab_ov.sim.views.OvPhysxView`: a
string-keyed binding manager over the per-tensor-type ``TensorBinding`` handles,
rather than the single opaque view object used by the PhysX and Newton backends.
Address attributes by their lowercased ``TensorType`` name (e.g.
``root_view.get_attribute("articulation_dof_stiffness")``) or by the
:class:`~isaaclab_ov.tensor_types.TensorType` member itself. For high-level
state access (instance counts, prim paths, transforms), prefer the
:attr:`num_instances`, :attr:`body_names`, and
:attr:`~ArticulationData.root_link_pose_w` accessors instead.
.. note::
Use this view with caution. It requires handling of tensors in a specific way.
"""
return self._root_view
@property
def instantaneous_wrench_composer(self) -> WrenchComposer:
"""Instantaneous wrench composer.
Returns a :class:`~isaaclab.utils.wrench_composer.WrenchComposer` instance. Wrenches added or set to this wrench
composer are only valid for the current simulation step. At the end of the simulation step, the wrenches set
to this object are discarded. This is useful to apply forces that change all the time, things like drag forces
for instance.
"""
return self._instantaneous_wrench_composer
@property
def permanent_wrench_composer(self) -> WrenchComposer:
"""Permanent wrench composer.
Returns a :class:`~isaaclab.utils.wrench_composer.WrenchComposer` instance. Wrenches added or set to this wrench
composer are persistent and are applied to the simulation at every step. This is useful to apply forces that
are constant over a period of time, things like the thrust of a motor for instance.
"""
return self._permanent_wrench_composer
"""
Operations.
"""
def reset(
self, env_ids: Sequence[int] | torch.Tensor | wp.array | None = None, env_mask: wp.array | None = None
) -> None:
"""Reset the articulation.
.. caution::
If both `env_ids` and `env_mask` are provided, then `env_mask` takes precedence over `env_ids`.
Args:
env_ids: Environment indices. If None, then all indices are used.
env_mask: Environment mask. If None, then all the instances are updated. Shape is (num_instances,).
"""
if (env_ids is None) or (env_ids == slice(None)):
env_ids = slice(None)
# reset external wrenches.
self._instantaneous_wrench_composer.reset(env_ids, env_mask)
self._permanent_wrench_composer.reset(env_ids, env_mask)
def write_data_to_sim(self) -> None:
"""Write external wrenches and joint commands to the simulation.
If any explicit actuators are present, then the actuator models are used to compute the
joint commands. Otherwise, the joint commands are directly set into the simulation.
.. note::
We write external wrench to the simulation here since this function is called before the simulation step.
This ensures that the external wrench is applied at every simulation step.
"""
# write external wrench
inst = self._instantaneous_wrench_composer
perm = self._permanent_wrench_composer
if inst.active or perm.active:
if inst.active:
if perm.active:
inst.add_raw_buffers_from(perm)
force_b = inst.out_force_b.warp
torque_b = inst.out_torque_b.warp
else:
force_b = perm.out_force_b.warp
torque_b = perm.out_torque_b.warp
# rotate body-frame wrenches into the world frame expected by ``LINK_WRENCH``.
# Read the link poses directly from the backend-order ``LINK_POSE`` buffer: the
# kernel indexes them in backend order (same physical body as the public wrench),
# so no public-order pose shadow refresh / reorder launch is needed here.
poses = self._data._backend_body_link_pose_w
has_body_ordering = self.data.has_body_ordering
wp.launch(
shared_kernels._body_wrench_to_world_ordered,
dim=(self._num_instances, self._num_bodies),
inputs=[force_b, torque_b, poses, self._body_user_to_backend_map(), has_body_ordering],
outputs=[self._wrench_buf],
device=self._device,
)
if self._get_binding(TT.LINK_WRENCH) is not None:
self._root_view.set_attribute(TT.LINK_WRENCH, self._wrench_buf)
if inst.active:
inst.reset()
# apply actuator models
self._apply_actuator_model()
# write actions into simulation (zeros are safe when no actuators are active).
# ``_applied_torque`` is the actuator-computed output (may differ from the raw
# commanded target, e.g. once clipped), so it must be reordered into its own
# scratch buffer rather than ``_joint_effort_target_backend``. The latter is the
# persistent mirror of the raw target that partial writes rely on for their
# unselected joints (see ``set_joint_effort_target_index``/``_mask``).
write_effort = self._can_write_effort
# position and velocity targets only for implicit actuators
write_pos = self._has_implicit_actuators and self._can_write_pos_target
write_vel = self._has_implicit_actuators and self._can_write_vel_target
if self.data.has_joint_ordering:
if write_effort or write_pos or write_vel:
# One fused gather replaces the per-target reorder launches. The
# fourth joint-acceleration output is disabled.
wp.launch(
ordering_kernels.reorder_joint_targets_user_to_backend,
dim=(self._num_instances, self._num_joints),
inputs=[
self._data._applied_torque,
self._data._joint_pos_target,
self._data._joint_vel_target,
self.data.joint_ordering.backend_to_user,
write_effort,
write_pos,
write_vel,
False,
],
outputs=[
self._applied_torque_backend,
self._joint_pos_target_backend,
self._joint_vel_target_backend,
None,
],
device=self._device,
)
effort = self._applied_torque_backend
pos_target = self._joint_pos_target_backend
vel_target = self._joint_vel_target_backend
else:
effort = self._data._applied_torque
pos_target = self._data._joint_pos_target
vel_target = self._data._joint_vel_target
if write_effort:
self._root_view.set_attribute(TT.DOF_ACTUATION_FORCE, effort)
if write_pos:
self._root_view.set_attribute(TT.DOF_POSITION_TARGET, pos_target)
if write_vel:
self._root_view.set_attribute(TT.DOF_VELOCITY_TARGET, vel_target)
def update(self, dt: float) -> None:
"""Updates the simulation data.
Args:
dt: The time step size in seconds.
"""
self._data.update(dt)
"""
Operations - Finders.
"""
def find_bodies(
self,
name_keys: str | Sequence[str],
preserve_order: bool = False,
*,
as_proxy: bool = False,
) -> tuple[list[int] | ProxyArray, list[str]]:
"""Find bodies in the articulation based on the name keys.
Please check the :func:`isaaclab.utils.string.resolve_matching_names` function for more
information on the name matching.
Args:
name_keys: A regular expression or a list of regular expressions to match the body names.
preserve_order: Whether to preserve the order of the name keys in the output. Defaults to False.
as_proxy: Whether to return cached proxy indices. Defaults to False.
Returns:
Matched body indices and names.
"""
body_ids, body_names = resolve_matching_names(name_keys, self.body_names, preserve_order)
resolved_ids = self._resolve_finder_indices(body_ids, domain="body", as_proxy=as_proxy, legacy_type="list")
return resolved_ids, body_names
def find_joints(
self,
name_keys: str | Sequence[str],
joint_subset: list[str] | None = None,
preserve_order: bool = False,
*,
as_proxy: bool = False,
) -> tuple[list[int] | ProxyArray, list[str]]:
"""Find joints in the articulation based on the name keys.
Please see the :func:`isaaclab.utils.string.resolve_matching_names` function for more information
on the name matching.
Args:
name_keys: A regular expression or a list of regular expressions to match the joint names.
joint_subset: A subset of joints to search for. Defaults to None, which means all joints
in the articulation are searched.
preserve_order: Whether to preserve the order of the name keys in the output. Defaults to False.
as_proxy: Whether to return cached proxy indices. Subset searches use asset-global proxy indices.
Defaults to False.
Returns:
Matched joint indices and names.
"""
if joint_subset is None:
joint_subset = self.joint_names
# find joints
joint_ids, joint_names = resolve_matching_names(name_keys, joint_subset, preserve_order)
proxy_joint_ids = [self.joint_names.index(name) for name in joint_names]
resolved_ids = self._resolve_finder_indices(
joint_ids,
domain="joint",
proxy_indices=proxy_joint_ids,
as_proxy=as_proxy,
legacy_type="list",
)
return resolved_ids, joint_names
def find_fixed_tendons(
self,
name_keys: str | Sequence[str],
tendon_subsets: list[str] | None = None,
preserve_order: bool = False,
*,
as_proxy: bool = False,
) -> tuple[list[int] | ProxyArray, list[str]]:
"""Find fixed tendons in the articulation based on the name keys.
Please see the :func:`isaaclab.utils.string.resolve_matching_names` function for more information
on the name matching.
Args:
name_keys: A regular expression or a list of regular expressions to match the
joint names with fixed tendons.
tendon_subsets: A subset of joints with fixed tendons to search for. Defaults to None, which means
all joints in the articulation are searched.
preserve_order: Whether to preserve the order of the name keys in the output. Defaults to False.
as_proxy: Whether to return cached proxy indices. Subset searches use asset-global proxy indices.
Defaults to False.
Returns:
Matched fixed-tendon indices and names.
"""
if tendon_subsets is None:
# tendons follow the joint names they are attached to
tendon_subsets = self.fixed_tendon_names
# find tendons
tendon_ids, tendon_names = resolve_matching_names(name_keys, tendon_subsets, preserve_order)
proxy_tendon_ids = [self.fixed_tendon_names.index(name) for name in tendon_names]
resolved_ids = self._resolve_finder_indices(
tendon_ids,
domain="fixed_tendon",
proxy_indices=proxy_tendon_ids,
as_proxy=as_proxy,
legacy_type="list",
)
return resolved_ids, tendon_names
def find_spatial_tendons(
self,
name_keys: str | Sequence[str],
tendon_subsets: list[str] | None = None,
preserve_order: bool = False,
*,
as_proxy: bool = False,
) -> tuple[list[int] | ProxyArray, list[str]]:
"""Find spatial tendons in the articulation based on the name keys.
Please see the :func:`isaaclab.utils.string.resolve_matching_names` function for more information
on the name matching.
Args:
name_keys: A regular expression or a list of regular expressions to match the tendon names.
tendon_subsets: A subset of tendons to search for. Defaults to None, which means all tendons
in the articulation are searched.
preserve_order: Whether to preserve the order of the name keys in the output. Defaults to False.
as_proxy: Whether to return cached proxy indices. Subset searches use asset-global proxy indices.
Defaults to False.
Returns:
Matched spatial-tendon indices and names.
"""
if tendon_subsets is None:
tendon_subsets = self.spatial_tendon_names
# find tendons
tendon_ids, tendon_names = resolve_matching_names(name_keys, tendon_subsets, preserve_order)
proxy_tendon_ids = [self.spatial_tendon_names.index(name) for name in tendon_names]
resolved_ids = self._resolve_finder_indices(
tendon_ids,
domain="spatial_tendon",
proxy_indices=proxy_tendon_ids,
as_proxy=as_proxy,
legacy_type="list",
)
return resolved_ids, tendon_names
"""
Operations - State Writers.
"""
def write_root_pose_to_sim_index(
self,
*,
root_pose: torch.Tensor | wp.array,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root pose over selected environment indices into the simulation.
The root pose comprises of the cartesian position and quaternion orientation in (x, y, z, w).
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_pose: Root poses in simulation frame. Shape is (len(env_ids), 7)
or (len(env_ids),) with dtype wp.transformf.
env_ids: Environment indices. If None, then all indices are used.
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
self.write_root_link_pose_to_sim_index(root_pose=root_pose, env_ids=env_ids, skip_forward=skip_forward)
def write_root_pose_to_sim_mask(
self,
*,
root_pose: torch.Tensor | wp.array,
env_mask: wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root pose over selected environment mask into the simulation.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_pose: Root poses in simulation frame. Shape is (num_instances, 7)
or (num_instances,) with dtype wp.transformf.
env_mask: Environment mask. If None, then all the instances are updated. Shape is (num_instances,).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
self.write_root_link_pose_to_sim_mask(root_pose=root_pose, env_mask=env_mask, skip_forward=skip_forward)
def write_root_link_pose_to_sim_index(
self,
*,
root_pose: torch.Tensor | wp.array,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root link pose over selected environment indices into the simulation.
The root pose comprises of the cartesian position and quaternion orientation in (x, y, z, w).
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_pose: Root link poses in simulation frame. Shape is (len(env_ids), 7)
or (len(env_ids),) with dtype wp.transformf.
env_ids: Environment indices. If None, then all indices are used.
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
env_ids = self._resolve_env_ids(env_ids)
self.assert_shape_and_dtype(root_pose, (env_ids.shape[0],), wp.transformf, "root_pose")
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.set_root_link_pose_to_sim_index_kernel(env_ids),
dim=env_ids.shape[0],
inputs=[root_pose, env_ids],
outputs=[self.data.root_link_pose_w, sim_env_ids],
device=self._device,
)
# Let the data class handle the invalidation of pose-dependent properties.
if not skip_forward:
self.data._reset_pose()
self._root_view.set_attribute(
TT.ROOT_POSE, self.data._root_link_pose_w.data.view(wp.float32), indices=sim_env_ids
)
def write_root_link_pose_to_sim_mask(
self,
*,
root_pose: torch.Tensor | wp.array,
env_mask: wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root link pose over selected environment mask into the simulation.
The root pose comprises of the cartesian position and quaternion orientation in (x, y, z, w).
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_pose: Root poses in simulation frame. Shape is (num_instances, 7)
or (num_instances,) with dtype wp.transformf.
env_mask: Environment mask. If None, then all the instances are updated. Shape is (num_instances,).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
env_mask_wp = self._resolve_env_mask(env_mask)
self.assert_shape_and_dtype(root_pose, (self._num_instances,), wp.transformf, "root_pose")
wp.launch(
shared_kernels.set_root_link_pose_to_sim_mask,
dim=self._num_instances,
inputs=[root_pose, env_mask_wp],
outputs=[self.data.root_link_pose_w],
device=self._device,
)
# Let the data class handle the invalidation of pose-dependent properties.
if not skip_forward:
self.data._reset_pose()
self._root_view.set_attribute(TT.ROOT_POSE, self.data._root_link_pose_w.data.view(wp.float32), mask=env_mask_wp)
def write_root_com_pose_to_sim_index(
self,
*,
root_pose: torch.Tensor | wp.array,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root center of mass pose over selected environment indices into the simulation.
The root pose comprises of the cartesian position and quaternion orientation in (x, y, z, w).
The orientation is the orientation of the principal axes of inertia.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_pose: Root center of mass poses in simulation frame. Shape is (len(env_ids), 7)
or (len(env_ids),) with dtype wp.transformf.
env_ids: Environment indices. If None, then all indices are used.
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
env_ids = self._resolve_env_ids(env_ids)
self.assert_shape_and_dtype(root_pose, (env_ids.shape[0],), wp.transformf, "root_pose")
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.set_root_com_pose_to_sim_index_kernel(env_ids),
dim=env_ids.shape[0],
inputs=[root_pose, self.data._backend_body_com_pose_b, env_ids],
outputs=[self.data.root_com_pose_w, self.data.root_link_pose_w, sim_env_ids],
device=self._device,
)
# Let the data class handle the invalidation of pose-dependent properties.
if not skip_forward:
self.data._reset_pose(from_link=False)
self._root_view.set_attribute(
TT.ROOT_POSE, self.data._root_link_pose_w.data.view(wp.float32), indices=sim_env_ids
)
def write_root_com_pose_to_sim_mask(
self,
*,
root_pose: torch.Tensor | wp.array,
env_mask: wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root center of mass pose over selected environment mask into the simulation.
The root pose comprises of the cartesian position and quaternion orientation in (x, y, z, w).
The orientation is the orientation of the principal axes of inertia.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_pose: Root center of mass poses in simulation frame. Shape is (num_instances, 7)
or (num_instances,) with dtype wp.transformf.
env_mask: Environment mask. If None, then all the instances are updated. Shape is (num_instances,).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
env_mask_wp = self._resolve_env_mask(env_mask)
self.assert_shape_and_dtype(root_pose, (self._num_instances,), wp.transformf, "root_pose")
wp.launch(
shared_kernels.set_root_com_pose_to_sim_mask,
dim=self._num_instances,
inputs=[root_pose, self.data._backend_body_com_pose_b, env_mask_wp],
outputs=[self.data.root_com_pose_w, self.data.root_link_pose_w],
device=self._device,
)
# Let the data class handle the invalidation of pose-dependent properties.
if not skip_forward:
self.data._reset_pose(from_link=False)
self._root_view.set_attribute(TT.ROOT_POSE, self.data._root_link_pose_w.data.view(wp.float32), mask=env_mask_wp)
def write_root_velocity_to_sim_index(
self,
*,
root_velocity: torch.Tensor | wp.array,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root center of mass velocity over selected environment indices into the simulation.
The velocity comprises linear velocity (x, y, z) and angular velocity (x, y, z) in that order.
.. note::
This sets the velocity of the root's center of mass rather than the root's frame.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_velocity: Root center of mass velocities in simulation world frame. Shape is (len(env_ids), 6)
or (len(env_ids),) with dtype wp.spatial_vectorf.
env_ids: Environment indices. If None, then all indices are used.
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
self.write_root_com_velocity_to_sim_index(
root_velocity=root_velocity, env_ids=env_ids, skip_forward=skip_forward
)
def write_root_velocity_to_sim_mask(
self,
*,
root_velocity: torch.Tensor | wp.array,
env_mask: wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root center of mass velocity over selected environment mask into the simulation.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_velocity: Root center of mass velocities in simulation world frame. Shape is (num_instances, 6)
or (num_instances,) with dtype wp.spatial_vectorf.
env_mask: Environment mask. If None, then all the instances are updated. Shape is (num_instances,).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
self.write_root_com_velocity_to_sim_mask(
root_velocity=root_velocity, env_mask=env_mask, skip_forward=skip_forward
)
def write_root_com_velocity_to_sim_index(
self,
*,
root_velocity: torch.Tensor | wp.array,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root center of mass velocity over selected environment indices into the simulation.
The velocity comprises linear velocity (x, y, z) and angular velocity (x, y, z) in that order.
.. note::
This sets the velocity of the root's center of mass rather than the root's frame.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_velocity: Root center of mass velocities in simulation world frame. Shape is (len(env_ids), 6)
or (len(env_ids),) with dtype wp.spatial_vectorf.
env_ids: Environment indices. If None, then all indices are used.
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
env_ids = self._resolve_env_ids(env_ids)
self.assert_shape_and_dtype(root_velocity, (env_ids.shape[0],), wp.spatial_vectorf, "root_velocity")
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.set_root_com_velocity_to_sim_index_kernel(env_ids),
dim=env_ids.shape[0],
inputs=[root_velocity, env_ids, self._num_bodies],
outputs=[self.data.root_com_vel_w, self.data.body_com_acc_w, sim_env_ids],
device=self._device,
)
# Let the data class handle the invalidation of velocity-dependent properties.
if not skip_forward:
self.data._reset_velocity()
self._root_view.set_attribute(
TT.ROOT_VELOCITY, self.data._root_com_vel_w.data.view(wp.float32), indices=sim_env_ids
)
def write_root_com_velocity_to_sim_mask(
self,
*,
root_velocity: torch.Tensor | wp.array,
env_mask: wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root center of mass velocity over selected environment mask into the simulation.
The velocity comprises linear velocity (x, y, z) and angular velocity (x, y, z) in that order.
.. note::
This sets the velocity of the root's center of mass rather than the root's frame.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_velocity: Root center of mass velocities in simulation world frame. Shape is (num_instances, 6)
or (num_instances,) with dtype wp.spatial_vectorf.
env_mask: Environment mask. If None, then all the instances are updated. Shape is (num_instances,).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
env_mask_wp = self._resolve_env_mask(env_mask)
self.assert_shape_and_dtype(root_velocity, (self._num_instances,), wp.spatial_vectorf, "root_velocity")
wp.launch(
shared_kernels.set_root_com_velocity_to_sim_mask,
dim=self._num_instances,
inputs=[root_velocity, env_mask_wp, self._num_bodies],
outputs=[self.data.root_com_vel_w, self.data.body_com_acc_w],
device=self._device,
)
# Let the data class handle the invalidation of velocity-dependent properties.
if not skip_forward:
self.data._reset_velocity()
self._root_view.set_attribute(
TT.ROOT_VELOCITY, self.data._root_com_vel_w.data.view(wp.float32), mask=env_mask_wp
)
def write_root_link_velocity_to_sim_index(
self,
*,
root_velocity: torch.Tensor | wp.array,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root link velocity over selected environment indices into the simulation.
The velocity comprises linear velocity (x, y, z) and angular velocity (x, y, z) in that order.
.. note::
This sets the velocity of the root's frame rather than the root's center of mass.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_velocity: Root frame velocities in simulation world frame. Shape is (len(env_ids), 6)
or (len(env_ids),) with dtype wp.spatial_vectorf.
env_ids: Environment indices. If None, then all indices are used.
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
env_ids = self._resolve_env_ids(env_ids)
self.assert_shape_and_dtype(root_velocity, (env_ids.shape[0],), wp.spatial_vectorf, "root_velocity")
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.set_root_link_velocity_to_sim_index_kernel(env_ids),
dim=env_ids.shape[0],
inputs=[
root_velocity,
self.data._backend_body_com_pose_b,
self.data.root_link_pose_w,
env_ids,
self._num_bodies,
],
outputs=[self.data.root_link_vel_w, self.data.root_com_vel_w, self.data.body_com_acc_w, sim_env_ids],
device=self._device,
)
# Let the data class handle the invalidation of velocity-dependent properties.
if not skip_forward:
self.data._reset_velocity(from_com=False)
self._root_view.set_attribute(
TT.ROOT_VELOCITY, self.data._root_com_vel_w.data.view(wp.float32), indices=sim_env_ids
)
def write_root_link_velocity_to_sim_mask(
self,
*,
root_velocity: torch.Tensor | wp.array,
env_mask: wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set the root link velocity over selected environment mask into the simulation.
The velocity comprises linear velocity (x, y, z) and angular velocity (x, y, z) in that order.
.. note::
This sets the velocity of the root's frame rather than the root's center of mass.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations. Performance is similar for both.
However, to allow graphed pipelines, the mask method must be used.
Args:
root_velocity: Root frame velocities in simulation world frame. Shape is (num_instances, 6)
or (num_instances,) with dtype wp.spatial_vectorf.
env_mask: Environment mask. If None, then all the instances are updated. Shape is (num_instances,).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
env_mask_wp = self._resolve_env_mask(env_mask)
self.assert_shape_and_dtype(root_velocity, (self._num_instances,), wp.spatial_vectorf, "root_velocity")
wp.launch(
shared_kernels.set_root_link_velocity_to_sim_mask,
dim=self._num_instances,
inputs=[
root_velocity,
self.data._backend_body_com_pose_b,
self.data.root_link_pose_w,
env_mask_wp,
self._num_bodies,
],
outputs=[self.data.root_link_vel_w, self.data.root_com_vel_w, self.data.body_com_acc_w],
device=self._device,
)
# Let the data class handle the invalidation of velocity-dependent properties.
if not skip_forward:
self.data._reset_velocity(from_com=False)
self._root_view.set_attribute(
TT.ROOT_VELOCITY, self.data._root_com_vel_w.data.view(wp.float32), mask=env_mask_wp
)
def write_joint_state_to_sim_index(
self,
*,
position: torch.Tensor | wp.array,
velocity: torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set joint positions and velocities over selected indices into the simulation.
The joint position and velocity caches, finite-difference velocity baseline, and
joint acceleration cache are updated in one device kernel.
.. note::
This method expects partial data.
Args:
position: Joint positions [m or rad, depending on joint type]. Shape is
(len(env_ids), len(joint_ids)) with dtype wp.float32.
velocity: Joint velocities [m/s or rad/s, depending on joint type]. Shape is
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
joint_selection_is_partial = joint_ids is not None
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
expected_shape = (env_ids.shape[0], joint_ids.shape[0])
self.assert_shape_and_dtype(position, expected_shape, wp.float32, "position")
self.assert_shape_and_dtype(velocity, expected_shape, wp.float32, "velocity")
if env_ids.shape[0] == 0 or joint_ids.shape[0] == 0:
return
joint_pos_backend = self._data._get_joint_pos_write_buffer(joint_selection_is_partial)
joint_vel_backend = self._data._get_joint_vel_write_buffer(joint_selection_is_partial)
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
write_joint_state_with_sim_ids_kernel(env_ids, joint_ids),
dim=expected_shape,
inputs=[
position,
velocity,
env_ids,
joint_ids,
self._joint_user_to_backend_map(),
self.data.has_joint_ordering,
],
outputs=[
self._data._joint_pos_buf.data,
self._data._joint_vel_buf.data,
self._data._previous_joint_vel,
self._data._joint_acc.data,
joint_pos_backend,
joint_vel_backend,
sim_env_ids,
],
device=self._device,
)
self._data._joint_acc.timestamp = self._data._sim_timestamp
if not skip_forward:
self._data._reset_pose()
self._data._reset_velocity()
self._root_view.set_attribute(TT.DOF_POSITION, joint_pos_backend, indices=sim_env_ids)
self._root_view.set_attribute(TT.DOF_VELOCITY, joint_vel_backend, indices=sim_env_ids)
def write_joint_position_to_sim_index(
self,
*,
position: torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set joint positions over selected env / joint indices into the simulation.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
position: Joint positions [m or rad, depending on joint type]. Shape is
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
joint_selection_is_partial = joint_ids is not None
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
self.assert_shape_and_dtype(position, (env_ids.shape[0], joint_ids.shape[0]), wp.float32, "position")
if env_ids.shape[0] == 0 or joint_ids.shape[0] == 0:
return
joint_pos_backend = self._data._get_joint_pos_write_buffer(joint_selection_is_partial)
has_joint_ordering = self.data.has_joint_ordering
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
write_joint_position_with_sim_ids_kernel(env_ids, joint_ids),
dim=(env_ids.shape[0], joint_ids.shape[0]),
inputs=[position, env_ids, joint_ids, self._joint_user_to_backend_map(), has_joint_ordering],
outputs=[self._data._joint_pos_buf.data, joint_pos_backend, sim_env_ids],
device=self._device,
)
# Let the data class handle the invalidation of pose- and velocity-dependent properties.
# A position write does not change joint velocities, so the joint acceleration is unaffected.
if not skip_forward:
self._data._reset_pose()
self._data._reset_velocity()
self._root_view.set_attribute(TT.DOF_POSITION, joint_pos_backend, indices=sim_env_ids)
def write_joint_position_to_sim_mask(
self,
*,
position: torch.Tensor | wp.array,
env_mask: wp.array | None = None,
joint_mask: wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set joint positions over selected env / joint masks into the simulation.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
position: Joint positions [m or rad, depending on joint type]. Shape is
(num_instances, num_joints) with dtype wp.float32.
env_mask: Environment mask. If None, all instances are updated. Shape is
(num_instances,).
joint_mask: Joint mask. If None, all joints are updated. Shape is
(num_joints,).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
joint_selection_is_partial = joint_mask is not None
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
self.assert_shape_and_dtype(position, (self._num_instances, self._num_joints), wp.float32, "position")
joint_pos_backend = self._data._get_joint_pos_write_buffer(joint_selection_is_partial)
has_joint_ordering = self.data.has_joint_ordering
ordering_kernels.write_float_user_to_backend_with_mask(
position,
env_mask_wp,
joint_mask_wp,
self._joint_user_to_backend_map(),
has_joint_ordering,
self._data._joint_pos_buf.data,
joint_pos_backend,
device=self._device,
)
# Let the data class handle the invalidation of pose- and velocity-dependent properties.
# A position write does not change joint velocities, so the joint acceleration is unaffected.
if not skip_forward:
self._data._reset_pose()
self._data._reset_velocity()
self._root_view.set_attribute(TT.DOF_POSITION, joint_pos_backend, mask=env_mask_wp)
def write_joint_velocity_to_sim_index(
self,
*,
velocity: torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set joint velocities over selected env / joint indices into the simulation.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
velocity: Joint velocities [m/s or rad/s, depending on joint type]. Shape is
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
joint_selection_is_partial = joint_ids is not None
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
self.assert_shape_and_dtype(velocity, (env_ids.shape[0], joint_ids.shape[0]), wp.float32, "velocity")
if env_ids.shape[0] == 0 or joint_ids.shape[0] == 0:
return
joint_vel_backend = self._data._get_joint_vel_write_buffer(joint_selection_is_partial)
has_joint_ordering = self.data.has_joint_ordering
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
write_joint_velocity_with_sim_ids_kernel(env_ids, joint_ids),
dim=(env_ids.shape[0], joint_ids.shape[0]),
inputs=[velocity, env_ids, joint_ids, self._joint_user_to_backend_map(), has_joint_ordering],
outputs=[
self._data._joint_vel_buf.data,
self._data._previous_joint_vel,
self._data._joint_acc.data,
joint_vel_backend,
sim_env_ids,
],
device=self._device,
)
self._data._joint_acc.timestamp = self._data._sim_timestamp
if not skip_forward:
self._data._reset_velocity()
self._root_view.set_attribute(TT.DOF_VELOCITY, joint_vel_backend, indices=sim_env_ids)
def write_joint_velocity_to_sim_mask(
self,
*,
velocity: torch.Tensor | wp.array,
env_mask: wp.array | None = None,
joint_mask: wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Set joint velocities over selected env / joint masks into the simulation.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
velocity: Joint velocities [m/s or rad/s, depending on joint type]. Shape is
(num_instances, num_joints) with dtype wp.float32.
env_mask: Environment mask. If None, all instances are updated. Shape is
(num_instances,).
joint_mask: Joint mask. If None, all joints are updated. Shape is
(num_joints,).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
joint_selection_is_partial = joint_mask is not None
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
self.assert_shape_and_dtype(velocity, (self._num_instances, self._num_joints), wp.float32, "velocity")
joint_vel_backend = self._data._get_joint_vel_write_buffer(joint_selection_is_partial)
has_joint_ordering = self.data.has_joint_ordering
wp.launch(
ordering_kernels.write_joint_vel_user_to_backend_with_mask,
dim=(self._num_instances, self._num_joints),
inputs=[velocity, env_mask_wp, joint_mask_wp, self._joint_user_to_backend_map(), has_joint_ordering],
outputs=[
self._data._joint_vel_buf.data,
self._data._previous_joint_vel,
self._data._joint_acc.data,
joint_vel_backend,
],
device=self._device,
)
self._data._joint_acc.timestamp = self._data._sim_timestamp
if not skip_forward:
self._data._reset_velocity()
self._root_view.set_attribute(TT.DOF_VELOCITY, joint_vel_backend, mask=env_mask_wp)
def write_joint_state_to_sim_mask(
self,
*,
position: torch.Tensor | wp.array,
velocity: torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
skip_forward: bool = False,
) -> None:
"""Write joint positions and velocities over selected environment mask into the simulation.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
position: Joint positions [m or rad, depending on joint type]. Shape is
(num_instances, num_joints) with dtype wp.float32.
velocity: Joint velocities [m/s or rad/s, depending on joint type]. Shape is
(num_instances, num_joints) with dtype wp.float32.
joint_mask: Joint mask. If None, all joints are updated. Shape is
(num_joints,).
env_mask: Environment mask. If None, all instances are updated. Shape is
(num_instances,).
skip_forward: Whether to skip invalidating cached data after the write. When True, the caller
must invalidate stale cached data before reading it back. Defaults to False.
"""
joint_selection_is_partial = joint_mask is not None
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
self.assert_shape_and_dtype(position, (self._num_instances, self._num_joints), wp.float32, "position")
self.assert_shape_and_dtype(velocity, (self._num_instances, self._num_joints), wp.float32, "velocity")
joint_pos_backend = self._data._get_joint_pos_write_buffer(joint_selection_is_partial)
joint_vel_backend = self._data._get_joint_vel_write_buffer(joint_selection_is_partial)
has_joint_ordering = self.data.has_joint_ordering
wp.launch(
ordering_kernels.write_joint_state_user_to_backend_with_mask,
dim=(self._num_instances, self._num_joints),
inputs=[
position,
velocity,
env_mask_wp,
joint_mask_wp,
self._joint_user_to_backend_map(),
has_joint_ordering,
],
outputs=[
self._data._joint_pos_buf.data,
self._data._joint_vel_buf.data,
self._data._previous_joint_vel,
self._data._joint_acc.data,
joint_pos_backend,
joint_vel_backend,
],
device=self._device,
)
self._data._joint_acc.timestamp = self._data._sim_timestamp
if not skip_forward:
self._data._reset_pose()
self._data._reset_velocity()
self._root_view.set_attribute(TT.DOF_POSITION, joint_pos_backend, mask=env_mask_wp)
self._root_view.set_attribute(TT.DOF_VELOCITY, joint_vel_backend, mask=env_mask_wp)
"""
Operations - Simulation Parameters Writers.
"""
def write_joint_stiffness_to_sim_index(
self,
*,
stiffness: float | torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set joint stiffness over selected env / joint indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_STIFFNESS`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
stiffness: Joint stiffness [N/m or N·m/rad, depending on joint type].
May be a scalar :class:`float` (broadcast), or shape
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
shape = (env_ids.shape[0], joint_ids.shape[0])
stiffness = self._broadcast_scalar_to_2d(stiffness, shape)
self.assert_shape_and_dtype(stiffness, shape, wp.float32, "stiffness")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, joint_ids),
dim=shape,
inputs=[stiffness, env_ids, joint_ids],
outputs=[self._data._joint_stiffness.data, sim_env_ids],
device=self._device,
)
self._push_joint_property(
TT.DOF_STIFFNESS,
self._data._joint_stiffness.data,
self._data._joint_stiffness_backend,
cpu_buffer=self.data._cpu_joint_stiffness,
indices=self._get_cpu_env_ids(env_ids, sim_env_ids),
)
def write_joint_stiffness_to_sim_mask(
self,
*,
stiffness: float | torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set joint stiffness over selected env / joint masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_STIFFNESS`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
stiffness: Joint stiffness [N/m or N·m/rad, depending on joint type].
May be a scalar :class:`float` (broadcast), or shape
(num_instances, num_joints) with dtype wp.float32.
joint_mask: Joint mask. If None, all joints are updated. Shape is (num_joints,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
shape = (self._num_instances, self._num_joints)
stiffness = self._broadcast_scalar_to_2d(stiffness, shape)
self.assert_shape_and_dtype(stiffness, shape, wp.float32, "stiffness")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[stiffness, env_mask_wp, joint_mask_wp],
outputs=[self._data._joint_stiffness.data],
device=self._device,
)
self._push_joint_property(
TT.DOF_STIFFNESS,
self._data._joint_stiffness.data,
self._data._joint_stiffness_backend,
cpu_buffer=self.data._cpu_joint_stiffness,
mask=self._get_cpu_env_mask(env_mask_wp),
)
def write_joint_damping_to_sim_index(
self,
*,
damping: float | torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set joint damping over selected env / joint indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_DAMPING`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
damping: Joint damping [N·s/m or N·m·s/rad, depending on joint type].
May be a scalar :class:`float` (broadcast), or shape
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
shape = (env_ids.shape[0], joint_ids.shape[0])
damping = self._broadcast_scalar_to_2d(damping, shape)
self.assert_shape_and_dtype(damping, shape, wp.float32, "damping")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, joint_ids),
dim=shape,
inputs=[damping, env_ids, joint_ids],
outputs=[self._data._joint_damping.data, sim_env_ids],
device=self._device,
)
self._push_joint_property(
TT.DOF_DAMPING,
self._data._joint_damping.data,
self._data._joint_damping_backend,
cpu_buffer=self.data._cpu_joint_damping,
indices=self._get_cpu_env_ids(env_ids, sim_env_ids),
)
def write_joint_damping_to_sim_mask(
self,
*,
damping: float | torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set joint damping over selected env / joint masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_DAMPING`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
damping: Joint damping [N·s/m or N·m·s/rad, depending on joint type].
May be a scalar :class:`float` (broadcast), or shape
(num_instances, num_joints) with dtype wp.float32.
joint_mask: Joint mask. If None, all joints are updated. Shape is (num_joints,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
shape = (self._num_instances, self._num_joints)
damping = self._broadcast_scalar_to_2d(damping, shape)
self.assert_shape_and_dtype(damping, shape, wp.float32, "damping")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[damping, env_mask_wp, joint_mask_wp],
outputs=[self._data._joint_damping.data],
device=self._device,
)
self._push_joint_property(
TT.DOF_DAMPING,
self._data._joint_damping.data,
self._data._joint_damping_backend,
cpu_buffer=self.data._cpu_joint_damping,
mask=self._get_cpu_env_mask(env_mask_wp),
)
def write_joint_position_limit_to_sim_index(
self,
*,
limits: torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
warn_limit_violation: bool = True,
) -> None:
"""Set joint position limits over selected env / joint indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_LIMIT`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limits: Joint position limits ``[lower, upper]``
[m or rad, depending on joint type]. Either shape
(len(env_ids), len(joint_ids), 2) with dtype wp.float32, or
shape (len(env_ids), len(joint_ids)) with dtype wp.vec2f.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
warn_limit_violation: If True, log a warning when the provided limits
are inconsistent (lower > upper). Defaults to True.
"""
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
# Position limits cannot be scalar-broadcast (they pair lower/upper);
# match PhysX which explicitly rejects floats here.
if isinstance(limits, float):
raise ValueError("Joint position limits must be a tensor or array, not a float.")
# Accept both wp.vec2f shape (N, J) and the legacy (N, J, 2) wp.float32
# form (canonical PhysX/Newton layout uses vec2f).
if isinstance(limits, wp.array) and limits.dtype == wp.vec2f:
self.assert_shape_and_dtype(limits, (env_ids.shape[0], joint_ids.shape[0]), wp.vec2f, "limits")
# Reinterpret the vec2f input as a (N, J, 2) float32 view for the kernel.
kernel_limits = wp.array(
ptr=limits.ptr,
shape=(env_ids.shape[0], joint_ids.shape[0], 2),
dtype=wp.float32,
device=str(limits.device),
copy=False,
)
else:
self.assert_shape_and_dtype(limits, (env_ids.shape[0], joint_ids.shape[0], 2), wp.float32, "limits")
kernel_limits = limits
if env_ids.shape[0] == 0 or joint_ids.shape[0] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
# Scatter [lower, upper] pairs into the vec2f cache buffer.
wp.launch(
shared_kernels.write_joint_position_limit_to_buffer_index_kernel(env_ids, joint_ids),
dim=(env_ids.shape[0], joint_ids.shape[0]),
inputs=[kernel_limits, env_ids, joint_ids],
outputs=[self._data._joint_pos_limits.data, sim_env_ids],
device=self._device,
)
# Clamp default_joint_pos to the new limits and refresh soft_joint_pos_limits.
clamped_count = wp.zeros(1, dtype=wp.int32, device=self._device)
wp.launch(
clamp_default_joint_pos_and_update_soft_limits_index_kernel(env_ids, joint_ids),
dim=(env_ids.shape[0], joint_ids.shape[0]),
inputs=[
self._data._joint_pos_limits.data,
env_ids,
joint_ids,
self.cfg.soft_joint_pos_limit_factor,
],
outputs=[
self._data._default_joint_pos,
self._data._soft_joint_pos_limits,
clamped_count,
],
device=self._device,
)
if clamped_count.numpy()[0] > 0:
violation_message = (
"Some default joint positions are outside of the range of the new joint limits. Default joint"
" positions will be clamped to be within the new joint limits."
)
if warn_limit_violation:
logger.warning(violation_message)
else:
logger.info(violation_message)
# Stage to pinned-host CPU: flatten the vec2f buffer to float32 view.
self._push_joint_property(
TT.DOF_LIMIT,
self._data._joint_pos_limits.data,
self._data._joint_pos_limits_backend,
cpu_buffer=self.data._cpu_joint_position_limit,
indices=self._get_cpu_env_ids(env_ids, sim_env_ids),
)
def write_joint_position_limit_to_sim_mask(
self,
*,
limits: torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
warn_limit_violation: bool = True,
) -> None:
"""Set joint position limits over selected env / joint masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_LIMIT`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limits: Joint position limits ``[lower, upper]``
[m or rad, depending on joint type]. Either shape
(num_instances, num_joints, 2) with dtype wp.float32, or shape
(num_instances, num_joints) with dtype wp.vec2f.
joint_mask: Joint mask. If None, all joints are updated. Shape is (num_joints,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
warn_limit_violation: If True, log a warning when the provided limits
are inconsistent (lower > upper). Defaults to True.
"""
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
# Position limits cannot be scalar-broadcast (they pair lower/upper);
# match PhysX which explicitly rejects floats here.
if isinstance(limits, float):
raise ValueError("Joint position limits must be a tensor or array, not a float.")
# Accept both wp.vec2f shape (N, J) and the legacy (N, J, 2) wp.float32
# form (canonical PhysX/Newton layout uses vec2f).
if isinstance(limits, wp.array) and limits.dtype == wp.vec2f:
self.assert_shape_and_dtype(limits, (self._num_instances, self._num_joints), wp.vec2f, "limits")
kernel_limits = wp.array(
ptr=limits.ptr,
shape=(self._num_instances, self._num_joints, 2),
dtype=wp.float32,
device=str(limits.device),
copy=False,
)
else:
self.assert_shape_and_dtype(limits, (self._num_instances, self._num_joints, 2), wp.float32, "limits")
kernel_limits = limits
wp.launch(
shared_kernels.write_joint_position_limit_to_buffer_mask,
dim=(self._num_instances, self._num_joints),
inputs=[kernel_limits, env_mask_wp, joint_mask_wp],
outputs=[self._data._joint_pos_limits.data],
device=self._device,
)
# Clamp default_joint_pos to the new limits and refresh soft_joint_pos_limits.
clamped_count = wp.zeros(1, dtype=wp.int32, device=self._device)
wp.launch(
clamp_default_joint_pos_and_update_soft_limits_mask,
dim=(self._num_instances, self._num_joints),
inputs=[
self._data._joint_pos_limits.data,
env_mask_wp,
joint_mask_wp,
self.cfg.soft_joint_pos_limit_factor,
],
outputs=[
self._data._default_joint_pos,
self._data._soft_joint_pos_limits,
clamped_count,
],
device=self._device,
)
if clamped_count.numpy()[0] > 0:
violation_message = (
"Some default joint positions are outside of the range of the new joint limits. Default joint"
" positions will be clamped to be within the new joint limits."
)
if warn_limit_violation:
logger.warning(violation_message)
else:
logger.info(violation_message)
self._push_joint_property(
TT.DOF_LIMIT,
self._data._joint_pos_limits.data,
self._data._joint_pos_limits_backend,
cpu_buffer=self.data._cpu_joint_position_limit,
mask=self._get_cpu_env_mask(env_mask_wp),
)
def write_joint_velocity_limit_to_sim_index(
self,
*,
limits: float | torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set joint velocity limits over selected env / joint indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_MAX_VELOCITY`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limits: Joint velocity limits [m/s or rad/s, depending on joint type].
May be a scalar :class:`float` (broadcast), or shape
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
shape = (env_ids.shape[0], joint_ids.shape[0])
limits = self._broadcast_scalar_to_2d(limits, shape)
self.assert_shape_and_dtype(limits, shape, wp.float32, "limits")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, joint_ids),
dim=shape,
inputs=[limits, env_ids, joint_ids],
outputs=[self._data._joint_vel_limits.data, sim_env_ids],
device=self._device,
)
self._push_joint_property(
TT.DOF_MAX_VELOCITY,
self._data._joint_vel_limits.data,
self._data._joint_vel_limits_backend,
cpu_buffer=self.data._cpu_joint_velocity_limit,
indices=self._get_cpu_env_ids(env_ids, sim_env_ids),
)
def write_joint_velocity_limit_to_sim_mask(
self,
*,
limits: float | torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set joint velocity limits over selected env / joint masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_MAX_VELOCITY`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limits: Joint velocity limits [m/s or rad/s, depending on joint type].
May be a scalar :class:`float` (broadcast), or shape
(num_instances, num_joints) with dtype wp.float32.
joint_mask: Joint mask. If None, all joints are updated. Shape is (num_joints,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
shape = (self._num_instances, self._num_joints)
limits = self._broadcast_scalar_to_2d(limits, shape)
self.assert_shape_and_dtype(limits, shape, wp.float32, "limits")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[limits, env_mask_wp, joint_mask_wp],
outputs=[self._data._joint_vel_limits.data],
device=self._device,
)
self._push_joint_property(
TT.DOF_MAX_VELOCITY,
self._data._joint_vel_limits.data,
self._data._joint_vel_limits_backend,
cpu_buffer=self.data._cpu_joint_velocity_limit,
mask=self._get_cpu_env_mask(env_mask_wp),
)
def write_joint_effort_limit_to_sim_index(
self,
*,
limits: float | torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set joint effort limits over selected env / joint indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_MAX_FORCE`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limits: Joint effort limits [N or N·m, depending on joint type].
May be a scalar :class:`float` (broadcast), or shape
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
shape = (env_ids.shape[0], joint_ids.shape[0])
limits = self._broadcast_scalar_to_2d(limits, shape)
self.assert_shape_and_dtype(limits, shape, wp.float32, "limits")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, joint_ids),
dim=shape,
inputs=[limits, env_ids, joint_ids],
outputs=[self._data._joint_effort_limits.data, sim_env_ids],
device=self._device,
)
self._push_joint_property(
TT.DOF_MAX_FORCE,
self._data._joint_effort_limits.data,
self._data._joint_effort_limits_backend,
cpu_buffer=self.data._cpu_joint_effort_limit,
indices=self._get_cpu_env_ids(env_ids, sim_env_ids),
)
def write_joint_effort_limit_to_sim_mask(
self,
*,
limits: float | torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set joint effort limits over selected env / joint masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_MAX_FORCE`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limits: Joint effort limits [N or N·m, depending on joint type].
May be a scalar :class:`float` (broadcast), or shape
(num_instances, num_joints) with dtype wp.float32.
joint_mask: Joint mask. If None, all joints are updated. Shape is (num_joints,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
shape = (self._num_instances, self._num_joints)
limits = self._broadcast_scalar_to_2d(limits, shape)
self.assert_shape_and_dtype(limits, shape, wp.float32, "limits")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[limits, env_mask_wp, joint_mask_wp],
outputs=[self._data._joint_effort_limits.data],
device=self._device,
)
self._push_joint_property(
TT.DOF_MAX_FORCE,
self._data._joint_effort_limits.data,
self._data._joint_effort_limits_backend,
cpu_buffer=self.data._cpu_joint_effort_limit,
mask=self._get_cpu_env_mask(env_mask_wp),
)
def write_joint_armature_to_sim_index(
self,
*,
armature: float | torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set joint armature over selected env / joint indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_ARMATURE`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
armature: Joint armature [kg·m²]. May be a scalar :class:`float`
(broadcast), or shape (len(env_ids), len(joint_ids)) with
dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
shape = (env_ids.shape[0], joint_ids.shape[0])
armature = self._broadcast_scalar_to_2d(armature, shape)
self.assert_shape_and_dtype(armature, shape, wp.float32, "armature")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, joint_ids),
dim=shape,
inputs=[armature, env_ids, joint_ids],
outputs=[self._data._joint_armature.data, sim_env_ids],
device=self._device,
)
self._push_joint_property(
TT.DOF_ARMATURE,
self._data._joint_armature.data,
self._data._joint_armature_backend,
cpu_buffer=self.data._cpu_joint_armature,
indices=self._get_cpu_env_ids(env_ids, sim_env_ids),
)
self._data._reset_dynamics(mass_matrix=True)
def write_joint_armature_to_sim_mask(
self,
*,
armature: float | torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set joint armature over selected env / joint masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``DOF_ARMATURE`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
armature: Joint armature [kg·m²]. May be a scalar :class:`float`
(broadcast), or shape (num_instances, num_joints) with dtype
wp.float32.
joint_mask: Joint mask. If None, all joints are updated. Shape is (num_joints,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
shape = (self._num_instances, self._num_joints)
armature = self._broadcast_scalar_to_2d(armature, shape)
self.assert_shape_and_dtype(armature, shape, wp.float32, "armature")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[armature, env_mask_wp, joint_mask_wp],
outputs=[self._data._joint_armature.data],
device=self._device,
)
self._push_joint_property(
TT.DOF_ARMATURE,
self._data._joint_armature.data,
self._data._joint_armature_backend,
cpu_buffer=self.data._cpu_joint_armature,
mask=self._get_cpu_env_mask(env_mask_wp),
)
self._data._reset_dynamics(mass_matrix=True)
def write_joint_friction_coefficient_to_sim_index(
self,
*,
joint_friction_coeff: float | torch.Tensor | wp.array,
joint_dynamic_friction_coeff: float | torch.Tensor | wp.array | None = None,
joint_viscous_friction_coeff: float | torch.Tensor | wp.array | None = None,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
r"""Write joint friction coefficients over selected env / joint indices into the simulation.
Mirrors :meth:`isaaclab_physx.assets.Articulation.write_joint_friction_coefficient_to_sim_index`:
Coulomb (static & dynamic) friction with an optional viscous term. Any of the three
components can be left unset by passing ``None``; the corresponding slot in the
combined ``DOF_FRICTION_PROPERTIES`` ``(N, J, 3)`` binding is preserved.
``DOF_FRICTION_PROPERTIES`` is a CPU-only OVPhysX binding, so the
write is routed through pinned-host staging.
.. note::
This method expects partial data. Each component, if provided,
may be a scalar :class:`float` (broadcast to
``(len(env_ids), len(joint_ids))``) or a 2D tensor / warp array.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
joint_friction_coeff: Static friction coefficient :math:`\mu_s` [dimensionless].
joint_dynamic_friction_coeff: Dynamic (Coulomb) friction coefficient
:math:`\mu_d`. If ``None``, the dynamic component is preserved.
joint_viscous_friction_coeff: Viscous friction coefficient :math:`c_v`.
If ``None``, the viscous component is preserved.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
shape = (env_ids.shape[0], joint_ids.shape[0])
joint_friction_coeff = self._broadcast_scalar_to_2d(joint_friction_coeff, shape)
if joint_dynamic_friction_coeff is not None:
joint_dynamic_friction_coeff = self._broadcast_scalar_to_2d(joint_dynamic_friction_coeff, shape)
if joint_viscous_friction_coeff is not None:
joint_viscous_friction_coeff = self._broadcast_scalar_to_2d(joint_viscous_friction_coeff, shape)
self.assert_shape_and_dtype(joint_friction_coeff, shape, wp.float32, "joint_friction_coeff")
if joint_dynamic_friction_coeff is not None:
self.assert_shape_and_dtype(joint_dynamic_friction_coeff, shape, wp.float32, "joint_dynamic_friction_coeff")
if joint_viscous_friction_coeff is not None:
self.assert_shape_and_dtype(joint_viscous_friction_coeff, shape, wp.float32, "joint_viscous_friction_coeff")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(shape[0])
# refresh the combined (N, J, 3) buffer from the binding so unchanged
# components are preserved on the round-trip
self._data._read_joint_friction_binding()
wp.launch(
write_joint_friction_data_to_buffer_index_kernel(env_ids, joint_ids),
dim=shape,
inputs=[
joint_friction_coeff,
joint_dynamic_friction_coeff,
joint_viscous_friction_coeff,
env_ids,
joint_ids,
],
outputs=[self._data._joint_friction_props_buf.data, sim_env_ids],
device=self._device,
)
# Stage the combined (N, J, 3) buffer to pinned-host CPU and write to the binding.
self._push_joint_property(
TT.DOF_FRICTION_PROPERTIES,
self._data._joint_friction_props_buf.data,
self._data._joint_friction_props_backend,
component_count=3,
indices=self._get_cpu_env_ids(env_ids, sim_env_ids),
)
def write_joint_friction_coefficient_to_sim_mask(
self,
*,
joint_friction_coeff: float | torch.Tensor | wp.array,
joint_dynamic_friction_coeff: float | torch.Tensor | wp.array | None = None,
joint_viscous_friction_coeff: float | torch.Tensor | wp.array | None = None,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
r"""Mask variant of :meth:`write_joint_friction_coefficient_to_sim_index`.
Args:
joint_friction_coeff: Static friction coefficient :math:`\mu_s`. Full data,
shape ``(num_instances, num_joints)``. May be a scalar :class:`float`.
joint_dynamic_friction_coeff: Dynamic friction. ``None`` to preserve.
joint_viscous_friction_coeff: Viscous friction. ``None`` to preserve.
joint_mask: Joint mask. If None, all joints are updated.
env_mask: Environment mask. If None, all instances are updated.
"""
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
shape = (self._num_instances, self._num_joints)
joint_friction_coeff = self._broadcast_scalar_to_2d(joint_friction_coeff, shape)
if joint_dynamic_friction_coeff is not None:
joint_dynamic_friction_coeff = self._broadcast_scalar_to_2d(joint_dynamic_friction_coeff, shape)
if joint_viscous_friction_coeff is not None:
joint_viscous_friction_coeff = self._broadcast_scalar_to_2d(joint_viscous_friction_coeff, shape)
self.assert_shape_and_dtype(joint_friction_coeff, shape, wp.float32, "joint_friction_coeff")
if joint_dynamic_friction_coeff is not None:
self.assert_shape_and_dtype(joint_dynamic_friction_coeff, shape, wp.float32, "joint_dynamic_friction_coeff")
if joint_viscous_friction_coeff is not None:
self.assert_shape_and_dtype(joint_viscous_friction_coeff, shape, wp.float32, "joint_viscous_friction_coeff")
# refresh the (N, J, 3) buffer first (see ``_index`` variant)
self._data._read_joint_friction_binding()
wp.launch(
write_joint_friction_data_to_buffer_mask,
dim=shape,
inputs=[
joint_friction_coeff,
joint_dynamic_friction_coeff,
joint_viscous_friction_coeff,
env_mask_wp,
joint_mask_wp,
],
outputs=[self._data._joint_friction_props_buf.data],
device=self._device,
)
self._push_joint_property(
TT.DOF_FRICTION_PROPERTIES,
self._data._joint_friction_props_buf.data,
self._data._joint_friction_props_backend,
component_count=3,
mask=self._get_cpu_env_mask(env_mask_wp),
)
def write_joint_dynamic_friction_coefficient_to_sim_index(
self,
*,
joint_dynamic_friction_coeff: float | torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
r"""Write joint dynamic friction coefficients over selected env / joint indices into the simulation.
Mirrors :meth:`isaaclab_physx.assets.Articulation.write_joint_dynamic_friction_coefficient_to_sim_index`:
updates only the dynamic (Coulomb) slot of the combined ``DOF_FRICTION_PROPERTIES`` ``(N, J, 3)``
binding; the static and viscous components are preserved.
``DOF_FRICTION_PROPERTIES`` is a CPU-only OVPhysX binding, so the
write is routed through pinned-host staging.
.. note::
This method expects partial data. ``joint_dynamic_friction_coeff`` may be a
scalar :class:`float` (broadcast to ``(len(env_ids), len(joint_ids))``) or a
2D tensor / warp array.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
joint_dynamic_friction_coeff: Dynamic (Coulomb) friction coefficient
:math:`\mu_d` [dimensionless]. Shape is ``(len(env_ids), len(joint_ids))``
with dtype wp.float32, or a scalar that is broadcast.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
shape = (env_ids.shape[0], joint_ids.shape[0])
joint_dynamic_friction_coeff = self._broadcast_scalar_to_2d(joint_dynamic_friction_coeff, shape)
self.assert_shape_and_dtype(joint_dynamic_friction_coeff, shape, wp.float32, "joint_dynamic_friction_coeff")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(shape[0])
# refresh the combined (N, J, 3) buffer from the binding so unchanged
# components are preserved on the round-trip
self._data._read_joint_friction_binding()
wp.launch(
write_joint_friction_data_to_buffer_index_kernel(env_ids, joint_ids),
dim=shape,
inputs=[
None, # in_static — preserved
joint_dynamic_friction_coeff,
None, # in_viscous — preserved
env_ids,
joint_ids,
],
outputs=[self._data._joint_friction_props_buf.data, sim_env_ids],
device=self._device,
)
self._push_joint_property(
TT.DOF_FRICTION_PROPERTIES,
self._data._joint_friction_props_buf.data,
self._data._joint_friction_props_backend,
component_count=3,
indices=self._get_cpu_env_ids(env_ids, sim_env_ids),
)
def write_joint_dynamic_friction_coefficient_to_sim_mask(
self,
*,
joint_dynamic_friction_coeff: float | torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
r"""Mask variant of :meth:`write_joint_dynamic_friction_coefficient_to_sim_index`.
Updates only the dynamic (Coulomb) slot of the combined ``DOF_FRICTION_PROPERTIES``
``(N, J, 3)`` binding; the static and viscous components are preserved.
Args:
joint_dynamic_friction_coeff: Dynamic (Coulomb) friction coefficient
:math:`\mu_d` [dimensionless]. Full data, shape
``(num_instances, num_joints)``. May be a scalar :class:`float`.
joint_mask: Joint mask. If None, all joints are updated.
env_mask: Environment mask. If None, all instances are updated.
"""
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
shape = (self._num_instances, self._num_joints)
joint_dynamic_friction_coeff = self._broadcast_scalar_to_2d(joint_dynamic_friction_coeff, shape)
self.assert_shape_and_dtype(joint_dynamic_friction_coeff, shape, wp.float32, "joint_dynamic_friction_coeff")
# refresh the (N, J, 3) buffer first (see ``_index`` variant)
self._data._read_joint_friction_binding()
wp.launch(
write_joint_friction_data_to_buffer_mask,
dim=shape,
inputs=[
None, # in_static — preserved
joint_dynamic_friction_coeff,
None, # in_viscous — preserved
env_mask_wp,
joint_mask_wp,
],
outputs=[self._data._joint_friction_props_buf.data],
device=self._device,
)
self._push_joint_property(
TT.DOF_FRICTION_PROPERTIES,
self._data._joint_friction_props_buf.data,
self._data._joint_friction_props_backend,
component_count=3,
mask=self._get_cpu_env_mask(env_mask_wp),
)
def write_joint_viscous_friction_coefficient_to_sim_index(
self,
*,
joint_viscous_friction_coeff: float | torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
r"""Write joint viscous friction coefficients over selected env / joint indices into the simulation.
Mirrors :meth:`isaaclab_physx.assets.Articulation.write_joint_viscous_friction_coefficient_to_sim_index`:
updates only the viscous slot of the combined ``DOF_FRICTION_PROPERTIES`` ``(N, J, 3)``
binding; the static and dynamic components are preserved.
``DOF_FRICTION_PROPERTIES`` is a CPU-only OVPhysX binding, so the
write is routed through pinned-host staging.
.. note::
This method expects partial data. ``joint_viscous_friction_coeff`` may be a
scalar :class:`float` (broadcast to ``(len(env_ids), len(joint_ids))``) or a
2D tensor / warp array.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
joint_viscous_friction_coeff: Viscous friction coefficient
:math:`c_v` [N·m·s/rad or N·s/m, depending on joint type].
Shape is ``(len(env_ids), len(joint_ids))`` with dtype wp.float32, or
a scalar that is broadcast.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
joint_ids = self._resolve_joint_ids(joint_ids)
shape = (env_ids.shape[0], joint_ids.shape[0])
joint_viscous_friction_coeff = self._broadcast_scalar_to_2d(joint_viscous_friction_coeff, shape)
self.assert_shape_and_dtype(joint_viscous_friction_coeff, shape, wp.float32, "joint_viscous_friction_coeff")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(shape[0])
# refresh the combined (N, J, 3) buffer from the binding so unchanged
# components are preserved on the round-trip
self._data._read_joint_friction_binding()
wp.launch(
write_joint_friction_data_to_buffer_index_kernel(env_ids, joint_ids),
dim=shape,
inputs=[
None, # in_static — preserved
None, # in_dynamic — preserved
joint_viscous_friction_coeff,
env_ids,
joint_ids,
],
outputs=[self._data._joint_friction_props_buf.data, sim_env_ids],
device=self._device,
)
self._push_joint_property(
TT.DOF_FRICTION_PROPERTIES,
self._data._joint_friction_props_buf.data,
self._data._joint_friction_props_backend,
component_count=3,
indices=self._get_cpu_env_ids(env_ids, sim_env_ids),
)
def write_joint_viscous_friction_coefficient_to_sim_mask(
self,
*,
joint_viscous_friction_coeff: float | torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
r"""Mask variant of :meth:`write_joint_viscous_friction_coefficient_to_sim_index`.
Updates only the viscous slot of the combined ``DOF_FRICTION_PROPERTIES``
``(N, J, 3)`` binding; the static and dynamic components are preserved.
Args:
joint_viscous_friction_coeff: Viscous friction coefficient
:math:`c_v` [N·m·s/rad or N·s/m, depending on joint type].
Full data, shape ``(num_instances, num_joints)``. May be a
scalar :class:`float`.
joint_mask: Joint mask. If None, all joints are updated.
env_mask: Environment mask. If None, all instances are updated.
"""
env_mask_wp = self._resolve_env_mask(env_mask)
joint_mask_wp = self._resolve_joint_mask(joint_mask)
shape = (self._num_instances, self._num_joints)
joint_viscous_friction_coeff = self._broadcast_scalar_to_2d(joint_viscous_friction_coeff, shape)
self.assert_shape_and_dtype(joint_viscous_friction_coeff, shape, wp.float32, "joint_viscous_friction_coeff")
# refresh the (N, J, 3) buffer first (see ``_index`` variant)
self._data._read_joint_friction_binding()
wp.launch(
write_joint_friction_data_to_buffer_mask,
dim=shape,
inputs=[
None, # in_static — preserved
None, # in_dynamic — preserved
joint_viscous_friction_coeff,
env_mask_wp,
joint_mask_wp,
],
outputs=[self._data._joint_friction_props_buf.data],
device=self._device,
)
self._push_joint_property(
TT.DOF_FRICTION_PROPERTIES,
self._data._joint_friction_props_buf.data,
self._data._joint_friction_props_backend,
component_count=3,
mask=self._get_cpu_env_mask(env_mask_wp),
)
"""
Operations - Setters.
"""
def set_masses_index(
self,
*,
masses: torch.Tensor | wp.array,
body_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set body masses over selected env / body indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``BODY_MASS`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized
implementations. Performance is similar for both. However, to
allow graphed pipelines, the mask method must be used.
Args:
masses: Body masses [kg]. Shape is (len(env_ids), len(body_ids))
with dtype wp.float32.
body_ids: Body indices. Defaults to None (all bodies).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
body_ids = self._resolve_body_ids(body_ids)
self.assert_shape_and_dtype(masses, (env_ids.shape[0], body_ids.shape[0]), wp.float32, "masses")
if env_ids.shape[0] == 0 or body_ids.shape[0] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
has_body_ordering = self.data.has_body_ordering
body_mass_backend = self._data._body_mass.data
if has_body_ordering:
backend_staging = self._data._body_mass_backend
body_mass_backend = backend_staging.data
ordering_kernels.write_float_user_to_backend_with_indices_and_sim_ids(
masses,
env_ids,
body_ids,
self._body_user_to_backend_map(),
has_body_ordering,
False,
self._data._body_mass.data,
body_mass_backend,
sim_env_ids,
device=self._device,
)
self._data._body_mass.timestamp = self._data._sim_timestamp
cpu_env_ids = self._get_cpu_env_ids(env_ids, sim_env_ids)
wp.copy(self.data._cpu_body_mass, body_mass_backend)
self._root_view.set_attribute(TT.BODY_MASS, self.data._cpu_body_mass, indices=cpu_env_ids)
self._data._reset_dynamics(mass_matrix=True, gravity_compensation=True)
def set_masses_mask(
self,
*,
masses: torch.Tensor | wp.array,
body_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set body masses over selected env / body masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``BODY_MASS`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized
implementations. Performance is similar for both. However, to
allow graphed pipelines, the mask method must be used.
Args:
masses: Body masses [kg]. Shape is (num_instances, num_bodies)
with dtype wp.float32.
body_mask: Body mask. If None, all bodies are updated.
Shape is (num_bodies,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
body_mask_wp = self._resolve_body_mask(body_mask)
self.assert_shape_and_dtype(masses, (self._num_instances, self._num_bodies), wp.float32, "masses")
has_body_ordering = self.data.has_body_ordering
body_mass_backend = self._data._body_mass.data
if has_body_ordering:
backend_staging = self._data._body_mass_backend
body_mass_backend = backend_staging.data
ordering_kernels.write_float_user_to_backend_with_mask(
masses,
env_mask_wp,
body_mask_wp,
self._body_user_to_backend_map(),
has_body_ordering,
self._data._body_mass.data,
body_mass_backend,
device=self._device,
)
self._data._body_mass.timestamp = self._data._sim_timestamp
wp.copy(self.data._cpu_body_mass, body_mass_backend)
self._root_view.set_attribute(TT.BODY_MASS, self.data._cpu_body_mass, mask=self._get_cpu_env_mask(env_mask_wp))
self._data._reset_dynamics(mass_matrix=True, gravity_compensation=True)
def _set_coms(
self,
coms: torch.Tensor | wp.array,
*,
env_sel: Sequence[int] | torch.Tensor | wp.array | None,
body_sel: Sequence[int] | torch.Tensor | wp.array | None,
all_envs_selected: bool,
all_bodies_selected: bool,
use_mask: bool,
) -> None:
"""Write body center-of-mass poses to the public buffer and push them to ``BODY_COM_POSE``.
Shared implementation behind :meth:`set_coms_index` and :meth:`set_coms_mask`. This is a
CPU-only write routed through pinned-host staging because ``BODY_COM_POSE`` is a CPU-only
OVPhysX binding. The public-order pose is always written; under a non-identity body ordering
the value is additionally scattered into backend-order staging, and the staging buffer is the
one copied to the pinned host and pushed to the binding.
The static COM cache validity is preserved: a full (all envs, all bodies) write leaves the
cache valid, while a partial write first refreshes the cache
(:meth:`~ArticulationData._ensure_body_com_pose_b_current`) and keeps it valid only if it was
already current, otherwise invalidating it via the ``0.0`` / ``-1.0`` timestamp sentinel.
Args:
coms: Body center-of-mass poses [m, quaternion (x, y, z, w)]. Shape is
(len(env_ids), len(body_ids)) for index selection or (num_instances, num_bodies) for
mask selection, with dtype wp.transformf.
env_sel: Environment indices (index selection) or mask (mask selection). None selects all.
body_sel: Body indices (index selection) or mask (mask selection). None selects all.
all_envs_selected: Whether every environment is selected (the original selector was None).
all_bodies_selected: Whether every body is selected (the original selector was None).
use_mask: Whether :paramref:`env_sel` and :paramref:`body_sel` are masks (True) or indices
(False).
"""
sim_env_ids = None
if use_mask:
env_sel = self._resolve_env_mask(env_sel)
body_sel = self._resolve_body_mask(body_sel)
self.assert_shape_and_dtype(coms, (self._num_instances, self._num_bodies), wp.transformf, "coms")
else:
env_sel = self._resolve_env_ids(env_sel)
body_sel = self._resolve_body_ids(body_sel)
self.assert_shape_and_dtype(coms, (env_sel.shape[0], body_sel.shape[0]), wp.transformf, "coms")
if env_sel.shape[0] == 0 or body_sel.shape[0] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_sel.shape[0])
has_body_ordering = self.data.has_body_ordering
backend_staging = self._data._body_com_pose_b_backend
body_com_backend = self._data._body_com_pose_b.data
if has_body_ordering:
body_com_backend = backend_staging.data
if not all_bodies_selected:
self._data._ensure_body_com_pose_b_current()
cache_is_valid = (all_envs_selected and all_bodies_selected) or (
self._data._body_com_pose_b.timestamp >= 0.0 and (not has_body_ordering or backend_staging.timestamp >= 0.0)
)
if use_mask:
ordering_kernels.write_2d_user_to_backend_with_mask(
coms,
env_sel,
body_sel,
self._body_user_to_backend_map(),
has_body_ordering,
self._data._body_com_pose_b.data,
body_com_backend,
dtype=wp.transformf,
device=self._device,
)
else:
ordering_kernels.write_2d_user_to_backend_with_indices_and_sim_ids(
coms,
env_sel,
body_sel,
self._body_user_to_backend_map(),
has_body_ordering,
False,
self._data._body_com_pose_b.data,
body_com_backend,
sim_env_ids,
dtype=wp.transformf,
device=self._device,
)
self._data._reset_body_com_pose_b_dependents()
validity = 0.0 if cache_is_valid else -1.0
self._data._body_com_pose_b.timestamp = validity
if has_body_ordering:
backend_staging.timestamp = validity
wp.copy(self.data._cpu_body_coms, body_com_backend.view(wp.float32))
if use_mask:
self._root_view.set_attribute(
TT.BODY_COM_POSE, self.data._cpu_body_coms, mask=self._get_cpu_env_mask(env_sel)
)
else:
self._root_view.set_attribute(
TT.BODY_COM_POSE,
self.data._cpu_body_coms,
indices=self._get_cpu_env_ids(env_sel, sim_env_ids),
)
def set_coms_index(
self,
*,
coms: torch.Tensor | wp.array,
body_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set body center-of-mass poses over selected env / body indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``BODY_COM_POSE`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized
implementations. Performance is similar for both. However, to
allow graphed pipelines, the mask method must be used.
Args:
coms: Body center-of-mass poses [m, quaternion (x, y, z, w)].
Shape is (len(env_ids), len(body_ids)) with dtype wp.transformf.
body_ids: Body indices. Defaults to None (all bodies).
env_ids: Environment indices. Defaults to None (all environments).
"""
self._set_coms(
coms,
env_sel=env_ids,
body_sel=body_ids,
all_envs_selected=env_ids is None,
all_bodies_selected=body_ids is None,
use_mask=False,
)
def set_coms_mask(
self,
*,
coms: torch.Tensor | wp.array,
body_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set body center-of-mass poses over selected env / body masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``BODY_COM_POSE`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized
implementations. Performance is similar for both. However, to
allow graphed pipelines, the mask method must be used.
Args:
coms: Body center-of-mass poses [m, quaternion (x, y, z, w)].
Shape is (num_instances, num_bodies) with dtype wp.transformf.
body_mask: Body mask. If None, all bodies are updated.
Shape is (num_bodies,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
self._set_coms(
coms,
env_sel=env_mask,
body_sel=body_mask,
all_envs_selected=env_mask is None,
all_bodies_selected=body_mask is None,
use_mask=True,
)
def set_inertias_index(
self,
*,
inertias: torch.Tensor | wp.array,
body_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set body inertia tensors over selected env / body indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``BODY_INERTIA`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized
implementations. Performance is similar for both. However, to
allow graphed pipelines, the mask method must be used.
Args:
inertias: Body inertia tensors [kg·m²]. Shape is
(len(env_ids), len(body_ids), 9) with dtype wp.float32.
body_ids: Body indices. Defaults to None (all bodies).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
body_ids = self._resolve_body_ids(body_ids)
self.assert_shape_and_dtype(inertias, (env_ids.shape[0], body_ids.shape[0], 9), wp.float32, "inertias")
if env_ids.shape[0] == 0 or body_ids.shape[0] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
has_body_ordering = self.data.has_body_ordering
body_inertia_backend = self._data._body_inertia.data
if has_body_ordering:
backend_staging = self._data._body_inertia_backend
body_inertia_backend = backend_staging.data
ordering_kernels.write_3d_user_to_backend_with_indices_and_sim_ids(
inertias,
env_ids,
body_ids,
self._body_user_to_backend_map(),
has_body_ordering,
False,
self._data._body_inertia.data,
body_inertia_backend,
sim_env_ids,
dtype=wp.float32,
device=self._device,
)
self._data._body_inertia.timestamp = self._data._sim_timestamp
cpu_env_ids = self._get_cpu_env_ids(env_ids, sim_env_ids)
wp.copy(self.data._cpu_body_inertia, body_inertia_backend)
self._root_view.set_attribute(TT.BODY_INERTIA, self.data._cpu_body_inertia, indices=cpu_env_ids)
self._data._reset_dynamics(mass_matrix=True)
def set_inertias_mask(
self,
*,
inertias: torch.Tensor | wp.array,
body_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set body inertia tensors over selected env / body masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``BODY_INERTIA`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized
implementations. Performance is similar for both. However, to
allow graphed pipelines, the mask method must be used.
Args:
inertias: Body inertia tensors [kg·m²]. Shape is
(num_instances, num_bodies, 9) with dtype wp.float32.
body_mask: Body mask. If None, all bodies are updated.
Shape is (num_bodies,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
body_mask_wp = self._resolve_body_mask(body_mask)
self.assert_shape_and_dtype(inertias, (self._num_instances, self._num_bodies, 9), wp.float32, "inertias")
has_body_ordering = self.data.has_body_ordering
body_inertia_backend = self._data._body_inertia.data
if has_body_ordering:
backend_staging = self._data._body_inertia_backend
body_inertia_backend = backend_staging.data
ordering_kernels.write_3d_user_to_backend_with_mask(
inertias,
env_mask_wp,
body_mask_wp,
self._body_user_to_backend_map(),
has_body_ordering,
self._data._body_inertia.data,
body_inertia_backend,
dtype=wp.float32,
device=self._device,
)
self._data._body_inertia.timestamp = self._data._sim_timestamp
wp.copy(self.data._cpu_body_inertia, body_inertia_backend)
self._root_view.set_attribute(
TT.BODY_INERTIA, self.data._cpu_body_inertia, mask=self._get_cpu_env_mask(env_mask_wp)
)
self._data._reset_dynamics(mass_matrix=True)
def _write_joint_target(
self,
target: torch.Tensor | wp.array,
*,
user_buffer: wp.array,
backend_buffer: wp.array | None,
tensor_type: TT.TensorType,
env_sel: Sequence[int] | torch.Tensor | wp.array | None,
joint_sel: Sequence[int] | torch.Tensor | wp.array | None,
use_mask: bool,
) -> None:
"""Write a joint target into the public buffer and push it to the backend binding.
Shared implementation behind the six
``set_joint_{position,velocity,effort}_target_{index,mask}`` setters. The public-order
target is always written to :paramref:`user_buffer`; under a non-identity joint ordering the
value is additionally scattered into :paramref:`backend_buffer` (backend-order staging), and
that staging buffer is the one pushed to the simulation. Otherwise :paramref:`user_buffer` is
pushed directly.
Args:
target: Joint targets [m, rad, m/s, rad/s, N, or N·m, depending on the setter and joint
type]. Shape is (len(env_ids), len(joint_ids)) for index selection or
(num_instances, num_joints) for mask selection, with dtype wp.float32.
user_buffer: Public-order destination buffer for the target.
backend_buffer: Backend-order staging destination, or None when the joint ordering is
identity. It is guaranteed non-None while a non-identity joint ordering is active
because :meth:`_ordering_configure_backend_staging` allocates it during
initialization.
tensor_type: Backend binding key the target is pushed to.
env_sel: Environment indices (index selection) or mask (mask selection). None selects all.
joint_sel: Joint indices (index selection) or mask (mask selection). None selects all.
use_mask: Whether :paramref:`env_sel` and :paramref:`joint_sel` are masks (True) or
indices (False).
"""
if use_mask:
env_sel = self._resolve_env_mask(env_sel)
joint_sel = self._resolve_joint_mask(joint_sel)
self.assert_shape_and_dtype(target, (self._num_instances, self._num_joints), wp.float32, "target")
else:
env_sel = self._resolve_env_ids(env_sel)
joint_sel = self._resolve_joint_ids(joint_sel)
self.assert_shape_and_dtype(target, (env_sel.shape[0], joint_sel.shape[0]), wp.float32, "target")
if env_sel.shape[0] == 0 or joint_sel.shape[0] == 0:
return
# Under a non-identity ordering the backend staging receives the reordered copy and is the
# buffer pushed to the binding; the identity case writes and pushes the public buffer.
has_joint_ordering = self.data.has_joint_ordering
if has_joint_ordering:
target_backend = backend_buffer
else:
target_backend = user_buffer
if use_mask:
ordering_kernels.write_float_user_to_backend_with_mask(
target,
env_sel,
joint_sel,
self._joint_user_to_backend_map(),
has_joint_ordering,
user_buffer,
target_backend,
device=self._device,
)
self._root_view.set_attribute(tensor_type, target_backend, mask=env_sel)
else:
sim_env_ids = self._sim_env_ids_view(env_sel.shape[0])
ordering_kernels.write_float_user_to_backend_with_indices_and_sim_ids(
target,
env_sel,
joint_sel,
self._joint_user_to_backend_map(),
has_joint_ordering,
False,
user_buffer,
target_backend,
sim_env_ids,
device=self._device,
)
self._root_view.set_attribute(
tensor_type, target_backend, indices=self._get_sim_env_ids(env_sel, sim_env_ids)
)
def set_joint_position_target_index(
self,
*,
target: torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set joint position targets into internal buffers using indices.
This function does not apply the joint targets to the simulation. It only fills the
buffers with the desired values. To apply the joint targets, call
:meth:`write_data_to_sim`.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
target: Joint position targets [m or rad, depending on joint type]. Shape is
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
self._write_joint_target(
target,
user_buffer=self._data._joint_pos_target,
backend_buffer=self._joint_pos_target_backend,
tensor_type=TT.DOF_POSITION_TARGET,
env_sel=env_ids,
joint_sel=joint_ids,
use_mask=False,
)
def set_joint_position_target_mask(
self,
*,
target: torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set joint position targets into internal buffers using masks.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
target: Joint position targets [m or rad, depending on joint type]. Shape is
(num_instances, num_joints) with dtype wp.float32.
joint_mask: Joint mask. If None, all joints are updated. Shape is (num_joints,).
env_mask: Environment mask. If None, all instances are updated. Shape is
(num_instances,).
"""
self._write_joint_target(
target,
user_buffer=self._data._joint_pos_target,
backend_buffer=self._joint_pos_target_backend,
tensor_type=TT.DOF_POSITION_TARGET,
env_sel=env_mask,
joint_sel=joint_mask,
use_mask=True,
)
def set_joint_velocity_target_index(
self,
*,
target: torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set joint velocity targets into internal buffers using indices.
This function does not apply the joint targets to the simulation. It only fills the
buffers with the desired values. To apply the joint targets, call
:meth:`write_data_to_sim`.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
target: Joint velocity targets [m/s or rad/s, depending on joint type]. Shape is
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
self._write_joint_target(
target,
user_buffer=self._data._joint_vel_target,
backend_buffer=self._joint_vel_target_backend,
tensor_type=TT.DOF_VELOCITY_TARGET,
env_sel=env_ids,
joint_sel=joint_ids,
use_mask=False,
)
def set_joint_velocity_target_mask(
self,
*,
target: torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set joint velocity targets into internal buffers using masks.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
target: Joint velocity targets [m/s or rad/s, depending on joint type]. Shape is
(num_instances, num_joints) with dtype wp.float32.
joint_mask: Joint mask. If None, all joints are updated. Shape is (num_joints,).
env_mask: Environment mask. If None, all instances are updated. Shape is
(num_instances,).
"""
self._write_joint_target(
target,
user_buffer=self._data._joint_vel_target,
backend_buffer=self._joint_vel_target_backend,
tensor_type=TT.DOF_VELOCITY_TARGET,
env_sel=env_mask,
joint_sel=joint_mask,
use_mask=True,
)
def set_joint_effort_target_index(
self,
*,
target: torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set joint effort targets into internal buffers using indices.
This function does not apply the joint targets to the simulation. It only fills the
buffers with the desired values. To apply the joint targets, call
:meth:`write_data_to_sim`.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
target: Joint effort targets [N or N·m, depending on joint type]. Shape is
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
self._write_joint_target(
target,
user_buffer=self._data._joint_effort_target,
backend_buffer=self._joint_effort_target_backend,
tensor_type=TT.DOF_ACTUATION_FORCE,
env_sel=env_ids,
joint_sel=joint_ids,
use_mask=False,
)
def set_joint_effort_target_mask(
self,
*,
target: torch.Tensor | wp.array,
joint_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set joint effort targets into internal buffers using masks.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
target: Joint effort targets [N or N·m, depending on joint type]. Shape is
(num_instances, num_joints) with dtype wp.float32.
joint_mask: Joint mask. If None, all joints are updated. Shape is (num_joints,).
env_mask: Environment mask. If None, all instances are updated. Shape is
(num_instances,).
"""
self._write_joint_target(
target,
user_buffer=self._data._joint_effort_target,
backend_buffer=self._joint_effort_target_backend,
tensor_type=TT.DOF_ACTUATION_FORCE,
env_sel=env_mask,
joint_sel=joint_mask,
use_mask=True,
)
"""
Operations - Tendons.
"""
def set_fixed_tendon_stiffness_index(
self,
*,
stiffness: float | torch.Tensor | wp.array,
fixed_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set fixed-tendon stiffness over selected env / tendon indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_STIFFNESS`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
stiffness: Fixed-tendon stiffness [N/m]. May be a scalar
:class:`float` (broadcast), or shape
(len(env_ids), len(fixed_tendon_ids)) with dtype wp.float32.
fixed_tendon_ids: Fixed-tendon indices. Defaults to None (all fixed tendons).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
tendon_ids = self._resolve_fixed_tendon_ids(fixed_tendon_ids)
shape = (env_ids.shape[0], tendon_ids.shape[0])
stiffness = self._broadcast_scalar_to_2d(stiffness, shape)
self.assert_shape_and_dtype(stiffness, shape, wp.float32, "stiffness")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, tendon_ids),
dim=shape,
inputs=[stiffness, env_ids, tendon_ids],
outputs=[self._data._fixed_tendon_stiffness.data, sim_env_ids],
device=self._device,
)
self._root_view.set_attribute(
TT.FIXED_TENDON_STIFFNESS,
self._data._fixed_tendon_stiffness.data,
indices=self._get_sim_env_ids(env_ids, sim_env_ids),
)
def set_fixed_tendon_stiffness_mask(
self,
*,
stiffness: float | torch.Tensor | wp.array,
fixed_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set fixed-tendon stiffness over selected env / tendon masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_STIFFNESS`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
stiffness: Fixed-tendon stiffness [N/m]. May be a scalar
:class:`float` (broadcast), or shape
(num_instances, num_fixed_tendons) with dtype wp.float32.
fixed_tendon_mask: Fixed-tendon mask. If None, all fixed tendons are updated.
Shape is (num_fixed_tendons,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
tendon_mask_wp = self._resolve_fixed_tendon_mask(fixed_tendon_mask)
shape = (self._num_instances, self._num_fixed_tendons)
stiffness = self._broadcast_scalar_to_2d(stiffness, shape)
self.assert_shape_and_dtype(stiffness, shape, wp.float32, "stiffness")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[stiffness, env_mask_wp, tendon_mask_wp],
outputs=[self._data._fixed_tendon_stiffness.data],
device=self._device,
)
self._root_view.set_attribute(
TT.FIXED_TENDON_STIFFNESS, self._data._fixed_tendon_stiffness.data, mask=env_mask_wp
)
def set_fixed_tendon_damping_index(
self,
*,
damping: float | torch.Tensor | wp.array,
fixed_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set fixed-tendon damping over selected env / tendon indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_DAMPING`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
damping: Fixed-tendon damping [N·s/m]. May be a scalar :class:`float`
(broadcast), or shape (len(env_ids), len(fixed_tendon_ids)) with
dtype wp.float32.
fixed_tendon_ids: Fixed-tendon indices. Defaults to None (all fixed tendons).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
tendon_ids = self._resolve_fixed_tendon_ids(fixed_tendon_ids)
shape = (env_ids.shape[0], tendon_ids.shape[0])
damping = self._broadcast_scalar_to_2d(damping, shape)
self.assert_shape_and_dtype(damping, shape, wp.float32, "damping")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, tendon_ids),
dim=shape,
inputs=[damping, env_ids, tendon_ids],
outputs=[self._data._fixed_tendon_damping.data, sim_env_ids],
device=self._device,
)
self._root_view.set_attribute(
TT.FIXED_TENDON_DAMPING,
self._data._fixed_tendon_damping.data,
indices=self._get_sim_env_ids(env_ids, sim_env_ids),
)
def set_fixed_tendon_damping_mask(
self,
*,
damping: float | torch.Tensor | wp.array,
fixed_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set fixed-tendon damping over selected env / tendon masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_DAMPING`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
damping: Fixed-tendon damping [N·s/m]. May be a scalar :class:`float`
(broadcast), or shape (num_instances, num_fixed_tendons) with
dtype wp.float32.
fixed_tendon_mask: Fixed-tendon mask. If None, all fixed tendons are updated.
Shape is (num_fixed_tendons,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
tendon_mask_wp = self._resolve_fixed_tendon_mask(fixed_tendon_mask)
shape = (self._num_instances, self._num_fixed_tendons)
damping = self._broadcast_scalar_to_2d(damping, shape)
self.assert_shape_and_dtype(damping, shape, wp.float32, "damping")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[damping, env_mask_wp, tendon_mask_wp],
outputs=[self._data._fixed_tendon_damping.data],
device=self._device,
)
self._root_view.set_attribute(TT.FIXED_TENDON_DAMPING, self._data._fixed_tendon_damping.data, mask=env_mask_wp)
def set_fixed_tendon_limit_stiffness_index(
self,
*,
limit_stiffness: float | torch.Tensor | wp.array,
fixed_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set fixed-tendon limit stiffness over selected env / tendon indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_LIMIT_STIFFNESS`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limit_stiffness: Fixed-tendon limit stiffness [N/m]. May be a
scalar :class:`float` (broadcast), or shape
(len(env_ids), len(fixed_tendon_ids)) with dtype wp.float32.
fixed_tendon_ids: Fixed-tendon indices. Defaults to None (all fixed tendons).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
tendon_ids = self._resolve_fixed_tendon_ids(fixed_tendon_ids)
shape = (env_ids.shape[0], tendon_ids.shape[0])
limit_stiffness = self._broadcast_scalar_to_2d(limit_stiffness, shape)
self.assert_shape_and_dtype(limit_stiffness, shape, wp.float32, "limit_stiffness")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, tendon_ids),
dim=shape,
inputs=[limit_stiffness, env_ids, tendon_ids],
outputs=[self._data._fixed_tendon_limit_stiffness.data, sim_env_ids],
device=self._device,
)
self._root_view.set_attribute(
TT.FIXED_TENDON_LIMIT_STIFFNESS,
self._data._fixed_tendon_limit_stiffness.data,
indices=self._get_sim_env_ids(env_ids, sim_env_ids),
)
def set_fixed_tendon_limit_stiffness_mask(
self,
*,
limit_stiffness: float | torch.Tensor | wp.array,
fixed_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set fixed-tendon limit stiffness over selected env / tendon masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_LIMIT_STIFFNESS`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limit_stiffness: Fixed-tendon limit stiffness [N/m]. May be a
scalar :class:`float` (broadcast), or shape
(num_instances, num_fixed_tendons) with dtype wp.float32.
fixed_tendon_mask: Fixed-tendon mask. If None, all fixed tendons are updated.
Shape is (num_fixed_tendons,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
tendon_mask_wp = self._resolve_fixed_tendon_mask(fixed_tendon_mask)
shape = (self._num_instances, self._num_fixed_tendons)
limit_stiffness = self._broadcast_scalar_to_2d(limit_stiffness, shape)
self.assert_shape_and_dtype(limit_stiffness, shape, wp.float32, "limit_stiffness")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[limit_stiffness, env_mask_wp, tendon_mask_wp],
outputs=[self._data._fixed_tendon_limit_stiffness.data],
device=self._device,
)
self._root_view.set_attribute(
TT.FIXED_TENDON_LIMIT_STIFFNESS, self._data._fixed_tendon_limit_stiffness.data, mask=env_mask_wp
)
def set_fixed_tendon_position_limit_index(
self,
*,
limit: torch.Tensor | wp.array,
fixed_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set fixed-tendon position limits over selected env / tendon indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_LIMIT`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limit: Fixed-tendon position limits ``[lower, upper]`` [m].
Shape is (len(env_ids), len(fixed_tendon_ids), 2) with dtype wp.float32.
fixed_tendon_ids: Fixed-tendon indices. Defaults to None (all fixed tendons).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
tendon_ids = self._resolve_fixed_tendon_ids(fixed_tendon_ids)
self.assert_shape_and_dtype(limit, (env_ids.shape[0], tendon_ids.shape[0], 2), wp.float32, "limit")
if env_ids.shape[0] == 0 or tendon_ids.shape[0] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
# Scatter [lower, upper] pairs into the vec2f cache buffer.
wp.launch(
shared_kernels.write_joint_position_limit_to_buffer_index_kernel(env_ids, tendon_ids),
dim=(env_ids.shape[0], tendon_ids.shape[0]),
inputs=[limit, env_ids, tendon_ids],
outputs=[self._data._fixed_tendon_pos_limits.data, sim_env_ids],
device=self._device,
)
# reinterpret the vec2f buffer as a (N, T, 2) float32 view for the binding
flat_src = wp.array(
ptr=self._data._fixed_tendon_pos_limits.data.ptr,
shape=(self._num_instances, self._num_fixed_tendons, 2),
dtype=wp.float32,
device=self._device,
copy=False,
)
self._root_view.set_attribute(
TT.FIXED_TENDON_LIMIT, flat_src, indices=self._get_sim_env_ids(env_ids, sim_env_ids)
)
def set_fixed_tendon_position_limit_mask(
self,
*,
limit: torch.Tensor | wp.array,
fixed_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set fixed-tendon position limits over selected env / tendon masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_LIMIT`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limit: Fixed-tendon position limits ``[lower, upper]`` [m].
Shape is (num_instances, num_fixed_tendons, 2) with dtype wp.float32.
fixed_tendon_mask: Fixed-tendon mask. If None, all fixed tendons are updated.
Shape is (num_fixed_tendons,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
tendon_mask_wp = self._resolve_fixed_tendon_mask(fixed_tendon_mask)
self.assert_shape_and_dtype(limit, (self._num_instances, self._num_fixed_tendons, 2), wp.float32, "limit")
wp.launch(
shared_kernels.write_joint_position_limit_to_buffer_mask,
dim=(self._num_instances, self._num_fixed_tendons),
inputs=[limit, env_mask_wp, tendon_mask_wp],
outputs=[self._data._fixed_tendon_pos_limits.data],
device=self._device,
)
flat_src = wp.array(
ptr=self._data._fixed_tendon_pos_limits.data.ptr,
shape=(self._num_instances, self._num_fixed_tendons, 2),
dtype=wp.float32,
device=self._device,
copy=False,
)
self._root_view.set_attribute(TT.FIXED_TENDON_LIMIT, flat_src, mask=env_mask_wp)
def set_fixed_tendon_rest_length_index(
self,
*,
rest_length: float | torch.Tensor | wp.array,
fixed_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set fixed-tendon rest lengths over selected env / tendon indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_REST_LENGTH`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
rest_length: Fixed-tendon rest lengths [m]. May be a scalar
:class:`float` (broadcast), or shape
(len(env_ids), len(fixed_tendon_ids)) with dtype wp.float32.
fixed_tendon_ids: Fixed-tendon indices. Defaults to None (all fixed tendons).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
tendon_ids = self._resolve_fixed_tendon_ids(fixed_tendon_ids)
shape = (env_ids.shape[0], tendon_ids.shape[0])
rest_length = self._broadcast_scalar_to_2d(rest_length, shape)
self.assert_shape_and_dtype(rest_length, shape, wp.float32, "rest_length")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, tendon_ids),
dim=shape,
inputs=[rest_length, env_ids, tendon_ids],
outputs=[self._data._fixed_tendon_rest_length.data, sim_env_ids],
device=self._device,
)
self._root_view.set_attribute(
TT.FIXED_TENDON_REST_LENGTH,
self._data._fixed_tendon_rest_length.data,
indices=self._get_sim_env_ids(env_ids, sim_env_ids),
)
def set_fixed_tendon_rest_length_mask(
self,
*,
rest_length: float | torch.Tensor | wp.array,
fixed_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set fixed-tendon rest lengths over selected env / tendon masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_REST_LENGTH`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
rest_length: Fixed-tendon rest lengths [m]. May be a scalar
:class:`float` (broadcast), or shape
(num_instances, num_fixed_tendons) with dtype wp.float32.
fixed_tendon_mask: Fixed-tendon mask. If None, all fixed tendons are updated.
Shape is (num_fixed_tendons,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
tendon_mask_wp = self._resolve_fixed_tendon_mask(fixed_tendon_mask)
shape = (self._num_instances, self._num_fixed_tendons)
rest_length = self._broadcast_scalar_to_2d(rest_length, shape)
self.assert_shape_and_dtype(rest_length, shape, wp.float32, "rest_length")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[rest_length, env_mask_wp, tendon_mask_wp],
outputs=[self._data._fixed_tendon_rest_length.data],
device=self._device,
)
self._root_view.set_attribute(
TT.FIXED_TENDON_REST_LENGTH, self._data._fixed_tendon_rest_length.data, mask=env_mask_wp
)
def set_fixed_tendon_offset_index(
self,
*,
offset: float | torch.Tensor | wp.array,
fixed_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set fixed-tendon offsets over selected env / tendon indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_OFFSET`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
offset: Fixed-tendon offsets [m]. May be a scalar :class:`float`
(broadcast), or shape (len(env_ids), len(fixed_tendon_ids))
with dtype wp.float32.
fixed_tendon_ids: Fixed-tendon indices. Defaults to None (all fixed tendons).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
tendon_ids = self._resolve_fixed_tendon_ids(fixed_tendon_ids)
shape = (env_ids.shape[0], tendon_ids.shape[0])
offset = self._broadcast_scalar_to_2d(offset, shape)
self.assert_shape_and_dtype(offset, shape, wp.float32, "offset")
if shape[0] == 0 or shape[1] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, tendon_ids),
dim=shape,
inputs=[offset, env_ids, tendon_ids],
outputs=[self._data._fixed_tendon_offset.data, sim_env_ids],
device=self._device,
)
self._root_view.set_attribute(
TT.FIXED_TENDON_OFFSET,
self._data._fixed_tendon_offset.data,
indices=self._get_sim_env_ids(env_ids, sim_env_ids),
)
def set_fixed_tendon_offset_mask(
self,
*,
offset: float | torch.Tensor | wp.array,
fixed_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set fixed-tendon offsets over selected env / tendon masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``FIXED_TENDON_OFFSET`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
offset: Fixed-tendon offsets [m]. May be a scalar :class:`float`
(broadcast), or shape (num_instances, num_fixed_tendons) with
dtype wp.float32.
fixed_tendon_mask: Fixed-tendon mask. If None, all fixed tendons are updated.
Shape is (num_fixed_tendons,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
tendon_mask_wp = self._resolve_fixed_tendon_mask(fixed_tendon_mask)
shape = (self._num_instances, self._num_fixed_tendons)
offset = self._broadcast_scalar_to_2d(offset, shape)
self.assert_shape_and_dtype(offset, shape, wp.float32, "offset")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[offset, env_mask_wp, tendon_mask_wp],
outputs=[self._data._fixed_tendon_offset.data],
device=self._device,
)
self._root_view.set_attribute(TT.FIXED_TENDON_OFFSET, self._data._fixed_tendon_offset.data, mask=env_mask_wp)
def write_fixed_tendon_properties_to_sim_index(
self,
*,
fixed_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Push the cached fixed-tendon properties to the simulation in a single batch.
PhysX exposes a single ``root_view.set_fixed_tendon_properties`` that writes all
six tendon property buffers at once. OVPhysX has no such batch setter, so this
method writes each ``FIXED_TENDON_*`` binding individually from the matching
``self._data._fixed_tendon_*`` buffer.
.. note::
Only env indices apply to the simulation write; ``fixed_tendon_ids`` is
accepted for API parity with PhysX but is unused (the simulation
writes all tendons of the selected envs).
Args:
fixed_tendon_ids: Accepted for PhysX API parity; ignored.
env_ids: Environment indices. If None, all environments are written.
"""
env_ids = self._resolve_env_ids(env_ids)
if env_ids.shape[0] == 0:
return
sim_env_ids = self._get_sim_env_ids(env_ids)
for tt, buf in (
(TT.FIXED_TENDON_STIFFNESS, self._data._fixed_tendon_stiffness),
(TT.FIXED_TENDON_DAMPING, self._data._fixed_tendon_damping),
(TT.FIXED_TENDON_LIMIT_STIFFNESS, self._data._fixed_tendon_limit_stiffness),
(TT.FIXED_TENDON_REST_LENGTH, self._data._fixed_tendon_rest_length),
(TT.FIXED_TENDON_OFFSET, self._data._fixed_tendon_offset),
):
if self._get_binding(tt) is not None:
self._root_view.set_attribute(tt, buf.data, indices=sim_env_ids)
# Position-limit binding consumes a flat (N, T, 2) float32 view.
binding = self._get_binding(TT.FIXED_TENDON_LIMIT)
if binding is not None:
flat_src = wp.array(
ptr=self._data._fixed_tendon_pos_limits.data.ptr,
shape=(self._num_instances, self._num_fixed_tendons, 2),
dtype=wp.float32,
device=self._device,
copy=False,
)
self._root_view.set_attribute(TT.FIXED_TENDON_LIMIT, flat_src, indices=sim_env_ids)
def write_fixed_tendon_properties_to_sim_mask(
self,
*,
fixed_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Mask variant of :meth:`write_fixed_tendon_properties_to_sim_index`.
Args:
fixed_tendon_mask: Accepted for PhysX API parity; ignored.
env_mask: Environment mask. If None, all environments are written.
"""
env_mask_wp = self._resolve_env_mask(env_mask)
for tt, buf in (
(TT.FIXED_TENDON_STIFFNESS, self._data._fixed_tendon_stiffness),
(TT.FIXED_TENDON_DAMPING, self._data._fixed_tendon_damping),
(TT.FIXED_TENDON_LIMIT_STIFFNESS, self._data._fixed_tendon_limit_stiffness),
(TT.FIXED_TENDON_REST_LENGTH, self._data._fixed_tendon_rest_length),
(TT.FIXED_TENDON_OFFSET, self._data._fixed_tendon_offset),
):
if self._get_binding(tt) is not None:
self._root_view.set_attribute(tt, buf.data, mask=env_mask_wp)
binding = self._get_binding(TT.FIXED_TENDON_LIMIT)
if binding is not None:
flat_src = wp.array(
ptr=self._data._fixed_tendon_pos_limits.data.ptr,
shape=(self._num_instances, self._num_fixed_tendons, 2),
dtype=wp.float32,
device=self._device,
copy=False,
)
self._root_view.set_attribute(TT.FIXED_TENDON_LIMIT, flat_src, mask=env_mask_wp)
def set_spatial_tendon_stiffness_index(
self,
*,
stiffness: float | torch.Tensor | wp.array,
spatial_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set spatial-tendon stiffness over selected env / tendon indices into the simulation.
``SPATIAL_TENDON_STIFFNESS`` is a sim-device binding on OVPhysX
(tendon properties are applied without a CPU clone), so the write
goes directly from the sim-device buffer to the binding.
.. note::
This method expects partial data. A scalar :class:`float` is
broadcast to ``(len(env_ids), len(spatial_tendon_ids))``.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
stiffness: Spatial-tendon stiffness [N/m]. Scalar :class:`float`,
or shape ``(len(env_ids), len(spatial_tendon_ids))`` with
dtype wp.float32.
spatial_tendon_ids: Spatial-tendon indices. Defaults to None (all spatial tendons).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
tendon_ids = self._resolve_spatial_tendon_ids(spatial_tendon_ids)
stiffness = self._broadcast_scalar_to_2d(stiffness, (env_ids.shape[0], tendon_ids.shape[0]))
self.assert_shape_and_dtype(stiffness, (env_ids.shape[0], tendon_ids.shape[0]), wp.float32, "stiffness")
if env_ids.shape[0] == 0 or tendon_ids.shape[0] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, tendon_ids),
dim=(env_ids.shape[0], tendon_ids.shape[0]),
inputs=[stiffness, env_ids, tendon_ids],
outputs=[self._data._spatial_tendon_stiffness.data, sim_env_ids],
device=self._device,
)
self._root_view.set_attribute(
TT.SPATIAL_TENDON_STIFFNESS,
self._data._spatial_tendon_stiffness.data,
indices=self._get_sim_env_ids(env_ids, sim_env_ids),
)
def set_spatial_tendon_stiffness_mask(
self,
*,
stiffness: float | torch.Tensor | wp.array,
spatial_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set spatial-tendon stiffness over selected env / tendon masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``SPATIAL_TENDON_STIFFNESS`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
stiffness: Spatial-tendon stiffness [N/m]. May be a scalar
:class:`float` (broadcast), or shape
(num_instances, num_spatial_tendons) with dtype wp.float32.
spatial_tendon_mask: Spatial-tendon mask. If None, all spatial tendons are updated.
Shape is (num_spatial_tendons,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
tendon_mask_wp = self._resolve_spatial_tendon_mask(spatial_tendon_mask)
shape = (self._num_instances, self._num_spatial_tendons)
stiffness = self._broadcast_scalar_to_2d(stiffness, shape)
self.assert_shape_and_dtype(stiffness, shape, wp.float32, "stiffness")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[stiffness, env_mask_wp, tendon_mask_wp],
outputs=[self._data._spatial_tendon_stiffness.data],
device=self._device,
)
self._root_view.set_attribute(
TT.SPATIAL_TENDON_STIFFNESS, self._data._spatial_tendon_stiffness.data, mask=env_mask_wp
)
def set_spatial_tendon_damping_index(
self,
*,
damping: float | torch.Tensor | wp.array,
spatial_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set spatial-tendon damping over selected env / tendon indices into the simulation.
This is a CPU-only write routed through pinned-host staging because
``SPATIAL_TENDON_DAMPING`` is a CPU-only OVPhysX binding.
.. note::
This method expects partial data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
damping: Spatial-tendon damping [N·s/m]. Shape is
(len(env_ids), len(spatial_tendon_ids)) with dtype wp.float32.
spatial_tendon_ids: Spatial-tendon indices. Defaults to None (all spatial tendons).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
tendon_ids = self._resolve_spatial_tendon_ids(spatial_tendon_ids)
damping = self._broadcast_scalar_to_2d(damping, (env_ids.shape[0], tendon_ids.shape[0]))
self.assert_shape_and_dtype(damping, (env_ids.shape[0], tendon_ids.shape[0]), wp.float32, "damping")
if env_ids.shape[0] == 0 or tendon_ids.shape[0] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, tendon_ids),
dim=(env_ids.shape[0], tendon_ids.shape[0]),
inputs=[damping, env_ids, tendon_ids],
outputs=[self._data._spatial_tendon_damping.data, sim_env_ids],
device=self._device,
)
self._root_view.set_attribute(
TT.SPATIAL_TENDON_DAMPING,
self._data._spatial_tendon_damping.data,
indices=self._get_sim_env_ids(env_ids, sim_env_ids),
)
def set_spatial_tendon_damping_mask(
self,
*,
damping: float | torch.Tensor | wp.array,
spatial_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set spatial-tendon damping over selected env / tendon masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``SPATIAL_TENDON_DAMPING`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
damping: Spatial-tendon damping [N·s/m]. May be a scalar
:class:`float` (broadcast), or shape
(num_instances, num_spatial_tendons) with dtype wp.float32.
spatial_tendon_mask: Spatial-tendon mask. If None, all spatial tendons are updated.
Shape is (num_spatial_tendons,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
tendon_mask_wp = self._resolve_spatial_tendon_mask(spatial_tendon_mask)
shape = (self._num_instances, self._num_spatial_tendons)
damping = self._broadcast_scalar_to_2d(damping, shape)
self.assert_shape_and_dtype(damping, shape, wp.float32, "damping")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[damping, env_mask_wp, tendon_mask_wp],
outputs=[self._data._spatial_tendon_damping.data],
device=self._device,
)
self._root_view.set_attribute(
TT.SPATIAL_TENDON_DAMPING, self._data._spatial_tendon_damping.data, mask=env_mask_wp
)
def set_spatial_tendon_limit_stiffness_index(
self,
*,
limit_stiffness: float | torch.Tensor | wp.array,
spatial_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set spatial-tendon limit stiffness over selected env / tendon indices into the simulation.
``SPATIAL_TENDON_LIMIT_STIFFNESS`` is a sim-device binding on OVPhysX;
the write goes directly from the sim-device buffer to the binding.
.. note::
This method expects partial data. A scalar :class:`float` is
broadcast to ``(len(env_ids), len(spatial_tendon_ids))``.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limit_stiffness: Spatial-tendon limit stiffness [N/m]. Scalar
:class:`float`, or shape ``(len(env_ids), len(spatial_tendon_ids))``
with dtype wp.float32.
spatial_tendon_ids: Spatial-tendon indices. Defaults to None (all spatial tendons).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
tendon_ids = self._resolve_spatial_tendon_ids(spatial_tendon_ids)
limit_stiffness = self._broadcast_scalar_to_2d(limit_stiffness, (env_ids.shape[0], tendon_ids.shape[0]))
self.assert_shape_and_dtype(
limit_stiffness, (env_ids.shape[0], tendon_ids.shape[0]), wp.float32, "limit_stiffness"
)
if env_ids.shape[0] == 0 or tendon_ids.shape[0] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, tendon_ids),
dim=(env_ids.shape[0], tendon_ids.shape[0]),
inputs=[limit_stiffness, env_ids, tendon_ids],
outputs=[self._data._spatial_tendon_limit_stiffness.data, sim_env_ids],
device=self._device,
)
self._root_view.set_attribute(
TT.SPATIAL_TENDON_LIMIT_STIFFNESS,
self._data._spatial_tendon_limit_stiffness.data,
indices=self._get_sim_env_ids(env_ids, sim_env_ids),
)
def set_spatial_tendon_limit_stiffness_mask(
self,
*,
limit_stiffness: float | torch.Tensor | wp.array,
spatial_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set spatial-tendon limit stiffness over selected env / tendon masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``SPATIAL_TENDON_LIMIT_STIFFNESS`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
limit_stiffness: Spatial-tendon limit stiffness [N/m]. May be a
scalar :class:`float` (broadcast), or shape
(num_instances, num_spatial_tendons) with dtype wp.float32.
spatial_tendon_mask: Spatial-tendon mask. If None, all spatial tendons are updated.
Shape is (num_spatial_tendons,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
tendon_mask_wp = self._resolve_spatial_tendon_mask(spatial_tendon_mask)
shape = (self._num_instances, self._num_spatial_tendons)
limit_stiffness = self._broadcast_scalar_to_2d(limit_stiffness, shape)
self.assert_shape_and_dtype(limit_stiffness, shape, wp.float32, "limit_stiffness")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[limit_stiffness, env_mask_wp, tendon_mask_wp],
outputs=[self._data._spatial_tendon_limit_stiffness.data],
device=self._device,
)
self._root_view.set_attribute(
TT.SPATIAL_TENDON_LIMIT_STIFFNESS, self._data._spatial_tendon_limit_stiffness.data, mask=env_mask_wp
)
def set_spatial_tendon_offset_index(
self,
*,
offset: float | torch.Tensor | wp.array,
spatial_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Set spatial-tendon offsets over selected env / tendon indices into the simulation.
``SPATIAL_TENDON_OFFSET`` is a sim-device binding on OVPhysX; the
write goes directly from the sim-device buffer to the binding.
.. note::
This method expects partial data. A scalar :class:`float` is
broadcast to ``(len(env_ids), len(spatial_tendon_ids))``.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
offset: Spatial-tendon offsets [m]. Scalar :class:`float`, or
shape ``(len(env_ids), len(spatial_tendon_ids))`` with
dtype wp.float32.
spatial_tendon_ids: Spatial-tendon indices. Defaults to None (all spatial tendons).
env_ids: Environment indices. Defaults to None (all environments).
"""
env_ids = self._resolve_env_ids(env_ids)
tendon_ids = self._resolve_spatial_tendon_ids(spatial_tendon_ids)
offset = self._broadcast_scalar_to_2d(offset, (env_ids.shape[0], tendon_ids.shape[0]))
self.assert_shape_and_dtype(offset, (env_ids.shape[0], tendon_ids.shape[0]), wp.float32, "offset")
if env_ids.shape[0] == 0 or tendon_ids.shape[0] == 0:
return
sim_env_ids = self._sim_env_ids_view(env_ids.shape[0])
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_indices_and_sim_ids_kernel(env_ids, tendon_ids),
dim=(env_ids.shape[0], tendon_ids.shape[0]),
inputs=[offset, env_ids, tendon_ids],
outputs=[self._data._spatial_tendon_offset.data, sim_env_ids],
device=self._device,
)
self._root_view.set_attribute(
TT.SPATIAL_TENDON_OFFSET,
self._data._spatial_tendon_offset.data,
indices=self._get_sim_env_ids(env_ids, sim_env_ids),
)
def set_spatial_tendon_offset_mask(
self,
*,
offset: float | torch.Tensor | wp.array,
spatial_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Set spatial-tendon offsets over selected env / tendon masks into the simulation.
This is a CPU-only write routed through pinned-host staging because
``SPATIAL_TENDON_OFFSET`` is a CPU-only OVPhysX binding.
.. note::
This method expects full data.
.. tip::
Both the index and mask methods have dedicated optimized implementations.
Performance is similar for both. However, to allow graphed pipelines, the
mask method must be used.
Args:
offset: Spatial-tendon offsets [m]. May be a scalar :class:`float`
(broadcast), or shape (num_instances, num_spatial_tendons) with
dtype wp.float32.
spatial_tendon_mask: Spatial-tendon mask. If None, all spatial tendons are updated.
Shape is (num_spatial_tendons,).
env_mask: Environment mask. If None, all instances are updated.
Shape is (num_instances,).
"""
env_mask_wp = self._resolve_env_mask(env_mask)
tendon_mask_wp = self._resolve_spatial_tendon_mask(spatial_tendon_mask)
shape = (self._num_instances, self._num_spatial_tendons)
offset = self._broadcast_scalar_to_2d(offset, shape)
self.assert_shape_and_dtype(offset, shape, wp.float32, "offset")
wp.launch(
shared_kernels.write_2d_data_to_buffer_with_mask,
dim=shape,
inputs=[offset, env_mask_wp, tendon_mask_wp],
outputs=[self._data._spatial_tendon_offset.data],
device=self._device,
)
self._root_view.set_attribute(
TT.SPATIAL_TENDON_OFFSET, self._data._spatial_tendon_offset.data, mask=env_mask_wp
)
def write_spatial_tendon_properties_to_sim_index(
self,
*,
spatial_tendon_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Push the cached spatial-tendon properties to the simulation in a single batch.
Mirrors :meth:`write_fixed_tendon_properties_to_sim_index` for
spatial tendons. Only the four wheel-supported tensor types are
written; ``ARTICULATION_SPATIAL_TENDON_LIMIT`` and
``ARTICULATION_SPATIAL_TENDON_REST_LENGTH`` are forward-compat
stubs (see ``docs/superpowers/specs/2026-04-28-ovphysx-wheel-gaps-for-marco.md``).
Args:
spatial_tendon_ids: Accepted for PhysX API parity; ignored.
env_ids: Environment indices. If None, all environments are written.
"""
env_ids = self._resolve_env_ids(env_ids)
if env_ids.shape[0] == 0:
return
sim_env_ids = self._get_sim_env_ids(env_ids)
for tt, buf in (
(TT.SPATIAL_TENDON_STIFFNESS, self._data._spatial_tendon_stiffness),
(TT.SPATIAL_TENDON_DAMPING, self._data._spatial_tendon_damping),
(TT.SPATIAL_TENDON_LIMIT_STIFFNESS, self._data._spatial_tendon_limit_stiffness),
(TT.SPATIAL_TENDON_OFFSET, self._data._spatial_tendon_offset),
):
if self._get_binding(tt) is not None:
self._root_view.set_attribute(tt, buf.data, indices=sim_env_ids)
def write_spatial_tendon_properties_to_sim_mask(
self,
*,
spatial_tendon_mask: wp.array | None = None,
env_mask: wp.array | None = None,
) -> None:
"""Mask variant of :meth:`write_spatial_tendon_properties_to_sim_index`."""
env_mask_wp = self._resolve_env_mask(env_mask)
for tt, buf in (
(TT.SPATIAL_TENDON_STIFFNESS, self._data._spatial_tendon_stiffness),
(TT.SPATIAL_TENDON_DAMPING, self._data._spatial_tendon_damping),
(TT.SPATIAL_TENDON_LIMIT_STIFFNESS, self._data._spatial_tendon_limit_stiffness),
(TT.SPATIAL_TENDON_OFFSET, self._data._spatial_tendon_offset),
):
if self._get_binding(tt) is not None:
self._root_view.set_attribute(tt, buf.data, mask=env_mask_wp)
"""
Internal helper.
"""
def _initialize_impl(self) -> None:
"""Initialize the articulation from the OVPhysX simulation backend."""
# obtain global simulation view
physx_instance = OvPhysxManager.get_physx_instance()
if physx_instance is None:
raise RuntimeError("OvPhysxManager has not been initialized yet.")
self._ovphysx = physx_instance
self._device = OvPhysxManager.get_device()
# Resolve the articulation root expression.
if self.cfg.articulation_root_prim_path is not None:
root_prim_path_expr = self.cfg.prim_path + self.cfg.articulation_root_prim_path
else:
def has_articulation_root_api(prim) -> bool:
return bool(prim.HasAPI(UsdPhysics.ArticulationRootAPI))
resolve_kwargs = {"predicate": has_articulation_root_api, "expected_num_matches": 1}
root_matches = sim_utils.resolve_matching_prims_from_source(self.cfg.prim_path, **resolve_kwargs)
_, root_prim_path_expr = root_matches[0]
# Validate the prim exists on the live stage -- ``create_tensor_binding`` silently
# returns a 0-count binding when the pattern matches nothing, surfacing as obscure
# AttributeErrors deep in property accessors. Also stash the concrete source-side
# root path for tendon discovery downstream.
stage = PhysicsManager._sim.stage
first_match = sim_utils.find_first_matching_prim(root_prim_path_expr, stage=stage)
if first_match is None:
raise RuntimeError(f"Failed to find articulation root prim at '{root_prim_path_expr}'.")
self._articulation_root_path = first_match.GetPath().pathString
# IsaacLab paths may use ``.*`` regex or ``{ENV_REGEX_NS}`` placeholder; ovphysx
# ``create_tensor_binding`` expects fnmatch globs.
pattern = re.sub(r"\{ENV_REGEX_NS\}", "*", root_prim_path_expr)
pattern = sim_utils.path_expr_to_glob(pattern)
self._binding_pattern = pattern
# eagerly create every binding the data container reads at init, so
# failures surface here rather than as KeyError downstream
eager_types = [
TT.ROOT_POSE,
TT.ROOT_VELOCITY,
TT.LINK_POSE,
TT.LINK_VELOCITY,
TT.LINK_ACCELERATION,
TT.DOF_POSITION,
TT.DOF_VELOCITY,
TT.JACOBIAN,
TT.MASS_MATRIX,
TT.GRAVITY_FORCE,
TT.DOF_STIFFNESS,
TT.DOF_DAMPING,
TT.DOF_LIMIT,
TT.DOF_MAX_VELOCITY,
TT.DOF_MAX_FORCE,
TT.DOF_ARMATURE,
TT.DOF_FRICTION_PROPERTIES,
TT.BODY_MASS,
TT.BODY_COM_POSE,
TT.BODY_INERTIA,
]
self._root_view = OvPhysxView(self._ovphysx, pattern=pattern, device=self._device)
# ``try_binding_for`` creates and caches each binding, returning ``None`` for tensor
# types that do not apply to these prims (so a minimal articulation that lacks some
# of these types is skipped rather than failing the whole init).
for tt in eager_types:
self._root_view.try_binding_for(tt)
if not self._root_view.available_attributes:
raise RuntimeError(
f"OVPhysX could not create any articulation bindings for pattern {pattern!r}. "
f"Check that prim_path={self.cfg.prim_path!r} matches at least one "
"UsdPhysics.ArticulationRootAPI prim."
)
# read metadata from the view (any instantiated binding carries it)
self._num_instances = self._root_view.count
self._num_joints = self._root_view.dof_count
self._num_bodies = self._root_view.body_count
self._is_fixed_base = self._root_view.is_fixed_base
self._joint_names = list(self._root_view.dof_names)
self._body_names = list(self._root_view.body_names)
# tendon counts/names must be resolved before buffer allocation
self._process_tendons()
# eagerly create tendon bindings now that the counts are known, through the same
# view, so ArticulationData reads them via a cached lookup (no lazy callback).
if self._num_fixed_tendons > 0:
for tt in (
TT.FIXED_TENDON_STIFFNESS,
TT.FIXED_TENDON_DAMPING,
TT.FIXED_TENDON_LIMIT_STIFFNESS,
TT.FIXED_TENDON_LIMIT,
TT.FIXED_TENDON_REST_LENGTH,
TT.FIXED_TENDON_OFFSET,
):
self._root_view.try_binding_for(tt)
if self._num_spatial_tendons > 0:
for tt in (
TT.SPATIAL_TENDON_STIFFNESS,
TT.SPATIAL_TENDON_DAMPING,
TT.SPATIAL_TENDON_LIMIT_STIFFNESS,
TT.SPATIAL_TENDON_OFFSET,
):
self._root_view.try_binding_for(tt)
# construct the data container; counts come from the view's bindings
joint_dof_signs = self._resolve_joint_dof_signs(stage)
self._data = ArticulationData(self._root_view, self._device)
if -1 in joint_dof_signs:
self._data._joint_dof_signs = wp.array(joint_dof_signs, dtype=wp.int32, device=self.device)
self._data._has_reversed_joints = True
self._resolve_and_install_ordering_maps()
self._data.fixed_tendon_names = self._fixed_tendon_names
self._data.spatial_tendon_names = self._spatial_tendon_names
# allocate asset-side buffers
self._create_buffers()
# apply initial state from config
self._process_cfg()
# build actuator instances and write drive properties to PhysX
self._process_actuators_cfg()
self._can_write_effort = self._get_binding(TT.DOF_ACTUATION_FORCE) is not None
self._can_write_pos_target = self._get_binding(TT.DOF_POSITION_TARGET) is not None
self._can_write_vel_target = self._get_binding(TT.DOF_VELOCITY_TARGET) is not None
# validate the resolved configuration AFTER actuator/tendon processing
# so the values reflect any overrides applied by the actuator models
self._validate_cfg()
# prime the data by performing the first read
self.update(0.0)
# mark data as ready
self._data.is_primed = True
def _resolve_joint_dof_signs(self, stage: Usd.Stage) -> tuple[int, ...]:
"""Resolve joint directions once from the source USD."""
source_prim = sim_utils.find_first_matching_prim(self.cfg.prim_path, stage=stage)
if source_prim is None:
return (1,) * self.num_joints
joint_prims_by_name = {}
for prim in Usd.PrimRange(source_prim, Usd.TraverseInstanceProxies()):
if not prim.IsA(UsdPhysics.Joint):
continue
joint_prims_by_name[_JOINT_KIND.resolve_target_name(prim)] = prim
body_indices = {name: index for index, name in enumerate(self._body_names)}
signs = []
for dof_name in self._joint_names:
canonical_dof_name = _canonical_joint_dof_name(dof_name)
matches = [
(name, prim)
for name, prim in joint_prims_by_name.items()
if canonical_dof_name == _canonical_joint_dof_name(name)
or canonical_dof_name.startswith(_canonical_joint_dof_name(name) + "_")
]
if not matches:
signs.append(1)
continue
joint = UsdPhysics.Joint(max(matches, key=lambda item: len(item[0]))[1])
body0 = joint.GetBody0Rel().GetTargets()
body1 = joint.GetBody1Rel().GetTargets()
body0_name = _BODY_KIND.resolve_target_name(stage.GetPrimAtPath(body0[0])) if body0 else ""
body1_name = _BODY_KIND.resolve_target_name(stage.GetPrimAtPath(body1[0])) if body1 else ""
body0_index = body_indices.get(body0_name)
body1_index = body_indices.get(body1_name)
signs.append(-1 if body0_index is not None and body1_index is not None and body0_index > body1_index else 1)
return tuple(signs)
def _create_buffers(self) -> None:
"""Allocate asset-side buffers (index/mask constants, wrench buf, pinned CPU staging)."""
N = self._num_instances
B = self._num_bodies
J = self._num_joints
FT = self._num_fixed_tendons
ST = self._num_spatial_tendons
device = self._device
# Index constants.
self._ALL_INDICES = wp.array(np.arange(N, dtype=np.int32), device=device)
self._ALL_BODY_INDICES = wp.array(np.arange(B, dtype=np.int32), device=device)
self._ALL_JOINT_INDICES = wp.array(np.arange(J, dtype=np.int32), device=device)
self._ALL_FIXED_TENDON_INDICES = wp.array(np.arange(FT, dtype=np.int32), device=device)
self._ALL_SPATIAL_TENDON_INDICES = wp.array(np.arange(ST, dtype=np.int32), device=device)
self._sim_env_ids = wp.empty(N, dtype=wp.int32, device=device)
self._sim_env_ids_views: dict[int, wp.array] = {}
self._joint_pos_target_backend: wp.array | None = None
self._joint_vel_target_backend: wp.array | None = None
self._joint_effort_target_backend: wp.array | None = None
self._applied_torque_backend: wp.array | None = None
self._ordering_configure_backend_staging()
# All-true masks.
self._ALL_TRUE_ENV_MASK = wp.array(np.ones(N, dtype=bool), dtype=wp.bool, device=device)
self._ALL_TRUE_BODY_MASK = wp.array(np.ones(B, dtype=bool), dtype=wp.bool, device=device)
self._ALL_TRUE_JOINT_MASK = wp.array(np.ones(J, dtype=bool), dtype=wp.bool, device=device)
self._ALL_TRUE_FIXED_TENDON_MASK = wp.array(np.ones(FT, dtype=bool), dtype=wp.bool, device=device)
self._ALL_TRUE_SPATIAL_TENDON_MASK = wp.array(np.ones(ST, dtype=bool), dtype=wp.bool, device=device)
# Wrench buffer (force, torque, position) per body, written by the
# ``_body_wrench_to_world_ordered`` kernel and consumed by the
# ``LINK_WRENCH`` binding which expects the 3D ``(N, B, 9)`` shape.
self._wrench_buf = wp.zeros((N, B, 9), dtype=wp.float32, device=device)
# Wrench composers.
self._instantaneous_wrench_composer = WrenchComposer(self)
self._permanent_wrench_composer = WrenchComposer(self)
# Wrench scratch buffer (used by _apply_external_wrenches, not yet allocated above).
# Joint-index arrays for each actuator (populated by _process_actuators_cfg).
self._joint_ids_per_actuator: dict[str, slice | torch.Tensor] = {}
# Pinned-host CPU staging for env ids/masks (PR #5329 pattern).
self._cpu_env_ids_all = wp.zeros(N, dtype=wp.int32, device="cpu", pinned=True)
wp.copy(self._cpu_env_ids_all, self._ALL_INDICES)
self._cpu_env_ids = wp.empty(N, dtype=wp.int32, device="cpu", pinned=True)
self._cpu_env_ids_views: dict[int, wp.array] = {}
self._cpu_env_mask = wp.zeros(N, dtype=wp.bool, device="cpu", pinned=True)
def _process_cfg(self) -> None:
"""Populate default state buffers from the config (mirrors RigidObject and Newton Articulation)."""
cfg = self.cfg
N = self._num_instances
D = self._num_joints
dev = self._device
# Default root state from config (matching PhysX pattern).
default_root_pose = tuple(cfg.init_state.pos) + tuple(cfg.init_state.rot)
default_root_vel = tuple(cfg.init_state.lin_vel) + tuple(cfg.init_state.ang_vel)
np_pose = np.tile(np.array(default_root_pose, dtype=np.float32), (N, 1))
np_vel = np.tile(np.array(default_root_vel, dtype=np.float32), (N, 1))
self._data.default_root_pose = wp.array(np_pose, dtype=wp.transformf, device=dev)
self._data.default_root_vel = wp.array(np_vel, dtype=wp.spatial_vectorf, device=dev)
# Default joint positions / velocities from config patterns.
# cfg.init_state.joint_pos is a dict[str, float] where keys are regex patterns
# matching joint names. We expand this into a (N, D) buffer.
self._resolve_joint_values(cfg.init_state.joint_pos, self._data._default_joint_pos)
self._resolve_joint_values(cfg.init_state.joint_vel, self._data._default_joint_vel)
# Compute soft joint position limits from the hard limits read from the binding
# (or zeros if no joints). This matches the PhysX/Newton path.
if D > 0:
wp.launch(
update_soft_joint_pos_limits,
dim=(N, D),
inputs=[self._data.joint_pos_limits, cfg.soft_joint_pos_limit_factor],
outputs=[self._data._soft_joint_pos_limits],
device=dev,
)
def _process_tendons(self) -> None:
"""Discover tendon counts from binding metadata and names from USD.
Tendon counts come from the ovphysx binding metadata. Tendon names are
recovered from the in-memory USD articulation subtree because ovphysx
exposes joint names/counts, but not the per-joint USD paths that the
PhysX backend can query directly.
"""
self._fixed_tendon_names = []
self._spatial_tendon_names = []
self._num_fixed_tendons = self._root_view.fixed_tendon_count
self._num_spatial_tendons = self._root_view.spatial_tendon_count
if self._num_fixed_tendons > 0 or self._num_spatial_tendons > 0:
stage_usda = OvPhysxManager._stage_usda
if stage_usda is not None:
try:
from pxr import Sdf, Usd
from isaaclab.sim.utils.queries import get_all_matching_child_prims
layer = Sdf.Layer.CreateAnonymous("isaaclab_ov.usda")
if not layer.ImportFromString(stage_usda):
raise RuntimeError("Failed to import the serialized OVPhysX stage.")
stage = Usd.Stage.Open(layer)
articulation_root_path = getattr(self, "_articulation_root_path", None)
if articulation_root_path is None:
joint_prims = stage.Traverse()
else:
joint_prims = get_all_matching_child_prims(
articulation_root_path,
predicate=lambda p: p.IsA(UsdPhysics.Joint),
stage=stage,
traverse_instance_prims=False,
)
for prim in joint_prims:
if not prim.IsA(UsdPhysics.Joint):
continue
schema_names = list(prim.GetAppliedSchemas())
metadata = prim.GetMetadata("apiSchemas")
if metadata is not None:
for field in ("prependedItems", "appendedItems", "explicitItems"):
items = getattr(metadata, field, None)
if items:
schema_names.extend(str(item) for item in items)
schemas_str = " ".join(schema_names)
name = prim.GetPath().name
if "PhysxTendonAxisRootAPI" in schemas_str:
self._fixed_tendon_names.append(name)
elif (
"PhysxTendonAttachmentRootAPI" in schemas_str
or "PhysxTendonAttachmentLeafAPI" in schemas_str
):
self._spatial_tendon_names.append(name)
except Exception:
logger.debug("Could not parse in-memory USD stage for tendon names", exc_info=True)
def _get_binding(self, tensor_type: int):
"""Return a cached TensorBinding, creating it on first access.
Bindings are lightweight handles (a pointer + shape metadata into
PhysX's shared GPU buffer). Creating one does NOT allocate new GPU
memory -- the underlying simulation buffers are allocated once by PhysX
regardless of how many bindings point into them. Still, we defer
creation so that tensor types the user never queries are never looked up.
Args:
tensor_type: The TensorType constant identifying which simulation
buffer to bind (e.g. :attr:`~isaaclab_ov.tensor_types.ROOT_POSE`).
Returns:
A TensorBinding object, or ``None`` if the binding could not be created.
"""
return self._root_view.try_binding_for(tensor_type)
def _resolve_joint_values(self, pattern_dict: dict[str, float], buffer: wp.array) -> None:
"""Resolve a ``{pattern: value}`` dict into a per-joint buffer.
Builds values on CPU then copies to buffer's device (GPU arrays'
``.numpy()`` returns a read-only copy, not a writable view).
Args:
pattern_dict: A mapping from regex pattern strings to scalar values.
Matches joint names returned by :attr:`joint_names`.
buffer: Target warp array of shape ``(num_instances, num_joints)``
to populate.
"""
buf_np = buffer.numpy()
modified = False
for pattern, value in pattern_dict.items():
for j, name in enumerate(self.joint_names):
if re.fullmatch(pattern, name):
buf_np[:, j] = value
modified = True
if modified:
wp.copy(buffer, wp.from_numpy(buf_np, dtype=buffer.dtype, device=str(buffer.device)))
def _n_envs_index(self, env_ids) -> int:
"""Return the number of environments from an ``env_ids`` argument."""
if env_ids is None:
return self._num_instances
if isinstance(env_ids, (list, tuple)):
return len(env_ids)
return env_ids.shape[0] if hasattr(env_ids, "shape") else len(env_ids)
def _nft(self) -> int:
"""Return the number of fixed tendons (0 if none)."""
return self._num_fixed_tendons
def _nst(self) -> int:
"""Return the number of spatial tendons (0 if none)."""
return self._num_spatial_tendons
"""
Internal simulation callbacks.
"""
def _invalidate_initialize_callback(self, event) -> None:
"""Invalidate the asset on simulation reset."""
super()._invalidate_initialize_callback(event)
# Drop the view (and the bindings it caches) on stop so a destroyed/stale binding is
# not held across the reset; ``_initialize_impl`` rebuilds a fresh view on the next play.
self._root_view = None
"""
Internal helpers -- Ordering.
"""
_ordering_joint_staging_names: tuple[str, ...] = (
"_joint_pos_target_backend",
"_joint_vel_target_backend",
"_joint_effort_target_backend",
"_applied_torque_backend",
)
"""Backend-order joint staging buffers managed by :meth:`_ordering_configure_backend_staging`.
``_joint_pos_target_backend`` / ``_joint_vel_target_backend`` / ``_joint_effort_target_backend``
are persistent backend-order mirrors of the corresponding user-order target buffers, kept
current by the partial :meth:`set_joint_position_target_index`-style setters.
``_applied_torque_backend`` is separate, purely transient scratch: :meth:`write_data_to_sim`
fully overwrites it every step with the backend-order actuator output, so it must not alias
``_joint_effort_target_backend`` (whose unselected rows a partial effort-target write relies
on to still hold the persisted target, not the last pushed applied torque).
"""
"""
Internal helpers -- Actuators.
"""
def _process_actuators_cfg(self) -> None:
"""Build actuator instances from the config and write drive properties to PhysX.
Mirrors the PhysX backend's ``_process_actuators_cfg``:
* For :class:`~isaaclab.actuators.ImplicitActuator`: write the configured
stiffness/damping to the PhysX drive so the solver uses exactly those values.
* For all explicit actuators: zero out PhysX stiffness/damping so USD-authored
drive gains cannot interfere with the explicit torque path.
* For all actuators: write :attr:`~isaaclab.actuators.ActuatorBase.effort_limit_sim`
and :attr:`~isaaclab.actuators.ActuatorBase.velocity_limit_sim`.
"""
from isaaclab.actuators import ImplicitActuator
self.actuators: dict[str, Any] = {}
self._has_implicit_actuators = False
for name, act_cfg in self.cfg.actuators.items():
joint_ids, joint_names = self.find_joints(act_cfg.joint_names_expr, as_proxy=True)
if not joint_names:
logger.warning("Actuator '%s': no joints matched '%s'", name, act_cfg.joint_names_expr)
continue
actuator_joint_ids = slice(None) if joint_names == self.joint_names else joint_ids.torch
torch_joint_ids = actuator_joint_ids
act_cfg_copy = act_cfg.copy()
# seed the actuator with the simulation's already-correct DOF defaults
# (USD-authored ``physxJoint:maxJointVelocity`` etc. parsed at scene-load).
# Without these the ActuatorBase constructor falls back to ``inf`` for unset
# cfg fields, and the ``write_joint_*_to_sim_index`` calls below then
# overwrite the correct values with ``inf``.
act = act_cfg_copy.class_type(
act_cfg_copy,
joint_names=joint_names,
joint_ids=actuator_joint_ids,
num_envs=self._num_instances,
device=self._device,
stiffness=self._data.joint_stiffness.torch[:, torch_joint_ids],
damping=self._data.joint_damping.torch[:, torch_joint_ids],
armature=self._data.joint_armature.torch[:, torch_joint_ids],
friction=self._data.joint_friction_coeff.torch[:, torch_joint_ids],
dynamic_friction=self._data.joint_dynamic_friction_coeff.torch[:, torch_joint_ids],
viscous_friction=self._data.joint_viscous_friction_coeff.torch[:, torch_joint_ids],
effort_limit=self._data.joint_effort_limits.torch[:, torch_joint_ids].clone(),
velocity_limit=self._data.joint_vel_limits.torch[:, torch_joint_ids],
)
self.actuators[name] = act
self._joint_ids_per_actuator[name] = actuator_joint_ids
# Write drive gains and limits to PhysX to match the actuator config.
# Without this, PhysX retains whatever stiffness/damping was authored in the
# USD file, which can produce large restoring forces when the USD gains differ
# from the actuator config.
if isinstance(act, ImplicitActuator):
self._has_implicit_actuators = True
stiffness = act.stiffness # torch (N, J)
damping = act.damping # torch (N, J)
else:
stiffness = wp.zeros((self._num_instances, len(joint_names)), dtype=wp.float32, device=self._device)
damping = wp.zeros((self._num_instances, len(joint_names)), dtype=wp.float32, device=self._device)
self.write_joint_stiffness_to_sim_index(stiffness=stiffness, joint_ids=actuator_joint_ids)
self.write_joint_damping_to_sim_index(damping=damping, joint_ids=actuator_joint_ids)
self.write_joint_effort_limit_to_sim_index(limits=act.effort_limit_sim, joint_ids=actuator_joint_ids)
self.write_joint_velocity_limit_to_sim_index(limits=act.velocity_limit_sim, joint_ids=actuator_joint_ids)
def _apply_actuator_model(self) -> None:
"""Run the actuator model to compute joint torques from user-supplied targets.
IsaacLab actuators are torch-based. The method converts Warp buffers to
torch via DLPack (zero-copy on GPU), runs each actuator's
:meth:`~isaaclab.actuators.ActuatorBase.compute` method, then writes the
computed effort back to the private ``_computed_torque`` / ``_applied_torque``
buffers of the data container. :meth:`write_data_to_sim` then pushes
``_applied_torque`` to the ``DOF_ACTUATION_FORCE`` binding in one shot.
"""
from isaaclab.utils.types import ArticulationActions
for name, act in self.actuators.items():
joint_ids = self._joint_ids_per_actuator[name]
all_joints = isinstance(joint_ids, slice)
torch_joint_ids = joint_ids
# Warp -> torch (zero-copy on same device via DLPack).
jp_target_full = self._data.joint_pos_target.torch
jv_target_full = self._data.joint_vel_target.torch
je_target_full = self._data.joint_effort_target.torch
jp_target = jp_target_full if all_joints else jp_target_full[:, torch_joint_ids]
jv_target = jv_target_full if all_joints else jv_target_full[:, torch_joint_ids]
je_target = je_target_full if all_joints else je_target_full[:, torch_joint_ids]
control_action = ArticulationActions(
joint_positions=jp_target,
joint_velocities=jv_target,
joint_efforts=je_target,
)
jp_cur_full = self._data.joint_pos.torch
jv_cur_full = self._data.joint_vel.torch
jp_cur = jp_cur_full if all_joints else jp_cur_full[:, torch_joint_ids]
jv_cur = jv_cur_full if all_joints else jv_cur_full[:, torch_joint_ids]
control_action = act.compute(control_action, jp_cur, jv_cur)
if act.computed_effort is not None:
ct = wp.to_torch(self._data._computed_torque)
at = wp.to_torch(self._data._applied_torque)
if all_joints:
ct[:] = act.computed_effort
at[:] = act.applied_effort
else:
ct[:, torch_joint_ids] = act.computed_effort
at[:, torch_joint_ids] = act.applied_effort
"""
Internal helpers -- Debugging.
"""
def _validate_cfg(self) -> None:
"""Validate the configuration after processing.
Mirrors :meth:`isaaclab_physx.assets.Articulation._validate_cfg` (raises
``ValueError`` with a per-joint message when any default joint position
is outside ``[lower, upper]`` or any default joint velocity exceeds the
per-joint max velocity). Reads come from :attr:`ArticulationData`
accessors instead of PhysX's ``root_view.get_dof_limits`` /
``get_dof_max_velocities`` because OVPhysX's ``root_view`` is the
per-tensor-type bindings dict.
.. note::
Must be called only after :meth:`_create_buffers` /
:meth:`_process_cfg` / :meth:`_process_actuators_cfg`, otherwise
limits and defaults may not yet reflect the final values.
"""
# check that the default joint positions are within the limits
joint_pos_limits = self._data.joint_pos_limits.torch[0] # (num_joints, 2)
default_joint_pos = self._data.default_joint_pos.torch[0] # (num_joints,)
out_of_range = default_joint_pos < joint_pos_limits[:, 0]
out_of_range |= default_joint_pos > joint_pos_limits[:, 1]
violated_indices = torch.nonzero(out_of_range, as_tuple=False).squeeze(-1)
if len(violated_indices) > 0:
msg = "The following joints have default positions out of the limits: \n"
for idx in violated_indices:
joint_name = self._data.joint_names[idx]
joint_limit = joint_pos_limits[idx]
joint_pos = default_joint_pos[idx]
msg += f"\t- '{joint_name}': {joint_pos:.3f} not in [{joint_limit[0]:.3f}, {joint_limit[1]:.3f}]\n"
raise ValueError(msg)
# check that the default joint velocities are within the limits
joint_max_vel = self._data.joint_vel_limits.torch[0] # (num_joints,)
default_joint_vel = self._data.default_joint_vel.torch[0] # (num_joints,)
out_of_range = torch.abs(default_joint_vel) > joint_max_vel
violated_indices = torch.nonzero(out_of_range, as_tuple=False).squeeze(-1)
if len(violated_indices) > 0:
msg = "The following joints have default velocities out of the limits: \n"
for idx in violated_indices:
joint_name = self._data.joint_names[idx]
joint_limit = [-joint_max_vel[idx], joint_max_vel[idx]]
joint_vel = default_joint_vel[idx]
msg += f"\t- '{joint_name}': {joint_vel:.3f} not in [{joint_limit[0]:.3f}, {joint_limit[1]:.3f}]\n"
raise ValueError(msg)
def _log_articulation_info(self) -> None:
pass
def _resolve_env_ids(self, env_ids: Sequence[int] | torch.Tensor | wp.array | None) -> wp.array | torch.Tensor:
"""Resolve environment indices on ``self._device``."""
if env_ids is None or (isinstance(env_ids, slice) and env_ids == slice(None)):
return self._ALL_INDICES
if isinstance(env_ids, ProxyArray):
raise TypeError("ProxyArray is output-only; pass .warp or .torch explicitly.")
if isinstance(env_ids, list):
return wp.array(env_ids, dtype=wp.int32, device=self._device)
if isinstance(env_ids, torch.Tensor):
return env_ids.to(device=self._device)
if isinstance(env_ids, wp.array) and str(env_ids.device) != self._device:
env_ids = wp.clone(env_ids, device=self._device)
return env_ids
def _resolve_body_ids(self, body_ids: Sequence[int] | torch.Tensor | wp.array | None) -> wp.array | torch.Tensor:
"""Resolve body indices to a Warp signed-integer array on ``self._device``."""
if isinstance(body_ids, ProxyArray):
raise TypeError("ProxyArray is output-only; pass .warp or .torch explicitly.")
if body_ids is None or body_ids == slice(None):
return self._ALL_BODY_INDICES
if isinstance(body_ids, list):
return wp.array(body_ids, dtype=wp.int32, device=self._device)
if isinstance(body_ids, torch.Tensor):
return body_ids.to(device=self._device)
if isinstance(body_ids, wp.array) and str(body_ids.device) != self._device:
body_ids = wp.clone(body_ids, device=self._device)
return body_ids
def _resolve_joint_ids(self, joint_ids: Sequence[int] | torch.Tensor | wp.array | None) -> wp.array | torch.Tensor:
"""Resolve joint indices to a Warp signed-integer array on ``self._device``."""
if isinstance(joint_ids, ProxyArray):
raise TypeError("ProxyArray is output-only; pass .warp or .torch explicitly.")
if joint_ids is None or joint_ids == slice(None):
return self._ALL_JOINT_INDICES
if isinstance(joint_ids, list):
return wp.array(joint_ids, dtype=wp.int32, device=self._device)
if isinstance(joint_ids, torch.Tensor):
return joint_ids.to(device=self._device)
if isinstance(joint_ids, wp.array) and str(joint_ids.device) != self._device:
joint_ids = wp.clone(joint_ids, device=self._device)
return joint_ids
def _resolve_fixed_tendon_ids(
self, tendon_ids: Sequence[int] | torch.Tensor | wp.array | None
) -> wp.array | torch.Tensor:
"""Resolve fixed-tendon indices to a Warp signed-integer array on ``self._device``."""
if isinstance(tendon_ids, ProxyArray):
raise TypeError("ProxyArray is output-only; pass .warp or .torch explicitly.")
if tendon_ids is None or tendon_ids == slice(None):
return self._ALL_FIXED_TENDON_INDICES
if isinstance(tendon_ids, list):
return wp.array(tendon_ids, dtype=wp.int32, device=self._device)
if isinstance(tendon_ids, torch.Tensor):
return tendon_ids.to(device=self._device)
if isinstance(tendon_ids, wp.array) and str(tendon_ids.device) != self._device:
tendon_ids = wp.clone(tendon_ids, device=self._device)
return tendon_ids
def _resolve_spatial_tendon_ids(
self, tendon_ids: Sequence[int] | torch.Tensor | wp.array | None
) -> wp.array | torch.Tensor:
"""Resolve spatial-tendon indices to a Warp signed-integer array on ``self._device``."""
if isinstance(tendon_ids, ProxyArray):
raise TypeError("ProxyArray is output-only; pass .warp or .torch explicitly.")
if tendon_ids is None or tendon_ids == slice(None):
return self._ALL_SPATIAL_TENDON_INDICES
if isinstance(tendon_ids, list):
return wp.array(tendon_ids, dtype=wp.int32, device=self._device)
if isinstance(tendon_ids, torch.Tensor):
return tendon_ids.to(device=self._device)
if isinstance(tendon_ids, wp.array) and str(tendon_ids.device) != self._device:
tendon_ids = wp.clone(tendon_ids, device=self._device)
return tendon_ids
def _broadcast_scalar_to_2d(
self, value: float | torch.Tensor | wp.array, shape: tuple[int, int]
) -> torch.Tensor | wp.array:
"""Broadcast a scalar :class:`float` to a ``(rows, cols)`` torch ``float32`` tensor.
Tendon and joint setters accept ``float | torch.Tensor | wp.array``; the
underlying ``shared_kernels.write_2d_data_to_buffer_*`` kernels only
accept 2D arrays. This helper expands a Python float into a constant
tensor on :attr:`_device`; tensor / warp inputs are returned as-is.
Mirrors the PhysX backend's ``isinstance(value, float)`` branching,
which dispatches to ``articulation_kernels.float_data_to_buffer_with_*``.
OVPhysX does not have those scalar kernels, so we materialize the
broadcast on the Python side.
Args:
value: Scalar float or 2D tensor / warp array.
shape: ``(rows, cols)`` target shape used when broadcasting a
scalar.
Returns:
A 2D :class:`torch.Tensor` on ``self._device`` if *value* was a
float; otherwise *value* unchanged.
"""
if isinstance(value, float):
return torch.full(shape, value, dtype=torch.float32, device=self._device)
return value
def _resolve_env_mask(self, env_mask: wp.array | None) -> wp.array:
"""Resolve an environment mask to a ``wp.bool`` array on ``self._device``.
OVPhysX (like Newton) writes through the view's ``set_attribute(mask=...)``, which
forwards to the binding's native masked write, so the mask is preserved end-to-end;
no ``torch.nonzero`` conversion is needed. ``None`` returns the pre-allocated
all-true mask.
"""
if env_mask is None:
return self._ALL_TRUE_ENV_MASK
if isinstance(env_mask, torch.Tensor):
return wp.from_torch(env_mask.to(torch.bool), dtype=wp.bool)
if isinstance(env_mask, wp.array) and str(env_mask.device) != self._device:
env_mask = wp.clone(env_mask, device=self._device)
return env_mask
def _sim_env_ids_view(self, count: int) -> wp.array:
"""Return a cached prefix view of the simulator-index scratch buffer."""
if count not in self._sim_env_ids_views:
self._sim_env_ids_views[count] = wp.array(
ptr=self._sim_env_ids.ptr,
shape=(count,),
dtype=wp.int32,
device=self._device,
copy=False,
)
return self._sim_env_ids_views[count]
def _resolve_body_mask(self, body_mask: wp.array | None) -> wp.array:
"""Resolve a body mask to a ``wp.bool`` array on ``self._device`` (Newton-style)."""
if body_mask is None:
return self._ALL_TRUE_BODY_MASK
if isinstance(body_mask, torch.Tensor):
return wp.from_torch(body_mask.to(torch.bool), dtype=wp.bool)
if isinstance(body_mask, wp.array) and str(body_mask.device) != self._device:
body_mask = wp.clone(body_mask, device=self._device)
return body_mask
def _resolve_joint_mask(self, joint_mask: wp.array | None) -> wp.array:
"""Resolve a joint mask to a ``wp.bool`` array on ``self._device``."""
if joint_mask is None:
return self._ALL_TRUE_JOINT_MASK
if isinstance(joint_mask, torch.Tensor):
return wp.from_torch(joint_mask.to(torch.bool), dtype=wp.bool)
if isinstance(joint_mask, wp.array) and str(joint_mask.device) != self._device:
joint_mask = wp.clone(joint_mask, device=self._device)
return joint_mask
def _resolve_fixed_tendon_mask(self, tendon_mask: wp.array | None) -> wp.array:
"""Resolve a fixed-tendon mask to a ``wp.bool`` array on ``self._device``."""
if tendon_mask is None:
return self._ALL_TRUE_FIXED_TENDON_MASK
if isinstance(tendon_mask, torch.Tensor):
return wp.from_torch(tendon_mask.to(torch.bool), dtype=wp.bool)
if isinstance(tendon_mask, wp.array) and str(tendon_mask.device) != self._device:
tendon_mask = wp.clone(tendon_mask, device=self._device)
return tendon_mask
def _resolve_spatial_tendon_mask(self, tendon_mask: wp.array | None) -> wp.array:
"""Resolve a spatial-tendon mask to a ``wp.bool`` array on ``self._device``."""
if tendon_mask is None:
return self._ALL_TRUE_SPATIAL_TENDON_MASK
if isinstance(tendon_mask, torch.Tensor):
return wp.from_torch(tendon_mask.to(torch.bool), dtype=wp.bool)
if isinstance(tendon_mask, wp.array) and str(tendon_mask.device) != self._device:
tendon_mask = wp.clone(tendon_mask, device=self._device)
return tendon_mask
def _get_cpu_env_mask(self, env_mask: wp.array) -> wp.array:
"""Return a pinned-host CPU copy of :paramref:`env_mask` for a CPU-only binding write.
:paramref:`env_mask` is normally on ``self._device``; ``set_attribute(mask=...)``
requires the mask on the binding's native device, which is CPU for mass / CoMs /
inertia. Reuses the pre-allocated ``_cpu_env_mask`` pinned buffer.
"""
wp.copy(self._cpu_env_mask, env_mask)
return self._cpu_env_mask
def _get_cpu_env_ids(self, env_ids: wp.array | torch.Tensor, sim_env_ids: wp.array | None = None) -> wp.array:
"""Return CPU int32 indices, using the pre-allocated pinned ``_cpu_env_ids_all``
fast path when *env_ids* matches ``_ALL_INDICES`` (PR #5329 pattern).
"""
if isinstance(env_ids, torch.Tensor):
if env_ids.dtype == torch.int64 and sim_env_ids is None:
return wp.from_torch(env_ids.to(device="cpu", dtype=torch.int32), dtype=wp.int32)
env_ids = wp.from_torch(env_ids)
if env_ids.ptr == self._ALL_INDICES.ptr:
return self._cpu_env_ids_all
if env_ids.dtype == wp.int64:
if sim_env_ids is None:
return wp.from_torch(wp.to_torch(env_ids).to(device="cpu", dtype=torch.int32), dtype=wp.int32)
env_ids = sim_env_ids
if str(env_ids.device) == "cpu":
return env_ids
cpu_env_ids = self._cpu_env_ids_view(env_ids.shape[0])
wp.copy(cpu_env_ids, env_ids)
return cpu_env_ids
def _get_sim_env_ids(self, env_ids: wp.array | torch.Tensor, sim_env_ids: wp.array | None = None) -> wp.array:
"""Return int32 environment indices for OVPhysX."""
if isinstance(env_ids, torch.Tensor):
if env_ids.dtype == torch.int64 and sim_env_ids is None:
return wp.from_torch(env_ids.to(device=self._device, dtype=torch.int32), dtype=wp.int32)
env_ids = wp.from_torch(env_ids)
if env_ids.dtype == wp.int64:
if sim_env_ids is None:
return wp.from_torch(wp.to_torch(env_ids).to(device=self._device, dtype=torch.int32), dtype=wp.int32)
return sim_env_ids
if str(env_ids.device) != self._device:
return wp.clone(env_ids, device=self._device)
return env_ids
def _cpu_env_ids_view(self, count: int) -> wp.array:
"""Return a cached prefix of the CPU simulator-index scratch buffer."""
if count not in self._cpu_env_ids_views:
self._cpu_env_ids_views[count] = wp.array(
ptr=self._cpu_env_ids.ptr,
shape=(count,),
dtype=wp.int32,
device="cpu",
copy=False,
)
return self._cpu_env_ids_views[count]
def _push_joint_property(
self,
tensor_type: int,
user_buffer: wp.array,
backend_buffer: wp.array | TimestampedBufferWarp | None,
*,
cpu_buffer: wp.array | None = None,
component_count: int | None = None,
indices: wp.array | None = None,
mask: wp.array | None = None,
) -> None:
"""Push a public-order joint property through backend and CPU staging."""
property_backend = self._get_backend_ordered_joint_buffer(
user_buffer,
backend_buffer,
component_count=component_count,
)
if cpu_buffer is None:
cpu_buffer = self._data._stage_to_pinned_cpu(tensor_type, "write", property_backend)
else:
source = property_backend
if source.dtype != wp.float32:
source = wp.array(
ptr=source.ptr,
shape=cpu_buffer.shape,
dtype=wp.float32,
device=str(source.device),
copy=False,
)
wp.copy(cpu_buffer, source)
if indices is not None:
self._root_view.set_attribute(tensor_type, cpu_buffer, indices=indices)
else:
self._root_view.set_attribute(tensor_type, cpu_buffer, mask=mask)
"""
Deprecated methods.
"""
def write_root_state_to_sim(
self,
root_state: torch.Tensor | wp.array,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Deprecated; use :meth:`write_root_link_pose_to_sim_index` and
:meth:`write_root_com_velocity_to_sim_index` instead.
Args:
root_state: Root state [m, m, m, qw, qx, qy, qz, m/s, m/s, m/s, rad/s, rad/s, rad/s].
Shape is (len(env_ids), 13) with dtype wp.float32.
env_ids: Environment indices. Defaults to None (all environments).
"""
warnings.warn(
"The function 'write_root_state_to_sim' will be deprecated in a future release. Please"
" use 'write_root_link_pose_to_sim_index' and 'write_root_com_velocity_to_sim_index' instead.",
DeprecationWarning,
stacklevel=2,
)
self.write_root_link_pose_to_sim_index(root_pose=root_state[:, :7], env_ids=env_ids)
self.write_root_com_velocity_to_sim_index(root_velocity=root_state[:, 7:], env_ids=env_ids)
def write_root_com_state_to_sim(
self,
root_state: torch.Tensor | wp.array,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Deprecated; use :meth:`write_root_com_pose_to_sim_index` and
:meth:`write_root_com_velocity_to_sim_index` instead.
Args:
root_state: Root CoM state [m, m, m, qw, qx, qy, qz, m/s, m/s, m/s, rad/s, rad/s, rad/s].
Shape is (len(env_ids), 13) with dtype wp.float32.
env_ids: Environment indices. Defaults to None (all environments).
"""
warnings.warn(
"The function 'write_root_com_state_to_sim' will be deprecated in a future release. Please"
" use 'write_root_com_pose_to_sim_index' and 'write_root_com_velocity_to_sim_index' instead.",
DeprecationWarning,
stacklevel=2,
)
self.write_root_com_pose_to_sim_index(root_pose=root_state[:, :7], env_ids=env_ids)
self.write_root_com_velocity_to_sim_index(root_velocity=root_state[:, 7:], env_ids=env_ids)
def write_root_link_state_to_sim(
self,
root_state: torch.Tensor | wp.array,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Deprecated; use :meth:`write_root_link_pose_to_sim_index` and
:meth:`write_root_link_velocity_to_sim_index` instead.
Args:
root_state: Root link state [m, m, m, qw, qx, qy, qz, m/s, m/s, m/s, rad/s, rad/s, rad/s].
Shape is (len(env_ids), 13) with dtype wp.float32.
env_ids: Environment indices. Defaults to None (all environments).
"""
warnings.warn(
"The function 'write_root_link_state_to_sim' will be deprecated in a future release. Please"
" use 'write_root_link_pose_to_sim_index' and 'write_root_link_velocity_to_sim_index' instead.",
DeprecationWarning,
stacklevel=2,
)
self.write_root_link_pose_to_sim_index(root_pose=root_state[:, :7], env_ids=env_ids)
self.write_root_link_velocity_to_sim_index(root_velocity=root_state[:, 7:], env_ids=env_ids)
def write_joint_state_to_sim(
self,
position: torch.Tensor | wp.array,
velocity: torch.Tensor | wp.array,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Deprecated combined joint-state write; use :meth:`write_joint_position_to_sim_index`
and :meth:`write_joint_velocity_to_sim_index` instead.
Args:
position: Joint positions [m or rad, depending on joint type]. Shape is
(len(env_ids), len(joint_ids)) with dtype wp.float32.
velocity: Joint velocities [m/s or rad/s, depending on joint type]. Shape is
(len(env_ids), len(joint_ids)) with dtype wp.float32.
joint_ids: Joint indices. Defaults to None (all joints).
env_ids: Environment indices. Defaults to None (all environments).
"""
warnings.warn(
"write_joint_state_to_sim is deprecated; use write_joint_position_to_sim_index"
" and write_joint_velocity_to_sim_index instead.",
DeprecationWarning,
stacklevel=2,
)
self.write_joint_position_to_sim_index(position=position, joint_ids=joint_ids, env_ids=env_ids)
self.write_joint_velocity_to_sim_index(velocity=velocity, joint_ids=joint_ids, env_ids=env_ids)
def write_joint_friction_coefficient_to_sim(
self,
joint_friction_coeff: torch.Tensor | wp.array | float,
joint_dynamic_friction_coeff: torch.Tensor | wp.array | float | None = None,
joint_viscous_friction_coeff: torch.Tensor | wp.array | float | None = None,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Deprecated, same as :meth:`write_joint_friction_coefficient_to_sim_index`."""
warnings.warn(
"The function 'write_joint_friction_coefficient_to_sim' will be deprecated in a future release. Please"
" use 'write_joint_friction_coefficient_to_sim_index' instead.",
DeprecationWarning,
stacklevel=2,
)
self.write_joint_friction_coefficient_to_sim_index(
joint_friction_coeff=joint_friction_coeff,
joint_dynamic_friction_coeff=joint_dynamic_friction_coeff,
joint_viscous_friction_coeff=joint_viscous_friction_coeff,
joint_ids=joint_ids,
env_ids=env_ids,
)
def write_joint_dynamic_friction_coefficient_to_sim(
self,
joint_dynamic_friction_coeff: torch.Tensor | wp.array | float,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Deprecated, same as :meth:`write_joint_dynamic_friction_coefficient_to_sim_index`."""
warnings.warn(
"The function 'write_joint_dynamic_friction_coefficient_to_sim' will be deprecated in a future release. "
"Please use 'write_joint_dynamic_friction_coefficient_to_sim_index' instead.",
DeprecationWarning,
stacklevel=2,
)
self.write_joint_dynamic_friction_coefficient_to_sim_index(
joint_dynamic_friction_coeff=joint_dynamic_friction_coeff,
joint_ids=joint_ids,
env_ids=env_ids,
)
def write_joint_viscous_friction_coefficient_to_sim(
self,
joint_viscous_friction_coeff: torch.Tensor | wp.array | float,
joint_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
env_ids: Sequence[int] | torch.Tensor | wp.array | None = None,
) -> None:
"""Deprecated, same as :meth:`write_joint_viscous_friction_coefficient_to_sim_index`."""
warnings.warn(
"The function 'write_joint_viscous_friction_coefficient_to_sim' will be deprecated in a future release. "
"Please use 'write_joint_viscous_friction_coefficient_to_sim_index' instead.",
DeprecationWarning,
stacklevel=2,
)
self.write_joint_viscous_friction_coefficient_to_sim_index(
joint_viscous_friction_coeff=joint_viscous_friction_coeff,
joint_ids=joint_ids,
env_ids=env_ids,
)
