





"""simple docstring"""




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 			__UpperCamelCase							(		a__						,			unittest.TestCase		):
							lowerCamelCase  :   Optional[Any]    =ShapEImgaImgPipeline
							lowerCamelCase  :   Union[str, Any]    =["""image"""]
							lowerCamelCase  :   Tuple    =["""image"""]
							lowerCamelCase  :   Optional[int]    =[
							    """num_images_per_prompt""",
							    """num_inference_steps""",
							    """generator""",
							    """latents""",
							    """guidance_scale""",
							    """frame_size""",
							    """output_type""",
							    """return_dict""",
							]
							lowerCamelCase  :   Tuple    =False






							@property
							def        __a  (  self )      ->					List[str]:
								return 32






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






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






							@property
							def        __a  (  self )      ->					Optional[Any]:
								return 8






							@property
							def        __a  (  self )      ->					Dict:
								torch.manual_seed(0 )
								a :						List[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 :						str      =							CLIPVisionModel(lowerCAmelCase__ )
								return model






							@property
							def        __a  (  self )      ->					int:
								a :						Any      =							CLIPImageProcessor(
								    crop_size=224 ,   do_center_crop=lowerCAmelCase__ ,   do_normalize=lowerCAmelCase__ ,   do_resize=lowerCAmelCase__ ,   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






							@property
							def        __a  (  self )      ->					List[Any]:
								torch.manual_seed(0 )

								a :						Optional[Any]      =							{
								    "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 :						List[str]      =							PriorTransformer(**lowerCAmelCase__ )
								return model






							@property
							def        __a  (  self )      ->					Optional[Any]:
								torch.manual_seed(0 )

								a :						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 :						str      =							ShapERenderer(**lowerCAmelCase__ )
								return model






							def        __a  (  self )      ->					Optional[Any]:
								a :						Any      =							self.dummy_prior
								a :						List[str]      =							self.dummy_image_encoder
								a :						Dict      =							self.dummy_image_processor
								a :						List[str]      =							self.dummy_renderer

								a :						Optional[Any]      =							HeunDiscreteScheduler(
								    beta_schedule="exp" ,   num_train_timesteps=1024 ,   prediction_type="sample" ,   use_karras_sigmas=lowerCAmelCase__ ,   clip_sample=lowerCAmelCase__ ,   clip_sample_range=1.0 ,   )
								a :						Dict      =							{
								    "prior": prior,
								    "image_encoder": image_encoder,
								    "image_processor": image_processor,
								    "renderer": renderer,
								    "scheduler": scheduler,
								}

								return components






							def        __a  (  self ,   lowerCAmelCase__ ,   lowerCAmelCase__=0 )      ->					str:
								a :						List[str]      =							floats_tensor((1, 3, 64, 64) ,   rng=random.Random(lowerCAmelCase__ ) ).to(lowerCAmelCase__ )

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






							def        __a  (  self )      ->					List[str]:
								a :						Optional[Any]      =							"cpu"

								a :						int      =							self.get_dummy_components()

								a :						Tuple      =							self.pipeline_class(**lowerCAmelCase__ )
								a :						List[str]      =							pipe.to(lowerCAmelCase__ )

								pipe.set_progress_bar_config(disable=lowerCAmelCase__ )

								a :						Tuple      =							pipe(**self.get_dummy_inputs(lowerCAmelCase__ ) )
								a :						Dict      =							output.images[0]
								a :						Optional[int]      =							image[0, -3:, -3:, -1]

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

								a :						Any      =							np.array(
								    [
								        0.00_039_216,
								        0.00_039_216,
								        0.00_039_216,
								        0.00_039_216,
								        0.00_039_216,
								        0.00_039_216,
								        0.00_039_216,
								        0.00_039_216,
								        0.00_039_216,
								    ] )

								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 )      ->					str:
								a :						Dict      =							torch_device == "cpu"
								a :						Optional[Any]      =							True
								self._test_inference_batch_single_identical(
								    batch_size=2 ,   test_max_difference=lowerCAmelCase__ ,   relax_max_difference=lowerCAmelCase__ ,   )






							def        __a  (  self )      ->					Any:
								a :						List[str]      =							self.get_dummy_components()
								a :						Any      =							self.pipeline_class(**lowerCAmelCase__ )
								a :						Union[str, Any]      =							pipe.to(lowerCAmelCase__ )
								pipe.set_progress_bar_config(disable=lowerCAmelCase__ )

								a :						int      =							1
								a :						Dict      =							2

								a :						Any      =							self.get_dummy_inputs(lowerCAmelCase__ )

								for key in inputs.keys():
									if key in self.batch_params:
										a :						Optional[int]      =							batch_size * [inputs[key]]

								a :						str      =							pipe(**lowerCAmelCase__ ,   num_images_per_prompt=lowerCAmelCase__ )[0]

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


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






							def        __a  (  self )      ->					Optional[Any]:
								# clean up the VRAM after each test
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()






							def        __a  (  self )      ->					Any:
								a :						Tuple      =							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 :						Optional[int]      =							ShapEImgaImgPipeline.from_pretrained("openai/shap-e-img2img" )
								a :						Dict      =							pipe.to(lowerCAmelCase__ )
								pipe.set_progress_bar_config(disable=lowerCAmelCase__ )

								a :						Union[str, Any]      =							torch.Generator(device=lowerCAmelCase__ ).manual_seed(0 )

								a :						int      =							pipe(
								    lowerCAmelCase__ ,   generator=lowerCAmelCase__ ,   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(lowerCAmelCase__ ,   lowerCAmelCase__ )









"""simple docstring"""




import unittest

from transformers import EsmConfig, is_torch_available
from transformers.testing_utils import TestCasePlus, require_torch, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
		import torch

		from transformers import EsmForMaskedLM, EsmForSequenceClassification, EsmForTokenClassification, EsmModel
		from transformers.models.esm.modeling_esm import (
		    ESM_PRETRAINED_MODEL_ARCHIVE_LIST,
		    EsmEmbeddings,
		    create_position_ids_from_input_ids,
		)


class 			__UpperCamelCase							:






							def __init__(  self ,   lowerCAmelCase__ ,   lowerCAmelCase__=13 ,   lowerCAmelCase__=7 ,   lowerCAmelCase__=False ,   lowerCAmelCase__=True ,   lowerCAmelCase__=False ,   lowerCAmelCase__=True ,   lowerCAmelCase__=33 ,   lowerCAmelCase__=32 ,   lowerCAmelCase__=5 ,   lowerCAmelCase__=4 ,   lowerCAmelCase__=37 ,   lowerCAmelCase__="gelu" ,   lowerCAmelCase__=0.1 ,   lowerCAmelCase__=0.1 ,   lowerCAmelCase__=512 ,   lowerCAmelCase__=16 ,   lowerCAmelCase__=2 ,   lowerCAmelCase__=0.02 ,   lowerCAmelCase__=3 ,   lowerCAmelCase__=4 ,   lowerCAmelCase__=None ,   )      ->					List[str]:
								a :						List[str]      =							parent
								a :						Optional[int]      =							batch_size
								a :						Optional[int]      =							seq_length
								a :						Tuple      =							is_training
								a :						Optional[int]      =							use_input_mask
								a :						List[Any]      =							use_token_type_ids
								a :						Tuple      =							use_labels
								a :						Union[str, Any]      =							vocab_size
								a :						str      =							hidden_size
								a :						List[Any]      =							num_hidden_layers
								a :						Dict      =							num_attention_heads
								a :						int      =							intermediate_size
								a :						str      =							hidden_act
								a :						Optional[int]      =							hidden_dropout_prob
								a :						str      =							attention_probs_dropout_prob
								a :						str      =							max_position_embeddings
								a :						List[str]      =							type_vocab_size
								a :						List[str]      =							type_sequence_label_size
								a :						Union[str, Any]      =							initializer_range
								a :						int      =							num_labels
								a :						Optional[int]      =							num_choices
								a :						Union[str, Any]      =							scope






							def        __a  (  self )      ->					Optional[Any]:
								a :						str      =							ids_tensor([self.batch_size, self.seq_length] ,   self.vocab_size )

								a :						Dict      =							None
								if self.use_input_mask:
									a :						str      =							random_attention_mask([self.batch_size, self.seq_length] )

								a :						Union[str, Any]      =							None
								a :						int      =							None
								a :						int      =							None
								if self.use_labels:
									a :						Optional[int]      =							ids_tensor([self.batch_size] ,   self.type_sequence_label_size )
									a :						Optional[int]      =							ids_tensor([self.batch_size, self.seq_length] ,   self.num_labels )
									a :						List[Any]      =							ids_tensor([self.batch_size] ,   self.num_choices )

								a :						str      =							self.get_config()

								return config, input_ids, input_mask, sequence_labels, token_labels, choice_labels






							def        __a  (  self )      ->					Optional[int]:
								return EsmConfig(
								    vocab_size=self.vocab_size ,   hidden_size=self.hidden_size ,   pad_token_id=1 ,   num_hidden_layers=self.num_hidden_layers ,   num_attention_heads=self.num_attention_heads ,   intermediate_size=self.intermediate_size ,   hidden_act=self.hidden_act ,   hidden_dropout_prob=self.hidden_dropout_prob ,   attention_probs_dropout_prob=self.attention_probs_dropout_prob ,   max_position_embeddings=self.max_position_embeddings ,   type_vocab_size=self.type_vocab_size ,   initializer_range=self.initializer_range ,   )






							def        __a  (  self ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ )      ->					str:
								a :						Union[str, Any]      =							EsmModel(config=lowerCAmelCase__ )
								model.to(lowerCAmelCase__ )
								model.eval()
								a :						Optional[int]      =							model(lowerCAmelCase__ ,   attention_mask=lowerCAmelCase__ )
								a :						List[str]      =							model(lowerCAmelCase__ )
								a :						Optional[Any]      =							model(lowerCAmelCase__ )

								self.parent.assertEqual(result.last_hidden_state.shape ,   (self.batch_size, self.seq_length, self.hidden_size) )
								self.parent.assertEqual(result.pooler_output.shape ,   (self.batch_size, self.hidden_size) )






							def        __a  (  self ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ )      ->					Tuple:
								a :						int      =							EsmForMaskedLM(config=lowerCAmelCase__ )
								model.to(lowerCAmelCase__ )
								model.eval()
								a :						Union[str, Any]      =							model(lowerCAmelCase__ ,   attention_mask=lowerCAmelCase__ ,   labels=lowerCAmelCase__ )
								self.parent.assertEqual(result.logits.shape ,   (self.batch_size, self.seq_length, self.vocab_size) )






							def        __a  (  self ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ ,   lowerCAmelCase__ )      ->					Optional[Any]:
								a :						str      =							self.num_labels
								a :						int      =							EsmForTokenClassification(config=lowerCAmelCase__ )
								model.to(lowerCAmelCase__ )
								model.eval()
								a :						int      =							model(lowerCAmelCase__ ,   attention_mask=lowerCAmelCase__ ,   labels=lowerCAmelCase__ )
								self.parent.assertEqual(result.logits.shape ,   (self.batch_size, self.seq_length, self.num_labels) )






							def        __a  (  self )      ->					Union[str, Any]:
								a :						str      =							self.prepare_config_and_inputs()
								(
								    (
								    a
								),			(
								    a
								),			(
								    a
								),			(
								    a
								),			(
								    a
								),			(
								    a
								),			
								) :						Optional[int]      =							config_and_inputs
								a :						int      =							{"input_ids": input_ids, "attention_mask": input_mask}
								return config, inputs_dict


@require_torch
class 			__UpperCamelCase							(		a__						,			a__						,			unittest.TestCase		):
							lowerCamelCase  :   Optional[Any]    =False

							lowerCamelCase  :   Tuple    =(
							    (
							        EsmForMaskedLM,
							        EsmModel,
							        EsmForSequenceClassification,
							        EsmForTokenClassification,
							    )
							    if is_torch_available()
							    else ()
							)
							lowerCamelCase  :   Tuple    =()
							lowerCamelCase  :   Optional[int]    =(
							    {
							        """feature-extraction""": EsmModel,
							        """fill-mask""": EsmForMaskedLM,
							        """text-classification""": EsmForSequenceClassification,
							        """token-classification""": EsmForTokenClassification,
							        """zero-shot""": EsmForSequenceClassification,
							    }
							    if is_torch_available()
							    else {}
							)
							lowerCamelCase  :   Any    =True






							def        __a  (  self )      ->					Tuple:
								a :						str      =							EsmModelTester(self )
								a :						Union[str, Any]      =							ConfigTester(self ,   config_class=lowerCAmelCase__ ,   hidden_size=37 )






							def        __a  (  self )      ->					Dict:
								self.config_tester.run_common_tests()






							def        __a  (  self )      ->					Union[str, Any]:
								a :						Tuple      =							self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*lowerCAmelCase__ )






							def        __a  (  self )      ->					List[str]:
								a :						Dict      =							self.model_tester.prepare_config_and_inputs()
								for type in ["absolute", "relative_key", "relative_key_query"]:
									a :						List[str]      =							type
									self.model_tester.create_and_check_model(*lowerCAmelCase__ )






							def        __a  (  self )      ->					Optional[Any]:
								a :						Dict      =							self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_masked_lm(*lowerCAmelCase__ )






							def        __a  (  self )      ->					Dict:
								a :						int      =							self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_token_classification(*lowerCAmelCase__ )






							@slow
							def        __a  (  self )      ->					Optional[Any]:
								for model_name in ESM_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
									a :						Tuple      =							EsmModel.from_pretrained(lowerCAmelCase__ )
									self.assertIsNotNone(lowerCAmelCase__ )






							def        __a  (  self )      ->					Optional[int]:
								a :						List[str]      =							self.model_tester.prepare_config_and_inputs()[0]
								a :						str      =							EsmEmbeddings(config=lowerCAmelCase__ )

								a :						Optional[Any]      =							torch.as_tensor([[12, 31, 13, model.padding_idx]] )
								a :						Tuple      =							torch.as_tensor(
								    [
								        [
								            0 + model.padding_idx + 1,
								            1 + model.padding_idx + 1,
								            2 + model.padding_idx + 1,
								            model.padding_idx,
								        ]
								    ] )
								a :						str      =							create_position_ids_from_input_ids(lowerCAmelCase__ ,   model.padding_idx )
								self.assertEqual(position_ids.shape ,   expected_positions.shape )
								self.assertTrue(torch.all(torch.eq(lowerCAmelCase__ ,   lowerCAmelCase__ ) ) )






							def        __a  (  self )      ->					List[Any]:
								a :						int      =							self.model_tester.prepare_config_and_inputs()[0]
								a :						Dict      =							EsmEmbeddings(config=lowerCAmelCase__ )

								a :						Any      =							torch.empty(2 ,   4 ,   30 )
								a :						str      =							[
								    0 + embeddings.padding_idx + 1,
								    1 + embeddings.padding_idx + 1,
								    2 + embeddings.padding_idx + 1,
								    3 + embeddings.padding_idx + 1,
								]
								a :						Optional[Any]      =							torch.as_tensor([expected_single_positions, expected_single_positions] )
								a :						str      =							embeddings.create_position_ids_from_inputs_embeds(lowerCAmelCase__ )
								self.assertEqual(position_ids.shape ,   expected_positions.shape )
								self.assertTrue(torch.all(torch.eq(lowerCAmelCase__ ,   lowerCAmelCase__ ) ) )






							@unittest.skip("Esm does not support embedding resizing" )
							def        __a  (  self )      ->					Any:
								pass






							@unittest.skip("Esm does not support embedding resizing" )
							def        __a  (  self )      ->					Dict:
								pass






							@unittest.skip("Will be fixed soon by reducing the size of the model used for common tests." )
							def        __a  (  self )      ->					str:
								pass


@require_torch
class 			__UpperCamelCase							(		a__		):






							@slow
							def        __a  (  self )      ->					int:
								with torch.no_grad():
									a :						str      =							EsmForMaskedLM.from_pretrained("facebook/esm2_t6_8M_UR50D" )
									model.eval()
									a :						List[str]      =							torch.tensor([[0, 1, 2, 3, 4, 5]] )
									a :						Any      =							model(lowerCAmelCase__ )[0]

									a :						Union[str, Any]      =							33

									a :						str      =							torch.Size((1, 6, vocab_size) )
									self.assertEqual(output.shape ,   lowerCAmelCase__ )

									a :						Optional[int]      =							torch.tensor(
									    [[[8.9_215, -10.5_898, -6.4_671], [-6.3_967, -13.9_114, -1.1_212], [-7.7_812, -13.9_516, -3.7_406]]] )
									self.assertTrue(torch.allclose(output[:, :3, :3] ,   lowerCAmelCase__ ,   atol=1E-4 ) )






							@slow
							def        __a  (  self )      ->					str:
								with torch.no_grad():
									a :						Optional[int]      =							EsmModel.from_pretrained("facebook/esm2_t6_8M_UR50D" )
									model.eval()

									a :						str      =							torch.tensor([[0, 6, 4, 13, 5, 4, 16, 12, 11, 7, 2]] )
									a :						Tuple      =							model(lowerCAmelCase__ )[0]
									# compare the actual values for a slice.
									a :						Union[str, Any]      =							torch.tensor(
									    [[[0.1_444, 0.5_413, 0.3_248], [0.3_034, 0.0_053, 0.3_108], [0.3_228, -0.2_499, 0.3_415]]] )
									self.assertTrue(torch.allclose(output[:, :3, :3] ,   lowerCAmelCase__ ,   atol=1E-4 ) )


