
"""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). """				,lowercase__				,)
class A_	(lowercase__      ):




  '''simple docstring'''
  SCREAMING_SNAKE_CASE__			:						Union[str, Any]       =		RobertaConfig
  SCREAMING_SNAKE_CASE__			:						int       =		"""roberta"""







  def __init__(		self				,   lowercase_  ):

     """simple docstring"""
     super().__init__(lowercase_  )

     UpperCAmelCase_					:					int			     =    RobertaEmbeddings(lowercase_  )
     self.init_weights()




@add_start_docstrings(
    """RoBERTa Model (with early exiting - DeeRoBERTa) with a classifier on top,
    also takes care of multi-layer training. """				,lowercase__				,)
class A_	(lowercase__      ):




  '''simple docstring'''
  SCREAMING_SNAKE_CASE__			:						Optional[Any]       =		RobertaConfig
  SCREAMING_SNAKE_CASE__			:						Optional[int]       =		"""roberta"""







  def __init__(		self				,   lowercase_  ):

     """simple docstring"""
     super().__init__(lowercase_  )
     UpperCAmelCase_					:					Any			     =    config.num_labels
     UpperCAmelCase_					:					Optional[Any]			     =    config.num_hidden_layers

     UpperCAmelCase_					:					int			     =    DeeRobertaModel(lowercase_  )
     UpperCAmelCase_					:					Tuple			     =    nn.Dropout(config.hidden_dropout_prob  )
     UpperCAmelCase_					:					Dict			     =    nn.Linear(config.hidden_size				,   self.config.num_labels  )







  @add_start_docstrings_to_model_forward(lowercase_  )
  def        UpperCamelCase__					(		self				,   lowercase_=None				,   lowercase_=None				,   lowercase_=None				,   lowercase_=None				,   lowercase_=None				,   lowercase_=None				,   lowercase_=None				,   lowercase_=-1				,   lowercase_=False				,   ):

     """simple docstring"""
     UpperCAmelCase_					:					str			     =    self.num_layers
     try:
        UpperCAmelCase_					:					str			     =    self.roberta(
            lowercase_				,   attention_mask=lowercase_				,   token_type_ids=lowercase_				,   position_ids=lowercase_				,   head_mask=lowercase_				,   inputs_embeds=lowercase_				,   )

        UpperCAmelCase_					:					List[str]			     =    outputs[1]

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

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

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

        if train_highway:
           UpperCAmelCase_					:					Tuple			     =    (sum(highway_losses[:-1]  ),) + outputs
           # exclude the final highway, of course
        else:
           UpperCAmelCase_					:					Any			     =    (loss,) + outputs
     if not self.training:
        UpperCAmelCase_					:					Any			     =    outputs + ((original_entropy, highway_entropy), exit_layer)
        if output_layer >= 0:
           UpperCAmelCase_					:					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






"""simple docstring"""


import warnings
from typing import Dict, List, Optional, Tuple

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


