code: def run(
      trainable: Callable,
      trainable_args: dict,
      search_space: dict,
      hyperparameter_tune_kwargs: Union[dict, str],
      metric: str,
      mode: str,
      save_dir: str,
      ray_tune_adapter: RayTuneAdapter,
      trainable_is_parallel: bool = False,
      total_resources: Optional[dict] = dict(),
      minimum_cpu_per_trial: int = 1,
      minimum_gpu_per_trial: float = 0.0,
      model_estimate_memory_usage: Optional[int] = None,
      time_budget_s: Optional[float] = None,
      verbose: int = 1,
      tune_config_kwargs: Optional[dict] = None,
      run_config_kwargs: Optional[dict] = None,
  ) -> tune.ExperimentAnalysis:
      """
      Parse hyperparameter_tune_kwargs
      Init necessary objects, i.e. searcher, scheduler, and ray
      Translate search space
      Calculate resource per trial and update trainable_args accordingly
      Finally tune.run
  
      Parameters
      ----------
      trainable
          The function used to train the model.
      trainable_args
          Args passed to trainable.
      search_space
          Search space for HPO.
      hyperparameter_tune_kwargs
          User specified HPO options.
      metric
          Name of the monitored metric for HPO.
      mode
          Determines whether objective is minimizing or maximizing the metric.
          Must be one of [min, max].
      save_dir
          Directory to save ray tune results. Ray will write artifacts to save_dir/trial_dir/
          While HPO, ray will chdir to this directory. Therefore, it's important to provide dataset or model saving path as absolute path.
          After HPO, we change back to the original working directory.
          trial_dir name has been overwritten to be trial_id.
          To provide custom trial_dir, pass a custom function to `trial_dirname_creator` as kwargs.
          For example of creating the custom function, refer to `_trial_dirname_creator`.
      ray_tune_adapter
          Adapter to provide necessary custom info to ray tune.
      trainable_is_parallel
          Whether the trainable itself will use ray to run parallel job or not.
      total_resources
          Total resources can be used for HPO.
          If not specified, will use all the resources by default.
      minimum_cpu_per_trial
          Specify minimum number of cpu to use per trial.
      minimum_gpu_per_trial
          Specify minimum number of gpu to use per trial.
          Ray supports usage of fraction of gpu.
      model_estimate_memory_usage
          Provide optional estimate of the model memory usage.
          Calculation of the resources_per_trial might use this info to better distribute resources
      time_budget_s
          Time limit for the HPO.
      verbose
          0 = silent, 1 = only status updates, 2 = status and brief trial results, 3 = status and detailed trial results.
      tune_config_kwargs
          Additional args being passed to tune.TuneConfig https://docs.ray.io/en/latest/ray-air/package-ref.html#ray.tune.tune_config.TuneConfig
      run_config_kwargs
          Additional args being passed to air.RunConfig https://docs.ray.io/en/latest/ray-air/package-ref.html#ray.air.config.RunConfig
      """
      assert mode in [
          MIN,
          MAX,
      ], f"mode {mode} is not a valid option. Options are {[MIN, MAX]}"
      if isinstance(hyperparameter_tune_kwargs, str):
          assert (
              hyperparameter_tune_kwargs in ray_tune_adapter.presets
          ), f"{hyperparameter_tune_kwargs} is not a valid option. Options are {ray_tune_adapter.presets.keys()}"
          hyperparameter_tune_kwargs = ray_tune_adapter.presets.get(
              hyperparameter_tune_kwargs
          )
      num_samples = hyperparameter_tune_kwargs.get("num_trials", None)
      if num_samples is None:
          num_samples = (
              1 if time_budget_s is None else 1000
          )  # if both num_samples and time_budget_s are None, we only run 1 trial
      if not any(
          isinstance(search_space[hyperparam], (ag_space.Space, Domain))
          for hyperparam in search_space
      ):
          raise EmptySearchSpace
      search_space, default_hyperparameters = _convert_search_space(
          search_space
      )
  
      searcher = _get_searcher(
          hyperparameter_tune_kwargs=hyperparameter_tune_kwargs,
          metric=metric,
          mode=mode,
          default_hyperparameters=default_hyperparameters,
          supported_searchers=ray_tune_adapter.get_supported_searchers(),
      )
      scheduler = _get_scheduler(
          hyperparameter_tune_kwargs=hyperparameter_tune_kwargs,
          supported_schedulers=ray_tune_adapter.get_supported_schedulers(),
      )
  
      if not ray.is_initialized():
          if DistributedContext.is_distributed_mode():
              ray.init(address="auto")
          else:
              ray.init(
                  log_to_driver=False,
                  runtime_env={
                      "env_vars": {"PL_DISABLE_FORK": "1"}
                  },  # https://github.com/ray-project/ray/issues/28197
                  logging_level=logging.ERROR,  # https://github.com/ray-project/ray/issues/29216
                  **total_resources,
              )
  
      resources_per_trial = hyperparameter_tune_kwargs.get(
          "resources_per_trial", None
      )
      resources_per_trial = ray_tune_adapter.get_resources_per_trial(
          total_resources=total_resources,
          num_samples=num_samples,
          resources_per_trial=resources_per_trial,
          minimum_gpu_per_trial=minimum_gpu_per_trial,
          minimum_cpu_per_trial=minimum_cpu_per_trial,
          model_estimate_memory_usage=model_estimate_memory_usage,
          wrap_resources_per_job_into_placement_group=trainable_is_parallel,
      )
      resources_per_trial = _validate_resources_per_trial(
          resources_per_trial
      )
      logger.debug(
          f"resources_per_trial to be dispatched by ray tune: {resources_per_trial}"
      )
      ray_tune_adapter.resources_per_trial = resources_per_trial
      trainable_args = ray_tune_adapter.trainable_args_update_method(
          trainable_args
      )
  
      original_path = os.getcwd()
      save_dir = os.path.normpath(save_dir)
      if tune_config_kwargs is None:
          tune_config_kwargs = dict()
      if run_config_kwargs is None:
          run_config_kwargs = dict()
      tuner = tune.Tuner(
          tune.with_resources(
              tune.with_parameters(trainable, **trainable_args),
              resources_per_trial,
          ),
          param_space=search_space,
          tune_config=tune.TuneConfig(
              num_samples=num_samples,
              search_alg=searcher,
              scheduler=scheduler,
              metric=metric,
              mode=mode,
              time_budget_s=time_budget_s,
              **tune_config_kwargs,
          ),
          run_config=air.RunConfig(
              name=os.path.basename(save_dir),
              local_dir=os.path.dirname(save_dir),
              verbose=verbose,
              **run_config_kwargs,
          ),
          _tuner_kwargs={
              "trial_name_creator": _trial_name_creator,
              "trial_dirname_creator": _trial_dirname_creator,
          },
      )
      results = tuner.fit()
  
      os.chdir(
          original_path
      )  # go back to the original directory to avoid relative path being broken
      return results._experiment_analysis
  
