



"""simple docstring"""


import inspect
import unittest
import warnings

from transformers import DeiTConfig
from transformers.models.auto import get_values
from transformers.testing_utils import (
    require_accelerate,
    require_torch,
    require_torch_gpu,
    require_vision,
    slow,
    torch_device,
)
from transformers.utils import cached_property, is_torch_available, is_vision_available

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 torch import nn

	from transformers import (
	    MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
	    MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
	    MODEL_MAPPING,
	    DeiTForImageClassification,
	    DeiTForImageClassificationWithTeacher,
	    DeiTForMaskedImageModeling,
	    DeiTModel,
	)
	from transformers.models.deit.modeling_deit import DEIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
	from PIL import Image

	from transformers import DeiTImageProcessor



class 			lowerCamelCase__						:







						def __init__(							self   :    Tuple			,			_lowercase   :    int			,			_lowercase   :    Tuple=13			,			_lowercase   :    List[str]=30			,			_lowercase   :    Tuple=2			,			_lowercase   :    Any=3			,			_lowercase   :    Dict=True			,			_lowercase   :    Optional[int]=True			,			_lowercase   :    int=32			,			_lowercase   :    List[str]=5			,			_lowercase   :    Any=4			,			_lowercase   :    Optional[int]=37			,			_lowercase   :    List[Any]="gelu"			,			_lowercase   :    int=0.1			,			_lowercase   :    Union[str, Any]=0.1			,			_lowercase   :    Optional[Any]=10			,			_lowercase   :    Dict=0.0_2			,			_lowercase   :    Tuple=3			,			_lowercase   :    Union[str, Any]=None			,			_lowercase   :    List[Any]=2			,			):
								A					       = parent
								A					       = batch_size
								A					       = image_size
								A					       = patch_size
								A					       = num_channels
								A					       = is_training
								A					       = use_labels
								A					       = hidden_size
								A					       = num_hidden_layers
								A					       = num_attention_heads
								A					       = intermediate_size
								A					       = hidden_act
								A					       = hidden_dropout_prob
								A					       = attention_probs_dropout_prob
								A					       = type_sequence_label_size
								A					       = initializer_range
								A					       = scope
								A					       = encoder_stride

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







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

								A					       = None
								if self.use_labels:
										A					       = ids_tensor([self.batch_size]			,			self.type_sequence_label_size		)

								A					       = self.get_config()

								return config, pixel_values, labels







						def __a						(							self   :    str		):
								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=__lowerCamelCase			,			initializer_range=self.initializer_range			,			encoder_stride=self.encoder_stride			,			)







						def __a						(							self   :    int			,			_lowercase   :    Optional[Any]			,			_lowercase   :    int			,			_lowercase   :    List[str]		):
								A					       = DeiTModel(config=__lowerCamelCase		)
								model.to(__lowerCamelCase		)
								model.eval()
								A					       = model(__lowerCamelCase		)
								self.parent.assertEqual(result.last_hidden_state.shape			,			(self.batch_size, self.seq_length, self.hidden_size)		)







						def __a						(							self   :    List[Any]			,			_lowercase   :    Optional[Any]			,			_lowercase   :    Dict			,			_lowercase   :    Any		):
								A					       = DeiTForMaskedImageModeling(config=__lowerCamelCase		)
								model.to(__lowerCamelCase		)
								model.eval()
								A					       = model(__lowerCamelCase		)
								self.parent.assertEqual(
								    result.reconstruction.shape			,			(self.batch_size, self.num_channels, self.image_size, self.image_size)		)

								# test greyscale images
								A					       = 1
								A					       = DeiTForMaskedImageModeling(__lowerCamelCase		)
								model.to(__lowerCamelCase		)
								model.eval()

								A					       = floats_tensor([self.batch_size, 1, self.image_size, self.image_size]		)
								A					       = model(__lowerCamelCase		)
								self.parent.assertEqual(result.reconstruction.shape			,			(self.batch_size, 1, self.image_size, self.image_size)		)







						def __a						(							self   :    Any			,			_lowercase   :    str			,			_lowercase   :    List[str]			,			_lowercase   :    int		):
								A					       = self.type_sequence_label_size
								A					       = DeiTForImageClassification(__lowerCamelCase		)
								model.to(__lowerCamelCase		)
								model.eval()
								A					       = model(__lowerCamelCase			,			labels=__lowerCamelCase		)
								self.parent.assertEqual(result.logits.shape			,			(self.batch_size, self.type_sequence_label_size)		)

								# test greyscale images
								A					       = 1
								A					       = DeiTForImageClassification(__lowerCamelCase		)
								model.to(__lowerCamelCase		)
								model.eval()

								A					       = floats_tensor([self.batch_size, 1, self.image_size, self.image_size]		)
								A					       = model(__lowerCamelCase			,			labels=__lowerCamelCase		)
								self.parent.assertEqual(result.logits.shape			,			(self.batch_size, self.type_sequence_label_size)		)







						def __a						(							self   :    Any		):
								A					       = self.prepare_config_and_inputs()
								(
								    A 
								)					       = config_and_inputs
								A					       = {"pixel_values": pixel_values}
								return config, inputs_dict



