





'''simple docstring'''






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,
)


__UpperCamelCase					    =	logging.getLogger(__name__)




class  _A (					__lowercase		):

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


					"""simple docstring"""

					__snake_case					:      List[Any]							      =     self.layer[current_layer](__magic_name__  ,      __magic_name__  ,      head_mask[current_layer]			)

					__snake_case					:      Any							      =     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		):

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


					"""simple docstring"""

					super().__init__(__magic_name__			)

					__snake_case					:      Optional[Any]							      =     BertEncoderWithPabee(__magic_name__			)

					self.init_weights()
					__snake_case					:      Union[str, Any]							      =     0
					__snake_case					:      Union[str, Any]							      =     0
					__snake_case					:      Optional[int]							      =     0

					__snake_case					:      List[str]							      =     0

		def 	lowercase__     (   self					:   Union[str, Any]  ,      __magic_name__					:   Tuple			)					->    Dict:


					"""simple docstring"""

					__snake_case					:      Tuple							      =     threshold

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


					"""simple docstring"""

					__snake_case					:      List[Any]							      =     patience

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


					"""simple docstring"""

					__snake_case					:      Optional[Any]							      =     0
					__snake_case					:      Dict							      =     0

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


					"""simple docstring"""

					__snake_case					:      Optional[Any]							      =     self.inference_layers_num / self.inference_instances_num
					__snake_case					:      Any							      =     (
					    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(__magic_name__			)

		@add_start_docstrings_to_model_forward(__magic_name__			)
		def 	lowercase__     (   self					:   List[Any]  ,      __magic_name__					:   List[str]=None  ,      __magic_name__					:   List[Any]=None  ,      __magic_name__					:   Optional[int]=None  ,      __magic_name__					:   Optional[Any]=None  ,      __magic_name__					:   int=None  ,      __magic_name__					:   Optional[int]=None  ,      __magic_name__					:   Optional[Any]=None  ,      __magic_name__					:   Optional[Any]=None  ,      __magic_name__					:   List[str]=None  ,      __magic_name__					:   Optional[Any]=None  ,      __magic_name__					:   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:
								__snake_case					:      Union[str, Any]							      =     input_ids.size()
					elif inputs_embeds is not None:
								__snake_case					:      List[str]							      =     inputs_embeds.size()[:-1]
					else:
								raise ValueError("""You have to specify either input_ids or inputs_embeds"""			)

					__snake_case					:      Tuple							      =     input_ids.device if input_ids is not None else inputs_embeds.device

					if attention_mask is None:
								__snake_case					:      int							      =     torch.ones(__magic_name__  ,      device=__magic_name__			)
					if token_type_ids is None:
								__snake_case					:      int							      =     torch.zeros(__magic_name__  ,      dtype=torch.long  ,      device=__magic_name__			)

					# 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.
					__snake_case					:      torch.Tensor							      =     self.get_extended_attention_mask(__magic_name__  ,      __magic_name__  ,      __magic_name__			)

					# 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:
								__snake_case      ,     __snake_case      ,     __snake_case					:      str							      =     encoder_hidden_states.size()
								__snake_case					:      List[Any]							      =     (encoder_batch_size, encoder_sequence_length)
								if encoder_attention_mask is None:
											__snake_case					:      List[Any]							      =     torch.ones(__magic_name__  ,      device=__magic_name__			)
								__snake_case					:      Optional[Any]							      =     self.invert_attention_mask(__magic_name__			)
					else:
								__snake_case					:      Any							      =     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]
					__snake_case					:      List[Any]							      =     self.get_head_mask(__magic_name__  ,      self.config.num_hidden_layers			)

					__snake_case					:      Optional[Any]							      =     self.embeddings(
					    input_ids=__magic_name__  ,      position_ids=__magic_name__  ,      token_type_ids=__magic_name__  ,      inputs_embeds=__magic_name__			)
					__snake_case					:      Optional[int]							      =     embedding_output

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

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

											__snake_case					:      Union[str, Any]							      =     self.pooler(__magic_name__			)
											__snake_case					:      str							      =     output_layers[i](__magic_name__			)
											if regression:
														__snake_case					:      str							      =     logits.detach()
														if patient_result is not None:
																	__snake_case					:      Any							      =     patient_result.detach()
														if (patient_result is not None) and torch.abs(patient_result - labels			) < self.regression_threshold:
																	patient_counter += 1
														else:
																	__snake_case					:      Any							      =     0
											else:
														__snake_case					:      int							      =     logits.detach().argmax(dim=1			)
														if patient_result is not None:
																	__snake_case					:      Any							      =     patient_result.detach().argmax(dim=1			)
														if (patient_result is not None) and torch.all(labels.eq(__magic_name__			)			):
																	patient_counter += 1
														else:
																	__snake_case					:      Tuple							      =     0

											__snake_case					:      Optional[int]							      =     logits
											if patient_counter == self.patience:
														break
								__snake_case					:      Tuple							      =     [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		):

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


					"""simple docstring"""

					super().__init__(__magic_name__			)
					__snake_case					:      List[str]							      =     config.num_labels

					__snake_case					:      Union[str, Any]							      =     BertModelWithPabee(__magic_name__			)
					__snake_case					:      str							      =     nn.Dropout(config.hidden_dropout_prob			)
					__snake_case					:      Union[str, Any]							      =     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(__magic_name__			)
		def 	lowercase__     (   self					:   List[str]  ,      __magic_name__					:   Dict=None  ,      __magic_name__					:   str=None  ,      __magic_name__					:   Union[str, Any]=None  ,      __magic_name__					:   str=None  ,      __magic_name__					:   Dict=None  ,      __magic_name__					:   str=None  ,      __magic_name__					:   Tuple=None  ,      )					->    Optional[Any]:


					"""simple docstring"""

					__snake_case					:      Any							      =     self.bert(
					    input_ids=__magic_name__  ,      attention_mask=__magic_name__  ,      token_type_ids=__magic_name__  ,      position_ids=__magic_name__  ,      head_mask=__magic_name__  ,      inputs_embeds=__magic_name__  ,      output_dropout=self.dropout  ,      output_layers=self.classifiers  ,      regression=self.num_labels == 1  ,      )

					__snake_case					:      Optional[int]							      =     (logits[-1],)

					if labels is not None:
								__snake_case					:      List[str]							      =     None
								__snake_case					:      Optional[Any]							      =     0
								for ix, logits_item in enumerate(__magic_name__			):
											if self.num_labels == 1:
														#  We are doing regression
														__snake_case					:      Optional[int]							      =     MSELoss()
														__snake_case					:      List[Any]							      =     loss_fct(logits_item.view(-1			)  ,      labels.view(-1			)			)
											else:
														__snake_case					:      Union[str, Any]							      =     CrossEntropyLoss()
														__snake_case					:      List[Any]							      =     loss_fct(logits_item.view(-1  ,      self.num_labels			)  ,      labels.view(-1			)			)
											if total_loss is None:
														__snake_case					:      List[Any]							      =     loss
											else:
														total_loss += loss * (ix + 1)
											total_weights += ix + 1
								__snake_case					:      List[Any]							      =     (total_loss / total_weights,) + outputs

					return outputs










"""simple docstring"""
import warnings
from typing import Dict

import numpy as np

from ..utils import ExplicitEnum, add_end_docstrings, is_tf_available, is_torch_available
from .base import PIPELINE_INIT_ARGS, GenericTensor, Pipeline


if is_tf_available():
							from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING

if is_torch_available():
							from ..models.auto.modeling_auto import MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING






def       __UpperCamelCase						(		snake_case__       ):
	return 1.0 / (1.0 + np.exp(-_outputs       ))






def       __UpperCamelCase						(		snake_case__       ):
	A_		:    Union[str, Any]						     = np.max(_outputs , axis=-1 , keepdims=snake_case__       )
	A_		:    str						     = np.exp(_outputs - maxes       )
	return shifted_exp / shifted_exp.sum(axis=-1 , keepdims=snake_case__       )



class   SCREAMING_SNAKE_CASE	(					_SCREAMING_SNAKE_CASE    ):





	"""simple docstring"""




	_A							:					Any 			=					"""sigmoid"""
	_A							:					Any 			=					"""softmax"""
	_A							:					Union[str, Any] 			=					"""none"""






@add_end_docstrings(
    _SCREAMING_SNAKE_CASE							,							R"""
        return_all_scores (`bool`, *optional*, defaults to `False`):
            Whether to return all prediction scores or just the one of the predicted class.
        function_to_apply (`str`, *optional*, defaults to `\"default\"`):
            The function to apply to the model outputs in order to retrieve the scores. Accepts four different values:

            - `\"default\"`: if the model has a single label, will apply the sigmoid function on the output. If the model
              has several labels, will apply the softmax function on the output.
            - `\"sigmoid\"`: Applies the sigmoid function on the output.
            - `\"softmax\"`: Applies the softmax function on the output.
            - `\"none\"`: Does not apply any function on the output.
    """							,							)
class   SCREAMING_SNAKE_CASE	(					_SCREAMING_SNAKE_CASE    ):





	"""simple docstring"""




	_A							:					Optional[int] 			=					False
	_A							:					Dict 			=					ClassificationFunction.NONE





	def __init__(self  ,  **lowerCAmelCase_				):
		super().__init__(**lowerCAmelCase_				)

		self.check_model_type(
		    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING
		    if self.framework == """tf"""
		    else MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING				)





	def  lowerCamelCase(self  ,  lowerCAmelCase_=None  ,  lowerCAmelCase_=None  ,  lowerCAmelCase_=""  ,  **lowerCAmelCase_				):
		# Using "" as default argument because we're going to use `top_k=None` in user code to declare
		# "No top_k"
		A_		:    Union[str, Any]						     = tokenizer_kwargs

		A_		:    List[str]						     = {}
		if hasattr(self.model.config  ,  """return_all_scores"""				) and return_all_scores is None:
			A_		:    Optional[int]						     = self.model.config.return_all_scores

		if isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				) or top_k is None:
			A_		:    Dict						     = top_k
			A_		:    Any						     = False
		elif return_all_scores is not None:
			warnings.warn(
			    """`return_all_scores` is now deprecated,  if want a similar functionality use `top_k=None` instead of"""
			    """ `return_all_scores=True` or `top_k=1` instead of `return_all_scores=False`."""  ,  lowerCAmelCase_  ,  )
			if return_all_scores:
				A_		:    List[Any]						     = None
			else:
				A_		:    List[str]						     = 1

		if isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				):
			A_		:    str						     = ClassificationFunction[function_to_apply.upper()]

		if function_to_apply is not None:
			A_		:    List[str]						     = function_to_apply
		return preprocess_params, {}, postprocess_params





	def __call__(self  ,  *lowerCAmelCase_  ,  **lowerCAmelCase_				):
		A_		:    List[str]						     = super().__call__(*lowerCAmelCase_  ,  **lowerCAmelCase_				)
		# TODO try and retrieve it in a nicer way from _sanitize_parameters.
		A_		:    Union[str, Any]						     = """top_k""" not in kwargs
		if isinstance(args[0]  ,  lowerCAmelCase_				) and _legacy:
			# This pipeline is odd, and return a list when single item is run
			return [result]
		else:
			return result





	def  lowerCamelCase(self  ,  lowerCAmelCase_  ,  **lowerCAmelCase_				):
		A_		:    Union[str, Any]						     = self.framework
		if isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				):
			return self.tokenizer(**lowerCAmelCase_  ,  return_tensors=lowerCAmelCase_  ,  **lowerCAmelCase_				)
		elif isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				) and len(lowerCAmelCase_				) == 1 and isinstance(inputs[0]  ,  lowerCAmelCase_				) and len(inputs[0]				) == 2:
			# It used to be valid to use a list of list of list for text pairs, keeping this path for BC
			return self.tokenizer(
			    text=inputs[0][0]  ,  text_pair=inputs[0][1]  ,  return_tensors=lowerCAmelCase_  ,  **lowerCAmelCase_				)
		elif isinstance(lowerCAmelCase_  ,  lowerCAmelCase_				):
			# This is likely an invalid usage of the pipeline attempting to pass text pairs.
			raise ValueError(
			    """The pipeline received invalid inputs, if you are trying to send text pairs, you can try to send a"""
			    """ dictionary `{\"text\": \"My text\", \"text_pair\": \"My pair\"}` in order to send a text pair."""				)
		return self.tokenizer(lowerCAmelCase_  ,  return_tensors=lowerCAmelCase_  ,  **lowerCAmelCase_				)





	def  lowerCamelCase(self  ,  lowerCAmelCase_				):
		return self.model(**lowerCAmelCase_				)





	def  lowerCamelCase(self  ,  lowerCAmelCase_  ,  lowerCAmelCase_=None  ,  lowerCAmelCase_=1  ,  lowerCAmelCase_=True				):
		# `_legacy` is used to determine if we're running the naked pipeline and in backward
		# compatibility mode, or if running the pipeline with `pipeline(..., top_k=1)` we're running
		# the more natural result containing the list.
		# Default value before `set_parameters`
		if function_to_apply is None:
			if self.model.config.problem_type == "multi_label_classification" or self.model.config.num_labels == 1:
				A_		:    Optional[Any]						     = ClassificationFunction.SIGMOID
			elif self.model.config.problem_type == "single_label_classification" or self.model.config.num_labels > 1:
				A_		:    Tuple						     = ClassificationFunction.SOFTMAX
			elif hasattr(self.model.config  ,  """function_to_apply"""				) and function_to_apply is None:
				A_		:    Any						     = self.model.config.function_to_apply
			else:
				A_		:    Dict						     = ClassificationFunction.NONE

		A_		:    Optional[Any]						     = model_outputs["""logits"""][0]
		A_		:    Tuple						     = outputs.numpy()

		if function_to_apply == ClassificationFunction.SIGMOID:
			A_		:    str						     = sigmoid(lowerCAmelCase_				)
		elif function_to_apply == ClassificationFunction.SOFTMAX:
			A_		:    Dict						     = softmax(lowerCAmelCase_				)
		elif function_to_apply == ClassificationFunction.NONE:
			A_		:    Optional[int]						     = outputs
		else:
			raise ValueError(f"""Unrecognized `function_to_apply` argument: {function_to_apply}"""				)

		if top_k == 1 and _legacy:
			return {"label": self.model.config.idalabel[scores.argmax().item()], "score": scores.max().item()}

		A_		:    Optional[Any]						     = [
		    {"""label""": self.model.config.idalabel[i], """score""": score.item()} for i, score in enumerate(lowerCAmelCase_				)
		]
		if not _legacy:
			dict_scores.sort(key=lambda lowerCAmelCase_				: x["score"]  ,  reverse=lowerCAmelCase_				)
			if top_k is not None:
				A_		:    str						     = dict_scores[:top_k]
		return dict_scores

