


"""simple docstring"""




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




		'''simple docstring'''







		def __init__(	self						,       lowerCAmelCase__						,       lowerCAmelCase__=12						,       lowerCAmelCase__=7						,       lowerCAmelCase__=True						,       lowerCAmelCase__=True						,       lowerCAmelCase__=True						,       lowerCAmelCase__=99						,       lowerCAmelCase__=32						,       lowerCAmelCase__=32						,       lowerCAmelCase__=2						,       lowerCAmelCase__=4						,       lowerCAmelCase__=37						,       lowerCAmelCase__=0.1						,       lowerCAmelCase__=0.1						,       lowerCAmelCase__=512						,       lowerCAmelCase__=0.02						,       lowerCAmelCase__=0						,       lowerCAmelCase__=None						,       )     ->							Any:
								SCREAMING_SNAKE_CASE													=					parent
								SCREAMING_SNAKE_CASE													=					batch_size
								SCREAMING_SNAKE_CASE													=					seq_length
								SCREAMING_SNAKE_CASE													=					is_training
								SCREAMING_SNAKE_CASE													=					use_input_mask
								SCREAMING_SNAKE_CASE													=					use_labels
								SCREAMING_SNAKE_CASE													=					vocab_size
								SCREAMING_SNAKE_CASE													=					hidden_size
								SCREAMING_SNAKE_CASE													=					projection_dim
								SCREAMING_SNAKE_CASE													=					num_hidden_layers
								SCREAMING_SNAKE_CASE													=					num_attention_heads
								SCREAMING_SNAKE_CASE													=					intermediate_size
								SCREAMING_SNAKE_CASE													=					dropout
								SCREAMING_SNAKE_CASE													=					attention_dropout
								SCREAMING_SNAKE_CASE													=					max_position_embeddings
								SCREAMING_SNAKE_CASE													=					initializer_range
								SCREAMING_SNAKE_CASE													=					scope
								SCREAMING_SNAKE_CASE													=					bos_token_id







		def 							__A						(	self				)     ->							Optional[int]:
								SCREAMING_SNAKE_CASE													=					ids_tensor([self.batch_size, self.seq_length]						,       self.vocab_size				)

								SCREAMING_SNAKE_CASE													=					None
								if self.use_input_mask:
														SCREAMING_SNAKE_CASE													=					random_attention_mask([self.batch_size, self.seq_length]				)

								if input_mask is not None:
														SCREAMING_SNAKE_CASE													=					input_mask.numpy()
														SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE													=					input_mask.shape
														SCREAMING_SNAKE_CASE													=					np.random.randint(1						,       seq_length - 1						,       size=(batch_size,)				)
														for batch_idx, start_index in enumerate(lowerCAmelCase__				):
																				SCREAMING_SNAKE_CASE													=					1
																				SCREAMING_SNAKE_CASE													=					0

								SCREAMING_SNAKE_CASE													=					self.get_config()

								return config, input_ids, tf.convert_to_tensor(lowerCAmelCase__				)







		def 							__A						(	self				)     ->							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 							__A						(	self						,       lowerCAmelCase__						,       lowerCAmelCase__						,       lowerCAmelCase__				)     ->							Optional[int]:
								SCREAMING_SNAKE_CASE													=					TFBlipTextModel(config=lowerCAmelCase__				)
								SCREAMING_SNAKE_CASE													=					model(lowerCAmelCase__						,       attention_mask=lowerCAmelCase__						,       training=lowerCAmelCase__				)
								SCREAMING_SNAKE_CASE													=					model(lowerCAmelCase__						,       training=lowerCAmelCase__				)
								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 							__A						(	self				)     ->							Tuple:
								SCREAMING_SNAKE_CASE													=					self.prepare_config_and_inputs()
								SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE							,				SCREAMING_SNAKE_CASE													=					config_and_inputs
								SCREAMING_SNAKE_CASE													=					{'input_ids': input_ids, 'attention_mask': input_mask}
								return config, inputs_dict





@require_tf
class 					lowerCAmelCase	(					_UpperCAmelCase	,    unittest.TestCase    ):




		'''simple docstring'''







		SCREAMING_SNAKE_CASE_						:     Optional[int]								=     (TFBlipTextModel,) if is_tf_available() else ()
		SCREAMING_SNAKE_CASE_						:     List[str]								=     False
		SCREAMING_SNAKE_CASE_						:     Union[str, Any]								=     False
		SCREAMING_SNAKE_CASE_						:     Union[str, Any]								=     False







		def 							__A						(	self				)     ->							List[Any]:
								SCREAMING_SNAKE_CASE													=					BlipTextModelTester(self				)
								SCREAMING_SNAKE_CASE													=					ConfigTester(self						,       config_class=lowerCAmelCase__						,       hidden_size=37				)







		def 							__A						(	self				)     ->							Any:
								self.config_tester.run_common_tests()







		def 							__A						(	self				)     ->							str:
								SCREAMING_SNAKE_CASE													=					self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*lowerCAmelCase__				)







		def 							__A						(	self				)     ->							List[str]:
								pass







		def 							__A						(	self				)     ->							Tuple:
								pass







		@unittest.skip(reason='Blip does not use inputs_embeds'				)
		def 							__A						(	self				)     ->							str:
								pass







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







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







		@slow
		def 							__A						(	self				)     ->							List[Any]:
								for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														SCREAMING_SNAKE_CASE													=					TFBlipTextModel.from_pretrained(lowerCAmelCase__				)
														self.assertIsNotNone(lowerCAmelCase__				)







		def 							__A						(	self						,       lowerCAmelCase__=True				)     ->							List[Any]:
								super().test_pt_tf_model_equivalence(allow_missing_keys=lowerCAmelCase__				)









