





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			:List[Any]	 =     logging.getLogger(__name__)






class 							_a(     __A     ):







		def 							lowercase							(     self			,    __snake_case			,    __snake_case			,    __snake_case=None			,    __snake_case=None					)   ->   Union[str, Any]:


							'''simple docstring'''

							if current_layer == 0:
												_snake_case							:       Optional[Any]												= self.embedding_hidden_mapping_in(__snake_case					)
							else:
												_snake_case							:       Optional[int]												= hidden_states[0]

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

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

							_snake_case							:       Tuple												= self.albert_layer_groups[group_idx](
							    __snake_case			,    __snake_case			,    head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]			,    )
							_snake_case							:       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 							_a(     __A     ):







		def __init__(     self			,    __snake_case					)   ->   Optional[int]:


							'''simple docstring'''

							super().__init__(__snake_case					)

							_snake_case							:       Any												= AlbertTransformerWithPabee(__snake_case					)

							self.init_weights()
							_snake_case							:       int												= 0
							_snake_case							:       Optional[int]												= 0
							_snake_case							:       List[str]												= 0

							_snake_case							:       List[Any]												= 0







		def 							lowercase							(     self			,    __snake_case					)   ->   str:


							'''simple docstring'''

							_snake_case							:       Optional[int]												= threshold







		def 							lowercase							(     self			,    __snake_case					)   ->   str:


							'''simple docstring'''

							_snake_case							:       Tuple												= patience







		def 							lowercase							(     self					)   ->   Optional[Any]:


							'''simple docstring'''

							_snake_case							:       List[Any]												= 0
							_snake_case							:       Dict												= 0







		def 							lowercase							(     self					)   ->   Dict:


							'''simple docstring'''

							_snake_case							:       List[str]												= self.inference_layers_num / self.inference_instances_num
							_snake_case							:       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(__snake_case					)







		@add_start_docstrings_to_model_forward(__snake_case					)
		def 							lowercase							(     self			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    __snake_case=False			,    )   ->   Dict:


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

							_snake_case							:       str												= input_ids.device if input_ids is not None else inputs_embeds.device

							if attention_mask is None:
												_snake_case							:       Tuple												= torch.ones(__snake_case			,    device=__snake_case					)
							if token_type_ids is None:
												_snake_case							:       int												= torch.zeros(__snake_case			,    dtype=torch.long			,    device=__snake_case					)

							_snake_case							:       Dict												= attention_mask.unsqueeze(1					).unsqueeze(2					)
							_snake_case							:       Union[str, Any]												= extended_attention_mask.to(dtype=self.dtype					)  # fp16 compatibility
							_snake_case							:       int												= (1.0 - extended_attention_mask) * -1_00_00.0
							_snake_case							:       Tuple												= self.get_head_mask(__snake_case			,    self.config.num_hidden_layers					)

							_snake_case							:       List[str]												= self.embeddings(
							    __snake_case			,    position_ids=__snake_case			,    token_type_ids=__snake_case			,    inputs_embeds=__snake_case					)
							_snake_case							:       str												= embedding_output

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

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

																	_snake_case							:       List[Any]												= self.pooler_activation(self.pooler(encoder_outputs[0][:, 0]					)					)
																	_snake_case							:       List[str]												= output_layers[i](__snake_case					)
																	if regression:
																						_snake_case							:       List[str]												= logits.detach()
																						if patient_result is not None:
																											_snake_case							:       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:
																											_snake_case							:       List[str]												= 0
																	else:
																						_snake_case							:       List[Any]												= logits.detach().argmax(dim=1					)
																						if patient_result is not None:
																											_snake_case							:       Optional[int]												= patient_result.detach().argmax(dim=1					)
																						if (patient_result is not None) and torch.all(labels.eq(__snake_case					)					):
																											patient_counter += 1
																						else:
																											_snake_case							:       List[str]												= 0

																	_snake_case							:       Dict												= logits
																	if patient_counter == self.patience:
																						break
												_snake_case							:       Union[str, 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\n    the pooled output) e.g. for GLUE tasks. '						, __A						, )
class 							_a(     __A     ):







		def __init__(     self			,    __snake_case					)   ->   int:


							'''simple docstring'''

							super().__init__(__snake_case					)
							_snake_case							:       str												= config.num_labels

							_snake_case							:       List[str]												= AlbertModelWithPabee(__snake_case					)
							_snake_case							:       int												= nn.Dropout(config.classifier_dropout_prob					)
							_snake_case							:       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(__snake_case					)
		def 							lowercase							(     self			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    __snake_case=None			,    )   ->   List[str]:


							'''simple docstring'''

							_snake_case							:       List[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			,    )

							_snake_case							:       Dict												= (logits[-1],)

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

							return outputs






def  A					(				UpperCAmelCase					):
					if not isinstance(UpperCAmelCase	,	UpperCAmelCase					):
										_snake_case							:       Optional[Any]												= F"""Input value of [number={number}] must be an integer"""
										raise TypeError(UpperCAmelCase					)
					if number < 1:
										raise ValueError("Input must be a positive integer"					)
					return number * (2 * number - 1)


if __name__ == "__main__":
							import doctest

							doctest.testmod()