
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       _snake_case   (	UpperCAmelCase_  ,						unittest.TestCase							):
	__lowerCAmelCase  :     Optional[Any]        =			ConsistencyModelPipeline
	__lowerCAmelCase  :     List[Any]        =			UNCONDITIONAL_IMAGE_GENERATION_PARAMS
	__lowerCAmelCase  :     Any        =			UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

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


	@property
	def      lowercase__     (					self):


			'''simple docstring'''




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


	@property
	def      lowercase__     (					self):


			'''simple docstring'''




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


	def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_=False):


			'''simple docstring'''




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

			# Default to CM multistep sampler
			lowercase__  :   Dict    		=      CMStochasticIterativeScheduler(
			    num_train_timesteps=40					,  sigma_min=0.0_0_2					,  sigma_max=80.0					,  )

			lowercase__  :   Dict    		=      {
			    """unet""": unet,
			    """scheduler""": scheduler,
			}

			return components


	def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_=0):


			'''simple docstring'''




			if str(_lowercase).startswith("""mps"""):
					lowercase__  :   Union[str, Any]    		=      torch.manual_seed(_lowercase)
			else:
					lowercase__  :   Tuple    		=      torch.Generator(device=_lowercase).manual_seed(_lowercase)

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

			return inputs


	def      lowercase__     (					self):


			'''simple docstring'''




			lowercase__  :   str    		=      """cpu"""  # ensure determinism for the device-dependent torch.Generator
			lowercase__  :   Tuple    		=      self.get_dummy_components()
			lowercase__  :   str    		=      ConsistencyModelPipeline(**_lowercase)
			lowercase__  :   int    		=      pipe.to(_lowercase)
			pipe.set_progress_bar_config(disable=_lowercase)

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

			lowercase__  :   Union[str, Any]    		=      image[0, -3:, -3:, -1]
			lowercase__  :   str    		=      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      lowercase__     (					self):


			'''simple docstring'''




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

			lowercase__  :   Optional[int]    		=      self.get_dummy_inputs(_lowercase)
			lowercase__  :   Dict    		=      0
			lowercase__  :   Tuple    		=      pipe(**_lowercase).images
			assert image.shape == (1, 32, 32, 3)

			lowercase__  :   int    		=      image[0, -3:, -3:, -1]
			lowercase__  :   Tuple    		=      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      lowercase__     (					self):


			'''simple docstring'''




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

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

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


			'''simple docstring'''




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

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

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



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


	def      lowercase__     (					self):


			'''simple docstring'''




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


	def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_=0					,  SCREAMING_SNAKE_CASE_=False					,  SCREAMING_SNAKE_CASE_="cpu"					,  SCREAMING_SNAKE_CASE_=torch.floataa					,  SCREAMING_SNAKE_CASE_=(1, 3, 64, 64)):


			'''simple docstring'''




			lowercase__  :   Tuple    		=      torch.manual_seed(_lowercase)

			lowercase__  :   Union[str, Any]    		=      {
			    """num_inference_steps""": None,
			    """timesteps""": [22, 0],
			    """class_labels""": 0,
			    """generator""": generator,
			    """output_type""": """np""",
			}

			if get_fixed_latents:
					lowercase__  :   Tuple    		=      self.get_fixed_latents(seed=_lowercase					,  device=_lowercase					,  dtype=_lowercase					,  shape=_lowercase)
					lowercase__  :   int    		=      latents

			return inputs


	def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_=0					,  SCREAMING_SNAKE_CASE_="cpu"					,  SCREAMING_SNAKE_CASE_=torch.floataa					,  SCREAMING_SNAKE_CASE_=(1, 3, 64, 64)):


			'''simple docstring'''




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


	def      lowercase__     (					self):


			'''simple docstring'''




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

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

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

			lowercase__  :   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      lowercase__     (					self):


			'''simple docstring'''




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

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

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

			lowercase__  :   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      lowercase__     (					self):


			'''simple docstring'''




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

			lowercase__  :   List[str]    		=      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):
					lowercase__  :   int    		=      pipe(**_lowercase).images
			assert image.shape == (1, 64, 64, 3)

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

			lowercase__  :   Optional[int]    		=      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      lowercase__     (					self):


			'''simple docstring'''




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

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

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

			lowercase__  :   Any    		=      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 dataclasses import dataclass
