




# 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(						UpperCamelCase							,				UpperCamelCase							,				UpperCamelCase							,				unittest.TestCase       ):



 '''simple docstring'''


 UpperCamelCase						       =						StableDiffusionControlNetImgaImgPipeline
 UpperCamelCase						       =						TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
 UpperCamelCase						       =						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 UpperCamelCase						       =						IMAGE_TO_IMAGE_IMAGE_PARAMS.union({'''control_image'''}       )
 UpperCamelCase						       =						IMAGE_TO_IMAGE_IMAGE_PARAMS

 def     a__			(    self :  Any  )       ->     List[str]:

       """simple docstring"""






       torch.manual_seed(0  )
       lowerCamelCase_							=   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  )
       lowerCamelCase_							=   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  )
       lowerCamelCase_							=   DDIMScheduler(
           beta_start=0.00085						,							beta_end=0.012						,							beta_schedule='scaled_linear'						,							clip_sample=A_						,							set_alpha_to_one=A_						,							)
       torch.manual_seed(0  )
       lowerCamelCase_							=   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  )
       lowerCamelCase_							=   CLIPTextConfig(
           bos_token_id=0						,							eos_token_id=2						,							hidden_size=32						,							intermediate_size=37						,							layer_norm_eps=1E-05						,							num_attention_heads=4						,							num_hidden_layers=5						,							pad_token_id=1						,							vocab_size=1000						,							)
       lowerCamelCase_							=   CLIPTextModel(A_  )
       lowerCamelCase_							=   CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  )

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

 def     a__			(    self :  Any						,							A_ :  Any						,							A_ :  str=0  )       ->     List[str]:

       """simple docstring"""






       if str(A_  ).startswith('mps'  ):
             lowerCamelCase_							=   torch.manual_seed(A_  )
       else:
             lowerCamelCase_							=   torch.Generator(device=A_  ).manual_seed(A_  )

       lowerCamelCase_							=   2
       lowerCamelCase_							=   randn_tensor(
           (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)						,							generator=A_						,							device=torch.device(A_  )						,							)
       lowerCamelCase_							=   floats_tensor(control_image.shape						,							rng=random.Random(A_  )  ).to(A_  )
       lowerCamelCase_							=   image.cpu().permute(0						,							2						,							3						,							1  )[0]
       lowerCamelCase_							=   Image.fromarray(np.uinta(A_  )  ).convert('RGB'  ).resize((64, 64)  )
       lowerCamelCase_							=   {
           '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     a__			(    self :  List[str]  )       ->     Union[str, Any]:

       """simple docstring"""






       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     a__			(    self :  Union[str, Any]  )       ->     Tuple:

       """simple docstring"""






       self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3  )





 def     a__			(    self :  Tuple  )       ->     List[str]:

       """simple docstring"""






       self._test_inference_batch_single_identical(expected_max_diff=2E-3  )







