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







     """simple docstring"""






     SCREAMING_SNAKE_CASE__		   =						ConsistencyModelPipeline
     SCREAMING_SNAKE_CASE__		   =						UNCONDITIONAL_IMAGE_GENERATION_PARAMS
     SCREAMING_SNAKE_CASE__		   =						UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

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







     @property
     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	List[Any]		    =  UNetaDModel.from_pretrained(
             'diffusers/consistency-models-test'     ,			subfolder='test_unet'     ,			)
         return unet







     @property
     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	Tuple		    =  UNetaDModel.from_pretrained(
             'diffusers/consistency-models-test'     ,			subfolder='test_unet_class_cond'     ,			)
         return unet







     def lowerCAmelCase     (		self     ,			_lowerCamelCase=False    ):
         if class_cond:
             __UpperCamelCase	:	int		    =  self.dummy_cond_unet
         else:
             __UpperCamelCase	:	List[str]		    =  self.dummy_uncond_unet

         # Default to CM multistep sampler
         __UpperCamelCase	:	int		    =  CMStochasticIterativeScheduler(
             num_train_timesteps=4_0     ,			sigma_min=0.0_0_2     ,			sigma_max=8_0.0     ,			)

         __UpperCamelCase	:	str		    =  {
             'unet': unet,
             'scheduler': scheduler,
         }

         return components







     def lowerCAmelCase     (		self     ,			_lowerCamelCase     ,			_lowerCamelCase=0    ):
         if str(_lowerCamelCase    ).startswith('mps'    ):
             __UpperCamelCase	:	Optional[int]		    =  torch.manual_seed(_lowerCamelCase    )
         else:
             __UpperCamelCase	:	str		    =  torch.Generator(device=_lowerCamelCase    ).manual_seed(_lowerCamelCase    )

         __UpperCamelCase	:	Optional[Any]		    =  {
             'batch_size': 1,
             'num_inference_steps': None,
             'timesteps': [2_2, 0],
             'generator': generator,
             'output_type': 'np',
         }

         return inputs







     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	Union[str, Any]		    =  'cpu'  # ensure determinism for the device-dependent torch.Generator
         __UpperCamelCase	:	Optional[Any]		    =  self.get_dummy_components()
         __UpperCamelCase	:	List[Any]		    =  ConsistencyModelPipeline(**_lowerCamelCase    )
         __UpperCamelCase	:	Any		    =  pipe.to(_lowerCamelCase    )
         pipe.set_progress_bar_config(disable=_lowerCamelCase    )

         __UpperCamelCase	:	Dict		    =  self.get_dummy_inputs(_lowerCamelCase    )
         __UpperCamelCase	:	str		    =  pipe(**_lowerCamelCase    ).images
         assert image.shape == (1, 3_2, 3_2, 3)

         __UpperCamelCase	:	Tuple		    =  image[0, -3:, -3:, -1]
         __UpperCamelCase	:	Any		    =  np.array([0.3_5_7_2, 0.6_2_7_3, 0.4_0_3_1, 0.3_9_6_1, 0.4_3_2_1, 0.5_7_3_0, 0.5_2_6_6, 0.4_7_8_0, 0.5_0_0_4]    )

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







     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	Any		    =  'cpu'  # ensure determinism for the device-dependent torch.Generator
         __UpperCamelCase	:	Optional[Any]		    =  self.get_dummy_components(class_cond=_lowerCamelCase    )
         __UpperCamelCase	:	Any		    =  ConsistencyModelPipeline(**_lowerCamelCase    )
         __UpperCamelCase	:	str		    =  pipe.to(_lowerCamelCase    )
         pipe.set_progress_bar_config(disable=_lowerCamelCase    )

         __UpperCamelCase	:	List[Any]		    =  self.get_dummy_inputs(_lowerCamelCase    )
         __UpperCamelCase	:	List[Any]		    =  0
         __UpperCamelCase	:	Tuple		    =  pipe(**_lowerCamelCase    ).images
         assert image.shape == (1, 3_2, 3_2, 3)

         __UpperCamelCase	:	int		    =  image[0, -3:, -3:, -1]
         __UpperCamelCase	:	Optional[Any]		    =  np.array([0.3_5_7_2, 0.6_2_7_3, 0.4_0_3_1, 0.3_9_6_1, 0.4_3_2_1, 0.5_7_3_0, 0.5_2_6_6, 0.4_7_8_0, 0.5_0_0_4]    )

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







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

         __UpperCamelCase	:	Optional[Any]		    =  self.get_dummy_inputs(_lowerCamelCase    )
         __UpperCamelCase	:	Dict		    =  1
         __UpperCamelCase	:	Optional[Any]		    =  None
         __UpperCamelCase	:	Any		    =  pipe(**_lowerCamelCase    ).images
         assert image.shape == (1, 3_2, 3_2, 3)

         __UpperCamelCase	:	Union[str, Any]		    =  image[0, -3:, -3:, -1]
         __UpperCamelCase	:	List[Any]		    =  np.array([0.5_0_0_4, 0.5_0_0_4, 0.4_9_9_4, 0.5_0_0_8, 0.4_9_7_6, 0.5_0_1_8, 0.4_9_9_0, 0.4_9_8_2, 0.4_9_8_7]    )

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







     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	str		    =  'cpu'  # ensure determinism for the device-dependent torch.Generator
         __UpperCamelCase	:	Any		    =  self.get_dummy_components(class_cond=_lowerCamelCase    )
         __UpperCamelCase	:	Any		    =  ConsistencyModelPipeline(**_lowerCamelCase    )
         __UpperCamelCase	:	int		    =  pipe.to(_lowerCamelCase    )
         pipe.set_progress_bar_config(disable=_lowerCamelCase    )

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

         __UpperCamelCase	:	List[Any]		    =  image[0, -3:, -3:, -1]
         __UpperCamelCase	:	Optional[int]		    =  np.array([0.5_0_0_4, 0.5_0_0_4, 0.4_9_9_4, 0.5_0_0_8, 0.4_9_7_6, 0.5_0_1_8, 0.4_9_9_0, 0.4_9_8_2, 0.4_9_8_7]    )

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






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







     """simple docstring"""






     def lowerCAmelCase     (		self    ):
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()







     def lowerCAmelCase     (		self     ,			_lowerCamelCase=0     ,			_lowerCamelCase=False     ,			_lowerCamelCase="cpu"     ,			_lowerCamelCase=torch.floataa     ,			_lowerCamelCase=(1, 3, 6_4, 6_4)    ):
         __UpperCamelCase	:	Any		    =  torch.manual_seed(_lowerCamelCase    )

         __UpperCamelCase	:	int		    =  {
             'num_inference_steps': None,
             'timesteps': [2_2, 0],
             'class_labels': 0,
             'generator': generator,
             'output_type': 'np',
         }

         if get_fixed_latents:
             __UpperCamelCase	:	Optional[Any]		    =  self.get_fixed_latents(seed=_lowerCamelCase     ,			device=_lowerCamelCase     ,			dtype=_lowerCamelCase     ,			shape=_lowerCamelCase    )
             __UpperCamelCase	:	int		    =  latents

         return inputs







     def lowerCAmelCase     (		self     ,			_lowerCamelCase=0     ,			_lowerCamelCase="cpu"     ,			_lowerCamelCase=torch.floataa     ,			_lowerCamelCase=(1, 3, 6_4, 6_4)    ):
         if type(_lowerCamelCase    ) == str:
             __UpperCamelCase	:	Union[str, Any]		    =  torch.device(_lowerCamelCase    )
         __UpperCamelCase	:	Dict		    =  torch.Generator(device=_lowerCamelCase    ).manual_seed(_lowerCamelCase    )
         __UpperCamelCase	:	Any		    =  randn_tensor(_lowerCamelCase     ,			generator=_lowerCamelCase     ,			device=_lowerCamelCase     ,			dtype=_lowerCamelCase    )
         return latents







     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	Optional[Any]		    =  UNetaDModel.from_pretrained('diffusers/consistency_models'     ,			subfolder='diffusers_cd_imagenet64_l2'    )
         __UpperCamelCase	:	Dict		    =  CMStochasticIterativeScheduler(
             num_train_timesteps=4_0     ,			sigma_min=0.0_0_2     ,			sigma_max=8_0.0     ,			)
         __UpperCamelCase	:	List[Any]		    =  ConsistencyModelPipeline(unet=_lowerCamelCase     ,			scheduler=_lowerCamelCase    )
         pipe.to(torch_device=_lowerCamelCase    )
         pipe.set_progress_bar_config(disable=_lowerCamelCase    )

         __UpperCamelCase	:	Any		    =  self.get_inputs()
         __UpperCamelCase	:	Union[str, Any]		    =  pipe(**_lowerCamelCase    ).images
         assert image.shape == (1, 6_4, 6_4, 3)

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

         __UpperCamelCase	:	List[str]		    =  np.array([0.0_8_8_8, 0.0_8_8_1, 0.0_6_6_6, 0.0_4_7_9, 0.0_2_9_2, 0.0_1_9_5, 0.0_2_0_1, 0.0_1_6_3, 0.0_2_5_4]    )

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







     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	Optional[Any]		    =  UNetaDModel.from_pretrained('diffusers/consistency_models'     ,			subfolder='diffusers_cd_imagenet64_l2'    )
         __UpperCamelCase	:	str		    =  CMStochasticIterativeScheduler(
             num_train_timesteps=4_0     ,			sigma_min=0.0_0_2     ,			sigma_max=8_0.0     ,			)
         __UpperCamelCase	:	List[Any]		    =  ConsistencyModelPipeline(unet=_lowerCamelCase     ,			scheduler=_lowerCamelCase    )
         pipe.to(torch_device=_lowerCamelCase    )
         pipe.set_progress_bar_config(disable=_lowerCamelCase    )

         __UpperCamelCase	:	List[Any]		    =  self.get_inputs()
         __UpperCamelCase	:	Optional[int]		    =  1
         __UpperCamelCase	:	str		    =  None
         __UpperCamelCase	:	Any		    =  pipe(**_lowerCamelCase    ).images
         assert image.shape == (1, 6_4, 6_4, 3)

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

         __UpperCamelCase	:	List[Any]		    =  np.array([0.0_3_4_0, 0.0_1_5_2, 0.0_0_6_3, 0.0_2_6_7, 0.0_2_2_1, 0.0_1_0_7, 0.0_4_1_6, 0.0_1_8_6, 0.0_2_1_7]    )

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







     @require_torch_a
     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	int		    =  UNetaDModel.from_pretrained('diffusers/consistency_models'     ,			subfolder='diffusers_cd_imagenet64_l2'    )
         __UpperCamelCase	:	str		    =  CMStochasticIterativeScheduler(
             num_train_timesteps=4_0     ,			sigma_min=0.0_0_2     ,			sigma_max=8_0.0     ,			)
         __UpperCamelCase	:	Any		    =  ConsistencyModelPipeline(unet=_lowerCamelCase     ,			scheduler=_lowerCamelCase    )
         pipe.to(torch_device=_lowerCamelCase     ,			torch_dtype=torch.floataa    )
         pipe.set_progress_bar_config(disable=_lowerCamelCase    )

         __UpperCamelCase	:	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    ):
             __UpperCamelCase	:	Union[str, Any]		    =  pipe(**_lowerCamelCase    ).images
         assert image.shape == (1, 6_4, 6_4, 3)

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

         __UpperCamelCase	:	Dict		    =  np.array([0.1_8_7_5, 0.1_4_2_8, 0.1_2_8_9, 0.2_1_5_1, 0.2_0_9_2, 0.1_4_7_7, 0.1_8_7_7, 0.1_6_4_1, 0.1_3_5_3]    )

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







     @require_torch_a
     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	Optional[Any]		    =  UNetaDModel.from_pretrained('diffusers/consistency_models'     ,			subfolder='diffusers_cd_imagenet64_l2'    )
         __UpperCamelCase	:	Optional[int]		    =  CMStochasticIterativeScheduler(
             num_train_timesteps=4_0     ,			sigma_min=0.0_0_2     ,			sigma_max=8_0.0     ,			)
         __UpperCamelCase	:	List[Any]		    =  ConsistencyModelPipeline(unet=_lowerCamelCase     ,			scheduler=_lowerCamelCase    )
         pipe.to(torch_device=_lowerCamelCase     ,			torch_dtype=torch.floataa    )
         pipe.set_progress_bar_config(disable=_lowerCamelCase    )

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

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

         __UpperCamelCase	:	List[str]		    =  np.array([0.1_6_6_3, 0.1_9_4_8, 0.2_2_7_5, 0.1_6_8_0, 0.1_2_0_4, 0.1_2_4_5, 0.1_8_5_8, 0.1_3_3_8, 0.2_0_9_5]    )

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