imports: import logging
  import os
  import shutil
  from autogluon.common import space
  from autogluon.common.utils.distribute_utils import DistributedContext
  from autogluon.common.utils.try_import import try_import_ray
  from abc import ABC, abstractmethod
  from typing import Callable, List, Optional, Union
  import ray
  from ray import air, tune
  from ray.tune import ExperimentAnalysis, PlacementGroupFactory
  from ray.tune.schedulers import TrialScheduler
  from ray.tune.search import SearchAlgorithm, Searcher
  from ray.tune.search.sample import Domain
  from autogluon.common.utils.resource_utils import ResourceManager
  from ..ray.resources_calculator import ResourceCalculator, ResourceCalculatorFactory
  from .constants import MAX, MIN
  from .exceptions import EmptySearchSpace
  from .ray_tune_constants import SCHEDULER_PRESETS, SEARCHER_PRESETS
  from .ray_tune_scheduler_factory import SchedulerFactory
  from .ray_tune_searcher_factory import SearcherFactory
  from .space_converter import RaySpaceConverterFactory
label_desc: Use image node_id 1 for calling a global function with example usage: run(trainable, trainable_args, search_space, hyperparameter_tune_kwargs, metric, mode, save_dir, ray_tune_adapter, trainable_is_parallel, total_resources, minimum_cpu_per_trial, minimum_gpu_per_trial, model_estimate_memory_usage, time_budget_s, verbose, tune_config_kwargs, run_config_kwargs) and returns: results