





from __future__ import absolute_import, division, print_function, unicode_literals

from torch import nn
from torch.nn import CrossEntropyLoss, MSELoss

from transformers import RobertaConfig
from transformers.file_utils import add_start_docstrings, add_start_docstrings_to_model_forward
from transformers.models.roberta.modeling_roberta import (
    ROBERTA_INPUTS_DOCSTRING,
    ROBERTA_START_DOCSTRING,
    RobertaEmbeddings,
)

from .modeling_highway_bert import BertPreTrainedModel, DeeBertModel, HighwayException, entropy




@add_start_docstrings(
    "The RoBERTa Model transformer with early exiting (DeeRoBERTa). "				,  UpperCamelCase__				,  )
class   UpperCamelCase_       (     UpperCamelCase__    ):
							lowerCamelCase_           =					RobertaConfig
							lowerCamelCase_           =					"roberta"


							def __init__(							self    :Optional[Any]							,				__A    :List[Any]		)   ->   Tuple:



									"""simple docstring"""



									super().__init__(__A		)

									SCREAMING_SNAKE_CASE__			=      RobertaEmbeddings(__A		)
									self.init_weights()




@add_start_docstrings(
    "RoBERTa Model (with early exiting - DeeRoBERTa) with a classifier on top,\n    also takes care of multi-layer training. "				,  UpperCamelCase__				,  )
class   UpperCamelCase_       (     UpperCamelCase__    ):
							lowerCamelCase_           =					RobertaConfig
							lowerCamelCase_           =					"roberta"


							def __init__(							self    :Optional[Any]							,				__A    :Union[str, Any]		)   ->   Any:



									"""simple docstring"""



									super().__init__(__A		)
									SCREAMING_SNAKE_CASE__			=      config.num_labels
									SCREAMING_SNAKE_CASE__			=      config.num_hidden_layers

									SCREAMING_SNAKE_CASE__			=      DeeRobertaModel(__A		)
									SCREAMING_SNAKE_CASE__			=      nn.Dropout(config.hidden_dropout_prob		)
									SCREAMING_SNAKE_CASE__			=      nn.Linear(config.hidden_size							,				self.config.num_labels		)


							@add_start_docstrings_to_model_forward(__A		)
							def      _snake_case  (							self    :Union[str, Any]							,				__A    :int=None							,				__A    :str=None							,				__A    :List[Any]=None							,				__A    :Any=None							,				__A    :List[Any]=None							,				__A    :Optional[int]=None							,				__A    :List[str]=None							,				__A    :Union[str, Any]=-1							,				__A    :int=False							,				)   ->   List[str]:



									"""simple docstring"""



									SCREAMING_SNAKE_CASE__			=      self.num_layers
									try:
											SCREAMING_SNAKE_CASE__			=      self.roberta(
											    __A							,				attention_mask=__A							,				token_type_ids=__A							,				position_ids=__A							,				head_mask=__A							,				inputs_embeds=__A							,				)

											SCREAMING_SNAKE_CASE__			=      outputs[1]

											SCREAMING_SNAKE_CASE__			=      self.dropout(__A		)
											SCREAMING_SNAKE_CASE__			=      self.classifier(__A		)
											SCREAMING_SNAKE_CASE__			=      (logits,) + outputs[2:]  # add hidden states and attention if they are here
									except HighwayException as e:
											SCREAMING_SNAKE_CASE__			=      e.message
											SCREAMING_SNAKE_CASE__			=      e.exit_layer
											SCREAMING_SNAKE_CASE__			=      outputs[0]

									if not self.training:
											SCREAMING_SNAKE_CASE__			=      entropy(__A		)
											SCREAMING_SNAKE_CASE__			=      []
											SCREAMING_SNAKE_CASE__			=      []
									if labels is not None:
											if self.num_labels == 1:
													#  We are doing regression
													SCREAMING_SNAKE_CASE__			=      MSELoss()
													SCREAMING_SNAKE_CASE__			=      loss_fct(logits.view(-1		)							,				labels.view(-1		)		)
											else:
													SCREAMING_SNAKE_CASE__			=      CrossEntropyLoss()
													SCREAMING_SNAKE_CASE__			=      loss_fct(logits.view(-1							,				self.num_labels		)							,				labels.view(-1		)		)

											# work with highway exits
											SCREAMING_SNAKE_CASE__			=      []
											for highway_exit in outputs[-1]:
													SCREAMING_SNAKE_CASE__			=      highway_exit[0]
													if not self.training:
															highway_logits_all.append(__A		)
															highway_entropy.append(highway_exit[2]		)
													if self.num_labels == 1:
															#  We are doing regression
															SCREAMING_SNAKE_CASE__			=      MSELoss()
															SCREAMING_SNAKE_CASE__			=      loss_fct(highway_logits.view(-1		)							,				labels.view(-1		)		)
													else:
															SCREAMING_SNAKE_CASE__			=      CrossEntropyLoss()
															SCREAMING_SNAKE_CASE__			=      loss_fct(highway_logits.view(-1							,				self.num_labels		)							,				labels.view(-1		)		)
													highway_losses.append(__A		)

											if train_highway:
													SCREAMING_SNAKE_CASE__			=      (sum(highway_losses[:-1]		),) + outputs
													# exclude the final highway, of course
											else:
													SCREAMING_SNAKE_CASE__			=      (loss,) + outputs
									if not self.training:
											SCREAMING_SNAKE_CASE__			=      outputs + ((original_entropy, highway_entropy), exit_layer)
											if output_layer >= 0:
													SCREAMING_SNAKE_CASE__			=      (
													    (outputs[0],) + (highway_logits_all[output_layer],) + outputs[2:]
													)  # use the highway of the last layer

									return outputs  # (loss), logits, (hidden_states), (attentions), entropy






