



'''simple docstring'''





import random
import unittest

import numpy as np
import torch
from PIL import Image
from torch import nn
from transformers import (
    CLIPImageProcessor,
    CLIPTextConfig,
    CLIPTextModelWithProjection,
    CLIPTokenizer,
    CLIPVisionConfig,
    CLIPVisionModelWithProjection,
)

from diffusers import KandinskyVaaPriorEmbaEmbPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import floats_tensor, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps

from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()



class       snake_case__		(  _SCREAMING_SNAKE_CASE   ,			unittest.TestCase							):
 A__							=      KandinskyVaaPriorEmbaEmbPipeline
 A__							=      ["prompt", "image"]
 A__							=      ["prompt", "image"]
 A__							=      [
     "num_images_per_prompt",
     "strength",
     "generator",
     "num_inference_steps",
     "latents",
     "negative_prompt",
     "guidance_scale",
     "output_type",
     "return_dict",
 ]
 A__							=      False
 @property
 def A_					(					self		:				Dict   )	->					Optional[Any]:







       '''simple docstring'''
       return 32
 @property
 def A_					(					self		:				int   )	->					Dict:







       '''simple docstring'''
       return 32
 @property
 def A_					(					self		:				Optional[int]   )	->					Dict:







       '''simple docstring'''
       return self.time_input_dim
 @property
 def A_					(					self		:				Optional[Any]   )	->					Dict:







       '''simple docstring'''
       return self.time_input_dim * 4
 @property
 def A_					(					self		:				Dict   )	->					Optional[Any]:







       '''simple docstring'''
       return 100
 @property
 def A_					(					self		:				Optional[int]   )	->					Union[str, Any]:







       '''simple docstring'''
       __snake_case      :					Optional[int]						=  CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'   )
       return tokenizer
 @property
 def A_					(					self		:				Any   )	->					Union[str, Any]:







       '''simple docstring'''
       torch.manual_seed(0   )
       __snake_case      :					str						=  CLIPTextConfig(
           bos_token_id=0	,			eos_token_id=2	,			hidden_size=self.text_embedder_hidden_size	,			projection_dim=self.text_embedder_hidden_size	,			intermediate_size=37	,			layer_norm_eps=1e-05	,			num_attention_heads=4	,			num_hidden_layers=5	,			pad_token_id=1	,			vocab_size=1000	,			)
       return CLIPTextModelWithProjection(__a   )
 @property
 def A_					(					self		:				str   )	->					Union[str, Any]:







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

       __snake_case      :					Tuple						=  {
           'num_attention_heads': 2,
           'attention_head_dim': 12,
           'embedding_dim': self.text_embedder_hidden_size,
           'num_layers': 1,
       }

       __snake_case      :					Any						=  PriorTransformer(**__a   )
       # clip_std and clip_mean is initialized to be 0 so PriorTransformer.post_process_latents will always return 0 - set clip_std to be 1 so it won't return 0
       __snake_case      :					Union[str, Any]						=  nn.Parameter(torch.ones(model.clip_std.shape   )   )
       return model
 @property
 def A_					(					self		:				List[str]   )	->					List[str]:







       '''simple docstring'''
       torch.manual_seed(0   )
       __snake_case      :					Union[str, Any]						=  CLIPVisionConfig(
           hidden_size=self.text_embedder_hidden_size	,			image_size=224	,			projection_dim=self.text_embedder_hidden_size	,			intermediate_size=37	,			num_attention_heads=4	,			num_channels=3	,			num_hidden_layers=5	,			patch_size=14	,			)

       __snake_case      :					Tuple						=  CLIPVisionModelWithProjection(__a   )
       return model
 @property
 def A_					(					self		:				Optional[int]   )	->					Optional[int]:







       '''simple docstring'''
       __snake_case      :					Union[str, Any]						=  CLIPImageProcessor(
           crop_size=224	,			do_center_crop=__a	,			do_normalize=__a	,			do_resize=__a	,			image_mean=[0.4_8_1_4_5_4_6_6, 0.4_5_7_8_2_7_5, 0.4_0_8_2_1_0_7_3]	,			image_std=[0.2_6_8_6_2_9_5_4, 0.2_6_1_3_0_2_5_8, 0.2_7_5_7_7_7_1_1]	,			resample=3	,			size=224	,			)

       return image_processor
 def A_					(					self		:				List[str]   )	->					Dict:







       '''simple docstring'''
       __snake_case      :					str						=  self.dummy_prior
       __snake_case      :					Dict						=  self.dummy_image_encoder
       __snake_case      :					Tuple						=  self.dummy_text_encoder
       __snake_case      :					Union[str, Any]						=  self.dummy_tokenizer
       __snake_case      :					List[str]						=  self.dummy_image_processor

       __snake_case      :					Any						=  UnCLIPScheduler(
           variance_type='fixed_small_log'	,			prediction_type='sample'	,			num_train_timesteps=1000	,			clip_sample=__a	,			clip_sample_range=1_0.0	,			)

       __snake_case      :					List[str]						=  {
           'prior': prior,
           'image_encoder': image_encoder,
           'text_encoder': text_encoder,
           'tokenizer': tokenizer,
           'scheduler': scheduler,
           'image_processor': image_processor,
       }

       return components
 def A_					(					self		:				str	,			__a		:				Dict	,			__a		:				Optional[Any]=0   )	->					Optional[int]:







       '''simple docstring'''
       if str(__a   ).startswith('mps'   ):
             __snake_case      :					Dict						=  torch.manual_seed(__a   )
       else:
             __snake_case      :					List[Any]						=  torch.Generator(device=__a   ).manual_seed(__a   )

       __snake_case      :					Optional[int]						=  floats_tensor((1, 3, 64, 64)	,			rng=random.Random(__a   )   ).to(__a   )
       __snake_case      :					List[Any]						=  image.cpu().permute(0	,			2	,			3	,			1   )[0]
       __snake_case      :					Any						=  Image.fromarray(np.uinta(__a   )   ).convert('RGB'   ).resize((256, 256)   )

       __snake_case      :					List[Any]						=  {
           'prompt': 'horse',
           'image': init_image,
           'strength': 0.5,
           'generator': generator,
           'guidance_scale': 4.0,
           'num_inference_steps': 2,
           'output_type': 'np',
       }
       return inputs
 def A_					(					self		:				str   )	->					List[str]:







       '''simple docstring'''
       __snake_case      :					Dict						=  'cpu'

       __snake_case      :					Union[str, Any]						=  self.get_dummy_components()

       __snake_case      :					List[Any]						=  self.pipeline_class(**__a   )
       __snake_case      :					Any						=  pipe.to(__a   )

       pipe.set_progress_bar_config(disable=__a   )

       __snake_case      :					Union[str, Any]						=  pipe(**self.get_dummy_inputs(__a   )   )
       __snake_case      :					str						=  output.image_embeds

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

       __snake_case      :					Dict						=  image[0, -10:]
       __snake_case      :					Union[str, Any]						=  image_from_tuple[0, -10:]

       assert image.shape == (1, 32)

       __snake_case      :					Any						=  np.array(
           [
               0.1_0_7_1_2_8_4,
               1.3_3_3_0_2_7_1,
               0.6_1_2_6_0_2_2_3,
               -0.6_6_9_1_0_6_5,
               -0.3_8_4_6_8_5_2,
               -1.0_3_0_3_6_6_1,
               0.2_2_7_1_6_1_1_1,
               0.0_3_3_4_8_9_0_1,
               0.3_0_0_4_0_6_7_5,
               -0.2_4_8_0_5_0_2_9,
           ]   )

       assert np.abs(image_slice.flatten() - expected_slice   ).max() < 1e-2
       assert np.abs(image_from_tuple_slice.flatten() - expected_slice   ).max() < 1e-2
 @skip_mps
 def A_					(					self		:				Union[str, Any]   )	->					Any:







       '''simple docstring'''
       __snake_case      :					Tuple						=  torch_device == 'cpu'
       __snake_case      :					Optional[Any]						=  True
       __snake_case      :					Optional[Any]						=  False

       self._test_inference_batch_single_identical(
           test_max_difference=__a	,			relax_max_difference=__a	,			test_mean_pixel_difference=__a	,			)





 @skip_mps
 def A_					(					self		:				List[str]   )	->					Any:







       '''simple docstring'''
       __snake_case      :					Optional[Any]						=  torch_device == 'cpu'
       __snake_case      :					str						=  False

       self._test_attention_slicing_forward_pass(
           test_max_difference=__a	,			test_mean_pixel_difference=__a	,			)









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


