






'''simple docstring'''






import torch
import torch.nn as nn
from transformers.modeling_utils import ModuleUtilsMixin
from transformers.models.ta.modeling_ta import TaBlock, TaConfig, TaLayerNorm

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin

class 							a__( __lowercase		,			__lowercase		,			__lowercase     ):







    @register_to_config
    def __init__(				self			:  Optional[Any]						,      __snake_case			:  int						,      __snake_case			:  int						,      __snake_case			:  int						,      __snake_case			:  float						,      __snake_case			:  int						,      __snake_case			:  int						,      __snake_case			:  int						,      __snake_case			:  int						,      __snake_case			:  str						,      __snake_case			:  bool = False						,      ):
      super().__init__()

      a						: Optional[int]					       =					nn.Embedding(snake_case_						,      snake_case_     )

      a						: int					       =					nn.Embedding(snake_case_						,      snake_case_     )
      a						: List[str]					       =					False

      a						: List[str]					       =					nn.Dropout(p=snake_case_     )

      a						: int					       =					TaConfig(
          vocab_size=snake_case_						,      d_model=snake_case_						,      num_heads=snake_case_						,      d_kv=snake_case_						,      d_ff=snake_case_						,      dropout_rate=snake_case_						,      feed_forward_proj=snake_case_						,      is_decoder=snake_case_						,      is_encoder_decoder=snake_case_						,      )

      a						: Optional[int]					       =					nn.ModuleList()
      for lyr_num in range(snake_case_     ):
        a						: Dict					       =					TaBlock(snake_case_     )
        self.encoders.append(snake_case_     )

      a						: str					       =					TaLayerNorm(snake_case_     )
      a						: Dict					       =					nn.Dropout(p=snake_case_     )







    def       lowercase_	(				self			:  Optional[Any]						,      __snake_case			:  Any						,      __snake_case			:  int     ):
      a						: Tuple					       =					self.token_embedder(snake_case_     )

      a						: Tuple					       =					encoder_input_tokens.shape[1]
      a						: str					       =					torch.arange(snake_case_						,      device=encoder_input_tokens.device     )
      x += self.position_encoding(snake_case_     )

      a						: Any					       =					self.dropout_pre(snake_case_     )

      # inverted the attention mask
      a						: Tuple					       =					encoder_input_tokens.size()
      a						: List[str]					       =					self.get_extended_attention_mask(snake_case_						,      snake_case_     )

      for lyr in self.encoders:
        a						: Tuple					       =					lyr(snake_case_						,      snake_case_     )[0]
      a						: List[str]					       =					self.layer_norm(snake_case_     )

      return self.dropout_post(snake_case_     ), encoder_inputs_mask




'''simple docstring'''






import re
from typing import Callable, List, Optional, Union

import tensorflow as tf


try:
     from tensorflow.keras.optimizers.legacy import Adam
except ImportError:
     from tensorflow.keras.optimizers import Adam

class 							a__( tf.keras.optimizers.schedules.LearningRateSchedule     ):







    def __init__(				self			:  Optional[Any]						,      __snake_case			:  float						,      __snake_case			:  Callable						,      __snake_case			:  int						,      __snake_case			:  float = 1.0						,      __snake_case			:  str = None						,      ):
      super().__init__()
      a						: Dict					       =					initial_learning_rate
      a						: Optional[int]					       =					warmup_steps
      a						: Tuple					       =					power
      a						: str					       =					decay_schedule_fn
      a						: Optional[int]					       =					name







    def __call__(				self			:  List[Any]						,      __snake_case			:  int     ):
      with tf.name_scope(self.name or 'WarmUp'     ) as name:
        # Implements polynomial warmup. i.e., if global_step < warmup_steps, the
        # learning rate will be `global_step/num_warmup_steps * init_lr`.
        a						: Optional[int]					       =					tf.cast(__snake_case						,      tf.floataa     )
        a						: Tuple					       =					tf.cast(self.warmup_steps						,      tf.floataa     )
        a						: str					       =					global_step_float / warmup_steps_float
        a						: Union[str, Any]					       =					self.initial_learning_rate * tf.math.pow(__snake_case						,      self.power     )
        return tf.cond(
            global_step_float < warmup_steps_float						,      lambda: warmup_learning_rate						,      lambda: self.decay_schedule_fn(step - self.warmup_steps     )						,      name=__snake_case						,      )







    def       lowercase_	(				self			:  Dict     ):
      return {
          "initial_learning_rate": self.initial_learning_rate,
          "decay_schedule_fn": self.decay_schedule_fn,
          "warmup_steps": self.warmup_steps,
          "power": self.power,
          "name": self.name,
      }
