"""simple docstring"""




from manim import *
class 		__A							(				__A	):






     def 					__A						(	self						):
       _lowerCAmelCase		: Optional[int]						    =  Rectangle(height=0.5       ,  width=0.5						)
       _lowerCAmelCase		: List[Any]						    =  Rectangle(height=0.2_5       ,  width=0.2_5						)
       _lowerCAmelCase		: Optional[int]						    =  Rectangle(height=0.4_6       ,  width=0.4_6						).set_stroke(width=0						)

       _lowerCAmelCase		: Optional[int]						    =  [mem.copy() for i in range(6						)]
       _lowerCAmelCase		: Optional[int]						    =  [mem.copy() for i in range(6						)]
       _lowerCAmelCase		: List[str]						    =  VGroup(*a__						).arrange(a__       ,  buff=0						)
       _lowerCAmelCase		: Tuple						    =  VGroup(*a__						).arrange(a__       ,  buff=0						)
       _lowerCAmelCase		: int						    =  VGroup(a__       ,  a__						).arrange(a__       ,  buff=0						)
       _lowerCAmelCase		: Optional[int]						    =  Text("""CPU"""       ,  font_size=24						)
       _lowerCAmelCase		: Any						    =  Group(a__       ,  a__						).arrange(a__       ,  buff=0.5       ,  aligned_edge=a__						)
       cpu.move_to([-2.5, -0.5, 0]						)
       self.add(a__						)

       _lowerCAmelCase		: Any						    =  [mem.copy() for i in range(4						)]
       _lowerCAmelCase		: List[str]						    =  VGroup(*a__						).arrange(a__       ,  buff=0						)
       _lowerCAmelCase		: int						    =  Text("""GPU"""       ,  font_size=24						)
       _lowerCAmelCase		: Optional[int]						    =  Group(a__       ,  a__						).arrange(a__       ,  buff=0.5       ,  aligned_edge=a__						)
       gpu.move_to([-1, -1, 0]						)
       self.add(a__						)

       _lowerCAmelCase		: Optional[Any]						    =  [mem.copy() for i in range(6						)]
       _lowerCAmelCase		: Optional[int]						    =  VGroup(*a__						).arrange(a__       ,  buff=0						)

       _lowerCAmelCase		: int						    =  Text("""Model"""       ,  font_size=24						)
       _lowerCAmelCase		: Any						    =  Group(a__       ,  a__						).arrange(a__       ,  buff=0.5       ,  aligned_edge=a__						)
       model.move_to([3, -1.0, 0]						)
       self.add(a__						)

       _lowerCAmelCase		: Optional[int]						    =  []
       _lowerCAmelCase		: Union[str, Any]						    =  []
       _lowerCAmelCase		: Dict						    =  []

       for i, rect in enumerate(a__						):
         rect.set_stroke(a__						)

         _lowerCAmelCase		: Tuple						    =  Rectangle(height=0.4_6 / 4       ,  width=0.4_6 / 3						).set_stroke(width=0.0						).set_fill(a__       ,  opacity=0.7						)

         if i == 0:
           cpu_target.next_to(cpu_left_col_base[0].get_corner(DOWN + LEFT						)       ,  buff=0.0_2       ,  direction=a__						)
           cpu_target.set_x(cpu_target.get_x() + 0.1						)
         elif i == 3:
           cpu_target.next_to(model_cpu_arr[0]       ,  direction=a__       ,  buff=0.0						)
         else:
           cpu_target.next_to(model_cpu_arr[i - 1]       ,  direction=a__       ,  buff=0.0						)
         self.add(a__						)
         model_cpu_arr.append(a__						)

       self.add(*a__       ,  *a__       ,  *a__						)

       _lowerCAmelCase		: Union[str, Any]						    =  [mem.copy() for i in range(6						)]
       _lowerCAmelCase		: Tuple						    =  VGroup(*a__						).arrange(a__       ,  buff=0						)

       _lowerCAmelCase		: int						    =  Text("""Loaded Checkpoint"""       ,  font_size=24						)
       _lowerCAmelCase		: Union[str, Any]						    =  Group(a__       ,  a__						).arrange(a__       ,  buff=0.5       ,  aligned_edge=a__						)
       checkpoint.move_to([3, 0.5, 0]						)

       self.add(a__						)

       _lowerCAmelCase		: Tuple						    =  []
       _lowerCAmelCase		: List[Any]						    =  []

       for i, rect in enumerate(a__						):
         _lowerCAmelCase		: Tuple						    =  fill.copy().set_fill(a__       ,  opacity=0.7						)
         target.move_to(a__						)
         ckpt_arr.append(a__						)

         _lowerCAmelCase		: Tuple						    =  target.copy()
         if i < 5:
           cpu_target.move_to(cpu_left_col_base[i + 1]						)
         else:
           cpu_target.move_to(cpu_right_col_base[i - 5]						)
         ckpt_cpu_arr.append(a__						)
       self.add(*a__       ,  *a__						)

       _lowerCAmelCase		: int						    =  Square(side_length=2.2						)
       key.move_to([-5, 2, 0]						)

       _lowerCAmelCase		: Tuple						    =  MarkupText(
           F"<b>Key:</b>\n\n<span fgcolor=\'{YELLOW}\'>●</span> Empty Model"       ,  font_size=18       ,  )

       key_text.move_to([-5, 2.4, 0]						)

       self.add(a__       ,  a__						)

       _lowerCAmelCase		: Any						    =  MarkupText(
           F"<span fgcolor=\'{BLUE}\'>●</span> Checkpoint"       ,  font_size=18       ,  )

       blue_text.next_to(a__       ,  DOWN * 2.4       ,  aligned_edge=key_text.get_left()						)
       self.add(a__						)

       _lowerCAmelCase		: Any						    =  MarkupText(
           F"Based on the passed in configuration, weights are stored in\na variety of np.memmaps on disk or to a particular device."       ,  font_size=24       ,  )
       step_a.move_to([2, 2, 0]						)

       _lowerCAmelCase		: Union[str, Any]						    =  [meta_mem.copy() for i in range(6						)]
       _lowerCAmelCase		: List[Any]						    =  [meta_mem.copy() for i in range(6						)]
       _lowerCAmelCase		: Union[str, Any]						    =  VGroup(*a__						).arrange(a__       ,  buff=0						)
       _lowerCAmelCase		: str						    =  VGroup(*a__						).arrange(a__       ,  buff=0						)
       _lowerCAmelCase		: Any						    =  VGroup(a__       ,  a__						).arrange(a__       ,  buff=0						)
       _lowerCAmelCase		: int						    =  Text("""Disk"""       ,  font_size=24						)
       _lowerCAmelCase		: Dict						    =  Group(a__       ,  a__						).arrange(a__       ,  buff=0.5       ,  aligned_edge=a__						)
       disk.move_to([-4.0, -1.2_5, 0]						)
       self.play(Write(a__       ,  run_time=3						)       ,  Write(a__       ,  run_time=1						)       ,  Create(a__       ,  run_time=1						)						)

       _lowerCAmelCase		: Optional[int]						    =  []
       for i, rect in enumerate(a__						):
         _lowerCAmelCase		: str						    =  rect.copy()
         target.generate_target()
         target.target.move_to(disk_left_col_base[i]						).scale(0.5						)
         animations.append(MoveToTarget(a__       ,  run_time=1.5						)						)
       self.play(*a__						)

       self.play(FadeOut(a__						)						)

       _lowerCAmelCase		: Any						    =  MarkupText(F"Then, the checkpoint is removed from memory\nthrough garbage collection."       ,  font_size=24						)
       step_a.move_to([2, 2, 0]						)

       self.play(Write(a__       ,  run_time=3						)						)

       self.play(
           FadeOut(a__       ,  a__       ,  *a__       ,  *a__						)       ,  )

       self.wait()





"""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__						)




