



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 							lowerCamelCase   (    self					):



    '''simple docstring'''
    __lowerCamelCase    				=     AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''					)
    __lowerCamelCase    				=     AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''					).to(__UpperCAmelCase					)
    __lowerCamelCase    				=     -1

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

    with CaptureStdout() as cs:
       __lowerCamelCase    				=     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
    __lowerCamelCase    				=     cs.out[:-1]

    self.assertEqual(__UpperCAmelCase						,					__UpperCAmelCase					)




 def 							lowerCamelCase   (    self					):



    '''simple docstring'''
    __lowerCamelCase    				=     AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''					)
    __lowerCamelCase    				=     AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''					).to(__UpperCAmelCase					)
    __lowerCamelCase    				=     -1

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

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

    self.assertEqual(__UpperCAmelCase						,					__UpperCAmelCase					)




 def 							lowerCamelCase   (    self					):



    '''simple docstring'''
    __lowerCamelCase    				=     AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''					)
    __lowerCamelCase    				=     AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''					).to(__UpperCAmelCase					)
    __lowerCamelCase    				=     -1

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

    with CaptureStdout() as cs:
       __lowerCamelCase    				=     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
    __lowerCamelCase    				=     cs.out[:-1]

    self.assertEqual(__UpperCAmelCase						,					__UpperCAmelCase					)




 def 							lowerCamelCase   (    self					):



    '''simple docstring'''
    # Tests that we can pass `decode_kwargs` to the streamer to control how the tokens are decoded. Must be tested
    # with actual models -- the dummy models' tokenizers are not aligned with their models, and
    # `skip_special_tokens=True` has no effect on them
    __lowerCamelCase    				=     AutoTokenizer.from_pretrained('''distilgpt2'''					)
    __lowerCamelCase    				=     AutoModelForCausalLM.from_pretrained('''distilgpt2'''					).to(__UpperCAmelCase					)
    __lowerCamelCase    				=     -1

    __lowerCamelCase    				=     torch.ones((1, 5)						,					device=__UpperCAmelCase					).long() * model.config.bos_token_id
    with CaptureStdout() as cs:
       __lowerCamelCase    				=     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
    __lowerCamelCase    				=     cs.out[:-1]  # Remove the final "\n"
    __lowerCamelCase    				=     tokenizer(__UpperCAmelCase						,					return_tensors='''pt'''					)
    self.assertEqual(streamer_text_tokenized.input_ids.shape						,					(1, 1)					)




 def 							lowerCamelCase   (    self					):



    '''simple docstring'''
    __lowerCamelCase    				=     AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''					)
    __lowerCamelCase    				=     AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''					).to(__UpperCAmelCase					)
    __lowerCamelCase    				=     -1

    __lowerCamelCase    				=     ids_tensor((1, 5)						,					vocab_size=model.config.vocab_size					).to(__UpperCAmelCase					)
    __lowerCamelCase    				=     TextIteratorStreamer(__UpperCAmelCase						,					timeout=0.001					)
    __lowerCamelCase    				=     {'''input_ids''': input_ids, '''max_new_tokens''': 10, '''do_sample''': False, '''streamer''': streamer}
    __lowerCamelCase    				=     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					):
       __lowerCamelCase    				=     ''''''
       for new_text in streamer:
          streamer_text += new_text











def 		a__      (		_UpperCamelCase      :			int				):
   __lowerCamelCase    				=     int(_UpperCamelCase				)
   if decimal in (0, 1):  # Exit cases for the recursion
      return str(_UpperCamelCase				)
   __lowerCamelCase				,__lowerCamelCase    				=     divmod(_UpperCamelCase			,2				)
   return binary_recursive(_UpperCamelCase				) + str(_UpperCamelCase				)





def 		a__      (		_UpperCamelCase      :			str				):
   __lowerCamelCase    				=     str(_UpperCamelCase				).strip()
   if not number:
      raise ValueError('''No input value was provided'''				)
   __lowerCamelCase    				=     '''-''' if number.startswith('''-'''				) else ''''''
   __lowerCamelCase    				=     number.lstrip('''-'''				)
   if not number.isnumeric():
      raise ValueError('''Input value is not an integer'''				)
   return F"""{negative}0b{binary_recursive(int(_UpperCamelCase				)				)}"""


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

  testmod()






