



import unittest

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

from diffusers import KandinskyVaaPriorPipeline, PriorTransformer, UnCLIPScheduler
from diffusers.utils import 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_ (lowercase__ , unittest.TestCase  ):




       """simple docstring"""






       _lowerCamelCase			    =				KandinskyVaaPriorPipeline
       _lowerCamelCase			    =				["""prompt"""]
       _lowerCamelCase			    =				["""prompt""", """negative_prompt"""]
       _lowerCamelCase			    =				[
           """num_images_per_prompt""",
           """generator""",
           """num_inference_steps""",
           """latents""",
           """negative_prompt""",
           """guidance_scale""",
           """output_type""",
           """return_dict""",
       ]
       _lowerCamelCase			    =				False



       @property
       def  A_   (     self):

             """simple docstring"""





             return 32



       @property
       def  A_   (     self):

             """simple docstring"""





             return 32



       @property
       def  A_   (     self):

             """simple docstring"""





             return self.time_input_dim



       @property
       def  A_   (     self):

             """simple docstring"""





             return self.time_input_dim * 4



       @property
       def  A_   (     self):

             """simple docstring"""





             return 100



       @property
       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	List[str]				      =		CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")
             return tokenizer



       @property
       def  A_   (     self):

             """simple docstring"""





             torch.manual_seed(0)
             UpperCAmelCase_  :	Tuple				      =		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(lowercase)



       @property
       def  A_   (     self):

             """simple docstring"""





             torch.manual_seed(0)

             UpperCAmelCase_  :	Union[str, Any]				      =		{
                 "num_attention_heads": 2,
                 "attention_head_dim": 12,
                 "embedding_dim": self.text_embedder_hidden_size,
                 "num_layers": 1,
             }

             UpperCAmelCase_  :	List[Any]				      =		PriorTransformer(**lowercase)
             # 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
             UpperCAmelCase_  :	List[Any]				      =		nn.Parameter(torch.ones(model.clip_std.shape))
             return model



       @property
       def  A_   (     self):

             """simple docstring"""





             torch.manual_seed(0)
             UpperCAmelCase_  :	Tuple				      =		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     ,)

             UpperCAmelCase_  :	Tuple				      =		CLIPVisionModelWithProjection(lowercase)
             return model



       @property
       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	Tuple				      =		CLIPImageProcessor(
                 crop_size=224     ,do_center_crop=lowercase     ,do_normalize=lowercase     ,do_resize=lowercase     ,image_mean=[0.4814_5466, 0.457_8275, 0.4082_1073]     ,image_std=[0.2686_2954, 0.2613_0258, 0.2757_7711]     ,resample=3     ,size=224     ,)

             return image_processor



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	List[Any]				      =		self.dummy_prior
             UpperCAmelCase_  :	Dict				      =		self.dummy_image_encoder
             UpperCAmelCase_  :	Optional[Any]				      =		self.dummy_text_encoder
             UpperCAmelCase_  :	Tuple				      =		self.dummy_tokenizer
             UpperCAmelCase_  :	Optional[Any]				      =		self.dummy_image_processor

             UpperCAmelCase_  :	Tuple				      =		UnCLIPScheduler(
                 variance_type="fixed_small_log"     ,prediction_type="sample"     ,num_train_timesteps=1000     ,clip_sample=lowercase     ,clip_sample_range=10.0     ,)

             UpperCAmelCase_  :	int				      =		{
                 "prior": prior,
                 "image_encoder": image_encoder,
                 "text_encoder": text_encoder,
                 "tokenizer": tokenizer,
                 "scheduler": scheduler,
                 "image_processor": image_processor,
             }

             return components



       def  A_   (     self     ,lowercase     ,lowercase=0):

             """simple docstring"""





             if str(lowercase).startswith("mps"):
                   UpperCAmelCase_  :	Any				      =		torch.manual_seed(lowercase)
             else:
                   UpperCAmelCase_  :	List[str]				      =		torch.Generator(device=lowercase).manual_seed(lowercase)
             UpperCAmelCase_  :	Tuple				      =		{
                 "prompt": "horse",
                 "generator": generator,
                 "guidance_scale": 4.0,
                 "num_inference_steps": 2,
                 "output_type": "np",
             }
             return inputs



       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	str				      =		"cpu"

             UpperCAmelCase_  :	List[Any]				      =		self.get_dummy_components()

             UpperCAmelCase_  :	Optional[Any]				      =		self.pipeline_class(**lowercase)
             UpperCAmelCase_  :	Optional[int]				      =		pipe.to(lowercase)

             pipe.set_progress_bar_config(disable=lowercase)

             UpperCAmelCase_  :	Dict				      =		pipe(**self.get_dummy_inputs(lowercase))
             UpperCAmelCase_  :	Optional[int]				      =		output.image_embeds

             UpperCAmelCase_  :	Any				      =		pipe(
                 **self.get_dummy_inputs(lowercase)     ,return_dict=lowercase     ,)[0]

             UpperCAmelCase_  :	Union[str, Any]				      =		image[0, -10:]
             UpperCAmelCase_  :	Tuple				      =		image_from_tuple[0, -10:]

             assert image.shape == (1, 32)

             UpperCAmelCase_  :	str				      =		np.array(
                 [-0.0532, 1.7120, 0.3656, -1.0852, -0.8946, -1.1756, 0.4348, 0.2482, 0.5146, -0.1156])

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

             """simple docstring"""





             UpperCAmelCase_  :	Optional[Any]				      =		torch_device == "cpu"
             UpperCAmelCase_  :	Union[str, Any]				      =		True
             UpperCAmelCase_  :	List[str]				      =		False

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



       @skip_mps
       def  A_   (     self):

             """simple docstring"""





             UpperCAmelCase_  :	List[Any]				      =		torch_device == "cpu"
             UpperCAmelCase_  :	str				      =		False

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





# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
   import torch


class   snake_case_ (TensorFormatter[Mapping, """torch.Tensor""", Mapping]  ):




       """simple docstring"""






       def __init__(     self     ,lowercase=None     ,**lowercase):

             """simple docstring"""





             super().__init__(features=lowercase)
             UpperCAmelCase_  :	Union[str, Any]				      =		torch_tensor_kwargs
             import torch  # noqa import torch at initialization



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             import torch

             if isinstance(lowercase     ,lowercase) and column:
                   if all(
                       isinstance(lowercase     ,torch.Tensor) and x.shape == column[0].shape and x.dtype == column[0].dtype
                       for x in column):
                         return torch.stack(lowercase)
             return column



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             import torch

             if isinstance(lowercase     ,(str, bytes, type(lowercase))):
                   return value
             elif isinstance(lowercase     ,(np.character, np.ndarray)) and np.issubdtype(value.dtype     ,np.character):
                   return value.tolist()

             UpperCAmelCase_  :	List[str]				      =		{}

             if isinstance(lowercase     ,(np.number, np.ndarray)) and np.issubdtype(value.dtype     ,np.integer):
                   UpperCAmelCase_  :	Dict				      =		{"dtype": torch.intaa}
             elif isinstance(lowercase     ,(np.number, np.ndarray)) and np.issubdtype(value.dtype     ,np.floating):
                   UpperCAmelCase_  :	Optional[Any]				      =		{"dtype": torch.floataa}
             elif config.PIL_AVAILABLE and "PIL" in sys.modules:
                   import PIL.Image

                   if isinstance(lowercase     ,PIL.Image.Image):
                         UpperCAmelCase_  :	List[str]				      =		np.asarray(lowercase)
             return torch.tensor(lowercase     ,**{**default_dtype, **self.torch_tensor_kwargs})



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             import torch

             # support for torch, tf, jax etc.
             if hasattr(lowercase     ,"__array__") and not isinstance(lowercase     ,torch.Tensor):
                   UpperCAmelCase_  :	int				      =		data_struct.__array__()
             # support for nested types like struct of list of struct
             if isinstance(lowercase     ,np.ndarray):
                   if data_struct.dtype == object:  # torch tensors cannot be instantied from an array of objects
                         return self._consolidate([self.recursive_tensorize(lowercase) for substruct in data_struct])
             elif isinstance(lowercase     ,(list, tuple)):
                   return self._consolidate([self.recursive_tensorize(lowercase) for substruct in data_struct])
             return self._tensorize(lowercase)



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             return map_nested(self._recursive_tensorize     ,lowercase     ,map_list=lowercase)



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             UpperCAmelCase_  :	List[Any]				      =		self.numpy_arrow_extractor().extract_row(lowercase)
             UpperCAmelCase_  :	Tuple				      =		self.python_features_decoder.decode_row(lowercase)
             return self.recursive_tensorize(lowercase)



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             UpperCAmelCase_  :	str				      =		self.numpy_arrow_extractor().extract_column(lowercase)
             UpperCAmelCase_  :	List[Any]				      =		self.python_features_decoder.decode_column(lowercase     ,pa_table.column_names[0])
             UpperCAmelCase_  :	List[Any]				      =		self.recursive_tensorize(lowercase)
             UpperCAmelCase_  :	List[str]				      =		self._consolidate(lowercase)
             return column



       def  A_   (     self     ,lowercase):

             """simple docstring"""





             UpperCAmelCase_  :	Any				      =		self.numpy_arrow_extractor().extract_batch(lowercase)
             UpperCAmelCase_  :	Optional[int]				      =		self.python_features_decoder.decode_batch(lowercase)
             UpperCAmelCase_  :	List[Any]				      =		self.recursive_tensorize(lowercase)
             for column_name in batch:
                   UpperCAmelCase_  :	Optional[int]				      =		self._consolidate(batch[column_name])
             return batch