A     :   str			    =			logging.get_logger(__name__)
class   _UpperCamelCase  (   lowerCAmelCase__		):
							'''simple docstring'''







							__UpperCAmelCase						:    str				      =["""input_ids""", """attention_mask"""]

							def __init__(     self    ,   __a="</s>"    ,   __a="<unk>"    ,   __a="<pad>"    ,   __a=1_25    ,   __a=None    ,   **__a    ,   ):
										# Add extra_ids to the special token list
										if extra_ids > 0 and additional_special_tokens is None:
													__lowerCAmelCase				    =			[f"<extra_id_{i}>" for i in range(__a	)]
										elif extra_ids > 0 and additional_special_tokens is not None:
													# Check that we have the right number of extra_id special tokens
													__lowerCAmelCase				    =			len(set(filter(lambda __a	: bool("extra_id" in str(__a	)	)    ,   __a	)	)	)
													if extra_tokens != extra_ids:
																raise ValueError(
																    f"Both extra_ids ({extra_ids}) and additional_special_tokens ({additional_special_tokens}) are"
																    " provided to ByT5Tokenizer. In this case the additional_special_tokens must include the"
																    " extra_ids tokens"	)

										__lowerCAmelCase				    =			AddedToken(__a    ,   lstrip=__a    ,   rstrip=__a	) if isinstance(__a    ,   __a	) else pad_token
										__lowerCAmelCase				    =			AddedToken(__a    ,   lstrip=__a    ,   rstrip=__a	) if isinstance(__a    ,   __a	) else eos_token
										__lowerCAmelCase				    =			AddedToken(__a    ,   lstrip=__a    ,   rstrip=__a	) if isinstance(__a    ,   __a	) else unk_token

										super().__init__(
										    eos_token=__a    ,   unk_token=__a    ,   pad_token=__a    ,   extra_ids=__a    ,   additional_special_tokens=__a    ,   **__a    ,   )

										__lowerCAmelCase				    =			extra_ids

										__lowerCAmelCase				    =			2**8  # utf is 8 bits

										# define special tokens dict
										__lowerCAmelCase				    =			{
										    self.pad_token: 0,
										    self.eos_token: 1,
										    self.unk_token: 2,
										}
										__lowerCAmelCase				    =			len(self.special_tokens_encoder	)
										__lowerCAmelCase				    =			len(__a	)
										for i, token in enumerate(__a	):
													__lowerCAmelCase				    =			self.vocab_size + i - n
										__lowerCAmelCase				    =			{v: k for k, v in self.special_tokens_encoder.items()}

							@property
							def 			snake_case				(     self	):
										return self._utf_vocab_size + self._num_special_tokens + self._extra_ids

							def 			snake_case				(     self    ,   __a    ,   __a = None    ,   __a = False	):
										if already_has_special_tokens:
													return super().get_special_tokens_mask(
													    token_ids_a=__a    ,   token_ids_a=__a    ,   already_has_special_tokens=__a	)

										# normal case: some special tokens
										if token_ids_a is None:
													return ([0] * len(__a	)) + [1]
										return ([0] * len(__a	)) + [1] + ([0] * len(__a	)) + [1]

							def 			snake_case				(     self    ,   __a	):
										if len(__a	) > 0 and token_ids[-1] == self.eos_token_id:
													warnings.warn(
													    f"This sequence already has {self.eos_token}. In future versions this behavior may lead to duplicated"
													    " eos tokens being added."	)
													return token_ids
										else:
													return token_ids + [self.eos_token_id]

							def 			snake_case				(     self    ,   __a    ,   __a = None	):
										__lowerCAmelCase				    =			[self.eos_token_id]

										if token_ids_a is None:
													return len(token_ids_a + eos	) * [0]
										return len(token_ids_a + eos + token_ids_a + eos	) * [0]

							def 			snake_case				(     self    ,   __a    ,   __a = None	):
										__lowerCAmelCase				    =			self._add_eos_if_not_present(__a	)
										if token_ids_a is None:
													return token_ids_a
										else:
													__lowerCAmelCase				    =			self._add_eos_if_not_present(__a	)
													return token_ids_a + token_ids_a

							def 			snake_case				(     self    ,   __a	):
										__lowerCAmelCase				    =			[chr(__a	) for i in text.encode("utf-8"	)]
										return tokens

							def 			snake_case				(     self    ,   __a	):
										if token in self.special_tokens_encoder:
													__lowerCAmelCase				    =			self.special_tokens_encoder[token]
										elif token in self.added_tokens_encoder:
													__lowerCAmelCase				    =			self.added_tokens_encoder[token]
										elif len(__a	) != 1:
													__lowerCAmelCase				    =			self.unk_token_id
										else:
													__lowerCAmelCase				    =			ord(__a	) + self._num_special_tokens
										return token_id

							def 			snake_case				(     self    ,   __a	):
										if index in self.special_tokens_decoder:
													__lowerCAmelCase				    =			self.special_tokens_decoder[index]
										else:
													__lowerCAmelCase				    =			chr(index - self._num_special_tokens	)
										return token

							def 			snake_case				(     self    ,   __a	):
										__lowerCAmelCase				    =			B""
										for token in tokens:
													if token in self.special_tokens_decoder:
																__lowerCAmelCase				    =			self.special_tokens_decoder[token].encode("utf-8"	)
													elif token in self.added_tokens_decoder:
																__lowerCAmelCase				    =			self.special_tokens_decoder[token].encode("utf-8"	)
													elif token in self.special_tokens_encoder:
																__lowerCAmelCase				    =			token.encode("utf-8"	)
													elif token in self.added_tokens_encoder:
																__lowerCAmelCase				    =			token.encode("utf-8"	)
													else:
																__lowerCAmelCase				    =			bytes([ord(__a	)]	)
													bstring += tok_string
										__lowerCAmelCase				    =			bstring.decode("utf-8"    ,   errors="ignore"	)
										return string



							def 			snake_case				(     self    ,   __a    ,   __a = None	):
										return ()