"""simple docstring"""





import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np

class 		lowercase					:





      def     _snake_case			(       self )		->     List[Any]:
            torch.manual_seed(0 )
            lowerCAmelCase  						=       TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5""" )

            torch.manual_seed(0 )
            lowerCAmelCase  						=       AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5""" )

            torch.manual_seed(0 )
            lowerCAmelCase  						=       UNetaDConditionModel(
                sample_size=32      ,						layers_per_block=1      ,						block_out_channels=[32, 64]      ,						down_block_types=[
                    """ResnetDownsampleBlock2D""",
                    """SimpleCrossAttnDownBlock2D""",
                ]      ,						mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""      ,						up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]      ,						in_channels=3      ,						out_channels=6      ,						cross_attention_dim=32      ,						encoder_hid_dim=32      ,						attention_head_dim=8      ,						addition_embed_type="""text"""      ,						addition_embed_type_num_heads=2      ,						cross_attention_norm="""group_norm"""      ,						resnet_time_scale_shift="""scale_shift"""      ,						act_fn="""gelu"""      ,						)
            unet.set_attn_processor(AttnAddedKVProcessor() )  # For reproducibility tests

            torch.manual_seed(0 )
            lowerCAmelCase  						=       DDPMScheduler(
                num_train_timesteps=1_000      ,						beta_schedule="""squaredcos_cap_v2"""      ,						beta_start=0.0_001      ,						beta_end=0.02      ,						thresholding=lowercase      ,						dynamic_thresholding_ratio=0.95      ,						sample_max_value=1.0      ,						prediction_type="""epsilon"""      ,						variance_type="""learned_range"""      ,						)

            torch.manual_seed(0 )
            lowerCAmelCase  						=       IFWatermarker()

            return {
                "text_encoder": text_encoder,
                "tokenizer": tokenizer,
                "unet": unet,
                "scheduler": scheduler,
                "watermarker": watermarker,
                "safety_checker": None,
                "feature_extractor": None,
            }





      def     _snake_case			(       self )		->     int:
            torch.manual_seed(0 )
            lowerCAmelCase  						=       TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5""" )

            torch.manual_seed(0 )
            lowerCAmelCase  						=       AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5""" )

            torch.manual_seed(0 )
            lowerCAmelCase  						=       UNetaDConditionModel(
                sample_size=32      ,						layers_per_block=[1, 2]      ,						block_out_channels=[32, 64]      ,						down_block_types=[
                    """ResnetDownsampleBlock2D""",
                    """SimpleCrossAttnDownBlock2D""",
                ]      ,						mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""      ,						up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]      ,						in_channels=6      ,						out_channels=6      ,						cross_attention_dim=32      ,						encoder_hid_dim=32      ,						attention_head_dim=8      ,						addition_embed_type="""text"""      ,						addition_embed_type_num_heads=2      ,						cross_attention_norm="""group_norm"""      ,						resnet_time_scale_shift="""scale_shift"""      ,						act_fn="""gelu"""      ,						class_embed_type="""timestep"""      ,						mid_block_scale_factor=1.414      ,						time_embedding_act_fn="""gelu"""      ,						time_embedding_dim=32      ,						)
            unet.set_attn_processor(AttnAddedKVProcessor() )  # For reproducibility tests

            torch.manual_seed(0 )
            lowerCAmelCase  						=       DDPMScheduler(
                num_train_timesteps=1_000      ,						beta_schedule="""squaredcos_cap_v2"""      ,						beta_start=0.0_001      ,						beta_end=0.02      ,						thresholding=lowercase      ,						dynamic_thresholding_ratio=0.95      ,						sample_max_value=1.0      ,						prediction_type="""epsilon"""      ,						variance_type="""learned_range"""      ,						)

            torch.manual_seed(0 )
            lowerCAmelCase  						=       DDPMScheduler(
                num_train_timesteps=1_000      ,						beta_schedule="""squaredcos_cap_v2"""      ,						beta_start=0.0_001      ,						beta_end=0.02      ,						)

            torch.manual_seed(0 )
            lowerCAmelCase  						=       IFWatermarker()

            return {
                "text_encoder": text_encoder,
                "tokenizer": tokenizer,
                "unet": unet,
                "scheduler": scheduler,
                "image_noising_scheduler": image_noising_scheduler,
                "watermarker": watermarker,
                "safety_checker": None,
                "feature_extractor": None,
            }





      def     _snake_case			(       self )		->     Union[str, Any]:
            lowerCAmelCase  						=       self.get_dummy_components()
            lowerCAmelCase  						=       self.pipeline_class(**lowercase )
            pipe.to(lowercase )
            pipe.set_progress_bar_config(disable=lowercase )

            lowerCAmelCase  						=       self.get_dummy_inputs(lowercase )

            lowerCAmelCase  						=       inputs["""prompt"""]
            lowerCAmelCase  						=       inputs["""generator"""]
            lowerCAmelCase  						=       inputs["""num_inference_steps"""]
            lowerCAmelCase  						=       inputs["""output_type"""]

            if "image" in inputs:
                  lowerCAmelCase  						=       inputs["""image"""]
            else:
                  lowerCAmelCase  						=       None

            if "mask_image" in inputs:
                  lowerCAmelCase  						=       inputs["""mask_image"""]
            else:
                  lowerCAmelCase  						=       None

            if "original_image" in inputs:
                  lowerCAmelCase  						=       inputs["""original_image"""]
            else:
                  lowerCAmelCase  						=       None

            lowerCAmelCase						,		lowerCAmelCase  						=       pipe.encode_prompt(lowercase )

            # inputs with prompt converted to embeddings
            lowerCAmelCase  						=       {
                """prompt_embeds""": prompt_embeds,
                """negative_prompt_embeds""": negative_prompt_embeds,
                """generator""": generator,
                """num_inference_steps""": num_inference_steps,
                """output_type""": output_type,
            }

            if image is not None:
                  lowerCAmelCase  						=       image

            if mask_image is not None:
                  lowerCAmelCase  						=       mask_image

            if original_image is not None:
                  lowerCAmelCase  						=       original_image

            # set all optional components to None
            for optional_component in pipe._optional_components:
                  setattr(lowercase      ,						lowercase      ,						lowercase )

            lowerCAmelCase  						=       pipe(**lowercase )[0]

            with tempfile.TemporaryDirectory() as tmpdir:
                  pipe.save_pretrained(lowercase )
                  lowerCAmelCase  						=       self.pipeline_class.from_pretrained(lowercase )
                  pipe_loaded.to(lowercase )
                  pipe_loaded.set_progress_bar_config(disable=lowercase )

            pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor() )  # For reproducibility tests

            for optional_component in pipe._optional_components:
                  self.assertTrue(
                      getattr(lowercase      ,						lowercase ) is None      ,						f'`{optional_component}` did not stay set to None after loading.'      ,						)

            lowerCAmelCase  						=       self.get_dummy_inputs(lowercase )

            lowerCAmelCase  						=       inputs["""generator"""]
            lowerCAmelCase  						=       inputs["""num_inference_steps"""]
            lowerCAmelCase  						=       inputs["""output_type"""]

            # inputs with prompt converted to embeddings
            lowerCAmelCase  						=       {
                """prompt_embeds""": prompt_embeds,
                """negative_prompt_embeds""": negative_prompt_embeds,
                """generator""": generator,
                """num_inference_steps""": num_inference_steps,
                """output_type""": output_type,
            }

            if image is not None:
                  lowerCAmelCase  						=       image

            if mask_image is not None:
                  lowerCAmelCase  						=       mask_image

            if original_image is not None:
                  lowerCAmelCase  						=       original_image

            lowerCAmelCase  						=       pipe_loaded(**lowercase )[0]

            lowerCAmelCase  						=       np.abs(to_np(lowercase ) - to_np(lowercase ) ).max()
            self.assertLess(lowercase      ,						1e-4 )





      def     _snake_case			(       self )		->     Optional[int]:
            lowerCAmelCase  						=       self.get_dummy_components()
            lowerCAmelCase  						=       self.pipeline_class(**lowercase )
            pipe.to(lowercase )
            pipe.set_progress_bar_config(disable=lowercase )

            lowerCAmelCase  						=       self.get_dummy_inputs(lowercase )
            lowerCAmelCase  						=       pipe(**lowercase )[0]

            with tempfile.TemporaryDirectory() as tmpdir:
                  pipe.save_pretrained(lowercase )
                  lowerCAmelCase  						=       self.pipeline_class.from_pretrained(lowercase )
                  pipe_loaded.to(lowercase )
                  pipe_loaded.set_progress_bar_config(disable=lowercase )

            pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor() )  # For reproducibility tests

            lowerCAmelCase  						=       self.get_dummy_inputs(lowercase )
            lowerCAmelCase  						=       pipe_loaded(**lowercase )[0]

            lowerCAmelCase  						=       np.abs(to_np(lowercase ) - to_np(lowercase ) ).max()
            self.assertLess(lowercase      ,						1e-4 )