a_      						=   logging.get_logger(__name__)

class        _lowerCamelCase  (	_SCREAMING_SNAKE_CASE      ):

      """simple docstring"""

      lowerCAmelCase__	:       Union[str, Any]			=    ["pixel_values"]







      def __init__(   self    :       Union[str, Any]       ,				snake_case    :       bool = True       ,				snake_case    :       Dict[str, int] = None       ,				snake_case    :       bool = True       ,				snake_case    :       Dict[str, int] = None       ,				snake_case    :       PILImageResampling = PILImageResampling.BICUBIC       ,				snake_case    :       bool = True       ,				snake_case    :       Union[int, float] = 1 / 255       ,				snake_case    :       bool = True       ,				snake_case    :       Optional[Union[float, List[float]]] = None       ,				snake_case    :       Optional[Union[float, List[float]]] = None       ,				**snake_case    :       Union[str, Any]       ,				):
          super().__init__(**snake_case )
          __UpperCamelCase     						=  crop_size if crop_size is not None else {'''height''': 256, '''width''': 256}
          __UpperCamelCase     						=  get_size_dict(snake_case       ,				param_name='''crop_size''' )
          __UpperCamelCase     						=  size if size is not None else {'''height''': 224, '''width''': 224}
          __UpperCamelCase     						=  get_size_dict(snake_case )

          __UpperCamelCase     						=  do_center_crop
          __UpperCamelCase     						=  crop_size
          __UpperCamelCase     						=  do_resize
          __UpperCamelCase     						=  size
          __UpperCamelCase     						=  resample
          __UpperCamelCase     						=  do_rescale
          __UpperCamelCase     						=  rescale_factor
          __UpperCamelCase     						=  do_normalize
          __UpperCamelCase     						=  image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
          __UpperCamelCase     						=  image_std if image_std is not None else IMAGENET_DEFAULT_STD







      def 					snake_case		(   self    :       Optional[Any]       ,				snake_case    :       np.ndarray       ,				snake_case    :       Dict[str, int]       ,				snake_case    :       Optional[int] = None       ,				snake_case    :       Optional[Union[str, ChannelDimension]] = None       ,				**snake_case    :       List[Any]       ,				):
          __UpperCamelCase     						=  self.size if size is None else size
          __UpperCamelCase     						=  get_size_dict(snake_case )
          __UpperCamelCase     						=  get_size_dict(snake_case       ,				param_name='''crop_size''' )

          __UpperCamelCase						,					__UpperCamelCase     						=  get_image_size(snake_case )
          __UpperCamelCase     						=  min(snake_case       ,				snake_case )
          __UpperCamelCase     						=  (size['''height'''] / crop_size['''height''']) * min_dim
          __UpperCamelCase     						=  (size['''width'''] / crop_size['''width''']) * min_dim
          return center_crop(snake_case       ,				size=(cropped_height, cropped_width)       ,				data_format=snake_case       ,				**snake_case )







      def 					snake_case		(   self    :       Union[str, Any]       ,				snake_case    :       np.ndarray       ,				snake_case    :       Dict[str, int]       ,				snake_case    :       PILImageResampling = PIL.Image.BICUBIC       ,				snake_case    :       Optional[Union[str, ChannelDimension]] = None       ,				**snake_case    :       List[str]       ,				):
          __UpperCamelCase     						=  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 					snake_case		(   self    :       Union[str, Any]       ,				snake_case    :       np.ndarray       ,				snake_case    :       Union[int, float]       ,				snake_case    :       Optional[Union[str, ChannelDimension]] = None       ,				**snake_case    :       Optional[int]       ,				):
          return rescale(snake_case       ,				scale=snake_case       ,				data_format=snake_case       ,				**snake_case )







      def 					snake_case		(   self    :       Optional[int]       ,				snake_case    :       np.ndarray       ,				snake_case    :       Union[float, List[float]]       ,				snake_case    :       Union[float, List[float]]       ,				snake_case    :       Optional[Union[str, ChannelDimension]] = None       ,				**snake_case    :       Any       ,				):
          return normalize(snake_case       ,				mean=snake_case       ,				std=snake_case       ,				data_format=snake_case       ,				**snake_case )







      def 					snake_case		(   self    :       List[Any]       ,				snake_case    :       ImageInput       ,				snake_case    :       Optional[bool] = None       ,				snake_case    :       Optional[Dict[str, int]] = None       ,				snake_case    :       Optional[bool] = None       ,				snake_case    :       Optional[Dict[str, int]] = None       ,				snake_case    :       PILImageResampling = None       ,				snake_case    :       Optional[bool] = None       ,				snake_case    :       Optional[float] = None       ,				snake_case    :       Optional[bool] = None       ,				snake_case    :       Optional[Union[float, List[float]]] = None       ,				snake_case    :       Optional[Union[float, List[float]]] = None       ,				snake_case    :       Optional[Union[str, TensorType]] = None       ,				snake_case    :       ChannelDimension = ChannelDimension.FIRST       ,				**snake_case    :       Tuple       ,				):
          __UpperCamelCase     						=  do_center_crop if do_center_crop is not None else self.do_center_crop
          __UpperCamelCase     						=  crop_size if crop_size is not None else self.crop_size
          __UpperCamelCase     						=  get_size_dict(snake_case       ,				param_name='''crop_size''' )
          __UpperCamelCase     						=  do_resize if do_resize is not None else self.do_resize
          __UpperCamelCase     						=  size if size is not None else self.size
          __UpperCamelCase     						=  get_size_dict(snake_case )
          __UpperCamelCase     						=  resample if resample is not None else self.resample
          __UpperCamelCase     						=  do_rescale if do_rescale is not None else self.do_rescale
          __UpperCamelCase     						=  rescale_factor if rescale_factor is not None else self.rescale_factor
          __UpperCamelCase     						=  do_normalize if do_normalize is not None else self.do_normalize
          __UpperCamelCase     						=  image_mean if image_mean is not None else self.image_mean
          __UpperCamelCase     						=  image_std if image_std is not None else self.image_std

          __UpperCamelCase     						=  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.
          __UpperCamelCase     						=  [to_numpy_array(snake_case ) for image in images]

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

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

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

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

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

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






