





import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaImgaImgPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
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

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()





class 			lowerCAmelCase__( __lowercase  ,     unittest.TestCase    ):






       '''simple docstring'''


       __snake_case 			=    KandinskyVaaImgaImgPipeline
       __snake_case 			=    ['image_embeds', 'negative_image_embeds', 'image']
       __snake_case 			=    [
           'image_embeds',
           'negative_image_embeds',
           'image',
       ]
       __snake_case 			=    [
           'generator',
           'height',
           'width',
           'strength',
           'guidance_scale',
           'num_inference_steps',
           'return_dict',
           'guidance_scale',
           'num_images_per_prompt',
           'output_type',
           'return_dict',
       ]
       __snake_case 			=    False




       @property
       def     UpperCamelCase_			(      self							)			->     Any:
            return 3_2




       @property
       def     UpperCamelCase_			(      self							)			->     Optional[Any]:
            return 3_2




       @property
       def     UpperCamelCase_			(      self							)			->     Dict:
            return self.time_input_dim




       @property
       def     UpperCamelCase_			(      self							)			->     Any:
            return self.time_input_dim * 4




       @property
       def     UpperCamelCase_			(      self							)			->     Tuple:
            return 1_0_0




       @property
       def     UpperCamelCase_			(      self							)			->     Dict:
            torch.manual_seed(0							)

            _SCREAMING_SNAKE_CASE       :	Tuple						    =							{
                "in_channels": 4,
                # Out channels is double in channels because predicts mean and variance
                "out_channels": 8,
                "addition_embed_type": "image",
                "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
                "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
                "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
                "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
                "layers_per_block": 1,
                "encoder_hid_dim": self.text_embedder_hidden_size,
                "encoder_hid_dim_type": "image_proj",
                "cross_attention_dim": self.cross_attention_dim,
                "attention_head_dim": 4,
                "resnet_time_scale_shift": "scale_shift",
                "class_embed_type": None,
            }

            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							UNetaDConditionModel(**__lowerCamelCase							)
            return model




       @property
       def     UpperCamelCase_			(      self							)			->     Union[str, Any]:
            return {
                "block_out_channels": [3_2, 6_4],
                "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
                "in_channels": 3,
                "latent_channels": 4,
                "layers_per_block": 1,
                "norm_num_groups": 8,
                "norm_type": "spatial",
                "num_vq_embeddings": 1_2,
                "out_channels": 3,
                "up_block_types": [
                    "AttnUpDecoderBlock2D",
                    "UpDecoderBlock2D",
                ],
                "vq_embed_dim": 4,
            }




       @property
       def     UpperCamelCase_			(      self							)			->     List[str]:
            torch.manual_seed(0							)
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							VQModel(**self.dummy_movq_kwargs							)
            return model




       def     UpperCamelCase_			(      self							)			->     str:
            _SCREAMING_SNAKE_CASE       :	int						    =							self.dummy_unet
            _SCREAMING_SNAKE_CASE       :	int						    =							self.dummy_movq

            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							{
                "num_train_timesteps": 1_0_0_0,
                "beta_schedule": "linear",
                "beta_start": 0.0_0085,
                "beta_end": 0.012,
                "clip_sample": False,
                "set_alpha_to_one": False,
                "steps_offset": 0,
                "prediction_type": "epsilon",
                "thresholding": False,
            }

            _SCREAMING_SNAKE_CASE       :	Dict						    =							DDIMScheduler(**__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE       :	int						    =							{
                "unet": unet,
                "scheduler": scheduler,
                "movq": movq,
            }

            return components




       def     UpperCamelCase_			(      self    ,						__lowerCamelCase    ,						__lowerCamelCase=0							)			->     int:
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							floats_tensor((1, self.text_embedder_hidden_size)    ,						rng=random.Random(__lowerCamelCase							)							).to(__lowerCamelCase							)
            _SCREAMING_SNAKE_CASE       :	int						    =							floats_tensor((1, self.text_embedder_hidden_size)    ,						rng=random.Random(seed + 1							)							).to(
                __lowerCamelCase							)
            # create init_image
            _SCREAMING_SNAKE_CASE       :	List[str]						    =							floats_tensor((1, 3, 6_4, 6_4)    ,						rng=random.Random(__lowerCamelCase							)							).to(__lowerCamelCase							)
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							image.cpu().permute(0    ,						2    ,						3    ,						1							)[0]
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							Image.fromarray(np.uinta(__lowerCamelCase							)							).convert("RGB"							).resize((2_5_6, 2_5_6)							)

            if str(__lowerCamelCase							).startswith("mps"							):
                 _SCREAMING_SNAKE_CASE       :	int						    =							torch.manual_seed(__lowerCamelCase							)
            else:
                 _SCREAMING_SNAKE_CASE       :	int						    =							torch.Generator(device=__lowerCamelCase							).manual_seed(__lowerCamelCase							)
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							{
                "image": init_image,
                "image_embeds": image_embeds,
                "negative_image_embeds": negative_image_embeds,
                "generator": generator,
                "height": 6_4,
                "width": 6_4,
                "num_inference_steps": 1_0,
                "guidance_scale": 7.0,
                "strength": 0.2,
                "output_type": "np",
            }
            return inputs







       def     UpperCamelCase_			(      self							)			->     Tuple:
            _SCREAMING_SNAKE_CASE       :	Dict						    =							"cpu"

            _SCREAMING_SNAKE_CASE       :	Tuple						    =							self.get_dummy_components()

            _SCREAMING_SNAKE_CASE       :	Dict						    =							self.pipeline_class(**__lowerCamelCase							)
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							pipe.to(__lowerCamelCase							)

            pipe.set_progress_bar_config(disable=__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE       :	List[str]						    =							pipe(**self.get_dummy_inputs(__lowerCamelCase							)							)
            _SCREAMING_SNAKE_CASE       :	Any						    =							output.images

            _SCREAMING_SNAKE_CASE       :	List[Any]						    =							pipe(
                **self.get_dummy_inputs(__lowerCamelCase							)    ,						return_dict=__lowerCamelCase    ,						)[0]

            _SCREAMING_SNAKE_CASE       :	int						    =							image[0, -3:, -3:, -1]
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							image_from_tuple[0, -3:, -3:, -1]

            assert image.shape == (1, 6_4, 6_4, 3)

            _SCREAMING_SNAKE_CASE       :	List[str]						    =							np.array(
                [0.619_9778, 0.6398_4406, 0.4614_5785, 0.6294_4984, 0.562_2215, 0.4730_6132, 0.4744_1456, 0.460_7606, 0.4871_9263]							)
            assert (
                np.abs(image_slice.flatten() - expected_slice							).max() < 1E-2
            ), F""" expected_slice {expected_slice}, but got {image_slice.flatten()}"""
            assert (
                np.abs(image_from_tuple_slice.flatten() - expected_slice							).max() < 1E-2
            ), F""" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"""





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






       '''simple docstring'''


       def     UpperCamelCase_			(      self							)			->     Optional[Any]:
            # clean up the VRAM after each test
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()







       def     UpperCamelCase_			(      self							)			->     Any:
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							load_numpy(
                "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
                "/kandinskyv22/kandinskyv22_img2img_frog.npy"							)

            _SCREAMING_SNAKE_CASE       :	Tuple						    =							load_image(
                "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"							)
            _SCREAMING_SNAKE_CASE       :	int						    =							"A red cartoon frog, 4k"

            _SCREAMING_SNAKE_CASE       :	Dict						    =							KandinskyVaaPriorPipeline.from_pretrained(
                "kandinsky-community/kandinsky-2-2-prior"    ,						torch_dtype=torch.floataa							)
            pipe_prior.to(__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE       :	List[str]						    =							KandinskyVaaImgaImgPipeline.from_pretrained(
                "kandinsky-community/kandinsky-2-2-decoder"    ,						torch_dtype=torch.floataa							)
            _SCREAMING_SNAKE_CASE       :	Tuple						    =							pipeline.to(__lowerCamelCase							)

            pipeline.set_progress_bar_config(disable=__lowerCamelCase							)

            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							torch.Generator(device="cpu"							).manual_seed(0							)
            _SCREAMING_SNAKE_CASE    ,			_SCREAMING_SNAKE_CASE       :	Any						    =							pipe_prior(
                __lowerCamelCase    ,						generator=__lowerCamelCase    ,						num_inference_steps=5    ,						negative_prompt=""    ,						).to_tuple()

            _SCREAMING_SNAKE_CASE       :	str						    =							pipeline(
                image=__lowerCamelCase    ,						image_embeds=__lowerCamelCase    ,						negative_image_embeds=__lowerCamelCase    ,						generator=__lowerCamelCase    ,						num_inference_steps=1_0_0    ,						height=7_6_8    ,						width=7_6_8    ,						strength=0.2    ,						output_type="np"    ,						)

            _SCREAMING_SNAKE_CASE       :	List[str]						    =							output.images[0]

            assert image.shape == (7_6_8, 7_6_8, 3)

            assert_mean_pixel_difference(__lowerCamelCase    ,						__lowerCamelCase							)






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=1_2    ,						__lowerCamelCase=7    ,						__lowerCamelCase=True    ,						__lowerCamelCase=True    ,						__lowerCamelCase=True    ,						__lowerCamelCase=9_9    ,						__lowerCamelCase=3_2    ,						__lowerCamelCase=3_2    ,						__lowerCamelCase=2    ,						__lowerCamelCase=4    ,						__lowerCamelCase=3_7    ,						__lowerCamelCase=0.1    ,						__lowerCamelCase=0.1    ,						__lowerCamelCase=5_1_2    ,						__lowerCamelCase=0.02    ,						__lowerCamelCase=0    ,						__lowerCamelCase=None    ,						)			->     Any:
            _SCREAMING_SNAKE_CASE       :	int						    =							parent
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							batch_size
            _SCREAMING_SNAKE_CASE       :	str						    =							seq_length
            _SCREAMING_SNAKE_CASE       :	str						    =							is_training
            _SCREAMING_SNAKE_CASE       :	str						    =							use_input_mask
            _SCREAMING_SNAKE_CASE       :	int						    =							use_labels
            _SCREAMING_SNAKE_CASE       :	List[Any]						    =							vocab_size
            _SCREAMING_SNAKE_CASE       :	Dict						    =							hidden_size
            _SCREAMING_SNAKE_CASE       :	int						    =							projection_dim
            _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							num_hidden_layers
            _SCREAMING_SNAKE_CASE       :	int						    =							num_attention_heads
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							intermediate_size
            _SCREAMING_SNAKE_CASE       :	int						    =							dropout
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							attention_dropout
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							max_position_embeddings
            _SCREAMING_SNAKE_CASE       :	Dict						    =							initializer_range
            _SCREAMING_SNAKE_CASE       :	Tuple						    =							scope
            _SCREAMING_SNAKE_CASE       :	Any						    =							bos_token_id




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

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

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

            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							self.get_config()

            return config, input_ids, tf.convert_to_tensor(__lowerCamelCase							)




       def     UpperCamelCase_			(      self							)			->     List[str]:
            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    ,						__lowerCamelCase    ,						__lowerCamelCase    ,						__lowerCamelCase							)			->     List[str]:
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							TFBlipTextModel(config=__lowerCamelCase							)
            _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							model(__lowerCamelCase    ,						attention_mask=__lowerCamelCase    ,						training=__lowerCamelCase							)
            _SCREAMING_SNAKE_CASE       :	Dict						    =							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     UpperCamelCase_			(      self							)			->     str:
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							self.prepare_config_and_inputs()
            _SCREAMING_SNAKE_CASE    ,			_SCREAMING_SNAKE_CASE    ,			_SCREAMING_SNAKE_CASE       :	int						    =							config_and_inputs
            _SCREAMING_SNAKE_CASE       :	List[str]						    =							{"input_ids": input_ids, "attention_mask": input_mask}
            return config, inputs_dict





@require_tf
class 			lowerCAmelCase__( __lowercase  ,     unittest.TestCase    ):






       '''simple docstring'''


       __snake_case 			=    (TFBlipTextModel,) if is_tf_available() else ()
       __snake_case 			=    False
       __snake_case 			=    False
       __snake_case 			=    False




       def     UpperCamelCase_			(      self							)			->     Any:
            _SCREAMING_SNAKE_CASE       :	Optional[Any]						    =							BlipTextModelTester(self							)
            _SCREAMING_SNAKE_CASE       :	Tuple						    =							ConfigTester(self    ,						config_class=__lowerCamelCase    ,						hidden_size=3_7							)




       def     UpperCamelCase_			(      self							)			->     Union[str, Any]:
            self.config_tester.run_common_tests()




       def     UpperCamelCase_			(      self							)			->     int:
            _SCREAMING_SNAKE_CASE       :	Optional[int]						    =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_model(*__lowerCamelCase							)




       def     UpperCamelCase_			(      self							)			->     int:
            pass




       def     UpperCamelCase_			(      self							)			->     Tuple:
            pass




       @unittest.skip(reason="Blip does not use inputs_embeds"							)
       def     UpperCamelCase_			(      self							)			->     Dict:
            pass




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




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




       @slow
       def     UpperCamelCase_			(      self							)			->     int:
            for model_name in TF_BLIP_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                 _SCREAMING_SNAKE_CASE       :	Union[str, Any]						    =							TFBlipTextModel.from_pretrained(__lowerCamelCase							)
                 self.assertIsNotNone(__lowerCamelCase							)







       def     UpperCamelCase_			(      self    ,						__lowerCamelCase=True							)			->     int:
            super().test_pt_tf_model_equivalence(allow_missing_keys=__lowerCamelCase							)