




import inspect
import unittest

from transformers import EfficientNetConfig
from transformers.testing_utils import is_pipeline_test, 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 EfficientNetForImageClassification, EfficientNetModel
							from transformers.models.efficientnet.modeling_efficientnet import EFFICIENTNET_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
							from PIL import Image

							from transformers import AutoImageProcessor


class 							__SCREAMING_SNAKE_CASE				:

					def __init__(							self					,			__lowercase					,			__lowercase=13					,			__lowercase=32					,			__lowercase=3					,			__lowercase=[3, 3, 5]					,			__lowercase=[32, 16, 24]					,			__lowercase=[16, 24, 40]					,			__lowercase=[1, 1, 2]					,			__lowercase=[1, 1, 2]					,			__lowercase=[1, 6, 6]					,			__lowercase=True					,			__lowercase=True					,			__lowercase=37					,			__lowercase="gelu"					,			__lowercase=10					,			):







						'''simple docstring'''




						lowerCamelCase_				:						Tuple				  =				parent
						lowerCamelCase_				:						Any				  =				batch_size
						lowerCamelCase_				:						int				  =				image_size
						lowerCamelCase_				:						List[Any]				  =				num_channels
						lowerCamelCase_				:						Optional[Any]				  =				kernel_sizes
						lowerCamelCase_				:						Any				  =				in_channels
						lowerCamelCase_				:						Union[str, Any]				  =				out_channels
						lowerCamelCase_				:						Tuple				  =				strides
						lowerCamelCase_				:						Optional[int]				  =				num_block_repeats
						lowerCamelCase_				:						List[str]				  =				expand_ratios
						lowerCamelCase_				:						Tuple				  =				is_training
						lowerCamelCase_				:						Optional[Any]				  =				hidden_act
						lowerCamelCase_				:						List[Any]				  =				num_labels
						lowerCamelCase_				:						List[str]				  =				use_labels

					def       lowercase_				(							self   ):







						'''simple docstring'''




						lowerCamelCase_				:						Optional[int]				  =				floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]   )

						lowerCamelCase_				:						int				  =				None
						if self.use_labels:
							lowerCamelCase_				:						str				  =				ids_tensor([self.batch_size]					,			self.num_labels   )

						lowerCamelCase_				:						Union[str, Any]				  =				self.get_config()
						return config, pixel_values, labels

					def       lowercase_				(							self   ):







						'''simple docstring'''




						return EfficientNetConfig(
						    num_channels=self.num_channels					,			kernel_sizes=self.kernel_sizes					,			in_channels=self.in_channels					,			out_channels=self.out_channels					,			strides=self.strides					,			num_block_repeats=self.num_block_repeats					,			expand_ratios=self.expand_ratios					,			hidden_act=self.hidden_act					,			num_labels=self.num_labels					,			)

					def       lowercase_				(							self					,			__lowercase					,			__lowercase					,			__lowercase   ):







						'''simple docstring'''




						lowerCamelCase_				:						Any				  =				EfficientNetModel(config=__lowercase   )
						model.to(__lowercase   )
						model.eval()
						lowerCamelCase_				:						Dict				  =				model(__lowercase   )
						# expected last hidden states: B, C, H // 4, W // 4
						self.parent.assertEqual(
						    result.last_hidden_state.shape					,			(self.batch_size, config.hidden_dim, self.image_size // 4, self.image_size // 4)					,			)

					def       lowercase_				(							self					,			__lowercase					,			__lowercase					,			__lowercase   ):







						'''simple docstring'''




						lowerCamelCase_				:						Tuple				  =				EfficientNetForImageClassification(__lowercase   )
						model.to(__lowercase   )
						model.eval()
						lowerCamelCase_				:						Union[str, Any]				  =				model(__lowercase					,			labels=__lowercase   )
						self.parent.assertEqual(result.logits.shape					,			(self.batch_size, self.num_labels)   )





					def       lowercase_				(							self   ):







						'''simple docstring'''




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


@require_torch
class 							__SCREAMING_SNAKE_CASE				(						lowerCamelCase  ,	lowerCamelCase  ,	unittest.TestCase    ):
					_lowerCamelCase						    =					(EfficientNetModel, EfficientNetForImageClassification) if is_torch_available() else ()
					_lowerCamelCase						    =					(
					    {"""feature-extraction""": EfficientNetModel, """image-classification""": EfficientNetForImageClassification}
					    if is_torch_available()
					    else {}
					)

					_lowerCamelCase						    =					False
					_lowerCamelCase						    =					False
					_lowerCamelCase						    =					False
					_lowerCamelCase						    =					False
					_lowerCamelCase						    =					False

					def       lowercase_				(							self   ):







						'''simple docstring'''




						lowerCamelCase_				:						Optional[Any]				  =				EfficientNetModelTester(self   )
						lowerCamelCase_				:						Dict				  =				ConfigTester(
						    self					,			config_class=__lowercase					,			has_text_modality=__lowercase					,			hidden_size=37   )

					def       lowercase_				(							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       lowercase_				(							self   ):







						'''simple docstring'''




						return

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







						'''simple docstring'''




						pass

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







						'''simple docstring'''




						pass

					@unittest.skip(reason='EfficientNet does not use feedforward chunking'   )
					def       lowercase_				(							self   ):







						'''simple docstring'''




						pass

					def       lowercase_				(							self   ):







						'''simple docstring'''




						lowerCamelCase_  ,     lowerCamelCase_				:						Dict				  =				self.model_tester.prepare_config_and_inputs_for_common()

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

							lowerCamelCase_				:						Any				  =				['pixel_values']
							self.assertListEqual(arg_names[:1]					,			__lowercase   )

					def       lowercase_				(							self   ):







						'''simple docstring'''




						lowerCamelCase_				:						Tuple				  =				self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_model(*__lowercase   )

					def       lowercase_				(							self   ):







						'''simple docstring'''




						def check_hidden_states_output(__lowercase					,			__lowercase					,			__lowercase   ):
							lowerCamelCase_				:						List[Any]				  =				model_class(__lowercase   )
							model.to(__lowercase   )
							model.eval()

							with torch.no_grad():
								lowerCamelCase_				:						str				  =				model(**self._prepare_for_class(__lowercase					,			__lowercase   )   )

							lowerCamelCase_				:						List[Any]				  =				outputs.encoder_hidden_states if config.is_encoder_decoder else outputs.hidden_states
							lowerCamelCase_				:						str				  =				sum(config.num_block_repeats   ) * 4
							self.assertEqual(len(__lowercase   )					,			__lowercase   )

							# EfficientNet's feature maps are of shape (batch_size, num_channels, height, width)
							self.assertListEqual(
							    list(hidden_states[0].shape[-2:]   )					,			[self.model_tester.image_size // 2, self.model_tester.image_size // 2]					,			)

						lowerCamelCase_  ,     lowerCamelCase_				:						str				  =				self.model_tester.prepare_config_and_inputs_for_common()

						for model_class in self.all_model_classes:
							lowerCamelCase_				:						Any				  =				True
							check_hidden_states_output(__lowercase					,			__lowercase					,			__lowercase   )

							# check that output_hidden_states also work using config
							del inputs_dict["output_hidden_states"]
							lowerCamelCase_				:						str				  =				True

							check_hidden_states_output(__lowercase					,			__lowercase					,			__lowercase   )

					def       lowercase_				(							self   ):







						'''simple docstring'''




						lowerCamelCase_				:						Optional[Any]				  =				self.model_tester.prepare_config_and_inputs()
						self.model_tester.create_and_check_for_image_classification(*__lowercase   )

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







						'''simple docstring'''




						pass

					@slow
					def       lowercase_				(							self   ):







						'''simple docstring'''




						for model_name in EFFICIENTNET_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
							lowerCamelCase_				:						int				  =				EfficientNetModel.from_pretrained(__lowercase   )
							self.assertIsNotNone(__lowercase   )





					@is_pipeline_test
					@require_vision
					@slow
					def       lowercase_				(							self   ):







						'''simple docstring'''




						super().test_pipeline_image_classification()






def    snake_case   (				)				->				List[Any]:




	'''simple docstring'''


	lowerCamelCase_				:						Optional[Any]				  =				Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'     )
	return image



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

					@cached_property
					def       lowercase_				(							self   ):







						'''simple docstring'''




						return AutoImageProcessor.from_pretrained('google/efficientnet-b7'   ) if is_vision_available() else None





					@slow
					def       lowercase_				(							self   ):







						'''simple docstring'''




						lowerCamelCase_				:						str				  =				EfficientNetForImageClassification.from_pretrained('google/efficientnet-b7'   ).to(__lowercase   )

						lowerCamelCase_				:						int				  =				self.default_image_processor
						lowerCamelCase_				:						Optional[Any]				  =				prepare_img()
						lowerCamelCase_				:						Optional[int]				  =				image_processor(images=__lowercase					,			return_tensors='pt'   ).to(__lowercase   )

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

						# verify the logits
						lowerCamelCase_				:						List[str]				  =				torch.Size((1, 1_000)   )
						self.assertEqual(outputs.logits.shape					,			__lowercase   )

						lowerCamelCase_				:						Tuple				  =				torch.tensor([0.00_01, 0.00_02, 0.00_02]   ).to(__lowercase   )
						self.assertTrue(torch.allclose(outputs.logits[0, :3]					,			__lowercase					,			atol=1E-4   )   )






'''simple docstring'''


import timeit
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_auto import MODEL_MAPPING, MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_torch_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_torch_available():
     import torch

     from .benchmark_args import PyTorchBenchmarkArguments


if is_pyanvml_available():
     import pyanvml.pyanvml as nvml


UpperCamelCase			     =					logging.get_logger(__name__)






class 						lowerCAmelCase_	(       UpperCAmelCase_							):







  '''simple docstring'''

  UpperCamelCase_			:      PyTorchBenchmarkArguments
  UpperCamelCase_			:      PretrainedConfig
  UpperCamelCase_			:      str										=				"PyTorch"



  @property
  def 					_snake_case     ( self				:							int	)			->		List[str]:







       '''simple docstring'''



       return torch.__version__



  def 					_snake_case     ( self				:							Dict					,  SCREAMING_SNAKE_CASE_				:							str					,  SCREAMING_SNAKE_CASE_				:							int					,  SCREAMING_SNAKE_CASE_				:							int	)			->		float:







       '''simple docstring'''



       A:		List[str]          =		self._prepare_inference_func(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
       return self._measure_speed(_inference	)



  def 					_snake_case     ( self				:							Dict					,  SCREAMING_SNAKE_CASE_				:							str					,  SCREAMING_SNAKE_CASE_				:							int					,  SCREAMING_SNAKE_CASE_				:							int	)			->		[Memory, Optional[MemorySummary]]:







       '''simple docstring'''



       A:		Any          =		self._prepare_inference_func(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
       return self._measure_memory(_inference	)



  def 					_snake_case     ( self				:							List[str]					,  SCREAMING_SNAKE_CASE_				:							str					,  SCREAMING_SNAKE_CASE_				:							int					,  SCREAMING_SNAKE_CASE_				:							int	)			->		float:







       '''simple docstring'''



       A:		List[Any]          =		self._prepare_train_func(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
       return self._measure_speed(_train	)



  def 					_snake_case     ( self				:							int					,  SCREAMING_SNAKE_CASE_				:							str					,  SCREAMING_SNAKE_CASE_				:							int					,  SCREAMING_SNAKE_CASE_				:							int	)			->		[Memory, Optional[MemorySummary]]:







       '''simple docstring'''



       A:		List[Any]          =		self._prepare_train_func(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
       return self._measure_memory(_train	)



  def 					_snake_case     ( self				:							Union[str, Any]					,  SCREAMING_SNAKE_CASE_				:							str					,  SCREAMING_SNAKE_CASE_				:							int					,  SCREAMING_SNAKE_CASE_				:							int	)			->		Callable[[], None]:







       '''simple docstring'''



       A:		int          =		self.config_dict[model_name]

       if self.args.torchscript:
            A:		int          =		True

       A:		Any          =		(
           hasattr(SCREAMING_SNAKE_CASE_					,  '''architectures'''	)
           and isinstance(config.architectures					,  SCREAMING_SNAKE_CASE_	)
           and len(config.architectures	) > 0
       )
       if not self.args.only_pretrain_model and has_model_class_in_config:
            try:
                 A:		Optional[int]          =		config.architectures[0]
                 A:		Optional[int]          =		__import__('''transformers'''					,  fromlist=[model_class]	)
                 A:		Dict          =		getattr(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
                 A:		Optional[Any]          =		model_cls(SCREAMING_SNAKE_CASE_	)
            except ImportError:
                 raise ImportError(
                     f"""{model_class} does not exist. If you just want to test the pretrained model, you might want to"""
                     ''' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'''	)
       else:
            A:		Union[str, Any]          =		MODEL_MAPPING[config.__class__](SCREAMING_SNAKE_CASE_	)

       model.eval()
       model.to(self.args.device	)

       # encoder-decoder has vocab size saved differently
       A:		Tuple          =		config.vocab_size if hasattr(SCREAMING_SNAKE_CASE_					,  '''vocab_size'''	) else config.encoder.vocab_size
       A:		List[Any]          =		torch.randint(SCREAMING_SNAKE_CASE_					,  (batch_size, sequence_length)					,  dtype=torch.long					,  device=self.args.device	)

       if self.args.fpaa:
            logger.info('''Running training in Mixed Precision...'''	)
            if not self.args.is_gpu:
                 raise ValueError('''Mixed precision is possible only for GPU.'''	)
            # amp seems to have memory leaks so that memory usage
            # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
            model.half()

       if self.args.torchscript:
            with torch.no_grad():
                 A:		Any          =		torch.jit.trace(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
       else:
            A:		Dict          =		model

       def encoder_decoder_forward():
            with torch.no_grad():
                 A:		List[str]          =		inference_model(SCREAMING_SNAKE_CASE_					,  decoder_input_ids=SCREAMING_SNAKE_CASE_	)
            return outputs

       def encoder_forward():
            with torch.no_grad():
                 A:		Union[str, Any]          =		inference_model(SCREAMING_SNAKE_CASE_	)
            return outputs

       A:		List[Any]          =		encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
       return _forward



  def 					_snake_case     ( self				:							Tuple					,  SCREAMING_SNAKE_CASE_				:							str					,  SCREAMING_SNAKE_CASE_				:							int					,  SCREAMING_SNAKE_CASE_				:							int	)			->		Callable[[], None]:







       '''simple docstring'''



       A:		int          =		self.config_dict[model_name]

       A:		str          =		(
           hasattr(SCREAMING_SNAKE_CASE_					,  '''architectures'''	)
           and isinstance(config.architectures					,  SCREAMING_SNAKE_CASE_	)
           and len(config.architectures	) > 0
       )
       if not self.args.only_pretrain_model and has_model_class_in_config:
            try:
                 A:		List[str]          =		config.architectures[0]
                 A:		List[str]          =		__import__('''transformers'''					,  fromlist=[model_class]	)
                 A:		Optional[int]          =		getattr(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	)
                 A:		Union[str, Any]          =		model_cls(SCREAMING_SNAKE_CASE_	)
            except ImportError:
                 raise ImportError(
                     f"""{model_class} does not exist. If you just want to test the pretrained model, you might want to"""
                     ''' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'''	)
       else:
            A:		Union[str, Any]          =		MODEL_WITH_LM_HEAD_MAPPING[config.__class__](SCREAMING_SNAKE_CASE_	)

       if self.args.torchscript:
            raise NotImplementedError('''Training for torchscript is currently not implemented'''	)
       else:
            A:		str          =		model

       model.train()
       model.to(self.args.device	)

       # encoder-decoder has vocab size saved differently
       A:		Tuple          =		config.vocab_size if hasattr(SCREAMING_SNAKE_CASE_					,  '''vocab_size'''	) else config.encoder.vocab_size
       A:		str          =		torch.randint(SCREAMING_SNAKE_CASE_					,  (batch_size, sequence_length)					,  dtype=torch.long					,  device=self.args.device	)

       if self.args.fpaa:
            logger.info('''Running training in Mixed Precision...'''	)
            if not self.args.is_gpu:
                 raise ValueError('''Mixed precision is possible only for GPU.'''	)

            # amp seems to have memory leaks so that memory usage
            # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
            model.half()

       def compute_loss_and_backprob_encoder():
            A:		str          =		train_model(SCREAMING_SNAKE_CASE_					,  labels=SCREAMING_SNAKE_CASE_	)[0]
            loss.backward()
            return loss

       def compute_loss_and_backprob_encoder_decoder():
            A:		Dict          =		train_model(SCREAMING_SNAKE_CASE_					,  decoder_input_ids=SCREAMING_SNAKE_CASE_					,  labels=SCREAMING_SNAKE_CASE_	)[0]
            loss.backward()
            return loss

       A:		str          =		(
           compute_loss_and_backprob_encoder_decoder
           if config.is_encoder_decoder
           else compute_loss_and_backprob_encoder
       )
       return _train



  def 					_snake_case     ( self				:							Any					,  SCREAMING_SNAKE_CASE_				:							Any	)			->		float:







       '''simple docstring'''



       try:
            if self.args.is_tpu or self.args.torchscript:
                 # run additional 10 times to stabilize compilation for tpu and torchscript
                 logger.info('''Do inference on TPU or torchscript. Running model 5 times to stabilize compilation'''	)
                 timeit.repeat(
                     SCREAMING_SNAKE_CASE_					,  repeat=1					,  number=5					,  )

            # as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
            A:		Any          =		timeit.repeat(
                SCREAMING_SNAKE_CASE_					,  repeat=self.args.repeat					,  number=10					,  )

            if self.args.is_tpu and self.args.torch_xla_tpu_print_metrics:
                 import torch_xla.debug.metrics as met

                 self.print_fn(met.metrics_report()	)

            return min(SCREAMING_SNAKE_CASE_	) / 10.0
       except RuntimeError as e:
            self.print_fn(f"""Doesn't fit on GPU. {e}"""	)
            return "N/A"




  def 					_snake_case     ( self				:							Optional[int]					,  SCREAMING_SNAKE_CASE_				:							Callable[[], None]	)			->		[Memory, MemorySummary]:







       '''simple docstring'''



       try:
            if self.args.trace_memory_line_by_line:
                 A:		List[Any]          =		start_memory_tracing('''transformers'''	)

            if self.args.is_tpu:
                 # tpu
                 raise NotImplementedError(
                     '''Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking with'''
                     ''' `--no-memory` or `args.memory=False`'''	)
            elif self.args.is_gpu:
                 if not is_pyanvml_available():
                      logger.warning(
                          '''py3nvml not installed, we won\'t log GPU memory usage. '''
                          '''Install py3nvml (pip install py3nvml) to log information about GPU.'''	)
                      A:		Optional[int]          =		'''N/A'''
                 else:
                      logger.info(
                          '''Measuring total GPU usage on GPU device. Make sure to not have additional processes running'''
                          ''' on the same GPU.'''	)
                      # init nvml
                      nvml.nvmlInit()
                      func()
                      A:		Union[str, Any]          =		nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx	)
                      A:		Union[str, Any]          =		nvml.nvmlDeviceGetMemoryInfo(SCREAMING_SNAKE_CASE_	)
                      A:		Tuple          =		meminfo.used
                      A:		Tuple          =		Memory(SCREAMING_SNAKE_CASE_	)
                      # shutdown nvml
                      nvml.nvmlShutdown()
            else:
                 # cpu
                 A:		int          =		measure_peak_memory_cpu(SCREAMING_SNAKE_CASE_	)
                 A:		str          =		Memory(SCREAMING_SNAKE_CASE_	) if isinstance(SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_	) else memory_bytes

            if self.args.trace_memory_line_by_line:
                 A:		Dict          =		stop_memory_tracing(SCREAMING_SNAKE_CASE_	)
            else:
                 A:		List[str]          =		None

            return memory, summary
       except RuntimeError as e:
            self.print_fn(f"""Doesn't fit on GPU. {e}"""	)
            return "N/A", None






