Source code for grid2op.Chronics.chronicsHandler

# 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 copy
import os
import numpy as np
from datetime import timedelta

from grid2op.dtypes import dt_int
from grid2op.Exceptions import Grid2OpException, ChronicsError
from grid2op.Space import RandomObject
from grid2op.Chronics.gridValue import GridValue
from grid2op.Chronics.changeNothing import ChangeNothing


[docs]class ChronicsHandler(RandomObject): """ Represents a Chronics handler that returns a grid state. As stated previously, it is not recommended to make an directly an object from the class :class:`GridValue`. This utility will ensure that the creation of such objects are properly made. The types of chronics used can be specified in the :attr:`ChronicsHandler.chronicsClass` attribute. Attributes ---------- chronicsClass: ``type``, optional Type of chronics that will be loaded and generated. Default is :class:`ChangeNothing` (*NB* the class, and not an object / instance of the class should be send here.) This should be a derived class from :class:`GridValue`. kwargs: ``dict``, optional key word arguments that will be used to build new chronics. max_iter: ``int``, optional Maximum number of iterations per episode. real_data: :class:`GridValue` An instance of type given by :attr:`ChronicsHandler.chronicsClass`. path: ``str`` (or None) path where the data are located. """ def __init__( self, chronicsClass=ChangeNothing, time_interval=timedelta(minutes=5), max_iter=-1, **kwargs ): RandomObject.__init__(self) if not isinstance(chronicsClass, type): raise Grid2OpException( 'Parameter "chronicsClass" used to build the ChronicsHandler should be a type ' "(a class) and not an object (an instance of a class). It is currently " '"{}"'.format(type(chronicsClass)) ) if not issubclass(chronicsClass, GridValue): raise ChronicsError( 'ChronicsHandler: the "chronicsClass" argument should be a derivative of the ' '"Grid2Op.GridValue" type and not {}.'.format(type(chronicsClass)) ) self.chronicsClass = chronicsClass self._kwargs = kwargs self.max_iter = max_iter self.path = None if "path" in kwargs: self.path = kwargs["path"] self._real_data = None try: self._real_data = self.chronicsClass( time_interval=time_interval, max_iter=self.max_iter, **self.kwargs ) except TypeError as exc_: raise ChronicsError( "Impossible to build a chronics of type {} with arguments in " "{}".format(chronicsClass, self.kwargs) ) from exc_ @property def kwargs(self): res = copy.deepcopy(self._kwargs) if self._real_data is not None: self._real_data.get_kwargs(res) return res @kwargs.setter def kwargs(self, new_value): raise ChronicsError('Impossible to set the "kwargs" attribute') @property def real_data(self): return self._real_data
[docs] def next_time_step(self): """ This method returns the modification of the powergrid at the next time step for the same episode. See definition of :func:`GridValue.load_next` for more information about this method. """ res = self._real_data.load_next() return res
[docs] def max_episode_duration(self): """ Returns ------- max_duration: ``int`` The maximum duration of the current episode Notes ----- Using this function (which we do not recommend) you will receive "-1" for "infinite duration" otherwise you will receive a positive integer """ tmp = self.max_iter if tmp == -1: # tmp = -1 means "infinite duration" but in this case, i can have a limit # due to the data used (especially if read from files) tmp = self._real_data.max_timestep() else: # i can also have a limit on the maximum number of data in the chronics (especially if read from files) tmp = min(tmp, self._real_data.max_timestep()) return tmp
[docs] def get_name(self): """ This method retrieve a unique name that is used to serialize episode data on disk. See definition of :mod:`EpisodeData` for more information about this method. """ return str(os.path.split(self.get_id())[-1])
[docs] def set_max_iter(self, max_iter): """ This function is used to set the maximum number of iterations possible before the chronics ends. Parameters ---------- max_iter: ``int`` The maximum number of steps that can be done before reaching the end of the episode """ if not isinstance(max_iter, int): raise Grid2OpException( "The maximum number of iterations possible for this chronics, before it ends." ) if max_iter == 0: raise Grid2OpException( "The maximum number of iteration should be > 0 (or -1 if you mean " '"don\'t limit it")' ) elif max_iter < -1: raise Grid2OpException( "The maximum number of iteration should be > 0 (or -1 if you mean " '"don\'t limit it")' ) self.max_iter = max_iter self._real_data.max_iter = max_iter
[docs] def seed(self, seed): """ Seed the chronics handler and the :class:`GridValue` that is used to generate the chronics. Parameters ---------- seed: ``int`` Set the seed for this instance and for the data it holds Returns ------- seed: ``int`` The seed used for this object seed_chronics: ``int`` The seed used for the real data """ super().seed(seed) max_int = np.iinfo(dt_int).max seed_chronics = self.space_prng.randint(max_int) self._real_data.seed(seed_chronics) return seed, seed_chronics
def __getattr__(self, name): if name in ["__getstate__", "__setstate__"]: # otherwise there is a recursion depth exceeded in multiprocessing # https://github.com/matplotlib/matplotlib/issues/7852/ return object.__getattr__(self, name) return getattr(self._real_data, name)