from manim import *




class 					_UpperCAmelCase			(     lowerCAmelCase  ):



 '''simple docstring'''




 def 		__UpperCAmelCase				(		self				:   List[Any])						->	Optional[int]:


   """simple docstring"""







   _UpperCamelCase           =    Rectangle(height=0.5							, width=0.5)
   _UpperCamelCase           =    Rectangle(height=0.46							, width=0.46).set_stroke(width=0)

   _UpperCamelCase           =    Rectangle(height=0.25							, width=0.25)

   _UpperCamelCase           =    [mem.copy() for i in range(6)]
   _UpperCamelCase           =    [mem.copy() for i in range(6)]
   _UpperCamelCase           =    VGroup(*lowercase_).arrange(lowercase_							, buff=0)
   _UpperCamelCase           =    VGroup(*lowercase_).arrange(lowercase_							, buff=0)
   _UpperCamelCase           =    VGroup(lowercase_							, lowercase_).arrange(lowercase_							, buff=0)
   _UpperCamelCase           =    Text("CPU"							, font_size=24)
   _UpperCamelCase           =    Group(lowercase_							, lowercase_).arrange(lowercase_							, buff=0.5							, aligned_edge=lowercase_)
   cpu.move_to([-2.5, -0.5, 0])
   self.add(lowercase_)

   _UpperCamelCase           =    [mem.copy() for i in range(4)]
   _UpperCamelCase           =    VGroup(*lowercase_).arrange(lowercase_							, buff=0)
   _UpperCamelCase           =    Text("GPU"							, font_size=24)
   _UpperCamelCase           =    Group(lowercase_							, lowercase_).arrange(lowercase_							, buff=0.5							, aligned_edge=lowercase_)
   gpu.move_to([-1, -1, 0])
   self.add(lowercase_)

   _UpperCamelCase           =    [mem.copy() for i in range(6)]
   _UpperCamelCase           =    VGroup(*lowercase_).arrange(lowercase_							, buff=0)

   _UpperCamelCase           =    Text("Model"							, font_size=24)
   _UpperCamelCase           =    Group(lowercase_							, lowercase_).arrange(lowercase_							, buff=0.5							, aligned_edge=lowercase_)
   model.move_to([3, -1.0, 0])
   self.add(lowercase_)

   _UpperCamelCase           =    []
   _UpperCamelCase           =    []

   for i, rect in enumerate(lowercase_):
     _UpperCamelCase           =    fill.copy().set_fill(lowercase_							, opacity=0.8)
     target.move_to(lowercase_)
     model_arr.append(lowercase_)

     _UpperCamelCase           =    Rectangle(height=0.46							, width=0.46).set_stroke(width=0.0).set_fill(lowercase_							, opacity=0.8)
     cpu_target.move_to(cpu_left_col_base[i])
     model_cpu_arr.append(lowercase_)

   self.add(*lowercase_							, *lowercase_)

   _UpperCamelCase           =    [meta_mem.copy() for i in range(6)]
   _UpperCamelCase           =    [meta_mem.copy() for i in range(6)]
   _UpperCamelCase           =    VGroup(*lowercase_).arrange(lowercase_							, buff=0)
   _UpperCamelCase           =    VGroup(*lowercase_).arrange(lowercase_							, buff=0)
   _UpperCamelCase           =    VGroup(lowercase_							, lowercase_).arrange(lowercase_							, buff=0)
   _UpperCamelCase           =    Text("Disk"							, font_size=24)
   _UpperCamelCase           =    Group(lowercase_							, lowercase_).arrange(lowercase_							, buff=0.5							, aligned_edge=lowercase_)
   disk.move_to([-4, -1.25, 0])
   self.add(lowercase_							, lowercase_)

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

   _UpperCamelCase           =    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(lowercase_							, lowercase_)

   _UpperCamelCase           =    MarkupText(
       f'<span fgcolor=\'{BLUE}\'>●</span> Checkpoint'							, font_size=18							, )

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

   _UpperCamelCase           =    MarkupText(
       f'Now watch as an input is passed through the model\nand how the memory is utilized and handled.'							, font_size=24							, )
   step_a.move_to([2, 2, 0])

   self.play(Write(lowercase_))

   _UpperCamelCase           =    Square(0.3)
   input.set_fill(lowercase_							, opacity=1.0)
   input.set_stroke(width=0.0)
   input.next_to(model_base[0]							, lowercase_							, buff=0.5)

   self.play(Write(lowercase_))

   input.generate_target()
   input.target.next_to(model_arr[0]							, direction=lowercase_							, buff=0.02)
   self.play(MoveToTarget(lowercase_))

   self.play(FadeOut(lowercase_))

   _UpperCamelCase           =    Arrow(start=lowercase_							, end=lowercase_							, color=lowercase_							, buff=0.5)
   a.next_to(model_arr[0].get_left()							, lowercase_							, buff=0.2)

   model_cpu_arr[0].generate_target()
   model_cpu_arr[0].target.move_to(gpu_rect[0])

   _UpperCamelCase           =    MarkupText(
       f'As the input reaches a layer, the hook triggers\nand weights are moved from the CPU\nto the GPU and back.'							, font_size=24							, )
   step_a.move_to([2, 2, 0])

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

   _UpperCamelCase           =    {"run_time": 1, "fade_in": True, "fade_out": True, "buff": 0.02}

   self.play(
       Write(lowercase_)							, Circumscribe(model_arr[0]							, color=lowercase_							, **lowercase_)							, Circumscribe(model_cpu_arr[0]							, color=lowercase_							, **lowercase_)							, Circumscribe(gpu_rect[0]							, color=lowercase_							, **lowercase_)							, )
   self.play(MoveToTarget(model_cpu_arr[0]))

   _UpperCamelCase           =    a.copy()
   for i in range(6):
     a_c.next_to(model_arr[i].get_right() + 0.02							, lowercase_							, buff=0.2)

     input.generate_target()
     input.target.move_to(model_arr[i].get_right() + 0.02)

     _UpperCamelCase           =    AnimationGroup(
         FadeOut(lowercase_							, run_time=0.5)							, MoveToTarget(lowercase_							, run_time=0.5)							, FadeIn(lowercase_							, run_time=0.5)							, lag_ratio=0.2)

     self.play(lowercase_)

     model_cpu_arr[i].generate_target()
     model_cpu_arr[i].target.move_to(cpu_left_col_base[i])

     if i < 5:
       model_cpu_arr[i + 1].generate_target()
       model_cpu_arr[i + 1].target.move_to(gpu_rect[0])
       if i >= 1:
         _UpperCamelCase           =    0.7

       self.play(
           Circumscribe(model_arr[i]							, **lowercase_)							, Circumscribe(cpu_left_col_base[i]							, **lowercase_)							, Circumscribe(cpu_left_col_base[i + 1]							, color=lowercase_							, **lowercase_)							, Circumscribe(gpu_rect[0]							, color=lowercase_							, **lowercase_)							, Circumscribe(model_arr[i + 1]							, color=lowercase_							, **lowercase_)							, )
       if i < 1:
         self.play(
             MoveToTarget(model_cpu_arr[i])							, MoveToTarget(model_cpu_arr[i + 1])							, )
       else:
         self.play(
             MoveToTarget(model_cpu_arr[i]							, run_time=0.7)							, MoveToTarget(model_cpu_arr[i + 1]							, run_time=0.7)							, )
     else:
       model_cpu_arr[i].generate_target()
       model_cpu_arr[i].target.move_to(cpu_left_col_base[-1])
       input.generate_target()
       input.target.next_to(model_arr[-1].get_right()							, RIGHT + 0.02							, buff=0.2)

       self.play(
           Circumscribe(model_arr[-1]							, color=lowercase_							, **lowercase_)							, Circumscribe(cpu_left_col_base[-1]							, color=lowercase_							, **lowercase_)							, Circumscribe(gpu_rect[0]							, color=lowercase_							, **lowercase_)							, )

       self.play(MoveToTarget(model_cpu_arr[i]))

     _UpperCamelCase           =    a_c
     _UpperCamelCase           =    a_c.copy()

   input.generate_target()
   input.target.next_to(model_base[-1]							, RIGHT + 0.02							, buff=0.5)
   self.play(
       FadeOut(lowercase_)							, FadeOut(lowercase_							, run_time=0.5)							, )

   _UpperCamelCase           =    MarkupText(f'Inference on a model too large for GPU memory\nis successfully completed.'							, font_size=24)
   step_a.move_to([2, 2, 0])

   self.play(Write(lowercase_							, run_time=3)							, MoveToTarget(lowercase_))

   self.wait()





