

'''simple docstring'''
import logging

import torch
from torch import nn
from torch.nn import CrossEntropyLoss, MSELoss

from transformers.file_utils import add_start_docstrings, add_start_docstrings_to_model_forward
from transformers.models.bert.modeling_bert import (
    BERT_INPUTS_DOCSTRING,
    BERT_START_DOCSTRING,
    BertEncoder,
    BertModel,
    BertPreTrainedModel,
)


_snake_case     =  logging.getLogger(__name__)


class    a__  (							lowerCamelCase_  ):


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase=None			,				_UpperCamelCase=None  ):





      """simple docstring"""





      _lowercase			:      Optional[int]			     =   self.layer[current_layer](_UpperCamelCase			,				_UpperCamelCase			,				head_mask[current_layer]  )

      _lowercase			:      Dict			     =   layer_outputs[0]

      return hidden_states




@add_start_docstrings(
    'The bare Bert Model transformer with PABEE outputting raw hidden-states without any specific head on top.'  ,      lowerCamelCase_  ,      )
class    a__  (							lowerCamelCase_  ):


   def __init__( self			,				_UpperCamelCase  ):





      """simple docstring"""





      super().__init__(_UpperCamelCase  )

      _lowercase			:      List[str]			     =   BertEncoderWithPabee(_UpperCamelCase  )

      self.init_weights()
      _lowercase			:      List[str]			     =   0
      _lowercase			:      Optional[int]			     =   0
      _lowercase			:      Tuple			     =   0

      _lowercase			:      Any			     =   0


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase  ):





      """simple docstring"""





      _lowercase			:      Dict			     =   threshold


   def 						_lowerCamelCase   ( self			,				_UpperCamelCase  ):





      """simple docstring"""





      _lowercase			:      Any			     =   patience


   def 						_lowerCamelCase   ( self  ):





      """simple docstring"""





      _lowercase			:      Optional[int]			     =   0
      _lowercase			:      int			     =   0


   def 						_lowerCamelCase   ( self  ):





      """simple docstring"""





      _lowercase			:      str			     =   self.inference_layers_num / self.inference_instances_num
      _lowercase			:      List[Any]			     =   (
          f'''*** Patience = {self.patience} Avg. Inference Layers = {avg_inf_layers:.2f} Speed Up ='''
          f''' {1 - avg_inf_layers / self.config.num_hidden_layers:.2f} ***'''
      )
      print(_UpperCamelCase  )






   @add_start_docstrings_to_model_forward(_UpperCamelCase  )
   def 						_lowerCamelCase   ( self			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=False			,				):





      """simple docstring"""





      if input_ids is not None and inputs_embeds is not None:
         raise ValueError("You cannot specify both input_ids and inputs_embeds at the same time"  )
      elif input_ids is not None:
         _lowercase			:      str			     =   input_ids.size()
      elif inputs_embeds is not None:
         _lowercase			:      Tuple			     =   inputs_embeds.size()[:-1]
      else:
         raise ValueError("You have to specify either input_ids or inputs_embeds"  )

      _lowercase			:      List[Any]			     =   input_ids.device if input_ids is not None else inputs_embeds.device

      if attention_mask is None:
         _lowercase			:      Union[str, Any]			     =   torch.ones(_UpperCamelCase			,				device=_UpperCamelCase  )
      if token_type_ids is None:
         _lowercase			:      List[str]			     =   torch.zeros(_UpperCamelCase			,				dtype=torch.long			,				device=_UpperCamelCase  )

      # We can provide a self-attention mask of dimensions [batch_size, from_seq_length, to_seq_length]
      # ourselves in which case we just need to make it broadcastable to all heads.
      _lowercase			:      str			     =   self.get_extended_attention_mask(_UpperCamelCase			,				_UpperCamelCase			,				_UpperCamelCase  )

      # If a 2D ou 3D attention mask is provided for the cross-attention
      # we need to make broadcastable to [batch_size, num_heads, seq_length, seq_length]
      if self.config.is_decoder and encoder_hidden_states is not None:
         _lowercase     ,      _lowercase     ,      _lowercase			:      Any			     =   encoder_hidden_states.size()
         _lowercase			:      Optional[Any]			     =   (encoder_batch_size, encoder_sequence_length)
         if encoder_attention_mask is None:
            _lowercase			:      Any			     =   torch.ones(_UpperCamelCase			,				device=_UpperCamelCase  )
         _lowercase			:      List[Any]			     =   self.invert_attention_mask(_UpperCamelCase  )
      else:
         _lowercase			:      Dict			     =   None

      # Prepare head mask if needed
      # 1.0 in head_mask indicate we keep the head
      # attention_probs has shape bsz x n_heads x N x N
      # input head_mask has shape [num_heads] or [num_hidden_layers x num_heads]
      # and head_mask is converted to shape [num_hidden_layers x batch x num_heads x seq_length x seq_length]
      _lowercase			:      Optional[Any]			     =   self.get_head_mask(_UpperCamelCase			,				self.config.num_hidden_layers  )

      _lowercase			:      Optional[Any]			     =   self.embeddings(
          input_ids=_UpperCamelCase			,				position_ids=_UpperCamelCase			,				token_type_ids=_UpperCamelCase			,				inputs_embeds=_UpperCamelCase  )
      _lowercase			:      List[str]			     =   embedding_output

      if self.training:
         _lowercase			:      int			     =   []
         for i in range(self.config.num_hidden_layers  ):
            _lowercase			:      int			     =   self.encoder.adaptive_forward(
                _UpperCamelCase			,				current_layer=_UpperCamelCase			,				attention_mask=_UpperCamelCase			,				head_mask=_UpperCamelCase  )

            _lowercase			:      Dict			     =   self.pooler(_UpperCamelCase  )
            _lowercase			:      Any			     =   output_layers[i](output_dropout(_UpperCamelCase  )  )
            res.append(_UpperCamelCase  )
      elif self.patience == 0:  # Use all layers for inference
         _lowercase			:      Optional[int]			     =   self.encoder(
             _UpperCamelCase			,				attention_mask=_UpperCamelCase			,				head_mask=_UpperCamelCase			,				encoder_hidden_states=_UpperCamelCase			,				encoder_attention_mask=_UpperCamelCase			,				)
         _lowercase			:      int			     =   self.pooler(encoder_outputs[0]  )
         _lowercase			:      Any			     =   [output_layers[self.config.num_hidden_layers - 1](_UpperCamelCase  )]
      else:
         _lowercase			:      Optional[int]			     =   0
         _lowercase			:      List[Any]			     =   None
         _lowercase			:      str			     =   0
         for i in range(self.config.num_hidden_layers  ):
            calculated_layer_num += 1
            _lowercase			:      int			     =   self.encoder.adaptive_forward(
                _UpperCamelCase			,				current_layer=_UpperCamelCase			,				attention_mask=_UpperCamelCase			,				head_mask=_UpperCamelCase  )

            _lowercase			:      Union[str, Any]			     =   self.pooler(_UpperCamelCase  )
            _lowercase			:      Any			     =   output_layers[i](_UpperCamelCase  )
            if regression:
               _lowercase			:      List[Any]			     =   logits.detach()
               if patient_result is not None:
                  _lowercase			:      int			     =   patient_result.detach()
               if (patient_result is not None) and torch.abs(patient_result - labels  ) < self.regression_threshold:
                  patient_counter += 1
               else:
                  _lowercase			:      Any			     =   0
            else:
               _lowercase			:      Union[str, Any]			     =   logits.detach().argmax(dim=1  )
               if patient_result is not None:
                  _lowercase			:      List[str]			     =   patient_result.detach().argmax(dim=1  )
               if (patient_result is not None) and torch.all(labels.eq(_UpperCamelCase  )  ):
                  patient_counter += 1
               else:
                  _lowercase			:      Dict			     =   0

            _lowercase			:      List[Any]			     =   logits
            if patient_counter == self.patience:
               break
         _lowercase			:      Optional[int]			     =   [patient_result]
         self.inference_layers_num += calculated_layer_num
         self.inference_instances_num += 1

      return res




