





import inspect
import unittest
import warnings
from math import ceil, floor

from transformers import LevitConfig
from transformers.file_utils import cached_property, is_torch_available, is_vision_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
       import torch

       from transformers import (
           MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
           MODEL_MAPPING,
           LevitForImageClassification,
           LevitForImageClassificationWithTeacher,
           LevitModel,
       )
       from transformers.models.levit.modeling_levit import LEVIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
       from PIL import Image

       from transformers import LevitImageProcessor







class       __A							(   lowerCamelCase__	):




      """simple docstring"""







      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  List[str]						=		self.config_class(**self.inputs_dict)
             self.parent.assertTrue(hasattr(a__   ,				'''hidden_sizes'''))
             self.parent.assertTrue(hasattr(a__   ,				'''num_attention_heads'''))







class       __A							:




      """simple docstring"""







      def __init__(		self   ,				a__   ,				a__=13   ,				a__=64   ,				a__=3   ,				a__=3   ,				a__=2   ,				a__=1   ,				a__=16   ,				a__=[128, 256, 384]   ,				a__=[4, 6, 8]   ,				a__=[2, 3, 4]   ,				a__=[16, 16, 16]   ,				a__=0   ,				a__=[2, 2, 2]   ,				a__=[2, 2, 2]   ,				a__=0.02   ,				a__=True   ,				a__=True   ,				a__=2   ,				):



             """simple docstring"""





             _lowerCamelCase				:  str						=		parent
             _lowerCamelCase				:  str						=		batch_size
             _lowerCamelCase				:  Dict						=		image_size
             _lowerCamelCase				:  Union[str, Any]						=		num_channels
             _lowerCamelCase				:  List[str]						=		kernel_size
             _lowerCamelCase				:  int						=		stride
             _lowerCamelCase				:  Tuple						=		padding
             _lowerCamelCase				:  Optional[Any]						=		hidden_sizes
             _lowerCamelCase				:  Union[str, Any]						=		num_attention_heads
             _lowerCamelCase				:  Tuple						=		depths
             _lowerCamelCase				:  int						=		key_dim
             _lowerCamelCase				:  Any						=		drop_path_rate
             _lowerCamelCase				:  Optional[Any]						=		patch_size
             _lowerCamelCase				:  Tuple						=		attention_ratio
             _lowerCamelCase				:  List[Any]						=		mlp_ratio
             _lowerCamelCase				:  Union[str, Any]						=		initializer_range
             _lowerCamelCase				:  List[Any]						=		[
                 ['''Subsample''', key_dim[0], hidden_sizes[0] // key_dim[0], 4, 2, 2],
                 ['''Subsample''', key_dim[0], hidden_sizes[1] // key_dim[0], 4, 2, 2],
             ]
             _lowerCamelCase				:  List[str]						=		is_training
             _lowerCamelCase				:  str						=		use_labels
             _lowerCamelCase				:  Optional[Any]						=		num_labels
             _lowerCamelCase				:  Optional[Any]						=		initializer_range


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Union[str, Any]						=		floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size])

             _lowerCamelCase				:  Any						=		None
             if self.use_labels:
                    _lowerCamelCase				:  Tuple						=		ids_tensor([self.batch_size]   ,				self.num_labels)

             _lowerCamelCase				:  int						=		self.get_config()
             return config, pixel_values, labels


      def 					__snake_case      (		self):



             """simple docstring"""





             return LevitConfig(
                 image_size=self.image_size   ,				num_channels=self.num_channels   ,				kernel_size=self.kernel_size   ,				stride=self.stride   ,				padding=self.padding   ,				patch_size=self.patch_size   ,				hidden_sizes=self.hidden_sizes   ,				num_attention_heads=self.num_attention_heads   ,				depths=self.depths   ,				key_dim=self.key_dim   ,				drop_path_rate=self.drop_path_rate   ,				mlp_ratio=self.mlp_ratio   ,				attention_ratio=self.attention_ratio   ,				initializer_range=self.initializer_range   ,				down_ops=self.down_ops   ,				)


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



             """simple docstring"""





             _lowerCamelCase				:  Any						=		LevitModel(config=a__)
             model.to(a__)
             model.eval()
             _lowerCamelCase				:  int						=		model(a__)
             _lowerCamelCase				:  List[Any]						=		(self.image_size, self.image_size)
             _lowerCamelCase,	_lowerCamelCase				:  Any						=		image_size[0], image_size[1]
             for _ in range(4):
                    _lowerCamelCase				:  Tuple						=		floor(((height + 2 * self.padding - self.kernel_size) / self.stride) + 1)
                    _lowerCamelCase				:  List[Any]						=		floor(((width + 2 * self.padding - self.kernel_size) / self.stride) + 1)
             self.parent.assertEqual(
                 result.last_hidden_state.shape   ,				(self.batch_size, ceil(height / 4) * ceil(width / 4), self.hidden_sizes[-1])   ,				)


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



             """simple docstring"""





             _lowerCamelCase				:  int						=		self.num_labels
             _lowerCamelCase				:  Dict						=		LevitForImageClassification(a__)
             model.to(a__)
             model.eval()
             _lowerCamelCase				:  Any						=		model(a__   ,				labels=a__)
             self.parent.assertEqual(result.logits.shape   ,				(self.batch_size, self.num_labels))



      def 					__snake_case      (		self):



             """simple docstring"""





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







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




      """simple docstring"""







      UpperCAmelCase__							=   (
          (LevitModel, LevitForImageClassification, LevitForImageClassificationWithTeacher)
          if is_torch_available()
          else ()
      )
      UpperCAmelCase__							=   (
          {
              """feature-extraction""": LevitModel,
              """image-classification""": (LevitForImageClassification, LevitForImageClassificationWithTeacher),
          }
          if is_torch_available()
          else {}
      )

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


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Optional[Any]						=		LevitModelTester(self)
             _lowerCamelCase				:  Union[str, Any]						=		ConfigTester(self   ,				config_class=a__   ,				has_text_modality=a__   ,				hidden_size=37)


      def 					__snake_case      (		self):



             """simple docstring"""





             self.create_and_test_config_common_properties()
             self.config_tester.create_and_test_config_to_json_string()
             self.config_tester.create_and_test_config_to_json_file()
             self.config_tester.create_and_test_config_from_and_save_pretrained()
             self.config_tester.create_and_test_config_with_num_labels()
             self.config_tester.check_config_can_be_init_without_params()
             self.config_tester.check_config_arguments_init()


      def 					__snake_case      (		self):



             """simple docstring"""





             return


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



             """simple docstring"""





             pass


      @unittest.skip(reason='''Levit does not support input and output embeddings''')
      def 					__snake_case      (		self):



             """simple docstring"""





             pass


      @unittest.skip(reason='''Levit does not output attentions''')
      def 					__snake_case      (		self):



             """simple docstring"""





             pass


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase,	_lowerCamelCase				:  List[str]						=		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				:  int						=		inspect.signature(model.forward)
                    # signature.parameters is an OrderedDict => so arg_names order is deterministic
                    _lowerCamelCase				:  List[Any]						=		[*signature.parameters.keys()]

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


      def 					__snake_case      (		self):



             """simple docstring"""





             def check_hidden_states_output(a__   ,				a__   ,				a__):
                    _lowerCamelCase				:  Any						=		model_class(a__)
                    model.to(a__)
                    model.eval()

                    with torch.no_grad():
                           _lowerCamelCase				:  str						=		model(**self._prepare_for_class(a__   ,				a__))

                    _lowerCamelCase				:  Optional[Any]						=		outputs.hidden_states

                    _lowerCamelCase				:  Tuple						=		len(self.model_tester.depths) + 1
                    self.assertEqual(len(a__)   ,				a__)

                    _lowerCamelCase				:  List[str]						=		(self.model_tester.image_size, self.model_tester.image_size)
                    _lowerCamelCase,	_lowerCamelCase				:  Union[str, Any]						=		image_size[0], image_size[1]
                    for _ in range(4):
                           _lowerCamelCase				:  Union[str, Any]						=		floor(
                               (
                                   (height + 2 * self.model_tester.padding - self.model_tester.kernel_size)
                                   / self.model_tester.stride
                               )
                               + 1)
                           _lowerCamelCase				:  int						=		floor(
                               (
                                   (width + 2 * self.model_tester.padding - self.model_tester.kernel_size)
                                   / self.model_tester.stride
                               )
                               + 1)
                    # verify the first hidden states (first block)
                    self.assertListEqual(
                        list(hidden_states[0].shape[-2:])   ,				[
                            height * width,
                            self.model_tester.hidden_sizes[0],
                        ]   ,				)

             _lowerCamelCase,	_lowerCamelCase				:  int						=		self.model_tester.prepare_config_and_inputs_for_common()

             for model_class in self.all_model_classes:
                    _lowerCamelCase				:  Optional[Any]						=		True
                    check_hidden_states_output(a__   ,				a__   ,				a__)

                    # check that output_hidden_states also work using config
                    del inputs_dict["output_hidden_states"]
                    _lowerCamelCase				:  Optional[Any]						=		True

                    check_hidden_states_output(a__   ,				a__   ,				a__)


      @unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.''')
      def 					__snake_case      (		self):



             """simple docstring"""





             pass


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



             """simple docstring"""





             _lowerCamelCase				:  Dict						=		super()._prepare_for_class(a__   ,				a__   ,				return_labels=a__)

             if return_labels:
                    if model_class.__name__ == "LevitForImageClassificationWithTeacher":
                           del inputs_dict["labels"]

             return inputs_dict


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  str						=		self.model_tester.prepare_config_and_inputs()
             self.model_tester.create_and_check_model(*a__)


      def 					__snake_case      (		self):



             """simple docstring"""





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


      def 					__snake_case      (		self):



             """simple docstring"""





             if not self.model_tester.is_training:
                    return

             _lowerCamelCase,	_lowerCamelCase				:  int						=		self.model_tester.prepare_config_and_inputs_for_common()
             _lowerCamelCase				:  List[str]						=		True

             for model_class in self.all_model_classes:
                    # LevitForImageClassificationWithTeacher supports inference-only
                    if (
                        model_class in get_values(a__)
                        or model_class.__name__ == "LevitForImageClassificationWithTeacher"
                    ):
                           continue
                    _lowerCamelCase				:  Union[str, Any]						=		model_class(a__)
                    model.to(a__)
                    model.train()
                    _lowerCamelCase				:  List[Any]						=		self._prepare_for_class(a__   ,				a__   ,				return_labels=a__)
                    _lowerCamelCase				:  List[Any]						=		model(**a__).loss
                    loss.backward()


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase,	_lowerCamelCase				:  Union[str, Any]						=		self.model_tester.prepare_config_and_inputs_for_common()
             if not self.model_tester.is_training:
                    return

             _lowerCamelCase				:  Any						=		False
             _lowerCamelCase				:  Optional[int]						=		True

             for model_class in self.all_model_classes:
                    if model_class in get_values(a__) or not model_class.supports_gradient_checkpointing:
                           continue
                    # LevitForImageClassificationWithTeacher supports inference-only
                    if model_class.__name__ == "LevitForImageClassificationWithTeacher":
                           continue
                    _lowerCamelCase				:  int						=		model_class(a__)
                    model.gradient_checkpointing_enable()
                    model.to(a__)
                    model.train()
                    _lowerCamelCase				:  Dict						=		self._prepare_for_class(a__   ,				a__   ,				return_labels=a__)
                    _lowerCamelCase				:  Optional[Any]						=		model(**a__).loss
                    loss.backward()


      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase,	_lowerCamelCase				:  List[str]						=		self.model_tester.prepare_config_and_inputs_for_common()

             _lowerCamelCase				:  Tuple						=		[
                 {'''title''': '''multi_label_classification''', '''num_labels''': 2, '''dtype''': torch.float},
                 {'''title''': '''single_label_classification''', '''num_labels''': 1, '''dtype''': torch.long},
                 {'''title''': '''regression''', '''num_labels''': 1, '''dtype''': torch.float},
             ]

             for model_class in self.all_model_classes:
                    if (
                        model_class
                        not in [
                            *get_values(a__),
                        ]
                        or model_class.__name__ == "LevitForImageClassificationWithTeacher"
                    ):
                           continue

                    for problem_type in problem_types:
                           with self.subTest(msg=F"""Testing {model_class} with {problem_type['title']}"""):
                                  _lowerCamelCase				:  Optional[int]						=		problem_type['''title''']
                                  _lowerCamelCase				:  Optional[int]						=		problem_type['''num_labels''']

                                  _lowerCamelCase				:  str						=		model_class(a__)
                                  model.to(a__)
                                  model.train()

                                  _lowerCamelCase				:  Dict						=		self._prepare_for_class(a__   ,				a__   ,				return_labels=a__)

                                  if problem_type["num_labels"] > 1:
                                         _lowerCamelCase				:  Any						=		inputs['''labels'''].unsqueeze(1).repeat(1   ,				problem_type['''num_labels'''])

                                  _lowerCamelCase				:  Union[str, Any]						=		inputs['''labels'''].to(problem_type['''dtype'''])

                                  # This tests that we do not trigger the warning form PyTorch "Using a target size that is different
                                  # to the input size. This will likely lead to incorrect results due to broadcasting. Please ensure
                                  # they have the same size." which is a symptom something in wrong for the regression problem.
                                  # See https://github.com/huggingface/transformers/issues/11780
                                  with warnings.catch_warnings(record=a__) as warning_list:
                                         _lowerCamelCase				:  Any						=		model(**a__).loss
                                  for w in warning_list:
                                         if "Using a target size that is different to the input size" in str(w.message):
                                                raise ValueError(
                                                    F"""Something is going wrong in the regression problem: intercepted {w.message}""")

                                  loss.backward()



      @slow
      def 					__snake_case      (		self):



             """simple docstring"""





             for model_name in LEVIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                    _lowerCamelCase				:  Optional[int]						=		LevitModel.from_pretrained(a__)
                    self.assertIsNotNone(a__)




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







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




      """simple docstring"""







      @cached_property
      def 					__snake_case      (		self):



             """simple docstring"""





             return LevitImageProcessor.from_pretrained(LEVIT_PRETRAINED_MODEL_ARCHIVE_LIST[0])



      @slow
      def 					__snake_case      (		self):



             """simple docstring"""





             _lowerCamelCase				:  Any						=		LevitForImageClassificationWithTeacher.from_pretrained(LEVIT_PRETRAINED_MODEL_ARCHIVE_LIST[0]).to(
                 a__)

             _lowerCamelCase				:  Any						=		self.default_image_processor
             _lowerCamelCase				:  Optional[Any]						=		prepare_img()
             _lowerCamelCase				:  Any						=		image_processor(images=a__   ,				return_tensors='''pt''').to(a__)

             # forward pass
             with torch.no_grad():
                    _lowerCamelCase				:  List[Any]						=		model(**a__)

             # verify the logits
             _lowerCamelCase				:  Union[str, Any]						=		torch.Size((1, 1000))
             self.assertEqual(outputs.logits.shape   ,				a__)

             _lowerCamelCase				:  str						=		torch.tensor([1.0448, -0.3745, -1.8317]).to(a__)

             self.assertTrue(torch.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__)

