# This model implementation is heavily based on:

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,
    StableDiffusionControlNetInpaintPipeline,
    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 (
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_PARAMS,
    TEXT_TO_IMAGE_IMAGE_PARAMS,
)
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
)


enable_full_determinism()



class     snake_case_			(						__A    ,__A    ,__A    ,unittest.TestCase			):
   __A      :						str   =							StableDiffusionControlNetInpaintPipeline
   __A      :						Union[str, Any]   =							TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
   __A      :						List[Any]   =							TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
   __A      :						Any   =							frozenset({"control_image"}			)  # skip `image` and `mask` for now, only test for control_image
   __A      :						Dict   =							TEXT_TO_IMAGE_IMAGE_PARAMS






   def  __UpperCamelCase	( self							:	Dict    )   ->		List[Any]:
       torch.manual_seed(0    )
       lowercase__					:  str          =							UNetaDConditionModel(
           block_out_channels=(32, 64)      ,		layers_per_block=2      ,		sample_size=32      ,		in_channels=9      ,		out_channels=4      ,		down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")      ,		up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")      ,		cross_attention_dim=32      ,		)
       torch.manual_seed(0    )
       lowercase__					:  Optional[int]          =							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    )
       lowercase__					:  Tuple          =							DDIMScheduler(
           beta_start=0.0_00_85      ,		beta_end=0.0_12      ,		beta_schedule="scaled_linear"      ,		clip_sample=lowercase_      ,		set_alpha_to_one=lowercase_      ,		)
       torch.manual_seed(0    )
       lowercase__					:  str          =							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    )
       lowercase__					:  Any          =							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=10_00      ,		)
       lowercase__					:  List[Any]          =							CLIPTextModel(lowercase_    )
       lowercase__					:  Tuple          =							CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"    )

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






   def  __UpperCamelCase	( self							:	Tuple      ,		lowercase_							:	List[Any]      ,		lowercase_							:	int=0    )   ->		Any:
       if str(lowercase_    ).startswith("mps"    ):
           lowercase__					:  Any          =							torch.manual_seed(lowercase_    )
       else:
           lowercase__					:  Optional[Any]          =							torch.Generator(device=lowercase_    ).manual_seed(lowercase_    )

       lowercase__					:  str          =							2
       lowercase__					:  Optional[Any]          =							randn_tensor(
           (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)      ,		generator=lowercase_      ,		device=torch.device(lowercase_    )      ,		)
       lowercase__					:  List[str]          =							floats_tensor((1, 3, 32, 32)      ,		rng=random.Random(lowercase_    )    ).to(lowercase_    )
       lowercase__					:  int          =							init_image.cpu().permute(0      ,		2      ,		3      ,		1    )[0]

       lowercase__					:  Any          =							Image.fromarray(np.uinta(lowercase_    )    ).convert("RGB"    ).resize((64, 64)    )
       lowercase__					:  List[str]          =							Image.fromarray(np.uinta(init_image + 4    )    ).convert("RGB"    ).resize((64, 64)    )

       lowercase__					:  Any          =							{
           "prompt": "A painting of a squirrel eating a burger",
           "generator": generator,
           "num_inference_steps": 2,
           "guidance_scale": 6.0,
           "output_type": "numpy",
           "image": image,
           "mask_image": mask_image,
           "control_image": control_image,
       }

       return inputs






   def  __UpperCamelCase	( self							:	Union[str, Any]    )   ->		Optional[int]:
       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  __UpperCamelCase	( self							:	Dict    )   ->		Optional[int]:
       self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3    )







   def  __UpperCamelCase	( self							:	Dict    )   ->		Any:
       self._test_inference_batch_single_identical(expected_max_diff=2E-3    )



