Source code for grid2op.Environment.MultiMixEnv

# Copyright (c) 2019-2020, RTE (https://www.rte-france.com)
# See AUTHORS.txt
# This Source Code Form is subject to the terms of the Mozilla Public License, version 2.0.
# If a copy of the Mozilla Public License, version 2.0 was not distributed with this file,
# you can obtain one at http://mozilla.org/MPL/2.0/.
# SPDX-License-Identifier: MPL-2.0
# This file is part of Grid2Op, Grid2Op a testbed platform to model sequential decision making in power systems.

import os
import warnings
import numpy as np
import copy

from grid2op.dtypes import dt_int, dt_float
from grid2op.Space import GridObjects, RandomObject
from grid2op.Exceptions import EnvError, Grid2OpException


[docs]class MultiMixEnvironment(GridObjects, RandomObject): """ This class represent a single powergrid configuration, backed by multiple environments parameters and chronics It implements most of the :class:`BaseEnv` public interface: so it can be used as a more classic environment. MultiMixEnvironment environments behave like a superset of the environment: they are made of sub environments (called mixes) that are grid2op regular :class:`Environment`. You might think the MultiMixEnvironment as a dictionary of :class:`Environment` that implements some of the :class:`BaseEnv` interface such as :func:`BaseEnv.step` or :func:`BaseEnv.reset`. By default, each time you call the "step" function a different mix is used. Mixes, by default are looped through always in the same order. You can see the Examples section for information about control of these Examples -------- In this section we present some common use of the MultiMix environment. **Basic Usage** You can think of a MultiMixEnvironment as any :class:`Environment`. So this is a perfectly valid way to use a MultiMix: .. code-block:: python import grid2op from grid2op.Agent import RandomAgent # we use an example of a multimix dataset attached with grid2op pacakage multimix_env = grid2op.make("l2rpn_neurips_2020_track2", test=True) # define an agent like in any environment agent = RandomAgent(multimix_env.action_space) # and now you can do the open ai gym loop NB_EPISODE = 10 for i in range(NB_EPISODE): obs = multimix_env.reset() # each time "reset" is called, another mix is used. reward = multimix_env.reward_range[0] done = False while not done: act = agent.act(obs, reward, done) obs, reward, done, info = multimix_env.step(act) **Use each mix one after the other** In case you want to study each mix independently, you can iterate through the MultiMix in a pythonic way. This makes it easy to perform, for example, 10 episode for a given mix before passing to the next one. .. code-block:: python import grid2op from grid2op.Agent import RandomAgent # we use an example of a multimix dataset attached with grid2op pacakage multimix_env = grid2op.make("l2rpn_neurips_2020_track2", test=True) NB_EPISODE = 10 for mix in multimix_env: # mix is a regular environment, you can do whatever you want with it # for example for i in range(NB_EPISODE): obs = multimix_env.reset() # each time "reset" is called, another mix is used. reward = multimix_env.reward_range[0] done = False while not done: act = agent.act(obs, reward, done) obs, reward, done, info = multimix_env.step(act) **Selecting a given Mix** Sometimes it might be interesting to study only a given mix. For that you can use the `[]` operator to select only a given mix (which is a grid2op environment) and use it as you would. This can be done with: .. code-block:: python import grid2op from grid2op.Agent import RandomAgent # we use an example of a multimix dataset attached with grid2op pacakage multimix_env = grid2op.make("l2rpn_neurips_2020_track2", test=True) # define an agent like in any environment agent = RandomAgent(multimix_env.action_space) # list all available mixes: mixes_names = list(multimix_env.keys()) # and now supposes we want to study only the first one mix = multimix_env[mixes_names[0]] # and now you can do the open ai gym loop, or anything you want with it NB_EPISODE = 10 for i in range(NB_EPISODE): obs = mix.reset() # each time "reset" is called, another mix is used. reward = mix.reward_range[0] done = False while not done: act = agent.act(obs, reward, done) obs, reward, done, info = mix.step(act) **Using the Runner** For MultiMixEnvironment using the :class:`grid2op.Runner.Runner` cannot be done in a straightforward manner. Here we give an example on how to do it. .. code-block:: python import os import grid2op from grid2op.Agent import RandomAgent # we use an example of a multimix dataset attached with grid2op pacakage multimix_env = grid2op.make("l2rpn_neurips_2020_track2", test=True) # you can use the runner as following PATH = "PATH/WHERE/YOU/WANT/TO/SAVE/THE/RESULTS" for mix in multimix_env: runner = Runner(**mix.get_params_for_runner(), agentClass=RandomAgent) runner.run(nb_episode=1, path_save=os.path.join(PATH,mix.name)) """ def __init__( self, envs_dir, logger=None, experimental_read_from_local_dir=False, _add_to_name="", # internal, for test only, do not use ! _compat_glop_version=None, # internal, for test only, do not use ! _test=False, **kwargs, ): GridObjects.__init__(self) RandomObject.__init__(self) self.current_env = None self.env_index = None self.mix_envs = [] self._env_dir = os.path.abspath(envs_dir) self.__closed = False # Special case handling for backend # TODO: with backend.copy() instead ! backendClass = None backend_kwargs = {} if "backend" in kwargs: backendClass = type(kwargs["backend"]) if hasattr(kwargs["backend"], "_my_kwargs"): # was introduced in grid2op 1.7.1 backend_kwargs = kwargs["backend"]._my_kwargs del kwargs["backend"] # Inline import to prevent cyclical import from grid2op.MakeEnv.Make import make # TODO reuse same observation_space and action_space in all the envs maybe ? try: for env_dir in sorted(os.listdir(envs_dir)): env_path = os.path.join(envs_dir, env_dir) if not os.path.isdir(env_path): continue this_logger = ( logger.getChild(f"MultiMixEnvironment_{env_dir}") if logger is not None else None ) # Special case for backend if backendClass is not None: try: # should pass with grid2op >= 1.7.1 bk = backendClass(**backend_kwargs) except TypeError as exc_: # with grid2Op version prior to 1.7.1 # you might have trouble with # "TypeError: __init__() got an unexpected keyword argument 'can_be_copied'" msg_ = ("Impossible to create a backend for each mix using the " "backend key-word arguments. Falling back to creating " "with no argument at all (default behaviour with grid2op <= 1.7.0).") warnings.warn(msg_) bk = backendClass() env = make( env_path, backend=bk, _add_to_name=_add_to_name, _compat_glop_version=_compat_glop_version, test=_test, logger=this_logger, experimental_read_from_local_dir=experimental_read_from_local_dir, **kwargs, ) else: env = make( env_path, _add_to_name=_add_to_name, _compat_glop_version=_compat_glop_version, test=_test, logger=this_logger, experimental_read_from_local_dir=experimental_read_from_local_dir, **kwargs, ) self.mix_envs.append(env) except Exception as exc_: err_msg = "MultiMix environment creation failed: {}".format(exc_) raise EnvError(err_msg) if len(self.mix_envs) == 0: err_msg = "MultiMix envs_dir did not contain any valid env" raise EnvError(err_msg) self.env_index = 0 self.current_env = self.mix_envs[self.env_index] # Make sure GridObject class attributes are set from first env # Should be fine since the grid is the same for all envs multi_env_name = os.path.basename(os.path.abspath(envs_dir)) + _add_to_name save_env_name = self.current_env.env_name self.current_env.env_name = multi_env_name self.__class__ = self.init_grid(self.current_env) self.current_env.env_name = save_env_name
[docs] def get_path_env(self): """ Get the path that allows to create this environment. It can be used for example in `grid2op.utils.underlying_statistics` to save the information directly inside the environment data. """ if self.__closed: raise EnvError("This environment is closed, you cannot use it.") return self._env_dir
@property def current_index(self): return self.env_index def __len__(self): return len(self.mix_envs) def __iter__(self): """ Operator __iter__ overload to make a ``MultiMixEnvironment`` iterable .. code-block:: python import grid2op from grid2op.Environment import MultiMixEnvironment from grid2op.Runner import Runner mm_env = MultiMixEnvironment("/path/to/multi/dataset/folder") for env in mm_env: run_p = env.get_params_for_runner() runner = Runner(**run_p) runner.run(nb_episode=1, max_iter=-1) """ self.env_index = 0 return self def __next__(self): if self.env_index < len(self.mix_envs): r = self.mix_envs[self.env_index] self.env_index = self.env_index + 1 return r else: self.env_index = 0 raise StopIteration def __getattr__(self, name): # TODO what if name is an integer ? make it possible to loop with integer here return getattr(self.current_env, name) def keys(self): for mix in self.mix_envs: yield mix.name def values(self): for mix in self.mix_envs: yield mix def items(self): for mix in self.mix_envs: yield mix.name, mix def copy(self): if self.__closed: raise EnvError("This environment is closed, you cannot use it.") mix_envs = self.mix_envs self.mix_envs = None current_env = self.current_env self.current_env = None cls = self.__class__ res = cls.__new__(cls) for k in self.__dict__: if k == "mix_envs" or k == "current_env": # this is handled elsewhere continue setattr(res, k, copy.deepcopy(getattr(self, k))) res.mix_envs = [mix.copy() for mix in mix_envs] res.current_env = res.mix_envs[res.env_index] self.mix_envs = mix_envs self.current_env = current_env return res def __getitem__(self, key): """ Operator [] overload for accessing underlying mixes by name .. code-block:: python import grid2op from grid2op.Environment import MultiMixEnvironment mm_env = MultiMixEnvironment("/path/to/multi/dataset/folder") mix1_env.name = mm_env["mix_1"] assert mix1_env == "mix_1" mix2_env.name = mm_env["mix_2"] assert mix2_env == "mix_2" """ if self.__closed: raise EnvError("This environment is closed, you cannot use it.") # Search for key for mix in self.mix_envs: if mix.name == key: return mix # Not found by name raise KeyError def reset(self, random=False): if self.__closed: raise EnvError("This environment is closed, you cannot use it.") if random: self.env_index = self.space_prng.randint(len(self.mix_envs)) else: self.env_index = (self.env_index + 1) % len(self.mix_envs) self.current_env = self.mix_envs[self.env_index] self.current_env.reset() return self.get_obs()
[docs] def seed(self, seed=None): """ Set the seed of this :class:`Environment` for a better control and to ease reproducible experiments. Parameters ---------- seed: ``int`` The seed to set. Returns --------- seeds: ``list`` The seed used to set the prng (pseudo random number generator) for all environments, and each environment ``tuple`` seeds """ if self.__closed: raise EnvError("This environment is closed, you cannot use it.") try: seed = np.array(seed).astype(dt_int) except Exception as e: raise Grid2OpException( "Cannot to seed with the seed provided." "Make sure it can be converted to a" "numpy 32 bits integer." ) s = super().seed(seed) seeds = [s] max_dt_int = np.iinfo(dt_int).max for env in self.mix_envs: env_seed = self.space_prng.randint(max_dt_int) env_seeds = env.seed(env_seed) seeds.append(env_seeds) return seeds
def set_chunk_size(self, new_chunk_size): if self.__closed: raise EnvError("This environment is closed, you cannot use it.") for mix in self.mix_envs: mix.set_chunk_size(new_chunk_size) def set_id(self, id_): if self.__closed: raise EnvError("This environment is closed, you cannot use it.") for mix in self.mix_envs: mix.set_id(id_) def deactivate_forecast(self): if self.__closed: raise EnvError("This environment is closed, you cannot use it.") for mix in self.mix_envs: mix.deactivate_forecast() def reactivate_forecast(self): if self.__closed: raise EnvError("This environment is closed, you cannot use it.") for mix in self.mix_envs: mix.reactivate_forecast()
[docs] def set_thermal_limit(self, thermal_limit): """ Set the thermal limit effectively. Will propagate to all underlying mixes """ if self.__closed: raise EnvError("This environment is closed, you cannot use it.") for mix in self.mix_envs: mix.set_thermal_limit(thermal_limit)
def __enter__(self): """ Support *with-statement* for the environment. """ return self def __exit__(self, *args): """ Support *with-statement* for the environment. """ self.close() # propagate exception return False def close(self): if self.__closed: return for mix in self.mix_envs: mix.close() self.__closed = True
[docs] def attach_layout(self, grid_layout): if self.__closed: raise EnvError("This environment is closed, you cannot use it.") for mix in self.mix_envs: mix.attach_layout(grid_layout)
def __del__(self): """when the environment is garbage collected, free all the memory, including cross reference to itself in the observation space.""" if not self.__closed: self.close() def generate_classes(self): # TODO this is not really a good idea, as the multi-mix itself is not read from the # files ! for mix in self.mix_envs: mix.generate_classes()