From fcb172dd82e425050ffd7241dc7d53a12fb8c29a Mon Sep 17 00:00:00 2001 From: Brian Delhaisse Date: Tue, 26 Mar 2019 19:49:26 +0100 Subject: [PATCH] update actions, states, approx, policies, and values --- pyrobolearn/actions/action.py | 19 +- .../actions/robot_actions/joint_actions.py | 32 ++ pyrobolearn/approximators/approximator.py | 94 ++-- pyrobolearn/approximators/nn_approximator.py | 70 +-- pyrobolearn/policies/basic_policy.py | 159 +++++-- pyrobolearn/policies/cpg_policy.py | 49 +- pyrobolearn/policies/dmp_policy.py | 79 ++-- pyrobolearn/policies/neat_policy.py | 86 ++-- pyrobolearn/policies/nn_policy.py | 60 ++- pyrobolearn/policies/policy.py | 426 +++++++++++++----- pyrobolearn/states/generators/__init__.py | 2 +- pyrobolearn/states/state.py | 23 +- pyrobolearn/values/README.md | 7 + pyrobolearn/values/__init__.py | 9 + pyrobolearn/values/basic_value.py | 97 ++++ pyrobolearn/values/nn_value.py | 198 ++++++++ pyrobolearn/values/value.py | 338 ++++++++++++++ 17 files changed, 1419 insertions(+), 329 deletions(-) create mode 100644 pyrobolearn/values/README.md create mode 100644 pyrobolearn/values/__init__.py create mode 100644 pyrobolearn/values/basic_value.py create mode 100644 pyrobolearn/values/nn_value.py create mode 100644 pyrobolearn/values/value.py diff --git a/pyrobolearn/actions/action.py b/pyrobolearn/actions/action.py index 3c632e5..9d38ce9 100644 --- a/pyrobolearn/actions/action.py +++ b/pyrobolearn/actions/action.py @@ -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): diff --git a/pyrobolearn/actions/robot_actions/joint_actions.py b/pyrobolearn/actions/robot_actions/joint_actions.py index 0478e6d..4d6e581 100644 --- a/pyrobolearn/actions/robot_actions/joint_actions.py +++ b/pyrobolearn/actions/robot_actions/joint_actions.py @@ -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 diff --git a/pyrobolearn/approximators/approximator.py b/pyrobolearn/approximators/approximator.py index 1720b6c..cf14f3d 100644 --- a/pyrobolearn/approximators/approximator.py +++ b/pyrobolearn/approximators/approximator.py @@ -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 # diff --git a/pyrobolearn/approximators/nn_approximator.py b/pyrobolearn/approximators/nn_approximator.py index 21254d7..8de908a 100644 --- a/pyrobolearn/approximators/nn_approximator.py +++ b/pyrobolearn/approximators/nn_approximator.py @@ -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.""" diff --git a/pyrobolearn/policies/basic_policy.py b/pyrobolearn/policies/basic_policy.py index 31ad49a..a131ef1 100644 --- a/pyrobolearn/policies/basic_policy.py +++ b/pyrobolearn/policies/basic_policy.py @@ -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 diff --git a/pyrobolearn/policies/cpg_policy.py b/pyrobolearn/policies/cpg_policy.py index dfdf902..8bb7605 100644 --- a/pyrobolearn/policies/cpg_policy.py +++ b/pyrobolearn/policies/cpg_policy.py @@ -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 = {: {'phi': , 'offset': , 'amplitude': , 'freq': , 'nodes': [{'id': , '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() diff --git a/pyrobolearn/policies/dmp_policy.py b/pyrobolearn/policies/dmp_policy.py index 3a5868d..80b37f9 100644 --- a/pyrobolearn/policies/dmp_policy.py +++ b/pyrobolearn/policies/dmp_policy.py @@ -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: diff --git a/pyrobolearn/policies/neat_policy.py b/pyrobolearn/policies/neat_policy.py index 58abe8c..3b827ba 100644 --- a/pyrobolearn/policies/neat_policy.py +++ b/pyrobolearn/policies/neat_policy.py @@ -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') diff --git a/pyrobolearn/policies/nn_policy.py b/pyrobolearn/policies/nn_policy.py index 8042ed9..6106e44 100644 --- a/pyrobolearn/policies/nn_policy.py +++ b/pyrobolearn/policies/nn_policy.py @@ -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) diff --git a/pyrobolearn/policies/policy.py b/pyrobolearn/policies/policy.py index f88690a..39f737c 100644 --- a/pyrobolearn/policies/policy.py +++ b/pyrobolearn/policies/policy.py @@ -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__ diff --git a/pyrobolearn/states/generators/__init__.py b/pyrobolearn/states/generators/__init__.py index 998a861..b281b4d 100644 --- a/pyrobolearn/states/generators/__init__.py +++ b/pyrobolearn/states/generators/__init__.py @@ -1,3 +1,3 @@ # import state generators -from state_generator import * +from .state_generator import * diff --git a/pyrobolearn/states/state.py b/pyrobolearn/states/state.py index bccec75..7ac05e5 100644 --- a/pyrobolearn/states/state.py +++ b/pyrobolearn/states/state.py @@ -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): diff --git a/pyrobolearn/values/README.md b/pyrobolearn/values/README.md new file mode 100644 index 0000000..d5424e9 --- /dev/null +++ b/pyrobolearn/values/README.md @@ -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`. diff --git a/pyrobolearn/values/__init__.py b/pyrobolearn/values/__init__.py new file mode 100644 index 0000000..1d1539c --- /dev/null +++ b/pyrobolearn/values/__init__.py @@ -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 * diff --git a/pyrobolearn/values/basic_value.py b/pyrobolearn/values/basic_value.py new file mode 100644 index 0000000..3093209 --- /dev/null +++ b/pyrobolearn/values/basic_value.py @@ -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) diff --git a/pyrobolearn/values/nn_value.py b/pyrobolearn/values/nn_value.py new file mode 100644 index 0000000..90e41bf --- /dev/null +++ b/pyrobolearn/values/nn_value.py @@ -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) diff --git a/pyrobolearn/values/value.py b/pyrobolearn/values/value.py new file mode 100644 index 0000000..91dce89 --- /dev/null +++ b/pyrobolearn/values/value.py @@ -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