





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.bert.modeling_bert import (
    BERT_INPUTS_DOCSTRING,
    BERT_START_DOCSTRING,
    BertEncoder,
    BertModel,
    BertPreTrainedModel,
)


lowercase_		:				List[str]				=						logging.getLogger(__name__)


class 			lowercase				(							a_     ):



						"""simple docstring"""

						def 		__UpperCAmelCase       (    self			:		Optional[Any] ,				lowerCamelCase_			:		Union[str, Any] ,				lowerCamelCase_			:		str ,				lowerCamelCase_			:		Tuple=None ,				lowerCamelCase_			:		Optional[int]=None					):





										'''simple docstring'''







										_snake_case :			List[Any]  						=   self.layer[current_layer](lowerCamelCase_ ,				lowerCamelCase_ ,				head_mask[current_layer]					)

										_snake_case :			Union[str, Any]  						=   layer_outputs[0]

										return hidden_states



@add_start_docstrings(
    "The bare Bert Model transformer with PABEE outputting raw hidden-states without any specific head on top."		,			a_		,			)
class 			lowercase				(							a_     ):



						"""simple docstring"""

						def __init__(    self			:		int ,				lowerCamelCase_			:		str					):





										'''simple docstring'''







										super().__init__(lowerCamelCase_					)

										_snake_case :			List[str]  						=   BertEncoderWithPabee(lowerCamelCase_					)

										self.init_weights()
										_snake_case :			Tuple  						=   0
										_snake_case :			List[str]  						=   0
										_snake_case :			Dict  						=   0

										_snake_case :			Optional[int]  						=   0







						def 		__UpperCAmelCase       (    self			:		List[Any] ,				lowerCamelCase_			:		Tuple					):





										'''simple docstring'''







										_snake_case :			Optional[Any]  						=   threshold







						def 		__UpperCAmelCase       (    self			:		Any ,				lowerCamelCase_			:		int					):





										'''simple docstring'''







										_snake_case :			Dict  						=   patience







						def 		__UpperCAmelCase       (    self			:		Optional[Any]					):





										'''simple docstring'''







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







						def 		__UpperCAmelCase       (    self			:		Dict					):





										'''simple docstring'''







										_snake_case :			Tuple  						=   self.inference_layers_num / self.inference_instances_num
										_snake_case :			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(lowerCamelCase_					)







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

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

										if attention_mask is None:
														_snake_case :			List[Any]  						=   torch.ones(lowerCamelCase_ ,				device=lowerCamelCase_					)
										if token_type_ids is None:
														_snake_case :			Optional[int]  						=   torch.zeros(lowerCamelCase_ ,				dtype=torch.long ,				device=lowerCamelCase_					)

										# We can provide a self-attention mask of dimensions [batch_size, from_seq_length, to_seq_length]
										# ourselves in which case we just need to make it broadcastable to all heads.
										_snake_case :			torch.Tensor  						=   self.get_extended_attention_mask(lowerCamelCase_ ,				lowerCamelCase_ ,				lowerCamelCase_					)

										# If a 2D ou 3D attention mask is provided for the cross-attention
										# we need to make broadcastable to [batch_size, num_heads, seq_length, seq_length]
										if self.config.is_decoder and encoder_hidden_states is not None:
														_snake_case :			Tuple  						=   encoder_hidden_states.size()
														_snake_case :			List[str]  						=   (encoder_batch_size, encoder_sequence_length)
														if encoder_attention_mask is None:
																		_snake_case :			Dict  						=   torch.ones(lowerCamelCase_ ,				device=lowerCamelCase_					)
														_snake_case :			int  						=   self.invert_attention_mask(lowerCamelCase_					)
										else:
														_snake_case :			str  						=   None

										# Prepare head mask if needed
										# 1.0 in head_mask indicate we keep the head
										# attention_probs has shape bsz x n_heads x N x N
										# input head_mask has shape [num_heads] or [num_hidden_layers x num_heads]
										# and head_mask is converted to shape [num_hidden_layers x batch x num_heads x seq_length x seq_length]
										_snake_case :			Any  						=   self.get_head_mask(lowerCamelCase_ ,				self.config.num_hidden_layers					)

										_snake_case :			Dict  						=   self.embeddings(
										    input_ids=lowerCamelCase_ ,				position_ids=lowerCamelCase_ ,				token_type_ids=lowerCamelCase_ ,				inputs_embeds=lowerCamelCase_					)
										_snake_case :			List[Any]  						=   embedding_output

										if self.training:
														_snake_case :			int  						=   []
														for i in range(self.config.num_hidden_layers					):
																		_snake_case :			Tuple  						=   self.encoder.adaptive_forward(
																		    lowerCamelCase_ ,				current_layer=lowerCamelCase_ ,				attention_mask=lowerCamelCase_ ,				head_mask=lowerCamelCase_					)

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

																		_snake_case :			List[str]  						=   self.pooler(lowerCamelCase_					)
																		_snake_case :			Dict  						=   output_layers[i](lowerCamelCase_					)
																		if regression:
																						_snake_case :			Union[str, Any]  						=   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 :			int  						=   0
																		else:
																						_snake_case :			Optional[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(lowerCamelCase_					)					):
																										patient_counter += 1
																						else:
																										_snake_case :			Optional[int]  						=   0

																		_snake_case :			Dict  						=   logits
																		if patient_counter == self.patience:
																						break
														_snake_case :			Any  						=   [patient_result]
														self.inference_layers_num += calculated_layer_num
														self.inference_instances_num += 1

										return res



