




import logging

import torch
from torch import nn
from torch.nn import CrossEntropyLoss, MSELoss

from transformers.file_utils import add_start_docstrings, add_start_docstrings_to_model_forward
from transformers.models.bert.modeling_bert import (
    BERT_INPUTS_DOCSTRING,
    BERT_START_DOCSTRING,
    BertEncoder,
    BertModel,
    BertPreTrainedModel,
)


lowerCamelCase							:						Union[str, Any]					       =  logging.getLogger(__name__)







class     A(		__lowercase  ):



 '''simple docstring'''


 def 							a__					(							self   :    int		,				A_   :    Dict		,				A_   :    Any		,				A_   :    Union[str, Any]=None		,				A_   :    Optional[Any]=None    )  ->      str:

       """simple docstring"""






       lowerCamelCase_							=       self.layer[current_layer](A_		,				A_		,				head_mask[current_layer]    )

       lowerCamelCase_							=       layer_outputs[0]

       return hidden_states







@add_start_docstrings(
    '''The bare Bert Model transformer with PABEE outputting raw hidden-states without any specific head on top.'''				, __lowercase				, )
class     A(		__lowercase  ):



 '''simple docstring'''


 def __init__(							self   :    Optional[Any]		,				A_   :    str    )  ->      Dict:

       """simple docstring"""






       super().__init__(A_    )

       lowerCamelCase_							=       BertEncoderWithPabee(A_    )

       self.init_weights()
       lowerCamelCase_							=       0
       lowerCamelCase_							=       0
       lowerCamelCase_							=       0

       lowerCamelCase_							=       0

 def 							a__					(							self   :    Union[str, Any]		,				A_   :    Tuple    )  ->      Dict:

       """simple docstring"""






       lowerCamelCase_							=       threshold

 def 							a__					(							self   :    List[str]		,				A_   :    List[Any]    )  ->      Tuple:

       """simple docstring"""






       lowerCamelCase_							=       patience

 def 							a__					(							self   :    Optional[Any]    )  ->      Tuple:

       """simple docstring"""






       lowerCamelCase_							=       0
       lowerCamelCase_							=       0

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

       """simple docstring"""






       lowerCamelCase_							=       self.inference_layers_num / self.inference_instances_num
       lowerCamelCase_							=       (
           f"""*** Patience = {self.patience} Avg. Inference Layers = {avg_inf_layers:.2f} Speed Up ="""
           f""" {1 - avg_inf_layers / self.config.num_hidden_layers:.2f} ***"""
       )
       print(A_    )





 @add_start_docstrings_to_model_forward(A_    )
 def 							a__					(							self   :    List[Any]		,				A_   :    List[str]=None		,				A_   :    List[Any]=None		,				A_   :    Optional[int]=None		,				A_   :    Optional[Any]=None		,				A_   :    int=None		,				A_   :    Optional[int]=None		,				A_   :    Optional[Any]=None		,				A_   :    Optional[Any]=None		,				A_   :    List[str]=None		,				A_   :    Optional[Any]=None		,				A_   :    Optional[int]=False		,				)  ->      str:

       """simple docstring"""






       if input_ids is not None and inputs_embeds is not None:
             raise ValueError('You cannot specify both input_ids and inputs_embeds at the same time'    )
       elif input_ids is not None:
             lowerCamelCase_							=       input_ids.size()
       elif inputs_embeds is not None:
             lowerCamelCase_							=       inputs_embeds.size()[:-1]
       else:
             raise ValueError('You have to specify either input_ids or inputs_embeds'    )

       lowerCamelCase_							=       input_ids.device if input_ids is not None else inputs_embeds.device

       if attention_mask is None:
             lowerCamelCase_							=       torch.ones(A_		,				device=A_    )
       if token_type_ids is None:
             lowerCamelCase_							=       torch.zeros(A_		,				dtype=torch.long		,				device=A_    )

       # We can provide a self-attention mask of dimensions [batch_size, from_seq_length, to_seq_length]
       # ourselves in which case we just need to make it broadcastable to all heads.
       lowerCamelCase_							=       self.get_extended_attention_mask(A_		,				A_		,				A_    )

       # If a 2D ou 3D attention mask is provided for the cross-attention
       # we need to make broadcastable to [batch_size, num_heads, seq_length, seq_length]
       if self.config.is_decoder and encoder_hidden_states is not None:
             lowerCamelCase_							=       encoder_hidden_states.size()
             lowerCamelCase_							=       (encoder_batch_size, encoder_sequence_length)
             if encoder_attention_mask is None:
                   lowerCamelCase_							=       torch.ones(A_		,				device=A_    )
             lowerCamelCase_							=       self.invert_attention_mask(A_    )
       else:
             lowerCamelCase_							=       None

       # Prepare head mask if needed
       # 1.0 in head_mask indicate we keep the head
       # attention_probs has shape bsz x n_heads x N x N
       # input head_mask has shape [num_heads] or [num_hidden_layers x num_heads]
       # and head_mask is converted to shape [num_hidden_layers x batch x num_heads x seq_length x seq_length]
       lowerCamelCase_							=       self.get_head_mask(A_		,				self.config.num_hidden_layers    )

       lowerCamelCase_							=       self.embeddings(
           input_ids=A_		,				position_ids=A_		,				token_type_ids=A_		,				inputs_embeds=A_    )
       lowerCamelCase_							=       embedding_output

       if self.training:
             lowerCamelCase_							=       []
             for i in range(self.config.num_hidden_layers    ):
                   lowerCamelCase_							=       self.encoder.adaptive_forward(
                       A_		,				current_layer=A_		,				attention_mask=A_		,				head_mask=A_    )

                   lowerCamelCase_							=       self.pooler(A_    )
                   lowerCamelCase_							=       output_layers[i](output_dropout(A_    )    )
                   res.append(A_    )
       elif self.patience == 0:  # Use all layers for inference
             lowerCamelCase_							=       self.encoder(
                 A_		,				attention_mask=A_		,				head_mask=A_		,				encoder_hidden_states=A_		,				encoder_attention_mask=A_		,				)
             lowerCamelCase_							=       self.pooler(encoder_outputs[0]    )
             lowerCamelCase_							=       [output_layers[self.config.num_hidden_layers - 1](A_    )]
       else:
             lowerCamelCase_							=       0
             lowerCamelCase_							=       None
             lowerCamelCase_							=       0
             for i in range(self.config.num_hidden_layers    ):
                   calculated_layer_num += 1
                   lowerCamelCase_							=       self.encoder.adaptive_forward(
                       A_		,				current_layer=A_		,				attention_mask=A_		,				head_mask=A_    )

                   lowerCamelCase_							=       self.pooler(A_    )
                   lowerCamelCase_							=       output_layers[i](A_    )
                   if regression:
                         lowerCamelCase_							=       logits.detach()
                         if patient_result is not None:
                               lowerCamelCase_							=       patient_result.detach()
                         if (patient_result is not None) and torch.abs(patient_result - labels    ) < self.regression_threshold:
                               patient_counter += 1
                         else:
                               lowerCamelCase_							=       0
                   else:
                         lowerCamelCase_							=       logits.detach().argmax(dim=1    )
                         if patient_result is not None:
                               lowerCamelCase_							=       patient_result.detach().argmax(dim=1    )
                         if (patient_result is not None) and torch.all(labels.eq(A_    )    ):
                               patient_counter += 1
                         else:
                               lowerCamelCase_							=       0

                   lowerCamelCase_							=       logits
                   if patient_counter == self.patience:
                         break
             lowerCamelCase_							=       [patient_result]
             self.inference_layers_num += calculated_layer_num
             self.inference_instances_num += 1

       return res







