

import importlib.util
import weakref
from functools import partialmethod

from .dependency_versions_check import dep_version_check
from .utils import is_accelerate_available, is_torch_available, logging


if is_torch_available():
    import torch

_UpperCamelCase     =    logging.get_logger(__name__)
def 		UpperCamelCase_(    ) ->				List[str]:
 return importlib.util.find_spec('deepspeed'     ) is not None


if is_accelerate_available() and is_deepspeed_available():
    from accelerate.utils.deepspeed import HfDeepSpeedConfig as DeepSpeedConfig
else:
    # Inherits from a dummy `object` if accelerate is not available, so that python succeeds to import this file.
    # Deepspeed glue code will never inherit this dummy object as it checks if accelerate is available.
    from builtins import object as DeepSpeedConfig



class 					lowercase		( _a      ):



 '''simple docstring'''


 def __init__(self , __a				)				->      List[Any]:






  """simple docstring"""





  set_hf_deepspeed_config(self				)
  dep_version_check('accelerate'				)
  dep_version_check('deepspeed'				)
  super().__init__(lowerCamelCase__				)







class 					lowercase		( _a      ):



 '''simple docstring'''


 def __init__(self , __a				)				->      Tuple:






  """simple docstring"""





  super().__init__(lowerCamelCase__				)
  UpperCAmelCase__           =       None
  UpperCAmelCase__           =       []




 def 						UpperCamelCase__							(self				)				->      Union[str, Any]:






  """simple docstring"""





  if self._dtype is None:
   raise ValueError('trainer_config_process() wasn\'t called yet to tell dtype'				)
  return self._dtype




 def 						UpperCamelCase__							(self , __a				)				->      List[Any]:






  """simple docstring"""





  UpperCAmelCase__           =       self.get_value(lowerCamelCase__				)
  if val is None:
   return False
  else:
   return val == "auto"




 def 						UpperCamelCase__							(self , __a , __a , __a=None , __a=True				)				->      List[Any]:






  """simple docstring"""





  UpperCAmelCase__           =       self.find_config_node(lowerCamelCase__				)
  if config is None:
   return

  if config.get(lowerCamelCase__				) == "auto":
   UpperCAmelCase__           =       hf_val
   return

  if not must_match:
   return

  UpperCAmelCase__           =       config.get(lowerCamelCase__				)
  if ds_val is not None and ds_val != hf_val:
   self.mismatches.append(F"- ds {ds_key_long}={ds_val} vs hf {hf_key}={hf_val}"				)

 __SCREAMING_SNAKE_CASE      							=       partialmethod(_a				,       must_match=_a      )




 def 						UpperCamelCase__							(self , __a				)				->      Any:






  """simple docstring"""





  UpperCAmelCase__           =       args.world_size * args.per_device_train_batch_size * args.gradient_accumulation_steps
  self.fill_match(
      'train_micro_batch_size_per_gpu' , args.per_device_train_batch_size , 'per_device_train_batch_size'				)
  self.fill_match('gradient_accumulation_steps' , args.gradient_accumulation_steps , 'gradient_accumulation_steps'				)
  self.fill_match('train_batch_size' , lowerCamelCase__ , 'train_batch_size (calculated)'				)
  self.fill_match('gradient_clipping' , args.max_grad_norm , 'max_grad_norm'				)

  self.fill_match('optimizer.params.lr' , args.learning_rate , 'learning_rate'				)
  self.fill_match('optimizer.params.betas' , [args.adam_betaa, args.adam_betaa] , 'adam_beta1+adam_beta2'				)
  self.fill_match('optimizer.params.eps' , args.adam_epsilon , 'adam_epsilon'				)
  self.fill_match('optimizer.params.weight_decay' , args.weight_decay , 'weight_decay'				)

  self.fill_only('scheduler.params.warmup_min_lr' , 0				)  # not a trainer arg
  self.fill_match('scheduler.params.warmup_max_lr' , args.learning_rate , 'learning_rate'				)
  # total_num_steps - will get set in trainer_config_finalize

  # fp16
  if args.fpaa or args.fpaa_full_eval:
   UpperCAmelCase__           =       "apex" if args.fpaa_backend == "apex" else "amp"
  else:
   UpperCAmelCase__           =       None

  if args.save_on_each_node:
   # deepspeed uses shared storage by default. Let's override this setting if save_on_each_node == True
   UpperCAmelCase__           =       self.config.get('checkpoint' , {}				)
   UpperCAmelCase__           =       args.save_on_each_node

  # amp: similar to the pytorch native amp - it has a bunch of optional params but we won't set
  # any here unless the user did the work
  self.fill_match(
      'fp16.enabled' , ((args.fpaa or args.fpaa_full_eval) and fpaa_backend == 'amp') , 'fp16|fp16_full_eval+fp16_backend(amp)' , )

  # apex: delegates amp work to apex (which needs to be available), but it cannot be used with any
  # ZeRO features
  self.fill_match('amp.enabled' , fpaa_backend == 'apex' , 'fp16+fp16_backend(apex)'				)
  self.fill_match('amp.opt_level' , args.fpaa_opt_level , 'fp16_opt_level'				)

  self.fill_match('bf16.enabled' , (args.bfaa or args.bfaa_full_eval) , 'bf16|bf16_full_eval'				)

  # deepspeed's default mode is fp16 unless there is a config that says differently
  if self.is_true('bf16.enabled'				):
   UpperCAmelCase__           =       torch.bfloataa
  elif self.is_false('fp16.enabled'				):
   UpperCAmelCase__           =       torch.floataa
  else:
   UpperCAmelCase__           =       torch.floataa




 def 						UpperCamelCase__							(self , __a , __a , __a				)				->      List[Any]:






  """simple docstring"""





  UpperCAmelCase__           =       [
      "zero_optimization.reduce_bucket_size",
      "zero_optimization.stage3_prefetch_bucket_size",
      "zero_optimization.stage3_param_persistence_threshold",
  ]
  UpperCAmelCase__           =       [x for x in hidden_size_based_keys if self.is_auto(lowerCamelCase__				)]

  if len(lowerCamelCase__				) > 0:
   if hasattr(model.config , 'hidden_size'				):
    UpperCAmelCase__           =       model.config.hidden_size
   elif hasattr(model.config , 'hidden_sizes'				):
    # if there are many hidden sizes pick the largest one
    UpperCAmelCase__           =       max(model.config.hidden_sizes				)
   else:
    raise ValueError(
        'The model\'s config file has neither `hidden_size` nor `hidden_sizes` entry, '
        'therefore it\'s not possible to automatically fill out the following `auto` entries '
        F"in the DeepSpeed config file: {hidden_size_auto_keys}. You can fix that by replacing "
        '`auto` values for these keys with an integer value of your choice.'				)

   self.fill_only('zero_optimization.reduce_bucket_size' , hidden_size * hidden_size				)
   if self.is_zeroa():
    # automatically assign the optimal config values based on model config
    self.fill_only('zero_optimization.stage3_prefetch_bucket_size' , 0.9 * hidden_size * hidden_size				)
    self.fill_only('zero_optimization.stage3_param_persistence_threshold' , 10 * hidden_size				)

        # scheduler
  self.fill_match('scheduler.params.total_num_steps' , lowerCamelCase__ , 'num_training_steps (calculated)'				)
  self.fill_match('scheduler.params.warmup_num_steps' , args.get_warmup_steps(lowerCamelCase__				) , 'warmup_steps'				)

  if len(self.mismatches				) > 0:
   UpperCAmelCase__           =       "\n".join(self.mismatches				)
   raise ValueError(
       'Please correct the following DeepSpeed config values that mismatch TrainingArguments'
       F" values:\n{mismatches}\nThe easiest method is to set these DeepSpeed config values to 'auto'."				)







