mirror of
https://github.com/wassname/pyrobolearn.git
synced 2026-09-09 11:31:38 +08:00
update actions, states, approx, policies, and values
This commit is contained in:
Brian Delhaisse
@@ -742,7 +742,7 @@ class Action(object):
|
||||
it checks that it is within the bounds.
|
||||
|
||||
Args:
|
||||
item (Action, list/tuple of action): check if given action(s) is(are) in the combined action
|
||||
item (Action, list/tuple of action, type): check if given action(s) is(are) in the combined action
|
||||
|
||||
Example:
|
||||
s1 = JntPositionAction(robot)
|
||||
@@ -753,12 +753,23 @@ class Action(object):
|
||||
print((s1, s2) in s) # output True
|
||||
"""
|
||||
# check type of item
|
||||
if not isinstance(item, (Action, np.ndarray)):
|
||||
raise TypeError("Expecting a action or numpy array.")
|
||||
if not isinstance(item, (Action, np.ndarray, type)):
|
||||
raise TypeError("Expecting an Action, a np.array, or a class type, instead got: {}".format(type(item)))
|
||||
|
||||
# if class type
|
||||
if isinstance(item, type):
|
||||
# if there is one action
|
||||
if self.has_data():
|
||||
return self.__class__ == item
|
||||
# the action has multiple actions, thus we go through each action
|
||||
for action in self.actions:
|
||||
if action.__class__ == item:
|
||||
return True
|
||||
return False
|
||||
|
||||
# check if action item is in the combined action
|
||||
if self._data is None and isinstance(item, Action):
|
||||
return (item in self._actions)
|
||||
return item in self._actions
|
||||
|
||||
# check if action/data is within the bounds
|
||||
if isinstance(item, Action):
|
||||
|
||||
@@ -85,6 +85,38 @@ class JointVelocityAction(JointAction):
|
||||
self.robot.set_joint_velocities(data, self.joints)
|
||||
|
||||
|
||||
class JointPositionAndVelocityAction(JointAction):
|
||||
r"""Joint position and velocity action
|
||||
|
||||
Set the joint position using position control using PD control, where the contraint error to be minimized is
|
||||
given by: :math:`error = kp * (q^* - q) - kd * (\dot{q}^* - \dot{q})`.
|
||||
"""
|
||||
|
||||
def __init__(self, robot, joint_ids=None, kp=None, kd=None, max_force=None):
|
||||
super(JointPositionAndVelocityAction, self).__init__(robot, joint_ids)
|
||||
self.kp, self.kd, self.max_force = kp, kd, max_force
|
||||
pos, vel = robot.get_joint_positions(self.joints), robot.get_joint_velocities(self.joints)
|
||||
self.data = np.concatenate((pos, vel))
|
||||
self.idx = len(pos)
|
||||
|
||||
def _write(self, data=None):
|
||||
if data is None:
|
||||
self.robot.set_joint_positions(self._data[:self.idx], self.joints, kp=self.kp, kd=self.kd,
|
||||
velocities=self._data[self.idx:], forces=self.max_force)
|
||||
else:
|
||||
self.robot.set_joint_positions(data[:self.idx], self.joints, kp=self.kp, kd=self.kd,
|
||||
velocities=data[self.idx:], forces=self.max_force)
|
||||
|
||||
|
||||
# class JointPositionVelocityAccelerationAction(JointAction):
|
||||
# r"""Set the joint positions, velocities, and accelerations.
|
||||
#
|
||||
# Set the joint positions, velocities, and accelerations by computing the necessary torques / forces using inverse
|
||||
# dynamics.
|
||||
# """
|
||||
# pass
|
||||
|
||||
|
||||
class JointForceAction(JointAction):
|
||||
r"""Joint Force Action
|
||||
|
||||
|
||||
@@ -366,8 +366,28 @@ class Approximator(object):
|
||||
processor.reset()
|
||||
self.model.reset()
|
||||
|
||||
def predict(self, x=None, to_numpy=True, return_logits=False):
|
||||
"""Predict the output given the input."""
|
||||
def __convert_to_numpy(self, x, to_numpy=True):
|
||||
"""Convert the given argument to a numpy array if specified."""
|
||||
if to_numpy and isinstance(x, torch.Tensor):
|
||||
if x.requires_grad:
|
||||
return x.detach().numpy()
|
||||
return x.numpy()
|
||||
return x
|
||||
|
||||
def predict(self, x=None, to_numpy=True, return_logits=False, set_output_data=True):
|
||||
"""Predict the output given the input.
|
||||
|
||||
Args:
|
||||
x (None, State, Action, (list of) np.array, (list of) torch.Tensor): input data. If None, it will get the
|
||||
data from the inputs that were given at the initialization.
|
||||
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 at the
|
||||
initialization.
|
||||
|
||||
Returns:
|
||||
(list of) np.array, list of (torch.Tensor): predicted output data.
|
||||
"""
|
||||
# if no input is given, take the provided inputs at the beginning
|
||||
if x is None:
|
||||
x = self.inputs
|
||||
@@ -382,45 +402,67 @@ class Approximator(object):
|
||||
for processor in self.preprocessors:
|
||||
x = processor(x)
|
||||
|
||||
# go through the model
|
||||
# predict output using the learning model
|
||||
x = self.model.predict(x, to_numpy=False)
|
||||
|
||||
# go through each postprocessor
|
||||
for processor in self.postprocessors:
|
||||
x = processor(x)
|
||||
|
||||
# set the output data
|
||||
if isinstance(self.outputs, (State, Action)): # TODO: think when multiple outputs and to set them
|
||||
if self.outputs.is_discrete() and not return_logits:
|
||||
if isinstance(x, np.ndarray):
|
||||
x = np.array([np.argmax(x)])
|
||||
elif isinstance(x, torch.Tensor):
|
||||
x = torch.argmax(x, dim=0, keepdim=True)
|
||||
# set the output data and convert it if specified
|
||||
if isinstance(self.outputs, (State, Action)) and (to_numpy or not return_logits or set_output_data):
|
||||
# if output data `x` is not a list, make it a list as we will iterate through it
|
||||
if not isinstance(x, list):
|
||||
x = [x]
|
||||
|
||||
# go through each output and output data
|
||||
for idx, (output, data) in enumerate(zip(self.outputs, x)):
|
||||
if isinstance(data, np.ndarray):
|
||||
if output.is_discrete():
|
||||
discrete_data = np.array([np.argmax(data)])
|
||||
if set_output_data:
|
||||
output.data = discrete_data
|
||||
if not return_logits:
|
||||
x[idx] = discrete_data
|
||||
elif set_output_data:
|
||||
output.data = data
|
||||
elif isinstance(data, torch.Tensor):
|
||||
if output.is_discrete():
|
||||
discrete_data = torch.argmax(data, dim=0, keepdim=True)
|
||||
if set_output_data:
|
||||
output.torch_data = discrete_data
|
||||
if return_logits:
|
||||
x[idx] = self.__convert_to_numpy(data, to_numpy=to_numpy)
|
||||
else:
|
||||
x[idx] = self.__convert_to_numpy(discrete_data, to_numpy=to_numpy)
|
||||
else:
|
||||
raise TypeError("Expecting `x` to be a numpy array, torch.Tensor, or a list of them, instead got: "
|
||||
"{}".format(type(x)))
|
||||
raise TypeError("Expecting `data` output to be a numpy array, torch.Tensor, or a list of them, "
|
||||
"instead got: {}".format(type(data)))
|
||||
|
||||
# set the data
|
||||
if isinstance(x, np.ndarray):
|
||||
self.outputs.data = x
|
||||
else: # isinstance(x, torch.Tensor):
|
||||
self.outputs.torch_data = x
|
||||
# if output is a list and has one element, return just that element
|
||||
if isinstance(x, list) and len(x) == 1:
|
||||
x = x[0]
|
||||
|
||||
# return the data
|
||||
# convert to numpy if specified
|
||||
if to_numpy and isinstance(x, torch.Tensor):
|
||||
if x.requires_grad:
|
||||
return x.detach().numpy()
|
||||
return x.numpy()
|
||||
# return the output data
|
||||
return x
|
||||
|
||||
def save(self, filename):
|
||||
"""save the inner model."""
|
||||
"""Save the inner model on the disk.
|
||||
|
||||
Args:
|
||||
filename (str): path to the file to save the model.
|
||||
"""
|
||||
self.model.save(filename)
|
||||
|
||||
def load(self, filename):
|
||||
"""load the inner model."""
|
||||
self.model.load(filename)
|
||||
"""Load the inner model from the disk.
|
||||
|
||||
Args:
|
||||
filename (str): path to the file which contains the model.
|
||||
"""
|
||||
# TODO: check if model is None
|
||||
self.model = self.model.load(filename)
|
||||
return self.model
|
||||
|
||||
#############
|
||||
# Operators #
|
||||
|
||||
@@ -112,26 +112,24 @@ class MLPApproximator(NNApproximator):
|
||||
return False
|
||||
return True
|
||||
|
||||
def predict(self, x):
|
||||
# convert given input to torch tensor
|
||||
if isinstance(x, (State, Action)):
|
||||
x = x.merged_data[0]
|
||||
x = torch.from_numpy(x).float()
|
||||
|
||||
# feed it to the model and get predicted output
|
||||
x = self.model(x)
|
||||
|
||||
# check output
|
||||
if isinstance(self.outputs, (State, Action)):
|
||||
# data
|
||||
self.outputs.train_data = x
|
||||
# convert back output from torch tensor to np array
|
||||
x = x.detach().numpy()
|
||||
self.outputs.data = x
|
||||
|
||||
return self.outputs
|
||||
|
||||
return x
|
||||
# def predict(self, x):
|
||||
# # convert given input to torch tensor
|
||||
# if isinstance(x, (State, Action)):
|
||||
# x = x.merged_data[0]
|
||||
# x = torch.from_numpy(x).float()
|
||||
#
|
||||
# # feed it to the model and get predicted output
|
||||
# x = self.model(x)
|
||||
#
|
||||
# # check output
|
||||
# if isinstance(self.outputs, (State, Action)):
|
||||
# # data
|
||||
# self.outputs.train_data = x
|
||||
# # convert back output from torch tensor to np array
|
||||
# x = x.detach().numpy()
|
||||
# self.outputs.data = x
|
||||
# return self.outputs
|
||||
# return x
|
||||
|
||||
|
||||
class NEATApproximator(Approximator):
|
||||
@@ -208,21 +206,23 @@ class NEATApproximator(Approximator):
|
||||
# Methods #
|
||||
###########
|
||||
|
||||
def predict(self, x, to_numpy=True):
|
||||
# x = self.preprocessors(x)
|
||||
x = self.model.predict(x.merged_data[0])
|
||||
|
||||
if isinstance(self.outputs, (State, Action)):
|
||||
if self.outputs.is_discrete():
|
||||
x = np.argmax(x)
|
||||
elif self.outputs.is_continuous():
|
||||
x = 2 * np.array(x) - 1
|
||||
else:
|
||||
raise NotImplementedError("The outputs are not discrete or continuous...")
|
||||
self.outputs.data = x
|
||||
# x = self.postprocessors(x)
|
||||
# return x
|
||||
return self.outputs
|
||||
# def predict(self, x, to_numpy=True, return_logits=True, set_output_data=False):
|
||||
# # x = self.preprocessors(x)
|
||||
# x = self.model.predict(x)
|
||||
#
|
||||
# if isinstance(self.outputs, (State, Action)):
|
||||
# if self.outputs.is_discrete():
|
||||
# print(x)
|
||||
# x = np.argmax(x)
|
||||
# print(x)
|
||||
# elif self.outputs.is_continuous():
|
||||
# x = 2 * np.array(x) - 1
|
||||
# else:
|
||||
# raise NotImplementedError("The outputs are not discrete or continuous...")
|
||||
# self.outputs.data = x
|
||||
# # x = self.postprocessors(x)
|
||||
# # return x
|
||||
# return self.outputs
|
||||
|
||||
def set_network(self, genome=None, config=None):
|
||||
"""Set the genome network."""
|
||||
|
||||
@@ -5,8 +5,10 @@ Define the various basic policies such as the random policy, linear policy, poli
|
||||
"""
|
||||
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
from pyrobolearn.policies.policy import Policy
|
||||
from pyrobolearn.values.value import ParametrizedStateOutputActionValue
|
||||
from pyrobolearn.approximators import LinearApproximator
|
||||
|
||||
|
||||
@@ -24,63 +26,150 @@ class RandomPolicy(Policy):
|
||||
"""Random policy
|
||||
"""
|
||||
|
||||
def __init__(self, states, actions, seed=None, *args, **kwargs):
|
||||
super(RandomPolicy, self).__init__(states, actions, *args, **kwargs)
|
||||
if seed is not None:
|
||||
np.random.seed(seed)
|
||||
class RandomModel(object):
|
||||
"""Random model"""
|
||||
def __init__(self, actions, seed=None):
|
||||
self.seed = seed
|
||||
self.actions = actions
|
||||
|
||||
def act(self, state=None, deterministic=False, to_numpy=True):
|
||||
# get the space of each action
|
||||
spaces = self.actions.space
|
||||
@property
|
||||
def seed(self):
|
||||
return self._seed
|
||||
|
||||
# sample from each space
|
||||
action_data = [space.sample() for space in spaces]
|
||||
@seed.setter
|
||||
def seed(self, seed):
|
||||
if seed is not None:
|
||||
np.random.seed(seed)
|
||||
self._seed = seed
|
||||
|
||||
# set the data for each action
|
||||
self.actions.data = action_data
|
||||
def predict(self, state=None, to_numpy=True):
|
||||
spaces = self.actions.space
|
||||
return [space.sample() for space in spaces]
|
||||
|
||||
return self.actions
|
||||
def __init__(self, states, actions, rate=1, seed=None, preprocessors=None, postprocessors=None, *args, **kwargs):
|
||||
"""
|
||||
Initialize the Random policy.
|
||||
|
||||
def sample(self, state):
|
||||
Args:
|
||||
actions (Action): At each step, by calling `policy.act(state)`, the `actions` are computed by the policy,
|
||||
and should be given to the environment. As with the `states`, the type and size/shape of each 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.
|
||||
states (State): By giving the `states` to the policy, it can automatically infer the type and size/shape
|
||||
of each state, and thus can be used to automatically build a policy. At each step, the `states`
|
||||
are 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 (Approximator, Model, None): inner model or approximator
|
||||
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 = self.RandomModel(actions, seed=seed)
|
||||
super(RandomPolicy, self).__init__(states, actions, model=model, rate=rate, preprocessors=preprocessors,
|
||||
postprocessors=postprocessors, *args, **kwargs)
|
||||
|
||||
def sample(self, state=None):
|
||||
return self.act(state)
|
||||
|
||||
def reset(self, seed=None):
|
||||
if seed is not None:
|
||||
np.random.seed(seed)
|
||||
|
||||
|
||||
class LinearPolicy(Policy):
|
||||
"""Linear Policy
|
||||
"""
|
||||
|
||||
def __init__(self, states, actions, *args, **kwargs):
|
||||
model = LinearApproximator(states, actions)
|
||||
super(LinearPolicy, self).__init__(states, actions, model, *args, **kwargs)
|
||||
def __init__(self, states, actions, rate=1, preprocessors=None, postprocessors=None, *args, **kwargs):
|
||||
"""
|
||||
Initialize the Linear Policy.
|
||||
|
||||
def act(self, state, deterministic=True, to_numpy=True):
|
||||
return self.model.predict(state, to_numpy=to_numpy)
|
||||
|
||||
def sample(self, state):
|
||||
pass
|
||||
Args:
|
||||
actions (Action): At each step, by calling `policy.act(state)`, the `actions` are computed by the policy,
|
||||
and should be given to the environment. As with the `states`, the type and size/shape of each 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.
|
||||
states (State): By giving the `states` to the policy, it can automatically infer the type and size/shape
|
||||
of each state, and thus can be used to automatically build a policy. At each step, the `states`
|
||||
are 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(states, actions, preprocessors=preprocessors, postprocessors=postprocessors)
|
||||
super(LinearPolicy, self).__init__(states, actions, model, rate=rate, *args, **kwargs)
|
||||
|
||||
|
||||
class PolicyFromValue(Policy):
|
||||
r"""Policy From Value Function Approximator
|
||||
r"""Policy From state-action value function approximator
|
||||
|
||||
.. math::
|
||||
|
||||
a = argmax_a Q^\pi(s,a)
|
||||
This computes the optimal discrete action :math:`a` using the underlying value function approximator
|
||||
:math:`Q(s,a)` which given the state as input computes the Q-value for each discrete action. The policy select
|
||||
the best action by computing :math:`a = argmax_a Q_{\phi}(s,a)`.
|
||||
|
||||
.. seealso::
|
||||
|
||||
* `value.py`
|
||||
"""
|
||||
|
||||
def __init__(self, states, actions, *args, **kwargs):
|
||||
super(PolicyFromValue, self).__init__(states, actions, *args, **kwargs)
|
||||
def __init__(self, value, rate=1, preprocessors=None, postprocessors=None, *args, **kwargs):
|
||||
"""
|
||||
Initialize the Policy from the value function approximator.
|
||||
|
||||
def act(self, state=None, deterministic=True, to_numpy=True):
|
||||
pass
|
||||
Args:
|
||||
value (ParametrizedStateOutputActionValue): trainable value function approximator.
|
||||
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
|
||||
"""
|
||||
self.value = value
|
||||
super(PolicyFromValue, self).__init__(value.state, value.action, model=value, rate=rate,
|
||||
preprocessors=preprocessors, postprocessors=postprocessors,
|
||||
*args, **kwargs)
|
||||
|
||||
def sample(self, state):
|
||||
pass
|
||||
##############
|
||||
# Properties #
|
||||
##############
|
||||
|
||||
@property
|
||||
def value(self):
|
||||
"""Return the value function approximator."""
|
||||
return self._value
|
||||
|
||||
@value.setter
|
||||
def value(self, value):
|
||||
"""Set the value function approximator."""
|
||||
# TODO: need to check that input and output dimensions of the new value function approximator match the ones
|
||||
# from the previous model.
|
||||
if not isinstance(value, ParametrizedStateOutputActionValue):
|
||||
raise TypeError("Expecting the given `value` function approximator to be an instance of "
|
||||
"`ParametrizedStateOutputActionValue`, instead got: {}".format(type(value)))
|
||||
self._value = value
|
||||
|
||||
###########
|
||||
# Methods #
|
||||
###########
|
||||
|
||||
def _predict(self, state, to_numpy=False, return_logits=True, set_output_data=False):
|
||||
"""Inner prediction step."""
|
||||
action = self.model.compute(state, to_numpy=to_numpy)
|
||||
if to_numpy:
|
||||
return np.argmax(action)
|
||||
return torch.argmax(action, dim=0, keepdim=True)
|
||||
|
||||
# def act(self, state=None, deterministic=True, to_numpy=True, return_logits=False, apply_action=True):
|
||||
# pass
|
||||
|
||||
# def sample(self, state):
|
||||
# pass
|
||||
|
||||
@@ -33,14 +33,16 @@ class CPGPolicy(Policy):
|
||||
child_coupling_weight=None, child_coupling_bias=0., update_amplitudes=True, update_offsets=True,
|
||||
update_init_phases=True, update_frequencies=True, update_weights=True, update_biases=True,
|
||||
amplitude_bounds=np.pi, offset_bounds=np.pi, phase_bounds=np.pi, frequency_bounds=5.,
|
||||
weight_bounds=2., bias_bounds=np.pi, *args, **kwargs):
|
||||
weight_bounds=2., bias_bounds=np.pi, preprocessors=None, postprocessors=None, *args, **kwargs):
|
||||
"""
|
||||
Initialize the CPG Network Policy.
|
||||
|
||||
Args:
|
||||
states (PhaseState): phase state.
|
||||
actions (JointAction): joint action. Normally, it will be JointPositionAction.
|
||||
rate (int): number of steps to wait before going to the next step with the policy.
|
||||
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.
|
||||
cpg_network (dict, None): dictionary describing the CPG network. The syntax is the following:
|
||||
cpg_network = {<node_id>: {'phi': <phi>, 'offset': <offset>, 'amplitude': <amplitude>, 'freq': <freq>,
|
||||
'nodes': [{'id': <coupling_node_id>, 'bias': <coupling_bias>,
|
||||
@@ -90,10 +92,13 @@ class CPGPolicy(Policy):
|
||||
frequency_bounds (float, tuple of float): bounds / limits to the frequency parameter (useful when training).
|
||||
weight_bounds (float, tuple of float): bounds / limits to the weight parameters (useful when training).
|
||||
bias_bounds (float, tuple of float): bounds / limits to the bias parameters (useful when training).
|
||||
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): other arguments given to the CPG network learning model.
|
||||
**kwargs (dict): other key + value arguments given to the CPG network learning model.
|
||||
"""
|
||||
super(CPGPolicy, self).__init__(states, actions, rate=rate, *args, **kwargs)
|
||||
super(CPGPolicy, self).__init__(states, actions, rate=rate, preprocessors=preprocessors,
|
||||
postprocessors=postprocessors, *args, **kwargs)
|
||||
|
||||
# check actions
|
||||
if not isinstance(actions, JointAction):
|
||||
@@ -191,32 +196,22 @@ class CPGPolicy(Policy):
|
||||
frequency_bounds=frequency_bounds, weight_bounds=weight_bounds,
|
||||
bias_bounds=bias_bounds, *args, **kwargs)
|
||||
|
||||
def _size(self, x):
|
||||
size = 0
|
||||
if isinstance(x, (State, Action)):
|
||||
if x.is_discrete():
|
||||
size = x.space[0].n
|
||||
else:
|
||||
size = x.total_size()
|
||||
elif isinstance(x, np.ndarray):
|
||||
size = x.size
|
||||
elif isinstance(x, torch.Tensor):
|
||||
size = x.numel()
|
||||
elif isinstance(x, int):
|
||||
size = x
|
||||
return size
|
||||
def _predict(self, state, to_numpy=False, return_logits=True, set_output_data=False):
|
||||
"""Inner prediction step."""
|
||||
action_data = self.model.step()
|
||||
return action_data
|
||||
|
||||
def act(self, state=None, deterministic=True, to_numpy=True, return_logits=False):
|
||||
if (self.cnt % self.rate) == 0:
|
||||
self.last_action = self.model.step()
|
||||
self.cnt += 1
|
||||
# angles = self.model.step()
|
||||
# self.actions.data = angles
|
||||
self.actions.data = self.last_action
|
||||
return self.actions
|
||||
# def act(self, state=None, deterministic=True, to_numpy=True, return_logits=False):
|
||||
# if (self.cnt % self.rate) == 0:
|
||||
# self.last_action = self.model.step()
|
||||
# self.cnt += 1
|
||||
# # angles = self.model.step()
|
||||
# # self.actions.data = angles
|
||||
# self.actions.data = self.last_action
|
||||
# return self.actions
|
||||
|
||||
def sample(self, state):
|
||||
pass
|
||||
# def sample(self, state):
|
||||
# pass
|
||||
|
||||
def phase_resetting(self):
|
||||
self.model.reset()
|
||||
|
||||
@@ -10,7 +10,8 @@ import torch
|
||||
from pyrobolearn.models import DMP, DiscreteDMP, RhythmicDMP, BioDiscreteDMP
|
||||
from pyrobolearn.policies.policy import Policy
|
||||
from pyrobolearn.states import State
|
||||
from pyrobolearn.actions import Action, JointPositionAction, JointVelocityAction, JointAccelerationAction
|
||||
from pyrobolearn.actions import Action, JointPositionAction, JointVelocityAction, JointAccelerationAction, \
|
||||
JointPositionAndVelocityAction
|
||||
|
||||
__author__ = "Brian Delhaisse"
|
||||
__copyright__ = "Copyright 2018, PyRoboLearn"
|
||||
@@ -30,49 +31,55 @@ class DMPPolicy(Policy):
|
||||
if not isinstance(model, DMP):
|
||||
raise TypeError("Expecting model to be an instance of DMP")
|
||||
super(DMPPolicy, self).__init__(states, actions, model, rate=rate, *args, **kwargs)
|
||||
self.y, self.dy, self.ddy = 0, 0, 0
|
||||
|
||||
def _size(self, x):
|
||||
size = 0
|
||||
if isinstance(x, (State, Action)):
|
||||
if x.is_discrete():
|
||||
size = x.space[0].n
|
||||
else:
|
||||
size = x.total_size()
|
||||
elif isinstance(x, np.ndarray):
|
||||
size = x.size
|
||||
elif isinstance(x, torch.Tensor):
|
||||
size = x.numel()
|
||||
elif isinstance(x, int):
|
||||
size = x
|
||||
return size
|
||||
# check actions
|
||||
self.is_joint_position_action = JointPositionAction in actions or JointPositionAndVelocityAction in actions
|
||||
self.is_joint_velocity_action = JointVelocityAction in actions or JointPositionAndVelocityAction in actions
|
||||
self.is_joint_acceleration_action = JointAccelerationAction in actions
|
||||
if not (self.is_joint_position_action or self.is_joint_velocity_action or self.is_joint_acceleration_action):
|
||||
raise ValueError("The actions do not have a joint position, velocity, or acceleration action.")
|
||||
|
||||
def act(self, state, deterministic=True, to_numpy=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 _predict(self, state, to_numpy=False, return_logits=True, set_output_data=False):
|
||||
"""Inner prediction step."""
|
||||
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 sample(self, state):
|
||||
pass
|
||||
# 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):
|
||||
def imitate(self, data): # TODO: improve this
|
||||
if len(data) > 0:
|
||||
print("Imitating with :", data.shape)
|
||||
# print("Imitating with :", data.shape)
|
||||
# y, dy, ddy = data
|
||||
y = data
|
||||
# if len(y.shape) == 1:
|
||||
|
||||
@@ -55,12 +55,36 @@ class NEATPolicy(Policy):
|
||||
"""
|
||||
|
||||
def __init__(self, states, actions, num_hidden=0, activation_fct='relu', network_type='feedforward',
|
||||
aggregation='sum', weights_limits=(-20, 20), bias_limits=(-20, 20), rate=1, *args, **kwargs):
|
||||
aggregation='sum', weights_limits=(-20, 20), bias_limits=(-20, 20), rate=1, preprocessors=None,
|
||||
postprocessors=None, *args, **kwargs):
|
||||
r"""Initialize the neural network policy for the NEAT algorithm.
|
||||
|
||||
Args:
|
||||
actions (Action): At each step, by calling `policy.act(state)`, the `actions` are computed by the policy,
|
||||
and should be given to the environment. As with the `states`, the type and size/shape of each 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.
|
||||
states (State): By giving the `states` to the policy, it can automatically infer the type and size/shape
|
||||
of each state, and thus can be used to automatically build a policy. At each step, the `states`
|
||||
are 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_hidden (int): number of units in the hidden layer
|
||||
activation_fct (str): activation function to use.
|
||||
network_type (str): type of neural network. Select between 'feedforward' and 'recurrent'.
|
||||
aggregation (str): how to aggregate the input signals of a node. Select between 'sum', 'product', 'max',
|
||||
'min', 'maxabs', 'median', and 'mean'.
|
||||
weights_limits (tuple): weight limits / bounds. The tuple contains the lower and upper bounds.
|
||||
bias_limits (tuple): bias limits / bounds. The tuple contains the lower and upper bounds.
|
||||
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
|
||||
"""
|
||||
model = NEATApproximator(states, actions, num_hidden=num_hidden, activation_fct=activation_fct,
|
||||
network_type=network_type, aggregation=aggregation, weights_limits=weights_limits,
|
||||
bias_limits=bias_limits)
|
||||
bias_limits=bias_limits, preprocessors=preprocessors, postprocessors=postprocessors)
|
||||
super(NEATPolicy, self).__init__(states, actions, model, rate=rate, *args, **kwargs)
|
||||
|
||||
##############
|
||||
@@ -79,18 +103,22 @@ class NEATPolicy(Policy):
|
||||
|
||||
@property
|
||||
def genome(self):
|
||||
"""Return the NEAT model's genome."""
|
||||
return self.model.genome
|
||||
|
||||
@genome.setter
|
||||
def genome(self, genome):
|
||||
"""Set the genome."""
|
||||
self.model.genome = genome
|
||||
|
||||
@property
|
||||
def network(self):
|
||||
"""Return the NEAT model's network."""
|
||||
return self.model.network
|
||||
|
||||
@property
|
||||
def population(self):
|
||||
"""Return the population used in NEAT."""
|
||||
return self.model.population
|
||||
|
||||
###########
|
||||
@@ -98,36 +126,38 @@ class NEATPolicy(Policy):
|
||||
###########
|
||||
|
||||
def update_config(self, config):
|
||||
"""Update the configuration file."""
|
||||
self.model.update_config(config)
|
||||
|
||||
def set_network(self, genome=None, config=None):
|
||||
"""Set the genome network."""
|
||||
self.model.set_network(genome, config)
|
||||
|
||||
def act(self, state, deterministic=True):
|
||||
if (self.cnt % self.rate) == 0:
|
||||
self.last_action = self.model.predict(state)
|
||||
self.cnt += 1
|
||||
return self.last_action
|
||||
|
||||
def sample(self, state):
|
||||
pass
|
||||
# def act(self, state, deterministic=True):
|
||||
# if (self.cnt % self.rate) == 0:
|
||||
# self.last_action = self.model.predict(state)
|
||||
# self.cnt += 1
|
||||
# return self.last_action
|
||||
#
|
||||
# def sample(self, state):
|
||||
# pass
|
||||
|
||||
|
||||
class NEATFeedForwardPolicy(NEATPolicy):
|
||||
r"""NEAT feed-forward policy
|
||||
|
||||
This creates a feed-forward network policy.
|
||||
"""
|
||||
|
||||
def __init__(self, states, actions, genome):
|
||||
super(NEATFeedForwardPolicy, self).__init__(states, actions, genome, network_type='feedforward')
|
||||
|
||||
|
||||
class NEATRecurrentPolicy(NEATPolicy):
|
||||
r"""NEAT recurrent policy
|
||||
|
||||
This creates a recurrent network policy.
|
||||
"""
|
||||
|
||||
def __init__(self, states, actions, genome):
|
||||
super(NEATRecurrentPolicy, self).__init__(states, actions, genome, network_type='recurrent')
|
||||
# class NEATFeedForwardPolicy(NEATPolicy):
|
||||
# r"""NEAT feed-forward policy
|
||||
#
|
||||
# This creates a feed-forward network policy.
|
||||
# """
|
||||
#
|
||||
# def __init__(self, states, actions, genome):
|
||||
# super(NEATFeedForwardPolicy, self).__init__(states, actions, genome, network_type='feedforward')
|
||||
#
|
||||
#
|
||||
# class NEATRecurrentPolicy(NEATPolicy):
|
||||
# r"""NEAT recurrent policy
|
||||
#
|
||||
# This creates a recurrent network policy.
|
||||
# """
|
||||
#
|
||||
# def __init__(self, states, actions, genome):
|
||||
# super(NEATRecurrentPolicy, self).__init__(states, actions, genome, network_type='recurrent')
|
||||
|
||||
@@ -21,15 +21,32 @@ __status__ = "Development"
|
||||
class NNPolicy(Policy):
|
||||
r"""Neural Network Policy
|
||||
|
||||
Defines the neural network policy. If the model is not given,
|
||||
|
||||
Examples:
|
||||
simulator = Bullet()
|
||||
robot = Robot(simulator)
|
||||
policy = NNPolicy(Robot, states=['joint_positions', 'joint_velocities'], actions=['joint_positions'])
|
||||
Defines the neural network policy.
|
||||
"""
|
||||
|
||||
def __init__(self, states, actions, model=None, *args, **kwargs):
|
||||
def __init__(self, states, actions, model=None, rate=1, preprocessors=None, postprocessors=None, *args, **kwargs):
|
||||
"""
|
||||
Initialize the Neural network policy.
|
||||
|
||||
Args:
|
||||
actions (Action): At each step, by calling `policy.act(state)`, the `actions` are computed by the policy,
|
||||
and should be given to the environment. As with the `states`, the type and size/shape of each 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.
|
||||
states (State): By giving the `states` to the policy, it can automatically infer the type and size/shape
|
||||
of each state, and thus can be used to automatically build a policy. At each step, the `states`
|
||||
are 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 (NN, NNApproximator): NN 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
|
||||
**kwargs (dict): dictionary of arguments
|
||||
"""
|
||||
if model is None:
|
||||
raise ValueError("Expecting a NN model for the NN policy")
|
||||
else:
|
||||
@@ -37,13 +54,14 @@ class NNPolicy(Policy):
|
||||
# checking the output dimension of the model and the dimension of actions
|
||||
pass
|
||||
|
||||
super(NNPolicy, self).__init__(states, actions, model, *args, **kwargs)
|
||||
super(NNPolicy, self).__init__(states, actions, model, rate=rate, preprocessors=preprocessors,
|
||||
postprocessors=postprocessors, *args, **kwargs)
|
||||
|
||||
def act(self, state, deterministic=True):
|
||||
pass
|
||||
|
||||
def sample(self, state):
|
||||
pass
|
||||
# def act(self, state, deterministic=True):
|
||||
# pass
|
||||
#
|
||||
# def sample(self, state):
|
||||
# pass
|
||||
|
||||
|
||||
class MLPPolicy(NNPolicy):
|
||||
@@ -53,9 +71,8 @@ class MLPPolicy(NNPolicy):
|
||||
activation functions.
|
||||
"""
|
||||
|
||||
def __init__(self, states, actions, hidden_units=(),
|
||||
activation_fct='linear', last_activation_fct=None, dropout_prob=None,
|
||||
preprocessors=None, postprocessors=None):
|
||||
def __init__(self, states, actions, hidden_units=(), activation_fct='linear', last_activation_fct=None,
|
||||
dropout_prob=None, rate=1, preprocessors=None, postprocessors=None):
|
||||
"""Initialize MLP policy.
|
||||
|
||||
Args:
|
||||
@@ -70,11 +87,16 @@ class MLPPolicy(NNPolicy):
|
||||
if it is in the list/tuple of activation functions provided for the
|
||||
previous argument.
|
||||
dropout_prob (None, float, or list/tuple of float/None): dropout probability.
|
||||
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
|
||||
"""
|
||||
model = MLPApproximator(states, actions, hidden_units=hidden_units,
|
||||
activation_fct=activation_fct, last_activation_fct=last_activation_fct,
|
||||
dropout_prob=dropout_prob, preprocessors=preprocessors, postprocessors=postprocessors)
|
||||
super(MLPPolicy, self).__init__(states, actions, model)
|
||||
super(MLPPolicy, self).__init__(states, actions, model, rate=rate)
|
||||
|
||||
def act(self, state, deterministic=True):
|
||||
return self.model.predict(state)
|
||||
# def act(self, state, deterministic=True):
|
||||
# return self.model.predict(state)
|
||||
|
||||
+314
-112
@@ -1,25 +1,27 @@
|
||||
#!/usr/bin/env python
|
||||
"""Define the basic Policy class.
|
||||
|
||||
A policy couples one or several learning model(s), the state, and action together. In this framework, the policy
|
||||
A policy couples one or several learning model(s), the action, and state together. In this framework, the policy
|
||||
usually represents the robot's "brain".
|
||||
|
||||
Dependencies:
|
||||
- `pyrobolearn.states`
|
||||
- `pyrobolearn.actions`
|
||||
- `pyrobolearn.approximators` (and thus `pyrobolearn.models`)
|
||||
- `pyrobolearn.models`
|
||||
- `pyrobolearn.approximators`
|
||||
- `pyrobolearn.exploration`
|
||||
"""
|
||||
|
||||
from abc import ABCMeta, abstractmethod
|
||||
import collections
|
||||
import pickle
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
from pyrobolearn.states import State
|
||||
from pyrobolearn.actions import Action
|
||||
|
||||
from pyrobolearn.models import Model
|
||||
from pyrobolearn.approximators import Approximator, NNApproximator
|
||||
|
||||
from pyrobolearn.approximators import Approximator
|
||||
|
||||
__author__ = "Brian Delhaisse"
|
||||
__copyright__ = "Copyright 2018, PyRoboLearn"
|
||||
@@ -35,22 +37,41 @@ class Policy(object):
|
||||
r"""Abstract `Policy` class.
|
||||
|
||||
A policy maps a state to an action, and is often denoted as :math:`\pi_{\theta}(a_t|s_t)`, where :math:`\theta`
|
||||
represents the policy parameters. It represents the cognition of the agent(s).
|
||||
In our framework, the policy groups the learning model, state, and action objects.
|
||||
represents the policy parameters. It represents the cognitive part of the agent(s).
|
||||
|
||||
Specifically, the policy is dissociated from the learning model, as a learning model can be used for different
|
||||
purposes. For instance, a neural network can be used to represent a policy but also a value function approximator,
|
||||
thus we separate these 2 notions (policy and learning model).
|
||||
In the PyRoboLearn (PRL) framework, the policy groups the learning model / function approximator, action, and
|
||||
state objects. Note that for some policies the state object is not required as they have an inner time state which
|
||||
allows to generate the next action each time they are called.
|
||||
|
||||
Anyway, the policy is dissociated from the learning model as this last one can be used for different purposes as
|
||||
well. For instance, a neural network can be used to represent a policy but also a value function approximator, or
|
||||
a dynamic transition probability function as well. We thus separate these 2 notions (policy and learning model /
|
||||
approximator).
|
||||
|
||||
The policy is also loosely dissociated from the simulator and more specifically from the agent's body, as this last
|
||||
one is seen as being part of the environment. The states and actions are what connects the policy with the
|
||||
environment (and thus the simulator). The states and actions are given to the policy, and allows to build
|
||||
automatically a learning model (if not given) by inferring the dimensions of the inputs and outputs of the model.
|
||||
automatically a learning model / approximator (if not given) by inferring the dimensions of the inputs and outputs
|
||||
of the model.
|
||||
|
||||
In PRL, there are 3 abstraction layers; the learning model, a possible function approximator, and the policy.
|
||||
The learning model is completely independent on concepts such as `State` and `Action` created in this framework.
|
||||
The `Approximator` then combines the `State`, `Action`, and `Model` together. Finally, the `Policy` is just an
|
||||
instance of that `Approximator` where the input is the `State` and the output is the `Action`. Policies can
|
||||
also directly uses the learning models without defining or using an `Approximator`. Approximators make sense when
|
||||
the learning models can be used for other function approximators such as value functions, dynamic transition
|
||||
probability functions, etc. Some models such as movement primitives are too specific and are not general function
|
||||
approximators and thus general approximators for them do not exist. Also, policies have a control update rate in
|
||||
the case we are running in real-time, or a number of time steps they are inactive and return the same action in
|
||||
the case we control when the simulator / environment performs a step.
|
||||
|
||||
Finally, note that in PRL, the policy is the one responsible to execute the action by calling it; i.e. `action()`.
|
||||
This is not performed by the environment nor the approximator.
|
||||
|
||||
.. note::
|
||||
|
||||
Exploration can be carried out by the policy, by specifying the exploration strategy (that is, exploration
|
||||
in the parameter or action space).
|
||||
in the parameter or action space). The Exploration strategy wraps the Policy.
|
||||
|
||||
Example::
|
||||
|
||||
@@ -61,14 +82,11 @@ class Policy(object):
|
||||
world = BasicWorld(simulator)
|
||||
|
||||
# create robot
|
||||
robot = world.loadRobot('robot_name')
|
||||
# or load the robot (via urdf) and spawns it in the simulator
|
||||
#robot = Robot(simulator)
|
||||
#world.loadRobot(robot) # a robot is part of the world (if not done, it will be done inside Env)
|
||||
robot = world.load_robot('robot_name')
|
||||
|
||||
# create states / actions
|
||||
states = JntPositionState(robot) + JntVelocityState(robot)
|
||||
actions = JntPositionAction(robot)
|
||||
states = JointPositionState(robot) + JointVelocityState(robot)
|
||||
actions = JointPositionAction(robot)
|
||||
|
||||
# optional: create learning model (if defined, it has to agree with the dimensions of states/actions)
|
||||
model = NN(...)
|
||||
@@ -77,13 +95,13 @@ class Policy(object):
|
||||
policy = Policy(states, actions, model)
|
||||
|
||||
# create rewards/costs (i.e. r(s,a,s')): gives robot, or state/actions
|
||||
reward = ForWardProgressReward(robot) - FallenCost(robot) - PowerConsumptionCost(robot)
|
||||
reward = ForWardProgressReward(robot) + FallenCost(robot) + PowerConsumptionCost(robot)
|
||||
|
||||
# create environment to interact with
|
||||
env = Env(world, states, rewards)
|
||||
|
||||
# create and run task
|
||||
task = Task(env, policy)
|
||||
# create and run a RL task
|
||||
task = RLTask(env, policy)
|
||||
task.run()
|
||||
|
||||
# Optional: create RL algo (see RL_Algo)
|
||||
@@ -93,31 +111,35 @@ class Policy(object):
|
||||
* `state.py`: describes the various states
|
||||
* `action.py`: describes the various actions
|
||||
* `model.py`: describes the abstract learning model class
|
||||
* `approximator.py`: describe the function approximator class (which is the intermediary layer between
|
||||
the policy and the learning model)
|
||||
* `exploration.py`: describes how to explore using the policy
|
||||
"""
|
||||
__metaclass__ = ABCMeta
|
||||
|
||||
def __init__(self, states, actions, model=None, rate=1, preprocessors=None, postprocessors=None,
|
||||
distribution=None, *args, **kwargs):
|
||||
r"""
|
||||
Initialize a policy, the learning model.
|
||||
Initialize a policy (and the inner approximator / learning model).
|
||||
|
||||
Args:
|
||||
states (State): By giving the `states` to the policy, it can automatically infer the type and size/shape of
|
||||
each state, and thus can be used to automatically build a policy. At each step, the `states`
|
||||
are 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.
|
||||
actions (Action): At each step, by calling `policy.act(state)`, the `actions` are computed by the policy,
|
||||
and should be given to the environment. As with the `states`, the type and size/shape of
|
||||
each 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.
|
||||
model (Model, Approximator, None): inner model or approximator
|
||||
rate (int): rate at which the policy operates
|
||||
and should be given to the environment. As with the `states`, the type and size/shape of each 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.
|
||||
states (State): By giving the `states` to the policy, it can automatically infer the type and size/shape
|
||||
of each state, and thus can be used to automatically build a policy. At each step, the `states`
|
||||
are 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 (Approximator, Model, None): inner model or approximator
|
||||
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
|
||||
distribution:
|
||||
args:
|
||||
kwargs:
|
||||
*args (list): list of arguments
|
||||
**kwargs (dict): dictionary of arguments
|
||||
"""
|
||||
self.states = states
|
||||
self.actions = actions
|
||||
@@ -125,7 +147,20 @@ class Policy(object):
|
||||
self.train_mode = False
|
||||
self.rate = rate
|
||||
self.cnt = 0
|
||||
self.last_action = None
|
||||
self.action_data = None
|
||||
|
||||
# preprocessors and postprocessors
|
||||
if preprocessors is None:
|
||||
preprocessors = []
|
||||
if not isinstance(preprocessors, collections.Iterable):
|
||||
preprocessors = [preprocessors]
|
||||
self.preprocessors = preprocessors
|
||||
|
||||
if postprocessors is None:
|
||||
postprocessors = []
|
||||
if not isinstance(postprocessors, collections.Iterable):
|
||||
postprocessors = [postprocessors]
|
||||
self.postprocessors = postprocessors
|
||||
|
||||
##############
|
||||
# Properties #
|
||||
@@ -133,10 +168,12 @@ class Policy(object):
|
||||
|
||||
@property
|
||||
def states(self):
|
||||
"""Return the states."""
|
||||
return self._states
|
||||
|
||||
@states.setter
|
||||
def states(self, states):
|
||||
"""Set the states."""
|
||||
if states is not None:
|
||||
if not isinstance(states, State):
|
||||
raise TypeError("Expecting states to be an instance of State.")
|
||||
@@ -144,26 +181,29 @@ class Policy(object):
|
||||
|
||||
@property
|
||||
def actions(self):
|
||||
"""Return the actions."""
|
||||
return self._actions
|
||||
|
||||
@actions.setter
|
||||
def actions(self, actions):
|
||||
"""Set the actions."""
|
||||
if not isinstance(actions, Action):
|
||||
raise TypeError("Expecting actions to be an instance of Action.")
|
||||
self._actions = actions
|
||||
|
||||
@property
|
||||
def model(self):
|
||||
"""Return the inner approximator / learning model."""
|
||||
return self._model
|
||||
|
||||
@model.setter
|
||||
def model(self, model):
|
||||
"""Set the inner approximator / learning model."""
|
||||
if model is not None and not isinstance(model, Approximator):
|
||||
# Try to wrap it with the corresponding Approximator
|
||||
if isinstance(model, Model):
|
||||
model = Approximator(inputs=self.states, outputs=self.actions, model=model)
|
||||
elif isinstance(model, torch.nn.Module):
|
||||
model = NNApproximator(inputs=self.states, outputs=self.actions, model=model)
|
||||
# preprocessors=self.preprocessors, postprocessors=self.postprocessors)
|
||||
# else:
|
||||
# raise TypeError("Expecting the model to be an instance of Model.")
|
||||
self._model = model
|
||||
@@ -181,40 +221,34 @@ class Policy(object):
|
||||
self._rate = rate
|
||||
|
||||
@property
|
||||
def parameters(self):
|
||||
"""
|
||||
Return an iterator over the learning model parameters.
|
||||
"""
|
||||
if self.model is None:
|
||||
return None
|
||||
return self.model.parameters()
|
||||
def input_size(self):
|
||||
"""Return the policy input size."""
|
||||
return self.model.input_size
|
||||
|
||||
@property
|
||||
def hyperparameters(self):
|
||||
"""
|
||||
Return an iterator over the learning model hyperparameters.
|
||||
"""
|
||||
if self.model is None:
|
||||
return None
|
||||
return self.model.hyperparameters()
|
||||
def output_size(self):
|
||||
"""Return the policy output size."""
|
||||
return self.model.output_size
|
||||
|
||||
@property
|
||||
def input_dims(self):
|
||||
"""
|
||||
Return the input dimension of the policy.
|
||||
"""
|
||||
if self.model is None:
|
||||
return None
|
||||
return self.model.get_input_dims()
|
||||
def input_shape(self):
|
||||
"""Return the policy input shape."""
|
||||
return self.model.input_shape
|
||||
|
||||
@property
|
||||
def output_dims(self):
|
||||
"""
|
||||
Return the output dimension of the policy.
|
||||
"""
|
||||
if self.model is None:
|
||||
return None
|
||||
return self.model.get_output_dims()
|
||||
def output_shape(self):
|
||||
"""Return the policy output shape."""
|
||||
return self.model.output_shape
|
||||
|
||||
@property
|
||||
def input_dim(self):
|
||||
"""Return the input dimension of the policy; i.e. len(input_shape)."""
|
||||
return self.model.input_dim
|
||||
|
||||
@property
|
||||
def output_dim(self):
|
||||
"""Return the output dimension of the policy; i.e. len(output_shape)."""
|
||||
return self.model.output_dim
|
||||
|
||||
@property
|
||||
def num_parameters(self):
|
||||
@@ -225,25 +259,43 @@ class Policy(object):
|
||||
# Methods #
|
||||
###########
|
||||
|
||||
def _size(self, items):
|
||||
"""Compute the size of the given argument :attr:`items`."""
|
||||
size = 0
|
||||
if not isinstance(items, (list, tuple)):
|
||||
items = [items]
|
||||
for item in items:
|
||||
if isinstance(item, (State, Action)):
|
||||
for element in item:
|
||||
if element.is_discrete():
|
||||
size += element.space[0].n
|
||||
else:
|
||||
size += element.total_size()
|
||||
elif isinstance(item, np.ndarray):
|
||||
size += item.size
|
||||
elif isinstance(item, torch.Tensor):
|
||||
size += item.numel()
|
||||
elif isinstance(item, int):
|
||||
size += item
|
||||
return size
|
||||
|
||||
def is_deterministic(self):
|
||||
"""
|
||||
Return True if the policy is deterministic; that is, given the same states result in the same actions.
|
||||
.. math:: a_t = f(s_t)
|
||||
|
||||
Returns:
|
||||
bool: True if the policy is deterministic
|
||||
"""
|
||||
return self.model.is_deterministic()
|
||||
|
||||
def is_stochastic(self):
|
||||
def is_probabilistic(self):
|
||||
"""
|
||||
Return True if the policy is stochastic; that is, given the same states can result in different actions.
|
||||
.. math:: a_t ~ p(a_t|s_t)
|
||||
|
||||
Returns:
|
||||
bool: True if the policy is stochastic
|
||||
"""
|
||||
return self.model.is_stochastic()
|
||||
return self.model.is_probabilistic()
|
||||
|
||||
def is_parametric(self):
|
||||
"""
|
||||
@@ -274,39 +326,184 @@ class Policy(object):
|
||||
"""
|
||||
raise self.model.is_recurrent()
|
||||
|
||||
def parameters(self):
|
||||
"""
|
||||
Return an iterator over the learning model parameters.
|
||||
"""
|
||||
if self.model is None:
|
||||
return []
|
||||
return self.model.parameters()
|
||||
|
||||
def named_parameters(self):
|
||||
"""
|
||||
Return an iterator over the learning model parameters; yielding both the name and the parameter itself.
|
||||
"""
|
||||
if self.model is None:
|
||||
return []
|
||||
return self.model.named_parameters()
|
||||
|
||||
def list_parameters(self):
|
||||
"""
|
||||
Return the learning model parameters.
|
||||
"""
|
||||
if self.model is None:
|
||||
return []
|
||||
return self.model.list_parameters()
|
||||
|
||||
def hyperparameters(self):
|
||||
"""
|
||||
Return an iterator over the learning model hyper-parameters.
|
||||
"""
|
||||
if self.model is None:
|
||||
return []
|
||||
return self.model.hyperparameters()
|
||||
|
||||
def named_hyperparameters(self):
|
||||
"""
|
||||
Return an iterator over the learning model hyper-parameters; yielding both the name and the hyper-parameter
|
||||
itself.
|
||||
"""
|
||||
if self.model is None:
|
||||
return []
|
||||
return self.model.named_hyperparameters()
|
||||
|
||||
def list_hyperparameters(self):
|
||||
"""
|
||||
Return the learning model hyper-parameters
|
||||
"""
|
||||
if self.model is None:
|
||||
return None
|
||||
return self.model.list_hyperparameters()
|
||||
|
||||
def get_vectorized_parameters(self, to_numpy=True):
|
||||
"""Get the parameters in a vectorized form."""
|
||||
return self.model.get_vectorized_parameters(to_numpy=to_numpy)
|
||||
|
||||
def set_vectorized_parameters(self, vector):
|
||||
"""Set the vectorized parameters."""
|
||||
self.model.set_vectorized_parameters(vector=vector)
|
||||
|
||||
@abstractmethod
|
||||
def act(self, state, deterministic=True, to_numpy=True):
|
||||
"""
|
||||
Perform the action given the state.
|
||||
def __convert_to_numpy(self, x, to_numpy=True):
|
||||
"""Convert the given argument to a numpy array if specified."""
|
||||
if to_numpy and isinstance(x, torch.Tensor):
|
||||
if x.requires_grad:
|
||||
return x.detach().numpy()
|
||||
return x.numpy()
|
||||
return x
|
||||
|
||||
def _predict(self, state, to_numpy=False, return_logits=True, set_output_data=False):
|
||||
"""Inner prediction step."""
|
||||
if isinstance(self.model, Approximator): # inner model is an approximator
|
||||
return self.model.predict(state, to_numpy=to_numpy, return_logits=return_logits,
|
||||
set_output_data=set_output_data)
|
||||
# inner model is a learning model
|
||||
return self.model.predict(state, to_numpy=to_numpy)
|
||||
|
||||
def act(self, state=None, deterministic=True, to_numpy=True, return_logits=False, apply_action=True):
|
||||
"""Perform the action given the state.
|
||||
|
||||
Args:
|
||||
state (State): current state
|
||||
deterministic (bool): True by default. It can only be set to False, if the policy is stochastic.
|
||||
to_numpy (bool): if True, return a np.array
|
||||
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.
|
||||
apply_action (bool): If True, it will call and execute the action.
|
||||
|
||||
Returns:
|
||||
Action: action
|
||||
(list of) np.array / torch.Tensor: action data
|
||||
"""
|
||||
if self.model is not None:
|
||||
if (self.cnt % self.rate) == 0:
|
||||
self.last_action = self.model.predict(state, to_numpy=to_numpy)
|
||||
self.cnt += 1
|
||||
return self.last_action
|
||||
# predict = act
|
||||
# if we should predict
|
||||
if (self.cnt % self.rate) == 0:
|
||||
|
||||
@abstractmethod
|
||||
def sample(self, state):
|
||||
# if no input is given, take the provided inputs at the beginning
|
||||
if state is None:
|
||||
state = self.states
|
||||
|
||||
# if the input is an instance of State, get the inner merged data.
|
||||
if isinstance(state, State):
|
||||
state = state.merged_data
|
||||
if len(state) == 1:
|
||||
state = state[0]
|
||||
|
||||
# go through each preprocessor
|
||||
for processor in self.preprocessors:
|
||||
state = processor(state)
|
||||
|
||||
# predict the output using the inner model
|
||||
self.action_data = self._predict(state, to_numpy=False, return_logits=True, set_output_data=False)
|
||||
# if isinstance(self.model, Approximator): # inner model is an approximator
|
||||
# self.action_data = self.model.predict(state, to_numpy=False, return_logits=True,
|
||||
# set_output_data=False)
|
||||
# else: # inner model is a learning model
|
||||
# self.action_data = self.model.predict(state, to_numpy=to_numpy)
|
||||
|
||||
# go through each postprocessor
|
||||
for processor in self.postprocessors:
|
||||
self.action_data = processor(self.action_data)
|
||||
|
||||
# set the action data
|
||||
|
||||
# if action data is not a list, make it a list as we will iterate through it
|
||||
if not isinstance(self.action_data, list):
|
||||
self.action_data = [self.action_data]
|
||||
|
||||
# go through each action and data
|
||||
for idx, (action, data) in enumerate(zip(self.actions, self.action_data)):
|
||||
if action.is_discrete(): # discrete action
|
||||
if isinstance(data, np.ndarray): # data action is a numpy array
|
||||
discrete_data = np.array([np.argmax(data)])
|
||||
action.data = discrete_data
|
||||
if not return_logits:
|
||||
self.action_data[idx] = discrete_data
|
||||
elif isinstance(data, torch.Tensor): # data action is a torch.Tensor
|
||||
discrete_data = torch.argmax(data, dim=0, keepdim=True)
|
||||
action.torch_data = discrete_data
|
||||
if not return_logits:
|
||||
self.action_data[idx] = self.__convert_to_numpy(discrete_data, to_numpy=to_numpy)
|
||||
else:
|
||||
self.action_data[idx] = self.__convert_to_numpy(data, to_numpy=to_numpy)
|
||||
# elif isinstance(data, (float, int)):
|
||||
# discrete_data = np.argmax(data)
|
||||
# action.data = discrete_data
|
||||
# if not return_logits:
|
||||
# self.action_data[idx] = discrete_data
|
||||
else:
|
||||
raise TypeError(
|
||||
"Expecting the `data` action to be a numpy array or torch.Tensor, instead got: "
|
||||
"{}".format(type(data)))
|
||||
else: # continuous action
|
||||
if isinstance(data, np.ndarray):
|
||||
action.data = data
|
||||
elif isinstance(data, torch.Tensor):
|
||||
action.torch_data = data
|
||||
self.action_data[idx] = self.__convert_to_numpy(data, to_numpy=to_numpy)
|
||||
# elif isinstance(data, (float, int)):
|
||||
# pass
|
||||
else:
|
||||
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(self.action_data, list) and len(self.action_data) == 1:
|
||||
self.action_data = self.action_data[0]
|
||||
|
||||
# apply action
|
||||
if apply_action:
|
||||
self.actions()
|
||||
|
||||
# increment tick counter
|
||||
self.cnt += 1
|
||||
|
||||
# return the action data
|
||||
return self.action_data
|
||||
|
||||
def sample(self, state=None):
|
||||
"""
|
||||
Given the state, sample from the policy. This only works if the inner model of the policy is stochastic.
|
||||
|
||||
Args:
|
||||
state (State, array): current state
|
||||
state (State, array, tensor, None): current state
|
||||
|
||||
Returns:
|
||||
array: sample
|
||||
@@ -319,43 +516,25 @@ class Policy(object):
|
||||
|
||||
Args:
|
||||
mode (bool): if True, set the policy in train mode.
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
self.train_mode = mode
|
||||
|
||||
def reset(self, *args, **kwargs):
|
||||
def reset(self, reset_processors=False, *args, **kwargs):
|
||||
"""
|
||||
Reset the policy.
|
||||
"""
|
||||
for processor in self.preprocessors:
|
||||
processor.reset()
|
||||
for processor in self.postprocessors:
|
||||
processor.reset()
|
||||
self.model.reset()
|
||||
|
||||
def get_params(self):
|
||||
"""
|
||||
Return the learning model parameters.
|
||||
"""
|
||||
if self.model is None:
|
||||
return None
|
||||
return self.model.get_params()
|
||||
|
||||
def get_hyperparams(self):
|
||||
"""
|
||||
Return the learning model hyperparameters
|
||||
"""
|
||||
if self.model is None:
|
||||
return None
|
||||
return self.model.get_hyperparams()
|
||||
|
||||
def save(self, filename):
|
||||
"""
|
||||
Save the policy in the given filename.
|
||||
|
||||
Args:
|
||||
filename (str): file to save the policy into
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
# self.model.save(filename)
|
||||
pickle.dump(self, open(filename, 'wb'))
|
||||
@@ -369,7 +548,7 @@ class Policy(object):
|
||||
filename (str): file to load the policy from
|
||||
|
||||
Returns:
|
||||
None
|
||||
Policy: the policy
|
||||
"""
|
||||
# self.model.load(filename)
|
||||
return pickle.load(open(filename, 'rb'))
|
||||
@@ -378,8 +557,31 @@ class Policy(object):
|
||||
# Operators #
|
||||
#############
|
||||
|
||||
def __call__(self, *args, **kwargs):
|
||||
def __call__(self, state=None, deterministic=True, to_numpy=True, return_logits=False, apply_action=True):
|
||||
"""Perform the action given the state.
|
||||
|
||||
Args:
|
||||
state (State): current state
|
||||
deterministic (bool): True by default. It can only be set to False, if the policy is stochastic.
|
||||
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.
|
||||
apply_action (bool): If True, it will call and execute the action.
|
||||
|
||||
Returns:
|
||||
Action: action
|
||||
"""
|
||||
return self.act(*args, **kwargs)
|
||||
|
||||
def __repr__(self):
|
||||
"""Return representation of python object."""
|
||||
if self.__class__.__name__ == 'Policy':
|
||||
if self.model is not None:
|
||||
return "{}({})".format(self.__class__.__name__, self.model.__str__())
|
||||
return self.__class__.__name__
|
||||
|
||||
def __str__(self):
|
||||
return self.model.__str__()
|
||||
"""Return string describing the policy."""
|
||||
if self.__class__.__name__ == 'Policy':
|
||||
if self.model is not None:
|
||||
return "{}({})".format(self.__class__.__name__, self.model.__str__())
|
||||
return self.__class__.__name__
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
|
||||
# import state generators
|
||||
from state_generator import *
|
||||
from .state_generator import *
|
||||
|
||||
@@ -763,23 +763,34 @@ class State(object):
|
||||
it checks that it is within the bounds.
|
||||
|
||||
Args:
|
||||
item (State, list/tuple of state): check if given state(s) is(are) in the combined state
|
||||
item (State, list/tuple of state, type): check if given state(s) is(are) in the combined state
|
||||
|
||||
Example:
|
||||
s1 = JntPositionState(robot)
|
||||
s2 = JntVelocityState(robot)
|
||||
s1 = JointPositionState(robot)
|
||||
s2 = JointVelocityState(robot)
|
||||
s = s1 + s2
|
||||
print(s1 in s) # output True
|
||||
print(s2 in s1) # output False
|
||||
print((s1, s2) in s) # output True
|
||||
"""
|
||||
# check type of item
|
||||
if not isinstance(item, (State, np.ndarray)):
|
||||
raise TypeError("Expecting a state or numpy array.")
|
||||
if not isinstance(item, (State, np.ndarray, type)):
|
||||
raise TypeError("Expecting a State, np.array, or a class type, instead got: {}".format(type(item)))
|
||||
|
||||
# if class type
|
||||
if isinstance(item, type):
|
||||
# if there is one state
|
||||
if self.has_data():
|
||||
return self.__class__ == item
|
||||
# the state has multiple states, thus we go through each state
|
||||
for state in self.states:
|
||||
if state.__class__ == item:
|
||||
return True
|
||||
return False
|
||||
|
||||
# check if state item is in the combined state
|
||||
if self._data is None and isinstance(item, State):
|
||||
return (item in self._states)
|
||||
return item in self._states
|
||||
|
||||
# check if state/data is within the bounds
|
||||
if isinstance(item, State):
|
||||
|
||||
@@ -0,0 +1,7 @@
|
||||
## Value function approximators
|
||||
|
||||
This folder contains the various value function approximators.
|
||||
|
||||
## What to look/check next?
|
||||
|
||||
You can have a look at `policies`, `approximators`, `actorcritics`, and `models`.
|
||||
@@ -0,0 +1,9 @@
|
||||
|
||||
# import values
|
||||
from .value import *
|
||||
|
||||
# import basic value function approximator (such as tables and linear)
|
||||
from .basic_value import *
|
||||
|
||||
# import NN value function approximators
|
||||
from .nn_value import *
|
||||
@@ -0,0 +1,97 @@
|
||||
#!/usr/bin/env python
|
||||
"""Provides the various basic value function approximators (e.g. table and linear value approximators)
|
||||
"""
|
||||
|
||||
from abc import ABCMeta
|
||||
import torch
|
||||
|
||||
# from pyrobolearn.models import Linear
|
||||
from pyrobolearn.approximators import LinearApproximator
|
||||
from pyrobolearn.values.value import ValueApproximator, ParametrizedValue, ParametrizedStateActionValue, \
|
||||
ParametrizedStateOutputActionValue
|
||||
|
||||
|
||||
__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 ValueTable(ValueApproximator):
|
||||
r"""Value Table
|
||||
|
||||
This is appropriate when the states/actions are discrete and have a low dimension.
|
||||
|
||||
Dynamic Programming is used to compute the table.
|
||||
"""
|
||||
|
||||
def __init__(self, state):
|
||||
super(ValueTable, self).__init__(state)
|
||||
|
||||
|
||||
class LinearStateValue(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(LinearStateValue, self).__init__(state, model=model)
|
||||
|
||||
|
||||
class LinearStateInputActionValue(ParametrizedStateActionValue):
|
||||
r"""Linear state - input action value function approximator
|
||||
|
||||
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(LinearStateInputActionValue, self).__init__(state, action, model=model)
|
||||
|
||||
|
||||
class LinearStateOutputActionValue(ParametrizedStateOutputActionValue):
|
||||
r"""Linear state - output action value function approximator
|
||||
|
||||
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(LinearStateOutputActionValue, self).__init__(state, action, model=model)
|
||||
@@ -0,0 +1,198 @@
|
||||
#!/usr/bin/env python
|
||||
"""Provides value function approximators based on neural networks.
|
||||
|
||||
For instance, a value network is a function represented by a neural network that maps a state to a value (real number).
|
||||
"""
|
||||
|
||||
from abc import ABCMeta
|
||||
import torch
|
||||
|
||||
from pyrobolearn.models import NN
|
||||
from pyrobolearn.approximators import NNApproximator, MLPApproximator
|
||||
from pyrobolearn.values.value import ParametrizedValue, ParametrizedStateActionValue, ParametrizedStateOutputActionValue
|
||||
|
||||
|
||||
__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 ValueNetwork(ParametrizedValue):
|
||||
# r"""Value Network
|
||||
#
|
||||
# Use a neural network to approximate the value function.
|
||||
# """
|
||||
# __metaclass__ = ABCMeta
|
||||
#
|
||||
# def __init__(self, state, model):
|
||||
# super(ValueNetwork, self).__init__(state)
|
||||
#
|
||||
# # Check the given model
|
||||
# if not isinstance(model, NNApproximator):
|
||||
# if isinstance(model, NN):
|
||||
# model = NNApproximator(state, torch.Tensor([1]), model)
|
||||
# else:
|
||||
# if isinstance(model, torch.nn.Module):
|
||||
# model = NN(model)
|
||||
# model = NNApproximator(state, torch.Tensor([1]), model)
|
||||
# else:
|
||||
# raise TypeError("The model for the neural network is not an instance of model.NN or "
|
||||
# "torch.nn.Module")
|
||||
#
|
||||
# self.model = model
|
||||
|
||||
|
||||
class StateValueNetwork(ParametrizedValue):
|
||||
r"""Stave Value Network
|
||||
|
||||
This is defined by :math:`V_{\psi}(s_t)` where :math:`\psi` represents the network parameters.
|
||||
"""
|
||||
|
||||
def __init__(self, state, model):
|
||||
"""
|
||||
Initialize the NN state value function approximator.
|
||||
|
||||
Args:
|
||||
state (State): input state.
|
||||
model (NN, NNApproximator): Neural Network model / approximator.
|
||||
"""
|
||||
super(StateValueNetwork, self).__init__(state, model)
|
||||
|
||||
|
||||
class StateInputActionValueNetwork(ParametrizedStateActionValue):
|
||||
r"""State Input Action Value Network
|
||||
|
||||
State-action value function :math:`Q_{\phi}(s, a)` approximated by a neural network, 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, model):
|
||||
"""
|
||||
Initialize the NN state-action value function approximator.
|
||||
|
||||
Args:
|
||||
state (State): input state.
|
||||
action (Action): input action.
|
||||
model (NN, NNApproximator): Neural Network model / approximator.
|
||||
"""
|
||||
super(StateInputActionValueNetwork, self).__init__(state, action, model)
|
||||
|
||||
|
||||
class StateOutputActionValueNetwork(ParametrizedStateOutputActionValue):
|
||||
r"""State Output Action Value Network
|
||||
|
||||
State-action value function :math:`Q_{\phi}(s, a)` approximated by a neural network, 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, model):
|
||||
"""
|
||||
Initialize the NN state-action value function approximator.
|
||||
|
||||
Args:
|
||||
state (State): input state.
|
||||
action (Action): output action.
|
||||
model (NN, NNApproximator): Neural Network model / approximator.
|
||||
"""
|
||||
super(StateOutputActionValueNetwork, self).__init__(state, action, model)
|
||||
|
||||
|
||||
class MLPStateValue(ParametrizedValue): # StateValueNetwork):
|
||||
r"""Multi-Layer Perceptron (MLP) State Value Function Approximator
|
||||
|
||||
This is defined by :math:`V_{\psi}(s_t)` where the function :math:`V` is approximated by a multilayer perceptron.
|
||||
"""
|
||||
|
||||
def __init__(self, state, hidden_units=(), activation_fct='linear', last_activation_fct=None, dropout_prob=None,
|
||||
preprocessors=None):
|
||||
"""Initialize the Value MLP approximator.
|
||||
|
||||
Args:
|
||||
state (State): 1D-states that is feed to the policy (the input dimensions will be inferred from the
|
||||
states)
|
||||
hidden_units (list/tuple of int): number of hidden units in the corresponding layer
|
||||
activation_fct (None, str, or list/tuple of str/None): activation function to be applied after each layer.
|
||||
If list/tuple, then it has to match the
|
||||
last_activation_fct (None or str): last activation function to be applied. If not specified, it will check
|
||||
if it is in the list/tuple of activation functions provided for the
|
||||
previous argument.
|
||||
dropout_prob (None, float, or list/tuple of float/None): dropout probability.
|
||||
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
|
||||
the inner model / function approximator.
|
||||
"""
|
||||
output = torch.Tensor([1.]) # torch.Tensor([[1.]])
|
||||
model = MLPApproximator(state, output, hidden_units=hidden_units, activation_fct=activation_fct,
|
||||
last_activation_fct=last_activation_fct, dropout_prob=dropout_prob,
|
||||
preprocessors=preprocessors)
|
||||
super(MLPStateValue, self).__init__(state, model)
|
||||
|
||||
|
||||
class MLPStateInputActionValue(ParametrizedStateActionValue):
|
||||
r"""MLP state - input action value function approximator
|
||||
|
||||
State-action value function :math:`Q_{\phi}(s, a)` approximated by a MLP 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, hidden_units=(), activation_fct='linear', last_activation_fct=None,
|
||||
dropout_prob=None, preprocessors=None):
|
||||
"""
|
||||
Initialize the MLP state-action value function approximator.
|
||||
|
||||
Args:
|
||||
state (State): input state.
|
||||
action (Action): input action.
|
||||
hidden_units (list/tuple of int): number of hidden units in the corresponding layer
|
||||
activation_fct (None, str, or list/tuple of str/None): activation function to be applied after each layer.
|
||||
If list/tuple, then it has to match the
|
||||
last_activation_fct (None or str): last activation function to be applied. If not specified, it will check
|
||||
if it is in the list/tuple of activation functions provided for the
|
||||
previous argument.
|
||||
dropout_prob (None, float, or list/tuple of float/None): dropout probability.
|
||||
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
|
||||
the inner model / function approximator.
|
||||
"""
|
||||
model = MLPApproximator(inputs=[state, action], outputs=torch.Tensor([1]), hidden_units=hidden_units,
|
||||
activation_fct=activation_fct, last_activation_fct=last_activation_fct,
|
||||
dropout_prob=dropout_prob, preprocessors=preprocessors)
|
||||
super(MLPStateInputActionValue, self).__init__(state, action, model=model)
|
||||
|
||||
|
||||
class MLPStateOutputActionValue(ParametrizedStateOutputActionValue):
|
||||
r"""MLP state - output action value function approximator
|
||||
|
||||
State-action value function :math:`Q_{\phi}(s, a)` approximated by a MLP 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, hidden_units=(), activation_fct='linear', last_activation_fct=None,
|
||||
dropout_prob=None, preprocessors=None):
|
||||
"""
|
||||
Initialize the MLP state-action value function approximator.
|
||||
|
||||
Args:
|
||||
state (State): input state.
|
||||
action (Action): output action.
|
||||
hidden_units (list/tuple of int): number of hidden units in the corresponding layer
|
||||
activation_fct (None, str, or list/tuple of str/None): activation function to be applied after each layer.
|
||||
If list/tuple, then it has to match the
|
||||
last_activation_fct (None or str): last activation function to be applied. If not specified, it will check
|
||||
if it is in the list/tuple of activation functions provided for the
|
||||
previous argument.
|
||||
dropout_prob (None, float, or list/tuple of float/None): dropout probability.
|
||||
preprocessors ((list of) Processor): pre-processors to be applied on the input state before being fed to
|
||||
the inner model / function approximator.
|
||||
"""
|
||||
model = MLPApproximator(inputs=state, outputs=action, hidden_units=hidden_units,
|
||||
activation_fct=activation_fct, last_activation_fct=last_activation_fct,
|
||||
dropout_prob=dropout_prob, preprocessors=preprocessors)
|
||||
super(MLPStateOutputActionValue, self).__init__(state, action, model=model)
|
||||
@@ -0,0 +1,338 @@
|
||||
#!/usr/bin/env python
|
||||
"""Provides the various abstract value function approximators.
|
||||
|
||||
It can use learning models as most of these models learn a function (that is how to map inputs to outputs).
|
||||
For instance, a value network is a function represented by a neural network that maps a state to a value (real number).
|
||||
|
||||
Dependencies:
|
||||
- `pyrobolearn.states`
|
||||
- `pyrobolearn.actions`
|
||||
- `pyrobolearn.approximators` (and thus `pyrobolearn.models`)
|
||||
"""
|
||||
|
||||
from abc import ABCMeta, abstractmethod
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
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 ValueApproximator(object):
|
||||
r"""Value Function Approximator
|
||||
|
||||
In reinforcement learning, this is known as value-based reinforcement learning. The other end being policy
|
||||
search. Several methods fall in the spectrum between these 2 approaches. For instance, actor-critic methods
|
||||
optimize a policy (aka the actor) using the value function (aka critic).
|
||||
|
||||
As for the Policy (see `policy.py`), the value function approximator groups a model (which can be learned as a
|
||||
neural network, or built as a table) with the states and actions.
|
||||
|
||||
There are mainly two types of value function:
|
||||
1. state value function denoted by :math:`V(s)` which represents the expected reward accumulated when
|
||||
starting at state :math:`s`.
|
||||
2. (state-)action value function denoted by :math:`Q(s,a)`, which represents the expected reward when
|
||||
starting at state :math:`s`, and performing action :math:`a`.
|
||||
"""
|
||||
__metaclass__ = ABCMeta
|
||||
|
||||
def __init__(self, state):
|
||||
"""
|
||||
Initialize the value function approximator.
|
||||
|
||||
Args:
|
||||
state (State, np.array, torch.Tensor): state input
|
||||
"""
|
||||
self.state = state
|
||||
self.value = None
|
||||
|
||||
##############
|
||||
# Properties #
|
||||
##############
|
||||
|
||||
@property
|
||||
def state(self):
|
||||
"""Return the state instance."""
|
||||
return self._state
|
||||
|
||||
@state.setter
|
||||
def state(self, state):
|
||||
"""Set the state input."""
|
||||
if isinstance(state, (int, float)):
|
||||
state = np.array([state])
|
||||
elif not isinstance(state, (State, torch.Tensor, np.ndarray)):
|
||||
raise TypeError("Expecting the state to be a State, torch.Tensor, or np.ndarray.")
|
||||
self._state = state
|
||||
|
||||
###########
|
||||
# Methods #
|
||||
###########
|
||||
|
||||
def compute(self, *args, **kwargs):
|
||||
"""Predict the value."""
|
||||
pass
|
||||
|
||||
def __call__(self, *args, **kwargs):
|
||||
"""Predict the value."""
|
||||
return self.compute(*args, **kwargs)
|
||||
|
||||
|
||||
class StateValueApproximator(ValueApproximator):
|
||||
r"""State Value Approximator
|
||||
|
||||
Compute :math:`V(s)`.
|
||||
"""
|
||||
__metaclass__ = ABCMeta
|
||||
|
||||
def __init__(self, state):
|
||||
super(StateValueApproximator, self).__init__(state)
|
||||
|
||||
|
||||
class ActionValueApproximator(ValueApproximator):
|
||||
r"""Action Value Approximator
|
||||
|
||||
Compute :math:`Q(s,a)`.
|
||||
"""
|
||||
__metaclass__ = ABCMeta
|
||||
|
||||
def __init__(self, state, actions):
|
||||
super(ActionValueApproximator, self).__init__(state)
|
||||
self.actions = actions
|
||||
|
||||
|
||||
class ParametrizedValue(ValueApproximator):
|
||||
r"""Parametrized (Learnable) Value Function Approximator
|
||||
|
||||
This value function approximator has parameters that can be optimized. By default it predicts :math:`V_{\phi}(s)`.
|
||||
"""
|
||||
|
||||
def __init__(self, state, model):
|
||||
"""
|
||||
Initialize the parametrized value function approximator.
|
||||
|
||||
Args:
|
||||
state (State): input state.
|
||||
model (Approximator): value function approximator.
|
||||
"""
|
||||
super(ParametrizedValue, self).__init__(state)
|
||||
self.model = model
|
||||
|
||||
##############
|
||||
# Properties #
|
||||
##############
|
||||
|
||||
@property
|
||||
def model(self):
|
||||
"""Return the model instance."""
|
||||
return self._model
|
||||
|
||||
@model.setter
|
||||
def model(self, model):
|
||||
"""Set the model / approximator instance."""
|
||||
if not isinstance(model, Approximator):
|
||||
raise TypeError("Expecting the model to be an instance of `Approximator`, instead got: "
|
||||
"{}".format(type(model)))
|
||||
self._model = model
|
||||
|
||||
@property
|
||||
def input_size(self):
|
||||
"""Return the policy input size."""
|
||||
return self.model.input_size
|
||||
|
||||
@property
|
||||
def output_size(self):
|
||||
"""Return the policy output size."""
|
||||
return self.model.output_size
|
||||
|
||||
@property
|
||||
def input_shape(self):
|
||||
"""Return the policy input shape."""
|
||||
return self.model.input_shape
|
||||
|
||||
@property
|
||||
def output_shape(self):
|
||||
"""Return the policy output shape."""
|
||||
return self.model.output_shape
|
||||
|
||||
@property
|
||||
def input_dim(self):
|
||||
"""Return the input dimension of the policy; i.e. len(input_shape)."""
|
||||
return self.model.input_dim
|
||||
|
||||
@property
|
||||
def output_dim(self):
|
||||
"""Return the output dimension of the policy; i.e. len(output_shape)."""
|
||||
return self.model.output_dim
|
||||
|
||||
@property
|
||||
def num_parameters(self):
|
||||
"""Return the total number of parameters"""
|
||||
return self.model.num_parameters
|
||||
|
||||
###########
|
||||
# Methods #
|
||||
###########
|
||||
|
||||
def parameters(self):
|
||||
"""
|
||||
Return an iterator over the learning model parameters.
|
||||
"""
|
||||
return self.model.parameters()
|
||||
|
||||
def named_parameters(self):
|
||||
"""
|
||||
Return an iterator over the learning model parameters; yielding both the name and the parameter itself.
|
||||
"""
|
||||
return self.model.named_parameters()
|
||||
|
||||
def list_parameters(self):
|
||||
"""
|
||||
Return the learning model parameters.
|
||||
"""
|
||||
return self.model.list_parameters()
|
||||
|
||||
def hyperparameters(self):
|
||||
"""
|
||||
Return an iterator over the learning model hyper-parameters.
|
||||
"""
|
||||
return self.model.hyperparameters()
|
||||
|
||||
def named_hyperparameters(self):
|
||||
"""
|
||||
Return an iterator over the learning model hyper-parameters; yielding both the name and the hyper-parameter
|
||||
itself.
|
||||
"""
|
||||
return self.model.named_hyperparameters()
|
||||
|
||||
def list_hyperparameters(self):
|
||||
"""
|
||||
Return the learning model hyper-parameters
|
||||
"""
|
||||
return self.model.list_hyperparameters()
|
||||
|
||||
def get_vectorized_parameters(self, to_numpy=True):
|
||||
"""
|
||||
Get the parameters in a vectorized form.
|
||||
|
||||
Args:
|
||||
to_numpy (bool): if True, it will convert the 1D parameter vector into a numpy array.
|
||||
"""
|
||||
return self.model.get_vectorized_parameters(to_numpy=to_numpy)
|
||||
|
||||
def set_vectorized_parameters(self, vector):
|
||||
"""
|
||||
Set the vectorized parameters.
|
||||
|
||||
Args:
|
||||
np.array, torch.Tensor: 1D parameter vector.
|
||||
"""
|
||||
self.model.set_vectorized_parameters(vector=vector)
|
||||
|
||||
def compute(self, state=None, to_numpy=True):
|
||||
"""Compute the output of the value function.
|
||||
|
||||
Args:
|
||||
state (None, State, (list of) np.array, (list of) torch.Tensor): state input data. If None, it will get
|
||||
the data from the inputs that were given at the initialization.
|
||||
to_numpy (bool): If True, it will convert the data (torch.Tensors) to numpy arrays.
|
||||
"""
|
||||
self.value = self.model.predict(state, to_numpy=to_numpy, return_logits=True, set_output_data=False)
|
||||
return self.value
|
||||
|
||||
def __call__(self, state=None, to_numpy=True):
|
||||
"""Predict the value."""
|
||||
return self.compute(state=state, to_numpy=to_numpy)
|
||||
|
||||
|
||||
class ParametrizedStateActionValue(ParametrizedValue):
|
||||
r"""Parametrized State Action Value Function Approximator
|
||||
|
||||
This value function approximator predicts the state-action value :math:`Q_{\phi}(s,a)`. Two different kind of
|
||||
state-action value function approximators can be defined:
|
||||
- where the action is given as input to the value function approximator. The action can be continuous or discrete.
|
||||
- where the action is given as output to the value function approximator. The output of such model is the
|
||||
state-action value for each DISCRETE action.
|
||||
|
||||
By default, this class implements the value function approximator :math:`Q_{\phi}(s,a)` where the states and
|
||||
actions are given as inputs.
|
||||
"""
|
||||
|
||||
def __init__(self, state, action, model):
|
||||
"""
|
||||
Initialize the parametrized state-action value function approximator.
|
||||
|
||||
Args:
|
||||
state (State): input state.
|
||||
action (Action): input / output action.
|
||||
model (Approximator): value function approximator.
|
||||
"""
|
||||
super(ParametrizedStateActionValue, self).__init__(state, model)
|
||||
self.action = action
|
||||
|
||||
##############
|
||||
# Properties #
|
||||
##############
|
||||
|
||||
@property
|
||||
def action(self):
|
||||
"""Return the action instance."""
|
||||
return self._action
|
||||
|
||||
@action.setter
|
||||
def action(self, action):
|
||||
"""Set the action input."""
|
||||
if isinstance(action, (int, float)):
|
||||
action = np.array([action])
|
||||
elif not isinstance(action, (Action, torch.Tensor, np.ndarray)):
|
||||
raise TypeError("Expecting the action to be an Action, torch.Tensor, or np.ndarray.")
|
||||
self._action = action
|
||||
|
||||
|
||||
class ParametrizedStateOutputActionValue(ParametrizedValue):
|
||||
r"""Parametrized State Output Action Value Function Approximator
|
||||
|
||||
This value function approximator predicts the state-action value :math:`Q_{\phi}(s,a)` for each discrete action.
|
||||
"""
|
||||
|
||||
def __init__(self, state, action, model):
|
||||
"""
|
||||
Initialize the parametrized state-action value function approximator.
|
||||
|
||||
Args:
|
||||
state (State): input state.
|
||||
action (Action): output DISCRETE state.
|
||||
model (Approximator): value function approximator.
|
||||
"""
|
||||
super(ParametrizedStateOutputActionValue, self).__init__(state, model)
|
||||
self.action = action
|
||||
|
||||
##############
|
||||
# Properties #
|
||||
##############
|
||||
|
||||
@property
|
||||
def action(self):
|
||||
"""Return the action instance."""
|
||||
return self._action
|
||||
|
||||
@action.setter
|
||||
def action(self, action):
|
||||
"""Set the action input."""
|
||||
if isinstance(action, (int, float)):
|
||||
action = np.array([action])
|
||||
elif isinstance(action, Action):
|
||||
if not action.is_discrete():
|
||||
raise ValueError("The given actions are not discrete: {}".format(action))
|
||||
elif not isinstance(action, (torch.Tensor, np.ndarray)):
|
||||
raise TypeError("Expecting the action to be an int, float, Action, torch.Tensor, or np.ndarray.")
|
||||
self._action = action
|
||||
Reference in New Issue
Block a user