@require_torch
class 			lowerCamelCase__						(   _A							,    _A							,    unittest.TestCase			):
						lowerCAmelCase       =     (
						    (
						        DeiTModel,
						        DeiTForImageClassification,
						        DeiTForImageClassificationWithTeacher,
						        DeiTForMaskedImageModeling,
						    )
						    if is_torch_available()
						    else ()
						)
						lowerCAmelCase       =     (
						    {
						        """feature-extraction""": DeiTModel,
						        """image-classification""": (DeiTForImageClassification, DeiTForImageClassificationWithTeacher),
						    }
						    if is_torch_available()
						    else {}
						)

						lowerCAmelCase       =     False
						lowerCAmelCase       =     False
						lowerCAmelCase       =     False







						def __a						(							self   :    int		):
								A					       = DeiTModelTester(self		)
								A					       = ConfigTester(self			,			config_class=__lowerCamelCase			,			has_text_modality=__lowerCamelCase			,			hidden_size=37		)







						def __a						(							self   :    Dict		):
								self.config_tester.run_common_tests()







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







						def __a						(							self   :    Any		):
								A					       = self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
										A					       = model_class(__lowerCamelCase		)
										self.assertIsInstance(model.get_input_embeddings()			,			(nn.Module)		)
										A					       = model.get_output_embeddings()
										self.assertTrue(x is None or isinstance(__lowerCamelCase			,			nn.Linear		)		)







						def __a						(							self   :    Any		):
								A					       = self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
										A					       = model_class(__lowerCamelCase		)
										A					       = inspect.signature(model.forward		)
										# signature.parameters is an OrderedDict => so arg_names order is deterministic
										A					       = [*signature.parameters.keys()]

										A					       = ["pixel_values"]
										self.assertListEqual(arg_names[:1]			,			__lowerCamelCase		)







						def __a						(							self   :    int		):
								A					       = self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*__lowerCamelCase		)







						def __a						(							self   :    Optional[Any]		):
								A					       = self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_masked_image_modeling(*__lowerCamelCase		)







						def __a						(							self   :    int		):
								A					       = self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_image_classification(*__lowerCamelCase		)







						def __a						(							self   :    Tuple			,			_lowercase   :    int			,			_lowercase   :    Optional[Any]			,			_lowercase   :    List[Any]=False		):
								A					       = super()._prepare_for_class(__lowerCamelCase			,			__lowerCamelCase			,			return_labels=__lowerCamelCase		)

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

								return inputs_dict







						def __a						(							self   :    Optional[int]		):
								if not self.model_tester.is_training:
										return

								A					       = self.model_tester.prepare_config_and_inputs_for_common()
								A					       = True

								for model_class in self.all_model_classes:
										# DeiTForImageClassificationWithTeacher supports inference-only
										if (
										    model_class in get_values(__lowerCamelCase		)
										    or model_class.__name__ == "DeiTForImageClassificationWithTeacher"
										):
												continue
										A					       = model_class(__lowerCamelCase		)
										model.to(__lowerCamelCase		)
										model.train()
										A					       = self._prepare_for_class(__lowerCamelCase			,			__lowerCamelCase			,			return_labels=__lowerCamelCase		)
										A					       = model(**__lowerCamelCase		).loss
										loss.backward()







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

								A					       = False
								A					       = True

								for model_class in self.all_model_classes:
										if model_class in get_values(__lowerCamelCase		) or not model_class.supports_gradient_checkpointing:
												continue
										# DeiTForImageClassificationWithTeacher supports inference-only
										if model_class.__name__ == "DeiTForImageClassificationWithTeacher":
												continue
										A					       = model_class(__lowerCamelCase		)
										model.gradient_checkpointing_enable()
										model.to(__lowerCamelCase		)
										model.train()
										A					       = self._prepare_for_class(__lowerCamelCase			,			__lowerCamelCase			,			return_labels=__lowerCamelCase		)
										A					       = model(**__lowerCamelCase		).loss
										loss.backward()







						def __a						(							self   :    str		):
								A					       = self.model_tester.prepare_config_and_inputs_for_common()

								A					       = [
								    {"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(__lowerCamelCase		),
										        *get_values(__lowerCamelCase		),
										    ]
										    or model_class.__name__ == "DeiTForImageClassificationWithTeacher"
										):
												continue

										for problem_type in problem_types:
												with self.subTest(msg=f'Testing {model_class} with {problem_type["title"]}'		):
														A					       = problem_type["title"]
														A					       = problem_type["num_labels"]

														A					       = model_class(__lowerCamelCase		)
														model.to(__lowerCamelCase		)
														model.train()

														A					       = self._prepare_for_class(__lowerCamelCase			,			__lowerCamelCase			,			return_labels=__lowerCamelCase		)

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

														A					       = 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=__lowerCamelCase		) as warning_list:
																A					       = model(**__lowerCamelCase		).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 __a						(							self   :    List[Any]		):
								for model_name in DEIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
										A					       = DeiTModel.from_pretrained(__lowerCamelCase		)
										self.assertIsNotNone(__lowerCamelCase		)