@add_start_docstrings(
    '''Bert Model transformer with PABEE and a sequence classification/regression head on top (a linear layer on top of
    the pooled output) e.g. for GLUE tasks. '''				, __lowercase				, )
class     A(		__lowercase  ):



 '''simple docstring'''


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

       """simple docstring"""






       super().__init__(A_    )
       lowerCamelCase_							=       config.num_labels

       lowerCamelCase_							=       BertModelWithPabee(A_    )
       lowerCamelCase_							=       nn.Dropout(config.hidden_dropout_prob    )
       lowerCamelCase_							=       nn.ModuleList(
           [nn.Linear(config.hidden_size		,				self.config.num_labels    ) for _ in range(config.num_hidden_layers    )]    )

       self.init_weights()





 @add_start_docstrings_to_model_forward(A_    )
 def 							a__					(							self   :    List[str]		,				A_   :    Dict=None		,				A_   :    str=None		,				A_   :    Union[str, Any]=None		,				A_   :    str=None		,				A_   :    Dict=None		,				A_   :    str=None		,				A_   :    Tuple=None		,				)  ->      Optional[Any]:

       """simple docstring"""






       lowerCamelCase_							=       self.bert(
           input_ids=A_		,				attention_mask=A_		,				token_type_ids=A_		,				position_ids=A_		,				head_mask=A_		,				inputs_embeds=A_		,				output_dropout=self.dropout		,				output_layers=self.classifiers		,				regression=self.num_labels == 1		,				)

       lowerCamelCase_							=       (logits[-1],)

       if labels is not None:
             lowerCamelCase_							=       None
             lowerCamelCase_							=       0
             for ix, logits_item in enumerate(A_    ):
                   if self.num_labels == 1:
                         #  We are doing regression
                         lowerCamelCase_							=       MSELoss()
                         lowerCamelCase_							=       loss_fct(logits_item.view(-1    )		,				labels.view(-1    )    )
                   else:
                         lowerCamelCase_							=       CrossEntropyLoss()
                         lowerCamelCase_							=       loss_fct(logits_item.view(-1		,				self.num_labels    )		,				labels.view(-1    )    )
                   if total_loss is None:
                         lowerCamelCase_							=       loss
                   else:
                         total_loss += loss * (ix + 1)
                   total_weights += ix + 1
             lowerCamelCase_							=       (total_loss / total_weights,) + outputs

       return outputs










