
"""simple docstring"""




import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
	import torch

if is_vision_available():
	from PIL import Image

	from transformers import DeiTImageProcessor




class 					UpperCamelCase     (	unittest.TestCase				):

				"""simple docstring"""



				def __init__(    self		,     lowercase__		,     lowercase__=7		,     lowercase__=3		,     lowercase__=18		,     lowercase__=30		,     lowercase__=400		,     lowercase__=True		,     lowercase__=None		,     lowercase__=True		,     lowercase__=None		,     lowercase__=True		,     lowercase__=[0.5, 0.5, 0.5]		,     lowercase__=[0.5, 0.5, 0.5]		,     )   ->   str:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     size if size is not None else {'height': 20, 'width': 20}
									SCREAMING_SNAKE_CASE           =     crop_size if crop_size is not None else {'height': 18, 'width': 18}

									SCREAMING_SNAKE_CASE           =     parent
									SCREAMING_SNAKE_CASE           =     batch_size
									SCREAMING_SNAKE_CASE           =     num_channels
									SCREAMING_SNAKE_CASE           =     image_size
									SCREAMING_SNAKE_CASE           =     min_resolution
									SCREAMING_SNAKE_CASE           =     max_resolution
									SCREAMING_SNAKE_CASE           =     do_resize
									SCREAMING_SNAKE_CASE           =     size
									SCREAMING_SNAKE_CASE           =     do_center_crop
									SCREAMING_SNAKE_CASE           =     crop_size
									SCREAMING_SNAKE_CASE           =     do_normalize
									SCREAMING_SNAKE_CASE           =     image_mean
									SCREAMING_SNAKE_CASE           =     image_std




				def 							A (    self      )   ->   Optional[Any]:
									"""simple docstring"""







									return {
									    "do_resize": self.do_resize,
									    "size": self.size,
									    "do_center_crop": self.do_center_crop,
									    "crop_size": self.crop_size,
									    "do_normalize": self.do_normalize,
									    "image_mean": self.image_mean,
									    "image_std": self.image_std,
									}






@require_torch
@require_vision
class 					UpperCamelCase     (	__magic_name__  ,				unittest.TestCase				):

				"""simple docstring"""



				UpperCAmelCase_    :				Optional[Any]						  =    DeiTImageProcessor if is_vision_available() else None
				UpperCAmelCase_    :				Union[str, Any]						  =    True


				def 							A (    self      )   ->   List[Any]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     DeiTImageProcessingTester(self      )


				@property
				def 							A (    self      )   ->   List[Any]:
									"""simple docstring"""







									return self.image_processor_tester.prepare_image_processor_dict()


				def 							A (    self      )   ->   List[str]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self.image_processing_class(**self.image_processor_dict      )
									self.assertTrue(hasattr(lowercase__		,     'do_resize'      )      )
									self.assertTrue(hasattr(lowercase__		,     'size'      )      )
									self.assertTrue(hasattr(lowercase__		,     'do_center_crop'      )      )
									self.assertTrue(hasattr(lowercase__		,     'center_crop'      )      )
									self.assertTrue(hasattr(lowercase__		,     'do_normalize'      )      )
									self.assertTrue(hasattr(lowercase__		,     'image_mean'      )      )
									self.assertTrue(hasattr(lowercase__		,     'image_std'      )      )


				def 							A (    self      )   ->   int:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self.image_processing_class.from_dict(self.image_processor_dict      )
									self.assertEqual(image_processor.size		,     {'height': 20, 'width': 20}      )
									self.assertEqual(image_processor.crop_size		,     {'height': 18, 'width': 18}      )

									SCREAMING_SNAKE_CASE           =     self.image_processing_class.from_dict(self.image_processor_dict		,     size=42		,     crop_size=84      )
									self.assertEqual(image_processor.size		,     {'height': 42, 'width': 42}      )
									self.assertEqual(image_processor.crop_size		,     {'height': 84, 'width': 84}      )


				def 							A (    self      )   ->   Optional[Any]:
									"""simple docstring"""







									pass


				def 							A (    self      )   ->   Optional[int]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self.image_processing_class(**self.image_processor_dict      )
									# create random PIL images
									SCREAMING_SNAKE_CASE           =     prepare_image_inputs(self.image_processor_tester		,     equal_resolution=lowercase__      )
									for image in image_inputs:
														self.assertIsInstance(lowercase__		,     Image.Image      )

									# Test not batched input
									SCREAMING_SNAKE_CASE           =     image_processing(image_inputs[0]		,     return_tensors='pt'      ).pixel_values
									self.assertEqual(
									    encoded_images.shape		,     (
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['height'],
									        self.image_processor_tester.crop_size['width'],
									    )		,     )

									# Test batched
									SCREAMING_SNAKE_CASE           =     image_processing(lowercase__		,     return_tensors='pt'      ).pixel_values
									self.assertEqual(
									    encoded_images.shape		,     (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['height'],
									        self.image_processor_tester.crop_size['width'],
									    )		,     )


				def 							A (    self      )   ->   int:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self.image_processing_class(**self.image_processor_dict      )
									# create random numpy tensors
									SCREAMING_SNAKE_CASE           =     prepare_image_inputs(self.image_processor_tester		,     equal_resolution=lowercase__		,     numpify=lowercase__      )
									for image in image_inputs:
														self.assertIsInstance(lowercase__		,     np.ndarray      )

									# Test not batched input
									SCREAMING_SNAKE_CASE           =     image_processing(image_inputs[0]		,     return_tensors='pt'      ).pixel_values
									self.assertEqual(
									    encoded_images.shape		,     (
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['height'],
									        self.image_processor_tester.crop_size['width'],
									    )		,     )

									# Test batched
									SCREAMING_SNAKE_CASE           =     image_processing(lowercase__		,     return_tensors='pt'      ).pixel_values
									self.assertEqual(
									    encoded_images.shape		,     (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['height'],
									        self.image_processor_tester.crop_size['width'],
									    )		,     )




				def 							A (    self      )   ->   Optional[int]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self.image_processing_class(**self.image_processor_dict      )
									# create random PyTorch tensors
									SCREAMING_SNAKE_CASE           =     prepare_image_inputs(self.image_processor_tester		,     equal_resolution=lowercase__		,     torchify=lowercase__      )
									for image in image_inputs:
														self.assertIsInstance(lowercase__		,     torch.Tensor      )

									# Test not batched input
									SCREAMING_SNAKE_CASE           =     image_processing(image_inputs[0]		,     return_tensors='pt'      ).pixel_values
									self.assertEqual(
									    encoded_images.shape		,     (
									        1,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['height'],
									        self.image_processor_tester.crop_size['width'],
									    )		,     )

									# Test batched
									SCREAMING_SNAKE_CASE           =     image_processing(lowercase__		,     return_tensors='pt'      ).pixel_values
									self.assertEqual(
									    encoded_images.shape		,     (
									        self.image_processor_tester.batch_size,
									        self.image_processor_tester.num_channels,
									        self.image_processor_tester.crop_size['height'],
									        self.image_processor_tester.crop_size['width'],
									    )		,     )


