Copy an existing environment configuration Args: env_name: Name of the environment to copy new_name: New name for the copied environment. If None, uses original name. new_version: New version for the copied environment. If None, increments version
(self,
env_name: str,
new_name: Optional[str] = None,
new_version: Optional[str] = None,
new_config: Optional[Dict[str, Any]] = None)
| 786 | raise |
| 787 | |
| 788 | async def copy(self, |
| 789 | env_name: str, |
| 790 | new_name: Optional[str] = None, |
| 791 | new_version: Optional[str] = None, |
| 792 | new_config: Optional[Dict[str, Any]] = None) -> EnvironmentConfig: |
| 793 | """Copy an existing environment configuration |
| 794 | |
| 795 | Args: |
| 796 | env_name: Name of the environment to copy |
| 797 | new_name: New name for the copied environment. If None, uses original name. |
| 798 | new_version: New version for the copied environment. If None, increments version. |
| 799 | new_config: New configuration dict for the copied environment. If None, uses original config. |
| 800 | |
| 801 | Returns: |
| 802 | EnvironmentConfig: New environment configuration |
| 803 | """ |
| 804 | try: |
| 805 | original_config = self._environment_configs.get(env_name) |
| 806 | if original_config is None: |
| 807 | raise ValueError(f"Environment {env_name} not found") |
| 808 | |
| 809 | if original_config.cls is None: |
| 810 | raise ValueError(f"Cannot copy environment {env_name}: no class provided") |
| 811 | |
| 812 | # Determine new name |
| 813 | if new_name is None: |
| 814 | new_name = env_name |
| 815 | |
| 816 | # Prepare config dict (merge original config with new config) |
| 817 | env_config_dict = original_config.config.copy() if original_config.config else {} |
| 818 | if new_config: |
| 819 | # Merge new config into original config |
| 820 | env_config_dict.update(new_config) |
| 821 | |
| 822 | # Instantiate environment instance (copy is a runtime operation) |
| 823 | try: |
| 824 | env_instance = original_config.cls(**env_config_dict) |
| 825 | except Exception as e: |
| 826 | logger.error(f"| ❌ Failed to create environment instance for {original_config.cls.__name__}: {e}") |
| 827 | raise ValueError(f"Failed to instantiate environment {original_config.cls.__name__} with provided config: {e}") |
| 828 | |
| 829 | # Apply name override if provided (after instantiation) |
| 830 | if new_name != env_name: |
| 831 | env_instance.name = new_name |
| 832 | |
| 833 | env_description = env_instance.description |
| 834 | env_metadata = getattr(env_instance, 'metadata', {}) |
| 835 | 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) |
| 836 | |
| 837 | # Determine new version from version_manager |
| 838 | if new_version is None: |
| 839 | if new_name == env_name: |
| 840 | # If copying with same name, get next version from version_manager |
| 841 | new_version = await version_manager.generate_next_version("environment", new_name, "patch") |
| 842 | else: |
| 843 | # If copying with different name, get or generate version for new name |
| 844 | new_version = await version_manager.get_version("environment", new_name) |
| 845 |
nothing calls this directly
no test coverage detected