






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). '''  ,						_a  ,						)
class 					SCREAMING_SNAKE_CASE_  (       _a ):
       """simple docstring"""


       __lowerCAmelCase    :		List[str]					=RobertaConfig
       __lowerCAmelCase    :		Optional[Any]					='''roberta'''





       def __init__(				self	:Dict,      snake_case	:List[Any]):



          """simple docstring"""

          super().__init__(snake_case)

          _lowercase              =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. '''  ,						_a  ,						)
class 					SCREAMING_SNAKE_CASE_  (       _a ):
       """simple docstring"""


       __lowerCAmelCase    :		List[Any]					=RobertaConfig
       __lowerCAmelCase    :		str					='''roberta'''





       def __init__(				self	:int,      snake_case	:Any):



          """simple docstring"""

          super().__init__(snake_case)
          _lowercase              =config.num_labels
          _lowercase              =config.num_hidden_layers

          _lowercase              =DeeRobertaModel(snake_case)
          _lowercase              =nn.Dropout(config.hidden_dropout_prob)
          _lowercase              =nn.Linear(config.hidden_size,      self.config.num_labels)





       @add_start_docstrings_to_model_forward(snake_case)
       def 				UpperCamelCase__						(				self	:List[str],      snake_case	:str=None,      snake_case	:int=None,      snake_case	:Tuple=None,      snake_case	:List[Any]=None,      snake_case	:Optional[int]=None,      snake_case	:Optional[Any]=None,      snake_case	:Tuple=None,      snake_case	:Tuple=-1,      snake_case	:List[str]=False,      ):



          """simple docstring"""

          _lowercase              =self.num_layers
          try:
             _lowercase              =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,      )

             _lowercase              =outputs[1]

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

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

             # work with highway exits
             _lowercase              =[]
             for highway_exit in outputs[-1]:
                _lowercase              =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
                   _lowercase              =MSELoss()
                   _lowercase              =loss_fct(highway_logits.view(-1),      labels.view(-1))
                else:
                   _lowercase              =CrossEntropyLoss()
                   _lowercase              =loss_fct(highway_logits.view(-1,      self.num_labels),      labels.view(-1))
                highway_losses.append(snake_case)

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

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType






class 					SCREAMING_SNAKE_CASE_  (       _a ):
       """simple docstring"""


       __lowerCAmelCase    :		Dict					=['''image_processor''', '''tokenizer''']
       __lowerCAmelCase    :		List[Any]					='''BlipImageProcessor'''
       __lowerCAmelCase    :		Union[str, Any]					=('''BertTokenizer''', '''BertTokenizerFast''')





       def __init__(				self	:Optional[int],      snake_case	:int,      snake_case	:Optional[int]):



          """simple docstring"""

          _lowercase              =False
          super().__init__(snake_case,      snake_case)
          _lowercase              =self.image_processor





       def __call__(				self	:Dict,      snake_case	:ImageInput = None,      snake_case	:Union[TextInput, PreTokenizedInput, List[TextInput], List[PreTokenizedInput]] = None,      snake_case	:bool = True,      snake_case	:Union[bool, str, PaddingStrategy] = False,      snake_case	:Union[bool, str, TruncationStrategy] = None,      snake_case	:Optional[int] = None,      snake_case	:int = 0,      snake_case	:Optional[int] = None,      snake_case	:Optional[bool] = None,      snake_case	:bool = False,      snake_case	:bool = False,      snake_case	:bool = False,      snake_case	:bool = False,      snake_case	:bool = False,      snake_case	:bool = True,      snake_case	:Optional[Union[str, TensorType]] = None,      **snake_case	:str,      ):



          """simple docstring"""

          if images is None and text is None:
             raise ValueError('You have to specify either images or text.')

          # Get only text
          if images is None:
             _lowercase              =self.tokenizer
             _lowercase              =self.tokenizer(
                 text=snake_case,      add_special_tokens=snake_case,      padding=snake_case,      truncation=snake_case,      max_length=snake_case,      stride=snake_case,      pad_to_multiple_of=snake_case,      return_attention_mask=snake_case,      return_overflowing_tokens=snake_case,      return_special_tokens_mask=snake_case,      return_offsets_mapping=snake_case,      return_token_type_ids=snake_case,      return_length=snake_case,      verbose=snake_case,      return_tensors=snake_case,      **snake_case,      )
             return text_encoding

          # add pixel_values
          _lowercase              =self.image_processor(snake_case,      return_tensors=snake_case)

          if text is not None:
             _lowercase              =self.tokenizer(
                 text=snake_case,      add_special_tokens=snake_case,      padding=snake_case,      truncation=snake_case,      max_length=snake_case,      stride=snake_case,      pad_to_multiple_of=snake_case,      return_attention_mask=snake_case,      return_overflowing_tokens=snake_case,      return_special_tokens_mask=snake_case,      return_offsets_mapping=snake_case,      return_token_type_ids=snake_case,      return_length=snake_case,      verbose=snake_case,      return_tensors=snake_case,      **snake_case,      )
          else:
             _lowercase              =None

          if text_encoding is not None:
             encoding_image_processor.update(snake_case)

          return encoding_image_processor





       def 				UpperCamelCase__						(				self	:List[Any],      *snake_case	:int,      **snake_case	:str):



          """simple docstring"""

          return self.tokenizer.batch_decode(*snake_case,      **snake_case)





       def 				UpperCamelCase__						(				self	:Tuple,      *snake_case	:List[str],      **snake_case	:str):



          """simple docstring"""

          return self.tokenizer.decode(*snake_case,      **snake_case)





       @property
       def 				UpperCamelCase__						(				self	:List[Any]):



          """simple docstring"""

          _lowercase              =self.tokenizer.model_input_names
          _lowercase              =self.image_processor.model_input_names
          return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names))


