

'''simple docstring'''




from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_tf_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_tf_available():
      import tensorflow as tf


_lowerCAmelCase         =						logging.get_logger(__name__)






@dataclass
class   _SCREAMING_SNAKE_CASE   (   __a     ):
 __SCREAMING_SNAKE_CASE					:Dict      	=					[
     """no_inference""",
     """no_cuda""",
     """no_tpu""",
     """no_speed""",
     """no_memory""",
     """no_env_print""",
     """no_multi_process""",
 ]



 def __init__(   self		:   int			,				**a__		:   List[Any]      ):
      for deprecated_arg in self.deprecated_args:
           if deprecated_arg in kwargs:
                __magic_name__						       =    deprecated_arg[3:]
                __magic_name__						       =    not kwargs.pop(a__      )
                logger.warning(
                    F'''{deprecated_arg} is depreciated. Please use --no-{positive_arg} or'''
                    F''' {positive_arg}={kwargs[positive_arg]}'''      )
      __magic_name__						       =    kwargs.pop('''tpu_name'''			,				self.tpu_name      )
      __magic_name__						       =    kwargs.pop('''device_idx'''			,				self.device_idx      )
      __magic_name__						       =    kwargs.pop('''eager_mode'''			,				self.eager_mode      )
      __magic_name__						       =    kwargs.pop('''use_xla'''			,				self.use_xla      )
      super().__init__(**a__      )

 __SCREAMING_SNAKE_CASE					:str      	=					field(
     default=__a		,metadata={"""help""": """Name of TPU"""}		,)
 __SCREAMING_SNAKE_CASE					:int      	=					field(
     default=0		,metadata={"""help""": """CPU / GPU device index. Defaults to 0."""}		,)
 __SCREAMING_SNAKE_CASE					:bool      	=					field(default=__a		,metadata={"""help""": """Benchmark models in eager model."""}     )
 __SCREAMING_SNAKE_CASE					:bool      	=					field(
     default=__a		,metadata={
         """help""": """Benchmark models using XLA JIT compilation. Note that `eager_model` has to be set to `False`."""
     }		,)



 @cached_property
 def 			snake_case__  (   self		:   int      ):
      requires_backends(self			,				['''tf''']      )
      __magic_name__						       =    None
      if self.tpu:
           try:
                if self.tpu_name:
                     __magic_name__						       =    tf.distribute.cluster_resolver.TPUClusterResolver(self.tpu_name      )
                else:
                     __magic_name__						       =    tf.distribute.cluster_resolver.TPUClusterResolver()
           except ValueError:
                __magic_name__						       =    None
      return tpu



 @cached_property
 def 			snake_case__  (   self		:   Optional[Any]      ):
      requires_backends(self			,				['''tf''']      )
      if self.is_tpu:
           tf.config.experimental_connect_to_cluster(self._setup_tpu      )
           tf.tpu.experimental.initialize_tpu_system(self._setup_tpu      )

           __magic_name__						       =    tf.distribute.TPUStrategy(self._setup_tpu      )
      else:
           # currently no multi gpu is allowed
           if self.is_gpu:
                # TODO: Currently only single GPU is supported
                tf.config.set_visible_devices(self.gpu_list[self.device_idx]			,				'''GPU'''      )
                __magic_name__						       =    tf.distribute.OneDeviceStrategy(device=F'''/gpu:{self.device_idx}'''      )
           else:
                tf.config.set_visible_devices([]			,				'''GPU'''      )  # disable GPU
                __magic_name__						       =    tf.distribute.OneDeviceStrategy(device=F'''/cpu:{self.device_idx}'''      )

      return strategy



 @property
 def 			snake_case__  (   self		:   List[str]      ):
      requires_backends(self			,				['''tf''']      )
      return self._setup_tpu is not None



 @property
 def 			snake_case__  (   self		:   int      ):
      requires_backends(self			,				['''tf''']      )
      return self._setup_strategy



 @property
 def 			snake_case__  (   self		:   str      ):
      requires_backends(self			,				['''tf''']      )
      return tf.config.list_physical_devices('''GPU'''      )



 @property
 def 			snake_case__  (   self		:   Dict      ):
      requires_backends(self			,				['''tf''']      )
      if self.cuda:
           return len(self.gpu_list      )
      return 0



 @property
 def 			snake_case__  (   self		:   Optional[Any]      ):
      return self.n_gpu > 0









