





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   ):
   lowerCAmelCase__        = ConsistencyModelPipeline
   lowerCAmelCase__        = UNCONDITIONAL_IMAGE_GENERATION_PARAMS
   lowerCAmelCase__        = UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

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




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




   @property
   def _lowerCAmelCase      (self	:int )->					Optional[Any]:
         __A          =				UNetaDModel.from_pretrained(
             '''diffusers/consistency-models-test'''    ,   subfolder='''test_unet_class_cond'''    ,   )
         return unet




   def _lowerCAmelCase      (self	:List[Any]    ,   _UpperCamelCase	:List[str]=False )->					Dict:
         if class_cond:
               __A          =				self.dummy_cond_unet
         else:
               __A          =				self.dummy_uncond_unet

         # Default to CM multistep sampler
         __A          =				CMStochasticIterativeScheduler(
             num_train_timesteps=40    ,   sigma_min=0.0_0_2    ,   sigma_max=8_0.0    ,   )

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

         return components




   def _lowerCAmelCase      (self	:int    ,   _UpperCamelCase	:int    ,   _UpperCamelCase	:Optional[Any]=0 )->					Tuple:
         if str(_UpperCamelCase ).startswith('''mps''' ):
               __A          =				torch.manual_seed(_UpperCamelCase )
         else:
               __A          =				torch.Generator(device=_UpperCamelCase ).manual_seed(_UpperCamelCase )

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

         return inputs




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

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

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

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

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

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

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

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

         __A          =				image[0, -3:, -3:, -1]
         __A          =				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 A_					(			unittest.TestCase   ):




   def _lowerCAmelCase      (self	:Any )->					Any:
         super().tearDown()
         gc.collect()
         torch.cuda.empty_cache()




   def _lowerCAmelCase      (self	:Union[str, Any]    ,   _UpperCamelCase	:Dict=0    ,   _UpperCamelCase	:List[str]=False    ,   _UpperCamelCase	:List[str]="cpu"    ,   _UpperCamelCase	:Optional[Any]=torch.floataa    ,   _UpperCamelCase	:List[str]=(1, 3, 64, 64) )->					Union[str, Any]:
         __A          =				torch.manual_seed(_UpperCamelCase )

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

         if get_fixed_latents:
               __A          =				self.get_fixed_latents(seed=_UpperCamelCase    ,   device=_UpperCamelCase    ,   dtype=_UpperCamelCase    ,   shape=_UpperCamelCase )
               __A          =				latents

         return inputs




   def _lowerCAmelCase      (self	:int    ,   _UpperCamelCase	:Any=0    ,   _UpperCamelCase	:Optional[int]="cpu"    ,   _UpperCamelCase	:List[Any]=torch.floataa    ,   _UpperCamelCase	:Tuple=(1, 3, 64, 64) )->					int:
         if type(_UpperCamelCase ) == str:
               __A          =				torch.device(_UpperCamelCase )
         __A          =				torch.Generator(device=_UpperCamelCase ).manual_seed(_UpperCamelCase )
         __A          =				randn_tensor(_UpperCamelCase    ,   generator=_UpperCamelCase    ,   device=_UpperCamelCase    ,   dtype=_UpperCamelCase )
         return latents




   def _lowerCAmelCase      (self	:Any )->					Dict:
         __A          =				UNetaDModel.from_pretrained('''diffusers/consistency_models'''    ,   subfolder='''diffusers_cd_imagenet64_l2''' )
         __A          =				CMStochasticIterativeScheduler(
             num_train_timesteps=40    ,   sigma_min=0.0_0_2    ,   sigma_max=8_0.0    ,   )
         __A          =				ConsistencyModelPipeline(unet=_UpperCamelCase    ,   scheduler=_UpperCamelCase )
         pipe.to(torch_device=_UpperCamelCase )
         pipe.set_progress_bar_config(disable=_UpperCamelCase )

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

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

         __A          =				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	:Optional[Any] )->					List[str]:
         __A          =				UNetaDModel.from_pretrained('''diffusers/consistency_models'''    ,   subfolder='''diffusers_cd_imagenet64_l2''' )
         __A          =				CMStochasticIterativeScheduler(
             num_train_timesteps=40    ,   sigma_min=0.0_0_2    ,   sigma_max=8_0.0    ,   )
         __A          =				ConsistencyModelPipeline(unet=_UpperCamelCase    ,   scheduler=_UpperCamelCase )
         pipe.to(torch_device=_UpperCamelCase )
         pipe.set_progress_bar_config(disable=_UpperCamelCase )

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

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

         __A          =				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	:Tuple )->					Union[str, Any]:
         __A          =				UNetaDModel.from_pretrained('''diffusers/consistency_models'''    ,   subfolder='''diffusers_cd_imagenet64_l2''' )
         __A          =				CMStochasticIterativeScheduler(
             num_train_timesteps=40    ,   sigma_min=0.0_0_2    ,   sigma_max=8_0.0    ,   )
         __A          =				ConsistencyModelPipeline(unet=_UpperCamelCase    ,   scheduler=_UpperCamelCase )
         pipe.to(torch_device=_UpperCamelCase    ,   torch_dtype=torch.floataa )
         pipe.set_progress_bar_config(disable=_UpperCamelCase )

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

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

         __A          =				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	:Any )->					Optional[Any]:
         __A          =				UNetaDModel.from_pretrained('''diffusers/consistency_models'''    ,   subfolder='''diffusers_cd_imagenet64_l2''' )
         __A          =				CMStochasticIterativeScheduler(
             num_train_timesteps=40    ,   sigma_min=0.0_0_2    ,   sigma_max=8_0.0    ,   )
         __A          =				ConsistencyModelPipeline(unet=_UpperCamelCase    ,   scheduler=_UpperCamelCase )
         pipe.to(torch_device=_UpperCamelCase    ,   torch_dtype=torch.floataa )
         pipe.set_progress_bar_config(disable=_UpperCamelCase )

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

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

         __A          =				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












from __future__ import annotations

from collections.abc import Iterable, Iterator
from dataclasses import dataclass

snake_case__					:							Union[str, Any]													=     (3, 9, -11, 0, 7, 5, 1, -1)
snake_case__					:							int													=     (4, 6, 2, 0, 8, 10, 3, -2)





@dataclass
class A_					:
   lowerCAmelCase__        = 42
   lowerCAmelCase__        = 42





class A_					:




   def __init__(self	:Dict    ,   _UpperCamelCase	:Iterable[int] )->					None:
         __A          =				None
         for i in sorted(_UpperCamelCase    ,   reverse=_UpperCamelCase ):
               __A          =				Node(_UpperCamelCase    ,   self.head )




   def __iter__(self	:List[str] )->					Iterator[int]:
         __A          =				self.head
         while node:
               yield node.data
               __A          =				node.next_node




   def __len__(self	:Union[str, Any] )->					int:
         return sum(1 for _ in self )






   def __str__(self	:List[Any] )->					str:
         return " -> ".join([str(_UpperCamelCase ) for node in self] )





def _a				(	lowerCamelCase:   SortedLinkedList					,				lowerCamelCase:   SortedLinkedList )	->						SortedLinkedList:







      '''simple docstring'''






      return SortedLinkedList(list(lowerCamelCase ) + list(lowerCamelCase ) )


if __name__ == "__main__":
       import doctest

       doctest.testmod()
       snake_case__					:							Any													=     SortedLinkedList
       print(merge_lists(SSL(test_data_odd), SSL(test_data_even)))





