






from manim import *







class 		_A						( UpperCAmelCase_ ):

  def    a					(    self						:	Dict			):
     """simple docstring"""

     __UpperCamelCase	:			Tuple					=					Rectangle(height=0.5   ,				width=0.5			)
     __UpperCamelCase	:			Tuple					=					Rectangle(height=0.46   ,				width=0.46			).set_stroke(width=0			)

     __UpperCamelCase	:			Optional[Any]					=					Rectangle(height=0.25   ,				width=0.25			)

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

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

     __UpperCamelCase	:			Dict					=					[mem.copy() for i in range(6			)]
     __UpperCamelCase	:			Tuple					=					VGroup(*lowerCamelCase__			).arrange(lowerCamelCase__   ,				buff=0			)

     __UpperCamelCase	:			Tuple					=					Text("""Model"""   ,				font_size=24			)
     __UpperCamelCase	:			List[str]					=					Group(lowerCamelCase__   ,				lowerCamelCase__			).arrange(lowerCamelCase__   ,				buff=0.5   ,				aligned_edge=lowerCamelCase__			)
     model.move_to([3, -1.0, 0]			)
     self.add(lowerCamelCase__			)

     __UpperCamelCase	:			Tuple					=					[]
     __UpperCamelCase	:			str					=					[]

     for i, rect in enumerate(lowerCamelCase__			):
        __UpperCamelCase	:			Optional[Any]					=					fill.copy().set_fill(lowerCamelCase__   ,				opacity=0.8			)
        target.move_to(lowerCamelCase__			)
        model_arr.append(lowerCamelCase__			)

        __UpperCamelCase	:			Tuple					=					Rectangle(height=0.46   ,				width=0.46			).set_stroke(width=0.0			).set_fill(lowerCamelCase__   ,				opacity=0.8			)
        cpu_target.move_to(cpu_left_col_base[i]			)
        model_cpu_arr.append(lowerCamelCase__			)

     self.add(*lowerCamelCase__   ,				*lowerCamelCase__			)

     __UpperCamelCase	:			Union[str, Any]					=					[meta_mem.copy() for i in range(6			)]
     __UpperCamelCase	:			Any					=					[meta_mem.copy() for i in range(6			)]
     __UpperCamelCase	:			int					=					VGroup(*lowerCamelCase__			).arrange(lowerCamelCase__   ,				buff=0			)
     __UpperCamelCase	:			Union[str, Any]					=					VGroup(*lowerCamelCase__			).arrange(lowerCamelCase__   ,				buff=0			)
     __UpperCamelCase	:			str					=					VGroup(lowerCamelCase__   ,				lowerCamelCase__			).arrange(lowerCamelCase__   ,				buff=0			)
     __UpperCamelCase	:			Optional[Any]					=					Text("""Disk"""   ,				font_size=24			)
     __UpperCamelCase	:			Optional[Any]					=					Group(lowerCamelCase__   ,				lowerCamelCase__			).arrange(lowerCamelCase__   ,				buff=0.5   ,				aligned_edge=lowerCamelCase__			)
     disk.move_to([-4, -1.25, 0]			)
     self.add(lowerCamelCase__   ,				lowerCamelCase__			)

     __UpperCamelCase	:			Union[str, Any]					=					Square(side_length=2.2			)
     key.move_to([-5, 2, 0]			)

     __UpperCamelCase	:			Optional[Any]					=					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(lowerCamelCase__   ,				lowerCamelCase__			)

     __UpperCamelCase	:			Optional[int]					=					MarkupText(
         f'<span fgcolor=\'{BLUE}\'>●</span> Checkpoint'   ,				font_size=18   ,				)

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

     __UpperCamelCase	:			Any					=					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(lowerCamelCase__			)			)

     __UpperCamelCase	:			List[str]					=					Square(0.3			)
     input.set_fill(lowerCamelCase__   ,				opacity=1.0			)
     input.set_stroke(width=0.0			)
     input.next_to(model_base[0]   ,				lowerCamelCase__   ,				buff=0.5			)

     self.play(Write(lowerCamelCase__			)			)

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

     self.play(FadeOut(lowerCamelCase__			)			)

     __UpperCamelCase	:			Dict					=					Arrow(start=lowerCamelCase__   ,				end=lowerCamelCase__   ,				color=lowerCamelCase__   ,				buff=0.5			)
     a.next_to(model_arr[0].get_left()   ,				lowerCamelCase__   ,				buff=0.2			)

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

     __UpperCamelCase	:			int					=					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(lowerCamelCase__   ,				run_time=3			)			)

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

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

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

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

        __UpperCamelCase	:			Tuple					=					AnimationGroup(
            FadeOut(lowerCamelCase__   ,				run_time=0.5			)   ,				MoveToTarget(lowerCamelCase__   ,				run_time=0.5			)   ,				FadeIn(lowerCamelCase__   ,				run_time=0.5			)   ,				lag_ratio=0.2			)

        self.play(lowerCamelCase__			)

        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	:			Any					=					0.7

           self.play(
               Circumscribe(model_arr[i]   ,				**lowerCamelCase__			)   ,				Circumscribe(cpu_left_col_base[i]   ,				**lowerCamelCase__			)   ,				Circumscribe(cpu_left_col_base[i + 1]   ,				color=lowerCamelCase__   ,				**lowerCamelCase__			)   ,				Circumscribe(gpu_rect[0]   ,				color=lowerCamelCase__   ,				**lowerCamelCase__			)   ,				Circumscribe(model_arr[i + 1]   ,				color=lowerCamelCase__   ,				**lowerCamelCase__			)   ,				)
           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=lowerCamelCase__   ,				**lowerCamelCase__			)   ,				Circumscribe(cpu_left_col_base[-1]   ,				color=lowerCamelCase__   ,				**lowerCamelCase__			)   ,				Circumscribe(gpu_rect[0]   ,				color=lowerCamelCase__   ,				**lowerCamelCase__			)   ,				)

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

        __UpperCamelCase	:			str					=					a_c
        __UpperCamelCase	:			List[Any]					=					a_c.copy()

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

     __UpperCamelCase	:			List[str]					=					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(lowerCamelCase__   ,				run_time=3			)   ,				MoveToTarget(lowerCamelCase__			)			)

     self.wait()











