




import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModel

from diffusers import HeunDiscreteScheduler, PriorTransformer, ShapEImgaImgPipeline
from diffusers.pipelines.shap_e import ShapERenderer
from diffusers.utils import floats_tensor, load_image, load_numpy, slow
from diffusers.utils.testing_utils import require_torch_gpu, torch_device

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference







class 						lowercase							(   UpperCamelCase__,unittest.TestCase		):
     _a							 =						ShapEImgaImgPipeline
     _a							 =						["image"]
     _a							 =						["image"]
     _a							 =						[
         "num_images_per_prompt",
         "num_inference_steps",
         "generator",
         "latents",
         "guidance_scale",
         "frame_size",
         "output_type",
         "return_dict",
     ]
     _a							 =						False

     @property
     def 					a__	(  self				)						->							Optional[int]:
           return 32

     @property
     def 					a__	(  self				)						->							Tuple:
           return 32

     @property
     def 					a__	(  self				)						->							Optional[Any]:
           return self.time_input_dim * 4

     @property
     def 					a__	(  self				)						->							Optional[int]:
           return 8

     @property
     def 					a__	(  self				)						->							List[Any]:
           torch.manual_seed(0				)
           _A							:     Union[str, Any]			=						CLIPVisionConfig(
               hidden_size=self.text_embedder_hidden_size		,     image_size=64		,     projection_dim=self.text_embedder_hidden_size		,     intermediate_size=37		,     num_attention_heads=4		,     num_channels=3		,     num_hidden_layers=5		,     patch_size=1		,     )

           _A							:     List[Any]			=						CLIPVisionModel(_a				)
           return model

     @property
     def 					a__	(  self				)						->							Dict:
           _A							:     int			=						CLIPImageProcessor(
               crop_size=224		,     do_center_crop=_a		,     do_normalize=_a		,     do_resize=_a		,     image_mean=[0.48145466, 0.4578275, 0.40821073]		,     image_std=[0.26862954, 0.26130258, 0.27577711]		,     resample=3		,     size=224		,     )

           return image_processor

     @property
     def 					a__	(  self				)						->							Union[str, Any]:
           torch.manual_seed(0				)

           _A							:     str			=						{
               """num_attention_heads""": 2,
               """attention_head_dim""": 16,
               """embedding_dim""": self.time_input_dim,
               """num_embeddings""": 32,
               """embedding_proj_dim""": self.text_embedder_hidden_size,
               """time_embed_dim""": self.time_embed_dim,
               """num_layers""": 1,
               """clip_embed_dim""": self.time_input_dim * 2,
               """additional_embeddings""": 0,
               """time_embed_act_fn""": """gelu""",
               """norm_in_type""": """layer""",
               """embedding_proj_norm_type""": """layer""",
               """encoder_hid_proj_type""": None,
               """added_emb_type""": None,
           }

           _A							:     Optional[int]			=						PriorTransformer(**_a				)
           return model

     @property
     def 					a__	(  self				)						->							Union[str, Any]:
           torch.manual_seed(0				)

           _A							:     List[str]			=						{
               """param_shapes""": (
                   (self.renderer_dim, 93),
                   (self.renderer_dim, 8),
                   (self.renderer_dim, 8),
                   (self.renderer_dim, 8),
               ),
               """d_latent""": self.time_input_dim,
               """d_hidden""": self.renderer_dim,
               """n_output""": 12,
               """background""": (
                   0.1,
                   0.1,
                   0.1,
               ),
           }
           _A							:     int			=						ShapERenderer(**_a				)
           return model

     def 					a__	(  self				)						->							int:
           _A							:     Optional[Any]			=						self.dummy_prior
           _A							:     Tuple			=						self.dummy_image_encoder
           _A							:     str			=						self.dummy_image_processor
           _A							:     Any			=						self.dummy_renderer

           _A							:     str			=						HeunDiscreteScheduler(
               beta_schedule="""exp"""		,     num_train_timesteps=1024		,     prediction_type="""sample"""		,     use_karras_sigmas=_a		,     clip_sample=_a		,     clip_sample_range=1.0		,     )
           _A							:     Union[str, Any]			=						{
               """prior""": prior,
               """image_encoder""": image_encoder,
               """image_processor""": image_processor,
               """renderer""": renderer,
               """scheduler""": scheduler,
           }

           return components

     def 					a__	(  self		,     _a		,     _a=0				)						->							int:
           _A							:     str			=						floats_tensor((1, 3, 64, 64)		,     rng=random.Random(_a				)				).to(_a				)

           if str(_a				).startswith("""mps"""				):
                 _A							:     List[Any]			=						torch.manual_seed(_a				)
           else:
                 _A							:     Tuple			=						torch.Generator(device=_a				).manual_seed(_a				)
           _A							:     List[Any]			=						{
               """image""": input_image,
               """generator""": generator,
               """num_inference_steps""": 1,
               """frame_size""": 32,
               """output_type""": """np""",
           }
           return inputs

     def 					a__	(  self				)						->							Tuple:
           _A							:     List[Any]			=						"""cpu"""

           _A							:     List[str]			=						self.get_dummy_components()

           _A							:     str			=						self.pipeline_class(**_a				)
           _A							:     Optional[int]			=						pipe.to(_a				)

           pipe.set_progress_bar_config(disable=_a				)

           _A							:     Tuple			=						pipe(**self.get_dummy_inputs(_a				)				)
           _A							:     Tuple			=						output.images[0]
           _A							:     Dict			=						image[0, -3:, -3:, -1]

           assert image.shape == (20, 32, 32, 3)

           _A							:     Dict			=						np.array(
               [
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
                   0.00039216,
               ]				)

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

     def 					a__	(  self				)						->							Tuple:
           # NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
           self._test_inference_batch_consistent(batch_sizes=[1, 2]				)

     def 					a__	(  self				)						->							Tuple:
           _A							:     List[str]			=						torch_device == """cpu"""
           _A							:     Union[str, Any]			=						True
           self._test_inference_batch_single_identical(
               batch_size=2		,     test_max_difference=_a		,     relax_max_difference=_a		,     )



     def 					a__	(  self				)						->							Tuple:
           _A							:     int			=						self.get_dummy_components()
           _A							:     Optional[int]			=						self.pipeline_class(**_a				)
           _A							:     int			=						pipe.to(_a				)
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     int			=						1
           _A							:     List[str]			=						2

           _A							:     str			=						self.get_dummy_inputs(_a				)

           for key in inputs.keys():
                 if key in self.batch_params:
                       _A							:     str			=						batch_size * [inputs[key]]

           _A							:     Union[str, Any]			=						pipe(**_a		,     num_images_per_prompt=_a				)[0]

           assert images.shape[0] == batch_size * num_images_per_prompt







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

     def 					a__	(  self				)						->							int:
           # clean up the VRAM after each test
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()



     def 					a__	(  self				)						->							Dict:
           _A							:     List[str]			=						load_image(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main""" """/shap_e/corgi.png"""				)
           _A							:     str			=						load_numpy(
               """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
               """/shap_e/test_shap_e_img2img_out.npy"""				)
           _A							:     List[Any]			=						ShapEImgaImgPipeline.from_pretrained("""openai/shap-e-img2img"""				)
           _A							:     Union[str, Any]			=						pipe.to(_a				)
           pipe.set_progress_bar_config(disable=_a				)

           _A							:     Tuple			=						torch.Generator(device=_a				).manual_seed(0				)

           _A							:     Optional[Any]			=						pipe(
               _a		,     generator=_a		,     guidance_scale=3.0		,     num_inference_steps=64		,     frame_size=64		,     output_type="""np"""		,     ).images[0]

           assert images.shape == (20, 64, 64, 3)

           assert_mean_pixel_difference(_a		,     _a				)