def 				lowerCamelCase__	(						_A			,      _A			,      _A			,      _A = 0.0			,      _A = 0.9			,      _A = 0.999			,      _A = 1E-8			,      _A = None			,      _A = None			,      _A = 0.0			,      _A = 1.0			,      _A = None			,      ):
  a						: Optional[int]					       =					tf.keras.optimizers.schedules.PolynomialDecay(
      initial_learning_rate=_A			,      decay_steps=num_train_steps - num_warmup_steps			,      end_learning_rate=init_lr * min_lr_ratio			,      power=_A			,      )
  if num_warmup_steps:
    a						: Tuple					       =					WarmUp(
        initial_learning_rate=_A			,      decay_schedule_fn=_A			,      warmup_steps=_A			,      )
  if weight_decay_rate > 0.0:
    a						: List[Any]					       =					AdamWeightDecay(
        learning_rate=_A			,      weight_decay_rate=_A			,      beta_a=_A			,      beta_a=_A			,      epsilon=_A			,      clipnorm=_A			,      global_clipnorm=_A			,      exclude_from_weight_decay=['LayerNorm', 'layer_norm', 'bias']			,      include_in_weight_decay=_A			,      )
  else:
    a						: Tuple					       =					tf.keras.optimizers.Adam(
        learning_rate=_A			,      beta_a=_A			,      beta_a=_A			,      epsilon=_A			,      clipnorm=_A			,      global_clipnorm=_A			,      )
  # We return the optimizer and the LR scheduler in order to better track the
  # evolution of the LR independently of the optimizer.
  return optimizer, lr_schedule

class 							a__( lowerCamelCase__     ):







    def __init__(				self			:  Union[str, Any]						,      __snake_case			:  Union[float, tf.keras.optimizers.schedules.LearningRateSchedule] = 0.001						,      __snake_case			:  float = 0.9						,      __snake_case			:  float = 0.999						,      __snake_case			:  float = 1e-7						,      __snake_case			:  bool = False						,      __snake_case			:  float = 0.0						,      __snake_case			:  Optional[List[str]] = None						,      __snake_case			:  Optional[List[str]] = None						,      __snake_case			:  str = "AdamWeightDecay"						,      **__snake_case			:  Optional[int]						,      ):
      super().__init__(__snake_case						,      __snake_case						,      __snake_case						,      __snake_case						,      __snake_case						,      __snake_case						,      **__snake_case     )
      a						: Tuple					       =					weight_decay_rate
      a						: Optional[Any]					       =					include_in_weight_decay
      a						: Any					       =					exclude_from_weight_decay







    @classmethod
    def       lowercase_	(				cls			:  Any						,      __snake_case			:  Optional[Any]     ):
      a						: Any					       =					{'WarmUp': WarmUp}
      return super(__snake_case						,      cls     ).from_config(__snake_case						,      custom_objects=__snake_case     )







    def       lowercase_	(				self			:  str						,      __snake_case			:  Optional[int]						,      __snake_case			:  Dict						,      __snake_case			:  Optional[int]     ):
      super(__snake_case						,      self     )._prepare_local(__snake_case						,      __snake_case						,      __snake_case     )
      a						: List[str]					       =					tf.constant(
          self.weight_decay_rate						,      name='adam_weight_decay_rate'     )







    def       lowercase_	(				self			:  int						,      __snake_case			:  Dict						,      __snake_case			:  List[Any]						,      __snake_case			:  str     ):
      a						: Optional[Any]					       =					self._do_use_weight_decay(var.name     )
      if do_decay:
        return var.assign_sub(
            learning_rate * var * apply_state[(var.device, var.dtype.base_dtype)]['weight_decay_rate']						,      use_locking=self._use_locking						,      )
      return tf.no_op()







    def       lowercase_	(				self			:  List[str]						,      __snake_case			:  List[str]						,      __snake_case			:  Any=None						,      **__snake_case			:  Tuple     ):
      a       ,     a						: Dict					       =					list(zip(*__snake_case     )     )
      return super(__snake_case						,      self     ).apply_gradients(zip(__snake_case						,      __snake_case     )						,      name=__snake_case						,      **__snake_case     )







    def       lowercase_	(				self			:  List[Any]						,      __snake_case			:  List[str]						,      __snake_case			:  Union[str, Any]						,      __snake_case			:  Optional[int]     ):
      if apply_state is None:
        return self._decayed_lr_t[var_dtype], {}

      a						: List[Any]					       =					apply_state or {}
      a						: Optional[Any]					       =					apply_state.get((var_device, var_dtype)     )
      if coefficients is None:
        a						: int					       =					self._fallback_apply_state(__snake_case						,      __snake_case     )
        a						: Any					       =					coefficients

      return coefficients["lr_t"], {"apply_state": apply_state}







    def       lowercase_	(				self			:  Dict						,      __snake_case			:  Union[str, Any]						,      __snake_case			:  Dict						,      __snake_case			:  Tuple=None     ):
      a       ,     a						: List[Any]					       =					self._get_lr(var.device						,      var.dtype.base_dtype						,      __snake_case     )
      a						: List[Any]					       =					self._decay_weights_op(__snake_case						,      __snake_case						,      __snake_case     )
      with tf.control_dependencies([decay]     ):
        return super(__snake_case						,      self     )._resource_apply_dense(__snake_case						,      __snake_case						,      **__snake_case     )







    def       lowercase_	(				self			:  Any						,      __snake_case			:  Optional[Any]						,      __snake_case			:  Optional[Any]						,      __snake_case			:  List[str]						,      __snake_case			:  int=None     ):
      a       ,     a						: Tuple					       =					self._get_lr(var.device						,      var.dtype.base_dtype						,      __snake_case     )
      a						: int					       =					self._decay_weights_op(__snake_case						,      __snake_case						,      __snake_case     )
      with tf.control_dependencies([decay]     ):
        return super(__snake_case						,      self     )._resource_apply_sparse(__snake_case						,      __snake_case						,      __snake_case						,      **__snake_case     )







    def       lowercase_	(				self			:  Optional[Any]     ):
      a						: Union[str, Any]					       =					super().get_config()
      config.update({'weight_decay_rate': self.weight_decay_rate}     )
      return config







    def       lowercase_	(				self			:  List[str]						,      __snake_case			:  Dict     ):
      if self.weight_decay_rate == 0:
        return False

      if self._include_in_weight_decay:
        for r in self._include_in_weight_decay:
          if re.search(__snake_case						,      __snake_case     ) is not None:
            return True

      if self._exclude_from_weight_decay:
        for r in self._exclude_from_weight_decay:
          if re.search(__snake_case						,      __snake_case     ) is not None:
            return False
      return True

