



'''simple docstring'''




import random
import timeit
from functools import wraps
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_tf_auto import TF_MODEL_MAPPING, TF_MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_tf_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_tf_available():
							import tensorflow as tf
							from tensorflow.python.framework.errors_impl import ResourceExhaustedError

							from .benchmark_args_tf import TensorFlowBenchmarkArguments

if is_pyanvml_available():
							import pyanvml.pyanvml as nvml

a   :			Tuple					  =    logging.get_logger(__name__)



def       __lowerCamelCase  (					_lowercase	,				_lowercase		)      ->			List[Any]:
							def run_func(_lowercase		):
														@wraps(_lowercase		)
														def run_in_eager_mode(*_lowercase	,				**_lowercase		):
																					return func(*_lowercase	,				**_lowercase		)

														@wraps(_lowercase		)
														@tf.function(experimental_compile=_lowercase		)
														def run_in_graph_mode(*_lowercase	,				**_lowercase		):
																					return func(*_lowercase	,				**_lowercase		)

														if do_eager_mode is True:
																					if use_xla is not False:
																												raise ValueError(
																												    """Cannot run model in XLA, if `args.eager_mode` is set to `True`. Please set `args.eager_mode=False`."""		)
																					return run_in_eager_mode
														else:
																					return run_in_graph_mode

							return run_func



def       __lowerCamelCase  (					_lowercase	,				_lowercase	,				_lowercase		)      ->			["tf.Tensor"]:
							UpperCAmelCase     :		str							     =		random.Random()
							UpperCAmelCase     :		Optional[Any]							     =		[rng.randint(0	,				vocab_size - 1		) for i in range(batch_size * sequence_length		)]
							return tf.constant(_lowercase	,				shape=(batch_size, sequence_length)	,				dtype=tf.intaa		)


