




'''simple docstring'''


from typing import Optional, Tuple, Union

import tensorflow as tf

from ...activations_tf import ACTaFN
from ...modeling_tf_outputs import (
    TFBaseModelOutputWithNoAttention,
    TFBaseModelOutputWithPoolingAndNoAttention,
    TFImageClassifierOutputWithNoAttention,
)
from ...modeling_tf_utils import TFPreTrainedModel, TFSequenceClassificationLoss, keras_serializable, unpack_inputs
from ...tf_utils import shape_list
from ...utils import add_code_sample_docstrings, add_start_docstrings, add_start_docstrings_to_model_forward, logging
from .configuration_resnet import ResNetConfig


UpperCamelCase			     =					logging.get_logger(__name__)

# General docstring
UpperCamelCase			     =					'''ResNetConfig'''

# Base docstring
UpperCamelCase			     =					'''microsoft/resnet-50'''
UpperCamelCase			     =					[1, 2048, 7, 7]

# Image classification docstring
UpperCamelCase			     =					'''microsoft/resnet-50'''
UpperCamelCase			     =					'''tiger cat'''

UpperCamelCase			     =					[
    '''microsoft/resnet-50''',
    # See all resnet models at https://huggingface.co/models?filter=resnet
]






class 						lowerCAmelCase_	(       tf.keras.layers.Layer							):







  '''simple docstring'''

  def __init__( self				:							Tuple					,  SCREAMING_SNAKE_CASE_				:							List[str]					,  SCREAMING_SNAKE_CASE_				:							Dict = 3					,  SCREAMING_SNAKE_CASE_				:							Optional[Any] = 1					,  SCREAMING_SNAKE_CASE_				:							List[Any] = "relu"					,  **SCREAMING_SNAKE_CASE_				:							str	)			->		None:







       '''simple docstring'''



       super().__init__(**__a	)
       A:		str          =		kernel_size // 2
       A:		Any          =		tf.keras.layers.ConvaD(
           __a					,  kernel_size=__a					,  strides=__a					,  padding='''valid'''					,  use_bias=__a					,  name='''convolution'''	)
       # Use same default momentum and epsilon as PyTorch equivalent
       A:		Union[str, Any]          =		tf.keras.layers.BatchNormalization(epsilon=1E-5					,  momentum=0.9					,  name='''normalization'''	)
       A:		int          =		ACTaFN[activation] if activation is not None else tf.keras.layers.Activation('''linear'''	)



  def 					_snake_case     ( self				:							Optional[int]					,  SCREAMING_SNAKE_CASE_				:							Union[str, Any]	)			->		tf.Tensor:







       '''simple docstring'''



       A:		Union[str, Any]          =		(self.pad_value, self.pad_value)
       A:		int          =		tf.pad(__a					,  [(0, 0), height_pad, width_pad, (0, 0)]	)
       A:		Any          =		self.conv(__a	)
       return hidden_state




  def 					_snake_case     ( self				:							Optional[int]					,  SCREAMING_SNAKE_CASE_				:							int					,  SCREAMING_SNAKE_CASE_				:							List[Any] = False	)			->		tf.Tensor:







       '''simple docstring'''



       A:		List[Any]          =		self.convolution(__a	)
       A:		Tuple          =		self.normalization(__a					,  training=__a	)
       A:		Union[str, Any]          =		self.activation(__a	)
       return hidden_state






