# Copyright (c) 2022-2026, The Isaac Lab Project Developers (https://github.com/isaac-sim/IsaacLab/blob/main/CONTRIBUTORS.md).
# All rights reserved.
#
# SPDX-License-Identifier: BSD-3-Clause
"""Base class for physics managers with unified callback system."""
from __future__ import annotations
import logging
import weakref
from abc import ABC, abstractmethod
from collections.abc import Callable
from enum import Enum
from typing import TYPE_CHECKING, Any, ClassVar
from isaaclab.sim.utils.stage import get_current_stage
from isaaclab.utils._device import set_cuda_device
if TYPE_CHECKING:
from isaaclab.scene_data import SceneDataBackend
from isaaclab.sim.simulation_context import SimulationContext
logger = logging.getLogger(__name__)
[docs]
class PhysicsEvent(Enum):
"""Physics simulation lifecycle events.
These are general events that apply across all physics backends.
Backend-specific events (e.g., PhysX step events, timeline events) are handled
by the respective manager classes via their own event enums (e.g., IsaacEvents).
Lifecycle order: MODEL_INIT -> PHYSICS_READY -> STOP
"""
MODEL_INIT = "model_init"
"""Physics model is being constructed.
Fired during scene building, before simulation can run. Use this to register
physics representations (rigid bodies, joints, constraints) with the solver.
"""
PHYSICS_READY = "physics_ready"
"""Physics is initialized and queryable.
Fired after all physics data structures are created and the simulation is
ready to step. Assets can now read initial state (positions, velocities).
"""
STOP = "stop"
"""Simulation is stopping."""
[docs]
class CallbackHandle:
"""Handle for a registered callback, allowing deregistration."""
[docs]
def __init__(self, callback_id: int, manager: type[PhysicsManager]):
self._id = callback_id
self._manager = manager
@property
def id(self) -> int:
return self._id
def deregister(self) -> None:
"""Remove this callback from the manager."""
self._manager.deregister_callback(self._id)
[docs]
class PhysicsManager(ABC):
"""Abstract base class for physics simulation managers.
Physics managers handle the lifecycle of a physics simulation backend,
including initialization, stepping, and cleanup.
This base class provides:
- Unified callback management system
- Common state variables (_sim, _cfg, _device)
- Default accessor implementations
Lifecycle: initialize() -> reset() -> step() (repeated) -> close()
"""
_sim: ClassVar[SimulationContext | None] = None
_cfg: ClassVar[Any] = None
_device: ClassVar[str] = "cuda:0"
_sim_time: ClassVar[float] = 0.0
_callbacks: ClassVar[dict[int, tuple[Any, Callable, int, str | None, Any]]] = {}
_callback_id: ClassVar[int] = 0
views: ClassVar[dict[tuple[type, str], Any]] = {}
@classmethod
def _prepare_stage_creation(cls) -> None:
"""Perform backend-specific setup required before the USD stage is created."""
pass
@classmethod
def fix_articulation_root(cls, articulation_prim: Any, stage: Any = None) -> Any:
"""Ensure that an articulation root has one enabled world fixed joint.
The base implementation leaves the root in place. Backends whose parser requires a different
root topology may relocate it and return the resulting root prim.
Args:
articulation_prim: The articulation-root prim to fix.
stage: The stage containing the prim. Defaults to the current stage.
Returns:
The articulation-root prim after backend normalization.
Raises:
NotImplementedError: If a new joint is needed and the root is not a rigid body.
"""
# Keep these imports local. Hoisting the isaaclab.sim ones closes a real import cycle:
# isaaclab.physics -> sim.schemas.schemas -> sim.utils.prims -> sim.utils.queries ->
# sim.simulation_context -> isaaclab.physics (partially initialized). Keeping pxr local
# also keeps USD out of the config-definition path, which env configs import through
# managers.manager_base before the simulation app starts.
from pxr import UsdPhysics # noqa: PLC0415
from isaaclab.sim.schemas.schemas import create_world_fixed_joint # noqa: PLC0415
from isaaclab.sim.utils import find_global_fixed_joint_prim # noqa: PLC0415
if stage is None:
stage = get_current_stage()
root_path = articulation_prim.GetPath().pathString
joint = find_global_fixed_joint_prim(root_path, stage=stage)
if joint is not None:
joint.GetJointEnabledAttr().Set(True)
return articulation_prim
if not articulation_prim.HasAPI(UsdPhysics.RigidBodyAPI):
raise NotImplementedError(f"Cannot fix non-rigid articulation root '{root_path}'.")
create_world_fixed_joint(articulation_prim, stage)
return articulation_prim
@staticmethod
def _relocate_articulation_root(
articulation_prim: Any,
companion_schema: str,
companion_namespace: str,
) -> Any:
"""Move root-bearing schemas and authored properties to the root link's parent."""
# Keep pxr local: this module is imported while environment configs load (via the manager
# classes), and config loading must not pull USD/omni modules before the simulation app
# starts.
from pxr import Usd, UsdPhysics # noqa: PLC0415
new_root = articulation_prim.GetParent()
if new_root.HasAPI(UsdPhysics.ArticulationRootAPI):
raise RuntimeError(
f"Cannot relocate '{articulation_prim.GetPath()}' to existing articulation root '{new_root.GetPath()}'."
)
# Keep this import local for the same reason as the pxr imports above.
from isaaclab.sim.schemas._backend_hooks import _articulation_root_companion_namespace # noqa: PLC0415
registry = Usd.SchemaRegistry()
root_schema = UsdPhysics.Tokens.PhysicsArticulationRootAPI
schemas_to_move = []
for schema_name in articulation_prim.GetPrimTypeInfo().GetAppliedAPISchemas():
definition = registry.FindAppliedAPIPrimDefinition(schema_name)
companion_namespace_override = _articulation_root_companion_namespace(schema_name)
if schema_name == companion_schema:
properties = list(articulation_prim.GetAuthoredPropertiesInNamespace(companion_namespace))
elif companion_namespace_override is not None:
# a backend-registered schema, possibly an unregistered token the registry cannot
# describe, so take the namespace the backend declared for it
properties = list(articulation_prim.GetAuthoredPropertiesInNamespace(companion_namespace_override))
elif schema_name == root_schema or (
definition is not None and root_schema in definition.GetAppliedAPISchemas()
):
properties = []
if definition is not None:
for property_name in definition.GetPropertyNames():
prop = articulation_prim.GetProperty(property_name)
if prop and prop.IsAuthored():
properties.append(prop)
else:
continue
schemas_to_move.append((schema_name, properties))
for schema_name, properties in schemas_to_move:
if not new_root.AddAppliedSchema(schema_name):
raise RuntimeError(f"Failed to apply '{schema_name}' to '{new_root.GetPath()}'.")
for prop in properties:
if not prop.FlattenTo(new_root):
raise RuntimeError(f"Failed to move '{prop.GetPath()}' to '{new_root.GetPath()}'.")
for schema_name, _ in schemas_to_move:
if not articulation_prim.RemoveAppliedSchema(schema_name):
raise RuntimeError(f"Failed to remove '{schema_name}' from '{articulation_prim.GetPath()}'.")
if articulation_prim.HasAPI(UsdPhysics.ArticulationRootAPI) or not new_root.HasAPI(
UsdPhysics.ArticulationRootAPI
):
raise RuntimeError(
f"Failed to relocate articulation root '{articulation_prim.GetPath()}' to '{new_root.GetPath()}'."
)
return new_root
@classmethod
def register_callback(
cls,
callback: Callable[[Any], None],
event: PhysicsEvent,
order: int = 0,
name: str | None = None,
wrap_weak_ref: bool = True,
) -> CallbackHandle:
"""Register a callback for a physics event.
Args:
callback: The callback function. Receives event payload as argument.
event: The event to listen for.
order: Priority order (lower = earlier). Default 0.
name: Optional name for debugging.
wrap_weak_ref: If True, wrap bound methods with weak references
to prevent preventing garbage collection. Default True.
Returns:
CallbackHandle that can be used to deregister the callback.
Example:
>>> def on_physics_ready(payload):
... print("Physics is ready!")
>>> handle = PhysxManager.register_callback(on_physics_ready, PhysicsEvent.PHYSICS_READY)
>>> # Later, to remove:
>>> handle.deregister()
"""
cid = cls._callback_id
cls._callback_id += 1
if wrap_weak_ref:
callback = cls._wrap_weak_ref(callback)
subscription = cls._subscribe_to_event(cid, callback, event, order, name)
cls._callbacks[cid] = (event, callback, order, name, subscription)
return CallbackHandle(cid, cls)
@classmethod
def deregister_callback(cls, callback_id: int | CallbackHandle) -> None:
"""Remove a registered callback.
Args:
callback_id: The ID or CallbackHandle returned by register_callback().
"""
cid = callback_id.id if isinstance(callback_id, CallbackHandle) else callback_id
if cid not in cls._callbacks:
return
event, callback, order, name, subscription = cls._callbacks.pop(cid)
cls._unsubscribe_from_event(cid, event, subscription)
@classmethod
def dispatch_event(cls, event: PhysicsEvent, payload: Any = None) -> None:
"""Dispatch an event to all registered callbacks.
This is the default implementation using simple callback lists.
Subclasses may override or extend with platform-specific dispatch.
Args:
event: The event to dispatch.
payload: Optional data to pass to callbacks.
"""
matching = [(cid, cb, order) for cid, (ev, cb, order, name, sub) in cls._callbacks.items() if ev == event]
matching.sort(key=lambda x: x[2])
for _, callback, _ in matching:
callback(payload)
@classmethod
def clear_callbacks(cls) -> None:
"""Remove all registered callbacks.
Do NOT reset ``_callback_id`` — handle IDs must remain monotonically
unique across the lifetime of the process. Resetting the counter
would let a future :meth:`register_callback` hand out an ID that an
old, still-alive :class:`CallbackHandle` (e.g. on a sensor that has
not been garbage-collected yet) holds, so when the old object
eventually finalizes its ``__del__`` would deregister the new
callback. This bit ovphysx's kitless multi-context tests where two
``InteractiveScene``s are created in sequence: the first scene's
sensor would post-GC deregister the second scene's
``_initialize_callback`` by ID collision, leaving the second sensor
forever uninitialized.
"""
for cid in list(cls._callbacks.keys()):
cls.deregister_callback(cid)
cls._callbacks.clear()
@classmethod
def _wrap_weak_ref(cls, callback: Callable) -> Callable:
"""Wrap bound methods with weak references to prevent leaks.
Args:
callback: The callback to wrap.
Returns:
Wrapped callback if it's a bound method, otherwise original.
"""
owner = getattr(callback, "__self__", None)
if owner is not None:
obj_ref = weakref.ref(owner)
method_name = callback.__name__
def weak_callback(payload: Any) -> Any:
obj = obj_ref()
if obj is None:
return None
return getattr(obj, method_name)(payload)
return weak_callback
return callback
@classmethod
def _subscribe_to_event(
cls,
callback_id: int,
callback: Callable,
event: PhysicsEvent,
order: int,
name: str | None,
) -> Any:
"""Subscribe to a platform-specific event.
Override in subclasses to integrate with platform event systems
(e.g., Omniverse event bus, timeline events).
Args:
callback_id: Unique ID for this callback.
callback: The callback function.
event: The event to subscribe to.
order: Priority order.
name: Optional name.
Returns:
Platform-specific subscription object (stored for cleanup).
"""
return None
@classmethod
def _unsubscribe_from_event(
cls,
callback_id: int,
event: PhysicsEvent,
subscription: Any,
) -> None:
"""Unsubscribe from a platform-specific event.
Override in subclasses to clean up platform subscriptions.
Args:
callback_id: The callback ID being removed.
event: The event that was subscribed to.
subscription: The subscription object from _subscribe_to_event().
"""
pass
@classmethod
@abstractmethod
def initialize(cls, sim_context: SimulationContext) -> None:
"""Initialize the physics manager with simulation context.
Subclasses should call super().initialize() first, then do backend-specific setup.
Args:
sim_context: Parent simulation context.
"""
# Set on PhysicsManager explicitly so PhysicsManager.get_*() works
# regardless of which subclass is active (Python class vars are per-class)
PhysicsManager._sim = sim_context
PhysicsManager._cfg = sim_context.cfg.physics
PhysicsManager._device = sim_context.cfg.device
PhysicsManager._sim_time = 0.0
# Synchronize the process-wide CUDA device before backend-specific
# initialization allocates state. PyTorch must select the device before
# Warp so that both runtimes retain the same primary CUDA context.
if "cuda" in PhysicsManager._device:
set_cuda_device(PhysicsManager._device)
@classmethod
@abstractmethod
def reset(cls, soft: bool = False) -> None:
"""Reset physics simulation.
Args:
soft: If True, skip full reinitialization.
"""
pass
@classmethod
@abstractmethod
def forward(cls) -> None:
"""Update kinematics without stepping physics (for rendering)."""
pass
@classmethod
@abstractmethod
def get_scene_data_backend(cls) -> SceneDataBackend:
"""Return the SceneDataBackend for the SceneDataProvider."""
pass
@classmethod
@abstractmethod
def step(cls) -> None:
"""Step physics simulation by one timestep (physics only, no rendering)."""
pass
@classmethod
def pre_render(cls) -> None:
"""Sync deferred physics state to the rendering backend.
Called by :meth:`~isaaclab.sim.SimulationContext.render` before cameras
and visualizers read scene data. The default implementation is a no-op.
Backends that defer transform writes (e.g. Newton's dirty-flag pattern)
should override this to flush pending updates.
"""
pass
@classmethod
def after_visualizers_render(cls) -> None:
"""Hook after visualizers have stepped during :meth:`~isaaclab.sim.SimulationContext.render`.
Use for physics-backend sync (e.g. fabric) if needed. Default is a no-op.
"""
pass
@classmethod
def video_capture_backend(cls) -> str | None:
"""Return the video capture backend identifier for this physics manager.
Used by :class:`~isaaclab.envs.utils.video_recorder.VideoRecorder` to select
how perspective video frames are captured when no visualizer is active.
Returns:
``"kit"`` for backends that use Kit/Replicator (e.g. :class:`~isaaclab_physx.physics.PhysxManager`),
``"newton_gl"`` for backends that use a headless Newton GL viewer
(e.g. :class:`~isaaclab_newton.physics.NewtonManager`),
or ``None`` if the backend does not support perspective video capture.
"""
return None
@classmethod
def close(cls) -> None:
"""Clean up physics resources.
Subclasses whose STOP listeners own backend handles should call
``super().close()`` before backend-specific cleanup so those listeners
can invalidate their handles while the backend is still live.
All STOP listeners are given a chance to run. If one or more listeners
fail, callback and shared simulation state is still cleared before an
aggregate :class:`RuntimeError` is raised from the first failure.
"""
sim = PhysicsManager._sim
# A config may declare its manager lazily as a ``"module:Class"`` string, which proxies
# attribute access but is a ``str``, so compare against that form as well as the class.
# The string must name the class's defining module; a config that pointed at a re-export
# path would not match here.
is_active_manager = sim is not None and (
sim.physics_manager is cls or sim.physics_manager == f"{cls.__module__}:{cls.__qualname__}"
)
callback_errors = cls._dispatch_event_collect_errors(PhysicsEvent.STOP) if is_active_manager else []
try:
cls.clear_callbacks()
finally:
if is_active_manager:
PhysicsManager.views.clear()
PhysicsManager._sim = None
PhysicsManager._cfg = None
PhysicsManager._sim_time = 0.0
if callback_errors:
raise RuntimeError(
f"{len(callback_errors)} callback(s) failed during PhysicsEvent.STOP dispatch."
) from callback_errors[0]
@classmethod
def _dispatch_event_collect_errors(cls, event: PhysicsEvent, payload: Any = None) -> list[Exception]:
"""Dispatch an event to every listener and collect direct or backend-stored failures."""
matching = [
(callback, order)
for registered_event, callback, order, _name, _subscription in cls._callbacks.values()
if registered_event == event
]
matching.sort(key=lambda item: item[1])
callback_errors: list[Exception] = []
raise_stored = getattr(cls, "raise_callback_exception_if_any", None)
def drain_stored_error() -> None:
if callable(raise_stored):
try:
raise_stored()
except Exception as exc:
callback_errors.append(exc)
for callback, _order in matching:
try:
callback(payload)
except Exception as exc:
callback_errors.append(exc)
drain_stored_error()
return callback_errors
@classmethod
def get_physics_dt(cls) -> float:
"""Get the physics timestep in seconds."""
return PhysicsManager._sim.cfg.dt if PhysicsManager._sim else 1.0 / 60.0
@classmethod
def get_device(cls) -> str:
"""Get the physics simulation device."""
return PhysicsManager._device
@classmethod
def get_simulation_time(cls) -> float:
"""Get the current simulation time in seconds."""
return PhysicsManager._sim_time
@classmethod
def get_physics_sim_view(cls) -> Any:
"""Get the physics simulation view. Override in subclasses."""
return None
@classmethod
def play(cls) -> None:
"""Start or resume physics simulation. Default is no-op."""
pass
@classmethod
def pause(cls) -> None:
"""Pause physics simulation. Default is no-op."""
pass
@classmethod
def stop(cls) -> None:
"""Stop physics simulation. Default is no-op."""
pass
@classmethod
def wait_for_playing(cls) -> None:
"""Block until the timeline is playing. Default is no-op."""
pass
@classmethod
def set_decimation(cls, decimation: int) -> None:
"""Inform the physics backend how many substeps the environment runs per policy step.
Backends that can fold the full decimation loop into a single
:meth:`step` call (e.g. Newton with all-graphable actuators) use this
to size their internal loop / CUDA graph. The default implementation
is a no-op.
Args:
decimation: Number of physics steps per environment step.
"""
pass
@classmethod
def handles_decimation(cls) -> bool:
"""``True`` when :meth:`step` executes the full decimation loop internally.
When this returns ``True`` the environment should call :meth:`step`
once per policy step instead of looping ``decimation`` times.
"""
return False
@classmethod
def get_backend(cls) -> str:
"""Get the tensor backend being used ("numpy" or "torch")."""
return "torch" if "cuda" in PhysicsManager._device else "numpy"
@staticmethod
def safe_callback_invoke(fn: Callable, *args, physics_manager: type[PhysicsManager] | None = None) -> None:
"""Invoke a callback, catching exceptions that would be swallowed by external event buses.
Ignores ``ReferenceError`` (from garbage-collected weakref proxies). All other
exceptions are forwarded to *physics_manager*.``store_callback_exception`` when
available (see note below), or re-raised immediately otherwise.
Note (Octi):
The carb event bus used by PhysX/Omniverse silently swallows exceptions raised
inside callbacks. ``PhysxManager`` works around this by storing the exception
and re-raising it after event dispatch completes (in ``reset()`` / ``step()``).
Backends that dispatch events directly (e.g. Newton) don't need this — exceptions
propagate normally — so ``store_callback_exception`` is not called for them.
This is a known wart; a cleaner solution is actively being explored.
"""
try:
fn(*args)
except ReferenceError:
pass
except Exception as e:
store_fn = getattr(physics_manager, "store_callback_exception", None)
if callable(store_fn):
store_fn(e)
else:
raise
