physics_modifier_function_wrapper

Module: GBC.gyms.isaaclab_45.managers.physics_modifier_function_wrapper

This module provides a sophisticated decorator system for physics modifier functions that enables dynamic parameter updates and adaptive curriculum learning strategies. The @update decorator transforms static physics modifier functions into intelligent, adaptive components that can modify their behavior based on training progress, performance metrics, and environmental conditions.

🎯 Core Functionality

The physics_modifier_function_wrapper provides:

• 🔄 Dynamic Parameter Override: Real-time modification of function parameters without code changes
• 📈 Adaptive Strategies: Custom update strategies for sophisticated curriculum learning
• 🧠 Stateful Updates: Maintains parameter state across function calls for progressive learning
• 🛡️ Robust Error Handling: Graceful fallback and parameter validation
• 🔧 Function Introspection: Intelligent parameter binding and signature management
• ⚡ Performance Optimization: Minimal overhead for high-frequency physics modifications

---

📚 Dependencies

import inspect
import functools
from collections.abc import Callable

---

🎨 @update Decorator

Module Name: GBC.gyms.isaaclab_45.managers.physics_modifier_function_wrapper.update

Decorator Signature:

def update(update_strategy: Callable | None = None, **kwargs):
    """Decorator to update the physics modifier function.

    Args:
        update_strategy: The update strategy to use. If None, the function will be updated by default
        **kwargs: Additional arguments to pass to the update strategy.

    Returns:
        A decorator that updates the physics modifier function.
    """

🔧 Decorator Parameters

update_strategy (Callable | None)

• Purpose: Custom function that defines how parameters should be updated
• Signature: update_strategy(current_overrides: dict, new_kwargs: dict) -> dict
• Default Behavior: Simple dictionary update when None
• Advanced Use: Complex curriculum learning strategies, performance-based adaptation

**kwargs (dict)

• Purpose: Additional arguments passed to the update strategy
• Usage: Configuration parameters for update behavior
• Examples: Learning rates, threshold values, adaptation schedules

🏗️ Decorator Architecture

Function Wrapping Process

@update(update_strategy=custom_strategy)
def physics_modifier_function(env, env_ids, param1, param2=default_value):
    # Original function implementation
    pass

🔄 Transformation Steps:

1. 📋 Signature Inspection: Analyzes original function signature
2. 🎭 Wrapper Creation: Creates parameter-aware wrapper function
3. 💾 State Initialization: Initializes _overrides dictionary for parameter storage
4. 🔧 Update Method: Attaches update() method for runtime parameter modification
5. 🏷️ Decorator Marking: Adds _is_update_decorated flag for validation

Runtime Parameter Override Logic

@functools.wraps(func)
def wrapper(*args, **kwargs):
    bound_args = sig.bind_partial(*args, **kwargs)

    # Apply overrides first to arguments not explicitly provided
    for param_key, override_val in wrapper._overrides.items():
        if param_key not in bound_args.arguments:
            bound_args.arguments[param_key] = override_val

    bound_args.apply_defaults()
    return func(*bound_args.args, **bound_args.kwargs)

🎯 Override Priority:

1. 🥇 Explicit Arguments: Directly passed parameters (highest priority)
2. 🥈 Override Values: Parameters set via update() method
3. 🥉 Default Values: Original function defaults (lowest priority)

---

⚙️ Update Method

Method Signature:

def update_func(**new_kwargs):
    if update_strategy:
        wrapper._overrides = update_strategy(wrapper._overrides, new_kwargs)
    else:
        wrapper._overrides.update(new_kwargs)

🔧 Update Behaviors

🎲 Default Update Strategy (update_strategy=None)

# Simple dictionary update
wrapper._overrides.update(new_kwargs)

# Example usage
@update()
def simple_modifier(env, env_ids, force_magnitude=100.0):
    # Apply force with current magnitude
    pass

# Runtime parameter modification
simple_modifier.update(force_magnitude=150.0)  # Increase force

🧠 Custom Update Strategy

def intelligent_strategy(current_overrides: dict, new_kwargs: dict) -> dict:
    """Advanced curriculum learning strategy"""
    updated = dict(current_overrides)
    
    # Custom logic for parameter adaptation
    if "performance_metric" in new_kwargs:
        performance = new_kwargs["performance_metric"]
        if performance > 0.8:  # High performance, increase difficulty
            updated["difficulty_multiplier"] = updated.get("difficulty_multiplier", 1.0) * 1.1
        elif performance < 0.4:  # Low performance, decrease difficulty
            updated["difficulty_multiplier"] = updated.get("difficulty_multiplier", 1.0) * 0.9
    
    return updated

