code: 
  
  
  
  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__	:  int 		=logging.get_logger(__name__)
  
  
  
  
  
  
  def 			__lowercase     (				)       ->				List[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      UpperCAmelCase_						(     __lowerCamelCase ):
  
  						'''simple docstring'''
  
  
  
  
  						def __init__(					self   ,							_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													set_hf_deepspeed_config(self	)
  													dep_version_check('accelerate'	)
  													dep_version_check('deepspeed'	)
  													super().__init__(_A	)
  
  
  
  
  
  
  
  class      UpperCAmelCase_						(     __lowerCamelCase ):
  
  						'''simple docstring'''
  
  
  
  
  						def __init__(					self   ,							_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													super().__init__(_A	)
  													__SCREAMING_SNAKE_CASE								=      None
  													__SCREAMING_SNAKE_CASE								=      []
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													if self._dtype is None:
  																				raise ValueError('trainer_config_process() wasn\'t called yet to tell dtype'	)
  													return self._dtype
  
  
  
  
  
  
  						def  _A			(					self   ,							_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      self.get_value(_A	)
  													if val is None:
  																				return False
  													else:
  																				return val == "auto"
  
  
  
  
  
  
  						def  _A			(					self   ,							_A   ,							_A   ,							_A=None   ,							_A=True	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE							, __SCREAMING_SNAKE_CASE								=      self.find_config_node(_A	)
  													if config is None:
  																				return
  
  													if config.get(_A	) == "auto":
  																				__SCREAMING_SNAKE_CASE								=      hf_val
  																				return
  
  													if not must_match:
  																				return
  
  													__SCREAMING_SNAKE_CASE								=      config.get(_A	)
  													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__    :				Union[str, Any]       						=						partialmethod(__lowerCamelCase					,					must_match=__lowerCamelCase )
  
  
  
  
  
  
  						def  _A			(					self   ,							_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      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'   ,							_A   ,							'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:
  																				__SCREAMING_SNAKE_CASE								=      'apex' if args.fpaa_backend == 'apex' else 'amp'
  													else:
  																				__SCREAMING_SNAKE_CASE								=      None
  
  													if args.save_on_each_node:
  																				# deepspeed uses shared storage by default. Let's override this setting if save_on_each_node == True
  																				__SCREAMING_SNAKE_CASE								=      self.config.get('checkpoint'   ,							{}	)
  																				__SCREAMING_SNAKE_CASE								=      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'	):
  																				__SCREAMING_SNAKE_CASE								=      torch.bfloataa
  													elif self.is_false('fp16.enabled'	):
  																				__SCREAMING_SNAKE_CASE								=      torch.floataa
  													else:
  																				__SCREAMING_SNAKE_CASE								=      torch.floataa
  
  
  
  
  
  
  						def  _A			(					self   ,							_A   ,							_A   ,							_A	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      [
  													    'zero_optimization.reduce_bucket_size',
  													    'zero_optimization.stage3_prefetch_bucket_size',
  													    'zero_optimization.stage3_param_persistence_threshold',
  													]
  													__SCREAMING_SNAKE_CASE								=      [x for x in hidden_size_based_keys if self.is_auto(_A	)]
  
  													if len(_A	) > 0:
  																				if hasattr(model.config   ,							'hidden_size'	):
  																											__SCREAMING_SNAKE_CASE								=      model.config.hidden_size
  																				elif hasattr(model.config   ,							'hidden_sizes'	):
  																											# if there are many hidden sizes pick the largest one
  																											__SCREAMING_SNAKE_CASE								=      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'   ,							_A   ,							'num_training_steps (calculated)'	)
  													self.fill_match('scheduler.params.warmup_num_steps'   ,							args.get_warmup_steps(_A	)   ,							'warmup_steps'	)
  
  													if len(self.mismatches	) > 0:
  																				__SCREAMING_SNAKE_CASE								=      '\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__	:  int 		=None
  
  
  
  
  
  
  def 			__lowercase     (				a__     )       ->				List[Any]:
  							global _hf_deepspeed_config_weak_ref
  							# will go away automatically when HfDeepSpeedConfig is destroyed (when TrainingArguments is destroyed)
  							__SCREAMING_SNAKE_CASE								=      weakref.ref(a__     )
  
  
  
  
  
  
  def 			__lowercase     (				)       ->				Optional[int]:
  							global _hf_deepspeed_config_weak_ref
  							__SCREAMING_SNAKE_CASE								=      None
  
  
  
  
  
  
  def 			__lowercase     (				)       ->				Dict:
  
  
  							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 			__lowercase     (				)       ->				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 			__lowercase     (				a__							,     a__							,     a__							,     a__							,     a__     )       ->				int:
  							from accelerate.utils import DummyOptim, DummyScheduler
  
  							__SCREAMING_SNAKE_CASE								=      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)
  
  							__SCREAMING_SNAKE_CASE								=      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.'     )
  														__SCREAMING_SNAKE_CASE								=      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.
  														__SCREAMING_SNAKE_CASE								=      trainer.create_optimizer()
  														# To use other optimizers requires voiding warranty with: `zero_allow_untested_optimizer`
  														__SCREAMING_SNAKE_CASE								=      True
  
  							__SCREAMING_SNAKE_CASE								=      None
  							if "scheduler" in config:
  														__SCREAMING_SNAKE_CASE								=      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.'     )
  														__SCREAMING_SNAKE_CASE								=      trainer.create_scheduler(num_training_steps=a__							,     optimizer=a__     )
  
  							return optimizer, lr_scheduler
  
  
  
  
  
  
  def 			__lowercase     (				a__							,     a__							,     a__=False     )       ->				Optional[int]:
  							from deepspeed.utils import logger as ds_logger
  
  							__SCREAMING_SNAKE_CASE								=      trainer.model
  							__SCREAMING_SNAKE_CASE								=      trainer.args
  
  							__SCREAMING_SNAKE_CASE								=      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'     )
  														__SCREAMING_SNAKE_CASE							, __SCREAMING_SNAKE_CASE								=      None, None
  														__SCREAMING_SNAKE_CASE								=      None
  							else:
  														__SCREAMING_SNAKE_CASE								=      None  # important for when deepspeed_init is used as re-init
  														__SCREAMING_SNAKE_CASE								=      list(filter(lambda a__     : p.requires_grad							,     model.parameters()     )     )
  														__SCREAMING_SNAKE_CASE							, __SCREAMING_SNAKE_CASE								=      deepspeed_optim_sched(
  														    a__							,     a__							,     a__							,     a__							,     a__     )
  
  							# keep for quick debug:
  							# from pprint import pprint; pprint(config)
  
  							return optimizer, lr_scheduler
  
  
  
  
  
  
  def 			__lowercase     (				a__							,     a__     )       ->				int:
  							import glob
  
  							__SCREAMING_SNAKE_CASE								=      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
  														__SCREAMING_SNAKE_CASE							, __SCREAMING_SNAKE_CASE								=      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}"""     )
  
  
  
  
  
  
  
style_context: 
  
  
  
  import unittest
  
  from cer import CER
  
  
  lowerCAmelCase__	:  Dict 		=CER()
  
  class      UpperCAmelCase_						(     unittest.TestCase ):
  
  						'''simple docstring'''
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      ['White House']
  													__SCREAMING_SNAKE_CASE								=      ['white house']
  													# S = 2, D = 0, I = 0, N = 11, CER = 2 / 11
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.1_8_1_8_1_8_1_8_1_8	) < 1e-6	)
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      ['were wolf']
  													__SCREAMING_SNAKE_CASE								=      ['werewolf']
  													# S = 0, D = 0, I = 1, N = 9, CER = 1 / 9
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.1_1_1_1_1_1_1	) < 1e-6	)
  
  													__SCREAMING_SNAKE_CASE								=      ['werewolf']
  													__SCREAMING_SNAKE_CASE								=      ['weae     wolf']
  													# S = 1, D = 1, I = 0, N = 8, CER = 0.25
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.2_5	) < 1e-6	)
  
  													# consecutive whitespaces case 1
  													__SCREAMING_SNAKE_CASE								=      ['were wolf']
  													__SCREAMING_SNAKE_CASE								=      ['were               wolf']
  													# S = 0, D = 0, I = 0, N = 9, CER = 0
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.0	) < 1e-6	)
  
  													# consecutive whitespaces case 2
  													__SCREAMING_SNAKE_CASE								=      ['were   wolf']
  													__SCREAMING_SNAKE_CASE								=      ['were               wolf']
  													# S = 0, D = 0, I = 0, N = 9, CER = 0
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.0	) < 1e-6	)
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      ['werewolf']
  													__SCREAMING_SNAKE_CASE								=      ['weaewolf']
  													# S = 1, D = 0, I = 0, N = 8, CER = 0.125
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.1_2_5	) < 1e-6	)
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      ['werewolf']
  													__SCREAMING_SNAKE_CASE								=      ['wereawolf']
  													# S = 0, D = 1, I = 0, N = 8, CER = 0.125
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.1_2_5	) < 1e-6	)
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      ['werewolf']
  													__SCREAMING_SNAKE_CASE								=      ['wereolf']
  													# S = 0, D = 0, I = 1, N = 8, CER = 0.125
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.1_2_5	) < 1e-6	)
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      ['werewolf']
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertEqual(_A   ,							0.0	)
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      ['werewolf', 'I am your father']
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertEqual(_A   ,							0.0	)
  
  													__SCREAMING_SNAKE_CASE								=      ['werewolf', 'I am your father', 'doge']
  													__SCREAMING_SNAKE_CASE								=      ['werxwolf', 'I       am your father', 'doge']
  													# S = 1, D = 0, I = 0, N = 28, CER = 1 / 28
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.0_3_5_7_1_4_2_8	) < 1e-6	)
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      ['My hovercraft', 'is full of eels']
  													__SCREAMING_SNAKE_CASE								=      ['My hovercraft is full', ' of eels']
  													# S = 0, D = 0, I = 2, N = 28, CER = 2 / 28
  													# whitespace at the front of " of eels" will be strip during preporcessing
  													# so need to insert 2 whitespaces
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A   ,							concatenate_texts=_A	)
  													self.assertTrue(abs(char_error_rate - 0.0_7_1_4_2_8	) < 1e-6	)
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      ['我能吞下玻璃而不伤身体']
  													__SCREAMING_SNAKE_CASE								=      [' 能吞虾玻璃而 不霜身体啦']
  													# S = 3, D = 2, I = 0, N = 11, CER = 5 / 11
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.4_5_4_5_4_5_4_5_4_5	) < 1e-6	)
  
  													__SCREAMING_SNAKE_CASE								=      ['我能吞下玻璃', '而不伤身体']
  													__SCREAMING_SNAKE_CASE								=      ['我    能 吞 下 玻 璃', '而不伤身体']
  													# S = 0, D = 5, I = 0, N = 11, CER = 5 / 11
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertTrue(abs(char_error_rate - 0.4_5_4_5_4_5_4_5_4_5_4_5	) < 1e-6	)
  
  													__SCREAMING_SNAKE_CASE								=      ['我能吞下玻璃而不伤身体']
  													__SCREAMING_SNAKE_CASE								=      cer.compute(predictions=_A   ,							references=_A	)
  													self.assertFalse(_A   ,							0.0	)
  
  
  
  
  
  
  						def  _A			(					self	):
  
  
  
  
  
  
  
  													'''simple docstring'''
  
  													__SCREAMING_SNAKE_CASE								=      ['']
  													__SCREAMING_SNAKE_CASE								=      ['Hypothesis']
  													with self.assertRaises(_A	):
  																				cer.compute(predictions=_A   ,							references=_A	)
  
  
  if __name__ == "__main__":
  			unittest.main()
  
  
  
  
  
  
  