

"""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_model_parallelism.py",
            "model_name_or_path": "roberta-large",
            "instance_type": "ml.p3dn.24xlarge",
            "results": {"train_runtime": 1_6_0_0, "eval_accuracy": 0.3, "eval_loss": 1.2},
        },
        {
            "framework": "pytorch",
            "script": "run_glue.py",
            "model_name_or_path": "roberta-large",
            "instance_type": "ml.p3dn.24xlarge",
            "results": {"train_runtime": 1_6_0_0, "eval_accuracy": 0.3, "eval_loss": 1.2},
        },
    ]   )
class 					_lowerCAmelCase  (	unittest.TestCase   ):







 """simple docstring"""







 def 					_lowercase (			self							:			Optional[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=UpperCAmelCase__,					)
    assert hasattr(self,					"env"	)


 def 					_lowercase (			self							:			str,					UpperCAmelCase__							:			List[Any]	):
    # configuration for running training on smdistributed Model Parallel
    __lowercase								=						{
        "enabled": True,
        "processes_per_host": 8,
    }
    __lowercase								=						{
        "enabled": True,
        "parameters": {
            "microbatches": 4,
            "placement_strategy": "spread",
            "pipeline": "interleaved",
            "optimize": "speed",
            "partitions": 4,
            "ddp": True,
        },
    }

    __lowercase								=						{"smdistributed": {"modelparallel": smp_options}, "mpi": mpi_options}

    __lowercase								=						"trainer" if self.script == "run_glue.py" else "smtrainer"
    # 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=F"""{self.env.base_job_name}-{instance_count}-smp-{name_extension}""",					instance_count=UpperCAmelCase__,					instance_type=self.instance_type,					debugger_hook_config=UpperCAmelCase__,					hyperparameters={
            **self.env.hyperparameters,
            "model_name_or_path": self.model_name_or_path,
            "max_steps": 5_0_0,
        },					metric_definitions=self.env.metric_definitions,					distribution=UpperCAmelCase__,					py_version="py36",					)


 def 					_lowercase (			self							:			Tuple,					UpperCAmelCase__							:			int	):
    TrainingJobAnalytics(UpperCAmelCase__	).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv"""	)



 @parameterized.expand([(1,)]	)
 def 					_lowercase (			self							:			str,					UpperCAmelCase__							:			Union[str, Any]	):
    # create estimator
    __lowercase								=						self.create_estimator(UpperCAmelCase__	)

    # run training
    estimator.fit()

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

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

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





"""simple docstring"""







import inspect
import unittest

from transformers import ViTConfig
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 ViTForImageClassification, ViTForMaskedImageModeling, ViTModel
       from transformers.models.vit.modeling_vit import VIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
       from PIL import Image

       from transformers import ViTImageProcessor