@update(update_strategy=intelligent_strategy)
def adaptive_modifier(env, env_ids, difficulty_multiplier=1.0):
    # Modifier adapts based on performance
    pass

---

🚀 Practical Examples

📊 Example 1: Scaling Modifier Strategy

Module Name: GBC.gyms.isaaclab_45.managers.physics_modifier_function_wrapper (Test Implementation)

def scaling_modifier(current_overrides: dict, new_kwargs: dict) -> dict:
    """Example update strategy that scales parameters"""
    updated = dict(current_overrides)
    
    if not new_kwargs and updated:  
        # No arguments provided, scale existing overrides
        for k in updated:
            updated[k] *= 0.6  # Reduce by 40%
    elif new_kwargs:  
        # New arguments provided, scale them directly
        for k, v in new_kwargs.items():
            if isinstance(v, (int, float)):
                updated[k] = v * 0.6  # Apply scaling factor
            else:
                raise ValueError(f"Scaling supports numeric only, got {k}={v}")
    
    return updated

@update(update_strategy=scaling_modifier)
def force_modifier(env, env_ids, max_force=1000.0, apply_duration=1.0):
    """Apply scaled external forces for curriculum learning"""
    # Force magnitude and duration automatically scaled based on progress
    pass

🎯 Example 2: Performance-Based Adaptation

class AdaptiveCurriculumUpdater:
    """Stateful update strategy for curriculum learning"""
    
    def __init__(self):
        self.performance_history = []
        self.adaptation_rate = 0.05
    
    def adaptive_update_strategy(self, current_overrides: dict, new_kwargs: dict) -> dict:
        """Adapts parameters based on performance metrics"""
        updated = dict(current_overrides)
        
        if "success_rate" in new_kwargs:
            success_rate = new_kwargs["success_rate"]
            self.performance_history.append(success_rate)
            
            # Calculate moving average
            if len(self.performance_history) > 10:
                avg_performance = sum(self.performance_history[-10:]) / 10
                
                if avg_performance > 0.8:  # High success, increase difficulty
                    updated["difficulty_scale"] = updated.get("difficulty_scale", 1.0) * (1 + self.adaptation_rate)
                elif avg_performance < 0.4:  # Low success, decrease difficulty
                    updated["difficulty_scale"] = updated.get("difficulty_scale", 1.0) * (1 - self.adaptation_rate)
                
                # Clamp to reasonable bounds
                updated["difficulty_scale"] = max(0.1, min(3.0, updated["difficulty_scale"]))
        
        return updated

# Usage with stateful updater
curriculum_updater = AdaptiveCurriculumUpdater()

@update(update_strategy=curriculum_updater.adaptive_update_strategy)
def curriculum_physics_modifier(env, env_ids, difficulty_scale=1.0, base_disturbance=0.1):
    """Physics modifier that adapts to training progress"""
    disturbance_magnitude = base_disturbance * difficulty_scale
    # Apply disturbance based on adaptive difficulty

🔄 Example 3: Multi-Parameter Coordination

def coordinated_update_strategy(current_overrides: dict, new_kwargs: dict) -> dict:
    """Coordinates multiple parameters for balanced curriculum"""
    updated = dict(current_overrides)
    
    if "curriculum_phase" in new_kwargs:
        phase = new_kwargs["curriculum_phase"]
        
        if phase == "easy":
            updated.update({
                "force_magnitude": 50.0,
                "noise_level": 0.01,
                "disturbance_frequency": 0.1
            })
        elif phase == "medium":
            updated.update({
                "force_magnitude": 100.0,
                "noise_level": 0.05,
                "disturbance_frequency": 0.3
            })
        elif phase == "hard":
            updated.update({
                "force_magnitude": 200.0,
                "noise_level": 0.1,
                "disturbance_frequency": 0.5
            })
    
    return updated

@update(update_strategy=coordinated_update_strategy)
def multi_param_modifier(env, env_ids, force_magnitude=100.0, noise_level=0.05, disturbance_frequency=0.3):
    """Coordinated multi-parameter physics modification"""
    # All parameters adapt together for consistent curriculum progression

