# 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: int):
"""
This function is used to set the maximum number of
iterations possible before the chronics ends.
You can reset this by setting it to `-1`.
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)