
'''simple docstring'''
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__ :		Optional[Any]					=logging.get_logger(__name__)







def 							A_  (    )					->							Optional[int]:





  """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 			__A					(			_SCREAMING_SNAKE_CASE      ):
 def __init__( self      :     Any					, lowerCamelCase      :     str							):



   """simple docstring"""





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







class 			__A					(			_SCREAMING_SNAKE_CASE      ):
 def __init__( self      :     Dict					, lowerCamelCase      :     Optional[Any]							):



   """simple docstring"""





   super().__init__(lowerCamelCase							)
   __A		:  Tuple       				=		None
   __A		:  Optional[Any]       				=		[]
 def 		lowercase_( self      :     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      :     str					, lowerCamelCase      :     Optional[Any]							):



   """simple docstring"""





   __A		:  Tuple       				=		self.get_value(lowerCamelCase							)
   if val is None:
     return False
   else:
     return val == "auto"
 def 		lowercase_( self      :     Any					, lowerCamelCase      :     int					, lowerCamelCase      :     Any					, lowerCamelCase      :     Dict=None					, lowerCamelCase      :     List[Any]=True							):



   """simple docstring"""





   __A				,       __A		:  int       				=		self.find_config_node(lowerCamelCase							)
   if config is None:
     return

   if config.get(lowerCamelCase							) == "auto":
     __A		:  Tuple       				=		hf_val
     return

   if not must_match:
     return

   __A		:  Dict       				=		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}"							)

 lowerCamelCase												=partialmethod(_SCREAMING_SNAKE_CASE     ,    must_match=_SCREAMING_SNAKE_CASE      )
 def 		lowercase_( self      :     List[Any]					, lowerCamelCase      :     List[Any]							):



   """simple docstring"""





   __A		:  Any       				=		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:
     __A		:  Optional[int]       				=		"""apex""" if args.fpaa_backend == """apex""" else """amp"""
   else:
     __A		:  List[str]       				=		None

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






 def 		lowercase_( self      :     Dict					, lowerCamelCase      :     List[str]					, lowerCamelCase      :     Union[str, Any]					, lowerCamelCase      :     Dict							):



   """simple docstring"""





   __A		:  Any       				=		[
       """zero_optimization.reduce_bucket_size""",
       """zero_optimization.stage3_prefetch_bucket_size""",
       """zero_optimization.stage3_param_persistence_threshold""",
   ]
   __A		:  Any       				=		[x for x in hidden_size_based_keys if self.is_auto(lowerCamelCase							)]

   if len(lowerCamelCase							) > 0:
     if hasattr(model.config					, """hidden_size"""							):
       __A		:  Optional[Any]       				=		model.config.hidden_size
     elif hasattr(model.config					, """hidden_sizes"""							):
       # if there are many hidden sizes pick the largest one
       __A		:  Optional[int]       				=		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:
     __A		:  int       				=		"""\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__ :		Tuple					=None







def 							A_  (    __SCREAMING_SNAKE_CASE    :	Optional[Any]			)					->							Optional[int]:





  """simple docstring"""

  global _hf_deepspeed_config_weak_ref
  # will go away automatically when HfDeepSpeedConfig is destroyed (when TrainingArguments is destroyed)
  __A		:  List[str]       				=		weakref.ref(__SCREAMING_SNAKE_CASE			)







def 							A_  (    )					->							Dict:





  """simple docstring"""

  global _hf_deepspeed_config_weak_ref
  __A		:  Any       				=		None







