class_docstr: Twin Delayed DDPG (TD3)
  Addressing Function Approximation Error in Actor-Critic Methods.
  
  Original implementation: https://github.com/sfujim/TD3
  Paper: https://arxiv.org/pdf/1802.09477.pdf
  Introduction to TD3: https://spinningup.openai.com/en/latest/algorithms/td3.html
  
  :param policy: (TD3Policy or str) The policy model to use (MlpPolicy, CnnPolicy, LnMlpPolicy, ...)
  :param env: (Gym environment or str) The environment to learn from (if registered in Gym, can be str)
  :param gamma: (float) the discount factor
  :param learning_rate: (float or callable) learning rate for adam optimizer,
      the same learning rate will be used for all networks (Q-Values and Actor networks)
      it can be a function of the current progress (from 1 to 0)
  :param buffer_size: (int) size of the replay buffer
  :param batch_size: (int) Minibatch size for each gradient update
  :param tau: (float) the soft update coefficient ("polyak update" of the target networks, between 0 and 1)
  :param policy_delay: (int) Policy and target networks will only be updated once every policy_delay steps
      per training steps. The Q values will be updated policy_delay more often (update every training step).
  :param action_noise: (ActionNoise) the action noise type. Cf DDPG for the different action noise type.
  :param target_policy_noise: (float) Standard deviation of Gaussian noise added to target policy
      (smoothing noise)
  :param target_noise_clip: (float) Limit for absolute value of target policy smoothing noise.
  :param train_freq: (int) Update the model every `train_freq` steps.
  :param learning_starts: (int) how many steps of the model to collect transitions for before learning starts
  :param gradient_steps: (int) How many gradient update after each step
  :param random_exploration: (float) Probability of taking a random action (as in an epsilon-greedy strategy)
      This is not needed for TD3 normally but can help exploring when using HER + TD3.
      This hack was present in the original OpenAI Baselines repo (DDPG + HER)
  :param verbose: (int) the verbosity level: 0 none, 1 training information, 2 tensorflow debug
  :param tensorboard_log: (str) the log location for tensorboard (if None, no logging)
  :param _init_setup_model: (bool) Whether or not to build the network at the creation of the instance
  :param policy_kwargs: (dict) additional arguments to be passed to the policy on creation
  :param full_tensorboard_log: (bool) enable additional logging when using tensorboard
      Note: this has no effect on TD3 logging for now
  :param seed: (int) Seed for the pseudo-random generators (python, numpy, tensorflow).
      If None (default), use random seed. Note that if you want completely deterministic
      results, you must set `n_cpu_tf_sess` to 1.
  :param n_cpu_tf_sess: (int) The number of threads for TensorFlow operations
      If None, the number of cpu of the current machine will be used.
