


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






						"""simple docstring"""
						__UpperCAmelCase   :		Optional[int]  					=	StableDiffusionControlNetInpaintPipeline
						__UpperCAmelCase   :		int  					=	TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
						__UpperCAmelCase   :		List[str]  					=	TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
						__UpperCAmelCase   :		Dict  					=	frozenset({'''control_image'''}       )  # skip `image` and `mask` for now, only test for control_image
						__UpperCAmelCase   :		Optional[Any]  					=	TEXT_TO_IMAGE_IMAGE_PARAMS




						def 					_UpperCamelCase			(   self     ):
							torch.manual_seed(0     )
							lowerCamelCase_			:							Any      =       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     )
							lowerCamelCase_			:							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)				,							)
							torch.manual_seed(0     )
							lowerCamelCase_			:							Any      =       DDIMScheduler(
							    beta_start=0.0_00_85				,							beta_end=0.0_12				,							beta_schedule="scaled_linear"				,							clip_sample=a_				,							set_alpha_to_one=a_				,							)
							torch.manual_seed(0     )
							lowerCamelCase_			:							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     )
							lowerCamelCase_			:							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=1000				,							)
							lowerCamelCase_			:							Any      =       CLIPTextModel(a_     )
							lowerCamelCase_			:							List[str]      =       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

							lowerCamelCase_			:							Optional[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				,							a_				,							a_=0     ):
							if str(a_     ).startswith("mps"     ):
								lowerCamelCase_			:							Dict      =       torch.manual_seed(a_     )
							else:
								lowerCamelCase_			:							Any      =       torch.Generator(device=a_     ).manual_seed(a_     )

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

							lowerCamelCase_			:							Optional[int]      =       Image.fromarray(np.uinta(a_     )     ).convert("RGB"     ).resize((64, 64)     )
							lowerCamelCase_			:							Dict      =       Image.fromarray(np.uinta(init_image + 4     )     ).convert("RGB"     ).resize((64, 64)     )

							lowerCamelCase_			:							Union[str, 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     ):
							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     ):
							self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3     )




						def 					_UpperCamelCase			(   self     ):
							self._test_inference_batch_single_identical(expected_max_diff=2E-3     )





class    lowerCAmelCase__				(			__lowerCamelCase       ):






						"""simple docstring"""
						__UpperCAmelCase   :		List[Any]  					=	StableDiffusionControlNetInpaintPipeline
						__UpperCAmelCase   :		Optional[int]  					=	TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
						__UpperCAmelCase   :		Optional[Any]  					=	TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
						__UpperCAmelCase   :		List[Any]  					=	frozenset([]       )




						def 					_UpperCamelCase			(   self     ):
							torch.manual_seed(0     )
							lowerCamelCase_			:							int      =       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_			:							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)				,							)
							torch.manual_seed(0     )
							lowerCamelCase_			:							Tuple      =       DDIMScheduler(
							    beta_start=0.0_00_85				,							beta_end=0.0_12				,							beta_schedule="scaled_linear"				,							clip_sample=a_				,							set_alpha_to_one=a_				,							)
							torch.manual_seed(0     )
							lowerCamelCase_			:							Union[str, Any]      =       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_			:							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=1000				,							)
							lowerCamelCase_			:							Tuple      =       CLIPTextModel(a_     )
							lowerCamelCase_			:							Tuple      =       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

							lowerCamelCase_			:							List[Any]      =       {
							    "unet": unet,
							    "controlnet": controlnet,
							    "scheduler": scheduler,
							    "vae": vae,
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "safety_checker": None,
							    "feature_extractor": None,
							}
							return components