'''simple docstring'''
class 			__lowercase		:







     """simple docstring"""






     def __init__(		self    ):
         __UpperCamelCase	:	Any		    =  0
         __UpperCamelCase	:	Any		    =  0
         __UpperCamelCase	:	Any		    =  {}







     def lowerCAmelCase     (		self     ,			_lowerCamelCase    ):
         if vertex not in self.adjacency:
             __UpperCamelCase	:	Tuple		    =  {}
             self.num_vertices += 1







     def lowerCAmelCase     (		self     ,			_lowerCamelCase     ,			_lowerCamelCase     ,			_lowerCamelCase    ):
         self.add_vertex(_lowerCamelCase    )
         self.add_vertex(_lowerCamelCase    )

         if head == tail:
             return

         __UpperCamelCase	:	List[Any]		    =  weight
         __UpperCamelCase	:	Optional[int]		    =  weight







     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	List[str]		    =  self.get_edges()
         for edge in edges:
             __UpperCamelCase    ,					__UpperCamelCase    ,					__UpperCamelCase	:	Optional[int]		    =  edge
             edges.remove((tail, head, weight)    )
         for i in range(len(_lowerCamelCase    )    ):
             __UpperCamelCase	:	Union[str, Any]		    =  list(edges[i]    )

         edges.sort(key=lambda _lowerCamelCase    : e[2]    )
         for i in range(len(_lowerCamelCase    ) - 1    ):
             if edges[i][2] >= edges[i + 1][2]:
                 __UpperCamelCase	:	Optional[int]		    =  edges[i][2] + 1
         for edge in edges:
             __UpperCamelCase    ,					__UpperCamelCase    ,					__UpperCamelCase	:	Optional[int]		    =  edge
             __UpperCamelCase	:	Tuple		    =  weight
             __UpperCamelCase	:	Optional[Any]		    =  weight







     def __str__(		self    ):
         __UpperCamelCase	:	Optional[Any]		    =  ''
         for tail in self.adjacency:
             for head in self.adjacency[tail]:
                 __UpperCamelCase	:	Tuple		    =  self.adjacency[head][tail]
                 string += f"""{head} -> {tail} == {weight}\n"""
         return string.rstrip('\n'    )







     def lowerCAmelCase     (		self    ):
         __UpperCamelCase	:	Any		    =  []
         for tail in self.adjacency:
             for head in self.adjacency[tail]:
                 output.append((tail, head, self.adjacency[head][tail])    )
         return output







     def lowerCAmelCase     (		self    ):
         return self.adjacency.keys()







     @staticmethod
     def lowerCAmelCase     (		_lowerCamelCase=None     ,			_lowerCamelCase=None    ):
         __UpperCamelCase	:	Tuple		    =  Graph()
         if vertices is None:
             __UpperCamelCase	:	List[Any]		    =  []
         if edges is None:
             __UpperCamelCase	:	Optional[int]		    =  []
         for vertex in vertices:
             g.add_vertex(_lowerCamelCase    )
         for edge in edges:
             g.add_edge(*_lowerCamelCase    )
         return g






     class 			__lowercase		:







          """simple docstring"""






          def __init__(		self    ):
              __UpperCamelCase	:	Union[str, Any]		    =  {}
              __UpperCamelCase	:	int		    =  {}







          def __len__(		self    ):
              return len(self.parent    )







          def lowerCAmelCase     (		self     ,			_lowerCamelCase    ):
              if item in self.parent:
                  return self.find(_lowerCamelCase    )

              __UpperCamelCase	:	Optional[Any]		    =  item
              __UpperCamelCase	:	List[str]		    =  0
              return item







          def lowerCAmelCase     (		self     ,			_lowerCamelCase    ):
              if item not in self.parent:
                  return self.make_set(_lowerCamelCase    )
              if item != self.parent[item]:
                  __UpperCamelCase	:	int		    =  self.find(self.parent[item]    )
              return self.parent[item]







          def lowerCAmelCase     (		self     ,			_lowerCamelCase     ,			_lowerCamelCase    ):
              __UpperCamelCase	:	Tuple		    =  self.find(_lowerCamelCase    )
              __UpperCamelCase	:	Any		    =  self.find(_lowerCamelCase    )

              if roota == roota:
                  return roota

              if self.rank[roota] > self.rank[roota]:
                  __UpperCamelCase	:	Dict		    =  roota
                  return roota

              if self.rank[roota] < self.rank[roota]:
                  __UpperCamelCase	:	Optional[Any]		    =  roota
                  return roota

              if self.rank[roota] == self.rank[roota]:
                  self.rank[roota] += 1
                  __UpperCamelCase	:	Dict		    =  roota
                  return roota
              return None







     @staticmethod
     def lowerCAmelCase     (		_lowerCamelCase    ):
         __UpperCamelCase	:	int		    =  graph.num_vertices

         __UpperCamelCase	:	str		    =  Graph.UnionFind()
         __UpperCamelCase	:	Tuple		    =  []
         while num_components > 1:
             __UpperCamelCase	:	Dict		    =  {}
             for vertex in graph.get_vertices():
                 __UpperCamelCase	:	str		    =  -1

             __UpperCamelCase	:	int		    =  graph.get_edges()
             for edge in edges:
                 __UpperCamelCase    ,					__UpperCamelCase    ,					__UpperCamelCase	:	List[str]		    =  edge
                 edges.remove((tail, head, weight)    )
             for edge in edges:
                 __UpperCamelCase    ,					__UpperCamelCase    ,					__UpperCamelCase	:	Dict		    =  edge
                 __UpperCamelCase	:	List[Any]		    =  union_find.find(_lowerCamelCase    )
                 __UpperCamelCase	:	str		    =  union_find.find(_lowerCamelCase    )
                 if seta != seta:
                     if cheap_edge[seta] == -1 or cheap_edge[seta][2] > weight:
                         __UpperCamelCase	:	str		    =  [head, tail, weight]

                     if cheap_edge[seta] == -1 or cheap_edge[seta][2] > weight:
                         __UpperCamelCase	:	str		    =  [head, tail, weight]
             for vertex in cheap_edge:
                 if cheap_edge[vertex] != -1:
                     __UpperCamelCase    ,					__UpperCamelCase    ,					__UpperCamelCase	:	List[str]		    =  cheap_edge[vertex]
                     if union_find.find(_lowerCamelCase    ) != union_find.find(_lowerCamelCase    ):
                         union_find.union(_lowerCamelCase     ,			_lowerCamelCase    )
                         mst_edges.append(cheap_edge[vertex]    )
                         __UpperCamelCase	:	int		    =  num_components - 1
         __UpperCamelCase	:	Union[str, Any]		    =  Graph.build(edges=_lowerCamelCase    )
         return mst



