
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

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

def a__   (	):






  '''simple docstring'''





  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 	UpperCAmelCase_      (			_A					):


  '''simple docstring'''







  def __init__(			self      :			Tuple      ,							UpperCamelCase__      :			Optional[int]   )      ->     Dict:







    """simple docstring"""






    set_hf_deepspeed_config(self   )
    dep_version_check("""accelerate"""   )
    dep_version_check("""deepspeed"""   )
    super().__init__(UpperCamelCase__   )







class 	UpperCAmelCase_      (			_A					):


  '''simple docstring'''







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







    """simple docstring"""






    super().__init__(UpperCamelCase__   )
    __magic_name__    					=	None
    __magic_name__    					=	[]
  def 							_lowercase							(			self      :			Optional[Any]   )      ->     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 							_lowercase							(			self      :			Any      ,							UpperCamelCase__      :			Any   )      ->     List[Any]:







    """simple docstring"""






    __magic_name__    					=	self.get_value(UpperCamelCase__   )
    if val is None:
      return False
    else:
      return val == "auto"
  def 							_lowercase							(			self      :			Optional[int]      ,							UpperCamelCase__      :			List[Any]      ,							UpperCamelCase__      :			List[str]      ,							UpperCamelCase__      :			Optional[Any]=None      ,							UpperCamelCase__      :			Tuple=True   )      ->     Dict:







    """simple docstring"""






    __magic_name__					,   __magic_name__    					=	self.find_config_node(UpperCamelCase__   )
    if config is None:
      return

    if config.get(UpperCamelCase__   ) == "auto":
      __magic_name__    					=	hf_val
      return

    if not must_match:
      return

    __magic_name__    					=	config.get(UpperCamelCase__   )
    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__   	=		partialmethod(_A  ,	must_match=_A					)
  def 							_lowercase							(			self      :			Any      ,							UpperCamelCase__      :			Optional[Any]   )      ->     List[Any]:







    """simple docstring"""






    __magic_name__    					=	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"""      ,							UpperCamelCase__      ,							"""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:
      __magic_name__    					=	"""apex""" if args.fpaa_backend == """apex""" else """amp"""
    else:
      __magic_name__    					=	None

    if args.save_on_each_node:
      # deepspeed uses shared storage by default. Let's override this setting if save_on_each_node == True
      __magic_name__    					=	self.config.get("""checkpoint"""      ,							{}   )
      __magic_name__    					=	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"""   ):
      __magic_name__    					=	torch.bfloataa
    elif self.is_false("""fp16.enabled"""   ):
      __magic_name__    					=	torch.floataa
    else:
      __magic_name__    					=	torch.floataa





  def 							_lowercase							(			self      :			List[str]      ,							UpperCamelCase__      :			List[str]      ,							UpperCamelCase__      :			Union[str, Any]      ,							UpperCamelCase__      :			Tuple   )      ->     Dict:







    """simple docstring"""






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

    if len(UpperCamelCase__   ) > 0:
      if hasattr(model.config      ,							"""hidden_size"""   ):
        __magic_name__    					=	model.config.hidden_size
      elif hasattr(model.config      ,							"""hidden_sizes"""   ):
        # if there are many hidden sizes pick the largest one
        __magic_name__    					=	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"""      ,							UpperCamelCase__      ,							"""num_training_steps (calculated)"""   )
    self.fill_match("""scheduler.params.warmup_num_steps"""      ,							args.get_warmup_steps(UpperCamelCase__   )      ,							"""warmup_steps"""   )

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







__lowerCAmelCase      :					Optional[int]   	=       None

def a__   (	A_ ):






  '''simple docstring'''





  global _hf_deepspeed_config_weak_ref
  # will go away automatically when HfDeepSpeedConfig is destroyed (when TrainingArguments is destroyed)
  __magic_name__    					=	weakref.ref(A_ )

def a__   (	):






  '''simple docstring'''





  global _hf_deepspeed_config_weak_ref
  __magic_name__    					=	None

def a__   (	):






  '''simple docstring'''





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






  '''simple docstring'''





  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 a__   (	A_,					A_,					A_,					A_,					A_ ):






  '''simple docstring'''





  from accelerate.utils import DummyOptim, DummyScheduler

  __magic_name__    					=	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)

  __magic_name__    					=	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.""" )
    __magic_name__    					=	DummyOptim(params=A_ )
  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.
    __magic_name__    					=	trainer.create_optimizer()
    # To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
    __magic_name__    					=	True

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

  return optimizer, lr_scheduler

def a__   (	A_,					A_,					A_=False ):






  '''simple docstring'''





  from deepspeed.utils import logger as ds_logger

  __magic_name__    					=	trainer.model
  __magic_name__    					=	trainer.args

  __magic_name__    					=	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(A_,					A_,					A_ )

  # 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""" )
    __magic_name__					,   __magic_name__    					=	None, None
    __magic_name__    					=	None
  else:
    __magic_name__    					=	None  # important for when deepspeed_init is used as re-init
    __magic_name__    					=	list(filter(lambda A_ : p.requires_grad,					model.parameters() ) )
    __magic_name__					,   __magic_name__    					=	deepspeed_optim_sched(
        A_,					A_,					A_,					A_,					A_ )

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

  return optimizer, lr_scheduler





def a__   (	A_,					A_ ):






  '''simple docstring'''





  import glob

  __magic_name__    					=	sorted(glob.glob(f'''{checkpoint_path}/global_step*''' ) )

  if len(A_ ) > 0:
    logger.info(f'''Attempting to resume from {checkpoint_path}''' )
    # this magically updates self.optimizer and self.lr_scheduler
    __magic_name__					,   __magic_name__    					=	deepspeed_engine.load_checkpoint(
        A_,					load_optimizer_states=A_,					load_lr_scheduler_states=A_ )
    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}''' )



class 	UpperCAmelCase_      :


  '''simple docstring'''







  def __init__(			self      :			Tuple      ,							UpperCamelCase__      :			Dict=None   )      ->     List[str]:







    """simple docstring"""






    __magic_name__    					=	data
    __magic_name__    					=	None





  def __repr__(			self      :			Optional[int]   )      ->     Optional[int]:







    """simple docstring"""






    __magic_name__    					=	""""""
    __magic_name__    					=	self
    while temp:
      string_rep += F'''<{temp.data}> ---> '''
      __magic_name__    					=	temp.next
    string_rep += "<END>"
    return string_rep







def a__   (	A_ ):






  '''simple docstring'''





  if not elements_list:
    raise Exception("""The Elements List is empty""" )

  # Set first element as Head
  __magic_name__    					=	Node(elements_list[0] )
  __magic_name__    					=	head
  # Loop through elements from position 1
  for data in elements_list[1:]:
    __magic_name__    					=	Node(A_ )
    __magic_name__    					=	current.next
  return head


__lowerCAmelCase      :					Union[str, Any]   	=       [1, 3, 5, 32, 44, 12, 43]
print(F'''List: {list_data}''')
print('Creating Linked List from List.')
__lowerCAmelCase      :					Optional[Any]   	=       make_linked_list(list_data)
print('Linked List:')





print(linked_list)