@add_start_docstrings(
    "Bert 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 			lowercase				(							a_     ):



						"""simple docstring"""

						def __init__(    self			:		Union[str, Any] ,				lowerCamelCase_			:		Tuple					):





										'''simple docstring'''







										super().__init__(lowerCamelCase_					)
										_snake_case :			List[str]  						=   config.num_labels

										_snake_case :			List[Any]  						=   BertModelWithPabee(lowerCamelCase_					)
										_snake_case :			List[str]  						=   nn.Dropout(config.hidden_dropout_prob					)
										_snake_case :			Optional[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(lowerCamelCase_					)
						def 		__UpperCAmelCase       (    self			:		Optional[Any] ,				lowerCamelCase_			:		Optional[Any]=None ,				lowerCamelCase_			:		int=None ,				lowerCamelCase_			:		Optional[Any]=None ,				lowerCamelCase_			:		Any=None ,				lowerCamelCase_			:		Tuple=None ,				lowerCamelCase_			:		Tuple=None ,				lowerCamelCase_			:		List[str]=None ,				):





										'''simple docstring'''







										_snake_case :			Tuple  						=   self.bert(
										    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 ,				)

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

										if labels is not None:
														_snake_case :			Dict  						=   None
														_snake_case :			Optional[Any]  						=   0
														for ix, logits_item in enumerate(lowerCamelCase_					):
																		if self.num_labels == 1:
																						#  We are doing regression
																						_snake_case :			int  						=   MSELoss()
																						_snake_case :			Optional[int]  						=   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 :			List[str]  						=   loss
																		else:
																						total_loss += loss * (ix + 1)
																		total_weights += ix + 1
														_snake_case :			Optional[int]  						=   (total_loss / total_weights,) + outputs

										return outputs













lowercase_		:				Tuple				=						'''
# Installazione di Transformers
! pip install transformers datasets
# Per installare dalla fonte invece dell\'ultima versione rilasciata, commenta il comando sopra e
# rimuovi la modalità commento al comando seguente.
# ! pip install git+https://github.com/huggingface/transformers.git
'''

lowercase_		:				Optional[int]				=						[{'''type''': '''code''', '''content''': INSTALL_CONTENT}]
lowercase_		:				str				=						{
    '''{processor_class}''': '''FakeProcessorClass''',
    '''{model_class}''': '''FakeModelClass''',
    '''{object_class}''': '''FakeObjectClass''',
}