class 						lowerCAmelCase_	(       tf.keras.layers.Layer							):







  '''simple docstring'''

  def __init__( self				:							Dict					,  SCREAMING_SNAKE_CASE_				:							Union[str, Any]					,  **SCREAMING_SNAKE_CASE_				:							str	)			->		None:







       '''simple docstring'''



       super().__init__(**__a	)
       A:		Any          =		TFResNetConvLayer(
           config.embedding_size					,  kernel_size=7					,  stride=2					,  activation=config.hidden_act					,  name='''embedder'''					,  )
       A:		Optional[Any]          =		tf.keras.layers.MaxPoolaD(pool_size=3					,  strides=2					,  padding='''valid'''					,  name='''pooler'''	)
       A:		Any          =		config.num_channels




  def 					_snake_case     ( self				:							Union[str, Any]					,  SCREAMING_SNAKE_CASE_				:							Tuple					,  SCREAMING_SNAKE_CASE_				:							Union[str, Any] = False	)			->		tf.Tensor:







       '''simple docstring'''



       A:		Dict          =		shape_list(__a	)
       if tf.executing_eagerly() and num_channels != self.num_channels:
            raise ValueError(
                '''Make sure that the channel dimension of the pixel values match with the one set in the configuration.'''	)
       A:		Optional[int]          =		pixel_values
       A:		int          =		self.embedder(__a	)
       A:		Any          =		tf.pad(__a					,  [[0, 0], [1, 1], [1, 1], [0, 0]]	)
       A:		Tuple          =		self.pooler(__a	)
       return hidden_state






class 						lowerCAmelCase_	(       tf.keras.layers.Layer							):







  '''simple docstring'''

  def __init__( self				:							Union[str, Any]					,  SCREAMING_SNAKE_CASE_				:							int					,  SCREAMING_SNAKE_CASE_				:							Dict = 2					,  **SCREAMING_SNAKE_CASE_				:							Tuple	)			->		None:







       '''simple docstring'''



       super().__init__(**__a	)
       A:		List[str]          =		tf.keras.layers.ConvaD(
           __a					,  kernel_size=1					,  strides=__a					,  use_bias=__a					,  name='''convolution'''	)
       # Use same default momentum and epsilon as PyTorch equivalent
       A:		Dict          =		tf.keras.layers.BatchNormalization(epsilon=1E-5					,  momentum=0.9					,  name='''normalization'''	)




  def 					_snake_case     ( self				:							Tuple					,  SCREAMING_SNAKE_CASE_				:							List[str]					,  SCREAMING_SNAKE_CASE_				:							List[Any] = False	)			->		tf.Tensor:







       '''simple docstring'''



       A:		List[Any]          =		x
       A:		int          =		self.convolution(__a	)
       A:		Tuple          =		self.normalization(__a					,  training=__a	)
       return hidden_state






class 						lowerCAmelCase_	(       tf.keras.layers.Layer							):







  '''simple docstring'''

  def __init__( self				:							List[Any]					,  SCREAMING_SNAKE_CASE_				:							Dict					,  SCREAMING_SNAKE_CASE_				:							int					,  SCREAMING_SNAKE_CASE_				:							Optional[int] = 1					,  SCREAMING_SNAKE_CASE_				:							Optional[int] = "relu"					,  **SCREAMING_SNAKE_CASE_				:							List[Any]	)			->		None:







       '''simple docstring'''



       super().__init__(**__a	)
       A:		Optional[int]          =		in_channels != out_channels or stride != 1
       A:		List[Any]          =		TFResNetConvLayer(__a					,  stride=__a					,  name='''layer.0'''	)
       A:		Optional[Any]          =		TFResNetConvLayer(__a					,  activation=__a					,  name='''layer.1'''	)
       A:		List[Any]          =		(
           TFResNetShortCut(__a					,  stride=__a					,  name='''shortcut'''	)
           if should_apply_shortcut
           else tf.keras.layers.Activation('''linear'''					,  name='''shortcut'''	)
       )
       A:		Tuple          =		ACTaFN[activation]




  def 					_snake_case     ( self				:							List[str]					,  SCREAMING_SNAKE_CASE_				:							str					,  SCREAMING_SNAKE_CASE_				:							Any = False	)			->		tf.Tensor:







       '''simple docstring'''



       A:		Dict          =		hidden_state
       A:		List[str]          =		self.conva(__a					,  training=__a	)
       A:		List[str]          =		self.conva(__a					,  training=__a	)
       A:		str          =		self.shortcut(__a					,  training=__a	)
       hidden_state += residual
       A:		List[Any]          =		self.activation(__a	)
       return hidden_state






