


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        lowerCAmelCase_			(							lowercase_      ,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 			_snake_case       (						self    : List[str]    )   ->Dict:






									"""simple docstring"""






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




						@property
						def 			_snake_case       (						self    : Tuple    )   ->Optional[Any]:






									"""simple docstring"""






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




						def 			_snake_case       (						self    : Optional[Any] ,  __A    : Optional[int]=False    )   ->Tuple:






									"""simple docstring"""






									if class_cond:
												a__			:int    	=     self.dummy_cond_unet
									else:
												a__			:Union[str, Any]    	=     self.dummy_uncond_unet

									# Default to CM multistep sampler
									a__			:Optional[int]    	=     CMStochasticIterativeScheduler(
									    num_train_timesteps=40 ,  sigma_min=0.002 ,  sigma_max=80.0 ,  )

									a__			:List[Any]    	=     {
									    "unet": unet,
									    "scheduler": scheduler,
									}

									return components




						def 			_snake_case       (						self    : Any ,  __A    : Union[str, Any] ,  __A    : Dict=0    )   ->str:






									"""simple docstring"""






									if str(UpperCamelCase_    ).startswith("mps"    ):
												a__			:str    	=     torch.manual_seed(UpperCamelCase_    )
									else:
												a__			:Tuple    	=     torch.Generator(device=UpperCamelCase_    ).manual_seed(UpperCamelCase_    )

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

									return inputs




						def 			_snake_case       (						self    : Tuple    )   ->Tuple:






									"""simple docstring"""






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

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

									a__			:Optional[int]    	=     image[0, -3:, -3:, -1]
									a__			:Optional[int]    	=     np.array([0.3_572, 0.6_273, 0.4_031, 0.3_961, 0.4_321, 0.5_730, 0.5_266, 0.4_780, 0.5_004]    )

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




						def 			_snake_case       (						self    : List[str]    )   ->Union[str, Any]:






									"""simple docstring"""






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

									a__			:Optional[int]    	=     self.get_dummy_inputs(UpperCamelCase_    )
									a__			:Union[str, Any]    	=     0
									a__			:Union[str, Any]    	=     pipe(**UpperCamelCase_    ).images
									assert image.shape == (1, 32, 32, 3)

									a__			:Optional[Any]    	=     image[0, -3:, -3:, -1]
									a__			:Any    	=     np.array([0.3_572, 0.6_273, 0.4_031, 0.3_961, 0.4_321, 0.5_730, 0.5_266, 0.4_780, 0.5_004]    )

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




						def 			_snake_case       (						self    : int    )   ->List[Any]:






									"""simple docstring"""






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

									a__			:List[Any]    	=     self.get_dummy_inputs(UpperCamelCase_    )
									a__			:Tuple    	=     1
									a__			:Optional[Any]    	=     None
									a__			:Optional[Any]    	=     pipe(**UpperCamelCase_    ).images
									assert image.shape == (1, 32, 32, 3)

									a__			:Optional[int]    	=     image[0, -3:, -3:, -1]
									a__			:List[str]    	=     np.array([0.5_004, 0.5_004, 0.4_994, 0.5_008, 0.4_976, 0.5_018, 0.4_990, 0.4_982, 0.4_987]    )

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






						def 			_snake_case       (						self    : str    )   ->List[str]:






									"""simple docstring"""






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

									a__			:str    	=     self.get_dummy_inputs(UpperCamelCase_    )
									a__			:int    	=     1
									a__			:Tuple    	=     None
									a__			:Any    	=     0
									a__			:Tuple    	=     pipe(**UpperCamelCase_    ).images
									assert image.shape == (1, 32, 32, 3)

									a__			:List[str]    	=     image[0, -3:, -3:, -1]
									a__			:List[Any]    	=     np.array([0.5_004, 0.5_004, 0.4_994, 0.5_008, 0.4_976, 0.5_018, 0.4_990, 0.4_982, 0.4_987]    )

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






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




						def 			_snake_case       (						self    : int    )   ->str:






									"""simple docstring"""






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




						def 			_snake_case       (						self    : Dict ,  __A    : List[Any]=0 ,  __A    : Any=False ,  __A    : Tuple="cpu" ,  __A    : Dict=torch.floataa ,  __A    : Dict=(1, 3, 64, 64)    )   ->List[str]:






									"""simple docstring"""






									a__			:Any    	=     torch.manual_seed(UpperCamelCase_    )

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

									if get_fixed_latents:
												a__			:List[Any]    	=     self.get_fixed_latents(seed=UpperCamelCase_ ,  device=UpperCamelCase_ ,  dtype=UpperCamelCase_ ,  shape=UpperCamelCase_    )
												a__			:int    	=     latents

									return inputs




						def 			_snake_case       (						self    : Tuple ,  __A    : List[str]=0 ,  __A    : Any="cpu" ,  __A    : Tuple=torch.floataa ,  __A    : str=(1, 3, 64, 64)    )   ->Tuple:






									"""simple docstring"""






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




						def 			_snake_case       (						self    : List[Any]    )   ->Tuple:






									"""simple docstring"""






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

									a__			:Union[str, Any]    	=     self.get_inputs()
									a__			:Optional[Any]    	=     pipe(**UpperCamelCase_    ).images
									assert image.shape == (1, 64, 64, 3)

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

									a__			:int    	=     np.array([0.0_888, 0.0_881, 0.0_666, 0.0_479, 0.0_292, 0.0_195, 0.0_201, 0.0_163, 0.0_254]    )

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




						def 			_snake_case       (						self    : Union[str, Any]    )   ->int:






									"""simple docstring"""






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

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

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

									a__			:Tuple    	=     np.array([0.0_340, 0.0_152, 0.0_063, 0.0_267, 0.0_221, 0.0_107, 0.0_416, 0.0_186, 0.0_217]    )

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




						@require_torch_a
						def 			_snake_case       (						self    : List[str]    )   ->Any:






									"""simple docstring"""






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

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

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

									a__			:List[str]    	=     np.array([0.1_875, 0.1_428, 0.1_289, 0.2_151, 0.2_092, 0.1_477, 0.1_877, 0.1_641, 0.1_353]    )

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






						@require_torch_a
						def 			_snake_case       (						self    : Any    )   ->Optional[Any]:






									"""simple docstring"""






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

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

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

									a__			:List[Any]    	=     np.array([0.1_663, 0.1_948, 0.2_275, 0.1_680, 0.1_204, 0.1_245, 0.1_858, 0.1_338, 0.2_095]    )

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










from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_sentencepiece_available,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


__a   :							Optional[int]						= {"configuration_xlnet": ["XLNET_PRETRAINED_CONFIG_ARCHIVE_MAP", "XLNetConfig"]}

try:
  if not is_sentencepiece_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  __a   :							Tuple						= ["XLNetTokenizer"]

try:
  if not is_tokenizers_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  __a   :							List[str]						= ["XLNetTokenizerFast"]

try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  __a   :							Any						= [
      "XLNET_PRETRAINED_MODEL_ARCHIVE_LIST",
      "XLNetForMultipleChoice",
      "XLNetForQuestionAnswering",
      "XLNetForQuestionAnsweringSimple",
      "XLNetForSequenceClassification",
      "XLNetForTokenClassification",
      "XLNetLMHeadModel",
      "XLNetModel",
      "XLNetPreTrainedModel",
      "load_tf_weights_in_xlnet",
  ]

try:
  if not is_tf_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  __a   :							Dict						= [
      "TF_XLNET_PRETRAINED_MODEL_ARCHIVE_LIST",
      "TFXLNetForMultipleChoice",
      "TFXLNetForQuestionAnsweringSimple",
      "TFXLNetForSequenceClassification",
      "TFXLNetForTokenClassification",
      "TFXLNetLMHeadModel",
      "TFXLNetMainLayer",
      "TFXLNetModel",
      "TFXLNetPreTrainedModel",
  ]


if TYPE_CHECKING:
  from .configuration_xlnet import XLNET_PRETRAINED_CONFIG_ARCHIVE_MAP, XLNetConfig

  try:
    if not is_sentencepiece_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .tokenization_xlnet import XLNetTokenizer

  try:
    if not is_tokenizers_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .tokenization_xlnet_fast import XLNetTokenizerFast

  try:
    if not is_torch_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_xlnet import (
        XLNET_PRETRAINED_MODEL_ARCHIVE_LIST,
        XLNetForMultipleChoice,
        XLNetForQuestionAnswering,
        XLNetForQuestionAnsweringSimple,
        XLNetForSequenceClassification,
        XLNetForTokenClassification,
        XLNetLMHeadModel,
        XLNetModel,
        XLNetPreTrainedModel,
        load_tf_weights_in_xlnet,
    )

  try:
    if not is_tf_available():
      raise OptionalDependencyNotAvailable()
  except OptionalDependencyNotAvailable:
    pass
  else:
    from .modeling_tf_xlnet import (
        TF_XLNET_PRETRAINED_MODEL_ARCHIVE_LIST,
        TFXLNetForMultipleChoice,
        TFXLNetForQuestionAnsweringSimple,
        TFXLNetForSequenceClassification,
        TFXLNetForTokenClassification,
        TFXLNetLMHeadModel,
        TFXLNetMainLayer,
        TFXLNetModel,
        TFXLNetPreTrainedModel,
    )

else:
  import sys

  __a   :							str						= _LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)




