




'''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				(   _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   A_ (		self    :					Optional[Any]					):
									snake_case_									=				UNetaDModel.from_pretrained(
									    '''diffusers/consistency-models-test'''       ,				subfolder='''test_unet'''       ,				)
									return unet


				@property
				def   A_ (		self    :					Any					):
									snake_case_									=				UNetaDModel.from_pretrained(
									    '''diffusers/consistency-models-test'''       ,				subfolder='''test_unet_class_cond'''       ,				)
									return unet


				def   A_ (		self    :					List[Any]       ,				lowercase_    :					Union[str, Any]=False					):
									if class_cond:
														snake_case_									=				self.dummy_cond_unet
									else:
														snake_case_									=				self.dummy_uncond_unet

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

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

									return components


				def   A_ (		self    :					Tuple       ,				lowercase_    :					Optional[int]       ,				lowercase_    :					Tuple=0					):
									if str(lowercase_					).startswith('''mps'''					):
														snake_case_									=				torch.manual_seed(lowercase_					)
									else:
														snake_case_									=				torch.Generator(device=lowercase_					).manual_seed(lowercase_					)

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

									return inputs


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

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

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

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

									snake_case_									=				image[0, -3:, -3:, -1]
									snake_case_									=				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   A_ (		self    :					str					):
									snake_case_									=				'''cpu'''  # ensure determinism for the device-dependent torch.Generator
									snake_case_									=				self.get_dummy_components()
									snake_case_									=				ConsistencyModelPipeline(**lowercase_					)
									snake_case_									=				pipe.to(lowercase_					)
									pipe.set_progress_bar_config(disable=lowercase_					)

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

									snake_case_									=				image[0, -3:, -3:, -1]
									snake_case_									=				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   A_ (		self    :					Dict					):
									snake_case_									=				'''cpu'''  # ensure determinism for the device-dependent torch.Generator
									snake_case_									=				self.get_dummy_components(class_cond=lowercase_					)
									snake_case_									=				ConsistencyModelPipeline(**lowercase_					)
									snake_case_									=				pipe.to(lowercase_					)
									pipe.set_progress_bar_config(disable=lowercase_					)

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

									snake_case_									=				image[0, -3:, -3:, -1]
									snake_case_									=				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   A_ (		self    :					str					):
									super().tearDown()
									gc.collect()
									torch.cuda.empty_cache()


				def   A_ (		self    :					Union[str, Any]       ,				lowercase_    :					Optional[Any]=0       ,				lowercase_    :					Dict=False       ,				lowercase_    :					Any="cpu"       ,				lowercase_    :					str=torch.floataa       ,				lowercase_    :					int=(1, 3, 64, 64)					):
									snake_case_									=				torch.manual_seed(lowercase_					)

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

									if get_fixed_latents:
														snake_case_									=				self.get_fixed_latents(seed=lowercase_       ,				device=lowercase_       ,				dtype=lowercase_       ,				shape=lowercase_					)
														snake_case_									=				latents

									return inputs


				def   A_ (		self    :					Optional[int]       ,				lowercase_    :					List[Any]=0       ,				lowercase_    :					int="cpu"       ,				lowercase_    :					Union[str, Any]=torch.floataa       ,				lowercase_    :					List[Any]=(1, 3, 64, 64)					):
									if type(lowercase_					) == str:
														snake_case_									=				torch.device(lowercase_					)
									snake_case_									=				torch.Generator(device=lowercase_					).manual_seed(lowercase_					)
									snake_case_									=				randn_tensor(lowercase_       ,				generator=lowercase_       ,				device=lowercase_       ,				dtype=lowercase_					)
									return latents


				def   A_ (		self    :					List[Any]					):
									snake_case_									=				UNetaDModel.from_pretrained('''diffusers/consistency_models'''       ,				subfolder='''diffusers_cd_imagenet64_l2'''					)
									snake_case_									=				CMStochasticIterativeScheduler(
									    num_train_timesteps=40       ,				sigma_min=0.002       ,				sigma_max=80.0       ,				)
									snake_case_									=				ConsistencyModelPipeline(unet=lowercase_       ,				scheduler=lowercase_					)
									pipe.to(torch_device=lowercase_					)
									pipe.set_progress_bar_config(disable=lowercase_					)

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

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

									snake_case_									=				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   A_ (		self    :					List[Any]					):
									snake_case_									=				UNetaDModel.from_pretrained('''diffusers/consistency_models'''       ,				subfolder='''diffusers_cd_imagenet64_l2'''					)
									snake_case_									=				CMStochasticIterativeScheduler(
									    num_train_timesteps=40       ,				sigma_min=0.002       ,				sigma_max=80.0       ,				)
									snake_case_									=				ConsistencyModelPipeline(unet=lowercase_       ,				scheduler=lowercase_					)
									pipe.to(torch_device=lowercase_					)
									pipe.set_progress_bar_config(disable=lowercase_					)

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

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

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

									snake_case_									=				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_					):
														snake_case_									=				pipe(**lowercase_					).images
									assert image.shape == (1, 64, 64, 3)

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

									snake_case_									=				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   A_ (		self    :					Any					):
									snake_case_									=				UNetaDModel.from_pretrained('''diffusers/consistency_models'''       ,				subfolder='''diffusers_cd_imagenet64_l2'''					)
									snake_case_									=				CMStochasticIterativeScheduler(
									    num_train_timesteps=40       ,				sigma_min=0.002       ,				sigma_max=80.0       ,				)
									snake_case_									=				ConsistencyModelPipeline(unet=lowercase_       ,				scheduler=lowercase_					)
									pipe.to(torch_device=lowercase_       ,				torch_dtype=torch.floataa					)
									pipe.set_progress_bar_config(disable=lowercase_					)

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

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

									snake_case_									=				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'''

import re


def 		__magic_name__    (	__UpperCAmelCase	)		->					bool:

					'''simple docstring'''

					snake_case_									=				re.compile(
					    r'''^(?:0|94|\+94|0{2}94)''' r'''7(0|1|2|4|5|6|7|8)''' r'''(-| |)''' r'''\d{7}$'''	)

					return bool(re.search(__UpperCAmelCase,      __UpperCAmelCase	)	)


if __name__ == "__main__":
							a  :				Any									=  '0094702343221'

							print(is_sri_lankan_phone_number(phone))

