






import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetImgaImgPipeline,
    KandinskyVaaPriorEmbaEmbPipeline,
    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 							_UpperCamelCase			(a_							,		unittest.TestCase			):
      snake_case_  =							KandinskyVaaControlnetImgaImgPipeline
      snake_case_  =							["""image_embeds""", """negative_image_embeds""", """image""", """hint"""]
      snake_case_  =							["""image_embeds""", """negative_image_embeds""", """image""", """hint"""]
      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	)->	Tuple:
           return 3_2


      @property
      def  __UpperCAmelCase		(	self	)->	List[str]:
           return 3_2


      @property
      def  __UpperCAmelCase		(	self	)->	Optional[int]:
           return self.time_input_dim


      @property
      def  __UpperCAmelCase		(	self	)->	Optional[int]:
           return self.time_input_dim * 4


      @property
      def  __UpperCAmelCase		(	self	)->	Union[str, Any]:
           return 1_0_0


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

           __lowerCAmelCase								=					{
               "in_channels": 8,
               # Out channels is double in channels because predicts mean and variance
               "out_channels": 8,
               "addition_embed_type": "image_hint",
               "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,
           }

           __lowerCAmelCase								=					UNetaDConditionModel(**__UpperCamelCase	)
           return model


      @property
      def  __UpperCAmelCase		(	self	)->	List[Any]:
           return {
               "block_out_channels": [3_2, 3_2, 6_4, 6_4],
               "down_block_types": [
                   "DownEncoderBlock2D",
                   "DownEncoderBlock2D",
                   "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", "UpDecoderBlock2D", "UpDecoderBlock2D"],
               "vq_embed_dim": 4,
           }


      @property
      def  __UpperCAmelCase		(	self	)->	Optional[int]:
           torch.manual_seed(0	)
           __lowerCAmelCase								=					VQModel(**self.dummy_movq_kwargs	)
           return model


      def  __UpperCAmelCase		(	self	)->	Union[str, Any]:
           __lowerCAmelCase								=					self.dummy_unet
           __lowerCAmelCase								=					self.dummy_movq

           __lowerCAmelCase								=					{
               "num_train_timesteps": 1_0_0_0,
               "beta_schedule": "linear",
               "beta_start": 0.0_0_0_8_5,
               "beta_end": 0.0_1_2,
               "clip_sample": False,
               "set_alpha_to_one": False,
               "steps_offset": 0,
               "prediction_type": "epsilon",
               "thresholding": False,
           }

           __lowerCAmelCase								=					DDIMScheduler(**__UpperCamelCase	)

           __lowerCAmelCase								=					{
               "unet": unet,
               "scheduler": scheduler,
               "movq": movq,
           }

           return components


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

           if str(__UpperCamelCase	).startswith("mps"	):
                __lowerCAmelCase								=					torch.manual_seed(__UpperCamelCase	)
           else:
                __lowerCAmelCase								=					torch.Generator(device=__UpperCamelCase	).manual_seed(__UpperCamelCase	)
           __lowerCAmelCase								=					{
               "image": init_image,
               "image_embeds": image_embeds,
               "negative_image_embeds": negative_image_embeds,
               "hint": hint,
               "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	)->	List[Any]:
           __lowerCAmelCase								=					"cpu"

           __lowerCAmelCase								=					self.get_dummy_components()

           __lowerCAmelCase								=					self.pipeline_class(**__UpperCamelCase	)
           __lowerCAmelCase								=					pipe.to(__UpperCamelCase	)

           pipe.set_progress_bar_config(disable=__UpperCamelCase	)

           __lowerCAmelCase								=					pipe(**self.get_dummy_inputs(__UpperCamelCase	)	)
           __lowerCAmelCase								=					output.images

           __lowerCAmelCase								=					pipe(
               **self.get_dummy_inputs(__UpperCamelCase	)					,						return_dict=__UpperCamelCase					,						)[0]

           __lowerCAmelCase								=					image[0, -3:, -3:, -1]
           __lowerCAmelCase								=					image_from_tuple[0, -3:, -3:, -1]

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

           __lowerCAmelCase								=					np.array(
               [0.5_4_9_8_5_0_3_4, 0.5_5_5_0_9_3_6_5, 0.5_2_5_6_1_5_0_4, 0.5_5_7_0_4_9_4, 0.5_5_9_3_8_1_8, 0.5_2_6_3_9_7_9, 0.5_0_2_8_5_6_4_3, 0.5_0_6_9_8_4_6, 0.5_1_1_9_6_7_3_6]	)
           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 							_UpperCamelCase			(unittest.TestCase			):


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


      def  __UpperCAmelCase		(	self	)->	Dict:
           __lowerCAmelCase								=					load_numpy(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
               "/kandinskyv22/kandinskyv22_controlnet_img2img_robotcat_fp16.npy"	)

           __lowerCAmelCase								=					load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main" "/kandinsky/cat.png"	)
           __lowerCAmelCase								=					init_image.resize((5_1_2, 5_1_2)	)

           __lowerCAmelCase								=					load_image(
               "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
               "/kandinskyv22/hint_image_cat.png"	)
           __lowerCAmelCase								=					torch.from_numpy(np.array(__UpperCamelCase	)	).float() / 2_5_5.0
           __lowerCAmelCase								=					hint.permute(2					,						0					,						1	).unsqueeze(0	)

           __lowerCAmelCase								=					"A robot, 4k photo"

           __lowerCAmelCase								=					KandinskyVaaPriorEmbaEmbPipeline.from_pretrained(
               "kandinsky-community/kandinsky-2-2-prior"					,						torch_dtype=torch.floataa	)
           pipe_prior.to(__UpperCamelCase	)

           __lowerCAmelCase								=					KandinskyVaaControlnetImgaImgPipeline.from_pretrained(
               "kandinsky-community/kandinsky-2-2-controlnet-depth"					,						torch_dtype=torch.floataa	)
           __lowerCAmelCase								=					pipeline.to(__UpperCamelCase	)

           pipeline.set_progress_bar_config(disable=__UpperCamelCase	)

           __lowerCAmelCase								=					torch.Generator(device="cpu"	).manual_seed(0	)

           __lowerCAmelCase			, __lowerCAmelCase								=					pipe_prior(
               __UpperCamelCase					,						image=__UpperCamelCase					,						strength=0.8_5					,						generator=__UpperCamelCase					,						negative_prompt=""					,						).to_tuple()

           __lowerCAmelCase								=					pipeline(
               image=__UpperCamelCase					,						image_embeds=__UpperCamelCase					,						negative_image_embeds=__UpperCamelCase					,						hint=__UpperCamelCase					,						generator=__UpperCamelCase					,						num_inference_steps=1_0_0					,						height=5_1_2					,						width=5_1_2					,						strength=0.5					,						output_type="np"					,						)

           __lowerCAmelCase								=					output.images[0]

           assert image.shape == (5_1_2, 5_1_2, 3)

           assert_mean_pixel_difference(__UpperCamelCase					,						__UpperCamelCase	)














import os
import sys
import unittest


lowerCamelCase			:					int         =	os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, '''utils'''))

