



"""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__					:						int												=			logging.get_logger(__name__)




def __lowercase    (							_a	,    _a					):
    def run_func(_a					):
        @wraps(_a					)
        def run_in_eager_mode(*_a	,    **_a					):
            return func(*_a	,    **_a					)

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

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






class  _UpperCAmelCase (    lowerCAmelCase__):
 _lowerCAmelCase     :    TensorFlowBenchmarkArguments
 _lowerCAmelCase     :    PretrainedConfig
 _lowerCAmelCase     :    str	      =	"TensorFlow"

 @property
 def 		_snake_case      (  self	: str       ):
     return tf.__version__

 def 		_snake_case      (  self	: Tuple						,  lowercase_	: str						,  lowercase_	: int						,  lowercase_	: int       ):
     # initialize GPU on separate process
     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_				:  Tuple       =			self._prepare_inference_func(lowercase_						,  lowercase_						,  lowercase_       )
     return self._measure_speed(_inference       )

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

 def 		_snake_case      (  self	: str						,  lowercase_	: str						,  lowercase_	: int						,  lowercase_	: 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]						,  lowercase_       )
     snake_case_				:  int       =			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(lowercase_						,  lowercase_						,  lowercase_       )
     return self._measure_memory(_inference       )

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

     snake_case_				:  Union[str, Any]       =			self._prepare_train_func(lowercase_						,  lowercase_						,  lowercase_       )
     return self._measure_memory(_train       )

 def 		_snake_case      (  self	: Optional[Any]						,  lowercase_	: str						,  lowercase_	: int						,  lowercase_	: int       ):
     snake_case_				:  Dict       =			self.config_dict[model_name]

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

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

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

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

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

     snake_case_				:  Union[str, Any]       =			encoder_decoder_forward if config.is_encoder_decoder else encoder_forward

     return _inference

 def 		_snake_case      (  self	: Tuple						,  lowercase_	: str						,  lowercase_	: int						,  lowercase_	: int       ):
     snake_case_				:  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.'''       )

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

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

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

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

     snake_case_				:  int       =			encoder_decoder_train if config.is_encoder_decoder else encoder_train

     return _train

 def 		_snake_case      (  self	: Optional[int]						,  lowercase_	: 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(lowercase_						,  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_				:  List[str]       =			timeit.repeat(
                 lowercase_						,  repeat=self.args.repeat						,  number=10						,  )

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






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










"""simple docstring"""

def __lowercase    (							_a					):
    snake_case_				:  Optional[Any]       =			int(_a					)
    if decimal in (0, 1):  # Exit cases for the recursion
        return str(_a					)
    snake_case_,			snake_case_				:  Optional[int]       =			divmod(_a	,    2					)
    return binary_recursive(_a					) + str(_a					)




def __lowercase    (							_a					):
    snake_case_				:  Dict       =			str(_a					).strip()
    if not number:
        raise ValueError('''No input value was provided'''					)
    snake_case_				:  List[str]       =			'''-''' if number.startswith('''-'''					) else ''''''
    snake_case_				:  Optional[Any]       =			number.lstrip('''-'''					)
    if not number.isnumeric():
        raise ValueError('''Input value is not an integer'''					)
    return f"{negative}0b{binary_recursive(int(_a					)					)}"


if __name__ == "__main__":
     from doctest import testmod

     testmod()





