




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


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







class 		__UpperCAmelCase(		A_ ):





     """simple docstring"""



     def 				__SCREAMING_SNAKE_CASE		(      self :       Dict						,				A :       Dict						,				A :       List[Any]						,				A :       str=None						,				A :       Optional[int]=None    )			->			List[str]:






         """simple docstring"""



         if current_layer == 0:
             a__      =       self.embedding_hidden_mapping_in(A    )
         else:
             a__      =       hidden_states[0]

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

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

         a__      =       self.albert_layer_groups[group_idx](
             A						,				A						,				head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]						,				)
         a__      =       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 		__UpperCAmelCase(		A_ ):





     """simple docstring"""



     def __init__(      self :       Union[str, Any]						,				A :       str    )			->			Dict:






         """simple docstring"""
         super().__init__(A    )

         a__      =       AlbertTransformerWithPabee(A    )

         self.init_weights()
         a__      =       0
         a__      =       0
         a__      =       0

         a__      =       0

     def 				__SCREAMING_SNAKE_CASE		(      self :       Any						,				A :       Tuple    )			->			Optional[Any]:






         """simple docstring"""
         a__      =       threshold

     def 				__SCREAMING_SNAKE_CASE		(      self :       Union[str, Any]						,				A :       Tuple    )			->			Union[str, Any]:






         """simple docstring"""
         a__      =       patience

     def 				__SCREAMING_SNAKE_CASE		(      self :       Tuple    )			->			Optional[Any]:






         """simple docstring"""
         a__      =       0
         a__      =       0

     def 				__SCREAMING_SNAKE_CASE		(      self :       Dict    )			->			Dict:






         """simple docstring"""
         a__      =       self.inference_layers_num / self.inference_instances_num
         a__      =       (
             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 				__SCREAMING_SNAKE_CASE		(      self :       List[Any]						,				A :       str=None						,				A :       str=None						,				A :       int=None						,				A :       Union[str, Any]=None						,				A :       Dict=None						,				A :       List[str]=None						,				A :       str=None						,				A :       Union[str, Any]=None						,				A :       Optional[Any]=False						,				)			->			str:






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

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

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

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

         a__      =       self.embeddings(
             A						,				position_ids=A						,				token_type_ids=A						,				inputs_embeds=A    )
         a__      =       embedding_output

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

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

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

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





     """simple docstring"""



     def __init__(      self :       Tuple						,				A :       Optional[Any]    )			->			Optional[int]:






         """simple docstring"""
         super().__init__(A    )
         a__      =       config.num_labels

         a__      =       AlbertModelWithPabee(A    )
         a__      =       nn.Dropout(config.classifier_dropout_prob    )
         a__      =       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 				__SCREAMING_SNAKE_CASE		(      self :       Union[str, Any]						,				A :       int=None						,				A :       List[str]=None						,				A :       str=None						,				A :       str=None						,				A :       Union[str, Any]=None						,				A :       str=None						,				A :       Optional[Any]=None						,				)			->			str:






         """simple docstring"""
         a__      =       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						,				)

         a__      =       (logits[-1],)

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

         return outputs












import unittest

from transformers import (
    MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
    PreTrainedTokenizer,
    is_vision_available,
)
from transformers.pipelines import ImageClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_torch_or_tf,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
 from PIL import Image
else:







 class 		__UpperCAmelCase:





      """simple docstring"""



      @staticmethod
      def 				__SCREAMING_SNAKE_CASE		(      *A :       str						,				**A :       int    )			->			Union[str, Any]:






          """simple docstring"""
          pass