class 						lowerCAmelCase_	(       tf.keras.layers.Layer							):







  '''simple docstring'''

  def __init__( self				:							List[str]					,  SCREAMING_SNAKE_CASE_				:							Any					,  SCREAMING_SNAKE_CASE_				:							Optional[int]					,  SCREAMING_SNAKE_CASE_				:							str = 1					,  SCREAMING_SNAKE_CASE_				:							Tuple = "relu"					,  SCREAMING_SNAKE_CASE_				:							Optional[int] = 4					,  **SCREAMING_SNAKE_CASE_				:							str					,  )			->		None:







       '''simple docstring'''



       super().__init__(**__a	)
       A:		Optional[Any]          =		in_channels != out_channels or stride != 1
       A:		Union[str, Any]          =		out_channels // reduction
       A:		Optional[int]          =		TFResNetConvLayer(__a					,  kernel_size=1					,  name='''layer.0'''	)
       A:		Dict          =		TFResNetConvLayer(__a					,  stride=__a					,  name='''layer.1'''	)
       A:		List[str]          =		TFResNetConvLayer(__a					,  kernel_size=1					,  activation=__a					,  name='''layer.2'''	)
       A:		Dict          =		(
           TFResNetShortCut(__a					,  stride=__a					,  name='''shortcut'''	)
           if should_apply_shortcut
           else tf.keras.layers.Activation('''linear'''					,  name='''shortcut'''	)
       )
       A:		Tuple          =		ACTaFN[activation]




  def 					_snake_case     ( self				:							Optional[Any]					,  SCREAMING_SNAKE_CASE_				:							Dict					,  SCREAMING_SNAKE_CASE_				:							Dict = False	)			->		tf.Tensor:







       '''simple docstring'''



       A:		str          =		hidden_state
       A:		List[str]          =		self.conva(__a					,  training=__a	)
       A:		Any          =		self.conva(__a					,  training=__a	)
       A:		str          =		self.conva(__a					,  training=__a	)
       A:		Tuple          =		self.shortcut(__a					,  training=__a	)
       hidden_state += residual
       A:		int          =		self.activation(__a	)
       return hidden_state






class 						lowerCAmelCase_	(       tf.keras.layers.Layer							):







  '''simple docstring'''

  def __init__( self				:							List[Any]					,  SCREAMING_SNAKE_CASE_				:							str					,  SCREAMING_SNAKE_CASE_				:							Optional[int]					,  SCREAMING_SNAKE_CASE_				:							Union[str, Any]					,  SCREAMING_SNAKE_CASE_				:							str = 2					,  SCREAMING_SNAKE_CASE_				:							Dict = 2					,  **SCREAMING_SNAKE_CASE_				:							List[Any]	)			->		None:







       '''simple docstring'''



       super().__init__(**__a	)

       A:		List[Any]          =		TFResNetBottleNeckLayer if config.layer_type == 'bottleneck' else TFResNetBasicLayer

       A:		int          =		[layer(__a					,  __a					,  stride=__a					,  activation=config.hidden_act					,  name='''layers.0'''	)]
       layers += [
           layer(__a					,  __a					,  activation=config.hidden_act					,  name=f"""layers.{i + 1}"""	)
           for i in range(depth - 1	)
       ]
       A:		Optional[int]          =		layers




  def 					_snake_case     ( self				:							Dict					,  SCREAMING_SNAKE_CASE_				:							Any					,  SCREAMING_SNAKE_CASE_				:							Optional[int] = False	)			->		tf.Tensor:







       '''simple docstring'''



       for layer in self.stage_layers:
            A:		Tuple          =		layer(__a					,  training=__a	)
       return hidden_state






