
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				:    Union[str, Any]           =			ShapEPipeline
      _UpperCAmelCase				:    str           =			["""prompt"""]
      _UpperCAmelCase				:    Dict           =			["""prompt"""]
      _UpperCAmelCase				:    Tuple           =			[
          """num_images_per_prompt""",
          """num_inference_steps""",
          """generator""",
          """latents""",
          """guidance_scale""",
          """frame_size""",
          """output_type""",
          """return_dict""",
      ]
      _UpperCAmelCase				:    List[str]           =			False
      @property
      def 						UpperCamelCase__							(      self      ):
       return 3_2
      @property
      def 						UpperCamelCase__							(      self      ):
       return 3_2
      @property
      def 						UpperCamelCase__							(      self      ):
       return self.time_input_dim * 4
      @property
      def 						UpperCamelCase__							(      self      ):
       return 8
      @property
      def 						UpperCamelCase__							(      self      ):
       lowerCamelCase		:	Union[str, Any]							=      CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""      )
       return tokenizer
      @property
      def 						UpperCamelCase__							(      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=3_7    ,  layer_norm_eps=1e-05    ,  num_attention_heads=4    ,  num_hidden_layers=5    ,  pad_token_id=1    ,  vocab_size=1_0_0_0    ,  )
       return CLIPTextModelWithProjection(__magic_name__      )
      @property
      def 						UpperCamelCase__							(      self      ):
       torch.manual_seed(0      )

       lowerCamelCase		:	Optional[int]							=      {
           """num_attention_heads""": 2,
           """attention_head_dim""": 1_6,
           """embedding_dim""": self.time_input_dim,
           """num_embeddings""": 3_2,
           """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		:	Dict							=      PriorTransformer(**__magic_name__      )
       return model
      @property
      def 						UpperCamelCase__							(      self      ):
       torch.manual_seed(0      )

       lowerCamelCase		:	str							=      {
           """param_shapes""": (
               (self.renderer_dim, 9_3),
               (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""": 1_2,
           """background""": (
               0.1,
               0.1,
               0.1,
           ),
       }
       lowerCamelCase		:	List[Any]							=      ShapERenderer(**__magic_name__      )
       return model
      def 						UpperCamelCase__							(      self      ):
       lowerCamelCase		:	List[Any]							=      self.dummy_prior
       lowerCamelCase		:	str							=      self.dummy_text_encoder
       lowerCamelCase		:	int							=      self.dummy_tokenizer
       lowerCamelCase		:	int							=      self.dummy_renderer

       lowerCamelCase		:	Tuple							=      HeunDiscreteScheduler(
           beta_schedule="""exp"""    ,  num_train_timesteps=1_0_2_4    ,  prediction_type="""sample"""    ,  use_karras_sigmas=__magic_name__    ,  clip_sample=__magic_name__    ,  clip_sample_range=1.0    ,  )
       lowerCamelCase		:	Optional[int]							=      {
           """prior""": prior,
           """text_encoder""": text_encoder,
           """tokenizer""": tokenizer,
           """renderer""": renderer,
           """scheduler""": scheduler,
       }

       return components
      def 						UpperCamelCase__							(      self    ,  __magic_name__    ,  __magic_name__=0      ):
       if str(__magic_name__      ).startswith("""mps"""      ):
        lowerCamelCase		:	List[str]							=      torch.manual_seed(__magic_name__      )
       else:
        lowerCamelCase		:	List[Any]							=      torch.Generator(device=__magic_name__      ).manual_seed(__magic_name__      )
       lowerCamelCase		:	int							=      {
           """prompt""": """horse""",
           """generator""": generator,
           """num_inference_steps""": 1,
           """frame_size""": 3_2,
           """output_type""": """np""",
       }
       return inputs
      def 						UpperCamelCase__							(      self      ):
       lowerCamelCase		:	List[Any]							=      """cpu"""

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

       lowerCamelCase		:	Optional[Any]							=      self.pipeline_class(**__magic_name__      )
       lowerCamelCase		:	List[str]							=      pipe.to(__magic_name__      )

       pipe.set_progress_bar_config(disable=__magic_name__      )

       lowerCamelCase		:	Optional[Any]							=      pipe(**self.get_dummy_inputs(__magic_name__      )      )
       lowerCamelCase		:	Optional[Any]							=      output.images[0]
       lowerCamelCase		:	Tuple							=      image[0, -3:, -3:, -1]

       assert image.shape == (2_0, 3_2, 3_2, 3)

       lowerCamelCase		:	List[str]							=      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 						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      ):
       lowerCamelCase		:	str							=      torch_device == """cpu"""
       lowerCamelCase		:	str							=      True

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






      def 						UpperCamelCase__							(      self      ):
       lowerCamelCase		:	Any							=      self.get_dummy_components()
       lowerCamelCase		:	int							=      self.pipeline_class(**__magic_name__      )
       lowerCamelCase		:	Optional[Any]							=      pipe.to(__magic_name__      )
       pipe.set_progress_bar_config(disable=__magic_name__      )

       lowerCamelCase		:	Optional[int]							=      1
       lowerCamelCase		:	Dict							=      2

       lowerCamelCase		:	Optional[Any]							=      self.get_dummy_inputs(__magic_name__      )

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

       lowerCamelCase		:	int							=      pipe(**__magic_name__    ,  num_images_per_prompt=__magic_name__      )[0]

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





