



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   snake_case_ (lowercase__ , unittest.TestCase  ):




       """simple docstring"""






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



       @property
       def  A_   (     self):

             """simple docstring"""





             return 32



       @property
       def  A_   (     self):

             """simple docstring"""





             return 32



       @property
       def  A_   (     self):

             """simple docstring"""





             return self.time_input_dim



       @property
       def  A_   (     self):

             """simple docstring"""





             return self.time_input_dim * 4



       @property
       def  A_   (     self):

             """simple docstring"""





             return 100



       @property
       def  A_   (     self):

             """simple docstring"""





             torch.manual_seed(0)

             UpperCAmelCase_  :	Union[str, Any]				      =		{
                 "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,
             }

             UpperCAmelCase_  :	int				      =		UNetaDConditionModel(**lowercase)
             return model



       @property
       def  A_   (     self):

             """simple docstring"""





             return {
                 "block_out_channels": [32, 64],
                 "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": 12,
                 "out_channels": 3,
                 "up_block_types": [
                     "AttnUpDecoderBlock2D",
                     "UpDecoderBlock2D",
                 ],
                 "vq_embed_dim": 4,
             }



       @property
       def  A_   (     self):

             """simple docstring"""





             torch.manual_seed(0)
             UpperCAmelCase_  :	Tuple				      =		VQModel(**self.dummy_movq_kwargs)
             return model



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	List[str]				      =		self.dummy_unet
             UpperCAmelCase_  :	Any				      =		self.dummy_movq

             UpperCAmelCase_  :	str				      =		{
                 "num_train_timesteps": 1000,
                 "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,
             }

             UpperCAmelCase_  :	List[Any]				      =		DDIMScheduler(**lowercase)

             UpperCAmelCase_  :	Union[str, Any]				      =		{
                 "unet": unet,
                 "scheduler": scheduler,
                 "movq": movq,
             }

             return components



       def  A_   (     self     ,lowercase     ,lowercase=0):

             """simple docstring"""





             UpperCAmelCase_  :	int				      =		floats_tensor((1, self.text_embedder_hidden_size)     ,rng=random.Random(lowercase)).to(lowercase)
             UpperCAmelCase_  :	Dict				      =		floats_tensor((1, self.text_embedder_hidden_size)     ,rng=random.Random(seed + 1)).to(
                 lowercase)
             # create init_image
             UpperCAmelCase_  :	Optional[Any]				      =		floats_tensor((1, 3, 64, 64)     ,rng=random.Random(lowercase)).to(lowercase)
             UpperCAmelCase_  :	List[str]				      =		image.cpu().permute(0     ,2     ,3     ,1)[0]
             UpperCAmelCase_  :	List[Any]				      =		Image.fromarray(np.uinta(lowercase)).convert("RGB").resize((256, 256))

             if str(lowercase).startswith("mps"):
                   UpperCAmelCase_  :	Optional[int]				      =		torch.manual_seed(lowercase)
             else:
                   UpperCAmelCase_  :	List[Any]				      =		torch.Generator(device=lowercase).manual_seed(lowercase)
             UpperCAmelCase_  :	List[str]				      =		{
                 "image": init_image,
                 "image_embeds": image_embeds,
                 "negative_image_embeds": negative_image_embeds,
                 "generator": generator,
                 "height": 64,
                 "width": 64,
                 "num_inference_steps": 10,
                 "guidance_scale": 7.0,
                 "strength": 0.2,
                 "output_type": "np",
             }
             return inputs



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	Dict				      =		"cpu"

             UpperCAmelCase_  :	Tuple				      =		self.get_dummy_components()

             UpperCAmelCase_  :	Any				      =		self.pipeline_class(**lowercase)
             UpperCAmelCase_  :	Optional[int]				      =		pipe.to(lowercase)

             pipe.set_progress_bar_config(disable=lowercase)

             UpperCAmelCase_  :	Union[str, Any]				      =		pipe(**self.get_dummy_inputs(lowercase))
             UpperCAmelCase_  :	Union[str, Any]				      =		output.images

             UpperCAmelCase_  :	Any				      =		pipe(
                 **self.get_dummy_inputs(lowercase)     ,return_dict=lowercase     ,)[0]

             UpperCAmelCase_  :	List[Any]				      =		image[0, -3:, -3:, -1]
             UpperCAmelCase_  :	Optional[int]				      =		image_from_tuple[0, -3:, -3:, -1]

             assert image.shape == (1, 64, 64, 3)

             UpperCAmelCase_  :	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   snake_case_ (unittest.TestCase  ):




       """simple docstring"""






       def  A_   (     self):

             """simple docstring"""





             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	List[str]				      =		load_numpy(
                 "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
                 "/kandinskyv22/kandinskyv22_img2img_frog.npy")

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

             UpperCAmelCase_  :	Optional[int]				      =		KandinskyVaaPriorPipeline.from_pretrained(
                 "kandinsky-community/kandinsky-2-2-prior"     ,torch_dtype=torch.floataa)
             pipe_prior.to(lowercase)

             UpperCAmelCase_  :	int				      =		KandinskyVaaImgaImgPipeline.from_pretrained(
                 "kandinsky-community/kandinsky-2-2-decoder"     ,torch_dtype=torch.floataa)
             UpperCAmelCase_  :	Dict				      =		pipeline.to(lowercase)

             pipeline.set_progress_bar_config(disable=lowercase)

             UpperCAmelCase_  :	Optional[Any]				      =		torch.Generator(device="cpu").manual_seed(0)
             UpperCAmelCase_  :	Any				      =		pipe_prior(
                 lowercase     ,generator=lowercase     ,num_inference_steps=5     ,negative_prompt=""     ,).to_tuple()

             UpperCAmelCase_  :	Optional[Any]				      =		pipeline(
                 image=lowercase     ,image_embeds=lowercase     ,negative_image_embeds=lowercase     ,generator=lowercase     ,num_inference_steps=100     ,height=768     ,width=768     ,strength=0.2     ,output_type="np"     ,)

             UpperCAmelCase_  :	str				      =		output.images[0]

             assert image.shape == (768, 768, 3)

             assert_mean_pixel_difference(lowercase     ,lowercase)





