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   lowerCamelCase					(__UpperCAmelCase					):
  """simple docstring"""







  lowerCamelCase__     =						RobertaConfig
  lowerCamelCase__     =						'''roberta'''







  def __init__(							self :       Optional[int]  ,     __magic_name__ :       str   )  ->	Optional[int]:
        super().__init__(__SCREAMING_SNAKE_CASE   )

        SCREAMING_SNAKE_CASE_          =  RobertaEmbeddings(__SCREAMING_SNAKE_CASE   )
        self.init_weights()



@add_start_docstrings(
    '''RoBERTa Model (with early exiting - DeeRoBERTa) with a classifier on top,
    also takes care of multi-layer training. '''  ,      __UpperCAmelCase  ,      )
class   lowerCamelCase					(__UpperCAmelCase					):
  """simple docstring"""







  lowerCamelCase__     =						RobertaConfig
  lowerCamelCase__     =						'''roberta'''







  def __init__(							self :       Dict  ,     __magic_name__ :       Any   )  ->	Tuple:
        super().__init__(__SCREAMING_SNAKE_CASE   )
        SCREAMING_SNAKE_CASE_          =  config.num_labels
        SCREAMING_SNAKE_CASE_          =  config.num_hidden_layers

        SCREAMING_SNAKE_CASE_          =  DeeRobertaModel(__SCREAMING_SNAKE_CASE   )
        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(__SCREAMING_SNAKE_CASE   )
  def    __A     (							self :       int  ,     __magic_name__ :       Any=None  ,     __magic_name__ :       Optional[int]=None  ,     __magic_name__ :       Dict=None  ,     __magic_name__ :       Dict=None  ,     __magic_name__ :       List[str]=None  ,     __magic_name__ :       Dict=None  ,     __magic_name__ :       List[Any]=None  ,     __magic_name__ :       List[Any]=-1  ,     __magic_name__ :       List[Any]=False  ,     )  ->	Any:
        SCREAMING_SNAKE_CASE_          =  self.num_layers
        try:
              SCREAMING_SNAKE_CASE_          =  self.roberta(
                  __SCREAMING_SNAKE_CASE  ,     attention_mask=__SCREAMING_SNAKE_CASE  ,     token_type_ids=__SCREAMING_SNAKE_CASE  ,     position_ids=__SCREAMING_SNAKE_CASE  ,     head_mask=__SCREAMING_SNAKE_CASE  ,     inputs_embeds=__SCREAMING_SNAKE_CASE  ,     )

              SCREAMING_SNAKE_CASE_          =  outputs[1]

              SCREAMING_SNAKE_CASE_          =  self.dropout(__SCREAMING_SNAKE_CASE   )
              SCREAMING_SNAKE_CASE_          =  self.classifier(__SCREAMING_SNAKE_CASE   )
              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(__SCREAMING_SNAKE_CASE   )
              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(__SCREAMING_SNAKE_CASE   )
                          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(__SCREAMING_SNAKE_CASE   )

              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



import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_sentencepiece, require_tokenizers, require_torch, slow


if is_torch_available():
      import torch

      from transformers import XLMRobertaModel


@require_sentencepiece
@require_tokenizers
@require_torch
class   lowerCamelCase					(unittest.TestCase					):
  """simple docstring"""







  @slow
  def    __A     (							self :       Dict   )  ->	Any:
        SCREAMING_SNAKE_CASE_          =  XLMRobertaModel.from_pretrained("xlm-roberta-base"   )
        SCREAMING_SNAKE_CASE_          =  torch.tensor([[0, 581, 10_269, 83, 99_942, 136, 60_742, 23, 70, 80_583, 18_276, 2]]   )
        # The dog is cute and lives in the garden house

        SCREAMING_SNAKE_CASE_          =  torch.Size((1, 12, 768)   )  # batch_size, sequence_length, embedding_vector_dim
        SCREAMING_SNAKE_CASE_          =  torch.tensor(
            [[-0.0101, 0.1218, -0.0803, 0.0801, 0.1327, 0.0776, -0.1215, 0.2383, 0.3338, 0.3106, 0.0300, 0.0252]]   )
        #  xlmr = torch.hub.load('pytorch/fairseq', 'xlmr.base')
        #  xlmr.eval()
        #  expected_output_values_last_dim = xlmr.extract_features(input_ids[0])[:, :, -1]
        with torch.no_grad():
              SCREAMING_SNAKE_CASE_          =  model(__magic_name__   )["last_hidden_state"].detach()
        self.assertEqual(output.shape  ,     __magic_name__   )
        # compare the actual values for a slice of last dim
        self.assertTrue(torch.allclose(output[:, :, -1]  ,     __magic_name__  ,     atol=1e-3   )   )







  @slow
  def    __A     (							self :       Tuple   )  ->	List[str]:
        SCREAMING_SNAKE_CASE_          =  XLMRobertaModel.from_pretrained("xlm-roberta-large"   )
        SCREAMING_SNAKE_CASE_          =  torch.tensor([[0, 581, 10_269, 83, 99_942, 136, 60_742, 23, 70, 80_583, 18_276, 2]]   )
        # The dog is cute and lives in the garden house

        SCREAMING_SNAKE_CASE_          =  torch.Size((1, 12, 1_024)   )  # batch_size, sequence_length, embedding_vector_dim
        SCREAMING_SNAKE_CASE_          =  torch.tensor(
            [[-0.0699, -0.0318, 0.0705, -0.1241, 0.0999, -0.0520, 0.1004, -0.1838, -0.4704, 0.1437, 0.0821, 0.0126]]   )
        #  xlmr = torch.hub.load('pytorch/fairseq', 'xlmr.large')
        #  xlmr.eval()
        #  expected_output_values_last_dim = xlmr.extract_features(input_ids[0])[:, :, -1]
        with torch.no_grad():
              SCREAMING_SNAKE_CASE_          =  model(__magic_name__   )["last_hidden_state"].detach()
        self.assertEqual(output.shape  ,     __magic_name__   )
        # compare the actual values for a slice of last dim
        self.assertTrue(torch.allclose(output[:, :, -1]  ,     __magic_name__  ,     atol=1e-3   )   )