@is_pipeline_test
@require_torch_or_tf
@require_vision
class 		__UpperCAmelCase(		unittest.TestCase ):





     """simple docstring"""



     __UpperCAmelCase	: Tuple						=    MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING
     __UpperCAmelCase	: Dict						=    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING

     def 				__SCREAMING_SNAKE_CASE		(      self :       str						,				A :       Optional[int]						,				A :       Tuple						,				A :       int    )			->			str:






         """simple docstring"""
         a__      =       ImageClassificationPipeline(model=A						,				image_processor=A						,				top_k=2    )
         a__      =       [
             Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""    ),
             """http://images.cocodataset.org/val2017/000000039769.jpg""",
         ]
         return image_classifier, examples

     def 				__SCREAMING_SNAKE_CASE		(      self :       Any						,				A :       List[str]						,				A :       Dict    )			->			Union[str, Any]:






         """simple docstring"""
         a__      =       image_classifier("""./tests/fixtures/tests_samples/COCO/000000039769.png"""    )

         self.assertEqual(
             A						,				[
                 {"""score""": ANY(A    ), """label""": ANY(A    )},
                 {"""score""": ANY(A    ), """label""": ANY(A    )},
             ]						,				)

         import datasets

         a__      =       datasets.load_dataset("""hf-internal-testing/fixtures_image_utils"""						,				"""image"""						,				split="""test"""    )

         # Accepts URL + PIL.Image + lists
         a__      =       image_classifier(
             [
                 Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""    ),
                 """http://images.cocodataset.org/val2017/000000039769.jpg""",
                 # RGBA
                 dataset[0]["""file"""],
                 # LA
                 dataset[1]["""file"""],
                 # L
                 dataset[2]["""file"""],
             ]    )
         self.assertEqual(
             A						,				[
                 [
                     {"""score""": ANY(A    ), """label""": ANY(A    )},
                     {"""score""": ANY(A    ), """label""": ANY(A    )},
                 ],
                 [
                     {"""score""": ANY(A    ), """label""": ANY(A    )},
                     {"""score""": ANY(A    ), """label""": ANY(A    )},
                 ],
                 [
                     {"""score""": ANY(A    ), """label""": ANY(A    )},
                     {"""score""": ANY(A    ), """label""": ANY(A    )},
                 ],
                 [
                     {"""score""": ANY(A    ), """label""": ANY(A    )},
                     {"""score""": ANY(A    ), """label""": ANY(A    )},
                 ],
                 [
                     {"""score""": ANY(A    ), """label""": ANY(A    )},
                     {"""score""": ANY(A    ), """label""": ANY(A    )},
                 ],
             ]						,				)

     @require_torch
     def 				__SCREAMING_SNAKE_CASE		(      self :       Optional[int]    )			->			Optional[Any]:






         """simple docstring"""
         a__      =       """hf-internal-testing/tiny-random-vit"""
         a__      =       pipeline("""image-classification"""						,				model=A    )

         a__      =       image_classifier("""http://images.cocodataset.org/val2017/000000039769.jpg"""    )
         self.assertEqual(
             nested_simplify(A						,				decimals=4    )						,				[{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}]						,				)

         a__      =       image_classifier(
             [
                 """http://images.cocodataset.org/val2017/000000039769.jpg""",
                 """http://images.cocodataset.org/val2017/000000039769.jpg""",
             ]						,				top_k=2						,				)
         self.assertEqual(
             nested_simplify(A						,				decimals=4    )						,				[
                 [{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}],
                 [{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}],
             ]						,				)

     @require_tf
     def 				__SCREAMING_SNAKE_CASE		(      self :       Dict    )			->			str:






         """simple docstring"""
         a__      =       """hf-internal-testing/tiny-random-vit"""
         a__      =       pipeline("""image-classification"""						,				model=A						,				framework="""tf"""    )

         a__      =       image_classifier("""http://images.cocodataset.org/val2017/000000039769.jpg"""    )
         self.assertEqual(
             nested_simplify(A						,				decimals=4    )						,				[{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}]						,				)

         a__      =       image_classifier(
             [
                 """http://images.cocodataset.org/val2017/000000039769.jpg""",
                 """http://images.cocodataset.org/val2017/000000039769.jpg""",
             ]						,				top_k=2						,				)
         self.assertEqual(
             nested_simplify(A						,				decimals=4    )						,				[
                 [{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}],
                 [{"""label""": """LABEL_1""", """score""": 0.5_7_4}, {"""label""": """LABEL_0""", """score""": 0.4_2_6}],
             ]						,				)

     def 				__SCREAMING_SNAKE_CASE		(      self :       Union[str, Any]    )			->			str:






         """simple docstring"""
         a__      =       PreTrainedTokenizer()

         # Assert that the pipeline can be initialized with a feature extractor that is not in any mapping
         a__      =       pipeline(
             """image-classification"""						,				model="""hf-internal-testing/tiny-random-vit"""						,				tokenizer=A    )

         self.assertIs(image_classifier.tokenizer						,				A    )






     @slow
     @require_torch
     def 				__SCREAMING_SNAKE_CASE		(      self :       Optional[Any]    )			->			Dict:






         """simple docstring"""
         # Perceiver is not tested by `run_pipeline_test` properly.
         # That is because the type of feature_extractor and model preprocessor need to be kept
         # in sync, which is not the case in the current design
         a__      =       pipeline("""image-classification"""						,				model="""deepmind/vision-perceiver-conv"""    )
         a__      =       image_classifier("""http://images.cocodataset.org/val2017/000000039769.jpg"""    )
         self.assertEqual(
             nested_simplify(A						,				decimals=4    )						,				[
                 {"""score""": 0.4_3_8_5, """label""": """tabby, tabby cat"""},
                 {"""score""": 0.3_2_1, """label""": """tiger cat"""},
                 {"""score""": 0.0_5_0_2, """label""": """Egyptian cat"""},
                 {"""score""": 0.0_1_3_7, """label""": """crib, cot"""},
                 {"""score""": 0.0_0_7, """label""": """radiator"""},
             ]						,				)

         a__      =       pipeline("""image-classification"""						,				model="""deepmind/vision-perceiver-fourier"""    )
         a__      =       image_classifier("""http://images.cocodataset.org/val2017/000000039769.jpg"""    )
         self.assertEqual(
             nested_simplify(A						,				decimals=4    )						,				[
                 {"""score""": 0.5_6_5_8, """label""": """tabby, tabby cat"""},
                 {"""score""": 0.1_3_0_9, """label""": """tiger cat"""},
                 {"""score""": 0.0_7_2_2, """label""": """Egyptian cat"""},
                 {"""score""": 0.0_7_0_7, """label""": """remote control, remote"""},
                 {"""score""": 0.0_0_8_2, """label""": """computer keyboard, keypad"""},
             ]						,				)

         a__      =       pipeline("""image-classification"""						,				model="""deepmind/vision-perceiver-learned"""    )
         a__      =       image_classifier("""http://images.cocodataset.org/val2017/000000039769.jpg"""    )
         self.assertEqual(
             nested_simplify(A						,				decimals=4    )						,				[
                 {"""score""": 0.3_0_2_2, """label""": """tabby, tabby cat"""},
                 {"""score""": 0.2_3_6_2, """label""": """Egyptian cat"""},
                 {"""score""": 0.1_8_5_6, """label""": """tiger cat"""},
                 {"""score""": 0.0_3_2_4, """label""": """remote control, remote"""},
                 {"""score""": 0.0_0_9_6, """label""": """quilt, comforter, comfort, puff"""},
             ]						,				)






