class_docstr: Implements the Population Based Training (PBT) algorithm.
  
  https://www.deepmind.com/blog/population-based-training-of-neural-networks
  
  PBT trains a group of models (or agents) in parallel. Periodically, poorly
  performing models clone the state of the top performers, and a random
  mutation is applied to their hyperparameters in the hopes of
  outperforming the current top models.
  
  Unlike other hyperparameter search algorithms, PBT mutates hyperparameters
  during training time. This enables very fast hyperparameter discovery and
  also automatically discovers good annealing schedules.
  
  This Tune PBT implementation considers all trials added as part of the
  PBT population. If the number of trials exceeds the cluster capacity,
  they will be time-multiplexed as to balance training progress across the
  population. To run multiple trials, use `tune.TuneConfig(num_samples=<int>)`.
  
  In {LOG_DIR}/{MY_EXPERIMENT_NAME}/, all mutations are logged in
  `pbt_global.txt` and individual policy perturbations are recorded
  in pbt_policy_{i}.txt. Tune logs: [target trial tag, clone trial tag,
  target trial iteration, clone trial iteration, old config, new config]
  on each perturbation step.
  
  Args:
      time_attr: The training result attr to use for comparing time.
          Note that you can pass in something non-temporal such as
          `training_iteration` as a measure of progress, the only requirement
          is that the attribute should increase monotonically.
      metric: The training result objective value attribute. Stopping
          procedures will use this attribute. If None but a mode was passed,
          the `ray.tune.result.DEFAULT_METRIC` will be used per default.
      mode: One of {min, max}. Determines whether objective is
          minimizing or maximizing the metric attribute.
      perturbation_interval: Models will be considered for
          perturbation at this interval of `time_attr`. Note that
          perturbation incurs checkpoint overhead, so you shouldn't set this
          to be too frequent.
      burn_in_period: Models will not be considered for
          perturbation before this interval of `time_attr` has passed. This
          guarantees that models are trained for at least a certain amount
          of time or timesteps before being perturbed.
      hyperparam_mutations: Hyperparams to mutate. The format is
          as follows: for each key, either a list, function,
          or a tune search space object (tune.loguniform, tune.uniform,
          etc.) can be provided. A list specifies an allowed set of
          categorical values. A function or tune search space object
          specifies the distribution of a continuous parameter. You must
          use tune.choice, tune.uniform, tune.loguniform, etc.. Arbitrary
          tune.sample_from objects are not supported.
          A key can also hold a dict for nested hyperparameters.
          You must specify at least one of `hyperparam_mutations` or
          `custom_explore_fn`.
          Tune will sample the search space provided by
          `hyperparam_mutations` for the initial hyperparameter values if the
          corresponding hyperparameters are not present in a trial's initial `config`.
      quantile_fraction: Parameters are transferred from the top
          `quantile_fraction` fraction of trials to the bottom
          `quantile_fraction` fraction. Needs to be between 0 and 0.5.
          Setting it to 0 essentially implies doing no exploitation at all.
      resample_probability: The probability of resampling from the
          original distribution when applying `hyperparam_mutations`. If not
          resampled, the value will be perturbed by a factor chosen from
          `perturbation_factors` if continuous, or changed to an adjacent value
          if discrete.
      perturbation_factors: Scaling factors to choose between when mutating
          a continuous hyperparameter.
      custom_explore_fn: You can also specify a custom exploration
          function. This function is invoked as `f(config)` after built-in
          perturbations from `hyperparam_mutations` are applied, and should
          return `config` updated as needed. You must specify at least one of
          `hyperparam_mutations` or `custom_explore_fn`.
      log_config: Whether to log the ray config of each model to
          local_dir at each exploit. Allows config schedule to be
          reconstructed.
      require_attrs: Whether to require time_attr and metric to appear
          in result for every iteration. If True, error will be raised
          if these values are not present in trial result.
      synch: If False, will use asynchronous implementation of
          PBT. Trial perturbations occur every perturbation_interval for each
          trial independently. If True, will use synchronous implementation
          of PBT. Perturbations will occur only after all trials are
          synced at the same time_attr every perturbation_interval.
          Defaults to False. See Appendix A.1 here
          https://arxiv.org/pdf/1711.09846.pdf.
  
  .. code-block:: python
  
      import random
      from ray import tune
      from ray.tune.schedulers import PopulationBasedTraining
  
      pbt = PopulationBasedTraining(
          time_attr="training_iteration",
          metric="episode_reward_mean",
          mode="max",
          perturbation_interval=10,  # every 10 `time_attr` units
                                     # (training_iterations in this case)
          hyperparam_mutations={
              # Perturb factor1 by scaling it by 0.8 or 1.2. Resampling
              # resets it to a value sampled from the lambda function.
              "factor_1": lambda: random.uniform(0.0, 20.0),
              # Alternatively, use tune search space primitives.
              # The search space for factor_1 is equivalent to factor_2.
              "factor_2": tune.uniform(0.0, 20.0),
              # Perturb factor3 by changing it to an adjacent value, e.g.
              # 10 -> 1 or 10 -> 100. Resampling will choose at random.
              "factor_3": [1, 10, 100, 1000, 10000],
              # Using tune.choice is NOT equivalent to the above.
              # factor_4 is treated as a continuous hyperparameter.
              "factor_4": tune.choice([1, 10, 100, 1000, 10000]),
          })
      tuner = tune.Tuner(
          trainable,
          tune_config=tune.TuneConfig(
              scheduler=pbt,
              num_samples=8,
          ),
      )
      tuner.fit()