class 							a__( lowerCamelCase__     ):







    def __init__(				self			:  List[str]     ):
      a						: List[Any]					       =					[]
      a						: Optional[int]					       =					None







    @property
    def       lowercase_	(				self			:  Any     ):
      if self._accum_steps is None:
        a						: List[Any]					       =					tf.Variable(
            tf.constant(0						,      dtype=tf.intaa     )						,      trainable=__snake_case						,      synchronization=tf.VariableSynchronization.ON_READ						,      aggregation=tf.VariableAggregation.ONLY_FIRST_REPLICA						,      )

      return self._accum_steps.value()







    @property
    def       lowercase_	(				self			:  Optional[int]     ):
      if not self._gradients:
        raise ValueError('The accumulator should be called first to initialize the gradients'     )
      return [gradient.value() if gradient is not None else gradient for gradient in self._gradients]







    def __call__(				self			:  List[Any]						,      __snake_case			:  Any     ):
      if not self._gradients:
        a						: Union[str, Any]					       =					self.step  # Create the step variable.
        self._gradients.extend(
            [
                tf.Variable(
                    tf.zeros_like(__snake_case     )						,      trainable=__snake_case						,      synchronization=tf.VariableSynchronization.ON_READ						,      aggregation=tf.VariableAggregation.ONLY_FIRST_REPLICA						,      )
                if gradient is not None
                else gradient
                for gradient in gradients
            ]     )
      if len(__snake_case     ) != len(self._gradients     ):
        raise ValueError(F"""Expected {len(self._gradients     )} gradients, but got {len(__snake_case     )}"""     )

      for accum_gradient, gradient in zip(self._gradients						,      __snake_case     ):
        if accum_gradient is not None and gradient is not None:
          accum_gradient.assign_add(__snake_case     )

      self._accum_steps.assign_add(1     )







    def       lowercase_	(				self			:  Tuple     ):
      if not self._gradients:
        return
      self._accum_steps.assign(0     )
      for gradient in self._gradients:
        if gradient is not None:
          gradient.assign(tf.zeros_like(__snake_case     )     )