






"""simple docstring"""




import inspect
import unittest

from transformers import MobileNetVaConfig
from transformers.testing_utils import require_torch, 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 transformers import MobileNetVaForImageClassification, MobileNetVaModel
     from transformers.models.mobilenet_va.modeling_mobilenet_va import MOBILENET_V1_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
     from PIL import Image

     from transformers import MobileNetVaImageProcessor




class    lowercase_    (_UpperCAmelCase     ):






      def 			lowerCamelCase__    (    self	)	->int:







        '''simple docstring'''


        _a							=	self.config_class(**self.inputs_dict	)
        self.parent.assertTrue(hasattr(_SCREAMING_SNAKE_CASE     ,     "tf_padding"	)	)
        self.parent.assertTrue(hasattr(_SCREAMING_SNAKE_CASE     ,     "depth_multiplier"	)	)







class    lowercase_    :






      def __init__(    self     ,     a_     ,     a_=1_3     ,     a_=3     ,     a_=3_2     ,     a_=0.25     ,     a_=8     ,     a_=True     ,     a_=1_0_2_4     ,     a_=3_2     ,     a_="relu6"     ,     a_=0.1     ,     a_=0.02     ,     a_=True     ,     a_=True     ,     a_=1_0     ,     a_=None     ,     )	->List[str]:







        '''simple docstring'''


        _a							=	parent
        _a							=	batch_size
        _a							=	num_channels
        _a							=	image_size
        _a							=	depth_multiplier
        _a							=	min_depth
        _a							=	tf_padding
        _a							=	int(last_hidden_size * depth_multiplier	)
        _a							=	output_stride
        _a							=	hidden_act
        _a							=	classifier_dropout_prob
        _a							=	use_labels
        _a							=	is_training
        _a							=	num_labels
        _a							=	initializer_range
        _a							=	scope






      def 			lowerCamelCase__    (    self	)	->int:







        '''simple docstring'''


        _a							=	floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]	)

        _a							=	None
        _a							=	None
        if self.use_labels:
          _a							=	ids_tensor([self.batch_size]     ,     self.num_labels	)
          _a							=	ids_tensor([self.batch_size, self.image_size, self.image_size]     ,     self.num_labels	)

        _a							=	self.get_config()

        return config, pixel_values, labels, pixel_labels






      def 			lowerCamelCase__    (    self	)	->Optional[Any]:







        '''simple docstring'''


        return MobileNetVaConfig(
            num_channels=self.num_channels     ,     image_size=self.image_size     ,     depth_multiplier=self.depth_multiplier     ,     min_depth=self.min_depth     ,     tf_padding=self.tf_padding     ,     hidden_act=self.hidden_act     ,     classifier_dropout_prob=self.classifier_dropout_prob     ,     initializer_range=self.initializer_range     ,     )






      def 			lowerCamelCase__    (    self     ,     a_     ,     a_     ,     a_     ,     a_	)	->int:







        '''simple docstring'''


        _a							=	MobileNetVaModel(config=_SCREAMING_SNAKE_CASE	)
        model.to(_SCREAMING_SNAKE_CASE	)
        model.eval()
        _a							=	model(_SCREAMING_SNAKE_CASE	)
        self.parent.assertEqual(
            result.last_hidden_state.shape     ,     (
                self.batch_size,
                self.last_hidden_size,
                self.image_size // self.output_stride,
                self.image_size // self.output_stride,
            )     ,     )






      def 			lowerCamelCase__    (    self     ,     a_     ,     a_     ,     a_     ,     a_	)	->List[Any]:







        '''simple docstring'''


        _a							=	self.num_labels
        _a							=	MobileNetVaForImageClassification(_SCREAMING_SNAKE_CASE	)
        model.to(_SCREAMING_SNAKE_CASE	)
        model.eval()
        _a							=	model(_SCREAMING_SNAKE_CASE     ,     labels=_SCREAMING_SNAKE_CASE	)
        self.parent.assertEqual(result.logits.shape     ,     (self.batch_size, self.num_labels)	)






      def 			lowerCamelCase__    (    self	)	->Any:







        '''simple docstring'''


        _a							=	self.prepare_config_and_inputs()
        _a			,	_a			,	_a			,	_a							=	config_and_inputs
        _a							=	{"pixel_values": pixel_values}
        return config, inputs_dict







