






'''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    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE       ,unittest.TestCase	):


   '''simple docstring'''





   __a							:							Optional[int]					     =					KandinskyVaaImgaImgPipeline
   __a							:							Dict					     =					["image_embeds", "negative_image_embeds", "image"]
   __a							:							Union[str, Any]					     =					[
       "image_embeds",
       "negative_image_embeds",
       "image",
   ]
   __a							:							str					     =					[
       "generator",
       "height",
       "width",
       "strength",
       "guidance_scale",
       "num_inference_steps",
       "return_dict",
       "guidance_scale",
       "num_images_per_prompt",
       "output_type",
       "return_dict",
   ]
   __a							:							Union[str, Any]					     =					False






   @property
   def 				A      (   self							:						Optional[int]			)							->    str:

         '''simple docstring'''







         return 3_2






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

         '''simple docstring'''







         return 3_2






   @property
   def 				A      (   self							:						Dict			)							->    List[str]:

         '''simple docstring'''







         return self.time_input_dim






   @property
   def 				A      (   self							:						Dict			)							->    str:

         '''simple docstring'''







         return self.time_input_dim * 4






   @property
   def 				A      (   self							:						List[str]			)							->    Tuple:

         '''simple docstring'''







         return 1_0_0






   @property
   def 				A      (   self							:						Tuple			)							->    str:

         '''simple docstring'''







         torch.manual_seed(0			)

         UpperCamelCase__         = {
             """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__         = UNetaDConditionModel(**UpperCamelCase__			)
         return model






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

         '''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 				A      (   self							:						Union[str, Any]			)							->    Union[str, Any]:

         '''simple docstring'''







         torch.manual_seed(0			)
         UpperCamelCase__         = VQModel(**self.dummy_movq_kwargs			)
         return model






   def 				A      (   self							:						Optional[Any]			)							->    Any:

         '''simple docstring'''







         UpperCamelCase__         = self.dummy_unet
         UpperCamelCase__         = self.dummy_movq

         UpperCamelCase__         = {
             """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,
         }

         UpperCamelCase__         = DDIMScheduler(**UpperCamelCase__			)

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

         return components






   def 				A      (   self							:						int ,      lowercase							:						Optional[Any] ,      lowercase							:						Dict=0			)							->    Any:

         '''simple docstring'''







         UpperCamelCase__         = floats_tensor((1, self.text_embedder_hidden_size) ,      rng=random.Random(UpperCamelCase__			)			).to(UpperCamelCase__			)
         UpperCamelCase__         = floats_tensor((1, self.text_embedder_hidden_size) ,      rng=random.Random(seed + 1			)			).to(
             UpperCamelCase__			)
         # create init_image
         UpperCamelCase__         = floats_tensor((1, 3, 6_4, 6_4) ,      rng=random.Random(UpperCamelCase__			)			).to(UpperCamelCase__			)
         UpperCamelCase__         = image.cpu().permute(0 ,      2 ,      3 ,      1			)[0]
         UpperCamelCase__         = Image.fromarray(np.uinta(UpperCamelCase__			)			).convert("""RGB"""			).resize((2_5_6, 2_5_6)			)

         if str(UpperCamelCase__			).startswith("""mps"""			):
               UpperCamelCase__         = torch.manual_seed(UpperCamelCase__			)
         else:
               UpperCamelCase__         = torch.Generator(device=UpperCamelCase__			).manual_seed(UpperCamelCase__			)
         UpperCamelCase__         = {
             """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 				A      (   self							:						Tuple			)							->    Optional[int]:

         '''simple docstring'''







         UpperCamelCase__         = """cpu"""

         UpperCamelCase__         = self.get_dummy_components()

         UpperCamelCase__         = self.pipeline_class(**UpperCamelCase__			)
         UpperCamelCase__         = pipe.to(UpperCamelCase__			)

         pipe.set_progress_bar_config(disable=UpperCamelCase__			)

         UpperCamelCase__         = pipe(**self.get_dummy_inputs(UpperCamelCase__			)			)
         UpperCamelCase__         = output.images

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

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

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

         UpperCamelCase__         = np.array(
             [0.6_1_9_9_7_7_8, 0.6_3_9_8_4_4_0_6, 0.4_6_1_4_5_7_8_5, 0.6_2_9_4_4_9_8_4, 0.5_6_2_2_2_1_5, 0.4_7_3_0_6_1_3_2, 0.4_7_4_4_1_4_5_6, 0.4_6_0_7_6_0_6, 0.4_8_7_1_9_2_6_3]			)
         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    _SCREAMING_SNAKE_CASE							(		unittest.TestCase	):


   '''simple docstring'''





   def 				A      (   self							:						Any			)							->    Dict:

         '''simple docstring'''







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







   def 				A      (   self							:						str			)							->    List[Any]:

         '''simple docstring'''







         UpperCamelCase__         = load_numpy(
             """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
             """/kandinskyv22/kandinskyv22_img2img_frog.npy"""			)

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

         UpperCamelCase__         = KandinskyVaaPriorPipeline.from_pretrained(
             """kandinsky-community/kandinsky-2-2-prior""" ,      torch_dtype=torch.floataa			)
         pipe_prior.to(UpperCamelCase__			)

         UpperCamelCase__         = KandinskyVaaImgaImgPipeline.from_pretrained(
             """kandinsky-community/kandinsky-2-2-decoder""" ,      torch_dtype=torch.floataa			)
         UpperCamelCase__         = pipeline.to(UpperCamelCase__			)

         pipeline.set_progress_bar_config(disable=UpperCamelCase__			)

         UpperCamelCase__         = torch.Generator(device="""cpu"""			).manual_seed(0			)
         UpperCamelCase__    ,   UpperCamelCase__         = pipe_prior(
             UpperCamelCase__ ,      generator=UpperCamelCase__ ,      num_inference_steps=5 ,      negative_prompt="""""" ,      ).to_tuple()

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

         UpperCamelCase__         = output.images[0]

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

         assert_mean_pixel_difference(UpperCamelCase__ ,      UpperCamelCase__			)








'''simple docstring'''


import enum
import os
from hashlib import shaaaa
from typing import Optional

from .. import config
from .logging import get_logger


lowerCamelCase_				: Any    	=							get_logger(__name__)
class    _SCREAMING_SNAKE_CASE							(		enum.Enum	):


			'''simple docstring'''





			__a							:							Tuple					     =					"all_checks"
			__a							:							List[Any]					     =					"basic_checks"
			__a							:							Dict					     =					"no_checks"
class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''












def 	__magic_name__( _A						,      _A						,      _A=None					):
						'''simple docstring'''





						if expected_checksums is None:
												logger.info("""Unable to verify checksums."""					)
												return
						if len(set(_A					) - set(_A					)					) > 0:
												raise ExpectedMoreDownloadedFiles(str(set(_A					) - set(_A					)					)					)
						if len(set(_A					) - set(_A					)					) > 0:
												raise UnexpectedDownloadedFile(str(set(_A					) - set(_A					)					)					)
						UpperCamelCase__         = [url for url in expected_checksums if expected_checksums[url] != recorded_checksums[url]]
						UpperCamelCase__         = """ for """ + verification_name if verification_name is not None else """"""
						if len(_A					) > 0:
												raise NonMatchingChecksumError(
												    f"Checksums didn't match{for_verification_name}:\n"
												    f"{bad_urls}\n"
												    """Set `verification_mode='no_checks'` to skip checksums verification and ignore this error"""					)
						logger.info("""All the checksums matched successfully""" + for_verification_name					)
class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''





class    _SCREAMING_SNAKE_CASE							(		SCREAMING_SNAKE_CASE	):


			'''simple docstring'''












def 	__magic_name__( _A						,      _A					):
						'''simple docstring'''





						if expected_splits is None:
												logger.info("""Unable to verify splits sizes."""					)
												return
						if len(set(_A					) - set(_A					)					) > 0:
												raise ExpectedMoreSplits(str(set(_A					) - set(_A					)					)					)
						if len(set(_A					) - set(_A					)					) > 0:
												raise UnexpectedSplits(str(set(_A					) - set(_A					)					)					)
						UpperCamelCase__         = [
						    {"""expected""": expected_splits[name], """recorded""": recorded_splits[name]}
						    for name in expected_splits
						    if expected_splits[name].num_examples != recorded_splits[name].num_examples
						]
						if len(_A					) > 0:
												raise NonMatchingSplitsSizesError(str(_A					)					)
						logger.info("""All the splits matched successfully."""					)







def 	__magic_name__( _A						,      _A = True					):
						'''simple docstring'''





						if record_checksum:
												UpperCamelCase__         = shaaaa()
												with open(_A						,      """rb"""					) as f:
																		for chunk in iter(lambda: f.read(1 << 20					)						,      B""""""					):
																								m.update(_A					)
																		UpperCamelCase__         = m.hexdigest()
						else:
												UpperCamelCase__         = None
						return {"num_bytes": os.path.getsize(_A					), "checksum": checksum}







def 	__magic_name__( _A					):
						'''simple docstring'''





						if dataset_size and config.IN_MEMORY_MAX_SIZE:
												return dataset_size < config.IN_MEMORY_MAX_SIZE
						else:
												return False
