isaac-sim · Mayankm96 · Oct 6, 2024 · Oct 6, 2024 · Oct 6, 2024 · Oct 6, 2024
@@ -21,7 +21,7 @@
 import omni.kit.app
 from omni.isaac.version import get_version
 
-from omni.isaac.lab.managers import EventManager
+from omni.isaac.lab.managers import EventManager, ManagerBase
 from omni.isaac.lab.scene import InteractiveScene
 from omni.isaac.lab.sim import SimulationContext
 from omni.isaac.lab.utils.noise import NoiseModel
@@ -113,6 +113,11 @@ def __init__(self, cfg: DirectMARLEnvCfg, render_mode: str | None = None, **kwar
             self._setup_scene()
         print("[INFO]: Scene manager: ", self.scene)
 
+        # randomization at scene level
+        if self.cfg.events:
+            with Timer("[INFO]: Time taken for scene randomization", "scene_randomization"):
+                self._apply_scene_randomization()
+
         # set up camera viewport controller
         # viewport is not available in other rendering modes so the function will throw a warning
         # FIXME: This needs to be fixed in the future when we unify the UI functionalities even for
@@ -563,6 +568,39 @@ def set_debug_vis(self, debug_vis: bool) -> bool:
     Helper functions.
     """
 
+    def _apply_scene_randomization(self):
+        """Apply scene-level randomization.
+
+        This function applies the scene-level randomization based on the configuration provided
+        to the event manager. Since the event manager is not initialized at this point, we mimic
+        the operations of the event manager to apply the scene-level randomization.
+
+        It must be called only before the simulation/physics is started.
+        """
+        # check if scene randomization is enabled
+        applied_scene_randomization = False
+        # iterate over all event terms
+        for term_name, term_cfg in self.cfg.events.__dict__.items():
+            # check for non config
+            if term_cfg is None:
+                continue
+            # call event terms corresponding to the scene-level randomization
+            if term_cfg.mode == "scene":
+                # resolve term config
+                ManagerBase.resolve_term_cfg(self, term_name, term_cfg, min_argc=2)
+                # enable scene randomization
+                applied_scene_randomization = True
+                # call the term
+                term_cfg.func(self, None, **term_cfg.params)
+
+        # warn the user that replicate physics may affect PhysX parsing
+        if self.scene.cfg.replicate_physics and applied_scene_randomization:
+            carb.log_warn(
+                "Replicate physics is enabled in the 'InteractiveScene' configuration."
+                " This may adversely affect PhysX parsing of scene information."
+                " We recommend disabling this property."
+            )
+
     def _configure_env_spaces(self):
         """Configure the spaces for the environment."""
         self.agents = self.cfg.possible_agents

@@ -21,7 +21,7 @@
 import omni.kit.app
 from omni.isaac.version import get_version
 
-from omni.isaac.lab.managers import EventManager
+from omni.isaac.lab.managers import EventManager, ManagerBase
 from omni.isaac.lab.scene import InteractiveScene
 from omni.isaac.lab.sim import SimulationContext
 from omni.isaac.lab.utils.noise import NoiseModel
@@ -118,6 +118,11 @@ def __init__(self, cfg: DirectRLEnvCfg, render_mode: str | None = None, **kwargs
             self._setup_scene()
         print("[INFO]: Scene manager: ", self.scene)
 
+        # randomization at scene level
+        if not self.cfg.events:
+            with Timer("[INFO]: Time taken for scene randomization", "scene_randomization"):
+                self._apply_scene_randomization()
+
         # set up camera viewport controller
         # viewport is not available in other rendering modes so the function will throw a warning
         # FIXME: This needs to be fixed in the future when we unify the UI functionalities even for
@@ -505,6 +510,39 @@ def set_debug_vis(self, debug_vis: bool) -> bool:
     Helper functions.
     """
 
+    def _apply_scene_randomization(self):
+        """Apply scene-level randomization.
+
+        This function applies the scene-level randomization based on the configuration provided
+        to the event manager. Since the event manager is not initialized at this point, we mimic
+        the operations of the event manager to apply the scene-level randomization.
+
+        It must be called only before the simulation/physics is started.
+        """
+        # check if scene randomization is enabled
+        applied_scene_randomization = False
+        # iterate over all event terms
+        for term_name, term_cfg in self.cfg.events.__dict__.items():
+            # check for non config
+            if term_cfg is None:
+                continue
+            # call event terms corresponding to the scene-level randomization
+            if term_cfg.mode == "scene":
+                # resolve term config
+                ManagerBase.resolve_term_cfg(self, term_name, term_cfg, min_argc=2)
+                # enable scene randomization
+                applied_scene_randomization = True
+                # call the term
+                term_cfg.func(self, None, **term_cfg.params)
+
+        # warn the user that replicate physics may affect PhysX parsing
+        if self.scene.cfg.replicate_physics and applied_scene_randomization:
+            carb.log_warn(
+                "Replicate physics is enabled in the 'InteractiveScene' configuration."
+                " This may adversely affect PhysX parsing of scene information."
+                " We recommend disabling this property."
+            )
+
     def _configure_gym_env_spaces(self):
         """Configure the action and observation spaces for the Gym environment."""
         # observation space (unbounded since we don't impose any limits)

@@ -11,7 +11,7 @@
 import carb
 import omni.isaac.core.utils.torch as torch_utils
 
-from omni.isaac.lab.managers import ActionManager, EventManager, ObservationManager
+from omni.isaac.lab.managers import ActionManager, EventManager, ManagerBase, ObservationManager
 from omni.isaac.lab.scene import InteractiveScene
 from omni.isaac.lab.sim import SimulationContext
 from omni.isaac.lab.utils.timer import Timer
@@ -116,6 +116,10 @@ def __init__(self, cfg: ManagerBasedEnvCfg):
             self.scene = InteractiveScene(self.cfg.scene)
         print("[INFO]: Scene manager: ", self.scene)
 
+        # randomization at scene level
+        with Timer("[INFO]: Time taken for scene randomization", "scene_randomization"):
+            self._apply_scene_randomization()
+
         # set up camera viewport controller
         # viewport is not available in other rendering modes so the function will throw a warning
         # FIXME: This needs to be fixed in the future when we unify the UI functionalities even for
@@ -346,6 +350,39 @@ def close(self):
     Helper functions.
     """
 
+    def _apply_scene_randomization(self):
+        """Apply scene-level randomization.
+
+        This function applies the scene-level randomization based on the configuration provided
+        to the event manager. Since the event manager is not initialized at this point, we mimic
+        the operations of the event manager to apply the scene-level randomization.
+
+        It must be called only before the simulation/physics is started.
+        """
+        # check if scene randomization is enabled
+        applied_scene_randomization = False
+        # iterate over all event terms
+        for term_name, term_cfg in self.cfg.events.__dict__.items():
+            # check for non config
+            if term_cfg is None:
+                continue
+            # call event terms corresponding to the scene-level randomization
+            if term_cfg.mode == "scene":
+                # resolve term config
+                ManagerBase.resolve_term_cfg(self, term_name, term_cfg, min_argc=2)
+                # enable scene randomization
+                applied_scene_randomization = True
+                # call the term
+                term_cfg.func(self, None, **term_cfg.params)
+
+        # warn the user that replicate physics may affect PhysX parsing
+        if self.scene.cfg.replicate_physics and applied_scene_randomization:
+            carb.log_warn(
+                "Replicate physics is enabled in the 'InteractiveScene' configuration."
+                " This may adversely affect PhysX parsing of scene information."
+                " We recommend disabling this property."
+            )
+
     def _reset_idx(self, env_ids: Sequence[int]):
         """Reset environments based on specified indices.
 

@@ -19,6 +19,8 @@
 
 import carb
 import omni.physics.tensors.impl.api as physx
+import omni.usd
+from pxr import Gf, Sdf
 
 import omni.isaac.lab.sim as sim_utils
 import omni.isaac.lab.utils.math as math_utils
@@ -31,6 +33,104 @@
     from omni.isaac.lab.envs import ManagerBasedEnv
 
 
+def randomize_shape_color(env: ManagerBasedEnv, env_ids: torch.Tensor | None, asset_cfg: SceneEntityCfg):
+    """Randomize the color of a shape.
+
+    This function randomizes the color of USD shapes created using :class:`omni.isaac.lab.sim.spawn.ShapeCfg`
+    class. It modifies the attribute: "geometry/material/Shader.inputs:diffuseColor" under the prims
+    corresponding to the asset. If the attribute does not exist, the function errors out.
+    """
+    # extract the used quantities (to enable type-hinting)
+    asset: RigidObject = env.scene[asset_cfg.name]
+
+    # resolve environment ids
+    if env_ids is None:
+        env_ids = torch.arange(env.scene.num_envs, device="cpu")
+    else:
+        env_ids = env_ids.cpu()
+
+    # acquire stage
+    stage = omni.usd.get_context().get_stage()
+    # resolve prim paths for spawning and cloning
+    prim_paths = sim_utils.find_matching_prim_paths(asset.cfg.prim_path)
+
+    # sample values
+    rand_samples = math_utils.sample_uniform(0.0, 1.0, (len(env_ids), 3), device="cpu").tolist()
+
+    # use sdf changeblock for faster processing of USD properties
+    with Sdf.ChangeBlock():
+        for i, env_id in enumerate(env_ids):
+            # spawn single instance
+            prim_spec = Sdf.CreatePrimInLayer(stage.GetRootLayer(), prim_paths[env_id])
+
+            # get the attribute to randomize
+            color_spec = prim_spec.GetAttributeAtPath(
+                prim_paths[env_id] + "/geometry/material/Shader.inputs:diffuseColor"
+            )
+            color_spec.default = Gf.Vec3f(*rand_samples[i])
+
+
+def randomize_scale(
+    env: ManagerBasedEnv,
+    env_ids: torch.Tensor | None,
+    scale_range: tuple[float, float] | dict[str, tuple[float, float]],
+    asset_cfg: SceneEntityCfg,
+):
+    """Randomize the scale of an asset along all dimensions.
+
+    This function modifies the "xformOp:scale" property of all the prims corresponding to the asset.
+
+    It takes a tuple or dictionary for the scale ranges. If it is a tuple, then the scaling along
+    individual axis is performed equally. If it is a dictionary, the scaling is independent across each dimension.
+
+    The keys of the dictionary are ``x``, ``y``, and ``z``. The values are tuples of the form ``(min, max)``.
+    If the dictionary does not contain a key, the range is set to one for that axis.
+
+    .. attention::
+        Since this function modifies USD properties, it should only be used before the simulation starts
+        playing. This corresponds to the event mode named "spawn". Using it at simulation time, may lead to
+        unpredictable behaviors.
+
+    .. note::
+        When randomizing the scale of individual assets, please make sure to set
+        :attr:`omni.isaac.lab.scene.InteractiveSceneCfg.replicate_physics` to False. This ensures that physics
+        parser will parse the individual asset properties separately.
+
+    """
+    # extract the used quantities (to enable type-hinting)
+    asset: RigidObject | Articulation = env.scene[asset_cfg.name]
+
+    # resolve environment ids
+    if env_ids is None:
+        env_ids = torch.arange(env.scene.num_envs, device="cpu")
+    else:
+        env_ids = env_ids.cpu()
+
+    # acquire stage
+    stage = omni.usd.get_context().get_stage()
+    # resolve prim paths for spawning and cloning
+    prim_paths = sim_utils.find_matching_prim_paths(asset.cfg.prim_path)
+
+    # sample scale values
+    if isinstance(scale_range, dict):
+        range_list = [scale_range.get(key, (1.0, 1.0)) for key in ["x", "y", "z"]]
+        ranges = torch.tensor(range_list, device="cpu")
+        rand_samples = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(env_ids), 3), device="cpu").tolist()
+    else:
+        rand_samples = math_utils.sample_uniform(*scale_range, (len(env_ids), 1), device="cpu")
+        rand_samples = rand_samples.repeat(1, 3).tolist()
+
+    # use sdf changeblock for faster processing of USD properties
+    with Sdf.ChangeBlock():
+        for i, env_id in enumerate(env_ids):
+            # spawn single instance
+            prim_spec = Sdf.CreatePrimInLayer(stage.GetRootLayer(), prim_paths[env_id])
+
+            # get the attribute to randomize
+            scale_spec = prim_spec.GetAttributeAtPath(prim_paths[env_id] + ".xformOp:scale")
+            scale_spec.default = Gf.Vec3f(*rand_samples[i])
+
+
 class randomize_rigid_body_material(ManagerTermBase):
     """Randomize the physics materials on all geometries of the asset.
 

@@ -38,7 +38,8 @@ class EventManager(ManagerBase):
 
     For a typical training process, you may want to apply events in the following modes:
 
-    - "startup": Event is applied once at the beginning of the training.
+    - "scene": Event is applied once after designing the scene.
+    - "startup": Event is applied once after the simulation starts playing.
     - "reset": Event is applied at every reset.
     - "interval": Event is applied at pre-specified intervals of time.
 
@@ -52,9 +53,6 @@ class EventManager(ManagerBase):
 
     """
 
-    _env: ManagerBasedEnv
-    """The environment instance."""
-
     def __init__(self, cfg: object, env: ManagerBasedEnv):
         """Initialize the event manager.