from __future__ import annotations

import unittest

import numpy as np

from transformers import BlipTextConfig
from transformers.testing_utils import require_tf, slow
from transformers.utils import is_tf_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask


if is_tf_available():
	import tensorflow as tf

	from transformers import TFBlipTextModel
	from transformers.models.blip.modeling_tf_blip import TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST

class 	UpperCamelCase    :



				def __init__(self       :   List[Any]			,						_A       :   List[Any]			,						_A       :   int=12			,						_A       :   List[str]=7			,						_A       :   List[Any]=True			,						_A       :   Union[str, Any]=True			,						_A       :   Union[str, Any]=True			,						_A       :   int=99			,						_A       :   Optional[Any]=32			,						_A       :   str=32			,						_A       :   List[Any]=2			,						_A       :   Any=4			,						_A       :   Tuple=37			,						_A       :   Dict=0.1			,						_A       :   Optional[Any]=0.1			,						_A       :   Union[str, Any]=5_12			,						_A       :   Any=0.02			,						_A       :   List[Any]=0			,						_A       :   Tuple=None			,						)					->			Dict:
					__snake_case    :							Optional[int]   	=				parent
					__snake_case    :							Any   	=				batch_size
					__snake_case    :							Any   	=				seq_length
					__snake_case    :							Union[str, Any]   	=				is_training
					__snake_case    :							Any   	=				use_input_mask
					__snake_case    :							Any   	=				use_labels
					__snake_case    :							int   	=				vocab_size
					__snake_case    :							Any   	=				hidden_size
					__snake_case    :							int   	=				projection_dim
					__snake_case    :							List[Any]   	=				num_hidden_layers
					__snake_case    :							str   	=				num_attention_heads
					__snake_case    :							Tuple   	=				intermediate_size
					__snake_case    :							Optional[int]   	=				dropout
					__snake_case    :							Tuple   	=				attention_dropout
					__snake_case    :							Optional[Any]   	=				max_position_embeddings
					__snake_case    :							Any   	=				initializer_range
					__snake_case    :							Any   	=				scope
					__snake_case    :							Any   	=				bos_token_id



				def       _lowercase  (self       :   Any)					->			Any:
					__snake_case    :							Optional[Any]   	=				ids_tensor([self.batch_size, self.seq_length]			,						self.vocab_size)

					__snake_case    :							List[str]   	=				None
					if self.use_input_mask:
						__snake_case    :							Any   	=				random_attention_mask([self.batch_size, self.seq_length])

					if input_mask is not None:
						__snake_case    :							str   	=				input_mask.numpy()
						__snake_case    :							Optional[Any]   	=				input_mask.shape
						__snake_case    :							List[str]   	=				np.random.randint(1			,						seq_length - 1			,						size=(batch_size,))
						for batch_idx, start_index in enumerate(_A):
							__snake_case    :							Dict   	=				1
							__snake_case    :							Dict   	=				0

					__snake_case    :							Tuple   	=				self.get_config()

					return config, input_ids, tf.convert_to_tensor(_A)



				def       _lowercase  (self       :   str)					->			Optional[int]:
					return BlipTextConfig(
					    vocab_size=self.vocab_size			,						hidden_size=self.hidden_size			,						projection_dim=self.projection_dim			,						num_hidden_layers=self.num_hidden_layers			,						num_attention_heads=self.num_attention_heads			,						intermediate_size=self.intermediate_size			,						dropout=self.dropout			,						attention_dropout=self.attention_dropout			,						max_position_embeddings=self.max_position_embeddings			,						initializer_range=self.initializer_range			,						bos_token_id=self.bos_token_id			,						)



				def       _lowercase  (self       :   Optional[Any]			,						_A       :   Tuple			,						_A       :   List[Any]			,						_A       :   Union[str, Any])					->			Optional[Any]:
					__snake_case    :							Optional[int]   	=				TFBlipTextModel(config=_A)
					__snake_case    :							Optional[int]   	=				model(_A			,						attention_mask=_A			,						training=_A)
					__snake_case    :							List[str]   	=				model(_A			,						training=_A)
					self.parent.assertEqual(result.last_hidden_state.shape			,						(self.batch_size, self.seq_length, self.hidden_size))
					self.parent.assertEqual(result.pooler_output.shape			,						(self.batch_size, self.hidden_size))






				def       _lowercase  (self       :   str)					->			Optional[int]:
					__snake_case    :							Union[str, Any]   	=				self.prepare_config_and_inputs()
					__snake_case    :							int   	=				config_and_inputs
					__snake_case    :							Any   	=				{'input_ids': input_ids, 'attention_mask': input_mask}
					return config, inputs_dict