def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							list[list[float]]    ):
		SCREAMING_SNAKE_CASE__			=      []
		for data in source_data:
				for i, el in enumerate(UpperCamelCase__    ):
						if len(UpperCamelCase__    ) < i + 1:
								data_lists.append([]    )
						data_lists[i].append(float(UpperCamelCase__    )    )
		return data_lists






def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							list[list[float]]					,   UpperCamelCase__:							list[int]    ):
		SCREAMING_SNAKE_CASE__			=      []
		for dlist, weight in zip(UpperCamelCase__					,   UpperCamelCase__    ):
				SCREAMING_SNAKE_CASE__			=      min(UpperCamelCase__    )
				SCREAMING_SNAKE_CASE__			=      max(UpperCamelCase__    )

				SCREAMING_SNAKE_CASE__			=      []
				# for weight 0 score is 1 - actual score
				if weight == 0:
						for item in dlist:
								try:
										score.append(1 - ((item - mind) / (maxd - mind))    )
								except ZeroDivisionError:
										score.append(1    )

				elif weight == 1:
						for item in dlist:
								try:
										score.append((item - mind) / (maxd - mind)    )
								except ZeroDivisionError:
										score.append(0    )

        # weight not 0 or 1
				else:
						SCREAMING_SNAKE_CASE__			=      f'''Invalid weight of {weight:f} provided'''
						raise ValueError(UpperCamelCase__    )

				score_lists.append(UpperCamelCase__    )

		return score_lists






def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							list[list[float]]    ):
		SCREAMING_SNAKE_CASE__			=      [0 for i in range(len(score_lists[0]    )    )]

		for slist in score_lists:
				for j, ele in enumerate(UpperCamelCase__    ):
						SCREAMING_SNAKE_CASE__			=      final_scores[j] + ele

		return final_scores






def 				SCREAMING_SNAKE_CASE__							(       UpperCamelCase__:							list[list[float]]					,   UpperCamelCase__:							list[int]    ):
		SCREAMING_SNAKE_CASE__			=      get_data(UpperCamelCase__    )
		SCREAMING_SNAKE_CASE__			=      calculate_each_score(UpperCamelCase__					,   UpperCamelCase__    )
		SCREAMING_SNAKE_CASE__			=      generate_final_scores(UpperCamelCase__    )

		# append scores to source data
		for i, ele in enumerate(UpperCamelCase__    ):
				source_data[i].append(UpperCamelCase__    )

		return source_data