'''simple docstring'''




import unittest

from transformers import (
    MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TextaTextGenerationPipeline,
    pipeline,
)
from transformers.testing_utils import is_pipeline_test, require_tf, require_torch
from transformers.utils import is_torch_available

from .test_pipelines_common import ANY


if is_torch_available():
      import torch






@is_pipeline_test
class   _SCREAMING_SNAKE_CASE   (   unittest.TestCase     ):
 __SCREAMING_SNAKE_CASE					:List[Any]      	=					MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
 __SCREAMING_SNAKE_CASE					:Any      	=					TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING



 def 			snake_case__  (   self		:   Tuple			,				a__		:   Tuple			,				a__		:   int			,				a__		:   int      ):
      __magic_name__						       =    TextaTextGenerationPipeline(model=a__			,				tokenizer=a__      )
      return generator, ["Something to write", "Something else"]



 def 			snake_case__  (   self		:   List[str]			,				a__		:   List[Any]			,				a__		:   List[str]      ):
      __magic_name__						       =    generator('''Something there'''      )
      self.assertEqual(a__			,				[{'''generated_text''': ANY(a__      )}]      )
      # These are encoder decoder, they don't just append to incoming string
      self.assertFalse(outputs[0]['''generated_text'''].startswith('''Something there'''      )      )

      __magic_name__						       =    generator(['''This is great !''', '''Something else''']			,				num_return_sequences=2			,				do_sample=a__      )
      self.assertEqual(
          a__			,				[
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
          ]			,				)

      __magic_name__						       =    generator(
          ['''This is great !''', '''Something else''']			,				num_return_sequences=2			,				batch_size=2			,				do_sample=a__      )
      self.assertEqual(
          a__			,				[
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
          ]			,				)

      with self.assertRaises(a__      ):
           generator(4      )



 @require_torch
 def 			snake_case__  (   self		:   Any      ):
      __magic_name__						       =    pipeline('''text2text-generation'''			,				model='''patrickvonplaten/t5-tiny-random'''			,				framework='''pt'''      )
      # do_sample=False necessary for reproducibility
      __magic_name__						       =    generator('''Something there'''			,				do_sample=a__      )
      self.assertEqual(a__			,				[{'''generated_text''': ''''''}]      )

      __magic_name__						       =    3
      __magic_name__						       =    generator(
          '''Something there'''			,				num_return_sequences=a__			,				num_beams=a__			,				)
      __magic_name__						       =    [
          {'''generated_text''': '''Beide Beide Beide Beide Beide Beide Beide Beide Beide'''},
          {'''generated_text''': '''Beide Beide Beide Beide Beide Beide Beide Beide'''},
          {'''generated_text''': ''''''},
      ]
      self.assertEqual(a__			,				a__      )

      __magic_name__						       =    generator('''This is a test'''			,				do_sample=a__			,				num_return_sequences=2			,				return_tensors=a__      )
      self.assertEqual(
          a__			,				[
              {'''generated_token_ids''': ANY(torch.Tensor      )},
              {'''generated_token_ids''': ANY(torch.Tensor      )},
          ]			,				)
      __magic_name__						       =    generator.model.config.eos_token_id
      __magic_name__						       =    '''<pad>'''
      __magic_name__						       =    generator(
          ['''This is a test''', '''This is a second test''']			,				do_sample=a__			,				num_return_sequences=2			,				batch_size=2			,				return_tensors=a__			,				)
      self.assertEqual(
          a__			,				[
              [
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
              ],
              [
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
              ],
          ]			,				)



 @require_tf
 def 			snake_case__  (   self		:   int      ):
      __magic_name__						       =    pipeline('''text2text-generation'''			,				model='''patrickvonplaten/t5-tiny-random'''			,				framework='''tf'''      )
      # do_sample=False necessary for reproducibility
      __magic_name__						       =    generator('''Something there'''			,				do_sample=a__      )
      self.assertEqual(a__			,				[{'''generated_text''': ''''''}]      )