class 			UpperCamelCase_ (							__magic_name__			):
	lowercase          = 42
	lowercase          = 42
	lowercase          = "TensorFlow"





	@property
	def  _lowercase(  self    )			->			Any:
								return tf.__version__





	def  _lowercase(  self		,		A		,		A		,		A    )			->			float:
								# initialize GPU on separate process
								UpperCAmelCase     :		Any							     =		self.args.strategy
								if strategy is None:
															raise ValueError("""A device strategy has to be initialized before using TensorFlow."""    )
								UpperCAmelCase     :		List[Any]							     =		self._prepare_inference_func(A		,		A		,		A    )
								return self._measure_speed(_inference    )





	def  _lowercase(  self		,		A		,		A		,		A    )			->			float:
								UpperCAmelCase     :		Tuple							     =		self.args.strategy
								if strategy is None:
															raise ValueError("""A device strategy has to be initialized before using TensorFlow."""    )
								UpperCAmelCase     :		Union[str, Any]							     =		self._prepare_train_func(A		,		A		,		A    )
								return self._measure_speed(_train    )





	def  _lowercase(  self		,		A		,		A		,		A    )			->			[Memory, Optional[MemorySummary]]:
								# initialize GPU on separate process
								if self.args.is_gpu:
															tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx]		,		A    )
								UpperCAmelCase     :		List[Any]							     =		self.args.strategy
								if strategy is None:
															raise ValueError("""A device strategy has to be initialized before using TensorFlow."""    )
								UpperCAmelCase     :		Any							     =		self._prepare_inference_func(A		,		A		,		A    )
								return self._measure_memory(_inference    )





	def  _lowercase(  self		,		A		,		A		,		A    )			->			[Memory, Optional[MemorySummary]]:
								if self.args.is_gpu:
															tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx]		,		A    )
								UpperCAmelCase     :		List[Any]							     =		self.args.strategy
								if strategy is None:
															raise ValueError("""A device strategy has to be initialized before using TensorFlow."""    )

								UpperCAmelCase     :		Dict							     =		self._prepare_train_func(A		,		A		,		A    )
								return self._measure_memory(_train    )





	def  _lowercase(  self		,		A		,		A		,		A    )			->			Callable[[], None]:
								UpperCAmelCase     :		Dict							     =		self.config_dict[model_name]

								if self.args.fpaa:
															raise NotImplementedError("""Mixed precision is currently not supported."""    )

								UpperCAmelCase     :		List[str]							     =		(
								    hasattr(A		,		"""architectures"""    )
								    and isinstance(config.architectures		,		A    )
								    and len(config.architectures    ) > 0
								)
								if not self.args.only_pretrain_model and has_model_class_in_config:
															try:
																						UpperCAmelCase     :		List[str]							     =		"""TF""" + config.architectures[0]  # prepend 'TF' for tensorflow model
																						UpperCAmelCase     :		int							     =		__import__("""transformers"""		,		fromlist=[model_class]    )
																						UpperCAmelCase     :		List[str]							     =		getattr(A		,		A    )
																						UpperCAmelCase     :		Optional[int]							     =		model_cls(A    )
															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:
															UpperCAmelCase     :		Optional[Any]							     =		TF_MODEL_MAPPING[config.__class__](A    )

								# encoder-decoder has vocab size saved differently
								UpperCAmelCase     :		Optional[Any]							     =		config.vocab_size if hasattr(A		,		"""vocab_size"""    ) else config.encoder.vocab_size
								UpperCAmelCase     :		Tuple							     =		random_input_ids(A		,		A		,		A    )

								@run_with_tf_optimizations(self.args.eager_mode		,		self.args.use_xla    )
								def encoder_decoder_forward():
															return model(A		,		decoder_input_ids=A		,		training=A    )

								@run_with_tf_optimizations(self.args.eager_mode		,		self.args.use_xla    )
								def encoder_forward():
															return model(A		,		training=A    )

								UpperCAmelCase     :		Tuple							     =		encoder_decoder_forward if config.is_encoder_decoder else encoder_forward

								return _inference





	def  _lowercase(  self		,		A		,		A		,		A    )			->			Callable[[], None]:
								UpperCAmelCase     :		Optional[int]							     =		self.config_dict[model_name]

								if self.args.eager_mode is not False:
															raise ValueError("""Training cannot be done in eager mode. Please make sure that `args.eager_mode = False`."""    )

								if self.args.fpaa:
															raise NotImplementedError("""Mixed precision is currently not supported."""    )

								UpperCAmelCase     :		Tuple							     =		(
								    hasattr(A		,		"""architectures"""    )
								    and isinstance(config.architectures		,		A    )
								    and len(config.architectures    ) > 0
								)
								if not self.args.only_pretrain_model and has_model_class_in_config:
															try:
																						UpperCAmelCase     :		Tuple							     =		"""TF""" + config.architectures[0]  # prepend 'TF' for tensorflow model
																						UpperCAmelCase     :		Dict							     =		__import__("""transformers"""		,		fromlist=[model_class]    )
																						UpperCAmelCase     :		List[str]							     =		getattr(A		,		A    )
																						UpperCAmelCase     :		Any							     =		model_cls(A    )
															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:
															UpperCAmelCase     :		int							     =		TF_MODEL_WITH_LM_HEAD_MAPPING[config.__class__](A    )

								# encoder-decoder has vocab size saved differently
								UpperCAmelCase     :		List[str]							     =		config.vocab_size if hasattr(A		,		"""vocab_size"""    ) else config.encoder.vocab_size
								UpperCAmelCase     :		List[str]							     =		random_input_ids(A		,		A		,		A    )

								@run_with_tf_optimizations(self.args.eager_mode		,		self.args.use_xla    )
								def encoder_decoder_train():
															UpperCAmelCase     :		List[Any]							     =		model(A		,		decoder_input_ids=A		,		labels=A		,		training=A    )[0]
															UpperCAmelCase     :		int							     =		tf.gradients(A		,		model.trainable_variables    )
															return gradients

								@run_with_tf_optimizations(self.args.eager_mode		,		self.args.use_xla    )
								def encoder_train():
															UpperCAmelCase     :		List[Any]							     =		model(A		,		labels=A		,		training=A    )[0]
															UpperCAmelCase     :		List[str]							     =		tf.gradients(A		,		model.trainable_variables    )
															return gradients

								UpperCAmelCase     :		List[str]							     =		encoder_decoder_train if config.is_encoder_decoder else encoder_train

								return _train





	def  _lowercase(  self		,		A    )			->			float:
								with self.args.strategy.scope():
															try:
																						if self.args.is_tpu or self.args.use_xla:
																													# run additional 10 times to stabilize compilation for tpu
																													logger.info("""Do inference on TPU. Running model 5 times to stabilize compilation"""    )
																													timeit.repeat(A		,		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
																						UpperCAmelCase     :		Any							     =		timeit.repeat(
																						    A		,		repeat=self.args.repeat		,		number=10		,		)

																						return min(A    ) / 1_0.0
															except ResourceExhaustedError as e:
																						self.print_fn(f'''Doesn\'t fit on GPU. {e}'''    )






	def  _lowercase(  self		,		A    )			->			[Memory, MemorySummary]:
								logger.info(
								    """Note that TensorFlow allocates more memory than """
								    """it might need to speed up computation. """
								    """The memory reported here corresponds to the memory """
								    """reported by `nvidia-smi`, which can vary depending """
								    """on total available memory on the GPU that is used."""    )
								with self.args.strategy.scope():
															try:
																						if self.args.trace_memory_line_by_line:
																													if not self.args.eager_mode:
																																				raise ValueError(
																																				    """`args.eager_mode` is set to `False`. Make sure to run model in eager mode to measure memory"""
																																				    """ consumption line by line."""    )
																													UpperCAmelCase     :		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 `args.memory=False`"""    )
																						elif self.args.is_gpu:
																													# 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."""    )
																																				UpperCAmelCase     :		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()
																																				UpperCAmelCase     :		List[str]							     =		nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx    )
																																				UpperCAmelCase     :		Union[str, Any]							     =		nvml.nvmlDeviceGetMemoryInfo(A    )
																																				UpperCAmelCase     :		Dict							     =		meminfo.used
																																				UpperCAmelCase     :		int							     =		Memory(A    )
																																				# shutdown nvml
																																				nvml.nvmlShutdown()
																						else:
																													# cpu
																													if self.args.trace_memory_line_by_line:
																																				logger.info(
																																				    """When enabling line by line tracing, the max peak memory for CPU is inaccurate in"""
																																				    """ TensorFlow."""    )
																																				UpperCAmelCase     :		Optional[Any]							     =		None
																													else:
																																				UpperCAmelCase     :		str							     =		measure_peak_memory_cpu(A    )
																																				UpperCAmelCase     :		Union[str, Any]							     =		Memory(A    ) if isinstance(A		,		A    ) else memory_bytes
																						if self.args.trace_memory_line_by_line:
																													UpperCAmelCase     :		Union[str, Any]							     =		stop_memory_tracing(A    )
																													if memory is None:
																																				UpperCAmelCase     :		str							     =		summary.total
																						else:
																													UpperCAmelCase     :		Any							     =		None

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






