Actuators integration

source/isaaclab/changelog.d/antoiner-refactor-pr5455-followups.minor.rst

@@ -0,0 +1,22 @@
+Added
+^^^^^
+
+* Added :func:`~isaaclab.sim.schemas.define_actuator_properties` to author
+  ``NewtonActuator`` USD prims from IsaacLab actuator configs. Lives alongside
+  the other ``define_*_properties`` schema writers and is invoked from the
+  schema-side post-spawn hook below.
+* Added :meth:`~isaaclab.assets.AssetBaseCfg._post_spawn` hook (no-op by
+  default) invoked by :class:`~isaaclab.assets.AssetBase` after spawning the
+  asset. :class:`~isaaclab.assets.ArticulationCfg` overrides it to author
+  Newton-native actuator prims from :attr:`~isaaclab.assets.ArticulationCfg.actuators`.
+
+Changed
+^^^^^^^
+
+* :class:`~isaaclab.assets.BaseArticulation` no longer imports
+  ``isaaclab_newton`` from its ``__init__``. Newton-native actuator authoring
+  now flows through the generic ``_post_spawn`` hook on
+  :class:`~isaaclab.assets.AssetBaseCfg`.
+* :meth:`~isaaclab.physics.PhysicsManager.set_decimation` now has an explicit
+  ``pass`` body so the base class is consistent with the other no-op
+  classmethods.

source/isaaclab/isaaclab/assets/articulation/articulation_cfg.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 from dataclasses import MISSING
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from isaaclab.actuators import ActuatorBaseCfg
 from isaaclab.utils.configclass import configclass
@@ -74,3 +74,25 @@ class InitialStateCfg(AssetBaseCfg.InitialStateCfg):
     actuator_value_resolution_debug_print = False
     """Print the resolution of actuator final value when input cfg is different from USD value, Defaults to False
     """
+
+    def _post_spawn(self, stage: Any) -> None:
+        """Author ``NewtonActuator`` USD prims from :attr:`actuators` after spawn.
+
+        Invoked by :class:`~isaaclab.assets.AssetBase` once the articulation's prims
+        exist on the stage. Delegates to
+        :func:`~isaaclab.sim.schemas.define_actuator_properties`, which gates itself
+        on ``sim_cfg.use_newton_actuators`` and silently no-ops when the simulation
+        is not configured for Newton-native actuators.
+        """
+        if self.actuators is MISSING:
+            return
+        from isaaclab.sim.schemas.schemas_actuators import define_actuator_properties  # noqa: PLC0415
+
+        # In InteractiveScene, articulated assets are often spawned first under
+        # a template path (for example ``/World/template/Robot``) and cloned
+        # into ``{ENV_REGEX_NS}`` later. Author NewtonActuator prims on the
+        # actual spawned source prim so clones inherit them.
+        author_prim_path = (
+            self.spawn.spawn_path if self.spawn is not None and self.spawn.spawn_path is not None else self.prim_path
+        )
+        define_actuator_properties(author_prim_path, self.actuators, stage=stage)

source/isaaclab/isaaclab/assets/articulation/base_articulation.py

@@ -16,6 +16,7 @@
 import torch
 import warp as wp
 
+from ...sim import SimulationContext
 from ...utils.leapp.leapp_semantics import OutputKindEnum, joint_names_resolver, leapp_tensor_semantics
 from ..asset_base import AssetBase
 
@@ -95,6 +96,8 @@ def __init__(self, cfg: ArticulationCfg):
             cfg: A configuration instance.
         """
         super().__init__(cfg)
+        sim_ctx = SimulationContext.instance()
+        self._sim_cfg = sim_ctx.cfg if sim_ctx is not None else None
 
     """
     Properties

source/isaaclab/isaaclab/assets/asset_base.py

@@ -15,6 +15,8 @@
 import torch
 import warp as wp
 
+from pxr import Usd
+
 import isaaclab.sim as sim_utils
 from isaaclab.physics import PhysicsEvent, PhysicsManager
 from isaaclab.sim.simulation_context import SimulationContext
@@ -79,7 +81,7 @@ def __init__(self, cfg: AssetBaseCfg):
         # flag for whether the asset is initialized
         self._is_initialized = False
         # get stage handle
-        self.stage = get_current_stage()
+        self.stage: Usd.Stage = get_current_stage()
 
         # spawn the asset
         # determine path where prims should exist after spawn