'''simple docstring'''






import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding




class  _A (					__lowercase		):
		lowercase__:   Any				     =   ['''image_processor''', '''tokenizer''']
		lowercase__:   Any				     =   '''CLIPImageProcessor'''
		lowercase__:   Optional[Any]				     =   ('''CLIPTokenizer''', '''CLIPTokenizerFast''')

		def __init__(   self					:   int  ,      __magic_name__					:   Dict=None  ,      __magic_name__					:   Dict=None  ,      **__magic_name__					:   Union[str, Any]			)					->    Any:


					"""simple docstring"""

					__snake_case					:      Optional[Any]							      =     None
					if "feature_extractor" in kwargs:
								warnings.warn(
								    """The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"""
								    """ instead."""  ,      __magic_name__  ,      )
								__snake_case					:      List[Any]							      =     kwargs.pop("""feature_extractor"""			)

					__snake_case					:      List[str]							      =     image_processor if image_processor is not None else feature_extractor
					if image_processor is None:
								raise ValueError("""You need to specify an `image_processor`."""			)
					if tokenizer is None:
								raise ValueError("""You need to specify a `tokenizer`."""			)

					super().__init__(__magic_name__  ,      __magic_name__			)

		def __call__(   self					:   int  ,      __magic_name__					:   List[str]=None  ,      __magic_name__					:   Tuple=None  ,      __magic_name__					:   Any=None  ,      **__magic_name__					:   Union[str, Any]			)					->    Optional[int]:


					"""simple docstring"""

					if text is None and images is None:
								raise ValueError("""You have to specify either text or images. Both cannot be none."""			)

					if text is not None:
								__snake_case					:      int							      =     self.tokenizer(__magic_name__  ,      return_tensors=__magic_name__  ,      **__magic_name__			)

					if images is not None:
								__snake_case					:      str							      =     self.image_processor(__magic_name__  ,      return_tensors=__magic_name__  ,      **__magic_name__			)

					if text is not None and images is not None:
								__snake_case					:      Union[str, Any]							      =     image_features.pixel_values
								return encoding
					elif text is not None:
								return encoding
					else:
								return BatchEncoding(data=dict(**__magic_name__			)  ,      tensor_type=__magic_name__			)

		def 	lowercase__     (   self					:   Optional[int]  ,      *__magic_name__					:   List[Any]  ,      **__magic_name__					:   Any			)					->    Optional[Any]:


					"""simple docstring"""

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

		def 	lowercase__     (   self					:   List[str]  ,      *__magic_name__					:   Tuple  ,      **__magic_name__					:   List[Any]			)					->    int:


					"""simple docstring"""

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

		@property
		def 	lowercase__     (   self					:   List[Any]			)					->    Optional[Any]:


					"""simple docstring"""

					__snake_case					:      Dict							      =     self.tokenizer.model_input_names
					__snake_case					:      str							      =     self.image_processor.model_input_names
					return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names			)			)

		@property
		def 	lowercase__     (   self					:   int			)					->    List[str]:


					"""simple docstring"""

					warnings.warn(
					    """`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."""  ,      __magic_name__  ,      )
					return self.image_processor_class

		@property
		def 	lowercase__     (   self					:   Union[str, Any]			)					->    Dict:


					"""simple docstring"""

					warnings.warn(
					    """`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."""  ,      __magic_name__  ,      )
					return self.image_processor