'''simple docstring'''




import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoImageProcessor, ViTImageProcessor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / """utils"""))

from test_module.custom_image_processing import CustomImageProcessor  # noqa E402


a   :			List[str]					  =    get_tests_dir("""fixtures""")

class 			UpperCamelCase_ (							unittest.TestCase			):





	def  _lowercase(  self    )			->			int:
								# A mock response for an HTTP head request to emulate server down
								UpperCAmelCase     :		Tuple							     =		mock.Mock()
								UpperCAmelCase     :		List[str]							     =		500
								UpperCAmelCase     :		Any							     =		{}
								UpperCAmelCase     :		List[str]							     =		HTTPError
								UpperCAmelCase     :		str							     =		{}

								# Download this model to make sure it's in the cache.
								UpperCAmelCase     :		Optional[int]							     =		ViTImageProcessor.from_pretrained("""hf-internal-testing/tiny-random-vit"""    )
								# Under the mock environment we get a 500 error when trying to reach the model.
								with mock.patch("""requests.Session.request"""		,		return_value=A    ) as mock_head:
															UpperCAmelCase     :		Optional[int]							     =		ViTImageProcessor.from_pretrained("""hf-internal-testing/tiny-random-vit"""    )
															# This check we did call the fake head request
															mock_head.assert_called()





	def  _lowercase(  self    )			->			Any:
								# This test is for deprecated behavior and can be removed in v5
								UpperCAmelCase     :		Tuple							     =		ViTImageProcessor.from_pretrained(
								    """https://huggingface.co/hf-internal-testing/tiny-random-vit/resolve/main/preprocessor_config.json"""    )






	def  _lowercase(  self    )			->			Union[str, Any]:
								with self.assertRaises(A    ):
															# config is in subfolder, the following should not work without specifying the subfolder
															UpperCAmelCase     :		Any							     =		AutoImageProcessor.from_pretrained("""hf-internal-testing/stable-diffusion-all-variants"""    )

								UpperCAmelCase     :		Union[str, Any]							     =		AutoImageProcessor.from_pretrained(
								    """hf-internal-testing/stable-diffusion-all-variants"""		,		subfolder="""feature_extractor"""    )

								self.assertIsNotNone(A    )




