




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 			__lowerCamelCase	(   UpperCamelCase__     ,    unittest.TestCase   ):




    """simple docstring"""

    snake_case__ =				ShapEImgaImgPipeline
    snake_case__ =				["image"]
    snake_case__ =				["image"]
    snake_case__ =				[
        "num_images_per_prompt",
        "num_inference_steps",
        "generator",
        "latents",
        "guidance_scale",
        "frame_size",
        "output_type",
        "return_dict",
    ]
    snake_case__ =				False


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


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


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


    @property
    def        a				(	self					:					List[Any]       )				->	List[str]:
         return 8


    @property
    def        a				(	self					:					Union[str, Any]       )				->	Dict:
         torch.manual_seed(0       )
         lowerCAmelCase__          =			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       ,     )

         lowerCAmelCase__          =			CLIPVisionModel(SCREAMING_SNAKE_CASE__       )
         return model


    @property
    def        a				(	self					:					str       )				->	List[Any]:
         lowerCAmelCase__          =			CLIPImageProcessor(
             crop_size=224       ,     do_center_crop=SCREAMING_SNAKE_CASE__       ,     do_normalize=SCREAMING_SNAKE_CASE__       ,     do_resize=SCREAMING_SNAKE_CASE__       ,     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					:					Optional[int]       )				->	Union[str, Any]:
         torch.manual_seed(0       )

         lowerCAmelCase__          =			{
             "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,
         }

         lowerCAmelCase__          =			PriorTransformer(**SCREAMING_SNAKE_CASE__       )
         return model


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

         lowerCAmelCase__          =			{
             "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,
             ),
         }
         lowerCAmelCase__          =			ShapERenderer(**SCREAMING_SNAKE_CASE__       )
         return model


    def        a				(	self					:					List[str]       )				->	int:
         lowerCAmelCase__          =			self.dummy_prior
         lowerCAmelCase__          =			self.dummy_image_encoder
         lowerCAmelCase__          =			self.dummy_image_processor
         lowerCAmelCase__          =			self.dummy_renderer

         lowerCAmelCase__          =			HeunDiscreteScheduler(
             beta_schedule="exp"       ,     num_train_timesteps=1_024       ,     prediction_type="sample"       ,     use_karras_sigmas=SCREAMING_SNAKE_CASE__       ,     clip_sample=SCREAMING_SNAKE_CASE__       ,     clip_sample_range=1.0       ,     )
         lowerCAmelCase__          =			{
             "prior": prior,
             "image_encoder": image_encoder,
             "image_processor": image_processor,
             "renderer": renderer,
             "scheduler": scheduler,
         }

         return components


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

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


    def        a				(	self					:					str       )				->	Dict:
         lowerCAmelCase__          =			"cpu"

         lowerCAmelCase__          =			self.get_dummy_components()

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

         pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			pipe(**self.get_dummy_inputs(SCREAMING_SNAKE_CASE__       )       )
         lowerCAmelCase__          =			output.images[0]
         lowerCAmelCase__          =			image[0, -3:, -3:, -1]

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

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


    def        a				(	self					:					Optional[int]       )				->	int:
         lowerCAmelCase__          =			self.get_dummy_components()
         lowerCAmelCase__          =			self.pipeline_class(**SCREAMING_SNAKE_CASE__       )
         lowerCAmelCase__          =			pipe.to(SCREAMING_SNAKE_CASE__       )
         pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__       )

         lowerCAmelCase__          =			1
         lowerCAmelCase__          =			2

         lowerCAmelCase__          =			self.get_dummy_inputs(SCREAMING_SNAKE_CASE__       )

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

         lowerCAmelCase__          =			pipe(**SCREAMING_SNAKE_CASE__       ,     num_images_per_prompt=SCREAMING_SNAKE_CASE__       )[0]

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






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




    """simple docstring"""

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


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

         lowerCAmelCase__          =			torch.Generator(device=SCREAMING_SNAKE_CASE__       ).manual_seed(0       )

         lowerCAmelCase__          =			pipe(
             SCREAMING_SNAKE_CASE__       ,     generator=SCREAMING_SNAKE_CASE__       ,     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(SCREAMING_SNAKE_CASE__       ,     SCREAMING_SNAKE_CASE__       )






import time
from contextlib import contextmanager
from pathlib import Path

import pytest
import requests
from huggingface_hub.hf_api import HfApi, HfFolder


UpperCamelCase								=					'__DUMMY_TRANSFORMERS_USER__'
UpperCamelCase								=					'Dummy User'
UpperCamelCase								=					'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'

UpperCamelCase								=					'https://hub-ci.huggingface.co'
UpperCamelCase								=					CI_HUB_ENDPOINT + '/datasets/{repo_id}/resolve/{revision}/{path}'
UpperCamelCase								=					CI_HUB_ENDPOINT + '/{repo_id}/resolve/{revision}/{filename}'
UpperCamelCase								=					Path('~/.huggingface/hub_ci_token').expanduser()



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: Dict      ):
     """simple docstring"""

     monkeypatch.setattr(
         "huggingface_hub.file_download.HUGGINGFACE_CO_URL_TEMPLATE"       ,    lowerCAmelCase_      )



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: int      ):
     """simple docstring"""

     monkeypatch.setattr("datasets.config.HF_ENDPOINT"       ,    lowerCAmelCase_      )
     monkeypatch.setattr("datasets.config.HUB_DATASETS_URL"       ,    lowerCAmelCase_      )



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: str      ):
     """simple docstring"""

     monkeypatch.setattr("huggingface_hub.hf_api.HfFolder.path_token"       ,    lowerCAmelCase_      )



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: int       ,    lowerCAmelCase_					: List[Any]      ):
     """simple docstring"""

     HfFolder.save_token(lowerCAmelCase_      )
     yield
     HfFolder.delete_token()



