






from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_torch_available, is_torch_tpu_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_torch_available():
	import torch

if is_torch_tpu_available(check_device=False):
	import torch_xla.core.xla_model as xm


__lowerCamelCase :      Union[str, Any]										=				logging.get_logger(__name__)



@dataclass
class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					a_     =		[
					    "no_inference",
					    "no_cuda",
					    "no_tpu",
					    "no_speed",
					    "no_memory",
					    "no_env_print",
					    "no_multi_process",
					]



					def __init__( self					:  List[str]							,		**__A					:  List[Any]	):
						for deprecated_arg in self.deprecated_args:
							if deprecated_arg in kwargs:
								snake_case__			:				Dict       =  deprecated_arg[3:]
								setattr(self							,		__A							,		not kwargs.pop(__A	)	)
								logger.warning(
								    f'''{deprecated_arg} is depreciated. Please use --no_{positive_arg} or'''
								    f''' {positive_arg}={kwargs[positive_arg]}'''	)

						snake_case__			:				str       =  kwargs.pop("torchscript"							,		self.torchscript	)
						snake_case__			:				Any       =  kwargs.pop("torch_xla_tpu_print_metrics"							,		self.torch_xla_tpu_print_metrics	)
						snake_case__			:				Optional[int]       =  kwargs.pop("fp16_opt_level"							,		self.fpaa_opt_level	)
						super().__init__(**__A	)

					a_     =		field(default=UpperCamelCase_			,       metadata={"help": "Trace the models using torchscript"}						)
					a_     =		field(default=UpperCamelCase_			,       metadata={"help": "Print Xla/PyTorch tpu metrics"}						)
					a_     =		field(
					    default="O1"			,       metadata={
					        "help": (
					            "For fp16: Apex AMP optimization level selected in ['O0', 'O1', 'O2', and 'O3']. "
					            "See details at https://nvidia.github.io/apex/amp.html"
					        )
					    }			,       )



					@cached_property
					def 	_lowercase					( self					:  List[str]	):
						requires_backends(self							,		["torch"]	)
						logger.info("PyTorch: setting up devices"	)
						if not self.cuda:
							snake_case__			:				List[Any]       =  torch.device("cpu"	)
							snake_case__			:				List[Any]       =  0
						elif is_torch_tpu_available():
							snake_case__			:				Any       =  xm.xla_device()
							snake_case__			:				Optional[Any]       =  0
						else:
							snake_case__			:				List[str]       =  torch.device("cuda" if torch.cuda.is_available() else "cpu"	)
							snake_case__			:				Optional[Any]       =  torch.cuda.device_count()
						return device, n_gpu



					@property
					def 	_lowercase					( self					:  Optional[int]	):
						return is_torch_tpu_available() and self.tpu



					@property
					def 	_lowercase					( self					:  List[str]	):
						requires_backends(self							,		["torch"]	)
						# TODO(PVP): currently only single GPU is supported
						return torch.cuda.current_device()



					@property
					def 	_lowercase					( self					:  Optional[int]	):
						requires_backends(self							,		["torch"]	)
						return self._setup_devices[0]



					@property
					def 	_lowercase					( self					:  Optional[int]	):
						requires_backends(self							,		["torch"]	)
						return self._setup_devices[1]






					@property
					def 	_lowercase					( self					:  str	):
						return self.n_gpu > 0











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

__lowerCamelCase :      List[Any]										=				logging.get_logger(__name__)







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							bool					,  snake_case_ :							bool       ):
	def run_func(snake_case_ :							str       ):
		@wraps(snake_case_       )
		def run_in_eager_mode(*snake_case_ :							str					,  **snake_case_ :							Union[str, Any]       ):
			return func(*snake_case_					,  **snake_case_       )

		@wraps(snake_case_       )
		@tf.function(experimental_compile=snake_case_       )
		def run_in_graph_mode(*snake_case_ :							List[Any]					,  **snake_case_ :							List[Any]       ):
			return func(*snake_case_					,  **snake_case_       )

		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 					SCREAMING_SNAKE_CASE    (			snake_case_ :							int					,  snake_case_ :							int					,  snake_case_ :							int       ):
	snake_case__			:				Dict       =  random.Random()
	snake_case__			:				List[Any]       =  [rng.randint(0					,  vocab_size - 1       ) for i in range(batch_size * sequence_length       )]
	return tf.constant(snake_case_					,  shape=(batch_size, sequence_length)					,  dtype=tf.intaa       )