class   A(						UpperCamelCase							,				UpperCamelCase							,				unittest.TestCase       ):



 '''simple docstring'''


 UpperCamelCase						       =						StableDiffusionControlNetImgaImgPipeline
 UpperCamelCase						       =						TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
 UpperCamelCase						       =						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
 UpperCamelCase						       =						frozenset([]       )  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess

 def     a__			(    self :  Dict  )       ->     Any:

       """simple docstring"""






       torch.manual_seed(0  )
       lowerCamelCase_							=   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(A_ :  Union[str, Any]  ):
             if isinstance(A_						,							torch.nn.Convad  ):
                   torch.nn.init.normal(m.weight  )
                   m.bias.data.fill_(1.0  )

       lowerCamelCase_							=   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(A_  )

       torch.manual_seed(0  )
       lowerCamelCase_							=   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(A_  )

       torch.manual_seed(0  )
       lowerCamelCase_							=   DDIMScheduler(
           beta_start=0.00085						,							beta_end=0.012						,							beta_schedule='scaled_linear'						,							clip_sample=A_						,							set_alpha_to_one=A_						,							)
       torch.manual_seed(0  )
       lowerCamelCase_							=   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  )
       lowerCamelCase_							=   CLIPTextConfig(
           bos_token_id=0						,							eos_token_id=2						,							hidden_size=32						,							intermediate_size=37						,							layer_norm_eps=1E-05						,							num_attention_heads=4						,							num_hidden_layers=5						,							pad_token_id=1						,							vocab_size=1000						,							)
       lowerCamelCase_							=   CLIPTextModel(A_  )
       lowerCamelCase_							=   CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'  )

       lowerCamelCase_							=   MultiControlNetModel([controlneta, controlneta]  )

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

 def     a__			(    self :  Union[str, Any]						,							A_ :  Any						,							A_ :  Tuple=0  )       ->     int:

       """simple docstring"""






       if str(A_  ).startswith('mps'  ):
             lowerCamelCase_							=   torch.manual_seed(A_  )
       else:
             lowerCamelCase_							=   torch.Generator(device=A_  ).manual_seed(A_  )

       lowerCamelCase_							=   2

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

       lowerCamelCase_							=   floats_tensor(control_image[0].shape						,							rng=random.Random(A_  )  ).to(A_  )
       lowerCamelCase_							=   image.cpu().permute(0						,							2						,							3						,							1  )[0]
       lowerCamelCase_							=   Image.fromarray(np.uinta(A_  )  ).convert('RGB'  ).resize((64, 64)  )
       lowerCamelCase_							=   {
           '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     a__			(    self :  Union[str, Any]  )       ->     Dict:

       """simple docstring"""






       lowerCamelCase_							=   self.get_dummy_components()
       lowerCamelCase_							=   self.pipeline_class(**A_  )
       pipe.to(A_  )

       lowerCamelCase_							=   10.0
       lowerCamelCase_							=   4

       lowerCamelCase_							=   self.get_dummy_inputs(A_  )
       lowerCamelCase_							=   steps
       lowerCamelCase_							=   scale
       lowerCamelCase_							=   pipe(**A_  )[0]

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

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

       lowerCamelCase_							=   self.get_dummy_inputs(A_  )
       lowerCamelCase_							=   steps
       lowerCamelCase_							=   scale
       lowerCamelCase_							=   pipe(**A_						,							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     a__			(    self :  Optional[int]  )       ->     Union[str, Any]:

       """simple docstring"""






       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     a__			(    self :  Tuple  )       ->     Tuple:

       """simple docstring"""






       self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3  )

 def     a__			(    self :  str  )       ->     Optional[Any]:

       """simple docstring"""






       self._test_inference_batch_single_identical(expected_max_diff=2E-3  )





 def     a__			(    self :  Tuple  )       ->     List[str]:

       """simple docstring"""






       lowerCamelCase_							=   self.get_dummy_components()
       lowerCamelCase_							=   self.pipeline_class(**A_  )
       pipe.to(A_  )
       pipe.set_progress_bar_config(disable=A_  )
       with tempfile.TemporaryDirectory() as tmpdir:
             try:
                   # save_pretrained is not implemented for Multi-ControlNet
                   pipe.save_pretrained(A_  )
             except NotImplementedError:
                   pass







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



 '''simple docstring'''


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

       """simple docstring"""






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





 def     a__			(    self :  List[str]  )       ->     Tuple:

       """simple docstring"""






       lowerCamelCase_							=   ControlNetModel.from_pretrained('lllyasviel/sd-controlnet-canny'  )

       lowerCamelCase_							=   StableDiffusionControlNetImgaImgPipeline.from_pretrained(
           'runwayml/stable-diffusion-v1-5'						,							safety_checker=A_						,							controlnet=A_  )
       pipe.enable_model_cpu_offload()
       pipe.set_progress_bar_config(disable=A_  )

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

       lowerCamelCase_							=   pipe(
           A_						,							A_						,							control_image=A_						,							generator=A_						,							output_type='np'						,							num_inference_steps=50						,							strength=0.6						,							)

       lowerCamelCase_							=   output.images[0]

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

       lowerCamelCase_							=   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









import math

def       _SCREAMING_SNAKE_CASE      (							lowercase     :	int					):


      '''simple docstring'''



      lowerCamelCase_							=   0
      lowerCamelCase_							=   0
      while num > 0:
            lowerCamelCase_							=   num % 8
            lowerCamelCase_							=   octal + (remainder * math.floor(math.pow(10     ,							lowercase					)					))
            counter += 1
            lowerCamelCase_							=   math.floor(num / 8					)  # basically /= 8 without remainder if any
            # This formatting removes trailing '.0' from `octal`.
      return f"""0o{int(lowercase					)}"""

def       _SCREAMING_SNAKE_CASE      (							):


      '''simple docstring'''



      print('\n2 in octal is:'					)
      print(decimal_to_octal(2					)					)  # = 2
      print('\n8 in octal is:'					)
      print(decimal_to_octal(8					)					)  # = 10
      print('\n65 in octal is:'					)
      print(decimal_to_octal(65					)					)  # = 101
      print('\n216 in octal is:'					)
      print(decimal_to_octal(2_16					)					)  # = 330
      print('\n512 in octal is:'					)
      print(decimal_to_octal(5_12					)					)  # = 1000
      print('\n'					)


if __name__ == "__main__":
 main()