class    lowerCAmelCase__				(			__lowerCamelCase,    __lowerCamelCase,    unittest.TestCase       ):






						"""simple docstring"""
						__UpperCAmelCase   :		str  					=	StableDiffusionControlNetInpaintPipeline
						__UpperCAmelCase   :		Optional[int]  					=	TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
						__UpperCAmelCase   :		Tuple  					=	TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS




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

							lowerCamelCase_			:							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)				,							)
							controlneta.controlnet_down_blocks.apply(a_     )

							torch.manual_seed(0     )
							lowerCamelCase_			:							List[str]      =       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_			:							Dict      =       DDIMScheduler(
							    beta_start=0.0_00_85				,							beta_end=0.0_12				,							beta_schedule="scaled_linear"				,							clip_sample=a_				,							set_alpha_to_one=a_				,							)
							torch.manual_seed(0     )
							lowerCamelCase_			:							Optional[Any]      =       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_			:							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=1000				,							)
							lowerCamelCase_			:							List[str]      =       CLIPTextModel(a_     )
							lowerCamelCase_			:							List[Any]      =       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"     )

							lowerCamelCase_			:							Dict      =       MultiControlNetModel([controlneta, controlneta]     )

							lowerCamelCase_			:							List[str]      =       {
							    "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				,							a_				,							a_=0     ):
							if str(a_     ).startswith("mps"     ):
								lowerCamelCase_			:							Optional[Any]      =       torch.manual_seed(a_     )
							else:
								lowerCamelCase_			:							List[Any]      =       torch.Generator(device=a_     ).manual_seed(a_     )

							lowerCamelCase_			:							Optional[Any]      =       2

							lowerCamelCase_			:							Dict      =       [
							    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_			:							List[str]      =       floats_tensor((1, 3, 32, 32)				,							rng=random.Random(a_     )     ).to(a_     )
							lowerCamelCase_			:							Dict      =       init_image.cpu().permute(0				,							2				,							3				,							1     )[0]

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

							lowerCamelCase_			:							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     ):
							lowerCamelCase_			:							Union[str, Any]      =       self.get_dummy_components()
							lowerCamelCase_			:							List[Any]      =       self.pipeline_class(**a_     )
							pipe.to(a_     )

							lowerCamelCase_			:							str      =       10.0
							lowerCamelCase_			:							Dict      =       4

							lowerCamelCase_			:							Tuple      =       self.get_dummy_inputs(a_     )
							lowerCamelCase_			:							Dict      =       steps
							lowerCamelCase_			:							Optional[Any]      =       scale
							lowerCamelCase_			:							int      =       pipe(**a_     )[0]

							lowerCamelCase_			:							Dict      =       self.get_dummy_inputs(a_     )
							lowerCamelCase_			:							Union[str, Any]      =       steps
							lowerCamelCase_			:							Dict      =       scale
							lowerCamelCase_			:							Optional[Any]      =       pipe(**a_				,							control_guidance_start=0.1				,							control_guidance_end=0.2     )[0]

							lowerCamelCase_			:							Tuple      =       self.get_dummy_inputs(a_     )
							lowerCamelCase_			:							Optional[int]      =       steps
							lowerCamelCase_			:							str      =       scale
							lowerCamelCase_			:							Optional[Any]      =       pipe(**a_				,							control_guidance_start=[0.1, 0.3]				,							control_guidance_end=[0.2, 0.7]     )[0]

							lowerCamelCase_			:							Tuple      =       self.get_dummy_inputs(a_     )
							lowerCamelCase_			:							Union[str, Any]      =       steps
							lowerCamelCase_			:							Union[str, Any]      =       scale
							lowerCamelCase_			:							Any      =       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 					_UpperCamelCase			(   self     ):
							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     ):
							self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3     )




						def 					_UpperCamelCase			(   self     ):
							self._test_inference_batch_single_identical(expected_max_diff=2E-3     )




						def 					_UpperCamelCase			(   self     ):
							lowerCamelCase_			:							Optional[int]      =       self.get_dummy_components()
							lowerCamelCase_			:							List[str]      =       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    lowerCAmelCase__				(			unittest.TestCase       ):






						"""simple docstring"""




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




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

							lowerCamelCase_			:							str      =       StableDiffusionControlNetInpaintPipeline.from_pretrained(
							    "runwayml/stable-diffusion-inpainting"				,							safety_checker=a_				,							controlnet=a_     )
							pipe.enable_model_cpu_offload()
							pipe.set_progress_bar_config(disable=a_     )

							lowerCamelCase_			:							List[Any]      =       torch.Generator(device="cpu"     ).manual_seed(0     )
							lowerCamelCase_			:							Tuple      =       load_image(
							    "https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png"     ).resize((512, 512)     )

							lowerCamelCase_			:							str      =       load_image(
							    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
							    "/stable_diffusion_inpaint/input_bench_mask.png"     ).resize((512, 512)     )

							lowerCamelCase_			:							Dict      =       "pitch black hole"

							lowerCamelCase_			:							int      =       load_image(
							    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png"     ).resize((512, 512)     )

							lowerCamelCase_			:							Tuple      =       pipe(
							    a_				,							image=a_				,							mask_image=a_				,							control_image=a_				,							generator=a_				,							output_type="np"				,							num_inference_steps=3				,							)

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

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

							lowerCamelCase_			:							Union[str, Any]      =       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     ):
							lowerCamelCase_			:							Optional[int]      =       ControlNetModel.from_pretrained("lllyasviel/control_v11p_sd15_inpaint"     )

							lowerCamelCase_			:							Optional[Any]      =       StableDiffusionControlNetInpaintPipeline.from_pretrained(
							    "runwayml/stable-diffusion-v1-5"				,							safety_checker=a_				,							controlnet=a_     )
							lowerCamelCase_			:							Optional[Any]      =       DDIMScheduler.from_config(pipe.scheduler.config     )
							pipe.enable_model_cpu_offload()
							pipe.set_progress_bar_config(disable=a_     )

							lowerCamelCase_			:							List[str]      =       torch.Generator(device="cpu"     ).manual_seed(33     )

							lowerCamelCase_			:							Dict      =       load_image(
							    "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_inpaint/boy.png"     )
							lowerCamelCase_			:							List[str]      =       init_image.resize((512, 512)     )

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

							lowerCamelCase_			:							int      =       "a handsome man with ray-ban sunglasses"

							def make_inpaint_condition(a_				,							a_     ):
								lowerCamelCase_			:							List[Any]      =       np.array(image.convert("RGB"     )     ).astype(np.floataa     ) / 255.0
								lowerCamelCase_			:							Dict      =       np.array(image_mask.convert("L"     )     ).astype(np.floataa     ) / 255.0

								assert image.shape[0:1] == image_mask.shape[0:1], "image and image_mask must have the same image size"
								lowerCamelCase_			:							str      =       -1.0  # set as masked pixel
								lowerCamelCase_			:							Any      =       np.expand_dims(a_				,							0     ).transpose(0				,							3				,							1				,							2     )
								lowerCamelCase_			:							List[Any]      =       torch.from_numpy(a_     )
								return image

							lowerCamelCase_			:							str      =       make_inpaint_condition(a_				,							a_     )

							lowerCamelCase_			:							int      =       pipe(
							    a_				,							image=a_				,							mask_image=a_				,							control_image=a_				,							guidance_scale=9.0				,							eta=1.0				,							generator=a_				,							num_inference_steps=20				,							output_type="np"				,							)
							lowerCamelCase_			:							Dict      =       output.images[0]

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

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





from __future__ import annotations





class    lowerCAmelCase__				:






						"""simple docstring"""




						def __init__(   self				,							a_     ):
							lowerCamelCase_			:							List[str]      =       value
							lowerCamelCase_			:							Node | None      =       None
							lowerCamelCase_			:							Node | None      =       None





class    lowerCAmelCase__				:






						"""simple docstring"""
						__UpperCAmelCase   :		int




						def __init__(   self     ):
							lowerCamelCase_			:							str      =       0




						def 					_UpperCamelCase			(   self				,							a_				,							a_     ):
							if node is None:
								return

							if path_sum == self.target:
								self.paths += 1

							if node.left:
								self.depth_first_search(node.left				,							path_sum + node.left.value     )
							if node.right:
								self.depth_first_search(node.right				,							path_sum + node.right.value     )




						def 					_UpperCamelCase			(   self				,							a_				,							a_ = None     ):
							if node is None:
								return 0
							if target is not None:
								lowerCamelCase_			:							List[str]      =       target

							self.depth_first_search(a_				,							node.value     )
							self.path_sum(node.left     )
							self.path_sum(node.right     )

							return self.paths


if __name__ == "__main__":
							import doctest

							doctest.testmod()