class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					a_     =		42
					a_     =		42
					a_     =		"TensorFlow"



					@property
					def 	_lowercase					( self					:  List[str]	):
						return tf.__version__



					def 	_lowercase					( self					:  List[str]							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						# initialize GPU on separate process
						snake_case__			:				str       =  self.args.strategy
						if strategy is None:
							raise ValueError("A device strategy has to be initialized before using TensorFlow."	)
						snake_case__			:				Dict       =  self._prepare_inference_func(__A							,		__A							,		__A	)
						return self._measure_speed(_inference	)



					def 	_lowercase					( self					:  Tuple							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						snake_case__			:				Optional[int]       =  self.args.strategy
						if strategy is None:
							raise ValueError("A device strategy has to be initialized before using TensorFlow."	)
						snake_case__			:				Any       =  self._prepare_train_func(__A							,		__A							,		__A	)
						return self._measure_speed(_train	)



					def 	_lowercase					( self					:  List[Any]							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						# 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	)
						snake_case__			:				List[str]       =  self.args.strategy
						if strategy is None:
							raise ValueError("A device strategy has to be initialized before using TensorFlow."	)
						snake_case__			:				Optional[Any]       =  self._prepare_inference_func(__A							,		__A							,		__A	)
						return self._measure_memory(_inference	)



					def 	_lowercase					( self					:  str							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						if self.args.is_gpu:
							tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx]							,		__A	)
						snake_case__			:				List[Any]       =  self.args.strategy
						if strategy is None:
							raise ValueError("A device strategy has to be initialized before using TensorFlow."	)

						snake_case__			:				int       =  self._prepare_train_func(__A							,		__A							,		__A	)
						return self._measure_memory(_train	)



					def 	_lowercase					( self					:  Union[str, Any]							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						snake_case__			:				int       =  self.config_dict[model_name]

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

						snake_case__			:				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:
								snake_case__			:				Dict       =  "TF" + config.architectures[0]  # prepend 'TF' for tensorflow model
								snake_case__			:				Union[str, Any]       =  __import__("transformers"							,		fromlist=[model_class]	)
								snake_case__			:				Any       =  getattr(__A							,		__A	)
								snake_case__			:				Dict       =  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:
							snake_case__			:				Dict       =  TF_MODEL_MAPPING[config.__class__](__A	)

						# encoder-decoder has vocab size saved differently
						snake_case__			:				Optional[int]       =  config.vocab_size if hasattr(__A							,		"vocab_size"	) else config.encoder.vocab_size
						snake_case__			:				List[Any]       =  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	)

						snake_case__			:				Optional[int]       =  encoder_decoder_forward if config.is_encoder_decoder else encoder_forward

						return _inference



					def 	_lowercase					( self					:  List[str]							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						snake_case__			:				Optional[Any]       =  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."	)

						snake_case__			:				Any       =  (
						    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:
								snake_case__			:				Dict       =  "TF" + config.architectures[0]  # prepend 'TF' for tensorflow model
								snake_case__			:				List[Any]       =  __import__("transformers"							,		fromlist=[model_class]	)
								snake_case__			:				Optional[int]       =  getattr(__A							,		__A	)
								snake_case__			:				str       =  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:
							snake_case__			:				Union[str, Any]       =  TF_MODEL_WITH_LM_HEAD_MAPPING[config.__class__](__A	)

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

						@run_with_tf_optimizations(self.args.eager_mode							,		self.args.use_xla	)
						def encoder_decoder_train():
							snake_case__			:				str       =  model(__A							,		decoder_input_ids=__A							,		labels=__A							,		training=__A	)[0]
							snake_case__			:				Dict       =  tf.gradients(__A							,		model.trainable_variables	)
							return gradients

						@run_with_tf_optimizations(self.args.eager_mode							,		self.args.use_xla	)
						def encoder_train():
							snake_case__			:				Optional[Any]       =  model(__A							,		labels=__A							,		training=__A	)[0]
							snake_case__			:				Dict       =  tf.gradients(__A							,		model.trainable_variables	)
							return gradients

						snake_case__			:				int       =  encoder_decoder_train if config.is_encoder_decoder else encoder_train

						return _train



					def 	_lowercase					( self					:  int							,		__A					:  List[Any]	):
						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
								snake_case__			:				Optional[Any]       =  timeit.repeat(
								    __A							,		repeat=self.args.repeat							,		number=1_0							,		)

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






					def 	_lowercase					( self					:  str							,		__A					:  Callable[[], None]	):
						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."	)
									snake_case__			:				Optional[int]       =  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."	)
										snake_case__			:				List[str]       =  "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__			:				Any       =  nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx	)
										snake_case__			:				Optional[Any]       =  nvml.nvmlDeviceGetMemoryInfo(__A	)
										snake_case__			:				Optional[int]       =  meminfo.used
										snake_case__			:				Any       =  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."	)
										snake_case__			:				int       =  None
									else:
										snake_case__			:				Any       =  measure_peak_memory_cpu(__A	)
										snake_case__			:				Tuple       =  Memory(__A	) if isinstance(__A							,		__A	) else memory_bytes
								if self.args.trace_memory_line_by_line:
									snake_case__			:				Optional[int]       =  stop_memory_tracing(__A	)
									if memory is None:
										snake_case__			:				Dict       =  summary.total
								else:
									snake_case__			:				List[str]       =  None

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



