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}"							)




import csv
from collections import defaultdict
from dataclasses import dataclass, field
from typing import List, Optional

import matplotlib.pyplot as plt
import numpy as np
from matplotlib.ticker import ScalarFormatter

from transformers import HfArgumentParser

def        SCREAMING_SNAKE_CASE_					(						snake_case__=None       ,  snake_case__=None							)					->     Optional[Any]:
							return field(default_factory=lambda: default       ,  metadata=snake_case__							)




@dataclass
class        lowercase_  :



	"""simple docstring"""





	UpperCAmelCase_  : str       =    field(
	    metadata={"""help""": """The csv file to plot."""}				,					)
	UpperCAmelCase_  : bool       =    field(
	    default=UpperCamelCase_				,					metadata={"""help""": """Whether to plot along batch size or sequence length. Defaults to sequence length."""}				,					)
	UpperCAmelCase_  : bool       =    field(
	    default=UpperCamelCase_				,					metadata={"""help""": """Whether the csv file has time results or memory results. Defaults to memory results."""}				,					)
	UpperCAmelCase_  : bool       =    field(
	    default=UpperCamelCase_				,					metadata={"""help""": """Disable logarithmic scale when plotting"""}				,					)
	UpperCAmelCase_  : bool       =    field(
	    default=UpperCamelCase_				,					metadata={
	        """help""": """Whether the csv file has training results or inference results. Defaults to inference results."""
	    }				,					)
	UpperCAmelCase_  : Optional[str]       =    field(
	    default=UpperCamelCase_				,					metadata={"""help""": """Filename under which the plot will be saved. If unused no plot is saved."""}				,					)
	UpperCAmelCase_  : Optional[List[str]]       =    list_field(
	    default=UpperCamelCase_				,					metadata={"""help""": """List of model names that are used instead of the ones in the csv file."""}						)

def        SCREAMING_SNAKE_CASE_					(						snake_case__							)					->     Dict:
							try:
														int(snake_case__							)
														return True
							except ValueError:
														return False

def        SCREAMING_SNAKE_CASE_					(						snake_case__							)					->     Union[str, Any]:
							try:
														float(snake_case__							)
														return True
							except ValueError:
														return False




class        lowercase_  :



	"""simple docstring"""





	def __init__(							self		,      __SCREAMING_SNAKE_CASE    )						->Optional[int]:
								lowerCAmelCase     		=		args
								lowerCAmelCase     		=		defaultdict(lambda: {"bsz": [], "seq_len": [], "result": {}}    )

								with open(self.args.csv_file		,      newline=''''''    ) as csv_file:
															lowerCAmelCase     		=		csv.DictReader(__SCREAMING_SNAKE_CASE    )
															for row in reader:
																						lowerCAmelCase     		=		row['''model''']
																						self.result_dict[model_name]["bsz"].append(int(row['''batch_size''']    )    )
																						self.result_dict[model_name]["seq_len"].append(int(row['''sequence_length''']    )    )
																						if can_convert_to_int(row['''result''']    ):
																													# value is not None
																													lowerCAmelCase     		=		int(row['''result''']    )
																						elif can_convert_to_float(row['''result''']    ):
																													# value is not None
																													lowerCAmelCase     		=		float(row['''result''']    )




	def       SCREAMING_SNAKE_CASE_    (							self    )						->Tuple:
								lowerCAmelCase		,				lowerCAmelCase     		=		plt.subplots()
								lowerCAmelCase     		=		'''Time usage''' if self.args.is_time else '''Memory usage'''
								lowerCAmelCase     		=		title_str + ''' for training''' if self.args.is_train else title_str + ''' for inference'''

								if not self.args.no_log_scale:
															# set logarithm scales
															ax.set_xscale('''log'''    )
															ax.set_yscale('''log'''    )

								for axis in [ax.xaxis, ax.yaxis]:
															axis.set_major_formatter(ScalarFormatter()    )

								for model_name_idx, model_name in enumerate(self.result_dict.keys()    ):
															lowerCAmelCase     		=		sorted(set(self.result_dict[model_name]['''bsz''']    )    )
															lowerCAmelCase     		=		sorted(set(self.result_dict[model_name]['''seq_len''']    )    )
															lowerCAmelCase     		=		self.result_dict[model_name]['''result''']

															((lowerCAmelCase)		,				(lowerCAmelCase))     		=		(
															    (batch_sizes, sequence_lengths) if self.args.plot_along_batch else (sequence_lengths, batch_sizes)
															)

															lowerCAmelCase     		=		(
															    model_name if self.args.short_model_names is None else self.args.short_model_names[model_name_idx]
															)

															for inner_loop_value in inner_loop_array:
																						if self.args.plot_along_batch:
																													lowerCAmelCase     		=		np.asarray(
																													    [results[(x, inner_loop_value)] for x in x_axis_array if (x, inner_loop_value) in results]		,      dtype=__SCREAMING_SNAKE_CASE		,      )
																						else:
																													lowerCAmelCase     		=		np.asarray(
																													    [results[(inner_loop_value, x)] for x in x_axis_array if (inner_loop_value, x) in results]		,      dtype=np.floataa		,      )

																						((lowerCAmelCase)		,				(lowerCAmelCase))     		=		(
																						    ('''batch_size''', '''len''') if self.args.plot_along_batch else ('''in #tokens''', '''bsz''')
																						)

																						lowerCAmelCase     		=		np.asarray(__SCREAMING_SNAKE_CASE		,      __SCREAMING_SNAKE_CASE    )[: len(__SCREAMING_SNAKE_CASE    )]
																						plt.scatter(
																						    __SCREAMING_SNAKE_CASE		,      __SCREAMING_SNAKE_CASE		,      label=F"{label_model_name} - {inner_loop_label}: {inner_loop_value}"    )
																						plt.plot(__SCREAMING_SNAKE_CASE		,      __SCREAMING_SNAKE_CASE		,      '''--'''    )

															title_str += F" {label_model_name} vs."

								lowerCAmelCase     		=		title_str[:-4]
								lowerCAmelCase     		=		'''Time in s''' if self.args.is_time else '''Memory in MB'''

								# plot
								plt.title(__SCREAMING_SNAKE_CASE    )
								plt.xlabel(__SCREAMING_SNAKE_CASE    )
								plt.ylabel(__SCREAMING_SNAKE_CASE    )
								plt.legend()

								if self.args.figure_png_file is not None:
															plt.savefig(self.args.figure_png_file    )
								else:
															plt.show()

def        SCREAMING_SNAKE_CASE_					(						)					->     List[str]:
							lowerCAmelCase     		=		HfArgumentParser(snake_case__							)
							lowerCAmelCase     		=		parser.parse_args_into_dataclasses()[0]
							lowerCAmelCase     		=		Plot(args=snake_case__							)
							plot.plot()


if __name__ == "__main__":
	main()