class 						lowerCAmelCase_	(       tf.keras.layers.Layer							):







  '''simple docstring'''

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







       '''simple docstring'''



       super().__init__(**__a	)
       # based on `downsample_in_first_stage` the first layer of the first stage may or may not downsample the input
       A:		str          =		[
           TFResNetStage(
               __a					,  config.embedding_size					,  config.hidden_sizes[0]					,  stride=2 if config.downsample_in_first_stage else 1					,  depth=config.depths[0]					,  name='''stages.0'''					,  )
       ]
       for i, (in_channels, out_channels, depth) in enumerate(
           zip(config.hidden_sizes					,  config.hidden_sizes[1:]					,  config.depths[1:]	)	):
            self.stages.append(TFResNetStage(__a					,  __a					,  __a					,  depth=__a					,  name=f"""stages.{i + 1}"""	)	)




  def 					_snake_case     ( self				:							Tuple					,  SCREAMING_SNAKE_CASE_				:							Any					,  SCREAMING_SNAKE_CASE_				:							Union[str, Any] = False					,  SCREAMING_SNAKE_CASE_				:							str = True					,  SCREAMING_SNAKE_CASE_				:							List[str] = False					,  )			->		TFBaseModelOutputWithNoAttention:







       '''simple docstring'''



       A:		Dict          =		() if output_hidden_states else None

       for stage_module in self.stages:
            if output_hidden_states:
                 A:		Any          =		hidden_states + (hidden_state,)

            A:		Union[str, Any]          =		stage_module(__a					,  training=__a	)

       if output_hidden_states:
            A:		Dict          =		hidden_states + (hidden_state,)

       if not return_dict:
            return tuple(v for v in [hidden_state, hidden_states] if v is not None	)

       return TFBaseModelOutputWithNoAttention(last_hidden_state=__a					,  hidden_states=__a	)






class 						lowerCAmelCase_	(       UpperCamelCase_							):







  '''simple docstring'''

  UpperCamelCase_			:      str										=				ResNetConfig
  UpperCamelCase_			:      str										=				"""resnet"""
  UpperCamelCase_			:      Dict										=				"""pixel_values"""



  @property
  def 					_snake_case     ( self				:							str	)			->		List[Any]:







       '''simple docstring'''



       return {"pixel_values": tf.TensorSpec(shape=(None, self.config.num_channels, 2_24, 2_24)					,  dtype=tf.floataa	)}






UpperCamelCase			     =					R'''
    This model is a TensorFlow
    [tf.keras.layers.Layer](https://www.tensorflow.org/api_docs/python/tf/keras/layers/Layer) sub-class. Use it as a
    regular TensorFlow Module and refer to the TensorFlow documentation for all matter related to general usage and
    behavior.

    Parameters:
        config ([`ResNetConfig`]): Model configuration class with all the parameters of the model.
            Initializing with a config file does not load the weights associated with the model, only the
            configuration. Check out the [`~TFPreTrainedModel.from_pretrained`] method to load the model weights.
'''


UpperCamelCase			     =					R'''
    Args:
        pixel_values (`tf.Tensor` of shape `(batch_size, num_channels, height, width)`):
            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See
            [`ConvNextImageProcessor.__call__`] for details.

        output_hidden_states (`bool`, *optional*):
            Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for
            more detail.
        return_dict (`bool`, *optional*):
            Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.
'''