@slow
@require_torch_gpu
class 							A__		(	unittest.TestCase):
      def 						UpperCamelCase__							(      self      ):
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()






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

       lowerCamelCase		:	List[Any]							=      torch.Generator(device=__magic_name__      ).manual_seed(0      )

       lowerCamelCase		:	Optional[int]							=      pipe(
           """a shark"""    ,  generator=__magic_name__    ,  guidance_scale=15.0    ,  num_inference_steps=6_4    ,  frame_size=6_4    ,  output_type="""np"""    ,  ).images[0]

       assert images.shape == (2_0, 6_4, 6_4, 3)

       assert_mean_pixel_difference(__magic_name__    ,  __magic_name__      )






import json
import os
import tempfile

import transformers

import datasets
from utils import generate_example_dataset, get_duration


_lowerCamelCase					   =5_0_0_0_0_0

_lowerCamelCase   ,  _lowerCamelCase					   =os.path.split(__file__)
_lowerCamelCase					   =os.path.join(RESULTS_BASEPATH, """results""", RESULTS_FILENAME.replace(""".py""", """.json"""))





@get_duration
def     _a					(						lowerCamelCase,	**lowerCamelCase							):
 lowerCamelCase		:	Optional[Any]							=      dataset.map(**lowerCamelCase							)





@get_duration
def     _a					(						lowerCamelCase,	**lowerCamelCase							):
 lowerCamelCase		:	Optional[Any]							=      dataset.filter(**lowerCamelCase							)





def     _a					(						):
 lowerCamelCase		:	Optional[Any]							=      {"""num examples""": SPEED_TEST_N_EXAMPLES}
 with tempfile.TemporaryDirectory() as tmp_dir:
  lowerCamelCase		:	Any							=      datasets.Features({"""text""": datasets.Value("""string"""							), """numbers""": datasets.Value("""float32"""							)}							)
  lowerCamelCase		:	Tuple							=      generate_example_dataset(
      os.path.join(lowerCamelCase,	"""dataset.arrow"""							),	lowerCamelCase,	num_examples=lowerCamelCase							)

  lowerCamelCase		:	Tuple							=      transformers.AutoTokenizer.from_pretrained("""bert-base-cased""",	use_fast=lowerCamelCase							)

  def tokenize(lowerCamelCase							):
   return tokenizer(examples["""text"""]							)

  lowerCamelCase		:	List[str]							=      map(lowerCamelCase							)

  lowerCamelCase		:	int							=      map(lowerCamelCase,	batched=lowerCamelCase							)

  lowerCamelCase		:	int							=      map(lowerCamelCase,	function=lambda lowerCamelCase							: None,	batched=lowerCamelCase							)

  with dataset.formatted_as(type="""numpy"""							):
   lowerCamelCase		:	Optional[int]							=      map(lowerCamelCase,	function=lambda lowerCamelCase							: None,	batched=lowerCamelCase							)

  with dataset.formatted_as(type="""pandas"""							):
   lowerCamelCase		:	List[str]							=      map(lowerCamelCase,	function=lambda lowerCamelCase							: None,	batched=lowerCamelCase							)

  with dataset.formatted_as(type="""torch""",	columns="""numbers"""							):
   lowerCamelCase		:	List[str]							=      map(lowerCamelCase,	function=lambda lowerCamelCase							: None,	batched=lowerCamelCase							)

  with dataset.formatted_as(type="""tensorflow""",	columns="""numbers"""							):
   lowerCamelCase		:	Optional[int]							=      map(lowerCamelCase,	function=lambda lowerCamelCase							: None,	batched=lowerCamelCase							)

  lowerCamelCase		:	int							=      map(lowerCamelCase,	function=lowerCamelCase,	batched=lowerCamelCase							)

  lowerCamelCase		:	Union[str, Any]							=      filter(lowerCamelCase							)

  # Activate later when tokenizer support batched inputs
  # with dataset.formatted_as(type='numpy'):
  #     times[func.__name__ + " fast-tokenizer batched numpy"] = func(dataset, function=tokenize, batched=True)

 with open(lowerCamelCase,	"""wb"""							) as f:
  f.write(json.dumps(lowerCamelCase							).encode("""utf-8"""							)							)


if __name__ == "__main__":  # useful to run the profiler
      benchmark_map_filter()




