
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.albert.modeling_albert import (
    ALBERT_INPUTS_DOCSTRING,
    ALBERT_START_DOCSTRING,
    AlbertModel,
    AlbertPreTrainedModel,
    AlbertTransformer,
)


_lowercase				:						Optional[int]												= logging.getLogger(__name__)






class 				lowerCAmelCase__     (						a_  ):






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None	):


											"""simple docstring"""





											if current_layer == 0:
																	lowercase_			:  List[Any]									= self.embedding_hidden_mapping_in(_snake_case	)
											else:
																	lowercase_			:  Tuple									= hidden_states[0]

											lowercase_			:  int									= int(self.config.num_hidden_layers / self.config.num_hidden_groups	)

											# Index of the hidden group
											lowercase_			:  Union[str, Any]									= int(current_layer / (self.config.num_hidden_layers / self.config.num_hidden_groups)	)

											lowercase_			:  Dict									= self.albert_layer_groups[group_idx](
											    _snake_case       ,							_snake_case       ,							head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]       ,							)
											lowercase_			:  Dict									= layer_group_output[0]

											return (hidden_states,)







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






					def __init__(     self       ,							__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											super().__init__(_snake_case	)

											lowercase_			:  int									= AlbertTransformerWithPabee(_snake_case	)

											self.init_weights()
											lowercase_			:  Optional[Any]									= 0
											lowercase_			:  List[str]									= 0
											lowercase_			:  Optional[int]									= 0

											lowercase_			:  Optional[int]									= 0






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											lowercase_			:  Optional[int]									= threshold






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											lowercase_			:  Dict									= patience






					def 			_snake_case					(     self	):


											"""simple docstring"""





											lowercase_			:  List[Any]									= 0
											lowercase_			:  Dict									= 0






					def 			_snake_case					(     self	):


											"""simple docstring"""





											lowercase_			:  Tuple									= self.inference_layers_num / self.inference_instances_num
											lowercase_			:  Optional[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(_snake_case	)






					@add_start_docstrings_to_model_forward(_snake_case	)
					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=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:
																	lowercase_			:  str									= input_ids.size()
											elif inputs_embeds is not None:
																	lowercase_			:  List[Any]									= inputs_embeds.size()[:-1]
											else:
																	raise ValueError('''You have to specify either input_ids or inputs_embeds'''	)

											lowercase_			:  Optional[int]									= input_ids.device if input_ids is not None else inputs_embeds.device

											if attention_mask is None:
																	lowercase_			:  int									= torch.ones(_snake_case       ,							device=_snake_case	)
											if token_type_ids is None:
																	lowercase_			:  List[Any]									= torch.zeros(_snake_case       ,							dtype=torch.long       ,							device=_snake_case	)

											lowercase_			:  Dict									= attention_mask.unsqueeze(1	).unsqueeze(2	)
											lowercase_			:  List[str]									= extended_attention_mask.to(dtype=self.dtype	)  # fp16 compatibility
											lowercase_			:  Optional[int]									= (1.0 - extended_attention_mask) * -1_00_00.0
											lowercase_			:  List[Any]									= self.get_head_mask(_snake_case       ,							self.config.num_hidden_layers	)

											lowercase_			:  Dict									= self.embeddings(
											    _snake_case       ,							position_ids=_snake_case       ,							token_type_ids=_snake_case       ,							inputs_embeds=_snake_case	)
											lowercase_			:  List[Any]									= embedding_output

											if self.training:
																	lowercase_			:  Optional[int]									= []
																	for i in range(self.config.num_hidden_layers	):
																							lowercase_			:  List[Any]									= self.encoder.adaptive_forward(
																							    _snake_case       ,							current_layer=_snake_case       ,							attention_mask=_snake_case       ,							head_mask=_snake_case       ,							)

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

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

																							lowercase_			:  Optional[int]									= logits
																							if patient_counter == self.patience:
																													break
																	lowercase_			:  Any									= [patient_result]
																	self.inference_layers_num += calculated_layer_num
																	self.inference_instances_num += 1

											return res







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






					def __init__(     self       ,							__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											super().__init__(_snake_case	)
											lowercase_			:  Any									= config.num_labels

											lowercase_			:  Optional[int]									= AlbertModelWithPabee(_snake_case	)
											lowercase_			:  Dict									= nn.Dropout(config.classifier_dropout_prob	)
											lowercase_			:  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(_snake_case	)
					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							):


											"""simple docstring"""





											lowercase_			:  List[str]									= self.albert(
											    input_ids=_snake_case       ,							attention_mask=_snake_case       ,							token_type_ids=_snake_case       ,							position_ids=_snake_case       ,							head_mask=_snake_case       ,							inputs_embeds=_snake_case       ,							output_dropout=self.dropout       ,							output_layers=self.classifiers       ,							regression=self.num_labels == 1       ,							)

											lowercase_			:  int									= (logits[-1],)

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

											return outputs





'''simple docstring'''







from typing import List

from ...processing_utils import ProcessorMixin
from ...utils import is_torch_available


if is_torch_available():
							import torch






class 				lowerCAmelCase__     (						lowerCamelCase_  ):
					lowerCAmelCase_       =  ['''image_processor''', '''tokenizer''']
					lowerCAmelCase_       =  '''OneFormerImageProcessor'''
					lowerCAmelCase_       =  ('''CLIPTokenizer''', '''CLIPTokenizerFast''')






					def __init__(     self       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE = 77       ,							__SCREAMING_SNAKE_CASE = 77       ,							**__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											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`.'''	)

											lowercase_			:  Dict									= max_seq_length
											lowercase_			:  Dict									= task_seq_length

											super().__init__(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	)






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=77	):


											"""simple docstring"""





											if text_list is None:
																	raise ValueError('''tokens cannot be None.'''	)

											lowercase_			:  str									= self.tokenizer(__SCREAMING_SNAKE_CASE       ,							padding='''max_length'''       ,							max_length=__SCREAMING_SNAKE_CASE       ,							truncation=__SCREAMING_SNAKE_CASE	)

											lowercase_		,				lowercase_			:  List[str]									= tokens['''attention_mask'''], tokens['''input_ids''']

											lowercase_			:  List[str]									= []
											for attn_mask, input_id in zip(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	):
																	lowercase_			:  Optional[Any]									= torch.tensor(__SCREAMING_SNAKE_CASE	) * torch.tensor(__SCREAMING_SNAKE_CASE	)
																	token_inputs.append(token.unsqueeze(0	)	)

											lowercase_			:  str									= torch.cat(__SCREAMING_SNAKE_CASE       ,							dim=0	)
											return token_inputs






					def __call__(     self       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							**__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											if task_inputs is None:
																	raise ValueError('''You have to specify the task_input. Found None.'''	)
											elif images is None:
																	raise ValueError('''You have to specify the image. Found None.'''	)

											if not all(task in ['''semantic''', '''instance''', '''panoptic'''] for task in task_inputs	):
																	raise ValueError('''task_inputs must be semantic, instance, or panoptic.'''	)

											lowercase_			:  Optional[int]									= self.image_processor(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	)

											if isinstance(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	):
																	lowercase_			:  List[str]									= [task_inputs]

											if isinstance(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	) and all(isinstance(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	) for task_input in task_inputs	):
																	lowercase_			:  List[str]									= []
																	for task in task_inputs:
																							lowercase_			:  Tuple									= F'''the task is {task}'''
																							task_token_inputs.append(__SCREAMING_SNAKE_CASE	)
																	lowercase_			:  Tuple									= self._preprocess_text(__SCREAMING_SNAKE_CASE       ,							max_length=self.task_seq_length	)
											else:
																	raise TypeError('''Task Inputs should be a string or a list of strings.'''	)

											if hasattr(__SCREAMING_SNAKE_CASE       ,							'''text_inputs'''	):
																	lowercase_			:  Any									= encoded_inputs.text_inputs

																	lowercase_			:  int									= []
																	for texts in texts_list:
																							lowercase_			:  Any									= self._preprocess_text(__SCREAMING_SNAKE_CASE       ,							max_length=self.max_seq_length	)
																							text_inputs.append(text_input_list.unsqueeze(0	)	)

																	lowercase_			:  Tuple									= torch.cat(__SCREAMING_SNAKE_CASE       ,							dim=0	)

											return encoded_inputs






					def 			_snake_case					(     self       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							__SCREAMING_SNAKE_CASE=None       ,							**__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											if task_inputs is None:
																	raise ValueError('''You have to specify the task_input. Found None.'''	)
											elif images is None:
																	raise ValueError('''You have to specify the image. Found None.'''	)

											if not all(task in ['''semantic''', '''instance''', '''panoptic'''] for task in task_inputs	):
																	raise ValueError('''task_inputs must be semantic, instance, or panoptic.'''	)

											lowercase_			:  Dict									= self.image_processor.encode_inputs(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	)

											if isinstance(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	):
																	lowercase_			:  List[Any]									= [task_inputs]

											if isinstance(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	) and all(isinstance(__SCREAMING_SNAKE_CASE       ,							__SCREAMING_SNAKE_CASE	) for task_input in task_inputs	):
																	lowercase_			:  int									= []
																	for task in task_inputs:
																							lowercase_			:  int									= F'''the task is {task}'''
																							task_token_inputs.append(__SCREAMING_SNAKE_CASE	)
																	lowercase_			:  str									= self._preprocess_text(__SCREAMING_SNAKE_CASE       ,							max_length=self.task_seq_length	)
											else:
																	raise TypeError('''Task Inputs should be a string or a list of strings.'''	)

											if hasattr(__SCREAMING_SNAKE_CASE       ,							'''text_inputs'''	):
																	lowercase_			:  Dict									= encoded_inputs.text_inputs

																	lowercase_			:  str									= []
																	for texts in texts_list:
																							lowercase_			:  List[str]									= self._preprocess_text(__SCREAMING_SNAKE_CASE       ,							max_length=self.max_seq_length	)
																							text_inputs.append(text_input_list.unsqueeze(0	)	)

																	lowercase_			:  Union[str, Any]									= torch.cat(__SCREAMING_SNAKE_CASE       ,							dim=0	)

											return encoded_inputs






					def 			_snake_case					(     self       ,							*__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											return self.image_processor.post_process_semantic_segmentation(*__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	)






					def 			_snake_case					(     self       ,							*__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											return self.image_processor.post_process_instance_segmentation(*__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	)






					def 			_snake_case					(     self       ,							*__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	):


											"""simple docstring"""





											return self.image_processor.post_process_panoptic_segmentation(*__SCREAMING_SNAKE_CASE       ,							**__SCREAMING_SNAKE_CASE	)



