






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

A_	:	List[Any]				=				logging.get_logger(__name__)

def 						snake_case      ():
  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        _lowerCAmelCase( SCREAMING_SNAKE_CASE__      ):





       """simple docstring"""







       def __init__(      self						,    _lowerCamelCase							):
         set_hf_deepspeed_config(self							)
         dep_version_check('accelerate'							)
         dep_version_check('deepspeed'							)
         super().__init__(a_							)







class        _lowerCAmelCase( SCREAMING_SNAKE_CASE__      ):





       """simple docstring"""







       def __init__(      self						,    _lowerCamelCase							):
         super().__init__(a_							)
         UpperCamelCase_:      Dict					   =       None
         UpperCamelCase_:      Optional[int]					   =       []





       def     _a					(      self							):



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





       def     _a					(      self						,    _lowerCamelCase							):
         UpperCamelCase_:      List[Any]					   =       self.get_value(a_							)
         if val is None:
           return False
         else:
           return val == "auto"





       def     _a					(      self						,    _lowerCamelCase						,    _lowerCamelCase						,    _lowerCamelCase=None						,    _lowerCamelCase=True							):
         UpperCamelCase_   ,UpperCamelCase_:      List[str]					   =       self.find_config_node(a_							)
         if config is None:
           return

         if config.get(a_							) == "auto":
           UpperCamelCase_:      Union[str, Any]					   =       hf_val
           return

         if not must_match:
           return

         UpperCamelCase_:      Union[str, Any]					   =       config.get(a_							)
         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}'''							)

       a       :	Dict								=partialmethod(SCREAMING_SNAKE_CASE__							,			must_match=SCREAMING_SNAKE_CASE__      )





       def     _a					(      self						,    _lowerCamelCase							):
         UpperCamelCase_:      Optional[int]					   =       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'						,    a_						,    '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_:      List[Any]					   =       'apex' if args.fpaa_backend == 'apex' else 'amp'
         else:
           UpperCamelCase_:      Dict					   =       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_:      Optional[int]					   =       self.config.get('checkpoint'						,    {}							)
           UpperCamelCase_:      int					   =       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_:      List[str]					   =       torch.bfloataa
         elif self.is_false('fp16.enabled'							):
           UpperCamelCase_:      List[Any]					   =       torch.floataa
         else:
           UpperCamelCase_:      str					   =       torch.floataa





       def     _a					(      self						,    _lowerCamelCase						,    _lowerCamelCase						,    _lowerCamelCase							):
         UpperCamelCase_:      Any					   =       [
             'zero_optimization.reduce_bucket_size',
             'zero_optimization.stage3_prefetch_bucket_size',
             'zero_optimization.stage3_param_persistence_threshold',
         ]
         UpperCamelCase_:      List[Any]					   =       [x for x in hidden_size_based_keys if self.is_auto(a_							)]

         if len(a_							) > 0:
           if hasattr(model.config						,    'hidden_size'							):
             UpperCamelCase_:      Optional[Any]					   =       model.config.hidden_size
           elif hasattr(model.config						,    'hidden_sizes'							):
             # if there are many hidden sizes pick the largest one
             UpperCamelCase_:      Tuple					   =       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'						,    1_0 * hidden_size							)

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

         if len(self.mismatches							) > 0:
           UpperCamelCase_:      Tuple					   =       '\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\'.'''							)







A_	:	Optional[int]				=				None

def 						snake_case      (UpperCAmelCase__						):
  global _hf_deepspeed_config_weak_ref
  # will go away automatically when HfDeepSpeedConfig is destroyed (when TrainingArguments is destroyed)
  UpperCamelCase_:      List[Any]					   =       weakref.ref(_UpperCamelCase						)

def 						snake_case      ():
  global _hf_deepspeed_config_weak_ref
  UpperCamelCase_:      Dict					   =       None

def 						snake_case      ():







  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 						snake_case      ():







  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 						snake_case      (UpperCAmelCase__			,					UpperCAmelCase__			,					UpperCAmelCase__			,					UpperCAmelCase__			,					UpperCAmelCase__						):
  from accelerate.utils import DummyOptim, DummyScheduler

  UpperCamelCase_:      str					   =       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_:      Optional[Any]					   =       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_:      int					   =       DummyOptim(params=_UpperCamelCase						)
  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_:      Any					   =       trainer.create_optimizer()
    # To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
    UpperCamelCase_:      Union[str, Any]					   =       True

  UpperCamelCase_:      Dict					   =       None
  if "scheduler" in config:
    UpperCamelCase_:      List[Any]					   =       DummyScheduler(_UpperCamelCase						)
  else:
    if isinstance(_UpperCamelCase			,					_UpperCamelCase						):
      raise ValueError(
          'Found `optimizer` configured in the DeepSpeed config, but no `scheduler`. '
          'Please configure a scheduler in the DeepSpeed config.'						)
    UpperCamelCase_:      Optional[Any]					   =       trainer.create_scheduler(num_training_steps=_UpperCamelCase			,					optimizer=_UpperCamelCase						)

  return optimizer, lr_scheduler

