



'''simple docstring'''






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). """	,				lowerCamelCase__	,				)
class 							a__(      lowerCamelCase__			):
    lowercase__				=					RobertaConfig
    lowercase__				=					"""roberta"""







    def __init__(   self				:						Union[str, Any]		,    __snake_case				:						Optional[int]			):
      super().__init__(__snake_case			)

      a		:      Optional[Any]       =  RobertaEmbeddings(__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. """	,				lowerCamelCase__	,				)
class 							a__(      lowerCamelCase__			):
    lowercase__				=					RobertaConfig
    lowercase__				=					"""roberta"""







    def __init__(   self				:						Union[str, Any]		,    __snake_case				:						str			):
      super().__init__(__snake_case			)
      a		:      int       =  config.num_labels
      a		:      int       =  config.num_hidden_layers

      a		:      Union[str, Any]       =  DeeRobertaModel(__snake_case			)
      a		:      str       =  nn.Dropout(config.hidden_dropout_prob			)
      a		:      Any       =  nn.Linear(config.hidden_size		,    self.config.num_labels			)







    @add_start_docstrings_to_model_forward(__snake_case			)
    def     lowercase_ (   self				:						List[Any]		,    __snake_case				:						Dict=None		,    __snake_case				:						int=None		,    __snake_case				:						List[Any]=None		,    __snake_case				:						Optional[int]=None		,    __snake_case				:						Any=None		,    __snake_case				:						Any=None		,    __snake_case				:						Optional[Any]=None		,    __snake_case				:						int=-1		,    __snake_case				:						Optional[Any]=False		,    ):
      a		:      str       =  self.num_layers
      try:
        a		:      List[Any]       =  self.roberta(
            __snake_case		,    attention_mask=__snake_case		,    token_type_ids=__snake_case		,    position_ids=__snake_case		,    head_mask=__snake_case		,    inputs_embeds=__snake_case		,    )

        a		:      Tuple       =  outputs[1]

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

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

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

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

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




'''simple docstring'''






import unittest

import numpy as np

from transformers import BertConfig, is_flax_available
from transformers.testing_utils import require_flax, slow

from ...test_modeling_flax_common import FlaxModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask


if is_flax_available():
     from transformers.models.bert.modeling_flax_bert import (
         FlaxBertForMaskedLM,
         FlaxBertForMultipleChoice,
         FlaxBertForNextSentencePrediction,
         FlaxBertForPreTraining,
         FlaxBertForQuestionAnswering,
         FlaxBertForSequenceClassification,
         FlaxBertForTokenClassification,
         FlaxBertModel,
     )