class        UpperCamelCase		:
   def __init__(			self     ,__UpperCamelCase     ,__UpperCamelCase=13     ,__UpperCamelCase=30     ,__UpperCamelCase=2     ,__UpperCamelCase=3     ,__UpperCamelCase=True     ,__UpperCamelCase=True     ,__UpperCamelCase=32     ,__UpperCamelCase=5     ,__UpperCamelCase=4     ,__UpperCamelCase=37     ,__UpperCamelCase="gelu"     ,__UpperCamelCase=0.1     ,__UpperCamelCase=0.1     ,__UpperCamelCase=10     ,__UpperCamelCase=0.02     ,__UpperCamelCase=None     ,__UpperCamelCase=2     ,)	->      List[Any]:


        '''simple docstring'''


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

        # in ViT, the seq length equals the number of patches + 1 (we add 1 for the [CLS] token)
        lowercase_	:				Any													=	(image_size // patch_size) ** 2
        lowercase_	:				Optional[int]													=	num_patches + 1
   def 			_UpperCAmelCase   (			self			)	->      List[Any]:


        '''simple docstring'''


        lowercase_	:				List[Any]													=	floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]			)

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

        lowercase_	:				Any													=	self.get_config()

        return config, pixel_values, labels
   def 			_UpperCAmelCase   (			self			)	->      Optional[Any]:


        '''simple docstring'''


        return ViTConfig(
            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=__UpperCamelCase     ,initializer_range=self.initializer_range     ,encoder_stride=self.encoder_stride     ,)
   def 			_UpperCAmelCase   (			self     ,__UpperCamelCase     ,__UpperCamelCase     ,__UpperCamelCase			)	->      Optional[int]:


        '''simple docstring'''


        lowercase_	:				List[Any]													=	ViTModel(config=__UpperCamelCase			)
        model.to(__UpperCamelCase			)
        model.eval()
        lowercase_	:				Optional[Any]													=	model(__UpperCamelCase			)
        self.parent.assertEqual(result.last_hidden_state.shape     ,(self.batch_size, self.seq_length, self.hidden_size)			)
   def 			_UpperCAmelCase   (			self     ,__UpperCamelCase     ,__UpperCamelCase     ,__UpperCamelCase			)	->      Any:


        '''simple docstring'''


        lowercase_	:				Optional[int]													=	ViTForMaskedImageModeling(config=__UpperCamelCase			)
        model.to(__UpperCamelCase			)
        model.eval()
        lowercase_	:				List[str]													=	model(__UpperCamelCase			)
        self.parent.assertEqual(
            result.reconstruction.shape     ,(self.batch_size, self.num_channels, self.image_size, self.image_size)			)

        # test greyscale images
        lowercase_	:				Any													=	1
        lowercase_	:				List[Any]													=	ViTForMaskedImageModeling(__UpperCamelCase			)
        model.to(__UpperCamelCase			)
        model.eval()

        lowercase_	:				Dict													=	floats_tensor([self.batch_size, 1, self.image_size, self.image_size]			)
        lowercase_	:				str													=	model(__UpperCamelCase			)
        self.parent.assertEqual(result.reconstruction.shape     ,(self.batch_size, 1, self.image_size, self.image_size)			)
   def 			_UpperCAmelCase   (			self     ,__UpperCamelCase     ,__UpperCamelCase     ,__UpperCamelCase			)	->      str:


        '''simple docstring'''


        lowercase_	:				Dict													=	self.type_sequence_label_size
        lowercase_	:				Dict													=	ViTForImageClassification(__UpperCamelCase			)
        model.to(__UpperCamelCase			)
        model.eval()
        lowercase_	:				int													=	model(__UpperCamelCase     ,labels=__UpperCamelCase			)
        self.parent.assertEqual(result.logits.shape     ,(self.batch_size, self.type_sequence_label_size)			)

        # test greyscale images
        lowercase_	:				Dict													=	1
        lowercase_	:				Any													=	ViTForImageClassification(__UpperCamelCase			)
        model.to(__UpperCamelCase			)
        model.eval()

        lowercase_	:				Dict													=	floats_tensor([self.batch_size, 1, self.image_size, self.image_size]			)
        lowercase_	:				Dict													=	model(__UpperCamelCase			)
        self.parent.assertEqual(result.logits.shape     ,(self.batch_size, self.type_sequence_label_size)			)







   def 			_UpperCAmelCase   (			self			)	->      List[str]:


        '''simple docstring'''


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







@require_torch
class        UpperCamelCase		(			lowercase_						,      lowercase_						,      unittest.TestCase  ):
   lowercase							=  (
       (
           ViTModel,
           ViTForImageClassification,
           ViTForMaskedImageModeling,
       )
       if is_torch_available()
       else ()
   )
   lowercase							=  (
       {'feature-extraction': ViTModel, 'image-classification': ViTForImageClassification}
       if is_torch_available()
       else {}
   )
   lowercase							=  True

   lowercase							=  False
   lowercase							=  False
   lowercase							=  False
   def 			_UpperCAmelCase   (			self			)	->      List[str]:


        '''simple docstring'''


        lowercase_	:				Optional[int]													=	ViTModelTester(self			)
        lowercase_	:				Dict													=	ConfigTester(self     ,config_class=__UpperCamelCase     ,has_text_modality=__UpperCamelCase     ,hidden_size=37			)
   def 			_UpperCAmelCase   (			self			)	->      Optional[Any]:


        '''simple docstring'''


        self.config_tester.run_common_tests()
   @unittest.skip(reason='ViT does not use inputs_embeds'			)
   def 			_UpperCAmelCase   (			self			)	->      Any:


        '''simple docstring'''


        pass
   def 			_UpperCAmelCase   (			self			)	->      Dict:


        '''simple docstring'''


        lowercase_							,  lowercase_	:				Any													=	self.model_tester.prepare_config_and_inputs_for_common()

        for model_class in self.all_model_classes:
             lowercase_	:				str													=	model_class(__UpperCamelCase			)
             self.assertIsInstance(model.get_input_embeddings()     ,(nn.Module)			)
             lowercase_	:				List[Any]													=	model.get_output_embeddings()
             self.assertTrue(x is None or isinstance(__UpperCamelCase     ,nn.Linear			)			)
   def 			_UpperCAmelCase   (			self			)	->      Union[str, Any]:


        '''simple docstring'''


        lowercase_							,  lowercase_	:				Union[str, Any]													=	self.model_tester.prepare_config_and_inputs_for_common()

        for model_class in self.all_model_classes:
             lowercase_	:				str													=	model_class(__UpperCamelCase			)
             lowercase_	:				List[Any]													=	inspect.signature(model.forward			)
             # signature.parameters is an OrderedDict => so arg_names order is deterministic
             lowercase_	:				Tuple													=	[*signature.parameters.keys()]

             lowercase_	:				Tuple													=	['pixel_values']
             self.assertListEqual(arg_names[:1]     ,__UpperCamelCase			)
   def 			_UpperCAmelCase   (			self			)	->      Any:


        '''simple docstring'''


        lowercase_	:				List[Any]													=	self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*__UpperCamelCase			)
   def 			_UpperCAmelCase   (			self			)	->      str:


        '''simple docstring'''


        lowercase_	:				str													=	self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_masked_image_modeling(*__UpperCamelCase			)
   def 			_UpperCAmelCase   (			self			)	->      str:


        '''simple docstring'''


        lowercase_	:				str													=	self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_image_classification(*__UpperCamelCase			)







   @slow
   def 			_UpperCAmelCase   (			self			)	->      Tuple:


        '''simple docstring'''


        for model_name in VIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
             lowercase_	:				List[Any]													=	ViTModel.from_pretrained(__UpperCamelCase			)
             self.assertIsNotNone(__UpperCamelCase			)






