


'''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 			lowerCamelCase_   (	unittest.TestCase      ):






      def lowercase_							(					self				:    Any ):


          '''simple docstring'''


          UpperCAmelCase__			:		int													=				AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2''' )
          UpperCAmelCase__			:		int													=				AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2''' ).to(_A )
          UpperCAmelCase__			:		Dict													=				-1

          UpperCAmelCase__			:		List[str]													=				ids_tensor((1, 5)						,						vocab_size=model.config.vocab_size ).to(_A )
          UpperCAmelCase__			:		List[str]													=				model.generate(_A						,						max_new_tokens=10						,						do_sample=_A )
          UpperCAmelCase__			:		Union[str, Any]													=				tokenizer.decode(greedy_ids[0] )

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

          self.assertEqual(_A						,						_A )






      def lowercase_							(					self				:    Any ):


          '''simple docstring'''


          UpperCAmelCase__			:		int													=				AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2''' )
          UpperCAmelCase__			:		Optional[int]													=				AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2''' ).to(_A )
          UpperCAmelCase__			:		List[Any]													=				-1

          UpperCAmelCase__			:		List[Any]													=				ids_tensor((1, 5)						,						vocab_size=model.config.vocab_size ).to(_A )
          UpperCAmelCase__			:		int													=				model.generate(_A						,						max_new_tokens=10						,						do_sample=_A )
          UpperCAmelCase__			:		Optional[int]													=				tokenizer.decode(greedy_ids[0] )

          UpperCAmelCase__			:		Union[str, Any]													=				TextIteratorStreamer(_A )
          UpperCAmelCase__			:		Tuple													=				{'''input_ids''': input_ids, '''max_new_tokens''': 10, '''do_sample''': False, '''streamer''': streamer}
          UpperCAmelCase__			:		Optional[int]													=				Thread(target=model.generate						,						kwargs=_A )
          thread.start()
          UpperCAmelCase__			:		Optional[int]													=				''''''
          for new_text in streamer:
              streamer_text += new_text

          self.assertEqual(_A						,						_A )






      def lowercase_							(					self				:    List[Any] ):


          '''simple docstring'''


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

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

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

          self.assertEqual(_A						,						_A )






      def lowercase_							(					self				:    List[str] ):


          '''simple docstring'''


          UpperCAmelCase__			:		Any													=				AutoTokenizer.from_pretrained('''distilgpt2''' )
          UpperCAmelCase__			:		Optional[Any]													=				AutoModelForCausalLM.from_pretrained('''distilgpt2''' ).to(_A )
          UpperCAmelCase__			:		List[str]													=				-1

          UpperCAmelCase__			:		Tuple													=				torch.ones((1, 5)						,						device=_A ).long() * model.config.bos_token_id
          with CaptureStdout() as cs:
              UpperCAmelCase__			:		List[Any]													=				TextStreamer(_A						,						skip_special_tokens=_A )
              model.generate(_A						,						max_new_tokens=1						,						do_sample=_A						,						streamer=_A )

          # 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
          UpperCAmelCase__			:		int													=				cs.out[:-1]  # Remove the final "\n"
          UpperCAmelCase__			:		str													=				tokenizer(_A						,						return_tensors='''pt''' )
          self.assertEqual(streamer_text_tokenized.input_ids.shape						,						(1, 1) )






      def lowercase_							(					self				:    Tuple ):


          '''simple docstring'''


          UpperCAmelCase__			:		List[Any]													=				AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2''' )
          UpperCAmelCase__			:		Any													=				AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2''' ).to(_A )
          UpperCAmelCase__			:		Dict													=				-1

          UpperCAmelCase__			:		Dict													=				ids_tensor((1, 5)						,						vocab_size=model.config.vocab_size ).to(_A )
          UpperCAmelCase__			:		str													=				TextIteratorStreamer(_A						,						timeout=0.0_0_1 )
          UpperCAmelCase__			:		Optional[int]													=				{'''input_ids''': input_ids, '''max_new_tokens''': 10, '''do_sample''': False, '''streamer''': streamer}
          UpperCAmelCase__			:		Any													=				Thread(target=model.generate						,						kwargs=_A )
          thread.start()

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




'''simple docstring'''
def 					a__       (					lowerCAmelCase__						)    ->	int:
    if a < 0:
        raise ValueError('''Input value must be a positive integer'''						)
    elif isinstance(lowerCAmelCase__  ,					lowerCAmelCase__						):
        raise TypeError('''Input value must be a \'int\' type'''						)
    return bin(lowerCAmelCase__						).count('''1'''						)


if __name__ == "__main__":
  import doctest

  doctest.testmod()