import check_dummies  # noqa: E402
from check_dummies import create_dummy_files, create_dummy_object, find_backend, read_init  # noqa: E402


# Align TRANSFORMERS_PATH in check_dummies with the current path
lowerCamelCase			:					Any         =	os.path.join(git_repo_path, '''src''', '''diffusers''')



class 							_UpperCamelCase			(unittest.TestCase			):


      def  __UpperCAmelCase		(	self	)->	Optional[int]:
           __lowerCAmelCase								=					find_backend("    if not is_torch_available():"	)
           self.assertEqual(__UpperCamelCase					,						"torch"	)

           # backend_with_underscore = find_backend("    if not is_tensorflow_text_available():")
           # self.assertEqual(backend_with_underscore, "tensorflow_text")

           __lowerCAmelCase								=					find_backend("    if not (is_torch_available() and is_transformers_available()):"	)
           self.assertEqual(__UpperCamelCase					,						"torch_and_transformers"	)

           # double_backend_with_underscore = find_backend(
           #    "    if not (is_sentencepiece_available() and is_tensorflow_text_available()):"
           # )
           # self.assertEqual(double_backend_with_underscore, "sentencepiece_and_tensorflow_text")

           __lowerCAmelCase								=					find_backend(
               "    if not (is_torch_available() and is_transformers_available() and is_onnx_available()):"	)
           self.assertEqual(__UpperCamelCase					,						"torch_and_transformers_and_onnx"	)


      def  __UpperCAmelCase		(	self	)->	Dict:
           __lowerCAmelCase								=					read_init()
           # We don't assert on the exact list of keys to allow for smooth grow of backend-specific objects
           self.assertIn("torch"					,						__UpperCamelCase	)
           self.assertIn("torch_and_transformers"					,						__UpperCamelCase	)
           self.assertIn("flax_and_transformers"					,						__UpperCamelCase	)
           self.assertIn("torch_and_transformers_and_onnx"					,						__UpperCamelCase	)

           # Likewise, we can't assert on the exact content of a key
           self.assertIn("UNet2DModel"					,						objects["torch"]	)
           self.assertIn("FlaxUNet2DConditionModel"					,						objects["flax"]	)
           self.assertIn("StableDiffusionPipeline"					,						objects["torch_and_transformers"]	)
           self.assertIn("FlaxStableDiffusionPipeline"					,						objects["flax_and_transformers"]	)
           self.assertIn("LMSDiscreteScheduler"					,						objects["torch_and_scipy"]	)
           self.assertIn("OnnxStableDiffusionPipeline"					,						objects["torch_and_transformers_and_onnx"]	)


      def  __UpperCAmelCase		(	self	)->	Any:
           __lowerCAmelCase								=					create_dummy_object("CONSTANT"					,						"'torch'"	)
           self.assertEqual(__UpperCamelCase					,						"\nCONSTANT = None\n"	)

           __lowerCAmelCase								=					create_dummy_object("function"					,						"'torch'"	)
           self.assertEqual(
               __UpperCamelCase					,						"\ndef function(*args, **kwargs):\n    requires_backends(function, 'torch')\n"	)

           __lowerCAmelCase								=					"\nclass FakeClass(metaclass=DummyObject):\n    _backends = 'torch'\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, 'torch')\n\n    @classmethod\n    def from_config(cls, *args, **kwargs):\n        requires_backends(cls, 'torch')\n\n    @classmethod\n    def from_pretrained(cls, *args, **kwargs):\n        requires_backends(cls, 'torch')\n"
           __lowerCAmelCase								=					create_dummy_object("FakeClass"					,						"'torch'"	)
           self.assertEqual(__UpperCamelCase					,						__UpperCamelCase	)


      def  __UpperCAmelCase		(	self	)->	Tuple:
           __lowerCAmelCase								=					"# This file is autogenerated by the command `make fix-copies`, do not edit.\nfrom ..utils import DummyObject, requires_backends\n\n\nCONSTANT = None\n\n\ndef function(*args, **kwargs):\n    requires_backends(function, [\"torch\"])\n\n\nclass FakeClass(metaclass=DummyObject):\n    _backends = [\"torch\"]\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, [\"torch\"])\n\n    @classmethod\n    def from_config(cls, *args, **kwargs):\n        requires_backends(cls, [\"torch\"])\n\n    @classmethod\n    def from_pretrained(cls, *args, **kwargs):\n        requires_backends(cls, [\"torch\"])\n"
           __lowerCAmelCase								=					create_dummy_files({"torch": ["CONSTANT", "function", "FakeClass"]}	)
           self.assertEqual(dummy_files["torch"]					,						__UpperCamelCase	)






