add states, actions, rewards

This commit is contained in:
Brian Delhaisse
2019-03-16 02:07:56 +01:00
parent 70f82978c1
commit 605674787a
27 changed files with 4265 additions and 0 deletions
+10
View File
@@ -0,0 +1,10 @@
## Actions
The `Action` is produced by the policy in response to a certain state/observation. From a programming point of view, compared to the `State` class, the action is a setter object. Thus, they have a very close relationship and share many functionalities. Some actions are mutually exclusive and cannot be executed at the same time.
An action is defined as something that affects the environment; that forces the environment to go to the next state. For instance, an action could be the desired joint positions, but also an abstract action such as 'open a door' which would then open a door in the simulator and load the next part of the world.
In the framework, the `Action` class is decoupled from the policy and environment rendering it more modular [1]. Nevertheless, the `Action` class still acts as a bridge between the policy and environment. In addition to be the output of a policy/controller, it can also be the input to some value estimators, dynamic models, reward functions, and so on.
References:
[1] "Wikipedia: Composition over Inheritance", https://en.wikipedia.org/wiki/Composition_over_inheritance
+9
View File
@@ -0,0 +1,9 @@
# import action
from action import Action
# import robot actions
from robot_actions import *
# import gym actions
from gym_actions import *
+670
View File
@@ -0,0 +1,670 @@
#!/usr/bin/env python
"""Define the Action class.
This file defines the `Action` class, which is returned by the policy and given to the environment.
"""
import numpy as np
import collections
from abc import ABCMeta, abstractmethod
import gym
from pyrobolearn.utils.data_structures.orderedset import OrderedSet
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class Action(object):
r"""Action class.
The `Action` is produced by the policy in response to a certain state/observation. From a programming point of
view, compared to the `State` class, the action is a setter object. Thus, they have a very close relationship
and share many functionalities. Some actions are mutually exclusive and cannot be executed at the same time.
An action is defined as something that affects the environment; that forces the environment to go to the next
state. For instance, an action could be the desired joint positions, but also an abstract action such as
'open a door' which would then open a door in the simulator and load the next part of the world.
In our framework, the `Action` class is decoupled from the policy and environment rendering it more modular [1].
Nevertheless, the `Action` class still acts as a bridge between the policy and environment. In addition to be
the output of a policy/controller, it can also be the input to some value estimators, dynamic models, reward
functions, and so on.
This class also describes the `action_space` which has initially been defined in `gym.Env` [2].
References:
[1] "Wikipedia: Composition over Inheritance", https://en.wikipedia.org/wiki/Composition_over_inheritance
[2] "OpenAI gym": https://gym.openai.com/ and https://github.com/openai/gym
"""
def __init__(self, actions=(), data=None, space=None, name=None):
"""
Initialize the action. The action contains some kind of data, or is a combination of other actions.
Args:
actions (list/tuple of Action): list of actions to be combined together (if given, we can not specified
data)
data (np.ndarray): data associated to this state
space (gym.space): space associated with the given data
Warning:
Both arguments can not be provided to the action.
"""
# Check arguments
if actions is None:
actions = tuple()
if not isinstance(actions, (list, tuple, set, OrderedSet)):
raise TypeError("Expecting a list, tuple, or (ordered) set of actions.")
if len(actions) > 0 and data is not None:
raise ValueError("Please specify only one of the argument `actions` xor `data`, but not both.")
# Check if data is given
if data is not None:
if not isinstance(data, np.ndarray):
if isinstance(data, (list, tuple)):
data = np.array(data)
elif isinstance(data, (int, float)):
data = np.array([data])
else:
raise TypeError("Expecting a numpy array, a list/tuple of int/float, or an int/float for 'data'")
# The following attributes should normally be set in the child classes
self._data = data
self._space = space
self._distribution = None # for sampling
self._normalizer = None
self._noiser = None # for noise
self._name = name
# create ordered set which is useful if this action is a combination of multiple actions
self._actions = OrderedSet()
if self._data is None:
self.add(actions)
# reset action
#self.reset()
##############################
# Properties (Getter/Setter) #
##############################
@property
def actions(self):
"""
Get the list of actions.
"""
return self._actions
@actions.setter
def actions(self, actions):
"""
Set the list of actions.
"""
if self.hasData():
raise AttributeError("Trying to add internal actions to the current action while it already has some data. "
"A action should be a combination of actions or should contain some kind of data, "
"but not both.")
if isinstance(actions, collections.Iterable):
for action in actions:
if not isinstance(action, Action):
raise TypeError("One of the given actions is not an instance of Action.")
self.add(action)
else:
raise TypeError("Expecting an iterator (e.g. list, tuple, OrderedSet, set,...) over actions")
@property
def data(self):
"""
Get the data associated to this particular action, or the combined data associated to each action.
Returns:
list of np.ndarray: list of data associated to the action
"""
if self.hasData():
return [self._data]
return [action._data for action in self._actions]
@data.setter
def data(self, data):
"""
Set the data associated to this particular action, or the combined data associated to each action.
Each data will be clipped if outside the range/bounds of the corresponding action.
Args:
data: the data to set
"""
# one action: change the data
if self.hasData():
if not isinstance(data, np.ndarray):
if isinstance(data, (list, tuple)):
data = np.array(data)
elif isinstance(data, (int, float)):
data = data * np.ones(self._data.shape)
else:
raise TypeError("Expecting a numpy array, a list/tuple of int/float, or an int/float for 'data'")
if self._data.shape != data.shape:
raise ValueError("The given data does not have the same shape as previously.")
# clip the value using the space
if self.hasSpace():
if self.isContinuous(): # continuous case
low, high = self._space.low, self._space.high
data = np.clip(data, low, high)
else: # discrete case
n = self._space.n
if data.size == 1:
data = np.clip(data, 0, n)
self._data = data
else: # combined action
if not isinstance(data, collections.Iterable):
raise TypeError("data is not an iterator")
if len(self._actions) != len(data):
raise ValueError("The number of actions is different from the number of data segments")
for action, d in zip(self._actions, data):
action.data = d
@property
def space(self):
"""
Get the corresponding space.
"""
if self.hasSpace():
return [self._space]
return [action._space for action in self._actions]
@space.setter
def space(self, space):
"""
Set the corresponding space. This can only be used one time!
"""
if self.hasData() and not self.hasSpace() and \
isinstance(space, (gym.spaces.Box, gym.spaces.Discrete)):
self._space = space
@property
def name(self):
"""
Return the name of the action.
"""
if self._name is None:
return self.__class__.__name__
return self._name
@name.setter
def name(self, name):
"""
Set the name of the action.
"""
if not isinstance(name, str):
raise TypeError("Expecting the name to be a string.")
self._name = name
@property
def shape(self):
"""
Return the shape of each action. Some actions, such as camera actions have more than 1 dimension.
"""
# if self.hasActions():
return [d.shape for d in self.data]
# return [self.data.shape]
@property
def size(self):
"""
Return the size of each action.
"""
# if self.hasActions():
return [d.size for d in self.data]
# return [len(self.data)]
@property
def dimension(self):
"""
Return the dimension (length of shape) of each action.
"""
return [len(d.shape) for d in self.data]
@property
def distribution(self):
"""
Get the current distribution used when sampling the action
"""
pass
@distribution.setter
def distribution(self, distribution):
"""
Set the distribution to the action.
"""
# check if distribution is discrete/continuous
pass
###########
# Methods #
###########
def isCombinedAction(self):
"""
Return a boolean value depending if the action is a combination of actions.
Returns:
bool: True if the action is a combination of actions, False otherwise.
"""
return len(self._actions) > 0
# alias
hasActions = isCombinedAction
def hasData(self):
return self._data is not None
def hasSpace(self):
return self._space is not None
def add(self, action):
"""
Add a action or a list of actions to the list of internal actions. Useful when combining different actions
together. This shouldn't be called if this action has some data set to it.
Args:
action (Action, list/tuple of Action): action(s) to add to the internal list of actions
"""
if self.hasData():
raise AttributeError("Undefined behavior: a action should be a combination of actions or should contain "
"some kind of data, but not both.")
if isinstance(action, Action):
self._actions.add(action)
elif isinstance(action, collections.Iterable):
for i, s in enumerate(action):
if not isinstance(s, Action):
raise TypeError("The item {} in the given list is not an instance of Action".format(i))
self._actions.add(s)
else:
raise TypeError("The 'other' argument should be an instance of Action, or an iterator over actions.")
# alias
append = add
extend = add
def _write(self, data=None):
pass
def write(self, data=None):
"""
Write the action values to the simulator for each action.
This has to be overwritten by the child class.
"""
if self.hasData(): # write the current action
self._write(data)
else: # read each action
if self.actions:
for action, d in zip(self.actions, data):
action._write(d)
# return the data
# return self.data
# def _reset(self):
# pass
#
# def reset(self):
# """
# Some actions need to be reset. It returns the initial action.
# This needs to be overwritten by the child class.
#
# Returns:
# initial action
# """
# if self.hasData(): # reset the current action
# self._reset()
# else: # reset each action
# for action in self.actions:
# action._reset()
#
# # return the first action data
# return self.write()
# def shape(self):
# """
# Return the shape of each action. Some actions, such as camera actions have more than 1 dimension.
# """
# return [d.shape for d in self.data]
#
# def dimension(self):
# """
# Return the dimension (length of shape) of each action.
# """
# return [len(d.shape) for d in self.data]
def maxDimension(self):
"""
Return the maximum dimension.
"""
return max(self.dimension)
# def size(self):
# """
# Return the size of each action.
# """
# return [d.size for d in self.data]
def totalSize(self):
"""
Return the total size of the combined action.
"""
return sum(self.size)
def hasDiscreteValues(self):
"""
Does the action have discrete values?
"""
if self._data is None:
return [isinstance(action._space, gym.spaces.Discrete) for action in self._actions]
if isinstance(self._space, gym.spaces.Discrete):
return [True]
return [False]
def isDiscrete(self):
"""
If all the actions are discrete, then it is discrete.
"""
return all(self.hasDiscreteValues())
def hasContinuousValues(self):
"""
Does the action have continuous values?
"""
if self._data is None:
return [isinstance(action._space, gym.spaces.Box) for action in self._actions]
if isinstance(self._space, gym.spaces.Box):
return [True]
return [False]
def isContinuous(self):
"""
If one of the action is continuous, then the action is considered to be continuous.
"""
return any(self.hasContinuousValues())
def bounds(self):
"""
If the action is continuous, it returns the lower and higher bounds of the action.
If the action is discrete, it returns the maximum number of discrete values that the action can take.
Returns:
list/tuple: list of bounds if multiple actions, or bounds of this action
"""
if self._data is None:
return [action.bounds() for action in self._actions]
if isinstance(self._space, gym.spaces.Box):
return (self._space.low, self._space.high)
elif isinstance(self._space, gym.spaces.Discrete):
return (self._space.n,)
raise NotImplementedError
def apply(self, fct):
"""
Apply the given fct to the data of the action, and set it to the action.
"""
self.data = fct(self.data)
def contains(self, x): # parameter dependent of the action
"""
Check if the argument is within the range/bound of the action.
"""
return self._space.contains(x)
def sample(self, distribution=None): # parameter dependent of the action (discrete and continuous distributions)
"""
Sample some values from the action based on the given distribution.
If no distribution is specified, it samples from a uniform distribution (default value).
"""
if self.isCombinedAction():
return [action.sample() for action in self._actions]
if self._distribution is None:
return
else:
pass
raise NotImplementedError
def addNoise(self, noise=None, replace=True): # parameter dependent of the action
"""
Add some noise to the action, and returns it.
Args:
noise (np.ndarray, fct): array to be added or function to be applied on the data
"""
if self._data is None:
# apply noise
for action in self._actions:
action.addNoise(noise=noise)
else:
# add noise to the data
noisy_data = self.data + noise
# clip such that the data is within the bounds
self.data = noisy_data
def normalize(self, normalizer=None, replace=True): # parameter dependent of the action
"""
Normalize using the action data using the provided normalizer.
Args:
normalizer (sklearn.preprocessing.Normalizer): the normalizer to apply to the data.
replace (bool): if True, it will replace the `data` attribute by the normalized data.
Returns:
the normalized data
"""
pass
########################
# Operator Overloading #
########################
def __repr__(self):
if self._data is None:
lst = [self.__class__.__name__ + '(']
for action in self.actions:
lst.append('\t' + action.__repr__() + ',')
lst.append(')')
return '\n'.join(lst)
else:
return '%s(%s)' % (self.name, self._data)
# def __str__(self):
# """
# String to represent the action. Need to be provided by each child class.
# """
# if self._data is None:
# return [str(action) for action in self._actions]
# return str(self)
def __call__(self, data=None):
"""
Compute/read the action and return it. It is an alias to the `self.write()` method.
"""
return self.write(data)
def __len__(self):
"""
Return the total number of actions contained in this class.
Example::
s1 = JntPositionAction(robot)
s2 = s1 + JntVelocityAction(robot)
print(len(s1)) # returns 1
print(len(s2)) # returns 2
"""
if self._data is None:
return len(self._actions)
return 1
def __iter__(self):
"""
Iterator over the actions.
"""
if self.isCombinedAction():
for action in self._actions:
yield action
else:
yield self
def __contains__(self, item):
"""
Check if the action item(s) is(are) in the combined action. If the item is the data associated with the action,
it checks that it is within the bounds.
Args:
item (Action, list/tuple of action): check if given action(s) is(are) in the combined action
Example:
s1 = JntPositionAction(robot)
s2 = JntVelocityAction(robot)
s = s1 + s2
print(s1 in s) # output True
print(s2 in s1) # output False
print((s1, s2) in s) # output True
"""
# check type of item
if not isinstance(item, (Action, np.ndarray)):
raise TypeError("Expecting a action or numpy array.")
# check if action item is in the combined action
if self._data is None and isinstance(item, Action):
return (item in self._actions)
# check if action/data is within the bounds
if isinstance(item, Action):
item = item.data
# check if continuous
# if self.isContinuous():
# low, high = self.bounds()
# return np.all(low <= item) and np.all(item <= high)
# else: # discrete case
# num = self.bounds()[0]
# # check the size of data
# if item.size > 1: # array
# return (item.size < num)
# else: # one number
# return (item[0] < num)
return self.contains(item)
def __getitem__(self, key):
"""
Get the corresponding item from the action(s)
"""
# if one action, slice the corresponding action data
if len(self._actions) == 0:
return self._data[key]
# if multiple actions
if isinstance(key, int):
# get one action
return self._actions[key]
elif isinstance(key, slice):
# get multiple actions
return Action(self._actions[key])
else:
raise TypeError("Expecting an int or slice for the key, but got instead {}".format(type(key)))
def __setitem__(self, key, value):
"""
Set the corresponding item/value to the corresponding key.
Args:
key (int, slice): index of the internal action, or index/indices for the action data
value (Action, int/float, array): value to be set
"""
if self.isCombinedAction():
# set/move the action to the specified key
if isinstance(value, Action) and isinstance(key, int):
self._actions[key] = value
else:
raise TypeError("Expecting key to be an int, and value to be a action.")
else:
# set the value on the data directly
self._data[key] = value
def __add__(self, other):
"""
Combine two different actions together. In this special case, the operation is not commutable.
This is the same as taking the union of the actions.
Args:
other (Action): another action
Returns:
Action: the combined action
Examples:
s1 = JntPositionAction(robot)
s2 = JntVelocityAction(robot)
s = s1 + s2 # = Action([JntPositionAction(robot), JntVelocityAction(robot)])
s1 = Action([JntPositionAction(robot), JntVelocityAction(robot)])
s2 = Action([JntPositionAction(robot), LinkPositionAction(robot)])
s = s1 + s2 # = Action([JntPositionAction(robot), JntVelocityAction(robot), LinkPositionAction(robot)])
"""
if not isinstance(other, Action):
raise TypeError("Expecting another action, instead got {}".format(type(other)))
s1 = self._actions if self._data is None else OrderedSet([self])
s2 = other._actions if other._data is None else OrderedSet([other])
s = s1 + s2
return Action(s)
def __iadd__(self, other):
"""
Add a action to the current one.
Args:
other (Action, list/tuple of Action): other action
Examples:
s = Action()
s += JntPositionAction(robot)
s += JntVelocityAction(robot)
"""
if self._data is not None:
raise AttributeError("The current class already has some data attached to it. This operation can not be "
"applied in this case.")
self.append(other)
def __sub__(self, other):
"""
Remove the other action(s) from the current action.
:param other:
:return:
"""
if not isinstance(other, Action):
raise TypeError("Expecting another action, instead got {}".format(type(other)))
s1 = self._actions if self._data is None else OrderedSet([self])
s2 = other._actions if other._data is None else OrderedSet([other])
s = s1 - s2
if len(s) == 1: # just one element
return s[0]
return Action(s)
def __isub__(self, other):
"""
Remove one or several actions from the combined action.
Args:
other:
"""
if not isinstance(other, Action):
raise TypeError("Expecting another action, instead got {}".format(type(other)))
if self._data is not None:
raise RuntimeError("This operation is only available for a combined action")
s = other._actions if other._data is None else OrderedSet([other])
self._actions -= s
+64
View File
@@ -0,0 +1,64 @@
#!/usr/bin/env python
"""Define the OpenAI Gym Action
This defines the OpenAI Gym action such that it is compatible with the pyrobolearn framework. It notably decouples
loosely the actions from the gym environment. Specifically, the `GymAction` allows to extract the shape of the action
from the gym environment, and keep it as an attribute of the class. This can then be used by other classes such as
the various policies defined in the pyrobolearn framework.
"""
import gym
from action import Action
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class GymAction(Action):
r"""OpenAI Gym Action.
"""
def __init__(self, gym_env):
"""
Initialize the OpenAI Gym action.
Args:
gym_env (gym.Env): OpenAI gym environment
"""
# check types
if not isinstance(gym_env, gym.Env):
raise TypeError("Expecting the `gym_env` argument to be an instance of the `gym.Env` class.")
self.env = gym_env
# set data and space
space = self.env.action_space
data = space.sample()
# call super constructor
super(GymAction, self).__init__(data=data, space=space)
def _write(self, data=None):
pass
# Tests
if __name__ == '__main__':
import gym
# create environment
env = gym.make('CartPole-v1')
# create gym action
actions = GymAction(env)
# print some information
print("Action: {}".format(actions))
print("Shape: {}".format(actions.shape))
print("Space: {}".format(actions.space))
@@ -0,0 +1,9 @@
# import the basic robot actions
from robot_actions import *
# import the joint actions
from joint_actions import *
# import the link / end-effector actions
from link_actions import *
@@ -0,0 +1,126 @@
#!/usr/bin/env python
"""Define the various joint actions
This includes notably the joint positions, velocities, and force/torque actions.
"""
import numpy as np
from abc import ABCMeta
from robot_actions import RobotAction
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class JointAction(RobotAction):
r"""Joint Action
"""
__metaclass__ = ABCMeta
def __init__(self, robot, joint_ids=None):
"""
Initialize the joint action.
Args:
robot (Robot): robot instance
joint_ids (int, int[N]): joint id or list of joint ids
"""
super(JointAction, self).__init__(robot)
# get the joints of the robot
if joint_ids is None:
joint_ids = robot.getJointIds()
elif isinstance(joint_ids, int):
joint_ids = [joint_ids]
self.joints = joint_ids
# @property
# def size(self):
# return len(self.joints)
def bounds(self):
return self.robot.getJointLimits(self.joints)
class JointPositionAction(JointAction):
r"""Joint Position Action
Set the joint positions using position control.
"""
def __init__(self, robot, joint_ids=None, kp=None, kd=None):
self.kp, self.kd = kp, kd
super(JointPositionAction, self).__init__(robot, joint_ids)
self._data = robot.getJointPositions(self.joints)
def _write(self, data=None):
if data is None:
self.robot.setJointPositions(self._data, self.joints, kp=self.kp, kd=self.kd)
else:
self.robot.setJointPositions(data, self.joints, kp=self.kp, kd=self.kd)
class JointVelocityAction(JointAction):
r"""Joint Velocity Action
Set the joint velocities using velocity control.
"""
def __init__(self, robot, joint_ids=None):
super(JointVelocityAction, self).__init__(robot, joint_ids)
self._data = robot.getJointVelocities(self.joints)
def _write(self, data=None):
if data is None:
self.robot.setJointVelocities(self._data, self.joints)
else:
self.robot.setJointVelocities(data, self.joints)
class JointForceAction(JointAction):
r"""Joint Force Action
Set the joint force/torque using force/torque control.
"""
def __init__(self, robot, joint_ids=None, f_min=-np.infty, f_max=np.infty):
super(JointForceAction, self).__init__(robot, joint_ids)
self._data = robot.getJointTorques(self.joints)
self.f_min = f_min
self.f_max = f_max
def _write(self, data=None):
if data is None:
self.robot.setJointTorques(self._data, self.joints)
else:
data = np.clip(data, self.f_min, self.f_max)
self.robot.setJointTorques(data, self.joints)
class JointAccelerationAction(JointAction):
r"""Joint Acceleration Action
Set the joint accelerations using force/torque control. In order to produce the given joint accelerations,
we use inverse dynamics which given the joint accelerations produce the corresponding joint forces/torques
to be applied.
"""
def __init__(self, robot, joint_ids=None, a_min=-np.infty, a_max=np.infty):
super(JointAccelerationAction, self).__init__(robot, joint_ids)
self._data = robot.getJointAccelerations(self.joints)
self.a_min = a_min
self.a_max = a_max
def _write(self, data=None):
if data is None:
self.robot.setJointAccelerations(self._data, self.joints)
else:
data = np.clip(data, self.a_min, self.a_max)
self.robot.setJointAccelerations(data, self.joints)
@@ -0,0 +1,84 @@
#!/usr/bin/env python
"""Define the various link / end-effector actions
This includes notably the link positions, velocities, and force/torque actions.
"""
from abc import ABCMeta
from robot_actions import RobotAction
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class LinkAction(RobotAction):
r"""Link Action
"""
__metaclass__ = ABCMeta
def __init__(self, robot, link_ids=None):
"""
Initialize the joint action.
Args:
robot (Robot): robot instance
jointIds (int, int[N]): joint id or list of joint ids
"""
super(LinkAction, self).__init__(robot)
# get the joints of the robot
if link_ids is None:
link_ids = robot.getLinkIds()
self.links = link_ids
class LinkPositionAction(LinkAction):
r"""Link position action
Set the link position(s) using IK.
"""
def __init__(self, robot, link_ids=None):
super(LinkPositionAction, self).__init__(robot, link_ids)
def _write(self, data=None):
if data is None:
self.robot.setLinkPositions(self.links, self._data)
else:
self.robot.setLinkPositions(self.links, data)
########################
# End Effector Actions #
########################
class EndEffectorAction(LinkAction):
def __init__(self, robot, end_effector_ids=None):
if end_effector_ids is None:
end_effector_ids = robot.getEndEffectorIds()
super(EndEffectorAction, self).__init__(robot, end_effector_ids)
class EndEffectorPositionAction(EndEffectorAction):
def __init__(self, robot, end_effector_ids=None):
super(EndEffectorPositionAction, self).__init__(robot, end_effector_ids)
class EndEffectorVelocityAction(EndEffectorAction):
def __init__(self, robot, end_effector_ids=None):
super(EndEffectorVelocityAction, self).__init__(robot, end_effector_ids)
class EndEffectorForceAction(EndEffectorAction):
def __init__(self, robot, end_effector_ids=None):
super(EndEffectorForceAction, self).__init__(robot, end_effector_ids)
@@ -0,0 +1,48 @@
#!/usr/bin/env python
"""Define the basic robot actions
Check also the joint, link, and end-effector actions.
Dependencies:
- `pyrobolearn.actions`
- `pyrobolearn.robots`
"""
from abc import ABCMeta
from pyrobolearn.actions import Action
from pyrobolearn.robots import Robot
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class RobotAction(Action):
"""Robot Action class.
This class defines how to map an action produced by the policy to the robot.
Each action of this type is associated to a particular robot.
"""
__metaclass__ = ABCMeta
def __init__(self, robot):
super(RobotAction, self).__init__()
if not isinstance(robot, Robot):
raise TypeError("The 'robot' parameter has to be an instance of Robot")
self._robot = robot
@property
def robot(self):
return self._robot
def isDiscrete(self):
return False
def isContinuous(self):
return True
+3
View File
@@ -0,0 +1,3 @@
## Rewards
In this folder, we define the most common rewards used in reinforcement learning and optimization.
+3
View File
@@ -0,0 +1,3 @@
# import rewards
from reward import *
+449
View File
@@ -0,0 +1,449 @@
#!/usr/bin/env python
"""Define most common costs used in reinforcement learning, control, and optimization.
A cost is defined as an objective that penalizes a certain behavior.
The `Cost` class inherits from the `Objective` class.
To see the documentation of a certain cost in the python interpreter, just type:
```python
from reward import <Cost>
print(<Cost>.__doc__)
```
Dependencies:
- `pyrobolearn.states`
- `pyrobolearn.actions`
"""
import copy
import numpy as np
import torch
# from objective import Objective
from reward import Reward
# class Cost(Objective):
class Cost(Reward):
r"""Abstract `Cost` class which inherits from the `Objective` class, and is set to be minimized.
Every classes that defines a cost inherits from this one. A cost is defined as an objective that
penalizes a certain behavior.
"""
def __init__(self):
super(Cost, self).__init__()
# super(Cost, self).__init__(maximize=False)
def logistic_kernel_function(error, alpha):
r"""
The logistic kernel function :math:`K(x|\alpha) = \frac{1}{(e^{\alpha x} + 2 + e^{-\alpha x})} \in [-0.25,0)`.
Args:
error (Cost, float): cost (e.g. error term)
alpha (float): sensitivity
Return:
callable, float: logistic kernel function
"""
if callable(error):
y = copy.copy(error) # shallow copy
y.compute = lambda: 1. / (np.exp(alpha * error() + 2. + np.exp(- alpha * error)))
return y
else:
return 1. / (np.exp(alpha * error) + 2. + np.exp(- alpha * error))
def min_angle_difference(q1, q2):
r"""
Return the minimum angle difference between two angles.
Args:
q1 (Cost, float): first angle
q2 (Cost, float): second angle
Returns:
callable, float: minimum angle difference
"""
return
class AngularVelocityErrorCost(Cost):
r"""Angular Velocity Error Cost
Return the angular velocity error cost which is defined in [1] as :math:`K(|\hat{\omega} - \omega|, alpha)` where
:math:`K` is the logistic kernel function given by
:math:`K(x|\alpha) = \frac{1}{(e^{\alpha x} + 2 + e^{-\alpha x})} \in [-0.25,0)`, with :math:`\alpha > 0.` being
the sensitivity factor, and :math:`\hat{\omega}` and :math:`\omega` being the target and current angular velocity.
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, angular_velocity_state, target_angular_velocity_state, sensitivity):
super(AngularVelocityErrorCost, self).__init__()
self.state = angular_velocity_state
self.target_state = target_angular_velocity_state
self.sensitivity = sensitivity
def compute(self):
error = np.linalg.norm(self.target_state.data - self.state.data)
return - logistic_kernel_function(error, self.sensitivity)
class LinearVelocityErrorCost(Cost):
r"""Linear Velocity Error Cost
Return the linear velocity error cost which is defined in [1] as :math:`K(|\hat{v} - v|, alpha)` where
:math:`K` is the logistic kernel function given by
:math:`K(x|\alpha) = \frac{1}{(e^{\alpha x} + 2 + e^{-\alpha x})} \in [-0.25,0)`, with :math:`\alpha > 0.` being
the sensitivity factor, and :math:`\hat{v}` and :math:`v` being the target and current linear velocity.
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, velocity_state, velocity_target_state, sensitivity):
super(LinearVelocityErrorCost, self).__init__()
self.state = velocity_state
self.target_state = velocity_target_state
self.sensitivity = sensitivity
def compute(self):
error = np.linalg.norm(self.target_state.data - self.state.data)
return - logistic_kernel_function(error, self.sensitivity)
class HeightCost(Cost):
r"""Height Cost
Height cost defined in [1] as :math:`cost = 1.0` if height < threshold, otherwise 0.
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, height_state, threshold):
super(HeightCost, self).__init__()
self.height = height_state
self.threshold = threshold
def compute(self):
if self.height.data < self.threshold:
return -1.
return 0
class JointPositionErrorCost(Cost):
r"""Joint Position Error Cost
Return the joint position error as defined in [1] as :math:`d(\hat{\phi}, \phi) \in [0, \pi]` where :math:`d(.,.)`
is the minimum angle difference between two angles, and :math:`\hat{\phi}` and :math:`\phi` are the target and
current angles.
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, joint_state, target_joint_state):
super(JointPositionErrorCost, self).__init__()
self.state = joint_state
self.target_state = target_joint_state
def compute(self):
return - min_angle_difference(self.state.data, self.target_state.data)
class OrientationGravityCost(Cost):
r"""Orientation Gravity Cost
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, gravity_state, gravity_vector=[0., 0., -1.]):
super(OrientationGravityCost, self).__init__()
self.gravity_state = gravity_state
self.gravity = np.array(gravity_vector)
def compute(self):
return np.linalg.norm(self.gravity_state.data - self.gravity)
class TorqueCost(Cost):
r"""Torque Cost
Return the cost due to the torques; :math:`cost = ||\tau||^2`.
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, torque_state):
super(TorqueCost, self).__init__()
self.tau = torque_state
def compute(self):
return - np.sum(self.tau.data**2)
class PowerCost(Cost):
r"""Power Consumption Cost
Return the power consumption cost, where the power is computed as the torque times the velocity.
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, torque_state, velocity_state):
super(PowerCost, self).__init__()
self.tau = torque_state
self.vel = velocity_state
def compute(self):
return - np.sum(np.maximum(self.tau.data * self.vel.data, 0))
class JointAccelerationCost(Cost):
r"""Joint Acceleration Cost
Return the joint acceleration cost defined notably in [1] as :math:`cost = ||\ddot{q}||^2`
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, joint_acceleration_state):
super(JointAccelerationCost, self).__init__()
self.ddq = joint_acceleration_state
def compute(self):
return - np.sum(self.ddq.data**2)
class JointSpeedCost(Cost):
r"""Joint Speed Cost
Return the joint speed cost as computed in [1].
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, joint_velocity_state, max_joint_speed=None):
super(JointSpeedCost, self).__init__()
self.dq = joint_velocity_state
self.dq_max = max_joint_speed
if max_joint_speed is None:
self.dq_max = joint_velocity_state.max
def compute(self):
return - np.sum(np.maximum(self.dq_max - np.abs(self.dq.data), 0)**2)
class BodyImpulseCost(Cost):
r"""Body Impulse Cost
Return the body impulse cost as computed in [1].
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, robot):
super(BodyImpulseCost, self).__init__()
self.robot = robot
def compute(self):
return
class BodySlippageCost(Cost):
r"""Body Slippage Cost
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self):
super(BodySlippageCost, self).__init__()
def compute(self):
pass
class FootSlippageCost(Cost):
r"""Foot Slippage Cost
'In mechanics, a unilateral contact denotes a mechanical constraint which prevents penetration between two bodies.
These bodies may be rigid or flexible. A unilateral contact is usually associated with a gap function g which
measures the distance between the two bodies and a contact force' [2]
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
[2] Unilateral Contact (Wikipedia): https://en.wikipedia.org/wiki/Unilateral_contact
"""
def __init__(self):
super(FootSlippageCost, self).__init__()
def compute(self):
pass
class FootClearanceCost(Cost):
r"""Foot Clearance Cost
'In mechanics, a unilateral contact denotes a mechanical constraint which prevents penetration between two bodies.
These bodies may be rigid or flexible. A unilateral contact is usually associated with a gap function g which
measures the distance between the two bodies and a contact force' [2]
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
[2] Unilateral Contact (Wikipedia): https://en.wikipedia.org/wiki/Unilateral_contact
"""
def __init__(self):
super(FootClearanceCost, self).__init__()
def compute(self):
pass
class SelfCollisionCost(Cost):
r"""Self Collision Cost
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self):
super(SelfCollisionCost, self).__init__()
def compute(self):
pass
class ActionDifferenceCost(Cost):
r"""Action Difference Cost
References:
[1] "Robust Recovery Controller for a Quadrupedal Robot using Deep Reinforcement Learning", Lee et al., 2019
"""
def __init__(self, action):
super(ActionDifferenceCost, self).__init__()
self.action = action
def compute(self):
return - (self.action.data - self.action.prev_data)**2
class PhysicsViolationCost(Cost):
"""Physics Violation Cost.
This cost defines ...
It was formally defined in [1]. It accepts two arguments.
[1] 'Automated Discovery and Learning of Complex Movement Behaviors' (PhD thesis), Mordatch, 2015
.. seealso: `cio.py` in 'pyrobolearn/optim' which uses this cost.
"""
def __init__(self):
super(PhysicsViolationCost, self).__init__()
class ContactInvariantCost(Cost):
"""Contact Invariant Cost.
This cost defines ... and is used in Contact Invariant Optimization (CIO).
It was formally defined in [1]. It accepts two arguments.
[1] 'Automated Discovery and Learning of Complex Movement Behaviors' (PhD thesis), Mordatch, 2015
.. seealso: `cio.py` in 'pyrobolearn/optim' which uses this cost.
"""
def __init__(self):
super(ContactInvariantCost, self).__init__()
class PowerConsumptionCost(Cost):
"""Power Consumption Cost.
It penalizes power consumption using u^TWu where W is a weight matrix, and u is the control vector.
"""
def __init__(self):
super(PowerConsumptionCost, self).__init__()
def loss(self, robot):
return np.dot(robot.getJointTorques(), robot.getJointVelocities())
class DistanceCost(Cost):
"""Distance Cost.
It penalizes the distance between 2 objects. One of the 2 objects must be movable in order for this
cost to change.
"""
def __init__(self):
super(DistanceCost, self).__init__()
def loss(self, object1, object2):
pass
class ImpactCost(Cost):
"""Impact cost.
Calculates the impact force using the kinetic energy.
"""
def __init__(self):
super(ImpactCost, self).__init__()
def loss(self, object1, object2):
pass
class DriftCost(Cost):
"""Drift cost.
Calculates the drift of a moving object wrt a direction.
"""
def __init__(self):
super(DriftCost, self).__init__()
def loss(self, object, direction):
pass
class ShakeCost(Cost):
"""Shake cost.
Calculates the
"""
def __init__(self):
super(ShakeCost, self).__init__()
def loss(self, object, direction):
pass
class SpeedCost(Cost):
"""Speed cost.
"""
def __init__(self):
pass
class JerkCost(Cost):
"""Jerk cost.
Calculates the jerk of an object.
"""
def __init__(self, object, dt):
self.object = object
self.dt = dt
class ZMPCost(Cost):
"""ZMP Cost.
"""
def __init__(self):
super(ZMPCost, self).__init__()
+121
View File
@@ -0,0 +1,121 @@
# This file defines the main `Objective` class, which is inherited by the `Reward` and `Cost` classes, and
# all subsequent child classes which define common objective/reward/cost functions.
# Objectives can be maximized or minimized, while rewards are maximized, and costs are minimized.
# Allow to define objectives with different frameworks: numpy, pytorch, tensorflow, theano,...
# That is, by giving the framework we want to use as input to the children of the `Objective` class,
# it will return the objective in the correct format. Thus we don't need to define various classes for each possible
# different frameworks.
#
# Some objectives needs to access the robot's and/or environment's information.
#
# Objectives can be added together using `+` and multiplied by a weight number using `*`
#
# To see the doc in python interpreter:
# from pyrobolearn.objectives.objective import Objective
# print(Objective.__doc__)
from abc import ABCMeta, abstractmethod
class Objective(object):
"""Abstract `Objective` class which defines the objective function to be maximized or minimized.
This class must be inherited by any classes which defines a reward or cost.
.. seealso:
"""
__metaclass__ = ABCMeta
def __init__(self, objective, maximize=True):
self._objective = objective
self.maximize = maximize
def __neg__(self):
"""
max f = min -f <--> min f = max -f
This function is useful when using optimizers which only accepts to maximize xor minimize,
thus by taking the negative of the objective, it will still respect the initial objective
to optimize.
:return: -objective
"""
return Objective(-self.objective, maximize=not self.maximize)
def __add__(self, other):
"""
Define how to add two objectives. Adding two rewards or two costs make sense,
but adding a reward with a cost does not.
By default, if one of the objectives is to be maximized while
the other one is to be minimized, it will return an objective to be maximized.
:param other:
:return:
"""
if (self.maximize == other.maximize):
return Objective(self.objective + other.objective, maximize=self.maximize)
else:
raise ValueError("Trying to add an objective to maximize with one to minimize. You probably meant: "
"objective_1 - objective_2.")
def __sub__(self, other):
"""
Substracting two rewards or two costs do not make any sense. However, substracting a reward
and a cost makes sense.
:param other:
:return:
"""
return self.__add__(-other)
def __mul__(self, other): # self * other
if other < 0:
return Objective(abs(other)*self.objective, maximize = not self.maximize)
return Objective(other * self.objective, maximize=self.maximize)
def __rmul__(self, other): # other * self
return self.__mul__(other)
@property
def objective(self):
return self._objective
@objective.setter
def objective(self, objective):
self._objective = objective
@abstractmethod
def __call__(self, *args, **kwargs):
pass
def loss_fct(self, *args, **kwargs):
"""
Returns the loss fct.
"""
pass
def loss_value(self, *args, **kwargs):
"""
Returns the loss value. It evaluate the loss function for a particular instance.
"""
pass
def fct(self, *args, **kwargs):
"""
Define the objective function.
"""
pass
@abstractmethod
def symbolic(self):
"""
Return symbolic expression of the objective function.
:return:
"""
pass
@abstractmethod
def latex(self):
"""
Returns the latex formula of the objective function.
:return:
"""
pass
+833
View File
@@ -0,0 +1,833 @@
#!/usr/bin/env python
"""Define most common rewards used in reinforcement learning and optimization.
A reward is defined as an objective that compliments/rewards a certain behavior.
The `Reward` class inherits from the `Objective` class.
To see the documentation of a certain reward in the python interpreter, just type:
```python
from reward import <Reward>
print(<Reward>.__doc__)
```
Dependencies:
- `pyrobolearn.states`
- `pyrobolearn.actions`
"""
import operator
import copy
import collections
# from objective import Objective
from pyrobolearn.states import *
from pyrobolearn.actions import *
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
# class Reward(Objective):
class Reward(object):
r"""Abstract `Reward` class which inherits from the `Objective` class, and is set to be maximized.
Every classes that defines a reward inherits from this one. A reward is defined as an objective
that compliments/rewards a certain behavior.
A reward is defined as [1]:
- r(s): given the state s, it returns the reward.
- r(s,a): given the state s and action a, it returns the reward.
- r(s,a,s'): given the state s, action a, and next state s', it returns the reward.
Note:
- In order to enable binary operators such as `add`, `multiply`, and so on, we can accomplish it using two
different approaches; the functional [2] and `eval` approach.
- Functional approach: we define for each operator, what functions to call and set them to the new created
reward.
- Eval approach: we build the string that represents the operations to carry out, and evaluate it when we
compute the reward.
Examples:
# create simulator
sim = Simulator()
# create world and load robot
world = World(sim)
robot = world.load(Robot())
# create state, action, and policy
state = JntPositionState(robot) + JntVelocityState(robot)
action = JntPositionAction(robot)
policy = Policy(state, action)
# create reward
reward = 2 * ForwardProgressReward(CoMState(robot)) + 4 * NotFallingReward(robot) + exp(FixedReward(-1))
# create env
env = Env(world, state, reward)
# create and run task with the current policy
task = Task(env, policy)
task.run(num_steps=1000) # run loop between 'env' and 'policy' for num_steps
# create algo to train the policy, and run it
algo = Algo(task)
algo.train()
# run task with the trained policy
task.run(num_steps=1000)
References:
[1] "Reinforcement Learning: An Introduction", Sutton and Barto, 1998
[2] https://turion.wordpress.com/2012/01/05/add-and-multiply-python-functions-operable-functions/
"""
def __init__(self, state=None, action=None, rewards=None):
# super(Reward, self).__init__(maximize=True)
self.state = state
self.action = action
self.rewards = rewards
# # create automatically binary operator methods
# op_names = ['__add__', '__div__', '__floordiv__', '__iadd__', '__idiv__', '__ifloordiv__', '__imod__',
# '__imul__', '__ipow__', '__isub__', '__itruediv__', '__mod__', '__mul__', '__pow__', '__radd__',
# '__rdiv__', '__rfloordiv__', '__rmod__', '__rmul__', '__rpow__', '__rsub__', '__rtruediv__',
# '__sub__', '__truediv__']
# for name in op_names:
# # define binary operator method
# def wrapper(op):
# def binary_operator(self, other):
# # built the internal list of rewards
# rewards = self._rewards if self.hasRewards() else [self]
# if isinstance(other, Reward):
# if other.hasRewards():
# rewards.extend(other._rewards)
# else:
# rewards.append(other)
#
# # create reward to return
# reward = Reward(rewards=rewards)
#
# # replace the `reward.compute` by the corresponding function
# if isinstance(other, Reward): # callable
# def compute():
# return op(self(), other())
# else:
# def compute():
# return op(self(), other)
# reward.compute = compute
# return reward
# return binary_operator
#
# op = getattr(operator, name) if not name.startswith('__r') else getattr(operator, name.replace('r','',1))
# setattr(self.__class__, name, wrapper(op))
# # create automatically binary comparison operator methods
# op_names = ['__eq__', '__ge__', '__gt__', '__le__', '__lt__', '__ne__']
# for name in op_names:
# # define binary operator method
# def wrapper(op):
# def binary_operator(self, other):
# if isinstance(other, Reward): # callable
# def compute():
# return op(self(), other())
# else:
# def compute():
# return op(self(), other)
# return compute()
# return binary_operator
#
# op = getattr(operator, name)
# setattr(self.__class__, name, wrapper(op))
# # create automatically unary operator methods
# op_names = ['__abs__', '__neg__', '__pos__']
# for name in op_names:
# # define unary method
# def wrapper(op):
# def unary_operator(self):
# reward = copy.copy(self) # shallow copy
# def compute():
# return op(self())
# reward.compute = compute
# return reward
# return unary_operator
#
# op = getattr(operator, name)
# setattr(self.__class__, name, wrapper(op))
##############
# Properties #
##############
@property
def state(self):
return self._state
@state.setter
def state(self, state):
if state is not None and not isinstance(state, State):
raise TypeError("Expecting state to be None or an instance of State.")
self._state = state
@property
def action(self):
return self._action
@action.setter
def action(self, action):
if action is not None and not isinstance(action, Action):
raise TypeError("Expecting action to be None or an instance of Action.")
self._action = action
@property
def rewards(self):
return self._rewards
@rewards.setter
def rewards(self, rewards):
if rewards is None:
rewards = []
elif isinstance(rewards, collections.Iterable):
for reward in rewards:
if not isinstance(reward, Reward):
raise TypeError("Expecting a Reward instance for each item in the iterator.")
else:
if not isinstance(rewards, Reward):
raise TypeError("Expecting rewards to be an instance of Reward.")
rewards = [rewards]
self._rewards = rewards
###########
# Methods #
###########
def has_rewards(self):
return len(self._rewards) > 0
@staticmethod
def is_maximized():
return True
def compute(self):
pass
#############
# Operators #
#############
def __repr__(self):
if not self.rewards or self.rewards is None:
return self.__class__.__name__
else:
lst = [reward.__repr__() for reward in self.rewards]
return ' + '.join(lst)
def __call__(self, *args, **kwargs):
return self.compute()
# for unary and binary operators, see `__init__()` method.
def __build_reward(self, other):
# built the internal list of rewards
rewards = self._rewards if self.has_rewards() else [self]
if isinstance(other, Reward):
if other.has_rewards():
rewards.extend(other._rewards)
else:
rewards.append(other)
return Reward(rewards=rewards)
def __get_operation(self, other, op):
if isinstance(other, Reward): # callable
def compute():
return op(self(), other())
else:
def compute():
return op(self(), other)
return compute
def __add__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__add__)
return reward
def __div__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__div__)
return reward
def __floordiv__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__floordiv__)
return reward
def __iadd__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__iadd__)
return reward
def __idiv__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__idiv__)
return reward
def __ifloordiv__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__ifloordiv__)
return reward
def __imod__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__imod__)
return reward
def __imul__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__imul__)
return reward
def __ipow__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__ipow__)
return reward
def __isub__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__isub__)
return reward
def __itruediv__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__itruediv__)
return reward
def __mod__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__mod__)
return reward
def __mul__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__mul__)
return reward
def __pow__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__pow__)
return reward
def __radd__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__add__)
return reward
def __rdiv__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__div__)
return reward
def __rfloordiv__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__floordiv__)
return reward
def __rmod__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__mod__)
return reward
def __rmul__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__mul__)
return reward
def __rpow__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__pow__)
return reward
def __rsub__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__sub__)
return reward
def __rtruediv__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__truediv__)
return reward
def __sub__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__sub__)
return reward
def __truediv__(self, other):
reward = self.__build_reward(other)
reward.compute = self.__get_operation(other, operator.__truediv__)
return reward
# binary comparison operators
def __eq__(self, other):
compute = self.__get_operation(other, operator.__eq__)
return compute()
def __ge__(self, other):
compute = self.__get_operation(other, operator.__ge__)
return compute()
def __gt__(self, other):
compute = self.__get_operation(other, operator.__gt__)
return compute()
def __le__(self, other):
compute = self.__get_operation(other, operator.__le__)
return compute()
def __lt__(self, other):
compute = self.__get_operation(other, operator.__lt__)
return compute()
def __ne__(self, other):
compute = self.__get_operation(other, operator.__ne__)
return compute()
# unary operators
def __abs__(self):
reward = copy.copy(self) # shallow copy
reward.compute = lambda: operator.__abs__(self())
return reward
def __neg__(self):
reward = copy.copy(self) # shallow copy
reward.compute = lambda: operator.__neg__(self())
return reward
def __pos__(self):
reward = copy.copy(self) # shallow copy
reward.compute = lambda: operator.__pos__(self())
return reward
######################################
# mathematical operations on rewards #
######################################
# import math
# import numpy as np
# names = []
# name_lst = [name for name in dir(math) if '__' not in name and name in dir(np)]
# name_lst.remove('e')
# for name in name_lst:
# op = getattr(np, name)
# if callable(op):
# try:
# op(1)
# except (ValueError, TypeError) as e:
# pass
# else:
# names.append(name)
# # define the following mathematical operations automatically
# names = ['ceil', 'cos', 'cosh', 'degrees', 'exp', 'expm1', 'fabs', 'floor', 'frexp', 'isinf', 'isnan', 'log', 'log10',
# 'log1p', 'modf', 'radians', 'sin', 'sinh', 'sqrt', 'tan', 'tanh', 'trunc']
# for name in names:
# def wrapper(op):
# def fct(x):
# if callable(x):
# y = copy.copy(x) # shallow copy
# def f():
# return op(x())
# y.compute = f
# return y
# else:
# return op(x)
# return fct
#
# globals()[name] = wrapper(getattr(np, name))
def ceil(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.ceil(x())
return y
else:
return np.ceil(x)
def cos(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.cos(x())
return y
else:
return np.cos(x)
def cosh(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.cosh(x())
return y
else:
return np.cosh(x)
def degrees(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.degrees(x())
return y
else:
return np.degrees(x)
def exp(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.exp(x())
return y
else:
return np.exp(x)
def expm1(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.expm1(x())
return y
else:
return np.expm1(x)
def floor(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.floor(x())
return y
else:
return np.floor(x)
def frexp(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.frexp(x())
return y
else:
return np.frexp(x)
def isinf(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.isinf(x())
return y
else:
return np.isinf(x)
def isnan(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.isnan(x())
return y
else:
return np.isnan(x)
def log(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.log(x())
return y
else:
return np.log(x)
def log10(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.log10(x())
return y
else:
return np.log10(x)
def log1p(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.log1p(x())
return y
else:
return np.log1p(x)
def modf(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.modf(x())
return y
else:
return np.modf(x)
def radians(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.radians(x())
return y
else:
return np.radians(x)
def sin(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.sin(x())
return y
else:
return np.sin(x)
def sinh(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.sinh(x())
return y
else:
return np.sinh(x)
def sqrt(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.sqrt(x())
return y
else:
return np.sqrt(x)
def tan(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.tan(x())
return y
else:
return np.tan(x)
def tanh(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.tanh(x())
return y
else:
return np.tanh(x)
def trunc(x):
if callable(x):
y = copy.copy(x) # shallow copy
y.compute = lambda: np.trunc(x())
return y
else:
return np.trunc(x)
##############################################################
# Rewards #
##############################################################
class FixedReward(Reward):
r"""Fixed reward.
This is a dummy class which always returns a fixed reward. This is fixed initially.
"""
def __init__(self, value):
super(FixedReward, self).__init__()
if not isinstance(value, (int, float)):
raise TypeError("Expecting a number")
self.value = value
def __repr__(self):
return '%s(%s)' % (self.__class__.__name__, str(self.value))
def compute(self):
return self.value
class FunctionalReward(Reward):
r"""Functional reward.
This is a reward class which calls a given function/class to compute the reward.
"""
def __init__(self, function):
super(FunctionalReward, self).__init__()
self.function = function
def __repr__(self):
return self.function.__name__
def compute(self):
return self.function()
class ForwardProgressReward(Reward):
r"""Forward progress reward
Compute the forward progress based on a forward direction, a previous and current positions.
"""
def __init__(self, state, direction=(1, 0, 0), normalize=False):
super(ForwardProgressReward, self).__init__(state=state)
# if direction is None:
# # takes the robot initial direction
# #direction = ...
# #init_pos
# pass
# if isinstance(direction, np.ndarray):
# pass
# elif isinstance(direction, Robot):
# # takes the
# pass
#
# self.direction = direction
# #self.init_pos = init_pos
self.direction = self.normalize(np.array(direction))
# TODO uncomment
# if not isinstance(state, (PositionState, BasePositionState)):
# raise ValueError("Expecting state to be a PositionState or BasePositionState")
self.init_pos = np.copy(self.state._data)
self.value = 0
@staticmethod
def normalize(x):
"""
Normalize the given vector.
"""
if np.allclose(x, 0):
return x
return x / np.linalg.norm(x)
def compute(self):
curr_pos = self.state._data
delta_pos = curr_pos - self.init_pos
self.value = self.direction.dot(delta_pos)
# self.value = curr_pos[0] - self.init_pos[0]
self.init_pos = np.copy(curr_pos)
return self.value
class DirectiveReward(Reward):
r"""Directive Reward
Provide reward if the vector state is in the specified direction. Specifically, it computes the dot product
between the state vector and the specified direction.
If normalize, the reward is between -1 and 1.
"""
def __init__(self, state, direction=(1, 0, 0), normalize=True):
super(DirectiveReward, self).__init__(state=state)
self.normalize = normalize
if self.normalize:
self.direction = self.norm(np.array(direction))
# TODO uncomment
# if not isinstance(state, (PositionState, BasePositionState)):
# raise ValueError("Expecting state to be a PositionState or BasePositionState")
self.value = 0
@staticmethod
def norm(x):
"""
Normalize the given vector.
"""
if np.allclose(x, 0):
return x
return x / np.linalg.norm(x)
def compute(self):
pos = self.state._data
if self.normalize:
pos = self.norm(pos)
self.value = self.direction.dot(pos)
return self.value
class L2SimilarityReward(Reward):
"""
Compute the square of the L2 norm between two vectors.
"""
def __init__(self):
super(L2SimilarityReward, self).__init__()
def value(self, vector1, vector2):
return np.dot(vector1, vector2)
class ImitationReward(Reward):
def __init__(self, human, robot):
super(ImitationReward, self).__init__()
self.human = human # instance of HumanKinematic class
self.robot = robot # instance of Robot class
def compute(self):
# check
pass
class GymReward(Reward):
r"""OpenAI Gym reward
This provides a wrapper
"""
def __init__(self, value):
super(GymReward, self).__init__()
if not isinstance(value, (int, float)):
raise TypeError("Expecting a number")
self.value = value
def __repr__(self):
return '%s(%s)' % (self.__class__.__name__, str(self.value))
def compute(self):
return self.value
# Test
if __name__ == '__main__':
reward = 2*FixedReward(10) + FixedReward(3)**2 - 10
reward += FixedReward(2)
print(reward())
print(isinstance(reward, Reward))
print(reward.rewards)
reward = FixedReward(-10)
print(reward())
reward = abs(reward)
print(reward())
print(FixedReward(10) == FixedReward(10))
print('')
reward = FixedReward(2) + FixedReward(1)
print(reward())
reward = cos(reward)
print(reward())
# print(type(reward))
# print(reward.rewards)
+3
View File
@@ -0,0 +1,3 @@
## Simulators
This folder contains the APIs to the various simulators. Currently, the main simulator being supported is PyBullet. Work is under progress for Gazebo+ROS, and OpenSIM.
+11
View File
@@ -0,0 +1,11 @@
## States
The `State` is returned by the environment and given to the policy. The state might include information about the state of one or several objects in the world, including robots.
It is the main bridge between the robots/objects in the environment and the policy. Specifically, it is given as an input to the policy which knows how to feed the state to the learning model. Usually, the user only has to instantiate a child of this class, and give it to the policy and environment, and that's it. In addition to the policy, the state can be given to a controller, dynamic model, value estimator, reward function, and so on.
To allow the framework to be modular, we favor composition over inheritance [1] leading the state to be decoupled from notions such as the environment, policy, rewards, etc. This class also describes the `state_space` which has initially been defined in `gym.Env` [2].
References:
[1] "Wikipedia: Composition over Inheritance", https://en.wikipedia.org/wiki/Composition_over_inheritance
[2] "OpenAI gym": https://gym.openai.com/ and https://github.com/openai/gym
+18
View File
@@ -0,0 +1,18 @@
# import state
from state import State
# import basic states
from basic_states import *
# import object states
from object_states import *
# import time/count states
from time_states import *
# import robot states
from robot_states import *
# import gym states
from gym_states import *
+137
View File
@@ -0,0 +1,137 @@
#!/usr/bin/env python
"""Define basic states
This includes notably the fixed, functional, and counter states.
"""
import numpy as np
from state import State
from pyrobolearn.actions import Action
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class FixedState(State):
r"""Fixed State.
This is a dummy fixed state which always returns the value it was initialized with.
"""
def __init__(self, value):
super(FixedState, self).__init__(data=value)
class FunctionalState(State):
r"""Functional State.
This is a state which accepts a function which has to output the data.
"""
def __init__(self, function, *args, **kwargs):
self.function = function
self.args, self.kwargs = args, kwargs
data = function(*args, **kwargs) # call one time to get data
super(FunctionalState, self).__init__(data=data)
def _reset(self):
self.data = self.function(*self.args, **self.kwargs)
def _read(self):
self.data = self.function(*self.args, **self.kwargs)
class CounterState(State):
r"""Counter State.
Counts the number of time this step has been called.
"""
def __init__(self, cnt=-1):
self.cnt = cnt
if isinstance(cnt, int):
cnt = np.array([cnt])
if not (isinstance(cnt, np.ndarray) and cnt.size == 1 and len(cnt.shape) == 1
and cnt.dtype.kind in np.typecodes['AllInteger']):
raise TypeError("Expecting an int, or a numpy array (integer) with size 1")
super(CounterState, self).__init__(data=cnt)
def _reset(self):
self.data = self.cnt
def _read(self):
self.data = self._data + 1
class PreviousActionState(State):
r"""Previous Action State
This state copies the previous action.
"""
def __init__(self, action):
if not isinstance(action, Action):
raise TypeError("Expecting the action to be an instance of Action, instead got {}".format(action))
self.action = action
super(PreviousActionState, self).__init__()
def _reset(self):
self.data = self.action.data
def _read(self):
self.data = self.action.data
# Tests
if __name__ == '__main__':
s1 = FixedState([1, 2])
s2 = FixedState(3)
s3 = FixedState([4, 5, 6])
s = s1 + s2 + s2 + s3 + s1
fused = s1 & s2
print("\nStates:")
print("s1 = {}".format(s1))
print("s2 = {}".format(s2))
print("s3 = {}".format(s3))
print("s = s1 + s2 + s2 + s3 + s1 = {}".format(s))
print("s1.fuse() = {}".format(s1.fuse()))
print("s.fuse() = {}".format(s.fuse()))
print("fused = s1 & s2 = {}".format(fused))
print("\nSome dimensions:")
print("s.shape: {}".format(s.shape))
print("s.dimension: {}".format(s.dimension))
print("s.maxDimension: {}".format(s.maxDimension()))
print("s.size: {}".format(s.size))
print("s.totalSize: {}".format(s.totalSize()))
print("len(s) = {}".format(len(s)))
print("len(s1) = {}".format(len(s1)))
# print(s2 + s1)
print("\nIndexing: ")
print("s[0] = {}".format(s[0]))
print("s[1:3] = {}".format(s[1:3]))
print("s1 = {}".format(s1))
s1[1] = 7
print("s1[1] = 7 --> {}".format(s1))
print("s = {}".format(s))
s[0] = FixedState([8, 9, 10, 11])
print("s[0] = [8,9] --> {}".format(s))
# for state in s:
# print(state)
print("\nDifference: ")
print("s - s2 = {}".format(s - s2))
s = CounterState()
print("\nCounter State:")
print(s.reset())
for i in range(10):
print(s())
+72
View File
@@ -0,0 +1,72 @@
#!/usr/bin/env python
"""Define the OpenAI Gym State
This defines the OpenAI Gym state such that it is compatible with the pyrobolearn framework. It notably decouples
loosely the states from the gym environment. Specifically, the `GymState` allows to extract the shape of the state
from the gym environment, and keep it as an attribute of the class. This can then be used by other classes such as
the various policies defined in the pyrobolearn framework.
"""
import gym
from state import State
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class GymState(State):
r"""OpenAI Gym State.
This defines the OpenAI Gym state such that it is compatible with the pyrobolearn framework. It notably decouples
loosely the states from the gym environment. Specifically, the `GymState` allows to extract the shape of the state
from the gym environment, and keep it as an attribute of the class. This can then be used by other classes such as
the various policies defined in the pyrobolearn framework.
See Also: `GymEnv`
"""
def __init__(self, gym_env):
"""
Initialize the OpenAI Gym state.
Args:
gym_env (gym.Env): OpenAI gym environment
"""
# check types
if not isinstance(gym_env, gym.Env):
raise TypeError("Expecting the `gym_env` argument to be an instance of the `gym.Env` class.")
self.env = gym_env
# set data and space
space = self.env.observation_space
data = space.sample()
# call super constructor
super(GymState, self).__init__(data=data, space=space)
def _reset(self):
pass
def _read(self):
pass
# Tests
if __name__ == '__main__':
# create environment
env = gym.make('CartPole-v1')
# create gym state
states = GymState(env)
# print some information
print("State: {}".format(states))
print("Shape: {}".format(states.shape))
print("Space: {}".format(states.space))
+77
View File
@@ -0,0 +1,77 @@
#!/usr/bin/env python
"""Define the various joint states
This includes notably the joint positions, velocities, and force/torque states.
"""
from abc import ABCMeta, abstractmethod
from state import State
from pyrobolearn.worlds import World
from pyrobolearn.robots import Object
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class ObjectState(State):
"""Object state (abstract)
"""
__metaclass__ = ABCMeta
def __init__(self, obj, world=None):
super(ObjectState, self).__init__()
if not isinstance(obj, (Object, int)):
raise TypeError("Expecting an instance of Object, or an identifier from the simulator/world.")
if isinstance(obj, int):
if not isinstance(world, World):
# try to look for the world in global variables
if 'world' in globals() and isinstance(globals()['world'], World): # O(1)
world = globals()['world']
else:
raise ValueError("When giving the object identifier, the world need to be given as well.")
obj = Object(world.getSimulator(), obj)
self.obj = obj
@abstractmethod
def _read(self):
pass
class PositionState(ObjectState):
"""Position of an object.
"""
def __init__(self, obj, world=None):
super(PositionState, self).__init__(obj, world)
self.data = self.obj.position
def _read(self):
self.data = self.obj.position
class OrientationState(ObjectState):
"""Orientation of an object.
"""
def __init__(self, obj, world=None):
super(OrientationState, self).__init__(obj, world)
self.data = self.obj.orientation
def _read(self):
self.data = self.obj.orientation
class VelocityState(ObjectState):
"""Velocity of an object.
"""
def __init__(self, obj, world=None):
super(VelocityState, self).__init__(obj, world)
self.data = self.obj.velocity
def _read(self):
self.data = self.obj.velocity
@@ -0,0 +1,12 @@
# import the basic robot states
from robot_states import *
# import the joint states
from joint_states import *
# import the link states
from link_states import *
# import the sensor states
from sensor_states import *
@@ -0,0 +1,97 @@
#!/usr/bin/env python
"""Define the various joint states
This includes notably the joint positions, velocities, and force/torque states.
"""
from abc import ABCMeta
from robot_states import RobotState
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class JointState(RobotState):
r"""Joint State of a robot (abstract class).
"""
__metaclass__ = ABCMeta
def __init__(self, robot, joint_ids=None):
"""
Initialize the joint state.
Args:
robot (Robot): robot instance
joint_ids (int, int[N]): joint id or list of joint ids
"""
super(JointState, self).__init__(robot)
# get the joints of the robot
if joint_ids is None:
joint_ids = robot.getJointIds()
elif isinstance(joint_ids, int):
joint_ids = [joint_ids]
self.joints = joint_ids
# read the data
self._read()
class JointPositionState(JointState):
r"""Joint Position State
Return the joint positions as the state.
"""
def __init__(self, robot, joint_ids=None):
super(JointPositionState, self).__init__(robot, joint_ids)
def _read(self):
self._data = self.robot.getJointPositions(self.joints)
class JointVelocityState(JointState):
r"""Joint Velocity State
Return the joint velocities as the state.
"""
def __init__(self, robot, joint_ids=None):
super(JointVelocityState, self).__init__(robot, joint_ids)
def _read(self):
self._data = self.robot.getJointVelocities(self.joints)
class JointForceTorqueState(JointState):
r"""Joint Force Torque State
Return the joint force and torques as the state.
"""
def __init__(self, robot, joint_ids=None):
super(JointForceTorqueState, self).__init__(robot, joint_ids)
def _read(self):
self._data = self.robot.getJointTorques(self.joints)
class JointAccelerationState(JointState):
r"""Joint Acceleration State.
Return the joint accelerations as the state. In order to produce the joint accelerations, we first read the
joint torques and then applied forward dynamics to get the corresponding joint accelerations.
"""
def __init__(self, robot, joint_ids=None):
super(JointAccelerationState, self).__init__(robot, joint_ids)
def _read(self):
self._data = self.robot.getJointAccelerations(self.joints)
@@ -0,0 +1,109 @@
#!/usr/bin/env python
"""Define the various link states
This includes notably the link positions and velocities.
"""
from abc import ABCMeta
from robot_states import RobotState
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class LinkState(RobotState):
r"""Link state of a robot
"""
__metaclass__ = ABCMeta
def __init__(self, robot, link_ids=None):
"""
Initialize the link state.
Args:
robot (Robot): robot instance
link_ids (int, int[N]): link id or list of link ids
"""
super(LinkState, self).__init__(robot)
# get links from robot
if link_ids is None:
link_ids = robot.getlink_ids()
self.links = link_ids
# read the data
self._read()
class LinkPositionState(LinkState):
r"""Link Position state
"""
def __init__(self, robot, link_ids=None, wrt_link_id=None):
self.wrt_link_id = wrt_link_id
super(LinkPositionState, self).__init__(robot, link_ids)
def _read(self):
self._data = self.robot.getLinkPositions(self.links, wrtLinkId=self.wrt_link_id)
class LinkWorldPositionState(LinkState):
r"""Link World Position state
"""
def __init__(self, robot, link_ids=None):
super(LinkWorldPositionState, self).__init__(robot, link_ids)
def _read(self):
self._data = self.robot.getLinkPositions(self.links)
class LinkOrientationState(LinkState):
r"""Link Orientation state
"""
def __init__(self, robot, link_ids=None):
super(LinkOrientationState, self).__init__(robot, link_ids)
def _read(self): # TODO: convert
self._data = self.robot.getLinkOrientations(self.links)
class LinkVelocityState(LinkState):
r"""Link velocity state
"""
def __init__(self, robot, link_ids=None):
super(LinkVelocityState, self).__init__(robot, link_ids)
def _read(self):
self._data = self.robot.getLinkVelocities(self.links)
class LinkLinearVelocityState(LinkState):
r"""Link linear velocity state
"""
def __init__(self, robot, link_ids=None):
super(LinkLinearVelocityState, self).__init__(robot, link_ids)
def _read(self):
self._data = self.robot.getLinkLinearVelocities(self.links)
class LinkAngularVelocityState(LinkState):
r"""Link angular velocity state
"""
def __init__(self, robot, link_ids=None):
super(LinkAngularVelocityState, self).__init__(robot, link_ids)
def _read(self):
self._data = self.robot.getLinkAngularVelocities(self.links)
@@ -0,0 +1,123 @@
#!/usr/bin/env python
"""Define the basic robot states
Check also the joint, link, and sensor states.
Dependencies:
- `pyrobolearn.states`
- `pyrobolearn.robots`
"""
from abc import ABCMeta, abstractmethod
import numpy as np
from pyrobolearn.states import State
from pyrobolearn.robots import Robot
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class RobotState(State):
r"""Robot state (abstract)
This class is inherited by all the states that described a robot state.
"""
__metaclass__ = ABCMeta
def __init__(self, robot):
"""
Initialize the robot state.
Args:
robot (Robot): instance of Robot which allows to access to the robot state
"""
super(RobotState, self).__init__()
if not isinstance(robot, Robot):
raise TypeError("The 'robot' parameter has to be an instance of Robot")
self._robot = robot
@property
def robot(self):
"""Return the robot instance"""
return self._robot
@abstractmethod
def _read(self):
pass
class BasePositionState(RobotState):
r"""Base position state
This is the state that returns the base position with respect to the world frame.
"""
def __init__(self, robot):
super(BasePositionState, self).__init__(robot)
self._read()
def _read(self):
self._data = self.robot.getBasePosition()
class BaseHeightState(RobotState):
r"""Base height state
This is the state that returns the base height with respect to the world frame.
"""
def __init__(self, robot):
super(BaseHeightState, self).__init__(robot)
self._read()
def _read(self):
self._data = np.array([self.robot.getBasePosition()[-1]])
class BaseOrientationState(RobotState):
r"""Base orientation state
This is the state that returns the base orientation with respect to the world frame.
"""
def __init__(self, robot):
super(BaseOrientationState, self).__init__(robot)
self._read()
def _read(self):
self._data = self.robot.getBaseOrientation(convert_to_numpy_quaternion=False)
class BaseLinearVelocityState(RobotState):
r"""Base linear velocity state
This is the state that returns the base linear velocity with respect to the world frame.
"""
def __init__(self, robot):
super(BaseLinearVelocityState, self).__init__(robot)
self._read()
def _read(self):
self._data = self.robot.getBaseLinearVelocity()
class BaseAngularVelocityState(RobotState):
r"""Base angular velocity state
This is the state that returns the base angular velocity with respect to the world frame.
"""
def __init__(self, robot):
super(BaseAngularVelocityState, self).__init__(robot)
self._read()
def _read(self):
self._data = self.robot.getBaseAngularVelocity()
@@ -0,0 +1,64 @@
#!/usr/bin/env python
"""Define the various sensor states
This includes notably the camera, contact, IMU, force/torque sensors and others.
"""
from abc import ABCMeta
from robot_states import RobotState
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class SensorState(RobotState): # TODO: define refresh_rate & frequency
r"""Sensor state (abstract class)
"""
__metaclass__ = ABCMeta
def __init__(self, robot):
super(SensorState, self).__init__(robot)
class CameraState(SensorState):
r"""Camera state
"""
def __init__(self, robot, camera=None):
super(CameraState, self).__init__(robot)
def _read(self):
pass
class ContactState(SensorState):
r"""Contact state
Return the contact states between a link of the robot and an object in the world (including the floor).
"""
def __init__(self, robot, contacts=None):
super(ContactState, self).__init__(robot)
def _read(self):
pass
class FeetContactState(ContactState):
r"""Feet Contact State
Return the contact states between
"""
def __init__(self, robot, contacts=None):
super(FeetContactState, self).__init__(robot, contacts)
def _read(self):
pass
+798
View File
@@ -0,0 +1,798 @@
#!/usr/bin/env python
"""Define the State class.
This file defines the `State` class, which is returned by the environment, and given as an input to several
models such as policies/controllers, dynamic transition functions, value estimators, reward/cost function, and so on.
"""
import numpy as np
import collections
from abc import ABCMeta, abstractmethod
import gym
from pyrobolearn.utils.data_structures.orderedset import OrderedSet
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class State(object):
r"""State class.
The `State` is returned by the environment and given to the policy. The state might include information
about the state of one or several objects in the world, including robots.
It is the main bridge between the robots/objects in the environment and the policy. Specifically, it is given
as an input to the policy which knows how to feed the state to the learning model. Usually, the user only has to
instantiate a child of this class, and give it to the policy and environment, and that's it.
In addition to the policy, the state can be given to a controller, dynamic model, value estimator, reward function,
and so on.
To allow our framework to be modular, we favor composition over inheritance [1] leading the state to be decoupled
from notions such as the environment, policy, rewards, etc. This class also describes the `state_space` which has
initially been defined in `gym.Env` [2].
Note that the policy does not represent in a strict sense the robot but more its brain, the sensors and actuators
are parts of the environments. Note also that any kind of data can be represented with numbers (e.g. binary code).
Example:
sim = Bullet()
robot = Robot(sim)
# Two ways to initialize states
states = State([JntPositionState(robot), JntVelocityState(robot)])
# or
states = JntPositionState(robot) + JntVelocityState(robot)
actions = JntPositionAction(robot)
policy = NNPolicy(states, actions)
References:
[1] "Wikipedia: Composition over Inheritance", https://en.wikipedia.org/wiki/Composition_over_inheritance
[2] "OpenAI gym": https://gym.openai.com/ and https://github.com/openai/gym
"""
def __init__(self, states=(), data=None, space=None, name=None):
"""
Initialize the state. The state contains some kind of data, or is a state combined of other states.
Args:
states (list/tuple of State): list of states to be combined together (if given, we can not specified data)
data (np.ndarray): data associated to this state
space (gym.space): space associated with the given data
Warning:
Both arguments can not be provided to the state.
"""
# Check arguments
if states is None:
states = tuple()
if not isinstance(states, (list, tuple, set, OrderedSet)):
# # TODO: should check that states is a list of state, however O(N)
# if data is None: # this is in the case someone calls `State(data)`
# data = states
# else:
raise TypeError("Expecting a list, tuple, or (ordered) set of states.")
if len(states) > 0 and data is not None:
raise ValueError("Please specify only one of the argument `states` xor `data`, but not both.")
# # Check if data is given
# if data is not None:
# if not isinstance(data, np.ndarray):
# if isinstance(data, (list, tuple)):
# data = np.array(data)
# elif isinstance(data, (int, float)):
# data = np.array([data])
# else:
# raise TypeError("Expecting a numpy array, a list/tuple of int/float, or an int/float for 'data'")
#
# if isinstance(states, collections.Iterable):
# if len(states) > 0 and data is not None: # check if both the states and data are specified
# raise ValueError("Please specify only one of the argument `states` xor `data`, but not both.")
#
# #
# for state in states:
# if not isinstance(state, State):
# if data is None: # in the case, someone calls `State(data)`
# data = states
# break
# else:
# raise ValueError("Please specify only one of the argument `states` xor `data`, but not both.")
# Check if data is given
if data is not None:
if not isinstance(data, np.ndarray):
if isinstance(data, (list, tuple)):
data = np.array(data)
elif isinstance(data, (int, float)):
data = np.array([data])
else:
raise TypeError("Expecting a numpy array, a list/tuple of int/float, or an int/float for 'data'")
# The following attributes should normally be set in the child classes
self._data = data
self._space = space
self._distribution = None # for sampling
self._normalizer = None
self._noiser = None # for noise
self._name = name
# create ordered set which is useful if this state is a combination of multiple states
self._states = OrderedSet()
if self._data is None:
self.add(states)
# reset state
self.reset()
##############################
# Properties (Getter/Setter) #
##############################
@property
def states(self):
"""
Get the list of states.
"""
return self._states
@states.setter
def states(self, states):
"""
Set the list of states.
"""
if self.hasData():
raise AttributeError("Trying to add internal states to the current state while it already has some data. "
"A state should be a combination of states or should contain some kind of data, "
"but not both.")
if isinstance(states, collections.Iterable):
for state in states:
if not isinstance(state, State):
raise TypeError("One of the given states is not an instance of State.")
self.add(state)
else:
raise TypeError("Expecting an iterator (e.g. list, tuple, OrderedSet, set,...) over states")
@property
def data(self):
"""
Get the data associated to this particular state, or the combined data associated to each state.
Returns:
list of np.ndarray: list of data associated to the state
"""
if self.hasData():
return [self._data]
return [state._data for state in self._states]
@data.setter
def data(self, data):
"""
Set the data associated to this particular state, or the combined data associated to each state.
Each data will be clipped if outside the range/bounds of the corresponding state.
Args:
data: the data to set
"""
# one state: change the data
if self.hasData():
if not isinstance(data, np.ndarray):
if isinstance(data, (list, tuple)):
data = np.array(data)
elif isinstance(data, (int, float)):
data = data * np.ones(self._data.shape)
else:
raise TypeError("Expecting a numpy array, a list/tuple of int/float, or an int/float for 'data'")
if self._data.shape != data.shape:
raise ValueError("The given data does not have the same shape as previously.")
# clip the value using the space
if self.hasSpace():
if self.isContinuous(): # continuous case
low, high = self._space.low, self._space.high
data = np.clip(data, low, high)
else: # discrete case
n = self._space.n
if data.size == 1:
data = np.clip(data, 0, n)
self._data = data
else: # combined state
if not isinstance(data, collections.Iterable):
raise TypeError("data is not an iterator")
if len(self._states) != len(data):
raise ValueError("The number of states is different from the number of data segments")
for state, d in zip(self._states, data):
state.data = d
@property
def merged_data(self):
"""
Return the merged data.
"""
# fuse the data
fused_state = self.fuse()
# return the data
return fused_state.data
@property
def space(self):
"""
Get the corresponding space.
"""
if self.hasSpace():
return [self._space]
return [state._space for state in self._states]
@space.setter
def space(self, space):
"""
Set the corresponding space. This can only be used one time!
"""
if self.hasData() and not self.hasSpace() and \
isinstance(space, (gym.spaces.Box, gym.spaces.Discrete)):
self._space = space
@property
def name(self):
"""
Return the name of the state.
"""
if self._name is None:
return self.__class__.__name__
return self._name
@name.setter
def name(self, name):
"""
Set the name of the state.
"""
if not isinstance(name, str):
raise TypeError("Expecting the name to be a string.")
self._name = name
@property
def shape(self):
"""
Return the shape of each state. Some states, such as camera states have more than 1 dimension.
"""
return [d.shape for d in self.data]
@property
def size(self):
"""
Return the size of each state.
"""
return [d.size for d in self.data]
@property
def dimension(self):
"""
Return the dimension (length of shape) of each state.
"""
return [len(d.shape) for d in self.data]
@property
def distribution(self):
"""
Get the current distribution used when sampling the state
"""
pass
@distribution.setter
def distribution(self, distribution):
"""
Set the distribution to the state.
"""
# check if distribution is discrete/continuous
pass
###########
# Methods #
###########
def isCombinedState(self):
"""
Return a boolean value depending if the state is a combination of states.
Returns:
bool: True if the state is a combination of states, False otherwise.
"""
return len(self._states) > 0
# alias
hasStates = isCombinedState
def hasData(self):
return self._data is not None
def hasSpace(self):
return self._space is not None
def add(self, state):
"""
Add a state or a list of states to the list of internal states. Useful when combining different states together.
This shouldn't be called if this state has some data set to it.
Args:
state (State, list/tuple of State): state(s) to add to the internal list of states
"""
if self.hasData():
raise AttributeError("Undefined behavior: a state should be a combination of states or should contain "
"some kind of data, but not both.")
if isinstance(state, State):
self._states.add(state)
elif isinstance(state, collections.Iterable):
for i, s in enumerate(state):
if not isinstance(s, State):
raise TypeError("The item {} in the given list is not an instance of State".format(i))
self._states.add(s)
else:
raise TypeError("The 'other' argument should be an instance of State, or an iterator over states.")
# alias
append = add
extend = add
def _read(self):
pass
def read(self):
"""
Read the state values from the simulator for each state, set it and return their values.
This has to be overwritten by the child class.
"""
if self.hasData(): # read the current state
self._read()
else: # read each state
for state in self.states:
state._read()
# return the data
return self.data
def _reset(self):
pass
def reset(self):
"""
Some states need to be reset. It returns the initial state.
This needs to be overwritten by the child class.
Returns:
initial state
"""
if self.hasData(): # reset the current state
self._reset()
else: # reset each state
for state in self.states:
state._reset()
# return the first state data
return self.read()
def maxDimension(self):
"""
Return the maximum dimension.
"""
return max(self.dimension)
def totalSize(self):
"""
Return the total size of the combined state.
"""
return sum(self.size)
def hasDiscreteValues(self):
"""
Does the state have discrete values?
"""
if self._data is None:
return [isinstance(state._space, gym.spaces.Discrete) for state in self._states]
if isinstance(self._space, gym.spaces.Discrete):
return [True]
return [False]
def isDiscrete(self):
"""
If all the states are discrete, then it is discrete.
"""
return all(self.hasDiscreteValues())
def hasContinuousValues(self):
"""
Does the state have continuous values?
"""
if self._data is None:
return [isinstance(state._space, gym.spaces.Box) for state in self._states]
if isinstance(self._space, gym.spaces.Box):
return [True]
return [False]
def isContinuous(self):
"""
If one of the state is continuous, then the state is considered to be continuous.
"""
return any(self.hasContinuousValues())
def bounds(self):
"""
If the state is continuous, it returns the lower and higher bounds of the state.
If the state is discrete, it returns the maximum number of discrete values that the state can take.
Returns:
list/tuple: list of bounds if multiple states, or bounds of this state
"""
if self._data is None:
return [state.bounds() for state in self._states]
if isinstance(self._space, gym.spaces.Box):
return (self._space.low, self._space.high)
elif isinstance(self._space, gym.spaces.Discrete):
return (self._space.n,)
raise NotImplementedError
def apply(self, fct):
"""
Apply the given fct to the data of the state, and set it to the state.
"""
self.data = fct(self.data)
def contains(self, x): # parameter dependent of the state
"""
Check if the argument is within the range/bound of the state.
"""
return self._space.contains(x)
def sample(self, distribution=None): # parameter dependent of the state (discrete and continuous distributions)
"""
Sample some values from the state based on the given distribution.
If no distribution is specified, it samples from a uniform distribution (default value).
"""
if self.isCombinedState():
return [state.sample() for state in self._states]
if self._distribution is None:
return
else:
pass
raise NotImplementedError
def addNoise(self, noise=None, replace=True): # parameter dependent of the state
"""
Add some noise to the state, and returns it.
Args:
noise (np.ndarray, fct): array to be added or function to be applied on the data
"""
if self._data is None:
# apply noise
for state in self._states:
state.addNoise(noise=noise)
else:
# add noise to the data
noisy_data = self.data + noise
# clip such that the data is within the bounds
self.data = noisy_data
def normalize(self, normalizer=None, replace=True): # parameter dependent of the state
"""
Normalize using the state data using the provided normalizer.
Args:
normalizer (sklearn.preprocessing.Normalizer): the normalizer to apply to the data.
replace (bool): if True, it will replace the `data` attribute by the normalized data.
Returns:
the normalized data
"""
pass
def fuse(self, other=None, axis=0):
"""
Fuse the states that have the same shape together. The axis specified along which axis we concatenate the data.
If multiple states with different shapes are present, the axis will be the one specified if possible, otherwise
it will be min(dimension, axis).
Examples:
s0 = JntPositionState(robot)
s1 = JntVelocityState(robot)
s = s0 & s1
print(s)
print(s.shape)
s = s0 + s1
s.fuse()
print(s)
print(s.shape)
"""
# check argument
if not (other is None or isinstance(other, State)):
raise TypeError("The 'other' argument should be None or another state.")
# build list of all the states
states = [self] if self.hasData() else self._states
if other is not None:
if other.hasData():
states.append(other)
else:
states.extend(other._states)
# check if only one state
if len(states) < 2:
return self # do nothing
# build the dictionary with key=dimension of shape, value=state
dic = {}
for state in states:
dic.setdefault(len(state._data.shape), []).append(state)
# traverse the dictionary and fuse corresponding shapes
states = []
for key, value in dic.items():
if len(value) > 1:
# fuse
data = [state._data for state in value]
names = [state.name for state in value]
s = State(data=np.concatenate(data, axis=min(axis, key)), name='+'.join(names))
states.append(s)
else:
# only one state
states.append(value[0])
# return the fused state
if len(states) == 1:
return states[0]
return State(states)
def lookfor(self, class_type):
"""
Look for the specified class type/name in the list of internal states, and returns it.
"""
if self.hasData():
return None
for state in self.states:
if state.__class__ == class_type:
return state
########################
# Operator Overloading #
########################
def __repr__(self):
if self._data is None:
lst = [self.__class__.__name__ + '(']
for state in self.states:
lst.append('\t' + state.__repr__() + ',')
lst.append(')')
return '\n'.join(lst)
else:
return '%s(%s)' % (self.name, self._data)
# def __str__(self):
# """
# String to represent the state. Need to be provided by each child class.
# """
# if self._data is None:
# return [str(state) for state in self._states]
# return str(self)
def __call__(self):
"""
Compute/read the state and return it. It is an alias to the `self.read()` method.
"""
return self.read()
def __len__(self):
"""
Return the total number of states contained in this class.
Example::
s1 = JntPositionState(robot)
s2 = s1 + JntVelocityState(robot)
print(len(s1)) # returns 1
print(len(s2)) # returns 2
"""
if self._data is None:
return len(self._states)
return 1
def __iter__(self):
"""
Iterator over the states.
"""
if self.isCombinedState():
for state in self._states:
yield state
else:
yield self
def __contains__(self, item):
"""
Check if the state item(s) is(are) in the combined state. If the item is the data associated with the state,
it checks that it is within the bounds.
Args:
item (State, list/tuple of state): check if given state(s) is(are) in the combined state
Example:
s1 = JntPositionState(robot)
s2 = JntVelocityState(robot)
s = s1 + s2
print(s1 in s) # output True
print(s2 in s1) # output False
print((s1, s2) in s) # output True
"""
# check type of item
if not isinstance(item, (State, np.ndarray)):
raise TypeError("Expecting a state or numpy array.")
# check if state item is in the combined state
if self._data is None and isinstance(item, State):
return (item in self._states)
# check if state/data is within the bounds
if isinstance(item, State):
item = item.data
# check if continuous
# if self.isContinuous():
# low, high = self.bounds()
# return np.all(low <= item) and np.all(item <= high)
# else: # discrete case
# num = self.bounds()[0]
# # check the size of data
# if item.size > 1: # array
# return (item.size < num)
# else: # one number
# return (item[0] < num)
return self.contains(item)
def __getitem__(self, key):
"""
Get the corresponding item from the state(s)
"""
# if one state, slice the corresponding state data
if len(self._states) == 0:
return self._data[key]
# if multiple states
if isinstance(key, int):
# get one state
return self._states[key]
elif isinstance(key, slice):
# get multiple states
return State(self._states[key])
else:
raise TypeError("Expecting an int or slice for the key, but got instead {}".format(type(key)))
def __setitem__(self, key, value):
"""
Set the corresponding item/value to the corresponding key.
Args:
key (int, slice): index of the internal state, or index/indices for the state data
value (State, int/float, array): value to be set
"""
if self.isCombinedState():
# set/move the state to the specified key
if isinstance(value, State) and isinstance(key, int):
self._states[key] = value
else:
raise TypeError("Expecting key to be an int, and value to be a state.")
else:
# set the value on the data directly
self._data[key] = value
def __add__(self, other):
"""
Combine two different states together. In this special case, the operation is not commutable.
This is the same as taking the union of the states.
Args:
other (State): another state
Returns:
State: the combined state
Examples:
s1 = JntPositionState(robot)
s2 = JntVelocityState(robot)
s = s1 + s2 # = State([JntPositionState(robot), JntVelocityState(robot)])
s1 = State([JntPositionState(robot), JntVelocityState(robot)])
s2 = State([JntPositionState(robot), LinkPositionState(robot)])
s = s1 + s2 # = State([JntPositionState(robot), JntVelocityState(robot), LinkPositionState(robot)])
"""
if not isinstance(other, State):
raise TypeError("Expecting another state, instead got {}".format(type(other)))
s1 = self._states if self._data is None else OrderedSet([self])
s2 = other._states if other._data is None else OrderedSet([other])
s = s1 + s2
return State(s)
def __iadd__(self, other):
"""
Add a state to the current one.
Args:
other (State, list/tuple of State): other state
Examples:
s = State()
s += JntPositionState(robot)
s += JntVelocityState(robot)
"""
if self._data is not None:
raise AttributeError("The current class already has some data attached to it. This operation can not be "
"applied in this case.")
self.append(other)
def __sub__(self, other):
"""
Remove the other state(s) from the current state.
:param other:
:return:
"""
if not isinstance(other, State):
raise TypeError("Expecting another state, instead got {}".format(type(other)))
s1 = self._states if self._data is None else OrderedSet([self])
s2 = other._states if other._data is None else OrderedSet([other])
s = s1 - s2
if len(s) == 1: # just one element
return s[0]
return State(s)
def __isub__(self, other):
"""
Remove one or several states from the combined state.
Args:
other:
"""
if not isinstance(other, State):
raise TypeError("Expecting another state, instead got {}".format(type(other)))
if self._data is not None:
raise RuntimeError("This operation is only available for a combined state")
s = other._states if other._data is None else OrderedSet([other])
self._states -= s
def __and__(self, other):
"""
Fuse two states together; only one data for the two states, instead of a data for each state as done
when combining the states. All the states must have the same dimensions, and it fuses the data along
the axis=0.
Args:
other: the other (combined) state
Returns:
State: the intersection of states
Examples:
s0 = JntPositionState(robot)
s1 = JntVelocityState(robot)
print(s0.shape)
print(s1.shape)
s = s0 + s1
print(s.shape) # prints [s0.shape, s1.shape]
s = s0 & s1
print(s.shape) # prints np.concatenate((s0,s1)).shape
"""
return self.fuse(other, axis=0)
# def __invert__(self):
# """
# Return the
# :return:
# """
# pass
+180
View File
@@ -0,0 +1,180 @@
from abc import ABCMeta, abstractmethod
class InitialStateGenerator(object):
r"""Initial State Generator
Initialize the state which will be given as the first state by the environment when calling ``env.reset()``.
If for instance, the state consists of joint positions and velocities, we can generate them from a distribution
that is given or learned from data.
The `state generator` is tightly coupled with a `state` object.
Sometimes a mapping between different states is necessary. For instance, the state generator might generate
human joint states that need first to be mapped to robot joint states in order to initialize the robot.
In this example, a kinematic mapping which is modeled mathematically or learned need to be provided additionally.
This is particularly significant as robot data are lacking, while human data is pretty abundant.
The mapping function has to return a `state` object.
"""
__metaclass__ = ABCMeta
def __init__(self):
pass
@abstractmethod
def generate(self):
pass
class FixedInitialStateGenerator(InitialStateGenerator):
r"""Fixed Initial State Generator
This generator returns the same initial state each time it is called.
"""
def __init__(self):
super(FixedInitialStateGenerator, self).__init__()
class FIFOQueueInitialStateGenerator(InitialStateGenerator):
r"""FIFO Queue Initial State Generator
Generate the initial state from a FIFO queue. If the queue is empty returns the default initial state.
The queue is filled by the user during training.
"""
def __init__(self):
super(FIFOQueueInitialStateGenerator, self).__init__()
class PriorityQueueInitialStateGenerator(InitialStateGenerator):
r"""Priority Queue Initial State Generator
Generate the initial state from a priority queue filled by the user. If empty, it returns the default initial
state. The queue can be filled for instance with states that have high/low uncertainty, or high/low rewards.
The queue has a limited capacity, and can be used to include states from which the agent/policy performed
poorly during the training.
"""
def __init__(self):
pass
class DistributedInitialStateGenerator():
r"""Distributed Initial State Generator
The initial states :math:`s` are generated by a probability distribution :math:`p(s)`, that is :math:`s \sim p(s)`.
The probability distribution can be learned using generative models.
"""
def __init__(self):
pass
class UniformInitialStateGenerator():
r"""Uniform Initial State Generator
The initial states are generated by a uniform distribution. If no upper/lower limits are specified, the limits
will be set to be the range of the states.
"""
def __init__(self):
pass
class NormalInitialStateGenerator():
r"""Normal Initial State Generator
The initial states are generated by a normal distribution, where the mean and standard deviation are specified.
The states are then truncated / clipped to be inside their corresponding range.
"""
def __init__(self):
pass
class LearnableInitialStateGenerator():
r"""Learnable Initial State Generator
This uses Generative Models.
"""
def __init__(self, model):
self.model = model
class AEBOInitialStateGenerator():
r"""AutoEncoder (AE) - Bayesian Optimization (BO) Initial State Generator
Using a pretrained AE on plausible states, and keeping the decoder allows us to explore in the lower dimensional
state space using BO (GP). The BO will provide the reduced state vector based on the uncertainty / objective
fct value. Then, the outputted vector can be fed to the decoder which will return the corresponding high-
dimensional state.
In addition, we fix a certain capacity to the kernel matrix of the GP underlying the BO. If when inserting a
new (low-dimensional) state, the capacity is exceeded, the oldest state is removed from the kernel to allow
the incoming state.
If the states have a certain range, we use the encoder part to get the corresponding low-dimensional state limits.
The exploration will then be carried out in the hyperrectangle formed by these 2 reduced state vector limits.
"""
def __init__(self, autoencoder, kernel_capacity=100):
pass
class VAEInitialStateGenerator():
r"""Variational Autoencoder (VAE) Initial State Generator
This uses the decoder a pretrained VAE to generate initial states.
"""
def __init__(self, states, model):
pass
class GANInitialStateGenerator():
r"""Generative Adversarial Network (GAN) Initial State Generator
This uses the generator of a trained GAN model to generate similar states.
"""
def __init__(self, states, model, distribution=None, mapping=None):
"""
:param states: states that need to be generated
:param model: GAN or generator of GAN
:param distribution: distribution over the noise vector
:param mapping:
"""
# checking and setting the model
if isinstance(model, GAN):
self.generator = model.getGenerator()
elif isinstance(model, Generator):
self.generator = model
else:
raise TypeError("The `model` parameter should be an instance of GAN or Generator.")
# checking and setting the distribution
if distribution is None:
# create normal distribution with dimension of the generator input
pass
else:
if not isinstance(distribution, Distribution):
raise TypeError("The given `distribution` is not an instance of Distribution.")
self.distribution = distribution
# setting mapping
self.mapping = mapping
def generate(self):
noise_vector = self.distribution.sample()
states = self.generator(noise_vector)
if self.mapping is not None:
return self.mapping(states)
return states
class GMMInitialStateGenerator():
r"""Gaussian Mixture Model Initial State Generator
This uses a pretrained GMM to generate the states.
"""
def __init__(self):
pass
+135
View File
@@ -0,0 +1,135 @@
#!/usr/bin/env python
"""Define the various time states
This includes notably the absolute, relative, and cumulative time states.
"""
from abc import ABCMeta
import time
import numpy as np
from state import State
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "(c) Brian Delhaisse"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class TimeState(State):
r"""Time state (abstract class)"""
__metaclass__ = ABCMeta
pass
class AbsoluteTimeState(TimeState):
"""Absolute time state
Returns the absolute time.
"""
def __init__(self):
data = np.array([time.time()])
super(AbsoluteTimeState, self).__init__(data=data)
def _read(self):
self._data[0] = time.time()
class RelativeTimeState(TimeState):
"""Relative time state
Returns the time difference from last time.
"""
def __init__(self):
data = np.array([0.0])
super(RelativeTimeState, self).__init__(data=data)
def _reset(self):
self.current_time = time.time()
self._data[0] = 0.0
def _read(self):
next_time = time.time()
self._data[0] = next_time - self.current_time
self.current_time = next_time
class CumulativeTimeState(TimeState):
r"""Cumulative time state
Return the cumulative time.
"""
def __init__(self):
data = np.array([0.0])
super(CumulativeTimeState, self).__init__(data=data)
def _reset(self):
self._data[0] = 0.0
self.current_time = time.time()
def _read(self):
next_time = time.time()
self._data[0] += (next_time - self.current_time)
self.current_time = next_time
class PhaseState(TimeState):
r"""Phase State
"""
def __init__(self, num_steps=100, max_value=1., rate=1):
data = np.array([0.0])
self.cnt = 0
self.rate = rate
self.max_value = max_value
if num_steps < 2:
num_steps = 2
self.dphase = float(max_value) / (num_steps - 1)
super(PhaseState, self).__init__(data=data)
def _reset(self):
self._data[0] = 0.0
self.cnt = 0
def _read(self):
if (self.cnt % self.rate) == 0:
if self._data[0] < self.max_value:
self._data[0] += self.dphase
self.cnt += 1
# Tests the different time states
if __name__ == '__main__':
s = AbsoluteTimeState()
print("\nAbsolute Time State:")
print(s.reset())
for i in range(10):
print(s())
s = RelativeTimeState()
print("\nRelative Time State:")
print(s.reset())
for i in range(10):
print(s())
s = CumulativeTimeState()
print("\nCumulative Time State:")
print(s.reset())
for i in range(10):
print(s())
combined = AbsoluteTimeState() + RelativeTimeState() + CumulativeTimeState()
fused = AbsoluteTimeState() & RelativeTimeState() & CumulativeTimeState()
print("\nCombined state: {}".format(combined))
print("\nFused state: {}".format(fused))
for i in range(4):
print(combined.read())
print(fused.read())