



'''simple docstring'''

import gc
import unittest

import numpy as np
import torch
from torch.backends.cuda import sdp_kernel

from diffusers import (
    CMStochasticIterativeScheduler,
    ConsistencyModelPipeline,
    UNetaDModel,
)
from diffusers.utils import randn_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_a, require_torch_gpu

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


enable_full_determinism()


class 	a__     (				_lowercase,			unittest.TestCase	):
	__magic_name__		:    Tuple	    =     ConsistencyModelPipeline
	__magic_name__		:    str	    =     UNCONDITIONAL_IMAGE_GENERATION_PARAMS
	__magic_name__		:    Optional[Any]	    =     UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

	# Override required_optional_params to remove num_images_per_prompt
	__magic_name__		:    Optional[int]	    =     frozenset(
	    [
	        "num_inference_steps",
	        "generator",
	        "latents",
	        "output_type",
	        "return_dict",
	        "callback",
	        "callback_steps",
	    ]	)


	@property
	def 				lowercase__  (self : List[Any]      )     ->			str:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						List[Any]      					= UNetaDModel.from_pretrained(
					    '''diffusers/consistency-models-test''',					subfolder='''test_unet''',					)
					return unet


	@property
	def 				lowercase__  (self : Any      )     ->			Dict:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= UNetaDModel.from_pretrained(
					    '''diffusers/consistency-models-test''',					subfolder='''test_unet_class_cond''',					)
					return unet


	def 				lowercase__  (self : List[str],					__UpperCAmelCase : Any=False      )     ->			List[Any]:

					"""simple docstring"""

					if class_cond:
									SCREAMING_SNAKE_CASE  :						Any      					= self.dummy_cond_unet
					else:
									SCREAMING_SNAKE_CASE  :						Tuple      					= self.dummy_uncond_unet

					# Default to CM multistep sampler
					SCREAMING_SNAKE_CASE  :						Tuple      					= CMStochasticIterativeScheduler(
					    num_train_timesteps=40,					sigma_min=0.002,					sigma_max=80.0,					)

					SCREAMING_SNAKE_CASE  :						List[Any]      					= {
					    '''unet''': unet,
					    '''scheduler''': scheduler,
					}

					return components


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

					"""simple docstring"""

					if str(__UpperCAmelCase      ).startswith('''mps'''      ):
									SCREAMING_SNAKE_CASE  :						str      					= torch.manual_seed(__UpperCAmelCase      )
					else:
									SCREAMING_SNAKE_CASE  :						str      					= torch.Generator(device=__UpperCAmelCase      ).manual_seed(__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						Optional[int]      					= {
					    '''batch_size''': 1,
					    '''num_inference_steps''': None,
					    '''timesteps''': [22, 0],
					    '''generator''': generator,
					    '''output_type''': '''np''',
					}

					return inputs


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

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						Any      					= '''cpu'''  # ensure determinism for the device-dependent torch.Generator
					SCREAMING_SNAKE_CASE  :						Optional[Any]      					= self.get_dummy_components()
					SCREAMING_SNAKE_CASE  :						Dict      					= ConsistencyModelPipeline(**__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						Any      					= pipe.to(__UpperCAmelCase      )
					pipe.set_progress_bar_config(disable=__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						Optional[Any]      					= self.get_dummy_inputs(__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						str      					= pipe(**__UpperCAmelCase      ).images
					assert image.shape == (1, 32, 32, 3)

					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= image[0, -3:, -3:, -1]
					SCREAMING_SNAKE_CASE  :						Dict      					= np.array([0.3572, 0.6273, 0.4031, 0.3961, 0.4321, 0.5730, 0.5266, 0.4780, 0.5004]      )

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


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

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						List[Any]      					= '''cpu'''  # ensure determinism for the device-dependent torch.Generator
					SCREAMING_SNAKE_CASE  :						List[Any]      					= self.get_dummy_components(class_cond=__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= ConsistencyModelPipeline(**__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						int      					= pipe.to(__UpperCAmelCase      )
					pipe.set_progress_bar_config(disable=__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						List[str]      					= self.get_dummy_inputs(__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						Tuple      					= 0
					SCREAMING_SNAKE_CASE  :						Tuple      					= pipe(**__UpperCAmelCase      ).images
					assert image.shape == (1, 32, 32, 3)

					SCREAMING_SNAKE_CASE  :						str      					= image[0, -3:, -3:, -1]
					SCREAMING_SNAKE_CASE  :						Any      					= np.array([0.3572, 0.6273, 0.4031, 0.3961, 0.4321, 0.5730, 0.5266, 0.4780, 0.5004]      )

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


	def 				lowercase__  (self : int      )     ->			Tuple:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						Tuple      					= '''cpu'''  # ensure determinism for the device-dependent torch.Generator
					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= self.get_dummy_components()
					SCREAMING_SNAKE_CASE  :						List[Any]      					= ConsistencyModelPipeline(**__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						List[Any]      					= pipe.to(__UpperCAmelCase      )
					pipe.set_progress_bar_config(disable=__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						Tuple      					= self.get_dummy_inputs(__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						Optional[int]      					= 1
					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= None
					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= pipe(**__UpperCAmelCase      ).images
					assert image.shape == (1, 32, 32, 3)

					SCREAMING_SNAKE_CASE  :						Tuple      					= image[0, -3:, -3:, -1]
					SCREAMING_SNAKE_CASE  :						Dict      					= np.array([0.5004, 0.5004, 0.4994, 0.5008, 0.4976, 0.5018, 0.4990, 0.4982, 0.4987]      )

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






	def 				lowercase__  (self : int      )     ->			int:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						Optional[Any]      					= '''cpu'''  # ensure determinism for the device-dependent torch.Generator
					SCREAMING_SNAKE_CASE  :						List[str]      					= self.get_dummy_components(class_cond=__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						Optional[int]      					= ConsistencyModelPipeline(**__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						Tuple      					= pipe.to(__UpperCAmelCase      )
					pipe.set_progress_bar_config(disable=__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= self.get_dummy_inputs(__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						List[str]      					= 1
					SCREAMING_SNAKE_CASE  :						str      					= None
					SCREAMING_SNAKE_CASE  :						List[str]      					= 0
					SCREAMING_SNAKE_CASE  :						List[str]      					= pipe(**__UpperCAmelCase      ).images
					assert image.shape == (1, 32, 32, 3)

					SCREAMING_SNAKE_CASE  :						Any      					= image[0, -3:, -3:, -1]
					SCREAMING_SNAKE_CASE  :						List[Any]      					= np.array([0.5004, 0.5004, 0.4994, 0.5008, 0.4976, 0.5018, 0.4990, 0.4982, 0.4987]      )

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



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


	def 				lowercase__  (self : Tuple      )     ->			Tuple:

					"""simple docstring"""

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


	def 				lowercase__  (self : Union[str, Any],					__UpperCAmelCase : Any=0,					__UpperCAmelCase : str=False,					__UpperCAmelCase : str="cpu",					__UpperCAmelCase : List[Any]=torch.floataa,					__UpperCAmelCase : Optional[Any]=(1, 3, 64, 64)      )     ->			Union[str, Any]:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						Optional[Any]      					= torch.manual_seed(__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						Optional[Any]      					= {
					    '''num_inference_steps''': None,
					    '''timesteps''': [22, 0],
					    '''class_labels''': 0,
					    '''generator''': generator,
					    '''output_type''': '''np''',
					}

					if get_fixed_latents:
									SCREAMING_SNAKE_CASE  :						Optional[Any]      					= self.get_fixed_latents(seed=__UpperCAmelCase,					device=__UpperCAmelCase,					dtype=__UpperCAmelCase,					shape=__UpperCAmelCase      )
									SCREAMING_SNAKE_CASE  :						List[Any]      					= latents

					return inputs


	def 				lowercase__  (self : str,					__UpperCAmelCase : int=0,					__UpperCAmelCase : Dict="cpu",					__UpperCAmelCase : Union[str, Any]=torch.floataa,					__UpperCAmelCase : Optional[Any]=(1, 3, 64, 64)      )     ->			List[Any]:

					"""simple docstring"""

					if type(__UpperCAmelCase      ) == str:
									SCREAMING_SNAKE_CASE  :						Optional[int]      					= torch.device(__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						Optional[Any]      					= torch.Generator(device=__UpperCAmelCase      ).manual_seed(__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						List[str]      					= randn_tensor(__UpperCAmelCase,					generator=__UpperCAmelCase,					device=__UpperCAmelCase,					dtype=__UpperCAmelCase      )
					return latents


	def 				lowercase__  (self : Any      )     ->			Any:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= UNetaDModel.from_pretrained('''diffusers/consistency_models''',					subfolder='''diffusers_cd_imagenet64_l2'''      )
					SCREAMING_SNAKE_CASE  :						List[Any]      					= CMStochasticIterativeScheduler(
					    num_train_timesteps=40,					sigma_min=0.002,					sigma_max=80.0,					)
					SCREAMING_SNAKE_CASE  :						List[str]      					= ConsistencyModelPipeline(unet=__UpperCAmelCase,					scheduler=__UpperCAmelCase      )
					pipe.to(torch_device=__UpperCAmelCase      )
					pipe.set_progress_bar_config(disable=__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						Tuple      					= self.get_inputs()
					SCREAMING_SNAKE_CASE  :						Optional[Any]      					= pipe(**__UpperCAmelCase      ).images
					assert image.shape == (1, 64, 64, 3)

					SCREAMING_SNAKE_CASE  :						List[Any]      					= image[0, -3:, -3:, -1]

					SCREAMING_SNAKE_CASE  :						List[str]      					= np.array([0.0888, 0.0881, 0.0666, 0.0479, 0.0292, 0.0195, 0.0201, 0.0163, 0.0254]      )

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


	def 				lowercase__  (self : int      )     ->			Any:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						int      					= UNetaDModel.from_pretrained('''diffusers/consistency_models''',					subfolder='''diffusers_cd_imagenet64_l2'''      )
					SCREAMING_SNAKE_CASE  :						Tuple      					= CMStochasticIterativeScheduler(
					    num_train_timesteps=40,					sigma_min=0.002,					sigma_max=80.0,					)
					SCREAMING_SNAKE_CASE  :						Dict      					= ConsistencyModelPipeline(unet=__UpperCAmelCase,					scheduler=__UpperCAmelCase      )
					pipe.to(torch_device=__UpperCAmelCase      )
					pipe.set_progress_bar_config(disable=__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						Optional[int]      					= self.get_inputs()
					SCREAMING_SNAKE_CASE  :						Optional[Any]      					= 1
					SCREAMING_SNAKE_CASE  :						int      					= None
					SCREAMING_SNAKE_CASE  :						str      					= pipe(**__UpperCAmelCase      ).images
					assert image.shape == (1, 64, 64, 3)

					SCREAMING_SNAKE_CASE  :						int      					= image[0, -3:, -3:, -1]

					SCREAMING_SNAKE_CASE  :						Any      					= np.array([0.0340, 0.0152, 0.0063, 0.0267, 0.0221, 0.0107, 0.0416, 0.0186, 0.0217]      )

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


	@require_torch_a
	def 				lowercase__  (self : Any      )     ->			Dict:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						str      					= UNetaDModel.from_pretrained('''diffusers/consistency_models''',					subfolder='''diffusers_cd_imagenet64_l2'''      )
					SCREAMING_SNAKE_CASE  :						int      					= CMStochasticIterativeScheduler(
					    num_train_timesteps=40,					sigma_min=0.002,					sigma_max=80.0,					)
					SCREAMING_SNAKE_CASE  :						Dict      					= ConsistencyModelPipeline(unet=__UpperCAmelCase,					scheduler=__UpperCAmelCase      )
					pipe.to(torch_device=__UpperCAmelCase,					torch_dtype=torch.floataa      )
					pipe.set_progress_bar_config(disable=__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						List[Any]      					= self.get_inputs(get_fixed_latents=__UpperCAmelCase,					device=__UpperCAmelCase      )
					# Ensure usage of flash attention in torch 2.0
					with sdp_kernel(enable_flash=__UpperCAmelCase,					enable_math=__UpperCAmelCase,					enable_mem_efficient=__UpperCAmelCase      ):
									SCREAMING_SNAKE_CASE  :						Tuple      					= pipe(**__UpperCAmelCase      ).images
					assert image.shape == (1, 64, 64, 3)

					SCREAMING_SNAKE_CASE  :						List[str]      					= image[0, -3:, -3:, -1]

					SCREAMING_SNAKE_CASE  :						int      					= np.array([0.1875, 0.1428, 0.1289, 0.2151, 0.2092, 0.1477, 0.1877, 0.1641, 0.1353]      )

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






	@require_torch_a
	def 				lowercase__  (self : int      )     ->			Tuple:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= UNetaDModel.from_pretrained('''diffusers/consistency_models''',					subfolder='''diffusers_cd_imagenet64_l2'''      )
					SCREAMING_SNAKE_CASE  :						Optional[Any]      					= CMStochasticIterativeScheduler(
					    num_train_timesteps=40,					sigma_min=0.002,					sigma_max=80.0,					)
					SCREAMING_SNAKE_CASE  :						Any      					= ConsistencyModelPipeline(unet=__UpperCAmelCase,					scheduler=__UpperCAmelCase      )
					pipe.to(torch_device=__UpperCAmelCase,					torch_dtype=torch.floataa      )
					pipe.set_progress_bar_config(disable=__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						Tuple      					= self.get_inputs(get_fixed_latents=__UpperCAmelCase,					device=__UpperCAmelCase      )
					SCREAMING_SNAKE_CASE  :						str      					= 1
					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= None
					# Ensure usage of flash attention in torch 2.0
					with sdp_kernel(enable_flash=__UpperCAmelCase,					enable_math=__UpperCAmelCase,					enable_mem_efficient=__UpperCAmelCase      ):
									SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= pipe(**__UpperCAmelCase      ).images
					assert image.shape == (1, 64, 64, 3)

					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= image[0, -3:, -3:, -1]

					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= np.array([0.1663, 0.1948, 0.2275, 0.1680, 0.1204, 0.1245, 0.1858, 0.1338, 0.2095]      )

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









'''simple docstring'''

from manim import *


class 	a__     (				_lowercase	):


	def 				lowercase__  (self : Optional[Any]      )     ->			str:

					"""simple docstring"""

					SCREAMING_SNAKE_CASE  :						List[Any]      					= Rectangle(height=0.5,					width=0.5      )
					SCREAMING_SNAKE_CASE  :						str      					= Rectangle(height=0.46,					width=0.46      ).set_stroke(width=0      )

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

					SCREAMING_SNAKE_CASE  :						Any      					= [mem.copy() for i in range(1      )]
					SCREAMING_SNAKE_CASE  :						Dict      					= VGroup(*__UpperCAmelCase      ).arrange(__UpperCAmelCase,					buff=0      )
					SCREAMING_SNAKE_CASE  :						List[str]      					= Text('''GPU''',					font_size=24      )
					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= Group(__UpperCAmelCase,					__UpperCAmelCase      ).arrange(__UpperCAmelCase,					buff=0.5,					aligned_edge=__UpperCAmelCase      )
					gpu.align_to(__UpperCAmelCase,					__UpperCAmelCase      )
					gpu.set_x(gpu.get_x() - 1      )

					self.add(__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						List[Any]      					= [mem.copy() for i in range(6      )]
					SCREAMING_SNAKE_CASE  :						Any      					= VGroup(*__UpperCAmelCase      ).arrange(__UpperCAmelCase,					buff=0      )

					SCREAMING_SNAKE_CASE  :						Union[str, Any]      					= Text('''Model''',					font_size=24      )
					SCREAMING_SNAKE_CASE  :						Dict      					= Group(__UpperCAmelCase,					__UpperCAmelCase      ).arrange(__UpperCAmelCase,					buff=0.5,					aligned_edge=__UpperCAmelCase      )
					model.move_to([3, -1.0, 0]      )

					self.play(
					    Create(__UpperCAmelCase,					run_time=1      ),					Create(__UpperCAmelCase,					run_time=1      ),					Create(__UpperCAmelCase,					run_time=1      ),					)

					SCREAMING_SNAKE_CASE  :						Tuple      					= MarkupText(
					    F'''First, an empty model skeleton is loaded\ninto <span fgcolor=\'{YELLOW}\'>memory</span> without using much RAM.''',					font_size=24,					)

					SCREAMING_SNAKE_CASE  :						Optional[Any]      					= Square(side_length=2.2      )
					key.move_to([-5, 2, 0]      )

					SCREAMING_SNAKE_CASE  :						Union[str, 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]      )

					step_a.move_to([2, 2, 0]      )
					self.play(Write(__UpperCAmelCase,					run_time=2.5      ),					Write(__UpperCAmelCase      ),					Write(__UpperCAmelCase      )      )

					self.add(__UpperCAmelCase      )

					SCREAMING_SNAKE_CASE  :						List[str]      					= []
					SCREAMING_SNAKE_CASE  :						List[str]      					= []
					SCREAMING_SNAKE_CASE  :						Dict      					= []
					for i, rect in enumerate(__UpperCAmelCase      ):
									SCREAMING_SNAKE_CASE  :						Optional[int]      					= Rectangle(height=0.46,					width=0.46      ).set_stroke(width=0.0      ).set_fill(__UpperCAmelCase,					opacity=0.7      )
									cpu_target.move_to(__UpperCAmelCase      )
									cpu_target.generate_target()
									SCREAMING_SNAKE_CASE  :						List[str]      					= 0.46 / 4
									SCREAMING_SNAKE_CASE  :						Any      					= 0.46 / 3

									if i == 0:
													cpu_target.target.next_to(cpu_left_col_base[0].get_corner(DOWN + LEFT      ),					buff=0.02,					direction=__UpperCAmelCase      )
													cpu_target.target.set_x(cpu_target.target.get_x() + 0.1      )
									elif i == 3:
													cpu_target.target.next_to(cpu_targs[0].target,					direction=__UpperCAmelCase,					buff=0.0      )
									else:
													cpu_target.target.next_to(cpu_targs[i - 1].target,					direction=__UpperCAmelCase,					buff=0.0      )
									cpu_targs.append(__UpperCAmelCase      )

									first_animations.append(rect.animate(run_time=0.5      ).set_stroke(__UpperCAmelCase      )      )
									second_animations.append(MoveToTarget(__UpperCAmelCase,					run_time=1.5      )      )

					self.play(*__UpperCAmelCase      )
					self.play(*__UpperCAmelCase      )

					self.wait()




