


from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import center_crop, normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    IMAGENET_DEFAULT_MEAN,
    IMAGENET_DEFAULT_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    get_image_size,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
 import PIL


_SCREAMING_SNAKE_CASE    :				Any       =					logging.get_logger(__name__)

class 						A__      (					snake_case__			):







    """simple docstring"""

    __magic_name__    						=	['pixel_values']
    def __init__( self    ,      __snake_case = True    ,      __snake_case = None    ,      __snake_case = True    ,      __snake_case = None    ,      __snake_case = PILImageResampling.BICUBIC    ,      __snake_case = True    ,      __snake_case = 1 / 2_5_5    ,      __snake_case = True    ,      __snake_case = None    ,      __snake_case = None    ,      **__snake_case    ,      ):
         super().__init__(**__snake_case					)
         snake_case					=       crop_size if crop_size is not None else {'''height''': 2_5_6, '''width''': 2_5_6}
         snake_case					=       get_size_dict(__snake_case    ,      param_name='''crop_size'''					)
         snake_case					=       size if size is not None else {'''height''': 2_2_4, '''width''': 2_2_4}
         snake_case					=       get_size_dict(__snake_case					)

         snake_case					=       do_center_crop
         snake_case					=       crop_size
         snake_case					=       do_resize
         snake_case					=       size
         snake_case					=       resample
         snake_case					=       do_rescale
         snake_case					=       rescale_factor
         snake_case					=       do_normalize
         snake_case					=       image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
         snake_case					=       image_std if image_std is not None else IMAGENET_DEFAULT_STD
    def 	a_    ( self    ,      __snake_case    ,      __snake_case    ,      __snake_case = None    ,      __snake_case = None    ,      **__snake_case    ,      ):
         snake_case					=       self.size if size is None else size
         snake_case					=       get_size_dict(__snake_case					)
         snake_case					=       get_size_dict(__snake_case    ,      param_name='''crop_size'''					)

         snake_case				,						snake_case					=       get_image_size(__snake_case					)
         snake_case					=       min(__snake_case    ,      __snake_case					)
         snake_case					=       (size['''height'''] / crop_size['''height''']) * min_dim
         snake_case					=       (size['''width'''] / crop_size['''width''']) * min_dim
         return center_crop(__snake_case    ,      size=(cropped_height, cropped_width)    ,      data_format=__snake_case    ,      **__snake_case					)
    def 	a_    ( self    ,      __snake_case    ,      __snake_case    ,      __snake_case = PIL.Image.BICUBIC    ,      __snake_case = None    ,      **__snake_case    ,      ):
         snake_case					=       get_size_dict(__snake_case					)
         if "height" not in size or "width" not in size:
              raise ValueError(F'''The size dictionary must contain the keys \'height\' and \'width\'. Got {size.keys()}'''					)
         return resize(
             __snake_case    ,      size=(size['''height'''], size['''width'''])    ,      resample=__snake_case    ,      data_format=__snake_case    ,      **__snake_case					)
    def 	a_    ( self    ,      __snake_case    ,      __snake_case    ,      __snake_case = None    ,      **__snake_case    ,      ):
         return rescale(__snake_case    ,      scale=__snake_case    ,      data_format=__snake_case    ,      **__snake_case					)
    def 	a_    ( self    ,      __snake_case    ,      __snake_case    ,      __snake_case    ,      __snake_case = None    ,      **__snake_case    ,      ):
         return normalize(__snake_case    ,      mean=__snake_case    ,      std=__snake_case    ,      data_format=__snake_case    ,      **__snake_case					)




    def 	a_    ( self    ,      __snake_case    ,      __snake_case = None    ,      __snake_case = None    ,      __snake_case = None    ,      __snake_case = None    ,      __snake_case = None    ,      __snake_case = None    ,      __snake_case = None    ,      __snake_case = None    ,      __snake_case = None    ,      __snake_case = None    ,      __snake_case = None    ,      __snake_case = ChannelDimension.FIRST    ,      **__snake_case    ,      ):
         snake_case					=       do_center_crop if do_center_crop is not None else self.do_center_crop
         snake_case					=       crop_size if crop_size is not None else self.crop_size
         snake_case					=       get_size_dict(__snake_case    ,      param_name='''crop_size'''					)
         snake_case					=       do_resize if do_resize is not None else self.do_resize
         snake_case					=       size if size is not None else self.size
         snake_case					=       get_size_dict(__snake_case					)
         snake_case					=       resample if resample is not None else self.resample
         snake_case					=       do_rescale if do_rescale is not None else self.do_rescale
         snake_case					=       rescale_factor if rescale_factor is not None else self.rescale_factor
         snake_case					=       do_normalize if do_normalize is not None else self.do_normalize
         snake_case					=       image_mean if image_mean is not None else self.image_mean
         snake_case					=       image_std if image_std is not None else self.image_std

         snake_case					=       make_list_of_images(__snake_case					)

         if not valid_images(__snake_case					):
              raise ValueError(
                  '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
                  '''torch.Tensor, tf.Tensor or jax.ndarray.'''					)

         if do_center_crop and crop_size is None:
              raise ValueError('''If `do_center_crop` is set to `True`, `crop_size` must be provided.'''					)

         if do_resize and size is None:
              raise ValueError('''Size must be specified if do_resize is True.'''					)

         if do_rescale and rescale_factor is None:
              raise ValueError('''Rescale factor must be specified if do_rescale is True.'''					)

         if do_normalize and (image_mean is None or image_std is None):
              raise ValueError('''Image mean and image standard deviation must be specified if do_normalize is True.'''					)

         # All transformations expect numpy arrays.
         snake_case					=       [to_numpy_array(__snake_case					) for image in images]

         if do_center_crop:
              snake_case					=       [self.center_crop(__snake_case    ,      __snake_case    ,      size=__snake_case					) for image in images]

         if do_resize:
              snake_case					=       [self.resize(image=__snake_case    ,      size=__snake_case    ,      resample=__snake_case					) for image in images]

         if do_rescale:
              snake_case					=       [self.rescale(image=__snake_case    ,      scale=__snake_case					) for image in images]

         if do_normalize:
              snake_case					=       [self.normalize(image=__snake_case    ,      mean=__snake_case    ,      std=__snake_case					) for image in images]

         snake_case					=       [to_channel_dimension_format(__snake_case    ,      __snake_case					) for image in images]

         snake_case					=       {'''pixel_values''': images}
         return BatchFeature(data=__snake_case    ,      tensor_type=__snake_case					)