_UpperCamelCase     =    None
def 		UpperCamelCase_(    snake_case__:						int     ) ->				Optional[int]:
 # this is a special weakref global object to allow us to get to Deepspeed config from APIs
 # that don't have an easy way to get to the Deepspeed config outside of the Trainer domain.
 global _hf_deepspeed_config_weak_ref
 # will go away automatically when HfDeepSpeedConfig is destroyed (when TrainingArguments is destroyed)
 UpperCAmelCase__           =       weakref.ref(snake_case__     )
def 		UpperCamelCase_(    ) ->				List[str]:
 # useful for unit tests to ensure the global state doesn't leak - call from `tearDown` method
 global _hf_deepspeed_config_weak_ref
 UpperCAmelCase__           =       None
def 		UpperCamelCase_(    ) ->				str:
 if _hf_deepspeed_config_weak_ref is not None and _hf_deepspeed_config_weak_ref() is not None:
  return _hf_deepspeed_config_weak_ref().is_zeroa()
 else:
  return False
def 		UpperCamelCase_(    ) ->				Any:
 if _hf_deepspeed_config_weak_ref is not None and _hf_deepspeed_config_weak_ref() is not None:
  return _hf_deepspeed_config_weak_ref().config
 else:
  return None
def 		UpperCamelCase_(    snake_case__:						Union[str, Any]						,					snake_case__:						int						,					snake_case__:						Optional[Any]						,					snake_case__:						List[str]						,					snake_case__:						Union[str, Any]     ) ->				List[str]:
 from accelerate.utils import DummyOptim, DummyScheduler

 UpperCAmelCase__           =       hf_deepspeed_config.config

 # Optimizer + Scheduler
 # Currently supported combos:
 # 1. DS scheduler + DS optimizer: Yes
 # 2. HF scheduler + HF optimizer: Yes
 # 3. DS scheduler + HF optimizer: Yes
 # 4. HF scheduler + DS optimizer: No
 #
 # Unless Offload is enabled in which case it's:
 # 1. DS scheduler + DS optimizer: Yes
 # 2. HF scheduler + HF optimizer: Mostly*
 # 3. DS scheduler + HF optimizer: Mostly*
 # 4. HF scheduler + DS optimizer: No
 #
 # Mostly*: All non-native DeepSpeed optimizers that have both CPU and GPU implementation should work (except LAMB)

 UpperCAmelCase__           =       None
 if "optimizer" in config:
  if args.adafactor:
   raise ValueError(
       '--adafactor was passed, but also found `optimizer` configured in the DeepSpeed config. '
       'Only one optimizer can be configured.'     )
  UpperCAmelCase__           =       DummyOptim(params=snake_case__     )
 else:
  if hf_deepspeed_config.is_offload():
   logger.info(
       'Detected ZeRO Offload and non-DeepSpeed optimizers: This combination should work as long as the'
       ' custom optimizer has both CPU and GPU implementation (except LAMB)'     )

  # ds supports Adam, OneBitAdam, and Lamb optimizers and can import other optimizers from torch.
  # But trainer uses AdamW by default.
  UpperCAmelCase__           =       trainer.create_optimizer()
  # To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
  UpperCAmelCase__           =       True

 UpperCAmelCase__           =       None
 if "scheduler" in config:
  UpperCAmelCase__           =       DummyScheduler(snake_case__     )
 else:
  if isinstance(snake_case__						,					snake_case__     ):
   raise ValueError(
       'Found `optimizer` configured in the DeepSpeed config, but no `scheduler`. '
       'Please configure a scheduler in the DeepSpeed config.'     )
  UpperCAmelCase__           =       trainer.create_scheduler(num_training_steps=snake_case__						,					optimizer=snake_case__     )

 return optimizer, lr_scheduler
