




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


_lowerCamelCase  :			Optional[Any]        =       logging.getLogger(__name__)

class        lowercase      (    __UpperCAmelCase):







				def 					a_  (							self   :					Optional[Any]      ,     _lowerCamelCase   :					Union[str, Any]      ,     _lowerCamelCase   :					Any      ,     _lowerCamelCase   :					List[Any]=None      ,     _lowerCamelCase   :					List[Any]=None       ):



					"""simple docstring"""

					A_  :				Union[str, Any]	   =		self.layer[current_layer](_lowerCamelCase      ,     _lowerCamelCase      ,     head_mask[current_layer]       )

					A_  :				List[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."""     ,	__UpperCAmelCase     ,	)
class        lowercase      (    __UpperCAmelCase):







				def __init__(							self   :					List[Any]      ,     _lowerCamelCase   :					int       ):



					"""simple docstring"""

					super().__init__(_lowerCamelCase       )

					A_  :				Optional[Any]	   =		BertEncoderWithPabee(_lowerCamelCase       )

					self.init_weights()
					A_  :				List[str]	   =		0
					A_  :				Optional[Any]	   =		0
					A_  :				Union[str, Any]	   =		0

					A_  :				Optional[Any]	   =		0







				def 					a_  (							self   :					Any      ,     _lowerCamelCase   :					int       ):



					"""simple docstring"""

					A_  :				Optional[Any]	   =		threshold







				def 					a_  (							self   :					Optional[int]      ,     _lowerCamelCase   :					List[Any]       ):



					"""simple docstring"""

					A_  :				Union[str, Any]	   =		patience







				def 					a_  (							self   :					str       ):



					"""simple docstring"""

					A_  :				Optional[Any]	   =		0
					A_  :				Any	   =		0







				def 					a_  (							self   :					Optional[int]       ):



					"""simple docstring"""

					A_  :				List[str]	   =		self.inference_layers_num / self.inference_instances_num
					A_  :				Tuple	   =		(
					    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(_lowerCamelCase       )







				@add_start_docstrings_to_model_forward(_lowerCamelCase       )
				def 					a_  (							self   :					str      ,     _lowerCamelCase   :					Dict=None      ,     _lowerCamelCase   :					List[str]=None      ,     _lowerCamelCase   :					Any=None      ,     _lowerCamelCase   :					Tuple=None      ,     _lowerCamelCase   :					str=None      ,     _lowerCamelCase   :					List[str]=None      ,     _lowerCamelCase   :					Optional[int]=None      ,     _lowerCamelCase   :					List[Any]=None      ,     _lowerCamelCase   :					List[str]=None      ,     _lowerCamelCase   :					List[str]=None      ,     _lowerCamelCase   :					Optional[int]=False      ,     ):



					"""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:
						A_  :				Tuple	   =		input_ids.size()
					elif inputs_embeds is not None:
						A_  :				List[str]	   =		inputs_embeds.size()[:-1]
					else:
						raise ValueError('''You have to specify either input_ids or inputs_embeds'''       )

					A_  :				str	   =		input_ids.device if input_ids is not None else inputs_embeds.device

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

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

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

					A_  :				Optional[Any]	   =		self.embeddings(
					    input_ids=_lowerCamelCase      ,     position_ids=_lowerCamelCase      ,     token_type_ids=_lowerCamelCase      ,     inputs_embeds=_lowerCamelCase       )
					A_  :				List[str]	   =		embedding_output

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

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

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

							A_  :				int	   =		logits
							if patient_counter == self.patience:
								break
						A_  :				Optional[Any]	   =		[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. """     ,	__UpperCAmelCase     ,	)
class        lowercase      (    __UpperCAmelCase):







				def __init__(							self   :					Any      ,     _lowerCamelCase   :					Union[str, Any]       ):



					"""simple docstring"""

					super().__init__(_lowerCamelCase       )
					A_  :				Optional[Any]	   =		config.num_labels

					A_  :				Optional[int]	   =		BertModelWithPabee(_lowerCamelCase       )
					A_  :				int	   =		nn.Dropout(config.hidden_dropout_prob       )
					A_  :				Optional[int]	   =		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(_lowerCamelCase       )
				def 					a_  (							self   :					str      ,     _lowerCamelCase   :					Optional[int]=None      ,     _lowerCamelCase   :					List[Any]=None      ,     _lowerCamelCase   :					Tuple=None      ,     _lowerCamelCase   :					str=None      ,     _lowerCamelCase   :					Dict=None      ,     _lowerCamelCase   :					Any=None      ,     _lowerCamelCase   :					Dict=None      ,     ):



					"""simple docstring"""

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

					A_  :				Union[str, Any]	   =		(logits[-1],)

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

					return outputs











"""simple docstring"""

from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available, is_vision_available


_lowerCamelCase  :			Optional[Any]        =       {
    'configuration_mask2former': [
        'MASK2FORMER_PRETRAINED_CONFIG_ARCHIVE_MAP',
        'Mask2FormerConfig',
    ],
}

try:
		if not is_vision_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_lowerCamelCase  :			int        =       ['Mask2FormerImageProcessor']

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_lowerCamelCase  :			Tuple        =       [
		    'MASK2FORMER_PRETRAINED_MODEL_ARCHIVE_LIST',
		    'Mask2FormerForUniversalSegmentation',
		    'Mask2FormerModel',
		    'Mask2FormerPreTrainedModel',
		]

if TYPE_CHECKING:
		from .configuration_maskaformer import MASK2FORMER_PRETRAINED_CONFIG_ARCHIVE_MAP, MaskaFormerConfig

		try:
				if not is_vision_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .image_processing_maskaformer import MaskaFormerImageProcessor

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_maskaformer import (
				    MASK2FORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
				    MaskaFormerForUniversalSegmentation,
				    MaskaFormerModel,
				    MaskaFormerPreTrainedModel,
				)


else:
		import sys

		_lowerCamelCase  :			Optional[Any]        =       _LazyModule(__name__, globals()['__file__'], _import_structure)





