



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_						(						UpperCamelCase_     ,unittest.TestCase				):
							__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 snake_case_  (   self      ):
									return 3_2






							@property
							def snake_case_  (   self      ):
									return 3_2






							@property
							def snake_case_  (   self      ):
									return self.time_input_dim






							@property
							def snake_case_  (   self      ):
									return self.time_input_dim * 4






							@property
							def snake_case_  (   self      ):
									return 1_0_0






							@property
							def snake_case_  (   self      ):
									a_     :      List[str] 						=     CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"      )
									return tokenizer






							@property
							def snake_case_  (   self      ):
									torch.manual_seed(0      )
									a_     :      Any 						=     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=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			,     )
									return CLIPTextModelWithProjection(a_      )






							@property
							def snake_case_  (   self      ):
									torch.manual_seed(0      )

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

									a_     :      Optional[int] 						=     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
									a_     :      Tuple 						=     nn.Parameter(torch.ones(model.clip_std.shape      )      )
									return model






							@property
							def snake_case_  (   self      ):
									torch.manual_seed(0      )
									a_     :      int 						=     CLIPVisionConfig(
									    hidden_size=self.text_embedder_hidden_size			,     image_size=2_2_4			,     projection_dim=self.text_embedder_hidden_size			,     intermediate_size=3_7			,     num_attention_heads=4			,     num_channels=3			,     num_hidden_layers=5			,     patch_size=1_4			,     )

									a_     :      int 						=     CLIPVisionModelWithProjection(a_      )
									return model






							@property
							def snake_case_  (   self      ):
									a_     :      Optional[int] 						=     CLIPImageProcessor(
									    crop_size=2_2_4			,     do_center_crop=a_			,     do_normalize=a_			,     do_resize=a_			,     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=2_2_4			,     )

									return image_processor






							def snake_case_  (   self      ):
									a_     :      List[str] 						=     self.dummy_prior
									a_     :      Tuple 						=     self.dummy_image_encoder
									a_     :      int 						=     self.dummy_text_encoder
									a_     :      Optional[int] 						=     self.dummy_tokenizer
									a_     :      Dict 						=     self.dummy_image_processor

									a_     :      List[Any] 						=     UnCLIPScheduler(
									    variance_type="fixed_small_log"			,     prediction_type="sample"			,     num_train_timesteps=1_0_0_0			,     clip_sample=a_			,     clip_sample_range=10.0			,     )

									a_     :      List[Any] 						=     {
									    '''prior''': prior,
									    '''image_encoder''': image_encoder,
									    '''text_encoder''': text_encoder,
									    '''tokenizer''': tokenizer,
									    '''scheduler''': scheduler,
									    '''image_processor''': image_processor,
									}

									return components






							def snake_case_  (   self			,     a_			,     a_=0      ):
									if str(a_      ).startswith("mps"      ):
											a_     :      List[str] 						=     torch.manual_seed(a_      )
									else:
											a_     :      Optional[int] 						=     torch.Generator(device=a_      ).manual_seed(a_      )
									a_     :      Optional[Any] 						=     {
									    '''prompt''': '''horse''',
									    '''generator''': generator,
									    '''guidance_scale''': 4.0,
									    '''num_inference_steps''': 2,
									    '''output_type''': '''np''',
									}
									return inputs






							def snake_case_  (   self      ):
									a_     :      int 						=     '''cpu'''

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

									a_     :      Dict 						=     self.pipeline_class(**a_      )
									a_     :      Union[str, Any] 						=     pipe.to(a_      )

									pipe.set_progress_bar_config(disable=a_      )

									a_     :      Optional[Any] 						=     pipe(**self.get_dummy_inputs(a_      )      )
									a_     :      Optional[Any] 						=     output.image_embeds

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

									a_     :      Tuple 						=     image[0, -1_0:]
									a_     :      Optional[int] 						=     image_from_tuple[0, -1_0:]

									assert image.shape == (1, 3_2)

									a_     :      str 						=     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 snake_case_  (   self      ):
									a_     :      Union[str, Any] 						=     torch_device == '''cpu'''
									a_     :      List[str] 						=     True
									a_     :      Any 						=     False

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






							@skip_mps
							def snake_case_  (   self      ):
									a_     :      str 						=     torch_device == '''cpu'''
									a_     :      Dict 						=     False

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




"""simple docstring"""
import unittest

import numpy as np
import torch

from diffusers import PNDMPipeline, PNDMScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()




class 			snake_case_						(						unittest.TestCase				):






							@property
							def snake_case_  (   self      ):
									torch.manual_seed(0      )
									a_     :      Any 						=     UNetaDModel(
									    block_out_channels=(3_2, 6_4)			,     layers_per_block=2			,     sample_size=3_2			,     in_channels=3			,     out_channels=3			,     down_block_types=("DownBlock2D", "AttnDownBlock2D")			,     up_block_types=("AttnUpBlock2D", "UpBlock2D")			,     )
									return model






							def snake_case_  (   self      ):
									a_     :      Any 						=     self.dummy_uncond_unet
									a_     :      Any 						=     PNDMScheduler()

									a_     :      List[Any] 						=     PNDMPipeline(unet=a_			,     scheduler=a_      )
									pndm.to(a_      )
									pndm.set_progress_bar_config(disable=a_      )

									a_     :      Tuple 						=     torch.manual_seed(0      )
									a_     :      Union[str, Any] 						=     pndm(generator=a_			,     num_inference_steps=2_0			,     output_type="numpy"      ).images

									a_     :      int 						=     torch.manual_seed(0      )
									a_     :      Optional[int] 						=     pndm(generator=a_			,     num_inference_steps=2_0			,     output_type="numpy"			,     return_dict=a_      )[0]

									a_     :      List[Any] 						=     image[0, -3:, -3:, -1]
									a_     :      Union[str, Any] 						=     image_from_tuple[0, -3:, -3:, -1]

									assert image.shape == (1, 3_2, 3_2, 3)
									a_     :      List[Any] 						=     np.array([1.0, 1.0, 0.0, 1.0, 0.0, 1.0, 0.0, 0.0, 0.0]      )

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




@slow
@require_torch
class 			snake_case_						(						unittest.TestCase				):






							def snake_case_  (   self      ):
									a_     :      Optional[Any] 						=     "google/ddpm-cifar10-32"

									a_     :      Optional[int] 						=     UNetaDModel.from_pretrained(a_      )
									a_     :      Union[str, Any] 						=     PNDMScheduler()

									a_     :      List[Any] 						=     PNDMPipeline(unet=a_			,     scheduler=a_      )
									pndm.to(a_      )
									pndm.set_progress_bar_config(disable=a_      )
									a_     :      Optional[int] 						=     torch.manual_seed(0      )
									a_     :      List[Any] 						=     pndm(generator=a_			,     output_type="numpy"      ).images

									a_     :      str 						=     image[0, -3:, -3:, -1]

									assert image.shape == (1, 3_2, 3_2, 3)
									a_     :      Optional[Any] 						=     np.array([0.1_564, 0.14_645, 0.1_406, 0.14_715, 0.12_425, 0.14_045, 0.13_115, 0.12_175, 0.125]      )

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