



'''simple docstring'''




import subprocess
import sys

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







class      _lowerCAmelCase							(      __UpperCAmelCase			):


	@require_torch
	def    _a     (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
				A_     :						int     =					"""
from transformers import BertConfig, BertModel, BertTokenizer, pipeline
        """

				A_     :						Tuple     =					"""
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\")
        """

				A_     :						Dict     =					"""
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
				A_     :						Optional[Any]     =					"""hf-internal-testing/tiny-random-bert"""
				BertConfig.from_pretrained(lowercase							)
				BertModel.from_pretrained(lowercase							)
				BertTokenizer.from_pretrained(lowercase							)
				pipeline(task="""fill-mask"""	,				model=lowercase							)

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

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


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

				A_     :						str     =					"""
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\")
        """

				A_     :						int     =					"""
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
				A_     :						Tuple     =					"""hf-internal-testing/tiny-random-bert"""
				BertConfig.from_pretrained(lowercase							)
				BertModel.from_pretrained(lowercase							)
				BertTokenizer.from_pretrained(lowercase							)
				pipeline(task="""fill-mask"""	,				model=lowercase							)

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

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


	@require_torch
	def    _a     (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
				A_     :						str     =					"""
from transformers import BertConfig, BertModel, BertTokenizer
        """

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

				A_     :						Optional[int]     =					"""
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
				A_     :						Union[str, Any]     =					[sys.executable, """-c""", """\n""".join([load, run]							)]

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

				# next emulate no network
				A_     :						int     =					[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
				A_     :						int     =					"""1"""
				A_     :						Optional[int]     =					subprocess.run(lowercase	,				env=lowercase	,				check=lowercase	,				capture_output=lowercase							)
				self.assertEqual(result.returncode	,				0	,				result.stderr							)
				self.assertIn("""success"""	,				result.stdout.decode()							)


	@require_torch
	def    _a     (self							):
				A_     :						Dict     =					"""
from transformers import pipeline
        """
				A_     :						Tuple     =					"""
mname = \"hf-internal-testing/tiny-random-bert\"
pipe = pipeline(model=mname)
        """

				A_     :						Optional[Any]     =					"""
import socket
def offline_socket(*args, **kwargs): raise socket.error(\"Offline mode is enabled\")
socket.socket = offline_socket
        """
				A_     :						List[str]     =					self.get_env()
				A_     :						Any     =					"""1"""
				A_     :						Optional[int]     =					[sys.executable, """-c""", """\n""".join([load, mock, run]							)]
				A_     :						List[str]     =					subprocess.run(lowercase	,				env=lowercase	,				check=lowercase	,				capture_output=lowercase							)
				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    _a     (self							):
				A_     :						List[Any]     =					"""
from transformers import AutoModel
        """
				A_     :						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
				A_     :						Union[str, Any]     =					[sys.executable, """-c""", """\n""".join([load, run]							)]

				# should succeed
				A_     :						List[Any]     =					self.get_env()
				A_     :						int     =					subprocess.run(lowercase	,				env=lowercase	,				check=lowercase	,				capture_output=lowercase							)
				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
				A_     :						List[Any]     =					"""1"""
				A_     :						str     =					subprocess.run(lowercase	,				env=lowercase	,				check=lowercase	,				capture_output=lowercase							)
				self.assertEqual(result.returncode	,				0	,				result.stderr							)
				self.assertIn("""success"""	,				result.stdout.decode()							)




'''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      _lowerCAmelCase							(      __UpperCAmelCase				,      unittest.TestCase			):
	__SCREAMING_SNAKE_CASE       :		Union[str, Any]		      = ShapEPipeline
	__SCREAMING_SNAKE_CASE       :		Optional[int]		      = ['prompt']
	__SCREAMING_SNAKE_CASE       :		str		      = ['prompt']
	__SCREAMING_SNAKE_CASE       :		Dict		      = [
	    'num_images_per_prompt',
	    'num_inference_steps',
	    'generator',
	    'latents',
	    'guidance_scale',
	    'frame_size',
	    'output_type',
	    'return_dict',
	]
	__SCREAMING_SNAKE_CASE       :		Union[str, Any]		      = False


	@property
	def    _a     (self							):
				return 32


	@property
	def    _a     (self							):
				return 32


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


	@property
	def    _a     (self							):
				return 8


	@property
	def    _a     (self							):
				A_     :						Dict     =					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""							)
				return tokenizer


	@property
	def    _a     (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=37	,				layer_norm_eps=1E-05	,				num_attention_heads=4	,				num_hidden_layers=5	,				pad_token_id=1	,				vocab_size=1000	,				)
				return CLIPTextModelWithProjection(lowercase							)


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

				A_     :						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""",
				    """encoder_hid_proj_type""": None,
				    """added_emb_type""": None,
				}

				A_     :						List[str]     =					PriorTransformer(**lowercase							)
				return model


	@property
	def    _a     (self							):
				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_     :						Dict     =					ShapERenderer(**lowercase							)
				return model


	def    _a     (self							):
				A_     :						Optional[int]     =					self.dummy_prior
				A_     :						Optional[int]     =					self.dummy_text_encoder
				A_     :						int     =					self.dummy_tokenizer
				A_     :						Dict     =					self.dummy_renderer

				A_     :						Tuple     =					HeunDiscreteScheduler(
				    beta_schedule="""exp"""	,				num_train_timesteps=1024	,				prediction_type="""sample"""	,				use_karras_sigmas=lowercase	,				clip_sample=lowercase	,				clip_sample_range=1.0	,				)
				A_     :						Union[str, Any]     =					{
				    """prior""": prior,
				    """text_encoder""": text_encoder,
				    """tokenizer""": tokenizer,
				    """renderer""": renderer,
				    """scheduler""": scheduler,
				}

				return components


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


	def    _a     (self							):
				A_     :						str     =					"""cpu"""

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

				A_     :						Optional[int]     =					self.pipeline_class(**lowercase							)
				A_     :						str     =					pipe.to(lowercase							)

				pipe.set_progress_bar_config(disable=lowercase							)

				A_     :						Dict     =					pipe(**self.get_dummy_inputs(lowercase							)							)
				A_     :						Dict     =					output.images[0]
				A_     :						int     =					image[0, -3:, -3:, -1]

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

				A_     :						Tuple     =					np.array(
				    [
				        0.00_03_92_16,
				        0.00_03_92_16,
				        0.00_03_92_16,
				        0.00_03_92_16,
				        0.00_03_92_16,
				        0.00_03_92_16,
				        0.00_03_92_16,
				        0.00_03_92_16,
				        0.00_03_92_16,
				    ]							)

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


	def    _a     (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    _a     (self							):
				A_     :						Tuple     =					torch_device == """cpu"""
				A_     :						Union[str, Any]     =					True

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


	def    _a     (self							):
				A_     :						List[Any]     =					self.get_dummy_components()
				A_     :						Any     =					self.pipeline_class(**lowercase							)
				A_     :						Any     =					pipe.to(lowercase							)
				pipe.set_progress_bar_config(disable=lowercase							)

				A_     :						int     =					1
				A_     :						Union[str, Any]     =					2

				A_     :						Dict     =					self.get_dummy_inputs(lowercase							)

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

				A_     :						List[Any]     =					pipe(**lowercase	,				num_images_per_prompt=lowercase							)[0]

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







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


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


	def    _a     (self							):
				A_     :						Union[str, Any]     =					load_numpy(
				    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
				    """/shap_e/test_shap_e_np_out.npy"""							)
				A_     :						Tuple     =					ShapEPipeline.from_pretrained("""openai/shap-e"""							)
				A_     :						int     =					pipe.to(lowercase							)
				pipe.set_progress_bar_config(disable=lowercase							)

				A_     :						int     =					torch.Generator(device=lowercase							).manual_seed(0							)

				A_     :						List[str]     =					pipe(
				    """a shark"""	,				generator=lowercase	,				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(lowercase	,				lowercase							)