




"""simple docstring"""

import unittest
from queue import Empty
from threading import Thread

from transformers import AutoTokenizer, TextIteratorStreamer, TextStreamer, is_torch_available
from transformers.testing_utils import CaptureStdout, require_torch, torch_device

from ..test_modeling_common import ids_tensor


if is_torch_available():
       import torch

       from transformers import AutoModelForCausalLM





@require_torch
class 							lowercase_			(	unittest.TestCase				):



 '''simple docstring'''

 def      lowerCAmelCase_				(					self			:			Union[str, Any]						):
       _A     =  AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2'						)
       _A     =  AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2'						).to(_UpperCAmelCase						)
       _A     =  -1

       _A     =  ids_tensor((1, 5)       ,				vocab_size=model.config.vocab_size						).to(_UpperCAmelCase						)
       _A     =  model.generate(_UpperCAmelCase       ,				max_new_tokens=10       ,				do_sample=_UpperCAmelCase						)
       _A     =  tokenizer.decode(greedy_ids[0]						)

       with CaptureStdout() as cs:
             _A     =  TextStreamer(_UpperCAmelCase						)
             model.generate(_UpperCAmelCase       ,				max_new_tokens=10       ,				do_sample=_UpperCAmelCase       ,				streamer=_UpperCAmelCase						)
       # The greedy text should be printed to stdout, except for the final "\n" in the streamer
       _A     =  cs.out[:-1]

       self.assertEqual(_UpperCAmelCase       ,				_UpperCAmelCase						)

 def      lowerCAmelCase_				(					self			:			str						):
       _A     =  AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2'						)
       _A     =  AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2'						).to(_UpperCAmelCase						)
       _A     =  -1

       _A     =  ids_tensor((1, 5)       ,				vocab_size=model.config.vocab_size						).to(_UpperCAmelCase						)
       _A     =  model.generate(_UpperCAmelCase       ,				max_new_tokens=10       ,				do_sample=_UpperCAmelCase						)
       _A     =  tokenizer.decode(greedy_ids[0]						)

       _A     =  TextIteratorStreamer(_UpperCAmelCase						)
       _A     =  {'input_ids': input_ids, 'max_new_tokens': 10, 'do_sample': False, 'streamer': streamer}
       _A     =  Thread(target=model.generate       ,				kwargs=_UpperCAmelCase						)
       thread.start()
       _A     =  ''
       for new_text in streamer:
             streamer_text += new_text

       self.assertEqual(_UpperCAmelCase       ,				_UpperCAmelCase						)

 def      lowerCAmelCase_				(					self			:			Dict						):
       _A     =  AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2'						)
       _A     =  AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2'						).to(_UpperCAmelCase						)
       _A     =  -1

       _A     =  ids_tensor((1, 5)       ,				vocab_size=model.config.vocab_size						).to(_UpperCAmelCase						)
       _A     =  model.generate(_UpperCAmelCase       ,				max_new_tokens=10       ,				do_sample=_UpperCAmelCase						)
       _A     =  greedy_ids[:, input_ids.shape[1] :]
       _A     =  tokenizer.decode(new_greedy_ids[0]						)

       with CaptureStdout() as cs:
             _A     =  TextStreamer(_UpperCAmelCase       ,				skip_prompt=_UpperCAmelCase						)
             model.generate(_UpperCAmelCase       ,				max_new_tokens=10       ,				do_sample=_UpperCAmelCase       ,				streamer=_UpperCAmelCase						)
       # The greedy text should be printed to stdout, except for the final "\n" in the streamer
       _A     =  cs.out[:-1]

       self.assertEqual(_UpperCAmelCase       ,				_UpperCAmelCase						)

 def      lowerCAmelCase_				(					self			:			Any						):
       _A     =  AutoTokenizer.from_pretrained('distilgpt2'						)
       _A     =  AutoModelForCausalLM.from_pretrained('distilgpt2'						).to(_UpperCAmelCase						)
       _A     =  -1

       _A     =  torch.ones((1, 5)       ,				device=_UpperCAmelCase						).long() * model.config.bos_token_id
       with CaptureStdout() as cs:
             _A     =  TextStreamer(_UpperCAmelCase       ,				skip_special_tokens=_UpperCAmelCase						)
             model.generate(_UpperCAmelCase       ,				max_new_tokens=1       ,				do_sample=_UpperCAmelCase       ,				streamer=_UpperCAmelCase						)

       # The prompt contains a special token, so the streamer should not print it. As such, the output text, when
       # re-tokenized, must only contain one token
       _A     =  cs.out[:-1]  # Remove the final "\n"
       _A     =  tokenizer(_UpperCAmelCase       ,				return_tensors='pt'						)
       self.assertEqual(streamer_text_tokenized.input_ids.shape       ,				(1, 1)						)



 def      lowerCAmelCase_				(					self			:			Any						):
       _A     =  AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2'						)
       _A     =  AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2'						).to(_UpperCAmelCase						)
       _A     =  -1

       _A     =  ids_tensor((1, 5)       ,				vocab_size=model.config.vocab_size						).to(_UpperCAmelCase						)
       _A     =  TextIteratorStreamer(_UpperCAmelCase       ,				timeout=0.001						)
       _A     =  {'input_ids': input_ids, 'max_new_tokens': 10, 'do_sample': False, 'streamer': streamer}
       _A     =  Thread(target=model.generate       ,				kwargs=_UpperCAmelCase						)
       thread.start()

       # The streamer will timeout after 0.001 seconds, so an exception will be raised
       with self.assertRaises(_UpperCAmelCase						):
             _A     =  ''
             for new_text in streamer:
                   streamer_text += new_text







