"""simple docstring"""




from dataclasses import dataclass, field
from typing import Tuple

from ..utils import cached_property, is_tf_available, logging, requires_backends
from .benchmark_args_utils import BenchmarkArguments


if is_tf_available():
     import tensorflow as tf


_a   :	int 					=     logging.get_logger(__name__)
@dataclass
class 		__A							(				SCREAMING_SNAKE_CASE_	):
     _UpperCamelCase							:    Any	      =	[
         "no_inference",
         "no_cuda",
         "no_tpu",
         "no_speed",
         "no_memory",
         "no_env_print",
         "no_multi_process",
     ]






     def __init__(	self       ,  **a__						):
       for deprecated_arg in self.deprecated_args:
         if deprecated_arg in kwargs:
           _lowerCAmelCase		: Union[str, Any]						    =  deprecated_arg[3:]
           _lowerCAmelCase		: Optional[Any]						    =  not kwargs.pop(a__						)
           logger.warning(
               F"{deprecated_arg} is depreciated. Please use --no-{positive_arg} or"
               F" {positive_arg}={kwargs[positive_arg]}"						)
       _lowerCAmelCase		: Dict						    =  kwargs.pop("""tpu_name"""       ,  self.tpu_name						)
       _lowerCAmelCase		: Optional[Any]						    =  kwargs.pop("""device_idx"""       ,  self.device_idx						)
       _lowerCAmelCase		: int						    =  kwargs.pop("""eager_mode"""       ,  self.eager_mode						)
       _lowerCAmelCase		: List[str]						    =  kwargs.pop("""use_xla"""       ,  self.use_xla						)
       super().__init__(**a__						)

     _UpperCamelCase							:    str	      =	field(
         default=SCREAMING_SNAKE_CASE_					,			metadata={"help": "Name of TPU"}					,			)
     _UpperCamelCase							:    int	      =	field(
         default=0					,			metadata={"help": "CPU / GPU device index. Defaults to 0."}					,			)
     _UpperCamelCase							:    bool	      =	field(default=SCREAMING_SNAKE_CASE_					,			metadata={"help": "Benchmark models in eager model."}	)
     _UpperCamelCase							:    bool	      =	field(
         default=SCREAMING_SNAKE_CASE_					,			metadata={
             "help": "Benchmark models using XLA JIT compilation. Note that `eager_model` has to be set to `False`."
         }					,			)






     @cached_property
     def 					__A						(	self						):
       requires_backends(self       ,  ["""tf"""]						)
       _lowerCAmelCase		: int						    =  None
       if self.tpu:
         try:
           if self.tpu_name:
             _lowerCAmelCase		: int						    =  tf.distribute.cluster_resolver.TPUClusterResolver(self.tpu_name						)
           else:
             _lowerCAmelCase		: int						    =  tf.distribute.cluster_resolver.TPUClusterResolver()
         except ValueError:
           _lowerCAmelCase		: List[Any]						    =  None
       return tpu






     @cached_property
     def 					__A						(	self						):
       requires_backends(self       ,  ["""tf"""]						)
       if self.is_tpu:
         tf.config.experimental_connect_to_cluster(self._setup_tpu						)
         tf.tpu.experimental.initialize_tpu_system(self._setup_tpu						)

         _lowerCAmelCase		: List[str]						    =  tf.distribute.TPUStrategy(self._setup_tpu						)
       else:
         # currently no multi gpu is allowed
         if self.is_gpu:
           # TODO: Currently only single GPU is supported
           tf.config.set_visible_devices(self.gpu_list[self.device_idx]       ,  """GPU"""						)
           _lowerCAmelCase		: Dict						    =  tf.distribute.OneDeviceStrategy(device=F"/gpu:{self.device_idx}"						)
         else:
           tf.config.set_visible_devices([]       ,  """GPU"""						)  # disable GPU
           _lowerCAmelCase		: str						    =  tf.distribute.OneDeviceStrategy(device=F"/cpu:{self.device_idx}"						)

       return strategy






     @property
     def 					__A						(	self						):
       requires_backends(self       ,  ["""tf"""]						)
       return self._setup_tpu is not None






     @property
     def 					__A						(	self						):
       requires_backends(self       ,  ["""tf"""]						)
       return self._setup_strategy






     @property
     def 					__A						(	self						):
       requires_backends(self       ,  ["""tf"""]						)
       return tf.config.list_physical_devices("""GPU"""						)






     @property
     def 					__A						(	self						):
       requires_backends(self       ,  ["""tf"""]						)
       if self.cuda:
         return len(self.gpu_list						)
       return 0






     @property
     def 					__A						(	self						):
       return self.n_gpu > 0





