





import unittest

from transformers import (
    MODEL_FOR_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_CAUSAL_LM_MAPPING,
    TextGenerationPipeline,
    logging,
    pipeline,
)
from transformers.testing_utils import (
    CaptureLogger,
    is_pipeline_test,
    require_accelerate,
    require_tf,
    require_torch,
    require_torch_gpu,
    require_torch_or_tf,
)

from .test_pipelines_common import ANY





@is_pipeline_test
@require_torch_or_tf
class  A__    (							unittest.TestCase	):




   """simple docstring"""



   __A   :				List[str]         =  MODEL_FOR_CAUSAL_LM_MAPPING
   __A   :				Optional[Any]         =  TF_MODEL_FOR_CAUSAL_LM_MAPPING

   @require_torch
   def 							__lowercase		(					self)			->						List[Any]:







          '''simple docstring'''

          a__						:	Tuple =				pipeline(task='text-generation'					,  model='sshleifer/tiny-ctrl'					,  framework='pt')
          # Using `do_sample=False` to force deterministic output
          a__						:	List[Any] =				text_generator('This is a test'					,  do_sample=lowercase)
          self.assertEqual(
              lowercase					,  [
                  {
                      'generated_text': (
                          'This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope.'
                          ' oscope. FiliFili@@'
                      )
                  }
              ]					,  )

          a__						:	str =				text_generator(['This is a test', 'This is a second test'])
          self.assertEqual(
              lowercase					,  [
                  [
                      {
                          'generated_text': (
                              'This is a test ☃ ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy oscope.'
                              ' oscope. FiliFili@@'
                          )
                      }
                  ],
                  [
                      {
                          'generated_text': (
                              'This is a second test ☃ segmental segmental segmental 议议eski eski flutter flutter Lacy'
                              ' oscope. oscope. FiliFili@@'
                          )
                      }
                  ],
              ]					,  )

          a__						:	Optional[Any] =				text_generator('This is a test'					,  do_sample=lowercase					,  num_return_sequences=2					,  return_tensors=lowercase)
          self.assertEqual(
              lowercase					,  [
                  {'generated_token_ids': ANY(lowercase)},
                  {'generated_token_ids': ANY(lowercase)},
              ]					,  )
          a__						:	Union[str, Any] =				text_generator.model.config.eos_token_id
          a__						:	str =				'<pad>'
          a__						:	List[str] =				text_generator(
              ['This is a test', 'This is a second test']					,  do_sample=lowercase					,  num_return_sequences=2					,  batch_size=2					,  return_tensors=lowercase					,  )
          self.assertEqual(
              lowercase					,  [
                  [
                      {'generated_token_ids': ANY(lowercase)},
                      {'generated_token_ids': ANY(lowercase)},
                  ],
                  [
                      {'generated_token_ids': ANY(lowercase)},
                      {'generated_token_ids': ANY(lowercase)},
                  ],
              ]					,  )

   @require_tf
   def 							__lowercase		(					self)			->						Optional[int]:







          '''simple docstring'''

          a__						:	Optional[Any] =				pipeline(task='text-generation'					,  model='sshleifer/tiny-ctrl'					,  framework='tf')

          # Using `do_sample=False` to force deterministic output
          a__						:	int =				text_generator('This is a test'					,  do_sample=lowercase)
          self.assertEqual(
              lowercase					,  [
                  {
                      'generated_text': (
                          'This is a test FeyFeyFey(Croatis.), s.), Cannes Cannes Cannes 閲閲Cannes Cannes Cannes 攵'
                          ' please,'
                      )
                  }
              ]					,  )

          a__						:	List[str] =				text_generator(['This is a test', 'This is a second test']					,  do_sample=lowercase)
          self.assertEqual(
              lowercase					,  [
                  [
                      {
                          'generated_text': (
                              'This is a test FeyFeyFey(Croatis.), s.), Cannes Cannes Cannes 閲閲Cannes Cannes Cannes 攵'
                              ' please,'
                          )
                      }
                  ],
                  [
                      {
                          'generated_text': (
                              'This is a second test Chieftain Chieftain prefecture prefecture prefecture Cannes Cannes'
                              ' Cannes 閲閲Cannes Cannes Cannes 攵 please,'
                          )
                      }
                  ],
              ]					,  )

   def 							__lowercase		(					self					,  lowercase					,  lowercase					,  lowercase)			->						Tuple:







          '''simple docstring'''

          a__						:	Optional[Any] =				TextGenerationPipeline(model=lowercase					,  tokenizer=lowercase)
          return text_generator, ["This is a test", "Another test"]

   def 							__lowercase		(					self)			->						Any:







          '''simple docstring'''

          a__						:	Union[str, Any] =				'Hello I believe in'
          a__						:	Optional[int] =				pipeline('text-generation'					,  model='hf-internal-testing/tiny-random-gpt2')
          a__						:	Optional[Any] =				text_generator(lowercase)
          self.assertEqual(
              lowercase					,  [{'generated_text': 'Hello I believe in fe fe fe fe fe fe fe fe fe fe fe fe'}]					,  )

          a__						:	List[Any] =				text_generator(lowercase					,  stop_sequence=' fe')
          self.assertEqual(lowercase					,  [{'generated_text': 'Hello I believe in fe'}])

   def 							__lowercase		(					self					,  lowercase					,  lowercase)			->						int:







          '''simple docstring'''

          a__						:	Any =				text_generator.model
          a__						:	Tuple =				text_generator.tokenizer

          a__						:	int =				text_generator('This is a test')
          self.assertEqual(lowercase					,  [{'generated_text': ANY(lowercase)}])
          self.assertTrue(outputs[0]['generated_text'].startswith('This is a test'))

          a__						:	List[Any] =				text_generator('This is a test'					,  return_full_text=lowercase)
          self.assertEqual(lowercase					,  [{'generated_text': ANY(lowercase)}])
          self.assertNotIn('This is a test'					,  outputs[0]['generated_text'])

          a__						:	List[str] =				pipeline(task='text-generation'					,  model=lowercase					,  tokenizer=lowercase					,  return_full_text=lowercase)
          a__						:	List[Any] =				text_generator('This is a test')
          self.assertEqual(lowercase					,  [{'generated_text': ANY(lowercase)}])
          self.assertNotIn('This is a test'					,  outputs[0]['generated_text'])

          a__						:	Dict =				text_generator('This is a test'					,  return_full_text=lowercase)
          self.assertEqual(lowercase					,  [{'generated_text': ANY(lowercase)}])
          self.assertTrue(outputs[0]['generated_text'].startswith('This is a test'))

          a__						:	Union[str, Any] =				text_generator(['This is great !', 'Something else']					,  num_return_sequences=2					,  do_sample=lowercase)
          self.assertEqual(
              lowercase					,  [
                  [{'generated_text': ANY(lowercase)}, {'generated_text': ANY(lowercase)}],
                  [{'generated_text': ANY(lowercase)}, {'generated_text': ANY(lowercase)}],
              ]					,  )

          if text_generator.tokenizer.pad_token is not None:
                 a__						:	Optional[int] =				text_generator(
                     ['This is great !', 'Something else']					,  num_return_sequences=2					,  batch_size=2					,  do_sample=lowercase)
                 self.assertEqual(
                     lowercase					,  [
                         [{'generated_text': ANY(lowercase)}, {'generated_text': ANY(lowercase)}],
                         [{'generated_text': ANY(lowercase)}, {'generated_text': ANY(lowercase)}],
                     ]					,  )

          with self.assertRaises(lowercase):
                 a__						:	int =				text_generator('test'					,  return_full_text=lowercase					,  return_text=lowercase)
          with self.assertRaises(lowercase):
                 a__						:	Any =				text_generator('test'					,  return_full_text=lowercase					,  return_tensors=lowercase)
          with self.assertRaises(lowercase):
                 a__						:	Union[str, Any] =				text_generator('test'					,  return_text=lowercase					,  return_tensors=lowercase)

          # Empty prompt is slighly special
          # it requires BOS token to exist.
          # Special case for Pegasus which will always append EOS so will
          # work even without BOS.
          if (
              text_generator.tokenizer.bos_token_id is not None
              or "Pegasus" in tokenizer.__class__.__name__
              or "Git" in model.__class__.__name__
          ):
                 a__						:	Optional[int] =				text_generator('')
                 self.assertEqual(lowercase					,  [{'generated_text': ANY(lowercase)}])
          else:
                 with self.assertRaises((ValueError, AssertionError)):
                        a__						:	Optional[int] =				text_generator('')

          if text_generator.framework == "tf":
                 # TF generation does not support max_new_tokens, and it's impossible
                 # to control long generation with only max_length without
                 # fancy calculation, dismissing tests for now.
                 return
          # We don't care about infinite range models.
          # They already work.
          # Skip this test for XGLM, since it uses sinusoidal positional embeddings which are resized on-the-fly.
          a__						:	Optional[int] =				['RwkvForCausalLM', 'XGLMForCausalLM', 'GPTNeoXForCausalLM']
          if (
              tokenizer.model_max_length < 1_0000
              and text_generator.model.__class__.__name__ not in EXTRA_MODELS_CAN_HANDLE_LONG_INPUTS
          ):
                 # Handling of large generations
                 with self.assertRaises((RuntimeError, IndexError, ValueError, AssertionError)):
                        text_generator('This is a test' * 500					,  max_new_tokens=20)

                 a__						:	Tuple =				text_generator('This is a test' * 500					,  handle_long_generation='hole'					,  max_new_tokens=20)
                 # Hole strategy cannot work
                 with self.assertRaises(lowercase):
                        text_generator(
                            'This is a test' * 500					,  handle_long_generation='hole'					,  max_new_tokens=tokenizer.model_max_length + 10					,  )

   @require_torch
   @require_accelerate
   @require_torch_gpu
   def 							__lowercase		(					self)			->						Optional[int]:







          '''simple docstring'''

          import torch

          # Classic `model_kwargs`
          a__						:	Tuple =				pipeline(
              model='hf-internal-testing/tiny-random-bloom'					,  model_kwargs={'device_map': 'auto', 'torch_dtype': torch.bfloataa}					,  )
          self.assertEqual(pipe.model.device					,  torch.device(0))
          self.assertEqual(pipe.model.lm_head.weight.dtype					,  torch.bfloataa)
          a__						:	Any =				pipe('This is a test')
          self.assertEqual(
              lowercase					,  [
                  {
                      'generated_text': (
                          'This is a test test test test test test test test test test test test test test test test'
                          ' test'
                      )
                  }
              ]					,  )

          # Upgraded those two to real pipeline arguments (they just get sent for the model as they're unlikely to mean anything else.)
          a__						:	int =				pipeline(model='hf-internal-testing/tiny-random-bloom'					,  device_map='auto'					,  torch_dtype=torch.bfloataa)
          self.assertEqual(pipe.model.device					,  torch.device(0))
          self.assertEqual(pipe.model.lm_head.weight.dtype					,  torch.bfloataa)
          a__						:	str =				pipe('This is a test')
          self.assertEqual(
              lowercase					,  [
                  {
                      'generated_text': (
                          'This is a test test test test test test test test test test test test test test test test'
                          ' test'
                      )
                  }
              ]					,  )

          # torch_dtype will be automatically set to float32 if not provided - check: https://github.com/huggingface/transformers/pull/20602
          a__						:	List[Any] =				pipeline(model='hf-internal-testing/tiny-random-bloom'					,  device_map='auto')
          self.assertEqual(pipe.model.device					,  torch.device(0))
          self.assertEqual(pipe.model.lm_head.weight.dtype					,  torch.floataa)
          a__						:	Optional[int] =				pipe('This is a test')
          self.assertEqual(
              lowercase					,  [
                  {
                      'generated_text': (
                          'This is a test test test test test test test test test test test test test test test test'
                          ' test'
                      )
                  }
              ]					,  )

   @require_torch
   @require_torch_gpu
   def 							__lowercase		(					self)			->						Dict:







          '''simple docstring'''

          import torch

          a__						:	Optional[Any] =				pipeline(model='hf-internal-testing/tiny-random-bloom'					,  device=0					,  torch_dtype=torch.floataa)
          pipe('This is a test')

   @require_torch
   @require_accelerate
   @require_torch_gpu
   def 							__lowercase		(					self)			->						Optional[Any]:







          '''simple docstring'''

          import torch

          a__						:	int =				pipeline(model='hf-internal-testing/tiny-random-bloom'					,  device_map='auto'					,  torch_dtype=torch.floataa)
          pipe('This is a test'					,  do_sample=lowercase					,  top_p=0.5)

   def 							__lowercase		(					self)			->						Optional[int]:







          '''simple docstring'''

          a__						:	List[str] =				'Hello world'
          a__						:	Optional[int] =				pipeline('text-generation'					,  model='hf-internal-testing/tiny-random-gpt2')
          if text_generator.model.framework == "tf":
                 a__						:	Any =				logging.get_logger('transformers.generation.tf_utils')
          else:
                 a__						:	str =				logging.get_logger('transformers.generation.utils')
          a__						:	Tuple =				'Both `max_new_tokens`'  # The beggining of the message to be checked in this test

          # Both are set by the user -> log warning
          with CaptureLogger(lowercase) as cl:
                 a__						:	Dict =				text_generator(lowercase					,  max_length=10					,  max_new_tokens=1)
          self.assertIn(lowercase					,  cl.out)

          # The user only sets one -> no warning
          with CaptureLogger(lowercase) as cl:
                 a__						:	Dict =				text_generator(lowercase					,  max_new_tokens=1)
          self.assertNotIn(lowercase					,  cl.out)

          with CaptureLogger(lowercase) as cl:
                 a__						:	int =				text_generator(lowercase					,  max_length=10)
          self.assertNotIn(lowercase					,  cl.out)









from collections.abc import Iterable
from typing import Generic, TypeVar

lowercase      :       Any									=  TypeVar("""_T""")





class  A__    (							Generic[_T]	):




   """simple docstring"""



   def __init__(					self					,  lowercase = None)			->						None:







          '''simple docstring'''

          a__						:	list[_T] =				list(iterable or [])
          a__						:	list[_T] =				[]

   def __len__(					self)			->						int:







          '''simple docstring'''

          return len(self._stacka) + len(self._stacka)

   def __repr__(					self)			->						str:







          '''simple docstring'''

          return F'Queue({tuple(self._stacka[::-1] + self._stacka)})'

   def 							__lowercase		(					self					,  lowercase)			->						None:







          '''simple docstring'''

          self._stacka.append(lowercase)

   def 							__lowercase		(					self)			->						_T:







          '''simple docstring'''

          a__						:	List[str] =				self._stacka.pop
          a__						:	Optional[int] =				self._stacka.append

          if not self._stacka:
                 while self._stacka:
                        stacka_append(stacka_pop())

          if not self._stacka:
                 raise IndexError('Queue is empty')
          return self._stacka.pop()


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

    testmod()


