





'''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). """				,		_a				,		)
class     A_      (    _a     ):
 _lowerCamelCase    :					str       			=				RobertaConfig
 _lowerCamelCase    :					List[str]       			=				"""roberta"""




 def __init__(   self			:       List[str]      ,     snake_case_			:       Any       ):
      super().__init__(snake_case_       )

      _UpperCAmelCase					=  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     A_      (    _a     ):
 _lowerCamelCase    :					Union[str, Any]       			=				RobertaConfig
 _lowerCamelCase    :					List[Any]       			=				"""roberta"""




 def __init__(   self			:       List[str]      ,     snake_case_			:       Optional[Any]       ):
      super().__init__(snake_case_       )
      _UpperCAmelCase					=  config.num_labels
      _UpperCAmelCase					=  config.num_hidden_layers

      _UpperCAmelCase					=  DeeRobertaModel(snake_case_       )
      _UpperCAmelCase					=  nn.Dropout(config.hidden_dropout_prob       )
      _UpperCAmelCase					=  nn.Linear(config.hidden_size      ,     self.config.num_labels       )




 @add_start_docstrings_to_model_forward(snake_case_       )
 def 	lowercase      (   self			:       Any      ,     snake_case_			:       Any=None      ,     snake_case_			:       str=None      ,     snake_case_			:       Tuple=None      ,     snake_case_			:       Dict=None      ,     snake_case_			:       Any=None      ,     snake_case_			:       Optional[Any]=None      ,     snake_case_			:       int=None      ,     snake_case_			:       Optional[int]=-1      ,     snake_case_			:       Tuple=False      ,     ):
      _UpperCAmelCase					=  self.num_layers
      try:
           _UpperCAmelCase					=  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_      ,     )

           _UpperCAmelCase					=  outputs[1]

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

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

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

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

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
	import torch

if is_vision_available():
	from PIL import Image

	from transformers import PoolFormerImageProcessor






class      UpperCamelCase(	unittest.TestCase ):


					def __init__(    self						:		Tuple  ,     SCREAMING_SNAKE_CASE						:		List[Any]  ,     SCREAMING_SNAKE_CASE						:		Optional[Any]=7  ,     SCREAMING_SNAKE_CASE						:		Dict=3  ,     SCREAMING_SNAKE_CASE						:		Any=3_0  ,     SCREAMING_SNAKE_CASE						:		List[str]=4_0_0  ,     SCREAMING_SNAKE_CASE						:		Any=True  ,     SCREAMING_SNAKE_CASE						:		List[Any]=None  ,     SCREAMING_SNAKE_CASE						:		int=0.9  ,     SCREAMING_SNAKE_CASE						:		Tuple=None  ,     SCREAMING_SNAKE_CASE						:		int=True  ,     SCREAMING_SNAKE_CASE						:		Union[str, Any]=[0.5, 0.5, 0.5]  ,     SCREAMING_SNAKE_CASE						:		Optional[Any]=[0.5, 0.5, 0.5]  ,     )      ->						Union[str, Any]:




											'''simple docstring'''

											__snake_case				=		size if size is not None else {"shortest_edge": 3_0}
											__snake_case				=		crop_size if crop_size is not None else {"height": 3_0, "width": 3_0}
											__snake_case				=		parent
											__snake_case				=		batch_size
											__snake_case				=		num_channels
											__snake_case				=		min_resolution
											__snake_case				=		max_resolution
											__snake_case				=		do_resize_and_center_crop
											__snake_case				=		size
											__snake_case				=		crop_pct
											__snake_case				=		crop_size
											__snake_case				=		do_normalize
											__snake_case				=		image_mean
											__snake_case				=		image_std






					def 				SCREAMING_SNAKE_CASE_	(    self						:		str    )      ->						Any:




											'''simple docstring'''

											return {
											    "size": self.size,
											    "do_resize_and_center_crop": self.do_resize_and_center_crop,
											    "crop_pct": self.crop_pct,
											    "crop_size": self.crop_size,
											    "do_normalize": self.do_normalize,
											    "image_mean": self.image_mean,
											    "image_std": self.image_std,
											}






@require_torch
@require_vision
class      UpperCamelCase(	_a  ,				unittest.TestCase ):
					snake_case_		:					List[str]    					=						PoolFormerImageProcessor if is_vision_available() else None


					def 				SCREAMING_SNAKE_CASE_	(    self						:		Union[str, Any]    )      ->						str:




											'''simple docstring'''

											__snake_case				=		PoolFormerImageProcessingTester(self    )


					@property
					def 				SCREAMING_SNAKE_CASE_	(    self						:		int    )      ->						Dict:




											'''simple docstring'''

											return self.image_processor_tester.prepare_image_processor_dict()


					def 				SCREAMING_SNAKE_CASE_	(    self						:		Dict    )      ->						List[str]:




											'''simple docstring'''

											__snake_case				=		self.image_processing_class(**self.image_processor_dict    )
											self.assertTrue(hasattr(SCREAMING_SNAKE_CASE  ,     "do_resize_and_center_crop"    )    )
											self.assertTrue(hasattr(SCREAMING_SNAKE_CASE  ,     "size"    )    )
											self.assertTrue(hasattr(SCREAMING_SNAKE_CASE  ,     "crop_pct"    )    )
											self.assertTrue(hasattr(SCREAMING_SNAKE_CASE  ,     "do_normalize"    )    )
											self.assertTrue(hasattr(SCREAMING_SNAKE_CASE  ,     "image_mean"    )    )
											self.assertTrue(hasattr(SCREAMING_SNAKE_CASE  ,     "image_std"    )    )


					def 				SCREAMING_SNAKE_CASE_	(    self						:		Union[str, Any]    )      ->						Any:




											'''simple docstring'''

											__snake_case				=		self.image_processing_class.from_dict(self.image_processor_dict    )
											self.assertEqual(image_processor.size  ,     {"shortest_edge": 3_0}    )
											self.assertEqual(image_processor.crop_size  ,     {"height": 3_0, "width": 3_0}    )

											__snake_case				=		self.image_processing_class.from_dict(self.image_processor_dict  ,     size=4_2  ,     crop_size=8_4    )
											self.assertEqual(image_processor.size  ,     {"shortest_edge": 4_2}    )
											self.assertEqual(image_processor.crop_size  ,     {"height": 8_4, "width": 8_4}    )


					def 				SCREAMING_SNAKE_CASE_	(    self						:		Optional[int]    )      ->						List[Any]:




											'''simple docstring'''

											pass


					def 				SCREAMING_SNAKE_CASE_	(    self						:		Optional[Any]    )      ->						Dict:




											'''simple docstring'''

											__snake_case				=		self.image_processing_class(**self.image_processor_dict    )
											# create random PIL images
											__snake_case				=		prepare_image_inputs(self.image_processor_tester  ,     equal_resolution=SCREAMING_SNAKE_CASE    )
											for image in image_inputs:
																	self.assertIsInstance(SCREAMING_SNAKE_CASE  ,     Image.Image    )

											# Test not batched input
											__snake_case				=		image_processing(image_inputs[0]  ,     return_tensors="pt"    ).pixel_values

											self.assertEqual(
											    encoded_images.shape  ,     (
											        1,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["height"],
											        self.image_processor_tester.crop_size["width"],
											    )  ,     )

											# Test batched
											__snake_case				=		image_processing(SCREAMING_SNAKE_CASE  ,     return_tensors="pt"    ).pixel_values
											self.assertEqual(
											    encoded_images.shape  ,     (
											        self.image_processor_tester.batch_size,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["height"],
											        self.image_processor_tester.crop_size["width"],
											    )  ,     )


					def 				SCREAMING_SNAKE_CASE_	(    self						:		Optional[Any]    )      ->						Optional[int]:




											'''simple docstring'''

											__snake_case				=		self.image_processing_class(**self.image_processor_dict    )
											# create random numpy tensors
											__snake_case				=		prepare_image_inputs(self.image_processor_tester  ,     equal_resolution=SCREAMING_SNAKE_CASE  ,     numpify=SCREAMING_SNAKE_CASE    )
											for image in image_inputs:
																	self.assertIsInstance(SCREAMING_SNAKE_CASE  ,     np.ndarray    )

											# Test not batched input
											__snake_case				=		image_processing(image_inputs[0]  ,     return_tensors="pt"    ).pixel_values
											self.assertEqual(
											    encoded_images.shape  ,     (
											        1,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["height"],
											        self.image_processor_tester.crop_size["width"],
											    )  ,     )

											# Test batched
											__snake_case				=		image_processing(SCREAMING_SNAKE_CASE  ,     return_tensors="pt"    ).pixel_values
											self.assertEqual(
											    encoded_images.shape  ,     (
											        self.image_processor_tester.batch_size,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["height"],
											        self.image_processor_tester.crop_size["width"],
											    )  ,     )






					def 				SCREAMING_SNAKE_CASE_	(    self						:		Optional[int]    )      ->						Union[str, Any]:




											'''simple docstring'''

											__snake_case				=		self.image_processing_class(**self.image_processor_dict    )
											# create random PyTorch tensors
											__snake_case				=		prepare_image_inputs(self.image_processor_tester  ,     equal_resolution=SCREAMING_SNAKE_CASE  ,     torchify=SCREAMING_SNAKE_CASE    )
											for image in image_inputs:
																	self.assertIsInstance(SCREAMING_SNAKE_CASE  ,     torch.Tensor    )

											# Test not batched input
											__snake_case				=		image_processing(image_inputs[0]  ,     return_tensors="pt"    ).pixel_values
											self.assertEqual(
											    encoded_images.shape  ,     (
											        1,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["height"],
											        self.image_processor_tester.crop_size["width"],
											    )  ,     )

											# Test batched
											__snake_case				=		image_processing(SCREAMING_SNAKE_CASE  ,     return_tensors="pt"    ).pixel_values
											self.assertEqual(
											    encoded_images.shape  ,     (
											        self.image_processor_tester.batch_size,
											        self.image_processor_tester.num_channels,
											        self.image_processor_tester.crop_size["height"],
											        self.image_processor_tester.crop_size["width"],
											    )  ,     )