def 			lowercase__(       ):
     lowercase_	:				Any													=	Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'      )
     return image







@require_torch
@require_vision
class        UpperCamelCase		(			unittest.TestCase  ):
   @cached_property
   def 			_UpperCAmelCase   (			self			)	->      Any:


        '''simple docstring'''


        return ViTImageProcessor.from_pretrained('google/vit-base-patch16-224'			) if is_vision_available() else None
   @slow
   def 			_UpperCAmelCase   (			self			)	->      int:


        '''simple docstring'''


        lowercase_	:				str													=	ViTForImageClassification.from_pretrained('google/vit-base-patch16-224'			).to(__UpperCamelCase			)

        lowercase_	:				Any													=	self.default_image_processor
        lowercase_	:				Dict													=	prepare_img()
        lowercase_	:				Tuple													=	image_processor(images=__UpperCamelCase     ,return_tensors='pt'			).to(__UpperCamelCase			)

        # forward pass
        with torch.no_grad():
             lowercase_	:				Any													=	model(**__UpperCamelCase			)

        # verify the logits
        lowercase_	:				Optional[Any]													=	torch.Size((1, 1000)			)
        self.assertEqual(outputs.logits.shape     ,__UpperCamelCase			)

        lowercase_	:				Optional[int]													=	torch.tensor([-0.2744, 0.8215, -0.0836]			).to(__UpperCamelCase			)

        self.assertTrue(torch.allclose(outputs.logits[0, :3]     ,__UpperCamelCase     ,atol=1e-4			)			)
   @slow
   def 			_UpperCAmelCase   (			self			)	->      Tuple:


        '''simple docstring'''


        lowercase_	:				str													=	ViTModel.from_pretrained('facebook/dino-vits8'			).to(__UpperCamelCase			)

        lowercase_	:				int													=	ViTImageProcessor.from_pretrained('facebook/dino-vits8'     ,size=480			)
        lowercase_	:				int													=	prepare_img()
        lowercase_	:				Dict													=	image_processor(images=__UpperCamelCase     ,return_tensors='pt'			)
        lowercase_	:				int													=	inputs.pixel_values.to(__UpperCamelCase			)

        # forward pass
        with torch.no_grad():
             lowercase_	:				int													=	model(__UpperCamelCase     ,interpolate_pos_encoding=__UpperCamelCase			)

        # verify the logits
        lowercase_	:				Any													=	torch.Size((1, 3601, 384)			)
        self.assertEqual(outputs.last_hidden_state.shape     ,__UpperCamelCase			)

        lowercase_	:				Any													=	torch.tensor(
            [[4.2340, 4.3906, -6.6692], [4.5463, 1.8928, -6.7257], [4.4429, 0.8496, -5.8585]]			).to(__UpperCamelCase			)

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







   @slow
   @require_accelerate
   @require_torch_gpu
   def 			_UpperCAmelCase   (			self			)	->      int:


        '''simple docstring'''


        lowercase_	:				Optional[int]													=	ViTModel.from_pretrained('facebook/dino-vits8'     ,torch_dtype=torch.floataa     ,device_map='auto'			)
        lowercase_	:				int													=	self.default_image_processor

        lowercase_	:				Optional[int]													=	prepare_img()
        lowercase_	:				Tuple													=	image_processor(images=__UpperCamelCase     ,return_tensors='pt'			)
        lowercase_	:				Union[str, Any]													=	inputs.pixel_values.to(__UpperCamelCase			)

        # forward pass to make sure inference works in fp16
        with torch.no_grad():
             lowercase_	:				Dict													=	model(__UpperCamelCase			)