@pytest.fixture(scope="session"      )
def 							_A       (						):
     """simple docstring"""

     return HfApi(endpoint=lowerCAmelCase_      )



@pytest.fixture(scope="session"      )
def 							_A       (						lowerCAmelCase_					: HfApi      ):
     """simple docstring"""

     lowerCAmelCase__          =			HfFolder.get_token()
     HfFolder.save_token(lowerCAmelCase_      )
     yield CI_HUB_USER_TOKEN
     if previous_token is not None:
          HfFolder.save_token(lowerCAmelCase_      )



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: Tuple      ):
     """simple docstring"""

     def _cleanup_repo(lowerCAmelCase_					: Optional[Any]      ):
          hf_api.delete_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"      )

     return _cleanup_repo



@pytest.fixture
def 							_A       (						lowerCAmelCase_					: Union[str, Any]      ):
     """simple docstring"""

     @contextmanager
     def _temporary_repo(lowerCAmelCase_					: str      ):
          try:
               yield repo_id
          finally:
               cleanup_repo(lowerCAmelCase_      )

     return _temporary_repo



@pytest.fixture(scope="session"      )
def 							_A       (						lowerCAmelCase_					: HfApi       ,    lowerCAmelCase_					: Union[str, Any]       ,    lowerCAmelCase_					: Any      ):
     """simple docstring"""

     lowerCAmelCase__          =			F'repo_txt_data-{int(time.time() * 1_0E3      )}'
     lowerCAmelCase__          =			F'{CI_HUB_USER}/{repo_name}'
     hf_api.create_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"       ,    private=lowerCAmelCase_      )
     hf_api.upload_file(
         token=lowerCAmelCase_       ,    path_or_fileobj=str(lowerCAmelCase_      )       ,    path_in_repo="data/text_data.txt"       ,    repo_id=lowerCAmelCase_       ,    repo_type="dataset"       ,    )
     yield repo_id
     try:
          hf_api.delete_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"      )
     except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
          pass



@pytest.fixture()
def 							_A       (						lowerCAmelCase_					: Optional[Any]       ,    lowerCAmelCase_					: Optional[Any]       ,    lowerCAmelCase_					: int      ):
     """simple docstring"""

     return hf_private_dataset_repo_txt_data_



@pytest.fixture(scope="session"      )
def 							_A       (						lowerCAmelCase_					: HfApi       ,    lowerCAmelCase_					: Tuple       ,    lowerCAmelCase_					: Tuple      ):
     """simple docstring"""

     lowerCAmelCase__          =			F'repo_zipped_txt_data-{int(time.time() * 1_0E3      )}'
     lowerCAmelCase__          =			F'{CI_HUB_USER}/{repo_name}'
     hf_api.create_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"       ,    private=lowerCAmelCase_      )
     hf_api.upload_file(
         token=lowerCAmelCase_       ,    path_or_fileobj=str(lowerCAmelCase_      )       ,    path_in_repo="data.zip"       ,    repo_id=lowerCAmelCase_       ,    repo_type="dataset"       ,    )
     yield repo_id
     try:
          hf_api.delete_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"      )
     except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
          pass



@pytest.fixture()
def 							_A       (						lowerCAmelCase_					: int       ,    lowerCAmelCase_					: List[str]       ,    lowerCAmelCase_					: Tuple      ):
     """simple docstring"""

     return hf_private_dataset_repo_zipped_txt_data_



@pytest.fixture(scope="session"      )
def 							_A       (						lowerCAmelCase_					: HfApi       ,    lowerCAmelCase_					: List[str]       ,    lowerCAmelCase_					: int      ):
     """simple docstring"""

     lowerCAmelCase__          =			F'repo_zipped_img_data-{int(time.time() * 1_0E3      )}'
     lowerCAmelCase__          =			F'{CI_HUB_USER}/{repo_name}'
     hf_api.create_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"       ,    private=lowerCAmelCase_      )
     hf_api.upload_file(
         token=lowerCAmelCase_       ,    path_or_fileobj=str(lowerCAmelCase_      )       ,    path_in_repo="data.zip"       ,    repo_id=lowerCAmelCase_       ,    repo_type="dataset"       ,    )
     yield repo_id
     try:
          hf_api.delete_repo(lowerCAmelCase_       ,    token=lowerCAmelCase_       ,    repo_type="dataset"      )
     except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
          pass






@pytest.fixture()
def 							_A       (						lowerCAmelCase_					: Optional[Any]       ,    lowerCAmelCase_					: Dict       ,    lowerCAmelCase_					: List[str]      ):
     """simple docstring"""

     return hf_private_dataset_repo_zipped_img_data_
