





"""simple docstring"""


import os
from datetime import datetime as dt

from github import Github


UpperCAmelCase__  		=						[
    """good first issue""",
    """feature request""",
    """wip""",
]






def     __UpperCAmelCase					(			):







  """simple docstring"""
  _UpperCAmelCase						 =      Github(os.environ["""GITHUB_TOKEN"""]		)
  _UpperCAmelCase						 =      g.get_repo("""huggingface/accelerate"""		)
  _UpperCAmelCase						 =      repo.get_issues(state="""open"""		)

  for issue in open_issues:
    _UpperCAmelCase						 =      sorted([comment for comment in issue.get_comments()]   ,key=lambda lowercase		: i.created_at   ,reverse=lowercase		)
    _UpperCAmelCase						 =      comments[0] if len(lowercase		) > 0 else None
    _UpperCAmelCase						 =      dt.utcnow()
    _UpperCAmelCase						 =      (current_time - issue.updated_at).days
    _UpperCAmelCase						 =      (current_time - issue.created_at).days
    if (
        last_comment is not None
        and last_comment.user.login == "github-actions[bot]"
        and days_since_updated > 7
        and days_since_creation >= 30
        and not any(label.name.lower() in LABELS_TO_EXEMPT for label in issue.get_labels()		)
    ):
      # Close issue since it has been 7 days of inactivity since bot mention.
      issue.edit(state="""closed"""		)
    elif (
        days_since_updated > 23
        and days_since_creation >= 30
        and not any(label.name.lower() in LABELS_TO_EXEMPT for label in issue.get_labels()		)
    ):
      # Add stale comment
      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/accelerate/blob/main/CONTRIBUTING.md) """
          """are likely to be ignored."""		)


if __name__ == "__main__":
    main()

"""simple docstring"""


# This model implementation is heavily inspired by https://github.com/haofanwang/ControlNet-for-Diffusers/

import gc
import random
import tempfile
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    ControlNetModel,
    DDIMScheduler,
    StableDiffusionControlNetImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion_controlnet import MultiControlNetModel
from diffusers.utils import floats_tensor, load_image, load_numpy, randn_tensor, slow, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
)


enable_full_determinism()


class        a				(		lowerCAmelCase_	,							lowerCAmelCase_	,							lowerCAmelCase_	,							unittest.TestCase   ):
    _snake_case     :  List[str]     							=							StableDiffusionControlNetImgaImgPipeline
    _snake_case     :  Optional[int]     							=							TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
    _snake_case     :  Optional[int]     							=							TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
    _snake_case     :  List[str]     							=							IMAGE_TO_IMAGE_IMAGE_PARAMS.union({'control_image'}   )
    _snake_case     :  Union[str, Any]     							=							IMAGE_TO_IMAGE_IMAGE_PARAMS


    def      lowerCAmelCase_  (					self	:  Optional[int]					):
      torch.manual_seed(0					)
      _UpperCAmelCase						 =      UNetaDConditionModel(
          block_out_channels=(32, 64)   ,				layers_per_block=2   ,				sample_size=32   ,				in_channels=4   ,				out_channels=4   ,				down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")   ,				up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")   ,				cross_attention_dim=32   ,				)
      torch.manual_seed(0					)
      _UpperCAmelCase						 =      ControlNetModel(
          block_out_channels=(32, 64)   ,				layers_per_block=2   ,				in_channels=4   ,				down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")   ,				cross_attention_dim=32   ,				conditioning_embedding_out_channels=(16, 32)   ,				)
      torch.manual_seed(0					)
      _UpperCAmelCase						 =      DDIMScheduler(
          beta_start=0.00_085   ,				beta_end=0.012   ,				beta_schedule="""scaled_linear"""   ,				clip_sample=__lowerCAmelCase   ,				set_alpha_to_one=__lowerCAmelCase   ,				)
      torch.manual_seed(0					)
      _UpperCAmelCase						 =      AutoencoderKL(
          block_out_channels=[32, 64]   ,				in_channels=3   ,				out_channels=3   ,				down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]   ,				up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]   ,				latent_channels=4   ,				)
      torch.manual_seed(0					)
      _UpperCAmelCase						 =      CLIPTextConfig(
          bos_token_id=0   ,				eos_token_id=2   ,				hidden_size=32   ,				intermediate_size=37   ,				layer_norm_eps=1e-0_5   ,				num_attention_heads=4   ,				num_hidden_layers=5   ,				pad_token_id=1   ,				vocab_size=1000   ,				)
      _UpperCAmelCase						 =      CLIPTextModel(__lowerCAmelCase					)
      _UpperCAmelCase						 =      CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""					)

      _UpperCAmelCase						 =      {
          """unet""": unet,
          """controlnet""": controlnet,
          """scheduler""": scheduler,
          """vae""": vae,
          """text_encoder""": text_encoder,
          """tokenizer""": tokenizer,
          """safety_checker""": None,
          """feature_extractor""": None,
      }
      return components


    def      lowerCAmelCase_  (					self	:  int   ,				__lowerCAmelCase	:  Tuple   ,				__lowerCAmelCase	:  List[str]=0					):
      if str(__lowerCAmelCase					).startswith("""mps"""					):
        _UpperCAmelCase						 =      torch.manual_seed(__lowerCAmelCase					)
      else:
        _UpperCAmelCase						 =      torch.Generator(device=__lowerCAmelCase					).manual_seed(__lowerCAmelCase					)

      _UpperCAmelCase						 =      2
      _UpperCAmelCase						 =      randn_tensor(
          (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)   ,				generator=__lowerCAmelCase   ,				device=torch.device(__lowerCAmelCase					)   ,				)
      _UpperCAmelCase						 =      floats_tensor(control_image.shape   ,				rng=random.Random(__lowerCAmelCase					)					).to(__lowerCAmelCase					)
      _UpperCAmelCase						 =      image.cpu().permute(0   ,				2   ,				3   ,				1					)[0]
      _UpperCAmelCase						 =      Image.fromarray(np.uinta(__lowerCAmelCase					)					).convert("""RGB"""					).resize((64, 64)					)
      _UpperCAmelCase						 =      {
          """prompt""": """A painting of a squirrel eating a burger""",
          """generator""": generator,
          """num_inference_steps""": 2,
          """guidance_scale""": 6.0,
          """output_type""": """numpy""",
          """image""": image,
          """control_image""": control_image,
      }

      return inputs


    def      lowerCAmelCase_  (					self	:  Dict					):
      return self._test_attention_slicing_forward_pass(expected_max_diff=2e-3					)


    @unittest.skipIf(
        torch_device != """cuda""" or not is_xformers_available()   ,				reason="""XFormers attention is only available with CUDA and `xformers` installed"""   ,				)
    def      lowerCAmelCase_  (					self	:  Tuple					):
      self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3					)


    def      lowerCAmelCase_  (					self	:  List[Any]					):
      self._test_inference_batch_single_identical(expected_max_diff=2e-3					)



class        a				(		lowerCAmelCase_	,							lowerCAmelCase_	,							unittest.TestCase   ):
    _snake_case     :  List[Any]     							=							StableDiffusionControlNetImgaImgPipeline
    _snake_case     :  int     							=							TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
    _snake_case     :  Any     							=							TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
    _snake_case     :  Optional[int]     							=							frozenset([]   )  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess


    def      lowerCAmelCase_  (					self	:  Any					):
      torch.manual_seed(0					)
      _UpperCAmelCase						 =      UNetaDConditionModel(
          block_out_channels=(32, 64)   ,				layers_per_block=2   ,				sample_size=32   ,				in_channels=4   ,				out_channels=4   ,				down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")   ,				up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")   ,				cross_attention_dim=32   ,				)
      torch.manual_seed(0					)

      def init_weights(__lowerCAmelCase	:  str					):
        if isinstance(__lowerCAmelCase   ,				torch.nn.Convad					):
          torch.nn.init.normal(m.weight					)
          m.bias.data.fill_(1.0					)

      _UpperCAmelCase						 =      ControlNetModel(
          block_out_channels=(32, 64)   ,				layers_per_block=2   ,				in_channels=4   ,				down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")   ,				cross_attention_dim=32   ,				conditioning_embedding_out_channels=(16, 32)   ,				)
      controlneta.controlnet_down_blocks.apply(__lowerCAmelCase					)

      torch.manual_seed(0					)
      _UpperCAmelCase						 =      ControlNetModel(
          block_out_channels=(32, 64)   ,				layers_per_block=2   ,				in_channels=4   ,				down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")   ,				cross_attention_dim=32   ,				conditioning_embedding_out_channels=(16, 32)   ,				)
      controlneta.controlnet_down_blocks.apply(__lowerCAmelCase					)

      torch.manual_seed(0					)
      _UpperCAmelCase						 =      DDIMScheduler(
          beta_start=0.00_085   ,				beta_end=0.012   ,				beta_schedule="""scaled_linear"""   ,				clip_sample=__lowerCAmelCase   ,				set_alpha_to_one=__lowerCAmelCase   ,				)
      torch.manual_seed(0					)
      _UpperCAmelCase						 =      AutoencoderKL(
          block_out_channels=[32, 64]   ,				in_channels=3   ,				out_channels=3   ,				down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]   ,				up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]   ,				latent_channels=4   ,				)
      torch.manual_seed(0					)
      _UpperCAmelCase						 =      CLIPTextConfig(
          bos_token_id=0   ,				eos_token_id=2   ,				hidden_size=32   ,				intermediate_size=37   ,				layer_norm_eps=1e-0_5   ,				num_attention_heads=4   ,				num_hidden_layers=5   ,				pad_token_id=1   ,				vocab_size=1000   ,				)
      _UpperCAmelCase						 =      CLIPTextModel(__lowerCAmelCase					)
      _UpperCAmelCase						 =      CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""					)

      _UpperCAmelCase						 =      MultiControlNetModel([controlneta, controlneta]					)

      _UpperCAmelCase						 =      {
          """unet""": unet,
          """controlnet""": controlnet,
          """scheduler""": scheduler,
          """vae""": vae,
          """text_encoder""": text_encoder,
          """tokenizer""": tokenizer,
          """safety_checker""": None,
          """feature_extractor""": None,
      }
      return components


    def      lowerCAmelCase_  (					self	:  Union[str, Any]   ,				__lowerCAmelCase	:  List[str]   ,				__lowerCAmelCase	:  List[str]=0					):
      if str(__lowerCAmelCase					).startswith("""mps"""					):
        _UpperCAmelCase						 =      torch.manual_seed(__lowerCAmelCase					)
      else:
        _UpperCAmelCase						 =      torch.Generator(device=__lowerCAmelCase					).manual_seed(__lowerCAmelCase					)

      _UpperCAmelCase						 =      2

      _UpperCAmelCase						 =      [
          randn_tensor(
              (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)   ,				generator=__lowerCAmelCase   ,				device=torch.device(__lowerCAmelCase					)   ,				),
          randn_tensor(
              (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)   ,				generator=__lowerCAmelCase   ,				device=torch.device(__lowerCAmelCase					)   ,				),
      ]

      _UpperCAmelCase						 =      floats_tensor(control_image[0].shape   ,				rng=random.Random(__lowerCAmelCase					)					).to(__lowerCAmelCase					)
      _UpperCAmelCase						 =      image.cpu().permute(0   ,				2   ,				3   ,				1					)[0]
      _UpperCAmelCase						 =      Image.fromarray(np.uinta(__lowerCAmelCase					)					).convert("""RGB"""					).resize((64, 64)					)
      _UpperCAmelCase						 =      {
          """prompt""": """A painting of a squirrel eating a burger""",
          """generator""": generator,
          """num_inference_steps""": 2,
          """guidance_scale""": 6.0,
          """output_type""": """numpy""",
          """image""": image,
          """control_image""": control_image,
      }

      return inputs


    def      lowerCAmelCase_  (					self	:  Optional[Any]					):
      _UpperCAmelCase						 =      self.get_dummy_components()
      _UpperCAmelCase						 =      self.pipeline_class(**__lowerCAmelCase					)
      pipe.to(__lowerCAmelCase					)

      _UpperCAmelCase						 =      10.0
      _UpperCAmelCase						 =      4

      _UpperCAmelCase						 =      self.get_dummy_inputs(__lowerCAmelCase					)
      _UpperCAmelCase						 =      steps
      _UpperCAmelCase						 =      scale
      _UpperCAmelCase						 =      pipe(**__lowerCAmelCase					)[0]

      _UpperCAmelCase						 =      self.get_dummy_inputs(__lowerCAmelCase					)
      _UpperCAmelCase						 =      steps
      _UpperCAmelCase						 =      scale
      _UpperCAmelCase						 =      pipe(**__lowerCAmelCase   ,				control_guidance_start=0.1   ,				control_guidance_end=0.2					)[0]

      _UpperCAmelCase						 =      self.get_dummy_inputs(__lowerCAmelCase					)
      _UpperCAmelCase						 =      steps
      _UpperCAmelCase						 =      scale
      _UpperCAmelCase						 =      pipe(**__lowerCAmelCase   ,				control_guidance_start=[0.1, 0.3]   ,				control_guidance_end=[0.2, 0.7]					)[0]

      _UpperCAmelCase						 =      self.get_dummy_inputs(__lowerCAmelCase					)
      _UpperCAmelCase						 =      steps
      _UpperCAmelCase						 =      scale
      _UpperCAmelCase						 =      pipe(**__lowerCAmelCase   ,				control_guidance_start=0.4   ,				control_guidance_end=[0.5, 0.8]					)[0]

      # make sure that all outputs are different
      assert np.sum(np.abs(output_a - output_a					)					) > 1e-3
      assert np.sum(np.abs(output_a - output_a					)					) > 1e-3
      assert np.sum(np.abs(output_a - output_a					)					) > 1e-3


    def      lowerCAmelCase_  (					self	:  Union[str, Any]					):
      return self._test_attention_slicing_forward_pass(expected_max_diff=2e-3					)


    @unittest.skipIf(
        torch_device != """cuda""" or not is_xformers_available()   ,				reason="""XFormers attention is only available with CUDA and `xformers` installed"""   ,				)
    def      lowerCAmelCase_  (					self	:  Dict					):
      self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3					)


    def      lowerCAmelCase_  (					self	:  Dict					):
      self._test_inference_batch_single_identical(expected_max_diff=2e-3					)


    def      lowerCAmelCase_  (					self	:  List[Any]					):
      _UpperCAmelCase						 =      self.get_dummy_components()
      _UpperCAmelCase						 =      self.pipeline_class(**__lowerCAmelCase					)
      pipe.to(__lowerCAmelCase					)
      pipe.set_progress_bar_config(disable=__lowerCAmelCase					)
      with tempfile.TemporaryDirectory() as tmpdir:
        try:
          # save_pretrained is not implemented for Multi-ControlNet
          pipe.save_pretrained(__lowerCAmelCase					)
        except NotImplementedError:
          pass



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


    def      lowerCAmelCase_  (					self	:  Optional[int]					):
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()


    def      lowerCAmelCase_  (					self	:  List[str]					):
      _UpperCAmelCase						 =      ControlNetModel.from_pretrained("""lllyasviel/sd-controlnet-canny"""					)

      _UpperCAmelCase						 =      StableDiffusionControlNetImgaImgPipeline.from_pretrained(
          """runwayml/stable-diffusion-v1-5"""   ,				safety_checker=__lowerCAmelCase   ,				controlnet=__lowerCAmelCase					)
      pipe.enable_model_cpu_offload()
      pipe.set_progress_bar_config(disable=__lowerCAmelCase					)

      _UpperCAmelCase						 =      torch.Generator(device="""cpu"""					).manual_seed(0					)
      _UpperCAmelCase						 =      """evil space-punk bird"""
      _UpperCAmelCase						 =      load_image(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png"""					).resize((512, 512)					)
      _UpperCAmelCase						 =      load_image(
          """https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png"""					).resize((512, 512)					)

      _UpperCAmelCase						 =      pipe(
          __lowerCAmelCase   ,				__lowerCAmelCase   ,				control_image=__lowerCAmelCase   ,				generator=__lowerCAmelCase   ,				output_type="""np"""   ,				num_inference_steps=50   ,				strength=0.6   ,				)

      _UpperCAmelCase						 =      output.images[0]

      assert image.shape == (512, 512, 3)

      _UpperCAmelCase						 =      load_numpy(
          """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/img2img.npy"""					)

      assert np.abs(expected_image - image					).max() < 9e-2