"""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


snake_case			       =		logging.get_logger(__name__)




class 					UpperCamelCase     (	__magic_name__				):

				"""simple docstring"""



				UpperCAmelCase_    :				PyTorchBenchmarkArguments
				UpperCAmelCase_    :				PretrainedConfig
				UpperCAmelCase_    :				str						  =    "PyTorch"


				@property
				def 							A (    self      )   ->   List[str]:
									"""simple docstring"""







									return torch.__version__


				def 							A (    self		,     lowercase__		,     lowercase__		,     lowercase__      )   ->   float:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self._prepare_inference_func(lowercase__		,     lowercase__		,     lowercase__      )
									return self._measure_speed(_inference      )


				def 							A (    self		,     lowercase__		,     lowercase__		,     lowercase__      )   ->   [Memory, Optional[MemorySummary]]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self._prepare_inference_func(lowercase__		,     lowercase__		,     lowercase__      )
									return self._measure_memory(_inference      )


				def 							A (    self		,     lowercase__		,     lowercase__		,     lowercase__      )   ->   float:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self._prepare_train_func(lowercase__		,     lowercase__		,     lowercase__      )
									return self._measure_speed(_train      )


				def 							A (    self		,     lowercase__		,     lowercase__		,     lowercase__      )   ->   [Memory, Optional[MemorySummary]]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self._prepare_train_func(lowercase__		,     lowercase__		,     lowercase__      )
									return self._measure_memory(_train      )


				def 							A (    self		,     lowercase__		,     lowercase__		,     lowercase__      )   ->   Callable[[], None]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self.config_dict[model_name]

									if self.args.torchscript:
														SCREAMING_SNAKE_CASE           =     True

									SCREAMING_SNAKE_CASE           =     (
									    hasattr(lowercase__		,     'architectures'      )
									    and isinstance(config.architectures		,     lowercase__      )
									    and len(config.architectures      ) > 0
									)
									if not self.args.only_pretrain_model and has_model_class_in_config:
														try:
																			SCREAMING_SNAKE_CASE           =     config.architectures[0]
																			SCREAMING_SNAKE_CASE           =     __import__('transformers'		,     fromlist=[model_class]      )
																			SCREAMING_SNAKE_CASE           =     getattr(lowercase__		,     lowercase__      )
																			SCREAMING_SNAKE_CASE           =     model_cls(lowercase__      )
														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:
														SCREAMING_SNAKE_CASE           =     MODEL_MAPPING[config.__class__](lowercase__      )

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

									# encoder-decoder has vocab size saved differently
									SCREAMING_SNAKE_CASE           =     config.vocab_size if hasattr(lowercase__		,     'vocab_size'      ) else config.encoder.vocab_size
									SCREAMING_SNAKE_CASE           =     torch.randint(lowercase__		,     (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():
																			SCREAMING_SNAKE_CASE           =     torch.jit.trace(lowercase__		,     lowercase__      )
									else:
														SCREAMING_SNAKE_CASE           =     model

									def encoder_decoder_forward():
														with torch.no_grad():
																			SCREAMING_SNAKE_CASE           =     inference_model(lowercase__		,     decoder_input_ids=lowercase__      )
														return outputs

									def encoder_forward():
														with torch.no_grad():
																			SCREAMING_SNAKE_CASE           =     inference_model(lowercase__      )
														return outputs

									SCREAMING_SNAKE_CASE           =     encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
									return _forward


				def 							A (    self		,     lowercase__		,     lowercase__		,     lowercase__      )   ->   Callable[[], None]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE           =     self.config_dict[model_name]

									SCREAMING_SNAKE_CASE           =     (
									    hasattr(lowercase__		,     'architectures'      )
									    and isinstance(config.architectures		,     lowercase__      )
									    and len(config.architectures      ) > 0
									)
									if not self.args.only_pretrain_model and has_model_class_in_config:
														try:
																			SCREAMING_SNAKE_CASE           =     config.architectures[0]
																			SCREAMING_SNAKE_CASE           =     __import__('transformers'		,     fromlist=[model_class]      )
																			SCREAMING_SNAKE_CASE           =     getattr(lowercase__		,     lowercase__      )
																			SCREAMING_SNAKE_CASE           =     model_cls(lowercase__      )
														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:
														SCREAMING_SNAKE_CASE           =     MODEL_WITH_LM_HEAD_MAPPING[config.__class__](lowercase__      )

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

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

									# encoder-decoder has vocab size saved differently
									SCREAMING_SNAKE_CASE           =     config.vocab_size if hasattr(lowercase__		,     'vocab_size'      ) else config.encoder.vocab_size
									SCREAMING_SNAKE_CASE           =     torch.randint(lowercase__		,     (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():
														SCREAMING_SNAKE_CASE           =     train_model(lowercase__		,     labels=lowercase__      )[0]
														loss.backward()
														return loss

									def compute_loss_and_backprob_encoder_decoder():
														SCREAMING_SNAKE_CASE           =     train_model(lowercase__		,     decoder_input_ids=lowercase__		,     labels=lowercase__      )[0]
														loss.backward()
														return loss

									SCREAMING_SNAKE_CASE           =     (
									    compute_loss_and_backprob_encoder_decoder
									    if config.is_encoder_decoder
									    else compute_loss_and_backprob_encoder
									)
									return _train


				def 							A (    self		,     lowercase__      )   ->   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(
																			    lowercase__		,     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
														SCREAMING_SNAKE_CASE           =     timeit.repeat(
														    lowercase__		,     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(lowercase__      ) / 10.0
									except RuntimeError as e:
														self.print_fn(f'''Doesn\'t fit on GPU. {e}'''      )
														return "N/A"




				def 							A (    self		,     lowercase__      )   ->   [Memory, MemorySummary]:
									"""simple docstring"""







									try:
														if self.args.trace_memory_line_by_line:
																			SCREAMING_SNAKE_CASE           =     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.'      )
																								SCREAMING_SNAKE_CASE           =     '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()
																								SCREAMING_SNAKE_CASE           =     nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx      )
																								SCREAMING_SNAKE_CASE           =     nvml.nvmlDeviceGetMemoryInfo(lowercase__      )
																								SCREAMING_SNAKE_CASE           =     meminfo.used
																								SCREAMING_SNAKE_CASE           =     Memory(lowercase__      )
																								# shutdown nvml
																								nvml.nvmlShutdown()
														else:
																			# cpu
																			SCREAMING_SNAKE_CASE           =     measure_peak_memory_cpu(lowercase__      )
																			SCREAMING_SNAKE_CASE           =     Memory(lowercase__      ) if isinstance(lowercase__		,     lowercase__      ) else memory_bytes

														if self.args.trace_memory_line_by_line:
																			SCREAMING_SNAKE_CASE           =     stop_memory_tracing(lowercase__      )
														else:
																			SCREAMING_SNAKE_CASE           =     None

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