from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_torch_available, is_torch_tpu_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_torch_available():
				import torch

if is_torch_tpu_available(check_device=False):
				import torch_xla.core.xla_model as xm


SCREAMING_SNAKE_CASE__     :     int		    =       logging.get_logger(__name__)



@dataclass
class       a_       (     SCREAMING_SNAKE_CASE__     ):
	A    							= [
	    '''no_inference''',
	    '''no_cuda''',
	    '''no_tpu''',
	    '''no_speed''',
	    '''no_memory''',
	    '''no_env_print''',
	    '''no_multi_process''',
	]





	def __init__(							self      ,     **SCREAMING_SNAKE_CASE      )  ->					Dict:




				"""simple docstring"""



				for deprecated_arg in self.deprecated_args:
							if deprecated_arg in kwargs:
										SCREAMING_SNAKE_CASE_								=						deprecated_arg[3:]
										setattr(self      ,     SCREAMING_SNAKE_CASE      ,     not kwargs.pop(SCREAMING_SNAKE_CASE      )      )
										logger.warning(
										    f'{deprecated_arg} is depreciated. Please use --no_{positive_arg} or'
										    f' {positive_arg}={kwargs[positive_arg]}'      )

				SCREAMING_SNAKE_CASE_								=						kwargs.pop('torchscript'      ,     self.torchscript      )
				SCREAMING_SNAKE_CASE_								=						kwargs.pop('torch_xla_tpu_print_metrics'      ,     self.torch_xla_tpu_print_metrics      )
				SCREAMING_SNAKE_CASE_								=						kwargs.pop('fp16_opt_level'      ,     self.fpaa_opt_level      )
				super().__init__(**SCREAMING_SNAKE_CASE      )

	A    							= field(default=SCREAMING_SNAKE_CASE__  ,    metadata={'''help''': '''Trace the models using torchscript'''}     )
	A    							= field(default=SCREAMING_SNAKE_CASE__  ,    metadata={'''help''': '''Print Xla/PyTorch tpu metrics'''}     )
	A    							= field(
	    default='''O1'''  ,    metadata={
	        '''help''': (
	            '''For fp16: Apex AMP optimization level selected in [\'O0\', \'O1\', \'O2\', and \'O3\']. '''
	            '''See details at https://nvidia.github.io/apex/amp.html'''
	        )
	    }  ,    )





	@cached_property
	def 						A_(							self      )  ->					Tuple["torch.device", int]:




				"""simple docstring"""



				requires_backends(self      ,     ['torch']      )
				logger.info('PyTorch: setting up devices'      )
				if not self.cuda:
							SCREAMING_SNAKE_CASE_								=						torch.device('cpu'      )
							SCREAMING_SNAKE_CASE_								=						0
				elif is_torch_tpu_available():
							SCREAMING_SNAKE_CASE_								=						xm.xla_device()
							SCREAMING_SNAKE_CASE_								=						0
				else:
							SCREAMING_SNAKE_CASE_								=						torch.device('cuda' if torch.cuda.is_available() else 'cpu'      )
							SCREAMING_SNAKE_CASE_								=						torch.cuda.device_count()
				return device, n_gpu





	@property
	def 						A_(							self      )  ->					Union[str, Any]:




				"""simple docstring"""



				return is_torch_tpu_available() and self.tpu





	@property
	def 						A_(							self      )  ->					int:




				"""simple docstring"""



				requires_backends(self      ,     ['torch']      )
				# TODO(PVP): currently only single GPU is supported
				return torch.cuda.current_device()





	@property
	def 						A_(							self      )  ->					"torch.device":




				"""simple docstring"""



				requires_backends(self      ,     ['torch']      )
				return self._setup_devices[0]





	@property
	def 						A_(							self      )  ->					Optional[Any]:




				"""simple docstring"""



				requires_backends(self      ,     ['torch']      )
				return self._setup_devices[1]





	@property
	def 						A_(							self      )  ->					Dict:




				"""simple docstring"""



				return self.n_gpu > 0
