




from typing import Callable





def 							A__   (snake_case      :		Dict							)    ->	Dict:

					def wrapper(snake_case      :		Dict							):
										__UpperCamelCase				:  Any								=			function_with_docstring.__doc__
										return target_function

					return wrapper












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__		    =    logging.get_logger(__name__)





def 							A__   ()    ->	int:
					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       SCREAMING_SNAKE_CASE_							(					_UpperCamelCase      ):



	"""simple docstring"""


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







						"""simple docstring"""







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






class       SCREAMING_SNAKE_CASE_							(					_UpperCamelCase      ):



	"""simple docstring"""


	def __init__(				self		:	Optional[Any]				,    lowerCAmelCase		:	Union[str, Any]  )  ->					int:







						"""simple docstring"""







						super().__init__(lowerCAmelCase  )
						__UpperCamelCase				:  Optional[int]								=			None
						__UpperCamelCase				:  Tuple								=			[]






	def 				lowerCamelCase__    (				self		:	List[str]  )  ->					List[Any]:







						"""simple docstring"""







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






	def 				lowerCamelCase__    (				self		:	int				,    lowerCAmelCase		:	Union[str, Any]  )  ->					List[Any]:







						"""simple docstring"""







						__UpperCamelCase				:  List[str]								=			self.get_value(lowerCAmelCase  )
						if val is None:
											return False
						else:
											return val == "auto"






	def 				lowerCamelCase__    (				self		:	str				,    lowerCAmelCase		:	Tuple				,    lowerCAmelCase		:	int				,    lowerCAmelCase		:	List[Any]=None				,    lowerCAmelCase		:	int=True  )  ->					int:







						"""simple docstring"""







						__UpperCamelCase			, __UpperCamelCase				:  Union[str, Any]								=			self.find_config_node(lowerCAmelCase  )
						if config is None:
											return

						if config.get(lowerCAmelCase  ) == "auto":
											__UpperCamelCase				:  List[str]								=			hf_val
											return

						if not must_match:
											return

						__UpperCamelCase				:  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}'''  )

	__magic_name__      :					Optional[Any]			  =		partialmethod(_UpperCamelCase  ,     must_match=_UpperCamelCase      )






	def 				lowerCamelCase__    (				self		:	Tuple				,    lowerCAmelCase		:	Optional[int]  )  ->					Union[str, Any]:







						"""simple docstring"""







						__UpperCamelCase				:  str								=			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				:  Union[str, Any]								=			"""apex""" if args.fpaa_backend == """apex""" else """amp"""
						else:
											__UpperCamelCase				:  Optional[int]								=			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				:  List[Any]								=			self.config.get("""checkpoint"""				,    {}  )
											__UpperCamelCase				:  Tuple								=			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[Any]								=			torch.bfloataa
						elif self.is_false("""fp16.enabled"""  ):
											__UpperCamelCase				:  Optional[int]								=			torch.floataa
						else:
											__UpperCamelCase				:  List[str]								=			torch.floataa






	def 				lowerCamelCase__    (				self		:	Optional[Any]				,    lowerCAmelCase		:	Tuple				,    lowerCAmelCase		:	Union[str, Any]				,    lowerCAmelCase		:	Union[str, Any]  )  ->					str:







						"""simple docstring"""







						__UpperCamelCase				:  Tuple								=			[
						    """zero_optimization.reduce_bucket_size""",
						    """zero_optimization.stage3_prefetch_bucket_size""",
						    """zero_optimization.stage3_param_persistence_threshold""",
						]
						__UpperCamelCase				:  str								=			[x for x in hidden_size_based_keys if self.is_auto(lowerCAmelCase  )]

						if len(lowerCAmelCase  ) > 0:
											if hasattr(model.config				,    """hidden_size"""  ):
																__UpperCamelCase				:  Any								=			model.config.hidden_size
											elif hasattr(model.config				,    """hidden_sizes"""  ):
																# if there are many hidden sizes pick the largest one
																__UpperCamelCase				:  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:
											__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__		    =    None





def 							A__   (snake_case      :		Tuple							)    ->	Union[str, Any]:
					# 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				:  str								=			weakref.ref(snake_case							)





def 							A__   ()    ->	Any:
					# useful for unit tests to ensure the global state doesn't leak - call from `tearDown` method
					global _hf_deepspeed_config_weak_ref
					__UpperCamelCase				:  List[Any]								=			None





def 							A__   ()    ->	Tuple:
					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__   ()    ->	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 							A__   (snake_case      :		Dict		,  snake_case      :		Optional[int]		,  snake_case      :		List[str]		,  snake_case      :		Dict		,  snake_case      :		List[str]							)    ->	int:
					from accelerate.utils import DummyOptim, DummyScheduler

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

					__UpperCamelCase				:  str								=			None
					if "scheduler" in config:
										__UpperCamelCase				:  str								=			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				:  List[Any]								=			trainer.create_scheduler(num_training_steps=snake_case		,  optimizer=snake_case							)

					return optimizer, lr_scheduler





def 							A__   (snake_case      :		Any		,  snake_case      :		Union[str, Any]		,  snake_case      :		str=False							)    ->	int:
					from deepspeed.utils import logger as ds_logger

					__UpperCamelCase				:  List[Any]								=			trainer.model
					__UpperCamelCase				:  Tuple								=			trainer.args

					__UpperCamelCase				:  Union[str, Any]								=			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			, __UpperCamelCase				:  str								=			None, None
										__UpperCamelCase				:  Dict								=			None
					else:
										__UpperCamelCase				:  Optional[Any]								=			None  # important for when deepspeed_init is used as re-init
										__UpperCamelCase				:  Optional[int]								=			list(filter(lambda snake_case							: p.requires_grad		,  model.parameters()							)							)
										__UpperCamelCase			, __UpperCamelCase				:  str								=			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 							A__   (snake_case      :		Optional[Any]		,  snake_case      :		Any							)    ->	str:
					# 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				:  Any								=			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			, __UpperCamelCase				:  Optional[int]								=			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}'''							)






