class_docstr: Interface for abstract model to executor HPO runs
docstr: Register total resources used for the experiment, and calculate resources per trial if user specified.
  User specified resources per trial will be validated against total resources and minimum resources required, and respected directly if legit.
  When HPO with bagging, user could specify resources per fold as well as resources per trial.
      Resources per fold will be checked against total resources, minimum resources required, and resources per trial (if specified).
      Resources per fold will have higher priority than resources per trial, and the corresponding resources per trial will be calculated accordingly.
  When no user specified resources present, we try to maximize trials running in parallel while respecting the minimum resources required.
  
  Parameters
  ----------
  initialized_model
      The model that will be performed HPO. This model MUST be initialized.
  num_cpus
      Total number of cpus available for the experiment.
  num_gpus
      Total number of gpus available for the experiment.
  k_fold
      Number of folds if bagging. Used to check if an individual trial is a bagged model.
  kwargs
      Any additional parameters being passed to `AbstractModel.hyperparameter_tune()`
      This function will pass these parameters to initialized model to get estimation of memory usage
docstr_tok: ["Register","total","resources","used","for","the","experiment",",","and","calculate","resources","per","trial","if","user","specified",".","User","specified","resources","per","trial","will","be","validated","against","total","resources","and","minimum","resources","required",",","and","respected","directly","if","legit",".","When","HPO","with","bagging",",","user","could","specify","resources","per","fold","as","well","as","resources","per","trial",".","Resources","per","fold","will","be","checked","against","total","resources",",","minimum","resources","required",",","and","resources","per","trial","(","if","specified",")",".","Resources","per","fold","will","have","higher","priority","than","resources","per","trial",",","and","the","corresponding","resources","per","trial","will","be","calculated","accordingly",".","When","no","user","specified","resources","present",",","we","try","to","maximize","trials","running","in","parallel","while","respecting","the","minimum","resources","required",".","Parameters","--","--","--","--","--","initialized_model","The","model","that","will","be","performed","HPO",".","This","model","MUST","be","initialized",".","num_cpus","Total","number","of","cpus","available","for","the","experiment",".","num_gpus","Total","number","of","gpus","available","for","the","experiment",".","k_fold","Number","of","folds","if","bagging",".","Used","to","check","if","an","individual","trial","is","a","bagged","model",".","kwargs","Any","additional","parameters","being","passed","to","`","AbstractModel.hyperparameter_tune","(",")","`","This","function","will","pass","these","parameters","to","initialized","model","to","get","estimation","of","memory","usage"]
code: def register_resources(
      self,
      initialized_model: AbstractModel,
      num_cpus: int,
      num_gpus: Union[int, float],
      k_fold: Optional[int] = None,
      **kwargs,
  ):
      """
      Register total resources used for the experiment, and calculate resources per trial if user specified.
      User specified resources per trial will be validated against total resources and minimum resources required, and respected directly if legit.
      When HPO with bagging, user could specify resources per fold as well as resources per trial.
          Resources per fold will be checked against total resources, minimum resources required, and resources per trial (if specified).
          Resources per fold will have higher priority than resources per trial, and the corresponding resources per trial will be calculated accordingly.
      When no user specified resources present, we try to maximize trials running in parallel while respecting the minimum resources required.
  
      Parameters
      ----------
      initialized_model
          The model that will be performed HPO. This model MUST be initialized.
      num_cpus
          Total number of cpus available for the experiment.
      num_gpus
          Total number of gpus available for the experiment.
      k_fold
          Number of folds if bagging. Used to check if an individual trial is a bagged model.
      kwargs
          Any additional parameters being passed to `AbstractModel.hyperparameter_tune()`
          This function will pass these parameters to initialized model to get estimation of memory usage
      """
      minimum_model_resources = initialized_model.get_minimum_resources(
          is_gpu_available=(num_gpus > 0)
      )
      minimum_model_num_cpus = minimum_model_resources.get(
          "num_cpus", 1
      )
      minimum_model_num_gpus = minimum_model_resources.get(
          "num_gpus", 0
      )
      initialized_model_params = initialized_model.get_params()
  
      if (
          "hyperparameters" in initialized_model_params
          and "ag_args_fit"
          in initialized_model_params["hyperparameters"]
      ):
          user_specified_trial_num_cpus = initialized_model_params[
              "hyperparameters"
          ]["ag_args_fit"].get("num_cpus", None)
          user_specified_trial_num_gpus = initialized_model_params[
              "hyperparameters"
          ]["ag_args_fit"].get("num_gpus", None)
          if (
              user_specified_trial_num_cpus is not None
              or user_specified_trial_num_gpus is not None
          ):
              num_trials_in_parallel_with_gpu = math.inf
              if user_specified_trial_num_cpus is None:
                  # If user didn't specify cpu per trial, we find the min based on gpu
                  num_trials_in_parallel_with_gpu = (
                      num_gpus // user_specified_trial_num_gpus
                  )
                  user_specified_trial_num_cpus = (
                      num_cpus // num_trials_in_parallel_with_gpu
                  )  # keep gpus per trial int to avoid complexity
              num_trials_in_parallel_with_cpu = math.inf
              if user_specified_trial_num_gpus is None:
                  # If user didn't specify gpu per trial, we find the min based on cpu
                  num_trials_in_parallel_with_cpu = (
                      num_cpus // user_specified_trial_num_cpus
                  )
                  user_specified_trial_num_gpus = (
                      num_gpus // num_trials_in_parallel_with_cpu
                  )  # keep gpus per trial int to avoid complexity
              assert (
                  user_specified_trial_num_cpus <= num_cpus
              ), f"Detected trial level cpu requirement = {user_specified_trial_num_cpus} > total cpu granted to AG predictor = {num_cpus}"
              assert (
                  user_specified_trial_num_cpus
                  >= minimum_model_num_cpus
              ), f"The trial requires minimum cpu {minimum_model_num_cpus}, but you only specified {user_specified_trial_num_cpus}"
              assert (
                  user_specified_trial_num_gpus <= num_gpus
              ), f"Detected trial level gpu requirement = {user_specified_trial_num_gpus} > total gpu granted to AG predictor = {num_gpus}"
              assert (
                  user_specified_trial_num_gpus
                  >= minimum_model_num_gpus
              ), f"The trial requires minimum gpu {minimum_model_num_gpus}, but you only specified {user_specified_trial_num_gpus}"
  
              # Custom backend should set its total resource to be resources_per_trial
              self.hyperparameter_tune_kwargs["resources_per_trial"] = {
                  "num_cpus": user_specified_trial_num_cpus,
                  "num_gpus": user_specified_trial_num_gpus,
              }
  
      model_base = initialized_model._get_model_base()
      if model_base != initialized_model:
          # This is an ensemble model
          total_num_cpus_per_trial = num_cpus
          total_num_gpus_per_trial = num_gpus
          if "resources_per_trial" in self.hyperparameter_tune_kwargs:
              resources_per_trial = self.hyperparameter_tune_kwargs[
                  "resources_per_trial"
              ]
              total_num_cpus_per_trial = resources_per_trial.get(
                  "num_cpus"
              )
              total_num_gpus_per_trial = resources_per_trial.get(
                  "num_gpus"
              )
          user_specified_fold_resources = model_base._user_params_aux
          user_specified_fold_num_cpus = (
              user_specified_fold_resources.get("num_cpus", None)
          )  # We shouldn't always use it
          user_specified_fold_num_gpus = (
              user_specified_fold_resources.get("num_gpus", None)
          )
          if (
              user_specified_fold_num_cpus is not None
              or user_specified_fold_num_gpus is not None
          ):
              num_folds_in_parallel_with_cpu = math.inf
              if minimum_model_num_cpus > 0:
                  num_folds_in_parallel_with_cpu = (
                      total_num_cpus_per_trial // minimum_model_num_cpus
                  )
              if user_specified_fold_num_cpus is not None:
                  assert (
                      user_specified_fold_num_cpus
                      <= total_num_cpus_per_trial
                  ), f"Detected fold level cpu requirement = {user_specified_fold_num_cpus} > total cpu granted to AG predictor per trial= {total_num_cpus_per_trial}"
                  assert (
                      user_specified_fold_num_cpus
                      >= minimum_model_num_cpus
                  ), f"The model requires minimum cpu {minimum_model_num_cpus}, but you only specified {user_specified_fold_num_cpus}"
                  num_folds_in_parallel_with_cpu = (
                      total_num_cpus_per_trial
                      // user_specified_fold_num_cpus
                  )
              num_folds_in_parallel_with_gpu = math.inf
              if minimum_model_num_gpus > 0:
                  num_folds_in_parallel_with_gpu = (
                      total_num_gpus_per_trial // minimum_model_num_gpus
                  )
              if user_specified_fold_num_gpus is not None:
                  assert (
                      user_specified_fold_num_gpus
                      <= total_num_gpus_per_trial
                  ), f"Detected fold level gpu requirement = {user_specified_fold_num_gpus} > total gpu granted to AG predictor per trial = {total_num_gpus_per_trial}"
                  assert (
                      user_specified_fold_num_gpus
                      >= minimum_model_num_gpus
                  ), f"The model requires minimum gpu {minimum_model_num_gpus}, but you only specified {user_specified_fold_num_gpus}"
                  if minimum_model_num_gpus > 0:
                      num_folds_in_parallel_with_gpu = (
                          total_num_gpus_per_trial
                          // user_specified_fold_num_gpus
                      )
              num_folds_in_parallel = min(
                  num_folds_in_parallel_with_cpu,
                  num_folds_in_parallel_with_gpu,
              )
  
              cpu_per_trial = user_specified_fold_num_cpus * min(
                  k_fold, num_folds_in_parallel
              )
              gpu_per_trial = user_specified_fold_num_gpus * min(
                  k_fold, num_folds_in_parallel
              )
  
              # Custom backend should set its total resource to be resources_per_trial
              self.hyperparameter_tune_kwargs["resources_per_trial"] = {
                  "num_cpus": cpu_per_trial,
                  "num_gpus": gpu_per_trial,
              }
      if "resources_per_trial" not in self.hyperparameter_tune_kwargs:
          # User didn't provide any requirements
          num_jobs_in_parallel_with_mem = math.inf
  
          if initialized_model.estimate_memory_usage is not None:
              model_estimate_memory_usage = (
                  initialized_model.estimate_memory_usage(**kwargs)
              )
              total_memory_available = (
                  ResourceManager.get_available_virtual_mem()
              )
              num_jobs_in_parallel_with_mem = (
                  total_memory_available // model_estimate_memory_usage
              )
  
          num_jobs_in_parallel_with_cpu = (
              num_cpus // minimum_model_num_cpus
          )
          num_jobs_in_parallel_with_gpu = math.inf
          if minimum_model_num_gpus > 0:
              num_jobs_in_parallel_with_gpu = (
                  num_gpus // minimum_model_num_gpus
              )
          num_jobs_in_parallel = min(
              num_jobs_in_parallel_with_mem,
              num_jobs_in_parallel_with_cpu,
              num_jobs_in_parallel_with_gpu,
          )
          if k_fold is not None and k_fold > 0:
              max_models = (
                  self.hyperparameter_tune_kwargs.get(
                      "num_trials", math.inf
                  )
                  * k_fold
              )
              num_jobs_in_parallel = min(
                  num_jobs_in_parallel, max_models
              )
          system_num_cpu = ResourceManager.get_cpu_count()
          system_num_gpu = ResourceManager.get_gpu_count()
          if model_base != initialized_model:
              # bagged model
              if num_jobs_in_parallel // k_fold < 1:
                  # We can only train 1 trial in parallel
                  num_trials_in_parallel = 1
              else:
                  num_trials_in_parallel = (
                      num_jobs_in_parallel // k_fold
                  )
              if self.executor_type == "custom":
                  # custom backend runs sequentially
                  num_trials_in_parallel = 1
              cpu_per_trial = int(num_cpus // num_trials_in_parallel)
              gpu_per_trial = num_gpus // num_trials_in_parallel
          else:
              num_trials = self.hyperparameter_tune_kwargs.get(
                  "num_trials", math.inf
              )
              if self.executor_type == "custom":
                  # custom backend runs sequentially
                  num_jobs_in_parallel = 1
              cpu_per_trial = int(
                  num_cpus // min(num_jobs_in_parallel, num_trials)
              )
              gpu_per_trial = num_gpus / min(
                  num_jobs_in_parallel, num_trials
              )
          # In distributed setting, a single trial could be scheduled with resources that's more than a single node causing hanging
          # Force it to be less than the current node. This works under the assumption that all nodes are of the same type
          cpu_per_trial = min(cpu_per_trial, system_num_cpu)
          gpu_per_trial = min(gpu_per_trial, system_num_gpu)
  
          self.hyperparameter_tune_kwargs["resources_per_trial"] = {
              "num_cpus": cpu_per_trial,
              "num_gpus": gpu_per_trial,
          }
  
      self.resources = dict(num_gpus=num_gpus, num_cpus=num_cpus)
  
imports: from __future__ import annotations
  import copy
  import logging
  import math
  import os
  import time
  from abc import ABC, abstractmethod
  from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Union
  import pandas
  from autogluon.common import space
  from autogluon.common.utils.resource_utils import ResourceManager
  from autogluon.common.utils.s3_utils import is_s3_url
  from ..ray.resources_calculator import ResourceCalculator
  from ..scheduler.scheduler_factory import scheduler_factory
  from ..utils.savers import save_pkl
  from .constants import CUSTOM_BACKEND, RAY_BACKEND
  from .exceptions import EmptySearchSpace
label_desc: Use image node_id 6 for calling the HpoExecutor obj's underlying member method code with example usage: obj.register_resources(initialized_model, num_cpus, num_gpus, k_fold) without return types