def 		UpperCamelCase_(    snake_case__:						List[Any]						,					snake_case__:						Tuple						,					snake_case__:						List[str]=False     ) ->				List[Any]:
 from deepspeed.utils import logger as ds_logger

 UpperCAmelCase__           =       trainer.model
 UpperCAmelCase__           =       trainer.args

 UpperCAmelCase__           =       trainer.accelerator.state.deepspeed_plugin.hf_ds_config

 # resume config update - some bits like `model` and `num_training_steps` only become available during train
 hf_deepspeed_config.trainer_config_finalize(snake_case__						,					snake_case__						,					snake_case__     )

 # set the Deepspeed log level consistent with the Trainer
 ds_logger.setLevel(args.get_process_log_level()     )

 if inference:
  # only Z3 makes sense for the inference
  if not hf_deepspeed_config.is_zeroa():
   raise ValueError('ZeRO inference only makes sense with ZeRO Stage 3 - please adjust your config'     )

  # in case the training config is re-used for inference
  hf_deepspeed_config.del_config_sub_tree('optimizer'     )
  hf_deepspeed_config.del_config_sub_tree('lr_scheduler'     )
  UpperCAmelCase__           =       None, None
  UpperCAmelCase__           =       None
 else:
  UpperCAmelCase__           =       None  # important for when deepspeed_init is used as re-init
  UpperCAmelCase__           =       list(filter(lambda snake_case__     : p.requires_grad						,					model.parameters()     )     )
  UpperCAmelCase__           =       deepspeed_optim_sched(
      snake_case__						,					snake_case__						,					snake_case__						,					snake_case__						,					snake_case__     )

 # keep for quick debug:
 # from pprint import pprint; pprint(config)

 return optimizer, lr_scheduler







def 		UpperCamelCase_(    snake_case__:						List[Any]						,					snake_case__:						Dict     ) ->				Optional[Any]:
 # it's possible that the user is trying to resume from model_path, which doesn't necessarily
 # contain a deepspeed checkpoint. e.g. examples just check if the dir exists and assume it's
 # a resume from a checkpoint and not just a local pretrained weight. So we check here if the
 # path contains what looks like a deepspeed checkpoint
 import glob

 UpperCAmelCase__           =       sorted(glob.glob(f"{checkpoint_path}/global_step*"     )     )

 if len(snake_case__     ) > 0:
  logger.info(f"Attempting to resume from {checkpoint_path}"     )
  # this magically updates self.optimizer and self.lr_scheduler
  UpperCAmelCase__           =       deepspeed_engine.load_checkpoint(
      snake_case__						,					load_optimizer_states=snake_case__						,					load_lr_scheduler_states=snake_case__     )
  if load_path is None:
   raise ValueError(f"[deepspeed] failed to resume from checkpoint {checkpoint_path}"     )
 else:
  raise ValueError(f"Can't find a valid checkpoint at {checkpoint_path}"     )






def 		UpperCamelCase_(    snake_case__:						int     ) ->				int:
 if not isinstance(snake_case__						,					snake_case__     ):
  UpperCAmelCase__           =       f"Input value of [number={number}] must be an integer"
  raise TypeError(snake_case__     )
 if number < 1:
  raise ValueError('Input must be a positive integer'     )
 return number * (2 * number - 1)


if __name__ == "__main__":
 import doctest

 doctest.testmod()