@add_start_docstrings(
    'Bert Model transformer with PABEE and a sequence classification/regression head on top (a linear layer on top of\n    the pooled output) e.g. for GLUE tasks. '  ,      lowerCamelCase_  ,      )
class    a__  (							lowerCamelCase_  ):


   def __init__( self			,				_UpperCamelCase  ):





      """simple docstring"""





      super().__init__(_UpperCamelCase  )
      _lowercase			:      List[str]			     =   config.num_labels

      _lowercase			:      Optional[int]			     =   BertModelWithPabee(_UpperCamelCase  )
      _lowercase			:      Dict			     =   nn.Dropout(config.hidden_dropout_prob  )
      _lowercase			:      List[str]			     =   nn.ModuleList(
          [nn.Linear(config.hidden_size			,				self.config.num_labels  ) for _ in range(config.num_hidden_layers  )]  )

      self.init_weights()






   @add_start_docstrings_to_model_forward(_UpperCamelCase  )
   def 						_lowerCamelCase   ( self			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				_UpperCamelCase=None			,				):





      """simple docstring"""





      _lowercase			:      Optional[Any]			     =   self.bert(
          input_ids=_UpperCamelCase			,				attention_mask=_UpperCamelCase			,				token_type_ids=_UpperCamelCase			,				position_ids=_UpperCamelCase			,				head_mask=_UpperCamelCase			,				inputs_embeds=_UpperCamelCase			,				output_dropout=self.dropout			,				output_layers=self.classifiers			,				regression=self.num_labels == 1			,				)

      _lowercase			:      Union[str, Any]			     =   (logits[-1],)

      if labels is not None:
         _lowercase			:      Optional[int]			     =   None
         _lowercase			:      Optional[Any]			     =   0
         for ix, logits_item in enumerate(_UpperCamelCase  ):
            if self.num_labels == 1:
               #  We are doing regression
               _lowercase			:      Optional[int]			     =   MSELoss()
               _lowercase			:      str			     =   loss_fct(logits_item.view(-1  )			,				labels.view(-1  )  )
            else:
               _lowercase			:      Dict			     =   CrossEntropyLoss()
               _lowercase			:      str			     =   loss_fct(logits_item.view(-1			,				self.num_labels  )			,				labels.view(-1  )  )
            if total_loss is None:
               _lowercase			:      Dict			     =   loss
            else:
               total_loss += loss * (ix + 1)
            total_weights += ix + 1
         _lowercase			:      Optional[Any]			     =   (total_loss / total_weights,) + outputs

      return outputs









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  __lowerCAmelCase     (					SCREAMING_SNAKE_CASE			,			SCREAMING_SNAKE_CASE			,			SCREAMING_SNAKE_CASE							):




  @register_to_config
  def __init__(				self							,   lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase							,   lowerCAmelCase = False							,   )  -> Tuple:
        '''simple docstring'''






        super().__init__()

        _lowercase	       =nn.Embedding(lowerCAmelCase							,   lowerCAmelCase					)

        _lowercase	       =nn.Embedding(lowerCAmelCase							,   lowerCAmelCase					)
        _lowercase	       =False

        _lowercase	       =nn.Dropout(p=lowerCAmelCase					)

        _lowercase	       =TaConfig(
            vocab_size=lowerCAmelCase							,   d_model=lowerCAmelCase							,   num_heads=lowerCAmelCase							,   d_kv=lowerCAmelCase							,   d_ff=lowerCAmelCase							,   dropout_rate=lowerCAmelCase							,   feed_forward_proj=lowerCAmelCase							,   is_decoder=lowerCAmelCase							,   is_encoder_decoder=lowerCAmelCase							,   )

        _lowercase	       =nn.ModuleList()
        for lyr_num in range(lowerCAmelCase					):
              _lowercase	       =TaBlock(lowerCAmelCase					)
              self.encoders.append(lowerCAmelCase					)

        _lowercase	       =TaLayerNorm(lowerCAmelCase					)
        _lowercase	       =nn.Dropout(p=lowerCAmelCase					)




  def   A__     (				self							,   lowerCAmelCase							,   lowerCAmelCase					)  -> Dict:
        '''simple docstring'''






        _lowercase	       =self.token_embedder(lowerCAmelCase					)

        _lowercase	       =encoder_input_tokens.shape[1]
        _lowercase	       =torch.arange(lowerCAmelCase							,   device=encoder_input_tokens.device					)
        x += self.position_encoding(lowerCAmelCase					)

        _lowercase	       =self.dropout_pre(lowerCAmelCase					)

        # inverted the attention mask
        _lowercase	       =encoder_input_tokens.size()
        _lowercase	       =self.get_extended_attention_mask(lowerCAmelCase							,   lowerCAmelCase					)

        for lyr in self.encoders:
              _lowercase	       =lyr(lowerCAmelCase							,   lowerCAmelCase					)[0]
        _lowercase	       =self.layer_norm(lowerCAmelCase					)

        return self.dropout_post(lowerCAmelCase					), encoder_inputs_mask