import unittest

import numpy as np
import torch

from diffusers import DDIMPipeline, DDIMScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, slow, torch_device

from ..pipeline_params import UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS, UNCONDITIONAL_IMAGE_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()




class 					_UpperCAmelCase			(     lowerCAmelCase,			unittest.TestCase  ):



 '''simple docstring'''




 __A			       =					DDIMPipeline
 __A			       =					UNCONDITIONAL_IMAGE_GENERATION_PARAMS
 __A			       =					PipelineTesterMixin.required_optional_params - {
     '''num_images_per_prompt''',
     '''latents''',
     '''callback''',
     '''callback_steps''',
 }
 __A			       =					UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS
 __A			       =					False







 def 		__UpperCAmelCase				(		self				:   int)						->	List[str]:


   """simple docstring"""







   torch.manual_seed(0)
   _UpperCamelCase           =    UNetaDModel(
       block_out_channels=(32, 64)							, layers_per_block=2							, sample_size=32							, in_channels=3							, out_channels=3							, down_block_types=("DownBlock2D", "AttnDownBlock2D")							, up_block_types=("AttnUpBlock2D", "UpBlock2D")							, )
   _UpperCamelCase           =    DDIMScheduler()
   _UpperCamelCase           =    {"unet": unet, "scheduler": scheduler}
   return components







 def 		__UpperCAmelCase				(		self				:   int							, lowercase_				:   List[str]							, lowercase_				:   Optional[Any]=0)						->	Optional[Any]:


   """simple docstring"""







   if str(lowercase_).startswith("mps"):
     _UpperCamelCase           =    torch.manual_seed(lowercase_)
   else:
     _UpperCamelCase           =    torch.Generator(device=lowercase_).manual_seed(lowercase_)
   _UpperCamelCase           =    {
       "batch_size": 1,
       "generator": generator,
       "num_inference_steps": 2,
       "output_type": "numpy",
   }
   return inputs







 def 		__UpperCAmelCase				(		self				:   Dict)						->	Tuple:


   """simple docstring"""







   _UpperCamelCase           =    "cpu"

   _UpperCamelCase           =    self.get_dummy_components()
   _UpperCamelCase           =    self.pipeline_class(**lowercase_)
   pipe.to(lowercase_)
   pipe.set_progress_bar_config(disable=lowercase_)

   _UpperCamelCase           =    self.get_dummy_inputs(lowercase_)
   _UpperCamelCase           =    pipe(**lowercase_).images
   _UpperCamelCase           =    image[0, -3:, -3:, -1]

   self.assertEqual(image.shape							, (1, 32, 32, 3))
   _UpperCamelCase           =    np.array(
       [1.0_0_0e0_0, 5.7_1_7e-0_1, 4.7_1_7e-0_1, 1.0_0_0e0_0, 0.0_0_0e0_0, 1.0_0_0e0_0, 3.0_0_0e-0_4, 0.0_0_0e0_0, 9.0_0_0e-0_4])
   _UpperCamelCase           =    np.abs(image_slice.flatten() - expected_slice).max()
   self.assertLessEqual(lowercase_							, 1e-3)







 def 		__UpperCAmelCase				(		self				:   Optional[int])						->	Union[str, Any]:


   """simple docstring"""







   super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3)







 def 		__UpperCAmelCase				(		self				:   List[Any])						->	Union[str, Any]:


   """simple docstring"""







   super().test_save_load_local(expected_max_difference=3e-3)







 def 		__UpperCAmelCase				(		self				:   int)						->	str:


   """simple docstring"""







   super().test_save_load_optional_components(expected_max_difference=3e-3)







 def 		__UpperCAmelCase				(		self				:   List[Any])						->	Dict:


   """simple docstring"""







   super().test_inference_batch_single_identical(expected_max_diff=3e-3)




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



 '''simple docstring'''




 def 		__UpperCAmelCase				(		self				:   List[Any])						->	str:


   """simple docstring"""







   _UpperCamelCase           =    "google/ddpm-cifar10-32"

   _UpperCamelCase           =    UNetaDModel.from_pretrained(lowercase_)
   _UpperCamelCase           =    DDIMScheduler()

   _UpperCamelCase           =    DDIMPipeline(unet=lowercase_							, scheduler=lowercase_)
   ddim.to(lowercase_)
   ddim.set_progress_bar_config(disable=lowercase_)

   _UpperCamelCase           =    torch.manual_seed(0)
   _UpperCamelCase           =    ddim(generator=lowercase_							, eta=0.0							, output_type="numpy").images

   _UpperCamelCase           =    image[0, -3:, -3:, -1]

   assert image.shape == (1, 32, 32, 3)
   _UpperCamelCase           =    np.array([0.17_23, 0.16_17, 0.16_00, 0.16_26, 0.14_97, 0.15_13, 0.15_05, 0.14_42, 0.14_53])

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







 def 		__UpperCAmelCase				(		self				:   Tuple)						->	Tuple:


   """simple docstring"""







   _UpperCamelCase           =    "google/ddpm-ema-bedroom-256"

   _UpperCamelCase           =    UNetaDModel.from_pretrained(lowercase_)
   _UpperCamelCase           =    DDIMScheduler.from_pretrained(lowercase_)

   _UpperCamelCase           =    DDIMPipeline(unet=lowercase_							, scheduler=lowercase_)
   ddpm.to(lowercase_)
   ddpm.set_progress_bar_config(disable=lowercase_)

   _UpperCamelCase           =    torch.manual_seed(0)
   _UpperCamelCase           =    ddpm(generator=lowercase_							, output_type="numpy").images

   _UpperCamelCase           =    image[0, -3:, -3:, -1]

   assert image.shape == (1, 256, 256, 3)
   _UpperCamelCase           =    np.array([0.00_60, 0.02_01, 0.03_44, 0.00_24, 0.00_18, 0.00_02, 0.00_22, 0.00_00, 0.00_69])

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




