






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      lowercase__     (			_UpperCAmelCase,					unittest.TestCase	):
 a_         =KandinskyVaaPriorPipeline
 a_         =["""prompt"""]
 a_         =["""prompt""", """negative_prompt"""]
 a_         =[
     """num_images_per_prompt""",
     """generator""",
     """num_inference_steps""",
     """latents""",
     """negative_prompt""",
     """guidance_scale""",
     """output_type""",
     """return_dict""",
 ]
 a_         =False

 @property
 def 		UpperCAmelCase  (							self       )->					Tuple:







       '''simple docstring'''





       return 32

 @property
 def 		UpperCAmelCase  (							self       )->					Optional[int]:







       '''simple docstring'''





       return 32

 @property
 def 		UpperCAmelCase  (							self       )->					List[str]:







       '''simple docstring'''





       return self.time_input_dim

 @property
 def 		UpperCAmelCase  (							self       )->					Any:







       '''simple docstring'''





       return self.time_input_dim * 4

 @property
 def 		UpperCAmelCase  (							self       )->					int:







       '''simple docstring'''





       return 100

 @property
 def 		UpperCAmelCase  (							self       )->					List[str]:







       '''simple docstring'''





       lowerCAmelCase__       =						CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"       )
       return tokenizer

 @property
 def 		UpperCAmelCase  (							self       )->					Union[str, Any]:







       '''simple docstring'''





       torch.manual_seed(0       )
       lowerCAmelCase__       =						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-0_5		,    num_attention_heads=4		,    num_hidden_layers=5		,    pad_token_id=1		,    vocab_size=1000		,    )
       return CLIPTextModelWithProjection(__UpperCAmelCase       )

 @property
 def 		UpperCAmelCase  (							self       )->					Optional[Any]:







       '''simple docstring'''





       torch.manual_seed(0       )

       lowerCAmelCase__       =						{
           "num_attention_heads": 2,
           "attention_head_dim": 12,
           "embedding_dim": self.text_embedder_hidden_size,
           "num_layers": 1,
       }

       lowerCAmelCase__       =						PriorTransformer(**__UpperCAmelCase       )
       # 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
       lowerCAmelCase__       =						nn.Parameter(torch.ones(model.clip_std.shape       )       )
       return model

 @property
 def 		UpperCAmelCase  (							self       )->					str:







       '''simple docstring'''





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

       lowerCAmelCase__       =						CLIPVisionModelWithProjection(__UpperCAmelCase       )
       return model

 @property
 def 		UpperCAmelCase  (							self       )->					List[Any]:







       '''simple docstring'''





       lowerCAmelCase__       =						CLIPImageProcessor(
           crop_size=224		,    do_center_crop=__UpperCAmelCase		,    do_normalize=__UpperCAmelCase		,    do_resize=__UpperCAmelCase		,    image_mean=[0.48_145_466, 0.4_578_275, 0.40_821_073]		,    image_std=[0.26_862_954, 0.26_130_258, 0.27_577_711]		,    resample=3		,    size=224		,    )

       return image_processor

 def 		UpperCAmelCase  (							self       )->					int:







       '''simple docstring'''





       lowerCAmelCase__       =						self.dummy_prior
       lowerCAmelCase__       =						self.dummy_image_encoder
       lowerCAmelCase__       =						self.dummy_text_encoder
       lowerCAmelCase__       =						self.dummy_tokenizer
       lowerCAmelCase__       =						self.dummy_image_processor

       lowerCAmelCase__       =						UnCLIPScheduler(
           variance_type="fixed_small_log"		,    prediction_type="sample"		,    num_train_timesteps=1000		,    clip_sample=__UpperCAmelCase		,    clip_sample_range=10.0		,    )

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

       return components

 def 		UpperCAmelCase  (							self		,    __UpperCAmelCase		,    __UpperCAmelCase=0       )->					List[str]:







       '''simple docstring'''





       if str(__UpperCAmelCase       ).startswith("mps"       ):
             lowerCAmelCase__       =						torch.manual_seed(__UpperCAmelCase       )
       else:
             lowerCAmelCase__       =						torch.Generator(device=__UpperCAmelCase       ).manual_seed(__UpperCAmelCase       )
       lowerCAmelCase__       =						{
           "prompt": "horse",
           "generator": generator,
           "guidance_scale": 4.0,
           "num_inference_steps": 2,
           "output_type": "np",
       }
       return inputs

 def 		UpperCAmelCase  (							self       )->					List[Any]:







       '''simple docstring'''





       lowerCAmelCase__       =						"cpu"

       lowerCAmelCase__       =						self.get_dummy_components()

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

       pipe.set_progress_bar_config(disable=__UpperCAmelCase       )

       lowerCAmelCase__       =						pipe(**self.get_dummy_inputs(__UpperCAmelCase       )       )
       lowerCAmelCase__       =						output.image_embeds

       lowerCAmelCase__       =						pipe(
           **self.get_dummy_inputs(__UpperCAmelCase       )		,    return_dict=__UpperCAmelCase		,    )[0]

       lowerCAmelCase__       =						image[0, -10:]
       lowerCAmelCase__       =						image_from_tuple[0, -10:]

       assert image.shape == (1, 32)

       lowerCAmelCase__       =						np.array(
           [-0.0_532, 1.7_120, 0.3_656, -1.0_852, -0.8_946, -1.1_756, 0.4_348, 0.2_482, 0.5_146, -0.1_156]       )

       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 		UpperCAmelCase  (							self       )->					Any:







       '''simple docstring'''





       lowerCAmelCase__       =						torch_device == "cpu"
       lowerCAmelCase__       =						True
       lowerCAmelCase__       =						False

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


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







       '''simple docstring'''





       lowerCAmelCase__       =						torch_device == "cpu"
       lowerCAmelCase__       =						False

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