import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class 						A__      (					snake_case__ ,      snake_case__ ,      unittest.TestCase			):







    """simple docstring"""

    __magic_name__    						=	StableDiffusionParadigmsPipeline
    __magic_name__    						=	TEXT_TO_IMAGE_PARAMS
    __magic_name__    						=	TEXT_TO_IMAGE_BATCH_PARAMS
    __magic_name__    						=	TEXT_TO_IMAGE_IMAGE_PARAMS
    __magic_name__    						=	TEXT_TO_IMAGE_IMAGE_PARAMS
    def 	a_    ( self					):
         torch.manual_seed(0					)
         snake_case					=       UNetaDConditionModel(
             block_out_channels=(3_2, 6_4)    ,      layers_per_block=2    ,      sample_size=3_2    ,      in_channels=4    ,      out_channels=4    ,      down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')    ,      up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')    ,      cross_attention_dim=3_2    ,      attention_head_dim=(2, 4)    ,      use_linear_projection=__snake_case    ,      )
         snake_case					=       DDIMParallelScheduler(
             beta_start=0.0_0085    ,      beta_end=0.012    ,      beta_schedule='''scaled_linear'''    ,      clip_sample=__snake_case    ,      set_alpha_to_one=__snake_case    ,      )
         torch.manual_seed(0					)
         snake_case					=       AutoencoderKL(
             block_out_channels=[3_2, 6_4]    ,      in_channels=3    ,      out_channels=3    ,      down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']    ,      up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']    ,      latent_channels=4    ,      sample_size=1_2_8    ,      )
         torch.manual_seed(0					)
         snake_case					=       CLIPTextConfig(
             bos_token_id=0    ,      eos_token_id=2    ,      hidden_size=3_2    ,      intermediate_size=3_7    ,      layer_norm_eps=1E-05    ,      num_attention_heads=4    ,      num_hidden_layers=5    ,      pad_token_id=1    ,      vocab_size=1_0_0_0    ,      hidden_act='''gelu'''    ,      projection_dim=5_1_2    ,      )
         snake_case					=       CLIPTextModel(__snake_case					)
         snake_case					=       CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

         snake_case					=       {
             '''unet''': unet,
             '''scheduler''': scheduler,
             '''vae''': vae,
             '''text_encoder''': text_encoder,
             '''tokenizer''': tokenizer,
             '''safety_checker''': None,
             '''feature_extractor''': None,
         }
         return components
    def 	a_    ( self    ,      __snake_case    ,      __snake_case=0					):
         if str(__snake_case					).startswith('''mps'''					):
              snake_case					=       torch.manual_seed(__snake_case					)
         else:
              snake_case					=       torch.Generator(device=__snake_case					).manual_seed(__snake_case					)
         snake_case					=       {
             '''prompt''': '''a photograph of an astronaut riding a horse''',
             '''generator''': generator,
             '''num_inference_steps''': 1_0,
             '''guidance_scale''': 6.0,
             '''output_type''': '''numpy''',
             '''parallel''': 3,
             '''debug''': True,
         }
         return inputs
    def 	a_    ( self					):
         snake_case					=       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
         snake_case					=       self.get_dummy_components()
         snake_case					=       StableDiffusionParadigmsPipeline(**__snake_case					)
         snake_case					=       sd_pipe.to(__snake_case					)
         sd_pipe.set_progress_bar_config(disable=__snake_case					)

         snake_case					=       self.get_dummy_inputs(__snake_case					)
         snake_case					=       sd_pipe(**__snake_case					).images
         snake_case					=       image[0, -3:, -3:, -1]
         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case					=       np.array([0.4773, 0.5417, 0.4723, 0.4925, 0.5631, 0.4752, 0.5240, 0.4935, 0.5023]					)

         assert np.abs(image_slice.flatten() - expected_slice					).max() < 1E-2
    def 	a_    ( self					):
         snake_case					=       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
         snake_case					=       self.get_dummy_components()
         torch.manual_seed(0					)
         snake_case					=       DDPMParallelScheduler()
         torch.manual_seed(0					)
         snake_case					=       StableDiffusionParadigmsPipeline(**__snake_case					)
         snake_case					=       sd_pipe.to(__snake_case					)
         sd_pipe.set_progress_bar_config(disable=__snake_case					)

         snake_case					=       self.get_dummy_inputs(__snake_case					)
         snake_case					=       sd_pipe(**__snake_case					).images
         snake_case					=       image[0, -3:, -3:, -1]
         assert image.shape == (1, 6_4, 6_4, 3)

         snake_case					=       np.array([0.3573, 0.4420, 0.4960, 0.4799, 0.3796, 0.3879, 0.4819, 0.4365, 0.4468]					)

         assert np.abs(image_slice.flatten() - expected_slice					).max() < 1E-2
    def 	a_    ( self					):
         super().test_inference_batch_consistent(batch_sizes=[1, 2]					)
    def 	a_    ( self					):
         super().test_inference_batch_single_identical(batch_size=2    ,      expected_max_diff=3E-3					)




    def 	a_    ( self					):
         snake_case					=       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
         snake_case					=       self.get_dummy_components()
         snake_case					=       StableDiffusionParadigmsPipeline(**__snake_case					)
         snake_case					=       sd_pipe.to(__snake_case					)
         sd_pipe.set_progress_bar_config(disable=__snake_case					)

         snake_case					=       self.get_dummy_inputs(__snake_case					)
         snake_case					=       '''french fries'''
         snake_case					=       sd_pipe(**__snake_case    ,      negative_prompt=__snake_case					)
         snake_case					=       output.images
         snake_case					=       image[0, -3:, -3:, -1]

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

         snake_case					=       np.array([0.4771, 0.5420, 0.4683, 0.4918, 0.5636, 0.4725, 0.5230, 0.4923, 0.5015]					)

         assert np.abs(image_slice.flatten() - expected_slice					).max() < 1E-2






