Replaces randomization of texture to functional (#452)

Enables the new replicator functional API path for color and texture
randomization for Isaac Sim 5.0. For backwards compatibility with Isaac
Sim 4.5, we maintain the previous omni.graph code path for 4.5 use.

- New feature (non-breaking change which adds functionality)

- [x] I have run the [`pre-commit` checks](https://pre-commit.com/) with
`./isaaclab.sh --format`
- [x] have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [ ] I have updated the changelog and the corresponding version in the
extension's `config/extension.toml` file
- [ ] I have added my name to the `CONTRIBUTORS.md` or my name already
exists there

<!--
As you go through the checklist above, you can mark something as done by
putting an x character in it

For example,
- [x] I have done this task
- [ ] I have not done this task
-->

---------
Signed-off-by: jiehanw <jiehanw@nvidia.com>
Co-authored-by: Kelly Guo <kellyg@nvidia.com>

source/isaaclab/config/extension.toml

 [package]
 
 # Note: Semantic Versioning is used: https://semver.org/
-version = "0.42.22"
+version = "0.42.23"
 
 # Description
 title = "Isaac Lab framework for Robot Learning"

source/isaaclab/docs/CHANGELOG.rst

 Changelog
 ---------
 
-0.42.22 (2025-06-25)
+0.42.23 (2025-06-25)
 ~~~~~~~~~~~~~~~~~~~~
 
 Added
@@ -12,7 +12,7 @@ Added
   env instance
 
 
-0.42.21 (2025-07-11)
+0.42.22 (2025-07-11)
 ~~~~~~~~~~~~~~~~~~~~
 
 Fixed
@@ -22,7 +22,7 @@ Fixed
 restricting the resetting joint indices be that user defined joint indices.
 
 
-0.42.20 (2025-07-11)
+0.42.21 (2025-07-11)
 ~~~~~~~~~~~~~~~~~~~~
 
 Fixed
@@ -32,7 +32,7 @@ Fixed
   env_ids are passed.
 
 
-0.42.19 (2025-07-09)
+0.42.20 (2025-07-09)
 ~~~~~~~~~~~~~~~~~~~~
 
 Added
@@ -49,7 +49,7 @@ Fixed
   buffer on recording.
 
 
-0.42.18 (2025-07-10)
+0.42.19 (2025-07-10)
 ~~~~~~~~~~~~~~~~~~~~
 
 Added
@@ -80,6 +80,15 @@ Changed
 * Changed the implementation of :func:`~isaaclab.utils.math.copysign` to better reflect the documented functionality.
 
 
+0.42.18 (2025-07-09)
+~~~~~~~~~~~~~~~~~~~~
+
+Changed
+^^^^^^^
+
+* Changed texture and color randomization to use new replicator functional APIs.
+
+
 0.42.17 (2025-07-08)
 ~~~~~~~~~~~~~~~~~~~~
 
@@ -119,7 +128,7 @@ Fixed
 * Fixed unittest tests that are floating inside pytests for articulation and rendering
 
 
-0.42.13 (2025-07-03)
+0.42.13 (2025-07-07)
 ~~~~~~~~~~~~~~~~~~~~
 
 Changed

source/isaaclab/isaaclab/envs/mdp/events.py

@@ -15,6 +15,7 @@ the event introduced by the function.
 from __future__ import annotations
 
 import math
+import re
 import torch
 from typing import TYPE_CHECKING, Literal
 
@@ -30,6 +31,7 @@ from isaaclab.actuators import ImplicitActuator
 from isaaclab.assets import Articulation, DeformableObject, RigidObject
 from isaaclab.managers import EventTermCfg, ManagerTermBase, SceneEntityCfg
 from isaaclab.terrains import TerrainImporter
+from isaaclab.utils.version import compare_versions
 
 if TYPE_CHECKING:
     from isaaclab.envs import ManagerBasedEnv
@@ -1204,17 +1206,6 @@ class randomize_visual_texture_material(ManagerTermBase):
         """
         super().__init__(cfg, env)
 
-        # enable replicator extension if not already enabled
-        enable_extension("omni.replicator.core")
-        # we import the module here since we may not always need the replicator
-        import omni.replicator.core as rep
-
-        # read parameters from the configuration
-        asset_cfg: SceneEntityCfg = cfg.params.get("asset_cfg")
-        texture_paths = cfg.params.get("texture_paths")
-        event_name = cfg.params.get("event_name")
-        texture_rotation = cfg.params.get("texture_rotation", (0.0, 0.0))
-
         # check to make sure replicate_physics is set to False, else raise error
         # note: We add an explicit check here since texture randomization can happen outside of 'prestartup' mode
         #   and the event manager doesn't check in that case.
@@ -1225,8 +1216,14 @@ class randomize_visual_texture_material(ManagerTermBase):
                 " by setting 'replicate_physics' to False in 'InteractiveSceneCfg'."
             )
 
-        # convert from radians to degrees
-        texture_rotation = tuple(math.degrees(angle) for angle in texture_rotation)
+        # enable replicator extension if not already enabled
+        enable_extension("omni.replicator.core")
+
+        # we import the module here since we may not always need the replicator
+        import omni.replicator.core as rep
+
+        # read parameters from the configuration
+        asset_cfg: SceneEntityCfg = cfg.params.get("asset_cfg")
 
         # obtain the asset entity
         asset = env.scene[asset_cfg.name]
@@ -1244,7 +1241,6 @@ class randomize_visual_texture_material(ManagerTermBase):
         # Check if the pattern with '/visuals' yields results when matching `body_names_regex`.
         # If not, fall back to a broader pattern without '/visuals'.
         asset_main_prim_path = asset.cfg.prim_path
-        # Try the pattern with '/visuals' first for the generic case
         pattern_with_visuals = f"{asset_main_prim_path}/{body_names_regex}/visuals"
         # Use sim_utils to check if any prims currently match this pattern
         matching_prims = sim_utils.find_matching_prim_paths(pattern_with_visuals)
@@ -1261,19 +1257,52 @@ class randomize_visual_texture_material(ManagerTermBase):
                 " randomization."
             )
 
-        # Create the omni-graph node for the randomization term
-        def rep_texture_randomization():
-            prims_group = rep.get.prims(path_pattern=prim_path)
+        # extract the replicator version
+        version = re.match(r"^(\d+\.\d+\.\d+)", rep.__file__.split("/")[-5][21:]).group(1)
 
-            with prims_group:
-                rep.randomizer.texture(
-                    textures=texture_paths, project_uvw=True, texture_rotate=rep.distribution.uniform(*texture_rotation)
-                )
-            return prims_group.node
+        # use different path for different version of replicator
+        if compare_versions(version, "1.12.4") < 0:
+            texture_paths = cfg.params.get("texture_paths")
+            event_name = cfg.params.get("event_name")
+            texture_rotation = cfg.params.get("texture_rotation", (0.0, 0.0))
+
+            # convert from radians to degrees
+            texture_rotation = tuple(math.degrees(angle) for angle in texture_rotation)
+
+            # Create the omni-graph node for the randomization term
+            def rep_texture_randomization():
+                prims_group = rep.get.prims(path_pattern=prim_path)
 
-        # Register the event to the replicator
-        with rep.trigger.on_custom_event(event_name=event_name):
-            rep_texture_randomization()
+                with prims_group:
+                    rep.randomizer.texture(
+                        textures=texture_paths,
+                        project_uvw=True,
+                        texture_rotate=rep.distribution.uniform(*texture_rotation),
+                    )
+                return prims_group.node
+
+            # Register the event to the replicator
+            with rep.trigger.on_custom_event(event_name=event_name):
+                rep_texture_randomization()
+        else:
+            # acquire stage
+            stage = get_current_stage()
+            prims_group = rep.functional.get.prims(path_pattern=prim_path, stage=stage)
+
+            num_prims = len(prims_group)
+            # rng that randomizes the texture and rotation
+            self.texture_rng = rep.rng.ReplicatorRNG()
+
+            # Create the material first and bind it to the prims
+            for i, prim in enumerate(prims_group):
+                # Disable instancble
+                if prim.IsInstanceable():
+                    prim.SetInstanceable(False)
+
+            # TODO: Should we specify the value when creating the material?
+            self.material_prims = rep.functional.create_batch.material(
+                mdl="OmniPBR.mdl", bind_prims=prims_group, count=num_prims, project_uvw=True
+            )
 
     def __call__(
         self,
@@ -1284,13 +1313,36 @@ class randomize_visual_texture_material(ManagerTermBase):
         texture_paths: list[str],
         texture_rotation: tuple[float, float] = (0.0, 0.0),
     ):
-        # import replicator
-        import omni.replicator.core as rep
-
-        # only send the event to the replicator
         # note: This triggers the nodes for all the environments.
         #   We need to investigate how to make it happen only for a subset based on env_ids.
-        rep.utils.send_og_event(event_name)
+        # we import the module here since we may not always need the replicator
+        import omni.replicator.core as rep
+
+        # extract the replicator version
+        version = re.match(r"^(\d+\.\d+\.\d+)", rep.__file__.split("/")[-5][21:]).group(1)
+
+        # use different path for different version of replicator
+        if compare_versions(version, "1.12.4") < 0:
+            rep.utils.send_og_event(event_name)
+        else:
+            # read parameters from the configuration
+            texture_paths = texture_paths if texture_paths else self._cfg.params.get("texture_paths")
+            texture_rotation = (
+                texture_rotation if texture_rotation else self._cfg.params.get("texture_rotation", (0.0, 0.0))
+            )
+
+            # convert from radians to degrees
+            texture_rotation = tuple(math.degrees(angle) for angle in texture_rotation)
+
+            num_prims = len(self.material_prims)
+            random_textures = self.texture_rng.generator.choice(texture_paths, size=num_prims)
+            random_rotations = self.texture_rng.generator.uniform(
+                texture_rotation[0], texture_rotation[1], size=num_prims
+            )
+
+            # modify the material properties
+            rep.functional.modify.attribute(self.material_prims, "diffuse_texture", random_textures)
+            rep.functional.modify.attribute(self.material_prims, "texture_rotate", random_rotations)
 
 
 class randomize_visual_color(ManagerTermBase):
@@ -1317,7 +1369,12 @@ class randomize_visual_color(ManagerTermBase):
     """
 
     def __init__(self, cfg: EventTermCfg, env: ManagerBasedEnv):
-        """Initialize the randomization term."""
+        """Initialize the randomization term.
+
+        Args:
+            cfg: The configuration of the event term.
+            env: The environment instance.
+        """
         super().__init__(cfg, env)
 
         # enable replicator extension if not already enabled
@@ -1327,8 +1384,6 @@ class randomize_visual_color(ManagerTermBase):
 
         # read parameters from the configuration
         asset_cfg: SceneEntityCfg = cfg.params.get("asset_cfg")
-        colors = cfg.params.get("colors")
-        event_name = cfg.params.get("event_name")
         mesh_name: str = cfg.params.get("mesh_name", "")  # type: ignore
 
         # check to make sure replicate_physics is set to False, else raise error
@@ -1350,27 +1405,51 @@ class randomize_visual_color(ManagerTermBase):
         mesh_prim_path = f"{asset.cfg.prim_path}{mesh_name}"
         # TODO: Need to make it work for multiple meshes.
 
-        # parse the colors into replicator format
-        if isinstance(colors, dict):
-            # (r, g, b) - low, high --> (low_r, low_g, low_b) and (high_r, high_g, high_b)
-            color_low = [colors[key][0] for key in ["r", "g", "b"]]
-            color_high = [colors[key][1] for key in ["r", "g", "b"]]
-            colors = rep.distribution.uniform(color_low, color_high)
-        else:
-            colors = list(colors)
+        # extract the replicator version
+        version = re.match(r"^(\d+\.\d+\.\d+)", rep.__file__.split("/")[-5][21:]).group(1)
+
+        # use different path for different version of replicator
+        if compare_versions(version, "1.12.4") < 0:
+            colors = cfg.params.get("colors")
+            event_name = cfg.params.get("event_name")
+
+            # parse the colors into replicator format
+            if isinstance(colors, dict):
+                # (r, g, b) - low, high --> (low_r, low_g, low_b) and (high_r, high_g, high_b)
+                color_low = [colors[key][0] for key in ["r", "g", "b"]]
+                color_high = [colors[key][1] for key in ["r", "g", "b"]]
+                colors = rep.distribution.uniform(color_low, color_high)
+            else:
+                colors = list(colors)
+
+            # Create the omni-graph node for the randomization term
+            def rep_color_randomization():
+                prims_group = rep.get.prims(path_pattern=mesh_prim_path)
+                with prims_group:
+                    rep.randomizer.color(colors=colors)
 
-        # Create the omni-graph node for the randomization term
-        def rep_texture_randomization():
-            prims_group = rep.get.prims(path_pattern=mesh_prim_path)
+                return prims_group.node
 
-            with prims_group:
-                rep.randomizer.color(colors=colors)
+            # Register the event to the replicator
+            with rep.trigger.on_custom_event(event_name=event_name):
+                rep_color_randomization()
+        else:
+            stage = get_current_stage()
+            prims_group = rep.functional.get.prims(path_pattern=mesh_prim_path, stage=stage)
 
-            return prims_group.node
+            num_prims = len(prims_group)
+            self.color_rng = rep.rng.ReplicatorRNG()
 
-        # Register the event to the replicator
-        with rep.trigger.on_custom_event(event_name=event_name):
-            rep_texture_randomization()
+            # Create the material first and bind it to the prims
+            for i, prim in enumerate(prims_group):
+                # Disable instancble
+                if prim.IsInstanceable():
+                    prim.SetInstanceable(False)
+
+            # TODO: Should we specify the value when creating the material?
+            self.material_prims = rep.functional.create_batch.material(
+                mdl="OmniPBR.mdl", bind_prims=prims_group, count=num_prims, project_uvw=True
+            )
 
     def __call__(
         self,
@@ -1381,11 +1460,33 @@ class randomize_visual_color(ManagerTermBase):
         colors: list[tuple[float, float, float]] | dict[str, tuple[float, float]],
         mesh_name: str = "",
     ):
-        # import replicator
+        # note: This triggers the nodes for all the environments.
+        #   We need to investigate how to make it happen only for a subset based on env_ids.
+
+        # we import the module here since we may not always need the replicator
         import omni.replicator.core as rep
 
-        # only send the event to the replicator
-        rep.utils.send_og_event(event_name)
+        version = re.match(r"^(\d+\.\d+\.\d+)", rep.__file__.split("/")[-5][21:]).group(1)
+
+        # use different path for different version of replicator
+        if compare_versions(version, "1.12.4") < 0:
+            rep.utils.send_og_event(event_name)
+        else:
+            colors = colors if colors else self._cfg.params.get("colors")
+
+            # parse the colors into replicator format
+            if isinstance(colors, dict):
+                # (r, g, b) - low, high --> (low_r, low_g, low_b) and (high_r, high_g, high_b)
+                color_low = [colors[key][0] for key in ["r", "g", "b"]]
+                color_high = [colors[key][1] for key in ["r", "g", "b"]]
+                colors = [color_low, color_high]
+            else:
+                colors = list(colors)
+
+            num_prims = len(self.material_prims)
+            random_colors = self.color_rng.generator.uniform(colors[0], colors[1], size=(num_prims, 3))
+
+            rep.functional.modify.attribute(self.material_prims, "diffuse_color_constant", random_colors)
 
 
 """

source/isaaclab/isaaclab/utils/__init__.py

@@ -14,3 +14,4 @@ from .modifiers import *
 from .string import *
 from .timer import Timer
 from .types import *
+from .version import *

source/isaaclab/isaaclab/utils/version.py

+# Copyright (c) 2022-2025, The Isaac Lab Project Developers (https://github.com/isaac-sim/IsaacLab/blob/main/CONTRIBUTORS.md).
+# All rights reserved.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""Utility function for version comparison."""
+
+
+def compare_versions(v1: str, v2: str) -> int:
+    parts1 = list(map(int, v1.split(".")))
+    parts2 = list(map(int, v2.split(".")))
+
+    # Pad the shorter version with zeros (e.g. 1.2 vs 1.2.0)
+    length = max(len(parts1), len(parts2))
+    parts1 += [0] * (length - len(parts1))
+    parts2 += [0] * (length - len(parts2))
+
+    if parts1 > parts2:
+        return 1  # v1 is greater
+    elif parts1 < parts2:
+        return -1  # v2 is greater
+    else:
+        return 0  # versions are equal

source/isaaclab/test/envs/test_color_randomization.py

+# Copyright (c) 2022-2025, The Isaac Lab Project Developers (https://github.com/isaac-sim/IsaacLab/blob/main/CONTRIBUTORS.md).
+# All rights reserved.
+#
+# SPDX-License-Identifier: BSD-3-Clause
+
+"""
+This script tests the functionality of texture randomization applied to the cartpole scene.
+"""
+
+"""Launch Isaac Sim Simulator first."""
+
+from isaaclab.app import AppLauncher
+
+# launch omniverse app
+app_launcher = AppLauncher(headless=True, enable_cameras=True)
+simulation_app = app_launcher.app
+
+"""Rest everything follows."""
+
+import math
+import torch
+
+import omni.usd
+import pytest
+from isaacsim.core.version import get_version
+
+import isaaclab.envs.mdp as mdp
+from isaaclab.envs import ManagerBasedEnv, ManagerBasedEnvCfg
+from isaaclab.managers import EventTermCfg as EventTerm
+from isaaclab.managers import ObservationGroupCfg as ObsGroup
+from isaaclab.managers import ObservationTermCfg as ObsTerm
+from isaaclab.managers import SceneEntityCfg
+from isaaclab.utils import configclass
+
+from isaaclab_tasks.manager_based.classic.cartpole.cartpole_env_cfg import CartpoleSceneCfg
+
+
+@configclass
+class ActionsCfg:
+    """Action specifications for the environment."""
+
+    joint_efforts = mdp.JointEffortActionCfg(asset_name="robot", joint_names=["slider_to_cart"], scale=5.0)
+
+
+@configclass
+class ObservationsCfg:
+    """Observation specifications for the environment."""
+
+    @configclass
+    class PolicyCfg(ObsGroup):
+        """Observations for policy group."""
+
+        # observation terms (order preserved)
+        joint_pos_rel = ObsTerm(func=mdp.joint_pos_rel)
+        joint_vel_rel = ObsTerm(func=mdp.joint_vel_rel)
+
+        def __post_init__(self) -> None:
+            self.enable_corruption = False
+            self.concatenate_terms = True
+
+    # observation groups
+    policy: PolicyCfg = PolicyCfg()
+
+
+@configclass
+class EventCfg:
+    """Configuration for events."""
+
+    # on prestartup apply a new set of textures
+    # note from @mayank: Changed from 'reset' to 'prestartup' to make test pass.
+    #   The error happens otherwise on Kit thread which is not the main thread.
+    cart_texture_randomizer = EventTerm(
+        func=mdp.randomize_visual_color,
+        mode="prestartup",
+        params={
+            "asset_cfg": SceneEntityCfg("robot", body_names=["cart"]),
+            "colors": {"r": (0.0, 1.0), "g": (0.0, 1.0), "b": (0.0, 1.0)},
+            "event_name": "cart_color_randomizer",
+        },
+    )
+
+    # on reset apply a new set of textures
+    pole_texture_randomizer = EventTerm(
+        func=mdp.randomize_visual_color,
+        mode="reset",
+        params={
+            "asset_cfg": SceneEntityCfg("robot", body_names=["pole"]),
+            "colors": {"r": (0.0, 1.0), "g": (0.0, 1.0), "b": (0.0, 1.0)},
+            "event_name": "pole_color_randomizer",
+        },
+    )
+
+    reset_cart_position = EventTerm(
+        func=mdp.reset_joints_by_offset,
+        mode="reset",
+        params={
+            "asset_cfg": SceneEntityCfg("robot", joint_names=["slider_to_cart"]),
+            "position_range": (-1.0, 1.0),
+            "velocity_range": (-0.1, 0.1),
+        },
+    )
+
+    reset_pole_position = EventTerm(
+        func=mdp.reset_joints_by_offset,
+        mode="reset",
+        params={
+            "asset_cfg": SceneEntityCfg("robot", joint_names=["cart_to_pole"]),
+            "position_range": (-0.125 * math.pi, 0.125 * math.pi),
+            "velocity_range": (-0.01 * math.pi, 0.01 * math.pi),
+        },
+    )
+
+
+@configclass
+class CartpoleEnvCfg(ManagerBasedEnvCfg):
+    """Configuration for the cartpole environment."""
+
+    # Scene settings
+    scene = CartpoleSceneCfg(env_spacing=2.5)
+
+    # Basic settings
+    actions = ActionsCfg()
+    observations = ObservationsCfg()
+    events = EventCfg()
+
+    def __post_init__(self):
+        """Post initialization."""
+        # viewer settings
+        self.viewer.eye = [4.5, 0.0, 6.0]
+        self.viewer.lookat = [0.0, 0.0, 2.0]
+        # step settings
+        self.decimation = 4  # env step every 4 sim steps: 200Hz / 4 = 50Hz
+        # simulation settings
+        self.sim.dt = 0.005  # sim step every 5ms: 200Hz
+
+
+@pytest.mark.parametrize("device", ["cpu", "cuda"])
+def test_color_randomization(device):
+    """Test color randomization for cartpole environment."""
+    # skip test if stage in memory is not supported
+    isaac_sim_version = float(".".join(get_version()[2]))
+    if isaac_sim_version < 5:
+        pytest.skip("Color randomization test hangs in this version of Isaac Sim")
+
+    # Create a new stage
+    omni.usd.get_context().new_stage()
+
+    try:
+        # Set the arguments
+        env_cfg = CartpoleEnvCfg()
+        env_cfg.scene.num_envs = 16
+        env_cfg.scene.replicate_physics = False
+        env_cfg.sim.device = device
+
+        # Setup base environment
+        env = ManagerBasedEnv(cfg=env_cfg)
+
+        try:
+            # Simulate physics
+            with torch.inference_mode():
+                for count in range(50):
+                    # Reset every few steps to check nothing breaks
+                    if count % 10 == 0:
+                        env.reset()
+                    # Sample random actions
+                    joint_efforts = torch.randn_like(env.action_manager.action)
+                    # Step the environment
+                    env.step(joint_efforts)
+        finally:
+            env.close()
+    finally:
+        # Clean up stage
+        omni.usd.get_context().close_stage()