



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__			(		_a ):


						def 	_lowerCamelCase      (			self    :	Union[str, Any]   ,							_a    :	int   ,							_a    :	Dict   ,							_a    :	str=None   ,							_a    :	Tuple=None      ):
													if current_layer == 0:
																				a__:							int									=self.embedding_hidden_mapping_in(snake_case_      )
													else:
																				a__:							List[Any]									=hidden_states[0]

													a__:							Any									=int(self.config.num_hidden_layers / self.config.num_hidden_groups      )

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

													a__:							int									=self.albert_layer_groups[group_idx](
													    snake_case_   ,							snake_case_   ,							head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]   ,							)
													a__:							Dict									=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 						lowerCamelCase__			(		_a ):


						def __init__(			self    :	Optional[Any]   ,							_a    :	str      ):
													super().__init__(snake_case_      )

													a__:							List[str]									=AlbertTransformerWithPabee(snake_case_      )

													self.init_weights()
													a__:							Tuple									=0
													a__:							str									=0
													a__:							List[Any]									=0

													a__:							Dict									=0


						def 	_lowerCamelCase      (			self    :	List[Any]   ,							_a    :	List[Any]      ):
													a__:							Optional[Any]									=threshold


						def 	_lowerCamelCase      (			self    :	Union[str, Any]   ,							_a    :	List[str]      ):
													a__:							Dict									=patience


						def 	_lowerCamelCase      (			self    :	Tuple      ):
													a__:							str									=0
													a__:							Union[str, Any]									=0


						def 	_lowerCamelCase      (			self    :	Dict      ):
													a__:							Dict									=self.inference_layers_num / self.inference_instances_num
													a__:							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(snake_case_      )







						@add_start_docstrings_to_model_forward(snake_case_      )
						def 	_lowerCamelCase      (			self    :	Any   ,							_a    :	Tuple=None   ,							_a    :	Dict=None   ,							_a    :	List[str]=None   ,							_a    :	str=None   ,							_a    :	Union[str, Any]=None   ,							_a    :	str=None   ,							_a    :	Dict=None   ,							_a    :	List[Any]=None   ,							_a    :	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:
																				a__:							Optional[int]									=input_ids.size()
													elif inputs_embeds is not None:
																				a__:							Optional[Any]									=inputs_embeds.size()[:-1]
													else:
																				raise ValueError("You have to specify either input_ids or inputs_embeds"      )

													a__:							Any									=input_ids.device if input_ids is not None else inputs_embeds.device

													if attention_mask is None:
																				a__:							Tuple									=torch.ones(snake_case_   ,							device=snake_case_      )
													if token_type_ids is None:
																				a__:							Tuple									=torch.zeros(snake_case_   ,							dtype=torch.long   ,							device=snake_case_      )

													a__:							int									=attention_mask.unsqueeze(1      ).unsqueeze(2      )
													a__:							List[str]									=extended_attention_mask.to(dtype=self.dtype      )  # fp16 compatibility
													a__:							str									=(1.0 - extended_attention_mask) * -1_0_0_0_0.0
													a__:							List[Any]									=self.get_head_mask(snake_case_   ,							self.config.num_hidden_layers      )

													a__:							Optional[int]									=self.embeddings(
													    snake_case_   ,							position_ids=snake_case_   ,							token_type_ids=snake_case_   ,							inputs_embeds=snake_case_      )
													a__:							int									=embedding_output

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

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

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

																											a__:							int									=logits
																											if patient_counter == self.patience:
																																		break
																				a__:							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 						lowerCamelCase__			(		_a ):


						def __init__(			self    :	Any   ,							_a    :	Union[str, Any]      ):
													super().__init__(snake_case_      )
													a__:							Tuple									=config.num_labels

													a__:							Tuple									=AlbertModelWithPabee(snake_case_      )
													a__:							Dict									=nn.Dropout(config.classifier_dropout_prob      )
													a__:							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 	_lowerCamelCase      (			self    :	Union[str, Any]   ,							_a    :	Optional[int]=None   ,							_a    :	Optional[int]=None   ,							_a    :	List[Any]=None   ,							_a    :	Any=None   ,							_a    :	Any=None   ,							_a    :	Union[str, Any]=None   ,							_a    :	int=None   ,							):
													a__:							str									=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   ,							)

													a__:							Tuple									=(logits[-1],)

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

													return outputs













"""simple docstring"""





import torch

from transformers import PreTrainedModel

from .custom_configuration import CustomConfig, NoSuperInitConfig
class  __lowerCamelCase     (	_a	):
   a   :					List[Any]			    =CustomConfig

   def __init__(       self			,					snake_case_   )		->							str:
       super().__init__(snake_case_   )
       UpperCamelCase__								=     torch.nn.Linear(config.hidden_size			,					config.hidden_size   )

   def     SCREAMING_SNAKE_CASE__			(       self			,					snake_case_   )		->							Optional[int]:
       return self.linear(snake_case_   )







   def     SCREAMING_SNAKE_CASE__			(       self			,					snake_case_   )		->							Union[str, Any]:
       pass







class  __lowerCamelCase     (	_a	):
   a   :					Optional[int]			    =NoSuperInitConfig

   def __init__(       self			,					snake_case_   )		->							Optional[int]:
       super().__init__(snake_case_   )
       UpperCamelCase__								=     torch.nn.Linear(config.attribute			,					config.attribute   )

   def     SCREAMING_SNAKE_CASE__			(       self			,					snake_case_   )		->							int:
       return self.linear(snake_case_   )







   def     SCREAMING_SNAKE_CASE__			(       self			,					snake_case_   )		->							Union[str, Any]:
       pass




