


'''simple docstring'''



import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, DDPMScheduler, StableDiffusionUpscalePipeline, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()


class 		_snake_case       (  unittest.TestCase						):




					def 					snake_case__    (   self):
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()




					@property
					def 					snake_case__    (   self):
								UpperCAmelCase__	:       int   =  1
								UpperCAmelCase__	:       Any   =  3
								UpperCAmelCase__	:       Any   =  (32, 32)

								UpperCAmelCase__	:       Optional[int]   =  floats_tensor((batch_size, num_channels) + sizes				,				rng=random.Random(0)).to(_lowerCamelCase)
								return image




					@property
					def 					snake_case__    (   self):
								torch.manual_seed(0)
								UpperCAmelCase__	:       Dict   =  UNetaDConditionModel(
								    block_out_channels=(32, 32, 64)				,				layers_per_block=2				,				sample_size=32				,				in_channels=7				,				out_channels=4				,				down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""", """CrossAttnDownBlock2D""")				,				up_block_types=("""CrossAttnUpBlock2D""", """CrossAttnUpBlock2D""", """UpBlock2D""")				,				cross_attention_dim=32				,				attention_head_dim=8				,				use_linear_projection=_lowerCamelCase				,				only_cross_attention=(True, True, False)				,				num_class_embeds=100				,				)
								return model




					@property
					def 					snake_case__    (   self):
								torch.manual_seed(0)
								UpperCAmelCase__	:       Optional[int]   =  AutoencoderKL(
								    block_out_channels=[32, 32, 64]				,				in_channels=3				,				out_channels=3				,				down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D""", """DownEncoderBlock2D"""]				,				up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D""", """UpDecoderBlock2D"""]				,				latent_channels=4				,				)
								return model




					@property
					def 					snake_case__    (   self):
								torch.manual_seed(0)
								UpperCAmelCase__	:       Union[str, Any]   =  CLIPTextConfig(
								    bos_token_id=0				,				eos_token_id=2				,				hidden_size=32				,				intermediate_size=37				,				layer_norm_eps=1e-0_5				,				num_attention_heads=4				,				num_hidden_layers=5				,				pad_token_id=1				,				vocab_size=1000				,				hidden_act="""gelu"""				,				projection_dim=512				,				)
								return CLIPTextModel(_lowerCamelCase)




					def 					snake_case__    (   self):
								UpperCAmelCase__	:       str   =  """cpu"""  # ensure determinism for the device-dependent torch.Generator
								UpperCAmelCase__	:       Optional[Any]   =  self.dummy_cond_unet_upscale
								UpperCAmelCase__	:       Any   =  DDPMScheduler()
								UpperCAmelCase__	:       List[str]   =  DDIMScheduler(prediction_type="""v_prediction""")
								UpperCAmelCase__	:       str   =  self.dummy_vae
								UpperCAmelCase__	:       List[Any]   =  self.dummy_text_encoder
								UpperCAmelCase__	:       Dict   =  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

								UpperCAmelCase__	:       List[Any]   =  self.dummy_image.cpu().permute(0				,				2				,				3				,				1)[0]
								UpperCAmelCase__	:       int   =  Image.fromarray(np.uinta(_lowerCamelCase)).convert("""RGB""").resize((64, 64))

								# make sure here that pndm scheduler skips prk
								UpperCAmelCase__	:       Any   =  StableDiffusionUpscalePipeline(
								    unet=_lowerCamelCase				,				low_res_scheduler=_lowerCamelCase				,				scheduler=_lowerCamelCase				,				vae=_lowerCamelCase				,				text_encoder=_lowerCamelCase				,				tokenizer=_lowerCamelCase				,				max_noise_level=350				,				)
								UpperCAmelCase__	:       Union[str, Any]   =  sd_pipe.to(_lowerCamelCase)
								sd_pipe.set_progress_bar_config(disable=_lowerCamelCase)

								UpperCAmelCase__	:       List[str]   =  """A painting of a squirrel eating a burger"""
								UpperCAmelCase__	:       Union[str, Any]   =  torch.Generator(device=_lowerCamelCase).manual_seed(0)
								UpperCAmelCase__	:       Tuple   =  sd_pipe(
								    [prompt]				,				image=_lowerCamelCase				,				generator=_lowerCamelCase				,				guidance_scale=6.0				,				noise_level=20				,				num_inference_steps=2				,				output_type="""np"""				,				)

								UpperCAmelCase__	:       Tuple   =  output.images

								UpperCAmelCase__	:       str   =  torch.Generator(device=_lowerCamelCase).manual_seed(0)
								UpperCAmelCase__	:       Any   =  sd_pipe(
								    [prompt]				,				image=_lowerCamelCase				,				generator=_lowerCamelCase				,				guidance_scale=6.0				,				noise_level=20				,				num_inference_steps=2				,				output_type="""np"""				,				return_dict=_lowerCamelCase				,				)[0]

								UpperCAmelCase__	:       Any   =  image[0, -3:, -3:, -1]
								UpperCAmelCase__	:       List[str]   =  image_from_tuple[0, -3:, -3:, -1]

								UpperCAmelCase__	:       str   =  low_res_image.size[0] * 4
								assert image.shape == (1, expected_height_width, expected_height_width, 3)
								UpperCAmelCase__	:       Union[str, Any]   =  np.array([0.3113, 0.3910, 0.4272, 0.4859, 0.5061, 0.4652, 0.5362, 0.5715, 0.5661])

								assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-2
								assert np.abs(image_from_tuple_slice.flatten() - expected_slice).max() < 1e-2




					def 					snake_case__    (   self):
								UpperCAmelCase__	:       Dict   =  """cpu"""  # ensure determinism for the device-dependent torch.Generator
								UpperCAmelCase__	:       Tuple   =  self.dummy_cond_unet_upscale
								UpperCAmelCase__	:       List[str]   =  DDPMScheduler()
								UpperCAmelCase__	:       List[str]   =  DDIMScheduler(prediction_type="""v_prediction""")
								UpperCAmelCase__	:       Optional[Any]   =  self.dummy_vae
								UpperCAmelCase__	:       Dict   =  self.dummy_text_encoder
								UpperCAmelCase__	:       Optional[Any]   =  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

								UpperCAmelCase__	:       Optional[Any]   =  self.dummy_image.cpu().permute(0				,				2				,				3				,				1)[0]
								UpperCAmelCase__	:       int   =  Image.fromarray(np.uinta(_lowerCamelCase)).convert("""RGB""").resize((64, 64))

								# make sure here that pndm scheduler skips prk
								UpperCAmelCase__	:       Dict   =  StableDiffusionUpscalePipeline(
								    unet=_lowerCamelCase				,				low_res_scheduler=_lowerCamelCase				,				scheduler=_lowerCamelCase				,				vae=_lowerCamelCase				,				text_encoder=_lowerCamelCase				,				tokenizer=_lowerCamelCase				,				max_noise_level=350				,				)
								UpperCAmelCase__	:       Optional[int]   =  sd_pipe.to(_lowerCamelCase)
								sd_pipe.set_progress_bar_config(disable=_lowerCamelCase)

								UpperCAmelCase__	:       str   =  """A painting of a squirrel eating a burger"""
								UpperCAmelCase__	:       str   =  sd_pipe(
								    2 * [prompt]				,				image=2 * [low_res_image]				,				guidance_scale=6.0				,				noise_level=20				,				num_inference_steps=2				,				output_type="""np"""				,				)
								UpperCAmelCase__	:       Optional[int]   =  output.images
								assert image.shape[0] == 2

								UpperCAmelCase__	:       Any   =  torch.Generator(device=_lowerCamelCase).manual_seed(0)
								UpperCAmelCase__	:       Union[str, Any]   =  sd_pipe(
								    [prompt]				,				image=_lowerCamelCase				,				generator=_lowerCamelCase				,				num_images_per_prompt=2				,				guidance_scale=6.0				,				noise_level=20				,				num_inference_steps=2				,				output_type="""np"""				,				)
								UpperCAmelCase__	:       str   =  output.images
								assert image.shape[0] == 2




					@unittest.skipIf(torch_device != """cuda"""				,				"""This test requires a GPU""")
					def 					snake_case__    (   self):
								UpperCAmelCase__	:       Any   =  self.dummy_cond_unet_upscale
								UpperCAmelCase__	:       List[Any]   =  DDPMScheduler()
								UpperCAmelCase__	:       Tuple   =  DDIMScheduler(prediction_type="""v_prediction""")
								UpperCAmelCase__	:       int   =  self.dummy_vae
								UpperCAmelCase__	:       Tuple   =  self.dummy_text_encoder
								UpperCAmelCase__	:       Union[str, Any]   =  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

								UpperCAmelCase__	:       List[Any]   =  self.dummy_image.cpu().permute(0				,				2				,				3				,				1)[0]
								UpperCAmelCase__	:       Tuple   =  Image.fromarray(np.uinta(_lowerCamelCase)).convert("""RGB""").resize((64, 64))

								# put models in fp16, except vae as it overflows in fp16
								UpperCAmelCase__	:       int   =  unet.half()
								UpperCAmelCase__	:       Any   =  text_encoder.half()

								# make sure here that pndm scheduler skips prk
								UpperCAmelCase__	:       Tuple   =  StableDiffusionUpscalePipeline(
								    unet=_lowerCamelCase				,				low_res_scheduler=_lowerCamelCase				,				scheduler=_lowerCamelCase				,				vae=_lowerCamelCase				,				text_encoder=_lowerCamelCase				,				tokenizer=_lowerCamelCase				,				max_noise_level=350				,				)
								UpperCAmelCase__	:       int   =  sd_pipe.to(_lowerCamelCase)
								sd_pipe.set_progress_bar_config(disable=_lowerCamelCase)

								UpperCAmelCase__	:       Dict   =  """A painting of a squirrel eating a burger"""
								UpperCAmelCase__	:       Optional[Any]   =  torch.manual_seed(0)
								UpperCAmelCase__	:       List[Any]   =  sd_pipe(
								    [prompt]				,				image=_lowerCamelCase				,				generator=_lowerCamelCase				,				num_inference_steps=2				,				output_type="""np"""				,				).images

								UpperCAmelCase__	:       Tuple   =  low_res_image.size[0] * 4
								assert image.shape == (1, expected_height_width, expected_height_width, 3)


@slow
@require_torch_gpu
class 		_snake_case       (  unittest.TestCase						):




					def 					snake_case__    (   self):
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()




					def 					snake_case__    (   self):
								UpperCAmelCase__	:       List[str]   =  load_image(
								    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
								    """/sd2-upscale/low_res_cat.png""")
								UpperCAmelCase__	:       Union[str, Any]   =  load_numpy(
								    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale"""
								    """/upsampled_cat.npy""")

								UpperCAmelCase__	:       str   =  """stabilityai/stable-diffusion-x4-upscaler"""
								UpperCAmelCase__	:       Tuple   =  StableDiffusionUpscalePipeline.from_pretrained(_lowerCamelCase)
								pipe.to(_lowerCamelCase)
								pipe.set_progress_bar_config(disable=_lowerCamelCase)
								pipe.enable_attention_slicing()

								UpperCAmelCase__	:       int   =  """a cat sitting on a park bench"""

								UpperCAmelCase__	:       Optional[int]   =  torch.manual_seed(0)
								UpperCAmelCase__	:       Any   =  pipe(
								    prompt=_lowerCamelCase				,				image=_lowerCamelCase				,				generator=_lowerCamelCase				,				output_type="""np"""				,				)
								UpperCAmelCase__	:       List[str]   =  output.images[0]

								assert image.shape == (512, 512, 3)
								assert np.abs(expected_image - image).max() < 1e-3




					def 					snake_case__    (   self):
								UpperCAmelCase__	:       Dict   =  load_image(
								    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
								    """/sd2-upscale/low_res_cat.png""")
								UpperCAmelCase__	:       str   =  load_numpy(
								    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-upscale"""
								    """/upsampled_cat_fp16.npy""")

								UpperCAmelCase__	:       Optional[Any]   =  """stabilityai/stable-diffusion-x4-upscaler"""
								UpperCAmelCase__	:       Tuple   =  StableDiffusionUpscalePipeline.from_pretrained(
								    _lowerCamelCase				,				torch_dtype=torch.floataa				,				)
								pipe.to(_lowerCamelCase)
								pipe.set_progress_bar_config(disable=_lowerCamelCase)
								pipe.enable_attention_slicing()

								UpperCAmelCase__	:       List[str]   =  """a cat sitting on a park bench"""

								UpperCAmelCase__	:       Any   =  torch.manual_seed(0)
								UpperCAmelCase__	:       Dict   =  pipe(
								    prompt=_lowerCamelCase				,				image=_lowerCamelCase				,				generator=_lowerCamelCase				,				output_type="""np"""				,				)
								UpperCAmelCase__	:       Optional[int]   =  output.images[0]

								assert image.shape == (512, 512, 3)
								assert np.abs(expected_image - image).max() < 5e-1




					def 					snake_case__    (   self):
								torch.cuda.empty_cache()
								torch.cuda.reset_max_memory_allocated()
								torch.cuda.reset_peak_memory_stats()

								UpperCAmelCase__	:       Any   =  load_image(
								    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
								    """/sd2-upscale/low_res_cat.png""")

								UpperCAmelCase__	:       Optional[Any]   =  """stabilityai/stable-diffusion-x4-upscaler"""
								UpperCAmelCase__	:       int   =  StableDiffusionUpscalePipeline.from_pretrained(
								    _lowerCamelCase				,				torch_dtype=torch.floataa				,				)
								pipe.to(_lowerCamelCase)
								pipe.set_progress_bar_config(disable=_lowerCamelCase)
								pipe.enable_attention_slicing(1)
								pipe.enable_sequential_cpu_offload()

								UpperCAmelCase__	:       int   =  """a cat sitting on a park bench"""

								UpperCAmelCase__	:       Any   =  torch.manual_seed(0)
								UpperCAmelCase__	:       Any   =  pipe(
								    prompt=_lowerCamelCase				,				image=_lowerCamelCase				,				generator=_lowerCamelCase				,				num_inference_steps=5				,				output_type="""np"""				,				)

								UpperCAmelCase__	:       int   =  torch.cuda.max_memory_allocated()
								# make sure that less than 2.9 GB is allocated
								assert mem_bytes < 2.9 * 10**9





'''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 	_snake_case:






 def __init__(self       :  Any    ,		a       :  str    ,		a       :  Union[str, Any]=12    ,		a       :  List[str]=7    ,		a       :  Dict=True    ,		a       :  Tuple=True    ,		a       :  Any=True    ,		a       :  Optional[Any]=99    ,		a       :  Optional[Any]=32    ,		a       :  Tuple=32    ,		a       :  List[Any]=2    ,		a       :  str=4    ,		a       :  Dict=37    ,		a       :  Optional[Any]=0.1    ,		a       :  List[Any]=0.1    ,		a       :  Dict=5_12    ,		a       :  List[Any]=0.02    ,		a       :  Any=0    ,		a       :  Optional[int]=None    ,		)       ->   List[Any]:






    """simple docstring"""


    A__       =     parent
    A__       =     batch_size
    A__       =     seq_length
    A__       =     is_training
    A__       =     use_input_mask
    A__       =     use_labels
    A__       =     vocab_size
    A__       =     hidden_size
    A__       =     projection_dim
    A__       =     num_hidden_layers
    A__       =     num_attention_heads
    A__       =     intermediate_size
    A__       =     dropout
    A__       =     attention_dropout
    A__       =     max_position_embeddings
    A__       =     initializer_range
    A__       =     scope
    A__       =     bos_token_id






 def 						_UpperCamelCase (self       :  Union[str, Any]    )       ->   List[Any]:






    """simple docstring"""


    A__       =     ids_tensor([self.batch_size, self.seq_length]    ,		self.vocab_size    )

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

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

    A__       =     self.get_config()

    return config, input_ids, tf.convert_to_tensor(a    )






 def 						_UpperCamelCase (self       :  Tuple    )       ->   Union[str, Any]:






    """simple docstring"""


    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 						_UpperCamelCase (self       :  Optional[int]    ,		a       :  Any    ,		a       :  List[Any]    ,		a       :  Union[str, Any]    )       ->   Any:






    """simple docstring"""


    A__       =     TFBlipTextModel(config=a    )
    A__       =     model(a    ,		attention_mask=a    ,		training=a    )
    A__       =     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 						_UpperCamelCase (self       :  str    )       ->   List[Any]:






    """simple docstring"""


    A__       =     self.prepare_config_and_inputs()
    A__  ,			A__  ,			A__       =     config_and_inputs
    A__       =     {'input_ids': input_ids, 'attention_mask': input_mask}
    return config, inputs_dict




@require_tf
class 	_snake_case(			UpperCAmelCase       , unittest.TestCase      ):
 __snake_case:							Optional[int]						=					(TFBlipTextModel,) if is_tf_available() else ()
 __snake_case:							Union[str, Any]						=					False
 __snake_case:							Any						=					False
 __snake_case:							Union[str, Any]						=					False






 def 						_UpperCamelCase (self       :  Optional[Any]    )       ->   List[Any]:






    """simple docstring"""


    A__       =     BlipTextModelTester(self    )
    A__       =     ConfigTester(self    ,		config_class=a    ,		hidden_size=37    )






 def 						_UpperCamelCase (self       :  Tuple    )       ->   str:






    """simple docstring"""


    self.config_tester.run_common_tests()






 def 						_UpperCamelCase (self       :  Any    )       ->   Dict:






    """simple docstring"""


    A__       =     self.model_tester.prepare_config_and_inputs()
    self.model_tester.create_and_check_model(*a    )






 def 						_UpperCamelCase (self       :  Dict    )       ->   Dict:






    """simple docstring"""


    pass






 def 						_UpperCamelCase (self       :  str    )       ->   int:






    """simple docstring"""


    pass






 @unittest.skip(reason='Blip does not use inputs_embeds'    )
 def 						_UpperCamelCase (self       :  Union[str, Any]    )       ->   str:






    """simple docstring"""


    pass






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






    """simple docstring"""


    pass






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






    """simple docstring"""


    pass






 @slow
 def 						_UpperCamelCase (self       :  int    )       ->   Optional[Any]:






    """simple docstring"""


    for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
       A__       =     TFBlipTextModel.from_pretrained(a    )
       self.assertIsNotNone(a    )






 def 						_UpperCamelCase (self       :  str    ,		a       :  Optional[int]=True    )       ->   List[Any]:






    """simple docstring"""


    super().test_pt_tf_model_equivalence(allow_missing_keys=a    )