from abc import ABC, abstractmethod
from typing import Optional, Union

from .. import Dataset, DatasetDict, Features, IterableDataset, IterableDatasetDict, NamedSplit
from ..utils.typing import NestedDataStructureLike, PathLike


class   snake_case_ (lowercase__  ):




       """simple docstring"""






       def __init__(     self     ,lowercase = None     ,lowercase = None     ,lowercase = None     ,lowercase = None     ,lowercase = False     ,lowercase = False     ,lowercase = None     ,**lowercase     ,):

             """simple docstring"""





             UpperCAmelCase_  :	Optional[Any]				      =		path_or_paths
             UpperCAmelCase_  :	Optional[int]				      =		split if split or isinstance(lowercase     ,lowercase) else "train"
             UpperCAmelCase_  :	Optional[int]				      =		features
             UpperCAmelCase_  :	Dict				      =		cache_dir
             UpperCAmelCase_  :	int				      =		keep_in_memory
             UpperCAmelCase_  :	Tuple				      =		streaming
             UpperCAmelCase_  :	Union[str, Any]				      =		num_proc
             UpperCAmelCase_  :	Union[str, Any]				      =		kwargs



       @abstractmethod
       def  A_   (     self):

             """simple docstring"""





             pass


class   snake_case_ (lowercase__  ):




       """simple docstring"""






       def __init__(     self     ,lowercase = None     ,lowercase = None     ,lowercase = False     ,lowercase = False     ,lowercase = None     ,**lowercase     ,):

             """simple docstring"""





             UpperCAmelCase_  :	List[Any]				      =		features
             UpperCAmelCase_  :	List[Any]				      =		cache_dir
             UpperCAmelCase_  :	List[str]				      =		keep_in_memory
             UpperCAmelCase_  :	Optional[int]				      =		streaming
             UpperCAmelCase_  :	Dict				      =		num_proc
             UpperCAmelCase_  :	str				      =		kwargs



       @abstractmethod
       def  A_   (     self):

             """simple docstring"""





             pass
