"""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    __snake_case       (     __lowerCAmelCase       ,      unittest.TestCase				):
      a__        =		ConsistencyModelPipeline
      a__        =		UNCONDITIONAL_IMAGE_GENERATION_PARAMS
      a__        =		UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

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



      @property
      def   lowerCamelCase_   (						self)	->			List[str]:





        '''simple docstring'''





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



      @property
      def   lowerCamelCase_   (						self)	->			List[Any]:





        '''simple docstring'''





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



      def   lowerCamelCase_   (						self    ,			lowercase=False)	->			Optional[Any]:





        '''simple docstring'''





        if class_cond:
          a__:     Optional[Any]										=	self.dummy_cond_unet
        else:
          a__:     int										=	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[str]										=	{
            'unet': unet,
            'scheduler': scheduler,
        }

        return components



      def   lowerCamelCase_   (						self    ,			lowercase    ,			lowercase=0)	->			Tuple:





        '''simple docstring'''





        if str(lowercase).startswith('mps'):
          a__:     Optional[Any]										=	torch.manual_seed(lowercase)
        else:
          a__:     List[Any]										=	torch.Generator(device=lowercase).manual_seed(lowercase)

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

        return inputs



      def   lowerCamelCase_   (						self)	->			Optional[Any]:





        '''simple docstring'''





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

        a__:     str										=	self.get_dummy_inputs(lowercase)
        a__:     Any										=	pipe(**lowercase).images
        assert image.shape == (1, 32, 32, 3)

        a__:     List[Any]										=	image[0, -3:, -3:, -1]
        a__:     Union[str, Any]										=	np.array([0.3572, 0.6273, 0.4031, 0.3961, 0.4321, 0.5730, 0.5266, 0.4780, 0.5004])

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



      def   lowerCamelCase_   (						self)	->			str:





        '''simple docstring'''





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

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

        a__:     Optional[int]										=	image[0, -3:, -3:, -1]
        a__:     List[Any]										=	np.array([0.3572, 0.6273, 0.4031, 0.3961, 0.4321, 0.5730, 0.5266, 0.4780, 0.5004])

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



      def   lowerCamelCase_   (						self)	->			int:





        '''simple docstring'''





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

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

        a__:     List[Any]										=	image[0, -3:, -3:, -1]
        a__:     Dict										=	np.array([0.5004, 0.5004, 0.4994, 0.5008, 0.4976, 0.5018, 0.4990, 0.4982, 0.4987])

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



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





        '''simple docstring'''





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

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

        a__:     List[Any]										=	image[0, -3:, -3:, -1]
        a__:     Dict										=	np.array([0.5004, 0.5004, 0.4994, 0.5008, 0.4976, 0.5018, 0.4990, 0.4982, 0.4987])

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







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



      def   lowerCamelCase_   (						self)	->			str:





        '''simple docstring'''





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



      def   lowerCamelCase_   (						self    ,			lowercase=0    ,			lowercase=False    ,			lowercase="cpu"    ,			lowercase=torch.floataa    ,			lowercase=(1, 3, 64, 64))	->			Dict:





        '''simple docstring'''





        a__:     Dict										=	torch.manual_seed(lowercase)

        a__:     Union[str, Any]										=	{
            '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=lowercase    ,			device=lowercase    ,			dtype=lowercase    ,			shape=lowercase)
          a__:     Any										=	latents

        return inputs



      def   lowerCamelCase_   (						self    ,			lowercase=0    ,			lowercase="cpu"    ,			lowercase=torch.floataa    ,			lowercase=(1, 3, 64, 64))	->			Union[str, Any]:





        '''simple docstring'''





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



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





        '''simple docstring'''





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

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

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

        a__:     List[str]										=	np.array([0.0888, 0.0881, 0.0666, 0.0479, 0.0292, 0.0195, 0.0201, 0.0163, 0.0254])

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



      def   lowerCamelCase_   (						self)	->			int:





        '''simple docstring'''





        a__:     Optional[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__:     List[Any]										=	ConsistencyModelPipeline(unet=lowercase    ,			scheduler=lowercase)
        pipe.to(torch_device=lowercase)
        pipe.set_progress_bar_config(disable=lowercase)

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

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

        a__:     Optional[int]										=	np.array([0.0340, 0.0152, 0.0063, 0.0267, 0.0221, 0.0107, 0.0416, 0.0186, 0.0217])

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



      @require_torch_a
      def   lowerCamelCase_   (						self)	->			Any:





        '''simple docstring'''





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

        a__:     int										=	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):
          a__:     Tuple										=	pipe(**lowercase).images
        assert image.shape == (1, 64, 64, 3)

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

        a__:     Dict										=	np.array([0.1875, 0.1428, 0.1289, 0.2151, 0.2092, 0.1477, 0.1877, 0.1641, 0.1353])

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



      @require_torch_a
      def   lowerCamelCase_   (						self)	->			int:





        '''simple docstring'''





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

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

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

        a__:     List[Any]										=	np.array([0.1663, 0.1948, 0.2275, 0.1680, 0.1204, 0.1245, 0.1858, 0.1338, 0.2095])

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



"""simple docstring"""
def 						__a   (	_SCREAMING_SNAKE_CASE		,     _SCREAMING_SNAKE_CASE   )	->int:
  while a != 0:
    a__							,    a__:     List[str]										=	b % a, a
  return b







def 						__a   (	_SCREAMING_SNAKE_CASE		,     _SCREAMING_SNAKE_CASE   )	->int:
  if gcd(_SCREAMING_SNAKE_CASE		,     _SCREAMING_SNAKE_CASE   ) != 1:
    a__:     Dict										=	F'mod inverse of {a!r} and {m!r} does not exist'
    raise ValueError(_SCREAMING_SNAKE_CASE   )
  a__							,    a__							,    a__:     Union[str, Any]										=	1, 0, a
  a__							,    a__							,    a__:     Any										=	0, 1, m
  while va != 0:
    a__:     int										=	ua // va
    a__							,    a__							,    a__							,    a__							,    a__							,    a__:     Any										=	(ua - q * va), (ua - q * va), (ua - q * va), va, va, va
  return ua % m


