text: """!
   * Copyright (c) Microsoft Corporation. All rights reserved.
   * Licensed under the MIT License. See LICENSE file in the
   * project root for license information.
  """
  from typing import Optional, Union, List, Callable, Tuple
  import numpy as np
  import datetime
  import time
  
  try:
      from ray import __version__ as ray_version
  
      assert ray_version >= "1.0.0"
      from ray.tune.analysis import ExperimentAnalysis as EA
  
      ray_import = True
  except (ImportError, AssertionError):
      ray_import = False
      from .analysis import ExperimentAnalysis as EA
  from .result import DEFAULT_METRIC
  import logging
  
  logger = logging.getLogger(__name__)
  
  
  _use_ray = True
  _runner = None
  _verbose = 0
  _running_trial = None
  _training_iteration = 0
  
  
  class ExperimentAnalysis(EA):
      """Class for storing the experiment results"""
  
      def __init__(self, trials, metric, mode):
          try:
              super().__init__(self, None, trials, metric, mode)
          except (TypeError, ValueError):
              self.trials = trials
              self.default_metric = metric or DEFAULT_METRIC
              self.default_mode = mode
  
  
  def report(_metric=None, **kwargs):
      """A function called by the HPO application to report final or intermediate
      results.
  
      Example:
  
      .. code-block:: python
  
          import time
          from flaml import tune
  
          def compute_with_config(config):
              current_time = time.time()
              metric2minimize = (round(config['x'])-95000)**2
              time2eval = time.time() - current_time
              tune.report(metric2minimize=metric2minimize, time2eval=time2eval)
  
          analysis = tune.run(
              compute_with_config,
              config={
                  'x': tune.lograndint(lower=1, upper=1000000),
                  'y': tune.randint(lower=1, upper=1000000)
              },
              metric='metric2minimize', mode='min',
              num_samples=1000000, time_budget_s=60, use_ray=False)
  
          print(analysis.trials[-1].last_result)
  
      Args:
          _metric: Optional default anonymous metric for ``tune.report(value)``.
              (For compatibility with ray.tune.report)
          **kwargs: Any key value pair to be reported.
      """
      global _use_ray
      global _verbose
      global _running_trial
      global _training_iteration
      if _use_ray:
          from ray import tune
  
          return tune.report(_metric, **kwargs)
      else:
          result = kwargs
          if _metric:
              result[DEFAULT_METRIC] = _metric
          trial = _runner.running_trial
          if _running_trial == trial:
              _training_iteration += 1
          else:
              _training_iteration = 0
              _running_trial = trial
          result["training_iteration"] = _training_iteration
          result["config"] = trial.config
          for key, value in trial.config.items():
              result["config/" + key] = value
          _runner.process_trial_result(_runner.running_trial, result)
          result["time_total_s"] = trial.last_update_time - trial.start_time
          if _verbose > 2:
              logger.info(f"result: {result}")
          if _runner.running_trial.is_finished():
              return None
          else:
              return True
  
  
  def run(
      training_function,
      config: Optional[dict] = None,
      low_cost_partial_config: Optional[dict] = None,
      cat_hp_cost: Optional[dict] = None,
      metric: Optional[str] = None,
      mode: Optional[str] = None,
      time_budget_s: Union[int, float] = None,
      points_to_evaluate: Optional[List[dict]] = None,
      evaluated_rewards: Optional[List] = None,
      prune_attr: Optional[str] = None,
      min_resource: Optional[float] = None,
      max_resource: Optional[float] = None,
      reduction_factor: Optional[float] = None,
      report_intermediate_result: Optional[bool] = False,
      search_alg=None,
      verbose: Optional[int] = 2,
      local_dir: Optional[str] = None,
      num_samples: Optional[int] = 1,
      resources_per_trial: Optional[dict] = None,
      config_constraints: Optional[
          List[Tuple[Callable[[dict], float], str, float]]
      ] = None,
      metric_constraints: Optional[List[Tuple[str, str, float]]] = None,
      max_failure: Optional[int] = 100,
      use_ray: Optional[bool] = False,
  ):
      """The trigger for HPO.
  
      Example:
  
      .. code-block:: python
  
          import time
          from flaml import tune
  
          def compute_with_config(config):
              current_time = time.time()
              metric2minimize = (round(config['x'])-95000)**2
              time2eval = time.time() - current_time
              tune.report(metric2minimize=metric2minimize, time2eval=time2eval)
  
          analysis = tune.run(
              compute_with_config,
              config={
                  'x': tune.lograndint(lower=1, upper=1000000),
                  'y': tune.randint(lower=1, upper=1000000)
              },
              metric='metric2minimize', mode='min',
              num_samples=-1, time_budget_s=60, use_ray=False)
  
          print(analysis.trials[-1].last_result)
  
      Args:
          training_function: A user-defined training function.
          config: A dictionary to specify the search space.
          low_cost_partial_config: A dictionary from a subset of
              controlled dimensions to the initial low-cost values.
              e.g.,
  
              .. code-block:: python
  
                  {'n_estimators': 4, 'max_leaves': 4}
  
          cat_hp_cost: A dictionary from a subset of categorical dimensions
              to the relative cost of each choice.
              e.g.,
  
              .. code-block:: python
  
                  {'tree_method': [1, 1, 2]}
  
              i.e., the relative cost of the
              three choices of 'tree_method' is 1, 1 and 2 respectively
          metric: A string of the metric name to optimize for.
          mode: A string in ['min', 'max'] to specify the objective as
              minimization or maximization.
          time_budget_s: int or float | The time budget in seconds.
          points_to_evaluate: A list of initial hyperparameter
              configurations to run first.
          evaluated_rewards (list): If you have previously evaluated the
              parameters passed in as points_to_evaluate you can avoid
              re-running those trials by passing in the reward attributes
              as a list so the optimiser can be told the results without
              needing to re-compute the trial. Must be the same length as
              points_to_evaluate.
              e.g.,
              .. code-block:: python
                  points_to_evaluate = [
                      {"b": .99, "cost_related": {"a": 3}},
                      {"b": .99, "cost_related": {"a": 2}},
                  ]
                  evaluated_rewards=[3.0, 1.0]
  
              means that you know the reward for the two configs in
              points_to_evaluate are 3.0 and 1.0 respectively and want to
              inform run()
  
          prune_attr: A string of the attribute used for pruning.
              Not necessarily in space.
              When prune_attr is in space, it is a hyperparameter, e.g.,
              'n_iters', and the best value is unknown.
              When prune_attr is not in space, it is a resource dimension,
              e.g., 'sample_size', and the peak performance is assumed
              to be at the max_resource.
          min_resource: A float of the minimal resource to use for the
              prune_attr; only valid if prune_attr is not in space.
          max_resource: A float of the maximal resource to use for the
              prune_attr; only valid if prune_attr is not in space.
          reduction_factor: A float of the reduction factor used for incremental
              pruning.
          report_intermediate_result: A boolean of whether intermediate results
              are reported. If so, early stopping and pruning can be used.
          search_alg: An instance of BlendSearch as the search algorithm
              to be used. The same instance can be used for iterative tuning.
              e.g.,
  
              .. code-block:: python
  
                  from flaml import BlendSearch
                  algo = BlendSearch(metric='val_loss', mode='min',
                          space=search_space,
                          low_cost_partial_config=low_cost_partial_config)
                  for i in range(10):
                      analysis = tune.run(compute_with_config,
                          search_alg=algo, use_ray=False)
                      print(analysis.trials[-1].last_result)
  
          verbose: 0, 1, 2, or 3. Verbosity mode for ray if ray backend is used.
              0 = silent, 1 = only status updates, 2 = status and brief trial
              results, 3 = status and detailed trial results. Defaults to 2.
          local_dir: A string of the local dir to save ray logs if ray backend is
              used; or a local dir to save the tuning log.
          num_samples: An integer of the number of configs to try. Defaults to 1.
          resources_per_trial: A dictionary of the hardware resources to allocate
              per trial, e.g., `{'cpu': 1}`. Only valid when using ray backend.
          config_constraints: A list of config constraints to be satisfied.
              e.g.,
  
              .. code-block: python
  
                  config_constraints = [(mem_size, '<=', 1024**3)]
  
              mem_size is a function which produces a float number for the bytes
              needed for a config.
              It is used to skip configs which do not fit in memory.
          metric_constraints: A list of metric constraints to be satisfied.
              e.g., `['precision', '>=', 0.9]`.
          max_failure: int | the maximal consecutive number of failures to sample
              a trial before the tuning is terminated.
          use_ray: A boolean of whether to use ray as the backend.
      """
      global _use_ray
      global _verbose
      if not use_ray:
          _verbose = verbose
          if verbose > 0:
              import os
  
              if local_dir:
                  os.makedirs(local_dir, exist_ok=True)
                  logger.addHandler(
                      logging.FileHandler(
                          local_dir
                          + "/tune_"
                          + str(datetime.datetime.now()).replace(":", "-")
                          + ".log"
                      )
                  )
              elif not logger.handlers:
                  # Add the console handler.
                  _ch = logging.StreamHandler()
                  logger_formatter = logging.Formatter(
                      "[%(name)s: %(asctime)s] {%(lineno)d} %(levelname)s - %(message)s",
                      "%m-%d %H:%M:%S",
                  )
                  _ch.setFormatter(logger_formatter)
                  logger.addHandler(_ch)
              if verbose <= 2:
                  logger.setLevel(logging.INFO)
              else:
                  logger.setLevel(logging.DEBUG)
          else:
              logger.setLevel(logging.CRITICAL)
  
      from ..searcher.blendsearch import BlendSearch
  
      if search_alg is None:
          search_alg = BlendSearch(
              metric=metric or DEFAULT_METRIC,
              mode=mode,
              space=config,
              points_to_evaluate=points_to_evaluate,
              evaluated_rewards=evaluated_rewards,
              low_cost_partial_config=low_cost_partial_config,
              cat_hp_cost=cat_hp_cost,
              time_budget_s=time_budget_s,
              num_samples=num_samples,
              prune_attr=prune_attr,
              min_resource=min_resource,
              max_resource=max_resource,
              reduction_factor=reduction_factor,
              config_constraints=config_constraints,
              metric_constraints=metric_constraints,
          )
      else:
          if metric is None or mode is None:
              metric = metric or search_alg.metric
              mode = mode or search_alg.mode
          if ray_import:
              from ray.tune.suggest import ConcurrencyLimiter
          else:
              from flaml.searcher.suggestion import ConcurrencyLimiter
          searcher = (
              search_alg.searcher
              if isinstance(search_alg, ConcurrencyLimiter)
              else search_alg
          )
          if isinstance(searcher, BlendSearch):
              setting = {}
              if time_budget_s:
                  setting["time_budget_s"] = time_budget_s
              if num_samples > 0:
                  setting["num_samples"] = num_samples
              searcher.set_search_properties(metric, mode, config, setting)
          else:
              searcher.set_search_properties(metric, mode, config)
      scheduler = None
      if report_intermediate_result:
          params = {}
          # scheduler resource_dimension=prune_attr
          if prune_attr:
              params["time_attr"] = prune_attr
          if max_resource:
              params["max_t"] = max_resource
          if min_resource:
              params["grace_period"] = min_resource
          if reduction_factor:
              params["reduction_factor"] = reduction_factor
          if ray_import:
              from ray.tune.schedulers import ASHAScheduler
  
              scheduler = ASHAScheduler(**params)
      if use_ray:
          try:
              from ray import tune
          except ImportError:
              raise ImportError(
                  "Failed to import ray tune. "
                  "Please install ray[tune] or set use_ray=False"
              )
          _use_ray = True
          return tune.run(
              training_function,
              metric=metric,
              mode=mode,
              search_alg=search_alg,
              scheduler=scheduler,
              time_budget_s=time_budget_s,
              verbose=verbose,
              local_dir=local_dir,
              num_samples=num_samples,
              resources_per_trial=resources_per_trial,
          )
  
      # simple sequential run without using tune.run() from ray
      time_start = time.time()
      _use_ray = False
      if scheduler:
          scheduler.set_search_properties(metric=metric, mode=mode)
      from .trial_runner import SequentialTrialRunner
  
      global _runner
      _runner = SequentialTrialRunner(
          search_alg=search_alg,
          scheduler=scheduler,
          metric=metric,
          mode=mode,
      )
      num_trials = 0
      if time_budget_s is None:
          time_budget_s = np.inf
      fail = 0
      ub = (len(evaluated_rewards) if evaluated_rewards else 0) + max_failure
      while (
          time.time() - time_start < time_budget_s
          and (num_samples < 0 or num_trials < num_samples)
          and fail < ub
      ):
          trial_to_run = _runner.step()
          if trial_to_run:
              num_trials += 1
              if verbose:
                  logger.info(f"trial {num_trials} config: {trial_to_run.config}")
              result = training_function(trial_to_run.config)
              if result is not None:
                  if isinstance(result, dict):
                      report(**result)
                  else:
                      report(_metric=result)
              _runner.stop_trial(trial_to_run)
              fail = 0
          else:
              fail += 1  # break with ub consecutive failures
      if fail == ub:
          logger.warning(
              f"fail to sample a trial for {max_failure} times in a row, stopping."
          )
      if verbose > 0:
          logger.handlers.clear()
      return ExperimentAnalysis(_runner.get_trials(), metric=metric, mode=mode)
  