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

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







				@property
				def 							__UpperCAmelCase					( self    :       int						):

					"""simple docstring"""






					_lowercase :    Dict							=							UNetaDModel.from_pretrained(
					    'diffusers/consistency-models-test'			,					subfolder='test_unet'			,					)
					return unet







				@property
				def 							__UpperCAmelCase					( self    :       List[str]						):

					"""simple docstring"""






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







				def 							__UpperCAmelCase					( self    :       Optional[Any]			,					lowerCamelCase_    :       Union[str, Any]=False						):

					"""simple docstring"""






					if class_cond:
						_lowercase :    Tuple							=							self.dummy_cond_unet
					else:
						_lowercase :    Dict							=							self.dummy_uncond_unet

					# Default to CM multistep sampler
					_lowercase :    Optional[Any]							=							CMStochasticIterativeScheduler(
					    num_train_timesteps=4_0			,					sigma_min=0.002			,					sigma_max=8_0.0			,					)

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

					return components







				def 							__UpperCAmelCase					( self    :       int			,					lowerCamelCase_    :       Optional[int]			,					lowerCamelCase_    :       Any=0						):

					"""simple docstring"""






					if str(lowerCamelCase_						).startswith('mps'						):
						_lowercase :    List[str]							=							torch.manual_seed(lowerCamelCase_						)
					else:
						_lowercase :    str							=							torch.Generator(device=lowerCamelCase_						).manual_seed(lowerCamelCase_						)

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

					return inputs







				def 							__UpperCAmelCase					( self    :       Optional[Any]						):

					"""simple docstring"""






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

					_lowercase :    str							=							self.get_dummy_inputs(lowerCamelCase_						)
					_lowercase :    str							=							pipe(**lowerCamelCase_						).images
					assert image.shape == (1, 3_2, 3_2, 3)

					_lowercase :    str							=							image[0, -3:, -3:, -1]
					_lowercase :    Tuple							=							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 							__UpperCAmelCase					( self    :       List[str]						):

					"""simple docstring"""






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

					_lowercase :    List[Any]							=							self.get_dummy_inputs(lowerCamelCase_						)
					_lowercase :    int							=							0
					_lowercase :    Union[str, Any]							=							pipe(**lowerCamelCase_						).images
					assert image.shape == (1, 3_2, 3_2, 3)

					_lowercase :    Tuple							=							image[0, -3:, -3:, -1]
					_lowercase :    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 							__UpperCAmelCase					( self    :       Any						):

					"""simple docstring"""






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

					_lowercase :    Union[str, Any]							=							self.get_dummy_inputs(lowerCamelCase_						)
					_lowercase :    Optional[Any]							=							1
					_lowercase :    Dict							=							None
					_lowercase :    List[Any]							=							pipe(**lowerCamelCase_						).images
					assert image.shape == (1, 3_2, 3_2, 3)

					_lowercase :    Tuple							=							image[0, -3:, -3:, -1]
					_lowercase :    Optional[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 							__UpperCAmelCase					( self    :       Optional[Any]						):

					"""simple docstring"""






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

					_lowercase :    Tuple							=							self.get_dummy_inputs(lowerCamelCase_						)
					_lowercase :    Dict							=							1
					_lowercase :    Tuple							=							None
					_lowercase :    Optional[Any]							=							0
					_lowercase :    str							=							pipe(**lowerCamelCase_						).images
					assert image.shape == (1, 3_2, 3_2, 3)

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



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







				def 							__UpperCAmelCase					( self    :       Optional[Any]						):

					"""simple docstring"""






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







				def 							__UpperCAmelCase					( self    :       Optional[Any]			,					lowerCamelCase_    :       Tuple=0			,					lowerCamelCase_    :       Optional[Any]=False			,					lowerCamelCase_    :       Optional[Any]="cpu"			,					lowerCamelCase_    :       Union[str, Any]=torch.floataa			,					lowerCamelCase_    :       Dict=(1, 3, 6_4, 6_4)						):

					"""simple docstring"""






					_lowercase :    Union[str, Any]							=							torch.manual_seed(lowerCamelCase_						)

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

					if get_fixed_latents:
						_lowercase :    Optional[int]							=							self.get_fixed_latents(seed=lowerCamelCase_			,					device=lowerCamelCase_			,					dtype=lowerCamelCase_			,					shape=lowerCamelCase_						)
						_lowercase :    Tuple							=							latents

					return inputs







				def 							__UpperCAmelCase					( self    :       str			,					lowerCamelCase_    :       Tuple=0			,					lowerCamelCase_    :       Tuple="cpu"			,					lowerCamelCase_    :       Tuple=torch.floataa			,					lowerCamelCase_    :       str=(1, 3, 6_4, 6_4)						):

					"""simple docstring"""






					if type(lowerCamelCase_						) == str:
						_lowercase :    str							=							torch.device(lowerCamelCase_						)
					_lowercase :    List[str]							=							torch.Generator(device=lowerCamelCase_						).manual_seed(lowerCamelCase_						)
					_lowercase :    Any							=							randn_tensor(lowerCamelCase_			,					generator=lowerCamelCase_			,					device=lowerCamelCase_			,					dtype=lowerCamelCase_						)
					return latents







				def 							__UpperCAmelCase					( self    :       str						):

					"""simple docstring"""






					_lowercase :    int							=							UNetaDModel.from_pretrained('diffusers/consistency_models'			,					subfolder='diffusers_cd_imagenet64_l2'						)
					_lowercase :    List[str]							=							CMStochasticIterativeScheduler(
					    num_train_timesteps=4_0			,					sigma_min=0.002			,					sigma_max=8_0.0			,					)
					_lowercase :    List[Any]							=							ConsistencyModelPipeline(unet=lowerCamelCase_			,					scheduler=lowerCamelCase_						)
					pipe.to(torch_device=lowerCamelCase_						)
					pipe.set_progress_bar_config(disable=lowerCamelCase_						)

					_lowercase :    Optional[Any]							=							self.get_inputs()
					_lowercase :    Dict							=							pipe(**lowerCamelCase_						).images
					assert image.shape == (1, 6_4, 6_4, 3)

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

					_lowercase :    Union[str, Any]							=							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 							__UpperCAmelCase					( self    :       str						):

					"""simple docstring"""






					_lowercase :    Optional[int]							=							UNetaDModel.from_pretrained('diffusers/consistency_models'			,					subfolder='diffusers_cd_imagenet64_l2'						)
					_lowercase :    Any							=							CMStochasticIterativeScheduler(
					    num_train_timesteps=4_0			,					sigma_min=0.002			,					sigma_max=8_0.0			,					)
					_lowercase :    Optional[int]							=							ConsistencyModelPipeline(unet=lowerCamelCase_			,					scheduler=lowerCamelCase_						)
					pipe.to(torch_device=lowerCamelCase_						)
					pipe.set_progress_bar_config(disable=lowerCamelCase_						)

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

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

					_lowercase :    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 							__UpperCAmelCase					( self    :       List[str]						):

					"""simple docstring"""






					_lowercase :    int							=							UNetaDModel.from_pretrained('diffusers/consistency_models'			,					subfolder='diffusers_cd_imagenet64_l2'						)
					_lowercase :    List[Any]							=							CMStochasticIterativeScheduler(
					    num_train_timesteps=4_0			,					sigma_min=0.002			,					sigma_max=8_0.0			,					)
					_lowercase :    Tuple							=							ConsistencyModelPipeline(unet=lowerCamelCase_			,					scheduler=lowerCamelCase_						)
					pipe.to(torch_device=lowerCamelCase_			,					torch_dtype=torch.floataa						)
					pipe.set_progress_bar_config(disable=lowerCamelCase_						)

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

					_lowercase :    str							=							image[0, -3:, -3:, -1]

					_lowercase :    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 							__UpperCAmelCase					( self    :       Optional[int]						):

					"""simple docstring"""






					_lowercase :    Optional[Any]							=							UNetaDModel.from_pretrained('diffusers/consistency_models'			,					subfolder='diffusers_cd_imagenet64_l2'						)
					_lowercase :    List[Any]							=							CMStochasticIterativeScheduler(
					    num_train_timesteps=4_0			,					sigma_min=0.002			,					sigma_max=8_0.0			,					)
					_lowercase :    Dict							=							ConsistencyModelPipeline(unet=lowerCamelCase_			,					scheduler=lowerCamelCase_						)
					pipe.to(torch_device=lowerCamelCase_			,					torch_dtype=torch.floataa						)
					pipe.set_progress_bar_config(disable=lowerCamelCase_						)

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

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

					_lowercase :    Optional[int]							=							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 __future__ import annotations


def      __lowerCAmelCase(		__UpperCAmelCase			):






	"""simple docstring"""





	_lowercase :    Any							=							[True] * limit
	_lowercase :    Union[str, Any]							=							False
	_lowercase :    Any							=							False
	_lowercase :    List[Any]							=							True

	for i in range(3     ,int(limit**0.5 + 1			)     ,2			):
		_lowercase :    Optional[Any]							=							i * 2
		while index < limit:
			_lowercase :    int							=							False
			_lowercase :    Tuple							=							index + i

	_lowercase :    Optional[Any]							=							[2]

	for i in range(3     ,__UpperCAmelCase     ,2			):
		if is_prime[i]:
			primes.append(__UpperCAmelCase			)

	return primes


def      __lowerCAmelCase(		__UpperCAmelCase = 1_000_000			):






	"""simple docstring"""





	_lowercase :    Optional[Any]							=							prime_sieve(__UpperCAmelCase			)
	_lowercase :    int							=							0
	_lowercase :    List[Any]							=							0

	for i in range(len(__UpperCAmelCase			)			):
		for j in range(i + length     ,len(__UpperCAmelCase			)			):
			_lowercase :    List[Any]							=							sum(primes[i:j]			)
			if sol >= ceiling:
				break

			if sol in primes:
				_lowercase :    Optional[int]							=							j - i
				_lowercase :    Tuple							=							sol

	return largest


if __name__ == "__main__":
						print(f"""{solution() = }""")



