

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 			__UpperCAmelCase    (	unittest.TestCase      ):
				__SCREAMING_SNAKE_CASE :			str	      =      MODEL_FOR_CAUSAL_LM_MAPPING
				__SCREAMING_SNAKE_CASE :			Union[str, Any]	      =      TF_MODEL_FOR_CAUSAL_LM_MAPPING




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

											snake_case_          =   text_generator(['This is a test', 'This is a second test']	)
											self.assertEqual(
											    _snake_case				,       [
											        [
											            {
											                '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@@'
											                )
											            }
											        ],
											    ]				,       )

											snake_case_          =   text_generator('This is a test'				,       do_sample=_snake_case				,       num_return_sequences=2				,       return_tensors=_snake_case	)
											self.assertEqual(
											    _snake_case				,       [
											        {'generated_token_ids': ANY(_snake_case	)},
											        {'generated_token_ids': ANY(_snake_case	)},
											    ]				,       )
											snake_case_          =   text_generator.model.config.eos_token_id
											snake_case_          =   "<pad>"
											snake_case_          =   text_generator(
											    ['This is a test', 'This is a second test']				,       do_sample=_snake_case				,       num_return_sequences=2				,       batch_size=2				,       return_tensors=_snake_case				,       )
											self.assertEqual(
											    _snake_case				,       [
											        [
											            {'generated_token_ids': ANY(_snake_case	)},
											            {'generated_token_ids': ANY(_snake_case	)},
											        ],
											        [
											            {'generated_token_ids': ANY(_snake_case	)},
											            {'generated_token_ids': ANY(_snake_case	)},
											        ],
											    ]				,       )




				@require_tf
				def    a							( self	):
											snake_case_          =   pipeline(task='text-generation'				,       model='sshleifer/tiny-ctrl'				,       framework='tf'	)

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

											snake_case_          =   text_generator(['This is a test', 'This is a second test']				,       do_sample=_snake_case	)
											self.assertEqual(
											    _snake_case				,       [
											        [
											            {
											                '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    a							( self				,       snake_case				,       snake_case				,       snake_case	):
											snake_case_          =   TextGenerationPipeline(model=_snake_case				,       tokenizer=_snake_case	)
											return text_generator, ["This is a test", "Another test"]




				def    a							( self	):
											snake_case_          =   "Hello I believe in"
											snake_case_          =   pipeline('text-generation'				,       model='hf-internal-testing/tiny-random-gpt2'	)
											snake_case_          =   text_generator(_snake_case	)
											self.assertEqual(
											    _snake_case				,       [{'generated_text': 'Hello I believe in fe fe fe fe fe fe fe fe fe fe fe fe'}]				,       )

											snake_case_          =   text_generator(_snake_case				,       stop_sequence=' fe'	)
											self.assertEqual(_snake_case				,       [{'generated_text': 'Hello I believe in fe'}]	)




				def    a							( self				,       snake_case				,       snake_case	):
											snake_case_          =   text_generator.model
											snake_case_          =   text_generator.tokenizer

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

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

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

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

											snake_case_          =   text_generator(['This is great !', 'Something else']				,       num_return_sequences=2				,       do_sample=_snake_case	)
											self.assertEqual(
											    _snake_case				,       [
											        [{'generated_text': ANY(_snake_case	)}, {'generated_text': ANY(_snake_case	)}],
											        [{'generated_text': ANY(_snake_case	)}, {'generated_text': ANY(_snake_case	)}],
											    ]				,       )

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

											with self.assertRaises(_snake_case	):
																		snake_case_          =   text_generator('test'				,       return_full_text=_snake_case				,       return_text=_snake_case	)
											with self.assertRaises(_snake_case	):
																		snake_case_          =   text_generator('test'				,       return_full_text=_snake_case				,       return_tensors=_snake_case	)
											with self.assertRaises(_snake_case	):
																		snake_case_          =   text_generator('test'				,       return_text=_snake_case				,       return_tensors=_snake_case	)

											# 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__
											):
																		snake_case_          =   text_generator(''	)
																		self.assertEqual(_snake_case				,       [{'generated_text': ANY(_snake_case	)}]	)
											else:
																		with self.assertRaises((ValueError, AssertionError)	):
																									snake_case_          =   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.
											snake_case_          =   ["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	)

																		snake_case_          =   text_generator('This is a test' * 500				,       handle_long_generation='hole'				,       max_new_tokens=20	)
																		# Hole strategy cannot work
																		with self.assertRaises(_snake_case	):
																									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    a							( self	):
											import torch

											# Classic `model_kwargs`
											snake_case_          =   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	)
											snake_case_          =   pipe('This is a test'	)
											self.assertEqual(
											    _snake_case				,       [
											        {
											            '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.)
											snake_case_          =   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	)
											snake_case_          =   pipe('This is a test'	)
											self.assertEqual(
											    _snake_case				,       [
											        {
											            '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
											snake_case_          =   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	)
											snake_case_          =   pipe('This is a test'	)
											self.assertEqual(
											    _snake_case				,       [
											        {
											            '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    a							( self	):
											import torch

											snake_case_          =   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    a							( self	):
											import torch

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




				def    a							( self	):
											snake_case_          =   "Hello world"
											snake_case_          =   pipeline('text-generation'				,       model='hf-internal-testing/tiny-random-gpt2'	)
											if text_generator.model.framework == "tf":
																		snake_case_          =   logging.get_logger('transformers.generation.tf_utils'	)
											else:
																		snake_case_          =   logging.get_logger('transformers.generation.utils'	)
											snake_case_          =   "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(_snake_case	) as cl:
																		snake_case_          =   text_generator(_snake_case				,       max_length=10				,       max_new_tokens=1	)
											self.assertIn(_snake_case				,       cl.out	)

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

											with CaptureLogger(_snake_case	) as cl:
																		snake_case_          =   text_generator(_snake_case				,       max_length=10	)
											self.assertNotIn(_snake_case				,       cl.out	)







from ...configuration_utils import PretrainedConfig
from ...utils import logging


_UpperCAmelCase		:       str						    =	logging.get_logger(__name__)

_UpperCAmelCase		:       int						    =	{
    """google/canine-s""": """https://huggingface.co/google/canine-s/resolve/main/config.json""",
    # See all CANINE models at https://huggingface.co/models?filter=canine
}







class 			lowercase    (	lowercase_      ):
				__SCREAMING_SNAKE_CASE :			int	      =      '''canine'''




				def __init__( self				,       snake_case=768				,       snake_case=12				,       snake_case=12				,       snake_case=3072				,       snake_case="gelu"				,       snake_case=0.1				,       snake_case=0.1				,       snake_case=1_6384				,       snake_case=16				,       snake_case=0.02				,       snake_case=1e-1_2				,       snake_case=0				,       snake_case=0xE000				,       snake_case=0xE001				,       snake_case=4				,       snake_case=4				,       snake_case=8				,       snake_case=1_6384				,       snake_case=128				,       **snake_case				,       ):
											super().__init__(pad_token_id=snake_case				,       bos_token_id=snake_case				,       eos_token_id=snake_case				,       **snake_case	)

											snake_case_          =   max_position_embeddings
											snake_case_          =   hidden_size
											snake_case_          =   num_hidden_layers
											snake_case_          =   num_attention_heads
											snake_case_          =   intermediate_size
											snake_case_          =   hidden_act
											snake_case_          =   hidden_dropout_prob
											snake_case_          =   attention_probs_dropout_prob
											snake_case_          =   initializer_range
											snake_case_          =   type_vocab_size
											snake_case_          =   layer_norm_eps

											# Character config:
											snake_case_          =   downsampling_rate
											snake_case_          =   upsampling_kernel_size
											snake_case_          =   num_hash_functions
											snake_case_          =   num_hash_buckets
											snake_case_          =   local_transformer_stride




