add documentations and tests

Author: ooctipus

source/isaaclab/isaaclab/envs/mdp/curriculums.py

@@ -46,7 +46,33 @@ class modify_env_param(ManagerTermBase):
     `cfg.params["address"]`) the first time it is called, then on each invocation
     reads the current value, applies a user-provided `modify_fn`, and writes back
     the result.
+    
+    usage example, modify physics material bucket:
+
+        def resample_bucket_range(
+            env, env_id, data, static_friction_range, dynamic_friction_range, restitution_range, num_steps
+        ):
+            if env.common_step_counter > num_steps:
+                range_list = [static_friction_range, dynamic_friction_range, restitution_range]
+                ranges = torch.tensor(range_list, device="cpu")
+                new_buckets = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(data), 3), device="cpu")
+                return new_buckets
+
+        object_physics_material_curriculum = CurrTerm(
+                func=mdp.modify_env_param,
+                params={
+                    "address": "event_manager.cfg.object_physics_material.func.material_buckets",
+                    "modify_fn": resample_bucket_range,
+                    "modify_params": {
+                        "static_friction_range": [.5, 1.],
+                        "dynamic_friction_range": [.3, 1.],
+                        "restitution_range": [0.0, 0.5],
+                        "num_step": 120000
+                    }
+                }
+            )
     """
+    NO_CHANGE = object()
 
     def __init__(self, cfg, env):
         """
@@ -91,7 +117,8 @@ def __call__(
 
         data = self.get_fn()
         new_val = modify_fn(self._env, env_ids, data, **modify_params)
-        self.set_fn(new_val)
+        if new_val is not self.NO_CHANGE:
+            self.set_fn(new_val)
 
     def _compile_accessors(self, root, path: str):
         """
@@ -163,6 +190,21 @@ class modify_term_cfg(modify_env_param):
     with "_manager.cfg.", and then behaves identically to ModifyEnvParam.
 
     for example: command_manager.cfg.object_pose.ranges.xpos -> commands.object_pose.ranges.xpos
+    
+    usage example:
+
+        def update_value_after_step(env, env_ids, data, value, num_steps):
+            if env.common_step_counter > num_steps:
+                return value
+        
+        command_object_pose_xrange_adr = CurrTerm(
+            func=mdp.modify_term_cfg,
+            params={
+                "address": "commands.object_pose.ranges.pos_x",   # note that `_manager.cfg` is omitted
+                "modify_fn": mdp.update_value_after_step,
+                "modify_params": {"value": (-.75, -.25), "num_steps": 12000}
+            }
+        )
     """
 
     def __init__(self, cfg, env):

source/isaaclab/test/envs/test_modify_env_param_curr_term.py

@@ -0,0 +1,109 @@
+# 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 unittest
+from unittest.mock import patch
+
+import omni.usd
+
+import isaaclab.envs.mdp as mdp
+import isaaclab.utils.math as math_utils
+from isaaclab.envs import ManagerBasedRLEnv
+from isaaclab_tasks.manager_based.classic.cartpole.cartpole_env_cfg import CartpoleEnvCfg
+from isaaclab.managers import CurriculumTermCfg as CurrTerm
+from isaaclab.utils import configclass
+from isaaclab.utils.assets import NVIDIA_NUCLEUS_DIR
+
+
+def resample_bucket_range(
+    env, env_id, data, static_friction_range, dynamic_friction_range, restitution_range, num_steps
+):
+    if env.common_step_counter > num_steps:
+        range_list = [static_friction_range, dynamic_friction_range, restitution_range]
+        ranges = torch.tensor(range_list, device="cpu")
+        new_buckets = math_utils.sample_uniform(ranges[:, 0], ranges[:, 1], (len(data), 3), device="cpu")
+        return new_buckets
+
+
+def replace_term(
+    env, env_id, data, term, num_steps
+):
+    if env.common_step_counter > num_steps and data != term:
+        return term
+    return mdp.modify_env_param.NO_CHANGE
+
+
+@configclass
+class CurriculumsCfg:
+    modify_observation_joint_pos = CurrTerm(
+        func=mdp.modify_term_cfg,
+        params={
+            "address": "observations.policy.joint_pos_rel.func",
+            "modify_fn": replace_term,
+            "modify_params": {
+                "term": mdp.joint_pos,
+                "num_steps": 1
+            }
+        }
+    )
+
+class TestCurriculumModifyEnvParam(unittest.TestCase):
+    """Test for texture randomization"""
+
+    """
+    Tests
+    """
+
+    def test_curriculum_modify_env_param(self):
+        """Test texture randomization fallback mechanism when /visuals pattern doesn't match."""
+
+        for device in ["cpu", "cuda"]:
+            with self.subTest(device=device):
+                # create a new stage
+                omni.usd.get_context().new_stage()
+
+                # set the arguments - use fallback config
+                env_cfg = CartpoleEnvCfg()
+                env_cfg.scene.num_envs = 16
+                env_cfg.scene.replicate_physics = False
+                env_cfg.curriculum = CurriculumsCfg()
+                env_cfg.sim.device = device
+
+                # This should trigger the fallback mechanism and log the fallback message
+                env = ManagerBasedRLEnv(cfg=env_cfg)
+
+                # simulate physics
+                with torch.inference_mode():
+                    for count in range(2):  # shorter test for fallback
+                        # reset every few steps to check nothing breaks
+                        env.reset()
+                        # sample random actions
+                        joint_efforts = torch.randn_like(env.action_manager.action)
+                        # step the environment
+                        env.step(joint_efforts)
+                        
+                        if count == 0:
+                            assert env.observation_manager.cfg.policy.joint_pos_rel.func == mdp.joint_pos_rel
+                        
+                        if count == 1:
+                            assert env.observation_manager.cfg.policy.joint_pos_rel.func == mdp.joint_pos
+
+                env.close()