import gc
import random
import unittest

import numpy as np
import torch

from diffusers import (
    DDIMScheduler,
    KandinskyVaaControlnetPipeline,
    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_			(						SCREAMING_SNAKE_CASE	,					unittest.TestCase						):






 '''simple docstring'''






 UpperCAmelCase_     			= KandinskyVaaControlnetPipeline
 UpperCAmelCase_     			= ['image_embeds', 'negative_image_embeds', 'hint']
 UpperCAmelCase_     			= ['image_embeds', 'negative_image_embeds', 'hint']
 UpperCAmelCase_     			= [
     'generator',
     'height',
     'width',
     'latents',
     'guidance_scale',
     'num_inference_steps',
     'return_dict',
     'guidance_scale',
     'num_images_per_prompt',
     'output_type',
     'return_dict',
 ]
 UpperCAmelCase_     			= False




 @property
 def 		__snake_case						(      self :  Optional[int]):







        '''simple docstring'''



        return 32




 @property
 def 		__snake_case						(      self :  str):







        '''simple docstring'''



        return 32




 @property
 def 		__snake_case						(      self :  str):







        '''simple docstring'''



        return self.time_input_dim




 @property
 def 		__snake_case						(      self :  Tuple):







        '''simple docstring'''



        return self.time_input_dim * 4




 @property
 def 		__snake_case						(      self :  str):







        '''simple docstring'''



        return 100




 @property
 def 		__snake_case						(      self :  List[str]):







        '''simple docstring'''



        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(**lowercase__)
        return model




 @property
 def 		__snake_case						(      self :  Optional[int]):







        '''simple docstring'''



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




 @property
 def 		__snake_case						(      self :  Union[str, Any]):







        '''simple docstring'''



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




 def 		__snake_case						(      self :  Tuple):







        '''simple docstring'''



        lowerCAmelCase__					     = self.dummy_unet
        lowerCAmelCase__					     = self.dummy_movq

        lowerCAmelCase__					     = DDIMScheduler(
            num_train_timesteps=1_000      ,   beta_schedule='linear'      ,   beta_start=0.00_085      ,   beta_end=0.012      ,   clip_sample=lowercase__      ,   set_alpha_to_one=lowercase__      ,   steps_offset=1      ,   prediction_type='epsilon'      ,   thresholding=lowercase__      ,   )

        lowerCAmelCase__					     = {
            'unet': unet,
            'scheduler': scheduler,
            'movq': movq,
        }
        return components




 def 		__snake_case						(      self :  Optional[int]      ,   lowercase__ :  Optional[Any]      ,   lowercase__ :  Dict=0):







        '''simple docstring'''



        lowerCAmelCase__					     = floats_tensor((1, self.text_embedder_hidden_size)      ,   rng=random.Random(lowercase__)).to(lowercase__)
        lowerCAmelCase__					     = floats_tensor((1, self.text_embedder_hidden_size)      ,   rng=random.Random(seed + 1)).to(
            lowercase__)

        # create hint
        lowerCAmelCase__					     = floats_tensor((1, 3, 64, 64)      ,   rng=random.Random(lowercase__)).to(lowercase__)

        if str(lowercase__).startswith('mps'):
               lowerCAmelCase__					     = torch.manual_seed(lowercase__)
        else:
               lowerCAmelCase__					     = torch.Generator(device=lowercase__).manual_seed(lowercase__)
        lowerCAmelCase__					     = {
            'image_embeds': image_embeds,
            'negative_image_embeds': negative_image_embeds,
            'hint': hint,
            'generator': generator,
            'height': 64,
            'width': 64,
            'guidance_scale': 4.0,
            'num_inference_steps': 2,
            'output_type': 'np',
        }
        return inputs




 def 		__snake_case						(      self :  int):







        '''simple docstring'''



        lowerCAmelCase__					     = 'cpu'

        lowerCAmelCase__					     = self.get_dummy_components()

        lowerCAmelCase__					     = self.pipeline_class(**lowercase__)
        lowerCAmelCase__					     = pipe.to(lowercase__)

        pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__					     = pipe(**self.get_dummy_inputs(lowercase__))
        lowerCAmelCase__					     = output.images

        lowerCAmelCase__					     = pipe(
            **self.get_dummy_inputs(lowercase__)      ,   return_dict=lowercase__      ,   )[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_959_826, 0.868_279, 0.7_558_092, 0.68_769_467, 0.85_805_804, 0.65_977_496, 0.44_885_302, 0.5_959_111, 0.4_251_595])

        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						):






 '''simple docstring'''






 def 		__snake_case						(      self :  Dict):







        '''simple docstring'''



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




 def 		__snake_case						(      self :  Optional[int]):







        '''simple docstring'''



        lowerCAmelCase__					     = load_numpy(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
            '/kandinskyv22/kandinskyv22_controlnet_robotcat_fp16.npy')

        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(lowercase__)).float() / 255.0
        lowerCAmelCase__					     = hint.permute(2      ,   0      ,   1).unsqueeze(0)

        lowerCAmelCase__					     = KandinskyVaaPriorPipeline.from_pretrained(
            'kandinsky-community/kandinsky-2-2-prior'      ,   torch_dtype=torch.floataa)
        pipe_prior.to(lowercase__)

        lowerCAmelCase__					     = KandinskyVaaControlnetPipeline.from_pretrained(
            'kandinsky-community/kandinsky-2-2-controlnet-depth'      ,   torch_dtype=torch.floataa)
        lowerCAmelCase__					     = pipeline.to(lowercase__)
        pipeline.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__					     = 'A robot, 4k photo'

        lowerCAmelCase__					     = torch.Generator(device='cuda').manual_seed(0)
        lowerCAmelCase__     , lowerCAmelCase__					     = pipe_prior(
            lowercase__      ,   generator=lowercase__      ,   num_inference_steps=5      ,   negative_prompt=''      ,   ).to_tuple()

        lowerCAmelCase__					     = torch.Generator(device='cuda').manual_seed(0)
        lowerCAmelCase__					     = pipeline(
            image_embeds=lowercase__      ,   negative_image_embeds=lowercase__      ,   hint=lowercase__      ,   generator=lowercase__      ,   num_inference_steps=100      ,   output_type='np'      ,   )

        lowerCAmelCase__					     = output.images[0]

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

        assert_mean_pixel_difference(lowercase__      ,   lowercase__)