@keras_serializable
class 						lowerCAmelCase_	(       tf.keras.layers.Layer							):







  '''simple docstring'''

  UpperCamelCase_			:      Tuple										=				ResNetConfig



  def __init__( self				:							Optional[Any]					,  SCREAMING_SNAKE_CASE_				:							Optional[Any]					,  **SCREAMING_SNAKE_CASE_				:							List[Any]	)			->		None:







       '''simple docstring'''



       super().__init__(**__a	)
       A:		Tuple          =		config
       A:		Dict          =		TFResNetEmbeddings(__a					,  name='''embedder'''	)
       A:		Union[str, Any]          =		TFResNetEncoder(__a					,  name='''encoder'''	)
       A:		int          =		tf.keras.layers.GlobalAveragePoolingaD(keepdims=__a	)




  @unpack_inputs
  def 					_snake_case     ( self				:							Dict					,  SCREAMING_SNAKE_CASE_				:							List[str]					,  SCREAMING_SNAKE_CASE_				:							List[Any] = None					,  SCREAMING_SNAKE_CASE_				:							Union[str, Any] = None					,  SCREAMING_SNAKE_CASE_				:							List[Any] = False					,  )			->		Union[Tuple[tf.Tensor], TFBaseModelOutputWithPoolingAndNoAttention]:







       '''simple docstring'''



       A:		Union[str, Any]          =		(
           output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
       )
       A:		Any          =		return_dict if return_dict is not None else self.config.use_return_dict

       # TF 2.0 image layers can't use NCHW format when running on CPU.
       # We transpose to NHWC format and then transpose back after the full forward pass.
       # (batch_size, num_channels, height, width) -> (batch_size, height, width, num_channels)
       A:		Optional[Any]          =		tf.transpose(__a					,  perm=[0, 2, 3, 1]	)
       A:		Union[str, Any]          =		self.embedder(__a					,  training=__a	)

       A:		Any          =		self.encoder(
           __a					,  output_hidden_states=__a					,  return_dict=__a					,  training=__a	)

       A:		Optional[Any]          =		encoder_outputs[0]

       A:		Any          =		self.pooler(__a	)

       # Transpose all the outputs to the NCHW format
       # (batch_size, height, width, num_channels) -> (batch_size, num_channels, height, width)
       A:		str          =		tf.transpose(__a					,  (0, 3, 1, 2)	)
       A:		int          =		tf.transpose(__a					,  (0, 3, 1, 2)	)
       A:		str          =		()
       for hidden_state in encoder_outputs[1:]:
            A:		str          =		hidden_states + tuple(tf.transpose(__a					,  (0, 3, 1, 2)	) for h in hidden_state	)

       if not return_dict:
            return (last_hidden_state, pooled_output) + hidden_states

       A:		Union[str, Any]          =		hidden_states if output_hidden_states else None

       return TFBaseModelOutputWithPoolingAndNoAttention(
           last_hidden_state=__a					,  pooler_output=__a					,  hidden_states=__a					,  )






@add_start_docstrings(
    """The bare ResNet model outputting raw features without any specific head on top."""       ,					UpperCamelCase_       ,					)
class 						lowerCAmelCase_	(       UpperCamelCase_							):







  '''simple docstring'''

  def __init__( self				:							int					,  SCREAMING_SNAKE_CASE_				:							Any					,  **SCREAMING_SNAKE_CASE_				:							List[Any]	)			->		None:







       '''simple docstring'''



       super().__init__(__a					,  **__a	)
       A:		Union[str, Any]          =		TFResNetMainLayer(config=__a					,  name='''resnet'''	)




  @add_start_docstrings_to_model_forward(__a	)
  @add_code_sample_docstrings(
      checkpoint=_CHECKPOINT_FOR_DOC					,  output_type=__a					,  config_class=_CONFIG_FOR_DOC					,  modality='''vision'''					,  expected_output=_EXPECTED_OUTPUT_SHAPE					,  )
  @unpack_inputs
  def 					_snake_case     ( self				:							Any					,  SCREAMING_SNAKE_CASE_				:							Tuple					,  SCREAMING_SNAKE_CASE_				:							Any = None					,  SCREAMING_SNAKE_CASE_				:							str = None					,  SCREAMING_SNAKE_CASE_				:							Union[str, Any] = False					,  )			->		Union[Tuple[tf.Tensor], TFBaseModelOutputWithPoolingAndNoAttention]:







       '''simple docstring'''



       A:		Optional[Any]          =		(
           output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
       )
       A:		Optional[int]          =		return_dict if return_dict is not None else self.config.use_return_dict

       A:		int          =		self.resnet(
           pixel_values=__a					,  output_hidden_states=__a					,  return_dict=__a					,  training=__a					,  )
       return resnet_outputs