@slow
@require_torch_gpu
class 						A__      (					unittest.TestCase			):







    """simple docstring"""

    def 	a_    ( self					):
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()
    def 	a_    ( self    ,      __snake_case=0					):
         snake_case					=       torch.Generator(device=__snake_case					).manual_seed(__snake_case					)
         snake_case					=       {
             '''prompt''': '''a photograph of an astronaut riding a horse''',
             '''generator''': generator,
             '''num_inference_steps''': 1_0,
             '''guidance_scale''': 7.5,
             '''output_type''': '''numpy''',
             '''parallel''': 3,
             '''debug''': True,
         }
         return inputs




    def 	a_    ( self					):
         snake_case					=       '''stabilityai/stable-diffusion-2-base'''
         snake_case					=       DDIMParallelScheduler.from_pretrained(__snake_case    ,      subfolder='''scheduler'''					)
         snake_case					=       StableDiffusionParadigmsPipeline.from_pretrained(__snake_case    ,      scheduler=__snake_case    ,      safety_checker=__snake_case					)
         pipe.to(__snake_case					)
         pipe.set_progress_bar_config(disable=__snake_case					)
         pipe.enable_attention_slicing()

         snake_case					=       self.get_inputs()
         snake_case					=       pipe(**__snake_case					).images
         snake_case					=       image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 5_1_2, 5_1_2, 3)

         snake_case					=       np.array([0.9622, 0.9602, 0.9748, 0.9591, 0.9630, 0.9691, 0.9661, 0.9631, 0.9741]					)

         assert np.abs(expected_slice - image_slice					).max() < 1E-2




