<filename>hmlf/algorithms/paddpg/paddpg.py
from typing import Any, Callable, Dict, Optional, Tuple, Type, Union

import numpy as np
import torch as th

from hmlf import spaces
from hmlf.algorithms.td3.policies import TD3Policy
from hmlf.algorithms.td3.td3 import TD3
from hmlf.common.noise import ActionNoise
from hmlf.common.off_policy_algorithm import OffPolicyAlgorithm
from hmlf.common.type_aliases import GymEnv, MaybeCallback
from hmlf.environments.wrap_environment import register_algorithm_for_wrap_environment, wrap_one_hot


class PADDPG(TD3):
    """
    Deep Deterministic Policy Gradient (DDPG).

    Deterministic Policy Gradient: http://proceedings.mlr.press/v32/silver14.pdf
    DDPG Paper: https://arxiv.org/abs/1509.02971
    Introduction to DDPG: https://spinningup.openai.com/en/latest/algorithms/ddpg.html

    Note: we treat DDPG as a special case of its successor TD3.

    :param policy: The policy model to use (MlpPolicy, CnnPolicy, ...)
    :param env: The environment to learn from (if registered in Gym, can be str)
    :param learning_rate: learning rate for adam optimizer,
        the same learning rate will be used for all networks (Q-Values, Actor and Value function)
        it can be a function of the current progress remaining (from 1 to 0)
    :param buffer_size: size of the replay buffer
    :param learning_starts: how many steps of the model to collect transitions for before learning starts
    :param batch_size: Minibatch size for each gradient update
    :param tau: the soft update coefficient ("Polyak update", between 0 and 1)
    :param gamma: the discount factor
    param train_freq: Update the model every ``train_freq`` steps. Alternatively pass a tuple of frequency and unit
        like ``(5, "step")`` or ``(2, "episode")``.
    :param gradient_steps: How many gradient steps to do after each rollout (see ``train_freq``)
        Set to ``-1`` means to do as many gradient steps as steps done in the environment
        during the rollout.
    :param action_noise: the action noise type (None by default), this can help
        for hard exploration problem. Cf common.noise for the different action noise type.
    :param optimize_memory_usage: Enable a memory efficient variant of the replay buffer
        at a cost of more complexity.
        See https://github.com/DLR-RM/stable-baselines3/issues/37#issuecomment-637501195
    :param create_eval_env: Whether to create a second environment that will be
        used for evaluating the agent periodically. (Only available when passing string for the environment)
    :param policy_kwargs: additional arguments to be passed to the policy on creation
    :param verbose: the verbosity level: 0 no output, 1 info, 2 debug
    :param seed: Seed for the pseudo random generators
    :param device: Device (cpu, cuda, ...) on which the code should be run.
        Setting it to auto, the code will be run on the GPU if possible.
    :param _init_setup_model: Whether or not to build the network at the creation of the instance
    """

    def __init__(
        self,
        policy: Type[TD3Policy],
        env: Union[GymEnv, str],
        learning_rate: Union[float, Callable] = 1e-3,
        buffer_size: int = int(1e6),
        learning_starts: int = 100,
        batch_size: int = 100,
        tau: float = 0.005,
        gamma: float = 0.99,
        train_freq: Union[int, Tuple[int, str]] = -1,
        gradient_steps: int = -1,
        action_noise: Optional[ActionNoise] = None,
        optimize_memory_usage: bool = False,
        tensorboard_log: Optional[str] = None,
        create_eval_env: bool = False,
        policy_kwargs: Dict[str, Any] = None,
        verbose: int = 0,
        seed: Optional[int] = None,
        device: Union[th.device, str] = "auto",
        _init_setup_model: bool = True,
    ):

        super(PADDPG, self).__init__(
            policy=policy,
            env=env,
            learning_rate=learning_rate,
            buffer_size=buffer_size,
            learning_starts=learning_starts,
            batch_size=batch_size,
            tau=tau,
            gamma=gamma,
            train_freq=train_freq,
            gradient_steps=gradient_steps,
            action_noise=action_noise,
            policy_kwargs=policy_kwargs,
            tensorboard_log=tensorboard_log,
            verbose=verbose,
            device=device,
            create_eval_env=create_eval_env,
            seed=seed,
            optimize_memory_usage=optimize_memory_usage,
            # Remove all tricks from TD3 to obtain DDPG:
            # we still need to specify target_policy_noise > 0 to avoid errors
            policy_delay=1,
            target_noise_clip=0.0,
            target_policy_noise=0.1,
            _init_setup_model=False,
        )

        # Use only one critic
        if "n_critics" not in self.policy_kwargs:
            self.policy_kwargs["n_critics"] = 1

        if _init_setup_model:
            self._setup_model()

    def learn(
        self,
        total_timesteps: int,
        callback: MaybeCallback = None,
        log_interval: int = 4,
        eval_env: Optional[GymEnv] = None,
        eval_freq: int = -1,
        n_eval_episodes: int = 5,
        tb_log_name: str = "PADDPG",
        eval_log_path: Optional[str] = None,
        reset_num_timesteps: bool = True,
    ) -> OffPolicyAlgorithm:

        return super(PADDPG, self).learn(
            total_timesteps=total_timesteps,
            callback=callback,
            log_interval=log_interval,
            eval_env=eval_env,
            eval_freq=eval_freq,
            n_eval_episodes=n_eval_episodes,
            tb_log_name=tb_log_name,
            eval_log_path=eval_log_path,
            reset_num_timesteps=reset_num_timesteps,
        )

    def _sample_action(
        self, learning_starts: int, action_noise: Optional[ActionNoise] = None
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Sample an action according to the exploration policy.
        This is either done by sampling the probability distribution of the policy,
        or sampling a random action (from a uniform distribution over the action space)
        or by adding noise to the deterministic output.

        :param action_noise: Action noise that will be used for exploration
            Required for deterministic policy (e.g. TD3). This can also be used
            in addition to the stochastic policy for SAC.
        :param learning_starts: Number of steps before learning for the warm-up phase.
        :return: action to take in the environment
            and scaled action that will be stored in the replay buffer.
            The two differs when the action space is not normalized (bounds are not [-1, 1]).
        """

        # Select action randomly or according to policy
        if self.num_timesteps < learning_starts and not (self.use_sde and self.use_sde_at_warmup):
            # Warmup phase
            # unscaled_action = np.hstack(self.action_space.sample()).astype(np.float32)
            unscaled_action = np.hstack(self.action_space.sample())
        else:
            # Note: when using continuous actions,
            # we assume that the policy uses tanh to scale the action
            # We use non-deterministic action in the case of SAC, for TD3, it does not matter
            # Note: when using continuous actions,
            # we assume that the policy uses tanh to scale the action
            # We use non-deterministic action in the case of SAC, for TD3, it does not matter
            observation = th.as_tensor(self._last_obs).to(self.device)
            with th.no_grad():
                actions = self.policy._predict(observation, deterministic=False)
            # Convert to numpy
            unscaled_action = actions.cpu().numpy()
            unscaled_action = unscaled_action[0]

        # # Rescale the action from [low, high] to [-1, 1]
        # if isinstance(self.action_space, spaces.Box):
        #     scaled_action = self.policy.scale_action(unscaled_action)

        #     # Add noise to the action (improve exploration)
        #     if action_noise is not None:
        #         scaled_action = np.clip(scaled_action + action_noise(), -1, 1)

        #     # We store the scaled action in the buffer
        #     buffer_action = scaled_action
        #     action = self.policy.unscale_action(scaled_action)
        # TODO Add scaling for the parameters
        if isinstance(self.action_space, spaces.Tuple):
            buffer_action = unscaled_action
            action = self.action_space.format_action(buffer_action.reshape(-1, self.action_space.get_dimension()))
        else:
            # Discrete case, no need to normalize or clip
            buffer_action = unscaled_action
            action = buffer_action

        return action, buffer_action


register_algorithm_for_wrap_environment(PADDPG, wrap_one_hot)

<filename>hmlf/algorithms/paddpg/paddpg.py
from typing import Any, Callable, Dict, Optional, Tuple, Type, Union

import numpy as np
import torch as th

from hmlf import spaces
from hmlf.algorithms.td3.policies import TD3Policy
from hmlf.algorithms.td3.td3 import TD3
from hmlf.common.noise import ActionNoise
from hmlf.common.off_policy_algorithm import OffPolicyAlgorithm
from hmlf.common.type_aliases import GymEnv, MaybeCallback
from hmlf.environments.wrap_environment import register_algorithm_for_wrap_environment, wrap_one_hot


class PADDPG(TD3):
    """
    Deep Deterministic Policy Gradient (DDPG).

    Deterministic Policy Gradient: http://proceedings.mlr.press/v32/silver14.pdf
    DDPG Paper: https://arxiv.org/abs/1509.02971
    Introduction to DDPG: https://spinningup.openai.com/en/latest/algorithms/ddpg.html

    Note: we treat DDPG as a special case of its successor TD3.

    :param policy: The policy model to use (MlpPolicy, CnnPolicy, ...)
    :param env: The environment to learn from (if registered in Gym, can be str)
    :param learning_rate: learning rate for adam optimizer,
        the same learning rate will be used for all networks (Q-Values, Actor and Value function)
        it can be a function of the current progress remaining (from 1 to 0)
    :param buffer_size: size of the replay buffer
    :param learning_starts: how many steps of the model to collect transitions for before learning starts
    :param batch_size: Minibatch size for each gradient update
    :param tau: the soft update coefficient ("Polyak update", between 0 and 1)
    :param gamma: the discount factor
    param train_freq: Update the model every ``train_freq`` steps. Alternatively pass a tuple of frequency and unit
        like ``(5, "step")`` or ``(2, "episode")``.
    :param gradient_steps: How many gradient steps to do after each rollout (see ``train_freq``)
        Set to ``-1`` means to do as many gradient steps as steps done in the environment
        during the rollout.
    :param action_noise: the action noise type (None by default), this can help
        for hard exploration problem. Cf common.noise for the different action noise type.
    :param optimize_memory_usage: Enable a memory efficient variant of the replay buffer
        at a cost of more complexity.
        See https://github.com/DLR-RM/stable-baselines3/issues/37#issuecomment-637501195
    :param create_eval_env: Whether to create a second environment that will be
        used for evaluating the agent periodically. (Only available when passing string for the environment)
    :param policy_kwargs: additional arguments to be passed to the policy on creation
    :param verbose: the verbosity level: 0 no output, 1 info, 2 debug
    :param seed: Seed for the pseudo random generators
    :param device: Device (cpu, cuda, ...) on which the code should be run.
        Setting it to auto, the code will be run on the GPU if possible.
    :param _init_setup_model: Whether or not to build the network at the creation of the instance
    """

    def __init__(
        self,
        policy: Type[TD3Policy],
        env: Union[GymEnv, str],
        learning_rate: Union[float, Callable] = 1e-3,
        buffer_size: int = int(1e6),
        learning_starts: int = 100,
        batch_size: int = 100,
        tau: float = 0.005,
        gamma: float = 0.99,
        train_freq: Union[int, Tuple[int, str]] = -1,
        gradient_steps: int = -1,
        action_noise: Optional[ActionNoise] = None,
        optimize_memory_usage: bool = False,
        tensorboard_log: Optional[str] = None,
        create_eval_env: bool = False,
        policy_kwargs: Dict[str, Any] = None,
        verbose: int = 0,
        seed: Optional[int] = None,
        device: Union[th.device, str] = "auto",
        _init_setup_model: bool = True,
    ):

        super(PADDPG, self).__init__(
            policy=policy,
            env=env,
            learning_rate=learning_rate,
            buffer_size=buffer_size,
            learning_starts=learning_starts,
            batch_size=batch_size,
            tau=tau,
            gamma=gamma,
            train_freq=train_freq,
            gradient_steps=gradient_steps,
            action_noise=action_noise,
            policy_kwargs=policy_kwargs,
            tensorboard_log=tensorboard_log,
            verbose=verbose,
            device=device,
            create_eval_env=create_eval_env,
            seed=seed,
            optimize_memory_usage=optimize_memory_usage,
            # Remove all tricks from TD3 to obtain DDPG:
            # we still need to specify target_policy_noise > 0 to avoid errors
            policy_delay=1,
            target_noise_clip=0.0,
            target_policy_noise=0.1,
            _init_setup_model=False,
        )

        # Use only one critic
        if "n_critics" not in self.policy_kwargs:
            self.policy_kwargs["n_critics"] = 1

        if _init_setup_model:
            self._setup_model()

    def learn(
        self,
        total_timesteps: int,
        callback: MaybeCallback = None,
        log_interval: int = 4,
        eval_env: Optional[GymEnv] = None,
        eval_freq: int = -1,
        n_eval_episodes: int = 5,
        tb_log_name: str = "PADDPG",
        eval_log_path: Optional[str] = None,
        reset_num_timesteps: bool = True,
    ) -> OffPolicyAlgorithm:

        return super(PADDPG, self).learn(
            total_timesteps=total_timesteps,
            callback=callback,
            log_interval=log_interval,
            eval_env=eval_env,
            eval_freq=eval_freq,
            n_eval_episodes=n_eval_episodes,
            tb_log_name=tb_log_name,
            eval_log_path=eval_log_path,
            reset_num_timesteps=reset_num_timesteps,
        )

    def _sample_action(
        self, learning_starts: int, action_noise: Optional[ActionNoise] = None
    ) -> Tuple[np.ndarray, np.ndarray]:
        """
        Sample an action according to the exploration policy.
        This is either done by sampling the probability distribution of the policy,
        or sampling a random action (from a uniform distribution over the action space)
        or by adding noise to the deterministic output.

        :param action_noise: Action noise that will be used for exploration
            Required for deterministic policy (e.g. TD3). This can also be used
            in addition to the stochastic policy for SAC.
        :param learning_starts: Number of steps before learning for the warm-up phase.
        :return: action to take in the environment
            and scaled action that will be stored in the replay buffer.
            The two differs when the action space is not normalized (bounds are not [-1, 1]).
        """

        # Select action randomly or according to policy
        if self.num_timesteps < learning_starts and not (self.use_sde and self.use_sde_at_warmup):
            # Warmup phase
            # unscaled_action = np.hstack(self.action_space.sample()).astype(np.float32)
            unscaled_action = np.hstack(self.action_space.sample())
        else:
            # Note: when using continuous actions,
            # we assume that the policy uses tanh to scale the action
            # We use non-deterministic action in the case of SAC, for TD3, it does not matter
            # Note: when using continuous actions,
            # we assume that the policy uses tanh to scale the action
            # We use non-deterministic action in the case of SAC, for TD3, it does not matter
            observation = th.as_tensor(self._last_obs).to(self.device)
            with th.no_grad():
                actions = self.policy._predict(observation, deterministic=False)
            # Convert to numpy
            unscaled_action = actions.cpu().numpy()
            unscaled_action = unscaled_action[0]

        # # Rescale the action from [low, high] to [-1, 1]
        # if isinstance(self.action_space, spaces.Box):
        #     scaled_action = self.policy.scale_action(unscaled_action)

        #     # Add noise to the action (improve exploration)
        #     if action_noise is not None:
        #         scaled_action = np.clip(scaled_action + action_noise(), -1, 1)

        #     # We store the scaled action in the buffer
        #     buffer_action = scaled_action
        #     action = self.policy.unscale_action(scaled_action)
        # TODO Add scaling for the parameters
        if isinstance(self.action_space, spaces.Tuple):
            buffer_action = unscaled_action
            action = self.action_space.format_action(buffer_action.reshape(-1, self.action_space.get_dimension()))
        else:
            # Discrete case, no need to normalize or clip
            buffer_action = unscaled_action
            action = buffer_action

        return action, buffer_action


register_algorithm_for_wrap_environment(PADDPG, wrap_one_hot)

Deep Deterministic Policy Gradient (DDPG).      Deterministic Policy Gradient: http://proceedings.mlr.press/v32/silver14.pdf     DDPG Paper: https://arxiv.org/abs/1509.02971     Introduction to DDPG: https://spinningup.openai.com/en/latest/algorithms/ddpg.html      Note: we treat DDPG as a special case of its successor TD3.      :param policy: The policy model to use (MlpPolicy, CnnPolicy, ...)     :param env: The environment to learn from (if registered in Gym, can be str)     :param learning_rate: learning rate for adam optimizer,         the same learning rate will be used for all networks (Q-Values, Actor and Value function)         it can be a function of the current progress remaining (from 1 to 0)     :param buffer_size: size of the replay buffer     :param learning_starts: how many steps of the model to collect transitions for before learning starts     :param batch_size: Minibatch size for each gradient update     :param tau: the soft update coefficient ("Polyak update", between 0 and 1)     :param gamma: the discount factor     param train_freq: Update the model every ``train_freq`` steps. Alternatively pass a tuple of frequency and unit         like ``(5, "step")`` or ``(2, "episode")``.     :param gradient_steps: How many gradient steps to do after each rollout (see ``train_freq``)         Set to ``-1`` means to do as many gradient steps as steps done in the environment         during the rollout.     :param action_noise: the action noise type (None by default), this can help         for hard exploration problem. Cf common.noise for the different action noise type.     :param optimize_memory_usage: Enable a memory efficient variant of the replay buffer         at a cost of more complexity.         See https://github.com/DLR-RM/stable-baselines3/issues/37#issuecomment-637501195     :param create_eval_env: Whether to create a second environment that will be         used for evaluating the agent periodically. (Only available when passing string for the environment)     :param policy_kwargs: additional arguments to be passed to the policy on creation     :param verbose: the verbosity level: 0 no output, 1 info, 2 debug     :param seed: Seed for the pseudo random generators     :param device: Device (cpu, cuda, ...) on which the code should be run.         Setting it to auto, the code will be run on the GPU if possible.     :param _init_setup_model: Whether or not to build the network at the creation of the instance # Remove all tricks from TD3 to obtain DDPG: # we still need to specify target_policy_noise > 0 to avoid errors # Use only one critic Sample an action according to the exploration policy.         This is either done by sampling the probability distribution of the policy,         or sampling a random action (from a uniform distribution over the action space)         or by adding noise to the deterministic output.          :param action_noise: Action noise that will be used for exploration             Required for deterministic policy (e.g. TD3). This can also be used             in addition to the stochastic policy for SAC.         :param learning_starts: Number of steps before learning for the warm-up phase.         :return: action to take in the environment             and scaled action that will be stored in the replay buffer.             The two differs when the action space is not normalized (bounds are not [-1, 1]). # Select action randomly or according to policy # Warmup phase # unscaled_action = np.hstack(self.action_space.sample()).astype(np.float32) # Note: when using continuous actions, # we assume that the policy uses tanh to scale the action # We use non-deterministic action in the case of SAC, for TD3, it does not matter # Note: when using continuous actions, # we assume that the policy uses tanh to scale the action # We use non-deterministic action in the case of SAC, for TD3, it does not matter # Convert to numpy # # Rescale the action from [low, high] to [-1, 1] # if isinstance(self.action_space, spaces.Box): #     scaled_action = self.policy.scale_action(unscaled_action) #     # Add noise to the action (improve exploration) #     if action_noise is not None: #         scaled_action = np.clip(scaled_action + action_noise(), -1, 1) #     # We store the scaled action in the buffer #     buffer_action = scaled_action #     action = self.policy.unscale_action(scaled_action) # TODO Add scaling for the parameters # Discrete case, no need to normalize or clip