






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 					_snake_case						(			_A       ,    unittest.TestCase ):
	_A          =   ConsistencyModelPipeline
	_A          =   UNCONDITIONAL_IMAGE_GENERATION_PARAMS
	_A          =   UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

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






	@property
	def    lowerCAmelCase_							(		self       )						->    Dict:
					snake_case__      :Tuple         =   UNetaDModel.from_pretrained(
					    "diffusers/consistency-models-test"					,subfolder="test_unet"					,)
					return unet






	@property
	def    lowerCAmelCase_							(		self       )						->    str:
					snake_case__      :Tuple         =   UNetaDModel.from_pretrained(
					    "diffusers/consistency-models-test"					,subfolder="test_unet_class_cond"					,)
					return unet






	def    lowerCAmelCase_							(		self					,UpperCamelCase=False       )						->    Tuple:
					if class_cond:
									snake_case__      :Tuple         =   self.dummy_cond_unet
					else:
									snake_case__      :Optional[int]         =   self.dummy_uncond_unet

					# Default to CM multistep sampler
					snake_case__      :List[str]         =   CMStochasticIterativeScheduler(
					    num_train_timesteps=40					,sigma_min=0.002					,sigma_max=80.0					,)

					snake_case__      :Tuple         =   {
					    "unet": unet,
					    "scheduler": scheduler,
					}

					return components






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase=0       )						->    str:
					if str(UpperCamelCase       ).startswith("mps"       ):
									snake_case__      :List[str]         =   torch.manual_seed(UpperCamelCase       )
					else:
									snake_case__      :Any         =   torch.Generator(device=UpperCamelCase       ).manual_seed(UpperCamelCase       )

					snake_case__      :Any         =   {
					    "batch_size": 1,
					    "num_inference_steps": None,
					    "timesteps": [22, 0],
					    "generator": generator,
					    "output_type": "np",
					}

					return inputs






	def    lowerCAmelCase_							(		self       )						->    List[Any]:
					snake_case__      :Optional[int]         =   "cpu"  # ensure determinism for the device-dependent torch.Generator
					snake_case__      :Optional[Any]         =   self.get_dummy_components()
					snake_case__      :List[Any]         =   ConsistencyModelPipeline(**UpperCamelCase       )
					snake_case__      :Tuple         =   pipe.to(UpperCamelCase       )
					pipe.set_progress_bar_config(disable=UpperCamelCase       )

					snake_case__      :Dict         =   self.get_dummy_inputs(UpperCamelCase       )
					snake_case__      :List[Any]         =   pipe(**UpperCamelCase       ).images
					assert image.shape == (1, 32, 32, 3)

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

					snake_case__      :List[Any]         =   self.get_dummy_inputs(UpperCamelCase       )
					snake_case__      :Optional[Any]         =   0
					snake_case__      :int         =   pipe(**UpperCamelCase       ).images
					assert image.shape == (1, 32, 32, 3)

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

					snake_case__      :int         =   self.get_dummy_inputs(UpperCamelCase       )
					snake_case__      :Optional[Any]         =   1
					snake_case__      :str         =   None
					snake_case__      :Dict         =   pipe(**UpperCamelCase       ).images
					assert image.shape == (1, 32, 32, 3)

					snake_case__      :Dict         =   image[0, -3:, -3:, -1]
					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






	def    lowerCAmelCase_							(		self       )						->    Union[str, Any]:
					snake_case__      :List[str]         =   "cpu"  # ensure determinism for the device-dependent torch.Generator
					snake_case__      :Union[str, Any]         =   self.get_dummy_components(class_cond=UpperCamelCase       )
					snake_case__      :Dict         =   ConsistencyModelPipeline(**UpperCamelCase       )
					snake_case__      :Any         =   pipe.to(UpperCamelCase       )
					pipe.set_progress_bar_config(disable=UpperCamelCase       )

					snake_case__      :Tuple         =   self.get_dummy_inputs(UpperCamelCase       )
					snake_case__      :Optional[Any]         =   1
					snake_case__      :List[Any]         =   None
					snake_case__      :Dict         =   0
					snake_case__      :Dict         =   pipe(**UpperCamelCase       ).images
					assert image.shape == (1, 32, 32, 3)

					snake_case__      :Any         =   image[0, -3:, -3:, -1]
					snake_case__      :Optional[int]         =   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 					_snake_case						(			unittest.TestCase ):






	def    lowerCAmelCase_							(		self       )						->    Optional[Any]:
					super().tearDown()
					gc.collect()
					torch.cuda.empty_cache()






	def    lowerCAmelCase_							(		self					,UpperCamelCase=0					,UpperCamelCase=False					,UpperCamelCase="cpu"					,UpperCamelCase=torch.floataa					,UpperCamelCase=(1, 3, 64, 64)       )						->    List[Any]:
					snake_case__      :str         =   torch.manual_seed(UpperCamelCase       )

					snake_case__      :str         =   {
					    "num_inference_steps": None,
					    "timesteps": [22, 0],
					    "class_labels": 0,
					    "generator": generator,
					    "output_type": "np",
					}

					if get_fixed_latents:
									snake_case__      :Any         =   self.get_fixed_latents(seed=UpperCamelCase					,device=UpperCamelCase					,dtype=UpperCamelCase					,shape=UpperCamelCase       )
									snake_case__      :Union[str, Any]         =   latents

					return inputs






	def    lowerCAmelCase_							(		self					,UpperCamelCase=0					,UpperCamelCase="cpu"					,UpperCamelCase=torch.floataa					,UpperCamelCase=(1, 3, 64, 64)       )						->    Any:
					if type(UpperCamelCase       ) == str:
									snake_case__      :int         =   torch.device(UpperCamelCase       )
					snake_case__      :List[Any]         =   torch.Generator(device=UpperCamelCase       ).manual_seed(UpperCamelCase       )
					snake_case__      :Tuple         =   randn_tensor(UpperCamelCase					,generator=UpperCamelCase					,device=UpperCamelCase					,dtype=UpperCamelCase       )
					return latents






	def    lowerCAmelCase_							(		self       )						->    Any:
					snake_case__      :List[str]         =   UNetaDModel.from_pretrained("diffusers/consistency_models"					,subfolder="diffusers_cd_imagenet64_l2"       )
					snake_case__      :Any         =   CMStochasticIterativeScheduler(
					    num_train_timesteps=40					,sigma_min=0.002					,sigma_max=80.0					,)
					snake_case__      :int         =   ConsistencyModelPipeline(unet=UpperCamelCase					,scheduler=UpperCamelCase       )
					pipe.to(torch_device=UpperCamelCase       )
					pipe.set_progress_bar_config(disable=UpperCamelCase       )

					snake_case__      :List[str]         =   self.get_inputs()
					snake_case__      :List[Any]         =   pipe(**UpperCamelCase       ).images
					assert image.shape == (1, 64, 64, 3)

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

					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    lowerCAmelCase_							(		self       )						->    Union[str, Any]:
					snake_case__      :Optional[int]         =   UNetaDModel.from_pretrained("diffusers/consistency_models"					,subfolder="diffusers_cd_imagenet64_l2"       )
					snake_case__      :Optional[Any]         =   CMStochasticIterativeScheduler(
					    num_train_timesteps=40					,sigma_min=0.002					,sigma_max=80.0					,)
					snake_case__      :Union[str, Any]         =   ConsistencyModelPipeline(unet=UpperCamelCase					,scheduler=UpperCamelCase       )
					pipe.to(torch_device=UpperCamelCase       )
					pipe.set_progress_bar_config(disable=UpperCamelCase       )

					snake_case__      :Union[str, Any]         =   self.get_inputs()
					snake_case__      :str         =   1
					snake_case__      :Tuple         =   None
					snake_case__      :str         =   pipe(**UpperCamelCase       ).images
					assert image.shape == (1, 64, 64, 3)

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

					snake_case__      :Union[str, 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:
					snake_case__      :List[str]         =   UNetaDModel.from_pretrained("diffusers/consistency_models"					,subfolder="diffusers_cd_imagenet64_l2"       )
					snake_case__      :Union[str, Any]         =   CMStochasticIterativeScheduler(
					    num_train_timesteps=40					,sigma_min=0.002					,sigma_max=80.0					,)
					snake_case__      :Union[str, Any]         =   ConsistencyModelPipeline(unet=UpperCamelCase					,scheduler=UpperCamelCase       )
					pipe.to(torch_device=UpperCamelCase					,torch_dtype=torch.floataa       )
					pipe.set_progress_bar_config(disable=UpperCamelCase       )

					snake_case__      :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       ):
									snake_case__      :Any         =   pipe(**UpperCamelCase       ).images
					assert image.shape == (1, 64, 64, 3)

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

					snake_case__      :List[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       )						->    Union[str, Any]:
					snake_case__      :Optional[int]         =   UNetaDModel.from_pretrained("diffusers/consistency_models"					,subfolder="diffusers_cd_imagenet64_l2"       )
					snake_case__      :int         =   CMStochasticIterativeScheduler(
					    num_train_timesteps=40					,sigma_min=0.002					,sigma_max=80.0					,)
					snake_case__      :Dict         =   ConsistencyModelPipeline(unet=UpperCamelCase					,scheduler=UpperCamelCase       )
					pipe.to(torch_device=UpperCamelCase					,torch_dtype=torch.floataa       )
					pipe.set_progress_bar_config(disable=UpperCamelCase       )

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

					snake_case__      :Dict         =   image[0, -3:, -3:, -1]

					snake_case__      :Tuple         =   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







def 					lowercase_ (    __snake_case     :							int      )     -> bool:





				'''simple docstring'''


				if p < 2:
								raise ValueError("p should not be less than 2!"      )
				elif p == 2:
								return True

				snake_case__      :List[str]         =   4
				snake_case__      :Optional[int]         =   (1 << p) - 1
				for _ in range(p - 2      ):
								snake_case__      :List[Any]         =   ((s * s) - 2) % m
				return s == 0


if __name__ == "__main__":
	print(lucas_lehmer_test(7))
	print(lucas_lehmer_test(1_1))