# ----------------------------------------------------------------------------
# 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
import os
import gc
import sys
import queue
import time
import threading
from contextlib import contextmanager
import dexsim
import torch
import numpy as np
import warp as wp
from pathlib import Path
from uuid import uuid4
from copy import deepcopy
from datetime import datetime
from functools import cached_property, partial
from numbers import Integral
from typing import (
TYPE_CHECKING,
Any,
Callable,
Dict,
Iterator,
List,
Literal,
Mapping,
Sequence,
Union,
)
from dataclasses import dataclass, asdict, field, MISSING
# Global cache directories
SIM_CACHE_DIR = Path.home() / ".cache" / "embodichain_cache"
MATERIAL_CACHE_DIR = SIM_CACHE_DIR / "mat_cache"
CONVEX_DECOMP_DIR = SIM_CACHE_DIR / "convex_decomposition"
REACHABLE_XPOS_DIR = SIM_CACHE_DIR / "robot_reachable_xpos"
def _is_usd_path(path: object | None) -> bool:
"""Return whether a source path is a USD stage."""
return path is not None and str(path).lower().endswith(
(".usd", ".usda", ".usdc", ".usdz")
)
from dexsim.types import (
ActorType,
Backend,
ThreadMode,
)
from dexsim.core import TASK_RETURN
from dexsim.engine import Material, ObjectManipulator
from dexsim.models import MeshObject
from dexsim.render import LightType, Windows
from embodichain.lab.sim.objects import (
RigidObject,
VisualObject,
RigidObjectGroup,
DeformableObject,
SurfaceDeformableObject,
VolumeDeformableObject,
Articulation,
Robot,
Light,
RigidConstraint,
)
from embodichain.lab.sim.objects.gizmo import Gizmo, GizmoCfg
from embodichain.lab.sim.sensors import (
SensorCfg,
BaseSensor,
Camera,
CameraCfg,
StereoCamera,
ContactSensor,
)
from embodichain.lab.sim.sensors.attachment import resolve_parent_nodes
from embodichain.lab.sim.cfg import (
RenderCfg,
PhysicsBackendCfg,
GPUMemoryCfg,
DefaultPhysicsCfg,
NewtonPhysicsCfg,
validate_physics_cfg,
MarkerCfg,
WindowRecordCfg,
WindowCameraPoseCfg,
LightCfg,
RigidObjectCfg,
VisualObjectCfg,
UsdSourceCfg,
DeformableObjectCfg,
SurfaceDeformableObjectCfg,
VolumeDeformableObjectCfg,
RigidObjectGroupCfg,
ArticulationCfg,
ArticulationRootPropertiesCfg,
RobotCfg,
RobotPresetCfg,
RigidConstraintCfg,
)
from embodichain.lab.sim.physics import make_physics_backend
from embodichain.lab.sim.cfg.source import _asset_file_path
from embodichain.lab.sim.spawn.descriptors import (
articulation_desc_from_cfg,
configure_articulation_desc,
rigid_desc_from_cfg,
visual_desc_from_cfg,
surface_deformable_desc_from_cfg,
volume_deformable_desc_from_cfg,
)
from embodichain.lab.sim.spawn.usd import (
articulation_desc_from_usd,
rigid_desc_from_usd,
visual_desc_from_usd,
)
from embodichain.lab.sim.spawn.scene import SpawnScene
from embodichain.lab.sim import VisualMaterial, VisualMaterialCfg
from embodichain.lab.sim.material import _material_desc_from_cfg
from embodichain.lab.sim.profiler import Profiler, ProfilerCfg
from embodichain.lab.visualization.cfg import VisualizationCfg
from embodichain.utils import configclass, logger
from embodichain.utils.math import (
look_at_to_pose,
matrix_from_quat,
pose_inv,
quat_wxyz_to_xyzw,
)
if TYPE_CHECKING:
from dexsim.engine import PhysicsScene
from dexsim.scene import Scene
from dexsim.interaction import EntityGizmoConfig, EntityGizmoManipulator
from embodichain.lab.visualization import (
RuntimeHealth,
RuntimeStats,
SceneManifest,
SceneOverlays,
VisualizationRuntime,
)
__all__ = [
"SimulationManager",
"SimulationManagerCfg",
"get_physics_scene",
"SIM_CACHE_DIR",
"MATERIAL_CACHE_DIR",
"CONVEX_DECOMP_DIR",
"REACHABLE_XPOS_DIR",
]
@contextmanager
def _temporary_warp_kernel_log_suppression(
physics_cfg: PhysicsBackendCfg,
) -> Iterator[None]:
"""Temporarily suppress informational Warp logs for Newton operations."""
if not (
isinstance(physics_cfg, NewtonPhysicsCfg)
and physics_cfg.suppress_warp_kernel_logs
):
yield
return
previous_log_level = wp.config.log_level
try:
# Warp emits its startup banner and module-load timers at INFO level.
# Keep warnings and errors visible.
wp.config.log_level = wp.LOG_WARNING
yield
finally:
wp.config.log_level = previous_log_level
def _initialize_warp_runtime(physics_cfg: PhysicsBackendCfg) -> None:
"""Initialize Warp while honoring Newton startup-log suppression."""
with _temporary_warp_kernel_log_suppression(physics_cfg):
wp.init()
# Deformable objects are Newton particle sets. The Default implementation is
# deliberately absent so stale soft/cloth configurations fail at declaration.
_DEFORMABLE_BACKEND_IMPLEMENTATIONS = {
"default": {},
"newton": {
"volume": (
VolumeDeformableObjectCfg,
VolumeDeformableObject,
volume_deformable_desc_from_cfg,
"soft_object",
),
"surface": (
SurfaceDeformableObjectCfg,
SurfaceDeformableObject,
surface_deformable_desc_from_cfg,
"cloth_object",
),
},
}
[docs]
@configclass
class SimulationManagerCfg:
"""Global robot simulation configuration."""
def __init__(
self,
width: int = 1920,
height: int = 1080,
headless: bool = False,
render_cfg: RenderCfg | None = None,
gpu_id: int = 0,
thread_mode: ThreadMode = ThreadMode.RENDER_SHARE_ENGINE,
cpu_num: int = 1,
num_envs: int = 1,
arena_space: float = 5.0,
enable_entity_gizmo: bool = True,
robot_ik_gizmo: GizmoCfg | dict | None = GizmoCfg(),
physics_dt: float | None = None,
device: str | torch.device | None = None,
physics_cfg: PhysicsBackendCfg | None = None,
sim_device: str | torch.device | None = None,
physics_config: PhysicsBackendCfg | None = None,
gpu_memory_config: GPUMemoryCfg | None = None,
profiler: ProfilerCfg | None = None,
visualization: VisualizationCfg | None = None,
window_record: WindowRecordCfg | None = None,
window_camera_pose: WindowCameraPoseCfg | None = None,
startup_summary: Literal["compact", "full", "off"] = "compact",
dexsim_startup_info: bool = False,
scene_node_capacity: int | None = None,
) -> None:
self.startup_summary = startup_summary
self.dexsim_startup_info = dexsim_startup_info
self.scene_node_capacity = scene_node_capacity
self.width = width
self.height = height
self.headless = headless
self.render_cfg = RenderCfg() if render_cfg is None else render_cfg
self.gpu_id = gpu_id
self.thread_mode = thread_mode
self.cpu_num = cpu_num
self.num_envs = num_envs
self.arena_space = arena_space
self.enable_entity_gizmo = enable_entity_gizmo
self.robot_ik_gizmo = deepcopy(robot_ik_gizmo)
if physics_cfg is None:
physics_cfg = (
DefaultPhysicsCfg() if physics_config is None else physics_config
)
self.physics_cfg = physics_cfg
if gpu_memory_config is not None:
if not isinstance(self.physics_cfg, DefaultPhysicsCfg):
logger.log_error(
"gpu_memory_config is only supported by the default physics backend.",
ValueError,
)
self.physics_cfg.gpu_memory = gpu_memory_config
self.profiler = profiler
self.visualization = (
VisualizationCfg() if visualization is None else visualization
)
self.window_record = (
WindowRecordCfg() if window_record is None else window_record
)
self.window_camera_pose = (
WindowCameraPoseCfg() if window_camera_pose is None else window_camera_pose
)
if physics_dt is not None:
self.physics_cfg.physics_dt = physics_dt
# ``None`` is an omission sentinel, not a request for the generic
# PhysicsBackendCfg default. Leave the concrete config untouched so
# NewtonPhysicsCfg's CUDA default remains authoritative. A non-None
# value is an intentional runtime override and is applied uniformly.
runtime_device = device if device is not None else sim_device
if runtime_device is not None:
self.physics_cfg.device = runtime_device
self.__post_init__()
width: int = 1920
"""The width of the simulation window."""
startup_summary: Literal["compact", "full", "off"] = "compact"
"""Startup table detail. ``off`` disables both simulation and Gym summaries."""
dexsim_startup_info: bool = False
"""Show DexSim's native startup information in addition to our summary.
Warnings and errors remain visible regardless of this setting. Requires
a DexSim build exposing ``WorldConfig.log_startup_info``.
"""
scene_node_capacity: int | None = None
"""Maximum scene nodes, or None to retain the DexSim World default.
Large batches of articulated robots may require more nodes than the
default 65,536. NoRender caps larger native-renderer reservations at its
131,071-node limit while retaining the authored capacity in this config.
Physical scene replication must fit the effective capacity. This changes
allocation capacity, not physics settings.
"""
height: int = 1080
"""The height of the simulation window."""
headless: bool = False
"""Whether to run without an automatically opened native window.
This is forced to ``True`` when the Viser backend is enabled. Viser and
the native DexSim window are mutually exclusive; browser Gizmos do not
require a native window.
"""
enable_entity_gizmo: bool = True
"""Enable native object interaction when the first native window opens.
Headless and Viser simulations do not automatically create native gizmos.
Explicit runtime enable/disable calls take precedence over this default;
reopening a window preserves the current DexSim controller state.
Robot IK interaction is controlled separately by ``robot_ik_gizmo``.
"""
robot_ik_gizmo: GizmoCfg | None = field(default_factory=GizmoCfg)
"""Automatically register robot parts with configured IK chain/TCP metadata.
Native controls activate on the first I press; Viser creates its IK solver
on the first drag and requires ``visualization.allow_commands``. Pure
headless and multi-environment simulations do not register controls.
Defaults to DexSim IK. Use ``GizmoCfg(ik_solver="embodichain")`` to reuse
configured solvers such as Pink, or ``None`` to disable automatic setup.
``GizmoCfg(ik_start_enabled=True)`` explicitly activates native controls on
their first update with an open window, without waiting for I.
"""
render_cfg: RenderCfg = field(default_factory=RenderCfg)
"""The rendering configuration parameters."""
gpu_id: int = 0
"""The gpu index that the simulation engine will be used.
Note: it will affect the gpu physics device if using gpu physics.
"""
thread_mode: ThreadMode = ThreadMode.RENDER_SHARE_ENGINE
"""The threading mode for the simulation engine.
- RENDER_SHARE_ENGINE: The rendering thread shares the same thread with the simulation engine.
- RENDER_SCENE_SHARE_ENGINE: The rendering thread and scene update thread share the same thread with the simulation engine.
"""
cpu_num: int = 1
"""The number of CPU threads to use for the simulation engine."""
num_envs: int = 1
"""The number of parallel environments (arenas) to simulate."""
arena_space: float = 5.0
"""The distance between each arena when building multiple arenas."""
physics_cfg: PhysicsBackendCfg = field(default_factory=DefaultPhysicsCfg)
"""Physics backend configuration (type selects default vs Newton backend).
The concrete config owns the default device: Default uses ``cpu`` and
Newton uses ``cuda:0``. The constructor's optional ``device``/``sim_device``
arguments are applied only when explicitly provided.
"""
profiler: ProfilerCfg | None = None
"""Optional simulation profiler. ``None`` disables profiling.
Standalone calls to :meth:`SimulationManager.update` are recorded below a
``sim_update`` root. When the manager is owned by an environment, the same
profiler instance composes with the environment's step/reset hierarchy.
"""
window_record: WindowRecordCfg = field(default_factory=WindowRecordCfg)
"""Viewer window recording settings (hotkey, paths, FPS, memory budget)."""
window_camera_pose: WindowCameraPoseCfg = field(default_factory=WindowCameraPoseCfg)
"""Interactive viewer camera-pose printing settings."""
visualization: VisualizationCfg = field(default_factory=VisualizationCfg)
"""Live browser visualization settings."""
def __post_init__(self) -> None:
"""Validate physics and apply visualization-dependent defaults."""
if self.startup_summary not in ("compact", "full", "off"):
raise ValueError("startup_summary must be 'compact', 'full', or 'off'.")
validate_physics_cfg(self.physics_cfg)
if isinstance(self.robot_ik_gizmo, dict):
self.robot_ik_gizmo = GizmoCfg(**self.robot_ik_gizmo)
if self.robot_ik_gizmo is not None and not isinstance(
self.robot_ik_gizmo, GizmoCfg
):
raise TypeError("robot_ik_gizmo must be a GizmoCfg, mapping, or None.")
if self.visualization.backend == "viser":
self.headless = True
@property
def physics_dt(self) -> float:
"""The time step for the physics simulation."""
return self.physics_cfg.physics_dt
@physics_dt.setter
def physics_dt(self, value: float) -> None:
self.physics_cfg.physics_dt = value
@property
def device(self) -> str | torch.device:
"""The device for the physics simulation."""
return self.physics_cfg.device
@device.setter
def device(self, value: str | torch.device) -> None:
self.physics_cfg.device = value
@property
def sim_device(self) -> str | torch.device:
"""Legacy alias for :attr:`device`."""
return self.device
@sim_device.setter
def sim_device(self, value: str | torch.device) -> None:
self.device = value
@property
def physics_config(self) -> PhysicsBackendCfg:
"""Legacy alias for :attr:`physics_cfg`."""
return self.physics_cfg
@physics_config.setter
def physics_config(self, value: PhysicsBackendCfg) -> None:
validate_physics_cfg(value)
self.physics_cfg = value
@property
def gpu_memory_config(self) -> GPUMemoryCfg | None:
"""Legacy alias for the default backend GPU-memory configuration."""
if not isinstance(self.physics_cfg, DefaultPhysicsCfg):
return None
return self.physics_cfg.gpu_memory
@gpu_memory_config.setter
def gpu_memory_config(self, value: GPUMemoryCfg) -> None:
if not isinstance(self.physics_cfg, DefaultPhysicsCfg):
raise AttributeError(
"gpu_memory_config is unavailable for the Newton physics backend."
)
self.physics_cfg.gpu_memory = value
@dataclass
class _WindowRecordState:
"""Internal state for simulation recording."""
time_step: float
max_memory_bytes: int
output_dir: str
video_name: str
save_kwargs: dict[str, object]
record_camera: object | None = None
pose_provider: Callable[[], np.ndarray] | None = None
fixed_pose: np.ndarray | None = None
frames: list[np.ndarray] = field(default_factory=list)
current_memory_bytes: int = 0
last_capture_time: float = field(default_factory=time.time)
accumulated_sim_time: float = 0.0
capture_from_sim_update: bool = False
task_status: int = TASK_RETURN.TASK_LOOP
loop_handle: object | None = None
@dataclass(frozen=True)
class _AxisMarkerGroup:
"""Native axis handles and their backend-neutral display dimensions."""
handles: tuple[MeshObject, ...]
arena_index: int
axis_length: float
axis_radius: float
[docs]
class SimulationManager:
r"""Global Embodied AI simulation manager.
This class is used to manage the global simulation environment and simulated assets.
- assets loading, creation, modification and deletion.
- assets include rigid objects, soft objects, articulations, robots, sensors and lights.
- manager the scenes and the simulation environment.
- parallel scenes simulation on both CPU and GPU.
- create and setup the rendering related settings, eg. environment map, lighting, materials, etc.
- physics simulation management, including time steps and explicit stepping.
- interactive control via gizmo and window callbacks events.
Args:
sim_config (SimulationManagerCfg, optional): simulation configuration. Defaults to SimulationManagerCfg().
defer_startup_summary: Let an owning environment emit one combined table
after task and manager initialization. Standalone callers keep False.
requires_native_renderer: Predeclared native render demand. False permits
auto to select NoRender; None preserves standalone GPU selection.
Windows, Viser, forced synchronization and renderer overrides retain
precedence. The renderer is fixed before the World is constructed.
"""
_instances = {}
# World cleanup does not reset DexSim's process-wide renderer.
_initialized_process_renderer = False
# DexSim NoRender uses the 17-bit EntityManager node allocator.
_NO_RENDER_SCENE_NODE_CAPACITY = 131071
_cleanup_queue: queue.Queue = queue.Queue()
_DEFAULT_PLANE_GIZMO_TARGET_ID = (1 << 64) - 1
SUPPORTED_SENSOR_TYPES = {
"Camera": Camera,
"StereoCamera": StereoCamera,
"ContactSensor": ContactSensor,
}
[docs]
def __new__(
cls,
sim_config: SimulationManagerCfg = SimulationManagerCfg(),
*,
defer_startup_summary: bool = False,
requires_native_renderer: bool | None = None,
):
"""Create or return the instance based on instance_id."""
n_instance = 0
while n_instance in cls._instances:
n_instance += 1
instance = super(SimulationManager, cls).__new__(cls)
# Store sim_config in the instance for use in __init__ or elsewhere
instance.sim_config = sim_config
instance._is_constructed = False
instance.instance_id = n_instance
cls._instances[n_instance] = instance
return instance
[docs]
def __init__(
self,
sim_config: SimulationManagerCfg = SimulationManagerCfg(),
*,
defer_startup_summary: bool = False,
requires_native_renderer: bool | None = None,
) -> None:
"""Construct the simulation World with a fixed renderer selection.
Failed construction queues cleanup for its owned World without exiting
the process. Flush the cleanup queue after constructor frames unwind.
Failures before World creation remove only their own registry entry.
Args:
sim_config: Simulation, physics and rendering configuration.
defer_startup_summary: Let the owning environment emit the summary.
requires_native_renderer: Predeclared render demand. ``False`` allows
auto to select NoRender when there are no window, browser or
forced-sync consumers. ``None`` preserves standalone GPU-based
selection. Explicit and global renderer overrides take precedence.
"""
try:
self._initialize_simulation(
sim_config,
defer_startup_summary=defer_startup_summary,
requires_native_renderer=requires_native_renderer,
)
except BaseException:
if getattr(self, "_world", None) is not None:
try:
self.destroy(exit_process=False)
except Exception as error:
logger.log_warning(
f"Failed to queue constructor cleanup: {error!r}"
)
elif type(self)._instances.get(self.instance_id) is self:
type(self)._instances.pop(self.instance_id)
raise
def _initialize_simulation(
self,
sim_config: SimulationManagerCfg,
*,
defer_startup_summary: bool,
requires_native_renderer: bool | None,
) -> None:
"""Initialize resources with teardown state ready before native allocation."""
self._world: dexsim.World | None = None
self._env = None
self._spawn_scene: SpawnScene | None = None
self._window_record_state: _WindowRecordState | None = None
self._window_record_save_threads: list[threading.Thread] = []
self._visualization_runtime = None
self._arenas: list[dexsim.environment.Arena] = []
self._gizmos: dict[str, object] = {}
self._markers: dict[str, _AxisMarkerGroup] = {}
self._constraints: dict[str, RigidConstraint] = {}
self._visual_materials: dict[str, VisualMaterial] = {}
self._texture_cache: dict[str, torch.Tensor | list[torch.Tensor]] = {}
self._defer_startup_summary = defer_startup_summary
self._requires_native_renderer = requires_native_renderer
self._renderer_selection_reason: str | None = None
self._startup_summary_logged = False
self._scene_summary_logged = False
self._requested_renderer = sim_config.render_cfg.renderer
solver_cfg = getattr(sim_config.physics_cfg, "solver_cfg", None)
self._requested_solver = str(
solver_cfg.get("solver_type", solver_cfg.get("class_type", "auto"))
if isinstance(solver_cfg, Mapping)
else getattr(solver_cfg, "solver_type", "auto")
)
# Cache paths
self._sim_cache_dir = SIM_CACHE_DIR
self._material_cache_dir = MATERIAL_CACHE_DIR
self._convex_decomp_dir = CONVEX_DECOMP_DIR
self._reachable_xpos_dir = REACHABLE_XPOS_DIR
# Setup cache file path.
for path in [
self._sim_cache_dir,
self._material_cache_dir,
self._convex_decomp_dir,
self._reachable_xpos_dir,
]:
os.makedirs(path, exist_ok=True)
self.sim_config = sim_config
self.device = torch.device("cpu")
# Initialize physics backend (selected by the type of physics_cfg).
# The backend is held as an instance member; SimulationManager delegates
# all backend-specific lifecycle/scene/capability logic to it instead of
# branching on a backend name throughout the manager.
self.physics = make_physics_backend(sim_config.physics_cfg, self)
world_config = self._convert_sim_config(sim_config)
self._world_config = world_config
self.profiler = Profiler(sim_config.profiler, self.device)
# Initialize Warp before creating the world. For Newton, honor the
# configured startup/kernel-log suppression from the very first init.
_initialize_warp_runtime(sim_config.physics_cfg)
self._render_device_name: str | None = None
if sim_config.startup_summary != "off":
# Warp already enumerated these devices during initialization. Read
# its cached names instead of lazily initializing PyTorch CUDA just
# to print diagnostics for CPU-physics/explicit-renderer programs.
for render_device in wp.get_cuda_devices():
if render_device.ordinal == sim_config.gpu_id:
self._render_device_name = render_device.name
break
self._world = dexsim.World(world_config)
SimulationManager._initialized_process_renderer = True
# The caller owns physics time, including while the scene is assembled.
self._world.set_manual_update(True)
self._window: Windows | None = None
self._auto_entity_gizmo_pending = sim_config.enable_entity_gizmo
self._window_record_camera: object | None = None
wr = sim_config.window_record
self._window_record_hotkey_cfg: dict[str, object] | None = (
{
"save_path": wr.save_path,
"fps": wr.fps,
"max_memory": wr.max_memory,
"video_prefix": wr.video_prefix,
}
if wr.enable_hotkey
else None
)
self._window_record_input_control: ObjectManipulator | None = None
wcp = sim_config.window_camera_pose
self._window_camera_pose_hotkey_cfg: dict[str, object] | None = (
{"convert_to_look_at": wcp.convert_to_look_at}
if wcp.enable_hotkey
else None
)
self._window_camera_pose_input_control: ObjectManipulator | None = None
self._world.set_delta_time(sim_config.physics_cfg.physics_dt)
self._world.show_coordinate_axis(False)
# Activate the physics backend now that the dexsim World exists.
self.physics.activate(sim_config)
# activate physics
self.enable_physics(True)
self._env = self._world.get_env()
# gizmo management
self._disabled_robot_gizmos: set[str] = set()
# ``(uid, control_part)`` of the Gizmo currently owned by the Viser
# click-to-pick feature, or ``None``. Only one picker Gizmo is kept at a
# time and user-created Gizmos are never touched.
self._picker_gizmo: tuple[str, str | None] | None = None
self._visual_objects: dict[str, VisualObject] = {}
self._rigid_objects: Dict[str, RigidObject] = dict()
self._rigid_object_groups: Dict[str, RigidObjectGroup] = dict()
self._deformable_objects: Dict[str, DeformableObject] = dict()
self._articulations: Dict[str, Articulation] = dict()
self._robots: Dict[str, Robot] = dict()
self._sensors: Dict[str, BaseSensor] = dict()
self._lights: Dict[str, Light] = dict()
self._spawn_scene = SpawnScene(
self._world,
num_envs=sim_config.num_envs,
spacing=(sim_config.arena_space, sim_config.arena_space, 0.0),
renderer=sim_config.render_cfg.renderer,
)
self._arenas = list(self._spawn_scene.builder.prepare_arenas())
self._prepared_spawn_topology_revision = -1
self._ready_spawn_topology_revision = -1
self._synced_spawn_render_topology_revision = -1
self._camera_attachment_topology_revision = -1
self._render_state_published: bool | None = None
self._pending_record_dt = 0.0
self._visualization_overlays: SceneOverlays | None = None
self._visualization_topology_revision = 0
self._visualization_manifest_topology_revision = -1
self._visualization_sim_step = 0
self._visualization_sim_time = 0.0
self._visualization_error_reported = False
# The plane material and visibility are authored before declaration so
# both eager Default loading and deferred Newton loading see them.
self._spawn_default_plane_visibility = True
self._spawn_default_plane_material = None
self._default_plane = None
if self.has_native_renderer:
self._init_sim_resources()
self.set_default_background()
self._declare_spawn_default_plane()
if self.has_native_renderer:
self.set_default_global_lighting()
# SpawnScene has already prepared the configured Arenas. Start the
# optional browser runtime after default resources are declared.
self.start_visualization()
if sim_config.headless is False:
self._window = self._world.get_windows()
self.is_window_opened = self._window is not None
self._enable_default_entity_gizmo()
self._is_constructed = True
if not self._defer_startup_summary:
self._log_startup_summary()
def _log_startup_summary(self) -> None:
"""Emit the engine snapshot once, after successful construction."""
if self.sim_config.startup_summary == "off" or self._startup_summary_logged:
return
from ._startup_summary import format_summary, simulation_rows
logger.log_info(
format_summary("Simulation initialized", simulation_rows(self)),
prefix=False,
)
self._startup_summary_logged = True
def _log_scene_summary(self) -> None:
"""Emit the first usable scene snapshot, never a partial prepare."""
if (
not getattr(self, "_is_constructed", False)
or getattr(self, "_defer_startup_summary", True)
or getattr(self, "_scene_summary_logged", True)
or self.sim_config.startup_summary == "off"
or self.spawn_result is None
):
return
from ._startup_summary import format_summary, scene_is_ready, scene_rows
if not scene_is_ready(self):
return
if (
self.physics.name == "newton"
and self.physics.cuda_graph_status == "pending"
):
return
logger.log_info(format_summary("Scene ready", scene_rows(self)), prefix=False)
self._scene_summary_logged = True
[docs]
@classmethod
def get_instance(cls, instance_id: int = 0) -> SimulationManager:
"""Get the instance of SimulationManager by id.
Args:
instance_id (int): The instance id. Defaults to 0.
Returns:
SimulationManager: The instance.
Raises:
RuntimeError: If the instance has not been created yet.
"""
if instance_id not in cls._instances:
logger.log_error(
f"SimulationManager (id={instance_id}) has not been instantiated yet. "
f"Create an instance first using SimulationManager(sim_config, instance_id={instance_id})."
)
return cls._instances[instance_id]
[docs]
@classmethod
def get_instance_num(cls) -> int:
"""Get the number of instantiated SimulationManager instances.
Returns:
int: The number of instances.
"""
return len(cls._instances)
[docs]
@classmethod
def reset(cls, instance_id: int = 0) -> None:
"""Reset the instance.
This allows creating a new instance with different configuration.
"""
if instance_id in cls._instances:
logger.log_debug(f"Resetting SimulationManager instance {instance_id}.")
del cls._instances[instance_id]
[docs]
@classmethod
def is_instantiated(cls, instance_id: int = 0) -> bool:
"""Check if the instance has been created.
Returns:
bool: True if the instance exists, False otherwise.
"""
return instance_id in cls._instances
[docs]
@classmethod
def set_default_renderer(cls, renderer: str = "auto", gpu_id: int = 0) -> str:
"""Set the global default renderer used by new simulations.
This updates :data:`embodichain.lab.sim.cfg.DEFAULT_RENDERER`, which is
consulted by :func:`embodichain.lab.sim.utility.render_utils.select_default_renderer`
when ``render_cfg.renderer="auto"`` is resolved during :class:`SimulationManager`
construction.
Args:
renderer: The renderer to set. One of ``"auto"``, ``"hybrid"``,
``"fast-rt"``, or ``"rt"``. When ``"auto"``, the renderer is
resolved immediately from the detected GPU via
:func:`embodichain.lab.sim.utility.render_utils.select_default_renderer`.
gpu_id: The CUDA device index to query when ``renderer="auto"``.
Returns:
The resolved renderer name that was set as the default.
"""
from embodichain.lab.sim import cfg
from embodichain.lab.sim.utility.render_utils import select_default_renderer
valid = {"auto", "no-render", "hybrid", "fast-rt", "rt"}
if renderer not in valid:
logger.log_error(
f"Invalid renderer '{renderer}'. Must be one of {sorted(valid)}."
)
if renderer == "auto":
# Force auto-detection regardless of any previously forced default.
cfg.DEFAULT_RENDERER = "auto"
resolved = select_default_renderer(gpu_id)
else:
resolved = renderer
cfg.DEFAULT_RENDERER = resolved
logger.log_info(f"Default renderer set to '{resolved}'.")
return resolved
@property
def has_native_renderer(self) -> bool:
"""Whether the World owns a native renderer and visual resources."""
return self.sim_config.render_cfg.renderer != "no-render"
def _require_native_renderer(self, operation: str) -> None:
if not self.has_native_renderer:
raise RuntimeError(
f"{operation} requires a native renderer; the World uses no-render. "
"Explicitly select renderer='hybrid', 'fast-rt', or 'rt' before "
"construction, or override BaseEnv._requires_native_renderer() "
"to reserve rendering for custom cameras."
)
@cached_property
def num_envs(self) -> int:
"""Get the number of arenas in the simulation.
Returns:
int: number of arenas.
"""
return self.sim_config.num_envs
@property
def spawn_result(self) -> "Scene | None":
"""Return the finalized Scene, or ``None`` before first prepare."""
spawn_scene = getattr(self, "_spawn_scene", None)
if spawn_scene is None or not spawn_scene.builder.is_finalized:
return None
return spawn_scene.builder.result
@property
def is_use_gpu_physics(self) -> bool:
"""Whether the active physics backend is running on GPU."""
return self.device.type == "cuda"
@property
def physics_backend(self) -> str:
"""Return the active physics backend name."""
return self.physics.name
@property
def is_default_backend(self) -> bool:
"""Whether the Default physics backend is active."""
return self.physics.name == "default"
@property
def is_newton_backend(self) -> bool:
"""Whether the Newton physics backend is active."""
return self.physics.name == "newton"
@property
def _active_newton_solver_type(self) -> str | None:
"""Return the active backend's resolved solver type, when available."""
return self.physics.solver_type if self.is_newton_backend else None
@property
def newton_manager(self):
"""Compatibility accessor for the removed NewtonManager API.
A non-Newton backend still returns ``None``. The Newton backend raises
an actionable error because Spawn owns its World-level runtime and no
independent NewtonManager exists.
"""
return self.physics.newton_manager
@property
def differentiable_runtime(self):
"""Return the differentiable facade over the Spawn-owned Newton runtime."""
runtime = self.physics.differentiable_runtime
if runtime is None:
raise RuntimeError(
f"The {self.physics.name} physics backend does not expose a "
"differentiable runtime."
)
return runtime
@property
def is_physics_manually_update(self) -> bool:
return self._world.is_physics_manually_update()
@property
def asset_uids(self) -> List[str]:
"""Get all assets uid in the simulation.
The assets include lights, sensors, visual objects, robots, rigid objects and articulations.
Returns:
List[str]: list of all assets uid.
"""
uid_list = ["default_plane"]
uid_list.extend(list(self._lights.keys()))
uid_list.extend(list(self._sensors.keys()))
uid_list.extend(list(self._robots.keys()))
uid_list.extend(getattr(self, "_visual_objects", {}))
uid_list.extend(list(self._rigid_objects.keys()))
uid_list.extend(list(self._rigid_object_groups.keys()))
uid_list.extend(list(self._deformable_objects.keys()))
uid_list.extend(list(self._articulations.keys()))
return uid_list
@property
def visualization_runtime(self) -> VisualizationRuntime | None:
"""Return the active visualization runtime, if one has been started."""
return self._visualization_runtime
@property
def visualization_health(self) -> RuntimeHealth:
"""Return current visualization service and client health."""
from embodichain.lab.visualization import RuntimeHealth
if self._visualization_runtime is not None:
return self._visualization_runtime.health
configured = self.sim_config.visualization.backend == "viser"
return RuntimeHealth(
status="stopped" if configured else "disabled",
running=False,
endpoint=None,
client_count=0,
published_scene_revision=0,
)
@property
def visualization_stats(self) -> RuntimeStats | None:
"""Return visualization telemetry, or ``None`` before startup."""
if self._visualization_runtime is None:
return None
return self._visualization_runtime.stats
@property
def visualization_overlays(self) -> SceneOverlays | None:
"""Return the overlays included in every Viser scene frame."""
return self._visualization_overlays
[docs]
def set_visualization_overlays(self, overlays: SceneOverlays | None) -> None:
"""Set persistent overlays for automatic Viser captures.
The overlays remain active across :meth:`update` calls until replaced
or cleared with ``None``. When Viser is running, the new overlays are
published immediately.
Args:
overlays: Backend-neutral overlays to publish with every frame, or
``None`` to clear all persistent overlays.
"""
self._visualization_overlays = overlays
if (
self.sim_config.visualization.backend == "viser"
and self._visualization_runtime is not None
):
self.capture_visualization_safely(force=True)
[docs]
def notify_visualization_topology_changed(self) -> int:
"""Mark scene topology dirty and return its new local revision."""
self._visualization_topology_revision += 1
return self._visualization_topology_revision
[docs]
def start_visualization(self) -> VisualizationRuntime | None:
"""Start the configured live visualizer and publish the current scene."""
if self.sim_config.visualization.backend == "none":
return None
visualizable_assets = (
getattr(self, "_visual_objects", {}),
getattr(self, "_rigid_objects", {}),
getattr(self, "_rigid_object_groups", {}),
getattr(self, "_deformable_objects", {}),
getattr(self, "_articulations", {}),
getattr(self, "_robots", {}),
getattr(self, "_sensors", {}),
)
if getattr(self, "_spawn_scene", None) is not None and any(visualizable_assets):
self.prepare()
if getattr(self, "is_window_opened", False):
raise RuntimeError(
"Cannot start the Viser backend while the native DexSim window "
"is open. Close the native window before starting Viser."
)
if self._visualization_runtime is not None:
if self._visualization_runtime.is_running:
return self._visualization_runtime
self._visualization_runtime.stop()
self._visualization_runtime = None
from embodichain.lab.visualization import SceneExporter, VisualizationRuntime
visualization_cfg = self.sim_config.visualization
if (
visualization_cfg.allow_commands
and visualization_cfg.viser_server.host
not in {"127.0.0.1", "localhost", "::1"}
):
logger.log_warning(
"Viser simulation commands are enabled on a non-loopback interface. "
"Only expose this endpoint behind a trusted, authenticated boundary."
)
runtime = VisualizationRuntime(
SceneExporter(self, visualization_cfg),
visualization_cfg,
)
self._visualization_runtime = runtime
runtime.start()
self._visualization_manifest_topology_revision = (
self._visualization_topology_revision
)
self._visualization_error_reported = False
logger.log_info(f"Viser visualization ready at {runtime.endpoint}")
before_capture = None
if (
getattr(self, "_spawn_scene", None) is not None
and self.spawn_result is not None
):
before_capture = self._ensure_render_state
runtime.capture(
sim_step=self._visualization_sim_step,
sim_time=self._visualization_sim_time,
overlays=self._visualization_overlays,
force=True,
before_capture=before_capture,
)
return runtime
[docs]
def refresh_visualization(self) -> SceneManifest | None:
"""Publish current scene topology when Viser is active."""
runtime = self.start_visualization()
if runtime is None:
return None
for _, gizmo in self.get_gizmo_items():
cancel = getattr(gizmo, "cancel_interaction", None)
if cancel is not None:
cancel("viser:")
manifest = runtime.refresh_scene()
self._visualization_manifest_topology_revision = (
self._visualization_topology_revision
)
return manifest
[docs]
def capture_visualization(
self,
force: bool = False,
*,
capture_camera_images: bool = True,
) -> bool:
"""Capture current scene data for the configured visualizer.
Args:
force: Whether to bypass visualization frame-rate limiting.
capture_camera_images: Whether camera images may be captured.
Returns:
Whether scene or camera data was captured.
"""
runtime = self.start_visualization()
if runtime is None:
return False
if (
self._visualization_manifest_topology_revision
!= self._visualization_topology_revision
):
self.refresh_visualization()
return runtime.capture(
sim_step=self._visualization_sim_step,
sim_time=self._visualization_sim_time,
overlays=self._visualization_overlays,
force=force,
capture_camera_images=capture_camera_images,
before_capture=self._ensure_render_state,
)
[docs]
def capture_visualization_safely(
self,
force: bool = False,
*,
capture_camera_images: bool = True,
) -> None:
"""Update visualization without allowing failures to stop simulation.
The first visualization failure is logged and subsequent captures are
skipped until the runtime is restarted.
Args:
force: Whether to bypass visualization frame-rate limiting.
capture_camera_images: Whether camera images may be captured.
"""
if self._visualization_error_reported:
return
try:
self.capture_visualization(
force=force,
capture_camera_images=capture_camera_images,
)
except Exception as error:
if not self._visualization_error_reported:
logger.log_warning(f"Viser visualization update failed: {error!r}")
self._visualization_error_reported = True
[docs]
def stop_visualization(self) -> None:
"""Stop the visualization server and release its worker thread."""
runtime = self._visualization_runtime
if runtime is None:
return
try:
runtime.stop()
finally:
if getattr(self, "_picker_gizmo", None) is not None:
self._release_picker_gizmo()
for _, gizmo in self.get_gizmo_items():
cancel = getattr(gizmo, "cancel_interaction", None)
if cancel is not None:
cancel("viser:")
self._visualization_runtime = None
@staticmethod
def _get_process_renderer() -> str | None:
"""Read the engine configuration, which outlives individual Worlds."""
renderer = dexsim.get_world_config().renderer
# Before engine initialization get_world_config returns the default
# (Hybrid); NoRender therefore also identifies an externally initialized
# engine after all of its Worlds have been released.
if not (
SimulationManager._initialized_process_renderer
or dexsim.get_world_num() > 0
or renderer == dexsim.types.Renderer.NORENDER
):
return None
for name in ("no-render", "hybrid", "fast-rt", "rt"):
if RenderCfg(renderer=name).to_dexsim_flags() == renderer:
return name
raise RuntimeError(f"Unsupported DexSim process renderer: {renderer}.")
def _convert_sim_config(
self, sim_config: SimulationManagerCfg
) -> dexsim.WorldConfig:
world_config = dexsim.WorldConfig()
world_config.log_startup_info = sim_config.dexsim_startup_info
if sim_config.scene_node_capacity is not None:
world_config.scene_node_capacity = sim_config.scene_node_capacity
win_config = dexsim.WindowsConfig()
win_config.width = sim_config.width
win_config.height = sim_config.height
world_config.cpu_num = sim_config.cpu_num
world_config.win_config = win_config
world_config.open_windows = not sim_config.headless
self.is_window_opened = not sim_config.headless
world_config.backend = Backend.VULKAN
world_config.thread_mode = sim_config.thread_mode
world_config.cache_path = str(self._material_cache_dir)
process_renderer = SimulationManager._get_process_renderer()
if sim_config.render_cfg.renderer == "auto":
from embodichain.lab.sim import cfg
from embodichain.lab.sim.utility.render_utils import (
select_default_renderer,
)
if (
cfg.DEFAULT_RENDERER == "auto"
and process_renderer is not None
and process_renderer != "no-render"
):
resolved_renderer = process_renderer
self._renderer_selection_reason = "reuse process renderer"
elif (
cfg.DEFAULT_RENDERER == "auto"
and getattr(self, "_requires_native_renderer", None) is False
and sim_config.headless
and sim_config.visualization.backend == "none"
and getattr(sim_config.physics_cfg, "sync_to_renderer", None)
is not True
):
resolved_renderer = process_renderer or "no-render"
self._renderer_selection_reason = (
"reuse process renderer"
if process_renderer
else "no native render consumers"
)
else:
resolved_renderer = select_default_renderer(sim_config.gpu_id)
logger.log_debug(
f"Auto-selected '{resolved_renderer}' renderer for gpu_id={sim_config.gpu_id}."
)
sim_config.render_cfg.renderer = resolved_renderer
if process_renderer is not None and (
sim_config.render_cfg.renderer == "no-render"
) != (process_renderer == "no-render"):
raise RuntimeError(
f"DexSim's process renderer is already '{process_renderer}', but this "
f"environment requires '{sim_config.render_cfg.renderer}'. Closing a "
"World does not reset the process renderer. Start a new process to "
"change renderer. If any environment needs cameras, select a native "
"renderer before constructing its first World."
)
if (
sim_config.render_cfg.renderer == "no-render"
and world_config.scene_node_capacity
> SimulationManager._NO_RENDER_SCENE_NODE_CAPACITY
):
world_config.scene_node_capacity = (
SimulationManager._NO_RENDER_SCENE_NODE_CAPACITY
)
logger.log_debug(
f"NoRender limits scene node capacity to {world_config.scene_node_capacity}; "
f"native-renderer reservation was {sim_config.scene_node_capacity}."
)
sim_config.render_cfg.apply_to_dexsim_config(world_config)
if sim_config.render_cfg.renderer == "no-render":
if not sim_config.headless:
raise ValueError("renderer='no-render' requires headless=True.")
if getattr(sim_config.physics_cfg, "sync_to_renderer", None) is True:
raise ValueError("sync_to_renderer=True requires a native renderer.")
if type(sim_config.device) is str:
self.device = torch.device(sim_config.device)
else:
self.device = sim_config.device
if self.device.type == "cuda":
if self.device.index is not None and sim_config.gpu_id != self.device.index:
logger.log_warning(
f"Conflict gpu_id {sim_config.gpu_id} and device index {self.device.index}. Using device index."
)
sim_config.gpu_id = self.device.index
self.device = torch.device(f"cuda:{sim_config.gpu_id}")
world_config.gpu_id = sim_config.gpu_id
# Apply backend-specific WorldConfig fields (default tolerances/GPU flags
# or the Newton cfg) via the active backend.
self.physics.configure_world(world_config, sim_config)
return world_config
def _init_sim_resources(self) -> None:
"""Initialize the default simulation resources."""
from embodichain.data.assets import SimResources
self._default_resources = SimResources()
[docs]
def register_kinematic_joint_trajectory(
self,
uid: str,
joint_positions: torch.Tensor | np.ndarray,
*,
fps: float | None = None,
root_poses: torch.Tensor | np.ndarray | None = None,
) -> None:
"""Register a Newton kinematic joint trajectory for every arena.
The leading trajectory dimension follows EmbodiChain's batched arena
layout. Each arena row is lowered to one DexSim runtime control with
the corresponding concrete Spawn articulation path. Row zero is the
initial sample; when ``fps`` is omitted, each call to :meth:`update`
advances to the next sample.
.. attention::
Declare the target robot or articulation first, then call this
method before :meth:`prepare`. Runtime controls are part of the
finalized Newton simulation pipeline and cannot be added later.
Args:
uid: UID of a robot or articulation declared on this manager.
joint_positions: Batched positions in the articulation's public
qpos order with shape ``(num_envs, frames, dof)``.
fps: Optional trajectory sample rate. When omitted, samples advance
once per EmbodiChain physics frame.
root_poses: Optional batched world-space root transforms with shape
``(num_envs, frames, 4, 4)``.
Raises:
RuntimeError: If the active backend is not Newton, the Spawn scene
is already finalized, or its arena count is inconsistent.
KeyError: If ``uid`` is not a declared robot or articulation.
ValueError: If an input has an invalid batch shape or non-finite
values.
"""
if not isinstance(uid, str) or not uid:
raise ValueError("uid must be a non-empty string.")
if not self.is_newton_backend:
raise RuntimeError(
"Kinematic joint trajectory controls require the Newton backend."
)
if uid not in self._robots and uid not in self._articulations:
raise KeyError(f"Robot or articulation {uid!r} is not declared.")
scene = self._spawn_scene
if scene.builder.is_finalized:
raise RuntimeError(
"Kinematic joint trajectories must be registered before "
"SimulationManager.prepare()."
)
if isinstance(joint_positions, torch.Tensor):
positions = joint_positions.detach().cpu().numpy()
else:
positions = np.asarray(joint_positions)
positions = np.asarray(positions, dtype=np.float32)
expected_prefix = (self.num_envs,)
if (
positions.ndim != 3
or positions.shape[:1] != expected_prefix
or positions.shape[1] == 0
or positions.shape[2] == 0
):
raise ValueError(
"joint_positions must have non-empty shape "
f"({self.num_envs}, frames, dof); got {positions.shape}."
)
if not np.isfinite(positions).all():
raise ValueError("joint_positions must contain only finite values.")
poses: np.ndarray | None = None
if root_poses is not None:
if isinstance(root_poses, torch.Tensor):
poses = root_poses.detach().cpu().numpy()
else:
poses = np.asarray(root_poses)
poses = np.asarray(poses, dtype=np.float32)
expected_shape = (self.num_envs, positions.shape[1], 4, 4)
if poses.shape != expected_shape:
raise ValueError(
f"root_poses must have shape {expected_shape}; got {poses.shape}."
)
if not np.isfinite(poses).all():
raise ValueError("root_poses must contain only finite values.")
arena_names = scene.arena_names
if len(arena_names) != self.num_envs:
raise RuntimeError(
"Spawn arena count does not match SimulationManager.num_envs: "
f"{len(arena_names)} != {self.num_envs}."
)
from dexsim.engine.newton_physics import KinematicJointTrajectoryControl
controls = tuple(
KinematicJointTrajectoryControl(
f"{arena_name}/{uid}",
positions[env_index],
fps=fps,
root_poses=None if poses is None else poses[env_index],
)
for env_index, arena_name in enumerate(arena_names)
)
for control in controls:
scene.builder.add_runtime_control(control)
[docs]
def register_particle_contact_material_schedule(
self,
keyframes: Mapping[str, Sequence[Sequence[float]]],
) -> None:
"""Register time-varying Newton particle contact properties.
Supported tracks are ``dynamic_friction``, ``stiffness``, and
``damping``. Values apply scene-wide to particle-versus-rigid contacts
and are sampled piecewise-constantly in simulation time.
.. attention::
Register this control before :meth:`prepare`. This host-side
control requires direct Newton stepping and therefore disables
CUDA Graph replay for the finalized simulation.
Args:
keyframes: Particle contact-property tracks containing ``(time,
value)`` pairs.
Raises:
RuntimeError: If the backend is not Newton or the scene is already
finalized.
ValueError: If the keyframes are invalid.
"""
if not self.is_newton_backend:
raise RuntimeError(
"Particle contact material schedules require the Newton backend."
)
scene = self._spawn_scene
if scene.builder.is_finalized:
raise RuntimeError(
"Particle contact material schedules must be registered before "
"SimulationManager.prepare()."
)
from dexsim.engine.newton_physics import ParticleContactMaterialSchedule
scene.builder.add_runtime_control(ParticleContactMaterialSchedule(keyframes))
[docs]
def register_kinematic_nodal_trajectory(
self,
uid: str,
node_indices: torch.Tensor | np.ndarray | Sequence[int],
position_offsets: torch.Tensor | np.ndarray,
*,
fps: float | None = None,
rebuild_self_contact_bvh: bool = False,
) -> None:
"""Register a Newton trajectory for selected deformable nodes.
Each selected node is fixed during Newton model construction and then
moved relative to the world position captured when the runtime control
initializes. The manager expands the batched offsets to one control per
Arena without exposing the private Spawn scene.
When ``fps`` is provided, samples are linearly interpolated at Newton
substep times. Otherwise one sample is consumed per substep. The final
sample is held after the trajectory ends.
.. attention::
Declare the deformable first and clear the Newton ``ACTIVE`` bit in
its ``particle_flags`` for every selected node. Then register this
control before :meth:`prepare`. This host-side control makes Newton
use direct substep launches instead of CUDA Graph replay. Surface
node indices can follow an array-backed mesh directly; volume node
indices refer to the generated tetrahedral simulation particles,
not source-mesh vertices.
Args:
uid: UID of a deformable declared on this manager.
node_indices: Shared one-dimensional simulation-particle indices.
position_offsets: Batched world-frame position offsets with shape
``(num_envs, samples, selected_nodes, 3)``.
fps: Optional trajectory sample rate in samples per second.
rebuild_self_contact_bvh: Whether to request a full solver BVH
rebuild at the start of every physics frame when supported.
Raises:
RuntimeError: If the active backend is not Newton, the Spawn scene
is already finalized, or its Arena count is inconsistent.
KeyError: If ``uid`` is not a declared deformable.
TypeError: If node indices, ``fps``, or the BVH option have invalid
types.
ValueError: If an input has an invalid shape or value, or selected
nodes were not configured as inactive particles.
"""
if not isinstance(uid, str) or not uid:
raise ValueError("uid must be a non-empty string.")
if not self.is_newton_backend:
raise RuntimeError(
"Kinematic nodal trajectory controls require the Newton backend."
)
if uid not in self._deformable_objects:
raise KeyError(f"Deformable object {uid!r} is not declared.")
scene = self._spawn_scene
if scene.builder.is_finalized:
raise RuntimeError(
"Kinematic nodal trajectories must be registered before "
"SimulationManager.prepare()."
)
if isinstance(node_indices, torch.Tensor):
raw_indices = node_indices.detach().cpu().numpy()
else:
raw_indices = np.asarray(node_indices)
if raw_indices.ndim != 1 or raw_indices.size == 0:
raise ValueError("node_indices must be a non-empty one-dimensional array.")
if raw_indices.dtype.kind not in "iu":
raise TypeError("node_indices must contain integers.")
if np.any(raw_indices < 0):
raise ValueError("node_indices must be non-negative.")
if np.any(raw_indices > np.iinfo(np.int32).max):
raise ValueError("node_indices exceed the supported int32 range.")
indices = np.asarray(raw_indices, dtype=np.int32)
if len(np.unique(indices)) != len(indices):
raise ValueError("node_indices must not contain duplicates.")
configured_flags = self._deformable_objects[uid].cfg.particle_flags
if configured_flags is None:
raise ValueError(
"Kinematic nodes require particle_flags with the Newton ACTIVE "
"bit cleared before model construction."
)
flags = np.asarray(configured_flags)
if flags.ndim == 0:
selected_flags = np.full(len(indices), int(flags), dtype=np.int64)
elif flags.ndim == 1:
if int(indices.max()) >= len(flags):
raise ValueError(
"node_indices exceed the configured particle_flags length: "
f"max index {int(indices.max())}, length {len(flags)}."
)
selected_flags = flags[indices]
else:
raise ValueError(
"Configured particle_flags must be scalar or one-dimensional."
)
newton_active_particle_flag = 1
if np.any(
np.asarray(selected_flags, dtype=np.int64) & newton_active_particle_flag
):
raise ValueError(
"Every kinematic node must have the Newton ACTIVE particle flag cleared."
)
if isinstance(position_offsets, torch.Tensor):
offsets = position_offsets.detach().cpu().numpy()
else:
offsets = np.asarray(position_offsets)
offsets = np.asarray(offsets, dtype=np.float32)
if (
offsets.ndim != 4
or offsets.shape[0] != self.num_envs
or offsets.shape[1] == 0
or offsets.shape[2:] != (len(indices), 3)
):
raise ValueError(
"position_offsets must have non-empty shape "
f"({self.num_envs}, samples, {len(indices)}, 3); got "
f"{offsets.shape}."
)
if not np.isfinite(offsets).all():
raise ValueError("position_offsets must contain only finite values.")
if fps is not None:
if isinstance(fps, bool) or not isinstance(
fps, (int, float, np.integer, np.floating)
):
raise TypeError("fps must be a finite positive number or None.")
fps = float(fps)
if not np.isfinite(fps) or fps <= 0.0:
raise ValueError("fps must be a finite positive number.")
if not isinstance(rebuild_self_contact_bvh, bool):
raise TypeError("rebuild_self_contact_bvh must be a bool.")
arena_names = scene.arena_names
if len(arena_names) != self.num_envs:
raise RuntimeError(
"Spawn arena count does not match SimulationManager.num_envs: "
f"{len(arena_names)} != {self.num_envs}."
)
from embodichain.lab.sim._runtime_controls import (
_KinematicNodalTrajectoryControl,
)
controls = tuple(
_KinematicNodalTrajectoryControl(
f"{arena_name}/{uid}",
indices,
offsets[env_index],
fps=fps,
rebuild_self_contact_bvh=rebuild_self_contact_bvh,
)
for env_index, arena_name in enumerate(arena_names)
)
for control in controls:
scene.builder.add_runtime_control(control)
[docs]
def prepare(self) -> None:
"""Materialize declarations, bind state, and restore camera parents."""
self._ready_spawn_topology_revision = -1
scene = self._spawn_scene
result = scene.builder.result
if (
not scene.builder.is_finalized
or result is None
or result.needs_rebuild
or scene.builder.has_pending_changes
):
if result is not None:
# These native cameras are owned outside Spawn. Detach them
# before a rebuild removes their articulation render parents.
for sensor in self._sensors.values():
if (
isinstance(sensor, Camera)
and sensor.cfg.extrinsics.parent is not None
):
sensor._detach_from_parent_nodes()
result = scene.commit()
self._env = result.get_arena("default")
self._arenas = [result.get_arena(name) for name in scene.arena_names]
self.__dict__.pop("arena_offsets", None)
if self._default_plane is None:
self._bind_default_plane(scene.handles("default_plane")[0])
# Runtime readiness belongs to the SimulationManager. Keep this and
# facade binding outside the topology-change branch so a failed call
# remains retryable without rematerializing the scene.
scene.prepare_runtime_config(result)
self._prepare_spawn_runtime(result)
scene.bind()
for visual in getattr(self, "_visual_objects", {}).values():
if visual.is_spawn_bound:
visual._sync_spawn_state()
self._sync_spawn_render_state(result)
topology_revision = int(result.topology_revision)
if (
getattr(self, "_camera_attachment_topology_revision", -1)
!= topology_revision
):
self._attach_parented_cameras()
self._camera_attachment_topology_revision = topology_revision
self._ready_spawn_topology_revision = topology_revision
def _prepare_spawn_runtime(self, result: Scene) -> None:
"""Prepare backend runtime buffers for one Spawn topology revision."""
topology_revision = int(result.topology_revision)
if getattr(self, "_prepared_spawn_topology_revision", -1) == topology_revision:
return
self.physics.prepare_spawn_runtime(result)
self._prepared_spawn_topology_revision = topology_revision
def _sync_spawn_render_state(self, result: Scene) -> None:
"""Publish newly bound state once for each Spawn topology revision."""
topology_revision = int(result.topology_revision)
if (
getattr(self, "_synced_spawn_render_topology_revision", -1)
== topology_revision
):
return
self.physics.sync_render_state(result)
self._synced_spawn_render_topology_revision = topology_revision
if getattr(self, "_render_state_published", None) is not None:
self._render_state_published = True
def _ensure_render_state(self) -> None:
"""Reuse publication only inside a caller-owned, read-only frame."""
if getattr(self, "_render_state_published", None) is True:
return
self.sync_render_state()
if getattr(self, "_render_state_published", None) is not None:
self._render_state_published = True
[docs]
@contextmanager
def render_frame(
self,
*,
force_visualization: bool = False,
capture_camera_images: bool = True,
) -> Iterator[None]:
"""Share one state publication across the frame's visual consumers.
Render camera observations inside this context. On exit, due recording
and Viser frames consume the same state. An open native window is a
consumer even when automatic Newton synchronization is disabled.
.. attention::
Complete physics steps, resets and direct state writes before
entering. If state must change inside the context, explicitly call
:meth:`sync_render_state` afterwards. Publication is never cached
across contexts or independent camera/visualization calls.
Args:
force_visualization: Bypass the Viser frame-rate limiter.
capture_camera_images: Allow Viser to capture camera images.
Yields:
Control to the caller rendering this frame's camera observations.
"""
previous = getattr(self, "_render_state_published", None)
self._render_state_published = False
record_dt = self._pending_record_dt
self._pending_record_dt = 0.0
observations_complete = False
try:
if self.is_window_opened or (
record_dt > 0 and self.physics._requires_step_render_sync
):
self._ensure_render_state()
yield
observations_complete = True
if (
self._window_record_state is not None
and self._window_record_state.capture_from_sim_update
and record_dt > 0
):
with self.profiler.section("window_record_capture"):
self._step_window_record_from_sim_update(
self._window_record_state, record_dt
)
if self.sim_config.visualization.backend == "viser":
with self.profiler.section("visualization_capture"):
self.capture_visualization_safely(
force=force_visualization,
capture_camera_images=capture_camera_images,
)
finally:
if not observations_complete:
self._pending_record_dt += record_dt
# A nested frame may contain an explicit publication after a state
# edit. Never restore an outer frame's obsolete publication flag.
self._render_state_published = None if previous is None else False
[docs]
def sync_render_state(self) -> None:
"""Publish current physics state to render resources without stepping.
Use this after direct state writes, such as an environment reset, and
before reading cameras or publishing a visualization frame.
Raises:
RuntimeError: If the Spawn scene has not been prepared.
"""
result = self.spawn_result
if result is None:
raise RuntimeError(
"Render-state synchronization requires a prepared Spawn scene."
)
self.physics.sync_render_state(result)
if getattr(self, "_render_state_published", None) is not None:
self._render_state_published = True
[docs]
def enable_physics(self, enable: bool) -> None:
"""Enable or disable physics simulation.
Args:
enable (bool): whether to enable physics simulation.
"""
self._world.enable_physics(enable)
[docs]
def init_gpu_physics(self) -> None:
"""Prepare the Spawn-owned physics runtime.
This backwards-compatible alias now has the same backend-neutral
behavior as :meth:`prepare`.
"""
self.prepare()
[docs]
def finalize_newton_physics(self) -> None:
"""Prepare the Spawn-owned physics runtime.
This backwards-compatible alias now has the same backend-neutral
behavior as :meth:`prepare`.
"""
self.prepare()
[docs]
def render_camera_group(self, group_ids: list[int]) -> None:
"""Synchronize physics state and render camera groups.
An empty group list skips camera rendering. An open native window still
consumes current state. Use :meth:`render_frame` to share publication
with recording and browser visualization.
Args:
group_ids (list[int]): The list of camera group ids to render.
Note: This interface is only valid when Ray Tracing rendering backend is enabled.
"""
if group_ids or self.is_window_opened:
self._ensure_render_state()
if group_ids:
self._world.render_camera_group(group_ids)
self._log_scene_summary()
[docs]
def update(
self,
physics_dt: float | None = None,
step: int = 10,
*,
after_substep: Callable[[float], None] | None = None,
render_final_step: bool = True,
) -> None:
"""Advance physics explicitly and publish the resulting simulation state.
Each substep applies pending Gizmo controls before advancing the world,
then updates simulation time, recording, and browser visualization.
Physics does not advance in the background while the caller is idle.
Args:
physics_dt (float | None, optional): the time step for physics simulation. Defaults to None.
step (int, optional): the number of :meth:`World.update` calls per invocation. Defaults to 10.
after_substep: Observer called with the elapsed physics time after
each world update, before recording and visualization. It runs
inside this update call and is not retained by the manager.
render_final_step: Render the final substep immediately. Gym defers
it until after interval events, then enters :meth:`render_frame`
around observations to share publication with other consumers.
"""
with self.profiler.section("sim_update", is_root=True):
with self.profiler.section("gpu_physics_check"):
self.prepare()
with self.profiler.section("physics_steps"):
if physics_dt is None:
with self.profiler.section("resolve_physics_dt"):
physics_dt = self.sim_config.physics_dt
for i in range(step):
with self.profiler.section("gizmo_update"):
self.update_gizmos()
with self.profiler.section("world_update"):
with _temporary_warp_kernel_log_suppression(
self.sim_config.physics_cfg
):
self.physics._step(physics_dt)
self._visualization_sim_step += 1
self._visualization_sim_time += physics_dt
if after_substep is not None:
with self.profiler.section("after_substep"):
after_substep(physics_dt)
self._pending_record_dt += physics_dt
if render_final_step or i < step - 1:
with self.render_frame(capture_camera_images=i == step - 1):
pass
if step > 0:
self._log_scene_summary()
[docs]
def get_env(self, arena_index: int = -1) -> dexsim.environment.Arena:
"""Get the arena or env by index.
If arena_index is -1, return the global env.
If arena_index is valid, return the corresponding arena.
Args:
arena_index (int, optional): the index of arena to get, -1 for global env. Defaults to -1.
Returns:
dexsim.environment.Arena: The arena or global env.
"""
if arena_index >= 0:
if arena_index > len(self._arenas) - 1:
logger.log_error(
f"Invalid arena index: {arena_index}. Current number of arenas: {len(self._arenas)}"
)
return self._arenas[arena_index]
else:
return self._env
[docs]
def visualize_point_cloud(
self,
points: torch.Tensor | np.ndarray,
colors: torch.Tensor | np.ndarray | None = None,
point_size: float = 2.0,
name: str = "point_cloud",
) -> dexsim.models.PointCloud:
"""Visualize a static point cloud in the native simulation viewer.
Each invocation creates a separate native point-cloud object. This
convenience API is intended for static data, not incremental or
streaming updates.
Args:
points: Point positions with shape ``(N, 3)``.
colors: Optional per-point RGB or RGBA colors with shape ``(N, 3)``
or ``(N, 4)``. Values in ``[0, 255]`` are normalized to
``[0, 1]``. The alpha channel of RGBA input is ignored by the
native renderer. Defaults to green.
point_size: Native renderer point size. Defaults to ``2.0``.
name: Name assigned to the native point-cloud object.
Returns:
The native DexSim point-cloud handle.
Raises:
RuntimeError: If there is no active simulation environment.
ValueError: If the points or colors do not have a supported shape.
"""
if isinstance(points, torch.Tensor):
points = points.detach().cpu().numpy()
points = np.asarray(points, dtype=np.float32)
if points.ndim != 2 or points.shape[1] != 3:
raise ValueError(f"Points must have shape (N, 3), got {points.shape}")
if len(points) == 0:
raise ValueError("Points array is empty")
if colors is None:
colors = np.tile(
np.array((0.0, 1.0, 0.0), dtype=np.float32), (len(points), 1)
)
else:
if isinstance(colors, torch.Tensor):
colors = colors.detach().cpu().numpy()
colors = np.asarray(colors)
if colors.ndim != 2 or colors.shape[0] != len(points):
raise ValueError(
f"Colors must have shape ({len(points)}, 3) or ({len(points)}, 4), "
f"got {colors.shape}"
)
if colors.shape[1] not in (3, 4):
raise ValueError(
"Colors must have 3 (RGB) or 4 (RGBA) channels, "
f"got {colors.shape[1]}"
)
if colors.max() > 1.0:
colors = colors / 255.0
colors = np.asarray(colors[:, :3], dtype=np.float32)
env = self.get_env()
if env is None:
raise RuntimeError("Simulation manager has no active simulation")
point_cloud = env.create_point_cloud(name=name)
point_cloud.add_points(points)
point_cloud.set_colors(colors)
point_cloud.set_point_size(point_size)
logger.log_info(
f"Created point cloud '{name}' with {len(points)} points "
f"(point_size={point_size})"
)
return point_cloud
[docs]
def get_world(self) -> dexsim.World:
return self._world
[docs]
def get_physics_scene(self) -> "PhysicsScene":
"""Return the Default backend's compatibility scene after Spawn preparation.
Newton has no ``PhysicsScene`` facade and raises with guidance to use
:attr:`spawn_result` instead.
"""
return self.physics.get_scene()
[docs]
def can_open_native_window(self) -> bool:
"""Return whether the native DexSim window may be opened.
The Viser backend owns visualization while it is configured or
running, so a native window must not be opened for the same simulation.
Returns:
``True`` unless the Viser backend is configured or running.
"""
return (
self.sim_config.visualization.backend != "viser"
and self._visualization_runtime is None
)
[docs]
def open_window(self) -> bool:
"""Open the native DexSim simulation window when allowed.
Viser owns visualization while it is configured or running. In that
case this method safely skips the native window so launchers do not
need a separate Viser condition.
Returns:
``True`` when the native window is open, otherwise ``False``.
"""
if not self.can_open_native_window():
logger.log_info(
"Skipping the native DexSim window because the Viser backend "
"is configured or running."
)
return False
if self.is_window_opened:
return True
self._world.open_window()
self._window = self._world.get_windows()
if self._window is None:
return False
self._enable_default_entity_gizmo()
if (
self._window_record_hotkey_cfg is not None
and self._window_record_input_control is None
):
self.enable_window_record_hotkey(**self._window_record_hotkey_cfg)
if (
self._window_camera_pose_hotkey_cfg is not None
and self._window_camera_pose_input_control is None
):
self.enable_window_camera_pose_hotkey(**self._window_camera_pose_hotkey_cfg)
self.is_window_opened = True
if self.spawn_result is not None:
self.sync_render_state()
self._log_scene_summary()
return True
[docs]
def close_window(self) -> None:
"""Close the simulation window."""
if self.is_window_recording():
self.stop_window_record()
for _, gizmo in self.get_gizmo_items():
gizmo.detach_native_window()
self._world.close_window()
self._window = None
self._window_record_input_control = None
self._window_camera_pose_input_control = None
self.is_window_opened = False
[docs]
def set_indirect_lighting(self, name: str) -> None:
"""Set indirect lighting.
Args:
name (str): name of path of the indirect lighting.
"""
if name.startswith("/") is False:
ibl_path = self._default_resources.get_ibl_path(name)
logger.log_info(f"Set IBL {name} from sim default resources.")
else:
ibl_path = name
logger.log_info(f"Set IBL {name} from custom path.")
self._env.set_IBL(ibl_path)
[docs]
def set_emission_light(
self, color: Sequence[float] | None = None, intensity: float | None = None
) -> None:
"""Set environment emission light.
Args:
color (Sequence[float] | None): color of the light.
intensity (float | None): intensity of the light.
"""
if color is not None:
self._env.set_env_light_emission(color)
if intensity is not None:
self._env.set_env_light_intensity(intensity)
def _declare_spawn_default_plane(self) -> None:
"""Declare the global ground in the World's Spawn scene."""
from dexsim.spawn import (
CollisionApproximation,
CollisionDesc,
DexsimCollisionDesc,
GeometryDesc,
NewtonCollisionDesc,
ObjectDesc,
RenderDesc,
RigidBodyPhysicsDesc,
)
default_length = 1000.0
geometry = GeometryDesc.plane(default_length)
repeat_uv_size = default_length / 2.0
render = RenderDesc.from_geometry(
geometry,
material=self._spawn_default_plane_material,
)
render.uv_coords = np.asarray(
[
[0.0, 0.0],
[repeat_uv_size, 0.0],
[repeat_uv_size, repeat_uv_size],
[0.0, repeat_uv_size],
],
dtype=np.float32,
)
collision = CollisionDesc.from_geometry(
geometry,
approximation=CollisionApproximation.NONE,
)
collision.dexsim = DexsimCollisionDesc(
dynamic_friction=0.5,
static_friction=0.5,
)
collision.newton = NewtonCollisionDesc(mu=0.5)
collision.render_source_index = 0
descriptor = ObjectDesc(
name="default_plane",
renders=[render],
collisions=[collision],
physics=RigidBodyPhysicsDesc.static(),
per_env=False,
)
self._spawn_scene.declare(
"rigid_object",
"default_plane",
descriptor,
)
handles = self._spawn_scene.handles("default_plane")
if handles:
self._bind_default_plane(handles[0])
def _bind_default_plane(self, plane: Any) -> None:
"""Retain the spawned ground plane and apply its visibility."""
self._default_plane = plane
if self.has_native_renderer:
plane.set_visible(self._spawn_default_plane_visibility)
[docs]
def set_default_global_lighting(self) -> None:
"""Set default global lighting for the scene.
Initialize environment emission without creating direct lights or
changing existing scene lights. Add direct lights explicitly through
:meth:`add_light` or the task's lighting configuration.
"""
self._require_native_renderer("Default global lighting")
self.set_emission_light([0.92, 0.95, 1.0], 80.0)
[docs]
def set_default_background(self) -> None:
"""Set default background."""
self._require_native_renderer("Default background")
mat_name = "plane_mat"
mat_path = self._default_resources.get_material_path("PlaneDark")
color_texture = os.path.join(mat_path, "PlaneDark_2K_Color.jpg")
roughness_texture = os.path.join(mat_path, "PlaneDark_2K_Roughness.jpg")
mat = self.create_visual_material(
cfg=VisualMaterialCfg(
uid=mat_name,
base_color_texture=color_texture,
roughness_texture=roughness_texture,
roughness=1.0,
)
)
material = mat.get_instance("plane_mat").mat
# Consumed by _declare_spawn_default_plane(). Keeping the native
# material in the descriptor preserves the VisualMaterial registry
# used by visual randomization without forcing finalization.
self._spawn_default_plane_material = material
self._visual_materials[mat_name] = mat
[docs]
def set_ground_plane_visibility(self, visible: bool) -> None:
"""Set native ground rendering visibility while preserving collisions.
Applies to native windows and cameras. The setting is retained until
the ground is materialized, and does not control the Viser grid.
NoRender stores the setting without accessing native visual resources.
Args:
visible: Whether the default ground plane is rendered.
"""
self._spawn_default_plane_visibility = bool(visible)
if self._default_plane is None or not self.has_native_renderer:
return
self._default_plane.set_visible(bool(visible))
[docs]
def set_texture_cache(
self, key: str, texture: Union[torch.Tensor, List[torch.Tensor]]
) -> None:
"""Set the texture to the global texture cache.
Args:
key (str): The key of the texture.
texture (Union[torch.Tensor, List[torch.Tensor]]): The texture data.
"""
self._texture_cache[key] = texture
[docs]
def get_texture_cache(
self, key: str | None = None
) -> torch.Tensor | list[torch.Tensor] | None:
"""Get the texture from the global texture cache.
Args:
key (str | None, optional): The key of the texture. If None, return None. Defaults to None.
Returns:
torch.Tensor | list[torch.Tensor] | None: The texture if found, otherwise None.
"""
if key is None:
return self._texture_cache
if key not in self._texture_cache:
logger.log_warning(f"Texture {key} not found in global texture cache.")
return None
return self._texture_cache[key]
[docs]
def get_asset(
self, uid: str
) -> (
Light
| BaseSensor
| Robot
| RigidObject
| VisualObject
| RigidObjectGroup
| DeformableObject
| Articulation
| None
):
"""Get an asset by its UID.
The asset can be a light, sensor, robot, visual object, rigid object, deformable, or
articulation.
Args:
uid (str): The UID of the asset.
Returns:
The asset instance if found, otherwise ``None``.
"""
if uid in self._lights:
return self._lights[uid]
if uid in self._sensors:
return self._sensors[uid]
if uid in self._robots:
return self._robots[uid]
if uid in getattr(self, "_visual_objects", {}):
return self._visual_objects[uid]
if uid in self._rigid_objects:
return self._rigid_objects[uid]
if uid in self._rigid_object_groups:
return self._rigid_object_groups[uid]
if uid in self._deformable_objects:
return self._deformable_objects[uid]
if uid in self._articulations:
return self._articulations[uid]
logger.log_warning(f"Asset {uid} not found.")
return None
_LIGHT_TYPE_MAP: dict[str, LightType] = {
"point": LightType.POINT,
"sun": LightType.SUN,
"direction": LightType.DIRECTION,
"spot": LightType.SPOT,
"rect": LightType.RECT,
"mesh": LightType.MESH,
}
_GLOBAL_LIGHT_TYPES: tuple[str, ...] = ("sun", "direction")
[docs]
def add_light(self, cfg: LightCfg) -> Light:
"""Create a light in the scene.
Supports six light types: ``"point"``, ``"sun"``, ``"direction"``,
``"spot"``, ``"rect"``, and ``"mesh"``. See :class:`LightCfg` for
type-specific configuration fields.
.. attention::
``"sun"`` and ``"direction"`` lights are global scene lights
(infinite-distance directional light sources). They are created
as a single instance on the root environment, not batched per
environment. All other types are created as per-environment
batched lights.
Args:
cfg (LightCfg): Configuration for the light, including type, color,
intensity, and type-specific properties.
Returns:
Light: The created light instance.
Raises:
ValueError: If ``cfg.light_type`` is not supported.
"""
if cfg.uid is None:
uid = "light"
cfg.uid = uid
else:
uid = cfg.uid
if uid in self._lights:
logger.log_error(f"Light {uid} already exists.")
light_type = self._LIGHT_TYPE_MAP.get(cfg.light_type)
if light_type is None:
supported = ", ".join(self._LIGHT_TYPE_MAP)
raise ValueError(
f"Unsupported light type {cfg.light_type!r}. "
f"Supported types: {supported}."
)
if cfg.light_type == "mesh" and not cfg.mesh_path:
logger.log_warning(
f"Mesh light '{uid}' has no mesh_path set. "
f"Use set_mesh() to assign a MeshObject."
)
if cfg.light_type == "rect" and (cfg.rect_width <= 0 or cfg.rect_height <= 0):
logger.log_warning(
f"Rect light '{uid}' has zero or negative dimensions "
f"(width={cfg.rect_width}, height={cfg.rect_height})."
)
if cfg.light_type in self._GLOBAL_LIGHT_TYPES:
batch_lights = Light(
cfg=cfg,
entities=[self._env.create_light(uid, light_type)],
)
else:
batch_lights = Light(
cfg=cfg,
entities=[
arena.create_light(f"{uid}_{index}", light_type)
for index, arena in enumerate(self._arenas)
],
)
self._lights[uid] = batch_lights
self.notify_visualization_topology_changed()
return batch_lights
[docs]
def get_light(self, uid: str) -> Light | None:
"""Get a light by its UID.
Args:
uid (str): The UID of the light.
Returns:
Light | None: The light instance if found, otherwise None.
"""
if uid not in self._lights:
logger.log_warning(f"Light {uid} not found.")
return None
return self._lights[uid]
[docs]
def get_light_uid_list(self) -> List[str]:
"""Get current light uid list
Returns:
List[str]: list of light uid.
"""
return list(self._lights.keys())
[docs]
def add_usd(
self,
name: str,
file_path: str,
*,
pose: np.ndarray | None = None,
robot_cfgs: dict[str, RobotCfg] | None = None,
) -> dict[str, VisualObject | RigidObject | Articulation | Robot]:
"""Declare the supported entities in a USD scene.
The returned facades are keyed by their USD prim paths. They remain in
declared state until :meth:`prepare` finalizes the shared Spawn scene,
then bind in place to the resulting DexSim handles.
Objects without physics or collision geometry become VisualObjects.
Conflicting render-only names receive a ``name_`` prefix; dictionary
keys retain their original USD prim paths. Other UID conflicts fail
before native declarations.
USD does not identify which articulations should expose EmbodiChain's
robot interface. Pass those explicitly through ``robot_cfgs``; all
other articulation descriptions become :class:`Articulation` objects.
Args:
name: Name passed to DexSim's USD scene parser.
file_path: USD, USDA, USDC, or USDZ file path.
pose: Optional scene-root transform.
robot_cfgs: Robot configurations keyed by USD prim path. These
provide robot-side metadata while physics remains authored by
the USD scene.
Returns:
Supported EmbodiChain facades keyed by USD prim path.
Raises:
RuntimeError: If called after the Spawn scene was finalized.
"""
if self.spawn_result is not None:
raise RuntimeError(
"add_usd() must be called before SimulationManager.prepare()."
)
from dexsim.spawn import ArticulationDesc, MeshObjectDesc
from dexsim.kit.usd import parse_usd
source = parse_usd(file_path, root_name=name)
occupied = set(self.asset_uids)
for descriptor in (
*getattr(self._spawn_scene.builder, "mesh_objects", ()),
*getattr(self._spawn_scene.builder, "scene_entities", ()),
):
occupied.add(descriptor.name)
# Exported scenes can contain a visual ground named like our global
# physical ground. Keep both and retain the original USD path as the key.
for descriptor in source.mesh_objects:
if (
descriptor.name in occupied
and descriptor.physics is None
and not descriptor.collisions
):
descriptor.name = f"{name}_{descriptor.name}"
for collection in (
"mesh_objects",
"articulations",
"particle_sets",
"cameras",
"lights",
"cables",
"records",
):
for descriptor in getattr(source, collection, ()):
if descriptor.name in occupied:
raise ValueError(
f"USD asset UID already exists: {descriptor.name!r}."
)
occupied.add(descriptor.name)
if pose is not None and hasattr(descriptor, "pose"):
descriptor.pose = (
np.asarray(pose, dtype=np.float32).reshape(4, 4)
[docs]
@ descriptor.pose
)
source.materials = self._spawn_scene._resolve_materials(source.materials)
descriptors = self._spawn_scene.builder.add_scene(source, per_env=True)
assets: dict[str, VisualObject | RigidObject | Articulation | Robot] = {}
robot_cfgs = robot_cfgs or {}
for descriptor in descriptors:
source_path = (
descriptor.usd.prim_path
if descriptor.usd is not None and descriptor.usd.prim_path
else descriptor.name
)
if type(descriptor) is MeshObjectDesc:
if descriptor.physics is None and not descriptor.collisions:
cfg = VisualObjectCfg(
uid=descriptor.name,
source=UsdSourceCfg(
file_path=file_path,
source_prim_path=(
descriptor.usd.prim_path
if descriptor.usd is not None
else None
),
),
init_local_pose=descriptor.pose.copy(),
body_scale=tuple(
float(value) for value in descriptor.body_scale
),
)
facade = VisualObject(cfg=cfg, device=self.device)
self._spawn_scene.track(
"visual_object", descriptor.name, descriptor, facade=facade
)
self._visual_objects[descriptor.name] = facade
assets[source_path] = facade
continue
body_type = "static"
if descriptor.physics is not None:
body_type = {
ActorType.DYNAMIC: "dynamic",
ActorType.KINEMATIC: "kinematic",
ActorType.STATIC: "static",
}[descriptor.physics.actor_type]
cfg = RigidObjectCfg(
uid=descriptor.name,
source=UsdSourceCfg(
file_path=file_path,
source_prim_path=(
descriptor.usd.prim_path
if descriptor.usd is not None
else None
),
),
init_local_pose=descriptor.pose.copy(),
body_type=body_type,
body_scale=tuple(float(value) for value in descriptor.body_scale),
asset_physics_mode="preserve",
)
facade = RigidObject(
cfg=cfg,
device=self.device,
)
self._spawn_scene.track(
"rigid_object",
descriptor.name,
descriptor,
facade=facade,
)
self._rigid_objects[descriptor.name] = facade
assets[source_path] = facade
continue
if isinstance(descriptor, ArticulationDesc):
robot_cfg = robot_cfgs.get(source_path)
facade_type: type[Articulation] = (
Robot if robot_cfg is not None else Articulation
)
cfg = (
deepcopy(robot_cfg)
if robot_cfg is not None
else ArticulationCfg(uid=descriptor.name)
)
cfg.uid = descriptor.name
cfg.fpath = None
cfg.source = UsdSourceCfg(
file_path=file_path,
source_prim_path=(
descriptor.usd.prim_path if descriptor.usd is not None else None
),
)
cfg.init_local_pose = descriptor.pose.copy()
cfg.asset_physics_mode = "preserve"
if robot_cfg is None:
cfg.root_props = ArticulationRootPropertiesCfg()
else:
cfg.root_props = cfg.root_props.copy()
cfg.root_props.fixed_base = bool(descriptor.fixed_base)
cfg.root_props.self_collision_enabled = descriptor.enable_self_collision
cfg.body_scale = tuple(float(value) for value in descriptor.body_scale)
cfg.build_pk_chain = False
facade = facade_type(
cfg=cfg,
device=self.device,
)
self._spawn_scene.track(
"articulation",
descriptor.name,
descriptor,
facade=facade,
)
registry = (
self._robots if robot_cfg is not None else self._articulations
)
registry[descriptor.name] = facade
assets[source_path] = facade
self.notify_visualization_topology_changed()
return assets
def add_visual_object(self, cfg: VisualObjectCfg) -> VisualObject:
"""Declare a render-only mesh or primitive, without physics or colliders.
Args:
cfg: Appearance and initial state. ``source=UsdSourceCfg(...)``
selects a source USD object; only its appearance is imported.
Returns:
The declared facade, bound after ``prepare()``.
"""
if cfg.uid is None or cfg.uid in self.asset_uids:
raise ValueError("Visual object requires a unique, nonempty UID.")
path = _asset_file_path(cfg)
if _is_usd_path(path):
descriptor, materials = visual_desc_from_usd(cfg)
else:
descriptor, materials = visual_desc_from_cfg(cfg)
visual = VisualObject(cfg, self.device)
was_materialized = self.spawn_result is not None
self._spawn_scene.builder.materials.update(
self._spawn_scene._resolve_materials(materials)
)
self._spawn_scene.declare("visual_object", cfg.uid, descriptor, facade=visual)
self._visual_objects[cfg.uid] = visual
self.notify_visualization_topology_changed()
if was_materialized:
self.prepare()
return visual
[docs]
def get_visual_object(self, uid: str) -> VisualObject | None:
"""Find a registered render-only object.
Args:
uid: Logical asset identifier.
Returns:
The visual object, or ``None`` if absent.
"""
return getattr(self, "_visual_objects", {}).get(uid)
[docs]
def get_visual_object_uid_list(self) -> list[str]:
"""Return the registered render-only object identifiers.
Returns:
Identifiers in declaration order.
"""
return list(getattr(self, "_visual_objects", {}))
[docs]
def add_rigid_object(
self,
cfg: RigidObjectCfg,
) -> RigidObject:
"""Add a rigid object to the scene.
Args:
cfg (RigidObjectCfg): Configuration for the rigid object.
Returns:
RigidObject: The added rigid object instance handle.
"""
return self._declare_rigid_object(cfg, replace=False)
[docs]
def replace_rigid_object(self, cfg: RigidObjectCfg) -> RigidObject:
"""Replace a rigid object with the same UID in one preparation.
The replacement uses its configured initial state and physical
properties. Previously returned handles for this object must not be
reused. Other scene objects retain their runtime state.
Args:
cfg: New configuration whose UID identifies an existing rigid object.
Returns:
The replacement object, ready for state access if the scene was
already materialized.
Raises:
KeyError: If the UID does not identify an existing rigid object.
Note:
Replacement applies to all instances of the logical object. A
backend preparation failure is propagated; this operation does
not provide rollback of native resources.
"""
return self._declare_rigid_object(cfg, replace=True)
[docs]
def replace_rigid_objects(
self, cfgs: Sequence[RigidObjectCfg]
) -> list[RigidObject]:
"""Replace multiple existing rigid objects in one preparation.
All UIDs and descriptors are checked before scene mutation. Returned
facades follow input order and are ready after the single preparation
when the scene was materialized. Run events needing the new objects
after this call. Old replaced handles must be discarded; other objects
retain their state. Native preparation failures do not roll back.
Args:
cfgs: Replacement configurations with distinct, existing UIDs.
Returns:
Replacement objects in input order, or an empty list for no input.
Raises:
ValueError: A UID is missing or duplicated, or translation fails.
KeyError: A UID does not identify an existing rigid object.
"""
return self._declare_rigid_objects(cfgs, replace=True)
def _declare_rigid_object(
self, cfg: RigidObjectCfg, *, replace: bool
) -> RigidObject:
return self._declare_rigid_objects([cfg], replace=replace)[0]
def _declare_rigid_objects(
self, cfgs: Sequence[RigidObjectCfg], *, replace: bool
) -> list[RigidObject]:
"""Translate the complete batch before changing the scene."""
cfgs = list(cfgs)
if not cfgs:
return []
seen: set[str] = set()
for cfg in cfgs:
uid = cfg.uid
if uid is None:
raise ValueError("Rigid object uid must be specified.")
if uid in seen:
raise ValueError(f"Duplicate rigid object uid {uid!r}.")
seen.add(uid)
if replace and uid not in self._rigid_objects:
raise KeyError(f"Rigid object {uid!r} does not exist.")
if not replace and uid in self._rigid_objects:
raise ValueError(f"Rigid object {uid!r} already exists.")
declarations = []
for cfg in cfgs:
source_path = _asset_file_path(cfg)
if _is_usd_path(source_path):
descriptor, materials = rigid_desc_from_usd(
cfg,
per_env=True,
newton_solver_type=self._active_newton_solver_type,
)
else:
descriptor, materials = rigid_desc_from_cfg(
cfg,
per_env=True,
newton_solver_type=self._active_newton_solver_type,
)
rigid_obj = RigidObject(cfg=cfg, device=self.device)
declarations.append((cfg.uid, descriptor, materials, rigid_obj))
was_materialized = self.spawn_result is not None
for uid, descriptor, materials, rigid_obj in declarations:
self._spawn_scene.builder.materials.update(
self._spawn_scene._resolve_materials(materials)
)
if replace:
self._spawn_scene.remove(uid)
self._spawn_scene.declare("rigid_object", uid, descriptor, facade=rigid_obj)
self._rigid_objects[uid] = rigid_obj
self.notify_visualization_topology_changed()
if was_materialized:
self.prepare()
return [item[3] for item in declarations]
[docs]
def get_rigid_object(self, uid: str) -> RigidObject | None:
"""Get a rigid object by its unique ID.
Args:
uid (str): The unique ID of the rigid object.
Returns:
RigidObject | None: The rigid object instance if found, otherwise None.
"""
if uid not in self._rigid_objects:
logger.log_warning(f"Rigid object {uid} not found.")
return None
return self._rigid_objects[uid]
[docs]
def get_rigid_object_uid_list(self) -> List[str]:
"""Get current rigid body uid list
Returns:
List[str]: list of rigid body uid.
"""
return list(self._rigid_objects.keys())
@staticmethod
def _broadcast_frame(
frame: np.ndarray | None,
num_envs: int,
env_ids: Sequence[int],
name: str,
) -> list[np.ndarray]:
"""Broadcast a local constraint frame to the selected environments."""
if frame is None:
identity = np.eye(4, dtype=np.float32)
return [identity for _ in env_ids]
frame_np = np.asarray(frame, dtype=np.float32)
if frame_np.shape == (4, 4):
return [frame_np for _ in env_ids]
if frame_np.ndim == 3 and frame_np.shape[1:] == (4, 4):
if frame_np.shape[0] != num_envs:
logger.log_error(
f"Constraint '{name}' local frame has shape {frame_np.shape} "
f"but num_envs is {num_envs}. Expected ({num_envs}, 4, 4)."
)
return [frame_np[i] for i in env_ids]
logger.log_error(
f"Constraint '{name}' local frame has invalid shape {frame_np.shape}. "
"Expected None, (4, 4), or (N, 4, 4)."
)
@staticmethod
def _normalize_env_ids(
env_ids: Sequence[int] | torch.Tensor | None,
num_envs: int,
) -> list[int]:
"""Normalize an ``env_ids`` spec to a plain ``list[int]``.
Accepts ``None`` (-> all envs), a ``torch.Tensor`` (as passed by the
:class:`EventManager`), or any ``Sequence[int]``, and returns a list of
Python ints. Normalizing here keeps the per-arena constraint names clean
(e.g. ``"weld_0"`` rather than relying on a tensor's string form) and
avoids depending on implicit tensor-to-int conversions downstream.
Args:
env_ids: None, a tensor, or a sequence of ints.
num_envs: Total number of arenas (used when env_ids is None).
Returns:
A list of int env indices.
"""
if env_ids is None:
return list(range(num_envs))
if isinstance(env_ids, torch.Tensor):
return env_ids.detach().cpu().tolist()
return [int(i) for i in env_ids]
[docs]
def create_rigid_constraint(
self,
cfg: RigidConstraintCfg,
env_ids: Sequence[int] | torch.Tensor | None = None,
) -> RigidConstraint:
"""Create a fixed constraint between two rigid objects.
Constraints are native Default-backend resources owned by each Arena.
Spawn owns the two actors; this method only borrows their native actor
handles while creating the constraint.
Args:
cfg: The constraint configuration.
env_ids: Target environment indices. Accepts a tensor (as passed by
the :class:`EventManager`) or a sequence of ints. None -> all arenas.
Returns:
The created constraint batch.
"""
if hasattr(self, "physics") and not self.physics.supports_rigid_constraints:
raise NotImplementedError(
f"The {self.physics.name} backend does not support rigid constraints."
)
if cfg.constraint_type != "fixed":
logger.log_error(
f"Constraint '{cfg.name}' has unsupported type "
f"'{cfg.constraint_type}'. Only 'fixed' is supported."
)
if cfg.rigid_object_a_uid not in self._rigid_objects:
logger.log_error(
f"RigidObject '{cfg.rigid_object_a_uid}' not found for constraint "
f"'{cfg.name}'. Available: {list(self._rigid_objects.keys())}."
)
if cfg.rigid_object_b_uid not in self._rigid_objects:
logger.log_error(
f"RigidObject '{cfg.rigid_object_b_uid}' not found for constraint "
f"'{cfg.name}'. Available: {list(self._rigid_objects.keys())}."
)
if cfg.name in self._constraints:
logger.log_error(
f"Constraint '{cfg.name}' already exists. Remove it before recreating."
)
rigid_object_a = self._rigid_objects[cfg.rigid_object_a_uid]
rigid_object_b = self._rigid_objects[cfg.rigid_object_b_uid]
if hasattr(self, "_spawn_scene"):
self.prepare()
num_envs = self.num_envs
if rigid_object_a.num_instances != num_envs:
logger.log_error(
f"RigidObject '{cfg.rigid_object_a_uid}' has "
f"{rigid_object_a.num_instances} instances but num_envs is {num_envs}."
)
if rigid_object_b.num_instances != num_envs:
logger.log_error(
f"RigidObject '{cfg.rigid_object_b_uid}' has "
f"{rigid_object_b.num_instances} instances but num_envs is {num_envs}."
)
target_env_ids = self._normalize_env_ids(env_ids, num_envs)
frames_a = self._broadcast_frame(
cfg.local_frame_a, num_envs, target_env_ids, cfg.name
)
if cfg.local_frame_b is None:
pose_a = rigid_object_a.get_local_pose(to_matrix=True)
pose_b = rigid_object_b.get_local_pose(to_matrix=True)
frame_b = (
torch.bmm(pose_inv(pose_b), pose_a).cpu().numpy().astype(np.float32)
)
frames_b = [frame_b[i] for i in target_env_ids]
else:
frames_b = self._broadcast_frame(
cfg.local_frame_b, num_envs, target_env_ids, cfg.name
)
handles: list = [None] * num_envs
try:
for index, env_id in enumerate(target_env_ids):
actor_a = rigid_object_a._entities[env_id]
actor_b = rigid_object_b._entities[env_id]
if getattr(rigid_object_a, "is_spawn_bound", False) is True:
actor_a = actor_a.native()
if getattr(rigid_object_b, "is_spawn_bound", False) is True:
actor_b = actor_b.native()
if actor_a is None or actor_b is None:
logger.log_error(
f"Constraint '{cfg.name}' references a released Spawn actor "
f"in environment {env_id}."
)
arena = self.get_env(env_id)
name = cfg.name if num_envs <= 1 else f"{cfg.name}_{env_id}"
handle = arena.create_fixed_constraint(
name,
actor_a,
actor_b,
frames_a[index],
frames_b[index],
)
if handle is None:
logger.log_error(
f"Failed to create constraint '{name}' in arena {env_id}."
)
handles[env_id] = handle
except Exception:
RigidConstraint(
cfg=cfg,
constraint_handles=handles,
rigid_object_a=rigid_object_a,
rigid_object_b=rigid_object_b,
device=self.device,
).destroy(env_ids=target_env_ids, arena_resolver=self.get_env)
raise
constraint = RigidConstraint(
cfg=cfg,
constraint_handles=handles,
rigid_object_a=rigid_object_a,
rigid_object_b=rigid_object_b,
device=self.device,
)
self._constraints[cfg.name] = constraint
return constraint
[docs]
def remove_rigid_constraint(
self,
name: str,
env_ids: Sequence[int] | torch.Tensor | None = None,
) -> bool:
"""Remove a rigid constraint by name.
With ``env_ids=None`` the constraint is removed from every arena and
dropped from the registry. With a subset, only those arenas are cleared;
the registry entry is kept until all handles become None.
Args:
name: The base constraint name.
env_ids: Subset of arenas to clear. Accepts a tensor (as passed by
the :class:`EventManager`) or a sequence of ints. None -> all.
Returns:
True if the constraint was found (and removed or partially removed),
False if the name is unknown.
"""
constraint = self._constraints.get(name, None)
if constraint is None:
logger.log_warning(f"Constraint '{name}' not found. Nothing to remove.")
return False
target_env_ids = self._normalize_env_ids(env_ids, constraint.num_envs)
constraint.destroy(env_ids=target_env_ids, arena_resolver=self.get_env)
# drop from registry if no handles remain active
if all(h is None for h in constraint.constraint_handles):
del self._constraints[name]
return True
[docs]
def get_rigid_constraint(self, name: str) -> RigidConstraint | None:
"""Get a rigid constraint by its base name.
Args:
name: The base constraint name.
Returns:
The constraint, or None if not found.
"""
if name not in self._constraints:
logger.log_warning(f"Constraint '{name}' not found.")
return None
return self._constraints[name]
[docs]
def get_rigid_constraint_uid_list(self) -> List[str]:
"""Get the list of registered constraint base names.
Returns:
List[str]: list of constraint names.
"""
return list(self._constraints.keys())
[docs]
def add_rigid_object_group(self, cfg: RigidObjectGroupCfg) -> RigidObjectGroup:
"""Add a rigid object group to the scene.
Args:
cfg (RigidObjectGroupCfg): Configuration for the rigid object group.
Returns:
The stable Group facade. During initial scene construction it is
bound to Spawn handles by :meth:`prepare`.
"""
if not self.physics.supports_rigid_object_group:
raise NotImplementedError(
f"The {self.physics.name} backend does not support rigid object groups."
)
uid = cfg.uid
if uid is None:
raise ValueError("Rigid object group uid must be specified.")
if uid in self._rigid_object_groups:
raise ValueError(f"Rigid object group {uid!r} already exists.")
if cfg.body_type == "static":
raise ValueError("Rigid object group cannot be static.")
if not cfg.rigid_objects:
raise ValueError("Rigid object group must contain at least one object.")
actor_type = {
"dynamic": ActorType.DYNAMIC,
"kinematic": ActorType.KINEMATIC,
}[cfg.body_type]
descriptors = []
for index, member in enumerate(cfg.rigid_objects.values()):
member_cfg = deepcopy(member)
member_cfg.uid = f"{uid}__member_{index}"
member_cfg.body_type = cfg.body_type
source_path = _asset_file_path(member_cfg)
if _is_usd_path(source_path):
descriptor, materials = rigid_desc_from_usd(
member_cfg,
per_env=True,
newton_solver_type=self._active_newton_solver_type,
)
else:
descriptor, materials = rigid_desc_from_cfg(
member_cfg,
per_env=True,
newton_solver_type=self._active_newton_solver_type,
)
if descriptor.physics is None:
raise ValueError(
f"Rigid object group member {index} has no rigid-body physics."
)
descriptor.physics.actor_type = actor_type
self._spawn_scene.builder.materials.update(
self._spawn_scene._resolve_materials(materials)
)
descriptors.append(descriptor)
group = RigidObjectGroup(
cfg,
device=self.device,
)
was_materialized = self.spawn_result is not None
self._spawn_scene.declare(
"rigid_object_group",
uid,
tuple(descriptors),
facade=group,
)
self._rigid_object_groups[uid] = group
self.notify_visualization_topology_changed()
if was_materialized:
self.prepare()
return group
[docs]
def get_rigid_object_group(self, uid: str) -> RigidObjectGroup | None:
"""Get a rigid object group by its unique ID.
Args:
uid (str): The unique ID of the rigid object group.
Returns:
RigidObjectGroup | None: The rigid object group instance if found, otherwise None.
"""
if uid not in self._rigid_object_groups:
logger.log_warning(f"Rigid object group {uid} not found.")
return None
return self._rigid_object_groups[uid]
[docs]
def get_rigid_object_group_uid_list(self) -> List[str]:
"""Get current rigid body group uid list
Returns:
List[str]: list of rigid body group uid.
"""
return list(self._rigid_object_groups.keys())
@cached_property
def arena_offsets(self) -> torch.Tensor:
"""Get the arena offsets for all arenas.
Returns:
torch.Tensor: The arena offsets of shape (num_arenas, 3).
"""
env_list = [self._env] if len(self._arenas) == 0 else self._arenas
arena_offsets = torch.zeros(
(len(env_list), 3), dtype=torch.float32, device=self.device
)
for i, env in enumerate(env_list):
arena_position = env.get_root_node().get_world_pose()[:3, 3]
arena_offsets[i] = torch.tensor(
arena_position, dtype=torch.float32, device=self.device
)
return arena_offsets
[docs]
def has_non_static_rigid_object(self) -> bool:
"""Check if there is any non-static rigid object in the simulation.
Returns:
bool: True if there is at least one non-static rigid object, False otherwise.
"""
for rigid_obj in self._rigid_objects.values():
if rigid_obj.body_type != "static":
return True
if len(self._rigid_object_groups) > 0:
return True
return False
[docs]
def add_articulation(
self,
cfg: ArticulationCfg,
) -> Articulation:
"""Add an articulation to the scene.
Args:
cfg (ArticulationCfg): Configuration for the articulation.
Returns:
Articulation: The added articulation instance handle.
"""
uid = cfg.uid
if uid is None:
source_path = _asset_file_path(cfg)
if source_path is None:
raise ValueError(
"Articulation configuration must provide fpath when uid "
"is not specified."
)
uid = os.path.splitext(os.path.basename(source_path))[0]
cfg.uid = uid
if uid in self._articulations:
raise ValueError(f"Articulation {uid!r} already exists.")
was_materialized = self.spawn_result is not None
articulation = self._declare_spawn_articulation(cfg, Articulation)
self._articulations[uid] = articulation
if was_materialized:
self.prepare()
return articulation
[docs]
def get_articulation(self, uid: str) -> Articulation | None:
"""Get an articulation by its unique ID.
Args:
uid (str): The unique ID of the articulation.
Returns:
Articulation | None: The articulation instance if found, otherwise None.
"""
if uid not in self._articulations:
logger.log_warning(f"Articulation {uid} not found.")
return None
return self._articulations[uid]
[docs]
def get_articulation_uid_list(self) -> List[str]:
"""Get current articulation uid list
Returns:
List[str]: list of articulation uid.
"""
return list(self._articulations.keys())
[docs]
def add_robot(self, cfg: RobotCfg | RobotPresetCfg) -> Robot | None:
"""Add a Robot to the scene.
Args:
cfg: A concrete robot configuration or a replace-only backend
preset. Presets are resolved from ``physics_cfg`` before the
robot is declared. Mounted cameras are created only when
``cfg.enable_sensor`` is True and attach during preparation.
Returns:
Robot | None: The added robot instance handle, or None if failed.
"""
if not self.physics.supports_robot:
logger.log_error(
f"Robot support is not enabled for the "
f"{self.physics.name} backend yet.",
error_type=NotImplementedError,
)
if isinstance(cfg, RobotPresetCfg):
cfg = cfg.resolve(
self.sim_config.physics_cfg,
newton_solver_type=self._active_newton_solver_type,
)
uid = cfg.uid
source_path = _asset_file_path(cfg)
if source_path is None:
if cfg.urdf_cfg is None:
logger.log_error(
"Robot configuration must have a valid fpath or urdf_cfg."
)
return None
cfg.fpath = cfg.urdf_cfg.assemble_urdf()
source_path = cfg.fpath
if cfg.solver_cfg is not None:
if isinstance(cfg.solver_cfg, dict):
for key, value in cfg.solver_cfg.items():
if hasattr(value, "urdf_path") and value.urdf_path is None:
value.urdf_path = cfg.fpath
if uid is None:
uid = os.path.splitext(os.path.basename(source_path))[0]
cfg.uid = uid
if uid in self._robots:
logger.log_error(f"Robot {uid} already exists.")
return self._robots[uid]
mounted_sensors = []
if cfg.enable_sensor:
sensor_uids = set(self._sensors)
for sensor_cfg in cfg.sensors:
if not isinstance(sensor_cfg, CameraCfg):
raise TypeError("RobotCfg.sensors supports mounted cameras only.")
if not isinstance(sensor_cfg.uid, str) or not sensor_cfg.uid.strip():
raise ValueError("Mounted sensors require a non-empty uid.")
if sensor_cfg.uid in sensor_uids:
raise ValueError(f"Sensor {sensor_cfg.uid!r} already exists.")
sensor_uids.add(sensor_cfg.uid)
parent = sensor_cfg.extrinsics.parent
if not isinstance(parent, str) or not parent.strip():
raise ValueError("Mounted sensors require a parent link.")
link_name = parent.removeprefix(f"{uid}/")
if not link_name or "/" in link_name:
raise ValueError(
f"Mounted sensor {sensor_cfg.uid!r} must refer to a "
f"local link on robot {uid!r}."
)
bound_cfg = deepcopy(sensor_cfg)
bound_cfg.extrinsics.parent = f"{uid}/{link_name}"
mounted_sensors.append(bound_cfg)
was_materialized = self.spawn_result is not None
robot = self._declare_spawn_articulation(cfg, Robot)
self._robots[uid] = robot
if was_materialized:
self.prepare()
for sensor_cfg in mounted_sensors:
self.add_sensor(sensor_cfg)
return robot
def _declare_spawn_articulation(
self,
cfg: ArticulationCfg,
facade_type: type[Articulation],
) -> Articulation:
"""Declare an articulation facade and bind its Batch after finalize.
DexSim remains the sole articulation source loader. EmbodiChain applies
regex/group configuration to the resolved descriptor before either
backend materializes it. Runtime Batch data is created at the shared
prepare boundary.
"""
if _is_usd_path(_asset_file_path(cfg)):
descriptor, materials = articulation_desc_from_usd(
cfg,
per_env=True,
newton_solver_type=self._active_newton_solver_type,
)
self._spawn_scene.builder.materials.update(
self._spawn_scene._resolve_materials(materials)
)
else:
descriptor = articulation_desc_from_cfg(
cfg,
per_env=True,
newton_solver_type=self._active_newton_solver_type,
)
if cfg.uid is None:
cfg.uid = descriptor.name
facade = facade_type(
cfg=cfg,
device=self.device,
)
self._spawn_scene.declare(
"articulation",
descriptor.name,
descriptor,
facade=facade,
configure_source=partial(
configure_articulation_desc,
cfg=cfg,
newton_solver_type=self._active_newton_solver_type,
),
)
self.notify_visualization_topology_changed()
return facade
[docs]
def get_robot(self, uid: str) -> Robot | None:
"""Get a Robot by its unique ID.
Args:
uid (str): The unique ID of the robot.
Returns:
Robot | None: The robot instance if found, otherwise None.
"""
if uid not in self._robots:
logger.log_warning(f"Robot {uid} not found.")
return None
return self._robots[uid]
[docs]
def get_robot_uid_list(self) -> List[str]:
"""
Retrieves a list of unique identifiers (UIDs) for all robots in the simulation system.
Returns:
list: A list containing the UIDs of the robots.
"""
return list(self._robots.keys())
[docs]
def enable_entity_gizmo(
self,
config: EntityGizmoConfig | None = None,
) -> EntityGizmoManipulator:
"""Enable DexSim entity control and exclude the EmbodiChain ground.
Called automatically on the first native window unless disabled in
:class:`SimulationManagerCfg`. Explicit calls also work headlessly.
DexSim retains this controller and its configuration across window
close/reopen; reopening does not reapply the default configuration.
Args:
config: Native DexSim entity-Gizmo configuration.
Returns:
The active world-owned entity Gizmo manipulator.
"""
controller = (
self._world.enable_entity_gizmo()
if config is None
else self._world.enable_entity_gizmo(config)
)
self._auto_entity_gizmo_pending = False
default_plane = getattr(self, "_default_plane", None)
if default_plane is None:
return controller
result = controller.register_external_target(
self._DEFAULT_PLANE_GIZMO_TARGET_ID,
dexsim.interaction.EntityGizmoTargetType.RIGID_BODY,
default_plane.native(),
ActorType.STATIC,
)
if result != dexsim.interaction.EntityGizmoResult.SUCCESS:
logger.log_warning(
"Failed to exclude the default plane from entity Gizmo "
f"control: {result}."
)
return controller
[docs]
def disable_entity_gizmo(self) -> None:
"""Disable native object interaction, including future window opens.
Call :meth:`enable_entity_gizmo` to explicitly re-enable interaction.
Robot IK controllers and Viser command permissions are independent.
"""
self._world.disable_entity_gizmo()
self._auto_entity_gizmo_pending = False
def _enable_default_entity_gizmo(self) -> None:
# Apply the startup default once; DexSim owns subsequent controller
# state, including explicit native disable calls and window reopens.
if self._window is not None and self._auto_entity_gizmo_pending:
self.enable_entity_gizmo()
[docs]
def enable_gizmo(
self,
uid: str,
control_part: str | None = None,
gizmo_cfg: GizmoCfg | None = None,
) -> Gizmo | None:
"""Register Gizmo control for a simulation target.
Robot controls support native windows and Viser; native IK activates
on the configured toggle key. Explicit configuration takes precedence
over automatically registered controls.
Args:
uid: UID of the robot, rigid object, or camera sensor.
control_part: Robot control part used for IK/FK.
gizmo_cfg: Gizmo appearance and robot IK configuration.
Returns:
The created Gizmo, or ``None`` if setup failed.
"""
# Create gizmo key combining uid and control_part
gizmo_key = f"{uid}:{control_part}" if control_part else uid
if hasattr(self, "_disabled_robot_gizmos"):
self._disabled_robot_gizmos.discard(gizmo_key)
# Check if gizmo already exists
if gizmo_key in self._gizmos:
if gizmo_cfg is not None:
self.disable_gizmo(uid, control_part)
if hasattr(self, "_disabled_robot_gizmos"):
self._disabled_robot_gizmos.discard(gizmo_key)
else:
return self._gizmos[gizmo_key]
# Search for target object in different collections
target = None
object_type = None
if uid in self._robots:
target = self._robots[uid]
object_type = "robot"
elif uid in self._rigid_objects:
target = self._rigid_objects[uid]
object_type = "rigid_object"
elif uid in self._sensors:
target = self._sensors[uid]
object_type = "sensor"
else:
logger.log_error(
f"Object with uid '{uid}' not found in any collection (robots, rigid_objects, sensors, articulations)."
)
return None
gizmo: Gizmo | None = None
try:
gizmo = Gizmo(target, gizmo_cfg, control_part)
self._gizmos[gizmo_key] = gizmo
self.notify_visualization_topology_changed()
logger.log_info(
f"Gizmo registered for {object_type} '{uid}' with "
f"control_part '{control_part}'."
)
except Exception as e:
if gizmo is not None:
gizmo.destroy()
logger.log_error(
f"Failed to create gizmo for {object_type} '{uid}' with control_part '{control_part}': {e}"
)
return None
return gizmo
[docs]
def disable_gizmo(self, uid: str, control_part: str | None = None) -> None:
"""Disable a Gizmo and prevent its automatic recreation.
Args:
uid: Target asset UID.
control_part: Robot control part. Omit to disable every part of a robot.
"""
gizmo_key = f"{uid}:{control_part}" if control_part else uid
if hasattr(self, "_disabled_robot_gizmos"):
self._disabled_robot_gizmos.add(gizmo_key)
robot = getattr(self, "_robots", {}).get(uid)
if robot is not None and control_part is None:
for key, gizmo in self.get_gizmo_items():
if key != gizmo_key and gizmo.target is robot:
self.disable_gizmo(key)
if gizmo_key not in self._gizmos:
if robot is not None:
return
logger.log_warning(
f"No gizmo found for '{uid}' with control_part '{control_part}'."
)
return
try:
gizmo = self._gizmos.pop(gizmo_key)
if self._picker_gizmo == (uid, control_part):
self._picker_gizmo = None
try:
if gizmo is not None:
gizmo.destroy()
finally:
self.notify_visualization_topology_changed()
logger.log_info(
f"Gizmo disabled for '{uid}' with control_part '{control_part}'"
)
except Exception as error:
logger.log_error(
f"Failed to disable gizmo for '{uid}' with control_part "
f"'{control_part}': {error}"
)
[docs]
def get_gizmo(
self,
uid: str,
control_part: str | None = None,
) -> Gizmo | None:
"""Return an active Gizmo.
Args:
uid: Target asset UID.
control_part: Robot control part, if applicable.
Returns:
Gizmo instance if found, otherwise ``None``.
"""
gizmo_key = f"{uid}:{control_part}" if control_part else uid
return self._gizmos.get(gizmo_key, None)
[docs]
def has_gizmo(self, uid: str, control_part: str | None = None) -> bool:
"""Check if a gizmo exists for the given UID and control part.
Args:
uid (str): Object UID to check
control_part (str | None, optional): Control part name for robots. Defaults to None.
Returns:
bool: True if gizmo exists, False otherwise.
"""
gizmo_key = f"{uid}:{control_part}" if control_part else uid
return gizmo_key in self._gizmos
[docs]
def list_gizmos(self) -> dict[str, bool]:
"""List active Gizmo IDs and availability.
Returns:
Mapping from ``uid[:control_part]`` to availability.
"""
return {
gizmo_key: (gizmo is not None) for gizmo_key, gizmo in self._gizmos.items()
}
[docs]
def get_gizmo_items(self) -> tuple[tuple[str, Gizmo], ...]:
"""Return a stable snapshot of active Gizmo IDs and controllers."""
return tuple(
(gizmo_key, gizmo)
for gizmo_key, gizmo in getattr(self, "_gizmos", {}).items()
if gizmo is not None
)
[docs]
def process_visualization_commands(self) -> int:
"""Apply queued Viser Gizmo commands on the simulation thread.
Returns:
Number of commands accepted for active Gizmos.
"""
runtime = self._visualization_runtime
if runtime is None or not getattr(
self.sim_config.visualization,
"allow_commands",
False,
):
return 0
accepted = 0
for command in runtime.drain_gizmo_commands():
if (
command.run_id != runtime.exporter.run_id
or command.scene_revision != runtime.exporter.scene_revision
):
continue
gizmo = self._gizmos.get(command.gizmo_id)
if gizmo is None:
continue
source_id = f"viser:{command.client_id}"
if command.phase in {"start", "update"} and not gizmo.begin_interaction(
source_id
):
continue
position = torch.as_tensor(
command.position,
dtype=torch.float32,
device=self.device,
)
position = position - self.arena_offsets[0]
xyzw = quat_wxyz_to_xyzw(
torch.as_tensor(
command.wxyz,
dtype=torch.float32,
device=self.device,
)
).unsqueeze(0)
pose = torch.eye(
4,
dtype=torch.float32,
device=self.device,
).unsqueeze(0)
pose[0, :3, :3] = matrix_from_quat(xyzw)[0]
pose[0, :3, 3] = position
if not gizmo.request_local_pose(pose, source_id=source_id):
continue
accepted += 1
if command.phase == "end":
gizmo.end_interaction(source_id)
return accepted
def _register_robot_gizmos(self) -> None:
"""Discover IK-capable parts without constructing another solver."""
cfg = getattr(self.sim_config, "robot_ik_gizmo", None)
if cfg is None or self.num_envs != 1:
return
native = self._window is not None
viser = (
self.sim_config.visualization.backend == "viser"
and self.sim_config.visualization.allow_commands
)
if not native and not viser:
return
for uid, robot in self._robots.items():
solver_cfgs = robot.cfg.solver_cfg
if not isinstance(solver_cfgs, dict):
continue
for part, solver_cfg in solver_cfgs.items():
key = f"{uid}:{part}"
if (
uid in self._disabled_robot_gizmos
or key in self._disabled_robot_gizmos
or key in self._gizmos
or part not in (robot.control_parts or {})
or not getattr(solver_cfg, "root_link_name", None)
or not getattr(solver_cfg, "end_link_name", None)
):
continue
if any(
gizmo.target is robot and gizmo.control_part == part
for _, gizmo in self.get_gizmo_items()
):
continue
try:
self.enable_gizmo(uid, part, deepcopy(cfg))
except Exception as error:
self._disabled_robot_gizmos.add(key)
logger.log_warning(
f"Could not register robot Gizmo '{key}': {error}"
)
[docs]
def update_gizmos(self) -> None:
"""Register robot controls, process interaction, and update active Gizmos."""
self._register_robot_gizmos()
self.process_pick_commands()
self.process_visualization_commands()
for gizmo_key, gizmo in list(
getattr(self, "_gizmos", {}).items()
): # Use list() to avoid modification during iteration
if gizmo is not None:
try:
if getattr(self, "_window", None) is not None:
gizmo.update_native(self._world)
gizmo.update()
except Exception as error:
logger.log_warning(f"Error updating gizmo '{gizmo_key}': {error}")
[docs]
def process_pick_commands(self) -> int:
"""Apply queued Viser click-pick commands on the simulation thread.
A non-empty pick attaches a picker-owned Gizmo to the clicked node;
an empty pick (clicking empty space) clears it. Only one picker-owned
Gizmo is kept at a time and user-created Gizmos are never touched.
Returns:
Number of pick commands drained from the visualization runtime.
"""
runtime = self._visualization_runtime
if runtime is None or not getattr(
self.sim_config.visualization, "allow_commands", False
):
return 0
processed = 0
for command in runtime.drain_pick_commands():
processed += 1
if (
command.run_id != runtime.exporter.run_id
or command.scene_revision != runtime.exporter.scene_revision
):
continue
if command.node_id is None:
# Clicking empty space detaches the picker-owned Gizmo.
self._release_picker_gizmo()
continue
resolved = runtime.exporter.resolve_node_target(command.node_id)
if resolved is None:
continue
uid, kind = resolved
if kind not in {"robot", "rigid"}:
logger.log_warning(
f"Pick target kind {kind!r} (uid {uid!r}) is not gizmo-able; "
"only rigid objects and robots can be picked."
)
self._release_picker_gizmo()
continue
# Re-clicking the already-picked target is a no-op (avoids flicker
# from recreating the same Gizmo, e.g. clicking near its handle).
if self._picker_gizmo is not None and self._picker_gizmo[0] == uid:
continue
self._release_picker_gizmo()
if any(key == uid or key.startswith(f"{uid}:") for key in self._gizmos):
continue
gizmo = self.enable_gizmo(uid=uid)
if gizmo is not None:
self._picker_gizmo = (uid, None)
return processed
def _release_picker_gizmo(self) -> None:
"""Detach the picker-owned Gizmo if one is currently attached."""
if self._picker_gizmo is None:
return
uid, control_part = self._picker_gizmo
self._picker_gizmo = None
if self.has_gizmo(uid, control_part=control_part):
self.disable_gizmo(uid, control_part=control_part)
[docs]
def toggle_gizmo_visibility(
self, uid: str, control_part: str | None = None
) -> bool | None:
"""Toggle Gizmo visibility and return the new state, if it exists."""
gizmo = self.get_gizmo(uid, control_part)
if gizmo is not None:
return gizmo.toggle_visibility()
return None
[docs]
def set_gizmo_visibility(
self, uid: str, visible: bool, control_part: str | None = None
) -> None:
"""Set Gizmo visibility by target UID and optional control part."""
gizmo = self.get_gizmo(uid, control_part)
if gizmo is not None:
gizmo.set_visible(visible)
[docs]
def add_sensor(self, sensor_cfg: SensorCfg) -> BaseSensor:
"""Create a sensor on the pre-created simulation Arenas.
Cameras keep EmbodiChain's native CameraGroup implementation. A camera
attached to an articulation link is created immediately and attached
after the physical Spawn scene is prepared. Contact sensors are created
after preparation and query contacts through the backend-neutral Spawn
Scene API.
Args:
sensor_cfg (SensorCfg): configuration for the sensor.
Returns:
BaseSensor: The added sensor instance handle.
"""
sensor_type = sensor_cfg.sensor_type
uid = sensor_cfg.uid
if uid is None:
uid = f"{sensor_type.lower()}_{len(self._sensors)}"
sensor_cfg.uid = uid
if uid in self._sensors:
raise ValueError(f"Sensor {uid!r} already exists.")
sensor_factory = self.SUPPORTED_SENSOR_TYPES.get(sensor_type)
if sensor_factory is None:
raise ValueError(
f"Unsupported sensor type {sensor_type!r}. Supported types: "
f"{sorted(self.SUPPORTED_SENSOR_TYPES)}."
)
if sensor_type == "ContactSensor" and not self.physics.supports_contact_sensor:
raise NotImplementedError(
f"ContactSensor is not supported by the {self.physics.name} "
"physics backend."
)
if isinstance(sensor_factory, type) and issubclass(sensor_factory, Camera):
if len(self._arenas) != self.num_envs:
raise RuntimeError(
"Camera creation requires all Spawn Arenas to be "
f"prepared ({len(self._arenas)} of {self.num_envs} ready)."
)
parent_nodes = None
if (
sensor_cfg.extrinsics.parent is not None
and self._spawn_scene.builder.result is not None
):
# Resolve before allocating native camera views so an invalid
# parent cannot leave partially constructed render resources.
parent_nodes = self._resolve_spawn_sensor_parent_nodes(
sensor_cfg.extrinsics.parent
)
sensor = sensor_factory(
sensor_cfg,
self.device,
owner=self,
)
if parent_nodes is not None:
sensor.attach_to_parent_nodes(parent_nodes)
elif isinstance(sensor_factory, type) and issubclass(
sensor_factory, ContactSensor
):
self.prepare()
# ``solver_type='auto'`` is resolved while preparing the Spawn
# scene, so the early capability check above is not sufficient.
if not self.physics.supports_contact_sensor:
raise NotImplementedError(
f"ContactSensor is not supported by the {self.physics.name} "
"physics backend with the resolved solver/device."
)
sensor = sensor_factory(
sensor_cfg,
self.device,
owner=self,
)
else:
# Custom native sensors require a prepared physics scene; cameras
# only depend on the pre-created Arenas.
self.prepare()
# Preserve custom test/plugin factories whose two-argument
# constructor predates the manager-owned render context.
sensor = sensor_factory(sensor_cfg, self.device)
self._sensors[uid] = sensor
self.notify_visualization_topology_changed()
return sensor
def _attach_camera_parent(self, sensor: Camera) -> None:
"""Resolve and attach one camera to its configured parent nodes."""
parent = sensor.cfg.extrinsics.parent
if parent is None:
return
parent_nodes = self._resolve_spawn_sensor_parent_nodes(parent)
sensor.attach_to_parent_nodes(parent_nodes)
def _attach_parented_cameras(self) -> None:
"""Restore parented cameras after a Spawn topology change."""
for sensor in self._sensors.values():
if isinstance(sensor, Camera) and sensor.cfg.extrinsics.parent is not None:
self._attach_camera_parent(sensor)
def _resolve_spawn_sensor_parent_nodes(self, parent: str) -> list[object]:
"""Resolve one canonical articulation link to a render node per Arena.
A plain link name remains compatible with existing CameraCfg values.
When more than one robot/articulation owns that link, callers can use
``"<asset_uid>/<link_name>"`` to disambiguate without introducing
backend clone suffixes.
"""
assets: dict[str, Articulation] = {
**self._articulations,
**self._robots,
}
return resolve_parent_nodes(parent, assets, self.num_envs)
[docs]
def get_sensor(self, uid: str) -> BaseSensor | None:
"""Get a sensor by its UID.
Args:
uid (str): The UID of the sensor.
Returns:
BaseSensor | None: The sensor instance if found, otherwise None.
"""
if uid not in self._sensors:
logger.log_warning(f"Sensor {uid} not found.")
return None
return self._sensors[uid]
[docs]
def get_sensor_uid_list(self) -> List[str]:
"""Get current sensor uid list
Returns:
List[str]: list of sensor uid.
"""
return list(self._sensors.keys())
[docs]
def remove_asset(self, uid: str) -> bool:
"""Remove an asset by its UID.
Native render lights are not removed by this method. Sensors and
Spawn-owned physical assets are supported. Removing a robot also
destroys its automatically initialized mounted cameras.
Args:
uid (str): The UID of the asset.
Returns:
bool: True if the asset is removed successfully, otherwise False.
"""
if uid in self._sensors:
sensor = self._sensors.pop(uid)
destroy = getattr(sensor, "destroy", None)
if callable(destroy):
destroy()
self.notify_visualization_topology_changed()
return True
scene = self._spawn_scene
if uid not in scene:
return False
if uid == "default_plane":
raise ValueError("The Spawn-owned default plane cannot be removed.")
robot = self._robots.get(uid)
if robot is not None and robot.cfg.enable_sensor:
for sensor_cfg in robot.cfg.sensors:
sensor = self._sensors.get(sensor_cfg.uid)
if (
isinstance(sensor, Camera)
and sensor.cfg.extrinsics.parent is not None
and sensor.cfg.extrinsics.parent.startswith(f"{uid}/")
):
self.remove_asset(sensor_cfg.uid)
was_materialized = scene.builder.is_finalized
scene.remove(uid)
visual = getattr(self, "_visual_objects", {}).get(uid)
if visual is not None:
visual._release_spawn_references()
if was_materialized:
self.prepare()
if robot is not None:
for key, gizmo in self.get_gizmo_items():
if gizmo.target is robot:
self.disable_gizmo(key)
self._disabled_robot_gizmos = {
key
for key in self._disabled_robot_gizmos
if key != uid and not key.startswith(f"{uid}:")
}
getattr(self, "_visual_objects", {}).pop(uid, None)
self._rigid_objects.pop(uid, None)
self._rigid_object_groups.pop(uid, None)
self._deformable_objects.pop(uid, None)
self._articulations.pop(uid, None)
self._robots.pop(uid, None)
self.notify_visualization_topology_changed()
return True
[docs]
def draw_marker(
self,
cfg: MarkerCfg,
) -> MeshObject:
"""Draw visual markers in the simulation scene for debugging and visualization.
Drawing markers does not advance physics. Call
:meth:`capture_visualization` to publish them to Viser without stepping.
Args:
cfg (MarkerCfg): Marker configuration with the following key parameters:
- name (str): Unique identifier for the marker group
- marker_type (str): Type of marker ("axis" currently supported)
- axis_xpos (np.ndarray | List[np.ndarray]): 4x4 transformation matrices
for marker positions and orientations
- axis_size (float): Thickness of axis arrows
- axis_len (float): Length of axis arrows
- arena_index (int): Arena index for placement (-1 for global)
Returns:
List[MeshObject]: List of created marker handles, False if invalid input,
None if no poses provided.
Example:
```python
cfg = MarkerCfg(name="test_axis", marker_type="axis", axis_xpos=np.eye(4))
markers = sim.draw_marker(cfg)
```
"""
# Validate marker type
if cfg.marker_type != "axis":
logger.log_error(
f"Unsupported marker type '{cfg.marker_type}'. Currently only 'axis' is supported."
)
return False
draw_xpos = deepcopy(cfg.axis_xpos)
if isinstance(draw_xpos, torch.Tensor):
draw_xpos = draw_xpos.detach().cpu().numpy()
elif isinstance(draw_xpos, (list, tuple)):
draw_xpos = [
item.detach().cpu().numpy() if isinstance(item, torch.Tensor) else item
for item in draw_xpos
]
draw_xpos = np.array(draw_xpos)
if draw_xpos.ndim == 2:
if draw_xpos.shape == (4, 4):
draw_xpos = np.expand_dims(draw_xpos, axis=0)
else:
logger.log_error(
f"axis_xpos must be of shape (N, 4, 4), got {draw_xpos.shape}."
)
return False
elif draw_xpos.ndim != 3 or draw_xpos.shape[1:] != (4, 4):
logger.log_error(
f"axis_xpos must be of shape (N, 4, 4), got {draw_xpos.shape}."
)
return False
original_name = cfg.name
name = original_name
count = 0
while name in self._markers:
count += 1
name = f"{original_name}_{count}"
if count > 0:
logger.log_warning(
f"Marker name '{original_name}' already exists. Using '{name}'."
)
marker_num = len(draw_xpos)
if marker_num == 0:
logger.log_warning(f"No marker poses provided.")
return None
if cfg.arena_index >= 0:
name = f"{name}_{cfg.arena_index}"
env = self.get_env(cfg.arena_index)
# Create markers based on marker type
marker_handles = []
if cfg.marker_type == "axis":
# Create coordinate axes
axis_option = dexsim.types.AxisOption(
lx=cfg.axis_len,
ly=cfg.axis_len,
lz=cfg.axis_len,
size=cfg.axis_size,
arrow_type=cfg.arrow_type,
corner_type=cfg.corner_type,
tag_type=dexsim.types.AxisTagType.NONE,
)
for i, pose in enumerate(draw_xpos):
axis_handle = env.create_axis(axis_option)
axis_handle.set_local_pose(pose)
marker_handles.append(axis_handle)
# TODO: Add support for other marker types in the future
# elif cfg.marker_type == "line":
# # Create line markers
# pass
# elif cfg.marker_type == "point":
# # Create point markers
# pass
self._markers[name] = _AxisMarkerGroup(
handles=tuple(marker_handles),
arena_index=cfg.arena_index,
axis_length=cfg.axis_len,
axis_radius=cfg.axis_size,
)
return marker_handles
[docs]
def remove_marker(self, name: str) -> bool:
"""Remove markers (including axis) with the given name.
Args:
name (str): The name of the marker to remove.
Returns:
bool: True if the marker was removed successfully, False otherwise.
"""
if name not in self._markers:
logger.log_warning(f"Marker {name} not found.")
return False
try:
marker_group = self._markers[name]
env = self.get_env(marker_group.arena_index)
for marker_handle in marker_group.handles:
if marker_handle is not None:
env.remove_actor(marker_handle.get_name())
self._markers.pop(name)
return True
except Exception as e:
logger.log_warning(f"Failed to remove marker {name}: {str(e)}")
return False
[docs]
def get_axis_marker_items(
self,
) -> tuple[tuple[str, tuple[MeshObject, ...], float, float], ...]:
"""Return active axes for backend-neutral visualization.
Returns:
Tuples containing the marker name, native handles, axis length, and
axis radius for each active marker group.
"""
return tuple(
(
name,
group.handles,
group.axis_length,
group.axis_radius,
)
for name, group in self._markers.items()
)
[docs]
def add_custom_window_control(self, controls: list[ObjectManipulator]) -> None:
"""Add one or more custom window input controls.
This method registers additional :class:`ObjectManipulator` instances
with the simulation window so they can handle input events alongside
any default controls.
Args:
controls (list[ObjectManipulator]): A list of initialized
ObjectManipulator instances to add to the current window.
Each control will be registered via ``window.add_input_control``.
If no window is available, the controls are not added and a
warning is logged.
"""
if self._window is None:
logger.log_warning("No window available to add custom controls.")
return
for control in controls:
self._window.add_input_control(control)
def _build_window_record_output(
self, save_path: str | None, video_prefix: str
) -> tuple[str, str]:
"""Resolve the output directory and file name for viewer recording."""
if save_path is None:
output_dir = os.path.join(os.getcwd(), "outputs", "videos")
timestamp = datetime.now().strftime("%Y-%m-%d-%H-%M-%S")
video_name = f"{video_prefix}_{timestamp}"
else:
output_dir = os.path.dirname(save_path) or os.getcwd()
video_name = Path(os.path.basename(save_path)).stem
return output_dir, video_name
[docs]
def is_window_recording(self) -> bool:
"""Check whether the viewer window is currently recording."""
return self._window_record_state is not None
def _build_window_record_pose_from_look_at(
self,
eye: Sequence[float],
target: Sequence[float],
up: Sequence[float] = (0.0, 0.0, 1.0),
) -> np.ndarray:
"""Build a camera pose matrix for the recorder from look-at inputs."""
pose = look_at_to_pose(eye, target, up)[0].cpu().numpy()
pose[:3, 1] = -pose[:3, 1]
pose[:3, 2] = -pose[:3, 2]
return np.asarray(pose, dtype=np.float32)
def _resolve_window_record_pose(
self, state: _WindowRecordState
) -> np.ndarray | None:
"""Resolve the camera pose used by the recorder for the current frame."""
if state.pose_provider is not None:
pose = state.pose_provider()
return np.asarray(pose, dtype=np.float32)
if state.fixed_pose is not None:
return np.asarray(state.fixed_pose, dtype=np.float32)
if self._window is not None:
return np.asarray(self._window.get_pose_matrix(), dtype=np.float32)
return None
def _step_window_record(self, state: _WindowRecordState) -> int:
"""Capture frames in the render thread without blocking the UI loop."""
if state.task_status != TASK_RETURN.TASK_LOOP:
return state.task_status
now = time.time()
if now - state.last_capture_time < state.time_step:
return state.task_status
state.last_capture_time = now
return self._capture_window_record_frame(state)
def _capture_window_record_frame(self, state: _WindowRecordState) -> int:
"""Render one frame for the active recording session."""
frame: np.ndarray | None = None
pose = self._resolve_window_record_pose(state)
if pose is not None and state.record_camera is not None:
state.record_camera.set_world_pose(pose)
state.record_camera.render()
rgb = np.asarray(state.record_camera.get_rgb_map())
if rgb.size != 0:
frame = np.array(rgb[..., :3], copy=True, order="C")
if frame is None:
return state.task_status
state.frames.append(frame)
state.current_memory_bytes += frame.nbytes
if state.current_memory_bytes > state.max_memory_bytes:
logger.log_warning(
"Viewer recording exceeded the configured memory budget. "
"Press 'r' again to flush the buffered frames to disk."
)
state.task_status = TASK_RETURN.TASK_EXIT
return state.task_status
def _step_window_record_from_sim_update(
self, state: _WindowRecordState, physics_dt: float
) -> int:
"""Capture recording frames based on simulation time progression."""
if state.task_status != TASK_RETURN.TASK_LOOP:
return state.task_status
state.accumulated_sim_time += physics_dt
if state.accumulated_sim_time + 1e-9 < state.time_step:
return state.task_status
state.accumulated_sim_time = max(
0.0, state.accumulated_sim_time - state.time_step
)
# Publish from the simulation thread only when a frame is due. The
# render-thread recorder must not wait on its own render publication.
self._ensure_render_state()
return self._capture_window_record_frame(state)
def _save_window_record_worker(
self,
frames: list[np.ndarray],
output_dir: str,
video_name: str,
save_kwargs: dict[str, object],
) -> None:
"""Encode buffered frames into a video file in a background thread."""
from dexsim.utility import images_to_video
try:
os.makedirs(output_dir, exist_ok=True)
images_to_video(
images=frames,
output_dir=output_dir,
video_name=video_name,
**save_kwargs,
)
logger.log_info(
f"Viewer recording saved to {os.path.join(output_dir, video_name + '.mp4')}"
)
except Exception as exc:
logger.log_error(f"Failed to save viewer recording: {exc}")
[docs]
def start_window_record(
self,
save_path: str | None = None,
fps: int = 20,
max_memory: int = 1024,
video_prefix: str = "viewer_record",
pose_provider: Callable[[], np.ndarray] | None = None,
fixed_pose: np.ndarray | None = None,
look_at: (
tuple[
Sequence[float],
Sequence[float],
Sequence[float],
]
| None
) = None,
use_sim_time: bool | None = None,
) -> bool:
"""Start asynchronously recording the simulation to a video buffer.
The recorder can either follow the live viewer camera or run without a
window by using a fixed pose or a pose callback supplied by the caller.
Args:
save_path: Optional output path for the recorded video.
fps: Target output frames per second. Must be positive.
max_memory: Maximum buffered frame memory in MB. Must be positive.
video_prefix: File name prefix used when ``save_path`` is not provided.
pose_provider: Optional callback that returns the current camera pose.
fixed_pose: Optional fixed 4x4 camera pose matrix.
look_at: Optional ``(eye, target, up)`` tuple used to derive a fixed pose.
use_sim_time: Whether to capture frames from simulation time instead of
wall time. Defaults to headless mode when no viewer window exists.
Returns:
bool: True if recording starts successfully, otherwise False.
"""
if self.is_window_recording():
logger.log_error(
"A viewer recording session is already active. Stop it before starting a new recording."
)
if fps <= 0:
logger.log_error(f"Viewer recording FPS must be positive, got {fps}.")
if max_memory <= 0:
logger.log_error(
f"Viewer recording max_memory must be positive, got {max_memory}."
)
if pose_provider is not None and fixed_pose is not None:
logger.log_error(
"Recorder accepts only one explicit pose source: `pose_provider` or `fixed_pose`."
)
if pose_provider is not None and look_at is not None:
logger.log_error(
"Recorder accepts only one explicit pose source: `pose_provider` or `look_at`."
)
if fixed_pose is not None and look_at is not None:
logger.log_error(
"Recorder accepts only one explicit pose source: `fixed_pose` or `look_at`."
)
if look_at is not None:
fixed_pose = self._build_window_record_pose_from_look_at(*look_at)
if pose_provider is None and fixed_pose is None and self._window is None:
logger.log_warning(
"No simulation window available for viewer recording. "
"Provide `pose_provider`, `fixed_pose`, or `look_at` to record in headless mode."
)
return False
if use_sim_time is None:
use_sim_time = self._window is None
width = self.sim_config.width
height = self.sim_config.height
if self._window_record_camera is None:
camera_name = f"viewer_record_camera_{self.instance_id}"
self._window_record_camera = self._env.create_camera(
camera_name, width, height
)
record_camera = self._window_record_camera
if hasattr(record_camera, "is_open") and record_camera.is_open() is False:
record_camera.open_camera()
time_step = 1.0 / float(fps)
output_dir, video_name = self._build_window_record_output(
save_path, video_prefix
)
state = _WindowRecordState(
time_step=time_step,
max_memory_bytes=max_memory * 1024 * 1024,
output_dir=output_dir,
video_name=video_name,
save_kwargs={"fps": fps},
record_camera=record_camera,
pose_provider=pose_provider,
fixed_pose=(
None if fixed_pose is None else np.asarray(fixed_pose, dtype=np.float32)
),
capture_from_sim_update=use_sim_time,
last_capture_time=time.time() - time_step,
)
if not state.capture_from_sim_update:
def _window_record_loop(_: float) -> int:
return self._step_window_record(state)
state.loop_handle = self._world.thread_rt().add_loop(
_window_record_loop, time_step
)
self._window_record_state = state
follow_source = (
"live viewer pose"
if pose_provider is None and fixed_pose is None and self._window is not None
else "custom pose source"
)
timing_source = (
"simulation time" if state.capture_from_sim_update else "wall time"
)
save_target = os.path.join(output_dir, video_name + ".mp4")
if self._window is not None:
logger.log_info(
f"Viewer recording started ({follow_source}, {timing_source}). Press 'r' again to stop and save to "
f"{save_target}"
)
else:
logger.log_info(
f"Viewer recording started ({follow_source}, {timing_source}). Call `stop_window_record()` to save to "
f"{save_target}"
)
return True
[docs]
def stop_window_record(self, save_path: str | None = None) -> bool:
"""Stop the active viewer recording and save frames in the background."""
if self._window_record_state is None:
logger.log_warning("No active viewer recording session found.")
return False
state = self._window_record_state
state.task_status = TASK_RETURN.TASK_EXIT
if save_path is not None:
output_dir, video_name = self._build_window_record_output(
save_path, "viewer_record"
)
else:
output_dir, video_name = state.output_dir, state.video_name
if state.record_camera is not None and hasattr(state.record_camera, "is_open"):
if state.record_camera.is_open():
state.record_camera.close_camera()
frames = list(state.frames)
self._window_record_state = None
if len(frames) == 0:
logger.log_warning(
"Viewer recording stopped, but no frames were captured. Skipping video export."
)
return False
self._window_record_save_threads = [
thread for thread in self._window_record_save_threads if thread.is_alive()
]
save_thread = threading.Thread(
target=self._save_window_record_worker,
args=(frames, output_dir, video_name, dict(state.save_kwargs)),
daemon=False,
)
save_thread.start()
self._window_record_save_threads.append(save_thread)
logger.log_info(
"Viewer recording stopped. Saving video to "
f"{os.path.join(output_dir, video_name + '.mp4')} in background."
)
return True
[docs]
def wait_window_record_saves(self) -> None:
"""Wait for all background video export threads to finish."""
for thread in self._window_record_save_threads:
thread.join()
self._window_record_save_threads = []
[docs]
def toggle_window_record(
self,
save_path: str | None = None,
fps: int = 20,
max_memory: int = 1024,
video_prefix: str = "viewer_record",
) -> bool:
"""Toggle viewer recording on or off."""
if self.is_window_recording():
return self.stop_window_record(save_path=save_path)
return self.start_window_record(
save_path=save_path,
fps=fps,
max_memory=max_memory,
video_prefix=video_prefix,
)
[docs]
def enable_window_record_hotkey(
self,
save_path: str | None = None,
fps: int = 20,
max_memory: int = 1024,
video_prefix: str = "viewer_record",
) -> bool:
"""Register the ``r`` key to start/stop viewer recording."""
self._window_record_hotkey_cfg = {
"save_path": save_path,
"fps": fps,
"max_memory": max_memory,
"video_prefix": video_prefix,
}
if self._window is None:
logger.log_warning(
"No simulation window available yet. The viewer record hotkey will be registered after `open_window()`."
)
return False
if self._window_record_input_control is not None:
return True
from dexsim.types import InputKey
sim = self
hotkey_cfg = dict(self._window_record_hotkey_cfg)
class WindowRecordEvent(ObjectManipulator):
def on_key_down(self, key):
if key == InputKey.SCANCODE_R.value:
sim.toggle_window_record(**hotkey_cfg)
self._window_record_input_control = WindowRecordEvent()
self._window.add_input_control(self._window_record_input_control)
logger.log_info(
"Viewer record hotkey registered. Press 'r' to start/stop recording."
)
return True
@staticmethod
def _window_camera_pose_to_look_at(
pose: np.ndarray,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
"""Convert a DexSim window model matrix to look-at vectors.
DexSim stores the viewer camera model matrix with columns
``[right, up, -forward]``. The local camera up axis changes while the
viewer orbits, but ``Windows.set_look_at`` uses a world-up reference.
Always use DexSim's default Z-up vector so a captured snippet retains
the standard viewer controls.
Args:
pose: A 4x4 homogeneous viewer camera pose matrix.
Returns:
The ``(eye, look_at, up)`` vectors accepted by
``Windows.set_look_at``.
Raises:
ValueError: If ``pose`` is not a 4x4 homogeneous matrix.
"""
matrix = np.asarray(pose, dtype=np.float64)
if matrix.shape != (4, 4):
raise ValueError(
f"Window camera pose must have shape (4, 4), got {matrix.shape}."
)
eye = matrix[:3, 3]
look_at = eye - matrix[:3, 2]
up = np.array([0.0, 0.0, 1.0], dtype=np.float64)
return eye, look_at, up
@staticmethod
def _format_window_camera_pose(
pose: np.ndarray, convert_to_look_at: bool = True
) -> str:
"""Format a DexSim window pose as an executable Python snippet.
Args:
pose: A 4x4 homogeneous viewer camera pose matrix.
convert_to_look_at: Print a ``set_look_at`` call when true;
otherwise print the raw pose matrix.
Returns:
An executable Python snippet containing the camera pose.
Raises:
ValueError: If ``pose`` is not a 4x4 homogeneous matrix.
"""
matrix = np.asarray(pose, dtype=np.float64)
if matrix.shape != (4, 4):
raise ValueError(
f"Window camera pose must have shape (4, 4), got {matrix.shape}."
)
def _format_float(value: float) -> str:
if abs(value) < 1e-12:
return "0.0"
formatted = format(value, ".8g")
if "e" not in formatted and "." not in formatted:
formatted += ".0"
return formatted
def _vector_literal(vector: np.ndarray) -> str:
values = ", ".join(_format_float(float(value)) for value in vector)
return f"np.array([{values}], dtype=np.float32)"
if convert_to_look_at:
eye, look_at, up = SimulationManager._window_camera_pose_to_look_at(matrix)
return (
"window.set_look_at("
f"eye={_vector_literal(eye)}, "
f"look_at={_vector_literal(look_at)}, "
f"up={_vector_literal(up)})"
)
rows = ",\n ".join(
"[" + ", ".join(_format_float(float(value)) for value in row) + "]"
for row in matrix
)
return f"window_pose = np.array([\n {rows}\n], dtype=np.float32)"
[docs]
def print_window_camera_pose(self, convert_to_look_at: bool = True) -> str | None:
"""Print the current viewer camera pose as reusable Python code.
Args:
convert_to_look_at: Print ``window.set_look_at(...)`` by default.
Set false to print the raw 4x4 pose matrix instead.
Returns:
The printed snippet, or ``None`` when no viewer window is open.
"""
if self._window is None:
logger.log_warning("No simulation window available to print its pose.")
return None
pose = np.asarray(self._window.get_pose_matrix(), dtype=np.float32)
snippet = self._format_window_camera_pose(pose, convert_to_look_at)
print(snippet)
return snippet
[docs]
def enable_window_camera_pose_hotkey(self, convert_to_look_at: bool = True) -> bool:
"""Register ``p`` to print the current viewer camera pose.
Args:
convert_to_look_at: Print a ``window.set_look_at(...)`` call when
true, which is the default. Set false to print the raw matrix.
Returns:
Whether the control is registered on an available window.
"""
self._window_camera_pose_hotkey_cfg = {"convert_to_look_at": convert_to_look_at}
if self._window is None:
logger.log_warning(
"No simulation window available yet. The camera pose print "
"hotkey will be registered after `open_window()`."
)
return False
if self._window_camera_pose_input_control is not None:
return True
from dexsim.types import InputKey
sim = self
hotkey_cfg = dict(self._window_camera_pose_hotkey_cfg)
class WindowCameraPoseEvent(ObjectManipulator):
def on_key_down(self, key):
if key == InputKey.SCANCODE_P.value:
sim.print_window_camera_pose(**hotkey_cfg)
self._window_camera_pose_input_control = WindowCameraPoseEvent()
self._window.add_input_control(self._window_camera_pose_input_control)
logger.log_info(
"Camera pose print hotkey registered. Press 'p' to print the "
"current viewer pose."
)
return True
[docs]
def create_visual_material(self, cfg: VisualMaterialCfg) -> VisualMaterial:
"""Create a visual material with given configuration.
Args:
cfg (VisualMaterialCfg): configuration for the visual material.
Returns:
VisualMaterial: the created visual material instance handle.
"""
self._require_native_renderer("Native visual materials")
cfg = cfg.replace(uid=cfg.uid or f"visual_material_{uuid4().hex}")
descriptor = _material_desc_from_cfg(cfg, cfg.uid)
definitions = self._spawn_scene.builder.materials
registered = self._spawn_scene._material_definition(cfg.uid)
if registered is not None and registered != descriptor:
raise ValueError(f"Conflicting visual material definition for {cfg.uid!r}.")
if cfg.uid in self._visual_materials:
existing = self._visual_materials[cfg.uid]
if _material_desc_from_cfg(existing.cfg, cfg.uid) != descriptor:
raise ValueError(
f"Conflicting visual material definition for {cfg.uid!r}."
)
return existing
if cfg.preset is not None or cfg.parameters:
self._spawn_scene._resolve_materials({cfg.uid: descriptor})
mat, native_inst = self._spawn_scene._material_resources.resolve(descriptor)
visual_mat = VisualMaterial(
cfg, mat, initialize_default=False, mat_inst=native_inst
)
else:
mat: Material | None = self._env.find_material(cfg.uid)
initialize_default = mat is None or registered is None
if mat is None:
mat = self._env.create_pbr_material(cfg.uid, True)
visual_mat = VisualMaterial(cfg, mat, initialize_default=initialize_default)
definitions[cfg.uid] = descriptor
self._visual_materials[cfg.uid] = visual_mat
return visual_mat
[docs]
def get_visual_material(self, uid: str) -> VisualMaterial | None:
"""Get visual material by UID.
Args:
uid (str): uid of visual material.
"""
if uid not in self._visual_materials:
logger.log_warning(f"Visual material {uid} not found.")
return None
return self._visual_materials[uid]
[docs]
def clean_materials(self) -> None:
"""Clear runtime wrappers and ask the native owner to release unused materials.
Explicit material definitions remain reserved in the Scene for the World
lifetime. Cleaning does not permit a conflicting definition to replace a
UID that declarations or live native handles may still reference.
"""
self._visual_materials = {}
if self._env:
self._env.clean_materials()
[docs]
def reset_objects_state(
self,
env_ids: Sequence[int] | None = None,
excluded_uids: Sequence[str] | None = None,
*,
excluded_sensor_uids: Sequence[str] | None = None,
) -> None:
"""Reset the state of the simulated assets given the environment IDs and excluded UIDs.
Args:
env_ids (Sequence[int] | None): The environment IDs to reset. If None, reset all environments.
excluded_uids (Sequence[str] | None): List of asset UIDs to exclude from resetting. If None, reset all assets.
excluded_sensor_uids: Sensor UIDs to exclude without excluding
physical assets with the same UID.
"""
excluded_uids = set(excluded_uids) if excluded_uids is not None else set()
excluded_sensor_uids = set(excluded_sensor_uids or ())
articulation_uids = tuple(self._robots) + tuple(self._articulations)
reset_articulation_uids = tuple(
uid for uid in articulation_uids if uid not in excluded_uids
)
use_coordinated_newton_clear = bool(
self.physics.name == "newton" and reset_articulation_uids
)
excluded_articulation_uids = excluded_uids.intersection(articulation_uids)
solver_type = getattr(self.physics, "solver_type", None)
requires_complete_world_clear = solver_type in {
None,
"auto",
"mujoco_warp",
}
if (
use_coordinated_newton_clear
and requires_complete_world_clear
and excluded_articulation_uids
):
raise NotImplementedError(
"Resetting selected worlds with excluded Newton articulations "
"is unsupported because solver state must be cleared for every "
"articulation in each selected world."
)
newton_articulation_batch = None
if use_coordinated_newton_clear:
newton_articulation_batch = self._newton_articulation_reset_batch(
reset_articulation_uids,
env_ids,
)
for uid, robot in self._robots.items():
if uid not in excluded_uids:
if use_coordinated_newton_clear:
robot.reset(env_ids, clear_dynamics=False)
else:
robot.reset(env_ids)
for uid, articulation in self._articulations.items():
if uid not in excluded_uids:
if use_coordinated_newton_clear:
articulation.reset(env_ids, clear_dynamics=False)
else:
articulation.reset(env_ids)
for uid, visual in getattr(self, "_visual_objects", {}).items():
if uid not in excluded_uids and visual.is_spawn_bound:
visual.reset(env_ids)
for uid, rigid_obj in self._rigid_objects.items():
if uid not in excluded_uids:
rigid_obj.reset(env_ids)
for uid, rigid_obj_group in self._rigid_object_groups.items():
if uid not in excluded_uids:
rigid_obj_group.reset(env_ids)
for uid, deformable_obj in self._deformable_objects.items():
if uid not in excluded_uids:
deformable_obj.reset(env_ids)
for uid, light in self._lights.items():
if uid not in excluded_uids:
light.reset(env_ids)
for uid, sensor in self._sensors.items():
if uid not in excluded_uids and uid not in excluded_sensor_uids:
sensor.reset(env_ids)
if use_coordinated_newton_clear:
self._clear_newton_articulation_dynamics(newton_articulation_batch)
def _newton_articulation_reset_batch(
self,
articulation_uids: Sequence[str],
env_ids: Sequence[int] | None,
) -> Any:
"""Create the complete Newton articulation batch for selected worlds."""
result = self.spawn_result
if result is None:
raise RuntimeError(
"Newton articulation reset requires a prepared Spawn scene."
)
selected_env_ids = (
list(range(self.num_envs))
if env_ids is None
else torch.as_tensor(env_ids, dtype=torch.long).reshape(-1).tolist()
)
invalid_env_ids = [
env_id
for env_id in selected_env_ids
if env_id < 0 or env_id >= self.num_envs
]
if invalid_env_ids:
raise IndexError(
f"Environment selection {invalid_env_ids} is outside "
f"[0, {self.num_envs})."
)
handles = []
for uid in articulation_uids:
uid_handles = self._spawn_scene.handles(uid)
if len(uid_handles) != self.num_envs:
raise RuntimeError(
f"Articulation {uid!r} exposes {len(uid_handles)} Spawn "
f"handles for {self.num_envs} environments."
)
handles.extend(uid_handles[env_id] for env_id in selected_env_ids)
return result.create_articulation_batch(handles)
@staticmethod
def _clear_newton_articulation_dynamics(batch: Any) -> None:
"""Clear complete native articulation state for selected worlds."""
status = batch.clear_dynamics()
if isinstance(status, Integral) and status < 0:
raise RuntimeError(
"DexSim Scene articulation clear_dynamics failed with "
f"status {status}."
)
[docs]
def export_usd(self, fpath: str) -> bool:
"""Export the current simulation scene to a USD file.
Args:
fpath (str): The file path to save the USD file.
Returns:
bool: True if export is successful, False otherwise.
"""
try:
self._env.export_to_usd_file(fpath)
logger.log_info(f"Simulation scene exported to USD file: {fpath}")
return True
except Exception as e:
logger.log_error(f"Failed to export simulation scene to USD: {e}")
return False
[docs]
@staticmethod
def wait_scene_destruction(timeout_ms: int = 10000) -> None:
"""A public helper to wait for the underlying C++ scenes (dexsim.World) to destruct completely."""
import dexsim
import gc
# Force garbage collection to break cycle references
gc.collect()
import time
wait_times = 0
scene_count = dexsim.get_world_num()
max_loops = timeout_ms // 10
while scene_count > 0 and wait_times < max_loops:
time.sleep(0.01)
scene_count = dexsim.get_world_num()
wait_times += 1
if wait_times % 50 == 0:
from embodichain.utils import logger
logger.log_info(
f"Waiting for dexsim.World scenes to destruct. Remaining scenes: {scene_count}"
)
if scene_count > 0:
from embodichain.utils import logger
logger.log_warning(
f"Scene destruction wait timeout, {scene_count} C++ scene(s) still alive!"
)
[docs]
def destroy(self, exit_process: bool | None = None) -> None:
"""
No longer destructs C++ objects in place due to lingering deep local variables;
instead, packages itself into a destruction task, submits to the cleanup queue,
and waits for top-level delayed consumption.
Args:
exit_process (bool | None): Whether to call os._exit(0) after queuing
the destruction task. If None, reads EMBODICHAIN_SIM_EXIT_PROCESS.
"""
try:
self.stop_visualization()
except Exception as error:
logger.log_warning(f"Failed to stop Viser visualization cleanly: {error!r}")
if exit_process is None:
exit_process = (
os.getenv("EMBODICHAIN_SIM_EXIT_PROCESS", "1").strip().lower()
)
exit_process = exit_process not in ("0", "false", "no", "off")
self._is_pending_kill = True
# Transfer the actual destruction logic to the cleanup queue
SimulationManager._cleanup_queue.put(self._deferred_destroy)
if exit_process:
os._exit(0)
def _deferred_destroy(self) -> None:
"""Destroy all simulated assets and release resources."""
# Clean up all gizmos before destroying the simulation
for uid in list(self._gizmos.keys()):
self.disable_gizmo(uid)
if self.is_window_recording():
self.stop_window_record()
self.wait_window_record_saves()
# Stop the render loop before releasing scene resources. Vulkan window
# presentation may otherwise continue acquiring swapchain images while
# Env::Clean tears down render objects used by the in-flight frame.
if getattr(self, "is_window_opened", False):
self.close_window()
import sys, gc
# Release backend-owned views before Scene closes the native
# resources that back them. Newton also synchronizes its device here.
self.physics.prepare_for_teardown()
# Run wrapper destructors while their World is still alive. The later
# collections continue to break cycles left by the native teardown.
gc.collect()
# Render-only cameras may be attached to Spawn articulation link
# nodes. Remove their Arena views before closing Scene, which
# releases those parent nodes, and before World.quit releases their
# CameraGroups.
for sensor in list(getattr(self, "_sensors", {}).values()):
try:
sensor.destroy()
except Exception as error:
logger.log_warning(
f"Failed to destroy sensor {getattr(sensor, 'uid', None)!r}: "
f"{error!r}"
)
if self._spawn_scene is not None:
# Release result-scoped batches/facades before closing the
# Scene and, finally, the World that owns native resources.
for registry_name in (
"_visual_objects",
"_rigid_objects",
"_rigid_object_groups",
"_deformable_objects",
"_articulations",
"_robots",
):
for asset in getattr(self, registry_name, {}).values():
if isinstance(asset, VisualObject):
asset._release_spawn_references()
if hasattr(asset, "_data"):
asset._data = None
if hasattr(asset, "_spawn_result"):
asset._spawn_result = None
if hasattr(asset, "_entities"):
asset._entities = []
try:
self._spawn_scene.close()
finally:
self._spawn_scene = None
self.clean_materials()
if self._env:
self._env.clean()
if self._world:
self._world.quit()
# REMOVE INSTANCE FROM POOL
instance_id = getattr(self, "instance_id", 0)
if SimulationManager._instances.get(instance_id) is self:
SimulationManager.reset(instance_id)
# Helper to aggressively decouple C++ wrapped objects
def _sever_wrapper_refs(obj_registry):
if not hasattr(self, obj_registry):
return
registry = getattr(self, obj_registry)
if not isinstance(registry, dict):
return
for uid, obj in registry.items():
if hasattr(obj, "_world"):
obj._world = None
if hasattr(obj, "_ps"):
obj._ps = None
if hasattr(obj, "_env"):
obj._env = None
if hasattr(obj, "_entities"):
obj._entities = []
registry.clear()
_sever_wrapper_refs("_gizmos")
_sever_wrapper_refs("_markers")
_sever_wrapper_refs("_visual_objects")
_sever_wrapper_refs("_rigid_objects")
_sever_wrapper_refs("_constraints")
_sever_wrapper_refs("_rigid_object_groups")
_sever_wrapper_refs("_deformable_objects")
_sever_wrapper_refs("_articulations")
_sever_wrapper_refs("_robots")
_sever_wrapper_refs("_sensors")
_sever_wrapper_refs("_lights")
# Explicitly clear Python references to trigger C++ object destructors
self._env = None
self._world = None
self._default_plane = None
# Try to break ANY possible frame cycle
gc.collect()
self._visual_materials.clear()
self._texture_cache.clear()
self._arenas.clear()
self._markers.clear()
self._gizmos.clear()
self._constraints.clear()
if SimulationManager._instances.get(self.instance_id) is self:
SimulationManager.reset(self.instance_id)
# Forcefully drop underlying C++ object wrappers
self._env = None
self._world = None
gc.collect()
[docs]
@staticmethod
def flush_cleanup_queue() -> None:
"""Run pending destruction tasks and wait for their scenes to disappear.
An empty queue means that no manager requested destruction. In that
case, returning immediately is important: other managers may still own
live worlds, and waiting for the global world count to reach zero would
block until the timeout even though there is nothing to clean up.
"""
import gc
drained_task = False
while True:
try:
task = SimulationManager._cleanup_queue.get_nowait()
except queue.Empty:
break
drained_task = True
try:
task()
except Exception as e:
from embodichain.utils import logger
logger.log_error(f"Error during delayed destruction: {e}")
if not drained_task:
return
# After the queue is emptied, perform a top-level full GC to thoroughly reclaim dead objects that haven't released their RefPtrs yet
gc.collect()
# Other live managers still own native worlds; their world count must
# not be mistaken for an incomplete teardown of the drained tasks.
if not SimulationManager._instances:
SimulationManager.wait_scene_destruction()
[docs]
def get_physics_scene(instance_id: int = 0):
"""Return the active physics scene from a SimulationManager instance.
This is the unified EmbodiChain access point for code that previously
reached through ``dexsim.default_world().get_physics_scene()``.
"""
return SimulationManager.get_instance(instance_id).get_physics_scene()