import flax.linen as nn
import jax.numpy as jnp

from .attention_flax import FlaxTransformeraDModel
from .resnet_flax import FlaxDownsampleaD, FlaxResnetBlockaD, FlaxUpsampleaD


class      lowercase__     (			nn.Module	):
	a_         =42
	a_         =42
	a_         =0.0
	a_         =1
	a_         =1
	a_         =True
	a_         =False
	a_         =False
	a_         =False
	a_         =jnp.floataa

	def 		UpperCAmelCase  (							self       )->					Optional[Any]:







							'''simple docstring'''





							lowerCAmelCase__       =						[]
							lowerCAmelCase__       =						[]

							for i in range(self.num_layers       ):
													lowerCAmelCase__       =						self.in_channels if i == 0 else self.out_channels

													lowerCAmelCase__       =						FlaxResnetBlockaD(
													    in_channels=__UpperCAmelCase		,    out_channels=self.out_channels		,    dropout_prob=self.dropout		,    dtype=self.dtype		,    )
													resnets.append(__UpperCAmelCase       )

													lowerCAmelCase__       =						FlaxTransformeraDModel(
													    in_channels=self.out_channels		,    n_heads=self.num_attention_heads		,    d_head=self.out_channels // self.num_attention_heads		,    depth=1		,    use_linear_projection=self.use_linear_projection		,    only_cross_attention=self.only_cross_attention		,    use_memory_efficient_attention=self.use_memory_efficient_attention		,    dtype=self.dtype		,    )
													attentions.append(__UpperCAmelCase       )

							lowerCAmelCase__       =						resnets
							lowerCAmelCase__       =						attentions

							if self.add_downsample:
													lowerCAmelCase__       =						FlaxDownsampleaD(self.out_channels		,    dtype=self.dtype       )


	def __call__(							self		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase=True       )->					str:







							'''simple docstring'''





							lowerCAmelCase__       =						()

							for resnet, attn in zip(self.resnets		,    self.attentions       ):
													lowerCAmelCase__       =						resnet(__UpperCAmelCase		,    __UpperCAmelCase		,    deterministic=__UpperCAmelCase       )
													lowerCAmelCase__       =						attn(__UpperCAmelCase		,    __UpperCAmelCase		,    deterministic=__UpperCAmelCase       )
													output_states += (hidden_states,)

							if self.add_downsample:
													lowerCAmelCase__       =						self.downsamplers_a(__UpperCAmelCase       )
													output_states += (hidden_states,)

							return hidden_states, output_states





class      lowercase__     (			nn.Module	):
	a_         =42
	a_         =42
	a_         =0.0
	a_         =1
	a_         =True
	a_         =jnp.floataa

	def 		UpperCAmelCase  (							self       )->					Dict:







							'''simple docstring'''





							lowerCAmelCase__       =						[]

							for i in range(self.num_layers       ):
													lowerCAmelCase__       =						self.in_channels if i == 0 else self.out_channels

													lowerCAmelCase__       =						FlaxResnetBlockaD(
													    in_channels=__UpperCAmelCase		,    out_channels=self.out_channels		,    dropout_prob=self.dropout		,    dtype=self.dtype		,    )
													resnets.append(__UpperCAmelCase       )
							lowerCAmelCase__       =						resnets

							if self.add_downsample:
													lowerCAmelCase__       =						FlaxDownsampleaD(self.out_channels		,    dtype=self.dtype       )


	def __call__(							self		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase=True       )->					int:







							'''simple docstring'''





							lowerCAmelCase__       =						()

							for resnet in self.resnets:
													lowerCAmelCase__       =						resnet(__UpperCAmelCase		,    __UpperCAmelCase		,    deterministic=__UpperCAmelCase       )
													output_states += (hidden_states,)

							if self.add_downsample:
													lowerCAmelCase__       =						self.downsamplers_a(__UpperCAmelCase       )
													output_states += (hidden_states,)

							return hidden_states, output_states





class      lowercase__     (			nn.Module	):
	a_         =42
	a_         =42
	a_         =42
	a_         =0.0
	a_         =1
	a_         =1
	a_         =True
	a_         =False
	a_         =False
	a_         =False
	a_         =jnp.floataa

	def 		UpperCAmelCase  (							self       )->					int:







							'''simple docstring'''





							lowerCAmelCase__       =						[]
							lowerCAmelCase__       =						[]

							for i in range(self.num_layers       ):
													lowerCAmelCase__       =						self.in_channels if (i == self.num_layers - 1) else self.out_channels
													lowerCAmelCase__       =						self.prev_output_channel if i == 0 else self.out_channels

													lowerCAmelCase__       =						FlaxResnetBlockaD(
													    in_channels=resnet_in_channels + res_skip_channels		,    out_channels=self.out_channels		,    dropout_prob=self.dropout		,    dtype=self.dtype		,    )
													resnets.append(__UpperCAmelCase       )

													lowerCAmelCase__       =						FlaxTransformeraDModel(
													    in_channels=self.out_channels		,    n_heads=self.num_attention_heads		,    d_head=self.out_channels // self.num_attention_heads		,    depth=1		,    use_linear_projection=self.use_linear_projection		,    only_cross_attention=self.only_cross_attention		,    use_memory_efficient_attention=self.use_memory_efficient_attention		,    dtype=self.dtype		,    )
													attentions.append(__UpperCAmelCase       )

							lowerCAmelCase__       =						resnets
							lowerCAmelCase__       =						attentions

							if self.add_upsample:
													lowerCAmelCase__       =						FlaxUpsampleaD(self.out_channels		,    dtype=self.dtype       )


	def __call__(							self		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase=True       )->					Dict:







							'''simple docstring'''





							for resnet, attn in zip(self.resnets		,    self.attentions       ):
													# pop res hidden states
													lowerCAmelCase__       =						res_hidden_states_tuple[-1]
													lowerCAmelCase__       =						res_hidden_states_tuple[:-1]
													lowerCAmelCase__       =						jnp.concatenate((hidden_states, res_hidden_states)		,    axis=-1       )

													lowerCAmelCase__       =						resnet(__UpperCAmelCase		,    __UpperCAmelCase		,    deterministic=__UpperCAmelCase       )
													lowerCAmelCase__       =						attn(__UpperCAmelCase		,    __UpperCAmelCase		,    deterministic=__UpperCAmelCase       )

							if self.add_upsample:
													lowerCAmelCase__       =						self.upsamplers_a(__UpperCAmelCase       )

							return hidden_states





class      lowercase__     (			nn.Module	):
	a_         =42
	a_         =42
	a_         =42
	a_         =0.0
	a_         =1
	a_         =True
	a_         =jnp.floataa

	def 		UpperCAmelCase  (							self       )->					int:







							'''simple docstring'''





							lowerCAmelCase__       =						[]

							for i in range(self.num_layers       ):
													lowerCAmelCase__       =						self.in_channels if (i == self.num_layers - 1) else self.out_channels
													lowerCAmelCase__       =						self.prev_output_channel if i == 0 else self.out_channels

													lowerCAmelCase__       =						FlaxResnetBlockaD(
													    in_channels=resnet_in_channels + res_skip_channels		,    out_channels=self.out_channels		,    dropout_prob=self.dropout		,    dtype=self.dtype		,    )
													resnets.append(__UpperCAmelCase       )

							lowerCAmelCase__       =						resnets

							if self.add_upsample:
													lowerCAmelCase__       =						FlaxUpsampleaD(self.out_channels		,    dtype=self.dtype       )


	def __call__(							self		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase=True       )->					int:







							'''simple docstring'''





							for resnet in self.resnets:
													# pop res hidden states
													lowerCAmelCase__       =						res_hidden_states_tuple[-1]
													lowerCAmelCase__       =						res_hidden_states_tuple[:-1]
													lowerCAmelCase__       =						jnp.concatenate((hidden_states, res_hidden_states)		,    axis=-1       )

													lowerCAmelCase__       =						resnet(__UpperCAmelCase		,    __UpperCAmelCase		,    deterministic=__UpperCAmelCase       )

							if self.add_upsample:
													lowerCAmelCase__       =						self.upsamplers_a(__UpperCAmelCase       )

							return hidden_states





class      lowercase__     (			nn.Module	):
	a_         =42
	a_         =0.0
	a_         =1
	a_         =1
	a_         =False
	a_         =False
	a_         =jnp.floataa

	def 		UpperCAmelCase  (							self       )->					Union[str, Any]:







							'''simple docstring'''





							lowerCAmelCase__       =						[
							    FlaxResnetBlockaD(
							        in_channels=self.in_channels		,    out_channels=self.in_channels		,    dropout_prob=self.dropout		,    dtype=self.dtype		,    )
							]

							lowerCAmelCase__       =						[]

							for _ in range(self.num_layers       ):
													lowerCAmelCase__       =						FlaxTransformeraDModel(
													    in_channels=self.in_channels		,    n_heads=self.num_attention_heads		,    d_head=self.in_channels // self.num_attention_heads		,    depth=1		,    use_linear_projection=self.use_linear_projection		,    use_memory_efficient_attention=self.use_memory_efficient_attention		,    dtype=self.dtype		,    )
													attentions.append(__UpperCAmelCase       )

													lowerCAmelCase__       =						FlaxResnetBlockaD(
													    in_channels=self.in_channels		,    out_channels=self.in_channels		,    dropout_prob=self.dropout		,    dtype=self.dtype		,    )
													resnets.append(__UpperCAmelCase       )

							lowerCAmelCase__       =						resnets
							lowerCAmelCase__       =						attentions


	def __call__(							self		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase		,    __UpperCAmelCase=True       )->					Union[str, Any]:







							'''simple docstring'''





							lowerCAmelCase__       =						self.resnets[0](__UpperCAmelCase		,    __UpperCAmelCase       )
							for attn, resnet in zip(self.attentions		,    self.resnets[1:]       ):
													lowerCAmelCase__       =						attn(__UpperCAmelCase		,    __UpperCAmelCase		,    deterministic=__UpperCAmelCase       )
													lowerCAmelCase__       =						resnet(__UpperCAmelCase		,    __UpperCAmelCase		,    deterministic=__UpperCAmelCase       )

							return hidden_states






