





'''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 					UpperCAmelCase	(     snake_case_				,						unittest.TestCase  ):
      _lowercase:     Tuple			=  KandinskyVaaImgaImgPipeline
      _lowercase:     Tuple			=  ['''image_embeds''', '''negative_image_embeds''', '''image''']
      _lowercase:     int			=  [
          '''image_embeds''',
          '''negative_image_embeds''',
          '''image''',
      ]
      _lowercase:     Tuple			=  [
          '''generator''',
          '''height''',
          '''width''',
          '''strength''',
          '''guidance_scale''',
          '''num_inference_steps''',
          '''return_dict''',
          '''guidance_scale''',
          '''num_images_per_prompt''',
          '''output_type''',
          '''return_dict''',
      ]
      _lowercase:     Any			=  False




      @property
      def 							lowercase__       (     self     :					List[str]    )							-> Union[str, Any]:
            return 32




      @property
      def 							lowercase__       (     self     :					Union[str, Any]    )							-> Union[str, Any]:
            return 32




      @property
      def 							lowercase__       (     self     :					List[Any]    )							-> List[Any]:
            return self.time_input_dim




      @property
      def 							lowercase__       (     self     :					List[str]    )							-> List[Any]:
            return self.time_input_dim * 4




      @property
      def 							lowercase__       (     self     :					str    )							-> Dict:
            return 1_00




      @property
      def 							lowercase__       (     self     :					List[str]    )							-> Optional[Any]:
            torch.manual_seed(0    )

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

            _lowerCAmelCase					=				UNetaDConditionModel(**__snake_case    )
            return model




      @property
      def 							lowercase__       (     self     :					str    )							-> List[Any]:
            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 							lowercase__       (     self     :					Any    )							-> Dict:
            torch.manual_seed(0    )
            _lowerCAmelCase					=				VQModel(**self.dummy_movq_kwargs    )
            return model




      def 							lowercase__       (     self     :					List[Any]    )							-> Optional[Any]:
            _lowerCAmelCase					=				self.dummy_unet
            _lowerCAmelCase					=				self.dummy_movq

            _lowerCAmelCase					=				{
                """num_train_timesteps""": 10_00,
                """beta_schedule""": """linear""",
                """beta_start""": 0.0_00_85,
                """beta_end""": 0.0_12,
                """clip_sample""": False,
                """set_alpha_to_one""": False,
                """steps_offset""": 0,
                """prediction_type""": """epsilon""",
                """thresholding""": False,
            }

            _lowerCAmelCase					=				DDIMScheduler(**__snake_case    )

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

            return components




      def 							lowercase__       (     self     :					Tuple     ,     __snake_case     :					Optional[int]     ,     __snake_case     :					Union[str, Any]=0    )							-> int:
            _lowerCAmelCase					=				floats_tensor((1, self.text_embedder_hidden_size)     ,     rng=random.Random(__snake_case    )    ).to(__snake_case    )
            _lowerCAmelCase					=				floats_tensor((1, self.text_embedder_hidden_size)     ,     rng=random.Random(seed + 1    )    ).to(
                __snake_case    )
            # create init_image
            _lowerCAmelCase					=				floats_tensor((1, 3, 64, 64)     ,     rng=random.Random(__snake_case    )    ).to(__snake_case    )
            _lowerCAmelCase					=				image.cpu().permute(0     ,     2     ,     3     ,     1    )[0]
            _lowerCAmelCase					=				Image.fromarray(np.uinta(__snake_case    )    ).convert("""RGB"""    ).resize((2_56, 2_56)    )

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

            _lowerCAmelCase					=				self.get_dummy_components()

            _lowerCAmelCase					=				self.pipeline_class(**__snake_case    )
            _lowerCAmelCase					=				pipe.to(__snake_case    )

            pipe.set_progress_bar_config(disable=__snake_case    )

            _lowerCAmelCase					=				pipe(**self.get_dummy_inputs(__snake_case    )    )
            _lowerCAmelCase					=				output.images

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

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

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

            _lowerCAmelCase					=				np.array(
                [0.6_19_97_78, 0.63_98_44_06, 0.46_14_57_85, 0.62_94_49_84, 0.5_62_22_15, 0.47_30_61_32, 0.47_44_14_56, 0.4_60_76_06, 0.48_71_92_63]    )
            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 							lowercase__       (     self     :					Optional[int]    )							-> Tuple:
            # clean up the VRAM after each test
            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()




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

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

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

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

            pipeline.set_progress_bar_config(disable=__snake_case    )

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

            _lowerCAmelCase					=				pipeline(
                image=__snake_case     ,     image_embeds=__snake_case     ,     negative_image_embeds=__snake_case     ,     generator=__snake_case     ,     num_inference_steps=1_00     ,     height=7_68     ,     width=7_68     ,     strength=0.2     ,     output_type="""np"""     ,     )

            _lowerCAmelCase					=				output.images[0]

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

            assert_mean_pixel_difference(__snake_case     ,     __snake_case    )













'''simple docstring'''






from typing import Callable, Optional

from .. import Features
from ..packaged_modules.generator.generator import Generator
from .abc import AbstractDatasetInputStream




class 					UpperCAmelCase	(     snake_case_  ):




      def __init__(     self     :					Union[str, Any]     ,     __snake_case     :					Callable     ,     __snake_case     :					Optional[Features] = None     ,     __snake_case     :					str = None     ,     __snake_case     :					bool = False     ,     __snake_case     :					bool = False     ,     __snake_case     :					Optional[dict] = None     ,     __snake_case     :					Optional[int] = None     ,     **__snake_case     :					Optional[Any]     ,     )							-> Union[str, Any]:
            super().__init__(
                features=__snake_case     ,     cache_dir=__snake_case     ,     keep_in_memory=__snake_case     ,     streaming=__snake_case     ,     num_proc=__snake_case     ,     **__snake_case     ,     )
            _lowerCAmelCase					=				Generator(
                cache_dir=__snake_case     ,     features=__snake_case     ,     generator=__snake_case     ,     gen_kwargs=__snake_case     ,     **__snake_case     ,     )




      def 							lowercase__       (     self     :					str    )							-> Dict:
            # Build iterable dataset
            if self.streaming:
                  _lowerCAmelCase					=				self.builder.as_streaming_dataset(split="""train"""    )
            # Build regular (map-style) dataset
            else:
                  _lowerCAmelCase					=				None
                  _lowerCAmelCase					=				None
                  _lowerCAmelCase					=				None
                  _lowerCAmelCase					=				None

                  self.builder.download_and_prepare(
                      download_config=__snake_case     ,     download_mode=__snake_case     ,     verification_mode=__snake_case     ,     base_path=__snake_case     ,     num_proc=self.num_proc     ,     )
                  _lowerCAmelCase					=				self.builder.as_dataset(
                      split="""train"""     ,     verification_mode=__snake_case     ,     in_memory=self.keep_in_memory    )
            return dataset