def  __snake_case       (   )      ->		str:







		"""simple docstring"""





		A					       = Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'	)
		return image



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







						@cached_property
						def __a						(							self   :    Tuple		):
								return (
								    DeiTImageProcessor.from_pretrained('facebook/deit-base-distilled-patch16-224'		)
								    if is_vision_available()
								    else None
								)







						@slow
						def __a						(							self   :    Any		):
								A					       = DeiTForImageClassificationWithTeacher.from_pretrained('facebook/deit-base-distilled-patch16-224'		).to(
								    __lowerCamelCase		)

								A					       = self.default_image_processor
								A					       = prepare_img()
								A					       = image_processor(images=__lowerCamelCase			,			return_tensors='pt'		).to(__lowerCamelCase		)

								# forward pass
								with torch.no_grad():
										A					       = model(**__lowerCamelCase		)

								# verify the logits
								A					       = torch.Size((1, 1_000)		)
								self.assertEqual(outputs.logits.shape			,			__lowerCamelCase		)

								A					       = torch.tensor([-1.0_2_6_6, 0.1_9_1_2, -1.2_8_6_1]		).to(__lowerCamelCase		)

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







						@slow
						@require_accelerate
						@require_torch_gpu
						def __a						(							self   :    Union[str, Any]		):
								A					       = DeiTModel.from_pretrained(
								    'facebook/deit-base-distilled-patch16-224'			,			torch_dtype=torch.floataa			,			device_map='auto'		)
								A					       = self.default_image_processor

								A					       = prepare_img()
								A					       = image_processor(images=__lowerCamelCase			,			return_tensors='pt'		)
								A					       = inputs.pixel_values.to(__lowerCamelCase		)

								# forward pass to make sure inference works in fp16
								with torch.no_grad():
										A					       = model(__lowerCamelCase		)









"""simple docstring"""


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

import pytest
from parameterized import parameterized, 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.p3.16xlarge""",
            """results""": {"""train_runtime""": 650, """eval_accuracy""": 0.7, """eval_loss""": 0.6},
        },
        {
            """framework""": """pytorch""",
            """script""": """run_ddp.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 600, """eval_accuracy""": 0.7, """eval_loss""": 0.6},
        },
        {
            """framework""": """tensorflow""",
            """script""": """run_tf_dist.py""",
            """model_name_or_path""": """distilbert-base-cased""",
            """instance_type""": """ml.p3.16xlarge""",
            """results""": {"""train_runtime""": 600, """eval_accuracy""": 0.6, """eval_loss""": 0.7},
        },
    ]			)
class 			lowerCamelCase__						(   unittest.TestCase			):







						def __a						(							self   :    int		):
								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=_lowercase			,			)
								assert hasattr(self			,			'env'		)







						def __a						(							self   :    Optional[int]			,			_lowercase   :    int		):
								A					       = f'{self.env.base_job_name}-{instance_count}-{"ddp" if "ddp" in self.script else "smd"}'
								# distributed data settings
								A					       = {'smdistributed': {'dataparallel': {'enabled': True}}} if self.script != 'run_ddp.py' else None

								# creates estimator
								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=_lowercase			,			instance_count=_lowercase			,			instance_type=self.instance_type			,			debugger_hook_config=_lowercase			,			hyperparameters={**self.env.distributed_hyperparameters, 'model_name_or_path': self.model_name_or_path}			,			metric_definitions=self.env.metric_definitions			,			distribution=_lowercase			,			py_version='py36'			,			)







						def __a						(							self   :    Tuple			,			_lowercase   :    List[str]		):
								TrainingJobAnalytics(_lowercase		).export_csv(f'{self.env.test_path}/{job_name}_metrics.csv'		)







						@parameterized.expand([(2,)]		)
						def __a						(							self   :    List[Any]			,			_lowercase   :    Union[str, Any]		):
								# create estimator
								A					       = self.create_estimator(_lowercase		)

								# run training
								estimator.fit()

								# result dataframe
								A					       = TrainingJobAnalytics(estimator.latest_training_job.name		).dataframe()

								# extract kpis
								A					       = list(result_metrics_df[result_metrics_df.metric_name == 'eval_accuracy']['value']		)
								A					       = list(result_metrics_df[result_metrics_df.metric_name == 'eval_loss']['value']		)
								# get train time from SageMaker job, this includes starting, preprocessing, stopping
								A					       = (
								    Session().describe_training_job(estimator.latest_training_job.name		).get('TrainingTimeInSeconds'			,			999_999		)
								)

								# 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}			,			_lowercase		)





