wassname/pyrobolearn
1
0
Fork 0
mirror of https://github.com/wassname/pyrobolearn.git synced 2026-09-09 11:31:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

update envs and dynamics

Browse Source
This commit is contained in:
Brian Delhaisse
2019-03-26 17:36:56 +01:00
parent 8968bd1bbf
commit 0dc3867033
9 changed files with 630 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyrobolearn/dynamics/__init__.py
+10 -1
View File
@@ -1,3 +1,12 @@
# import the transition dynamic models
from dynamic import *
from .dynamic import *
# import basic dynamic models (such as linear dynamic models)
from .basic_dynamic import *
# import robot dynamic models
from .robot_dynamic import *
# import neural network dynamic models
from .nn_dynamic import *
pyrobolearn/dynamics/basic_dynamic.py
+74
View File
@@ -0,0 +1,74 @@
#!/usr/bin/env python
"""Provides the various basic dynamic transition function approximators (e.g. table and linear approximators)
It can use learning models as most of these models learn a function (that is how to map inputs to outputs).
For instance, a dynamic network is a function represented by a neural network that maps a state-action to the next
state.
"""
from pyrobolearn.approximators import LinearApproximator
from pyrobolearn.dynamics.dynamic import ParametrizedDynamicModel
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class LinearDynamicModel(ParametrizedDynamicModel):
r"""Linear Dynamic Model
Pros: easy to implement and learn
Cons: very limited
"""
def __init__(self, states, actions, next_states=None, distributions=None, preprocessors=None, postprocessors=None):
"""
Initialize the linear dynamic transition function / probability :math:`p(s_{t+1} | s_t, a_t)`.
Args:
states (State): state inputs.
actions (Action): action inputs.
next_states (State, None): state outputs. If None, it will take the state inputs as the outputs.
distributions (torch.distributions.Distribution): distribution to use to sample the next state. If None,
it will be deterministic.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
"""
if next_states is None:
next_states = states
model = LinearApproximator(inputs=[states, actions], outputs=next_states, preprocessors=preprocessors,
postprocessors=postprocessors)
super(LinearDynamicModel, self).__init__(states, actions, model=model, next_states=next_states,
distributions=distributions)
class PieceWiseLinearDynamicModel(ParametrizedDynamicModel):
r"""Piecewise linear dynamic model
Pros: easy to implement, often good predictions in local regions
Cons: poor scalability
"""
def __init__(self, states, actions, next_states=None, distributions=None, preprocessors=None, postprocessors=None):
"""
Initialize the piece wise linear dynamic transition function / probability :math:`p(s_{t+1} | s_t, a_t)`.
Args:
states (State): state inputs.
actions (Action): action inputs.
next_states (State, None): state outputs. If None, it will take the state inputs as the outputs.
distributions (torch.distributions.Distribution): distribution to use to sample the next state. If None,
it will be deterministic.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
"""
# TODO
model = None
super(PieceWiseLinearDynamicModel, self).__init__(states, actions, model=model, next_states=next_states,
distributions=distributions)
pyrobolearn/dynamics/dynamic.py
+309 -130
View File
@@ -1,7 +1,8 @@
#!/usr/bin/env python
"""Provides the `transition`/`dynamic` function approximators in RL.
Dynamic models allows to compute the next state given the current state and action; that is, p(s_{t+1} | s_t, a_t).
Dynamic models allows to compute the next state given the current state and action; that is,
:math:`s_{t+1} = f(s_t, a_t)` (if deterministic) or :math:`s_{t+1} \sim p(.| s_t, a_t)`.
Dependencies:
- `pyrobolearn.states`
@@ -10,10 +11,13 @@ Dependencies:
"""
from abc import ABCMeta, abstractmethod
import collections
import torch
from pyrobolearn.states import State
from pyrobolearn.actions import Action
# from pyrobolearn.approximators import Approximator
from pyrobolearn.approximators import Approximator
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
@@ -28,155 +32,330 @@ __status__ = "Development"
class DynamicModel(object):
r"""Dynamic/Transition Model
In the reinforcement learning setting, the dynamic model is the transition function associated to the environment
which describes how ... The agent/policy has no control over it. However, it can often be learned from data samples
acquired by interacting with the environment. This allows to model the environment and then perform internal
simulations which are then more sample efficient in the real environment.
In the reinforcement learning setting, the dynamic model is the transition probability function associated to the
environment which provides the next state given the current state and action. The agent / policy has no control
over it. However, it can often be learned from data samples acquired by interacting with the environment.
This allows to model the environment and then perform internal simulations requiring thus less samples from the
real environment.
When a dynamic model is involved, this is known as "Model-based Reinforcement Learning". These methods usually
requires less samples as they learn the model of the environment.
When a dynamic model is involved, this is known as "Model-based Reinforcement Learning", also known as "Optimal
Control". These methods usually require less samples as they learn a model of the environment. This contrasts with
model-free on-policy search algorithms that do not exploit the data collected in previous episodes, and thus
require a lot of samples per episode. Note that off-policy methods while exploiting past data are particular to
a specific task (due to the specific reward function).
.. math::
Dynamic models allows one also to plan. Additionally, using a *differentiable* dynamic model with a
*differentiable* policy allows to unfold the consequences of selected actions for a certain time horizon, and
to take the gradient of the loss with respect to the first action which can then be used to estimate a better
action to undertake by taking a small step (using the gradient) towards an action that optimize better the loss.
This is notably useful for *model predictive control* (MPC).
P_{\varphi}(s_{t+1} | s_t, a_t)
Dynamic models can be deterministic :math:`s_{t+1} = f_{\varphi}(s_t, a_t)` or stochastic
:math:`s_{t+1} \sim P_{\varphi}(s_{t+1} | s_t, a_t)`, where :math:`\varphi` is the possible set of parameters if
the dynamic model is a trainable model, :math:`s_t` and :math:`a_t` are the current state and action respectively,
and :math:`s_{t+1}` is the predicted next state.
They are 2 main ways to build a dynamic model:
1. build it from a mathematical model
Pros: mathematical guarantees (such as stability,...), predictable,...
Cons: linearization, unmodeled phenomenon, assumptions that might be violated (rigid body), complex...
2. learn it from the data, by letting the policy interacts with the environment
Pros:
Cons: usually requires a lot of samples to be accurate, mismatch between the real and the learned dynamic
model, often no guarantees and could be unpredictable
1. build it using a mathematical model
Pros: mathematical guarantees (such as stability), predictable, ...
Cons: linearization, unmodeled phenomenon, assumptions that might be violated (rigid body), and so on
2. learn it from data, by letting the policy interacts with the environment
Pros: data-driven approach and thus more flexible and potentially more accurate.
Cons: might require a lot of samples to be accurate, might overfit which could lead to a mismatch between
the real and the learned dynamic model, often no guarantees and could be unpredictable.
Note that learning a wrong dynamic model can have drastic consequences on the learned policy. Indeed, learning
a dynamic model in the simulator can be completely to a learned .
Some papers have worked on simulators that generates ...
Note that learning a wrong dynamic model can have drastic consequences on the learned policy as this last one
depends on the returned predicted states by the environment.
References:
[1] https://spinningup.openai.com/en/latest/spinningup/rl_intro.html
[2] "Optimal control theory: An introduction", Kirk, 2004
"""
__metaclass__ = ABCMeta
def __init__(self, states, actions, model=None):
self.states = self._check_states(states)
self.actions = self._check_actions(actions)
def __init__(self, states, actions, next_states=None):
"""
Initialize the dynamic transition probability :math:`p(s_{t+1} | s_t, a_t)`, or dynamic transition function
:math:`s_{t+1} = f(s_t, a_t)`.
@staticmethod
def _check_states(states):
Args:
states (State): state inputs.
actions (Action): action inputs.
next_states (State, None): state outputs. If None, it will take the state inputs as the outputs.
"""
Check if the states are valid (i.e. it is an instance of `State`, a list/tuple of `State` instances, or None).
:param states: states to be checked.
:return: states
"""
if isinstance(states, State):
states = [states]
elif isinstance(states, (list, tuple)):
for state in states:
if not isinstance(state, State):
raise ValueError("Each state in the list/tuple must be a `State` object.")
elif states is None: # some policies don't need the state information
states = []
else:
raise ValueError("The `states` parameter must be a `State` object or a list/tuple of `State` objects.")
return states
# set inputs
self.states = states
self.actions = actions
@staticmethod
def _check_actions(actions):
"""
Check if the actions are valid (i.e. it is an instance of `Action`, a list/tuple of `Action` instances).
:param actions: actions to be checked.
:return: actions
"""
if isinstance(actions, Action):
actions = [actions]
elif isinstance(actions, (list, tuple)):
for action in actions:
if not isinstance(action, Action):
raise ValueError("Each action in the list/tuple must be an instance of the `Action` class.")
else:
raise ValueError("The `actions` parameter must be an `Action` object or a list/tuple of `Action` objects.")
return actions
# set outputs
self.next_states = next_states
##############
# Properties #
##############
@property
def states(self):
"""Return the state instance."""
return self._states
@states.setter
def states(self, states):
"""Set the states."""
if not isinstance(states, State):
raise TypeError("Expecting the given states to be an instance of `State`, instead got: "
"{}".format(type(states)))
self._states = states
@property
def actions(self):
"""Return thge action instance."""
return self._actions
@actions.setter
def actions(self, actions):
"""Set the actions."""
if not isinstance(actions, Action):
raise TypeError("Expecting the given actions to be an instance of `Action`, instead got: "
"{}".format(type(actions)))
self._actions = actions
@property
def next_states(self):
"""Return the next state instance."""
return self._next_states
@next_states.setter
def next_states(self, states):
"""Set the next states."""
if states is None:
states = self.states
elif not isinstance(states, State):
raise TypeError("Expecting the given next_states to be an instance of `State`, instead got: "
"{}".format(type(states)))
self._next_states = states
###########
# Methods #
###########
@abstractmethod
def predict(self, states=None, actions=None, deterministic=False):
"""
Predict the next state given the current state and action.
Args:
states (None, State, (list of) np.array, (list of) torch.Tensor): input states.
actions (None, Action, (list of) np.array, (list of) torch.Tensor): input actions.
Returns:
(list of) np.array, (list of) torch.Tensor: predicted next state data.
"""
pass
def __call__(self, states, actions):
"""
Return predicted state given the current state and action
:param state:
:param action:
:return:
Return predicted next state given the current state and action.
Args:
states (None, State, (list of) np.array, (list of) torch.Tensor): input states.
actions (None, Action, (list of) np.array, (list of) torch.Tensor): input actions.
Returns:
(list of) np.array, (list of) torch.Tensor: predicted next state data.
"""
pass
def save(self, filename):
pass
def load(self, filename):
pass
return self.predict(states, actions)
class PhysicalDynamicModel(DynamicModel):
r"""Physical Dynamic Model
class ParametrizedDynamicModel(DynamicModel):
r"""Learnable Parametrized Dynamic Model
Dynamic model described by mathematical/physical equations.
Dynamic model that can be trained.
"""
def __init__(self, states, actions):
super(PhysicalDynamicModel, self).__init__(states, actions)
def __init__(self, states, actions, model, next_states=None, distributions=None):
"""
Initialize the dynamic transition probability :math:`p(s_{t+1} | s_t, a_t)`.
Args:
states (State): state inputs.
actions (Action): action inputs.
model (Approximator): approximator (inner learning model).
next_states (State, None): state outputs. If None, it will take the state inputs as the outputs.
distributions ((list of) torch.distributions.Distribution, None): distribution to use to sample the next
state. If None, it will be deterministic.
"""
# set inner model
super(ParametrizedDynamicModel, self).__init__(states, actions, next_states)
self.model = model
self.distributions = distributions
##############
# Properties #
##############
@property
def model(self):
"""Return the model instance."""
return self._model
@model.setter
def model(self, model):
"""Set the approximator model."""
if not isinstance(model, Approximator):
raise TypeError("Expecting the model to be an instance of `Approximator`, instead got: "
"{}".format(type(model)))
self._model = model
@property
def distributions(self):
"""Return the list of distributions."""
return self._distributions
@distributions.setter
def distributions(self, distributions):
if distributions is None:
distributions = []
elif isinstance(distributions, torch.distributions.Distribution):
distributions = [distributions]
if len(distributions) != 0 and len(distributions) != len(self.next_states):
raise ValueError("Expecting the number of distributions (={}) to match the number of states (={})"
".".format(len(distributions), len(self.next_states)))
self._distributions = distributions
@property
def input_size(self):
"""Return the policy input size."""
return self.model.input_size
@property
def output_size(self):
"""Return the policy output size."""
return self.model.output_size
@property
def input_shape(self):
"""Return the policy input shape."""
return self.model.input_shape
@property
def output_shape(self):
"""Return the policy output shape."""
return self.model.output_shape
@property
def input_dim(self):
"""Return the input dimension of the policy; i.e. len(input_shape)."""
return self.model.input_dim
@property
def output_dim(self):
"""Return the output dimension of the policy; i.e. len(output_shape)."""
return self.model.output_dim
@property
def num_parameters(self):
"""Return the total number of parameters"""
return self.model.num_parameters
###########
# Methods #
###########
def parameters(self):
"""
Return an iterator over the learning model parameters.
"""
return self.model.parameters()
def named_parameters(self):
"""
Return an iterator over the learning model parameters; yielding both the name and the parameter itself.
"""
return self.model.named_parameters()
def list_parameters(self):
"""
Return the learning model parameters.
"""
return self.model.list_parameters()
def hyperparameters(self):
"""
Return an iterator over the learning model hyper-parameters.
"""
return self.model.hyperparameters()
def named_hyperparameters(self):
"""
Return an iterator over the learning model hyper-parameters; yielding both the name and the hyper-parameter
itself.
"""
return self.model.named_hyperparameters()
def list_hyperparameters(self):
"""
Return the learning model hyper-parameters
"""
return self.model.list_hyperparameters()
def get_vectorized_parameters(self, to_numpy=True):
"""
Get the parameters in a vectorized form.
Args:
to_numpy (bool): if True, it will convert the 1D parameter vector into a numpy array.
"""
return self.model.get_vectorized_parameters(to_numpy=to_numpy)
def set_vectorized_parameters(self, vector):
"""
Set the vectorized parameters.
Args:
np.array, torch.Tensor: 1D parameter vector.
"""
self.model.set_vectorized_parameters(vector=vector)
def predict(self, states=None, actions=None, deterministic=False, to_numpy=True, set_next_state_data=True):
"""
Predict the next state given the current state and action.
Args:
states (None, State, (list of) np.array, (list of) torch.Tensor): input states.
actions (None, Action, (list of) np.array, (list of) torch.Tensor): input actions.
deterministic (bool): if True, the prediction will be deterministic. If False and if a distribution was
given at initialization, then the predicted output will be stochastic.
to_numpy (bool): If True, it will return a (list of) np.array.
set_next_state_data (bool): if True, it will set the next state data.
Returns:
(list of) np.array, (list of) torch.Tensor: predicted next state data.
"""
# if no input is given, take the provided inputs at the beginning
data = self.model.predict([states, actions], to_numpy=to_numpy, return_logits=True,
set_output_data=False)
# return predicted next state
if deterministic or len(self.distributions) == 0:
return data
data = [distribution(datum).sample() for distribution, datum in zip(self.distributions, data)]
if set_next_state_data:
self.next_states.data = data
return data
class RobotDynamicModel(PhysicalDynamicModel):
r"""Robot Dynamical Model
This is the mathematical model of the robots.
Limitations:
* mathematical assumptions such as rigid bodies
* the states/actions have to be robot states/actions
"""
def __init__(self, states, actions):
super(RobotDynamicModel, self).__init__(states, actions)
class LinearDynamicModel(DynamicModel):
r"""Linear Dynamic Model
Pros: easy to implement
Cons: very limited
"""
def __init__(self, states, actions):
super(LinearDynamicModel, self).__init__(states, actions)
class PieceWiseLinearDynamicModel(DynamicModel):
r"""Piecewise linear dynamic model
Pros: easy to implement, often good predictions in local regions
Cons: poor scalability
"""
def __init__(self, states, actions):
super(PieceWiseLinearDynamicModel, self).__init__(states, actions)
class NNDynamicModel(DynamicModel):
r"""Neural Network Dynamic Model
Dynamic model using neural networks.
Pros:
Cons: requires lot of samples, overfitting,...
"""
def __init__(self, states, actions):
super(NNDynamicModel, self).__init__(states, actions)
class GPDynamicModel(DynamicModel):
r"""Gaussian Process Dynamic Model
Dynamic model using Gaussian Processes.
Pros: good from a mathematical point of view: integrate uncertainty on the dynamic model
Cons:
..seealso: PILCO
"""
def __init__(self, states, actions):
super(GPDynamicModel, self).__init__(states, actions)
# class GPDynamicModel(ParametrizedDynamicModel):
# r"""Gaussian Process Dynamic Model
#
# Dynamic model using Gaussian Processes.
#
# Pros: good from a mathematical point of view: integrate uncertainty on the dynamic model
# Cons:
#
# ..seealso: PILCO
# """
#
# def __init__(self, states, actions):
# super(GPDynamicModel, self).__init__(states, actions)
pyrobolearn/dynamics/nn_dynamic.py
+88
View File
@@ -0,0 +1,88 @@
#!/usr/bin/env python
"""Provides dynamic transition neural network approximators
For instance, a dynamic network is a function represented by a neural network that maps a state-action to the next
state.
"""
from pyrobolearn.approximators import NNApproximator, MLPApproximator
from pyrobolearn.dynamics.dynamic import ParametrizedDynamicModel
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class NNDynamicModel(ParametrizedDynamicModel):
r"""Neural Network Dynamic Model
Dynamic model using neural networks.
Pros:
Cons: requires lot of samples, overfitting,...
"""
def __init__(self, states, actions, model, next_states=None, distributions=None, preprocessors=None,
postprocessors=None):
"""
Initialize the NN dynamic model.
Args:
states (State): state inputs.
actions (Action): action inputs.
next_states (State, None): state outputs. If None, it will take the state inputs as the outputs.
model (NNApproximator, NN): neural network model.
distributions (torch.distributions.Distribution): distribution to use to sample the next state. If None,
it will be deterministic.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
"""
if model is None:
raise TypeError("Expecting the model to be a neural network and not None.")
elif not isinstance(model, NNApproximator):
if next_states is None:
next_states = states
model = NNApproximator(inputs=[states, actions], outputs=next_states, model=model,
preprocessors=preprocessors, postprocessors=postprocessors)
super(NNDynamicModel, self).__init__(states, actions, model=model, next_states=next_states,
distributions=distributions)
class MLPDynamicModel(NNDynamicModel):
r"""MLP Dynamic Model
"""
def __init__(self, states, actions, next_states=None, hidden_units=(), activation_fct='Linear',
last_activation_fct=None, dropout_prob=None, distributions=None, preprocessors=None,
postprocessors=None):
"""
Initialize the multi-layer perceptron model.
Args:
states (State): state inputs.
actions (Action): action inputs.
next_states (State, None): state outputs. If None, it will take the state inputs as the outputs.
hidden_units (tuple, list of int): number of hidden units in each layer
activation_fct (str): activation function to apply on each layer
last_activation_fct (str, None): activation function to apply on the last layer
dropout_prob (None, float): dropout probability
distributions (torch.distributions.Distribution): distribution to use to sample the next state. If None,
it will be deterministic.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
"""
if next_states is None:
next_states = states
model = MLPApproximator(inputs=[states, actions], outputs=next_states, hidden_units=hidden_units,
activation_fct=activation_fct, last_activation_fct=last_activation_fct,
dropout_prob=dropout_prob)
super(MLPDynamicModel, self).__init__(states, actions, model=model, next_states=next_states,
distributions=distributions, preprocessors=preprocessors,
postprocessors=postprocessors)
pyrobolearn/dynamics/robot_dynamic.py
+59
View File
@@ -0,0 +1,59 @@
#!/usr/bin/env python
"""Provides robot dynamic transition functions
"""
from pyrobolearn.robots.robot import Robot
from pyrobolearn.dynamics.dynamic import DynamicModel
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class PhysicalDynamicModel(DynamicModel):
r"""Physical Dynamic Model
Dynamic model described by mathematical/physical equations.
"""
def __init__(self, states, actions):
super(PhysicalDynamicModel, self).__init__(states, actions)
class RobotDynamicModel(PhysicalDynamicModel):
r"""Robot Dynamical Model
This is the mathematical model of the robots.
Limitations:
* mathematical assumptions such as rigid bodies
* approximation of the dynamics
* the states/actions have to be robot states/actions
"""
def __init__(self, states, actions, robot):
super(RobotDynamicModel, self).__init__(states, actions)
self.robot = robot
##############
# Properties #
##############
@property
def robot(self):
"""Return the robot instance."""
return self._robot
@robot.setter
def robot(self, robot):
"""Set the robot instance."""
if not isinstance(robot, Robot):
raise TypeError("Expecting the given robot to be an instance of `Robot`, instead got: "
"{}".format(type(robot)))
self._robot = robot
pyrobolearn/envs/__init__.py
-1
View File
@@ -4,4 +4,3 @@ from .env import Env, BasicEnv
# define wrapper for the gym environment
from . import gym_wrapper as gym
pyrobolearn/envs/env.py
+86 -34
View File
@@ -19,6 +19,8 @@ from pyrobolearn.rewards import Reward
from pyrobolearn.terminal_conditions import TerminalCondition
from pyrobolearn.physics import PhysicsRandomizer
from pyrobolearn.states.generators import StateGenerator
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
@@ -52,24 +54,22 @@ class Env(object): # gym.Env):
"""
def __init__(self, world, states, rewards=None, terminal_conditions=None, initial_state_distribution=None,
physics_randomizer=None, extra_info=None):
def __init__(self, world, states, rewards=None, terminal_conditions=None, initial_state_generators=None,
physics_randomizers=None, extra_info=None):
"""
Initialize the environment.
Args:
world (World): world of the environment. The world contains all the objects (including robots), and has
access to the simulator.
access to the simulator.
states (State): states that are returned by the environment at each time step.
rewards (None, Reward): The rewards can be None when for instance we are in an imitation learning setting,
instead of a reinforcement learning one. If None, only the state is returned by
the environment.
terminal_conditions (None, callable): A callable function or object that check if the policy has failed
or succeeded the task.
initial_state_distribution (None, callable): A callable function or object that is called at the beginning
when resetting the environment to generate the initial state
distribution.
physics_randomizer (None, PhysicsRandomizer, list of PhysicsRandomizer): physics randomizers. This will be
instead of a reinforcement learning one. If None, only the state is returned by the environment.
terminal_conditions (None, callable, TerminalCondition, list of TerminalCondition): A callable function or
object that check if the policy has failed or succeeded the task.
initial_state_generators (None, StateGenerator, list of StateGenerator): state generators which are used
when resetting the environment to generate the initial states.
physics_randomizers (None, PhysicsRandomizer, list of PhysicsRandomizer): physics randomizers. This will be
called each time you reset the environment.
extra_info (None, callable): Extra info returned by the environment at each time step.
"""
@@ -78,7 +78,8 @@ class Env(object): # gym.Env):
self.states = states
self.rewards = rewards
self.terminal_conditions = terminal_conditions
self.physics_randomizers = physics_randomizer
self.physics_randomizers = physics_randomizers
self.state_generators = initial_state_generators
self.extra_info = extra_info if extra_info is not None else lambda: False
self.rendering = False # check with simulator
@@ -114,10 +115,18 @@ class Env(object): # gym.Env):
return self._states
@states.setter
def states(self, states):
def states(self, states): # TODO: make it a list of states
"""Set the states."""
if not isinstance(states, State):
raise TypeError("Expecting the 'states' argument to be an instance of State.")
if isinstance(states, State):
states = [states]
elif isinstance(states, (list, tuple)):
for idx, state in enumerate(states):
if not isinstance(state, State):
raise TypeError("The {} item is not an instance of `State`, but instead: "
"{}".format(idx, type(state)))
else:
raise TypeError("Expecting the 'states' argument to be an instance of `State` or a list of `State`, "
"instead got: {}".format(type(states)))
self._states = states
@property
@@ -179,10 +188,47 @@ class Env(object): # gym.Env):
"instead got: {}".format(type(randomizers)))
self._physics_randomizers = randomizers
@property
def state_generators(self):
"""Return the initial state generator instance."""
return self._state_generators
@state_generators.setter
def state_generators(self, generators):
"""Set the initial state generator."""
if generators is None:
generators = []
elif isinstance(generators, StateGenerator):
generators = [generators]
elif isinstance(generators, (list, tuple)):
for generator in generators:
if not isinstance(generator, StateGenerator):
raise TypeError("Expecting the generator to be an instance of `StateGenerator`, instead got "
"{}".format(generator))
else:
raise TypeError("Expecting the given generators to be None, a `StateGenerator`, or a list of them; "
"instead got: {}".format(type(generators)))
self._state_generators = generators
###########
# Methods #
###########
def _convert_state_to_data(self, states, convert=True):
"""Convert a `State` to a list of numpy arrays or a numpy array."""
if convert:
data = []
for state in states:
if isinstance(state, State):
state = state.merged_data
if isinstance(state, list) and len(state) == 1:
state = state[0]
data.append(state)
if len(data) == 1:
data = data[0]
return data
return states
def reset(self):
"""
Reset the environment; reset the world and states.
@@ -197,15 +243,19 @@ class Env(object): # gym.Env):
for randomizer in self.physics_randomizers:
randomizer.randomize()
# generate initial states
for generator in self.state_generators:
generator()
# reset states and return first states/observations
return self.states.reset()
states = [state.reset() for state in self.states]
return self._convert_state_to_data(states)
def step(self, actions=None):
"""
Run one timestep of the environment's dynamics. When end of
episode is reached, you are responsible for calling `reset()`
to reset this environment's state.
Accepts an action and returns a tuple (observation, reward, done, info).
Run one timestep of the environment's dynamics. When end of episode is reached, you are responsible for
calling `reset()` to reset this environment's state. Accepts an action and returns a tuple (observation,
reward, done, info).
Args:
action (Action, None): an action provided by the policy(ies) to the environment
@@ -219,8 +269,8 @@ class Env(object): # gym.Env):
"""
# if not isinstance(actions, (list, tuple)):
# actions = [actions]
if actions is not None and not isinstance(actions, Action):
raise TypeError("Expecting actions to be an instance of Action.")
# if actions is not None and not isinstance(actions, Action):
# raise TypeError("Expecting actions to be an instance of Action.")
# apply each policy's action in the environment
# for action in actions:
@@ -228,8 +278,8 @@ class Env(object): # gym.Env):
# TODO: calling the actions should be done inside the policy(ies), and not in the environments. The policy
# decided when to execute an action. Think about when there are multiple policies, when using multiprocessing,
# or when the environment runs in real-time.
if actions is not None and isinstance(actions, Action):
actions()
# if actions is not None and isinstance(actions, Action):
# actions()
# perform a step forward in the simulation which computes all the dynamics
self.world.step()
@@ -239,20 +289,20 @@ class Env(object): # gym.Env):
rewards = self.rewards()
# compute terminating condition
# done = [reward.is_done() for reward in self.rewards]
done = any([condition() for condition in self.terminal_conditions])
# get next state/obs for each policy
# states = [state() for state in self.states]
# TODO: this should be before computing the rewards as some rewards need the next state
self.states()
states = [state() for state in self.states]
states = self._convert_state_to_data(states, convert=True)
# get extra information
info = self.extra_info()
return self.states, rewards, done, info
return states, rewards, done, info
def render(self, mode='human'):
"""Renders the environment (show the GUI)."""
# This is dependent on the simulator. Some simulators allow to show the GUI at any point in time,
# while others like pybullet requires to specify it at the beginning (thus see SimuRealInterface).
@@ -262,11 +312,12 @@ class Env(object): # gym.Env):
pass
def hide(self):
# hide the GUI
"""hide the GUI."""
# self.sim.configureDebugVisualizer(self.sim.COV_ENABLE_RENDERING, 0)
pass
def close(self):
"""Close the environment."""
pass
def seed(self, seed=None):
@@ -284,16 +335,17 @@ class Env(object): # gym.Env):
randomizer.seed(seed)
class BasicEnv(Env):
"""Basic Environment class.
It creates a basic environment with a basic world (a floor and with gravity), no rewards, and no states.
"""
def __init__(self, states=None, rewards=None):
world = BasicWorld()
super(BasicEnv, self).__init__(world, states, rewards)
def __init__(self, sim, states=None, rewards=None, terminal_conditions=None, initial_state_generators=None,
physics_randomizers=None, extra_info=None):
world = BasicWorld(sim)
super(BasicEnv, self).__init__(world, states, rewards, terminal_conditions, initial_state_generators,
physics_randomizers, extra_info)
# Tests
@@ -306,7 +358,7 @@ if __name__ == '__main__':
# create world
world = BasicWorld(sim)
robot = world.loadRobot('coman', useFixedBase=True)
robot = world.load_robot('coman', fixed_base=True)
# create states
states = State()
pyrobolearn/envs/gym_wrapper.py
+3
View File
@@ -129,6 +129,9 @@ class GymEnvWrapper(gym.Env):
def render(self, mode='human'):
self.env.render(mode)
def hide(self):
pass
def __repr__(self):
return self.env.__repr__()
pyrobolearn/envs/locomotion_env.py
+1 -1
View File
@@ -4,7 +4,7 @@
Define the environment to perform a locomotion task; it mainly defines the reward function.
"""
from env import Env
from pyrobolearn.envs.env import Env
from pyrobolearn.worlds import BasicWorld
from pyrobolearn.states import State
from pyrobolearn.policies import Policy