from typing import Optional

import pyspark

from .. import Features, NamedSplit
from ..download import DownloadMode
from ..packaged_modules.spark.spark import Spark
from .abc import AbstractDatasetReader
class     a_			(						SCREAMING_SNAKE_CASE						):






 '''simple docstring'''






 def __init__(      self :  str      ,   lowercase__ :  pyspark.sql.DataFrame      ,   lowercase__ :  Optional[NamedSplit] = None      ,   lowercase__ :  Optional[Features] = None      ,   lowercase__ :  bool = True      ,   lowercase__ :  str = None      ,   lowercase__ :  bool = False      ,   lowercase__ :  str = None      ,   lowercase__ :  bool = True      ,   lowercase__ :  str = "arrow"      ,   **lowercase__ :  Any      ,   ):







        '''simple docstring'''



        super().__init__(
            split=lowercase__      ,   features=lowercase__      ,   cache_dir=lowercase__      ,   keep_in_memory=lowercase__      ,   streaming=lowercase__      ,   **lowercase__      ,   )
        lowerCAmelCase__					     = load_from_cache_file
        lowerCAmelCase__					     = file_format
        lowerCAmelCase__					     = Spark(
            df=lowercase__      ,   features=lowercase__      ,   cache_dir=lowercase__      ,   working_dir=lowercase__      ,   **lowercase__      ,   )




 def 		__snake_case						(      self :  Tuple):







        '''simple docstring'''



        if self.streaming:
               return self.builder.as_streaming_dataset(split=self.split)
        lowerCAmelCase__					     = None if self._load_from_cache_file else DownloadMode.FORCE_REDOWNLOAD
        self.builder.download_and_prepare(
            download_mode=lowercase__      ,   file_format=self._file_format      ,   )
        return self.builder.as_dataset(split=self.split)
