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

update/add approximators, dynamics, values, policies (ongoing)

Browse Source
This commit is contained in:
Brian Delhaisse
2019-11-14 08:29:29 +01:00
parent d36a3ba63d
commit d7becef6dd
37 changed files with 1755 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyrobolearn/approximators/__init__.py
+15 -4
View File
@@ -4,11 +4,22 @@
from .approximator import *
# import basic function approximators (random, linear, polynomial)
from .basic_approximator import *
from .basic import *
# import linear function approximator
from .linear import LinearApproximator
# import polynomial function approximator
from .polynomial import PolynomialApproximator
# import nn function approximators
from .nn_approximator import *
from .nn import *
# import gp function approximators
# from .gp_approximator import *
# import gpr function approximators
# from .gpr import *
# import gmr function approximators
# from .gmr import *
# import kmp function approximators
# from .kmp import *
pyrobolearn/approximators/basic_approximator.py → pyrobolearn/approximators/basic.py
View File
pyrobolearn/approximators/gmr.py
+164
View File
@@ -0,0 +1,164 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Define Gaussian Process function approximator.
Dependencies:
- `pyrobolearn.models`
- `pyrobolearn.states`
- `pyrobolearn.actions`
"""
from pyrobolearn.approximators.approximator import Approximator
from pyrobolearn.models.gmm import GMM
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class GMRApproximator(Approximator):
r"""Gaussian Mixture Regression Approximator
The Gaussian mixture regression (GMR) approximator depends on the Gaussian mixture model (GMM).
GMM
---
The Gaussian Mixture Model (GMM) is a semi-parametric, probabilistic and generative model [1,2].
In robotics, for instance, this is often used to model trajectories by jointly encoding the time and state
(position and velocity) [3,4,5,6].
It is mathematically described by:
.. math:: p(x) = \sum_{k=1}^{K} \pi_k \mathcal{N}(\mu_k, \Sigma_k)
where :math:`K` is the number of components, :math:`\pi_k` are prior probabilities (that is
:math:`0 \leq \pi_k \leq 1`) that sums to 1 (i.e. :math:`\sum_{k=1}^K \pi_k = 1`),
:math:`\mathcal{N}(\mu_k, \Sigma_k)` is the multivariate Gaussian (aka Normal) distribution, with mean
:math:`\mu_k` and covariance :math:`\Sigma_k`. The priors, means, and covariances are grouped to form the
parameter set :math:`\theta = \{\pi_k, \mu_k, \Sigma_k\}_{k=1}^K`.
Learning from data
------------------
There are three main ways to learn the parameters: maximum likelihood estimate (MLE), maximum a posteriori
estimate (MAP), and bayesian inference (using variational inference). Here, we will focus on MLE.
Given a dataset :math:`X \in \mathbb{R}^{N \times D}`, the log-likelihood of the GMM is given by:
.. math::
\mathcal{L}(\theta) &= \log p(X | \theta) = \log p(X | \pi, \mu, \Sigma) \\
&= \sum_{n=1}^N \log \sum_{k=1}^K \pi_k \mathcal{N}(x_n | \mu_k, \Sigma_k)
The summation inside the logarithm in the above loss does not allow for a closed-form solution. We thus turn our
attention to an iterative algorithm that maximizes this last one.
The Expectation-Maximization (EM) algorithm [1,2] allows to find the maximum likelihood estimate for models having
latent variables. This algorithm consists of 4 main steps:
1. Initialize the parameters :math:`\theta = \{\pi_k, \mu_k, \Sigma_k\}_{k=1}^K`
2. Expectation step: evaluate the posterior :math:`p(Z | X, \theta_{old})` while fixing the parameters.
3. Maximization step: maximize the expected value of the complete-data log-likelihood under the posterior
distribution of the latent variables (found during the Expectation step). That is,
:math:`\max_\theta Q(\theta, \theta_{old}) = \max_\theta \sum_{Z} p(Z | X, \theta_{old}) \log p(X,Z | \theta)`.
4. Evaluate the log-likelihood loss, and check if it converged. If it didn't, go back to step 2.
To be more specific, the EM algorithm alternatively computes a lower bound on the log-likelihood for the current
parameters, and then maximize this bound to obtain the new parameter values (see [1], sec 9.4 for more details).
This results in the above algorithm.
Few notes with respect to the EM algorithm:
* this guarantees an improvement over the but the initialization is quite important. In the literature, we can
often initialize it using the K-means algorithm.
* while other learning algorithms such as gradient ascent could be used, one of the major problem is that they
do not enforce constraints on the priors and covariance matrices during the optimization.
For other variants of the EM algorithm, please refer to [2], section 11.4.9.
Gaussian Mixture Regression (i.e. conditioned GMM)
--------------------------------------------------
Gaussian Mixture Regression [3,4] consists to condition the GMM (that models the joint distribution over the input
and output variables :math:`p(x^I, x^O)`) on a part of the variables (for instance, the input variables
:math:`p(x^O | x^I`). Let's :math:`x = [x^I, x^O]`, :math:`\mu_k = [\mu_k^I, \mu_k^O]`, and :math:`\Sigma_k =
\left[ \begin{array}{cc} \Sigma_k^I & \Sigma_k^{IO} \\ \Sigma_k^{OI} & \Sigma_k^O \end{array} \right]`, where
:math:`I` and :math:`O` are the superscripts to refer the input and output respectively.
.. math::
p(x^O | x^I) &= \sum_{k=1}^K p(z_k=1 | x^I) p(x^O | x^I, z_k=1) \\
&= \sum_{k=1}^K r_k(x^I) \mathcal{N}(\hat{\mu}_k^O(x^I), \hat{\Sigma}_k^O)
where :math:`r_k(x^I) = \frac{\pi_k \mathcal{N}(x^I|\mu_k^I, \Sigma_k^I)}{\sum_{j=1}^{K} \pi_j
\mathcal{N}(x^I|\mu_j^I,\Sigma_j^I)}` are the responsibilities, :math:`\hat{\mu}_k^O(x^I) =
\mu_k^O + \Sigma_k^{OI} \Sigma_k^I^{-1} (x^I - \mu_k^I)` and :math:`\hat{\Sigma}_k^O = \Sigma_k^O -
\Sigma_k^{OI} (\Sigma_k^I)^{-1} \Sigma_k^{IO}` are the resulting conditioned means and covariances,
respectively.
This results in another GMM, which can be approximated by a simple Gaussian (see [4] for more info, or the
documentation of the corresponding method: `approximate_by_single_gaussian`):
.. math::
p(x^O | x^I) \approx \mathcal{N}(x^O | \hat{\mu}^O(x^I), \hat{\Sigma}^O(x^I))
where :math:`\hat{\mu}^O(x^I) = \sum_{k=1}^K r_k(x^I) \hat{\mu}_k^O(x^I)` and :math:`\hat{\Sigma}^O(x^I) =
\sum_{k=1}^K r_k(x^I) (\hat{\Sigma}_k^O + \hat{\mu}_k^O(x^I) \hat{\mu}_k^O(x^I)^T) - \hat{\mu}^O(x^I)
\hat{\mu}^O(x^I)^T`.
Other miscellaneous information
-------------------------------
The conjugate prior of the GMM is the Dirichlet process.
References:
- [1] "Pattern Recognition and Machine Learning" (chap 2, 3, 9, and 10), Bishop, 2006
- [2] "Machine Learning: a Probabilistic Perspective" (chap 3 and 11), Murphy, 2012
- [3] "Robot Programming by Demonstration: a Probabilistic Approach" (chap 2), Calinon, 2009
- [4] "A Tutorial on Task-Parameterized Movement Learning and Retrieval", Calinon, 2015
- [5] "Programming by Demonstration on Riemannian Manifolds" (PhD thesis, chap 1 and 2), Zeerstraten, 2017
- [6] "Learning Control", Calinon et al., 2018
"""
def __init__(self, inputs, outputs, num_components=1, priors=None, means=None, covariances=None, gaussians=None,
preprocessors=None, postprocessors=None):
"""
Initialize the Gaussian Mixture regression approximator.
Args:
inputs (State, Action, np.array, torch.Tensor): inputs of the inner models (instance of Action/State)
outputs (State, Action, np.array, torch.Tensor): outputs of the inner models (instance of Action/State)
num_components (int): the number of components/gaussians (this argument should be provided if
no priors, means, covariances, or gaussians are provided)
priors (list/tuple of float, None): prior probabilities (they have to be positives). If not provided,
it will be a uniform distribution.
means (list of np.array[float[D]], None): list of means
covariances (list of np.array[float[D,D]], None): list of covariances
gaussians (list of Gaussian, None): list of gaussians. If provided, the `means` and `covariances`
parameters don't have to be provided.
preprocessors (None, Processor, list of Processor): the inputs are first given to the preprocessors then
to the model.
postprocessors (None, Processor, list of Processor): the predicted outputs by the model are given to the
processors before being returned.
"""
# create inner model
num_inputs, num_outputs = self._size(inputs), self._size(outputs)
dimensionality = num_inputs + num_outputs
model = GMM(num_components=num_components, priors=priors, means=means, covariances=covariances,
gaussians=gaussians, dimensionality=dimensionality)
# call parent class
super(GMRApproximator, self).__init__(inputs, outputs, model=model, preprocessors=preprocessors,
postprocessors=postprocessors)
pyrobolearn/approximators/gpr.py
+123
View File
@@ -0,0 +1,123 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Define Gaussian Process function approximator.
Dependencies:
- `pyrobolearn.models`
- `pyrobolearn.states`
- `pyrobolearn.actions`
"""
from pyrobolearn.approximators.approximator import Approximator
from pyrobolearn.models.gp import GPR
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class GPRApproximator(Approximator):
r"""Gaussian Process Regression Approximator
The Gaussian process is a generalization of the multivariate Gaussian distribution. It is a non-parametric,
probabilistic, and discriminative model.
This works by putting a prior distribution on the function:
.. math:: f|X ~ GP(0, K(X,X))
where :math:`K(\cdot, \cdot)` is the kernel matrix where each entry contains :math:`k(x_i, x_j)`, i.e. the kernel
function evaluated at the corresponding points. As it can be seen the kernel matrix grows with the number of
samples. The kernel function often has hyperparameters :math:`\Phi` that will be optimized.
The likelihood is given by:
.. math:: p(y | f) = \mathcal{N}(y | f, \sigma^2 I)
Learning the hyperparameters of the kernel are carried out by maximizing the marginal log likelihood, which is
given by:
.. math::
\log p(y | X; \Phi) &= \int p(y | f) p(f | X; \Phi) df \\
\log p(y | X; \Phi) &= -\frac{1}{2} y^\top (K + \sigma^2 I)^{-1} y - \frac{1}{2} \log |K + \sigma^2 I| -
\frac{n}{2} \log 2\pi
That is, we optimize the hyperparameters of the kernel function by maximizing the marginal log likelihood:
.. math::
\Phi^* = \arg \max_{\Phi} p(Y | X; \Phi)
The predictive distribution is then carried out by assuming that the observed target values :math:`y` and
the function values :math:`f^*` at the test locations :math:`X^*` are from the same joint Gaussian distribution.
By conditioning this distribution with respect to the old dataset :math:`X, y` and the test locations :math:`X^*`,
we can derive :math:`p(f^* | X, y, X^*)` which is the predictive output distribution given the new data points
:math:`X^*`.
Notes:
* GP takes into account correlations in the input domain but not in the output space
* The time complexity to learn a GP is :math:`O(N^3)` because of the matrix inversion during training.
* GMM vs GP:
* Both are probabilistic models.
* GMM is a generative semi-parametric model while GP is a discriminative non-parametric model.
* GMM captures the correlation between the inputs and outputs, while GP only captures correlation in the
input space.
* GMR models the variability/correlation between the predicted outputs while the GP provides uncertainty
on the predicted outputs. The predicted outputs in a GP are independent unless using a heteroscedastic
GP or a generalized Wishart process is used.
GPyTorch::
For most GP regression models, you will need to construct the following GPyTorch objects:
1. A GP Model (`gpytorch.models.ExactGP`) - This handles most of the inference.
2. A Likelihood (`gpytorch.likelihoods.GaussianLikelihood`) - This is the most common likelihood used for GP
regression.
3. A Mean - This defines the prior mean of the GP. If you don't know which mean to use, a
`gpytorch.means.ConstantMean` is a good place to start.
4. A Kernel - This defines the prior covariance of the GP. If you don't know which kernel to use, a
`gpytorch.kernels.ScaleKernel(gpytorch.kernels.RBFKernel())` is a good place to start.
5. A MultivariateNormal Distribution (`gpytorch.distributions.MultivariateNormal`) - This is the object used to
represent multivariate normal distributions.
References:
- [1] "Gaussian Processes for Machine Learning", Rasmussen and Williams, 2006
- [2] GPy: https://gpy.readthedocs.io/en/deploy/
- [3] GPyTorch: https://github.com/cornellius-gp/gpytorch
- [4] GPFlow: http://gpflow.readthedocs.io/en/latest/intro.html
"""
def __init__(self, inputs, outputs, mean=None, kernel=None, model=None, likelihood=None,
preprocessors=None, postprocessors=None):
"""
Initialize the Gaussian process regression approximator.
Args:
inputs (State, Action, np.array, torch.Tensor): inputs of the inner models (instance of Action/State)
outputs (State, Action, np.array, torch.Tensor): outputs of the inner models (instance of Action/State)
mean (None, gpytorch.means.Mean): mean prior. If None, it will be set to `gpytorch.means.ConstantMean()`.
kernel (None, gpytorch.kernels.Kernel): kernel prior. If None it will be set to
`gpytorch.kernels.ScaleKernel(gpytorch.kernels.RBFKernel() + gpytorch.kernels.WhiteNoiseKernel())`
model (None, gpytorch.module.Module): the prior GP model. If None, it will create `ExactGPModel()`, a GP
model using the provided mean, kernel, and likelihood.
likelihood (None, gpytorch.likelihoods.Likelihood): the likelihood pdf. If None, it will use the
`gpytorch.likelihoods.GaussianLikelihood()`.
preprocessors (None, Processor, list of Processor): the inputs are first given to the preprocessors then
to the model.
postprocessors (None, Processor, list of Processor): the predicted outputs by the model are given to the
processors before being returned.
"""
# create inner model
num_inputs, num_outputs = self._size(inputs), self._size(outputs)
model = GPR(mean=mean, kernel=kernel, model=model, likelihood=likelihood)
# call parent class
super(GPRApproximator, self).__init__(inputs, outputs, model=model, preprocessors=preprocessors,
postprocessors=postprocessors)
pyrobolearn/approximators/kmp.py
+58
View File
@@ -0,0 +1,58 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Define Kernel Movement Primitive (KMP) function approximator.
Dependencies:
- `pyrobolearn.models`
- `pyrobolearn.states`
- `pyrobolearn.actions`
"""
from pyrobolearn.approximators.approximator import Approximator
from pyrobolearn.models.kmp import KMP
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class KMPApproximator(Approximator):
r"""Kernel Movement Primitive Approximator
"""
def __init__(self, inputs, outputs, num_components=1, priors=None, means=None, covariances=None, gaussians=None,
preprocessors=None, postprocessors=None):
"""
Initialize the Kernel Movement Primitive approximator.
Args:
inputs (State, Action, np.array, torch.Tensor): inputs of the inner models (instance of Action/State)
outputs (State, Action, np.array, torch.Tensor): outputs of the inner models (instance of Action/State)
num_components (int): the number of components/gaussians (this argument should be provided if
no priors, means, covariances, or gaussians are provided)
priors (list/tuple of float, None): prior probabilities (they have to be positives). If not provided,
it will be a uniform distribution.
means (list of np.array[float[D]], None): list of means
covariances (list of np.array[float[D,D]], None): list of covariances
gaussians (list of Gaussian, None): list of gaussians. If provided, the `means` and `covariances`
parameters don't have to be provided.
preprocessors (None, Processor, list of Processor): the inputs are first given to the preprocessors then
to the model.
postprocessors (None, Processor, list of Processor): the predicted outputs by the model are given to the
processors before being returned.
"""
# create inner model
num_inputs, num_outputs = self._size(inputs), self._size(outputs)
dimensionality = num_inputs + num_outputs
model = KMP()
# call parent class
super(KMPApproximator, self).__init__(inputs, outputs, model=model, preprocessors=preprocessors,
postprocessors=postprocessors)
pyrobolearn/approximators/linear.py
+49
View File
@@ -0,0 +1,49 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Define linear function approximator.
Dependencies:
- `pyrobolearn.models`
- `pyrobolearn.states`
- `pyrobolearn.actions`
"""
from pyrobolearn.approximators.approximator import Approximator
from pyrobolearn.models.basics.linear import Linear
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class LinearApproximator(Approximator):
r"""Linear Function Approximator
The linear function approximator is a linear parametric model: :math:`y = W x + b` where :math:`x` and :math:`y`
are respectively the input and output vectors, :math:`W` is the weight matrix, and :math:`b` is the bias/intercept.
"""
def __init__(self, inputs, outputs, preprocessors=None, postprocessors=None):
"""
Initialize the linear approximator.
Args:
inputs (State, Action, np.array, torch.Tensor): inputs of the inner models (instance of Action/State)
outputs (State, Action, np.array, torch.Tensor): outputs of the inner models (instance of Action/State)
preprocessors (None, Processor, list of Processor): the inputs are first given to the preprocessors then
to the model.
postprocessors (None, Processor, list of Processor): the predicted outputs by the model are given to the
processors before being returned.
"""
# create inner model
model = Linear(num_inputs=self._size(inputs), num_outputs=self._size(outputs), add_bias=True)
# call parent class
super(LinearApproximator, self).__init__(inputs, outputs, model=model, preprocessors=preprocessors,
postprocessors=postprocessors)
pyrobolearn/approximators/nn_approximator.py → pyrobolearn/approximators/nn.py
View File
pyrobolearn/dynamics/README.md
-5
View File
@@ -1,5 +0,0 @@
## Dynamic models
Dynamic models are models that given the current state (or a history of states), and the current action compute the next state. If dynamic models are provided or trained, we are often in a model-based reinforcement learning (also known as optimal control) paradigm.
Dynamic models will be soon added to the framework.
pyrobolearn/dynamics/README.rst
+9
View File
@@ -0,0 +1,9 @@
Dynamic models
==============
Dynamic models are models that given the current state (or a history of states), and the current action compute the
next state. If dynamic models are provided or trained, we are often in a model-based reinforcement learning (also
known as optimal control) setting.
Warnings: Dynamic models have not been tested yet in the current framework. They will be added shortly to the
framework.
pyrobolearn/dynamics/__init__.py
+2 -2
View File
@@ -4,10 +4,10 @@
from .dynamic import *
# import basic dynamic models (such as linear dynamic models)
from .basic_dynamic import *
from .basic import *
# import robot dynamic models
from .robot_dynamic import *
# import neural network dynamic models
from .nn_dynamic import *
from .nn import *
pyrobolearn/dynamics/basic_dynamic.py → pyrobolearn/dynamics/basic.py
View File
pyrobolearn/dynamics/dynamic.py
+1 -1
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Provides the `transition`/`dynamic` function approximators in RL.
r"""Provides the `transition`/`dynamic` function approximators in RL.
Dynamic models allows to compute the next state given the current state and action; that is,
:math:`s_{t+1} = f(s_t, a_t)` (if deterministic) or :math:`s_{t+1} \sim p(.| s_t, a_t)`.
pyrobolearn/dynamics/gmr.py
+46
View File
@@ -0,0 +1,46 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Provides the GMR dynamic transition function approximators
The GMR dynamic model predicts using a GMR model the next state given the current state and action.
"""
from pyrobolearn.approximators.gmr import GMRApproximator
from pyrobolearn.dynamics.dynamic import ParametrizedDynamicModel
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class GMRDynamicModel(ParametrizedDynamicModel):
r"""GMR Dynamic Model
The GMR dynamic model predicts using a GMR model the next state given the current state and action.
"""
def __init__(self, state, action, next_state=None, distributions=None, preprocessors=None, postprocessors=None):
"""
Initialize the GMR dynamic transition function / probability :math:`p(s_{t+1} | s_t, a_t)`.
Args:
state (State): state inputs.
action (Action): action inputs.
next_state (State, None): state outputs. If None, it will take the state inputs as the outputs.
distributions (torch.distributions.Distribution): distribution to use to sample the next state. If None,
it will be deterministic.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
"""
if next_state is None:
next_state = state
model = GMRApproximator(inputs=[state, action], outputs=next_state, preprocessors=preprocessors,
postprocessors=postprocessors)
super(GMRDynamicModel, self).__init__(state, action, model=model, next_state=next_state,
distributions=distributions)
pyrobolearn/dynamics/gp.py
+46
View File
@@ -0,0 +1,46 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Provides the GPR dynamic transition function approximators
The GPR dynamic model predicts using a GPR model the next state given the current state and action.
"""
from pyrobolearn.approximators.gp import GPRApproximator
from pyrobolearn.dynamics.dynamic import ParametrizedDynamicModel
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class GPRDynamicModel(ParametrizedDynamicModel):
r"""GPR Dynamic Model
The GPR dynamic model predicts using a GPR model the next state given the current state and action.
"""
def __init__(self, state, action, next_state=None, distributions=None, preprocessors=None, postprocessors=None):
"""
Initialize the GPR dynamic transition function / probability :math:`p(s_{t+1} | s_t, a_t)`.
Args:
state (State): state inputs.
action (Action): action inputs.
next_state (State, None): state outputs. If None, it will take the state inputs as the outputs.
distributions (torch.distributions.Distribution): distribution to use to sample the next state. If None,
it will be deterministic.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
"""
if next_state is None:
next_state = state
model = GPRApproximator(inputs=[state, action], outputs=next_state, preprocessors=preprocessors,
postprocessors=postprocessors)
super(GPRDynamicModel, self).__init__(state, action, model=model, next_state=next_state,
distributions=distributions)
pyrobolearn/dynamics/linear.py
+49
View File
@@ -0,0 +1,49 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Provides the linear dynamic transition function approximators
The linear dynamic model predicts using a linear model the next state given the current state and action.
"""
from pyrobolearn.approximators import LinearApproximator
from pyrobolearn.dynamics.dynamic import ParametrizedDynamicModel
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class LinearDynamicModel(ParametrizedDynamicModel):
r"""Linear Dynamic Model
The linear dynamic model predicts using a linear model the next state given the current state and action.
Pros: easy to implement and learn
Cons: very limited
"""
def __init__(self, state, action, next_state=None, distributions=None, preprocessors=None, postprocessors=None):
"""
Initialize the linear dynamic transition function / probability :math:`p(s_{t+1} | s_t, a_t)`.
Args:
state (State): state inputs.
action (Action): action inputs.
next_state (State, None): state outputs. If None, it will take the state inputs as the outputs.
distributions (torch.distributions.Distribution): distribution to use to sample the next state. If None,
it will be deterministic.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
"""
if next_state is None:
next_state = state
model = LinearApproximator(inputs=[state, action], outputs=next_state, preprocessors=preprocessors,
postprocessors=postprocessors)
super(LinearDynamicModel, self).__init__(state, action, model=model, next_state=next_state,
distributions=distributions)
pyrobolearn/dynamics/nn_dynamic.py → pyrobolearn/dynamics/nn.py
View File
pyrobolearn/dynamics/polynomial.py
+1 -1
View File
@@ -10,7 +10,7 @@ from pyrobolearn.dynamics.dynamic import ParametrizedDynamicModel
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
pyrobolearn/dynamics/robot_dynamic.py
+3 -1
View File
@@ -3,12 +3,14 @@
"""Provides robot dynamic transition functions
"""
# TODO: to implement
from pyrobolearn.robots.robot import Robot
from pyrobolearn.dynamics.dynamic import DynamicModel
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
pyrobolearn/policies/README.md
+6 -2
View File
@@ -1,5 +1,9 @@
## Policies
Policies in this framework are controllers that can be trained and map states to actions. They use directly the learning model or the `Approximator` class (which uses the learning model).
Policies in this framework are controllers that can be trained and map states to actions. They use directly the
learning model or the `Approximator` class (which uses the learning model).
In this framework, `State` and `Action` instances should be given to the `Policy` which would infer its input and output dimensions and build the model with the correct number of inputs/outputs. In contrast to the learning model, the policy should know how to feed the various input states to the inner learning model, such that if a picture and joint states are given to the policy it knows where to feed the corresponding input observations.
In this framework, `State` and `Action` instances should be given to the `Policy` which would infer its input and
output dimensions and build the model with the correct number of inputs/outputs. In contrast to the learning model,
the policy should know how to feed the various input states to the inner learning model, such that if a picture and
joint states are given to the policy it knows where to feed the corresponding input observations.
pyrobolearn/policies/__init__.py
+5 -5
View File
@@ -4,16 +4,16 @@
from .policy import Policy
# import basic policies
from .basic_policy import *
from .basic import *
# import nn policies
from .nn_policy import *
from .nn import *
# import dmp policies
from .dmp_policy import *
from .dmp import *
# import cpg policies
from .cpg_policy import *
from .cpg import *
# import neat policies
from .neat_policy import *
from .neat import *
pyrobolearn/policies/basic_policy.py → pyrobolearn/policies/basic.py
View File
pyrobolearn/policies/cpg_policy.py → pyrobolearn/policies/cpg.py
View File
pyrobolearn/policies/dmp_policy.py → pyrobolearn/policies/dmp.py
+65 -28
View File
@@ -34,25 +34,25 @@ class DMPPolicy(Policy):
Args:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
model (DMP): DMP model
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
*args (list): list of arguments (this is not used in this class).
**kwargs (dict): dictionary of arguments (this is not used in this class).
"""
if not isinstance(model, DMP):
raise TypeError("Expecting model to be an instance of DMP")
raise TypeError("Expecting model to be an instance of `DMP`, but got instead: {}".format(type(model)))
super(DMPPolicy, self).__init__(state=state, action=action, model=model, rate=rate,
preprocessors=preprocessors, postprocessors=postprocessors, *args, **kwargs)
@@ -68,10 +68,12 @@ class DMPPolicy(Policy):
Args:
state ((list of) torch.Tensor, (list of) np.array): state data.
deterministic (bool): if True, it will predict in a deterministic way. Setting it to False, only works
with stochastic models.
to_numpy (bool): If True, it will convert the data (torch.Tensors) to numpy arrays.
return_logits (bool): If True, in the case of discrete outputs, it will return the logits.
set_output_data (bool): If True, it will set the predicted output data to the outputs given to the
approximator.
approximator.
Returns:
(list of) torch.Tensor, (list of) np.array: predicted action data.
@@ -116,14 +118,22 @@ class DMPPolicy(Policy):
if len(data) > 0:
# print("Imitating with :", data.shape)
# y, dy, ddy = data
y = data
# y = data
# if len(y.shape) == 1:
# y = y.reshape(1, -1)
# if len(dy.shape) == 1:
# dy = dy.reshape(1, -1)
# if len(ddy.shape) == 1:
# ddy = ddy.reshape(1, -1)
self.model.imitate(y, plot=False) # dy, ddy, plot=True) # dy, ddy)
if self.is_joint_position_action:
if self.is_joint_velocity_action:
self.model.imitate(y, dy, plot=False)
else:
y = data
self.model.imitate(y, plot=False) # dy, ddy, plot=True) # dy, ddy)
else:
raise NotImplementedError
else:
print("Nothing to imitate.")
@@ -147,7 +157,7 @@ class DiscreteDMPPolicy(DMPPolicy):
See Also: see documentation in `pyrobolearn.models.dmp.discrete_dmp.py`
References:
[1] "Dynamical movement primitives: Learning attractor models for motor behaviors", Ijspeert et al., 2013
- [1] "Dynamical movement primitives: Learning attractor models for motor behaviors", Ijspeert et al., 2013
"""
def __init__(self, action, state=None, num_basis=20, dt=0.01, y0=0, goal=1, forcing_terms=None,
@@ -156,8 +166,15 @@ class DiscreteDMPPolicy(DMPPolicy):
Initialize the discrete DMP policy.
Args:
action:
state:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
num_basis (int): number of basis functions
dt (float): step integration for Euler's method
y0 (float, np.array): initial position(s)
@@ -165,7 +182,9 @@ class DiscreteDMPPolicy(DMPPolicy):
forcing_terms (list, ForcingTerm): the forcing terms (which can have different basis functions)
stiffness (float): stiffness coefficient
damping (float): damping coefficient
rate:
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
"""
if not isinstance(action, Action):
raise TypeError("Expecting actions to be an instance of the 'Action' class.")
@@ -180,7 +199,7 @@ class RhythmicDMPPolicy(DMPPolicy):
See Also: see documentation in `pyrobolearn.models.dmp.rhythmic_dmp.py`
References:
[1] "Dynamical movement primitives: Learning attractor models for motor behaviors", Ijspeert et al., 2013
- [1] "Dynamical movement primitives: Learning attractor models for motor behaviors", Ijspeert et al., 2013
"""
def __init__(self, action, state=None, num_basis=20, dt=0.01, y0=0, goal=1, forcing_terms=None,
@@ -189,8 +208,15 @@ class RhythmicDMPPolicy(DMPPolicy):
Initialize the Rhythmic DMP policy.
Args:
action:
state:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
num_basis (int): number of basis functions
dt (float): step integration for Euler's method
y0 (float, np.array): initial position(s)
@@ -198,7 +224,9 @@ class RhythmicDMPPolicy(DMPPolicy):
forcing_terms (list, ForcingTerm): the forcing terms (which can have different basis functions)
stiffness (float): stiffness coefficient
damping (float): damping coefficient
rate:
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
"""
model = RhythmicDMP(num_dmps=self._size(action), num_basis=num_basis, dt=dt, y0=y0, goal=goal,
forcing_terms=forcing_terms, stiffness=stiffness, damping=damping)
@@ -211,10 +239,10 @@ class BioDiscreteDMPPolicy(DMPPolicy):
See Also: see documentation in `pyrobolearn.models.dmp.biodiscrete_dmp.py`
References:
[1] "Dynamical movement primitives: Learning attractor models for motor behaviors", Ijspeert et al., 2013
[2] "Biologically-inspired Dynamical Systems for Movement Generation: Automatic Real-time Goal Adaptation
and Obstacle Avoidance", Hoffmann et al., 2009
[3] "Learning and Generalization of Motor Skills by Learning from Demonstration", Pastor et al., 2009
- [1] "Dynamical movement primitives: Learning attractor models for motor behaviors", Ijspeert et al., 2013
- [2] "Biologically-inspired Dynamical Systems for Movement Generation: Automatic Real-time Goal Adaptation
and Obstacle Avoidance", Hoffmann et al., 2009
- [3] "Learning and Generalization of Motor Skills by Learning from Demonstration", Pastor et al., 2009
"""
def __init__(self, action, state=None, num_basis=20, dt=0.01, y0=0, goal=1, forcing_terms=None,
@@ -223,8 +251,15 @@ class BioDiscreteDMPPolicy(DMPPolicy):
Initialize the biologically-inspired DMP policy.
Args:
action:
state:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
num_basis (int): number of basis functions
dt (float): step integration for Euler's method
y0 (float, np.array): initial position(s)
@@ -232,7 +267,9 @@ class BioDiscreteDMPPolicy(DMPPolicy):
forcing_terms (list, ForcingTerm): the forcing terms (which can have different basis functions)
stiffness (float): stiffness coefficient
damping (float): damping coefficient
rate:
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
"""
model = BioDiscreteDMP(num_dmps=self._size(action), num_basis=num_basis, dt=dt, y0=y0, goal=goal,
forcing_terms=forcing_terms, stiffness=stiffness, damping=damping)
pyrobolearn/policies/gmr.py
+128
View File
@@ -0,0 +1,128 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Define the Gaussian Mixture Regression (GMR) Policy.
Define the GMR policy that can be used.
"""
import numpy as np
import torch
from pyrobolearn.approximators.gmr import GMRApproximator
from pyrobolearn.policies.policy import Policy
from pyrobolearn.states import State
from pyrobolearn.actions import Action
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class GMRPolicy(Policy):
r"""Gaussian Mixture Regression (GMR) policy
"""
def __init__(self, state, action, num_components=1, priors=None, means=None, covariances=None,
gaussians=None, rate=1, preprocessors=None, postprocessors=None, *args, **kwargs):
"""
Initialize the GMR policy.
Args:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
num_components (int): the number of components/gaussians (this argument should be provided if
no priors, means, covariances, or gaussians are provided)
priors (list/tuple of float, None): prior probabilities (they have to be positives). If not provided,
it will be a uniform distribution.
means (list of np.array[float[D]], None): list of means
covariances (list of np.array[float[D,D]], None): list of covariances
gaussians (list of Gaussian, None): list of gaussians. If provided, the `means` and `covariances`
parameters don't have to be provided.
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
*args (list): list of arguments (this is not used in this class).
**kwargs (dict): dictionary of arguments (this is not used in this class).
"""
model = GMRApproximator(inputs=state, outputs=action, num_components=num_components, priors=priors,
means=means, covariances=covariances, gaussians=gaussians, preprocessors=preprocessors,
postprocessors=postprocessors)
super(GMRPolicy, self).__init__(state, action, model, rate=rate, *args, **kwargs)
def inner_predict(self, state, deterministic=True, to_numpy=False, return_logits=True, set_output_data=False):
"""Inner prediction step.
Args:
state ((list of) torch.Tensor, (list of) np.array): state data.
deterministic (bool): if True, it will predict in a deterministic way. Setting it to False, only works
with stochastic models.
to_numpy (bool): If True, it will convert the data (torch.Tensors) to numpy arrays.
return_logits (bool): If True, in the case of discrete outputs, it will return the logits.
set_output_data (bool): If True, it will set the predicted output data to the outputs given to the
approximator.
Returns:
(list of) torch.Tensor, (list of) np.array: predicted action data.
"""
if isinstance(state, (np.ndarray, list, tuple)):
state = state[0]
y = self.model.condition(x_in=state, idx_out=None) # TODO
return y
# def act(self, state=None, deterministic=True, to_numpy=True, return_logits=False, apply_action=True):
# # return self.model.predict(state, to_numpy=to_numpy)
# if (self.cnt % self.rate) == 0:
# # print("Policy state value: {}".format(state.data[0][0]))
# self.y, self.dy, self.ddy = self.model.step(state.data[0][0])
# self.cnt += 1
# # y, dy, ddy = self.model.step()
# # return np.array([y, dy, ddy])
# if isinstance(self.actions, JointPositionAction):
# # print("DMP action: {}".format(self.y))
# self.actions.data = self.y
# elif isinstance(self.actions, JointVelocityAction):
# self.actions.data = self.dy
# elif isinstance(self.actions, JointAccelerationAction):
# self.actions.data = self.ddy
# return self.actions
# def sample(self, state):
# pass
def rollout(self):
"""Perform a rollout with the movement primitive."""
return self.model.rollout()
def imitate(self, data): # TODO: improve this
if len(data) > 0:
raise NotImplementedError
else:
print("Nothing to imitate.")
def plot_rollout(self, nrows=1, ncols=1, suptitle=None, titles=None, show=True):
"""
Plot the rollouts using the DMPs.
Args:
nrows (int): number of rows in the subplot.
ncols (int): number of columns in the subplot.
suptitle (str): main title for the subplots.
titles (str, list[str]): title for each subplot.
show (bool): if True, it will show and block the plot.
"""
self.model.plot_rollout(nrows=nrows, ncols=ncols, suptitle=suptitle, titles=titles, show=show)
pyrobolearn/policies/gp.py
+127
View File
@@ -0,0 +1,127 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Define the Gaussian Process Regression (GPR) Policy.
Define the GPR policy that can be used.
"""
import numpy as np
import torch
from pyrobolearn.approximators.gp import GPRApproximator
from pyrobolearn.policies.policy import Policy
from pyrobolearn.states import State
from pyrobolearn.actions import Action
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class GPRPolicy(Policy):
r"""Gaussian Process Regression (GPR) policy
"""
def __init__(self, state, action, mean=None, kernel=None, model=None, likelihood=None, rate=1, preprocessors=None,
postprocessors=None, *args, **kwargs):
"""
Initialize the GPR policy.
Args:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
mean (None, gpytorch.means.Mean): mean prior. If None, it will be set to `gpytorch.means.ConstantMean()`.
kernel (None, gpytorch.kernels.Kernel): kernel prior. If None it will be set to
`gpytorch.kernels.ScaleKernel(gpytorch.kernels.RBFKernel() + gpytorch.kernels.WhiteNoiseKernel())`
model (None, gpytorch.module.Module): the prior GP model. If None, it will create `ExactGPModel()`, a GP
model using the provided mean, kernel, and likelihood.
likelihood (None, gpytorch.likelihoods.Likelihood): the likelihood pdf. If None, it will use the
`gpytorch.likelihoods.GaussianLikelihood()`.
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
*args (list): list of arguments (this is not used in this class).
**kwargs (dict): dictionary of arguments (this is not used in this class).
"""
model = GPRApproximator(inputs=state, outputs=action, mean=mean, kernel=kernel, model=model,
likelihood=likelihood, preprocessors=preprocessors,
postprocessors=postprocessors)
super(GPRPolicy, self).__init__(state, action, model, rate=rate, *args, **kwargs)
def inner_predict(self, state, deterministic=True, to_numpy=False, return_logits=True, set_output_data=False):
"""Inner prediction step.
Args:
state ((list of) torch.Tensor, (list of) np.array): state data.
deterministic (bool): if True, it will predict in a deterministic way. Setting it to False, only works
with stochastic models.
to_numpy (bool): If True, it will convert the data (torch.Tensors) to numpy arrays.
return_logits (bool): If True, in the case of discrete outputs, it will return the logits.
set_output_data (bool): If True, it will set the predicted output data to the outputs given to the
approximator.
Returns:
(list of) torch.Tensor, (list of) np.array: predicted action data.
"""
if isinstance(state, (np.ndarray, list, tuple)):
state = state[0]
y = self.model.condition(x_in=state, idx_out=None) # TODO
return y
# def act(self, state=None, deterministic=True, to_numpy=True, return_logits=False, apply_action=True):
# # return self.model.predict(state, to_numpy=to_numpy)
# if (self.cnt % self.rate) == 0:
# # print("Policy state value: {}".format(state.data[0][0]))
# self.y, self.dy, self.ddy = self.model.step(state.data[0][0])
# self.cnt += 1
# # y, dy, ddy = self.model.step()
# # return np.array([y, dy, ddy])
# if isinstance(self.actions, JointPositionAction):
# # print("DMP action: {}".format(self.y))
# self.actions.data = self.y
# elif isinstance(self.actions, JointVelocityAction):
# self.actions.data = self.dy
# elif isinstance(self.actions, JointAccelerationAction):
# self.actions.data = self.ddy
# return self.actions
# def sample(self, state):
# pass
def rollout(self):
"""Perform a rollout with the movement primitive."""
return self.model.rollout()
def imitate(self, data): # TODO: improve this
if len(data) > 0:
raise NotImplementedError
else:
print("Nothing to imitate.")
def plot_rollout(self, nrows=1, ncols=1, suptitle=None, titles=None, show=True):
"""
Plot the rollouts using the DMPs.
Args:
nrows (int): number of rows in the subplot.
ncols (int): number of columns in the subplot.
suptitle (str): main title for the subplots.
titles (str, list[str]): title for each subplot.
show (bool): if True, it will show and block the plot.
"""
self.model.plot_rollout(nrows=nrows, ncols=ncols, suptitle=suptitle, titles=titles, show=show)
pyrobolearn/policies/kmp.py
+128
View File
@@ -0,0 +1,128 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Define the Kernel Movement Primitive (KMP) Policy.
Define the KMP policy that can be used.
"""
import numpy as np
import torch
from pyrobolearn.approximators.kmp import KMPApproximator
from pyrobolearn.policies.policy import Policy
from pyrobolearn.states import State
from pyrobolearn.actions import Action
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class KMPPolicy(Policy):
r"""Kernel Movement Primitive (KMP) policy
"""
def __init__(self, state, action, num_components=1, priors=None, means=None, covariances=None,
gaussians=None, rate=1, preprocessors=None, postprocessors=None, *args, **kwargs):
"""
Initialize the KMP policy.
Args:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
num_components (int): the number of components/gaussians (this argument should be provided if
no priors, means, covariances, or gaussians are provided)
priors (list/tuple of float, None): prior probabilities (they have to be positives). If not provided,
it will be a uniform distribution.
means (list of np.array[float[D]], None): list of means
covariances (list of np.array[float[D,D]], None): list of covariances
gaussians (list of Gaussian, None): list of gaussians. If provided, the `means` and `covariances`
parameters don't have to be provided.
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
*args (list): list of arguments (this is not used in this class).
**kwargs (dict): dictionary of arguments (this is not used in this class).
"""
model = KMPApproximator(inputs=state, outputs=action, num_components=num_components, priors=priors,
means=means, covariances=covariances, gaussians=gaussians, preprocessors=preprocessors,
postprocessors=postprocessors)
super(KMPPolicy, self).__init__(state, action, model, rate=rate, *args, **kwargs)
def inner_predict(self, state, deterministic=True, to_numpy=False, return_logits=True, set_output_data=False):
"""Inner prediction step.
Args:
state ((list of) torch.Tensor, (list of) np.array): state data.
deterministic (bool): if True, it will predict in a deterministic way. Setting it to False, only works
with stochastic models.
to_numpy (bool): If True, it will convert the data (torch.Tensors) to numpy arrays.
return_logits (bool): If True, in the case of discrete outputs, it will return the logits.
set_output_data (bool): If True, it will set the predicted output data to the outputs given to the
approximator.
Returns:
(list of) torch.Tensor, (list of) np.array: predicted action data.
"""
if isinstance(state, (np.ndarray, list, tuple)):
state = state[0]
y = self.model.condition(x_in=state, idx_out=None) # TODO
return y
# def act(self, state=None, deterministic=True, to_numpy=True, return_logits=False, apply_action=True):
# # return self.model.predict(state, to_numpy=to_numpy)
# if (self.cnt % self.rate) == 0:
# # print("Policy state value: {}".format(state.data[0][0]))
# self.y, self.dy, self.ddy = self.model.step(state.data[0][0])
# self.cnt += 1
# # y, dy, ddy = self.model.step()
# # return np.array([y, dy, ddy])
# if isinstance(self.actions, JointPositionAction):
# # print("DMP action: {}".format(self.y))
# self.actions.data = self.y
# elif isinstance(self.actions, JointVelocityAction):
# self.actions.data = self.dy
# elif isinstance(self.actions, JointAccelerationAction):
# self.actions.data = self.ddy
# return self.actions
# def sample(self, state):
# pass
def rollout(self):
"""Perform a rollout with the movement primitive."""
return self.model.rollout()
def imitate(self, data): # TODO: improve this
if len(data) > 0:
raise NotImplementedError
else:
print("Nothing to imitate.")
def plot_rollout(self, nrows=1, ncols=1, suptitle=None, titles=None, show=True):
"""
Plot the rollouts using the DMPs.
Args:
nrows (int): number of rows in the subplot.
ncols (int): number of columns in the subplot.
suptitle (str): main title for the subplots.
titles (str, list[str]): title for each subplot.
show (bool): if True, it will show and block the plot.
"""
self.model.plot_rollout(nrows=nrows, ncols=ncols, suptitle=suptitle, titles=titles, show=show)
pyrobolearn/policies/linear.py
+66
View File
@@ -0,0 +1,66 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Provide the linear policy.
The linear policy uses a linear parametric approximator to predict the action vector based on the state vector.
"""
from pyrobolearn.policies.policy import Policy
from pyrobolearn.approximators import LinearApproximator
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class LinearPolicy(Policy):
r"""Linear Policy
The linear policy uses a linear parametric approximator: :math:`y = W x + b` where :math:`x` is the state vector,
and :math:`y` is the action vector, :math:`W` is the weight matrix, and :math:`b` is the bias/intercept.
"""
def __init__(self, state, action, rate=1, preprocessors=None, postprocessors=None, *args, **kwargs):
"""
Initialize the Linear Policy.
Args:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
*args (list): list of arguments
**kwargs (dict): dictionary of arguments
"""
model = LinearApproximator(state, action, preprocessors=preprocessors, postprocessors=postprocessors)
super(LinearPolicy, self).__init__(state, action, model, rate=rate, *args, **kwargs)
# Tests
if __name__ == '__main__':
import copy
from pyrobolearn.states import FixedState
from pyrobolearn.actions import FixedAction
# check linear policy
policy = LinearPolicy(state=FixedState(range(4)), action=FixedAction(range(2)))
print(policy)
target = copy.deepcopy(policy)
print(target)
pyrobolearn/policies/neat_policy.py → pyrobolearn/policies/neat.py
View File
pyrobolearn/policies/nn_policy.py → pyrobolearn/policies/nn.py
View File
pyrobolearn/policies/policy.py
+5 -9
View File
@@ -531,21 +531,17 @@ class Policy(object):
# if not return_logits:
# action_data[idx] = discrete_data
else:
raise TypeError(
"Expecting the `data` action to be an int, numpy array, torch.Tensor, instead got: "
"{}".format(type(data)))
raise TypeError("Expecting the `data` action to be an int, numpy array, torch.Tensor, instead "
"got: {}".format(type(data)))
else: # continuous action
if isinstance(data, np.ndarray):
if isinstance(data, (np.ndarray, float, int)):
action.data = data
elif isinstance(data, torch.Tensor):
action.torch_data = data
action_data[idx] = self.__convert_to_numpy(data, to_numpy=to_numpy)
elif isinstance(data, (float, int)):
action.data = data
else:
raise TypeError(
"Expecting `data` to be a numpy array or torch.Tensor, instead got: "
"{}".format(type(data)))
raise TypeError("Expecting `data` to be a numpy array or torch.Tensor, instead got: "
"{}".format(type(data)))
# if action_data is a list and has one element, return just that element
if isinstance(action_data, list) and len(action_data) == 1:
pyrobolearn/policies/promp.py
+224
View File
@@ -0,0 +1,224 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Define the Probabilistic Movement Primitive (ProMP) Policy.
Define the various ProMP policies that can be used.
"""
import numpy as np
import torch
from pyrobolearn.models import ProMP, DiscreteProMP, RhythmicProMP
from pyrobolearn.policies.policy import Policy
from pyrobolearn.states import State
from pyrobolearn.actions import Action, JointPositionAction, JointVelocityAction, JointPositionAndVelocityAction
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class ProMPPolicy(Policy):
r"""Probabilistic Movement Primitive (ProMP) policy
"""
def __init__(self, state, action, model, rate=1, preprocessors=None, postprocessors=None, *args, **kwargs):
"""
Initialize the ProMP policy.
Args:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
model (ProMP): ProMP model
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
preprocessors (Processor, list of Processor, None): pre-processors to be applied to the given input
postprocessors (Processor, list of Processor, None): post-processors to be applied to the output
*args (list): list of arguments (this is not used in this class).
**kwargs (dict): dictionary of arguments (this is not used in this class).
"""
if not isinstance(model, ProMP):
raise TypeError("Expecting model to be an instance of `ProMP`, but got instead: {}".format(type(model)))
super(ProMPPolicy, self).__init__(state=state, action=action, model=model, rate=rate,
preprocessors=preprocessors, postprocessors=postprocessors, *args, **kwargs)
# check actions
self.is_joint_position_action = JointPositionAction in action or JointPositionAndVelocityAction in action
self.is_joint_velocity_action = JointVelocityAction in action or JointPositionAndVelocityAction in action
if not (self.is_joint_position_action or self.is_joint_velocity_action):
raise ValueError("The actions do not have a joint position or velocity action.")
def inner_predict(self, state, deterministic=True, to_numpy=False, return_logits=True, set_output_data=False):
"""Inner prediction step.
Args:
state ((list of) torch.Tensor, (list of) np.array): state data.
deterministic (bool): if True, it will predict in a deterministic way. Setting it to False, only works
with stochastic models.
to_numpy (bool): If True, it will convert the data (torch.Tensors) to numpy arrays.
return_logits (bool): If True, in the case of discrete outputs, it will return the logits.
set_output_data (bool): If True, it will set the predicted output data to the outputs given to the
approximator.
Returns:
(list of) torch.Tensor, (list of) np.array: predicted action data.
"""
if isinstance(state, (np.ndarray, list, tuple)):
state = state[0]
y, dy, ddy = self.model.step(state)
if self.is_joint_position_action:
if self.is_joint_velocity_action:
return np.concatenate((y, dy))
return y
elif self.is_joint_velocity_action:
return dy
else: # self.is_joint_acceleration_action
return ddy
# def act(self, state=None, deterministic=True, to_numpy=True, return_logits=False, apply_action=True):
# # return self.model.predict(state, to_numpy=to_numpy)
# if (self.cnt % self.rate) == 0:
# # print("Policy state value: {}".format(state.data[0][0]))
# self.y, self.dy, self.ddy = self.model.step(state.data[0][0])
# self.cnt += 1
# # y, dy, ddy = self.model.step()
# # return np.array([y, dy, ddy])
# if isinstance(self.actions, JointPositionAction):
# # print("ProMP action: {}".format(self.y))
# self.actions.data = self.y
# elif isinstance(self.actions, JointVelocityAction):
# self.actions.data = self.dy
# elif isinstance(self.actions, JointAccelerationAction):
# self.actions.data = self.ddy
# return self.actions
# def sample(self, state):
# pass
def rollout(self):
"""Perform a rollout with the movement primitive."""
return self.model.rollout()
def imitate(self, data): # TODO: improve this
if len(data) > 0:
# print("Imitating with :", data.shape)
# y, dy, ddy = data
# y = data
# if len(y.shape) == 1:
# y = y.reshape(1, -1)
# if len(dy.shape) == 1:
# dy = dy.reshape(1, -1)
# if len(ddy.shape) == 1:
# ddy = ddy.reshape(1, -1)
if self.is_joint_position_action:
if self.is_joint_velocity_action:
self.model.imitate(y, dy, plot=False)
else:
y = data
self.model.imitate(y, plot=False) # dy, ddy, plot=True) # dy, ddy)
else:
raise NotImplementedError
else:
print("Nothing to imitate.")
def plot_rollout(self, nrows=1, ncols=1, suptitle=None, titles=None, show=True):
"""
Plot the rollouts using the ProMPs.
Args:
nrows (int): number of rows in the subplot.
ncols (int): number of columns in the subplot.
suptitle (str): main title for the subplots.
titles (str, list[str]): title for each subplot.
show (bool): if True, it will show and block the plot.
"""
self.model.plot_rollout(nrows=nrows, ncols=ncols, suptitle=suptitle, titles=titles, show=show)
class DiscreteProMPPolicy(ProMPPolicy):
r"""Discrete ProMP Policy
See Also: see documentation in `pyrobolearn.models.promp.discrete_promp.py`
References:
- [1] "Probabilistic Movement Primitives", Paraschos et al., 2013
- [2] "Using Probabilistic Movement Primitives in Robotics", Paraschos et al., 2018
"""
def __init__(self, action, state=None, num_basis=20, weights=None, canonical_system=None, noise_covariance=1.,
basis_width=None, rate=1):
"""
Initialize the discrete ProMP policy.
Args:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
num_basis (int): number of basis functions
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
"""
if not isinstance(action, Action):
raise TypeError("Expecting actions to be an instance of the 'Action' class.")
model = DiscreteProMP(num_dofs=self._size(action), num_basis=num_basis, weights=weights,
canonical_system=canonical_system, noise_covariance=noise_covariance,
basis_width=basis_width)
super(DiscreteProMPPolicy, self).__init__(state, action, model, rate=rate)
class RhythmicProMPPolicy(ProMPPolicy):
r"""Rhythmic ProMP Policy
See Also: see documentation in `pyrobolearn.models.promp.rhythmic_promp.py`
References:
- [1] "Probabilistic Movement Primitives", Paraschos et al., 2013
- [2] "Using Probabilistic Movement Primitives in Robotics", Paraschos et al., 2018
"""
def __init__(self, action, state=None, num_basis=20, weights=None, canonical_system=None, noise_covariance=1.,
basis_width=None, rate=1):
"""
Initialize the Rhythmic ProMP policy.
Args:
action (Action): At each step, by calling `policy.act(state)`, the `action` is computed by the policy,
and can be given to the environment. As with the `state`, the type and size/shape of each inner
action can be inferred and could be used to automatically build a policy. The `action` connects the
policy with a controllable object (such as a robot) in the environment.
state (State): By giving the `state` to the policy, it can automatically infer the type and size/shape
of each inner state, and thus can be used to automatically build a policy. At each step, the `state`
is filled by the environment, and read by the policy. The `state` connects the policy with one or
several objects (including robots) in the environment. Note that some policies don't use any state
information.
num_basis (int): number of basis functions
rate (int, float): rate (float) at which the policy operates if we are operating in real-time. If we are
stepping deterministically in the simulator, it represents the number of ticks (int) to sleep before
executing the model.
"""
model = RhythmicProMP(num_dofs=self._size(action), num_basis=num_basis, weights=weights,
canonical_system=canonical_system, noise_covariance=noise_covariance,
basis_width=basis_width)
super(RhythmicProMPPolicy, self).__init__(state, action, model, rate=rate)
pyrobolearn/values/__init__.py
+2 -2
View File
@@ -4,7 +4,7 @@
from .value import *
# import basic value function approximator (such as tables and linear)
from .basic_value import *
from .basic import *
# import NN value function approximators
from .nn_value import *
from .nn import *
pyrobolearn/values/basic_value.py → pyrobolearn/values/basic.py
View File
pyrobolearn/values/gmr.py
+145
View File
@@ -0,0 +1,145 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Define the Gaussian Mixture Regression (GMR) value function approximators.
Define the GMR value that can be used.
"""
import numpy as np
import torch
from pyrobolearn.approximators.gmr import GMRApproximator
from pyrobolearn.values.value import ParametrizedValue, ParametrizedQValue, ParametrizedQValueOutput
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class GMRValue(ParametrizedValue):
r"""GMR State Value Function Approximator
State value function :math:`V_{\phi}(s)` approximated by a GMR model, where :math:`\phi` represents
the parameters of that model.
"""
def __init__(self, state, preprocessors=None):
"""
Initialize the GMR state value function approximator.
Args:
state (State): input state.
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
the inner model / function approximator.
"""
model = GMRApproximator(inputs=state, outputs=torch.Tensor([1]), preprocessors=preprocessors)
super(GMRValue, self).__init__(state, model=model)
def __copy__(self):
"""Return a shallow copy of the value approximator. This can be overridden in the child class."""
return self.__class__(state=self.state, preprocessors=self.model.preprocessors)
def __deepcopy__(self, memo={}):
"""Return a deep copy of the value approximator. This can be overridden in the child class.
Args:
memo (dict): memo dictionary of objects already copied during the current copying pass
"""
if self in memo:
return memo[self]
state = copy.deepcopy(self.state, memo)
preprocessors = [copy.deepcopy(preprocessor, memo) for preprocessor in self.model.preprocessors]
value = self.__class__(state=state, preprocessors=preprocessors)
memo[self] = value
return value
class GMRQValue(ParametrizedQValue):
r"""GMR Q-value function approximator (which accepts as inputs the states and actions)
State-action value function :math:`Q_{\phi}(s, a)` approximated by a GMR model, where :math:`\phi` represents
the parameters of that model. This approximator accepts as inputs the states :math:`s` and actions :math:`a`,
and outputs the value :math:`Q(s,a)`. This can be used for continuous actions as well as discrete actions.
"""
def __init__(self, state, action, preprocessors=None):
"""
Initialize the GMR state-action value function approximator.
Args:
state (State): input state.
action (Action): input action.
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
the inner model / function approximator.
"""
model = GMRApproximator(inputs=[state, action], outputs=torch.Tensor([1]), preprocessors=preprocessors)
super(GMRQValue, self).__init__(state, action, model=model)
def __copy__(self):
"""Return a shallow copy of the value approximator. This can be overridden in the child class."""
return self.__class__(state=self.state, action=self.action, preprocessors=self.model.preprocessors)
def __deepcopy__(self, memo={}):
"""Return a deep copy of the value approximator. This can be overridden in the child class.
Args:
memo (dict): memo dictionary of objects already copied during the current copying pass
"""
if self in memo:
return memo[self]
state = copy.deepcopy(self.state, memo)
action = copy.deepcopy(self.action, memo)
preprocessors = [copy.deepcopy(preprocessor, memo) for preprocessor in self.model.preprocessors]
value = self.__class__(state=state, action=action, preprocessors=preprocessors)
memo[self] = value
return value
class GMRQValueOutput(ParametrizedQValueOutput):
r"""GMR Q-value function approximator (which accepts as inputs the states and outputs a Q-value for each
discrete action)
State-action value function :math:`Q_{\phi}(s, a)` approximated by a GMR model, where :math:`\phi` represents
the parameters of that model. This approximator accepts as inputs the states :math:`s` and outputs the value
:math:`Q(s,a)` for each discrete action. This can NOT be used with continuous actions.
"""
def __init__(self, state, action, preprocessors=None):
"""
Initialize the GMR state-action value function approximator.
Args:
state (State): input state.
action (Action): output action.
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
the inner model / function approximator.
"""
model = GMRApproximator(inputs=state, outputs=action, preprocessors=preprocessors)
super(GMRQValueOutput, self).__init__(state, action, model=model)
def __copy__(self):
"""Return a shallow copy of the value approximator. This can be overridden in the child class."""
return self.__class__(state=self.state, action=self.action, preprocessors=self.model.preprocessors)
def __deepcopy__(self, memo={}):
"""Return a deep copy of the value approximator. This can be overridden in the child class.
Args:
memo (dict): memo dictionary of objects already copied during the current copying pass
"""
if self in memo:
return memo[self]
state = copy.deepcopy(self.state, memo)
action = copy.deepcopy(self.action, memo)
preprocessors = [copy.deepcopy(preprocessor, memo) for preprocessor in self.model.preprocessors]
value = self.__class__(state=state, action=action, preprocessors=preprocessors)
memo[self] = value
return value
pyrobolearn/values/gpr.py
+145
View File
@@ -0,0 +1,145 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Define the Gaussian Process Regression (GPR) value function approximators.
Define the GPR value that can be used.
"""
import numpy as np
import torch
from pyrobolearn.approximators.gp import GPRApproximator
from pyrobolearn.values.value import ParametrizedValue, ParametrizedQValue, ParametrizedQValueOutput
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2019, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class GPRValue(ParametrizedValue):
r"""GPR State Value Function Approximator
State value function :math:`V_{\phi}(s)` approximated by a GPR model, where :math:`\phi` represents
the parameters of that model.
"""
def __init__(self, state, preprocessors=None):
"""
Initialize the GPR state value function approximator.
Args:
state (State): input state.
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
the inner model / function approximator.
"""
model = GPRApproximator(inputs=state, outputs=torch.Tensor([1]), preprocessors=preprocessors)
super(GPRValue, self).__init__(state, model=model)
def __copy__(self):
"""Return a shallow copy of the value approximator. This can be overridden in the child class."""
return self.__class__(state=self.state, preprocessors=self.model.preprocessors)
def __deepcopy__(self, memo={}):
"""Return a deep copy of the value approximator. This can be overridden in the child class.
Args:
memo (dict): memo dictionary of objects already copied during the current copying pass
"""
if self in memo:
return memo[self]
state = copy.deepcopy(self.state, memo)
preprocessors = [copy.deepcopy(preprocessor, memo) for preprocessor in self.model.preprocessors]
value = self.__class__(state=state, preprocessors=preprocessors)
memo[self] = value
return value
class GPRQValue(ParametrizedQValue):
r"""GPR Q-value function approximator (which accepts as inputs the states and actions)
State-action value function :math:`Q_{\phi}(s, a)` approximated by a GPR model, where :math:`\phi` represents
the parameters of that model. This approximator accepts as inputs the states :math:`s` and actions :math:`a`,
and outputs the value :math:`Q(s,a)`. This can be used for continuous actions as well as discrete actions.
"""
def __init__(self, state, action, preprocessors=None):
"""
Initialize the GPR state-action value function approximator.
Args:
state (State): input state.
action (Action): input action.
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
the inner model / function approximator.
"""
model = GPRApproximator(inputs=[state, action], outputs=torch.Tensor([1]), preprocessors=preprocessors)
super(GPRQValue, self).__init__(state, action, model=model)
def __copy__(self):
"""Return a shallow copy of the value approximator. This can be overridden in the child class."""
return self.__class__(state=self.state, action=self.action, preprocessors=self.model.preprocessors)
def __deepcopy__(self, memo={}):
"""Return a deep copy of the value approximator. This can be overridden in the child class.
Args:
memo (dict): memo dictionary of objects already copied during the current copying pass
"""
if self in memo:
return memo[self]
state = copy.deepcopy(self.state, memo)
action = copy.deepcopy(self.action, memo)
preprocessors = [copy.deepcopy(preprocessor, memo) for preprocessor in self.model.preprocessors]
value = self.__class__(state=state, action=action, preprocessors=preprocessors)
memo[self] = value
return value
class GPRQValueOutput(ParametrizedQValueOutput):
r"""GPR Q-value function approximator (which accepts as inputs the states and outputs a Q-value for each
discrete action)
State-action value function :math:`Q_{\phi}(s, a)` approximated by a GPR model, where :math:`\phi` represents
the parameters of that model. This approximator accepts as inputs the states :math:`s` and outputs the value
:math:`Q(s,a)` for each discrete action. This can NOT be used with continuous actions.
"""
def __init__(self, state, action, preprocessors=None):
"""
Initialize the GPR state-action value function approximator.
Args:
state (State): input state.
action (Action): output action.
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
the inner model / function approximator.
"""
model = GPRApproximator(inputs=state, outputs=action, preprocessors=preprocessors)
super(GPRQValueOutput, self).__init__(state, action, model=model)
def __copy__(self):
"""Return a shallow copy of the value approximator. This can be overridden in the child class."""
return self.__class__(state=self.state, action=self.action, preprocessors=self.model.preprocessors)
def __deepcopy__(self, memo={}):
"""Return a deep copy of the value approximator. This can be overridden in the child class.
Args:
memo (dict): memo dictionary of objects already copied during the current copying pass
"""
if self in memo:
return memo[self]
state = copy.deepcopy(self.state, memo)
action = copy.deepcopy(self.action, memo)
preprocessors = [copy.deepcopy(preprocessor, memo) for preprocessor in self.model.preprocessors]
value = self.__class__(state=state, action=action, preprocessors=preprocessors)
memo[self] = value
return value
pyrobolearn/values/linear.py
+143
View File
@@ -0,0 +1,143 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""Provides the linear value function approximator.
"""
import copy
import torch
from pyrobolearn.approximators import LinearApproximator
from pyrobolearn.values.value import ParametrizedValue, ParametrizedQValue, ParametrizedQValueOutput
__author__ = "Brian Delhaisse"
__copyright__ = "Copyright 2018, PyRoboLearn"
__credits__ = ["Brian Delhaisse"]
__license__ = "GNU GPLv3"
__version__ = "1.0.0"
__maintainer__ = "Brian Delhaisse"
__email__ = "briandelhaisse@gmail.com"
__status__ = "Development"
class LinearValue(ParametrizedValue):
r"""Linear State Value Function Approximator
State value function :math:`V_{\phi}(s)` approximated by a linear model, where :math:`\phi` represents
the parameters of that model.
"""
def __init__(self, state, preprocessors=None):
"""
Initialize the linear state value function approximator.
Args:
state (State): input state.
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
the inner model / function approximator.
"""
model = LinearApproximator(inputs=state, outputs=torch.Tensor([1]), preprocessors=preprocessors)
super(LinearValue, self).__init__(state, model=model)
def __copy__(self):
"""Return a shallow copy of the value approximator. This can be overridden in the child class."""
return self.__class__(state=self.state, preprocessors=self.model.preprocessors)
def __deepcopy__(self, memo={}):
"""Return a deep copy of the value approximator. This can be overridden in the child class.
Args:
memo (dict): memo dictionary of objects already copied during the current copying pass
"""
if self in memo:
return memo[self]
state = copy.deepcopy(self.state, memo)
preprocessors = [copy.deepcopy(preprocessor, memo) for preprocessor in self.model.preprocessors]
value = self.__class__(state=state, preprocessors=preprocessors)
memo[self] = value
return value
class LinearQValue(ParametrizedQValue):
r"""Linear Q-value function approximator (which accepts as inputs the states and actions)
State-action value function :math:`Q_{\phi}(s, a)` approximated by a linear model, where :math:`\phi` represents
the parameters of that model. This approximator accepts as inputs the states :math:`s` and actions :math:`a`,
and outputs the value :math:`Q(s,a)`. This can be used for continuous actions as well as discrete actions.
"""
def __init__(self, state, action, preprocessors=None):
"""
Initialize the linear state-action value function approximator.
Args:
state (State): input state.
action (Action): input action.
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
the inner model / function approximator.
"""
model = LinearApproximator(inputs=[state, action], outputs=torch.Tensor([1]), preprocessors=preprocessors)
super(LinearQValue, self).__init__(state, action, model=model)
def __copy__(self):
"""Return a shallow copy of the value approximator. This can be overridden in the child class."""
return self.__class__(state=self.state, action=self.action, preprocessors=self.model.preprocessors)
def __deepcopy__(self, memo={}):
"""Return a deep copy of the value approximator. This can be overridden in the child class.
Args:
memo (dict): memo dictionary of objects already copied during the current copying pass
"""
if self in memo:
return memo[self]
state = copy.deepcopy(self.state, memo)
action = copy.deepcopy(self.action, memo)
preprocessors = [copy.deepcopy(preprocessor, memo) for preprocessor in self.model.preprocessors]
value = self.__class__(state=state, action=action, preprocessors=preprocessors)
memo[self] = value
return value
class LinearQValueOutput(ParametrizedQValueOutput):
r"""Linear Q-value function approximator (which accepts as inputs the states and outputs a Q-value for each
discrete action)
State-action value function :math:`Q_{\phi}(s, a)` approximated by a linear model, where :math:`\phi` represents
the parameters of that model. This approximator accepts as inputs the states :math:`s` and outputs the value
:math:`Q(s,a)` for each discrete action. This can NOT be used with continuous actions.
"""
def __init__(self, state, action, preprocessors=None):
"""
Initialize the linear state-action value function approximator.
Args:
state (State): input state.
action (Action): output action.
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
the inner model / function approximator.
"""
model = LinearApproximator(inputs=state, outputs=action, preprocessors=preprocessors)
super(LinearQValueOutput, self).__init__(state, action, model=model)
def __copy__(self):
"""Return a shallow copy of the value approximator. This can be overridden in the child class."""
return self.__class__(state=self.state, action=self.action, preprocessors=self.model.preprocessors)
def __deepcopy__(self, memo={}):
"""Return a deep copy of the value approximator. This can be overridden in the child class.
Args:
memo (dict): memo dictionary of objects already copied during the current copying pass
"""
if self in memo:
return memo[self]
state = copy.deepcopy(self.state, memo)
action = copy.deepcopy(self.action, memo)
preprocessors = [copy.deepcopy(preprocessor, memo) for preprocessor in self.model.preprocessors]
value = self.__class__(state=state, action=action, preprocessors=preprocessors)
memo[self] = value
return value
pyrobolearn/values/nn_value.py → pyrobolearn/values/nn.py
View File