




'''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"""
from collections import OrderedDict
from typing import Mapping

from packaging import version

from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig
from ...utils import logging


SCREAMING_SNAKE_CASE :    Dict									=				logging.get_logger(__name__)

SCREAMING_SNAKE_CASE :    Optional[int]									=				{
    """hustvl/yolos-small""": """https://huggingface.co/hustvl/yolos-small/resolve/main/config.json""",
    # See all YOLOS models at https://huggingface.co/models?filter=yolos
}



class 			_UpperCAmelCase (							__snake_case				):






		'''simple docstring'''

		lowerCamelCase__				='yolos'



		def __init__(self				,       a_=7_68				,       a_=12				,       a_=12				,       a_=30_72				,       a_="gelu"				,       a_=0.0				,       a_=0.0				,       a_=0.02				,       a_=1E-12				,       a_=[5_12, 8_64]				,       a_=16				,       a_=3				,       a_=True				,       a_=1_00				,       a_=True				,       a_=False				,       a_=1				,       a_=5				,       a_=2				,       a_=5				,       a_=2				,       a_=0.1				,       **a_				,       ):





			'''simple docstring'''

			super().__init__(**a_					)

			__snake_case	:						Union[str, Any]						=				hidden_size
			__snake_case	:						Union[str, Any]						=				num_hidden_layers
			__snake_case	:						Dict						=				num_attention_heads
			__snake_case	:						str						=				intermediate_size
			__snake_case	:						List[str]						=				hidden_act
			__snake_case	:						Tuple						=				hidden_dropout_prob
			__snake_case	:						Optional[int]						=				attention_probs_dropout_prob
			__snake_case	:						Union[str, Any]						=				initializer_range
			__snake_case	:						Tuple						=				layer_norm_eps
			__snake_case	:						List[Any]						=				image_size
			__snake_case	:						Tuple						=				patch_size
			__snake_case	:						str						=				num_channels
			__snake_case	:						Tuple						=				qkv_bias
			__snake_case	:						Union[str, Any]						=				num_detection_tokens
			__snake_case	:						List[str]						=				use_mid_position_embeddings
			__snake_case	:						Tuple						=				auxiliary_loss
			# Hungarian matcher
			__snake_case	:						List[str]						=				class_cost
			__snake_case	:						int						=				bbox_cost
			__snake_case	:						int						=				giou_cost
			# Loss coefficients
			__snake_case	:						Optional[int]						=				bbox_loss_coefficient
			__snake_case	:						List[str]						=				giou_loss_coefficient
			__snake_case	:						List[Any]						=				eos_coefficient



class 			_UpperCAmelCase (							__snake_case				):






		'''simple docstring'''

		lowerCamelCase__				=version.parse('1.11'				)



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			return OrderedDict(
			    [
			        ('''pixel_values''', {0: '''batch''', 1: '''num_channels''', 2: '''height''', 3: '''width'''}),
			    ]					)



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			return 1E-4



		@property
		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			return 12