@add_start_docstrings(
    """
    ResNet Model with an image classification head on top (a linear layer on top of the pooled features), e.g. for
    ImageNet.
    """       ,					UpperCamelCase_       ,					)
class 						lowerCAmelCase_	(       UpperCamelCase_       ,					UpperCamelCase_							):







  '''simple docstring'''

  def __init__( self				:							str					,  SCREAMING_SNAKE_CASE_				:							str					,  **SCREAMING_SNAKE_CASE_				:							int	)			->		None:







       '''simple docstring'''



       super().__init__(__a					,  **__a	)
       A:		Union[str, Any]          =		config.num_labels
       A:		List[str]          =		TFResNetMainLayer(__a					,  name='''resnet'''	)
       # classification head
       A:		Tuple          =		(
           tf.keras.layers.Dense(config.num_labels					,  name='''classifier.1'''	)
           if config.num_labels > 0
           else tf.keras.layers.Activation('''linear'''					,  name='''classifier.1'''	)
       )



  def 					_snake_case     ( self				:							Tuple					,  SCREAMING_SNAKE_CASE_				:							Dict	)			->		tf.Tensor:







       '''simple docstring'''



       A:		Dict          =		tf.keras.layers.Flatten()(__a	)
       A:		str          =		self.classifier_layer(__a	)
       return logits




  @add_start_docstrings_to_model_forward(__a	)
  @add_code_sample_docstrings(
      checkpoint=_IMAGE_CLASS_CHECKPOINT					,  output_type=__a					,  config_class=_CONFIG_FOR_DOC					,  expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT					,  )
  @unpack_inputs
  def 					_snake_case     ( self				:							Tuple					,  SCREAMING_SNAKE_CASE_				:							int = None					,  SCREAMING_SNAKE_CASE_				:							List[Any] = None					,  SCREAMING_SNAKE_CASE_				:							Optional[Any] = None					,  SCREAMING_SNAKE_CASE_				:							int = None					,  SCREAMING_SNAKE_CASE_				:							Union[str, Any] = False					,  )			->		Union[Tuple[tf.Tensor], TFImageClassifierOutputWithNoAttention]:







       '''simple docstring'''



       A:		Dict          =		return_dict if return_dict is not None else self.config.use_return_dict

       A:		Optional[int]          =		self.resnet(
           __a					,  output_hidden_states=__a					,  return_dict=__a					,  training=__a	)

       A:		Union[str, Any]          =		outputs.pooler_output if return_dict else outputs[1]

       A:		Optional[Any]          =		self.classifier(__a	)

       A:		List[str]          =		None if labels is None else self.hf_compute_loss(__a					,  __a	)

       if not return_dict:
            A:		List[Any]          =		(logits,) + outputs[2:]
            return (loss,) + output if loss is not None else output

       return TFImageClassifierOutputWithNoAttention(loss=__a					,  logits=__a					,  hidden_states=outputs.hidden_states	)














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