class 							a__(      unittest.TestCase			):







    def __init__(   self				:						Dict		,    __snake_case				:						Union[str, Any]		,    __snake_case				:						Any=13		,    __snake_case				:						Any=7		,    __snake_case				:						int=True		,    __snake_case				:						List[Any]=True		,    __snake_case				:						int=True		,    __snake_case				:						List[Any]=True		,    __snake_case				:						Optional[Any]=99		,    __snake_case				:						List[str]=32		,    __snake_case				:						List[Any]=5		,    __snake_case				:						int=4		,    __snake_case				:						List[str]=37		,    __snake_case				:						int="gelu"		,    __snake_case				:						Tuple=0.1		,    __snake_case				:						Union[str, Any]=0.1		,    __snake_case				:						Any=5_12		,    __snake_case				:						Tuple=16		,    __snake_case				:						Union[str, Any]=2		,    __snake_case				:						str=0.02		,    __snake_case				:						Union[str, Any]=4		,    ):
      a		:      List[Any]       =  parent
      a		:      List[str]       =  batch_size
      a		:      Dict       =  seq_length
      a		:      str       =  is_training
      a		:      Optional[int]       =  use_attention_mask
      a		:      Union[str, Any]       =  use_token_type_ids
      a		:      List[str]       =  use_labels
      a		:      Dict       =  vocab_size
      a		:      Tuple       =  hidden_size
      a		:      Optional[Any]       =  num_hidden_layers
      a		:      List[str]       =  num_attention_heads
      a		:      Tuple       =  intermediate_size
      a		:      Dict       =  hidden_act
      a		:      List[str]       =  hidden_dropout_prob
      a		:      str       =  attention_probs_dropout_prob
      a		:      str       =  max_position_embeddings
      a		:      Tuple       =  type_vocab_size
      a		:      Optional[Any]       =  type_sequence_label_size
      a		:      Union[str, Any]       =  initializer_range
      a		:      Union[str, Any]       =  num_choices







    def     lowercase_ (   self				:						Any			):
      a		:      List[Any]       =  ids_tensor([self.batch_size, self.seq_length]		,    self.vocab_size			)

      a		:      List[str]       =  None
      if self.use_attention_mask:
        a		:      List[Any]       =  random_attention_mask([self.batch_size, self.seq_length]			)

      a		:      Dict       =  None
      if self.use_token_type_ids:
        a		:      List[str]       =  ids_tensor([self.batch_size, self.seq_length]		,    self.type_vocab_size			)

      a		:      Dict       =  BertConfig(
          vocab_size=self.vocab_size		,    hidden_size=self.hidden_size		,    num_hidden_layers=self.num_hidden_layers		,    num_attention_heads=self.num_attention_heads		,    intermediate_size=self.intermediate_size		,    hidden_act=self.hidden_act		,    hidden_dropout_prob=self.hidden_dropout_prob		,    attention_probs_dropout_prob=self.attention_probs_dropout_prob		,    max_position_embeddings=self.max_position_embeddings		,    type_vocab_size=self.type_vocab_size		,    is_decoder=__snake_case		,    initializer_range=self.initializer_range		,    )

      return config, input_ids, token_type_ids, attention_mask







    def     lowercase_ (   self				:						Tuple			):
      a		:      Optional[int]       =  self.prepare_config_and_inputs()
      a     ,			a     ,			a     ,			a		:      Union[str, Any]       =  config_and_inputs
      a		:      str       =  {'input_ids': input_ids, 'token_type_ids': token_type_ids, 'attention_mask': attention_mask}
      return config, inputs_dict







    def     lowercase_ (   self				:						List[str]			):
      a		:      int       =  self.prepare_config_and_inputs()
      a     ,			a     ,			a     ,			a		:      Tuple       =  config_and_inputs

      a		:      int       =  True
      a		:      int       =  floats_tensor([self.batch_size, self.seq_length, self.hidden_size]			)
      a		:      Optional[Any]       =  ids_tensor([self.batch_size, self.seq_length]		,    vocab_size=2			)

      return (
          config,
          input_ids,
          attention_mask,
          encoder_hidden_states,
          encoder_attention_mask,
      )

@require_flax
class 							a__(      lowerCamelCase__	,				unittest.TestCase			):
    lowercase__				=					True

    lowercase__				=					(
        (
            FlaxBertModel,
            FlaxBertForPreTraining,
            FlaxBertForMaskedLM,
            FlaxBertForMultipleChoice,
            FlaxBertForQuestionAnswering,
            FlaxBertForNextSentencePrediction,
            FlaxBertForSequenceClassification,
            FlaxBertForTokenClassification,
            FlaxBertForQuestionAnswering,
        )
        if is_flax_available()
        else ()
    )







    def     lowercase_ (   self				:						Union[str, Any]			):
      a		:      str       =  FlaxBertModelTester(self			)







    @slow
    def     lowercase_ (   self				:						Any			):
      # Only check this for base model, not necessary for all model classes.
      # This will also help speed-up tests.
      a		:      Optional[Any]       =  FlaxBertModel.from_pretrained('bert-base-cased'			)
      a		:      Union[str, Any]       =  model(np.ones((1, 1)			)			)
      self.assertIsNotNone(__snake_case			)