



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

lowerCAmelCase:      Tuple		 =		logging.get_logger(__name__)
def    lowerCamelCase__ (		_A				,    _A       ):


		def run_func(_A       ):
				@wraps(_lowerCamelCase       )
				def run_in_eager_mode(*_A				,    **_A       ):
						return func(*_lowerCamelCase				,    **_lowerCamelCase       )

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

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

class 							a__(      lowercase__			):
				lowercase__				=					42
				lowercase__				=					42
				lowercase__				=					"""TensorFlow"""







				@property
				def     lowercase_ (   self				:						Optional[Any]			):
						return tf.__version__







				def     lowercase_ (   self				:						Optional[Any]		,    __snake_case				:						str		,    __snake_case				:						int		,    __snake_case				:						int			):
						# initialize GPU on separate process
						a		:      Dict       =  self.args.strategy
						if strategy is None:
								raise ValueError('A device strategy has to be initialized before using TensorFlow.'			)
						a		:      Any       =  self._prepare_inference_func(__snake_case		,    __snake_case		,    __snake_case			)
						return self._measure_speed(_inference			)







				def     lowercase_ (   self				:						Optional[Any]		,    __snake_case				:						str		,    __snake_case				:						int		,    __snake_case				:						int			):
						a		:      Optional[Any]       =  self.args.strategy
						if strategy is None:
								raise ValueError('A device strategy has to be initialized before using TensorFlow.'			)
						a		:      Any       =  self._prepare_train_func(__snake_case		,    __snake_case		,    __snake_case			)
						return self._measure_speed(_train			)







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







				def     lowercase_ (   self				:						Dict		,    __snake_case				:						str		,    __snake_case				:						int		,    __snake_case				:						int			):
						if self.args.is_gpu:
								tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx]		,    __snake_case			)
						a		:      List[str]       =  self.args.strategy
						if strategy is None:
								raise ValueError('A device strategy has to be initialized before using TensorFlow.'			)

						a		:      Dict       =  self._prepare_train_func(__snake_case		,    __snake_case		,    __snake_case			)
						return self._measure_memory(_train			)







				def     lowercase_ (   self				:						Optional[Any]		,    __snake_case				:						str		,    __snake_case				:						int		,    __snake_case				:						int			):
						a		:      List[str]       =  self.config_dict[model_name]

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

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

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

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

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

						a		:      Tuple       =  encoder_decoder_forward if config.is_encoder_decoder else encoder_forward

						return _inference







				def     lowercase_ (   self				:						str		,    __snake_case				:						str		,    __snake_case				:						int		,    __snake_case				:						int			):
						a		:      Union[str, 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.'			)

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

						# encoder-decoder has vocab size saved differently
						a		:      List[str]       =  config.vocab_size if hasattr(__snake_case		,    'vocab_size'			) else config.encoder.vocab_size
						a		:      List[Any]       =  random_input_ids(__snake_case		,    __snake_case		,    __snake_case			)

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

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

						a		:      Optional[int]       =  encoder_decoder_train if config.is_encoder_decoder else encoder_train

						return _train







				def     lowercase_ (   self				:						Union[str, Any]		,    __snake_case				:						str			):
						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(__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		:      List[Any]       =  timeit.repeat(
										    __snake_case		,    repeat=self.args.repeat		,    number=10		,    )

										return min(__snake_case			) / 10.0
								except ResourceExhaustedError as e:
										self.print_fn(F"""Doesn't fit on GPU. {e}"""			)







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






from ..utils import DummyObject, requires_backends

class 							a__(      metaclass=lowerCamelCase__			):
    lowercase__				=					["""torch""", """torchsde"""]







    def __init__(   self				:						Any		,    *__snake_case				:						List[str]		,    **__snake_case				:						Tuple			):
      requires_backends(self		,    ['torch', 'torchsde']			)







    @classmethod
    def     lowercase_ (   cls				:						List[Any]		,    *__snake_case				:						str		,    **__snake_case				:						List[Any]			):
      requires_backends(cls		,    ['torch', 'torchsde']			)







    @classmethod
    def     lowercase_ (   cls				:						Dict		,    *__snake_case				:						Optional[Any]		,    **__snake_case				:						str			):
      requires_backends(cls		,    ['torch', 'torchsde']			)