


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


snake_case__  				=			logging.getLogger(__name__)



class      snake_case__	(   _A	):
				def 			UpperCamelCase_    (	self  ,   _UpperCAmelCase  ,   _UpperCAmelCase  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None   ):




										"""simple docstring"""




										if current_layer == 0:
																__magic_name__					:				int				=   self.embedding_hidden_mapping_in(_UpperCAmelCase   )
										else:
																__magic_name__					:				List[str]				=   hidden_states[0]

										__magic_name__					:				Optional[Any]				=   int(self.config.num_hidden_layers / self.config.num_hidden_groups   )

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

										__magic_name__					:				List[Any]				=   self.albert_layer_groups[group_idx](
										    _UpperCAmelCase  ,   _UpperCAmelCase  ,   head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]  ,   )
										__magic_name__					:				Union[str, Any]				=   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	):
				def __init__(	self  ,   _UpperCAmelCase   ):




										"""simple docstring"""




										super().__init__(_UpperCAmelCase   )

										__magic_name__					:				List[Any]				=   AlbertTransformerWithPabee(_UpperCAmelCase   )

										self.init_weights()
										__magic_name__					:				Optional[Any]				=   0
										__magic_name__					:				Optional[int]				=   0
										__magic_name__					:				Union[str, Any]				=   0

										__magic_name__					:				Dict				=   0
				def 			UpperCamelCase_    (	self  ,   _UpperCAmelCase   ):




										"""simple docstring"""




										__magic_name__					:				Dict				=   threshold
				def 			UpperCamelCase_    (	self  ,   _UpperCAmelCase   ):




										"""simple docstring"""




										__magic_name__					:				Optional[Any]				=   patience
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				str				=   0
										__magic_name__					:				int				=   0
				def 			UpperCamelCase_    (	self   ):




										"""simple docstring"""




										__magic_name__					:				List[Any]				=   self.inference_layers_num / self.inference_instances_num
										__magic_name__					:				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(_UpperCAmelCase   )





				@add_start_docstrings_to_model_forward(_UpperCAmelCase   )
				def 			UpperCamelCase_    (	self  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=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:
																__magic_name__					:				List[str]				=   input_ids.size()
										elif inputs_embeds is not None:
																__magic_name__					:				str				=   inputs_embeds.size()[:-1]
										else:
																raise ValueError('You have to specify either input_ids or inputs_embeds'   )

										__magic_name__					:				Union[str, Any]				=   input_ids.device if input_ids is not None else inputs_embeds.device

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

										__magic_name__					:				Any				=   attention_mask.unsqueeze(1   ).unsqueeze(2   )
										__magic_name__					:				Optional[int]				=   extended_attention_mask.to(dtype=self.dtype   )  # fp16 compatibility
										__magic_name__					:				Optional[Any]				=   (1.0 - extended_attention_mask) * -10000.0
										__magic_name__					:				Optional[Any]				=   self.get_head_mask(_UpperCAmelCase  ,   self.config.num_hidden_layers   )

										__magic_name__					:				int				=   self.embeddings(
										    _UpperCAmelCase  ,   position_ids=_UpperCAmelCase  ,   token_type_ids=_UpperCAmelCase  ,   inputs_embeds=_UpperCAmelCase   )
										__magic_name__					:				Union[str, Any]				=   embedding_output

										if self.training:
																__magic_name__					:				List[Any]				=   []
																for i in range(self.config.num_hidden_layers   ):
																						__magic_name__					:				Optional[int]				=   self.encoder.adaptive_forward(
																						    _UpperCAmelCase  ,   current_layer=_UpperCAmelCase  ,   attention_mask=_UpperCAmelCase  ,   head_mask=_UpperCAmelCase  ,   )

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

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

																						__magic_name__					:				Union[str, Any]				=   logits
																						if patient_counter == self.patience:
																												break
																__magic_name__					:				Optional[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      snake_case__	(   _A	):
				def __init__(	self  ,   _UpperCAmelCase   ):




										"""simple docstring"""




										super().__init__(_UpperCAmelCase   )
										__magic_name__					:				str				=   config.num_labels

										__magic_name__					:				Dict				=   AlbertModelWithPabee(_UpperCAmelCase   )
										__magic_name__					:				Any				=   nn.Dropout(config.classifier_dropout_prob   )
										__magic_name__					:				Dict				=   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 			UpperCamelCase_    (	self  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   _UpperCAmelCase=None  ,   ):




										"""simple docstring"""




										__magic_name__					:				int				=   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  ,   )

										__magic_name__					:				List[str]				=   (logits[-1],)

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

										return outputs






import math
def 			__SCREAMING_SNAKE_CASE							(					lowercase__	:int					)							->	list:
						if n <= 0:
												raise ValueError('Only positive integers have prime factors'					)
						__magic_name__					:				Union[str, Any]				=   []
						while n % 2 == 0:
												pf.append(2					)
												__magic_name__					:				Union[str, Any]				=   int(n / 2					)
						for i in range(3   ,    int(math.sqrt(lowercase__					)					) + 1   ,    2					):
												while n % i == 0:
																		pf.append(lowercase__					)
																		__magic_name__					:				List[str]				=   int(n / i					)
						if n > 2:
												pf.append(lowercase__					)
						return pf
def 			__SCREAMING_SNAKE_CASE							(					lowercase__	:int					)							->	int:
						if n <= 0:
												raise ValueError('Only positive numbers are accepted'					)
						__magic_name__					:				Union[str, Any]				=   1
						__magic_name__					:				Tuple				=   1
						while n % 2 == 0:
												temp += 1
												__magic_name__					:				Any				=   int(n / 2					)
						div *= temp
						for i in range(3   ,    int(math.sqrt(lowercase__					)					) + 1   ,    2					):
												__magic_name__					:				List[Any]				=   1
												while n % i == 0:
																		temp += 1
																		__magic_name__					:				Any				=   int(n / i					)
												div *= temp
						if n > 1:
												div *= 2
						return div
def 			__SCREAMING_SNAKE_CASE							(					lowercase__	:int					)							->	int:
						if n <= 0:
												raise ValueError('Only positive numbers are accepted'					)
						__magic_name__					:				int				=   1
						__magic_name__					:				List[str]				=   1
						while n % 2 == 0:
												temp += 1
												__magic_name__					:				List[Any]				=   int(n / 2					)
						if temp > 1:
												s *= (2**temp - 1) / (2 - 1)
						for i in range(3   ,    int(math.sqrt(lowercase__					)					) + 1   ,    2					):
												__magic_name__					:				int				=   1
												while n % i == 0:
																		temp += 1
																		__magic_name__					:				Optional[Any]				=   int(n / i					)
												if temp > 1:
																		s *= (i**temp - 1) / (i - 1)
						return int(lowercase__					)
def 			__SCREAMING_SNAKE_CASE							(					lowercase__	:int					)							->	int:
						__magic_name__					:				Dict				=   n
						for x in set(prime_factors(lowercase__					)					):
												s *= (x - 1) / x
						return int(lowercase__					)


if __name__ == "__main__":
			import doctest

			doctest.testmod()


