





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 							_UpperCAmelCase       (       _UpperCamelCase						,       unittest.TestCase       ):

  """simple docstring"""


  a_              =					ShapEPipeline
  a_              =					["""prompt"""]
  a_              =					["""prompt"""]
  a_              =					[
      """num_images_per_prompt""",
      """num_inference_steps""",
      """generator""",
      """latents""",
      """guidance_scale""",
      """frame_size""",
      """output_type""",
      """return_dict""",
  ]
  a_              =					False






  @property
  def 							lowercase						(							self					:						int	)				->		Any:
     return 3_2






  @property
  def 							lowercase						(							self					:						int	)				->		Optional[int]:
     return 3_2






  @property
  def 							lowercase						(							self					:						Union[str, Any]	)				->		Optional[int]:
     return self.time_input_dim * 4






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






  @property
  def 							lowercase						(							self					:						Optional[int]	)				->		List[str]:
     __lowerCAmelCase				      =      CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'	)
     return tokenizer






  @property
  def 							lowercase						(							self					:						Optional[Any]	)				->		Optional[int]:
     torch.manual_seed(0	)
     __lowerCAmelCase				      =      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(lowerCAmelCase_	)






  @property
  def 							lowercase						(							self					:						str	)				->		Optional[Any]:
     torch.manual_seed(0	)

     __lowerCAmelCase				      =      {
         '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				      =      PriorTransformer(**lowerCAmelCase_	)
     return model






  @property
  def 							lowercase						(							self					:						Union[str, Any]	)				->		List[Any]:
     torch.manual_seed(0	)

     __lowerCAmelCase				      =      {
         '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				      =      ShapERenderer(**lowerCAmelCase_	)
     return model






  def 							lowercase						(							self					:						Tuple	)				->		Dict:
     __lowerCAmelCase				      =      self.dummy_prior
     __lowerCAmelCase				      =      self.dummy_text_encoder
     __lowerCAmelCase				      =      self.dummy_tokenizer
     __lowerCAmelCase				      =      self.dummy_renderer

     __lowerCAmelCase				      =      HeunDiscreteScheduler(
         beta_schedule='exp'   ,    num_train_timesteps=1_0_2_4   ,    prediction_type='sample'   ,    use_karras_sigmas=lowerCAmelCase_   ,    clip_sample=lowerCAmelCase_   ,    clip_sample_range=1.0   ,    )
     __lowerCAmelCase				      =      {
         'prior': prior,
         'text_encoder': text_encoder,
         'tokenizer': tokenizer,
         'renderer': renderer,
         'scheduler': scheduler,
     }

     return components






  def 							lowercase						(							self					:						List[Any]   ,    lowerCAmelCase_					:						Any   ,    lowerCAmelCase_					:						Optional[int]=0	)				->		int:
     if str(lowerCAmelCase_	).startswith('mps'	):
        __lowerCAmelCase				      =      torch.manual_seed(lowerCAmelCase_	)
     else:
        __lowerCAmelCase				      =      torch.Generator(device=lowerCAmelCase_	).manual_seed(lowerCAmelCase_	)
     __lowerCAmelCase				      =      {
         'prompt': 'horse',
         'generator': generator,
         'num_inference_steps': 1,
         'frame_size': 3_2,
         'output_type': 'np',
     }
     return inputs






  def 							lowercase						(							self					:						List[str]	)				->		str:
     __lowerCAmelCase				      =      'cpu'

     __lowerCAmelCase				      =      self.get_dummy_components()

     __lowerCAmelCase				      =      self.pipeline_class(**lowerCAmelCase_	)
     __lowerCAmelCase				      =      pipe.to(lowerCAmelCase_	)

     pipe.set_progress_bar_config(disable=lowerCAmelCase_	)

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

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

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






  def 							lowercase						(							self					:						Any	)				->		str:
     __lowerCAmelCase				      =      torch_device == 'cpu'
     __lowerCAmelCase				      =      True

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






  def 							lowercase						(							self					:						List[Any]	)				->		Optional[Any]:
     __lowerCAmelCase				      =      self.get_dummy_components()
     __lowerCAmelCase				      =      self.pipeline_class(**lowerCAmelCase_	)
     __lowerCAmelCase				      =      pipe.to(lowerCAmelCase_	)
     pipe.set_progress_bar_config(disable=lowerCAmelCase_	)

     __lowerCAmelCase				      =      1
     __lowerCAmelCase				      =      2

     __lowerCAmelCase				      =      self.get_dummy_inputs(lowerCAmelCase_	)

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

     __lowerCAmelCase				      =      pipe(**lowerCAmelCase_   ,    num_images_per_prompt=lowerCAmelCase_	)[0]

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





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

  """simple docstring"""


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






  def 							lowercase						(							self					:						int	)				->		Dict:
     __lowerCAmelCase				      =      load_numpy(
         'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
         '/shap_e/test_shap_e_np_out.npy'	)
     __lowerCAmelCase				      =      ShapEPipeline.from_pretrained('openai/shap-e'	)
     __lowerCAmelCase				      =      pipe.to(lowerCAmelCase_	)
     pipe.set_progress_bar_config(disable=lowerCAmelCase_	)

     __lowerCAmelCase				      =      torch.Generator(device=lowerCAmelCase_	).manual_seed(0	)

     __lowerCAmelCase				      =      pipe(
         'a shark'   ,    generator=lowerCAmelCase_   ,    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(lowerCAmelCase_   ,    lowerCAmelCase_	)









# Author: OMKAR PATHAK, Nwachukwu Chidiebere

# Use a Python dictionary to construct the graph.
from __future__ import annotations

from pprint import pformat
from typing import Generic, TypeVar

_SCREAMING_SNAKE_CASE   =  TypeVar("T")






class 					SCREAMING_SNAKE_CASE_  (       Generic[T] ):
       """simple docstring"""


       def __init__(				self	:List[Any],      snake_case	:bool = True):



          """simple docstring"""

          _lowercase              ={}  # dictionary of lists
          _lowercase              =directed





       def 				UpperCamelCase__						(				self	:Optional[Any],      snake_case	:T,      snake_case	:T):



          """simple docstring"""

          if not self.directed:  # For undirected graphs
             # if both source vertex and destination vertex are both present in the
             # adjacency list, add destination vertex to source vertex list of adjacent
             # vertices and add source vertex to destination vertex list of adjacent
             # vertices.
             if source_vertex in self.adj_list and destination_vertex in self.adj_list:
                self.adj_list[source_vertex].append(snake_case)
                self.adj_list[destination_vertex].append(snake_case)
             # if only source vertex is present in adjacency list, add destination vertex
             # to source vertex list of adjacent vertices, then create a new vertex with
             # destination vertex as key and assign a list containing the source vertex
             # as it's first adjacent vertex.
             elif source_vertex in self.adj_list:
                self.adj_list[source_vertex].append(snake_case)
                _lowercase              =[source_vertex]
             # if only destination vertex is present in adjacency list, add source vertex
             # to destination vertex list of adjacent vertices, then create a new vertex
             # with source vertex as key and assign a list containing the source vertex
             # as it's first adjacent vertex.
             elif destination_vertex in self.adj_list:
                self.adj_list[destination_vertex].append(snake_case)
                _lowercase              =[destination_vertex]
             # if both source vertex and destination vertex are not present in adjacency
             # list, create a new vertex with source vertex as key and assign a list
             # containing the destination vertex as it's first adjacent vertex also
             # create a new vertex with destination vertex as key and assign a list
             # containing the source vertex as it's first adjacent vertex.
             else:
                _lowercase              =[destination_vertex]
                _lowercase              =[source_vertex]
          else:  # For directed graphs
             # if both source vertex and destination vertex are present in adjacency
             # list, add destination vertex to source vertex list of adjacent vertices.
             if source_vertex in self.adj_list and destination_vertex in self.adj_list:
                self.adj_list[source_vertex].append(snake_case)
             # if only source vertex is present in adjacency list, add destination
             # vertex to source vertex list of adjacent vertices and create a new vertex
             # with destination vertex as key, which has no adjacent vertex
             elif source_vertex in self.adj_list:
                self.adj_list[source_vertex].append(snake_case)
                _lowercase              =[]
             # if only destination vertex is present in adjacency list, create a new
             # vertex with source vertex as key and assign a list containing destination
             # vertex as first adjacent vertex
             elif destination_vertex in self.adj_list:
                _lowercase              =[destination_vertex]
             # if both source vertex and destination vertex are not present in adjacency
             # list, create a new vertex with source vertex as key and a list containing
             # destination vertex as it's first adjacent vertex. Then create a new vertex
             # with destination vertex as key, which has no adjacent vertex
             else:
                _lowercase              =[destination_vertex]
                _lowercase              =[]

          return self





       def __repr__(				self	:Optional[int]):



          """simple docstring"""

          return pformat(self.adj_list)


