Update an existing environment with new configuration and create a new version Args: env_cls: New environment class with updated implementation env_config_dict: Configuration dict for environment initialization If None, will try to get from
(self,
env_cls: Type[Environment],
env_config_dict: Optional[Dict[str, Any]] = None,
new_version: Optional[str] = None,
description: Optional[str] = None,
code: Optional[str] = None)
| 648 | |
| 649 | |
| 650 | async def update(self, |
| 651 | env_cls: Type[Environment], |
| 652 | env_config_dict: Optional[Dict[str, Any]] = None, |
| 653 | new_version: Optional[str] = None, |
| 654 | description: Optional[str] = None, |
| 655 | code: Optional[str] = None) -> EnvironmentConfig: |
| 656 | """Update an existing environment with new configuration and create a new version |
| 657 | |
| 658 | Args: |
| 659 | env_cls: New environment class with updated implementation |
| 660 | env_config_dict: Configuration dict for environment initialization |
| 661 | If None, will try to get from global config |
| 662 | new_version: New version string. If None, auto-increments from current version. |
| 663 | description: Description for this version update |
| 664 | code: Optional source code string. If provided, uses this instead of extracting from env_cls. |
| 665 | This is useful when env_cls is dynamically created from code string. |
| 666 | |
| 667 | Returns: |
| 668 | EnvironmentConfig: Updated environment configuration |
| 669 | """ |
| 670 | try: |
| 671 | if env_config_dict is None: |
| 672 | # Fallback to global config by class name |
| 673 | env_config_key = inflection.underscore(env_cls.__name__) |
| 674 | env_config_dict = getattr(config, env_config_key, {}) if hasattr(config, env_config_key) else {} |
| 675 | |
| 676 | # Instantiate environment immediately (update is a runtime operation) |
| 677 | try: |
| 678 | env_instance = env_cls(**env_config_dict) |
| 679 | except Exception as e: |
| 680 | logger.error(f"| ❌ Failed to create environment instance for {env_cls.__name__}: {e}") |
| 681 | raise ValueError(f"Failed to instantiate environment {env_cls.__name__} with provided config: {e}") |
| 682 | |
| 683 | env_name = env_instance.name |
| 684 | |
| 685 | # Check if environment exists |
| 686 | original_config = self._environment_configs.get(env_name) |
| 687 | if original_config is None: |
| 688 | raise ValueError(f"Environment {env_name} not found. Use register() to register a new environment.") |
| 689 | |
| 690 | env_description = env_instance.description |
| 691 | env_metadata = getattr(env_instance, 'metadata', {}) |
| 692 | env_require_grad = env_config_dict.get("require_grad", getattr(env_instance, 'require_grad', False)) if env_config_dict and "require_grad" in env_config_dict else getattr(env_instance, 'require_grad', False) |
| 693 | |
| 694 | # Determine new version from version_manager |
| 695 | if new_version is None: |
| 696 | # Get current version from version_manager and generate next patch version |
| 697 | new_version = await version_manager.generate_next_version("environment", env_name, "patch") |
| 698 | |
| 699 | # Get environment code - use provided code if available (for dynamically created classes) |
| 700 | if code is not None: |
| 701 | env_code = code |
| 702 | else: |
| 703 | env_code = dynamic_manager.get_full_module_source(env_cls) |
| 704 | |
| 705 | # Build actions from environment class (same as register) |
| 706 | actions = {} |
| 707 | for attr_name in dir(env_cls): |
no test coverage detected