


'''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      lowerCamelCase_  (      __a	,			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 lowercase_				(   self							:       str					):


          '''simple docstring'''


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






      @property
      def lowercase_				(   self							:       List[str]					):


          '''simple docstring'''


          UpperCAmelCase__     :						List[Any]						=							UNetaDModel.from_pretrained(
              '''diffusers/consistency-models-test'''		,		subfolder='''test_unet_class_cond'''		,		)
          return unet






      def lowercase_				(   self							:       Optional[Any]		,		_A							:       int=False					):


          '''simple docstring'''


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

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

          UpperCAmelCase__     :						Any						=							{
              '''unet''': unet,
              '''scheduler''': scheduler,
          }

          return components






      def lowercase_				(   self							:       Tuple		,		_A							:       str		,		_A							:       List[Any]=0					):


          '''simple docstring'''


          if str(_A					).startswith('''mps'''					):
              UpperCAmelCase__     :						Union[str, Any]						=							torch.manual_seed(_A					)
          else:
              UpperCAmelCase__     :						int						=							torch.Generator(device=_A					).manual_seed(_A					)

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

          return inputs






      def lowercase_				(   self							:       Union[str, Any]					):


          '''simple docstring'''


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

          UpperCAmelCase__     :						Optional[int]						=							self.get_dummy_inputs(_A					)
          UpperCAmelCase__     :						int						=							pipe(**_A					).images
          assert image.shape == (1, 32, 32, 3)

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


          '''simple docstring'''


          UpperCAmelCase__     :						Any						=							'''cpu'''  # ensure determinism for the device-dependent torch.Generator
          UpperCAmelCase__     :						List[str]						=							self.get_dummy_components(class_cond=_A					)
          UpperCAmelCase__     :						List[Any]						=							ConsistencyModelPipeline(**_A					)
          UpperCAmelCase__     :						str						=							pipe.to(_A					)
          pipe.set_progress_bar_config(disable=_A					)

          UpperCAmelCase__     :						List[str]						=							self.get_dummy_inputs(_A					)
          UpperCAmelCase__     :						Tuple						=							0
          UpperCAmelCase__     :						Dict						=							pipe(**_A					).images
          assert image.shape == (1, 32, 32, 3)

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


          '''simple docstring'''


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

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

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


          '''simple docstring'''


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

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

          UpperCAmelCase__     :						Optional[Any]						=							image[0, -3:, -3:, -1]
          UpperCAmelCase__     :						Tuple						=							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      lowerCamelCase_  (      unittest.TestCase ):






      def lowercase_				(   self							:       Tuple					):


          '''simple docstring'''


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






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


          '''simple docstring'''


          UpperCAmelCase__     :						Tuple						=							torch.manual_seed(_A					)

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

          if get_fixed_latents:
              UpperCAmelCase__     :						Any						=							self.get_fixed_latents(seed=_A		,		device=_A		,		dtype=_A		,		shape=_A					)
              UpperCAmelCase__     :						Optional[Any]						=							latents

          return inputs






      def lowercase_				(   self							:       List[str]		,		_A							:       Tuple=0		,		_A							:       int="cpu"		,		_A							:       Optional[int]=torch.floataa		,		_A							:       List[str]=(1, 3, 64, 64)					):


          '''simple docstring'''


          if type(_A					) == str:
              UpperCAmelCase__     :						int						=							torch.device(_A					)
          UpperCAmelCase__     :						Union[str, Any]						=							torch.Generator(device=_A					).manual_seed(_A					)
          UpperCAmelCase__     :						List[str]						=							randn_tensor(_A		,		generator=_A		,		device=_A		,		dtype=_A					)
          return latents






      def lowercase_				(   self							:       str					):


          '''simple docstring'''


          UpperCAmelCase__     :						Any						=							UNetaDModel.from_pretrained('''diffusers/consistency_models'''		,		subfolder='''diffusers_cd_imagenet64_l2'''					)
          UpperCAmelCase__     :						Dict						=							CMStochasticIterativeScheduler(
              num_train_timesteps=40		,		sigma_min=0.0_0_2		,		sigma_max=8_0.0		,		)
          UpperCAmelCase__     :						Any						=							ConsistencyModelPipeline(unet=_A		,		scheduler=_A					)
          pipe.to(torch_device=_A					)
          pipe.set_progress_bar_config(disable=_A					)

          UpperCAmelCase__     :						str						=							self.get_inputs()
          UpperCAmelCase__     :						List[Any]						=							pipe(**_A					).images
          assert image.shape == (1, 64, 64, 3)

          UpperCAmelCase__     :						Optional[int]						=							image[0, -3:, -3:, -1]

          UpperCAmelCase__     :						Optional[int]						=							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							:       Union[str, Any]					):


          '''simple docstring'''


          UpperCAmelCase__     :						Optional[Any]						=							UNetaDModel.from_pretrained('''diffusers/consistency_models'''		,		subfolder='''diffusers_cd_imagenet64_l2'''					)
          UpperCAmelCase__     :						Tuple						=							CMStochasticIterativeScheduler(
              num_train_timesteps=40		,		sigma_min=0.0_0_2		,		sigma_max=8_0.0		,		)
          UpperCAmelCase__     :						Any						=							ConsistencyModelPipeline(unet=_A		,		scheduler=_A					)
          pipe.to(torch_device=_A					)
          pipe.set_progress_bar_config(disable=_A					)

          UpperCAmelCase__     :						str						=							self.get_inputs()
          UpperCAmelCase__     :						Optional[Any]						=							1
          UpperCAmelCase__     :						Tuple						=							None
          UpperCAmelCase__     :						str						=							pipe(**_A					).images
          assert image.shape == (1, 64, 64, 3)

          UpperCAmelCase__     :						Union[str, Any]						=							image[0, -3:, -3:, -1]

          UpperCAmelCase__     :						Union[str, 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							:       Union[str, Any]					):


          '''simple docstring'''


          UpperCAmelCase__     :						Dict						=							UNetaDModel.from_pretrained('''diffusers/consistency_models'''		,		subfolder='''diffusers_cd_imagenet64_l2'''					)
          UpperCAmelCase__     :						Optional[int]						=							CMStochasticIterativeScheduler(
              num_train_timesteps=40		,		sigma_min=0.0_0_2		,		sigma_max=8_0.0		,		)
          UpperCAmelCase__     :						List[Any]						=							ConsistencyModelPipeline(unet=_A		,		scheduler=_A					)
          pipe.to(torch_device=_A		,		torch_dtype=torch.floataa					)
          pipe.set_progress_bar_config(disable=_A					)

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

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

          UpperCAmelCase__     :						str						=							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							:       Any					):


          '''simple docstring'''


          UpperCAmelCase__     :						str						=							UNetaDModel.from_pretrained('''diffusers/consistency_models'''		,		subfolder='''diffusers_cd_imagenet64_l2'''					)
          UpperCAmelCase__     :						Tuple						=							CMStochasticIterativeScheduler(
              num_train_timesteps=40		,		sigma_min=0.0_0_2		,		sigma_max=8_0.0		,		)
          UpperCAmelCase__     :						int						=							ConsistencyModelPipeline(unet=_A		,		scheduler=_A					)
          pipe.to(torch_device=_A		,		torch_dtype=torch.floataa					)
          pipe.set_progress_bar_config(disable=_A					)

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

          UpperCAmelCase__     :						Optional[int]						=							image[0, -3:, -3:, -1]

          UpperCAmelCase__     :						Dict						=							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'''
