

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_		(	SCREAMING_SNAKE_CASE      ,							unittest.TestCase    ):
							_UpperCAmelCase   :  List[str]										=				ConsistencyModelPipeline
							_UpperCAmelCase   :  Dict										=				UNCONDITIONAL_IMAGE_GENERATION_PARAMS
							_UpperCAmelCase   :  int										=				UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

							# Override required_optional_params to remove num_images_per_prompt
							_UpperCAmelCase   :  Dict										=				frozenset(
							    [
							        '''num_inference_steps''',
							        '''generator''',
							        '''latents''',
							        '''output_type''',
							        '''return_dict''',
							        '''callback''',
							        '''callback_steps''',
							    ]    )
							@property
							def    lowerCAmelCase					(		self		:      Optional[int]):
								__lowerCamelCase    :    Any 	=			UNetaDModel.from_pretrained(
								    'diffusers/consistency-models-test'       ,subfolder='test_unet'       ,)
								return unet
							@property
							def    lowerCAmelCase					(		self		:      List[Any]):
								__lowerCamelCase    :    List[Any] 	=			UNetaDModel.from_pretrained(
								    'diffusers/consistency-models-test'       ,subfolder='test_unet_class_cond'       ,)
								return unet
							def    lowerCAmelCase					(		self		:      List[str]       ,SCREAMING_SNAKE_CASE__		:      Optional[int]=False):
								if class_cond:
									__lowerCamelCase    :    Optional[int] 	=			self.dummy_cond_unet
								else:
									__lowerCamelCase    :    Optional[int] 	=			self.dummy_uncond_unet

								# Default to CM multistep sampler
								__lowerCamelCase    :    Optional[Any] 	=			CMStochasticIterativeScheduler(
								    num_train_timesteps=4_0       ,sigma_min=0.002       ,sigma_max=80.0       ,)

								__lowerCamelCase    :    List[str] 	=			{
								    'unet': unet,
								    'scheduler': scheduler,
								}

								return components
							def    lowerCAmelCase					(		self		:      List[Any]       ,SCREAMING_SNAKE_CASE__		:      Dict       ,SCREAMING_SNAKE_CASE__		:      str=0):
								if str(SCREAMING_SNAKE_CASE__).startswith('mps'):
									__lowerCamelCase    :    Tuple 	=			torch.manual_seed(SCREAMING_SNAKE_CASE__)
								else:
									__lowerCamelCase    :    Union[str, Any] 	=			torch.Generator(device=SCREAMING_SNAKE_CASE__).manual_seed(SCREAMING_SNAKE_CASE__)

								__lowerCamelCase    :    Union[str, Any] 	=			{
								    'batch_size': 1,
								    'num_inference_steps': None,
								    'timesteps': [2_2, 0],
								    'generator': generator,
								    'output_type': 'np',
								}

								return inputs
							def    lowerCAmelCase					(		self		:      Optional[int]):
								__lowerCamelCase    :    List[str] 	=			'cpu'  # ensure determinism for the device-dependent torch.Generator
								__lowerCamelCase    :    Dict 	=			self.get_dummy_components()
								__lowerCamelCase    :    str 	=			ConsistencyModelPipeline(**SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    int 	=			pipe.to(SCREAMING_SNAKE_CASE__)
								pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__)

								__lowerCamelCase    :    Tuple 	=			self.get_dummy_inputs(SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    List[str] 	=			pipe(**SCREAMING_SNAKE_CASE__).images
								assert image.shape == (1, 3_2, 3_2, 3)

								__lowerCamelCase    :    Any 	=			image[0, -3:, -3:, -1]
								__lowerCamelCase    :    Union[str, 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    lowerCAmelCase					(		self		:      str):
								__lowerCamelCase    :    int 	=			'cpu'  # ensure determinism for the device-dependent torch.Generator
								__lowerCamelCase    :    Optional[Any] 	=			self.get_dummy_components(class_cond=SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    Any 	=			ConsistencyModelPipeline(**SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    List[str] 	=			pipe.to(SCREAMING_SNAKE_CASE__)
								pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__)

								__lowerCamelCase    :    Union[str, Any] 	=			self.get_dummy_inputs(SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    Tuple 	=			0
								__lowerCamelCase    :    Optional[Any] 	=			pipe(**SCREAMING_SNAKE_CASE__).images
								assert image.shape == (1, 3_2, 3_2, 3)

								__lowerCamelCase    :    List[str] 	=			image[0, -3:, -3:, -1]
								__lowerCamelCase    :    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    lowerCAmelCase					(		self		:      Union[str, Any]):
								__lowerCamelCase    :    Dict 	=			'cpu'  # ensure determinism for the device-dependent torch.Generator
								__lowerCamelCase    :    Dict 	=			self.get_dummy_components()
								__lowerCamelCase    :    Optional[Any] 	=			ConsistencyModelPipeline(**SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    List[Any] 	=			pipe.to(SCREAMING_SNAKE_CASE__)
								pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__)

								__lowerCamelCase    :    List[Any] 	=			self.get_dummy_inputs(SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    Dict 	=			1
								__lowerCamelCase    :    Optional[int] 	=			None
								__lowerCamelCase    :    Any 	=			pipe(**SCREAMING_SNAKE_CASE__).images
								assert image.shape == (1, 3_2, 3_2, 3)

								__lowerCamelCase    :    str 	=			image[0, -3:, -3:, -1]
								__lowerCamelCase    :    Union[str, 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


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

								__lowerCamelCase    :    Union[str, Any] 	=			self.get_dummy_inputs(SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    Dict 	=			1
								__lowerCamelCase    :    int 	=			None
								__lowerCamelCase    :    Optional[int] 	=			0
								__lowerCamelCase    :    Any 	=			pipe(**SCREAMING_SNAKE_CASE__).images
								assert image.shape == (1, 3_2, 3_2, 3)

								__lowerCamelCase    :    Optional[int] 	=			image[0, -3:, -3:, -1]
								__lowerCamelCase    :    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    lowerCAmelCase					(		self		:      Union[str, Any]):
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()
							def    lowerCAmelCase					(		self		:      Optional[int]       ,SCREAMING_SNAKE_CASE__		:      int=0       ,SCREAMING_SNAKE_CASE__		:      List[str]=False       ,SCREAMING_SNAKE_CASE__		:      Any="cpu"       ,SCREAMING_SNAKE_CASE__		:      str=torch.floataa       ,SCREAMING_SNAKE_CASE__		:      Dict=(1, 3, 6_4, 6_4)):
								__lowerCamelCase    :    Any 	=			torch.manual_seed(SCREAMING_SNAKE_CASE__)

								__lowerCamelCase    :    Any 	=			{
								    'num_inference_steps': None,
								    'timesteps': [2_2, 0],
								    'class_labels': 0,
								    'generator': generator,
								    'output_type': 'np',
								}

								if get_fixed_latents:
									__lowerCamelCase    :    List[str] 	=			self.get_fixed_latents(seed=SCREAMING_SNAKE_CASE__       ,device=SCREAMING_SNAKE_CASE__       ,dtype=SCREAMING_SNAKE_CASE__       ,shape=SCREAMING_SNAKE_CASE__)
									__lowerCamelCase    :    Dict 	=			latents

								return inputs
							def    lowerCAmelCase					(		self		:      int       ,SCREAMING_SNAKE_CASE__		:      Optional[int]=0       ,SCREAMING_SNAKE_CASE__		:      List[Any]="cpu"       ,SCREAMING_SNAKE_CASE__		:      Optional[int]=torch.floataa       ,SCREAMING_SNAKE_CASE__		:      Optional[int]=(1, 3, 6_4, 6_4)):
								if type(SCREAMING_SNAKE_CASE__) == str:
									__lowerCamelCase    :    str 	=			torch.device(SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    str 	=			torch.Generator(device=SCREAMING_SNAKE_CASE__).manual_seed(SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    Tuple 	=			randn_tensor(SCREAMING_SNAKE_CASE__       ,generator=SCREAMING_SNAKE_CASE__       ,device=SCREAMING_SNAKE_CASE__       ,dtype=SCREAMING_SNAKE_CASE__)
								return latents
							def    lowerCAmelCase					(		self		:      Tuple):
								__lowerCamelCase    :    Tuple 	=			UNetaDModel.from_pretrained('diffusers/consistency_models'       ,subfolder='diffusers_cd_imagenet64_l2')
								__lowerCamelCase    :    Optional[Any] 	=			CMStochasticIterativeScheduler(
								    num_train_timesteps=4_0       ,sigma_min=0.002       ,sigma_max=80.0       ,)
								__lowerCamelCase    :    Tuple 	=			ConsistencyModelPipeline(unet=SCREAMING_SNAKE_CASE__       ,scheduler=SCREAMING_SNAKE_CASE__)
								pipe.to(torch_device=SCREAMING_SNAKE_CASE__)
								pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__)

								__lowerCamelCase    :    Union[str, Any] 	=			self.get_inputs()
								__lowerCamelCase    :    List[Any] 	=			pipe(**SCREAMING_SNAKE_CASE__).images
								assert image.shape == (1, 6_4, 6_4, 3)

								__lowerCamelCase    :    Optional[int] 	=			image[0, -3:, -3:, -1]

								__lowerCamelCase    :    Optional[int] 	=			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    lowerCAmelCase					(		self		:      Union[str, Any]):
								__lowerCamelCase    :    Dict 	=			UNetaDModel.from_pretrained('diffusers/consistency_models'       ,subfolder='diffusers_cd_imagenet64_l2')
								__lowerCamelCase    :    List[str] 	=			CMStochasticIterativeScheduler(
								    num_train_timesteps=4_0       ,sigma_min=0.002       ,sigma_max=80.0       ,)
								__lowerCamelCase    :    Any 	=			ConsistencyModelPipeline(unet=SCREAMING_SNAKE_CASE__       ,scheduler=SCREAMING_SNAKE_CASE__)
								pipe.to(torch_device=SCREAMING_SNAKE_CASE__)
								pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__)

								__lowerCamelCase    :    Dict 	=			self.get_inputs()
								__lowerCamelCase    :    Optional[Any] 	=			1
								__lowerCamelCase    :    Union[str, Any] 	=			None
								__lowerCamelCase    :    Optional[Any] 	=			pipe(**SCREAMING_SNAKE_CASE__).images
								assert image.shape == (1, 6_4, 6_4, 3)

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

								__lowerCamelCase    :    List[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    lowerCAmelCase					(		self		:      str):
								__lowerCamelCase    :    List[str] 	=			UNetaDModel.from_pretrained('diffusers/consistency_models'       ,subfolder='diffusers_cd_imagenet64_l2')
								__lowerCamelCase    :    Optional[int] 	=			CMStochasticIterativeScheduler(
								    num_train_timesteps=4_0       ,sigma_min=0.002       ,sigma_max=80.0       ,)
								__lowerCamelCase    :    Dict 	=			ConsistencyModelPipeline(unet=SCREAMING_SNAKE_CASE__       ,scheduler=SCREAMING_SNAKE_CASE__)
								pipe.to(torch_device=SCREAMING_SNAKE_CASE__       ,torch_dtype=torch.floataa)
								pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__)

								__lowerCamelCase    :    List[str] 	=			self.get_inputs(get_fixed_latents=SCREAMING_SNAKE_CASE__       ,device=SCREAMING_SNAKE_CASE__)
								# Ensure usage of flash attention in torch 2.0
								with sdp_kernel(enable_flash=SCREAMING_SNAKE_CASE__       ,enable_math=SCREAMING_SNAKE_CASE__       ,enable_mem_efficient=SCREAMING_SNAKE_CASE__):
									__lowerCamelCase    :    Optional[int] 	=			pipe(**SCREAMING_SNAKE_CASE__).images
								assert image.shape == (1, 6_4, 6_4, 3)

								__lowerCamelCase    :    Tuple 	=			image[0, -3:, -3:, -1]

								__lowerCamelCase    :    str 	=			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    lowerCAmelCase					(		self		:      Dict):
								__lowerCamelCase    :    Dict 	=			UNetaDModel.from_pretrained('diffusers/consistency_models'       ,subfolder='diffusers_cd_imagenet64_l2')
								__lowerCamelCase    :    Union[str, Any] 	=			CMStochasticIterativeScheduler(
								    num_train_timesteps=4_0       ,sigma_min=0.002       ,sigma_max=80.0       ,)
								__lowerCamelCase    :    List[str] 	=			ConsistencyModelPipeline(unet=SCREAMING_SNAKE_CASE__       ,scheduler=SCREAMING_SNAKE_CASE__)
								pipe.to(torch_device=SCREAMING_SNAKE_CASE__       ,torch_dtype=torch.floataa)
								pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE__)

								__lowerCamelCase    :    Dict 	=			self.get_inputs(get_fixed_latents=SCREAMING_SNAKE_CASE__       ,device=SCREAMING_SNAKE_CASE__)
								__lowerCamelCase    :    int 	=			1
								__lowerCamelCase    :    Dict 	=			None
								# Ensure usage of flash attention in torch 2.0
								with sdp_kernel(enable_flash=SCREAMING_SNAKE_CASE__       ,enable_math=SCREAMING_SNAKE_CASE__       ,enable_mem_efficient=SCREAMING_SNAKE_CASE__):
									__lowerCamelCase    :    str 	=			pipe(**SCREAMING_SNAKE_CASE__).images
								assert image.shape == (1, 6_4, 6_4, 3)

								__lowerCamelCase    :    Tuple 	=			image[0, -3:, -3:, -1]

								__lowerCamelCase    :    str 	=			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







from collections.abc import Callable
from math import pi, sqrt
from random import uniform
from statistics import mean






def        SCREAMING_SNAKE_CASE__	(  lowerCamelCase__      )  ->					List[Any]:

	# A local function to see if a dot lands in the circle.
	def is_in_circle(lowerCamelCase__     ,			lowerCamelCase__      ) -> bool:
		__lowerCamelCase    :    Dict 	=			sqrt((x**2) + (y**2)      )
		# Our circle has a radius of 1, so a distance
		# greater than 1 would land outside the circle.
		return distance_from_centre <= 1

	# The proportion of guesses that landed in the circle
	__lowerCamelCase    :    Optional[int] 	=			mean(
	    int(is_in_circle(uniform(-1.0     ,			1.0      )     ,			uniform(-1.0     ,			1.0      )      )      )
	    for _ in range(lowerCamelCase__      )      )
	# The ratio of the area for circle to square is pi/4.
	__lowerCamelCase    :    Any 	=			proportion * 4
	print(F"The estimated value of pi is {pi_estimate}"      )
	print(F"The numpy value of pi is {pi}"      )
	print(F"The total error is {abs(pi - pi_estimate      )}"      )






def        SCREAMING_SNAKE_CASE__	(  lowerCamelCase__     ,			lowerCamelCase__     ,			lowerCamelCase__ = 0.0     ,			lowerCamelCase__ = 1.0     ,			)  ->					float:
	return mean(
	    function_to_integrate(uniform(lowerCamelCase__     ,			lowerCamelCase__      )      ) for _ in range(lowerCamelCase__      )      ) * (max_value - min_value)






def        SCREAMING_SNAKE_CASE__	(  lowerCamelCase__     ,			lowerCamelCase__ = 0.0     ,			lowerCamelCase__ = 1.0      )  ->					None:

	def identity_function(lowerCamelCase__      ) -> float:
		return x

	__lowerCamelCase    :    str 	=			area_under_curve_estimator(
	    lowerCamelCase__     ,			lowerCamelCase__     ,			lowerCamelCase__     ,			lowerCamelCase__      )
	__lowerCamelCase    :    int 	=			(max_value * max_value - min_value * min_value) / 2

	print('******************'      )
	print(F"Estimating area under y=x where x varies from {min_value} to {max_value}"      )
	print(F"Estimated value is {estimated_value}"      )
	print(F"Expected value is {expected_value}"      )
	print(F"Total error is {abs(estimated_value - expected_value      )}"      )
	print('******************'      )






def        SCREAMING_SNAKE_CASE__	(  lowerCamelCase__      )  ->					None:

	def function_to_integrate(lowerCamelCase__      ) -> float:
		return sqrt(4.0 - x * x      )

	__lowerCamelCase    :    Any 	=			area_under_curve_estimator(
	    lowerCamelCase__     ,			lowerCamelCase__     ,			0.0     ,			2.0      )

	print('******************'      )
	print('Estimating pi using area_under_curve_estimator'      )
	print(F"Estimated value is {estimated_value}"      )
	print(F"Expected value is {pi}"      )
	print(F"Total error is {abs(estimated_value - pi      )}"      )
	print('******************'      )


if __name__ == "__main__":
					import doctest

					doctest.testmod()