@require_tf
class 	UpperCamelCase    (					lowercase   ,			unittest.TestCase		):
				UpperCAmelCase       :   Optional[Any]					=	(TFBlipTextModel,) if is_tf_available() else ()
				UpperCAmelCase       :   List[Any]					=	False
				UpperCAmelCase       :   Optional[int]					=	False
				UpperCAmelCase       :   Optional[int]					=	False



				def       _lowercase  (self       :   Optional[Any])					->			List[Any]:
					__snake_case    :							Any   	=				BlipTextModelTester(self)
					__snake_case    :							Optional[Any]   	=				ConfigTester(self			,						config_class=_A			,						hidden_size=37)



				def       _lowercase  (self       :   str)					->			Dict:
					self.config_tester.run_common_tests()



				def       _lowercase  (self       :   Tuple)					->			Any:
					__snake_case    :							Tuple   	=				self.model_tester.prepare_config_and_inputs()
					self.model_tester.create_and_check_model(*_A)



				def       _lowercase  (self       :   Optional[Any])					->			Union[str, Any]:
					pass



				def       _lowercase  (self       :   Union[str, Any])					->			Optional[int]:
					pass



				@unittest.skip(reason='Blip does not use inputs_embeds')
				def       _lowercase  (self       :   Dict)					->			List[Any]:
					pass



				@unittest.skip(reason='BlipTextModel has no base class and is not available in MODEL_MAPPING')
				def       _lowercase  (self       :   int)					->			List[Any]:
					pass



				@unittest.skip(reason='BlipTextModel has no base class and is not available in MODEL_MAPPING')
				def       _lowercase  (self       :   str)					->			int:
					pass



				@slow
				def       _lowercase  (self       :   Tuple)					->			Optional[Any]:
					for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
						__snake_case    :							Union[str, Any]   	=				TFBlipTextModel.from_pretrained(_A)
						self.assertIsNotNone(_A)






				def       _lowercase  (self       :   Any			,						_A       :   Tuple=True)					->			List[str]:
					super().test_pt_tf_model_equivalence(allow_missing_keys=_A)






"""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       _lowercase  (self       :   List[Any])					->			Dict:
					__snake_case    :							int   	=				AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2')
					__snake_case    :							Optional[Any]   	=				AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2').to(_A)
					__snake_case    :							Union[str, Any]   	=				-1

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

					with CaptureStdout() as cs:
						__snake_case    :							List[str]   	=				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
					__snake_case    :							Optional[Any]   	=				cs.out[:-1]

					self.assertEqual(_A			,						_A)



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

					__snake_case    :							str   	=				ids_tensor((1, 5)			,						vocab_size=model.config.vocab_size).to(_A)
					__snake_case    :							int   	=				model.generate(_A			,						max_new_tokens=10			,						do_sample=_A)
					__snake_case    :							int   	=				tokenizer.decode(greedy_ids[0])

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

					self.assertEqual(_A			,						_A)



				def       _lowercase  (self       :   Any)					->			List[str]:
					__snake_case    :							List[str]   	=				AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2')
					__snake_case    :							Any   	=				AutoModelForCausalLM.from_pretrained('hf-internal-testing/tiny-random-gpt2').to(_A)
					__snake_case    :							Union[str, Any]   	=				-1

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

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

					self.assertEqual(_A			,						_A)



				def       _lowercase  (self       :   List[str])					->			Dict:
					# 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
					__snake_case    :							int   	=				AutoTokenizer.from_pretrained('distilgpt2')
					__snake_case    :							Dict   	=				AutoModelForCausalLM.from_pretrained('distilgpt2').to(_A)
					__snake_case    :							Union[str, Any]   	=				-1

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






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

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





