'''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__      (						_UpperCamelCase	,   unittest.TestCase     ):




					'''simple docstring'''



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



						'''simple docstring'''





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




					@property
					def 						__UpperCAmelCase     (  self				:	int					)  ->       Any:



						'''simple docstring'''





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




					def 						__UpperCAmelCase     (  self				:	Tuple     ,  UpperCamelCase_				:	Optional[Any]=False					)  ->       Tuple:



						'''simple docstring'''





						if class_cond:
							_lowercase					:       Union[str, Any]       =		self.dummy_cond_unet
						else:
							_lowercase					:       Any       =		self.dummy_uncond_unet

						# Default to CM multistep sampler
						_lowercase					:       List[Any]       =		CMStochasticIterativeScheduler(
						    num_train_timesteps=40     ,  sigma_min=0.0_02     ,  sigma_max=80.0     ,  )

						_lowercase					:       Optional[int]       =		{
						    "unet": unet,
						    "scheduler": scheduler,
						}

						return components




					def 						__UpperCAmelCase     (  self				:	Tuple     ,  UpperCamelCase_				:	Any     ,  UpperCamelCase_				:	Any=0					)  ->       Optional[int]:



						'''simple docstring'''





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

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

						return inputs




					def 						__UpperCAmelCase     (  self				:	str					)  ->       int:



						'''simple docstring'''





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

						_lowercase					:       Dict       =		self.get_dummy_inputs(__a					)
						_lowercase					:       Optional[Any]       =		pipe(**__a					).images
						assert image.shape == (1, 32, 32, 3)

						_lowercase					:       Optional[int]       =		image[0, -3:, -3:, -1]
						_lowercase					:       List[str]       =		np.array([0.35_72, 0.62_73, 0.40_31, 0.39_61, 0.43_21, 0.57_30, 0.52_66, 0.47_80, 0.50_04]					)

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




					def 						__UpperCAmelCase     (  self				:	Any					)  ->       Union[str, Any]:



						'''simple docstring'''





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

						_lowercase					:       Tuple       =		self.get_dummy_inputs(__a					)
						_lowercase					:       Tuple       =		0
						_lowercase					:       List[str]       =		pipe(**__a					).images
						assert image.shape == (1, 32, 32, 3)

						_lowercase					:       Union[str, Any]       =		image[0, -3:, -3:, -1]
						_lowercase					:       Optional[int]       =		np.array([0.35_72, 0.62_73, 0.40_31, 0.39_61, 0.43_21, 0.57_30, 0.52_66, 0.47_80, 0.50_04]					)

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




					def 						__UpperCAmelCase     (  self				:	Tuple					)  ->       List[Any]:



						'''simple docstring'''





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

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

						_lowercase					:       str       =		image[0, -3:, -3:, -1]
						_lowercase					:       Optional[Any]       =		np.array([0.50_04, 0.50_04, 0.49_94, 0.50_08, 0.49_76, 0.50_18, 0.49_90, 0.49_82, 0.49_87]					)

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




					def 						__UpperCAmelCase     (  self				:	List[str]					)  ->       Union[str, Any]:



						'''simple docstring'''





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

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

						_lowercase					:       Tuple       =		image[0, -3:, -3:, -1]
						_lowercase					:       Any       =		np.array([0.50_04, 0.50_04, 0.49_94, 0.50_08, 0.49_76, 0.50_18, 0.49_90, 0.49_82, 0.49_87]					)

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






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




					'''simple docstring'''



					def 						__UpperCAmelCase     (  self				:	Optional[Any]					)  ->       int:



						'''simple docstring'''





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




					def 						__UpperCAmelCase     (  self				:	Dict     ,  UpperCamelCase_				:	List[str]=0     ,  UpperCamelCase_				:	Optional[Any]=False     ,  UpperCamelCase_				:	Optional[Any]="cpu"     ,  UpperCamelCase_				:	int=torch.floataa     ,  UpperCamelCase_				:	Union[str, Any]=(1, 3, 64, 64)					)  ->       Tuple:



						'''simple docstring'''





						_lowercase					:       Optional[int]       =		torch.manual_seed(__a					)

						_lowercase					:       Optional[Any]       =		{
						    "num_inference_steps": None,
						    "timesteps": [22, 0],
						    "class_labels": 0,
						    "generator": generator,
						    "output_type": "np",
						}

						if get_fixed_latents:
							_lowercase					:       List[str]       =		self.get_fixed_latents(seed=__a     ,  device=__a     ,  dtype=__a     ,  shape=__a					)
							_lowercase					:       List[Any]       =		latents

						return inputs




					def 						__UpperCAmelCase     (  self				:	Tuple     ,  UpperCamelCase_				:	str=0     ,  UpperCamelCase_				:	int="cpu"     ,  UpperCamelCase_				:	List[str]=torch.floataa     ,  UpperCamelCase_				:	Tuple=(1, 3, 64, 64)					)  ->       Dict:



						'''simple docstring'''





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




					def 						__UpperCAmelCase     (  self				:	List[str]					)  ->       Any:



						'''simple docstring'''





						_lowercase					:       Optional[Any]       =		UNetaDModel.from_pretrained('diffusers/consistency_models'     ,  subfolder='diffusers_cd_imagenet64_l2'					)
						_lowercase					:       List[Any]       =		CMStochasticIterativeScheduler(
						    num_train_timesteps=40     ,  sigma_min=0.0_02     ,  sigma_max=80.0     ,  )
						_lowercase					:       List[Any]       =		ConsistencyModelPipeline(unet=__a     ,  scheduler=__a					)
						pipe.to(torch_device=__a					)
						pipe.set_progress_bar_config(disable=__a					)

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

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

						_lowercase					:       Tuple       =		np.array([0.08_88, 0.08_81, 0.06_66, 0.04_79, 0.02_92, 0.01_95, 0.02_01, 0.01_63, 0.02_54]					)

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




					def 						__UpperCAmelCase     (  self				:	Union[str, Any]					)  ->       Any:



						'''simple docstring'''





						_lowercase					:       Optional[Any]       =		UNetaDModel.from_pretrained('diffusers/consistency_models'     ,  subfolder='diffusers_cd_imagenet64_l2'					)
						_lowercase					:       Union[str, Any]       =		CMStochasticIterativeScheduler(
						    num_train_timesteps=40     ,  sigma_min=0.0_02     ,  sigma_max=80.0     ,  )
						_lowercase					:       List[Any]       =		ConsistencyModelPipeline(unet=__a     ,  scheduler=__a					)
						pipe.to(torch_device=__a					)
						pipe.set_progress_bar_config(disable=__a					)

						_lowercase					:       List[str]       =		self.get_inputs()
						_lowercase					:       List[str]       =		1
						_lowercase					:       Optional[Any]       =		None
						_lowercase					:       Optional[int]       =		pipe(**__a					).images
						assert image.shape == (1, 64, 64, 3)

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

						_lowercase					:       int       =		np.array([0.03_40, 0.01_52, 0.00_63, 0.02_67, 0.02_21, 0.01_07, 0.04_16, 0.01_86, 0.02_17]					)

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




					@require_torch_a
					def 						__UpperCAmelCase     (  self				:	List[Any]					)  ->       Any:



						'''simple docstring'''





						_lowercase					:       List[str]       =		UNetaDModel.from_pretrained('diffusers/consistency_models'     ,  subfolder='diffusers_cd_imagenet64_l2'					)
						_lowercase					:       Union[str, Any]       =		CMStochasticIterativeScheduler(
						    num_train_timesteps=40     ,  sigma_min=0.0_02     ,  sigma_max=80.0     ,  )
						_lowercase					:       Any       =		ConsistencyModelPipeline(unet=__a     ,  scheduler=__a					)
						pipe.to(torch_device=__a     ,  torch_dtype=torch.floataa					)
						pipe.set_progress_bar_config(disable=__a					)

						_lowercase					:       Optional[int]       =		self.get_inputs(get_fixed_latents=__a     ,  device=__a					)
						# Ensure usage of flash attention in torch 2.0
						with sdp_kernel(enable_flash=__a     ,  enable_math=__a     ,  enable_mem_efficient=__a					):
							_lowercase					:       Union[str, Any]       =		pipe(**__a					).images
						assert image.shape == (1, 64, 64, 3)

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

						_lowercase					:       Tuple       =		np.array([0.18_75, 0.14_28, 0.12_89, 0.21_51, 0.20_92, 0.14_77, 0.18_77, 0.16_41, 0.13_53]					)

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




					@require_torch_a
					def 						__UpperCAmelCase     (  self				:	Tuple					)  ->       str:



						'''simple docstring'''





						_lowercase					:       List[Any]       =		UNetaDModel.from_pretrained('diffusers/consistency_models'     ,  subfolder='diffusers_cd_imagenet64_l2'					)
						_lowercase					:       Optional[Any]       =		CMStochasticIterativeScheduler(
						    num_train_timesteps=40     ,  sigma_min=0.0_02     ,  sigma_max=80.0     ,  )
						_lowercase					:       Optional[Any]       =		ConsistencyModelPipeline(unet=__a     ,  scheduler=__a					)
						pipe.to(torch_device=__a     ,  torch_dtype=torch.floataa					)
						pipe.set_progress_bar_config(disable=__a					)

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

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

						_lowercase					:       Any       =		np.array([0.16_63, 0.19_48, 0.22_75, 0.16_80, 0.12_04, 0.12_45, 0.18_58, 0.13_38, 0.20_95]					)

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





'''simple docstring'''

_A				:    Optional[Any]     						='''ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/'''







def 						__UpperCamelCase				(							_lowercase					) ->			bytes:
	# Make sure the supplied data is a bytes-like object
	if not isinstance(_lowercase,				_lowercase					):
		_lowercase					:       Union[str, Any]       =		f'''a bytes-like object is required, not \'{data.__class__.__name__}\''''
		raise TypeError(_lowercase					)

	_lowercase					:       int       =		''.join(bin(_lowercase					)[2:].zfill(8					) for byte in data					)

	_lowercase					:       Dict       =		len(_lowercase					) % 6 != 0

	if padding_needed:
		# The padding that will be added later
		_lowercase					:       Optional[Any]       =		B'=' * ((6 - len(_lowercase					) % 6) // 2)

		# Append binary_stream with arbitrary binary digits (0's by default) to make its
		# length a multiple of 6.
		binary_stream += "0" * (6 - len(_lowercase					) % 6)
	else:
		_lowercase					:       Optional[int]       =		B''

	# Encode every 6 binary digits to their corresponding Base64 character
	return (
	    "".join(
	        B64_CHARSET[int(binary_stream[index : index + 6],				2					)]
	        for index in range(0,				len(_lowercase					),				6					)					).encode()
	    + padding
	)







def 						__UpperCamelCase				(							_lowercase					) ->			bytes:
	# Make sure encoded_data is either a string or a bytes-like object
	if not isinstance(_lowercase,				_lowercase					) and not isinstance(_lowercase,				_lowercase					):
		_lowercase					:       int       =		(
		    'argument should be a bytes-like object or ASCII string, '
		    f'''not \'{encoded_data.__class__.__name__}\''''
		)
		raise TypeError(_lowercase					)

	# In case encoded_data is a bytes-like object, make sure it contains only
	# ASCII characters so we convert it to a string object
	if isinstance(_lowercase,				_lowercase					):
		try:
			_lowercase					:       Optional[int]       =		encoded_data.decode('utf-8'					)
		except UnicodeDecodeError:
			raise ValueError('base64 encoded data should only contain ASCII characters'					)

	_lowercase					:       Optional[int]       =		encoded_data.count('='					)

	# Check if the encoded string contains non base64 characters
	if padding:
		assert all(
		    char in B64_CHARSET for char in encoded_data[:-padding]					), "Invalid base64 character(s) found."
	else:
		assert all(
		    char in B64_CHARSET for char in encoded_data					), "Invalid base64 character(s) found."

	# Check the padding
	assert len(_lowercase					) % 4 == 0 and padding < 3, "Incorrect padding"

	if padding:
		# Remove padding if there is one
		_lowercase					:       str       =		encoded_data[:-padding]

		_lowercase					:       Tuple       =		''.join(
		    bin(B64_CHARSET.index(_lowercase					)					)[2:].zfill(6					) for char in encoded_data					)[: -padding * 2]
	else:
		_lowercase					:       Union[str, Any]       =		''.join(
		    bin(B64_CHARSET.index(_lowercase					)					)[2:].zfill(6					) for char in encoded_data					)

	_lowercase					:       List[str]       =		[
	    int(binary_stream[index : index + 8],				2					)
	    for index in range(0,				len(_lowercase					),				8					)
	]

	return bytes(_lowercase					)


if __name__ == "__main__":
		import doctest

		doctest.testmod()
