






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				):



	'''simple docstring'''




	lowerCAmelCase													= MODEL_FOR_CAUSAL_LM_MAPPING
	lowerCAmelCase													= TF_MODEL_FOR_CAUSAL_LM_MAPPING




	@require_torch
	def 							_UpperCamelCase			(       self					)       ->    Tuple:
								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=a					)
								self.assertEqual(
								    a       ,			[
								        {
								            '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(
								    a       ,			[
								        [
								            {
								                '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=a       ,			num_return_sequences=2       ,			return_tensors=a					)
								self.assertEqual(
								    a       ,			[
								        {'generated_token_ids': ANY(a					)},
								        {'generated_token_ids': ANY(a					)},
								    ]       ,			)
								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=a       ,			num_return_sequences=2       ,			batch_size=2       ,			return_tensors=a       ,			)
								self.assertEqual(
								    a       ,			[
								        [
								            {'generated_token_ids': ANY(a					)},
								            {'generated_token_ids': ANY(a					)},
								        ],
								        [
								            {'generated_token_ids': ANY(a					)},
								            {'generated_token_ids': ANY(a					)},
								        ],
								    ]       ,			)




	@require_tf
	def 							_UpperCamelCase			(       self					)       ->    str:
								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=a					)
								self.assertEqual(
								    a       ,			[
								        {
								            '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=a					)
								self.assertEqual(
								    a       ,			[
								        [
								            {
								                '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 							_UpperCamelCase			(       self       ,			a       ,			a       ,			a					)       ->    Any:
								snake_case_				=							TextGenerationPipeline(model=a       ,			tokenizer=a					)
								return text_generator, ["This is a test", "Another test"]




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

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




	def 							_UpperCamelCase			(       self       ,			a       ,			a					)       ->    Optional[int]:
								snake_case_				=							text_generator.model
								snake_case_				=							text_generator.tokenizer

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

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

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

								snake_case_				=							text_generator('This is a test'       ,			return_full_text=a					)
								self.assertEqual(a       ,			[{'generated_text': ANY(a					)}]					)
								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=a					)
								self.assertEqual(
								    a       ,			[
								        [{'generated_text': ANY(a					)}, {'generated_text': ANY(a					)}],
								        [{'generated_text': ANY(a					)}, {'generated_text': ANY(a					)}],
								    ]       ,			)

								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=a					)
															self.assertEqual(
															    a       ,			[
															        [{'generated_text': ANY(a					)}, {'generated_text': ANY(a					)}],
															        [{'generated_text': ANY(a					)}, {'generated_text': ANY(a					)}],
															    ]       ,			)

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

								# 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(a       ,			[{'generated_text': ANY(a					)}]					)
								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_00_00
								    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' * 5_00       ,			max_new_tokens=20					)

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




	@require_torch
	@require_accelerate
	@require_torch_gpu
	def 							_UpperCamelCase			(       self					)       ->    Union[str, Any]:
								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(
								    a       ,			[
								        {
								            '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(
								    a       ,			[
								        {
								            '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(
								    a       ,			[
								        {
								            '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 							_UpperCamelCase			(       self					)       ->    str:
								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 							_UpperCamelCase			(       self					)       ->    Union[str, Any]:
								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=a       ,			top_p=0.5					)






	def 							_UpperCamelCase			(       self					)       ->    List[Any]:
								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(a					) as cl:
															snake_case_				=							text_generator(a       ,			max_length=10       ,			max_new_tokens=1					)
								self.assertIn(a       ,			cl.out					)

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

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













from typing import List, Optional, Union

from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType




class  UpperCamelCase_      (			snake_case_				):



	'''simple docstring'''




	lowerCAmelCase													= ['''image_processor''', '''tokenizer''']
	lowerCAmelCase													= '''BlipImageProcessor'''
	lowerCAmelCase													= ('''BertTokenizer''', '''BertTokenizerFast''')




	def __init__(       self       ,			a       ,			a					)       ->    Tuple:
								snake_case_				=							False
								super().__init__(a       ,			a					)
								snake_case_				=							self.image_processor




	def __call__(       self       ,			a = None       ,			a = None       ,			a = True       ,			a = False       ,			a = None       ,			a = None       ,			a = 0       ,			a = None       ,			a = None       ,			a = False       ,			a = False       ,			a = False       ,			a = False       ,			a = False       ,			a = True       ,			a = None       ,			**a       ,			)       ->    BatchEncoding:
								if images is None and text is None:
															raise ValueError('You have to specify either images or text.'					)

								# Get only text
								if images is None:
															snake_case_				=							self.tokenizer
															snake_case_				=							self.tokenizer(
															    text=a       ,			add_special_tokens=a       ,			padding=a       ,			truncation=a       ,			max_length=a       ,			stride=a       ,			pad_to_multiple_of=a       ,			return_attention_mask=a       ,			return_overflowing_tokens=a       ,			return_special_tokens_mask=a       ,			return_offsets_mapping=a       ,			return_token_type_ids=a       ,			return_length=a       ,			verbose=a       ,			return_tensors=a       ,			**a       ,			)
															return text_encoding

								# add pixel_values
								snake_case_				=							self.image_processor(a       ,			return_tensors=a					)

								if text is not None:
															snake_case_				=							self.tokenizer(
															    text=a       ,			add_special_tokens=a       ,			padding=a       ,			truncation=a       ,			max_length=a       ,			stride=a       ,			pad_to_multiple_of=a       ,			return_attention_mask=a       ,			return_overflowing_tokens=a       ,			return_special_tokens_mask=a       ,			return_offsets_mapping=a       ,			return_token_type_ids=a       ,			return_length=a       ,			verbose=a       ,			return_tensors=a       ,			**a       ,			)
								else:
															snake_case_				=							None

								if text_encoding is not None:
															encoding_image_processor.update(a					)

								return encoding_image_processor




	def 							_UpperCamelCase			(       self       ,			*a       ,			**a					)       ->    int:
								return self.tokenizer.batch_decode(*a       ,			**a					)




	def 							_UpperCamelCase			(       self       ,			*a       ,			**a					)       ->    Any:
								return self.tokenizer.decode(*a       ,			**a					)






	@property
	def 							_UpperCamelCase			(       self					)       ->    List[str]:
								snake_case_				=							self.tokenizer.model_input_names
								snake_case_				=							self.image_processor.model_input_names
								return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names					)					)





