



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


UpperCamelCase_	       =   logging.getLogger(__name__)







class 							a_  (		_snake_case				):

						def 	__a       (    self	:int   ,						_lowercase	:List[Any]   ,						_lowercase	:Dict   ,						_lowercase	:Optional[Any]=None   ,						_lowercase	:str=None)		->					Tuple:
												if current_layer == 0:
																		UpperCAmelCase_     							=     self.embedding_hidden_mapping_in(_lowercase)
												else:
																		UpperCAmelCase_     							=     hidden_states[0]

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

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

												UpperCAmelCase_     							=     self.albert_layer_groups[group_idx](
												    _lowercase   ,						_lowercase   ,						head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]   ,						)
												UpperCAmelCase_     							=     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 							a_  (		_snake_case				):

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

												UpperCAmelCase_     							=     AlbertTransformerWithPabee(_lowercase)

												self.init_weights()
												UpperCAmelCase_     							=     0
												UpperCAmelCase_     							=     0
												UpperCAmelCase_     							=     0

												UpperCAmelCase_     							=     0

						def 	__a       (    self	:str   ,						_lowercase	:int)		->					int:
												UpperCAmelCase_     							=     threshold

						def 	__a       (    self	:Union[str, Any]   ,						_lowercase	:str)		->					Union[str, Any]:
												UpperCAmelCase_     							=     patience

						def 	__a       (    self	:str)		->					List[str]:
												UpperCAmelCase_     							=     0
												UpperCAmelCase_     							=     0

						def 	__a       (    self	:Union[str, Any])		->					Optional[Any]:
												UpperCAmelCase_     							=     self.inference_layers_num / self.inference_instances_num
												UpperCAmelCase_     							=     (
												    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(_lowercase)




						@add_start_docstrings_to_model_forward(_lowercase)
						def 	__a       (    self	:List[str]   ,						_lowercase	:int=None   ,						_lowercase	:Optional[Any]=None   ,						_lowercase	:Tuple=None   ,						_lowercase	:Tuple=None   ,						_lowercase	:Tuple=None   ,						_lowercase	:Any=None   ,						_lowercase	:Union[str, Any]=None   ,						_lowercase	:Optional[int]=None   ,						_lowercase	:Tuple=False   ,						)		->					int:

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

												UpperCAmelCase_     							=     input_ids.device if input_ids is not None else inputs_embeds.device

												if attention_mask is None:
																		UpperCAmelCase_     							=     torch.ones(_lowercase   ,						device=_lowercase)
												if token_type_ids is None:
																		UpperCAmelCase_     							=     torch.zeros(_lowercase   ,						dtype=torch.long   ,						device=_lowercase)

												UpperCAmelCase_     							=     attention_mask.unsqueeze(1).unsqueeze(2)
												UpperCAmelCase_     							=     extended_attention_mask.to(dtype=self.dtype)  # fp16 compatibility
												UpperCAmelCase_     							=     (1.0 - extended_attention_mask) * -10_000.0
												UpperCAmelCase_     							=     self.get_head_mask(_lowercase   ,						self.config.num_hidden_layers)

												UpperCAmelCase_     							=     self.embeddings(
												    _lowercase   ,						position_ids=_lowercase   ,						token_type_ids=_lowercase   ,						inputs_embeds=_lowercase)
												UpperCAmelCase_     							=     embedding_output

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

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

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

																								UpperCAmelCase_     							=     logits
																								if patient_counter == self.patience:
																														break
																		UpperCAmelCase_     							=     [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. "					, _snake_case					, )
class 							a_  (		_snake_case				):

						def __init__(    self	:str   ,						_lowercase	:int)		->					Union[str, Any]:
												super().__init__(_lowercase)
												UpperCAmelCase_     							=     config.num_labels

												UpperCAmelCase_     							=     AlbertModelWithPabee(_lowercase)
												UpperCAmelCase_     							=     nn.Dropout(config.classifier_dropout_prob)
												UpperCAmelCase_     							=     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(_lowercase)
						def 	__a       (    self	:str   ,						_lowercase	:List[Any]=None   ,						_lowercase	:Union[str, Any]=None   ,						_lowercase	:str=None   ,						_lowercase	:Tuple=None   ,						_lowercase	:int=None   ,						_lowercase	:List[str]=None   ,						_lowercase	:str=None   ,						)		->					List[Any]:
												UpperCAmelCase_     							=     self.albert(
												    input_ids=_lowercase   ,						attention_mask=_lowercase   ,						token_type_ids=_lowercase   ,						position_ids=_lowercase   ,						head_mask=_lowercase   ,						inputs_embeds=_lowercase   ,						output_dropout=self.dropout   ,						output_layers=self.classifiers   ,						regression=self.num_labels == 1   ,						)

												UpperCAmelCase_     							=     (logits[-1],)

												if labels is not None:
																		UpperCAmelCase_     							=     None
																		UpperCAmelCase_     							=     0
																		for ix, logits_item in enumerate(_lowercase):
																								if self.num_labels == 1:
																														#  We are doing regression
																														UpperCAmelCase_     							=     MSELoss()
																														UpperCAmelCase_     							=     loss_fct(logits_item.view(-1)   ,						labels.view(-1))
																								else:
																														UpperCAmelCase_     							=     CrossEntropyLoss()
																														UpperCAmelCase_     							=     loss_fct(logits_item.view(-1   ,						self.num_labels)   ,						labels.view(-1))
																								if total_loss is None:
																														UpperCAmelCase_     							=     loss
																								else:
																														total_loss += loss * (ix + 1)
																								total_weights += ix + 1
																		UpperCAmelCase_     							=     (total_loss / total_weights,) + outputs

												return outputs










import unittest
from pathlib import Path

from transformers import is_vision_available, load_tool
from transformers.testing_utils import get_tests_dir

from .test_tools_common import ToolTesterMixin


if is_vision_available():
							from PIL import Image







class 							a_  (		unittest.TestCase					, _snake_case				):

						def 	__a       (    self	:Union[str, Any])		->					Optional[Any]:
												UpperCAmelCase_     							=     load_tool('''image-question-answering''')
												self.tool.setup()
												UpperCAmelCase_     							=     load_tool('''image-question-answering'''   ,						remote=_lowercase)

						def 	__a       (    self	:int)		->					Union[str, Any]:
												UpperCAmelCase_     							=     Image.open(Path(get_tests_dir('''fixtures/tests_samples/COCO''')) / '''000000039769.png''')
												UpperCAmelCase_     							=     self.tool(_lowercase   ,						'''How many cats are sleeping on the couch?''')
												self.assertEqual(_lowercase   ,						'''2''')

						def 	__a       (    self	:int)		->					Dict:
												UpperCAmelCase_     							=     Image.open(Path(get_tests_dir('''fixtures/tests_samples/COCO''')) / '''000000039769.png''')
												UpperCAmelCase_     							=     self.remote_tool(_lowercase   ,						'''How many cats are sleeping on the couch?''')
												self.assertEqual(_lowercase   ,						'''2''')

						def 	__a       (    self	:Optional[int])		->					int:
												UpperCAmelCase_     							=     Image.open(Path(get_tests_dir('''fixtures/tests_samples/COCO''')) / '''000000039769.png''')
												UpperCAmelCase_     							=     self.tool(image=_lowercase   ,						question='''How many cats are sleeping on the couch?''')
												self.assertEqual(_lowercase   ,						'''2''')




						def 	__a       (    self	:Optional[Any])		->					List[str]:
												UpperCAmelCase_     							=     Image.open(Path(get_tests_dir('''fixtures/tests_samples/COCO''')) / '''000000039769.png''')
												UpperCAmelCase_     							=     self.remote_tool(image=_lowercase   ,						question='''How many cats are sleeping on the couch?''')
												self.assertEqual(_lowercase   ,						'''2''')