class_docstr_tok: ["Twin","Delayed","DDPG","(","TD3",")","Addressing","Function","Approximation","Error","in","Actor-Critic","Methods",".","Original","implementation",":","https",":","\/\/github.com\/sfujim\/TD3","Paper",":","https",":","\/\/arxiv.org\/pdf\/1802.09477.pdf","Introduction","to","TD3",":","https",":","\/\/spinningup.openai.com\/en\/latest\/algorithms\/td3.html",":","param","policy",":","(","TD3Policy","or","str",")","The","policy","model","to","use","(","MlpPolicy",",","CnnPolicy",",","LnMlpPolicy",",","...",")",":","param","env",":","(","Gym","environment","or","str",")","The","environment","to","learn","from","(","if","registered","in","Gym",",","can","be","str",")",":","param","gamma",":","(","float",")","the","discount","factor",":","param","learning_rate",":","(","float","or","callable",")","learning","rate","for","adam","optimizer",",","the","same","learning","rate","will","be","used","for","all","networks","(","Q-Values","and","Actor","networks",")","it","can","be","a","function","of","the","current","progress","(","from","1","to","0",")",":","param","buffer_size",":","(","int",")","size","of","the","replay","buffer",":","param","batch_size",":","(","int",")","Minibatch","size","for","each","gradient","update",":","param","tau",":","(","float",")","the","soft","update","coefficient","(","``","polyak","update","''","of","the","target","networks",",","between","0","and","1",")",":","param","policy_delay",":","(","int",")","Policy","and","target","networks","will","only","be","updated","once","every","policy_delay","steps","per","training","steps",".","The","Q","values","will","be","updated","policy_delay","more","often","(","update","every","training","step",")",".",":","param","action_noise",":","(","ActionNoise",")","the","action","noise","type",".","Cf","DDPG","for","the","different","action","noise","type",".",":","param","target_policy_noise",":","(","float",")","Standard","deviation","of","Gaussian","noise","added","to","target","policy","(","smoothing","noise",")",":","param","target_noise_clip",":","(","float",")","Limit","for","absolute","value","of","target","policy","smoothing","noise",".",":","param","train_freq",":","(","int",")","Update","the","model","every","`","train_freq","`","steps",".",":","param","learning_starts",":","(","int",")","how","many","steps","of","the","model","to","collect","transitions","for","before","learning","starts",":","param","gradient_steps",":","(","int",")","How","many","gradient","update","after","each","step",":","param","random_exploration",":","(","float",")","Probability","of","taking","a","random","action","(","as","in","an","epsilon-greedy","strategy",")","This","is","not","needed","for","TD3","normally","but","can","help","exploring","when","using","HER","+","TD3",".","This","hack","was","present","in","the","original","OpenAI","Baselines","repo","(","DDPG","+","HER",")",":","param","verbose",":","(","int",")","the","verbosity","level",":","0","none",",","1","training","information",",","2","tensorflow","debug",":","param","tensorboard_log",":","(","str",")","the","log","location","for","tensorboard","(","if","None",",","no","logging",")",":","param","_init_setup_model",":","(","bool",")","Whether","or","not","to","build","the","network","at","the","creation","of","the","instance",":","param","policy_kwargs",":","(","dict",")","additional","arguments","to","be","passed","to","the","policy","on","creation",":","param","full_tensorboard_log",":","(","bool",")","enable","additional","logging","when","using","tensorboard","Note",":","this","has","no","effect","on","TD3","logging","for","now",":","param","seed",":","(","int",")","Seed","for","the","pseudo-random","generators","(","python",",","numpy",",","tensorflow",")",".","If","None","(","default",")",",","use","random","seed",".","Note","that","if","you","want","completely","deterministic","results",",","you","must","set","`","n_cpu_tf_sess","`","to","1.",":","param","n_cpu_tf_sess",":","(","int",")","The","number","of","threads","for","TensorFlow","operations","If","None",",","the","number","of","cpu","of","the","current","machine","will","be","used","."]
code: def learn(
      self,
      total_timesteps,
      callback=None,
      log_interval=4,
      tb_log_name="TD3",
      reset_num_timesteps=True,
      replay_wrapper=None,
  ):
      new_tb_log = self._init_num_timesteps(reset_num_timesteps)
      callback = self._init_callback(callback)
  
      if replay_wrapper is not None:
          self.replay_buffer = replay_wrapper(self.replay_buffer)
  
      with SetVerbosity(self.verbose), TensorboardWriter(
          self.graph, self.tensorboard_log, tb_log_name, new_tb_log
      ) as writer:
          self._setup_learn()
  
          # Transform to callable if needed
          self.learning_rate = get_schedule_fn(self.learning_rate)
          # Initial learning rate
          current_lr = self.learning_rate(1)
  
          start_time = time.time()
          episode_rewards = [0.0]
          episode_successes = []
          if self.action_noise is not None:
              self.action_noise.reset()
          obs = self.env.reset()
          # Retrieve unnormalized observation for saving into the buffer
          if self._vec_normalize_env is not None:
              obs_ = (
                  self._vec_normalize_env.get_original_obs().squeeze()
              )
          n_updates = 0
          infos_values = []
  
          callback.on_training_start(locals(), globals())
          callback.on_rollout_start()
  
          for step in range(total_timesteps):
              # Before training starts, randomly sample actions
              # from a uniform distribution for better exploration.
              # Afterwards, use the learned policy
              # if random_exploration is set to 0 (normal setting)
              if (
                  self.num_timesteps < self.learning_starts
                  or np.random.rand() < self.random_exploration
              ):
                  # actions sampled from action space are from range specific to the environment
                  # but algorithm operates on tanh-squashed actions therefore simple scaling is used
                  unscaled_action = self.env.action_space.sample()
                  action = scale_action(
                      self.action_space, unscaled_action
                  )
              else:
                  action = self.policy_tf.step(obs[None]).flatten()
                  # Add noise to the action, as the policy
                  # is deterministic, this is required for exploration
                  if self.action_noise is not None:
                      action = np.clip(
                          action + self.action_noise(), -1, 1
                      )
                  # Rescale from [-1, 1] to the correct bounds
                  unscaled_action = unscale_action(
                      self.action_space, action
                  )
  
              assert action.shape == self.env.action_space.shape
  
              new_obs, reward, done, info = self.env.step(
                  unscaled_action
              )
  
              self.num_timesteps += 1
  
              # Only stop training if return value is False, not when it is None. This is for backwards
              # compatibility with callbacks that have no return statement.
              callback.update_locals(locals())
              if callback.on_step() is False:
                  break
  
              # Store only the unnormalized version
              if self._vec_normalize_env is not None:
                  new_obs_ = (
                      self._vec_normalize_env.get_original_obs().squeeze()
                  )
                  reward_ = (
                      self._vec_normalize_env.get_original_reward().squeeze()
                  )
              else:
                  # Avoid changing the original ones
                  obs_, new_obs_, reward_ = obs, new_obs, reward
  
              # Store transition in the replay buffer.
              self.replay_buffer_add(
                  obs_, action, reward_, new_obs_, done, info
              )
              obs = new_obs
              # Save the unnormalized observation
              if self._vec_normalize_env is not None:
                  obs_ = new_obs_
  
              # Retrieve reward and episode length if using Monitor wrapper
              maybe_ep_info = info.get("episode")
              if maybe_ep_info is not None:
                  self.ep_info_buf.extend([maybe_ep_info])
  
              if writer is not None:
                  # Write reward per episode to tensorboard
                  ep_reward = np.array([reward_]).reshape((1, -1))
                  ep_done = np.array([done]).reshape((1, -1))
                  tf_util.total_episode_reward_logger(
                      self.episode_reward,
                      ep_reward,
                      ep_done,
                      writer,
                      self.num_timesteps,
                  )
  
              if self.num_timesteps % self.train_freq == 0:
                  callback.on_rollout_end()
  
                  mb_infos_vals = []
                  # Update policy, critics and target networks
                  for grad_step in range(self.gradient_steps):
                      # Break if the warmup phase is not over
                      # or if there are not enough samples in the replay buffer
                      if (
                          not self.replay_buffer.can_sample(
                              self.batch_size
                          )
                          or self.num_timesteps < self.learning_starts
                      ):
                          break
                      n_updates += 1
                      # Compute current learning_rate
                      frac = 1.0 - step / total_timesteps
                      current_lr = self.learning_rate(frac)
                      # Update policy and critics (q functions)
                      # Note: the policy is updated less frequently than the Q functions
                      # this is controlled by the `policy_delay` parameter
                      mb_infos_vals.append(
                          self._train_step(
                              step,
                              writer,
                              current_lr,
                              (step + grad_step) % self.policy_delay
                              == 0,
                          )
                      )
  
                  # Log losses and entropy, useful for monitor training
                  if len(mb_infos_vals) > 0:
                      infos_values = np.mean(mb_infos_vals, axis=0)
  
                  callback.on_rollout_start()
  
              episode_rewards[-1] += reward_
              if done:
                  if self.action_noise is not None:
                      self.action_noise.reset()
                  if not isinstance(self.env, VecEnv):
                      obs = self.env.reset()
                  episode_rewards.append(0.0)
  
                  maybe_is_success = info.get("is_success")
                  if maybe_is_success is not None:
                      episode_successes.append(float(maybe_is_success))
  
              if len(episode_rewards[-101:-1]) == 0:
                  mean_reward = -np.inf
              else:
                  mean_reward = round(
                      float(np.mean(episode_rewards[-101:-1])), 1
                  )
  
              # substract 1 as we appended a new term just now
              num_episodes = len(episode_rewards) - 1
              # Display training infos
              if (
                  self.verbose >= 1
                  and done
                  and log_interval is not None
                  and num_episodes % log_interval == 0
              ):
                  fps = int(step / (time.time() - start_time))
                  logger.logkv("episodes", num_episodes)
                  logger.logkv("mean 100 episode reward", mean_reward)
                  if (
                      len(self.ep_info_buf) > 0
                      and len(self.ep_info_buf[0]) > 0
                  ):
                      logger.logkv(
                          "ep_rewmean",
                          safe_mean(
                              [
                                  ep_info["r"]
                                  for ep_info in self.ep_info_buf
                              ]
                          ),
                      )
                      logger.logkv(
                          "eplenmean",
                          safe_mean(
                              [
                                  ep_info["l"]
                                  for ep_info in self.ep_info_buf
                              ]
                          ),
                      )
                  logger.logkv("n_updates", n_updates)
                  logger.logkv("current_lr", current_lr)
                  logger.logkv("fps", fps)
                  logger.logkv(
                      "time_elapsed", int(time.time() - start_time)
                  )
                  if len(episode_successes) > 0:
                      logger.logkv(
                          "success rate",
                          np.mean(episode_successes[-100:]),
                      )
                  if len(infos_values) > 0:
                      for name, val in zip(
                          self.infos_names, infos_values
                      ):
                          logger.logkv(name, val)
                  logger.logkv("total timesteps", self.num_timesteps)
                  logger.dumpkvs()
                  # Reset infos:
                  infos_values = []
  
          callback.on_training_end()
          return self
  
imports: import time
  import warnings
  import numpy
  import tensorflow
  from stable_baselines import logger
  from stable_baselines.common import tf_util, OffPolicyRLModel, SetVerbosity, TensorboardWriter
  from stable_baselines.common.vec_env import VecEnv
  from stable_baselines.common.math_util import safe_mean, unscale_action, scale_action
  from stable_baselines.common.schedules import get_schedule_fn
  from stable_baselines.common.buffers import ReplayBuffer
  from stable_baselines.td3.policies import TD3Policy
label_desc: Use image node_id 5 for calling the TD3 obj's underlying member method code with example usage: obj.learn(total_timesteps, callback, log_interval, tb_log_name, reset_num_timesteps, replay_wrapper) and returns: self