






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,
)


_a    :   Optional[Any]        =			logging.getLogger(__name__)







class 	_UpperCAmelCase  (  _A		):




      """simple docstring"""



      def 			snake_case_      (   self    ,	_lowerCAmelCase    ,	_lowerCAmelCase    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None		):



          '''simple docstring'''
          lowerCAmelCase__   :int      							=				self.layer[current_layer](_lowerCAmelCase    ,	_lowerCAmelCase    ,	head_mask[current_layer]		)

          lowerCAmelCase__   :Optional[Any]      							=				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.'''				,							_A				,							)
class 	_UpperCAmelCase  (  _A		):




      """simple docstring"""



      def __init__(   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          super().__init__(_lowerCAmelCase		)

          lowerCAmelCase__   :Optional[Any]      							=				BertEncoderWithPabee(_lowerCAmelCase		)

          self.init_weights()
          lowerCAmelCase__   :Dict      							=				0
          lowerCAmelCase__   :List[str]      							=				0
          lowerCAmelCase__   :Union[str, Any]      							=				0

          lowerCAmelCase__   :Dict      							=				0
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :Union[str, Any]      							=				threshold
      def 			snake_case_      (   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          lowerCAmelCase__   :List[Any]      							=				patience
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :Tuple      							=				0
          lowerCAmelCase__   :str      							=				0
      def 			snake_case_      (   self		):



          '''simple docstring'''
          lowerCAmelCase__   :Union[str, Any]      							=				self.inference_layers_num / self.inference_instances_num
          lowerCAmelCase__   :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(_lowerCAmelCase		)



      @add_start_docstrings_to_model_forward(_lowerCAmelCase		)
      def 			snake_case_      (   self    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=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:
              lowerCAmelCase__   :List[str]      							=				input_ids.size()
          elif inputs_embeds is not None:
              lowerCAmelCase__   :Any      							=				inputs_embeds.size()[:-1]
          else:
              raise ValueError("You have to specify either input_ids or inputs_embeds"		)

          lowerCAmelCase__   :Optional[int]      							=				input_ids.device if input_ids is not None else inputs_embeds.device

          if attention_mask is None:
              lowerCAmelCase__   :int      							=				torch.ones(_lowerCAmelCase    ,	device=_lowerCAmelCase		)
          if token_type_ids is None:
              lowerCAmelCase__   :List[Any]      							=				torch.zeros(_lowerCAmelCase    ,	dtype=torch.long    ,	device=_lowerCAmelCase		)

          # 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.
          lowerCAmelCase__   :torch.Tensor      							=				self.get_extended_attention_mask(_lowerCAmelCase    ,	_lowerCAmelCase    ,	_lowerCAmelCase		)

          # 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:
              lowerCAmelCase__   :Dict      							=				encoder_hidden_states.size()
              lowerCAmelCase__   :Optional[Any]      							=				(encoder_batch_size, encoder_sequence_length)
              if encoder_attention_mask is None:
                  lowerCAmelCase__   :List[str]      							=				torch.ones(_lowerCAmelCase    ,	device=_lowerCAmelCase		)
              lowerCAmelCase__   :List[str]      							=				self.invert_attention_mask(_lowerCAmelCase		)
          else:
              lowerCAmelCase__   :Tuple      							=				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]
          lowerCAmelCase__   :Tuple      							=				self.get_head_mask(_lowerCAmelCase    ,	self.config.num_hidden_layers		)

          lowerCAmelCase__   :Optional[int]      							=				self.embeddings(
              input_ids=_lowerCAmelCase    ,	position_ids=_lowerCAmelCase    ,	token_type_ids=_lowerCAmelCase    ,	inputs_embeds=_lowerCAmelCase		)
          lowerCAmelCase__   :List[Any]      							=				embedding_output

          if self.training:
              lowerCAmelCase__   :Any      							=				[]
              for i in range(self.config.num_hidden_layers		):
                  lowerCAmelCase__   :Tuple      							=				self.encoder.adaptive_forward(
                      _lowerCAmelCase    ,	current_layer=_lowerCAmelCase    ,	attention_mask=_lowerCAmelCase    ,	head_mask=_lowerCAmelCase		)

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

                  lowerCAmelCase__   :Optional[Any]      							=				self.pooler(_lowerCAmelCase		)
                  lowerCAmelCase__   :Optional[int]      							=				output_layers[i](_lowerCAmelCase		)
                  if regression:
                      lowerCAmelCase__   :Tuple      							=				logits.detach()
                      if patient_result is not None:
                          lowerCAmelCase__   :Optional[Any]      							=				patient_result.detach()
                      if (patient_result is not None) and torch.abs(patient_result - labels		) < self.regression_threshold:
                          patient_counter += 1
                      else:
                          lowerCAmelCase__   :Optional[Any]      							=				0
                  else:
                      lowerCAmelCase__   :Tuple      							=				logits.detach().argmax(dim=1		)
                      if patient_result is not None:
                          lowerCAmelCase__   :Any      							=				patient_result.detach().argmax(dim=1		)
                      if (patient_result is not None) and torch.all(labels.eq(_lowerCAmelCase		)		):
                          patient_counter += 1
                      else:
                          lowerCAmelCase__   :Tuple      							=				0

                  lowerCAmelCase__   :Union[str, Any]      							=				logits
                  if patient_counter == self.patience:
                      break
              lowerCAmelCase__   :Union[str, Any]      							=				[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
    the pooled output) e.g. for GLUE tasks. '''				,							_A				,							)
class 	_UpperCAmelCase  (  _A		):




      """simple docstring"""



      def __init__(   self    ,	_lowerCAmelCase		):



          '''simple docstring'''
          super().__init__(_lowerCAmelCase		)
          lowerCAmelCase__   :Union[str, Any]      							=				config.num_labels

          lowerCAmelCase__   :List[Any]      							=				BertModelWithPabee(_lowerCAmelCase		)
          lowerCAmelCase__   :Dict      							=				nn.Dropout(config.hidden_dropout_prob		)
          lowerCAmelCase__   :Dict      							=				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(_lowerCAmelCase		)
      def 			snake_case_      (   self    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	_lowerCAmelCase=None    ,	):



          '''simple docstring'''
          lowerCAmelCase__   :Dict      							=				self.bert(
              input_ids=_lowerCAmelCase    ,	attention_mask=_lowerCAmelCase    ,	token_type_ids=_lowerCAmelCase    ,	position_ids=_lowerCAmelCase    ,	head_mask=_lowerCAmelCase    ,	inputs_embeds=_lowerCAmelCase    ,	output_dropout=self.dropout    ,	output_layers=self.classifiers    ,	regression=self.num_labels == 1    ,	)

          lowerCAmelCase__   :Tuple      							=				(logits[-1],)

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

          return outputs









def 				snake_case__							(							UpperCAmelCase							:  float       ):

    if edge <= 0 or not isinstance(UpperCAmelCase    ,			UpperCAmelCase       ):
        raise ValueError("Length must be a positive."       )
    return 3 * ((2_5 + 1_0 * (5 ** (1 / 2))) ** (1 / 2)) * (edge**2)


def 				snake_case__							(							UpperCAmelCase							:  float       ):

    if edge <= 0 or not isinstance(UpperCAmelCase    ,			UpperCAmelCase       ):
        raise ValueError("Length must be a positive."       )
    return ((1_5 + (7 * (5 ** (1 / 2)))) / 4) * (edge**3)


if __name__ == "__main__":
 import doctest

 doctest.testmod()



