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        lowerCamelCase__				(   UpperCAmelCase       ,unittest.TestCase				):
			UpperCamelCase__    							=ConsistencyModelPipeline
			UpperCamelCase__    							=UNCONDITIONAL_IMAGE_GENERATION_PARAMS
			UpperCamelCase__    							=UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

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





			@property
			def   SCREAMING_SNAKE_CASE__     (       self	:	List[str]						):
										_lowerCAmelCase							=							UNetaDModel.from_pretrained(
										    'diffusers/consistency-models-test'  ,  subfolder='test_unet'  ,  )
										return unet





			@property
			def   SCREAMING_SNAKE_CASE__     (       self	:	str						):
										_lowerCAmelCase							=							UNetaDModel.from_pretrained(
										    'diffusers/consistency-models-test'  ,  subfolder='test_unet_class_cond'  ,  )
										return unet





			def   SCREAMING_SNAKE_CASE__     (       self	:	Tuple  ,  lowercase__	:	Optional[int]=False						):
										if class_cond:
																	_lowerCAmelCase							=							self.dummy_cond_unet
										else:
																	_lowerCAmelCase							=							self.dummy_uncond_unet

										# Default to CM multistep sampler
										_lowerCAmelCase							=							CMStochasticIterativeScheduler(
										    num_train_timesteps=40  ,  sigma_min=0.0_0_2  ,  sigma_max=8_0.0  ,  )

										_lowerCAmelCase							=							{
										    'unet': unet,
										    'scheduler': scheduler,
										}

										return components





			def   SCREAMING_SNAKE_CASE__     (       self	:	str  ,  lowercase__	:	str  ,  lowercase__	:	Optional[Any]=0						):
										if str(lowercase__						).startswith('mps'						):
																	_lowerCAmelCase							=							torch.manual_seed(lowercase__						)
										else:
																	_lowerCAmelCase							=							torch.Generator(device=lowercase__						).manual_seed(lowercase__						)

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

										return inputs





			def   SCREAMING_SNAKE_CASE__     (       self	:	List[Any]						):
										_lowerCAmelCase							=							'cpu'  # ensure determinism for the device-dependent torch.Generator
										_lowerCAmelCase							=							self.get_dummy_components()
										_lowerCAmelCase							=							ConsistencyModelPipeline(**lowercase__						)
										_lowerCAmelCase							=							pipe.to(lowercase__						)
										pipe.set_progress_bar_config(disable=lowercase__						)

										_lowerCAmelCase							=							self.get_dummy_inputs(lowercase__						)
										_lowerCAmelCase							=							pipe(**lowercase__						).images
										assert image.shape == (1, 32, 32, 3)

										_lowerCAmelCase							=							image[0, -3:, -3:, -1]
										_lowerCAmelCase							=							np.array([0.3_5_7_2, 0.6_2_7_3, 0.4_0_3_1, 0.3_9_6_1, 0.4_3_2_1, 0.5_7_3_0, 0.5_2_6_6, 0.4_7_8_0, 0.5_0_0_4]						)

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





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

										_lowerCAmelCase							=							self.get_dummy_inputs(lowercase__						)
										_lowerCAmelCase							=							0
										_lowerCAmelCase							=							pipe(**lowercase__						).images
										assert image.shape == (1, 32, 32, 3)

										_lowerCAmelCase							=							image[0, -3:, -3:, -1]
										_lowerCAmelCase							=							np.array([0.3_5_7_2, 0.6_2_7_3, 0.4_0_3_1, 0.3_9_6_1, 0.4_3_2_1, 0.5_7_3_0, 0.5_2_6_6, 0.4_7_8_0, 0.5_0_0_4]						)

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





			def   SCREAMING_SNAKE_CASE__     (       self	:	int						):
										_lowerCAmelCase							=							'cpu'  # ensure determinism for the device-dependent torch.Generator
										_lowerCAmelCase							=							self.get_dummy_components()
										_lowerCAmelCase							=							ConsistencyModelPipeline(**lowercase__						)
										_lowerCAmelCase							=							pipe.to(lowercase__						)
										pipe.set_progress_bar_config(disable=lowercase__						)

										_lowerCAmelCase							=							self.get_dummy_inputs(lowercase__						)
										_lowerCAmelCase							=							1
										_lowerCAmelCase							=							None
										_lowerCAmelCase							=							pipe(**lowercase__						).images
										assert image.shape == (1, 32, 32, 3)

										_lowerCAmelCase							=							image[0, -3:, -3:, -1]
										_lowerCAmelCase							=							np.array([0.5_0_0_4, 0.5_0_0_4, 0.4_9_9_4, 0.5_0_0_8, 0.4_9_7_6, 0.5_0_1_8, 0.4_9_9_0, 0.4_9_8_2, 0.4_9_8_7]						)

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





			def   SCREAMING_SNAKE_CASE__     (       self	:	Any						):
										_lowerCAmelCase							=							'cpu'  # ensure determinism for the device-dependent torch.Generator
										_lowerCAmelCase							=							self.get_dummy_components(class_cond=lowercase__						)
										_lowerCAmelCase							=							ConsistencyModelPipeline(**lowercase__						)
										_lowerCAmelCase							=							pipe.to(lowercase__						)
										pipe.set_progress_bar_config(disable=lowercase__						)

										_lowerCAmelCase							=							self.get_dummy_inputs(lowercase__						)
										_lowerCAmelCase							=							1
										_lowerCAmelCase							=							None
										_lowerCAmelCase							=							0
										_lowerCAmelCase							=							pipe(**lowercase__						).images
										assert image.shape == (1, 32, 32, 3)

										_lowerCAmelCase							=							image[0, -3:, -3:, -1]
										_lowerCAmelCase							=							np.array([0.5_0_0_4, 0.5_0_0_4, 0.4_9_9_4, 0.5_0_0_8, 0.4_9_7_6, 0.5_0_1_8, 0.4_9_9_0, 0.4_9_8_2, 0.4_9_8_7]						)

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







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





			def   SCREAMING_SNAKE_CASE__     (       self	:	int						):
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()





			def   SCREAMING_SNAKE_CASE__     (       self	:	List[Any]  ,  lowercase__	:	int=0  ,  lowercase__	:	str=False  ,  lowercase__	:	Union[str, Any]="cpu"  ,  lowercase__	:	Dict=torch.floataa  ,  lowercase__	:	List[str]=(1, 3, 64, 64)						):
										_lowerCAmelCase							=							torch.manual_seed(lowercase__						)

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

										if get_fixed_latents:
																	_lowerCAmelCase							=							self.get_fixed_latents(seed=lowercase__  ,  device=lowercase__  ,  dtype=lowercase__  ,  shape=lowercase__						)
																	_lowerCAmelCase							=							latents

										return inputs





			def   SCREAMING_SNAKE_CASE__     (       self	:	Tuple  ,  lowercase__	:	Dict=0  ,  lowercase__	:	Any="cpu"  ,  lowercase__	:	Union[str, Any]=torch.floataa  ,  lowercase__	:	Tuple=(1, 3, 64, 64)						):
										if type(lowercase__						) == str:
																	_lowerCAmelCase							=							torch.device(lowercase__						)
										_lowerCAmelCase							=							torch.Generator(device=lowercase__						).manual_seed(lowercase__						)
										_lowerCAmelCase							=							randn_tensor(lowercase__  ,  generator=lowercase__  ,  device=lowercase__  ,  dtype=lowercase__						)
										return latents





			def   SCREAMING_SNAKE_CASE__     (       self	:	Optional[Any]						):
										_lowerCAmelCase							=							UNetaDModel.from_pretrained('diffusers/consistency_models'  ,  subfolder='diffusers_cd_imagenet64_l2'						)
										_lowerCAmelCase							=							CMStochasticIterativeScheduler(
										    num_train_timesteps=40  ,  sigma_min=0.0_0_2  ,  sigma_max=8_0.0  ,  )
										_lowerCAmelCase							=							ConsistencyModelPipeline(unet=lowercase__  ,  scheduler=lowercase__						)
										pipe.to(torch_device=lowercase__						)
										pipe.set_progress_bar_config(disable=lowercase__						)

										_lowerCAmelCase							=							self.get_inputs()
										_lowerCAmelCase							=							pipe(**lowercase__						).images
										assert image.shape == (1, 64, 64, 3)

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

										_lowerCAmelCase							=							np.array([0.0_8_8_8, 0.0_8_8_1, 0.0_6_6_6, 0.0_4_7_9, 0.0_2_9_2, 0.0_1_9_5, 0.0_2_0_1, 0.0_1_6_3, 0.0_2_5_4]						)

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





			def   SCREAMING_SNAKE_CASE__     (       self	:	int						):
										_lowerCAmelCase							=							UNetaDModel.from_pretrained('diffusers/consistency_models'  ,  subfolder='diffusers_cd_imagenet64_l2'						)
										_lowerCAmelCase							=							CMStochasticIterativeScheduler(
										    num_train_timesteps=40  ,  sigma_min=0.0_0_2  ,  sigma_max=8_0.0  ,  )
										_lowerCAmelCase							=							ConsistencyModelPipeline(unet=lowercase__  ,  scheduler=lowercase__						)
										pipe.to(torch_device=lowercase__						)
										pipe.set_progress_bar_config(disable=lowercase__						)

										_lowerCAmelCase							=							self.get_inputs()
										_lowerCAmelCase							=							1
										_lowerCAmelCase							=							None
										_lowerCAmelCase							=							pipe(**lowercase__						).images
										assert image.shape == (1, 64, 64, 3)

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

										_lowerCAmelCase							=							np.array([0.0_3_4_0, 0.0_1_5_2, 0.0_0_6_3, 0.0_2_6_7, 0.0_2_2_1, 0.0_1_0_7, 0.0_4_1_6, 0.0_1_8_6, 0.0_2_1_7]						)

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





			@require_torch_a
			def   SCREAMING_SNAKE_CASE__     (       self	:	Union[str, Any]						):
										_lowerCAmelCase							=							UNetaDModel.from_pretrained('diffusers/consistency_models'  ,  subfolder='diffusers_cd_imagenet64_l2'						)
										_lowerCAmelCase							=							CMStochasticIterativeScheduler(
										    num_train_timesteps=40  ,  sigma_min=0.0_0_2  ,  sigma_max=8_0.0  ,  )
										_lowerCAmelCase							=							ConsistencyModelPipeline(unet=lowercase__  ,  scheduler=lowercase__						)
										pipe.to(torch_device=lowercase__  ,  torch_dtype=torch.floataa						)
										pipe.set_progress_bar_config(disable=lowercase__						)

										_lowerCAmelCase							=							self.get_inputs(get_fixed_latents=lowercase__  ,  device=lowercase__						)
										# Ensure usage of flash attention in torch 2.0
										with sdp_kernel(enable_flash=lowercase__  ,  enable_math=lowercase__  ,  enable_mem_efficient=lowercase__						):
																	_lowerCAmelCase							=							pipe(**lowercase__						).images
										assert image.shape == (1, 64, 64, 3)

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

										_lowerCAmelCase							=							np.array([0.1_8_7_5, 0.1_4_2_8, 0.1_2_8_9, 0.2_1_5_1, 0.2_0_9_2, 0.1_4_7_7, 0.1_8_7_7, 0.1_6_4_1, 0.1_3_5_3]						)

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





			@require_torch_a
			def   SCREAMING_SNAKE_CASE__     (       self	:	List[str]						):
										_lowerCAmelCase							=							UNetaDModel.from_pretrained('diffusers/consistency_models'  ,  subfolder='diffusers_cd_imagenet64_l2'						)
										_lowerCAmelCase							=							CMStochasticIterativeScheduler(
										    num_train_timesteps=40  ,  sigma_min=0.0_0_2  ,  sigma_max=8_0.0  ,  )
										_lowerCAmelCase							=							ConsistencyModelPipeline(unet=lowercase__  ,  scheduler=lowercase__						)
										pipe.to(torch_device=lowercase__  ,  torch_dtype=torch.floataa						)
										pipe.set_progress_bar_config(disable=lowercase__						)

										_lowerCAmelCase							=							self.get_inputs(get_fixed_latents=lowercase__  ,  device=lowercase__						)
										_lowerCAmelCase							=							1
										_lowerCAmelCase							=							None
										# Ensure usage of flash attention in torch 2.0
										with sdp_kernel(enable_flash=lowercase__  ,  enable_math=lowercase__  ,  enable_mem_efficient=lowercase__						):
																	_lowerCAmelCase							=							pipe(**lowercase__						).images
										assert image.shape == (1, 64, 64, 3)

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

										_lowerCAmelCase							=							np.array([0.1_6_6_3, 0.1_9_4_8, 0.2_2_7_5, 0.1_6_8_0, 0.1_2_0_4, 0.1_2_4_5, 0.1_8_5_8, 0.1_3_3_8, 0.2_0_9_5]						)

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







def _lowerCamelCase  (		snake_case = 10  ):
							if not isinstance(snake_case     , snake_case  ) or n < 0:
														raise ValueError('Invalid input'  )
							_lowerCAmelCase							=							10**n
							_lowerCAmelCase							=							28_433 * (pow(2     , 7_830_457     , snake_case  )) + 1
							return str(number % modulus  )


if __name__ == "__main__":
				from doctest import testmod

				testmod()
				print(f"""{solution(1_0) = }""")