class_docstr_tok: ["Implements","the","Population","Based","Training","(","PBT",")","algorithm",".","https",":","\/\/www.deepmind.com\/blog\/population-based-training-of-neural-networks","PBT","trains","a","group","of","models","(","or","agents",")","in","parallel",".","Periodically",",","poorly","performing","models","clone","the","state","of","the","top","performers",",","and","a","random","mutation","is","applied","to","their","hyperparameters","in","the","hopes","of","outperforming","the","current","top","models",".","Unlike","other","hyperparameter","search","algorithms",",","PBT","mutates","hyperparameters","during","training","time",".","This","enables","very","fast","hyperparameter","discovery","and","also","automatically","discovers","good","annealing","schedules",".","This","Tune","PBT","implementation","considers","all","trials","added","as","part","of","the","PBT","population",".","If","the","number","of","trials","exceeds","the","cluster","capacity",",","they","will","be","time-multiplexed","as","to","balance","training","progress","across","the","population",".","To","run","multiple","trials",",","use","`","tune.TuneConfig","(","num_samples=","<","int",">",")","`",".","In","{","LOG_DIR","}","\/","{","MY_EXPERIMENT_NAME","}","\/",",","all","mutations","are","logged","in","`","pbt_global.txt","`","and","individual","policy","perturbations","are","recorded","in","pbt_policy_","{","i","}",".txt",".","Tune","logs",":","[","target","trial","tag",",","clone","trial","tag",",","target","trial","iteration",",","clone","trial","iteration",",","old","config",",","new","config","]","on","each","perturbation","step",".","Args",":","time_attr",":","The","training","result","attr","to","use","for","comparing","time",".","Note","that","you","can","pass","in","something","non-temporal","such","as","`","training_iteration","`","as","a","measure","of","progress",",","the","only","requirement","is","that","the","attribute","should","increase","monotonically",".","metric",":","The","training","result","objective","value","attribute",".","Stopping","procedures","will","use","this","attribute",".","If","None","but","a","mode","was","passed",",","the","`","ray.tune.result.DEFAULT_METRIC","`","will","be","used","per","default",".","mode",":","One","of","{","min",",","max","}",".","Determines","whether","objective","is","minimizing","or","maximizing","the","metric","attribute",".","perturbation_interval",":","Models","will","be","considered","for","perturbation","at","this","interval","of","`","time_attr","`",".","Note","that","perturbation","incurs","checkpoint","overhead",",","so","you","should","n't","set","this","to","be","too","frequent",".","burn_in_period",":","Models","will","not","be","considered","for","perturbation","before","this","interval","of","`","time_attr","`","has","passed",".","This","guarantees","that","models","are","trained","for","at","least","a","certain","amount","of","time","or","timesteps","before","being","perturbed",".","hyperparam_mutations",":","Hyperparams","to","mutate",".","The","format","is","as","follows",":","for","each","key",",","either","a","list",",","function",",","or","a","tune","search","space","object","(","tune.loguniform",",","tune.uniform",",","etc",".",")","can","be","provided",".","A","list","specifies","an","allowed","set","of","categorical","values",".","A","function","or","tune","search","space","object","specifies","the","distribution","of","a","continuous","parameter",".","You","must","use","tune.choice",",","tune.uniform",",","tune.loguniform",",","etc","..","Arbitrary","tune.sample_from","objects","are","not","supported",".","A","key","can","also","hold","a","dict","for","nested","hyperparameters",".","You","must","specify","at","least","one","of","`","hyperparam_mutations","`","or","`","custom_explore_fn","`",".","Tune","will","sample","the","search","space","provided","by","`","hyperparam_mutations","`","for","the","initial","hyperparameter","values","if","the","corresponding","hyperparameters","are","not","present","in","a","trial","'s","initial","`","config","`",".","quantile_fraction",":","Parameters","are","transferred","from","the","top","`","quantile_fraction","`","fraction","of","trials","to","the","bottom","`","quantile_fraction","`","fraction",".","Needs","to","be","between","0","and","0.5",".","Setting","it","to","0","essentially","implies","doing","no","exploitation","at","all",".","resample_probability",":","The","probability","of","resampling","from","the","original","distribution","when","applying","`","hyperparam_mutations","`",".","If","not","resampled",",","the","value","will","be","perturbed","by","a","factor","chosen","from","`","perturbation_factors","`","if","continuous",",","or","changed","to","an","adjacent","value","if","discrete",".","perturbation_factors",":","Scaling","factors","to","choose","between","when","mutating","a","continuous","hyperparameter",".","custom_explore_fn",":","You","can","also","specify","a","custom","exploration","function",".","This","function","is","invoked","as","`","f","(","config",")","`","after","built-in","perturbations","from","`","hyperparam_mutations","`","are","applied",",","and","should","return","`","config","`","updated","as","needed",".","You","must","specify","at","least","one","of","`","hyperparam_mutations","`","or","`","custom_explore_fn","`",".","log_config",":","Whether","to","log","the","ray","config","of","each","model","to","local_dir","at","each","exploit",".","Allows","config","schedule","to","be","reconstructed",".","require_attrs",":","Whether","to","require","time_attr","and","metric","to","appear","in","result","for","every","iteration",".","If","True",",","error","will","be","raised","if","these","values","are","not","present","in","trial","result",".","synch",":","If","False",",","will","use","asynchronous","implementation","of","PBT",".","Trial","perturbations","occur","every","perturbation_interval","for","each","trial","independently",".","If","True",",","will","use","synchronous","implementation","of","PBT",".","Perturbations","will","occur","only","after","all","trials","are","synced","at","the","same","time_attr","every","perturbation_interval",".","Defaults","to","False",".","See","Appendix","A.1","here","https",":","\/\/arxiv.org\/pdf\/1711.09846.pdf",".","..","code-block",":",":","python","import","random","from","ray","import","tune","from","ray.tune.schedulers","import","PopulationBasedTraining","pbt","=","PopulationBasedTraining","(","time_attr=","''","training_iteration","''",",","metric=","''","episode_reward_mean","''",",","mode=","''","max","''",",","perturbation_interval=10",",","#","every","10","`","time_attr","`","units","#","(","training_iterations","in","this","case",")","hyperparam_mutations=","{","#","Perturb","factor1","by","scaling","it","by","0.8","or","1.2",".","Resampling","#","resets","it","to","a","value","sampled","from","the","lambda","function",".","``","factor_1","''",":","lambda",":","random.uniform","(","0.0",",","20.0",")",",","#","Alternatively",",","use","tune","search","space","primitives",".","#","The","search","space","for","factor_1","is","equivalent","to","factor_2",".","``","factor_2","''",":","tune.uniform","(","0.0",",","20.0",")",",","#","Perturb","factor3","by","changing","it","to","an","adjacent","value",",","e.g",".","#","10","-",">","1","or","10","-",">","100",".","Resampling","will","choose","at","random",".","``","factor_3","''",":","[","1",",","10",",","100",",","1000",",","10000","]",",","#","Using","tune.choice","is","NOT","equivalent","to","the","above",".","#","factor_4","is","treated","as","a","continuous","hyperparameter",".","``","factor_4","''",":","tune.choice","(","[","1",",","10",",","100",",","1000",",","10000","]",")",",","}",")","tuner","=","tune.Tuner","(","trainable",",","tune_config=tune.TuneConfig","(","scheduler=pbt",",","num_samples=8",",",")",",",")","tuner.fit","(",")"]
code: def on_trial_result(
      self,
      tune_controller: "TuneController",
      trial: Trial,
      result: Dict,
  ) -> str:
      if self._time_attr not in result:
          time_missing_msg = (
              "Cannot find time_attr {} "
              "in trial result {}. Make sure that this "
              "attribute is returned in the "
              "results of your Trainable.".format(
                  self._time_attr, result
              )
          )
          if self._require_attrs:
              raise RuntimeError(
                  time_missing_msg
                  + "If this error is expected, you can change this to "
                  "a warning message by "
                  "setting PBT(require_attrs=False)"
              )
          else:
              if log_once("pbt-time_attr-error"):
                  logger.warning(time_missing_msg)
      if self._metric not in result:
          metric_missing_msg = (
              "Cannot find metric {} in trial result {}. "
              "Make sure that this attribute is returned "
              "in the "
              "results of your Trainable.".format(self._metric, result)
          )
          if self._require_attrs:
              raise RuntimeError(
                  metric_missing_msg + "If this error is expected, "
                  "you can change this to a warning message by "
                  "setting PBT(require_attrs=False)"
              )
          else:
              if log_once("pbt-metric-error"):
                  logger.warning(metric_missing_msg)
  
      if self._metric not in result or self._time_attr not in result:
          return TrialScheduler.CONTINUE
  
      time = result[self._time_attr]
      state = self._trial_state[trial]
  
      # Continue training if burn-in period has not been reached, yet.
      if time < self._burn_in_period:
          logger.debug(
              f"Still in burn-in period: {time} < {self._burn_in_period}"
          )
          return TrialScheduler.CONTINUE
  
      # Continue training if perturbation interval has not been reached, yet.
      time_since_perturb = time - state.last_perturbation_time
      if time_since_perturb < self._perturbation_interval:
          logger.debug(
              f"Perturbation interval not reached: "
              f"{time_since_perturb} < {self._perturbation_interval}"
          )
          return TrialScheduler.CONTINUE  # avoid checkpoint overhead
  
      logger.debug(
          f"Updating trial state for trial {trial} at time {time}"
      )
      self._save_trial_state(state, time, result, trial)
  
      if not self._synch:
          state.last_perturbation_time = time
          lower_quantile, upper_quantile = self._quantiles()
          decision = TrialScheduler.CONTINUE
          for other_trial in tune_controller.get_trials():
              if other_trial.status in [Trial.PENDING, Trial.PAUSED]:
                  decision = TrialScheduler.PAUSE
                  break
          self._checkpoint_or_exploit(
              trial, tune_controller, upper_quantile, lower_quantile
          )
          return (
              TrialScheduler.NOOP
              if trial.status == Trial.PAUSED
              else decision
          )
      else:
          # Synchronous mode.
          if any(
              self._trial_state[t].last_train_time
              < self._next_perturbation_sync
              and t != trial
              for t in tune_controller.get_live_trials()
          ):
              logger.debug(
                  f"Sync: Other trials are not at perturb time, yet. "
                  f"Pausing trial {trial} to wait."
              )
          else:
              # All trials are synced at the same timestep.
              logger.debug("Sync: All trials are at perturb time.")
              lower_quantile, upper_quantile = self._quantiles()
              all_trials = tune_controller.get_trials()
              not_in_quantile = []
              for t in all_trials:
                  if (
                      t not in lower_quantile
                      and t not in upper_quantile
                  ):
                      not_in_quantile.append(t)
  
              logger.debug(
                  "Trial statistics\n"
                  f"Upper quantile: {upper_quantile}\n"
                  f"Lower quantile: {lower_quantile}\n"
                  f"Not in quantile: {not_in_quantile}"
              )
  
              # Move upper quantile trials to beginning and lower quantile
              # to end. This ensures that checkpointing of strong trials
              # occurs before exploiting of weaker ones.
              all_trials = (
                  upper_quantile + not_in_quantile + lower_quantile
              )
              for t in all_trials:
                  logger.debug(f"Perturbing trial {t}")
                  self._trial_state[t].last_perturbation_time = time
                  self._checkpoint_or_exploit(
                      t, tune_controller, upper_quantile, lower_quantile
                  )
  
              all_train_times = [
                  self._trial_state[t].last_train_time
                  for t in tune_controller.get_trials()
              ]
              max_last_train_time = max(all_train_times)
              self._next_perturbation_sync = max(
                  self._next_perturbation_sync
                  + self._perturbation_interval,
                  max_last_train_time,
              )
              logger.debug(
                  f"Next perturb at time {self._next_perturbation_sync}"
              )
          # In sync mode we should pause all trials once result comes in.
          # Once a perturbation step happens for all trials, they should
          # still all be paused.
          # choose_trial_to_run will then pick the next trial to run out of
          # the paused trials.
          return (
              TrialScheduler.NOOP
              if trial.status == Trial.PAUSED
              else TrialScheduler.PAUSE
          )
  
imports: import copy
  import json
  import logging
  import math
  import os
  import random
  import shutil
  import warnings
  from pathlib import Path
  from typing import Callable, Dict, List, Optional, Tuple, Union, TYPE_CHECKING
  from ray.air.constants import TRAINING_ITERATION
  from ray.train import Checkpoint
  from ray.train._internal.session import _TrainingResult, _FutureTrainingResult
  from ray.tune.error import TuneError
  from ray.tune.result import DEFAULT_METRIC
  from ray.tune.search import SearchGenerator
  from ray.tune.utils.util import SafeFallbackEncoder
  from ray.tune.search.sample import Domain, Function
  from ray.tune.schedulers import FIFOScheduler, TrialScheduler
  from ray.tune.search.variant_generator import format_vars
  from ray.tune.experiment import Trial
  from ray.util import PublicAPI
  from ray.util.debug import log_once
label_desc: Use image node_id 4 for calling the PopulationBasedTraining obj's underlying member method code with example usage: obj.on_trial_result(tune_controller, trial, result) and returns: TrialScheduler, TrialScheduler, TrialScheduler, unknown, unknown