






'''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_					=	logging.get_logger(__name__)





def 	_a(					UpperCamelCase__			:  bool,		UpperCamelCase__			:  bool							):




      '''simple docstring'''






      def run_func(UpperCamelCase__			:  Optional[Any]							):
            @wraps(UpperCamelCase__							)
            def run_in_eager_mode(*UpperCamelCase__			:  int,		**UpperCamelCase__			:  Dict							):
                  return func(*UpperCamelCase__,		**UpperCamelCase__							)

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

            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 	_a(					UpperCamelCase__			:  int,		UpperCamelCase__			:  int,		UpperCamelCase__			:  int							):




      '''simple docstring'''






      SCREAMING_SNAKE_CASE__       :			Optional[Any]													=random.Random()
      SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=[rng.randint(0,		vocab_size - 1							) for i in range(batch_size * sequence_length							)]
      return tf.constant(UpperCamelCase__,		shape=(batch_size, sequence_length),		dtype=tf.intaa							)







class        __SCREAMING_SNAKE_CASE (     lowerCamelCase			):
 snake_case_												=							42
 snake_case_												=							42
 snake_case_												=							"TensorFlow"




 @property
 def        __magic_name__		( self		:							List[Any] )		->  int:
       return tf.__version__




 def        __magic_name__		( self		:							Union[str, Any]   , __lowercase		:							str   , __lowercase		:							int   , __lowercase		:							int )		->  str:
       # initialize GPU on separate process
       SCREAMING_SNAKE_CASE__       :			List[str]													=self.args.strategy
       if strategy is None:
             raise ValueError('''A device strategy has to be initialized before using TensorFlow.''' )
       SCREAMING_SNAKE_CASE__       :			Optional[int]													=self._prepare_inference_func(_a   , _a   , _a )
       return self._measure_speed(_inference )




 def        __magic_name__		( self		:							Tuple   , __lowercase		:							str   , __lowercase		:							int   , __lowercase		:							int )		->  str:
       SCREAMING_SNAKE_CASE__       :			Any													=self.args.strategy
       if strategy is None:
             raise ValueError('''A device strategy has to be initialized before using TensorFlow.''' )
       SCREAMING_SNAKE_CASE__       :			Optional[int]													=self._prepare_train_func(_a   , _a   , _a )
       return self._measure_speed(_train )




 def        __magic_name__		( self		:							Optional[Any]   , __lowercase		:							str   , __lowercase		:							int   , __lowercase		:							int )		->  Optional[Any]:
       # 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 )
       SCREAMING_SNAKE_CASE__       :			Optional[Any]													=self.args.strategy
       if strategy is None:
             raise ValueError('''A device strategy has to be initialized before using TensorFlow.''' )
       SCREAMING_SNAKE_CASE__       :			Tuple													=self._prepare_inference_func(_a   , _a   , _a )
       return self._measure_memory(_inference )




 def        __magic_name__		( self		:							Union[str, Any]   , __lowercase		:							str   , __lowercase		:							int   , __lowercase		:							int )		->  Tuple:
       if self.args.is_gpu:
             tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx]   , _a )
       SCREAMING_SNAKE_CASE__       :			int													=self.args.strategy
       if strategy is None:
             raise ValueError('''A device strategy has to be initialized before using TensorFlow.''' )

       SCREAMING_SNAKE_CASE__       :			Optional[Any]													=self._prepare_train_func(_a   , _a   , _a )
       return self._measure_memory(_train )




 def        __magic_name__		( self		:							Optional[Any]   , __lowercase		:							str   , __lowercase		:							int   , __lowercase		:							int )		->  Dict:
       SCREAMING_SNAKE_CASE__       :			Tuple													=self.config_dict[model_name]

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

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

       # encoder-decoder has vocab size saved differently
       SCREAMING_SNAKE_CASE__       :			Optional[Any]													=config.vocab_size if hasattr(_a   , '''vocab_size''' ) else config.encoder.vocab_size
       SCREAMING_SNAKE_CASE__       :			Optional[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 )

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

       return _inference




 def        __magic_name__		( self		:							int   , __lowercase		:							str   , __lowercase		:							int   , __lowercase		:							int )		->  List[str]:
       SCREAMING_SNAKE_CASE__       :			Dict													=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.''' )

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

       # encoder-decoder has vocab size saved differently
       SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=config.vocab_size if hasattr(_a   , '''vocab_size''' ) else config.encoder.vocab_size
       SCREAMING_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():
             SCREAMING_SNAKE_CASE__       :			List[Any]													=model(_a   , decoder_input_ids=_a   , labels=_a   , training=_a )[0]
             SCREAMING_SNAKE_CASE__       :			Optional[Any]													=tf.gradients(_a   , model.trainable_variables )
             return gradients

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

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

       return _train




 def        __magic_name__		( self		:							Optional[int]   , __lowercase		:							Tuple )		->  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
                   SCREAMING_SNAKE_CASE__       :			Union[str, Any]													=timeit.repeat(
                       _a   , repeat=self.args.repeat   , number=10   , )

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




 def        __magic_name__		( self		:							Union[str, Any]   , __lowercase		:							Callable[[], None] )		->  str:
       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.''' )
                         SCREAMING_SNAKE_CASE__       :			List[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.''' )
                               SCREAMING_SNAKE_CASE__       :			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()
                               SCREAMING_SNAKE_CASE__       :			Any													=nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx )
                               SCREAMING_SNAKE_CASE__       :			Any													=nvml.nvmlDeviceGetMemoryInfo(_a )
                               SCREAMING_SNAKE_CASE__       :			List[str]													=meminfo.used
                               SCREAMING_SNAKE_CASE__       :			Optional[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.''' )
                               SCREAMING_SNAKE_CASE__       :			Optional[int]													=None
                         else:
                               SCREAMING_SNAKE_CASE__       :			str													=measure_peak_memory_cpu(_a )
                               SCREAMING_SNAKE_CASE__       :			Tuple													=Memory(_a ) if isinstance(_a   , _a ) else memory_bytes
                   if self.args.trace_memory_line_by_line:
                         SCREAMING_SNAKE_CASE__       :			int													=stop_memory_tracing(_a )
                         if memory is None:
                               SCREAMING_SNAKE_CASE__       :			Dict													=summary.total
                   else:
                         SCREAMING_SNAKE_CASE__       :			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'''
# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import torch

from ..models.auto import AutoModelForSequenceClassification, AutoTokenizer
from .base import PipelineTool






class        __SCREAMING_SNAKE_CASE (     lowerCamelCase			):
 snake_case_												=							"""facebook/bart-large-mnli"""
 snake_case_												=							(
     """This is a tool that classifies an English text using provided labels. It takes two inputs: `text`, which """
     """should be the text to classify, and `labels`, which should be the list of labels to use for classification. """
     """It returns the most likely label in the list of provided `labels` for the input text."""
 )
 snake_case_												=							"""text_classifier"""
 snake_case_												=							AutoTokenizer
 snake_case_												=							AutoModelForSequenceClassification

 snake_case_												=							["""text""", ["""text"""]]
 snake_case_												=							["""text"""]




 def        __magic_name__		( self		:							Any )		->  Any:
       super().setup()
       SCREAMING_SNAKE_CASE__       :			int													=self.model.config
       SCREAMING_SNAKE_CASE__       :			int													=-1
       for idx, label in config.idalabel.items():
             if label.lower().startswith('''entail''' ):
                   SCREAMING_SNAKE_CASE__       :			Any													=int(__lowercase )
       if self.entailment_id == -1:
             raise ValueError('''Could not determine the entailment ID from the model config, please pass it at init.''' )




 def        __magic_name__		( self		:							str   , __lowercase		:							Any   , __lowercase		:							Optional[Any] )		->  Optional[Any]:
       SCREAMING_SNAKE_CASE__       :			Dict													=labels
       return self.pre_processor(
           [text] * len(__lowercase )   , [F"This example is {label}" for label in labels]   , return_tensors='''pt'''   , padding='''max_length'''   , )




 def        __magic_name__		( self		:							List[Any]   , __lowercase		:							str )		->  List[Any]:
       SCREAMING_SNAKE_CASE__       :			int													=outputs.logits
       SCREAMING_SNAKE_CASE__       :			Any													=torch.argmax(logits[:, 2] ).item()
       return self._labels[label_id]