def 						snake_case      (UpperCAmelCase__			,					UpperCAmelCase__			,					UpperCAmelCase__=False						):
  from deepspeed.utils import logger as ds_logger

  UpperCamelCase_:      Union[str, Any]					   =       trainer.model
  UpperCamelCase_:      Optional[Any]					   =       trainer.args

  UpperCamelCase_:      Dict					   =       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(_UpperCamelCase			,					_UpperCamelCase			,					_UpperCamelCase						)

  # 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_   ,UpperCamelCase_:      Any					   =       None, None
    UpperCamelCase_:      Optional[Any]					   =       None
  else:
    UpperCamelCase_:      str					   =       None  # important for when deepspeed_init is used as re-init
    UpperCamelCase_:      Optional[Any]					   =       list(filter(lambda UpperCAmelCase__						: p.requires_grad			,					model.parameters()						)						)
    UpperCamelCase_   ,UpperCamelCase_:      Union[str, Any]					   =       deepspeed_optim_sched(
        _UpperCamelCase			,					_UpperCamelCase			,					_UpperCamelCase			,					_UpperCamelCase			,					_UpperCamelCase						)

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

  return optimizer, lr_scheduler

def 						snake_case      (UpperCAmelCase__			,					UpperCAmelCase__						):
  import glob

  UpperCamelCase_:      Tuple					   =       sorted(glob.glob(F'''{checkpoint_path}/global_step*'''						)						)

  if len(_UpperCamelCase						) > 0:
    logger.info(F'''Attempting to resume from {checkpoint_path}'''						)
    # this magically updates self.optimizer and self.lr_scheduler
    UpperCamelCase_   ,UpperCamelCase_:      Optional[Any]					   =       deepspeed_engine.load_checkpoint(
        _UpperCamelCase			,					load_optimizer_states=_UpperCamelCase			,					load_lr_scheduler_states=_UpperCamelCase						)
    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}'''						)







# Lint as: python3

import hashlib
from typing import Union

def 						snake_case      (UpperCAmelCase__						) ->       bytes:
  if isinstance(UpperCAmelCase__			,					UpperCAmelCase__						):
    # Data already in bytes, returns as it as
    return hash_data
  elif isinstance(UpperCAmelCase__			,					UpperCAmelCase__						):
    # We keep the data as it as for it ot be later encoded to UTF-8
    # However replace `\\` with `/` for Windows compatibility
    UpperCamelCase_:      List[str]					   =       hash_data.replace('\\'			,					'/'						)
  elif isinstance(UpperCAmelCase__			,					UpperCAmelCase__						):
    UpperCamelCase_:      Union[str, Any]					   =       str(UpperCAmelCase__						)
  else:
    # If data is not of the required type, raise error
    raise InvalidKeyError(UpperCAmelCase__						)

  return hash_data.encode('utf-8'						)




class        _lowerCAmelCase( UpperCAmelCase_      ):





       """simple docstring"""







       def __init__(      self						,    _lowerCamelCase							):
         UpperCamelCase_:      int					   =       '\nFAILURE TO GENERATE DATASET: Invalid key type detected'
         UpperCamelCase_:      Dict					   =       f'''\nFound Key {hash_data} of type {type(_lowerCamelCase							)}'''
         UpperCamelCase_:      Any					   =       '\nKeys should be either str, int or bytes type'
         super().__init__(f'''{self.prefix}{self.err_msg}{self.suffix}'''							)







class        _lowerCAmelCase( UpperCAmelCase_      ):





       """simple docstring"""







       def __init__(      self						,    _lowerCamelCase						,    _lowerCamelCase						,    _lowerCamelCase=""							):
         UpperCamelCase_:      int					   =       key
         UpperCamelCase_:      Dict					   =       duplicate_key_indices
         UpperCamelCase_:      int					   =       fix_msg
         UpperCamelCase_:      List[Any]					   =       'Found multiple examples generated with the same key'
         if len(_lowerCamelCase							) <= 2_0:
           UpperCamelCase_:      Tuple					   =       f'''\nThe examples at index {', '.join(_lowerCamelCase							)} have the key {key}'''
         else:
           UpperCamelCase_:      Optional[int]					   =       f'''\nThe examples at index {', '.join(duplicate_key_indices[:2_0]							)}... ({len(_lowerCamelCase							) - 2_0} more) have the key {key}'''
         UpperCamelCase_:      List[Any]					   =       '\n' + fix_msg if fix_msg else ''
         super().__init__(f'''{self.prefix}{self.err_msg}{self.suffix}'''							)







class        _lowerCAmelCase:





       """simple docstring"""







       def __init__(      self						,    _lowerCamelCase							):
         UpperCamelCase_:      int					   =       hashlib.mda(_as_bytes(_lowerCamelCase							)							)





       def     _a					(      self						,    _lowerCamelCase							):
         UpperCamelCase_:      Optional[int]					   =       self._split_mda.copy()
         UpperCamelCase_:      str					   =       _as_bytes(_lowerCamelCase							)
         mda.update(_lowerCamelCase							)
         # Convert to integer with hexadecimal conversion
         return int(mda.hexdigest()						,    1_6							)