from ...configuration_utils import PretrainedConfig
from ...utils import logging


UpperCAmelCase_			:  Optional[int]										=   logging.get_logger(__name__)

UpperCAmelCase_			:  List[str]										=   {
    'google/fnet-base': 'https://huggingface.co/google/fnet-base/resolve/main/config.json',
    'google/fnet-large': 'https://huggingface.co/google/fnet-large/resolve/main/config.json'
    # See all FNet models at https://huggingface.co/models?filter=fnet
}

class     _lowerCamelCase       (     snake_case_					):




 '''simple docstring'''
 __lowercase					:		List[Any]     				=    '''fnet'''


 def __init__(   self   ,   __lowercase=32_000   ,   __lowercase=768   ,   __lowercase=12   ,   __lowercase=3_072   ,   __lowercase="gelu_new"   ,   __lowercase=0.1   ,   __lowercase=512   ,   __lowercase=4   ,   __lowercase=0.0_2   ,   __lowercase=1E-12   ,   __lowercase=False   ,   __lowercase=512   ,   __lowercase=3   ,   __lowercase=1   ,   __lowercase=2   ,   **__lowercase   ,   ):

  """simple docstring"""







  super().__init__(pad_token_id=__lowercase   ,   bos_token_id=__lowercase   ,   eos_token_id=__lowercase   ,   **__lowercase    )

  __A	:		Union[str, Any]	  =      vocab_size
  __A	:		Dict	  =      max_position_embeddings
  __A	:		List[str]	  =      hidden_size
  __A	:		Tuple	  =      num_hidden_layers
  __A	:		Optional[int]	  =      intermediate_size
  __A	:		Dict	  =      hidden_act
  __A	:		List[str]	  =      hidden_dropout_prob
  __A	:		str	  =      initializer_range
  __A	:		Dict	  =      type_vocab_size
  __A	:		int	  =      layer_norm_eps
  __A	:		Tuple	  =      use_tpu_fourier_optimizations
  __A	:		Optional[Any]	  =      tpu_short_seq_length






