




"""simple docstring"""





# DISCLAIMER: This file is strongly influenced by https://github.com/ermongroup/ddim

from dataclasses import dataclass
from typing import Optional, Tuple, Union

import flax
import jax
import jax.numpy as jnp

from ..configuration_utils import ConfigMixin, register_to_config
from .scheduling_utils_flax import (
    CommonSchedulerState,
    FlaxKarrasDiffusionSchedulers,
    FlaxSchedulerMixin,
    FlaxSchedulerOutput,
    add_noise_common,
    get_velocity_common,
)



@flax.struct.dataclass
class      a__				:
			lowercase_    =	42

			# setable values
			lowercase_    =	42
			lowercase_    =	42
			lowercase_    =	None


			@classmethod
			def       a_     (      cls      :    List[str]      ,  UpperCamelCase_      :    CommonSchedulerState      ,  UpperCamelCase_      :    jnp.ndarray      ,  UpperCamelCase_      :    jnp.ndarray):


						"""simple docstring"""




						return cls(common=UpperCamelCase_      ,  init_noise_sigma=UpperCamelCase_      ,  timesteps=UpperCamelCase_)






@dataclass
class      a__				(	__magic_name__   ):
			lowercase_    =	42






class      a__				(	__magic_name__							,    __magic_name__   ):
			lowercase_    =	[e.name for e in FlaxKarrasDiffusionSchedulers]

			lowercase_    =	42


			@property
			def       a_     (      self      :    Optional[int]):


						"""simple docstring"""




						return True


			@register_to_config
			def __init__(      self      :    str      ,  UpperCamelCase_      :    int = 1000      ,  UpperCamelCase_      :    float = 0.0001      ,  UpperCamelCase_      :    float = 0.02      ,  UpperCamelCase_      :    str = "linear"      ,  UpperCamelCase_      :    Optional[jnp.ndarray] = None      ,  UpperCamelCase_      :    str = "fixed_small"      ,  UpperCamelCase_      :    bool = True      ,  UpperCamelCase_      :    str = "epsilon"      ,  UpperCamelCase_      :    jnp.dtype = jnp.floataa      ,  ):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[int]		=					dtype


			def       a_     (      self      :    Optional[Any]      ,  UpperCamelCase_      :    Optional[CommonSchedulerState] = None):


						"""simple docstring"""




						if common is None:
									__UpperCAmelCase	: Tuple		=					CommonSchedulerState.create(self)

						# standard deviation of the initial noise distribution
						__UpperCAmelCase	: Tuple		=					jnp.array(1.0      ,  dtype=self.dtype)

						__UpperCAmelCase	: Any		=					jnp.arange(0      ,  self.config.num_train_timesteps).round()[::-1]

						return DDPMSchedulerState.create(
						    common=UpperCamelCase_      ,  init_noise_sigma=UpperCamelCase_      ,  timesteps=UpperCamelCase_      ,  )


			def       a_     (      self      :    Optional[Any]      ,  UpperCamelCase_      :    DDPMSchedulerState      ,  UpperCamelCase_      :    jnp.ndarray      ,  UpperCamelCase_      :    Optional[int] = None):


						"""simple docstring"""




						return sample


			def       a_     (      self      :    Any      ,  UpperCamelCase_      :    DDPMSchedulerState      ,  UpperCamelCase_      :    int      ,  UpperCamelCase_      :    Tuple = ()):


						"""simple docstring"""




						__UpperCAmelCase	: List[str]		=					self.config.num_train_timesteps // num_inference_steps
						# creates integer timesteps by multiplying by ratio
						# rounding to avoid issues when num_inference_step is power of 3
						__UpperCAmelCase	: List[str]		=					(jnp.arange(0      ,  UpperCamelCase_) * step_ratio).round()[::-1]

						return state.replace(
						    num_inference_steps=UpperCamelCase_      ,  timesteps=UpperCamelCase_      ,  )


			def       a_     (      self      :    Any      ,  UpperCamelCase_      :    DDPMSchedulerState      ,  UpperCamelCase_      :    Optional[Any]      ,  UpperCamelCase_      :    Union[str, Any]=None      ,  UpperCamelCase_      :    Optional[int]=None):


						"""simple docstring"""




						__UpperCAmelCase	: List[str]		=					state.common.alphas_cumprod[t]
						__UpperCAmelCase	: Optional[int]		=					jnp.where(t > 0      ,  state.common.alphas_cumprod[t - 1]      ,  jnp.array(1.0      ,  dtype=self.dtype))

						# For t > 0, compute predicted variance βt (see formula (6) and (7) from https://arxiv.org/pdf/2006.11239.pdf)
						# and sample from it to get previous sample
						# x_{t-1} ~ N(pred_prev_sample, variance) == add variance to pred_sample
						__UpperCAmelCase	: Tuple		=					(1 - alpha_prod_t_prev) / (1 - alpha_prod_t) * state.common.betas[t]

						if variance_type is None:
									__UpperCAmelCase	: Optional[Any]		=					self.config.variance_type

						# hacks - were probably added for training stability
						if variance_type == "fixed_small":
									__UpperCAmelCase	: str		=					jnp.clip(UpperCamelCase_      ,  a_min=1e-20)
						# for rl-diffuser https://arxiv.org/abs/2205.09991
						elif variance_type == "fixed_small_log":
									__UpperCAmelCase	: Optional[int]		=					jnp.log(jnp.clip(UpperCamelCase_      ,  a_min=1e-20))
						elif variance_type == "fixed_large":
									__UpperCAmelCase	: Tuple		=					state.common.betas[t]
						elif variance_type == "fixed_large_log":
									# Glide max_log
									__UpperCAmelCase	: str		=					jnp.log(state.common.betas[t])
						elif variance_type == "learned":
									return predicted_variance
						elif variance_type == "learned_range":
									__UpperCAmelCase	: Any		=					variance
									__UpperCAmelCase	: Union[str, Any]		=					state.common.betas[t]
									__UpperCAmelCase	: List[str]		=					(predicted_variance + 1) / 2
									__UpperCAmelCase	: int		=					frac * max_log + (1 - frac) * min_log

						return variance


			def       a_     (      self      :    Optional[int]      ,  UpperCamelCase_      :    DDPMSchedulerState      ,  UpperCamelCase_      :    jnp.ndarray      ,  UpperCamelCase_      :    int      ,  UpperCamelCase_      :    jnp.ndarray      ,  UpperCamelCase_      :    Optional[jax.random.KeyArray] = None      ,  UpperCamelCase_      :    bool = True      ,  ):


						"""simple docstring"""




						__UpperCAmelCase	: Dict		=					timestep

						if key is None:
									__UpperCAmelCase	: List[str]		=					jax.random.PRNGKey(0)

						if model_output.shape[1] == sample.shape[1] * 2 and self.config.variance_type in ["learned", "learned_range"]:
									__UpperCAmelCase	,     __UpperCAmelCase	: int		=					jnp.split(UpperCamelCase_      ,  sample.shape[1]      ,  axis=1)
						else:
									__UpperCAmelCase	: List[str]		=					None

						# 1. compute alphas, betas
						__UpperCAmelCase	: str		=					state.common.alphas_cumprod[t]
						__UpperCAmelCase	: str		=					jnp.where(t > 0      ,  state.common.alphas_cumprod[t - 1]      ,  jnp.array(1.0      ,  dtype=self.dtype))
						__UpperCAmelCase	: Tuple		=					1 - alpha_prod_t
						__UpperCAmelCase	: int		=					1 - alpha_prod_t_prev

						# 2. compute predicted original sample from predicted noise also called
						# "predicted x_0" of formula (15) from https://arxiv.org/pdf/2006.11239.pdf
						if self.config.prediction_type == "epsilon":
									__UpperCAmelCase	: Any		=					(sample - beta_prod_t ** 0.5 * model_output) / alpha_prod_t ** 0.5
						elif self.config.prediction_type == "sample":
									__UpperCAmelCase	: Optional[int]		=					model_output
						elif self.config.prediction_type == "v_prediction":
									__UpperCAmelCase	: Optional[int]		=					(alpha_prod_t**0.5) * sample - (beta_prod_t**0.5) * model_output
						else:
									raise ValueError(
									    F"prediction_type given as {self.config.prediction_type} must be one of `epsilon`, `sample` "
									    " for the FlaxDDPMScheduler.")

						# 3. Clip "predicted x_0"
						if self.config.clip_sample:
									__UpperCAmelCase	: List[Any]		=					jnp.clip(UpperCamelCase_      ,  -1      ,  1)

						# 4. Compute coefficients for pred_original_sample x_0 and current sample x_t
						# See formula (7) from https://arxiv.org/pdf/2006.11239.pdf
						__UpperCAmelCase	: Union[str, Any]		=					(alpha_prod_t_prev ** 0.5 * state.common.betas[t]) / beta_prod_t
						__UpperCAmelCase	: Union[str, Any]		=					state.common.alphas[t] ** 0.5 * beta_prod_t_prev / beta_prod_t

						# 5. Compute predicted previous sample µ_t
						# See formula (7) from https://arxiv.org/pdf/2006.11239.pdf
						__UpperCAmelCase	: Optional[Any]		=					pred_original_sample_coeff * pred_original_sample + current_sample_coeff * sample

						# 6. Add noise
						def random_variance():
									__UpperCAmelCase	: int		=					jax.random.split(UpperCamelCase_      ,  num=1)
									__UpperCAmelCase	: Any		=					jax.random.normal(UpperCamelCase_      ,  shape=model_output.shape      ,  dtype=self.dtype)
									return (self._get_variance(UpperCamelCase_      ,  UpperCamelCase_      ,  predicted_variance=UpperCamelCase_) ** 0.5) * noise

						__UpperCAmelCase	: Tuple		=					jnp.where(t > 0      ,  random_variance()      ,  jnp.zeros(model_output.shape      ,  dtype=self.dtype))

						__UpperCAmelCase	: Tuple		=					pred_prev_sample + variance

						if not return_dict:
									return (pred_prev_sample, state)

						return FlaxDDPMSchedulerOutput(prev_sample=UpperCamelCase_      ,  state=UpperCamelCase_)


			def       a_     (      self      :    Optional[Any]      ,  UpperCamelCase_      :    DDPMSchedulerState      ,  UpperCamelCase_      :    jnp.ndarray      ,  UpperCamelCase_      :    jnp.ndarray      ,  UpperCamelCase_      :    jnp.ndarray      ,  ):


						"""simple docstring"""




						return add_noise_common(state.common      ,  UpperCamelCase_      ,  UpperCamelCase_      ,  UpperCamelCase_)


			def       a_     (      self      :    Union[str, Any]      ,  UpperCamelCase_      :    DDPMSchedulerState      ,  UpperCamelCase_      :    jnp.ndarray      ,  UpperCamelCase_      :    jnp.ndarray      ,  UpperCamelCase_      :    jnp.ndarray      ,  ):


						"""simple docstring"""




						return get_velocity_common(state.common      ,  UpperCamelCase_      ,  UpperCamelCase_      ,  UpperCamelCase_)


			def __len__(      self      :    int):


						"""simple docstring"""




						return self.config.num_train_timesteps








