


'''simple docstring'''







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 A__       (						UpperCAmelCase__ ,     unittest.TestCase      ):
      __UpperCamelCase       :  Tuple	       =							KandinskyVaaImgaImgPipeline
      __UpperCamelCase       :  Dict	       =							["image_embeds", "negative_image_embeds", "image"]
      __UpperCamelCase       :  Tuple	       =							[
          "image_embeds",
          "negative_image_embeds",
          "image",
      ]
      __UpperCamelCase       :  Any	       =							[
          "generator",
          "height",
          "width",
          "strength",
          "guidance_scale",
          "num_inference_steps",
          "return_dict",
          "guidance_scale",
          "num_images_per_prompt",
          "output_type",
          "return_dict",
      ]
      __UpperCamelCase       :  Tuple	       =							False






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




             '''simple docstring'''
             return 3_2






      @property
      def __UpperCAmelCase       ( self			:str	)       ->					str:




             '''simple docstring'''
             return 3_2






      @property
      def __UpperCAmelCase       ( self			:Dict	)       ->					Any:




             '''simple docstring'''
             return self.time_input_dim






      @property
      def __UpperCAmelCase       ( self			:Optional[int]	)       ->					Any:




             '''simple docstring'''
             return self.time_input_dim * 4






      @property
      def __UpperCAmelCase       ( self			:List[Any]	)       ->					List[str]:




             '''simple docstring'''
             return 1_0_0






      @property
      def __UpperCAmelCase       ( self			:Dict	)       ->					List[Any]:




             '''simple docstring'''
             torch.manual_seed(0	)

             _a    : str     					={
                 """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,
             }

             _a    : Union[str, Any]     					=UNetaDConditionModel(**SCREAMING_SNAKE_CASE	)
             return model






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




             '''simple docstring'''
             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			:Any	)       ->					Union[str, Any]:




             '''simple docstring'''
             torch.manual_seed(0	)
             _a    : Optional[Any]     					=VQModel(**self.dummy_movq_kwargs	)
             return model






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




             '''simple docstring'''
             _a    : Optional[int]     					=self.dummy_unet
             _a    : List[Any]     					=self.dummy_movq

             _a    : List[str]     					={
                 """num_train_timesteps""": 1_0_0_0,
                 """beta_schedule""": """linear""",
                 """beta_start""": 0.00_085,
                 """beta_end""": 0.012,
                 """clip_sample""": False,
                 """set_alpha_to_one""": False,
                 """steps_offset""": 0,
                 """prediction_type""": """epsilon""",
                 """thresholding""": False,
             }

             _a    : List[Any]     					=DDIMScheduler(**SCREAMING_SNAKE_CASE	)

             _a    : str     					={
                 """unet""": unet,
                 """scheduler""": scheduler,
                 """movq""": movq,
             }

             return components






      def __UpperCAmelCase       ( self			:str     ,						SCREAMING_SNAKE_CASE			:List[Any]     ,						SCREAMING_SNAKE_CASE			:Tuple=0	)       ->					Dict:




             '''simple docstring'''
             _a    : str     					=floats_tensor((1, self.text_embedder_hidden_size)     ,						rng=random.Random(SCREAMING_SNAKE_CASE	)	).to(SCREAMING_SNAKE_CASE	)
             _a    : Union[str, Any]     					=floats_tensor((1, self.text_embedder_hidden_size)     ,						rng=random.Random(seed + 1	)	).to(
                 SCREAMING_SNAKE_CASE	)
             # create init_image
             _a    : Optional[int]     					=floats_tensor((1, 3, 6_4, 6_4)     ,						rng=random.Random(SCREAMING_SNAKE_CASE	)	).to(SCREAMING_SNAKE_CASE	)
             _a    : Any     					=image.cpu().permute(0     ,						2     ,						3     ,						1	)[0]
             _a    : Union[str, Any]     					=Image.fromarray(np.uinta(SCREAMING_SNAKE_CASE	)	).convert("""RGB"""	).resize((2_5_6, 2_5_6)	)

             if str(SCREAMING_SNAKE_CASE	).startswith("""mps"""	):
                    _a    : Any     					=torch.manual_seed(SCREAMING_SNAKE_CASE	)
             else:
                    _a    : Union[str, Any]     					=torch.Generator(device=SCREAMING_SNAKE_CASE	).manual_seed(SCREAMING_SNAKE_CASE	)
             _a    : List[Any]     					={
                 """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			:List[Any]	)       ->					List[str]:




             '''simple docstring'''
             _a    : Dict     					="""cpu"""

             _a    : Tuple     					=self.get_dummy_components()

             _a    : List[str]     					=self.pipeline_class(**SCREAMING_SNAKE_CASE	)
             _a    : int     					=pipe.to(SCREAMING_SNAKE_CASE	)

             pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE	)

             _a    : Tuple     					=pipe(**self.get_dummy_inputs(SCREAMING_SNAKE_CASE	)	)
             _a    : Optional[int]     					=output.images

             _a    : Optional[int]     					=pipe(
                 **self.get_dummy_inputs(SCREAMING_SNAKE_CASE	)     ,						return_dict=SCREAMING_SNAKE_CASE     ,						)[0]

             _a    : List[str]     					=image[0, -3:, -3:, -1]
             _a    : Optional[int]     					=image_from_tuple[0, -3:, -3:, -1]

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

             _a    : List[Any]     					=np.array(
                 [0.6_199_778, 0.63_984_406, 0.46_145_785, 0.62_944_984, 0.5_622_215, 0.47_306_132, 0.47_441_456, 0.4_607_606, 0.48_719_263]	)
             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 A__       (						unittest.TestCase      ):






      def __UpperCAmelCase       ( self			:int	)       ->					Dict:




             '''simple docstring'''
             # clean up the VRAM after each test
             super().tearDown()
             gc.collect()
             torch.cuda.empty_cache()






      def __UpperCAmelCase       ( self			:Dict	)       ->					Optional[int]:




             '''simple docstring'''
             _a    : Any     					=load_numpy(
                 """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
                 """/kandinskyv22/kandinskyv22_img2img_frog.npy"""	)

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

             _a    : str     					=KandinskyVaaPriorPipeline.from_pretrained(
                 """kandinsky-community/kandinsky-2-2-prior"""     ,						torch_dtype=torch.floataa	)
             pipe_prior.to(SCREAMING_SNAKE_CASE	)

             _a    : Optional[Any]     					=KandinskyVaaImgaImgPipeline.from_pretrained(
                 """kandinsky-community/kandinsky-2-2-decoder"""     ,						torch_dtype=torch.floataa	)
             _a    : List[str]     					=pipeline.to(SCREAMING_SNAKE_CASE	)

             pipeline.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE	)

             _a    : Dict     					=torch.Generator(device="""cpu"""	).manual_seed(0	)
             _a					,    _a    : List[Any]     					=pipe_prior(
                 SCREAMING_SNAKE_CASE     ,						generator=SCREAMING_SNAKE_CASE     ,						num_inference_steps=5     ,						negative_prompt=""""""     ,						).to_tuple()

             _a    : Any     					=pipeline(
                 image=SCREAMING_SNAKE_CASE     ,						image_embeds=SCREAMING_SNAKE_CASE     ,						negative_image_embeds=SCREAMING_SNAKE_CASE     ,						generator=SCREAMING_SNAKE_CASE     ,						num_inference_steps=1_0_0     ,						height=7_6_8     ,						width=7_6_8     ,						strength=0.2     ,						output_type="""np"""     ,						)

             _a    : str     					=output.images[0]

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

             assert_mean_pixel_difference(SCREAMING_SNAKE_CASE     ,						SCREAMING_SNAKE_CASE	)






'''simple docstring'''







from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, Encoder, VectorQuantizer







@dataclass
class A__       (						UpperCAmelCase__      ):
      __UpperCamelCase       :  torch.FloatTensor







class A__       (						UpperCAmelCase__ ,     UpperCAmelCase__      ):






      @register_to_config
      def __init__( self			:Optional[Any]     ,						SCREAMING_SNAKE_CASE			:int = 3     ,						SCREAMING_SNAKE_CASE			:int = 3     ,						SCREAMING_SNAKE_CASE			:Tuple[str] = ("DownEncoderBlock2D",)     ,						SCREAMING_SNAKE_CASE			:Tuple[str] = ("UpDecoderBlock2D",)     ,						SCREAMING_SNAKE_CASE			:Tuple[int] = (6_4,)     ,						SCREAMING_SNAKE_CASE			:int = 1     ,						SCREAMING_SNAKE_CASE			:str = "silu"     ,						SCREAMING_SNAKE_CASE			:int = 3     ,						SCREAMING_SNAKE_CASE			:int = 3_2     ,						SCREAMING_SNAKE_CASE			:int = 2_5_6     ,						SCREAMING_SNAKE_CASE			:int = 3_2     ,						SCREAMING_SNAKE_CASE			:Optional[int] = None     ,						SCREAMING_SNAKE_CASE			:float = 0.18_215     ,						SCREAMING_SNAKE_CASE			:str = "group"     ,						)       ->					Optional[int]:




             '''simple docstring'''
             super().__init__()

             # pass init params to Encoder
             _a    : Union[str, Any]     					=Encoder(
                 in_channels=SCREAMING_SNAKE_CASE     ,						out_channels=SCREAMING_SNAKE_CASE     ,						down_block_types=SCREAMING_SNAKE_CASE     ,						block_out_channels=SCREAMING_SNAKE_CASE     ,						layers_per_block=SCREAMING_SNAKE_CASE     ,						act_fn=SCREAMING_SNAKE_CASE     ,						norm_num_groups=SCREAMING_SNAKE_CASE     ,						double_z=SCREAMING_SNAKE_CASE     ,						)

             _a    : Optional[int]     					=vq_embed_dim if vq_embed_dim is not None else latent_channels

             _a    : Optional[int]     					=nn.Convad(SCREAMING_SNAKE_CASE     ,						SCREAMING_SNAKE_CASE     ,						1	)
             _a    : str     					=VectorQuantizer(SCREAMING_SNAKE_CASE     ,						SCREAMING_SNAKE_CASE     ,						beta=0.25     ,						remap=SCREAMING_SNAKE_CASE     ,						sane_index_shape=SCREAMING_SNAKE_CASE	)
             _a    : List[str]     					=nn.Convad(SCREAMING_SNAKE_CASE     ,						SCREAMING_SNAKE_CASE     ,						1	)

             # pass init params to Decoder
             _a    : List[str]     					=Decoder(
                 in_channels=SCREAMING_SNAKE_CASE     ,						out_channels=SCREAMING_SNAKE_CASE     ,						up_block_types=SCREAMING_SNAKE_CASE     ,						block_out_channels=SCREAMING_SNAKE_CASE     ,						layers_per_block=SCREAMING_SNAKE_CASE     ,						act_fn=SCREAMING_SNAKE_CASE     ,						norm_num_groups=SCREAMING_SNAKE_CASE     ,						norm_type=SCREAMING_SNAKE_CASE     ,						)






      @apply_forward_hook
      def __UpperCAmelCase       ( self			:Optional[Any]     ,						SCREAMING_SNAKE_CASE			:torch.FloatTensor     ,						SCREAMING_SNAKE_CASE			:bool = True	)       ->					VQEncoderOutput:




             '''simple docstring'''
             _a    : Optional[int]     					=self.encoder(SCREAMING_SNAKE_CASE	)
             _a    : int     					=self.quant_conv(SCREAMING_SNAKE_CASE	)

             if not return_dict:
                    return (h,)

             return VQEncoderOutput(latents=SCREAMING_SNAKE_CASE	)






      @apply_forward_hook
      def __UpperCAmelCase       ( self			:List[Any]     ,						SCREAMING_SNAKE_CASE			:torch.FloatTensor     ,						SCREAMING_SNAKE_CASE			:bool = False     ,						SCREAMING_SNAKE_CASE			:bool = True	)       ->					Union[DecoderOutput, torch.FloatTensor]:




             '''simple docstring'''
             # also go through quantization layer
             if not force_not_quantize:
                    _a					,    _a					,    _a    : Tuple     					=self.quantize(SCREAMING_SNAKE_CASE	)
             else:
                    _a    : str     					=h
             _a    : Dict     					=self.post_quant_conv(SCREAMING_SNAKE_CASE	)
             _a    : Union[str, Any]     					=self.decoder(SCREAMING_SNAKE_CASE     ,						quant if self.config.norm_type == """spatial""" else None	)

             if not return_dict:
                    return (dec,)

             return DecoderOutput(sample=SCREAMING_SNAKE_CASE	)






      def __UpperCAmelCase       ( self			:Optional[Any]     ,						SCREAMING_SNAKE_CASE			:torch.FloatTensor     ,						SCREAMING_SNAKE_CASE			:bool = True	)       ->					Union[DecoderOutput, torch.FloatTensor]:




             '''simple docstring'''
             _a    : Tuple     					=sample
             _a    : int     					=self.encode(SCREAMING_SNAKE_CASE	).latents
             _a    : List[Any]     					=self.decode(SCREAMING_SNAKE_CASE	).sample

             if not return_dict:
                    return (dec,)

             return DecoderOutput(sample=SCREAMING_SNAKE_CASE	)


