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

add envs, tasks, experiments

Browse Source
This commit is contained in:
Brian Delhaisse
2019-03-16 02:45:23 +01:00
parent a1627975f9
commit 062e8395d4
32 changed files with 2428 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyrobolearn/control/__init__.py
+8
View File
@@ -0,0 +1,8 @@
# import some control algorithms
# PID
from pid import PID
# LQR
from lqr import LQR
pyrobolearn/control/ddp.py
+17
View File
@@ -0,0 +1,17 @@
# DDP: Differential Dynamic Programming
class DDP(object):
r"""Differential Dynamic Programming
Type: Model-based (optimal control)
References:
[1]
"""
def __init__(self):
pass
def compute(self):
pass
pyrobolearn/control/dp.py
+27
View File
@@ -0,0 +1,27 @@
# This file describes the Dynamic Programming algorithm
class DP(object):
r"""Dynamic Programming (DP)
Type: model-based
"Dynamic programming usually refers to simplifying a decision by breaking it down into a sequence of decision
steps over time" (Wikipedia)
Bellman's equations:
.. math:: V(s_{t+1}) =
References:
[1] "Dynamic Programming", Bellman, 1957
[2] "Reinforcement Learning: An Introduction", Sutton and Barto, 1998 (chap4)
[3] "Optimal Control Theory: An Introduction", Kirk, 1970
[4] "Dynamic Programming and Optimal Control", Bertsekas, 1987
"""
def __init__(self):
pass
def compute(self):
pass
pyrobolearn/control/ilqg.py
+25
View File
@@ -0,0 +1,25 @@
# This file describes the iterative Linear Quadratic Gaussian (iLQG)
class ILQG(object):
r"""Iterative Linear Quadratic Gaussian (iLQG)
Type: Model-based (optimal control)
Notes: It assumes that the dynamics are described by a linear system of differential equations
References:
[1]
See also:
- `lqr.py`: LQR
- `lqg.py`: LQG = LQR + LQE
- `ilqr.py`: iterative LQR
"""
def __init__(self):
pass
def compute(self):
pass
pyrobolearn/control/ilqr.py
+24
View File
@@ -0,0 +1,24 @@
# This file describes the iterative Linear Quadratic Regulator (iLQR)
class ILQR(object):
r"""Iterative Linear Quadratic Regulator
Type: Model-based (optimal control)
Notes: It assumes that the dynamics are described by a linear system of differential equations
References:
[1]
See also:
- `lqr.py`: LQR
- `lqg.py`: LQG = LQR + LQE
- `ilqg.py`: iterative LQG
"""
def __init__(self):
pass
def compute(self):
pass
pyrobolearn/control/lqg.py
+31
View File
@@ -0,0 +1,31 @@
# This file describes the Linear Quadratic Gaussian
class LQG(object):
r"""Linear Quadratic Gaussian
Type: Model-based (optimal control)
Notes: It assumes that the dynamics are described by a linear system of differential equations
"LQG concerns uncertain linear systems disturbed by additive white Gaussian noise, having incomplete
state information (i.e. not all the state variables are measured and available for feedback) and
undergoing control subject to quadratic costs. Moreover, the solution is unique and constitutes a linear
dynamic feedback control law that is easily computed and implemented."
LQG = LQE + LQR, where LQE is a Linear Quadratic Estimator (i.e. Kalman Filter), and LQR is a Linear Quadratic
Regressor.
References:
[1]
See also:
- `lqr.py`: LQR
- `ilqr.py`: iterative LQR
- `ilqg.py`: iterative LQG
"""
def __init__(self):
pass
def compute(self):
pass
pyrobolearn/control/lqr.py
+75
View File
@@ -0,0 +1,75 @@
# This file describes the linear quadratic regulator
import control
from scipy.linalg import solve_continuous_are
import numpy as np
class LQR(object):
r"""Linear Quadratic Regulator
Type: Model-based (optimal control)
LQR assumes that the dynamics are described by a set of linear differential equations, and a quadratic cost.
That is, the dynamics can written as :math:`\dot{x} = A x + B u`, where :math:`x` is the state vector, and
:math:`u` is the control vector, and the cost is given by:
.. math:: J = x(T)^T F(T) x(T) + \int_0^T (x(t)^T Q x(t) + u(t)^T R u(t) + 2 x(t)^T N u(t)) dt
where :math:`Q` and :math:`R` represents weight matrices which allows to specify the relative importance
of each state/control variable. These are normally set by the user.
The goal is to find the feedback control law :math:`u` that minimizes the above cost :math:`J`. Solving it
gives us :math:`u = -K x`, where :math:`K = R^{-1} (B^T S + N^T)` with :math:`S` is found by solving the
continuous time Riccati differential equation :math:`S A + A^T S - (S B + N) R^{-1} (B^T S + N^T) + Q = 0`.
Thus, LQR requires thus the model/dynamics of the system to be given (i.e. :math:`A` and :math:`B`).
If the dynamical system is described by a set of nonlinear differential equations, we first have to linearize
them around fixed points.
Time complexity: O(M^3) where M is the size of the state vector
Note: A desired state xd can also be given to the system: u = -K (x - xd) (P control)
See also:
- `ilqr.py`: iterative LQR
- `lqg.py`: LQG = LQR + LQE
- `ilqg.py`: iterative LQG
"""
def __init__(self, A, B, Q=None, R=None, N=None):
if not self.isControllable(A,B):
raise ValueError("The system is not controllable")
self.A = A
self.B = B
if Q is None: Q = np.identity(A.shape[1])
self.Q = Q
if R is None: R = np.identity(B.shape[1])
self.R = R
self.N = N
self.K = None
@staticmethod
def isControllable(A, B):
return np.linalg.matrix_rank(control.ctrb(A,B)) == A.shape[0]
def getRiccatiSolution(self):
S = solve_continuous_are(self.A, self.B, self.Q, self.R, s=self.N)
return S
def getGainK(self):
#S = self.getRiccatiSolution()
#S1 = self.B.T.dot(S)
#if self.N is not None: S1 += self.N.T
#K = np.linalg.inv(self.R).dot(S1)
K, S, E = control.lqr(self.A, self.B, self.Q, self.R, self.N)
return K
def compute(self, x, xd=None):
"""Return the u."""
if self.K is None:
self.K = self.getGainK()
if xd is None:
return self.K.dot(x)
else:
return self.K.dot(xd - x)
pyrobolearn/control/mpc.py
+28
View File
@@ -0,0 +1,28 @@
# MPC: Model Predictive Control
class MPC(object):
r"""Model Predictive Control
Type: Optimal Control
MPC optimizes for a finite time-horizon :math:`T` (i.e. compute the optimal :math:`\{u_t, u_{t+1},..., u_T\}`
given the dynamical system :math:`x_{t+1} = f(x_t, u_t)` and cost :math:`c(x_t,u_t)`), executes the first best
found control law :math:`u_t`, lets the system goes to the next state :math:`x_{t+1}`, and then re-optimize again
for each next time step. The finite time-horizon allows to take into account close future events, while
re-optimizing at each time step allows to deal with the discrepancy between the modeled and real dynamical systems.
Notes:
* MPC vs LQR: LQR assumes a linear dynamical system and optimizes for the whole time horizon providing us
the single optimal solution, while MPC optimizes in a receding time window at each time step
resulting in a suboptimal solution but more robust to various perturbations, uncertainties, and so on
not accounted/modeled by our dynamical system.
References:
[1]
"""
def __init__(self):
pass
def compute(self, x):
pass
pyrobolearn/control/optimal_control.py
+32
View File
@@ -0,0 +1,32 @@
class OptimalControlAlgo(object):
"""Optimal Control Algorithm
Any optimal control schemes inherit from this class. Optimal control is also known as model-based reinforcement
learning in the computer science and machine learning communities. They however use a different vocabulary and
have different notations:
* they minimize 'costs' :math:`c_t` instead of maximizing 'rewards' :math:`r_t`
* 'controller' instead of 'policy' or 'agent'
* 'controlled system'/'plant' instead of 'environment'
* 'control signal' :math:`u_t` instead of 'actions' :math:`a_t`
* 'states' are denoted by :math:`x_t` instead of `s_t`
Optimal control assumes that the dynamic model :math:`p(x_{t+1}|x_{t},u_{t})` is given, and use this knowledge
to optimize the controller.
.. seealso::
* `lqr.py`: Linear Quadratic Regulator
* `ilqr.py`: iterative Linear Quadratic Regulator
* `lqg.py`: Linear Quadratic Gaussian
* `ilqg.py`: iterative Linear Quadratic Gaussian
* `dp.py`: Dynamic Programming
* `ddp.py`: Differential Dynamic Programming
* `mpc.py`: Model Predictive Control
References:
[1] python-control: https://github.com/python-control/python-control
"""
def __init__(self, physical_model, states, actions):
pass
pyrobolearn/control/pid.py
+50
View File
@@ -0,0 +1,50 @@
# This defines the PID controller
class PID(object):
r"""Proportional-Integral-Derivative Controller
The PID control scheme is given by:
.. math: u(t) = K_p e(t) + K_i \int_0^t e(\tau) d\tau + K_d \frac{d e(t)}{dt}
where :math:`u(t)` is the controller output, :math:`K_p, K_i, K_d` are respectively the proportional, integral,
and derivative tuning gains (set by the user or an algorithm), :math:`e(t) = (x_{des} - x(t))` is the error
between the desired point :math:`x_{des}` and the current point :math:`x(t)`.
"""
def __init__(self, kp=0, kd=0, ki=0, dt=0.001):
"""
Initialize the PID controller
Args:
kp (float): proportional gain
kd (float): derivative gain
ki (float): integral gain
dt (float): time step
"""
self.kp = kp
self.kd = kd
self.ki = ki
self.dt = dt
self.errorI = 0
self.prev_error = 0
def compute(self, xd, x, dt=None):
"""
Compute the controller output using PID control scheme.
Args:
xd (float, array): desired point
x (float, array): current point
dt (float): time step
Returns:
float, array: control output
"""
error = xd - x
self.errorI += error
errorD = (error - self.prev_error) / dt
self.prev_error = error
u = self.kp * error + self.ki * self.errorI + self.kd * errorD
return u
pyrobolearn/controllers/__init__.py
View File
pyrobolearn/controllers/controller.py
View File
pyrobolearn/dynamics/__init__.py
+3
View File
@@ -0,0 +1,3 @@
# import the transition dynamic models
from dynamic import *
pyrobolearn/dynamics/dynamic.py
+182
View File
@@ -0,0 +1,182 @@
#!/usr/bin/env python
"""Provides the `transition`/`dynamic` function approximators in RL.
Dynamic models allows to compute the next state given the current state and action; that is, p(s_{t+1} | s_t, a_t).
Dependencies:
- `pyrobolearn.states`
- `pyrobolearn.actions`
- `pyrobolearn.approximators` (and thus `pyrobolearn.models`)
"""
from abc import ABCMeta, abstractmethod
from pyrobolearn.states import State
from pyrobolearn.actions import Action
# from pyrobolearn.approximators import Approximator
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class DynamicModel(object):
r"""Dynamic/Transition Model
In the reinforcement learning setting, the dynamic model is the transition function associated to the environment
which describes how ... The agent/policy has no control over it. However, it can often be learned from data samples
acquired by interacting with the environment. This allows to model the environment and then perform internal
simulations which are then more sample efficient in the real environment.
When a dynamic model is involved, this is known as "Model-based Reinforcement Learning". These methods usually
requires less samples as they learn the model of the environment.
.. math::
P_{\varphi}(s_{t+1} | s_t, a_t)
They are 2 main ways to build a dynamic model:
1. build it from a mathematical model
Pros: mathematical guarantees (such as stability,...), predictable,...
Cons: linearization, unmodeled phenomenon, assumptions that might be violated (rigid body), complex...
2. learn it from the data, by letting the policy interacts with the environment
Pros:
Cons: usually requires a lot of samples to be accurate, mismatch between the real and the learned dynamic
model, often no guarantees and could be unpredictable
Note that learning a wrong dynamic model can have drastic consequences on the learned policy. Indeed, learning
a dynamic model in the simulator can be completely to a learned .
Some papers have worked on simulators that generates ...
"""
__metaclass__ = ABCMeta
def __init__(self, states, actions, model=None):
self.states = self._check_states(states)
self.actions = self._check_actions(actions)
@staticmethod
def _check_states(states):
"""
Check if the states are valid (i.e. it is an instance of `State`, a list/tuple of `State` instances, or None).
:param states: states to be checked.
:return: states
"""
if isinstance(states, State):
states = [states]
elif isinstance(states, (list, tuple)):
for state in states:
if not isinstance(state, State):
raise ValueError("Each state in the list/tuple must be a `State` object.")
elif states is None: # some policies don't need the state information
states = []
else:
raise ValueError("The `states` parameter must be a `State` object or a list/tuple of `State` objects.")
return states
@staticmethod
def _check_actions(actions):
"""
Check if the actions are valid (i.e. it is an instance of `Action`, a list/tuple of `Action` instances).
:param actions: actions to be checked.
:return: actions
"""
if isinstance(actions, Action):
actions = [actions]
elif isinstance(actions, (list, tuple)):
for action in actions:
if not isinstance(action, Action):
raise ValueError("Each action in the list/tuple must be an instance of the `Action` class.")
else:
raise ValueError("The `actions` parameter must be an `Action` object or a list/tuple of `Action` objects.")
return actions
@abstractmethod
def __call__(self, states, actions):
"""
Return predicted state given the current state and action
:param state:
:param action:
:return:
"""
pass
def save(self, filename):
pass
def load(self, filename):
pass
class PhysicalDynamicModel(DynamicModel):
r"""Physical Dynamic Model
Dynamic model described by mathematical/physical equations.
"""
def __init__(self, states, actions):
super(PhysicalDynamicModel, self).__init__(states, actions)
class RobotDynamicModel(PhysicalDynamicModel):
r"""Robot Dynamical Model
This is the mathematical model of the robots.
Limitations:
* mathematical assumptions such as rigid bodies
* the states/actions have to be robot states/actions
"""
def __init__(self, states, actions):
super(RobotDynamicModel, self).__init__(states, actions)
class LinearDynamicModel(DynamicModel):
r"""Linear Dynamic Model
Pros: easy to implement
Cons: very limited
"""
def __init__(self, states, actions):
super(LinearDynamicModel, self).__init__(states, actions)
class PieceWiseLinearDynamicModel(DynamicModel):
r"""Piecewise linear dynamic model
Pros: easy to implement, often good predictions in local regions
Cons: poor scalability
"""
def __init__(self, states, actions):
super(PieceWiseLinearDynamicModel, self).__init__(states, actions)
class NNDynamicModel(DynamicModel):
r"""Neural Network Dynamic Model
Dynamic model using neural networks.
Pros:
Cons: requires lot of samples, overfitting,...
"""
def __init__(self, states, actions):
super(NNDynamicModel, self).__init__(states, actions)
class GPDynamicModel(DynamicModel):
r"""Gaussian Process Dynamic Model
Dynamic model using Gaussian Processes.
Pros: good from a mathematical point of view: integrate uncertainty on the dynamic model
Cons:
..seealso: PILCO
"""
def __init__(self, states, actions):
super(GPDynamicModel, self).__init__(states, actions)
pyrobolearn/envs/__init__.py
+9
View File
@@ -0,0 +1,9 @@
# import from envs
from env import Env, BasicEnv
# define wrapper for the gym environment
import gym_wrapper as gym
# import terminal conditions
from terminating_condition import *
pyrobolearn/envs/env.py
+270
View File
@@ -0,0 +1,270 @@
#!/usr/bin/env python
"""Define the `Env` class class which defines the world, states, and possible rewards. This is the main object
a policy interacts with.
Dependencies:
- `pyrobolearn.worlds`
- `pyrobolearn.states`
- `pyrobolearn.actions`
- (`pyrobolearn.rewards`)
"""
# import gym
from pyrobolearn.worlds import World, BasicWorld
from pyrobolearn.states import State
from pyrobolearn.actions import Action
from pyrobolearn.rewards import Reward
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class Env(object): # gym.Env):
r"""Environment class.
This class defines the environment as it described in a reinforcement learning setting [1]. That is, given an
action :math:`a_t` performed by an agent (i.e. policy), the environment computes and returns the next state
:math:`s_{t+1}` and reward :math:`r(s_t, a_t, s_{t+1})`. A policy can then interact with this environment,
and be trained.
The environment defines the world, the rewards, and the states that are returned by it.
To allow our framework to be generic and modular, the world, rewards, and states are decoupled from the environment,
and can be defined outside of this one and then provided as inputs to the `Env` class. That is, we favor
'composition over inheritance' (see [2]). This is a different approach compared to what is usually done using
the OpenAI gym framework (see [3]). That said, in order to be compatible with this framework, we inherit from
the `gym.Env` class (see `core.py` in `https://github.com/openai/gym/blob/master/gym/core.py`).
References:
[1] "Reinforcement Learning: An Introduction", Sutton and Barto, 1998
[2] "Wikipedia: Composition over Inheritance", https://en.wikipedia.org/wiki/Composition_over_inheritance
[3] "OpenAI gym": https://gym.openai.com/ and https://github.com/openai/gym
"""
def __init__(self, world, states, rewards=None, terminal_condition=None, initial_state_distribution=None,
extra_info=None):
"""
Initialize the environment.
Args:
world (World): world of the environment. The world contains all the objects (including robots), and has
access to the simulator.
states (State): states that are returned by the environment at each time step.
rewards (None, Reward): The rewards can be None when for instance we are in an imitation learning setting,
instead of a reinforcement learning one. If None, only the state is returned by
the environment.
terminal_condition (None, callable): A callable function or object that check if the policy has failed
or succeeded the task.
initial_state_distribution (None, callable): A callable function or object that is called at the beginning
when resetting the environment to generate the initial state
distribution.
extra_info (None, callable): Extra info returned by the environment at each time step.
"""
# Check and set parameters (see corresponding properties)
self.world = world
self.states = states
self.rewards = rewards
self.terminal_condition = terminal_condition
self.extra_info = extra_info if extra_info is not None else lambda: False
self.rendering = False # check with simulator
# save the world state in memory
self.world.save()
##############
# Properties #
##############
@property
def world(self):
"""Return an instance of the world."""
return self._world
@world.setter
def world(self, world):
"""Set the world."""
if not isinstance(world, World):
raise TypeError("Expecting the 'world' argument to be an instance of World.")
self._world = world
self.sim = self._world.simulator
@property
def simulator(self):
"""Return an instance of the simulator."""
return self._world.simulator
@property
def states(self):
"""Return the states."""
return self._states
@states.setter
def states(self, states):
"""Set the states."""
if not isinstance(states, State):
raise TypeError("Expecting the 'states' argument to be an instance of State.")
self._states = states
@property
def rewards(self):
"""Return the rewards."""
return self._rewards
@rewards.setter
def rewards(self, rewards):
"""Set the rewards."""
if rewards is not None:
if not isinstance(rewards, Reward):
raise TypeError("Expecting the 'rewards' argument to be an instance of Reward.")
else:
rewards = lambda: None
self._rewards = rewards
@property
def terminal_condition(self):
"""Return the terminal condition."""
return self._terminal_condition
@terminal_condition.setter
def terminal_condition(self, condition):
"""Set the terminal condition."""
if condition is None:
condition = lambda: False
if not callable(condition):
raise TypeError("Expecting the terminal condition to be callable.")
self._terminal_condition = condition
###########
# Methods #
###########
def reset(self):
"""
Reset the environment; reset the world and states.
Returns:
list/np.array: list of state data
"""
# reset world
self.world.reset()
# reset states and return first states/observations
return self.states.reset()
def step(self, actions=None):
"""
Run one timestep of the environment's dynamics. When end of
episode is reached, you are responsible for calling `reset()`
to reset this environment's state.
Accepts an action and returns a tuple (observation, reward, done, info).
Args:
action (Action, None): an action provided by the policy(ies) to the environment
Returns:
observation (object): agent's observation of the current environment
reward (float) : amount of reward returned after previous action
done (boolean): whether the episode has ended, in which case further step() calls will return undefined
results
info (dict): contains auxiliary diagnostic information (helpful for debugging, and sometimes learning)
"""
# if not isinstance(actions, (list, tuple)):
# actions = [actions]
if actions is not None and not isinstance(actions, Action):
raise TypeError("Expecting actions to be an instance of Action.")
# apply each policy's action in the environment
# for action in actions:
# action()
if actions is not None:
actions()
# perform a step forward in the simulation
self.world.step()
# compute reward
# rewards = [reward.compute() for reward in self.rewards]
rewards = self.rewards()
# compute terminating condition
# done = [reward.is_done() for reward in self.rewards]
done = self.terminal_condition()
# get next state/obs for each policy
# states = [state() for state in self.states]
# TODO: this should be before computing the rewards as some rewards need the next state
self.states()
# get extra information
info = self.extra_info()
return self.states, rewards, done, info
def render(self, mode='human'):
# This is dependent on the simulator. Some simulators allow to show the GUI at any point in time,
# while others like pybullet requires to specify it at the beginning (thus see SimuRealInterface).
# Bullet: do nothing
# if isinstance(self.sim, Bullet): pass
pass
def close(self):
pass
def seed(self, seed=None):
"""
Set the given seed for the simulator.
Args:
seed (int): seed for the random generator used in the simulator.
"""
self.sim.setSeed(seed)
class BasicEnv(Env):
"""Basic Environment class.
It creates a basic environment with a basic world (a floor and with gravity), no rewards, and no states.
"""
def __init__(self, states=None, rewards=None):
world = BasicWorld()
super(BasicEnv, self).__init__(world, states, rewards)
# Tests
if __name__ == '__main__':
from pyrobolearn.simulators import BulletSim
import time
# create simulator
sim = BulletSim()
# create world
world = BasicWorld(sim)
robot = world.loadRobot('coman', useFixedBase=True)
# create states
states = State()
# create rewards
reward = Reward()
# create Env
env = Env(world, states, reward)
# dummy action
action = Action()
# run the environment for n steps
for _ in range(10000):
env.step(action)
time.sleep(1./240)
pyrobolearn/envs/gym_wrapper.py
+145
View File
@@ -0,0 +1,145 @@
#!/usr/bin/env python
"""Provide a wrapper around the gym framework and the `gym.Env` class
This provides a wrapper around the gym framework such that is compatible with the pyrobolearn framework.
With this, users can use everything the various tools (models, algos, etc) defined in the pyrobolearn framework
on the OpenAI gym library.
"""
import inspect
import functools
import numpy as np
import gym
from gym import *
import warnings
warnings.simplefilter("ignore")
from pyrobolearn.states.gym_states import GymState
from pyrobolearn.actions.gym_actions import GymAction, Action
# from pyrobolearn.rewards import GymReward
# from terminating_condition import GymTerminatingCondition
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
def make(env_id):
"""Create the OpenAI Gym environment and return a wrapped version of it."""
env = gym.make(env_id)
env = GymEnvWrapper(env)
return env
def create(env_id):
"""Create the OpenAI Gym environment and return a wrapped version of it with the associated state and action."""
env = gym.make(env_id)
env = GymEnvWrapper(env)
return env, env.state, env.action
class GymEnvWrapper(gym.Env):
r"""Gym Environment wrapper
This update the data of the GymState, GymAction, and GymReward when performing a step in a gym environment.
"""
def __init__(self, env, state=None, action=None):
# set environment
self.env = env
# set state and action
self.state = state
self.action = action
# define the observation and action space
self.observation_space = self.env.observation_space
self.action_space = self.env.action_space
# self.reward = GymReward(1)
# self.done = GymTerminatingCondition(done=False)
@property
def env(self):
return self._env
@env.setter
def env(self, environment):
if not isinstance(environment, gym.Env):
raise TypeError("Expecting the given environment to be an instance of `gym.Env`")
self._env = environment
@property
def state(self):
return self._state
@state.setter
def state(self, s):
if s is None:
s = GymState(self.env)
self._state = s
@property
def action(self):
return self._action
@action.setter
def action(self, a):
if a is None:
a = GymAction(self.env)
self._action = a
def __getattr__(self, name):
"""Get the functions from the Gym Environment"""
attribute = getattr(self.env, name)
if inspect.isbuiltin(attribute):
attribute = functools.partial(attribute)
return attribute
def step(self, actions):
"""perform a step in the environment and set the data for the GymState and GymAction"""
if isinstance(actions, Action):
actions = actions.data[0]
if isinstance(self.action_space, gym.spaces.Discrete) and isinstance(actions, np.ndarray):
actions = actions[0]
observations, reward, done, info = self.env.step(actions)
self.state.data = observations
self.action.data = actions
# self.reward.value = reward
# self.done.done = done
return self.state, reward, done, info
def reset(self):
observation = self.env.reset()
self.state.data = observation
return self.state
def render(self, mode='human'):
self.env.render(mode)
def __repr__(self):
return self.env.__repr__()
def __str__(self):
return self.env.__str__()
# Test
if __name__ == '__main__':
env, state, action = create('CartPole-v1')
print("Env: {}".format(env))
print("\nState: {}".format(state))
print("-- shape: {}".format(state.shape))
print("-- space: {}".format(state.space))
print("\nAction: {}".format(action))
print("-- shape: {}".format(action.shape))
print("-- space: {}".format(action.space))
pyrobolearn/envs/locomotion_env.py
+67
View File
@@ -0,0 +1,67 @@
#!/usr/bin/env python
"""Define the locomotion environment.
Define the environment to perform a locomotion task; it mainly defines the reward function.
"""
from env import Env
from pyrobolearn.worlds import BasicWorld
from pyrobolearn.states import State
from pyrobolearn.policies import Policy
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class LocomotionEnv(Env):
r"""Locomotion environment
Define a simple environment for a locomotion task.
"""
def __init__(self, simulator, states, world=None, terminal_condition='default'):
r"""Initialize the locomotion environment
Args:
simulator (Simulator): simulator
states (State, Policy): states that environment must return. If the policy is given, it takes the states
that are given as input to the policy.
world (World, None): the world for the locomotion task. If None, it creates a basic world.
terminal_condition (str, default): the terminating condition criterion to stop if the policy failed
the task.
"""
# check parameters
if not isinstance(simulator, Simulator):
raise TypeError("Expecting the `simulator` parameter to be an instance of Simulator.")
if not isinstance(states, (State, Policy)):
raise TypeError("Expecting the `states` parameter to be an instance of State or Policy.")
# define world
if world is None:
world = BasicWorld(simulator)
# get states/actions if policy
actions = None
if isinstance(states, Policy):
actions = states.actions
states = states.states
# define reward based on states/actions
rewards = None
# define terminating condition criterion
if terminal_condition == 'default':
terminal_condition = None
terminal_condition = None
super(LocomotionEnv, self).__init__(world, states, rewards=rewards,
terminal_condition=terminal_condition, extra_info=None)
pyrobolearn/envs/terminating_condition.py
+158
View File
@@ -0,0 +1,158 @@
#!/usr/bin/env python
"""Define some common terminating condition for the environment.
"""
import numpy as np
from pyrobolearn.robots import Robot
from pyrobolearn.states import LinkState
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class TerminatingCondition(object):
r"""Terminating Condition
This class provides the basic layout for the `TerminatingCondition` class and its child classes.
This one can be used to check when an environment has fulfilled certain conditions and can be terminated.
This class can be further subdivided into two categories: failed and succeeded conditions.
* Failed conditions determine when a policy has failed to perform a certain task
* Succeeded conditions determine when a policy has succeeded to perform a certain task
In order to compute this condition, they can access to more information than the state provided to the agent(s).
"""
# TODO: we should be able to combine different conditions using 'OR' and 'AND'
def check(self):
"""
Check if the terminating condition has been fulfilled, and return True or False accordingly
"""
return False
def __repr__(self):
return self.__class__.__name__
def __call__(self, *args, **kwargs):
return self.check()
def __bool__(self):
return self.check()
__nonzero__ = __bool__
class FailedCondition(TerminatingCondition):
r"""Failed Terminating Condition
This determines when a policy or multiple ones have failed to perform a certain task.
"""
pass
class SucceededCondition(TerminatingCondition):
r"""Succeeded Terminating Condition
This determines when a policy or multiple ones have succeeded to perform a certain task.
"""
pass
class GymTerminatingCondition(TerminatingCondition):
r"""OpenAI Gym Terminating Condition
Returns if the OpenAI Gym environment has terminated. This does not provide any information if the environment
terminated because the policy succeeded or failed to perform the task.
"""
def __init__(self, done=False):
self.done = done
def check(self):
return self.done
class HasFallen(FailedCondition):
r"""Has Fallen Condition
Check if the given robot has fallen, by checking if its base is below a certain threshold.
"""
def __init__(self, robot, threshold=None):
self.robot = robot
self.threshold = threshold if threshold is not None else robot.height/4.
def check(self):
return self.robot.getBasePosition()[2] < self.threshold
def __repr__(self):
return self.__class__.__name__ + '(threshold=' + str(self.threshold) + ')'
class HasReached(SucceededCondition):
r"""Has Reached Condition
Check if the robot or a part of it has reached a certain position, configuration, or state for a certain amount
of time/steps.
"""
pass
class LinkInSpecifiedDirection(HasReached):
r"""Check if the specified link is in certain direction for a certain amount of time steps.
Specifically, it checks if the position of the link with respect to the world or another link is in a certain
direction for a certain amount of time steps. The position vector is considered to be in the good direction if
it belongs to the specified cone domain.
"""
def __init__(self, state, direction, domain=(0.95, 1.), total_steps=0):
if not isinstance(state, LinkState):
raise TypeError("Expecting the state to be an instance of LinkState, instead got: {}".format(type(state)))
self.state = state
self.direction = self.normalize(np.array(direction))
self.total_steps = total_steps
if len(domain) != 2:
raise ValueError("Expecting the domain to be a tuple or list of 2 values.")
self.domain = np.array(domain)
self.cnt = 0
@staticmethod
def normalize(x):
"""
Normalize the given vector.
"""
if np.allclose(x, 0):
return x
return x / np.linalg.norm(x)
def check(self):
"""
Check if the position of the link is in certain direction between the specified interval for a certain
amount of time steps.
Returns:
bool: True if the condition is satisfied.
"""
pos = self.state._data
pos = self.normalize(pos)
value = np.dot(pos, self.direction)
# check if the direction belongs to the cone domain
if self.domain[0] <= value <= self.domain[1]:
self.cnt += 1
# if we are in the cone domain for a certain amount of steps
if self.cnt > self.total_steps:
return True
else:
self.cnt = 0
return False
def __repr__(self):
return self.__class__.__name__ + '(direction=' + str(self.direction) + ')'
pyrobolearn/experiments/__init__.py
+3
View File
@@ -0,0 +1,3 @@
# import experiments
from experiment import Experiment
pyrobolearn/experiments/experiment.py
+151
View File
@@ -0,0 +1,151 @@
#!/usr/bin/env python
"""Define the `Experiment` class, which allows the user to define an experiment.
We define the `Experiment` class here which is the highest-level class of our framework. More specifically, it allows
to organize which tasks to run, which metrics to use, and allows to easily compare different models, algos, methods,
and so on.
An experiment should be well-defined, and clearly demonstrate the results.
Dependencies:
- `pyrobolearn.tasks` (thus `pyrobolearn.envs`, `pyrobolearn.policies`, `pyrobolearn.approximators`,...)
- `pyrobolearn.metrics`
- `pyrobolearn.algos`
"""
from abc import ABCMeta, abstractmethod
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class Experiment(object):
r"""Experiment class.
An experiment allows to organize which tasks to run (and in which order), which metrics to use to
evaluate the experiment, and to easily compare different learning models/rl, algos, and so on.
This represents the highest level of our `pyrobolearn` framework. Sometimes, the robot, policy, or algo
can be as well specified as an argument when creating an experiment.
.. seealso: the second highest level of our framework is the `Task` class which defines the policy and
environment, and is independent of the learning algorithm as well as the metrics.
Examples:
# 1st Example
sim = Bullet()
experiment = Experiment(sim) # this creates the task (states/actions, rl, environment), algo, metrics,...
results = experiment.run(train=True, evaluate_metrics=True)
experiment.plot_metrics()
reward = experiment.evaluate_policy()
# 2nd Example
sim = Bullet()
robots = [Robot(), Robot()]
algo = Algo()
experiment = Experiment(sim, robots, algo)
results = experiment(train=True, evaluate_metrics=True)
experiment.plot_metrics()
"""
__metaclass__ = ABCMeta
def __init__(self, tasks, policies, algos, metrics):
"""
Initialize an experiment.
Args:
tasks: defines the various tasks. Each task describe the world, the rewards/costs,
policies:
algos:
metrics:
"""
self.tasks = tasks
self.policies = policies
self.algos = algos
self.metrics = metrics
def run(self, train=False, evaluate_metrics=False):
"""
Run the experiment.
Args:
train (bool): if False, it does not train the policy(ies) using the algo(s).
evaluate_metrics (bool): if False, it does not evaluate the policy using the metric.
"""
# algos: train the rl on the given tasks
if train:
pass
# metrics: evaluate the rl/algos using the given metrics
if evaluate_metrics:
pass
def evaluate_policy(self):
"""
Evaluate the policies on the tasks using their corresponding objective functions.
"""
pass
def plot_metrics(self):
"""
Plot the results. This is let to the user to define this function.
"""
pass
def get_task(self, idx=None):
pass
def get_policy(self, idx=None):
pass
def get_state(self, idx=None, policy_idx=None):
pass
def get_action(self, idx=None, policy_idx=None):
pass
def get_environment(self, idx=None):
pass
def get_world(self, env_idx=None):
pass
def get_reward(self, env_idx=None, reward_idx=None):
pass
def get_algorithm(self, idx=None):
pass
def get_metric(self, idx=None):
pass
def change_simulator(self, simulator):
pass
class GymExperiment(Experiment):
r"""Gym Environment Experiment"""
def __init__(self, gym_env, policy, algo):
pass
class WalkingExperiment(Experiment):
def __init__(self, policies=None, algos=None):
tasks = [WalkingTask()]
metrics = None
if policies is None:
pass
if algos is None:
algos = [PPO()]
super(WalkingExperiment, self).__init__(tasks, metrics)
pyrobolearn/experiments/monitor.py
+8
View File
@@ -0,0 +1,8 @@
class Monitor(object):
r"""Monitor class.
Monitor an experiment.
"""
pass
pyrobolearn/tasks/__init__.py
+21
View File
@@ -0,0 +1,21 @@
# import from task
from task import Task
# import imitation learning task
from imitation import ILTask
# import reinforcement learning task
from reinforcement import RLTask
# import active learning task
from active import ALTask
# import transfer learning task
from transfer import TLTask
# import inverse reinforcement learning task
from inverse_reinforcement import IRLTask
# import curriculum learning task
from curriculum import CLTask
pyrobolearn/tasks/active.py
+36
View File
@@ -0,0 +1,36 @@
#!/usr/bin/env python
"""Define the active learning task.
"""
from imitation import ILTask
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class ALTask(ILTask):
r"""Active Learning Task
Task used for active learning. This is pretty similar to imitation learning with the exception that the policy
can decide to interact with the user (to ask for more demonstrations for instance). Thus the output of the policy
is interpreted by the interface.
"""
def __init__(self, environment, policies, interface=None, recorder=None):
"""Initialize the active learning task.
Args:
environment (Env): environment of the task (which contains the world)
policies (Policy): rl to be trained by the task
interface (Interface): input/output interface that allows to interact with the world.
recorder (Recorder): if the interface doesn't have a recorder, it can be supplemented here.
If the interface doesn't have a recorder, and it is not specified, it will create a recorder that
record the states and actions (inferred from the rl).
"""
super(ALTask, self).__init__(environment, policies, interface, recorder)
pyrobolearn/tasks/curriculum.py
+103
View File
@@ -0,0 +1,103 @@
#!/usr/bin/env python
"""Define the curriculum learning task.
This type of tasks starts by training agents on simple tasks/environments first and then increase the difficulty
level of tasks progressively.
This can be achieved in three different ways; discretely, continuously, or a mixture of both.
For instance, in a locomotion task where a robot has to learn how to walk, we can progressively increase the
difficulty level by:
- having different discrete type of worlds; starting from a world with a flat floor, passing to smooth terrains with
ups and downs, to a world filled with inanimate obstacles, to finally a lively world with different moving agents.
- modifying in a continuous manner the reward/cost function landscape. This can be achieved by increasing / decreasing
the coefficient values of some rewards/costs as the number of episodes/iterations progresses.
References:
[1] "Curriculum learning", Bengio et al., 2009
"""
import collections
from pyrobolearn.worlds import World
from task import Task, Env
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class CLTask(Task):
r"""Curriculum Learning Task
This type of tasks starts by training agents on simple tasks/environments first and then increase the difficulty
level of tasks progressively.
This can be achieved in three different ways; discretely, continuously, or a mixture of both.
For instance, in a locomotion task where a robot has to learn how to walk, we can progressively increase the
difficulty level by:
- having different discrete type of worlds; starting from a world with a flat floor, passing to smooth terrains
with ups and downs, to a world filled with inanimate obstacles, to finally a lively world with different moving
agents.
- modifying in a continuous manner the reward/cost function landscape. This can be achieved by increasing /
decreasing the coefficient values of some rewards/costs as the number of episodes/iterations progresses.
References:
[1] "Curriculum learning", Bengio et al., 2009
"""
def __init__(self, environments, policies):
self.environments = environments
first_environment = self.environments[0]
super(CLTask, self).__init__(first_environment, policies)
##############
# Properties #
##############
@property
def environments(self):
"""Return the list of environments ordered by complexities."""
return self._environments
@environments.setter
def environments(self, environments):
"""Set the environments sorted by increasing order of difficulty level."""
# TODO use an ordered dictionary
if isinstance(environments, Env):
environments = [environments]
elif isinstance(environments, collections.Iterable):
if len(environments) < 1:
raise ValueError("Expecting the list of environments to at least a length of one.")
for i, env in enumerate(environments):
if not isinstance(env, Env):
raise TypeError("Expecting 'environments' to be a list of environments, instead the {} item "
"in the list has a type of {}".format(i, type(env)))
else:
raise TypeError("Expecting 'environments' to be a list of environments, instead got "
"{}".format(type(environments)))
self._environments = environments
@property
def num_environments(self):
"""Return the number of environments."""
return len(self.environments)
###########
# Methods #
###########
def add_environment(self, environment, index=-1):
"""Add an environment at the specified index."""
if not isinstance(environment, Env):
raise TypeError("Expecting the given 'environment' to be an instance of Environment, instead got "
"{}".format(type(environment)))
self.environments.insert(index, environment)
def remove_environment(self, index=-1):
"""Remove and return the specified environment from the ordered list of environments."""
return self.environments.pop(index)
pyrobolearn/tasks/imitation.py
+432
View File
@@ -0,0 +1,432 @@
#!/usr/bin/env python
"""Define the imitation learning task.
Dependencies:
- `pyrobolearn.recorders`
- `pyrobolearn.tools` (interfaces / bridges to be used to demonstrate a certain skill).
"""
import collections
from itertools import count
import time
import numpy as np
from task import Task
from pyrobolearn.recorders import Recorder, StateRecorder, ActionRecorder
# from pyrobolearn.tools.interfaces import Interface
from pyrobolearn.tools.bridges import Bridge
from pyrobolearn.tools.bridges.mouse_keyboard import BridgeMouseKeyboardImitationTask
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class ILTask(Task):
r"""Imitation Learning Task
Imitation learning consists for an agent to generalize to reproduce/perform a certain task from possibly few
demonstrations [1].
References:
[1] "Learning from Humans", Billard et al., 2016
"""
def __init__(self, environment, policies, interface=None, recorders=None):
"""
Initialize the imitation learning task.
Args:
environment (Env): environment of the task (which contains the world)
policies (Policy): rl to be trained by the task
interface (Bridge, Interface): input/output interface that allows to interact with the world.
If None is provided, it will create by default the `MouseKeyboardInterface`.
recorders (Recorder, [Recorder]): if the interface doesn't have a recorder, it can be supplemented here.
If the interface doesn't have a recorder, and it is not specified, it will create a recorder that
record the states and actions (inferred from the rl).
"""
super(ILTask, self).__init__(environment, policies)
# recorder
self.recorders = recorders
# interface
self.bridge = interface
# few useful variables (that are accessed by the bridge)
self.recording_enabled = False
self.end_recording = False
self.training_enabled = False
self.end_training = False
self.testing_enabled = False
self.end_testing = False
self.end_task = False
# use the following code to check the caller class and method
# import inspect
# stack = inspect.stack()
# the_class = stack[1][0].f_locals["self"].__class__
# the_method = stack[1][0].f_code.co_name
##############
# Properties #
##############
@property
def recording_enabled(self):
return self._recording_enabled
@recording_enabled.setter
def recording_enabled(self, boolean):
self._recording_enabled = boolean
# if you are recording, you can not train or test
if self._recording_enabled:
self._training_enabled = False
self._testing_enabled = False
@property
def training_enabled(self):
return self._training_enabled
@training_enabled.setter
def training_enabled(self, boolean):
self._training_enabled = boolean
# if you are training, you can not record or test
if self._training_enabled:
self._recording_enabled = False
self._testing_enabled = False
@property
def testing_enabled(self):
return self._testing_enabled
@testing_enabled.setter
def testing_enabled(self, boolean):
self._testing_enabled = boolean
# if you are testing, you can not record or train
if self._testing_enabled:
self._recording_enabled = False
self._training_enabled = False
@property
def recorders(self):
"""Return the list of recorders."""
return self._recorders
@recorders.setter
def recorders(self, recorders):
"""Set the recorders."""
if recorders is None:
recorders = [StateRecorder(self.policies.states), ActionRecorder(self.policies.actions)]
elif isinstance(recorders, Recorder): # or not isinstance(recorders, collections.Iterable):
recorders = [recorders]
if not isinstance(recorders, collections.Iterable):
raise TypeError("Expecting a list of recorders.")
# check each recorder
for recorder in recorders:
if not isinstance(recorder, Recorder):
raise TypeError("Expecting each recorder to be an instance of `pyrobolearn.recorders.Recorder`, "
"instead got: {}".format(type(recorder)))
self._recorders = recorders
@property
def bridge(self):
"""Return the bridge to the interface."""
return self._bridge
@bridge.setter
def bridge(self, bridge):
"""Set the bridge to the interface. If None, create a Bridge to a mouse-keyboard interface."""
# If no bridge given, create one
if bridge is None:
bridge = BridgeMouseKeyboardImitationTask()
# elif isinstance(bridge, Interface):
# pass
# check if bridge instance
if not isinstance(bridge, Bridge):
raise TypeError("Expecting the bridge to be an instance of Bridge, got instead {}".format(type(bridge)))
# set bridge and task
self._bridge = bridge
self._bridge.task = self
@property
def interface(self):
"""Return the interface instance associated with the bridge."""
return self._bridge.interface
###########
# Methods #
###########
def step_bridge(self, update_interface=True):
"""
Perform one step with the interface and bridge.
Args:
update_interface (bool): if True, it will perform one step with the interface.
"""
# perform one step with the interface and bridge
self.bridge.step(update_interface=update_interface)
def step(self, deterministic=True, render=True):
"""
Perform one step with the environment and the policies.
"""
if render:
self.env.render()
for policy in self.policies:
# prev_obs = copy.deepcopy(policy.states.data)
if self.testing_enabled:
actions = policy.act(policy.states, deterministic=deterministic)
else:
actions = None
obs, rew, done, info = self.env.step(actions)
self._done = done
# d = {'prev_obs': prev_obs, 'actions': copy.deepcopy(actions.data),
# 'obs': copy.deepcopy(policy.states.data), 'rew': rew, 'done': done}
# results.append(d)
def save_recorders(self):
"""
Save data from recorders.
"""
for recorder in self.recorders:
recorder.save()
def reset_recorders(self):
"""
Reset the recorders.
"""
for recorder in self.recorders:
recorder.reset()
def add_data_row_in_recorder(self):
"""
Add a new data row in the recorder's data "matrix".
"""
for recorder in self.recorders:
recorder.add_row()
def discard_last_recorded_data(self):
"""
Discard/remove the last recorded piece of data.
"""
for recorder in self.recorders:
recorder.remove_last_entry()
def record_step(self):
"""Perform one step in the recording."""
if self.recording_enabled:
# perform a step with the recorder
for recorder in self.recorders:
recorder.record()
def record(self, num_steps=None, dt=1./240, signal_from_interface=True): # , render=True):
"""
Record the states / actions using the recorders.
Args:
num_steps (int, None): total number of steps to take in the environment. If None, it will not end.
dt (float, None): time to sleep before going to the next step. If None, it will be 1./240.
render (bool): If True, it will render the environment.
signal_from_interface (bool): If True, it is assumed that a signal is sent by the interface to start/stop
the recording.
"""
# check if the recording is enabled
self.recording_enabled = not signal_from_interface
self.bridge.enable_recording = signal_from_interface
# check the number of steps
if num_steps is None:
num_steps = np.infty
# check dt
if dt is None or dt < 0.:
dt = 1./240
# run several steps in the world
self.reset()
for t in count():
if t >= num_steps or self.end_recording:
self.end_recording = False
break
# perform one step with the interface and bridge
self.bridge.step(update_interface=True)
# record if specified
self.record_step()
# perform a step in the world
self.world.step(sleep_dt=dt)
def train_step(self):
"""
Perform one step in the training of the policy(ies) using the recorded data.
"""
if self.training_enabled:
# print(self.recorders[0].data)
# print(np.array(self.recorders[0].data).shape)
# take data from recorder
data = np.array([data[0] for data in self.recorders[0].data]).T
# train policy
for policy in self.policies:
policy.imitate(data)
def train(self, num_iters=1000, signal_from_interface=False):
"""
Train the policy(ies) using the recorded data.
Args:
num_iters (int): number of iterations to train the model.
signal_from_interface (bool): If True, it is assumed that a signal is sent by the interface to start/stop
the training.
"""
# check if the training is enabled
self.training_enabled = not signal_from_interface
self.bridge.enable_training = signal_from_interface
# check the number of steps
if num_iters is None or num_iters < 0:
num_iters = 1000
# run several steps in the environment
for i in range(num_iters):
if self.end_training:
self.end_training = False
break
# perform one step with the interface and bridge
self.bridge.step(update_interface=True)
# perform a step in the training process
self.train_step()
# TODO one step learning
if self.training_enabled:
self.end_training = True
self.training_enabled = False
def test_step(self):
"""
Perform one step in the test process.
"""
if self.testing_enabled:
# perform a step in the environment
self.step(render=True)
def test(self, num_steps=None, dt=1./240, signal_from_interface=False):
"""
Test the policy(ies) in the environment.
Args:
num_steps (int, None): total number of steps to take in the environment. If None, it will not end.
dt (float, None): time to sleep before going to the next step. If None, it will be 1./240.
render (bool): If True, it will render the environment.
signal_from_interface (bool): If True, it is assumed that a signal is sent by the interface to start/stop
the testing.
"""
# check if the recording is enabled
self.testing_enabled = not signal_from_interface
self.bridge.enable_testing = signal_from_interface
# check the number of steps
if num_steps is None:
num_steps = np.infty
# check dt
if dt is None or dt < 0.:
dt = 1. / 240
# run several steps in the environment
self.reset()
for t in count():
if t >= num_steps or self.end_testing:
self.end_testing = False
break
# perform one step with the interface and bridge
self.bridge.step(update_interface=True)
# perform a step in the environment
if self.testing_enabled:
self.step(render=True)
else:
self.world.step()
# sleep
time.sleep(dt)
def run(self, num_steps=None, dt=1./240, render=True):
"""
Reset and run the task until it is done, or the current time step matches num_steps.
It allows to record data, train, and test a policy using the bridge between the interface and this task.
Warnings: It is assumed that the bridge will send the signals to record the data, train and test the policy,
and end the task.
Args:
num_steps (int, None): total number of steps to take in the environment. If None, it will not end.
dt (float, None): time to sleep before going to the next step. If None, it will be 1./240.
render (bool): If True, it will render the environment. We always render in an imitation task, as it does
not make sense to not render in this setting.
"""
# We use the interface
self.recording_enabled = False
self.training_enabled = False
self.testing_enabled = False
self.bridge.enable_recording = True
self.bridge.enable_training = True
self.bridge.enable_testing = True
# check the number of steps
if num_steps is None:
num_steps = np.infty
# check dt
if dt is None or dt < 0.:
dt = 1. / 240
self.reset()
for t in count():
if t >= num_steps or self.end_task:
self.end_task = False
break
# perform one step with the interface and bridge
self.bridge.step(update_interface=True)
# record
if self.recording_enabled:
self.record_step()
# perform a step forward in the simulation
self.world.step(sleep_dt=dt)
# train
if self.training_enabled:
self.train_step()
# perform a step forward in the simulation (without sleeping?) TODO
self.world.step()
# test
if self.testing_enabled:
self.test_step()
# perform a step in the environment
self.step(render=True)
time.sleep(dt)
# if not recording, training or testing, just perform a step in the world
if not (self.recording_enabled or self.training_enabled or self.testing_enabled):
self.world.step(sleep_dt=dt)
pyrobolearn/tasks/inverse_reinforcement.py
+36
View File
@@ -0,0 +1,36 @@
#!/usr/bin/env python
"""Define the inverse reinforcement learning task.
"""
from imitation import ILTask
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class IRLTask(ILTask):
r"""Inverse Reinforcement Learning Task.
The goal of the inverse reinforcement learning task is to approximate the reward function from demonstrations
by the user.
"""
def __init__(self, environment, policies, reward_approximator, interface=None, recorder=None):
"""Initialize the inverse reinforcement learning task.
Args:
environment (Env): environment of the task (which contains the world)
policies (Policy): rl to be trained by the task
reward_approximator (Approximator): reward function approximator
interface (Interface): input/output interface that allows to interact with the world.
recorder (Recorder): if the interface doesn't have a recorder, it can be supplemented here.
If the interface doesn't have a recorder, and it is not specified, it will create a recorder that
record the states and actions (inferred from the rl).
"""
super(IRLTask, self).__init__(environment, policies, interface, recorder)
pyrobolearn/tasks/misc.py
+46
View File
@@ -0,0 +1,46 @@
#!/usr/bin/env python
"""Define the miscellaneous tasks.
"""
from tasks import Task, ILTask, RLTask
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class CombinedTask(Task):
def __init__(self, tasks):
super(CombinedTask, self).__init__(simulator, env)
def __rshift__(self, other):
"""
Add another task in sequence.
Args:
other (Task):
"""
pass
class WalkingTask(RLTask):
def __init__(self, simulator, robot=None, policy=None):
# define world
world = World(simulator)
world.setGravity()
# define reward
rewards = [Reward()]
# define env
env = Env(simulator, world, policies, rewards)
super(WalkingTask, self).__init__(env)
pyrobolearn/tasks/reinforcement.py
+29
View File
@@ -0,0 +1,29 @@
#!/usr/bin/env python
"""Define the reinforcement learning task.
"""
from task import Task
import gym
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class RLTask(Task):
r"""Reinforcement Learning Task
Reinforcement learning consists for an agent to learn to perform a certain task by maximizing the expected total
reward [1,2].
References:
[1] "Reinforcement Learning: An Introduction", Sutton and Barto, 2018
[2] "A Survey on Policy Search for Robotics", Deisenroth et al., 2013
"""
def __init__(self, environment, policies):
super(RLTask, self).__init__(environment, policies)
pyrobolearn/tasks/scheduler.py
+53
View File
@@ -0,0 +1,53 @@
#!/usr/bin/env python
"""Define the `Scheduler` class which deals on how to run multiple tasks in a sequential or parallel way.
A user can create its own task/scenario independently of the rest. Once a task is done, which is known when
the environment returns `done=True`, the next task is loaded into memory. When sequencing tasks, if a policy
is defined for Task1 and no policy is defined for Task2, it will use the same policy. If another policy is
defined it will sequence this one with the previous policy. The same rationale applies for the world, robots,
and so on.
"""
from task import Task
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class Scheduler(object):
r"""Scheduler class.
The scheduler is responsible on how to run multiple tasks might it be in a sequential or parallel manner.
Some tasks can only be run after certain conditions are met, thus in our framework, we represent the scheduler
as a directed graph.
If the user has only one task, this class is not useful. The scheduler can be dynamically built as the agent(s)
progress(es) in the various tasks.
"""
def __init__(self, tasks):
self.tasks = tasks
self.graph = {}
def add_task(self, task, previous_tasks=None, next_tasks=None):
pass
class NodeTask(object):
def __init__(self, task):
self.task = task
# references to the parent/children nodes
self.parents = []
self.children = []
def is_done(self):
return self.task.is_done()
pyrobolearn/tasks/task.py
+270
View File
@@ -0,0 +1,270 @@
#!/usr/bin/env python
"""Define the abstract 'Task'/'Scenario' class.
A task represents a certain learning paradigm, and use specific metrics with respect to that paradigm.
Learning paradigms include imitation learning (IL), reinforcement learning (RL), transfer learning (TL),
active learning (AL), etc.
IL, AL, and RL tasks groups the environment and policies together. The task allows you thus to run the process
that happens between the environment and the policies. That is, the environment produces the states and possible
rewards while the policies take the states and produce actions.
Tasks can be sequenced one after another and represented as a directed graph / state machine. For instance,
the first task might be to climb stairs, then the second one might be to open a door, etc. This allows to combine
different scenarios in a modular way.
Dependencies:
- `pyrobolearn.envs`
- `pyrobolearn.policies`
"""
import collections
import copy
import time
from abc import ABCMeta
from itertools import count
import numpy as np
from pyrobolearn.envs import Env, gym
from pyrobolearn.policies import Policy
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class Task(object):
r"""Task class.
The Task defines the policy (and thus the learning model), environment (with the world and rewards), the states,
and actions. For some tasks, the robot and/or policy can be provided as an input to the task.
The `Task` represents the second highest level of our `pyrobolearn` framework. It is notably independent of the
learning algorithm that is used to train the policy (i.e. learning model).
For instance, a 'walking task' should be independent of the robot, policy, learning algorithm, and sometimes,
of the particular terrain. However, the task already defines the various rewards that might be useful for this
one along with a default world.
.. seealso: * The highest level of our framework is the `Experiment` class which defines in addition the metrics
used to evaluate our tasks and algos. The algorithm is often defined in that class but can sometimes
be given as an input.
* The next lower level of our framework is the `Environment` and `Policy` classes.
* If there are multiple tasks, the `Scheduler` class which organizes how to run them might interest
the user.
The task is often given to the learning algorithm, which can then train the policy in the corresponding
environment.
"""
__metaclass__ = ABCMeta
def __init__(self, environment, policies):
if not isinstance(environment, (Env, gym.Env)):
raise TypeError("Expecting 'environment' to be an instance of Env or gym.Env")
if isinstance(policies, collections.Iterable):
for policy in policies:
if not isinstance(policy, Policy):
raise TypeError("Expecting 'policies' to be a list/tuple of Policy instances")
elif isinstance(policies, Policy):
policies = [policies]
else:
raise TypeError("Expecting 'policies' to be an instance of Policy, or list/tuple of policies")
self.env = environment
self.policies = policies
self._done = False
self._succeeded = False
##############
# Properties #
##############
@property
def done(self):
"""
Return if the task is done or not.
"""
return self._done
@property
def succeeded(self):
"""
Return if the task succeeded or not.
"""
return self._succeeded
@property
def failed(self):
"""
Check if the task failed or not.
"""
return not self.succeeded
@property
def simulator(self):
"""
Return the simulator.
"""
return self.env.simulator
@property
def world(self):
"""
Return the world instance.
"""
return self.env.world
@property
def policy(self):
"""
Return the policies.
"""
if len(self.policies) == 1:
return self.policies[0]
return self.policies
@property
def learning_model(self):
"""
Return the learning models.
"""
if len(self.policies) == 1:
return self.policies[0].model
return [policy.model for policy in self.policies]
@property
def environment(self):
"""
Return the environment.
"""
return self.env
@property
def rewards(self):
"""
Return the rewards.
"""
return self.env.rewards
@property
def states(self):
"""
Return the states.
"""
return self.env.states
@property
def actions(self):
"""
Return the actions.
"""
if len(self.policies) == 1:
return self.policies[0].actions
return [policy.actions for policy in self.policies]
###########
# Methods #
###########
def is_finished(self):
"""
Check if the task is finished or not.
"""
return self.done
def has_succeeded(self):
"""
Check if the task has succeeded.
"""
return self.succeeded
def has_failed(self):
"""
Check if the task has failed.
"""
return self.failed
def reset(self):
"""
Reset the task; reset the environment and policies
"""
# reset variables
self._done = False
self._succeeded = False
# reset env and policies
self.env.reset()
for policy in self.policies:
policy.reset()
def run(self, num_steps=None, dt=0, use_terminating_condition=False, render=False):
"""
Reset and run the task until it is done, or the current time step matches num_steps.
"""
if num_steps is None:
num_steps = np.infty
# results = []
total_rewards = np.zeros(len(self.policies))
self.reset()
# for t in range(4):
# for policy in self.policies:
# actions = policy.act(policy.states)
# self.simulator.stepSimulation()
# time.sleep(2.)
for t in count():
if t >= num_steps:
break
rewards = self.step(render=render)
# result = self.step(render=render)
# results.append(result)
total_rewards += rewards
if use_terminating_condition and self._done:
break
time.sleep(dt)
# return results
if total_rewards.size == 1:
return total_rewards[0]
return total_rewards
def step(self, deterministic=True, render=False):
"""
Perform one step.
"""
if render:
self.env.render()
# results = []
rewards = []
for policy in self.policies:
# prev_obs = copy.deepcopy(policy.states.data)
actions = policy.act(policy.states, deterministic=deterministic)
obs, rew, done, info = self.env.step(actions)
self._done = done
# d = {'prev_obs': prev_obs, 'actions': copy.deepcopy(actions.data),
# 'obs': copy.deepcopy(policy.states.data), 'rew': rew, 'done': done}
# results.append(d)
rewards.append(rew)
# return results
return np.array(rewards)
def get_policy(self, idx=None):
if idx is None:
return self.policies
return self.policies[idx]
def get_learning_model(self, idx=None):
if idx is None:
return [policy.model for policy in self.policies]
return self.policies[idx].model
def save_task(self, filename):
pass
# alias
Scenario = Task
pyrobolearn/tasks/transfer.py
+89
View File
@@ -0,0 +1,89 @@
#!/usr/bin/env python
"""Define the transfer learning task.
"""
from task import Task
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "MIT"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class TLTask(object):
r"""Transfer Learning task
Transfer learning consists to transfer the knowledge acquired by the agent while solving a problem to another
different but similar problem [1,2].
References:
[1] "A Survey on Transfer Learning", Pan et al., 2010
[2]" Transfer Learning for Reinforcement Learning Domains: A Survey", Taylor et al., 2009
"""
def __init__(self, domain_task, target_task):
# super(TLTask, self).__init__(environment, policies)
self.domain_task = domain_task
self.target_task = target_task
##############
# Properties #
##############
@property
def domain_task(self):
"""Return the domain task."""
return self._domain_task
@domain_task.setter
def domain_task(self, task):
"""Set the domain task."""
if not isinstance(task, Task):
raise TypeError("Expecting the domain task to be an instance of Task, instead got {}".format(type(task)))
self._domain_task = task
@property
def target_task(self):
"""Return the target task."""
return self._target_task
@target_task.setter
def target_task(self, task):
"""Set the target task."""
if not isinstance(task, Task):
raise TypeError("Expecting the target task to be an instance of Task, instead got {}".format(type(task)))
self._target_task = task
@property
def domain_environment(self):
"""Return the domain environment."""
return self.domain_task.environment
@property
def target_environment(self):
"""Return the target environment."""
return self.target_task.environment
@property
def domain_policies(self):
"""Return the domain policies."""
return self.domain_task.policies
@property
def target_policies(self):
"""Return the target policies."""
return self.target_task.policies
###########
# Methods #
###########
def train(self):
pass
def test(self):
pass