'''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 ):
      '''simple docstring'''


      UpperCAmelCase_   :   Optional[int]   				= ConsistencyModelPipeline
      UpperCAmelCase_   :   Tuple   				= UNCONDITIONAL_IMAGE_GENERATION_PARAMS
      UpperCAmelCase_   :   Optional[Any]   				= UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS

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




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



         '''simple docstring'''



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




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



         '''simple docstring'''



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




      def 		lowerCAmelCase__       (					self       ,lowerCamelCase_=False				)      ->      str:



         '''simple docstring'''



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

         # Default to CM multistep sampler
         UpperCAmelCase__				:					Tuple 					=	CMStochasticIterativeScheduler(
             num_train_timesteps=40       ,sigma_min=0.002       ,sigma_max=80.0       ,)

         UpperCAmelCase__				:					List[Any] 					=	{
             '''unet''': unet,
             '''scheduler''': scheduler,
         }

         return components




      def 		lowerCAmelCase__       (					self       ,lowerCamelCase_       ,lowerCamelCase_=0				)      ->      Any:



         '''simple docstring'''



         if str(lowerCamelCase_				).startswith('''mps'''				):
            UpperCAmelCase__				:					str 					=	torch.manual_seed(lowerCamelCase_				)
         else:
            UpperCAmelCase__				:					List[Any] 					=	torch.Generator(device=lowerCamelCase_				).manual_seed(lowerCamelCase_				)

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

         return inputs




      def 		lowerCAmelCase__       (					self				)      ->      Any:



         '''simple docstring'''



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

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

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



         '''simple docstring'''



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

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

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



         '''simple docstring'''



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

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

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



         '''simple docstring'''



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

         UpperCAmelCase__				:					Union[str, Any] 					=	self.get_dummy_inputs(lowerCamelCase_				)
         UpperCAmelCase__				:					List[Any] 					=	1
         UpperCAmelCase__				:					Dict 					=	None
         UpperCAmelCase__				:					int 					=	0
         UpperCAmelCase__				:					Union[str, Any] 					=	pipe(**lowerCamelCase_				).images
         assert image.shape == (1, 32, 32, 3)

         UpperCAmelCase__				:					str 					=	image[0, -3:, -3:, -1]
         UpperCAmelCase__				:					List[str] 					=	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 					_lowercase						(   unittest.TestCase ):
      '''simple docstring'''


      def 		lowerCAmelCase__       (					self				)      ->      List[Any]:



         '''simple docstring'''



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




      def 		lowerCAmelCase__       (					self       ,lowerCamelCase_=0       ,lowerCamelCase_=False       ,lowerCamelCase_="cpu"       ,lowerCamelCase_=torch.floataa       ,lowerCamelCase_=(1, 3, 64, 64)				)      ->      List[str]:



         '''simple docstring'''



         UpperCAmelCase__				:					Union[str, Any] 					=	torch.manual_seed(lowerCamelCase_				)

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

         if get_fixed_latents:
            UpperCAmelCase__				:					Tuple 					=	self.get_fixed_latents(seed=lowerCamelCase_       ,device=lowerCamelCase_       ,dtype=lowerCamelCase_       ,shape=lowerCamelCase_				)
            UpperCAmelCase__				:					Optional[int] 					=	latents

         return inputs




      def 		lowerCAmelCase__       (					self       ,lowerCamelCase_=0       ,lowerCamelCase_="cpu"       ,lowerCamelCase_=torch.floataa       ,lowerCamelCase_=(1, 3, 64, 64)				)      ->      Any:



         '''simple docstring'''



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




      def 		lowerCAmelCase__       (					self				)      ->      Dict:



         '''simple docstring'''



         UpperCAmelCase__				:					Any 					=	UNetaDModel.from_pretrained('''diffusers/consistency_models'''       ,subfolder='''diffusers_cd_imagenet64_l2'''				)
         UpperCAmelCase__				:					str 					=	CMStochasticIterativeScheduler(
             num_train_timesteps=40       ,sigma_min=0.002       ,sigma_max=80.0       ,)
         UpperCAmelCase__				:					str 					=	ConsistencyModelPipeline(unet=lowerCamelCase_       ,scheduler=lowerCamelCase_				)
         pipe.to(torch_device=lowerCamelCase_				)
         pipe.set_progress_bar_config(disable=lowerCamelCase_				)

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

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

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



         '''simple docstring'''



         UpperCAmelCase__				:					Optional[int] 					=	UNetaDModel.from_pretrained('''diffusers/consistency_models'''       ,subfolder='''diffusers_cd_imagenet64_l2'''				)
         UpperCAmelCase__				:					Tuple 					=	CMStochasticIterativeScheduler(
             num_train_timesteps=40       ,sigma_min=0.002       ,sigma_max=80.0       ,)
         UpperCAmelCase__				:					Dict 					=	ConsistencyModelPipeline(unet=lowerCamelCase_       ,scheduler=lowerCamelCase_				)
         pipe.to(torch_device=lowerCamelCase_				)
         pipe.set_progress_bar_config(disable=lowerCamelCase_				)

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

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

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



         '''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.002       ,sigma_max=80.0       ,)
         UpperCAmelCase__				:					Tuple 					=	ConsistencyModelPipeline(unet=lowerCamelCase_       ,scheduler=lowerCamelCase_				)
         pipe.to(torch_device=lowerCamelCase_       ,torch_dtype=torch.floataa				)
         pipe.set_progress_bar_config(disable=lowerCamelCase_				)

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

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

         UpperCAmelCase__				:					Optional[Any] 					=	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				)      ->      List[Any]:



         '''simple docstring'''



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

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

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

         UpperCAmelCase__				:					List[str] 					=	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'''

import os
from datetime import datetime as dt

from github import Github


UpperCamelCase__  :						str					=      [
    'good first issue',
    'good second issue',
    'good difficult issue',
    'enhancement',
    'new pipeline/model',
    'new scheduler',
    'wip',
]







def __UpperCamelCase( ):
   '''simple docstring'''
   UpperCAmelCase__				:					str 					=	Github(os.environ['''GITHUB_TOKEN''']     )
   UpperCAmelCase__				:					Union[str, Any] 					=	g.get_repo('''huggingface/diffusers'''     )
   UpperCAmelCase__				:					Union[str, Any] 					=	repo.get_issues(state='''open'''     )

   for issue in open_issues:
      UpperCAmelCase__				:					int 					=	sorted(issue.get_comments()	,      key=lambda _A     : i.created_at	,      reverse=_A     )
      UpperCAmelCase__				:					int 					=	comments[0] if len(_A     ) > 0 else None
      if (
          last_comment is not None
          and last_comment.user.login == "github-actions[bot]"
          and (dt.utcnow() - issue.updated_at).days > 7
          and (dt.utcnow() - issue.created_at).days >= 30
          and not any(label.name.lower() in LABELS_TO_EXEMPT for label in issue.get_labels()     )
      ):
         # Closes the issue after 7 days of inactivity since the Stalebot notification.
         issue.edit(state='''closed'''     )
      elif (
          "stale" in issue.get_labels()
          and last_comment is not None
          and last_comment.user.login != "github-actions[bot]"
      ):
         # Opens the issue if someone other than Stalebot commented.
         issue.edit(state='''open'''     )
         issue.remove_from_labels('''stale'''     )
      elif (
          (dt.utcnow() - issue.updated_at).days > 23
          and (dt.utcnow() - issue.created_at).days >= 30
          and not any(label.name.lower() in LABELS_TO_EXEMPT for label in issue.get_labels()     )
      ):
         # Post a Stalebot notification after 23 days of inactivity.
         issue.create_comment(
             '''This issue has been automatically marked as stale because it has not had '''
             '''recent activity. If you think this still needs to be addressed '''
             '''please comment on this thread.\n\nPlease note that issues that do not follow the '''
             '''[contributing guidelines](https://github.com/huggingface/diffusers/blob/main/CONTRIBUTING.md) '''
             '''are likely to be ignored.'''     )
         issue.add_to_labels('''stale'''     )


if __name__ == "__main__":
  main()




