





from __future__ import annotations

import inspect
import unittest

import numpy as np

from transformers import DeiTConfig
from transformers.testing_utils import require_tf, require_vision, slow
from transformers.utils import cached_property, is_tf_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
       import tensorflow as tf

       from transformers import (
           TFDeiTForImageClassification,
           TFDeiTForImageClassificationWithTeacher,
           TFDeiTForMaskedImageModeling,
           TFDeiTModel,
       )
       from transformers.models.deit.modeling_tf_deit import TF_DEIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
       from PIL import Image

       from transformers import DeiTImageProcessor







class       __A							:




      """simple docstring"""







      def __init__(		self   ,				a__   ,				a__=13   ,				a__=30   ,				a__=2   ,				a__=3   ,				a__=True   ,				a__=True   ,				a__=32   ,				a__=2   ,				a__=4   ,				a__=37   ,				a__="gelu"   ,				a__=0.1   ,				a__=0.1   ,				a__=10   ,				a__=0.02   ,				a__=3   ,				a__=None   ,				a__=2   ,				):



             """simple docstring"""





             _lowerCamelCase				:  Union[str, Any]						=		parent
             _lowerCamelCase				:  List[Any]						=		batch_size
             _lowerCamelCase				:  int						=		image_size
             _lowerCamelCase				:  str						=		patch_size
             _lowerCamelCase				:  List[Any]						=		num_channels
             _lowerCamelCase				:  Tuple						=		is_training
             _lowerCamelCase				:  Optional[Any]						=		use_labels
             _lowerCamelCase				:  int						=		hidden_size
             _lowerCamelCase				:  int						=		num_hidden_layers
             _lowerCamelCase				:  Union[str, Any]						=		num_attention_heads
             _lowerCamelCase				:  str						=		intermediate_size
             _lowerCamelCase				:  int						=		hidden_act
             _lowerCamelCase				:  Union[str, Any]						=		hidden_dropout_prob
             _lowerCamelCase				:  Dict						=		attention_probs_dropout_prob
             _lowerCamelCase				:  Union[str, Any]						=		type_sequence_label_size
             _lowerCamelCase				:  List[Any]						=		initializer_range
             _lowerCamelCase				:  Optional[Any]						=		scope
             _lowerCamelCase				:  str						=		encoder_stride

             # in DeiT, the seq length equals the number of patches + 2 (we add 2 for the [CLS] and distilation tokens)
             _lowerCamelCase				:  List[str]						=		(image_size // patch_size) ** 2
             _lowerCamelCase				:  int						=		num_patches + 2


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Tuple						=		floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size])

             _lowerCamelCase				:  Dict						=		None
             if self.use_labels:
                    _lowerCamelCase				:  int						=		ids_tensor([self.batch_size]   ,				self.type_sequence_label_size)

             _lowerCamelCase				:  Tuple						=		self.get_config()

             return config, pixel_values, labels


      def 					__snake_case      (		self):



             """simple docstring"""





             return DeiTConfig(
                 image_size=self.image_size   ,				patch_size=self.patch_size   ,				num_channels=self.num_channels   ,				hidden_size=self.hidden_size   ,				num_hidden_layers=self.num_hidden_layers   ,				num_attention_heads=self.num_attention_heads   ,				intermediate_size=self.intermediate_size   ,				hidden_act=self.hidden_act   ,				hidden_dropout_prob=self.hidden_dropout_prob   ,				attention_probs_dropout_prob=self.attention_probs_dropout_prob   ,				is_decoder=a__   ,				initializer_range=self.initializer_range   ,				encoder_stride=self.encoder_stride   ,				)


      def 					__snake_case      (		self   ,				a__   ,				a__   ,				a__):



             """simple docstring"""





             _lowerCamelCase				:  List[Any]						=		TFDeiTModel(config=a__)
             _lowerCamelCase				:  Optional[Any]						=		model(a__)
             self.parent.assertEqual(result.last_hidden_state.shape   ,				(self.batch_size, self.seq_length, self.hidden_size))


      def 					__snake_case      (		self   ,				a__   ,				a__   ,				a__):



             """simple docstring"""





             _lowerCamelCase				:  int						=		TFDeiTForMaskedImageModeling(config=a__)
             _lowerCamelCase				:  List[Any]						=		model(a__)
             self.parent.assertEqual(
                 result.reconstruction.shape   ,				(self.batch_size, self.num_channels, self.image_size, self.image_size))

             # test greyscale images
             _lowerCamelCase				:  Dict						=		1
             _lowerCamelCase				:  Optional[Any]						=		TFDeiTForMaskedImageModeling(a__)

             _lowerCamelCase				:  int						=		floats_tensor([self.batch_size, 1, self.image_size, self.image_size])
             _lowerCamelCase				:  List[Any]						=		model(a__)
             self.parent.assertEqual(result.reconstruction.shape   ,				(self.batch_size, 1, self.image_size, self.image_size))


      def 					__snake_case      (		self   ,				a__   ,				a__   ,				a__):



             """simple docstring"""





             _lowerCamelCase				:  Dict						=		self.type_sequence_label_size
             _lowerCamelCase				:  List[str]						=		TFDeiTForImageClassification(a__)
             _lowerCamelCase				:  List[Any]						=		model(a__   ,				labels=a__)
             self.parent.assertEqual(result.logits.shape   ,				(self.batch_size, self.type_sequence_label_size))

             # test greyscale images
             _lowerCamelCase				:  int						=		1
             _lowerCamelCase				:  Tuple						=		TFDeiTForImageClassification(a__)

             _lowerCamelCase				:  Dict						=		floats_tensor([self.batch_size, 1, self.image_size, self.image_size])
             _lowerCamelCase				:  Tuple						=		model(a__   ,				labels=a__)
             self.parent.assertEqual(result.logits.shape   ,				(self.batch_size, self.type_sequence_label_size))



      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Tuple						=		self.prepare_config_and_inputs()
             _lowerCamelCase,	_lowerCamelCase,	_lowerCamelCase				:  List[Any]						=		config_and_inputs
             _lowerCamelCase				:  Dict						=		{'''pixel_values''': pixel_values}
             return config, inputs_dict







