






"""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 					_UpperCAmelCase       ( unittest.TestCase							):




					def   A  (				self     :						Optional[int]   )  ->  Tuple:
							lowercase_ : Optional[int]							       =   AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''   )
							lowercase_ : List[str]							       =   AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''   ).to(A   )
							lowercase_ : Any							       =   -1

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

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

							self.assertEqual(A    ,						A   )




					def   A  (				self     :						List[Any]   )  ->  int:
							lowercase_ : int							       =   AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''   )
							lowercase_ : List[Any]							       =   AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''   ).to(A   )
							lowercase_ : Optional[int]							       =   -1

							lowercase_ : Dict							       =   ids_tensor((1, 5)    ,						vocab_size=model.config.vocab_size   ).to(A   )
							lowercase_ : List[str]							       =   model.generate(A    ,						max_new_tokens=10    ,						do_sample=A   )
							lowercase_ : List[str]							       =   tokenizer.decode(greedy_ids[0]   )

							lowercase_ : Any							       =   TextIteratorStreamer(A   )
							lowercase_ : int							       =   {'''input_ids''': input_ids, '''max_new_tokens''': 10, '''do_sample''': False, '''streamer''': streamer}
							lowercase_ : int							       =   Thread(target=model.generate    ,						kwargs=A   )
							thread.start()
							lowercase_ : List[Any]							       =   ''''''
							for new_text in streamer:
									streamer_text += new_text

							self.assertEqual(A    ,						A   )




					def   A  (				self     :						Optional[Any]   )  ->  Dict:
							lowercase_ : Dict							       =   AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''   )
							lowercase_ : List[Any]							       =   AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''   ).to(A   )
							lowercase_ : Any							       =   -1

							lowercase_ : List[str]							       =   ids_tensor((1, 5)    ,						vocab_size=model.config.vocab_size   ).to(A   )
							lowercase_ : List[str]							       =   model.generate(A    ,						max_new_tokens=10    ,						do_sample=A   )
							lowercase_ : Union[str, Any]							       =   greedy_ids[:, input_ids.shape[1] :]
							lowercase_ : Optional[int]							       =   tokenizer.decode(new_greedy_ids[0]   )

							with CaptureStdout() as cs:
									lowercase_ : Union[str, 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
							lowercase_ : List[str]							       =   cs.out[:-1]

							self.assertEqual(A    ,						A   )




					def   A  (				self     :						Any   )  ->  Any:
							# 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
							lowercase_ : int							       =   AutoTokenizer.from_pretrained('''distilgpt2'''   )
							lowercase_ : str							       =   AutoModelForCausalLM.from_pretrained('''distilgpt2'''   ).to(A   )
							lowercase_ : str							       =   -1

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






					def   A  (				self     :						List[str]   )  ->  List[Any]:
							lowercase_ : List[Any]							       =   AutoTokenizer.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''   )
							lowercase_ : Optional[Any]							       =   AutoModelForCausalLM.from_pretrained('''hf-internal-testing/tiny-random-gpt2'''   ).to(A   )
							lowercase_ : Optional[int]							       =   -1

							lowercase_ : Optional[Any]							       =   ids_tensor((1, 5)    ,						vocab_size=model.config.vocab_size   ).to(A   )
							lowercase_ : Optional[Any]							       =   TextIteratorStreamer(A    ,						timeout=0.001   )
							lowercase_ : List[Any]							       =   {'''input_ids''': input_ids, '''max_new_tokens''': 10, '''do_sample''': False, '''streamer''': streamer}
							lowercase_ : Optional[int]							       =   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   ):
									lowercase_ : Union[str, Any]							       =   ''''''
									for new_text in streamer:
											streamer_text += new_text











"""simple docstring"""



__A		:       Any  				=		{
    '''Pillow''': '''Pillow''',
    '''accelerate''': '''accelerate>=0.11.0''',
    '''compel''': '''compel==0.1.8''',
    '''black''': '''black~=23.1''',
    '''datasets''': '''datasets''',
    '''filelock''': '''filelock''',
    '''flax''': '''flax>=0.4.1''',
    '''hf-doc-builder''': '''hf-doc-builder>=0.3.0''',
    '''huggingface-hub''': '''huggingface-hub>=0.13.2''',
    '''requests-mock''': '''requests-mock==1.10.0''',
    '''importlib_metadata''': '''importlib_metadata''',
    '''invisible-watermark''': '''invisible-watermark''',
    '''isort''': '''isort>=5.5.4''',
    '''jax''': '''jax>=0.2.8,!=0.3.2''',
    '''jaxlib''': '''jaxlib>=0.1.65''',
    '''Jinja2''': '''Jinja2''',
    '''k-diffusion''': '''k-diffusion>=0.0.12''',
    '''torchsde''': '''torchsde''',
    '''note_seq''': '''note_seq''',
    '''librosa''': '''librosa''',
    '''numpy''': '''numpy''',
    '''omegaconf''': '''omegaconf''',
    '''parameterized''': '''parameterized''',
    '''protobuf''': '''protobuf>=3.20.3,<4''',
    '''pytest''': '''pytest''',
    '''pytest-timeout''': '''pytest-timeout''',
    '''pytest-xdist''': '''pytest-xdist''',
    '''ruff''': '''ruff>=0.0.241''',
    '''safetensors''': '''safetensors''',
    '''sentencepiece''': '''sentencepiece>=0.1.91,!=0.1.92''',
    '''scipy''': '''scipy''',
    '''onnx''': '''onnx''',
    '''regex''': '''regex!=2019.12.17''',
    '''requests''': '''requests''',
    '''tensorboard''': '''tensorboard''',
    '''torch''': '''torch>=1.4''',
    '''torchvision''': '''torchvision''',
    '''transformers''': '''transformers>=4.25.1''',
    '''urllib3''': '''urllib3<=2.0.0''',
}



