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


lowercase_     :   Any 				=				logging.getLogger(__name__)






class 		_lowerCamelCase		(   UpperCamelCase_    ):
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase=None , lowerCAmelCase=None   )       ->					Tuple:
     SCREAMING_SNAKE_CASE__:   List[str]=      self.layer[current_layer](lowerCAmelCase , lowerCAmelCase , head_mask[current_layer]   )

     SCREAMING_SNAKE_CASE__:   List[str]=      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."    ,     UpperCamelCase_    ,     )
class 		_lowerCamelCase		(   UpperCamelCase_    ):
    def __init__(  self , lowerCAmelCase   )       ->					Any:
     super().__init__(lowerCAmelCase   )

     SCREAMING_SNAKE_CASE__:   Optional[int]=      BertEncoderWithPabee(lowerCAmelCase   )

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

     SCREAMING_SNAKE_CASE__:   Optional[Any]=      0
    def 						UpperCamelCase_				(  self , lowerCAmelCase   )       ->					Tuple:
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      threshold
    def 						UpperCamelCase_				(  self , lowerCAmelCase   )       ->					Dict:
     SCREAMING_SNAKE_CASE__:   Optional[Any]=      patience
    def 						UpperCamelCase_				(  self   )       ->					Any:
     SCREAMING_SNAKE_CASE__:   str=      0
     SCREAMING_SNAKE_CASE__:   str=      0
    def 						UpperCamelCase_				(  self   )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   Optional[int]=      self.inference_layers_num / self.inference_instances_num
     SCREAMING_SNAKE_CASE__:   Tuple=      (
         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 						UpperCamelCase_				(  self , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=False , )       ->					Union[str, Any]:

     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:
      SCREAMING_SNAKE_CASE__:   List[str]=      input_ids.size()
     elif inputs_embeds is not None:
      SCREAMING_SNAKE_CASE__:   List[Any]=      inputs_embeds.size()[:-1]
     else:
      raise ValueError('''You have to specify either input_ids or inputs_embeds'''   )

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

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

     SCREAMING_SNAKE_CASE__:   Tuple=      self.embeddings(
         input_ids=lowerCAmelCase , position_ids=lowerCAmelCase , token_type_ids=lowerCAmelCase , inputs_embeds=lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   List[str]=      embedding_output

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

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

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

       SCREAMING_SNAKE_CASE__:   Any=      logits
       if patient_counter == self.patience:
        break
      SCREAMING_SNAKE_CASE__:   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. "    ,     UpperCamelCase_    ,     )
class 		_lowerCamelCase		(   UpperCamelCase_    ):
    def __init__(  self , lowerCAmelCase   )       ->					List[str]:
     super().__init__(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Dict=      config.num_labels

     SCREAMING_SNAKE_CASE__:   str=      BertModelWithPabee(lowerCAmelCase   )
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      nn.Dropout(config.hidden_dropout_prob   )
     SCREAMING_SNAKE_CASE__:   Any=      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 						UpperCamelCase_				(  self , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , lowerCAmelCase=None , )       ->					Dict:
     SCREAMING_SNAKE_CASE__:   int=      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 , )

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

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

     return outputs


def 			A__      (	snake_case_ :		str      ):
 if not head:
  return True
 # split the list to two parts
 SCREAMING_SNAKE_CASE__,	SCREAMING_SNAKE_CASE__:   Optional[int]=      head.next, head
 while fast and fast.next:
  SCREAMING_SNAKE_CASE__:   Dict=      fast.next.next
  SCREAMING_SNAKE_CASE__:   Union[str, Any]=      slow.next
 SCREAMING_SNAKE_CASE__:   Union[str, Any]=      slow.next
 SCREAMING_SNAKE_CASE__:   Union[str, Any]=      None  # Don't forget here! But forget still works!
 # reverse the second part
 SCREAMING_SNAKE_CASE__:   Optional[int]=      None
 while second:
  SCREAMING_SNAKE_CASE__:   Any=      second.next
  SCREAMING_SNAKE_CASE__:   int=      node
  SCREAMING_SNAKE_CASE__:   Optional[Any]=      second
  SCREAMING_SNAKE_CASE__:   Any=      nxt
 # compare two parts
 # second part has the same or one less node
 while node:
  if node.val != head.val:
   return False
  SCREAMING_SNAKE_CASE__:   Tuple=      node.next
  SCREAMING_SNAKE_CASE__:   Optional[int]=      head.next
 return True







def 			A__      (	snake_case_ :		Optional[Any]      ):
 if not head or not head.next:
  return True

 # 1. Get the midpoint (slow)
 SCREAMING_SNAKE_CASE__:   List[Any]=      head
 while fast and fast.next:
  SCREAMING_SNAKE_CASE__,	SCREAMING_SNAKE_CASE__:   Any=      fast.next.next, slow.next

 # 2. Push the second half into the stack
 SCREAMING_SNAKE_CASE__:   Optional[Any]=      [slow.val]
 while slow.next:
  SCREAMING_SNAKE_CASE__:   Optional[int]=      slow.next
  stack.append(slow.val      )

 # 3. Comparison
 while stack:
  if stack.pop() != cur.val:
   return False
  SCREAMING_SNAKE_CASE__:   Tuple=      cur.next

 return True







def 			A__      (	snake_case_ :		Any      ):
 if not head or not head.next:
  return True
 SCREAMING_SNAKE_CASE__:   Optional[int]=      {}
 SCREAMING_SNAKE_CASE__:   Union[str, Any]=      0
 while head:
  if head.val in d:
   d[head.val].append(snake_case_      )
  else:
   SCREAMING_SNAKE_CASE__:   Optional[int]=      [pos]
  SCREAMING_SNAKE_CASE__:   Dict=      head.next
  pos += 1
 SCREAMING_SNAKE_CASE__:   Dict=      pos - 1
 SCREAMING_SNAKE_CASE__:   str=      0
 for v in d.values():
  if len(snake_case_      ) % 2 != 0:
   middle += 1
  else:
   SCREAMING_SNAKE_CASE__:   List[Any]=      0
   for i in range(0       ,				len(snake_case_      )      ):
    if v[i] + v[len(snake_case_      ) - 1 - step] != checksum:
     return False
    step += 1
  if middle > 1:
   return False
 return True