def 							_SCREAMING_SNAKE_CASE	(					)     ->			Optional[Any]:







   """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_     (     UpperCamelCase_				):







   def __init__(self       , __a)      ->   Optional[Any]:
      """simple docstring"""






      set_hf_deepspeed_config(self)
      dep_version_check('accelerate')
      dep_version_check('deepspeed')
      super().__init__(__a)




class a_     (     UpperCamelCase_				):







   def __init__(self       , __a)      ->   List[str]:
      """simple docstring"""






      super().__init__(__a)
      __snake_case					:   Dict  				=			None
      __snake_case					:   Optional[Any]  				=			[]







   def      SCREAMING_SNAKE_CASE__       (self)      ->   Dict:
      """simple docstring"""






      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       , __a)      ->   int:
      """simple docstring"""






      __snake_case					:   Dict  				=			self.get_value(__a)
      if val is None:
         return False
      else:
         return val == "auto"







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a=None       , __a=True)      ->   int:
      """simple docstring"""






      __snake_case				,__snake_case					:   str  				=			self.find_config_node(__a)
      if config is None:
         return

      if config.get(__a) == "auto":
         __snake_case					:   List[Any]  				=			hf_val
         return

      if not must_match:
         return

      __snake_case					:   Optional[Any]  				=			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}""")

   _snake_case	   =				partialmethod(UpperCamelCase_		,   must_match=UpperCamelCase_				)







   def      SCREAMING_SNAKE_CASE__       (self       , __a)      ->   Union[str, Any]:
      """simple docstring"""






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







   def      SCREAMING_SNAKE_CASE__       (self       , __a       , __a       , __a)      ->   Any:
      """simple docstring"""






      __snake_case					:   Union[str, Any]  				=			[
          'zero_optimization.reduce_bucket_size',
          'zero_optimization.stage3_prefetch_bucket_size',
          'zero_optimization.stage3_param_persistence_threshold',
      ]
      __snake_case					:   int  				=			[x for x in hidden_size_based_keys if self.is_auto(__a)]

      if len(__a) > 0:
         if hasattr(model.config       , 'hidden_size'):
            __snake_case					:   Optional[int]  				=			model.config.hidden_size
         elif hasattr(model.config       , 'hidden_sizes'):
            # if there are many hidden sizes pick the largest one
            __snake_case					:   Tuple  				=			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'       , 1_0 * 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:
         __snake_case					:   Any  				=			'\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 							_SCREAMING_SNAKE_CASE	(					A	:							Dict	)     ->			Union[str, Any]:







   """simple docstring"""
   # 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)
   __snake_case					:   List[Any]  				=			weakref.ref(A	)




def 							_SCREAMING_SNAKE_CASE	(					)     ->			Union[str, Any]:







   """simple docstring"""
   # useful for unit tests to ensure the global state doesn't leak - call from `tearDown` method
   global _hf_deepspeed_config_weak_ref
   __snake_case					:   Optional[Any]  				=			None




def 							_SCREAMING_SNAKE_CASE	(					)     ->			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 							_SCREAMING_SNAKE_CASE	(					)     ->			List[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 							_SCREAMING_SNAKE_CASE	(					A	:							Dict	,		A	:							int	,		A	:							Optional[Any]	,		A	:							Optional[Any]	,		A	:							str	)     ->			Union[str, Any]:







   """simple docstring"""
   from accelerate.utils import DummyOptim, DummyScheduler

   __snake_case					:   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)

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

   __snake_case					:   List[Any]  				=			None
   if "scheduler" in config:
      __snake_case					:   Optional[int]  				=			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.'	)
      __snake_case					:   int  				=			trainer.create_scheduler(num_training_steps=A	,		optimizer=A	)

   return optimizer, lr_scheduler




def 							_SCREAMING_SNAKE_CASE	(					A	:							Dict	,		A	:							List[Any]	,		A	:							str=False	)     ->			Union[str, Any]:







   """simple docstring"""
   from deepspeed.utils import logger as ds_logger

   __snake_case					:   int  				=			trainer.model
   __snake_case					:   Any  				=			trainer.args

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

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

   return optimizer, lr_scheduler





def 							_SCREAMING_SNAKE_CASE	(					A	:							Any	,		A	:							int	)     ->			List[Any]:







   """simple docstring"""
   # 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

   __snake_case					:   int  				=			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
      __snake_case				,__snake_case					:   Optional[Any]  				=			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}"""	)