



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__  :  Union[str, Any]       = logging.getLogger(__name__)



class     __snake_case						(					snake_case_						):




							def        UpperCAmelCase__      (	self					:	Union[str, Any]    ,    A_					:	List[str]    ,    A_					:	int    ,    A_					:	Optional[int]=None    ,    A_					:	Tuple=None):


									if current_layer == 0:
											lowerCAmelCase_      :					str     =			self.embedding_hidden_mapping_in(A_)
									else:
											lowerCAmelCase_      :					List[Any]     =			hidden_states[0]

									lowerCAmelCase_      :					Optional[Any]     =			int(self.config.num_hidden_layers / self.config.num_hidden_groups)

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

									lowerCAmelCase_      :					Union[str, Any]     =			self.albert_layer_groups[group_idx](
									    A_    ,    A_    ,    head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]    ,    )
									lowerCAmelCase_      :					Tuple     =			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.'''						,snake_case_						,)
class     __snake_case						(					snake_case_						):




							def __init__(	self					:	int    ,    A_					:	Optional[int]):
									super().__init__(A_)

									lowerCAmelCase_      :					Optional[int]     =			AlbertTransformerWithPabee(A_)

									self.init_weights()
									lowerCAmelCase_      :					List[str]     =			0
									lowerCAmelCase_      :					List[str]     =			0
									lowerCAmelCase_      :					Any     =			0

									lowerCAmelCase_      :					Union[str, Any]     =			0




							def        UpperCAmelCase__      (	self					:	Any    ,    A_					:	Dict):
									lowerCAmelCase_      :					Optional[Any]     =			threshold




							def        UpperCAmelCase__      (	self					:	List[str]    ,    A_					:	int):
									lowerCAmelCase_      :					int     =			patience




							def        UpperCAmelCase__      (	self					:	List[str]):
									lowerCAmelCase_      :					Optional[Any]     =			0
									lowerCAmelCase_      :					Tuple     =			0




							def        UpperCAmelCase__      (	self					:	List[str]):
									lowerCAmelCase_      :					int     =			self.inference_layers_num / self.inference_instances_num
									lowerCAmelCase_      :					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(A_)







							@add_start_docstrings_to_model_forward(A_)
							def        UpperCAmelCase__      (	self					:	List[str]    ,    A_					:	Dict=None    ,    A_					:	Optional[Any]=None    ,    A_					:	Dict=None    ,    A_					:	List[Any]=None    ,    A_					:	Tuple=None    ,    A_					:	List[Any]=None    ,    A_					:	str=None    ,    A_					:	List[str]=None    ,    A_					:	Optional[Any]=False    ,    ):


									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_      :					Dict     =			input_ids.size()
									elif inputs_embeds is not None:
											lowerCAmelCase_      :					Optional[int]     =			inputs_embeds.size()[:-1]
									else:
											raise ValueError('''You have to specify either input_ids or inputs_embeds''')

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

									if attention_mask is None:
											lowerCAmelCase_      :					Dict     =			torch.ones(A_    ,    device=A_)
									if token_type_ids is None:
											lowerCAmelCase_      :					List[Any]     =			torch.zeros(A_    ,    dtype=torch.long    ,    device=A_)

									lowerCAmelCase_      :					List[Any]     =			attention_mask.unsqueeze(1).unsqueeze(2)
									lowerCAmelCase_      :					Tuple     =			extended_attention_mask.to(dtype=self.dtype)  # fp16 compatibility
									lowerCAmelCase_      :					Union[str, Any]     =			(1.0 - extended_attention_mask) * -1_0_0_0_0.0
									lowerCAmelCase_      :					Tuple     =			self.get_head_mask(A_    ,    self.config.num_hidden_layers)

									lowerCAmelCase_      :					Optional[int]     =			self.embeddings(
									    A_    ,    position_ids=A_    ,    token_type_ids=A_    ,    inputs_embeds=A_)
									lowerCAmelCase_      :					Tuple     =			embedding_output

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

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

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

													lowerCAmelCase_      :					Optional[Any]     =			logits
													if patient_counter == self.patience:
															break
											lowerCAmelCase_      :					Dict     =			[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. '''						,snake_case_						,)
