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


__A					=			logging.getLogger(__name__)







class 						lowerCamelCase__   (   __magic_name__							):





			'''simple docstring'''



			def    _lowerCAmelCase						(				self	,	__UpperCAmelCase	,	__UpperCAmelCase	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None    )					->	str:
				if current_layer == 0:
					_lowerCAmelCase								=self.embedding_hidden_mapping_in(__UpperCAmelCase    )
				else:
					_lowerCAmelCase								=hidden_states[0]

				_lowerCAmelCase								=int(self.config.num_hidden_layers / self.config.num_hidden_groups    )

				# Index of the hidden group
				_lowerCAmelCase								=int(current_layer / (self.config.num_hidden_layers / self.config.num_hidden_groups)    )

				_lowerCAmelCase								=self.albert_layer_groups[group_idx](
				    __UpperCAmelCase	,	__UpperCAmelCase	,	head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]	,	)
				_lowerCAmelCase								=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.'''					,      __magic_name__					,      )
class 						lowerCamelCase__   (   __magic_name__							):





			'''simple docstring'''



			def __init__(				self	,	__UpperCAmelCase    )					->	Union[str, Any]:
				super().__init__(__UpperCAmelCase    )

				_lowerCAmelCase								=AlbertTransformerWithPabee(__UpperCAmelCase    )

				self.init_weights()
				_lowerCAmelCase								=0
				_lowerCAmelCase								=0
				_lowerCAmelCase								=0

				_lowerCAmelCase								=0



			def    _lowerCAmelCase						(				self	,	__UpperCAmelCase    )					->	List[Any]:
				_lowerCAmelCase								=threshold



			def    _lowerCAmelCase						(				self	,	__UpperCAmelCase    )					->	str:
				_lowerCAmelCase								=patience



			def    _lowerCAmelCase						(				self    )					->	str:
				_lowerCAmelCase								=0
				_lowerCAmelCase								=0



			def    _lowerCAmelCase						(				self    )					->	int:
				_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(__UpperCAmelCase    )



			@add_start_docstrings_to_model_forward(__UpperCAmelCase    )
			def    _lowerCAmelCase						(				self	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=False	,	)					->	Any:

				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(__UpperCAmelCase	,	device=__UpperCAmelCase    )
				if token_type_ids is None:
					_lowerCAmelCase								=torch.zeros(__UpperCAmelCase	,	dtype=torch.long	,	device=__UpperCAmelCase    )

				_lowerCAmelCase								=attention_mask.unsqueeze(1    ).unsqueeze(2    )
				_lowerCAmelCase								=extended_attention_mask.to(dtype=self.dtype    )  # fp16 compatibility
				_lowerCAmelCase								=(1.0 - extended_attention_mask) * -1_0_0_0_0.0
				_lowerCAmelCase								=self.get_head_mask(__UpperCAmelCase	,	self.config.num_hidden_layers    )

				_lowerCAmelCase								=self.embeddings(
				    __UpperCAmelCase	,	position_ids=__UpperCAmelCase	,	token_type_ids=__UpperCAmelCase	,	inputs_embeds=__UpperCAmelCase    )
				_lowerCAmelCase								=embedding_output

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

						_lowerCAmelCase								=self.pooler_activation(self.pooler(encoder_outputs[0][:, 0]    )    )
						_lowerCAmelCase								=output_layers[i](output_dropout(__UpperCAmelCase    )    )
						res.append(__UpperCAmelCase    )
				elif self.patience == 0:  # Use all layers for inference
					_lowerCAmelCase								=self.encoder(__UpperCAmelCase	,	__UpperCAmelCase	,	head_mask=__UpperCAmelCase    )
					_lowerCAmelCase								=self.pooler_activation(self.pooler(encoder_outputs[0][:, 0]    )    )
					_lowerCAmelCase								=[output_layers[self.config.num_hidden_layers - 1](__UpperCAmelCase    )]
				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(
						    __UpperCAmelCase	,	current_layer=__UpperCAmelCase	,	attention_mask=__UpperCAmelCase	,	head_mask=__UpperCAmelCase	,	)

						_lowerCAmelCase								=self.pooler_activation(self.pooler(encoder_outputs[0][:, 0]    )    )
						_lowerCAmelCase								=output_layers[i](__UpperCAmelCase    )
						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(__UpperCAmelCase    )    ):
								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(
    '''Albert 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. '''					,      __magic_name__					,      )
class 						lowerCamelCase__   (   __magic_name__							):





			'''simple docstring'''



			def __init__(				self	,	__UpperCAmelCase    )					->	str:
				super().__init__(__UpperCAmelCase    )
				_lowerCAmelCase								=config.num_labels

				_lowerCAmelCase								=AlbertModelWithPabee(__UpperCAmelCase    )
				_lowerCAmelCase								=nn.Dropout(config.classifier_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(__UpperCAmelCase    )
			def    _lowerCAmelCase						(				self	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	__UpperCAmelCase=None	,	)					->	Union[str, Any]:
				_lowerCAmelCase								=self.albert(
				    input_ids=__UpperCAmelCase	,	attention_mask=__UpperCAmelCase	,	token_type_ids=__UpperCAmelCase	,	position_ids=__UpperCAmelCase	,	head_mask=__UpperCAmelCase	,	inputs_embeds=__UpperCAmelCase	,	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(__UpperCAmelCase    ):
						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 unittest

import numpy as np
import PIL
import torch

from diffusers.image_processor import VaeImageProcessor







class 						lowerCamelCase__   (   unittest.TestCase							):





			'''simple docstring'''



			@property
			def    _lowerCAmelCase						(				self    )					->	List[str]:
				_lowerCAmelCase								=1
				_lowerCAmelCase								=3
				_lowerCAmelCase								=8
				_lowerCAmelCase								=8

				_lowerCAmelCase								=torch.rand((batch_size, num_channels, height, width)    )

				return sample



			def    _lowerCAmelCase						(				self	,	__UpperCAmelCase    )					->	Any:
				if isinstance(image[0]	,	PIL.Image.Image    ):
					return np.stack([np.array(__UpperCAmelCase    ) for i in image]	,	axis=0    )
				elif isinstance(__UpperCAmelCase	,	torch.Tensor    ):
					return image.cpu().numpy().transpose(0	,	2	,	3	,	1    )
				return image



			def    _lowerCAmelCase						(				self    )					->	Tuple:
				_lowerCAmelCase								=VaeImageProcessor(do_resize=__UpperCAmelCase	,	do_normalize=__UpperCAmelCase    )

				_lowerCAmelCase								=self.dummy_sample
				_lowerCAmelCase								=self.to_np(__UpperCAmelCase    )

				for output_type in ["pt", "np", "pil"]:
					_lowerCAmelCase								=image_processor.postprocess(
					    image_processor.preprocess(__UpperCAmelCase    )	,	output_type=__UpperCAmelCase	,	)
					_lowerCAmelCase								=self.to_np(__UpperCAmelCase    )
					_lowerCAmelCase								=(input_np * 2_55).round() if output_type == """pil""" else input_np
					assert (
					    np.abs(in_np - out_np    ).max() < 1e-6
					), f'''decoded output does not match input for output_type {output_type}'''



			def    _lowerCAmelCase						(				self    )					->	Union[str, Any]:
				_lowerCAmelCase								=VaeImageProcessor(do_resize=__UpperCAmelCase	,	do_normalize=__UpperCAmelCase    )
				_lowerCAmelCase								=self.dummy_sample.cpu().numpy().transpose(0	,	2	,	3	,	1    )

				for output_type in ["pt", "np", "pil"]:
					_lowerCAmelCase								=image_processor.postprocess(image_processor.preprocess(__UpperCAmelCase    )	,	output_type=__UpperCAmelCase    )

					_lowerCAmelCase								=self.to_np(__UpperCAmelCase    )
					_lowerCAmelCase								=(input_np * 2_55).round() if output_type == """pil""" else input_np
					assert (
					    np.abs(in_np - out_np    ).max() < 1e-6
					), f'''decoded output does not match input for output_type {output_type}'''



			def    _lowerCAmelCase						(				self    )					->	List[str]:
				_lowerCAmelCase								=VaeImageProcessor(do_resize=__UpperCAmelCase	,	do_normalize=__UpperCAmelCase    )

				_lowerCAmelCase								=self.dummy_sample.cpu().numpy().transpose(0	,	2	,	3	,	1    )
				_lowerCAmelCase								=image_processor.numpy_to_pil(__UpperCAmelCase    )

				for output_type in ["pt", "np", "pil"]:
					_lowerCAmelCase								=image_processor.postprocess(image_processor.preprocess(__UpperCAmelCase    )	,	output_type=__UpperCAmelCase    )
					for i, o in zip(__UpperCAmelCase	,	__UpperCAmelCase    ):
						_lowerCAmelCase								=np.array(__UpperCAmelCase    )
						_lowerCAmelCase								=self.to_np(__UpperCAmelCase    ) if output_type == """pil""" else (self.to_np(__UpperCAmelCase    ) * 2_55).round()
						assert (
						    np.abs(in_np - out_np    ).max() < 1e-6
						), f'''decoded output does not match input for output_type {output_type}'''



			def    _lowerCAmelCase						(				self    )					->	Tuple:
				_lowerCAmelCase								=VaeImageProcessor(do_resize=__UpperCAmelCase	,	do_normalize=__UpperCAmelCase    )

				_lowerCAmelCase								=self.dummy_sample
				_lowerCAmelCase								=input_pt_ad.squeeze(0    )

				_lowerCAmelCase								=image_processor.postprocess(
				    image_processor.preprocess(__UpperCAmelCase    )	,	output_type="""np"""	,	)
				_lowerCAmelCase								=image_processor.postprocess(
				    image_processor.preprocess(__UpperCAmelCase    )	,	output_type="""np"""	,	)

				_lowerCAmelCase								=self.to_np(self.dummy_sample    )
				_lowerCAmelCase								=input_np_ad.squeeze(0    )

				_lowerCAmelCase								=image_processor.postprocess(
				    image_processor.preprocess(__UpperCAmelCase    )	,	output_type="""np"""	,	)
				_lowerCAmelCase								=image_processor.postprocess(
				    image_processor.preprocess(__UpperCAmelCase    )	,	output_type="""np"""	,	)

				assert np.abs(out_pt_ad - out_pt_ad    ).max() < 1e-6
				assert np.abs(out_np_ad - out_np_ad    ).max() < 1e-6



			def    _lowerCAmelCase						(				self    )					->	Optional[Any]:
				_lowerCAmelCase								=VaeImageProcessor(do_resize=__UpperCAmelCase	,	do_normalize=__UpperCAmelCase    )

				_lowerCAmelCase								=self.dummy_sample
				_lowerCAmelCase								=list(__UpperCAmelCase    )

				_lowerCAmelCase								=image_processor.postprocess(
				    image_processor.preprocess(__UpperCAmelCase    )	,	output_type="""np"""	,	)

				_lowerCAmelCase								=image_processor.postprocess(
				    image_processor.preprocess(__UpperCAmelCase    )	,	output_type="""np"""	,	)

				_lowerCAmelCase								=self.to_np(self.dummy_sample    )
				list(__UpperCAmelCase    )

				_lowerCAmelCase								=image_processor.postprocess(
				    image_processor.preprocess(__UpperCAmelCase    )	,	output_type="""np"""	,	)

				_lowerCAmelCase								=image_processor.postprocess(
				    image_processor.preprocess(__UpperCAmelCase    )	,	output_type="""np"""	,	)

				assert np.abs(out_pt_ad - out_pt_list    ).max() < 1e-6
				assert np.abs(out_np_ad - out_np_list    ).max() < 1e-6