@@ -96,6 +98,8 @@ def __init__(self, cfg: AssetBaseCfg):
             matching_prims = sim_utils.find_matching_prims(check_path)
             if len(matching_prims) == 0:
                 raise RuntimeError(f"Could not find prim with path {check_path}.")
+            # schema-side post-spawn hook (e.g. ArticulationCfg authors NewtonActuator prims here)
+            self.cfg._post_spawn(self.stage)
         else:
             # asset should exist at run time
             check_path = self.cfg.prim_path

source/isaaclab/isaaclab/assets/asset_base_cfg.py

@@ -6,7 +6,7 @@
 from __future__ import annotations
 
 from dataclasses import MISSING
-from typing import Literal
+from typing import Any, Literal
 
 from isaaclab.sim import SpawnerCfg
 from isaaclab.utils.configclass import configclass
@@ -88,3 +88,17 @@ class InitialStateCfg:
     When ``None`` (the default), shape checks follow Python's ``__debug__``
     flag — enabled in normal mode, disabled with ``python -O``.
     """
+
+    def _post_spawn(self, stage: Any) -> None:
+        """Hook invoked by :class:`~isaaclab.assets.AssetBase` after the asset's prims are
+        spawned and verified to exist on the stage.
+
+        The default implementation is a no-op. Subclasses that need to author additional
+        USD schemas tied to this asset (for example, :class:`~isaaclab.assets.ArticulationCfg`
+        which authors ``NewtonActuator`` prims from its ``actuators`` mapping) should
+        override this method.
+
+        Args:
+            stage: The USD stage on which the asset was spawned.
+        """
+        pass

source/isaaclab/isaaclab/envs/direct_marl_env.py

@@ -92,6 +92,7 @@ def __init__(self, cfg: DirectMARLEnvCfg, render_mode: str | None = None, **kwar
         self.render_mode = render_mode
         # initialize internal variables
         self._is_closed = False
+        self._physics_handles_decimation = False
 
         # set the seed for the environment
         if self.cfg.seed is not None:
@@ -197,6 +198,10 @@ def _init_sim(self, render_mode: str | None = None, **kwargs):
             # this shouldn't cause an issue since later on, users do a reset over all the environments
             # so the lazy buffers would be reset.
             self.scene.update(dt=self.physics_dt)
+        # let the physics backend know about the env decimation so it can
+        # fold the full loop into a single step() when possible
+        self.sim.physics_manager.set_decimation(self.cfg.decimation)
+        self._physics_handles_decimation = self.sim.physics_manager.handles_decimation()
 
         # check if debug visualization is has been implemented by the environment
         source_code = inspect.getsource(self._set_debug_vis_impl)
@@ -433,21 +438,32 @@ def step(self, actions: dict[AgentID, ActionType]) -> EnvStepReturn:
         is_rendering = self.sim.is_rendering
 
         # perform physics stepping
-        for _ in range(self.cfg.decimation):
-            self._sim_step_counter += 1
-            # set actions into buffers
+        if self._physics_handles_decimation:
+            self._sim_step_counter += self.cfg.decimation
             self._apply_action()
-            # set actions into simulator
             self.scene.write_data_to_sim()
-            # simulate
             self.sim.step(render=False)
-            # render between steps only if the GUI or an RTX sensor needs it.
-            # When render_enabled is False, Kit visualizer (camera/GUI) is skipped
-            # but standalone visualizers (Newton, Rerun, Viser) still update.
+            # render only when a render_interval boundary falls within this decimation block,
+            # mirroring the per-sub-step check in the else branch.
             if self._sim_step_counter % self.cfg.sim.render_interval == 0 and is_rendering:
                 self.sim.render(skip_app_pumping=not self.render_enabled)
-            # update buffers at sim dt
-            self.scene.update(dt=self.physics_dt)
+            self.scene.update(dt=self.step_dt)
+        else:
+            for _ in range(self.cfg.decimation):
+                self._sim_step_counter += 1
+                # set actions into buffers
+                self._apply_action()
+                # set actions into simulator
+                self.scene.write_data_to_sim()
+                # simulate
+                self.sim.step(render=False)
+                # render between steps only if the GUI or an RTX sensor needs it.
+                # When render_enabled is False, Kit visualizer (camera/GUI) is skipped
+                # but standalone visualizers (Newton, Rerun, Viser) still update.
+                if self._sim_step_counter % self.cfg.sim.render_interval == 0 and is_rendering:
+                    self.sim.render(skip_app_pumping=not self.render_enabled)
+                # update buffers at sim dt
+                self.scene.update(dt=self.physics_dt)
 
         # post-step:
         # -- update env counters (used for curriculum generation)

source/isaaclab/isaaclab/envs/direct_rl_env.py

@@ -97,6 +97,7 @@ def __init__(self, cfg: DirectRLEnvCfg, render_mode: str | None = None, **kwargs
         self.render_mode = render_mode
         # initialize internal variables
         self._is_closed = False
+        self._physics_handles_decimation = False
 
         # set the seed for the environment
         if self.cfg.seed is not None:
@@ -203,6 +204,10 @@ def _init_sim(self, render_mode: str | None = None, **kwargs):
             # this shouldn't cause an issue since later on, users do a reset over all the environments
             # so the lazy buffers would be reset.
             self.scene.update(dt=self.physics_dt)
+        # let the physics backend know about the env decimation so it can
+        # fold the full loop into a single step() when possible
+        self.sim.physics_manager.set_decimation(self.cfg.decimation)
+        self._physics_handles_decimation = self.sim.physics_manager.handles_decimation()
 
         # check if debug visualization is has been implemented by the environment
         source_code = inspect.getsource(self._set_debug_vis_impl)
@@ -424,21 +429,32 @@ def step(self, action: torch.Tensor) -> VecEnvStepReturn:
         is_rendering = self.sim.is_rendering
 
         # perform physics stepping
-        for _ in range(self.cfg.decimation):
-            self._sim_step_counter += 1
-            # set actions into buffers
+        if self._physics_handles_decimation:
+            self._sim_step_counter += self.cfg.decimation
             self._apply_action()
-            # set actions into simulator
             self.scene.write_data_to_sim()
-            # simulate
             self.sim.step(render=False)
-            # render between steps only if the GUI or an RTX sensor needs it.
-            # When render_enabled is False, Kit visualizer (camera/GUI) is skipped
-            # but standalone visualizers (Newton, Rerun, Viser) still update.
+            # render only when a render_interval boundary falls within this decimation block,
+            # mirroring the per-sub-step check in the else branch.
             if self._sim_step_counter % self.cfg.sim.render_interval == 0 and is_rendering:
                 self.sim.render(skip_app_pumping=not self.render_enabled)
-            # update buffers at sim dt
-            self.scene.update(dt=self.physics_dt)
+            self.scene.update(dt=self.step_dt)
+        else:
+            for _ in range(self.cfg.decimation):
+                self._sim_step_counter += 1
+                # set actions into buffers
+                self._apply_action()
+                # set actions into simulator
+                self.scene.write_data_to_sim()
+                # simulate
+                self.sim.step(render=False)
+                # render between steps only if the GUI or an RTX sensor needs it.
+                # When render_enabled is False, Kit visualizer (camera/GUI) is skipped
+                # but standalone visualizers (Newton, Rerun, Viser) still update.
+                if self._sim_step_counter % self.cfg.sim.render_interval == 0 and is_rendering:
+                    self.sim.render(skip_app_pumping=not self.render_enabled)
+                # update buffers at sim dt
+                self.scene.update(dt=self.physics_dt)
 
         # post-step:
         # -- update env counters (used for curriculum generation)

source/isaaclab/isaaclab/envs/manager_based_env.py

@@ -94,6 +94,7 @@ def __init__(self, cfg: ManagerBasedEnvCfg):
         self.cfg = cfg
         # initialize internal variables
         self._is_closed = False
+        self._physics_handles_decimation = False
 
         # set the seed for the environment
         if self.cfg.seed is not None:
@@ -216,6 +217,10 @@ def _init_sim(self):
             # this shouldn't cause an issue since later on, users do a reset over all the environments
             # so the lazy buffers would be reset.
             self.scene.update(dt=self.physics_dt)
+        # let the physics backend know about the env decimation so it can
+        # fold the full loop into a single step() when possible
+        self.sim.physics_manager.set_decimation(self.cfg.decimation)
+        self._physics_handles_decimation = self.sim.physics_manager.handles_decimation()
         # add timeline event to load managers
         self.load_managers()
 
@@ -530,21 +535,32 @@ def step(self, action: torch.Tensor) -> tuple[VecEnvObs, dict]:
         is_rendering = self.sim.is_rendering
 
         # perform physics stepping
-        for _ in range(self.cfg.decimation):
-            self._sim_step_counter += 1
-            # set actions into buffers
+        if self._physics_handles_decimation:
+            self._sim_step_counter += self.cfg.decimation
             self.action_manager.apply_action()
-            # set actions into simulator
             self.scene.write_data_to_sim()
-            # simulate
             self.sim.step(render=False)
-            # render between steps only if the GUI or an RTX sensor needs it.
-            # When render_enabled is False, Kit visualizer (camera/GUI) is skipped
-            # but standalone visualizers (Newton, Rerun, Viser) still update.
+            # render only when a render_interval boundary falls within this decimation block,
+            # mirroring the per-sub-step check in the else branch.
             if self._sim_step_counter % self.cfg.sim.render_interval == 0 and is_rendering:
                 self.sim.render(skip_app_pumping=not self.render_enabled)
-            # update buffers at sim dt
-            self.scene.update(dt=self.physics_dt)
+            self.scene.update(dt=self.step_dt)
+        else:
+            for _ in range(self.cfg.decimation):
+                self._sim_step_counter += 1
+                # set actions into buffers
+                self.action_manager.apply_action()
+                # set actions into simulator
+                self.scene.write_data_to_sim()
+                # simulate
+                self.sim.step(render=False)
+                # render between steps only if the GUI or an RTX sensor needs it.
+                # When render_enabled is False, Kit visualizer (camera/GUI) is skipped
+                # but standalone visualizers (Newton, Rerun, Viser) still update.
+                if self._sim_step_counter % self.cfg.sim.render_interval == 0 and is_rendering:
+                    self.sim.render(skip_app_pumping=not self.render_enabled)
+                # update buffers at sim dt
+                self.scene.update(dt=self.physics_dt)
 
         # post-step: step interval event
         if "interval" in self.event_manager.available_modes:

source/isaaclab/isaaclab/envs/manager_based_rl_env.py

@@ -199,22 +199,34 @@ def step(self, action: torch.Tensor) -> VecEnvStepReturn:
         is_rendering = self.sim.is_rendering
 
         # perform physics stepping
-        for _ in range(self.cfg.decimation):
-            self._sim_step_counter += 1
-            # set actions into buffers
+        if self._physics_handles_decimation:
+            self._sim_step_counter += self.cfg.decimation
             self.action_manager.apply_action()
-            # set actions into simulator
             self.scene.write_data_to_sim()
-            # simulate
             self.sim.step(render=False)
             self.recorder_manager.record_post_physics_decimation_step()
-            # render between steps only if the GUI or an RTX sensor needs it.
-            # When render_enabled is False, Kit visualizer (camera/GUI) is skipped
-            # but standalone visualizers (Newton, Rerun, Viser) still update.
+            # render only when a render_interval boundary falls within this decimation block,
+            # mirroring the per-sub-step check in the else branch.
             if self._sim_step_counter % self.cfg.sim.render_interval == 0 and is_rendering:
                 self.sim.render(skip_app_pumping=not self.render_enabled)
-            # update buffers at sim dt
-            self.scene.update(dt=self.physics_dt)
+            self.scene.update(dt=self.step_dt)
+        else:
+            for _ in range(self.cfg.decimation):
+                self._sim_step_counter += 1
+                # set actions into buffers
+                self.action_manager.apply_action()
+                # set actions into simulator
+                self.scene.write_data_to_sim()
+                # simulate
+                self.sim.step(render=False)
+                self.recorder_manager.record_post_physics_decimation_step()
+                # render between steps only if the GUI or an RTX sensor needs it.
+                # When render_enabled is False, Kit visualizer (camera/GUI) is skipped
+                # but standalone visualizers (Newton, Rerun, Viser) still update.
+                if self._sim_step_counter % self.cfg.sim.render_interval == 0 and is_rendering:
+                    self.sim.render(skip_app_pumping=not self.render_enabled)
+                # update buffers at sim dt
+                self.scene.update(dt=self.physics_dt)
 
         # post-step:
         # -- update env counters (used for curriculum generation)