


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    lowerCamelCase__     ( lowerCAmelCase):





     def   __A      (self	,  UpperCAmelCase	,  UpperCAmelCase	,  UpperCAmelCase=None	,  UpperCAmelCase=None )  ->     Optional[int]:
      if current_layer == 0:
       _lowercase							   =self.embedding_hidden_mapping_in(UpperCAmelCase )
      else:
       _lowercase							   =hidden_states[0]

      _lowercase							   =int(self.config.num_hidden_layers / self.config.num_hidden_groups )

      # Index of the hidden group
      _lowercase							   =int(current_layer / (self.config.num_hidden_layers / self.config.num_hidden_groups) )

      _lowercase							   =self.albert_layer_groups[group_idx](
          UpperCAmelCase	,  UpperCAmelCase	,  head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]	,  )
      _lowercase							   =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.'''     ,							lowerCAmelCase     ,							)
class    lowerCamelCase__     ( lowerCAmelCase):





     def __init__(self	,  UpperCAmelCase )  ->     Union[str, Any]:
      super().__init__(UpperCAmelCase )

      _lowercase							   =AlbertTransformerWithPabee(UpperCAmelCase )

      self.init_weights()
      _lowercase							   =0
      _lowercase							   =0
      _lowercase							   =0

      _lowercase							   =0





     def   __A      (self	,  UpperCAmelCase )  ->     Optional[Any]:
      _lowercase							   =threshold





     def   __A      (self	,  UpperCAmelCase )  ->     Dict:
      _lowercase							   =patience





     def   __A      (self )  ->     Optional[Any]:
      _lowercase							   =0
      _lowercase							   =0





     def   __A      (self )  ->     Any:
      _lowercase							   =self.inference_layers_num / self.inference_instances_num
      _lowercase							   =(
          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   __A      (self	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=False	,  )  ->     List[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:
       _lowercase							   =input_ids.size()
      elif inputs_embeds is not None:
       _lowercase							   =inputs_embeds.size()[:-1]
      else:
       raise ValueError('''You have to specify either input_ids or inputs_embeds''' )

      _lowercase							   =input_ids.device if input_ids is not None else inputs_embeds.device

      if attention_mask is None:
       _lowercase							   =torch.ones(UpperCAmelCase	,  device=UpperCAmelCase )
      if token_type_ids is None:
       _lowercase							   =torch.zeros(UpperCAmelCase	,  dtype=torch.long	,  device=UpperCAmelCase )

      _lowercase							   =attention_mask.unsqueeze(1 ).unsqueeze(2 )
      _lowercase							   =extended_attention_mask.to(dtype=self.dtype )  # fp16 compatibility
      _lowercase							   =(1.0 - extended_attention_mask) * -1_0000.0
      _lowercase							   =self.get_head_mask(UpperCAmelCase	,  self.config.num_hidden_layers )

      _lowercase							   =self.embeddings(
          UpperCAmelCase	,  position_ids=UpperCAmelCase	,  token_type_ids=UpperCAmelCase	,  inputs_embeds=UpperCAmelCase )
      _lowercase							   =embedding_output

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

        _lowercase							   =self.pooler_activation(self.pooler(encoder_outputs[0][:, 0] ) )
        _lowercase							   =output_layers[i](output_dropout(UpperCAmelCase ) )
        res.append(UpperCAmelCase )
      elif self.patience == 0:  # Use all layers for inference
       _lowercase							   =self.encoder(UpperCAmelCase	,  UpperCAmelCase	,  head_mask=UpperCAmelCase )
       _lowercase							   =self.pooler_activation(self.pooler(encoder_outputs[0][:, 0] ) )
       _lowercase							   =[output_layers[self.config.num_hidden_layers - 1](UpperCAmelCase )]
      else:
       _lowercase							   =0
       _lowercase							   =None
       _lowercase							   =0
       for i in range(self.config.num_hidden_layers ):
        calculated_layer_num += 1
        _lowercase							   =self.encoder.adaptive_forward(
            UpperCAmelCase	,  current_layer=UpperCAmelCase	,  attention_mask=UpperCAmelCase	,  head_mask=UpperCAmelCase	,  )

        _lowercase							   =self.pooler_activation(self.pooler(encoder_outputs[0][:, 0] ) )
        _lowercase							   =output_layers[i](UpperCAmelCase )
        if regression:
         _lowercase							   =logits.detach()
         if patient_result is not None:
          _lowercase							   =patient_result.detach()
         if (patient_result is not None) and torch.abs(patient_result - labels ) < self.regression_threshold:
          patient_counter += 1
         else:
          _lowercase							   =0
        else:
         _lowercase							   =logits.detach().argmax(dim=1 )
         if patient_result is not None:
          _lowercase							   =patient_result.detach().argmax(dim=1 )
         if (patient_result is not None) and torch.all(labels.eq(UpperCAmelCase ) ):
          patient_counter += 1
         else:
          _lowercase							   =0

        _lowercase							   =logits
        if patient_counter == self.patience:
         break
       _lowercase							   =[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. '''     ,							lowerCAmelCase     ,							)
class    lowerCamelCase__     ( lowerCAmelCase):





     def __init__(self	,  UpperCAmelCase )  ->     List[Any]:
      super().__init__(UpperCAmelCase )
      _lowercase							   =config.num_labels

      _lowercase							   =AlbertModelWithPabee(UpperCAmelCase )
      _lowercase							   =nn.Dropout(config.classifier_dropout_prob )
      _lowercase							   =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   __A      (self	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  UpperCAmelCase=None	,  )  ->     Union[str, Any]:
      _lowercase							   =self.albert(
          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							   =(logits[-1],)

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

      return outputs





import unittest

import numpy as np

from transformers import is_tf_available
from transformers.testing_utils import require_tf


if is_tf_available():
  import tensorflow as tf

  from transformers.activations_tf import get_tf_activation







@require_tf
class    lowerCamelCase__     ( unittest.TestCase):





     def   __A      (self )  ->     Optional[Any]:
      _lowercase							   =tf.constant([-1_0_0, -1.0, -0.1, 0, 0.1, 1.0, 100.0] )
      _lowercase							   =get_tf_activation('''gelu''' )
      _lowercase							   =get_tf_activation('''gelu_10''' )

      _lowercase							   =gelu(UpperCAmelCase )
      _lowercase							   =geluaa(UpperCAmelCase )

      _lowercase							   =tf.where(y_gelu_aa < 10.0	,  1.0	,  0.0 )

      self.assertEqual(tf.math.reduce_max(UpperCAmelCase ).numpy().item()	,  10.0 )
      self.assertTrue(np.allclose(y_gelu * clipped_mask	,  y_gelu_aa * clipped_mask ) )





     def   __A      (self )  ->     Optional[Any]:
      get_tf_activation('''gelu''' )
      get_tf_activation('''gelu_10''' )
      get_tf_activation('''gelu_fast''' )
      get_tf_activation('''gelu_new''' )
      get_tf_activation('''glu''' )
      get_tf_activation('''mish''' )
      get_tf_activation('''quick_gelu''' )
      get_tf_activation('''relu''' )
      get_tf_activation('''sigmoid''' )
      get_tf_activation('''silu''' )
      get_tf_activation('''swish''' )
      get_tf_activation('''tanh''' )
      with self.assertRaises(UpperCAmelCase ):
       get_tf_activation('''bogus''' )
      with self.assertRaises(UpperCAmelCase ):
       get_tf_activation(UpperCAmelCase )