@is_staging_test
class 			UpperCamelCase_ (							unittest.TestCase			):





	@classmethod
	def  _lowercase(  cls    )			->			Dict:
								UpperCAmelCase     :		Tuple							     =		TOKEN
								HfFolder.save_token(A    )





	@classmethod
	def  _lowercase(  cls    )			->			List[str]:
								try:
															delete_repo(token=cls._token		,		repo_id="""test-image-processor"""    )
								except HTTPError:
															pass

								try:
															delete_repo(token=cls._token		,		repo_id="""valid_org/test-image-processor-org"""    )
								except HTTPError:
															pass

								try:
															delete_repo(token=cls._token		,		repo_id="""test-dynamic-image-processor"""    )
								except HTTPError:
															pass





	def  _lowercase(  self    )			->			Optional[int]:
								UpperCAmelCase     :		Union[str, Any]							     =		ViTImageProcessor.from_pretrained(A    )
								image_processor.push_to_hub("""test-image-processor"""		,		use_auth_token=self._token    )

								UpperCAmelCase     :		Optional[int]							     =		ViTImageProcessor.from_pretrained(f'''{USER}/test-image-processor'''    )
								for k, v in image_processor.__dict__.items():
															self.assertEqual(A		,		getattr(A		,		A    )    )

								# Reset repo
								delete_repo(token=self._token		,		repo_id="""test-image-processor"""    )

								# Push to hub via save_pretrained
								with tempfile.TemporaryDirectory() as tmp_dir:
															image_processor.save_pretrained(
															    A		,		repo_id="""test-image-processor"""		,		push_to_hub=A		,		use_auth_token=self._token    )

								UpperCAmelCase     :		Tuple							     =		ViTImageProcessor.from_pretrained(f'''{USER}/test-image-processor'''    )
								for k, v in image_processor.__dict__.items():
															self.assertEqual(A		,		getattr(A		,		A    )    )





	def  _lowercase(  self    )			->			List[str]:
								UpperCAmelCase     :		List[str]							     =		ViTImageProcessor.from_pretrained(A    )
								image_processor.push_to_hub("""valid_org/test-image-processor"""		,		use_auth_token=self._token    )

								UpperCAmelCase     :		Tuple							     =		ViTImageProcessor.from_pretrained("""valid_org/test-image-processor"""    )
								for k, v in image_processor.__dict__.items():
															self.assertEqual(A		,		getattr(A		,		A    )    )

								# Reset repo
								delete_repo(token=self._token		,		repo_id="""valid_org/test-image-processor"""    )

								# Push to hub via save_pretrained
								with tempfile.TemporaryDirectory() as tmp_dir:
															image_processor.save_pretrained(
															    A		,		repo_id="""valid_org/test-image-processor-org"""		,		push_to_hub=A		,		use_auth_token=self._token    )

								UpperCAmelCase     :		int							     =		ViTImageProcessor.from_pretrained("""valid_org/test-image-processor-org"""    )
								for k, v in image_processor.__dict__.items():
															self.assertEqual(A		,		getattr(A		,		A    )    )






	def  _lowercase(  self    )			->			Optional[int]:
								CustomImageProcessor.register_for_auto_class()
								UpperCAmelCase     :		Optional[Any]							     =		CustomImageProcessor.from_pretrained(A    )

								image_processor.push_to_hub("""test-dynamic-image-processor"""		,		use_auth_token=self._token    )

								# This has added the proper auto_map field to the config
								self.assertDictEqual(
								    image_processor.auto_map		,		{"""AutoImageProcessor""": """custom_image_processing.CustomImageProcessor"""}		,		)

								UpperCAmelCase     :		Union[str, Any]							     =		AutoImageProcessor.from_pretrained(
								    f'''{USER}/test-dynamic-image-processor'''		,		trust_remote_code=A    )
								# Can't make an isinstance check because the new_image_processor is from the CustomImageProcessor class of a dynamic module
								self.assertEqual(new_image_processor.__class__.__name__		,		"""CustomImageProcessor"""    )

