Source code for embodichain.lab.sim.cfg.physics

# ----------------------------------------------------------------------------
# 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.
# ----------------------------------------------------------------------------

"""Physics backend configuration and conversion helpers."""

from __future__ import annotations

import inspect
import math

from collections.abc import Mapping
from dataclasses import field, fields
from typing import Any, Literal, Sequence, TYPE_CHECKING

import numpy as np
import torch

from embodichain.utils import configclass, logger

__all__ = [
    "GPUMemoryCfg",
    "PhysicsBackendCfg",
    "DefaultPhysicsCfg",
    "NewtonCollisionPipelineCfg",
    "NewtonPhysicsCfg",
    "physics_cfg_for_backend",
    "physics_backend_from_cfg",
    "validate_physics_cfg",
]


[docs] @configclass class GPUMemoryCfg: """GPU buffer capacities for the Default backend's GPU dynamics pipeline. Default-backend GPU buffers cannot all grow dynamically. Values that are too small may therefore produce overflow warnings, dropped contacts, or an invalid simulation. These settings are applied only when the Default backend runs on CUDA; they have no effect on Default CPU or Newton. """ temp_buffer_capacity: int = 2**24 """Temporary pinned-host buffer capacity in bytes. Increase this when the Default backend reports a pinned-host linear allocator overflow. """ max_rigid_contact_count: int = 2**19 """Maximum number of rigid-contact records in the GPU contact stream. Increase this when the Default backend reports ``Contact buffer overflow detected``. """ max_rigid_patch_count: int = ( 2**18 ) # 81920 is DexSim default but most tasks work with 2**18 """Maximum number of rigid-contact patches in the GPU patch stream. A patch groups nearby contact points that share a contact normal. Increase this when the Default backend reports ``Patch buffer overflow detected``. """ heap_capacity: int = 2**26 """Initial capacity in bytes of the GPU and pinned-host memory heaps.""" found_lost_pairs_capacity: int = ( 2**25 ) # 262144 is DexSim default but most tasks work with 2**25 """Capacity of broad-phase found/lost pair records.""" found_lost_aggregate_pairs_capacity: int = 2**10 """Capacity of found/lost pair records generated by aggregates.""" total_aggregate_pairs_capacity: int = 2**10 """Capacity of all aggregate-pair records in the GPU pipeline."""
def _gravity_vector( gravity: Sequence[float] | np.ndarray, ) -> list[float]: """Validate and normalize a backend-neutral gravity vector.""" values = np.asarray(gravity, dtype=np.float64).reshape(-1) if values.size != 3 or not np.all(np.isfinite(values)): raise ValueError("Gravity must contain three finite values.") return values.tolist()
[docs] @configclass class PhysicsBackendCfg: """Backend-neutral simulation timing, device, and gravity configuration. Concrete backend configs inherit this class. The config type selects the backend; no independent backend string can disagree with it. """ physics_dt: float = 1.0 / 100.0 """Duration of one physics step in seconds. Environment control steps may contain multiple physics steps. For Newton, this interval is further divided by :attr:`NewtonPhysicsCfg.num_substeps`. """ device: str | torch.device = "cpu" """Compute device used to build and step the selected physics backend. Concrete backend configurations may redeclare this field when their native runtime has a different default. In particular, :class:`NewtonPhysicsCfg` intentionally shadows this CPU default with ``"cuda:0"``. Callers can still provide an explicit device override. """ gravity: Sequence[float] | np.ndarray = field( default_factory=lambda: np.array([0.0, 0.0, -9.81]) ) """World-frame gravity vector in meters per second squared."""
[docs] @configclass class DefaultPhysicsCfg(PhysicsBackendCfg): """Configuration selector for the Default physics backend.""" cache_material: bool = False """Reuse identical physics materials across objects and environment clones. Enable this for large batches to avoid exhausting the native material table. """ bounce_threshold: float = 2.0 """Relative normal-speed threshold below which contacts do not bounce [m/s].""" enable_ccd: bool = False """Whether to enable scene-level continuous collision detection (CCD). A rigid body must also set :attr:`DefaultRigidBodyPropertiesCfg.enable_ccd` for CCD to be used on that body. """ length_tolerance: float = 0.05 """Representative scene length used by the Default backend's tolerance scale [m]. Set this near the characteristic size of simulated objects. It is a scene scale, not an accuracy knob, and must be configured before world creation. """ speed_tolerance: float = 0.25 """Representative scene speed used by the Default backend's tolerance scale [m/s]. The backend derives several internal thresholds from this value and :attr:`length_tolerance`. """ gpu_memory: GPUMemoryCfg = field(default_factory=GPUMemoryCfg) """Fixed-capacity GPU buffers used by Default-backend CUDA simulation.""" def __post_init__(self) -> None: """Decode mapping input for the nested GPU-memory configuration.""" if isinstance(self.gpu_memory, Mapping): self.gpu_memory = GPUMemoryCfg(**self.gpu_memory) if not isinstance(self.gpu_memory, GPUMemoryCfg): raise TypeError("gpu_memory must be a GPUMemoryCfg or mapping.")
[docs] def to_dexsim_args(self) -> dict[str, Any]: """Convert to DexSim physics arguments. Solver implementation details that are not exposed by :class:`DefaultPhysicsCfg` retain their established defaults here. """ args = { "cache_material": self.cache_material, "gravity": _gravity_vector(self.gravity), "bounce_threshold": self.bounce_threshold, "enable_ccd": self.enable_ccd, "enable_enhanced_determinism": False, "enable_friction_every_iteration": True, } return args
[docs] @configclass class NewtonCollisionPipelineCfg: """Newton collision-pipeline settings owned at scene scope. Construction values map to DexSim's ``NewtonCollisionPipelineCfg``. ``update_interval`` controls the DexSim-side scheduling of that external pipeline. Per-shape contact and SDF values belong to :class:`NewtonCollisionPropertiesCfg` instead. The pipeline performs broad-phase pair selection, narrow-phase contact generation, and optional contact reduction for the complete scene. See the `Newton collision guide <https://newton-physics.github.io/newton/latest/concepts/collisions.html>`_ for the native pipeline semantics. """ reduce_contacts: bool = True """Whether to reduce dense mesh contacts to a representative subset. Reduction lowers contact count and usually improves performance and solver stability for mesh-heavy scenes. """ rigid_contact_max: int | None = None """Maximum number of allocated rigid contacts. ``None`` uses the model-provided capacity when available and otherwise lets Newton estimate it from the scene's shapes and candidate pairs. """ max_triangle_pairs: int = 4_000_000 """Maximum triangle-pair candidates allocated by the narrow phase. Increase this only when complex meshes or heightfields report triangle-pair overflow. EmbodiChain intentionally uses a larger default than upstream Newton for mesh-heavy robotics scenes. """ soft_contact_max: int | None = None """Maximum number of allocated particle/soft contacts. ``None`` lets Newton derive the capacity from shape and particle counts. """ soft_contact_margin: float = 0.01 """Particle/soft contact detection distance [m].""" broad_phase: Literal["nxn", "sap", "explicit"] | Any | None = None """Built-in broad-phase mode or a prebuilt Newton broad-phase object. ``"explicit"`` tests precomputed pairs, ``"nxn"`` performs an all-pairs test, and ``"sap"`` uses sweep-and-prune. ``None`` keeps Newton's default. A prebuilt object is an expert path and must be compatible with :attr:`narrow_phase`. """ shape_pairs_filtered: Any | None = None """Optional precomputed pairs for ``"explicit"`` broad phase. When provided, this must be a Warp array of shape-index pairs with ``dtype=wp.vec2i``. ``None`` uses the model's contact-pair list. """ narrow_phase: Any | None = None """Optional prebuilt Newton narrow-phase object for expert pipelines.""" sdf_hydroelastic_config: Any | None = None """Optional Newton ``HydroelasticSDF.Config``-compatible object. ``None`` disables the hydroelastic pipeline. Individual participating procedural meshes must also opt in through :attr:`~embodichain.lab.sim.shapes.MeshCollisionCfg.is_hydroelastic`. """ update_interval: int | None = 1 """External-pipeline updates within one EmbodiChain physics step. The default ``1`` updates before every solver substep. An integer ``k >= 1`` updates at substeps ``0, k, 2k, ...``; ``None`` updates once at the first solver substep of each physics step. """ def __post_init__(self) -> None: """Validate external collision-pipeline scheduling.""" if self.update_interval is not None and ( type(self.update_interval) is not int or self.update_interval < 1 ): raise ValueError( "NewtonCollisionPipelineCfg.update_interval must be a positive " "integer or None." )
[docs] @configclass class NewtonPhysicsCfg(PhysicsBackendCfg): """Configuration selector for the Newton physics backend. DexSim wraps and extends Newton for EmbodiChain. The selected solver and collision pipeline are scene-wide. Shape, contact, material, and joint values are configured separately on object and articulation configs and compiled into DexSim Spawn descriptors. """ device: str | torch.device = "cuda:0" """Warp device used to build and step Newton, for example ``"cuda:0"``. This redeclaration intentionally takes precedence over :class:`PhysicsBackendCfg.device`, so ``NewtonPhysicsCfg()`` always starts on CUDA unless the caller explicitly supplies another device. """ num_substeps: int = 10 """Number of Newton solver substeps per EmbodiChain physics step. The effective solver interval is ``physics_dt / num_substeps``. """ requires_grad: bool = False """Whether to finalize the Newton model with differentiable state enabled. EmbodiChain currently requires the Semi-implicit solver for this mode and disables CUDA graph capture when gradients are enabled. """ use_cuda_graph: bool = True """Whether to capture Newton stepping in a CUDA graph when supported. This is ignored for gradient mode and is unavailable on a CPU device. """ sync_to_renderer: bool | None = None """Newton-to-render publication policy. ``None`` and ``False`` let :class:`SimulationManager` publish on demand for windows, cameras, recording and browser visualization. ``True`` also requests publication on steps without a visual consumer. The manager batches consumers within each render frame. Direct DexSim ``World.update`` callers retain the underlying automatic policy: window-aware for ``None``, always for ``True``, and disabled for ``False``. """ debug_mode: bool = False """Whether to enable additional Newton runtime diagnostics.""" suppress_warp_kernel_logs: bool = True """Whether to hide Warp startup and kernel compile/load messages. Genuine Newton/Warp warnings and errors are not suppressed. """ solver_cfg: Mapping[str, Any] | NewtonSolverCfg | None = None """Optional Newton solver configuration. A mapping is converted to the matching DexSim Newton solver config. Include ``solver_type`` or ``class_type`` to select the solver, then add any parameters accepted by that DexSim solver config. If omitted, EmbodiChain preserves DexSim's scene-aware ``AutoSolverCfg`` default. A DexSim build exporting ``AutoSolverCfg`` is required; no concrete-solver fallback is used. """ collision_cfg: NewtonCollisionPipelineCfg | Mapping[str, Any] | None = field( default_factory=NewtonCollisionPipelineCfg ) """Optional external collision-pipeline configuration. The default preserves external contact generation for ordinary rigid-body scenes. Set to ``None`` to avoid constructing or executing the external pipeline, for example when particle solvers use rigid shapes solely as one-way SDF boundaries. """ def __post_init__(self) -> None: """Normalize dictionary collision settings at the config boundary.""" if isinstance(self.collision_cfg, Mapping): self.collision_cfg = NewtonCollisionPipelineCfg(**self.collision_cfg)
[docs] def to_dexsim_cfg( self, gpu_id: int, ) -> NewtonCfg: """Convert this config to ``dexsim.engine.newton_physics.NewtonCfg``.""" from dexsim.engine.newton_physics import ( AutoSolverCfg, DexUniSolverCfg, FeatherstoneSolverCfg, MJWarpSolverCfg, NewtonCfg, NewtonCollisionPipelineCfg as DexsimNewtonCollisionPipelineCfg, SemiImplicitSolverCfg, VBDSolverCfg, XPBDSolverCfg, ) torch_device = ( torch.device(self.device) if isinstance(self.device, str) else self.device ) device = ( f"cuda:{gpu_id}" if torch_device.type == "cuda" and torch_device.index is None else str(torch_device) ) solver_cfg_map: dict[str, type] = { "auto": AutoSolverCfg, "dexuni": DexUniSolverCfg, "mujoco_warp": MJWarpSolverCfg, "xpbd": XPBDSolverCfg, "semi_implicit": SemiImplicitSolverCfg, "featherstone": FeatherstoneSolverCfg, "vbd": VBDSolverCfg, } solver_cfg = _newton_solver_cfg_to_dexsim( solver_cfg=self.solver_cfg, solver_cfg_map=solver_cfg_map, ) if self.requires_grad and ( solver_cfg is None or solver_cfg.solver_type != "semi_implicit" ): logger.log_error( "Newton gradient mode requires an explicit " "solver_type='semi_implicit'; AutoSolver does not select a " "differentiable solver." ) collision_pipeline_cfg = None if self.collision_cfg is not None: collision_values = { item.name: getattr(self.collision_cfg, item.name) for item in fields(self.collision_cfg) } collision_values["requires_grad"] = self.requires_grad # Newton 1.6 renamed this field; published DexSim builds also use # Newton 1.4, whose descriptor retains soft_contact_margin. if "soft_contact_gap" in { item.name for item in fields(DexsimNewtonCollisionPipelineCfg) }: collision_values["soft_contact_gap"] = collision_values.pop( "soft_contact_margin" ) collision_pipeline_cfg = DexsimNewtonCollisionPipelineCfg( **collision_values ) newton_cfg_args: dict[str, Any] = { "dt": self.physics_dt, "num_substeps": self.num_substeps, "device": device, "gravity": _gravity_vector(self.gravity), "debug_mode": self.debug_mode, "requires_grad": self.requires_grad, "suppress_warp_kernel_logs": self.suppress_warp_kernel_logs, "collision_pipeline_cfg": collision_pipeline_cfg, "sync_to_dexsim": self.sync_to_renderer, } if solver_cfg is not None: newton_cfg_args["solver_cfg"] = solver_cfg cfg = NewtonCfg( **newton_cfg_args, ) cfg.use_cuda_graph = self.use_cuda_graph and not self.requires_grad return cfg
def _normalize_newton_solver_type(solver_type: str) -> str: """Normalize public EmbodiChain and DexSim Newton solver aliases.""" key = solver_type.replace("-", "_").lower() aliases = { "auto": "auto", "autosolver": "auto", "autosolvercfg": "auto", "auto_solver": "auto", "auto_solver_cfg": "auto", "mjwarp": "mujoco_warp", "mjwarpsolver": "mujoco_warp", "mjwarpsolvercfg": "mujoco_warp", "mjwarp_solver": "mujoco_warp", "mjwarp_solver_cfg": "mujoco_warp", "mujoco_warp": "mujoco_warp", "mujocowarp": "mujoco_warp", "mujocowarpsolver": "mujoco_warp", "mujocowarpsolvercfg": "mujoco_warp", "dexuni": "dexuni", "dexunisolver": "dexuni", "dexunisolvercfg": "dexuni", "dexuni_solver": "dexuni", "dexuni_solver_cfg": "dexuni", "xpbdsolver": "xpbd", "xpbdsolvercfg": "xpbd", "xpbd": "xpbd", "semiimplicit": "semi_implicit", "semi_implicit": "semi_implicit", "semiimplicitsolver": "semi_implicit", "semiimplicitsolvercfg": "semi_implicit", "featherstone": "featherstone", "featherstonesolver": "featherstone", "featherstonesolvercfg": "featherstone", "vbd": "vbd", "vbdsolver": "vbd", "vbdsolvercfg": "vbd", } if key not in aliases: logger.log_error( f"Unsupported Newton solver type '{solver_type}'. " "Expected one of 'auto', 'dexuni', 'mjwarp', 'xpbd', 'semi_implicit', " "'featherstone', or 'vbd'." ) return aliases[key] def _newton_solver_cfg_to_dexsim( solver_cfg: Mapping[str, Any] | object | None, solver_cfg_map: Mapping[str, type], ) -> object | None: """Convert EmbodiChain Newton solver config input to a DexSim config.""" if solver_cfg is None: return None if not isinstance(solver_cfg, Mapping): if not hasattr(solver_cfg, "solver_type"): logger.log_error( "Newton solver_cfg must be a mapping or a DexSim Newton solver " "config object with a 'solver_type' attribute." ) return solver_cfg solver_cfg_data = dict(solver_cfg) configured_solver_type = ( solver_cfg_data.pop("solver_type", None) or solver_cfg_data.pop("class_type", None) or "auto" ) normalized_solver_type = _normalize_newton_solver_type(str(configured_solver_type)) solver_cfg_type = solver_cfg_map[normalized_solver_type] if normalized_solver_type == "dexuni": try: uses_grouped_options = ( "vbd_options" in inspect.signature(solver_cfg_type).parameters ) except (TypeError, ValueError): uses_grouped_options = True if not uses_grouped_options: # DexSim 0.5.0 still owns the legacy flat DexUni dataclass. return solver_cfg_type(**solver_cfg_data) # DexSim 0.5.0 exposed these options directly on DexUniSolverCfg. The # 0.5.1rc1 facade groups them by owned solver path, so keep existing # EmbodiChain mappings source-compatible while forwarding the values to # the native option contract. legacy_values: dict[str, Any] = {} vbd_options = dict(solver_cfg_data.get("vbd_options", {})) if "iterations" in solver_cfg_data: legacy_values["iterations"] = solver_cfg_data.pop("iterations") vbd_options.setdefault("iterations", legacy_values["iterations"]) if "step_rigid_bodies" in solver_cfg_data: legacy_values["step_rigid_bodies"] = solver_cfg_data.pop( "step_rigid_bodies" ) # The new option describes the inverse operation: when true, VBD # integrates rigid state supplied by an external rigid solver. vbd_options.setdefault( "integrate_with_external_rigid_solver", not legacy_values["step_rigid_bodies"], ) if vbd_options: solver_cfg_data["vbd_options"] = vbd_options native_cfg = solver_cfg_type(**solver_cfg_data) # Preserve read-only compatibility for callers that still inspect the # legacy attributes after conversion. Native DexSim dataclasses are # not slotted, so these aliases do not affect its option validation. for name, value in legacy_values.items(): setattr(native_cfg, name, value) return native_cfg return solver_cfg_type(**solver_cfg_data)
[docs] def physics_cfg_for_backend( backend: Literal["default", "newton"], ) -> DefaultPhysicsCfg | NewtonPhysicsCfg: """Return a default physics configuration instance for the given backend.""" if backend == "newton": return NewtonPhysicsCfg() if backend == "default": return DefaultPhysicsCfg() raise ValueError( f"Unsupported physics backend {backend!r}; expected 'default' or 'newton'." )
[docs] def physics_backend_from_cfg( physics_cfg: PhysicsBackendCfg, ) -> Literal["default", "newton"]: """Infer the physics backend name from a physics configuration instance.""" if isinstance(physics_cfg, NewtonPhysicsCfg): return "newton" if isinstance(physics_cfg, DefaultPhysicsCfg): return "default" logger.log_error( f"Unsupported physics_cfg type '{type(physics_cfg).__name__}'. " "Expected DefaultPhysicsCfg or NewtonPhysicsCfg." )
[docs] def validate_physics_cfg(physics_cfg: PhysicsBackendCfg) -> None: """Validate that ``physics_cfg`` is a supported backend configuration.""" physics_backend_from_cfg(physics_cfg)