


'''simple docstring'''


from diffusers.utils.testing_utils import require_onnxruntime







@require_onnxruntime
class 							__magic_name__	:

     """simple docstring"""






     pass





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


_UpperCamelCase 							=  logging.getLogger(__name__)







class 							__magic_name__	(       lowerCAmelCase ):

     """simple docstring"""






     def lowerCAmelCase__  (   self			,     lowerCamelCase			,     lowerCamelCase			,     lowerCamelCase=None			,     lowerCamelCase=None		):


          '''simple docstring'''





          __A	:       List[Any]      =							self.layer[current_layer](lowerCamelCase			,     lowerCamelCase			,     head_mask[current_layer]		)

          __A	:       Tuple      =							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 							__magic_name__	(       lowerCAmelCase ):

     """simple docstring"""






     def __init__(   self			,     lowerCamelCase		):


          '''simple docstring'''





          super().__init__(lowerCamelCase		)

          __A	:       Any      =							BertEncoderWithPabee(lowerCamelCase		)

          self.init_weights()
          __A	:       Optional[Any]      =							0
          __A	:       Union[str, Any]      =							0
          __A	:       Dict      =							0

          __A	:       List[Any]      =							0


     def lowerCAmelCase__  (   self			,     lowerCamelCase		):


          '''simple docstring'''





          __A	:       Any      =							threshold


     def lowerCAmelCase__  (   self			,     lowerCamelCase		):


          '''simple docstring'''





          __A	:       List[Any]      =							patience


     def lowerCAmelCase__  (   self		):


          '''simple docstring'''





          __A	:       List[str]      =							0
          __A	:       int      =							0


     def lowerCAmelCase__  (   self		):


          '''simple docstring'''





          __A	:       Optional[Any]      =							self.inference_layers_num / self.inference_instances_num
          __A	:       int      =							(
              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 lowerCAmelCase__  (   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:
               __A	:       Optional[int]      =							input_ids.size()
          elif inputs_embeds is not None:
               __A	:       Any      =							inputs_embeds.size()[:-1]
          else:
               raise ValueError("You have to specify either input_ids or inputs_embeds"		)

          __A	:       List[str]      =							input_ids.device if input_ids is not None else inputs_embeds.device

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

          __A	:       Tuple      =							self.embeddings(
              input_ids=lowerCamelCase			,     position_ids=lowerCamelCase			,     token_type_ids=lowerCamelCase			,     inputs_embeds=lowerCamelCase		)
          __A	:       str      =							embedding_output

          if self.training:
               __A	:       Dict      =							[]
               for i in range(self.config.num_hidden_layers		):
                    __A	:       str      =							self.encoder.adaptive_forward(
                        lowerCamelCase			,     current_layer=lowerCamelCase			,     attention_mask=lowerCamelCase			,     head_mask=lowerCamelCase		)

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

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

                    __A	:       int      =							logits
                    if patient_counter == self.patience:
                         break
               __A	:       List[str]      =							[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 							__magic_name__	(       lowerCAmelCase ):

     """simple docstring"""






     def __init__(   self			,     lowerCamelCase		):


          '''simple docstring'''





          super().__init__(lowerCamelCase		)
          __A	:       Union[str, Any]      =							config.num_labels

          __A	:       Union[str, Any]      =							BertModelWithPabee(lowerCamelCase		)
          __A	:       Union[str, Any]      =							nn.Dropout(config.hidden_dropout_prob		)
          __A	:       int      =							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 lowerCAmelCase__  (   self			,     lowerCamelCase=None			,     lowerCamelCase=None			,     lowerCamelCase=None			,     lowerCamelCase=None			,     lowerCamelCase=None			,     lowerCamelCase=None			,     lowerCamelCase=None			,     ):


          '''simple docstring'''





          __A	:       Optional[Any]      =							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			,     )

          __A	:       Any      =							(logits[-1],)

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

          return outputs

