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

lowercase_      :				Tuple   					=    logging.get_logger(__name__)







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      bool ,  lowercase_    :      bool						):


						def run_func(lowercase_    :      Optional[Any]						):
												@wraps(_lowerCamelCase						)
												def run_in_eager_mode(*lowercase_    :      Optional[Any] ,  **lowercase_    :      List[str]						):
																		return func(*_lowerCamelCase ,  **_lowerCamelCase						)

												@wraps(_lowerCamelCase						)
												@tf.function(experimental_compile=_lowerCamelCase						)
												def run_in_graph_mode(*lowercase_    :      Optional[int] ,  **lowercase_    :      List[str]						):
																		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      SCREAMING_SNAKE_CASE				(				lowercase_    :      int ,  lowercase_    :      int ,  lowercase_    :      int						):
						lowercase        =							random.Random()
						lowercase        =							[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      __UpperCamelCase       (lowercase__			):
	__A			  =					42
	__A			  =					42
	__A			  =					'''TensorFlow'''







	@property
	def   _a   (    self						)    ->    List[Any]:
							'''simple docstring'''






							return tf.__version__







	def   _a   (    self ,				_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)    ->    float:
							'''simple docstring'''






							lowercase        =							self.args.strategy
							if strategy is None:
													raise ValueError("""A device strategy has to be initialized before using TensorFlow."""						)
							lowercase        =							self._prepare_inference_func(_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)
							return self._measure_speed(_inference						)







	def   _a   (    self ,				_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)    ->    float:
							'''simple docstring'''






							lowercase        =							self.args.strategy
							if strategy is None:
													raise ValueError("""A device strategy has to be initialized before using TensorFlow."""						)
							lowercase        =							self._prepare_train_func(_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)
							return self._measure_speed(_train						)







	def   _a   (    self ,				_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)    ->    [Memory, Optional[MemorySummary]]:
							'''simple docstring'''






							if self.args.is_gpu:
													tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx] ,				_lowerCAmelCase						)
							lowercase        =							self.args.strategy
							if strategy is None:
													raise ValueError("""A device strategy has to be initialized before using TensorFlow."""						)
							lowercase        =							self._prepare_inference_func(_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)
							return self._measure_memory(_inference						)







	def   _a   (    self ,				_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)    ->    [Memory, Optional[MemorySummary]]:
							'''simple docstring'''






							if self.args.is_gpu:
													tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx] ,				_lowerCAmelCase						)
							lowercase        =							self.args.strategy
							if strategy is None:
													raise ValueError("""A device strategy has to be initialized before using TensorFlow."""						)

							lowercase        =							self._prepare_train_func(_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)
							return self._measure_memory(_train						)







	def   _a   (    self ,				_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)    ->    Callable[[], None]:
							'''simple docstring'''






							lowercase        =							self.config_dict[model_name]

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

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

							# encoder-decoder has vocab size saved differently
							lowercase        =							config.vocab_size if hasattr(_lowerCAmelCase ,				"""vocab_size"""						) else config.encoder.vocab_size
							lowercase        =							random_input_ids(_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)

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

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

							lowercase        =							encoder_decoder_forward if config.is_encoder_decoder else encoder_forward

							return _inference







	def   _a   (    self ,				_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)    ->    Callable[[], None]:
							'''simple docstring'''






							lowercase        =							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."""						)

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

							# encoder-decoder has vocab size saved differently
							lowercase        =							config.vocab_size if hasattr(_lowerCAmelCase ,				"""vocab_size"""						) else config.encoder.vocab_size
							lowercase        =							random_input_ids(_lowerCAmelCase ,				_lowerCAmelCase ,				_lowerCAmelCase						)

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

							@run_with_tf_optimizations(self.args.eager_mode ,				self.args.use_xla						)
							def encoder_train():
													lowercase        =							model(_lowerCAmelCase ,				labels=_lowerCAmelCase ,				training=_lowerCAmelCase						)[0]
													lowercase        =							tf.gradients(_lowerCAmelCase ,				model.trainable_variables						)
													return gradients

							lowercase        =							encoder_decoder_train if config.is_encoder_decoder else encoder_train

							return _train







	def   _a   (    self ,				_lowerCAmelCase						)    ->    float:
							'''simple docstring'''






							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(_lowerCAmelCase ,				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
																			lowercase        =							timeit.repeat(
																			    _lowerCAmelCase ,				repeat=self.args.repeat ,				number=10 ,				)

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







	def   _a   (    self ,				_lowerCAmelCase						)    ->    [Memory, MemorySummary]:
							'''simple docstring'''






							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."""						)
																									lowercase        =							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."""						)
																															lowercase        =							"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()
																															lowercase        =							nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx						)
																															lowercase        =							nvml.nvmlDeviceGetMemoryInfo(_lowerCAmelCase						)
																															lowercase        =							meminfo.used
																															lowercase        =							Memory(_lowerCAmelCase						)
																															# 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."""						)
																															lowercase        =							None
																									else:
																															lowercase        =							measure_peak_memory_cpu(_lowerCAmelCase						)
																															lowercase        =							Memory(_lowerCAmelCase						) if isinstance(_lowerCAmelCase ,				_lowerCAmelCase						) else memory_bytes
																			if self.args.trace_memory_line_by_line:
																									lowercase        =							stop_memory_tracing(_lowerCAmelCase						)
																									if memory is None:
																															lowercase        =							summary.total
																			else:
																									lowercase        =							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 __future__ import annotations

import os
from typing import Any

import requests

lowercase_      :				List[str]   					=    '''https://api.github.com'''

# https://docs.github.com/en/free-pro-team@latest/rest/reference/users#get-the-authenticated-user
lowercase_      :				Any   					=    BASE_URL + '''/user'''

# https://github.com/settings/tokens
lowercase_      :				Union[str, Any]   					=    os.environ.get('''USER_TOKEN''', '''''')







def      SCREAMING_SNAKE_CASE				(				lowercase_    :      str						):
						lowercase        =							{
						    """Authorization""": F"""token {auth_token}""",
						    """Accept""": """application/vnd.github.v3+json""",
						}
						return requests.get(lowercase_ ,  headers=lowercase_						).json()


if __name__ == "__main__":  # pragma: no cover
						if USER_TOKEN:
												for key, value in fetch_github_info(USER_TOKEN).items():
																		print(f'''{key}: {value}''')
						else:
												raise ValueError('''\'USER_TOKEN\' field cannot be empty.''')