@require_torch
class    lowercase_    (_UpperCAmelCase,    _UpperCAmelCase,    unittest.TestCase     ):
      A__	:		Any       			=				(MobileNetVaModel, MobileNetVaForImageClassification) if is_torch_available() else ()
      A__	:		Union[str, Any]       			=				(
          {'''feature-extraction''': MobileNetVaModel, '''image-classification''': MobileNetVaForImageClassification}
          if is_torch_available()
          else {}
      )

      A__	:		Union[str, Any]       			=				False
      A__	:		Any       			=				False
      A__	:		str       			=				False
      A__	:		Tuple       			=				False






      def 			lowerCamelCase__    (    self	)	->int:







        '''simple docstring'''


        _a							=	MobileNetVaModelTester(self	)
        _a							=	MobileNetVaConfigTester(self     ,     config_class=_SCREAMING_SNAKE_CASE     ,     has_text_modality=_SCREAMING_SNAKE_CASE	)






      def 			lowerCamelCase__    (    self	)	->Optional[Any]:







        '''simple docstring'''


        self.config_tester.run_common_tests()






      @unittest.skip(reason="MobileNetV1 does not use inputs_embeds"	)
      def 			lowerCamelCase__    (    self	)	->int:







        '''simple docstring'''


        pass






      @unittest.skip(reason="MobileNetV1 does not support input and output embeddings"	)
      def 			lowerCamelCase__    (    self	)	->int:







        '''simple docstring'''


        pass






      @unittest.skip(reason="MobileNetV1 does not output attentions"	)
      def 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


        pass






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







        '''simple docstring'''


        _a			,	_a							=	self.model_tester.prepare_config_and_inputs_for_common()

        for model_class in self.all_model_classes:
          _a							=	model_class(_SCREAMING_SNAKE_CASE	)
          _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]     ,     _SCREAMING_SNAKE_CASE	)






      def 			lowerCamelCase__    (    self	)	->str:







        '''simple docstring'''


        _a							=	self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_model(*_SCREAMING_SNAKE_CASE	)






      def 			lowerCamelCase__    (    self	)	->int:







        '''simple docstring'''


        def check_hidden_states_output(a_     ,     a_     ,     a_	):
          _a							=	model_class(_SCREAMING_SNAKE_CASE	)
          model.to(_SCREAMING_SNAKE_CASE	)
          model.eval()

          with torch.no_grad():
            _a							=	model(**self._prepare_for_class(_SCREAMING_SNAKE_CASE     ,     _SCREAMING_SNAKE_CASE	)	)

          _a							=	outputs.hidden_states

          _a							=	2_6
          self.assertEqual(len(_SCREAMING_SNAKE_CASE	)     ,     _SCREAMING_SNAKE_CASE	)

        _a			,	_a							=	self.model_tester.prepare_config_and_inputs_for_common()

        for model_class in self.all_model_classes:
          _a							=	True
          check_hidden_states_output(_SCREAMING_SNAKE_CASE     ,     _SCREAMING_SNAKE_CASE     ,     _SCREAMING_SNAKE_CASE	)

          # check that output_hidden_states also work using config
          del inputs_dict["output_hidden_states"]
          _a							=	True

          check_hidden_states_output(_SCREAMING_SNAKE_CASE     ,     _SCREAMING_SNAKE_CASE     ,     _SCREAMING_SNAKE_CASE	)






      def 			lowerCamelCase__    (    self	)	->str:







        '''simple docstring'''


        _a							=	self.model_tester.prepare_config_and_inputs()
        self.model_tester.create_and_check_for_image_classification(*_SCREAMING_SNAKE_CASE	)






      @slow
      def 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


        for model_name in MOBILENET_V1_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
          _a							=	MobileNetVaModel.from_pretrained(_SCREAMING_SNAKE_CASE	)
          self.assertIsNotNone(_SCREAMING_SNAKE_CASE	)







def        lowerCAmelCase (      )       ->							Optional[Any]:







  '''simple docstring'''







  _a							=	Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"					)
  return image




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






      @cached_property
      def 			lowerCamelCase__    (    self	)	->Dict:







        '''simple docstring'''


        return (
            MobileNetVaImageProcessor.from_pretrained("google/mobilenet_v1_1.0_224"	) if is_vision_available() else None
        )






      @slow
      def 			lowerCamelCase__    (    self	)	->Union[str, Any]:







        '''simple docstring'''


        _a							=	MobileNetVaForImageClassification.from_pretrained("google/mobilenet_v1_1.0_224"	).to(_SCREAMING_SNAKE_CASE	)

        _a							=	self.default_image_processor
        _a							=	prepare_img()
        _a							=	image_processor(images=_SCREAMING_SNAKE_CASE     ,     return_tensors="pt"	).to(_SCREAMING_SNAKE_CASE	)

        # forward pass
        with torch.no_grad():
          _a							=	model(**_SCREAMING_SNAKE_CASE	)

        # verify the logits
        _a							=	torch.Size((1, 1_0_0_1)	)
        self.assertEqual(outputs.logits.shape     ,     _SCREAMING_SNAKE_CASE	)

        _a							=	torch.tensor([-4.1_739, -1.1_233, 3.1_205]	).to(_SCREAMING_SNAKE_CASE	)

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










"""simple docstring"""




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    lowercase_    (unittest.TestCase     ):






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







        '''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 			lowerCamelCase__    (    self     ,     a_=1	)	->int:







        '''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 			lowerCamelCase__    (    self     ,     a_	)	->str:







        '''simple docstring'''


        TrainingJobAnalytics(a_	).export_csv(f'''{self.env.test_path}/{job_name}_metrics.csv'''	)






      def 			lowerCamelCase__    (    self	)	->Any:







        '''simple docstring'''


        _a							=	self.create_estimator()

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


