Source code for embodichain.lab.sim.objects.backends.base

# ----------------------------------------------------------------------------
# Copyright (c) 2021-2026 DexForce Technology Co., Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ----------------------------------------------------------------------------
from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Sequence
from functools import cached_property

import torch

__all__ = ["RigidBodyViewBase", "ArticulationViewBase"]


[docs] class RigidBodyViewBase(ABC): """Abstract interface for physics-backend rigid body data access. All pose/velocity/acceleration data uses EmbodiChain convention: ``(x, y, z, qx, qy, qz, qw)``. """ # -- Lifecycle & State -------------------------------------------------- @property @abstractmethod def is_ready(self) -> bool: """Whether the backend simulation is finalized and data can be accessed.""" ... @property def can_apply_pose(self) -> bool: """Whether world poses can be written through the backend view.""" return self.is_ready @property def can_fetch_pose(self) -> bool: """Whether world poses can be read through the backend view.""" return self.is_ready @property def is_newton_backend(self) -> bool: """Whether this view targets the DexSim Newton backend.""" return False # -- Body ID Management ------------------------------------------------- @cached_property @abstractmethod def body_ids(self) -> list[int]: """Backend body IDs for all managed entities.""" ... @cached_property @abstractmethod def body_ids_tensor(self) -> torch.Tensor: """Body IDs as an int32 tensor on ``device``.""" ...
[docs] @abstractmethod def select_body_ids(self, indices: Sequence[int] | torch.Tensor) -> torch.Tensor: """Return body IDs for the given entity indices.""" ...
# -- Pose ---------------------------------------------------------------
[docs] @abstractmethod def fetch_pose( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch poses into ``data`` as ``(N, 7)`` in ``(x, y, z, qx, qy, qz, qw)``.""" ...
[docs] @abstractmethod def apply_pose(self, pose: torch.Tensor, body_ids: torch.Tensor) -> None: """Apply poses from ``(N, 7)`` tensor in ``(x, y, z, qx, qy, qz, qw)``.""" ...
# -- Center of Mass (local) ---------------------------------------------
[docs] @abstractmethod def fetch_com_local_pose( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch COM-local poses as ``(x, y, z, qx, qy, qz, qw)``.""" ...
[docs] @abstractmethod def apply_com_local_pose(self, data: torch.Tensor, body_ids: torch.Tensor) -> None: """Apply COM-local poses from ``(x, y, z, qx, qy, qz, qw)``.""" ...
# -- Velocity -----------------------------------------------------------
[docs] @abstractmethod def fetch_linear_velocity( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch linear velocities into ``data`` as ``(N, 3)``.""" ...
[docs] @abstractmethod def fetch_angular_velocity( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch angular velocities into ``data`` as ``(N, 3)``.""" ...
[docs] @abstractmethod def apply_linear_velocity(self, data: torch.Tensor, body_ids: torch.Tensor) -> None: """Set linear velocities from ``(N, 3)`` tensor.""" ...
[docs] @abstractmethod def apply_angular_velocity( self, data: torch.Tensor, body_ids: torch.Tensor ) -> None: """Set angular velocities from ``(N, 3)`` tensor.""" ...
# -- Acceleration -------------------------------------------------------
[docs] @abstractmethod def fetch_linear_acceleration( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch linear accelerations into ``data`` as ``(N, 3)``.""" ...
[docs] @abstractmethod def fetch_angular_acceleration( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch angular accelerations into ``data`` as ``(N, 3)``.""" ...
# -- Force & Torque -----------------------------------------------------
[docs] @abstractmethod def apply_force(self, data: torch.Tensor, body_ids: torch.Tensor) -> None: """Apply external forces ``(N, 3)``. One-shot — consumed on next step.""" ...
[docs] @abstractmethod def apply_torque(self, data: torch.Tensor, body_ids: torch.Tensor) -> None: """Apply external torques ``(N, 3)``. One-shot — consumed on next step.""" ...
# -- Physical Properties -------------------------------------------------
[docs] @abstractmethod def fetch_mass( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch masses into ``data`` as ``(N, 1)``.""" ...
[docs] @abstractmethod def apply_mass(self, data: torch.Tensor, body_ids: torch.Tensor) -> None: """Apply masses from ``(N, 1)`` tensor.""" ...
[docs] @abstractmethod def fetch_inertia_diagonal( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch inertia diagonals into ``data`` as ``(N, 3)``.""" ...
[docs] @abstractmethod def apply_inertia_diagonal( self, data: torch.Tensor, body_ids: torch.Tensor ) -> None: """Apply inertia diagonals from ``(N, 3)`` tensor.""" ...
[docs] @abstractmethod def fetch_friction( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch friction coefficients into ``data`` as ``(N, 1)``.""" ...
[docs] @abstractmethod def apply_friction(self, data: torch.Tensor, body_ids: torch.Tensor) -> None: """Apply friction coefficients from ``(N, 1)`` tensor.""" ...
[docs] @abstractmethod def fetch_restitution( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch restitution coefficients into ``data`` as ``(N, 1)``.""" ...
[docs] @abstractmethod def apply_restitution(self, data: torch.Tensor, body_ids: torch.Tensor) -> None: """Apply restitution coefficients from ``(N, 1)`` tensor.""" ...
[docs] @abstractmethod def fetch_contact_offset( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch contact offsets into ``data`` as ``(N, 1)``.""" ...
[docs] @abstractmethod def apply_contact_offset(self, data: torch.Tensor, body_ids: torch.Tensor) -> None: """Apply contact offsets from ``(N, 1)`` tensor.""" ...
[docs] def fetch_damping( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch linear/angular damping into ``data`` as ``(N, 2)``.""" raise NotImplementedError("This backend view does not expose damping.")
[docs] def apply_damping(self, data: torch.Tensor, body_ids: torch.Tensor) -> None: """Apply linear/angular damping from an ``(N, 2)`` tensor.""" raise NotImplementedError("This backend view does not expose damping.")
[docs] def fetch_collision_filter( self, data: torch.Tensor, body_ids: torch.Tensor | None = None ) -> None: """Fetch collision-filter rows into ``data`` as ``(N, 4)``.""" raise NotImplementedError( "This backend view does not expose collision filters." )
[docs] def apply_collision_filter( self, data: torch.Tensor, body_ids: torch.Tensor ) -> None: """Apply collision-filter rows from an ``(N, 4)`` tensor.""" raise NotImplementedError( "This backend view does not expose collision filters." )
[docs] class ArticulationViewBase(ABC): """Abstract interface for physics-backend articulation data access. Public root/link poses use EmbodiChain convention: ``(x, y, z, qx, qy, qz, qw)``. """ dof: int """Scalar degree-of-freedom width exposed by the view.""" num_links: int """Link width exposed by the view.""" joint_names: list[str] """Active joint names in public flattened-DOF order.""" link_names: list[str] """Link names in public link-buffer order.""" @property @abstractmethod def is_ready(self) -> bool: """Whether backend runtime data can be accessed through batch APIs.""" ... @property def is_newton_backend(self) -> bool: """Whether this view targets the DexSim Newton backend.""" return False @property @abstractmethod def articulation_ids_tensor(self) -> torch.Tensor | None: """Backend articulation ids as an int32 tensor, if the backend uses ids.""" ...
[docs] @abstractmethod def select_articulation_ids( self, env_ids: Sequence[int] | torch.Tensor ) -> torch.Tensor: """Return backend articulation ids for the given environment ids.""" ...
[docs] def fetch_joint_properties( self, *, position_limits: torch.Tensor | None = None, velocity_limit: torch.Tensor | None = None, effort_limit: torch.Tensor | None = None, stiffness: torch.Tensor | None = None, damping: torch.Tensor | None = None, friction: torch.Tensor | None = None, armature: torch.Tensor | None = None, ) -> None: """Read current joint parameters in public DOF order. Args: position_limits: Lower/upper output, shape ``(N, dof, 2)``. velocity_limit: Velocity-limit output, shape ``(N, dof)``. effort_limit: Effort-limit output, shape ``(N, dof)``. stiffness: Drive-stiffness output, shape ``(N, dof)``. damping: Drive-damping output, shape ``(N, dof)``. friction: Joint-friction output, shape ``(N, dof)``. armature: Joint-armature output, shape ``(N, dof)``. Outputs are caller-owned. Omitted fields are not read; at least one output is required. Values follow the backend's current model, with retained descriptor changes becoming visible after Scene rebuild. Raises: NotImplementedError: If this view has no batch property reader. """ raise NotImplementedError("This view does not expose batch joint properties.")
[docs] def fetch_state( self, qpos: torch.Tensor, qvel: torch.Tensor, root_pose: torch.Tensor, root_lin_vel: torch.Tensor, root_ang_vel: torch.Tensor, ) -> None: """Read joint and root state into caller-owned buffers. Args: qpos: Joint positions, shape ``(N, dof)``. qvel: Joint velocities, shape ``(N, dof)``. root_pose: Root poses, shape ``(N, 7)``, in ``xyz + xyzw`` order. root_lin_vel: Root world-frame linear velocities, shape ``(N, 3)``. root_ang_vel: Root world-frame angular velocities, shape ``(N, 3)``. """ self.fetch_qpos(qpos) self.fetch_qvel(qvel) self.fetch_root_pose(root_pose) self.fetch_root_linear_velocity(root_lin_vel) self.fetch_root_angular_velocity(root_ang_vel)
[docs] @abstractmethod def fetch_root_pose(self, data: torch.Tensor) -> torch.Tensor: """Fetch root poses into ``data`` and return a view/result tensor.""" ...
[docs] @abstractmethod def fetch_root_linear_velocity(self, data: torch.Tensor) -> torch.Tensor: """Fetch root linear velocities into ``data`` and return a tensor.""" ...
[docs] @abstractmethod def fetch_root_angular_velocity(self, data: torch.Tensor) -> torch.Tensor: """Fetch root angular velocities into ``data`` and return a tensor.""" ...
[docs] @abstractmethod def fetch_qpos(self, data: torch.Tensor) -> torch.Tensor: """Fetch current joint positions into ``data``.""" ...
[docs] @abstractmethod def fetch_target_qpos(self, data: torch.Tensor) -> torch.Tensor: """Fetch target joint positions into ``data``.""" ...
[docs] @abstractmethod def fetch_qvel(self, data: torch.Tensor) -> torch.Tensor: """Fetch current joint velocities into ``data``.""" ...
[docs] @abstractmethod def fetch_target_qvel(self, data: torch.Tensor) -> torch.Tensor: """Fetch target joint velocities into ``data``.""" ...
[docs] @abstractmethod def fetch_qacc(self, data: torch.Tensor) -> torch.Tensor: """Fetch current joint accelerations into ``data``.""" ...
[docs] @abstractmethod def fetch_qf(self, data: torch.Tensor) -> torch.Tensor: """Fetch current joint forces into ``data``.""" ...
[docs] @abstractmethod def apply_root_pose( self, pose: torch.Tensor, env_ids: Sequence[int] | torch.Tensor ) -> None: """Apply root poses from EmbodiChain ``xyz + xyzw`` tensors.""" ...
[docs] @abstractmethod def apply_root_velocity( self, velocity: torch.Tensor, env_ids: Sequence[int] | torch.Tensor ) -> None: """Apply world-frame root linear and angular velocities. Args: velocity: Selected root velocities with shape ``(N, 6)``. env_ids: Environment rows in the same order as ``velocity``. """ ...
[docs] @abstractmethod def apply_state( self, env_ids: Sequence[int] | torch.Tensor | None = None, joint_ids: Sequence[int] | torch.Tensor | None = None, *, root_pose: torch.Tensor | None = None, qpos: torch.Tensor | None = None, target_qpos: torch.Tensor | None = None, qvel: torch.Tensor | None = None, target_qvel: torch.Tensor | None = None, qf: torch.Tensor | None = None, root_velocity: torch.Tensor | None = None, clear_dynamics: bool = False, ) -> None: """Submit selected final physical state without intermediate propagation. Args: env_ids: Environment rows; None selects all rows. joint_ids: Joint columns; None selects all joints. root_pose: Environment-local xyz + xyzw poses, shape ``(N, 7)``. qpos: Joint positions, shape ``(N, J)``. target_qpos: Joint position targets, shape ``(N, J)``. qvel: Joint velocities, shape ``(N, J)``. target_qvel: Joint velocity targets, shape ``(N, J)``. qf: Joint efforts, shape ``(N, J)``. root_velocity: World linear + angular velocities, shape ``(N, 6)``. clear_dynamics: Clear selected rows' dynamic state and hold their final joint positions. Explicit fields override these defaults. """ ...
[docs] @abstractmethod def apply_qpos( self, qpos: torch.Tensor, env_ids: Sequence[int] | torch.Tensor | None, joint_ids: Sequence[int] | torch.Tensor | None, *, target: bool, ) -> None: """Apply joint positions for selected envs and joints.""" ...
[docs] @abstractmethod def apply_qvel( self, qvel: torch.Tensor, env_ids: Sequence[int] | torch.Tensor | None, joint_ids: Sequence[int] | torch.Tensor | None, *, target: bool, ) -> None: """Apply joint velocities for selected envs and joints.""" ...
[docs] @abstractmethod def apply_qf( self, qf: torch.Tensor, env_ids: Sequence[int] | torch.Tensor | None, joint_ids: Sequence[int] | torch.Tensor | None, ) -> None: """Apply joint forces for selected envs and joints.""" ...
[docs] @abstractmethod def clear_dynamics(self, env_ids: Sequence[int] | torch.Tensor) -> None: """Clear backend-owned articulation dynamics for selected environments.""" ...
[docs] @abstractmethod def compute_kinematics(self, env_ids: Sequence[int] | torch.Tensor) -> None: """Refresh articulation kinematics if required by the backend.""" ...