




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


__lowerCamelCase								=						logging.getLogger(__name__)



class  _snake_case				(							A__       ):

				'''simple docstring'''




				def 				snake_case_ (							self      :  Union[str, Any]					,							snake_case      :  Tuple					,							snake_case      :  List[Any]					,							snake_case      :  Optional[int]=None					,							snake_case      :  Optional[int]=None     ):
								if current_layer == 0:
												UpperCAmelCase_					:Optional[Any]      			=      self.embedding_hidden_mapping_in(snake_case     )
								else:
												UpperCAmelCase_					:Union[str, Any]      			=      hidden_states[0]

								UpperCAmelCase_					:Dict      			=      int(self.config.num_hidden_layers / self.config.num_hidden_groups     )

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

								UpperCAmelCase_					:List[Any]      			=      self.albert_layer_groups[group_idx](
								    snake_case					,							snake_case					,							head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]					,							)
								UpperCAmelCase_					:str      			=      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  _snake_case				(							A__       ):

				'''simple docstring'''




				def __init__(							self      :  Optional[int]					,							snake_case      :  Dict     ):
								super().__init__(snake_case     )

								UpperCAmelCase_					:Dict      			=      AlbertTransformerWithPabee(snake_case     )

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

								UpperCAmelCase_					:str      			=      0





				def 				snake_case_ (							self      :  List[str]					,							snake_case      :  Any     ):
								UpperCAmelCase_					:List[Any]      			=      threshold





				def 				snake_case_ (							self      :  int					,							snake_case      :  str     ):
								UpperCAmelCase_					:int      			=      patience





				def 				snake_case_ (							self      :  Dict     ):
								UpperCAmelCase_					:Dict      			=      0
								UpperCAmelCase_					:List[str]      			=      0





				def 				snake_case_ (							self      :  Optional[int]     ):
								UpperCAmelCase_					:Union[str, Any]      			=      self.inference_layers_num / self.inference_instances_num
								UpperCAmelCase_					:str      			=      (
								    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(snake_case     )





				@add_start_docstrings_to_model_forward(snake_case     )
				def 				snake_case_ (							self      :  Optional[Any]					,							snake_case      :  str=None					,							snake_case      :  Dict=None					,							snake_case      :  Dict=None					,							snake_case      :  Tuple=None					,							snake_case      :  Any=None					,							snake_case      :  str=None					,							snake_case      :  Tuple=None					,							snake_case      :  List[str]=None					,							snake_case      :  List[Any]=False					,							):

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

								UpperCAmelCase_					:Tuple      			=      input_ids.device if input_ids is not None else inputs_embeds.device

								if attention_mask is None:
												UpperCAmelCase_					:Union[str, Any]      			=      torch.ones(snake_case					,							device=snake_case     )
								if token_type_ids is None:
												UpperCAmelCase_					:Tuple      			=      torch.zeros(snake_case					,							dtype=torch.long					,							device=snake_case     )

								UpperCAmelCase_					:Optional[Any]      			=      attention_mask.unsqueeze(1     ).unsqueeze(2     )
								UpperCAmelCase_					:Union[str, Any]      			=      extended_attention_mask.to(dtype=self.dtype     )  # fp16 compatibility
								UpperCAmelCase_					:Dict      			=      (1.0 - extended_attention_mask) * -10_000.0
								UpperCAmelCase_					:Any      			=      self.get_head_mask(snake_case					,							self.config.num_hidden_layers     )

								UpperCAmelCase_					:int      			=      self.embeddings(
								    snake_case					,							position_ids=snake_case					,							token_type_ids=snake_case					,							inputs_embeds=snake_case     )
								UpperCAmelCase_					:Any      			=      embedding_output

								if self.training:
												UpperCAmelCase_					:Optional[Any]      			=      []
												for i in range(self.config.num_hidden_layers     ):
																UpperCAmelCase_					:Union[str, Any]      			=      self.encoder.adaptive_forward(
																    snake_case					,							current_layer=snake_case					,							attention_mask=snake_case					,							head_mask=snake_case					,							)

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

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

																UpperCAmelCase_					:Any      			=      logits
																if patient_counter == self.patience:
																				break
												UpperCAmelCase_					:List[str]      			=      [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  _snake_case				(							A__       ):

				'''simple docstring'''




				def __init__(							self      :  int					,							snake_case      :  int     ):
								super().__init__(snake_case     )
								UpperCAmelCase_					:List[Any]      			=      config.num_labels

								UpperCAmelCase_					:Optional[int]      			=      AlbertModelWithPabee(snake_case     )
								UpperCAmelCase_					:List[str]      			=      nn.Dropout(config.classifier_dropout_prob     )
								UpperCAmelCase_					:Optional[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(snake_case     )
				def 				snake_case_ (							self      :  int					,							snake_case      :  str=None					,							snake_case      :  Dict=None					,							snake_case      :  str=None					,							snake_case      :  int=None					,							snake_case      :  Any=None					,							snake_case      :  Optional[Any]=None					,							snake_case      :  Dict=None					,							):
								UpperCAmelCase_					:Optional[Any]      			=      self.albert(
								    input_ids=snake_case					,							attention_mask=snake_case					,							token_type_ids=snake_case					,							position_ids=snake_case					,							head_mask=snake_case					,							inputs_embeds=snake_case					,							output_dropout=self.dropout					,							output_layers=self.classifiers					,							regression=self.num_labels == 1					,							)

								UpperCAmelCase_					:int      			=      (logits[-1],)

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

								return outputs











"""simple docstring"""

def 				a (						__snake_case     :				int  ):



				'''simple docstring'''





				if not isinstance(__snake_case,    __snake_case  ) or number < 0:
								raise ValueError('''Input must be a non-negative integer'''  )

				UpperCAmelCase_					:Optional[Any]      			=      number & ~(number - 1)
				UpperCAmelCase_					:List[str]      			=      0
				while intermediate:
								intermediate >>= 1
								index += 1
				return index - 1


if __name__ == "__main__":
						import doctest

						doctest.testmod(verbose=True)