@require_tf
class       __A							(   lowerCamelCase__			,lowerCamelCase__			,unittest.TestCase	):




      """simple docstring"""







      UpperCAmelCase__							=   (
          (
              TFDeiTModel,
              TFDeiTForImageClassification,
              TFDeiTForImageClassificationWithTeacher,
              TFDeiTForMaskedImageModeling,
          )
          if is_tf_available()
          else ()
      )
      UpperCAmelCase__							=   (
          {
              """feature-extraction""": TFDeiTModel,
              """image-classification""": (TFDeiTForImageClassification, TFDeiTForImageClassificationWithTeacher),
          }
          if is_tf_available()
          else {}
      )

      UpperCAmelCase__							=   False
      UpperCAmelCase__							=   False
      UpperCAmelCase__							=   False
      UpperCAmelCase__							=   False


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Optional[Any]						=		TFDeiTModelTester(self)
             _lowerCamelCase				:  Tuple						=		ConfigTester(self   ,				config_class=a__   ,				has_text_modality=a__   ,				hidden_size=37)


      def 					__snake_case      (		self):



             """simple docstring"""





             self.config_tester.run_common_tests()


      @unittest.skip(reason='''DeiT does not use inputs_embeds''')
      def 					__snake_case      (		self):



             """simple docstring"""





             pass


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase,	_lowerCamelCase				:  Optional[Any]						=		self.model_tester.prepare_config_and_inputs_for_common()

             for model_class in self.all_model_classes:
                    _lowerCamelCase				:  Optional[Any]						=		model_class(a__)
                    self.assertIsInstance(model.get_input_embeddings()   ,				(tf.keras.layers.Layer))
                    _lowerCamelCase				:  Any						=		model.get_output_embeddings()
                    self.assertTrue(x is None or isinstance(a__   ,				tf.keras.layers.Dense))


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase,	_lowerCamelCase				:  Optional[Any]						=		self.model_tester.prepare_config_and_inputs_for_common()

             for model_class in self.all_model_classes:
                    _lowerCamelCase				:  Union[str, Any]						=		model_class(a__)
                    _lowerCamelCase				:  Union[str, Any]						=		inspect.signature(model.call)
                    # signature.parameters is an OrderedDict => so arg_names order is deterministic
                    _lowerCamelCase				:  Optional[int]						=		[*signature.parameters.keys()]

                    _lowerCamelCase				:  Dict						=		['''pixel_values''']
                    self.assertListEqual(arg_names[:1]   ,				a__)


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Union[str, Any]						=		self.model_tester.prepare_config_and_inputs()
             self.model_tester.create_and_check_model(*a__)


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  List[str]						=		self.model_tester.prepare_config_and_inputs()
             self.model_tester.create_and_check_for_masked_image_modeling(*a__)


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  int						=		self.model_tester.prepare_config_and_inputs()
             self.model_tester.create_and_check_for_image_classification(*a__)


      def 					__snake_case      (		self   ,				a__   ,				a__   ,				a__=False):



             """simple docstring"""





             _lowerCamelCase				:  Union[str, Any]						=		super()._prepare_for_class(a__   ,				a__   ,				return_labels=a__)

             if return_labels:
                    if "labels" in inputs_dict and "labels" not in inspect.signature(model_class.call).parameters:
                           del inputs_dict["labels"]

             return inputs_dict



      @slow
      def 					__snake_case      (		self):



             """simple docstring"""





             for model_name in TF_DEIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                    _lowerCamelCase				:  str						=		TFDeiTModel.from_pretrained(a__)
                    self.assertIsNotNone(a__)




