



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


						def      _lowerCamelCase	(       self  :     Dict					,				_a  :     int					,				_a  :     Optional[int]					,				_a  :     Optional[int]=None					,				_a  :     Union[str, Any]=None	):




													if current_layer == 0:
																				a__:				Dict 							=self.embedding_hidden_mapping_in(lowerCAmelCase__	)
													else:
																				a__:				Optional[Any] 							=hidden_states[0]

													a__:				List[str] 							=int(self.config.num_hidden_layers / self.config.num_hidden_groups	)

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

													a__:				List[Any] 							=self.albert_layer_groups[group_idx](
													    lowerCAmelCase__					,				lowerCAmelCase__					,				head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]					,				)
													a__:				List[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.''' ,    __SCREAMING_SNAKE_CASE ,    )
class 		lowerCamelCase__       (						__SCREAMING_SNAKE_CASE      ):


						def __init__(       self  :     Optional[int]					,				_a  :     str	):
													super().__init__(lowerCAmelCase__	)

													a__:				Dict 							=AlbertTransformerWithPabee(lowerCAmelCase__	)

													self.init_weights()
													a__:				str 							=0
													a__:				Optional[int] 							=0
													a__:				Optional[Any] 							=0

													a__:				Any 							=0


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


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


						def      _lowerCamelCase	(       self  :     int	):
													a__:				str 							=0
													a__:				Tuple 							=0


						def      _lowerCamelCase	(       self  :     Optional[int]	):
													a__:				Any 							=self.inference_layers_num / self.inference_instances_num
													a__:				Tuple 							=(
													    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(lowerCAmelCase__	)







						@add_start_docstrings_to_model_forward(lowerCAmelCase__	)
						def      _lowerCamelCase	(       self  :     List[Any]					,				_a  :     int=None					,				_a  :     Optional[Any]=None					,				_a  :     str=None					,				_a  :     int=None					,				_a  :     Any=None					,				_a  :     str=None					,				_a  :     Tuple=None					,				_a  :     Optional[Any]=None					,				_a  :     str=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__:				List[Any] 							=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__:				List[Any] 							=input_ids.device if input_ids is not None else inputs_embeds.device

													if attention_mask is None:
																				a__:				Optional[Any] 							=torch.ones(lowerCAmelCase__					,				device=lowerCAmelCase__	)
													if token_type_ids is None:
																				a__:				List[str] 							=torch.zeros(lowerCAmelCase__					,				dtype=torch.long					,				device=lowerCAmelCase__	)

													a__:				Dict 							=attention_mask.unsqueeze(1	).unsqueeze(2	)
													a__:				int 							=extended_attention_mask.to(dtype=self.dtype	)  # fp16 compatibility
													a__:				Tuple 							=(1.0 - extended_attention_mask) * -1_0_0_0_0.0
													a__:				int 							=self.get_head_mask(lowerCAmelCase__					,				self.config.num_hidden_layers	)

													a__:				Any 							=self.embeddings(
													    lowerCAmelCase__					,				position_ids=lowerCAmelCase__					,				token_type_ids=lowerCAmelCase__					,				inputs_embeds=lowerCAmelCase__	)
													a__:				Any 							=embedding_output

													if self.training:
																				a__:				Dict 							=[]
																				for i in range(self.config.num_hidden_layers	):
																											a__:				Tuple 							=self.encoder.adaptive_forward(
																											    lowerCAmelCase__					,				current_layer=lowerCAmelCase__					,				attention_mask=lowerCAmelCase__					,				head_mask=lowerCAmelCase__					,				)

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

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

																											a__:				List[str] 							=logits
																											if patient_counter == self.patience:
																																		break
																				a__:				List[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. ''' ,    __SCREAMING_SNAKE_CASE ,    )
class 		lowerCamelCase__       (						__SCREAMING_SNAKE_CASE      ):


						def __init__(       self  :     Tuple					,				_a  :     Tuple	):
													super().__init__(lowerCAmelCase__	)
													a__:				List[str] 							=config.num_labels

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

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

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

													return outputs









from __future__ import annotations






def 		a__       (							A__							):
			if len(A__							) == 0:
						raise ValueError('find_max() arg is an empty sequence'							)
			SCREAMING_SNAKE_CASE_						:			str       						= nums[0]
			for x in nums:
						if x > max_num:
									SCREAMING_SNAKE_CASE_						:			Optional[int]       						= x
			return max_num


if __name__ == "__main__":
				import doctest

				doctest.testmod(verbose=True)





