



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


__a     =    logging.get_logger(__name__)
class     lowerCamelCase  (      _lowerCAmelCase    ):






					'''simple docstring'''





					_A		:      PyTorchBenchmarkArguments
					_A		:      PretrainedConfig
					_A		:      str						=					"PyTorch"






					@property
					def        lowerCAmelCase_    (	self:							Any						)       ->     int:
											return torch.__version__






					def        lowerCAmelCase_    (	self:							Union[str, Any]						,					snake_case:							str						,					snake_case:							int						,					snake_case:							int						)       ->     float:
											snake_case_			:Tuple						=      self._prepare_inference_func(snake_case						,					snake_case						,					snake_case						)
											return self._measure_speed(_inference						)






					def        lowerCAmelCase_    (	self:							Union[str, Any]						,					snake_case:							str						,					snake_case:							int						,					snake_case:							int						)       ->     [Memory, Optional[MemorySummary]]:
											snake_case_			:Union[str, Any]						=      self._prepare_inference_func(snake_case						,					snake_case						,					snake_case						)
											return self._measure_memory(_inference						)






					def        lowerCAmelCase_    (	self:							Union[str, Any]						,					snake_case:							str						,					snake_case:							int						,					snake_case:							int						)       ->     float:
											snake_case_			:List[str]						=      self._prepare_train_func(snake_case						,					snake_case						,					snake_case						)
											return self._measure_speed(_train						)






					def        lowerCAmelCase_    (	self:							Union[str, Any]						,					snake_case:							str						,					snake_case:							int						,					snake_case:							int						)       ->     [Memory, Optional[MemorySummary]]:
											snake_case_			:List[str]						=      self._prepare_train_func(snake_case						,					snake_case						,					snake_case						)
											return self._measure_memory(_train						)






					def        lowerCAmelCase_    (	self:							Dict						,					snake_case:							str						,					snake_case:							int						,					snake_case:							int						)       ->     Callable[[], None]:
											snake_case_			:str						=      self.config_dict[model_name]

											if self.args.torchscript:
																	snake_case_			:Dict						=      True

											snake_case_			:List[Any]						=      (
											    hasattr(snake_case						,					"""architectures"""						)
											    and isinstance(config.architectures						,					snake_case						)
											    and len(config.architectures						) > 0
											)
											if not self.args.only_pretrain_model and has_model_class_in_config:
																	try:
																							snake_case_			:Optional[int]						=      config.architectures[0]
																							snake_case_			:List[str]						=      __import__("""transformers"""						,					fromlist=[model_class]						)
																							snake_case_			:List[Any]						=      getattr(snake_case						,					snake_case						)
																							snake_case_			:Optional[int]						=      model_cls(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:
																	snake_case_			:Tuple						=      MODEL_MAPPING[config.__class__](snake_case						)

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

											# encoder-decoder has vocab size saved differently
											snake_case_			:Tuple						=      config.vocab_size if hasattr(snake_case						,					"""vocab_size"""						) else config.encoder.vocab_size
											snake_case_			:List[str]						=      torch.randint(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():
																							snake_case_			:Any						=      torch.jit.trace(snake_case						,					snake_case						)
											else:
																	snake_case_			:Optional[Any]						=      model

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

											def encoder_forward():
																	with torch.no_grad():
																							snake_case_			:Any						=      inference_model(snake_case						)
																	return outputs

											snake_case_			:List[str]						=      encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
											return _forward






					def        lowerCAmelCase_    (	self:							Union[str, Any]						,					snake_case:							str						,					snake_case:							int						,					snake_case:							int						)       ->     Callable[[], None]:
											snake_case_			:List[Any]						=      self.config_dict[model_name]

											snake_case_			:str						=      (
											    hasattr(snake_case						,					"""architectures"""						)
											    and isinstance(config.architectures						,					snake_case						)
											    and len(config.architectures						) > 0
											)
											if not self.args.only_pretrain_model and has_model_class_in_config:
																	try:
																							snake_case_			:str						=      config.architectures[0]
																							snake_case_			:Tuple						=      __import__("""transformers"""						,					fromlist=[model_class]						)
																							snake_case_			:Optional[Any]						=      getattr(snake_case						,					snake_case						)
																							snake_case_			:Optional[Any]						=      model_cls(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:
																	snake_case_			:str						=      MODEL_WITH_LM_HEAD_MAPPING[config.__class__](snake_case						)

											if self.args.torchscript:
																	raise NotImplementedError("""Training for torchscript is currently not implemented"""						)
											else:
																	snake_case_			:int						=      model

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

											# encoder-decoder has vocab size saved differently
											snake_case_			:Tuple						=      config.vocab_size if hasattr(snake_case						,					"""vocab_size"""						) else config.encoder.vocab_size
											snake_case_			:Optional[int]						=      torch.randint(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():
																	snake_case_			:Any						=      train_model(snake_case						,					labels=snake_case						)[0]
																	loss.backward()
																	return loss

											def compute_loss_and_backprob_encoder_decoder():
																	snake_case_			:List[Any]						=      train_model(snake_case						,					decoder_input_ids=snake_case						,					labels=snake_case						)[0]
																	loss.backward()
																	return loss

											snake_case_			:Optional[int]						=      (
											    compute_loss_and_backprob_encoder_decoder
											    if config.is_encoder_decoder
											    else compute_loss_and_backprob_encoder
											)
											return _train






					def        lowerCAmelCase_    (	self:							str						,					snake_case:							int						)       ->     float:
											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(
																							    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
																	snake_case_			:Any						=      timeit.repeat(
																	    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(snake_case						) / 10.0
											except RuntimeError as e:
																	self.print_fn(f"""Doesn't fit on GPU. {e}"""						)
																	return "N/A"






					def        lowerCAmelCase_    (	self:							Dict						,					snake_case:							Callable[[], None]						)       ->     [Memory, MemorySummary]:
											try:
																	if self.args.trace_memory_line_by_line:
																							snake_case_			:str						=      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."""						)
																													snake_case_			: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()
																													snake_case_			:str						=      nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx						)
																													snake_case_			:Tuple						=      nvml.nvmlDeviceGetMemoryInfo(snake_case						)
																													snake_case_			:List[Any]						=      meminfo.used
																													snake_case_			:Any						=      Memory(snake_case						)
																													# shutdown nvml
																													nvml.nvmlShutdown()
																	else:
																							# cpu
																							snake_case_			:List[str]						=      measure_peak_memory_cpu(snake_case						)
																							snake_case_			:int						=      Memory(snake_case						) if isinstance(snake_case						,					snake_case						) else memory_bytes

																	if self.args.trace_memory_line_by_line:
																							snake_case_			:Any						=      stop_memory_tracing(snake_case						)
																	else:
																							snake_case_			:Optional[int]						=      None

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





"""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 ViTImageProcessor
class     lowerCamelCase  (      unittest.TestCase    ):






					'''simple docstring'''





					def __init__(	self:							int						,					snake_case:							List[str]						,					snake_case:							List[str]=7						,					snake_case:							Union[str, Any]=3						,					snake_case:							List[str]=18						,					snake_case:							Any=30						,					snake_case:							Any=400						,					snake_case:							Any=True						,					snake_case:							List[str]=None						,					snake_case:							Union[str, Any]=True						,					snake_case:							int=[0.5, 0.5, 0.5]						,					snake_case:							Tuple=[0.5, 0.5, 0.5]						,					)       ->     int:
											snake_case_			:Dict						=      size if size is not None else {"""height""": 18, """width""": 18}
											snake_case_			:Optional[Any]						=      parent
											snake_case_			:Tuple						=      batch_size
											snake_case_			:Tuple						=      num_channels
											snake_case_			:Optional[int]						=      image_size
											snake_case_			:Tuple						=      min_resolution
											snake_case_			:str						=      max_resolution
											snake_case_			:Any						=      do_resize
											snake_case_			:Union[str, Any]						=      size
											snake_case_			:Tuple						=      do_normalize
											snake_case_			:int						=      image_mean
											snake_case_			:List[Any]						=      image_std






					def        lowerCAmelCase_    (	self:							List[Any]						)       ->     List[Any]:
											return {
											    "image_mean": self.image_mean,
											    "image_std": self.image_std,
											    "do_normalize": self.do_normalize,
											    "do_resize": self.do_resize,
											    "size": self.size,
											}

@require_torch
@require_vision
class     lowerCamelCase  (      _lowerCAmelCase   ,						unittest.TestCase    ):






					'''simple docstring'''





					_A		:      Tuple						=					ViTImageProcessor if is_vision_available() else None






					def        lowerCAmelCase_    (	self:							str						)       ->     Optional[Any]:
											snake_case_			:Dict						=      ViTImageProcessingTester(self						)






					@property
					def        lowerCAmelCase_    (	self:							Tuple						)       ->     Optional[Any]:
											return self.image_processor_tester.prepare_image_processor_dict()






					def        lowerCAmelCase_    (	self:							str						)       ->     List[str]:
											snake_case_			:List[str]						=      self.image_processing_class(**self.image_processor_dict						)
											self.assertTrue(hasattr(snake_case						,					"""image_mean"""						)						)
											self.assertTrue(hasattr(snake_case						,					"""image_std"""						)						)
											self.assertTrue(hasattr(snake_case						,					"""do_normalize"""						)						)
											self.assertTrue(hasattr(snake_case						,					"""do_resize"""						)						)
											self.assertTrue(hasattr(snake_case						,					"""size"""						)						)






					def        lowerCAmelCase_    (	self:							Dict						)       ->     Union[str, Any]:
											snake_case_			:int						=      self.image_processing_class.from_dict(self.image_processor_dict						)
											self.assertEqual(image_processor.size						,					{"""height""": 18, """width""": 18}						)

											snake_case_			:int						=      self.image_processing_class.from_dict(self.image_processor_dict						,					size=42						)
											self.assertEqual(image_processor.size						,					{"""height""": 42, """width""": 42}						)






					def        lowerCAmelCase_    (	self:							Tuple						)       ->     int:
											pass






					def        lowerCAmelCase_    (	self:							Dict						)       ->     Union[str, Any]:
											# Initialize image_processing
											snake_case_			:Any						=      self.image_processing_class(**self.image_processor_dict						)
											# create random PIL images
											snake_case_			:str						=      prepare_image_inputs(self.image_processor_tester						,					equal_resolution=snake_case						)
											for image in image_inputs:
																	self.assertIsInstance(snake_case						,					Image.Image						)

											# Test not batched input
											snake_case_			:List[str]						=      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.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)

											# Test batched
											snake_case_			:Dict						=      image_processing(snake_case						,					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.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)






					def        lowerCAmelCase_    (	self:							Optional[Any]						)       ->     Tuple:
											# Initialize image_processing
											snake_case_			:int						=      self.image_processing_class(**self.image_processor_dict						)
											# create random numpy tensors
											snake_case_			:str						=      prepare_image_inputs(self.image_processor_tester						,					equal_resolution=snake_case						,					numpify=snake_case						)
											for image in image_inputs:
																	self.assertIsInstance(snake_case						,					np.ndarray						)

											# Test not batched input
											snake_case_			:Any						=      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.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)

											# Test batched
											snake_case_			:Any						=      image_processing(snake_case						,					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.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)






					def        lowerCAmelCase_    (	self:							List[Any]						)       ->     Optional[int]:
											# Initialize image_processing
											snake_case_			:Union[str, Any]						=      self.image_processing_class(**self.image_processor_dict						)
											# create random PyTorch tensors
											snake_case_			:List[Any]						=      prepare_image_inputs(self.image_processor_tester						,					equal_resolution=snake_case						,					torchify=snake_case						)
											for image in image_inputs:
																	self.assertIsInstance(snake_case						,					torch.Tensor						)

											# Test not batched input
											snake_case_			:Optional[Any]						=      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.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)

											# Test batched
											snake_case_			:List[Any]						=      image_processing(snake_case						,					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.size["""height"""],
											        self.image_processor_tester.size["""width"""],
											    )						,					)