"""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__				(	__magic_name__							,    unittest.TestCase   ):
			lowercase_    =	ConsistencyModelPipeline
			lowercase_    =	UNCONDITIONAL_IMAGE_GENERATION_PARAMS
			lowercase_    =	UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

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


			@property
			def       a_     (      self      :    Optional[Any]):


						"""simple docstring"""




						__UpperCAmelCase	: Union[str, Any]		=					UNetaDModel.from_pretrained(
						    "diffusers/consistency-models-test"      ,  subfolder="test_unet"      ,  )
						return unet


			@property
			def       a_     (      self      :    List[Any]):


						"""simple docstring"""




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


			def       a_     (      self      :    Any      ,  UpperCamelCase_      :    int=False):


						"""simple docstring"""




						if class_cond:
									__UpperCAmelCase	: List[Any]		=					self.dummy_cond_unet
						else:
									__UpperCAmelCase	: Optional[int]		=					self.dummy_uncond_unet

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

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

						return components


			def       a_     (      self      :    List[str]      ,  UpperCamelCase_      :    Tuple      ,  UpperCamelCase_      :    List[Any]=0):


						"""simple docstring"""




						if str(UpperCamelCase_).startswith("mps"):
									__UpperCAmelCase	: str		=					torch.manual_seed(UpperCamelCase_)
						else:
									__UpperCAmelCase	: Optional[Any]		=					torch.Generator(device=UpperCamelCase_).manual_seed(UpperCamelCase_)

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

						return inputs


			def       a_     (      self      :    int):


						"""simple docstring"""




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

						__UpperCAmelCase	: Any		=					self.get_dummy_inputs(UpperCamelCase_)
						__UpperCAmelCase	: str		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 32, 32, 3)

						__UpperCAmelCase	: Dict		=					image[0, -3:, -3:, -1]
						__UpperCAmelCase	: List[Any]		=					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      :    Optional[int]):


						"""simple docstring"""




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

						__UpperCAmelCase	: Tuple		=					self.get_dummy_inputs(UpperCamelCase_)
						__UpperCAmelCase	: Dict		=					0
						__UpperCAmelCase	: int		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 32, 32, 3)

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


						"""simple docstring"""




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

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

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


						"""simple docstring"""




						__UpperCAmelCase	: Dict		=					"cpu"  # ensure determinism for the device-dependent torch.Generator
						__UpperCAmelCase	: List[str]		=					self.get_dummy_components(class_cond=UpperCamelCase_)
						__UpperCAmelCase	: Tuple		=					ConsistencyModelPipeline(**UpperCamelCase_)
						__UpperCAmelCase	: int		=					pipe.to(UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: Tuple		=					self.get_dummy_inputs(UpperCamelCase_)
						__UpperCAmelCase	: Optional[int]		=					1
						__UpperCAmelCase	: Optional[int]		=					None
						__UpperCAmelCase	: Tuple		=					0
						__UpperCAmelCase	: Tuple		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 32, 32, 3)

						__UpperCAmelCase	: List[str]		=					image[0, -3:, -3:, -1]
						__UpperCAmelCase	: 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      a__				(	unittest.TestCase   ):


			def       a_     (      self      :    Optional[int]):


						"""simple docstring"""




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


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


						"""simple docstring"""




						__UpperCAmelCase	: Optional[int]		=					torch.manual_seed(UpperCamelCase_)

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

						if get_fixed_latents:
									__UpperCAmelCase	: int		=					self.get_fixed_latents(seed=UpperCamelCase_      ,  device=UpperCamelCase_      ,  dtype=UpperCamelCase_      ,  shape=UpperCamelCase_)
									__UpperCAmelCase	: Optional[int]		=					latents

						return inputs


			def       a_     (      self      :    Union[str, Any]      ,  UpperCamelCase_      :    int=0      ,  UpperCamelCase_      :    Tuple="cpu"      ,  UpperCamelCase_      :    Tuple=torch.floataa      ,  UpperCamelCase_      :    Optional[Any]=(1, 3, 64, 64)):


						"""simple docstring"""




						if type(UpperCamelCase_) == str:
									__UpperCAmelCase	: Union[str, Any]		=					torch.device(UpperCamelCase_)
						__UpperCAmelCase	: Union[str, Any]		=					torch.Generator(device=UpperCamelCase_).manual_seed(UpperCamelCase_)
						__UpperCAmelCase	: Optional[int]		=					randn_tensor(UpperCamelCase_      ,  generator=UpperCamelCase_      ,  device=UpperCamelCase_      ,  dtype=UpperCamelCase_)
						return latents


			def       a_     (      self      :    List[Any]):


						"""simple docstring"""




						__UpperCAmelCase	: Optional[int]		=					UNetaDModel.from_pretrained("diffusers/consistency_models"      ,  subfolder="diffusers_cd_imagenet64_l2")
						__UpperCAmelCase	: Dict		=					CMStochasticIterativeScheduler(
						    num_train_timesteps=40      ,  sigma_min=0.002      ,  sigma_max=80.0      ,  )
						__UpperCAmelCase	: Dict		=					ConsistencyModelPipeline(unet=UpperCamelCase_      ,  scheduler=UpperCamelCase_)
						pipe.to(torch_device=UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: Dict		=					self.get_inputs()
						__UpperCAmelCase	: List[str]		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 64, 64, 3)

						__UpperCAmelCase	: List[Any]		=					image[0, -3:, -3:, -1]

						__UpperCAmelCase	: 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       a_     (      self      :    Tuple):


						"""simple docstring"""




						__UpperCAmelCase	: List[str]		=					UNetaDModel.from_pretrained("diffusers/consistency_models"      ,  subfolder="diffusers_cd_imagenet64_l2")
						__UpperCAmelCase	: Optional[Any]		=					CMStochasticIterativeScheduler(
						    num_train_timesteps=40      ,  sigma_min=0.002      ,  sigma_max=80.0      ,  )
						__UpperCAmelCase	: Any		=					ConsistencyModelPipeline(unet=UpperCamelCase_      ,  scheduler=UpperCamelCase_)
						pipe.to(torch_device=UpperCamelCase_)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

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

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

						__UpperCAmelCase	: 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       a_     (      self      :    Tuple):


						"""simple docstring"""




						__UpperCAmelCase	: int		=					UNetaDModel.from_pretrained("diffusers/consistency_models"      ,  subfolder="diffusers_cd_imagenet64_l2")
						__UpperCAmelCase	: Optional[int]		=					CMStochasticIterativeScheduler(
						    num_train_timesteps=40      ,  sigma_min=0.002      ,  sigma_max=80.0      ,  )
						__UpperCAmelCase	: int		=					ConsistencyModelPipeline(unet=UpperCamelCase_      ,  scheduler=UpperCamelCase_)
						pipe.to(torch_device=UpperCamelCase_      ,  torch_dtype=torch.floataa)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

						__UpperCAmelCase	: Tuple		=					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_):
									__UpperCAmelCase	: List[str]		=					pipe(**UpperCamelCase_).images
						assert image.shape == (1, 64, 64, 3)

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

						__UpperCAmelCase	: Optional[int]		=					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      :    str):


						"""simple docstring"""




						__UpperCAmelCase	: Tuple		=					UNetaDModel.from_pretrained("diffusers/consistency_models"      ,  subfolder="diffusers_cd_imagenet64_l2")
						__UpperCAmelCase	: Union[str, Any]		=					CMStochasticIterativeScheduler(
						    num_train_timesteps=40      ,  sigma_min=0.002      ,  sigma_max=80.0      ,  )
						__UpperCAmelCase	: List[Any]		=					ConsistencyModelPipeline(unet=UpperCamelCase_      ,  scheduler=UpperCamelCase_)
						pipe.to(torch_device=UpperCamelCase_      ,  torch_dtype=torch.floataa)
						pipe.set_progress_bar_config(disable=UpperCamelCase_)

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

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

						__UpperCAmelCase	: List[str]		=					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


