

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.albert.modeling_albert import (
    ALBERT_INPUTS_DOCSTRING,
    ALBERT_START_DOCSTRING,
    AlbertModel,
    AlbertPreTrainedModel,
    AlbertTransformer,
)


__UpperCAmelCase     					=  logging.getLogger(__name__)





class A__     (			A			):



 """simple docstring"""







 def  __magic_name__							(		self			:						Union[str, Any]  ,    A_			:						Optional[Any]  ,    A_			:						List[str]  ,    A_			:						List[Any]=None  ,    A_			:						Optional[Any]=None							):






      '''simple docstring'''




      if current_layer == 0:
           _lowerCAmelCase :       Dict	   =     self.embedding_hidden_mapping_in(A_							)
      else:
           _lowerCAmelCase :       Optional[int]	   =     hidden_states[0]

      _lowerCAmelCase :       Tuple	   =     int(self.config.num_hidden_layers / self.config.num_hidden_groups							)

      # Index of the hidden group
      _lowerCAmelCase :       List[Any]	   =     int(current_layer / (self.config.num_hidden_layers / self.config.num_hidden_groups)							)

      _lowerCAmelCase :       Tuple	   =     self.albert_layer_groups[group_idx](
          A_  ,    A_  ,    head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]  ,    )
      _lowerCAmelCase :       Optional[int]	   =     layer_group_output[0]

      return (hidden_states,)





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



 """simple docstring"""







 def __init__(		self			:						Optional[Any]  ,    A_			:						int							):






      '''simple docstring'''




      super().__init__(A_							)

      _lowerCAmelCase :       List[Any]	   =     AlbertTransformerWithPabee(A_							)

      self.init_weights()
      _lowerCAmelCase :       Tuple	   =     0
      _lowerCAmelCase :       Any	   =     0
      _lowerCAmelCase :       Union[str, Any]	   =     0

      _lowerCAmelCase :       Union[str, Any]	   =     0

 def  __magic_name__							(		self			:						str  ,    A_			:						int							):






      '''simple docstring'''




      _lowerCAmelCase :       List[Any]	   =     threshold

 def  __magic_name__							(		self			:						Any  ,    A_			:						Optional[int]							):






      '''simple docstring'''




      _lowerCAmelCase :       Union[str, Any]	   =     patience

 def  __magic_name__							(		self			:						List[Any]							):






      '''simple docstring'''




      _lowerCAmelCase :       str	   =     0
      _lowerCAmelCase :       List[str]	   =     0

 def  __magic_name__							(		self			:						Optional[Any]							):






      '''simple docstring'''




      _lowerCAmelCase :       Tuple	   =     self.inference_layers_num / self.inference_instances_num
      _lowerCAmelCase :       Optional[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(A_							)


 @add_start_docstrings_to_model_forward(A_							)
 def  __magic_name__							(		self			:						Tuple  ,    A_			:						int=None  ,    A_			:						Optional[Any]=None  ,    A_			:						List[str]=None  ,    A_			:						Any=None  ,    A_			:						Tuple=None  ,    A_			:						Dict=None  ,    A_			:						Union[str, Any]=None  ,    A_			:						List[str]=None  ,    A_			:						List[str]=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 :       Optional[int]	   =     input_ids.size()
      elif inputs_embeds is not None:
           _lowerCAmelCase :       str	   =     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 :       Any	   =     torch.ones(A_  ,    device=A_							)
      if token_type_ids is None:
           _lowerCAmelCase :       int	   =     torch.zeros(A_  ,    dtype=torch.long  ,    device=A_							)

      _lowerCAmelCase :       Union[str, Any]	   =     attention_mask.unsqueeze(1							).unsqueeze(2							)
      _lowerCAmelCase :       Any	   =     extended_attention_mask.to(dtype=self.dtype							)  # fp16 compatibility
      _lowerCAmelCase :       str	   =     (1.0 - extended_attention_mask) * -10000.0
      _lowerCAmelCase :       str	   =     self.get_head_mask(A_  ,    self.config.num_hidden_layers							)

      _lowerCAmelCase :       str	   =     self.embeddings(
          A_  ,    position_ids=A_  ,    token_type_ids=A_  ,    inputs_embeds=A_							)
      _lowerCAmelCase :       Optional[int]	   =     embedding_output

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

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

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

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

      return res





@add_start_docstrings(
    '''Albert 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 A__     (			A			):



 """simple docstring"""







 def __init__(		self			:						Tuple  ,    A_			:						Dict							):






      '''simple docstring'''




      super().__init__(A_							)
      _lowerCAmelCase :       str	   =     config.num_labels

      _lowerCAmelCase :       str	   =     AlbertModelWithPabee(A_							)
      _lowerCAmelCase :       Optional[Any]	   =     nn.Dropout(config.classifier_dropout_prob							)
      _lowerCAmelCase :       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(A_							)
 def  __magic_name__							(		self			:						Tuple  ,    A_			:						List[str]=None  ,    A_			:						Optional[Any]=None  ,    A_			:						int=None  ,    A_			:						Any=None  ,    A_			:						str=None  ,    A_			:						Any=None  ,    A_			:						Optional[int]=None  ,    ):






      '''simple docstring'''




      _lowerCAmelCase :       Any	   =     self.albert(
          input_ids=A_  ,    attention_mask=A_  ,    token_type_ids=A_  ,    position_ids=A_  ,    head_mask=A_  ,    inputs_embeds=A_  ,    output_dropout=self.dropout  ,    output_layers=self.classifiers  ,    regression=self.num_labels == 1  ,    )

      _lowerCAmelCase :       List[Any]	   =     (logits[-1],)

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

      return outputs









def 						_snake_case		(     SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE					)							->   Optional[int]:

     """simple docstring"""







     if power < 0:
          return -1
     base %= mod
     _lowerCAmelCase :       Tuple	   =     1

     while power > 0:
          if power & 1:
               _lowerCAmelCase :       Optional[Any]	   =     (result * base) % mod
          _lowerCAmelCase :       int	   =     power >> 1
          _lowerCAmelCase :       Optional[int]	   =     (base * base) % mod

     return result







def 						_snake_case		(     )							->   Optional[int]:

     """simple docstring"""







     print(modular_exponential(3		,				200		,				13					)					)


if __name__ == "__main__":
  import doctest

  doctest.testmod()

  main()