"""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 		__A							(				SCREAMING_SNAKE_CASE_					,			unittest.TestCase	):
     _UpperCamelCase							:    Optional[int]	      =	ShapEPipeline
     _UpperCamelCase							:    Optional[Any]	      =	["prompt"]
     _UpperCamelCase							:    Tuple	      =	["prompt"]
     _UpperCamelCase							:    Dict	      =	[
         "num_images_per_prompt",
         "num_inference_steps",
         "generator",
         "latents",
         "guidance_scale",
         "frame_size",
         "output_type",
         "return_dict",
     ]
     _UpperCamelCase							:    str	      =	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						):
       _lowerCAmelCase		: Union[str, Any]						    =  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""						)
       return tokenizer






     @property
     def 					__A						(	self						):
       torch.manual_seed(0						)
       _lowerCAmelCase		: List[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(a__						)






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

       _lowerCAmelCase		: Tuple						    =  {
           """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,
       }

       _lowerCAmelCase		: Any						    =  PriorTransformer(**a__						)
       return model






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

       _lowerCAmelCase		: Tuple						    =  {
           """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		: Dict						    =  ShapERenderer(**a__						)
       return model






     def 					__A						(	self						):
       _lowerCAmelCase		: Union[str, Any]						    =  self.dummy_prior
       _lowerCAmelCase		: Any						    =  self.dummy_text_encoder
       _lowerCAmelCase		: List[Any]						    =  self.dummy_tokenizer
       _lowerCAmelCase		: Dict						    =  self.dummy_renderer

       _lowerCAmelCase		: List[Any]						    =  HeunDiscreteScheduler(
           beta_schedule="""exp"""       ,  num_train_timesteps=1024       ,  prediction_type="""sample"""       ,  use_karras_sigmas=a__       ,  clip_sample=a__       ,  clip_sample_range=1.0       ,  )
       _lowerCAmelCase		: List[Any]						    =  {
           """prior""": prior,
           """text_encoder""": text_encoder,
           """tokenizer""": tokenizer,
           """renderer""": renderer,
           """scheduler""": scheduler,
       }

       return components






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






     def 					__A						(	self						):
       _lowerCAmelCase		: List[Any]						    =  """cpu"""

       _lowerCAmelCase		: List[Any]						    =  self.get_dummy_components()

       _lowerCAmelCase		: str						    =  self.pipeline_class(**a__						)
       _lowerCAmelCase		: List[Any]						    =  pipe.to(a__						)

       pipe.set_progress_bar_config(disable=a__						)

       _lowerCAmelCase		: Optional[int]						    =  pipe(**self.get_dummy_inputs(a__						)						)
       _lowerCAmelCase		: List[str]						    =  output.images[0]
       _lowerCAmelCase		: Optional[Any]						    =  image[0, -3:, -3:, -1]

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

       _lowerCAmelCase		: Union[str, Any]						    =  np.array(
           [
               0.0_0_0_3_9_2_1_6,
               0.0_0_0_3_9_2_1_6,
               0.0_0_0_3_9_2_1_6,
               0.0_0_0_3_9_2_1_6,
               0.0_0_0_3_9_2_1_6,
               0.0_0_0_3_9_2_1_6,
               0.0_0_0_3_9_2_1_6,
               0.0_0_0_3_9_2_1_6,
               0.0_0_0_3_9_2_1_6,
           ]						)

       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						):
       _lowerCAmelCase		: Any						    =  torch_device == """cpu"""
       _lowerCAmelCase		: Dict						    =  True

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






     def 					__A						(	self						):
       _lowerCAmelCase		: int						    =  self.get_dummy_components()
       _lowerCAmelCase		: Optional[Any]						    =  self.pipeline_class(**a__						)
       _lowerCAmelCase		: int						    =  pipe.to(a__						)
       pipe.set_progress_bar_config(disable=a__						)

       _lowerCAmelCase		: str						    =  1
       _lowerCAmelCase		: Optional[Any]						    =  2

       _lowerCAmelCase		: List[Any]						    =  self.get_dummy_inputs(a__						)

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

       _lowerCAmelCase		: Tuple						    =  pipe(**a__       ,  num_images_per_prompt=a__						)[0]

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







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






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






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

       _lowerCAmelCase		: Optional[int]						    =  torch.Generator(device=a__						).manual_seed(0						)

       _lowerCAmelCase		: Any						    =  pipe(
           """a shark"""       ,  generator=a__       ,  guidance_scale=1_5.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__						)