---

🧪 Testing and Validation

🔬 Test Implementation (Module Main Section)

The module includes comprehensive testing demonstrating both simple and advanced update strategies:

if __name__ == "__main__":
    # Test functions with different update strategies
    
    @update(update_strategy=scaling_modifier)
    def test_func(a, b=2, c=3):
        print(f"a: {a}, b: {b}, c: {c}")

    @update()  # Default update strategy
    def test_func_2(a, b=2, c=3):
        print(f"a: {a}, b: {b}, c: {c}")

    # Test sequence demonstrating parameter evolution
    test_func(1)           # a: 1, b: 2, c: 3 (initial)
    test_func_2(1)         # a: 1, b: 2, c: 3 (initial)
    
    test_func.update(b=4)  # Update with scaling
    test_func_2.update(b=4)# Direct update
    
    test_func(1)           # a: 1, b: 2.4, c: 3 (scaled)
    test_func_2(1)         # a: 1, b: 4, c: 3 (direct)

✅ Validation Features

1. 🔍 Parameter Binding: Verifies correct parameter override precedence
2. 🎭 Function Wrapping: Ensures original function behavior preservation
3. 📊 Update Logic: Tests both default and custom update strategies
4. 🛡️ Error Handling: Validates type checking and error recovery

---

🔧 Integration with PhysicsModifierManager

🏗️ Manager Validation

The PhysicsModifierManager validates decorated functions:

# In PhysicsModifierManager._prepare_terms()
if not hasattr(term_cfg.func, "update"):
    raise AttributeError(
        f"Physics modifier function '{term_name}' does not have 'update' attribute."
        f"Decorate your function with @update(update_strategy) from physics_modifier_function_wrapper.py."
    )

🔄 Runtime Update Cycle

# In PhysicsModifierManager.update()
for name, term_cfg in zip(self._term_names, self._term_cfgs):
    new_kwargs = {
        "env": self._env,
        "env_ids": env_ids,
    }
    new_kwargs = {**new_kwargs, **term_cfg.params}
    
    # Calls the decorated function's update method
    term_cfg.func.update(**new_kwargs)

---

💡 Best Practices

✅ Decorator Design Guidelines

1. 🎯 Single Responsibility: Each update strategy should focus on specific parameter adaptation logic
2. 🛡️ Robust Bounds: Always implement parameter bounds checking in update strategies
3. 📊 State Management: Use class-based update strategies for complex stateful behavior
4. 🔄 Idempotent Updates: Ensure update strategies can be called multiple times safely

🔧 Performance Optimization

1. ⚡ Lightweight Updates: Keep update strategies computationally efficient
2. 💾 Minimal State: Store only necessary state information in update strategy classes
3. 🎭 Function Overhead: Decorator adds minimal runtime overhead to function calls
4. 📋 Signature Caching: Function signatures are inspected once during decoration

🧪 Testing Strategies

# Recommended testing pattern
def test_physics_modifier():
    @update(update_strategy=test_strategy)
    def test_modifier(env, env_ids, param1=1.0):
        return param1
    
    # Test initial state
    assert test_modifier._overrides == {}
    
    # Test parameter update
    test_modifier.update(param1=2.0)
    assert test_modifier._overrides["param1"] == 2.0
    
    # Test function call with override
    result = test_modifier(mock_env, mock_env_ids)
    assert result == 2.0  # Uses override value

🚨 Common Pitfalls

1. ❌ Missing Decoration: All physics modifier functions must use @update decorator
2. ❌ Parameter Name Mismatch: Update parameters must match function signature
3. ❌ State Leakage: Avoid sharing mutable state between different modifier instances
4. ❌ Type Safety: Validate parameter types in update strategies

---

🔗 Related Components

• 🏗️ PhysicsModifierManager: Main manager that validates and uses decorated functions
• 📄 PhysicsModifierTermCfg: Configuration class that references decorated modifier functions
• 🎯 Update Strategies: Custom functions defining parameter adaptation logic
• ⚙️ IsaacLab Integration: Seamless integration with IsaacLab's physics simulation system

This decorator system forms the foundation for sophisticated curriculum learning and adaptive physics modification in reinforcement learning and imitation learning environments, enabling intelligent parameter adaptation based on training progress and performance metrics.