


'''simple docstring'''


import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import HeunDiscreteScheduler, PriorTransformer, ShapEPipeline
from diffusers.pipelines.shap_e import ShapERenderer
from diffusers.utils import 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__    (		_snake_case	,	unittest.TestCase   ):





							'''simple docstring'''


							A_				: Optional[Any]       =    ShapEPipeline
							A_				: Tuple       =    ["""prompt"""]
							A_				: Tuple       =    ["""prompt"""]
							A_				: Any       =    [
							    """num_images_per_prompt""",
							    """num_inference_steps""",
							    """generator""",
							    """latents""",
							    """guidance_scale""",
							    """frame_size""",
							    """output_type""",
							    """return_dict""",
							]
							A_				: Dict       =    False






							@property
							def UpperCAmelCase_      (       self    ):
											return 32






							@property
							def UpperCAmelCase_      (       self    ):
											return 32






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






							@property
							def UpperCAmelCase_      (       self    ):
											return 8






							@property
							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Tuple							= CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""    )
											return tokenizer






							@property
							def UpperCAmelCase_      (       self    ):
											torch.manual_seed(0    )
											_SCREAMING_SNAKE_CASE					:			int							= 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-05       ,   num_attention_heads=4       ,   num_hidden_layers=5       ,   pad_token_id=1       ,   vocab_size=1000       ,   )
											return CLIPTextModelWithProjection(__snake_case    )






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

											_SCREAMING_SNAKE_CASE					:			Optional[int]							= {
											    """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""",
											    """encoder_hid_proj_type""": None,
											    """added_emb_type""": None,
											}

											_SCREAMING_SNAKE_CASE					:			str							= PriorTransformer(**__snake_case    )
											return model






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

											_SCREAMING_SNAKE_CASE					:			Optional[int]							= {
											    """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,
											    ),
											}
											_SCREAMING_SNAKE_CASE					:			str							= ShapERenderer(**__snake_case    )
											return model






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= self.dummy_prior
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.dummy_text_encoder
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= self.dummy_tokenizer
											_SCREAMING_SNAKE_CASE					:			Any							= self.dummy_renderer

											_SCREAMING_SNAKE_CASE					:			Tuple							= HeunDiscreteScheduler(
											    beta_schedule="""exp"""       ,   num_train_timesteps=1024       ,   prediction_type="""sample"""       ,   use_karras_sigmas=__snake_case       ,   clip_sample=__snake_case       ,   clip_sample_range=1.0       ,   )
											_SCREAMING_SNAKE_CASE					:			str							= {
											    """prior""": prior,
											    """text_encoder""": text_encoder,
											    """tokenizer""": tokenizer,
											    """renderer""": renderer,
											    """scheduler""": scheduler,
											}

											return components






							def UpperCAmelCase_      (       self       ,   __snake_case       ,   __snake_case=0    ):
											if str(__snake_case    ).startswith("""mps"""    ):
															_SCREAMING_SNAKE_CASE					:			Tuple							= torch.manual_seed(__snake_case    )
											else:
															_SCREAMING_SNAKE_CASE					:			List[str]							= torch.Generator(device=__snake_case    ).manual_seed(__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Tuple							= {
											    """prompt""": """horse""",
											    """generator""": generator,
											    """num_inference_steps""": 1,
											    """frame_size""": 32,
											    """output_type""": """np""",
											}
											return inputs






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			int							= """cpu"""

											_SCREAMING_SNAKE_CASE					:			Dict							= self.get_dummy_components()

											_SCREAMING_SNAKE_CASE					:			Optional[int]							= self.pipeline_class(**__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Any							= pipe.to(__snake_case    )

											pipe.set_progress_bar_config(disable=__snake_case    )

											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= pipe(**self.get_dummy_inputs(__snake_case    )    )
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= output.images[0]
											_SCREAMING_SNAKE_CASE					:			List[Any]							= image[0, -3:, -3:, -1]

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

											_SCREAMING_SNAKE_CASE					:			int							= 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 UpperCAmelCase_      (       self    ):
											# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
											self._test_inference_batch_consistent(batch_sizes=[1, 2]    )






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= torch_device == """cpu"""
											_SCREAMING_SNAKE_CASE					:			int							= True

											self._test_inference_batch_single_identical(
											    batch_size=2       ,   test_max_difference=__snake_case       ,   relax_max_difference=__snake_case       ,   )






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			str							= self.get_dummy_components()
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= self.pipeline_class(**__snake_case    )
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= pipe.to(__snake_case    )
											pipe.set_progress_bar_config(disable=__snake_case    )

											_SCREAMING_SNAKE_CASE					:			List[str]							= 1
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= 2

											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_dummy_inputs(__snake_case    )

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

											_SCREAMING_SNAKE_CASE					:			Optional[int]							= pipe(**__snake_case       ,   num_images_per_prompt=__snake_case    )[0]

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





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





							'''simple docstring'''


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






							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
											    """/shap_e/test_shap_e_np_out.npy"""    )
											_SCREAMING_SNAKE_CASE					:			Any							= ShapEPipeline.from_pretrained("""openai/shap-e"""    )
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= pipe.to(__snake_case    )
											pipe.set_progress_bar_config(disable=__snake_case    )

											_SCREAMING_SNAKE_CASE					:			List[Any]							= torch.Generator(device=__snake_case    ).manual_seed(0    )

											_SCREAMING_SNAKE_CASE					:			str							= pipe(
											    """a shark"""       ,   generator=__snake_case       ,   guidance_scale=15.0       ,   num_inference_steps=64       ,   frame_size=64       ,   output_type="""np"""       ,   ).images[0]

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

											assert_mean_pixel_difference(__snake_case       ,   __snake_case    )








'''simple docstring'''


import subprocess
import sys

from transformers import BertConfig, BertModel, BertTokenizer, pipeline
from transformers.testing_utils import TestCasePlus, require_torch





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							@require_torch
							def UpperCAmelCase_      (       self    ):
											# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
											# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
											# while running an external program

											# python one-liner segments

											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

											_SCREAMING_SNAKE_CASE					:			Dict							= """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise RuntimeError(\"Offline mode is enabled, we shouldn't access internet\")
socket.socket = offline_socket
        """

											# Force fetching the files so that we can use the cache
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """hf-internal-testing/tiny-random-bert"""
											BertConfig.from_pretrained(__snake_case    )
											BertModel.from_pretrained(__snake_case    )
											BertTokenizer.from_pretrained(__snake_case    )
											pipeline(task="""fill-mask"""       ,   model=__snake_case    )

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			Union[str, Any]							= [sys.executable, """-c""", """\n""".join([load, run, mock]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_env()
											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			int							= """1"""
											_SCREAMING_SNAKE_CASE					:			str							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											# python one-liner segments
											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

											_SCREAMING_SNAKE_CASE					:			Any							= """
mname = \"hf-internal-testing/tiny-random-bert\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
BertTokenizer.from_pretrained(mname)
pipe = pipeline(task=\"fill-mask\", model=mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			List[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Faking flaky internet\")
socket.socket = offline_socket
        """

											# Force fetching the files so that we can use the cache
											_SCREAMING_SNAKE_CASE					:			Tuple							= """hf-internal-testing/tiny-random-bert"""
											BertConfig.from_pretrained(__snake_case    )
											BertModel.from_pretrained(__snake_case    )
											BertTokenizer.from_pretrained(__snake_case    )
											pipeline(task="""fill-mask"""       ,   model=__snake_case    )

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			Tuple							= [sys.executable, """-c""", """\n""".join([load, run, mock]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			str							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Tuple							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											# this test is a bit tricky since TRANSFORMERS_OFFLINE can only be changed before
											# `transformers` is loaded, and it's too late for inside pytest - so we are changing it
											# while running an external program

											# python one-liner segments

											# this must be loaded before socket.socket is monkey-patched
											_SCREAMING_SNAKE_CASE					:			Tuple							= """
from transformers import BertConfig, BertModel, BertTokenizer
        """

											_SCREAMING_SNAKE_CASE					:			str							= """
mname = \"hf-internal-testing/tiny-random-bert-sharded\"
BertConfig.from_pretrained(mname)
BertModel.from_pretrained(mname)
print(\"success\")
        """

											_SCREAMING_SNAKE_CASE					:			List[Any]							= """
import socket
def offline_socket(*args, **kwargs): raise ValueError(\"Offline mode is enabled\")
socket.socket = offline_socket
        """

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [sys.executable, """-c""", """\n""".join([load, run]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			List[str]							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )

											# next emulate no network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [sys.executable, """-c""", """\n""".join([load, mock, run]    )]

											# Doesn't fail anymore since the model is in the cache due to other tests, so commenting this.
											# env["TRANSFORMERS_OFFLINE"] = "0"
											# result = subprocess.run(cmd, env=env, check=False, capture_output=True)
											# self.assertEqual(result.returncode, 1, result.stderr)

											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			Dict							= """1"""
											_SCREAMING_SNAKE_CASE					:			int							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import pipeline
        """
											_SCREAMING_SNAKE_CASE					:			int							= """
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

											_SCREAMING_SNAKE_CASE					:			Any							= """
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
											_SCREAMING_SNAKE_CASE					:			Dict							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			Any							= """1"""
											_SCREAMING_SNAKE_CASE					:			str							= [sys.executable, """-c""", """\n""".join([load, mock, run]    )]
											_SCREAMING_SNAKE_CASE					:			List[Any]							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   1       ,   result.stderr    )
											self.assertIn(
											    """You cannot infer task automatically within `pipeline` when using offline mode"""       ,   result.stderr.decode().replace("""\n"""       ,   """"""    )       ,   )






							@require_torch
							def UpperCAmelCase_      (       self    ):
											_SCREAMING_SNAKE_CASE					:			Optional[int]							= """
from transformers import AutoModel
        """
											_SCREAMING_SNAKE_CASE					:			Tuple							= """
mname = \"hf-internal-testing/test_dynamic_model\"
AutoModel.from_pretrained(mname, trust_remote_code=True)
print(\"success\")
        """

											# baseline - just load from_pretrained with normal network
											_SCREAMING_SNAKE_CASE					:			List[str]							= [sys.executable, """-c""", """\n""".join([load, run]    )]

											# should succeed
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= self.get_env()
											_SCREAMING_SNAKE_CASE					:			List[str]							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )

											# should succeed as TRANSFORMERS_OFFLINE=1 tells it to use local files
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= """1"""
											_SCREAMING_SNAKE_CASE					:			Any							= subprocess.run(__snake_case       ,   env=__snake_case       ,   check=__snake_case       ,   capture_output=__snake_case    )
											self.assertEqual(result.returncode       ,   0       ,   result.stderr    )
											self.assertIn("""success"""       ,   result.stdout.decode()    )