def 							A_  (    )					->							str:





  """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_  (    )					->							Union[str, Any]:





  """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_  (    __SCREAMING_SNAKE_CASE    :	List[str]      ,  __SCREAMING_SNAKE_CASE    :	Optional[int]      ,  __SCREAMING_SNAKE_CASE    :	Tuple      ,  __SCREAMING_SNAKE_CASE    :	Optional[Any]      ,  __SCREAMING_SNAKE_CASE    :	Any			)					->							Dict:





  """simple docstring"""

  from accelerate.utils import DummyOptim, DummyScheduler

  __A		:  Optional[int]       				=		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)

  __A		:  Tuple       				=		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."""			)
    __A		:  Dict       				=		DummyOptim(params=__SCREAMING_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.
    __A		:  Optional[Any]       				=		trainer.create_optimizer()
    # To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
    __A		:  Optional[Any]       				=		True

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

  return optimizer, lr_scheduler







def 							A_  (    __SCREAMING_SNAKE_CASE    :	List[Any]      ,  __SCREAMING_SNAKE_CASE    :	str      ,  __SCREAMING_SNAKE_CASE    :	List[str]=False			)					->							Optional[Any]:





  """simple docstring"""

  from deepspeed.utils import logger as ds_logger

  __A		:  List[str]       				=		trainer.model
  __A		:  Tuple       				=		trainer.args

  __A		:  str       				=		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(__SCREAMING_SNAKE_CASE      ,  __SCREAMING_SNAKE_CASE      ,  __SCREAMING_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"""			)
    __A				,       __A		:  Tuple       				=		None, None
    __A		:  Dict       				=		None
  else:
    __A		:  Dict       				=		None  # important for when deepspeed_init is used as re-init
    __A		:  Any       				=		list(filter(lambda __SCREAMING_SNAKE_CASE			: p.requires_grad      ,  model.parameters()			)			)
    __A				,       __A		:  int       				=		deepspeed_optim_sched(
        __SCREAMING_SNAKE_CASE      ,  __SCREAMING_SNAKE_CASE      ,  __SCREAMING_SNAKE_CASE      ,  __SCREAMING_SNAKE_CASE      ,  __SCREAMING_SNAKE_CASE			)

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

  return optimizer, lr_scheduler







def 							A_  (    __SCREAMING_SNAKE_CASE    :	List[str]      ,  __SCREAMING_SNAKE_CASE    :	Dict			)					->							Dict:





  """simple docstring"""

  import glob

  __A		:  int       				=		sorted(glob.glob(F"{checkpoint_path}/global_step*"			)			)

  if len(__SCREAMING_SNAKE_CASE			) > 0:
    logger.info(F"Attempting to resume from {checkpoint_path}"			)
    # this magically updates self.optimizer and self.lr_scheduler
    __A				,       __A		:  str       				=		deepspeed_engine.load_checkpoint(
        __SCREAMING_SNAKE_CASE      ,  load_optimizer_states=__SCREAMING_SNAKE_CASE      ,  load_lr_scheduler_states=__SCREAMING_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}"			)





'''simple docstring'''
A__ :		dict[str, float]					={
    "kilogram": 1,
    "gram": pow(10, 3),
    "milligram": pow(10, 6),
    "metric-ton": pow(10, -3),
    "long-ton": 0.0_0_0_9_8_4_2_0_7_3,
    "short-ton": 0.0_0_1_1_0_2_3_1_2_2,
    "pound": 2.2_0_4_6_2_4_4_2_0_2,
    "stone": 0.1_5_7_4_7_3_1_7_2_8,
    "ounce": 3_5.2_7_3_9_9_0_7_2_3,
    "carrat": 5_000,
    "atomic-mass-unit": 6.0_2_2_1_3_6_6_5_2E2_6,
}

A__ :		dict[str, float]					={
    "kilogram": 1,
    "gram": pow(10, -3),
    "milligram": pow(10, -6),
    "metric-ton": pow(10, 3),
    "long-ton": 1_0_1_6.0_4_6_0_8,
    "short-ton": 9_0_7.1_8_4,
    "pound": 0.4_5_3_5_9_2,
    "stone": 6.3_5_0_2_9,
    "ounce": 0.0_2_8_3_4_9_5,
    "carrat": 0.0_0_0_2,
    "atomic-mass-unit": 1.6_6_0_5_4_0_1_9_9E-2_7,
}







def 							A_  (    __SCREAMING_SNAKE_CASE    :	str      ,  __SCREAMING_SNAKE_CASE    :	str      ,  __SCREAMING_SNAKE_CASE    :	float			)					->							float:





  """simple docstring"""

  if to_type not in KILOGRAM_CHART or from_type not in WEIGHT_TYPE_CHART:
    __A		:  Tuple       				=		(
        F"Invalid 'from_type' or 'to_type' value: {from_type!r}, {to_type!r}\n"
        F"Supported values are: {', '.join(__SCREAMING_SNAKE_CASE			)}"
    )
    raise ValueError(__SCREAMING_SNAKE_CASE			)
  return value * KILOGRAM_CHART[to_type] * WEIGHT_TYPE_CHART[from_type]


if __name__ == "__main__":
     import doctest

     doctest.testmod()



