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

lowercase__      :    Any          =     logging.get_logger(__name__)

def        SCREAMING_SNAKE_CASE_					(						)					->     Optional[Any]:
							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_  (  UpperCamelCase_						):



	"""simple docstring"""





	def __init__(							self		,      __SCREAMING_SNAKE_CASE    )						->Dict:
								# set global weakref object
								set_hf_deepspeed_config(self    )
								dep_version_check('''accelerate'''    )
								dep_version_check('''deepspeed'''    )
								super().__init__(__SCREAMING_SNAKE_CASE    )




class        lowercase_  (  UpperCamelCase_						):



	"""simple docstring"""





	def __init__(							self		,      __SCREAMING_SNAKE_CASE    )						->str:
								super().__init__(__SCREAMING_SNAKE_CASE    )
								lowerCAmelCase     		=		None
								lowerCAmelCase     		=		[]


	def       SCREAMING_SNAKE_CASE_    (							self    )						->List[str]:
								if self._dtype is None:
															raise ValueError('''trainer_config_process() wasn\'t called yet to tell dtype'''    )
								return self._dtype


	def       SCREAMING_SNAKE_CASE_    (							self		,      __SCREAMING_SNAKE_CASE    )						->Optional[Any]:
								lowerCAmelCase     		=		self.get_value(__SCREAMING_SNAKE_CASE    )
								if val is None:
															return False
								else:
															return val == "auto"


	def       SCREAMING_SNAKE_CASE_    (							self		,      __SCREAMING_SNAKE_CASE		,      __SCREAMING_SNAKE_CASE		,      __SCREAMING_SNAKE_CASE=None		,      __SCREAMING_SNAKE_CASE=True    )						->List[Any]:
								lowerCAmelCase		,				lowerCAmelCase     		=		self.find_config_node(__SCREAMING_SNAKE_CASE    )
								if config is None:
															return

								if config.get(__SCREAMING_SNAKE_CASE    ) == "auto":
															lowerCAmelCase     		=		hf_val
															return

								if not must_match:
															return

								lowerCAmelCase     		=		config.get(__SCREAMING_SNAKE_CASE    )
								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}"    )

	UpperCAmelCase_  : int       =    partialmethod(UpperCamelCase_				,					must_match=UpperCamelCase_						)


	def       SCREAMING_SNAKE_CASE_    (							self		,      __SCREAMING_SNAKE_CASE    )						->Union[str, Any]:
								lowerCAmelCase     		=		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'''		,      __SCREAMING_SNAKE_CASE		,      '''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:
															lowerCAmelCase     		=		'''apex''' if args.fpaa_backend == '''apex''' else '''amp'''
								else:
															lowerCAmelCase     		=		None

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




	def       SCREAMING_SNAKE_CASE_    (							self		,      __SCREAMING_SNAKE_CASE		,      __SCREAMING_SNAKE_CASE		,      __SCREAMING_SNAKE_CASE    )						->Optional[Any]:
								lowerCAmelCase     		=		[
								    '''zero_optimization.reduce_bucket_size''',
								    '''zero_optimization.stage3_prefetch_bucket_size''',
								    '''zero_optimization.stage3_param_persistence_threshold''',
								]
								lowerCAmelCase     		=		[x for x in hidden_size_based_keys if self.is_auto(__SCREAMING_SNAKE_CASE    )]

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

								if len(self.mismatches    ) > 0:
															lowerCAmelCase     		=		'''\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'."    )
lowercase__      :    List[str]          =     None

def        SCREAMING_SNAKE_CASE_					(						snake_case__							)					->     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)
							lowerCAmelCase     		=		weakref.ref(snake_case__							)

def        SCREAMING_SNAKE_CASE_					(						)					->     int:
							# useful for unit tests to ensure the global state doesn't leak - call from `tearDown` method
							global _hf_deepspeed_config_weak_ref
							lowerCAmelCase     		=		None

def        SCREAMING_SNAKE_CASE_					(						)					->     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        SCREAMING_SNAKE_CASE_					(						)					->     Union[str, 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        SCREAMING_SNAKE_CASE_					(						snake_case__       ,  snake_case__       ,  snake_case__       ,  snake_case__       ,  snake_case__							)					->     int:
							from accelerate.utils import DummyOptim, DummyScheduler

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

							lowerCAmelCase     		=		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.'''							)
														lowerCAmelCase     		=		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.
														lowerCAmelCase     		=		trainer.create_optimizer()
														# To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
														lowerCAmelCase     		=		True

							lowerCAmelCase     		=		None
							if "scheduler" in config:
														lowerCAmelCase     		=		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.'''							)
														lowerCAmelCase     		=		trainer.create_scheduler(num_training_steps=snake_case__       ,  optimizer=snake_case__							)

							return optimizer, lr_scheduler

def        SCREAMING_SNAKE_CASE_					(						snake_case__       ,  snake_case__       ,  snake_case__=False							)					->     Tuple:
							from deepspeed.utils import logger as ds_logger

							lowerCAmelCase     		=		trainer.model
							lowerCAmelCase     		=		trainer.args

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

							lowerCAmelCase     		=		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
														lowerCAmelCase		,				lowerCAmelCase     		=		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}"							)




lowercase__      :    dict[str, float]          =     {
    "kilogram": 1,
    "gram": pow(1_0, 3),
    "milligram": pow(1_0, 6),
    "metric-ton": pow(1_0, -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_0_0_0,
    "atomic-mass-unit": 6.0_22_13_66_52e26,
}

lowercase__      :    dict[str, float]          =     {
    "kilogram": 1,
    "gram": pow(1_0, -3),
    "milligram": pow(1_0, -6),
    "metric-ton": pow(1_0, 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_60_54_01_99e-27,
}

def        SCREAMING_SNAKE_CASE_					(						snake_case__       ,  snake_case__       ,  snake_case__							)					->     float:
							if to_type not in KILOGRAM_CHART or from_type not in WEIGHT_TYPE_CHART:
														lowerCAmelCase     		=		(
														    f"Invalid 'from_type' or 'to_type' value: {from_type!r}, {to_type!r}\n"
														    f"Supported values are: {', '.join(snake_case__							)}"
														)
														raise ValueError(snake_case__							)
							return value * KILOGRAM_CHART[to_type] * WEIGHT_TYPE_CHART[from_type]


if __name__ == "__main__":
	import doctest

	doctest.testmod()