from typing import Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, randn_tensor
from .scheduling_utils import SchedulerMixin



@dataclass
class       _snake_case   (	UpperCAmelCase_							):
 __lowerCAmelCase  :     torch.FloatTensor
 __lowerCAmelCase  :     torch.FloatTensor
 __lowerCAmelCase  :     Optional[torch.FloatTensor]        =			None



class       _snake_case   (	UpperCAmelCase_  ,						UpperCAmelCase_							):
 __lowerCAmelCase  :     Union[str, Any]        =			2


 @register_to_config
 def __init__(					self					,  SCREAMING_SNAKE_CASE_ = 0.0_2					,  SCREAMING_SNAKE_CASE_ = 1_00					,  SCREAMING_SNAKE_CASE_ = 1.0_0_7					,  SCREAMING_SNAKE_CASE_ = 80					,  SCREAMING_SNAKE_CASE_ = 0.0_5					,  SCREAMING_SNAKE_CASE_ = 50					,  ):


   '''simple docstring'''




   lowercase__  :   List[Any]    		=      sigma_max

   # setable values
   lowercase__  :   int    		=      None
   lowercase__  :   np.IntTensor    		=      None
   lowercase__  :   torch.FloatTensor    		=      None  # sigma(t_i)


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_ = None):


   '''simple docstring'''




   return sample


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_ = None):


   '''simple docstring'''




   lowercase__  :   Dict    		=      num_inference_steps
   lowercase__  :   List[Any]    		=      np.arange(0					,  self.num_inference_steps)[::-1].copy()
   lowercase__  :   Optional[Any]    		=      torch.from_numpy(SCREAMING_SNAKE_CASE_).to(SCREAMING_SNAKE_CASE_)
   lowercase__  :   str    		=      [
       (
           self.config.sigma_max**2
           * (self.config.sigma_min**2 / self.config.sigma_max**2) ** (i / (num_inference_steps - 1))
       )
       for i in self.timesteps
   ]
   lowercase__  :   int    		=      torch.tensor(SCREAMING_SNAKE_CASE_					,  dtype=torch.floataa					,  device=SCREAMING_SNAKE_CASE_)


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_ = None):


   '''simple docstring'''




   if self.config.s_min <= sigma <= self.config.s_max:
     lowercase__  :   Optional[Any]    		=      min(self.config.s_churn / self.num_inference_steps					,  2**0.5 - 1)
   else:
     lowercase__  :   Tuple    		=      0

   # sample eps ~ N(0, S_noise^2 * I)
   lowercase__  :   Dict    		=      self.config.s_noise * randn_tensor(sample.shape					,  generator=SCREAMING_SNAKE_CASE_).to(sample.device)
   lowercase__  :   int    		=      sigma + gamma * sigma
   lowercase__  :   Optional[int]    		=      sample + ((sigma_hat**2 - sigma**2) ** 0.5 * eps)

   return sample_hat, sigma_hat


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_ = True					,  ):


   '''simple docstring'''




   lowercase__  :   str    		=      sample_hat + sigma_hat * model_output
   lowercase__  :   str    		=      (sample_hat - pred_original_sample) / sigma_hat
   lowercase__  :   Optional[Any]    		=      sample_hat + (sigma_prev - sigma_hat) * derivative

   if not return_dict:
     return (sample_prev, derivative)

   return KarrasVeOutput(
       prev_sample=SCREAMING_SNAKE_CASE_					,  derivative=SCREAMING_SNAKE_CASE_					,  pred_original_sample=SCREAMING_SNAKE_CASE_)


 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_ = True					,  ):


   '''simple docstring'''




   lowercase__  :   str    		=      sample_prev + sigma_prev * model_output
   lowercase__  :   Dict    		=      (sample_prev - pred_original_sample) / sigma_prev
   lowercase__  :   Optional[int]    		=      sample_hat + (sigma_prev - sigma_hat) * (0.5 * derivative + 0.5 * derivative_corr)

   if not return_dict:
     return (sample_prev, derivative)

   return KarrasVeOutput(
       prev_sample=SCREAMING_SNAKE_CASE_					,  derivative=SCREAMING_SNAKE_CASE_					,  pred_original_sample=SCREAMING_SNAKE_CASE_)



 def      lowercase__     (					self					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_					,  SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   raise NotImplementedError()






