




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 		lowercase							(     UpperCamelCase__	):
     _a											=		RobertaConfig
     _a											=		"roberta"

     def __init__(						self	,  _a       )		->							int:
           super().__init__(_a       )

           _A							:  Optional[Any]    =		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 		lowercase							(     UpperCamelCase__	):
     _a											=		RobertaConfig
     _a											=		"roberta"

     def __init__(						self	,  _a       )		->							Any:
           super().__init__(_a       )
           _A							:  List[str]    =		config.num_labels
           _A							:  Optional[Any]    =		config.num_hidden_layers

           _A							:  List[Any]    =		DeeRobertaModel(_a       )
           _A							:  str    =		nn.Dropout(config.hidden_dropout_prob       )
           _A							:  Optional[int]    =		nn.Linear(config.hidden_size	,  self.config.num_labels       )



     @add_start_docstrings_to_model_forward(_a       )
     def       a__			(						self	,  _a=None	,  _a=None	,  _a=None	,  _a=None	,  _a=None	,  _a=None	,  _a=None	,  _a=-1	,  _a=False	,  )		->							List[str]:
           _A							:  Union[str, Any]    =		self.num_layers
           try:
                 _A							:  Dict    =		self.roberta(
                     _a	,  attention_mask=_a	,  token_type_ids=_a	,  position_ids=_a	,  head_mask=_a	,  inputs_embeds=_a	,  )

                 _A							:  Union[str, Any]    =		outputs[1]

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

           if not self.training:
                 _A							:  Optional[int]    =		entropy(_a       )
                 _A							:  List[Any]    =		[]
                 _A							:  Tuple    =		[]
           if labels is not None:
                 if self.num_labels == 1:
                       #  We are doing regression
                       _A							:  Tuple    =		MSELoss()
                       _A							:  Dict    =		loss_fct(logits.view(-1       )	,  labels.view(-1       )       )
                 else:
                       _A							:  List[str]    =		CrossEntropyLoss()
                       _A							:  Union[str, Any]    =		loss_fct(logits.view(-1	,  self.num_labels       )	,  labels.view(-1       )       )

                 # work with highway exits
                 _A							:  Optional[Any]    =		[]
                 for highway_exit in outputs[-1]:
                       _A							:  str    =		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
                             _A							:  str    =		MSELoss()
                             _A							:  List[str]    =		loss_fct(highway_logits.view(-1       )	,  labels.view(-1       )       )
                       else:
                             _A							:  int    =		CrossEntropyLoss()
                             _A							:  Dict    =		loss_fct(highway_logits.view(-1	,  self.num_labels       )	,  labels.view(-1       )       )
                       highway_losses.append(_a       )

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

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







from unittest import TestCase

from datasets import Sequence, Value
from datasets.arrow_dataset import Dataset







class 		lowercase							(     UpperCamelCase__	):

     def       a__			(						self       )		->							Optional[int]:
           return [
               {"col_1": 3, "col_2": "a"},
               {"col_1": 2, "col_2": "b"},
               {"col_1": 1, "col_2": "c"},
               {"col_1": 0, "col_2": "d"},
           ]

     def       a__			(						self       )		->							Optional[int]:
           _A							:  Union[str, Any]    =		{"""col_1""": [3, 2, 1, 0], """col_2""": ["""a""", """b""", """c""", """d"""]}
           return Dataset.from_dict(_a       )

     def       a__			(						self       )		->							Any:
           _A							:  str    =		self._create_example_records()
           _A							:  List[Any]    =		Dataset.from_list(_a       )
           self.assertListEqual(dset.column_names	,  ["""col_1""", """col_2"""]       )
           for i, r in enumerate(_a       ):
                 self.assertDictEqual(_a	,  example_records[i]       )

     def       a__			(						self       )		->							List[str]:
           _A							:  Dict    =		self._create_example_records()
           _A							:  List[str]    =		Dataset.from_list(_a       )
           _A							:  str    =		Dataset.from_dict({k: [r[k] for r in example_records] for k in example_records[0]}       )
           self.assertEqual(dset.info	,  dset_from_dict.info       )

     def       a__			(						self       )		->							str:  # checks what happens with missing columns
           _A							:  List[str]    =		[{"""col_1""": 1}, {"""col_2""": """x"""}]
           _A							:  List[str]    =		Dataset.from_list(_a       )
           self.assertDictEqual(dset[0]	,  {"""col_1""": 1}       )
           self.assertDictEqual(dset[1]	,  {"""col_1""": None}       )  # NB: first record is used for columns

     def       a__			(						self       )		->							Dict:  # checks if the type can be inferred from the second record
           _A							:  List[Any]    =		[{"""col_1""": []}, {"""col_1""": [1, 2]}]
           _A							:  str    =		Dataset.from_list(_a       )
           self.assertEqual(dset.info.features["""col_1"""]	,  Sequence(Value("""int64"""       )       )       )



     def       a__			(						self       )		->							Dict:
           _A							:  List[str]    =		Dataset.from_list([]       )
           self.assertEqual(len(_a       )	,  0       )
           self.assertListEqual(dset.column_names	,  []       )

