import logging
import os
import abc
from pathlib import Path
from os.path import expanduser
logger = logging.getLogger(__name__)
[docs]class Env:
"""The main Energym class to describe an abstract simulation environment.
It encapsulates an environment with arbitrary behind-the-scenes
dynamics. An environment can be partially or fully observed.
Notes
-----
Accept field keys or identify them? Maybe not a good idea... # TODO
Attributes
----------
energym_path : str
Absolute path to the energym folder.
runs_path : str
Path to store the information of the simulation run.
Methods
-------
step(action)
Advances the simulation by one timestep. (not implemented)
reset()
Resets the simulation environment. (not implemented)
close()
Closes the simulation environment. (not implemented)
get_forecast()
Gets the forecasts for external parameters.
get_output()
Gets the outputs of the last simulation step.
"""
def __init__(self):
""" """
self.energym_path = Path(__file__).resolve().parent.parent.parent
home = expanduser("~")
self.runs_path = os.path.join(home, "Energym_runs")
if not os.path.isdir(self.runs_path):
try:
os.mkdir(self.runs_path)
except BaseException as e:
logger.exception(f"Unable to create folder 'Energym_runs'. {e}")
[docs] @abc.abstractmethod
def step(self, action):
"""Advances the simulation by one timestep.
Not implemented. Subclasses should override this method.
Parameters
----------
action : dict
An action provided by the controller
Raises
------
NotImplementedError
"""
pass
[docs] @abc.abstractmethod
def reset(self):
"""Resets the simulation environment.
Not implemented. Subclasses should override this method.
Raises
------
NotImplementedError
"""
pass
[docs] @abc.abstractmethod
def close(self):
"""Closes the simulation environment.
Not implemented. Subclasses should override this method.
Raises
------
NotImplementedError
"""
pass
[docs] @abc.abstractmethod
def get_forecast(self):
"""Return forecasts for the environment.
Not implemented. Subclasses should override this method.
Raises
------
NotImplementedError
"""
pass
[docs] @abc.abstractmethod
def get_output(self):
"""Gets the outputs of the last simulation step.
Not implemented. Subclasses should override this method.
Raises NotImplementedError
"""
pass
@property
def unwrapped(self):
"""Completely unwrap this env.
Returns
-------
energym.Env
The base non-wrapped energym.Env instance.
"""
return self
def __str__(self):
if self.spec is None:
return "<{} instance>".format(type(self).__name__)
else:
return "<{}<{}>>".format(type(self).__name__, self.spec.id)
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
# propagate exception
return False
[docs]class Wrapper(Env):
"""Wraps the environment to allow a modular transformation.
This class is the base class for all wrappers. The subclass could override
some methods to change the behavior of the original environment without touching the
original code.
.. note::
Don't forget to call ``super().__init__(env)`` if the subclass overrides :meth:`__init__`.
"""
def __init__(self, env):
self.env = env
self.input_space = self.env.input_space
self.output_space = self.env.output_space
def __getattr__(self, name):
if name.startswith("_"):
raise AttributeError(
"attempted to get missing private attribute '{}'".format(name)
)
return getattr(self.env, name)
@classmethod
def class_name(cls):
return cls.__name__
[docs] def step(self, inputs):
return self.env.step(inputs)
[docs] def get_forecast(self, **kwargs):
return self.env.get_forecast(**kwargs)
[docs] def get_output(self):
return self.env.get_output()
[docs] def reset(self, **kwargs):
return self.env.reset(**kwargs)
[docs] def close(self):
return self.env.close()
def __str__(self):
return "<{}{}>".format(type(self).__name__, self.env)
def __repr__(self):
return str(self)
@property
def unwrapped(self):
return self.env.unwrapped
[docs]class OutputsWrapper(Wrapper):
"""Wrapper to transform simulations outputs"""
[docs] def reset(self, **kwargs):
outputs = self.env.reset(**kwargs)
return self.outputs(outputs)
[docs] def step(self, inputs):
outputs = self.env.step(inputs)
return self.outputs(outputs)
[docs] def get_forecast(self, **kwargs):
forecast = self.env.get_forecast(**kwargs)
return self.outputs(forecast) # forecasts always have to be part of the outputs
[docs] def get_output(self):
return self.outputs(self.env.get_output())
@abc.abstractmethod
def outputs(self, outputs):
pass
@abc.abstractmethod
def revert_outputs(self, outputs):
pass
[docs]class StepWrapper(Wrapper):
"""Wrapper to transform steps"""
[docs] def reset(self, **kwargs):
return self.env.reset(**kwargs)
[docs] @abc.abstractmethod
def step(self, inputs):
pass