def  __UpperCAmelCase(   ):
       _lowerCamelCase				:  Optional[int]						=		Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''   )
       return image







@require_tf
@require_vision
class       __A							(   unittest.TestCase	):




      """simple docstring"""







      @cached_property
      def 					__snake_case      (		self):



             """simple docstring"""





             return (
                 DeiTImageProcessor.from_pretrained('''facebook/deit-base-distilled-patch16-224''')
                 if is_vision_available()
                 else None
             )



      @slow
      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Dict						=		TFDeiTForImageClassificationWithTeacher.from_pretrained('''facebook/deit-base-distilled-patch16-224''')

             _lowerCamelCase				:  Optional[Any]						=		self.default_image_processor
             _lowerCamelCase				:  List[str]						=		prepare_img()
             _lowerCamelCase				:  Optional[int]						=		image_processor(images=a__   ,				return_tensors='''tf''')

             # forward pass
             _lowerCamelCase				:  Tuple						=		model(**a__)

             # verify the logits
             _lowerCamelCase				:  Tuple						=		tf.TensorShape((1, 1000))
             self.assertEqual(outputs.logits.shape   ,				a__)

             _lowerCamelCase				:  Optional[Any]						=		tf.constant([-1.0266, 0.1912, -1.2861])

             self.assertTrue(np.allclose(outputs.logits[0, :3]   ,				a__   ,				atol=1e-4))








import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
       from sagemaker import Session, TrainingJobAnalytics
       from sagemaker.huggingface import HuggingFace







@pytest.mark.skipif(
    literal_eval(os.getenv("""TEST_SAGEMAKER"""			,"""False"""	)	) is not True			,reason="""Skipping test because should only be run when releasing minor transformers version"""			,)
@pytest.mark.usefixtures("""sm_env"""	)
@parameterized_class(
    [
        {
            """framework""": """pytorch""",
            """script""": """run_glue.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.g4dn.xlarge""",
            """results""": {"""train_runtime""": 650, """eval_accuracy""": 0.6, """eval_loss""": 0.9},
        },
        {
            """framework""": """tensorflow""",
            """script""": """run_tf.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.g4dn.xlarge""",
            """results""": {"""train_runtime""": 600, """eval_accuracy""": 0.3, """eval_loss""": 0.9},
        },
    ]	)
class       __A							(   unittest.TestCase	):




      """simple docstring"""







      def 					__snake_case      (		self):



             """simple docstring"""





             if self.framework == "pytorch":
                    subprocess.run(
                        F"""cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py""".split()   ,				encoding='''utf-8'''   ,				check=a__   ,				)
             assert hasattr(self   ,				'''env''')


      def 					__snake_case      (		self   ,				a__=1):



             """simple docstring"""





             return HuggingFace(
                 entry_point=self.script   ,				source_dir=self.env.test_path   ,				role=self.env.role   ,				image_uri=self.env.image_uri   ,				base_job_name=F"""{self.env.base_job_name}-single"""   ,				instance_count=a__   ,				instance_type=self.instance_type   ,				debugger_hook_config=a__   ,				hyperparameters={**self.env.hyperparameters, '''model_name_or_path''': self.model_name_or_path}   ,				metric_definitions=self.env.metric_definitions   ,				py_version='''py36'''   ,				)


      def 					__snake_case      (		self   ,				a__):



             """simple docstring"""





             TrainingJobAnalytics(a__).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv""")



      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Dict						=		self.create_estimator()

             # run training
             estimator.fit()

             # result dataframe
             _lowerCamelCase				:  Tuple						=		TrainingJobAnalytics(estimator.latest_training_job.name).dataframe()

             # extract kpis
             _lowerCamelCase				:  Tuple						=		list(result_metrics_df[result_metrics_df.metric_name == '''eval_accuracy''']['''value'''])
             _lowerCamelCase				:  Dict						=		list(result_metrics_df[result_metrics_df.metric_name == '''eval_loss''']['''value'''])
             # get train time from SageMaker job, this includes starting, preprocessing, stopping
             _lowerCamelCase				:  Optional[Any]						=		(
                 Session().describe_training_job(estimator.latest_training_job.name).get('''TrainingTimeInSeconds'''   ,				99_9999)
             )

             # assert kpis
             assert train_runtime <= self.results["train_runtime"]
             assert all(t >= self.results['''eval_accuracy'''] for t in eval_accuracy)
             assert all(t <= self.results['''eval_loss'''] for t in eval_loss)

             # dump tests result into json file to share in PR
             with open(F"""{estimator.latest_training_job.name}.json"""   ,				'''w''') as outfile:
                    json.dump({'''train_time''': train_runtime, '''eval_accuracy''': eval_accuracy, '''eval_loss''': eval_loss}   ,				a__)