class     snake_case_			(						__A			):
   __A      :						List[Any]   =							StableDiffusionControlNetInpaintPipeline
   __A      :						int   =							TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
   __A      :						Dict   =							TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
   __A      :						Optional[Any]   =							frozenset([]			)






   def  __UpperCamelCase	( self							:	Optional[Any]    )   ->		int:
       torch.manual_seed(0    )
       lowercase__					:  Tuple          =							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    )
       lowercase__					:  Any          =							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    )
       lowercase__					:  Tuple          =							DDIMScheduler(
           beta_start=0.0_00_85      ,		beta_end=0.0_12      ,		beta_schedule="scaled_linear"      ,		clip_sample=lowercase_      ,		set_alpha_to_one=lowercase_      ,		)
       torch.manual_seed(0    )
       lowercase__					:  Optional[int]          =							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    )
       lowercase__					:  Tuple          =							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=10_00      ,		)
       lowercase__					:  Any          =							CLIPTextModel(lowercase_    )
       lowercase__					:  List[Any]          =							CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"    )

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



class     snake_case_			(						__A    ,__A    ,unittest.TestCase			):
   __A      :						Union[str, Any]   =							StableDiffusionControlNetInpaintPipeline
   __A      :						List[Any]   =							TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
   __A      :						List[Any]   =							TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS






   def  __UpperCamelCase	( self							:	Tuple    )   ->		Optional[int]:
       torch.manual_seed(0    )
       lowercase__					:  Tuple          =							UNetaDConditionModel(
           block_out_channels=(32, 64)      ,		layers_per_block=2      ,		sample_size=32      ,		in_channels=9      ,		out_channels=4      ,		down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")      ,		up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")      ,		cross_attention_dim=32      ,		)
       torch.manual_seed(0    )

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

       lowercase__					:  Tuple          =							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(lowercase_    )

       torch.manual_seed(0    )
       lowercase__					:  Any          =							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(lowercase_    )

       torch.manual_seed(0    )
       lowercase__					:  Tuple          =							DDIMScheduler(
           beta_start=0.0_00_85      ,		beta_end=0.0_12      ,		beta_schedule="scaled_linear"      ,		clip_sample=lowercase_      ,		set_alpha_to_one=lowercase_      ,		)
       torch.manual_seed(0    )
       lowercase__					:  Dict          =							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    )
       lowercase__					:  List[str]          =							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=10_00      ,		)
       lowercase__					:  int          =							CLIPTextModel(lowercase_    )
       lowercase__					:  str          =							CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"    )

       lowercase__					:  Tuple          =							MultiControlNetModel([controlneta, controlneta]    )

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






   def  __UpperCamelCase	( self							:	Optional[Any]      ,		lowercase_							:	List[str]      ,		lowercase_							:	Any=0    )   ->		Optional[int]:
       if str(lowercase_    ).startswith("mps"    ):
           lowercase__					:  Dict          =							torch.manual_seed(lowercase_    )
       else:
           lowercase__					:  Any          =							torch.Generator(device=lowercase_    ).manual_seed(lowercase_    )

       lowercase__					:  int          =							2

       lowercase__					:  Any          =							[
           randn_tensor(
               (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)      ,		generator=lowercase_      ,		device=torch.device(lowercase_    )      ,		),
           randn_tensor(
               (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)      ,		generator=lowercase_      ,		device=torch.device(lowercase_    )      ,		),
       ]
       lowercase__					:  List[str]          =							floats_tensor((1, 3, 32, 32)      ,		rng=random.Random(lowercase_    )    ).to(lowercase_    )
       lowercase__					:  List[Any]          =							init_image.cpu().permute(0      ,		2      ,		3      ,		1    )[0]

       lowercase__					:  Tuple          =							Image.fromarray(np.uinta(lowercase_    )    ).convert("RGB"    ).resize((64, 64)    )
       lowercase__					:  List[Any]          =							Image.fromarray(np.uinta(init_image + 4    )    ).convert("RGB"    ).resize((64, 64)    )

       lowercase__					:  Dict          =							{
           "prompt": "A painting of a squirrel eating a burger",
           "generator": generator,
           "num_inference_steps": 2,
           "guidance_scale": 6.0,
           "output_type": "numpy",
           "image": image,
           "mask_image": mask_image,
           "control_image": control_image,
       }

       return inputs






   def  __UpperCamelCase	( self							:	Any    )   ->		List[str]:
       lowercase__					:  Dict          =							self.get_dummy_components()
       lowercase__					:  int          =							self.pipeline_class(**lowercase_    )
       pipe.to(lowercase_    )

       lowercase__					:  Dict          =							10.0
       lowercase__					:  List[str]          =							4

       lowercase__					:  str          =							self.get_dummy_inputs(lowercase_    )
       lowercase__					:  List[Any]          =							steps
       lowercase__					:  int          =							scale
       lowercase__					:  List[str]          =							pipe(**lowercase_    )[0]

       lowercase__					:  List[Any]          =							self.get_dummy_inputs(lowercase_    )
       lowercase__					:  Any          =							steps
       lowercase__					:  List[str]          =							scale
       lowercase__					:  List[str]          =							pipe(**lowercase_      ,		control_guidance_start=0.1      ,		control_guidance_end=0.2    )[0]

       lowercase__					:  Optional[int]          =							self.get_dummy_inputs(lowercase_    )
       lowercase__					:  int          =							steps
       lowercase__					:  int          =							scale
       lowercase__					:  List[str]          =							pipe(**lowercase_      ,		control_guidance_start=[0.1, 0.3]      ,		control_guidance_end=[0.2, 0.7]    )[0]

       lowercase__					:  List[Any]          =							self.get_dummy_inputs(lowercase_    )
       lowercase__					:  List[str]          =							steps
       lowercase__					:  Dict          =							scale
       lowercase__					:  Dict          =							pipe(**lowercase_      ,		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  __UpperCamelCase	( self							:	Tuple    )   ->		Tuple:
       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  __UpperCamelCase	( self							:	int    )   ->		Optional[Any]:
       self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3    )






   def  __UpperCamelCase	( self							:	Any    )   ->		Tuple:
       self._test_inference_batch_single_identical(expected_max_diff=2E-3    )







   def  __UpperCamelCase	( self							:	Optional[Any]    )   ->		Dict:
       lowercase__					:  Tuple          =							self.get_dummy_components()
       lowercase__					:  List[Any]          =							self.pipeline_class(**lowercase_    )
       pipe.to(lowercase_    )
       pipe.set_progress_bar_config(disable=lowercase_    )
       with tempfile.TemporaryDirectory() as tmpdir:
           try:
               # save_pretrained is not implemented for Multi-ControlNet
               pipe.save_pretrained(lowercase_    )
           except NotImplementedError:
               pass



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






   def  __UpperCamelCase	( self							:	Any    )   ->		List[Any]:
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()






   def  __UpperCamelCase	( self							:	List[Any]    )   ->		Tuple:
       lowercase__					:  List[str]          =							ControlNetModel.from_pretrained("lllyasviel/sd-controlnet-canny"    )

       lowercase__					:  Any          =							StableDiffusionControlNetInpaintPipeline.from_pretrained(
           "runwayml/stable-diffusion-inpainting"      ,		safety_checker=lowercase_      ,		controlnet=lowercase_    )
       pipe.enable_model_cpu_offload()
       pipe.set_progress_bar_config(disable=lowercase_    )

       lowercase__					:  Dict          =							torch.Generator(device="cpu"    ).manual_seed(0    )
       lowercase__					:  Optional[Any]          =							load_image(
           "https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png"    ).resize((5_12, 5_12)    )

       lowercase__					:  List[str]          =							load_image(
           "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
           "/stable_diffusion_inpaint/input_bench_mask.png"    ).resize((5_12, 5_12)    )

       lowercase__					:  Any          =							"pitch black hole"

       lowercase__					:  Optional[Any]          =							load_image(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png"    ).resize((5_12, 5_12)    )

       lowercase__					:  Any          =							pipe(
           lowercase_      ,		image=lowercase_      ,		mask_image=lowercase_      ,		control_image=lowercase_      ,		generator=lowercase_      ,		output_type="np"      ,		num_inference_steps=3      ,		)

       lowercase__					:  Union[str, Any]          =							output.images[0]

       assert image.shape == (5_12, 5_12, 3)

       lowercase__					:  Optional[int]          =							load_numpy(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/inpaint.npy"    )

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







   def  __UpperCamelCase	( self							:	Union[str, Any]    )   ->		Dict:
       lowercase__					:  List[Any]          =							ControlNetModel.from_pretrained("lllyasviel/control_v11p_sd15_inpaint"    )

       lowercase__					:  Any          =							StableDiffusionControlNetInpaintPipeline.from_pretrained(
           "runwayml/stable-diffusion-v1-5"      ,		safety_checker=lowercase_      ,		controlnet=lowercase_    )
       lowercase__					:  int          =							DDIMScheduler.from_config(pipe.scheduler.config    )
       pipe.enable_model_cpu_offload()
       pipe.set_progress_bar_config(disable=lowercase_    )

       lowercase__					:  int          =							torch.Generator(device="cpu"    ).manual_seed(33    )

       lowercase__					:  List[Any]          =							load_image(
           "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_inpaint/boy.png"    )
       lowercase__					:  Tuple          =							init_image.resize((5_12, 5_12)    )

       lowercase__					:  Any          =							load_image(
           "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_inpaint/boy_mask.png"    )
       lowercase__					:  Any          =							mask_image.resize((5_12, 5_12)    )

       lowercase__					:  Union[str, Any]          =							"a handsome man with ray-ban sunglasses"

       def make_inpaint_condition(lowercase_							:	List[str]      ,		lowercase_							:	Any    ):
           lowercase__					:  Optional[int]          =							np.array(image.convert("RGB"    )    ).astype(np.floataa    ) / 2_55.0
           lowercase__					:  Any          =							np.array(image_mask.convert("L"    )    ).astype(np.floataa    ) / 2_55.0

           assert image.shape[0:1] == image_mask.shape[0:1], "image and image_mask must have the same image size"
           lowercase__					:  int          =							-1.0  # set as masked pixel
           lowercase__					:  Dict          =							np.expand_dims(lowercase_      ,		0    ).transpose(0      ,		3      ,		1      ,		2    )
           lowercase__					:  Optional[int]          =							torch.from_numpy(lowercase_    )
           return image

       lowercase__					:  Tuple          =							make_inpaint_condition(lowercase_      ,		lowercase_    )

       lowercase__					:  List[Any]          =							pipe(
           lowercase_      ,		image=lowercase_      ,		mask_image=lowercase_      ,		control_image=lowercase_      ,		guidance_scale=9.0      ,		eta=1.0      ,		generator=lowercase_      ,		num_inference_steps=20      ,		output_type="np"      ,		)
       lowercase__					:  Optional[Any]          =							output.images[0]

       assert image.shape == (5_12, 5_12, 3)

       lowercase__					:  Optional[Any]          =							load_numpy(
           "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/boy_ray_ban.npy"    )

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




class     snake_case_			:






   def __init__( self							:	Optional[int]      ,		lowercase_							:	Optional[int]=None    )   ->		Optional[int]:
       lowercase__					:  List[Any]          =							data
       lowercase__					:  Any          =							None







   def __repr__( self							:	Optional[Any]    )   ->		int:
       lowercase__					:  Optional[int]          =							""
       lowercase__					:  Any          =							self
       while temp:
           string_rep += F'''<{temp.data}> ---> '''
           lowercase__					:  List[Any]          =							temp.next
       string_rep += "<END>"
       return string_rep



def      lowercase_							(     _lowerCamelCase				:			List[Any]):

    # if elements_list is empty
    if not elements_list:
        raise Exception("The Elements List is empty")

    # Set first element as Head
    lowercase__					:  List[str]          =							Node(elements_list[0])
    lowercase__					:  Tuple          =							head
    # Loop through elements from position 1
    for data in elements_list[1:]:
        lowercase__					:  Tuple          =							Node(_lowerCamelCase)
        lowercase__					:  Union[str, Any]          =							current.next
    return head


UpperCamelCase    				=		[1, 3, 5, 32, 44, 12, 43]
print(f"List: {list_data}")
print('''Creating Linked List from List.''')
UpperCamelCase    				=		make_linked_list(list_data)
print('''Linked List:''')

print(linked_list)