class     __snake_case						(					snake_case_						):




							def __init__(	self					:	Union[str, Any]    ,    A_					:	Optional[Any]):
									super().__init__(A_)
									lowerCAmelCase_      :					Any     =			config.num_labels

									lowerCAmelCase_      :					List[str]     =			AlbertModelWithPabee(A_)
									lowerCAmelCase_      :					Optional[Any]     =			nn.Dropout(config.classifier_dropout_prob)
									lowerCAmelCase_      :					List[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(A_)
							def        UpperCAmelCase__      (	self					:	List[Any]    ,    A_					:	str=None    ,    A_					:	Optional[Any]=None    ,    A_					:	int=None    ,    A_					:	Tuple=None    ,    A_					:	Any=None    ,    A_					:	str=None    ,    A_					:	Any=None    ,    ):
									lowerCAmelCase_      :					List[Any]     =			self.albert(
									    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_      :					List[str]     =			(logits[-1],)

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

									return outputs









import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_video_inputs


if is_torch_available():
						import torch

if is_vision_available():
						from PIL import Image

						from transformers import VideoMAEImageProcessor



class     __snake_case						(					unittest.TestCase						):




							def __init__(	self					:	str    ,    A_					:	str    ,    A_					:	int=7    ,    A_					:	List[Any]=3    ,    A_					:	Dict=1_0    ,    A_					:	List[str]=1_8    ,    A_					:	List[Any]=3_0    ,    A_					:	str=4_0_0    ,    A_					:	str=True    ,    A_					:	Tuple=None    ,    A_					:	Optional[Any]=True    ,    A_					:	Tuple=[0.5, 0.5, 0.5]    ,    A_					:	Any=[0.5, 0.5, 0.5]    ,    A_					:	List[str]=None    ,    ):
									lowerCAmelCase_      :					Optional[Any]     =			size if size is not None else {'''shortest_edge''': 1_8}
									lowerCAmelCase_      :					Optional[int]     =			crop_size if crop_size is not None else {'''height''': 1_8, '''width''': 1_8}

									lowerCAmelCase_      :					Any     =			parent
									lowerCAmelCase_      :					Optional[Any]     =			batch_size
									lowerCAmelCase_      :					List[Any]     =			num_channels
									lowerCAmelCase_      :					Optional[int]     =			num_frames
									lowerCAmelCase_      :					Optional[int]     =			image_size
									lowerCAmelCase_      :					List[Any]     =			min_resolution
									lowerCAmelCase_      :					str     =			max_resolution
									lowerCAmelCase_      :					List[str]     =			do_resize
									lowerCAmelCase_      :					int     =			size
									lowerCAmelCase_      :					Union[str, Any]     =			do_normalize
									lowerCAmelCase_      :					Union[str, Any]     =			image_mean
									lowerCAmelCase_      :					List[str]     =			image_std
									lowerCAmelCase_      :					Any     =			crop_size







							def        UpperCAmelCase__      (	self					:	Any):
									return {
									    "image_mean": self.image_mean,
									    "image_std": self.image_std,
									    "do_normalize": self.do_normalize,
									    "do_resize": self.do_resize,
									    "size": self.size,
									    "crop_size": self.crop_size,
									}





@require_torch
@require_vision
class     __snake_case						(					UpperCamelCase_						,unittest.TestCase						):
							_a      				=			VideoMAEImageProcessor if is_vision_available() else None




							def        UpperCAmelCase__      (	self					:	Tuple):
									lowerCAmelCase_      :					int     =			VideoMAEImageProcessingTester(self)




							@property
							def        UpperCAmelCase__      (	self					:	Tuple):
									return self.image_processor_tester.prepare_image_processor_dict()




							def        UpperCAmelCase__      (	self					:	Optional[int]):
									lowerCAmelCase_      :					int     =			self.image_processing_class(**self.image_processor_dict)
									self.assertTrue(hasattr(A_    ,    '''image_mean'''))
									self.assertTrue(hasattr(A_    ,    '''image_std'''))
									self.assertTrue(hasattr(A_    ,    '''do_normalize'''))
									self.assertTrue(hasattr(A_    ,    '''do_resize'''))
									self.assertTrue(hasattr(A_    ,    '''do_center_crop'''))
									self.assertTrue(hasattr(A_    ,    '''size'''))




							def        UpperCAmelCase__      (	self					:	Dict):
									lowerCAmelCase_      :					int     =			self.image_processing_class.from_dict(self.image_processor_dict)
									self.assertEqual(image_processor.size    ,    {'''shortest_edge''': 1_8})
									self.assertEqual(image_processor.crop_size    ,    {'''height''': 1_8, '''width''': 1_8})

									lowerCAmelCase_      :					str     =			self.image_processing_class.from_dict(self.image_processor_dict    ,    size=4_2    ,    crop_size=8_4)
									self.assertEqual(image_processor.size    ,    {'''shortest_edge''': 4_2})
									self.assertEqual(image_processor.crop_size    ,    {'''height''': 8_4, '''width''': 8_4})




							def        UpperCAmelCase__      (	self					:	Dict):
									pass




							def        UpperCAmelCase__      (	self					:	str):
									# Initialize image_processing
									lowerCAmelCase_      :					int     =			self.image_processing_class(**self.image_processor_dict)
									# create random PIL videos
									lowerCAmelCase_      :					Optional[int]     =			prepare_video_inputs(self.image_processor_tester    ,    equal_resolution=A_)
									for video in video_inputs:
											self.assertIsInstance(A_    ,    A_)
											self.assertIsInstance(video[0]    ,    Image.Image)

									# Test not batched input
									lowerCAmelCase_      :					int     =			image_processing(video_inputs[0]    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        1,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )

									# Test batched
									lowerCAmelCase_      :					Tuple     =			image_processing(A_    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )




							def        UpperCAmelCase__      (	self					:	int):
									# Initialize image_processing
									lowerCAmelCase_      :					Dict     =			self.image_processing_class(**self.image_processor_dict)
									# create random numpy tensors
									lowerCAmelCase_      :					str     =			prepare_video_inputs(self.image_processor_tester    ,    equal_resolution=A_    ,    numpify=A_)
									for video in video_inputs:
											self.assertIsInstance(A_    ,    A_)
											self.assertIsInstance(video[0]    ,    np.ndarray)

									# Test not batched input
									lowerCAmelCase_      :					Tuple     =			image_processing(video_inputs[0]    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        1,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )

									# Test batched
									lowerCAmelCase_      :					Union[str, Any]     =			image_processing(A_    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )







							def        UpperCAmelCase__      (	self					:	List[str]):
									# Initialize image_processing
									lowerCAmelCase_      :					Tuple     =			self.image_processing_class(**self.image_processor_dict)
									# create random PyTorch tensors
									lowerCAmelCase_      :					Tuple     =			prepare_video_inputs(self.image_processor_tester    ,    equal_resolution=A_    ,    torchify=A_)
									for video in video_inputs:
											self.assertIsInstance(A_    ,    A_)
											self.assertIsInstance(video[0]    ,    torch.Tensor)

									# Test not batched input
									lowerCAmelCase_      :					List[str]     =			image_processing(video_inputs[0]    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        1,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )

									# Test batched
									lowerCAmelCase_      :					List[Any]     =			image_processing(A_    ,    return_tensors='''pt''').pixel_values
									self.assertEqual(
									    encoded_videos.shape    ,    (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_frames,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['''height'''],
									        self.image_processor_tester.crop_size['''width'''],
									    )    ,    )