import gc
import random
import unittest

import numpy as np
import torch

from diffusers import DDIMScheduler, KandinskyVaaPipeline, KandinskyVaaPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()







class 		_A						( UpperCAmelCase_	,				unittest.TestCase ):
  lowercase_						:			int						=					KandinskyVaaPipeline
  lowercase_						:			Union[str, Any]						=					[
      '''image_embeds''',
      '''negative_image_embeds''',
  ]
  lowercase_						:			int						=					['''image_embeds''', '''negative_image_embeds''']
  lowercase_						:			str						=					[
      '''generator''',
      '''height''',
      '''width''',
      '''latents''',
      '''guidance_scale''',
      '''num_inference_steps''',
      '''return_dict''',
      '''guidance_scale''',
      '''num_images_per_prompt''',
      '''output_type''',
      '''return_dict''',
  ]
  lowercase_						:			str						=					False

  @property
  def    a					(    self						:	int			):
     """simple docstring"""

     return 32

  @property
  def    a					(    self						:	str			):
     """simple docstring"""

     return 32

  @property
  def    a					(    self						:	Optional[int]			):
     """simple docstring"""

     return self.time_input_dim

  @property
  def    a					(    self						:	Optional[int]			):
     """simple docstring"""

     return self.time_input_dim * 4

  @property
  def    a					(    self						:	Optional[Any]			):
     """simple docstring"""

     return 1_00

  @property
  def    a					(    self						:	str			):
     """simple docstring"""

     torch.manual_seed(0			)

     __UpperCamelCase	:			int					=					{
         """in_channels""": 4,
         # Out channels is double in channels because predicts mean and variance
         """out_channels""": 8,
         """addition_embed_type""": """image""",
         """down_block_types""": ("""ResnetDownsampleBlock2D""", """SimpleCrossAttnDownBlock2D"""),
         """up_block_types""": ("""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""),
         """mid_block_type""": """UNetMidBlock2DSimpleCrossAttn""",
         """block_out_channels""": (self.block_out_channels_a, self.block_out_channels_a * 2),
         """layers_per_block""": 1,
         """encoder_hid_dim""": self.text_embedder_hidden_size,
         """encoder_hid_dim_type""": """image_proj""",
         """cross_attention_dim""": self.cross_attention_dim,
         """attention_head_dim""": 4,
         """resnet_time_scale_shift""": """scale_shift""",
         """class_embed_type""": None,
     }

     __UpperCamelCase	:			Any					=					UNetaDConditionModel(**lowerCamelCase__			)
     return model

  @property
  def    a					(    self						:	Union[str, Any]			):
     """simple docstring"""

     return {
         "block_out_channels": [32, 64],
         "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
         "in_channels": 3,
         "latent_channels": 4,
         "layers_per_block": 1,
         "norm_num_groups": 8,
         "norm_type": "spatial",
         "num_vq_embeddings": 12,
         "out_channels": 3,
         "up_block_types": [
             "AttnUpDecoderBlock2D",
             "UpDecoderBlock2D",
         ],
         "vq_embed_dim": 4,
     }

  @property
  def    a					(    self						:	int			):
     """simple docstring"""

     torch.manual_seed(0			)
     __UpperCamelCase	:			Tuple					=					VQModel(**self.dummy_movq_kwargs			)
     return model

  def    a					(    self						:	Tuple			):
     """simple docstring"""

     __UpperCamelCase	:			str					=					self.dummy_unet
     __UpperCamelCase	:			Optional[int]					=					self.dummy_movq

     __UpperCamelCase	:			List[str]					=					DDIMScheduler(
         num_train_timesteps=10_00   ,				beta_schedule="""linear"""   ,				beta_start=0.0_0085   ,				beta_end=0.012   ,				clip_sample=lowerCamelCase__   ,				set_alpha_to_one=lowerCamelCase__   ,				steps_offset=1   ,				prediction_type="""epsilon"""   ,				thresholding=lowerCamelCase__   ,				)

     __UpperCamelCase	:			str					=					{
         """unet""": unet,
         """scheduler""": scheduler,
         """movq""": movq,
     }
     return components

  def    a					(    self						:	Optional[int]   ,				lowerCamelCase__						:	Optional[Any]   ,				lowerCamelCase__						:	Any=0			):
     """simple docstring"""

     __UpperCamelCase	:			Tuple					=					floats_tensor((1, self.text_embedder_hidden_size)   ,				rng=random.Random(lowerCamelCase__			)			).to(lowerCamelCase__			)
     __UpperCamelCase	:			Optional[int]					=					floats_tensor((1, self.text_embedder_hidden_size)   ,				rng=random.Random(seed + 1			)			).to(
         lowerCamelCase__			)
     if str(lowerCamelCase__			).startswith("""mps"""			):
        __UpperCamelCase	:			Optional[int]					=					torch.manual_seed(lowerCamelCase__			)
     else:
        __UpperCamelCase	:			int					=					torch.Generator(device=lowerCamelCase__			).manual_seed(lowerCamelCase__			)
     __UpperCamelCase	:			Dict					=					{
         """image_embeds""": image_embeds,
         """negative_image_embeds""": negative_image_embeds,
         """generator""": generator,
         """height""": 64,
         """width""": 64,
         """guidance_scale""": 4.0,
         """num_inference_steps""": 2,
         """output_type""": """np""",
     }
     return inputs

  def    a					(    self						:	Dict			):
     """simple docstring"""

     __UpperCamelCase	:			Dict					=					"""cpu"""

     __UpperCamelCase	:			Dict					=					self.get_dummy_components()

     __UpperCamelCase	:			Union[str, Any]					=					self.pipeline_class(**lowerCamelCase__			)
     __UpperCamelCase	:			int					=					pipe.to(lowerCamelCase__			)

     pipe.set_progress_bar_config(disable=lowerCamelCase__			)

     __UpperCamelCase	:			List[Any]					=					pipe(**self.get_dummy_inputs(lowerCamelCase__			)			)
     __UpperCamelCase	:			List[Any]					=					output.images

     __UpperCamelCase	:			int					=					pipe(
         **self.get_dummy_inputs(lowerCamelCase__			)   ,				return_dict=lowerCamelCase__   ,				)[0]

     __UpperCamelCase	:			List[str]					=					image[0, -3:, -3:, -1]
     __UpperCamelCase	:			Tuple					=					image_from_tuple[0, -3:, -3:, -1]

     assert image.shape == (1, 64, 64, 3)

     __UpperCamelCase	:			int					=					np.array(
         [0.623_7976, 1.0, 0.3644_1332, 1.0, 0.7063_9634, 0.2987_7186, 0.8565_2125, 0.521_6843, 0.5445_4046]			)

     assert (
         np.abs(image_slice.flatten() - expected_slice			).max() < 1e-2
     ), f' expected_slice {expected_slice}, but got {image_slice.flatten()}'

     assert (
         np.abs(image_from_tuple_slice.flatten() - expected_slice			).max() < 1e-2
     ), f' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'







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

  def    a					(    self						:	Optional[int]			):
     """simple docstring"""

     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()

  def    a					(    self						:	List[Any]			):
     """simple docstring"""

     __UpperCamelCase	:			int					=					load_numpy(
         """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
         """/kandinskyv22/kandinskyv22_text2img_cat_fp16.npy"""			)

     __UpperCamelCase	:			List[str]					=					KandinskyVaaPriorPipeline.from_pretrained(
         """kandinsky-community/kandinsky-2-2-prior"""   ,				torch_dtype=torch.floataa			)
     pipe_prior.to(lowerCamelCase__			)

     __UpperCamelCase	:			int					=					KandinskyVaaPipeline.from_pretrained(
         """kandinsky-community/kandinsky-2-2-decoder"""   ,				torch_dtype=torch.floataa			)
     __UpperCamelCase	:			Optional[int]					=					pipeline.to(lowerCamelCase__			)
     pipeline.set_progress_bar_config(disable=lowerCamelCase__			)

     __UpperCamelCase	:			Optional[Any]					=					"""red cat, 4k photo"""

     __UpperCamelCase	:			Tuple					=					torch.Generator(device="""cuda"""			).manual_seed(0			)
     __UpperCamelCase	,     __UpperCamelCase	:			Union[str, Any]					=					pipe_prior(
         lowerCamelCase__   ,				generator=lowerCamelCase__   ,				num_inference_steps=5   ,				negative_prompt=""""""   ,				).to_tuple()

     __UpperCamelCase	:			Any					=					torch.Generator(device="""cuda"""			).manual_seed(0			)
     __UpperCamelCase	:			Union[str, Any]					=					pipeline(
         image_embeds=lowerCamelCase__   ,				negative_image_embeds=lowerCamelCase__   ,				generator=lowerCamelCase__   ,				num_inference_steps=1_00   ,				output_type="""np"""   ,				)

     __UpperCamelCase	:			Any					=					output.images[0]

     assert image.shape == (5_12, 5_12, 3)

     assert_mean_pixel_difference(lowerCamelCase__   ,				lowerCamelCase__			)



