




'''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        lowercase_      (						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     :					str			):
           snake_case__       :   Optional[int]	      =    UNetaDModel.from_pretrained(
               'diffusers/consistency-models-test'     ,							subfolder='test_unet'     ,							)
           return unet





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





     def 			_lowerCAmelCase      (     self     :					str     ,							__lowerCamelCase     :					Optional[Any]=False			):
           if class_cond:
                 snake_case__       :   Tuple	      =    self.dummy_cond_unet
           else:
                 snake_case__       :   Optional[Any]	      =    self.dummy_uncond_unet

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

           snake_case__       :   Optional[Any]	      =    {
               'unet': unet,
               'scheduler': scheduler,
           }

           return components





     def 			_lowerCAmelCase      (     self     :					Optional[int]     ,							__lowerCamelCase     :					str     ,							__lowerCamelCase     :					List[Any]=0			):
           if str(__lowerCamelCase			).startswith('mps'			):
                 snake_case__       :   str	      =    torch.manual_seed(__lowerCamelCase			)
           else:
                 snake_case__       :   Tuple	      =    torch.Generator(device=__lowerCamelCase			).manual_seed(__lowerCamelCase			)

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

           return inputs





     def 			_lowerCAmelCase      (     self     :					int			):
           snake_case__       :   str	      =    'cpu'  # ensure determinism for the device-dependent torch.Generator
           snake_case__       :   Dict	      =    self.get_dummy_components()
           snake_case__       :   List[str]	      =    ConsistencyModelPipeline(**__lowerCamelCase			)
           snake_case__       :   Tuple	      =    pipe.to(__lowerCamelCase			)
           pipe.set_progress_bar_config(disable=__lowerCamelCase			)

           snake_case__       :   Any	      =    self.get_dummy_inputs(__lowerCamelCase			)
           snake_case__       :   List[str]	      =    pipe(**__lowerCamelCase			).images
           assert image.shape == (1, 32, 32, 3)

           snake_case__       :   int	      =    image[0, -3:, -3:, -1]
           snake_case__       :   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 			_lowerCAmelCase      (     self     :					Tuple			):
           snake_case__       :   Dict	      =    'cpu'  # ensure determinism for the device-dependent torch.Generator
           snake_case__       :   List[Any]	      =    self.get_dummy_components(class_cond=__lowerCamelCase			)
           snake_case__       :   Tuple	      =    ConsistencyModelPipeline(**__lowerCamelCase			)
           snake_case__       :   Union[str, Any]	      =    pipe.to(__lowerCamelCase			)
           pipe.set_progress_bar_config(disable=__lowerCamelCase			)

           snake_case__       :   List[str]	      =    self.get_dummy_inputs(__lowerCamelCase			)
           snake_case__       :   Union[str, Any]	      =    0
           snake_case__       :   List[Any]	      =    pipe(**__lowerCamelCase			).images
           assert image.shape == (1, 32, 32, 3)

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

           snake_case__       :   Dict	      =    self.get_dummy_inputs(__lowerCamelCase			)
           snake_case__       :   Dict	      =    1
           snake_case__       :   int	      =    None
           snake_case__       :   int	      =    pipe(**__lowerCamelCase			).images
           assert image.shape == (1, 32, 32, 3)

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

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

           snake_case__       :   int	      =    image[0, -3:, -3:, -1]
           snake_case__       :   List[str]	      =    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        lowercase_      (						unittest.TestCase							):





     def 			_lowerCAmelCase      (     self     :					Tuple			):
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()





     def 			_lowerCAmelCase      (     self     :					Tuple     ,							__lowerCamelCase     :					List[str]=0     ,							__lowerCamelCase     :					str=False     ,							__lowerCamelCase     :					List[str]="cpu"     ,							__lowerCamelCase     :					List[Any]=torch.floataa     ,							__lowerCamelCase     :					Union[str, Any]=(1, 3, 64, 64)			):
           snake_case__       :   Any	      =    torch.manual_seed(__lowerCamelCase			)

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

           if get_fixed_latents:
                 snake_case__       :   Optional[int]	      =    self.get_fixed_latents(seed=__lowerCamelCase     ,							device=__lowerCamelCase     ,							dtype=__lowerCamelCase     ,							shape=__lowerCamelCase			)
                 snake_case__       :   List[str]	      =    latents

           return inputs





     def 			_lowerCAmelCase      (     self     :					Optional[Any]     ,							__lowerCamelCase     :					str=0     ,							__lowerCamelCase     :					Dict="cpu"     ,							__lowerCamelCase     :					Union[str, Any]=torch.floataa     ,							__lowerCamelCase     :					int=(1, 3, 64, 64)			):
           if type(__lowerCamelCase			) == str:
                 snake_case__       :   Any	      =    torch.device(__lowerCamelCase			)
           snake_case__       :   Any	      =    torch.Generator(device=__lowerCamelCase			).manual_seed(__lowerCamelCase			)
           snake_case__       :   Union[str, Any]	      =    randn_tensor(__lowerCamelCase     ,							generator=__lowerCamelCase     ,							device=__lowerCamelCase     ,							dtype=__lowerCamelCase			)
           return latents





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

           snake_case__       :   Dict	      =    self.get_inputs()
           snake_case__       :   Optional[int]	      =    pipe(**__lowerCamelCase			).images
           assert image.shape == (1, 64, 64, 3)

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

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

           snake_case__       :   Optional[int]	      =    self.get_inputs()
           snake_case__       :   Any	      =    1
           snake_case__       :   List[Any]	      =    None
           snake_case__       :   Tuple	      =    pipe(**__lowerCamelCase			).images
           assert image.shape == (1, 64, 64, 3)

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

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

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

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

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

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

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

           snake_case__       :   Union[str, 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






'''simple docstring'''






from math import pi






def 		UpperCamelCase__      (		__SCREAMING_SNAKE_CASE				,      __SCREAMING_SNAKE_CASE		)    ->   float:
      return 2 * pi * radius * (angle / 360)


if __name__ == "__main__":
      print(arc_length(90, 10))