import torch

from diffusers import DPMSolverSDEScheduler
from diffusers.utils import torch_device
from diffusers.utils.testing_utils import require_torchsde

from .test_schedulers import SchedulerCommonTest







@require_torchsde
class 						lowercase							(   UpperCamelCase__		):
     _a							 =						(DPMSolverSDEScheduler,)
     _a							 =						1_0

     def 					a__	(  self		,     **_a				)						->							Optional[Any]:
           _A							:     str			=						{
               """num_train_timesteps""": 1100,
               """beta_start""": 0.0001,
               """beta_end""": 0.02,
               """beta_schedule""": """linear""",
               """noise_sampler_seed""": 0,
           }

           config.update(**_a				)
           return config

     def 					a__	(  self				)						->							Tuple:
           for timesteps in [10, 50, 100, 1000]:
                 self.check_over_configs(num_train_timesteps=_a				)

     def 					a__	(  self				)						->							Optional[int]:
           for beta_start, beta_end in zip([0.00001, 0.0001, 0.001]		,     [0.0002, 0.002, 0.02]				):
                 self.check_over_configs(beta_start=_a		,     beta_end=_a				)

     def 					a__	(  self				)						->							Any:
           for schedule in ["linear", "scaled_linear"]:
                 self.check_over_configs(beta_schedule=_a				)

     def 					a__	(  self				)						->							Optional[int]:
           for prediction_type in ["epsilon", "v_prediction"]:
                 self.check_over_configs(prediction_type=_a				)

     def 					a__	(  self				)						->							Optional[int]:
           _A							:     Any			=						self.scheduler_classes[0]
           _A							:     List[str]			=						self.get_scheduler_config()
           _A							:     Optional[Any]			=						scheduler_class(**_a				)

           scheduler.set_timesteps(self.num_inference_steps				)

           _A							:     Dict			=						self.dummy_model()
           _A							:     Any			=						self.dummy_sample_deter * scheduler.init_noise_sigma
           _A							:     Dict			=						sample.to(_a				)

           for i, t in enumerate(scheduler.timesteps				):
                 _A							:     Optional[int]			=						scheduler.scale_model_input(_a		,     _a				)

                 _A							:     str			=						model(_a		,     _a				)

                 _A							:     List[Any]			=						scheduler.step(_a		,     _a		,     _a				)
                 _A							:     Optional[int]			=						output.prev_sample

           _A							:     Dict			=						torch.sum(torch.abs(_a				)				)
           _A							:     Dict			=						torch.mean(torch.abs(_a				)				)

           if torch_device in ["mps"]:
                 assert abs(result_sum.item() - 167.47821044921875				) < 1e-2
                 assert abs(result_mean.item() - 0.2178705964565277				) < 1e-3
           elif torch_device in ["cuda"]:
                 assert abs(result_sum.item() - 171.59352111816406				) < 1e-2
                 assert abs(result_mean.item() - 0.22342906892299652				) < 1e-3
           else:
                 assert abs(result_sum.item() - 162.52383422851562				) < 1e-2
                 assert abs(result_mean.item() - 0.211619570851326				) < 1e-3

     def 					a__	(  self				)						->							Optional[Any]:
           _A							:     Dict			=						self.scheduler_classes[0]
           _A							:     Optional[int]			=						self.get_scheduler_config(prediction_type="""v_prediction"""				)
           _A							:     Optional[Any]			=						scheduler_class(**_a				)

           scheduler.set_timesteps(self.num_inference_steps				)

           _A							:     Tuple			=						self.dummy_model()
           _A							:     int			=						self.dummy_sample_deter * scheduler.init_noise_sigma
           _A							:     Tuple			=						sample.to(_a				)

           for i, t in enumerate(scheduler.timesteps				):
                 _A							:     int			=						scheduler.scale_model_input(_a		,     _a				)

                 _A							:     Tuple			=						model(_a		,     _a				)

                 _A							:     Dict			=						scheduler.step(_a		,     _a		,     _a				)
                 _A							:     Optional[int]			=						output.prev_sample

           _A							:     Optional[Any]			=						torch.sum(torch.abs(_a				)				)
           _A							:     List[Any]			=						torch.mean(torch.abs(_a				)				)

           if torch_device in ["mps"]:
                 assert abs(result_sum.item() - 124.77149200439453				) < 1e-2
                 assert abs(result_mean.item() - 0.16226289014816284				) < 1e-3
           elif torch_device in ["cuda"]:
                 assert abs(result_sum.item() - 128.1663360595703				) < 1e-2
                 assert abs(result_mean.item() - 0.16688326001167297				) < 1e-3
           else:
                 assert abs(result_sum.item() - 119.8487548828125				) < 1e-2
                 assert abs(result_mean.item() - 0.1560530662536621				) < 1e-3

     def 					a__	(  self				)						->							List[str]:
           _A							:     Union[str, Any]			=						self.scheduler_classes[0]
           _A							:     List[Any]			=						self.get_scheduler_config()
           _A							:     List[str]			=						scheduler_class(**_a				)

           scheduler.set_timesteps(self.num_inference_steps		,     device=_a				)

           _A							:     Union[str, Any]			=						self.dummy_model()
           _A							:     Optional[Any]			=						self.dummy_sample_deter.to(_a				) * scheduler.init_noise_sigma

           for t in scheduler.timesteps:
                 _A							:     int			=						scheduler.scale_model_input(_a		,     _a				)

                 _A							:     List[Any]			=						model(_a		,     _a				)

                 _A							:     Dict			=						scheduler.step(_a		,     _a		,     _a				)
                 _A							:     Dict			=						output.prev_sample

           _A							:     str			=						torch.sum(torch.abs(_a				)				)
           _A							:     str			=						torch.mean(torch.abs(_a				)				)

           if torch_device in ["mps"]:
                 assert abs(result_sum.item() - 167.46957397460938				) < 1e-2
                 assert abs(result_mean.item() - 0.21805934607982635				) < 1e-3
           elif torch_device in ["cuda"]:
                 assert abs(result_sum.item() - 171.59353637695312				) < 1e-2
                 assert abs(result_mean.item() - 0.22342908382415771				) < 1e-3
           else:
                 assert abs(result_sum.item() - 162.52383422851562				) < 1e-2
                 assert abs(result_mean.item() - 0.211619570851326				) < 1e-3



     def 					a__	(  self				)						->							Union[str, Any]:
           _A							:     List[Any]			=						self.scheduler_classes[0]
           _A							:     Optional[Any]			=						self.get_scheduler_config()
           _A							:     int			=						scheduler_class(**_a		,     use_karras_sigmas=_a				)

           scheduler.set_timesteps(self.num_inference_steps		,     device=_a				)

           _A							:     Optional[Any]			=						self.dummy_model()
           _A							:     Dict			=						self.dummy_sample_deter.to(_a				) * scheduler.init_noise_sigma
           _A							:     str			=						sample.to(_a				)

           for t in scheduler.timesteps:
                 _A							:     Optional[int]			=						scheduler.scale_model_input(_a		,     _a				)

                 _A							:     List[Any]			=						model(_a		,     _a				)

                 _A							:     Dict			=						scheduler.step(_a		,     _a		,     _a				)
                 _A							:     List[str]			=						output.prev_sample

           _A							:     str			=						torch.sum(torch.abs(_a				)				)
           _A							:     List[str]			=						torch.mean(torch.abs(_a				)				)

           if torch_device in ["mps"]:
                 assert abs(result_sum.item() - 176.66974135742188				) < 1e-2
                 assert abs(result_mean.item() - 0.23003872730981811				) < 1e-2
           elif torch_device in ["cuda"]:
                 assert abs(result_sum.item() - 177.63653564453125				) < 1e-2
                 assert abs(result_mean.item() - 0.23003872730981811				) < 1e-2
           else:
                 assert abs(result_sum.item() - 170.3135223388672				) < 1e-2
                 assert abs(result_mean.item() - 0.23003872730981811				) < 1e-2

