





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       (  __snake_case      ,   __snake_case      ,   __snake_case      ,   unittest.TestCase      ):
				__A								=			StableDiffusionControlNetImgaImgPipeline
				__A								=			TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
				__A								=			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
				__A								=			IMAGE_TO_IMAGE_IMAGE_PARAMS.union({"""control_image"""}      )
				__A								=			IMAGE_TO_IMAGE_IMAGE_PARAMS


				def _snake_case    (						self							):
								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.0_0_0_8_5				,	beta_end=0.0_1_2				,	beta_schedule="""scaled_linear"""				,	clip_sample=UpperCamelCase__				,	set_alpha_to_one=UpperCamelCase__				,	)
								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(UpperCamelCase__							)
								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 _snake_case    (						self				,	UpperCAmelCase_				,	UpperCAmelCase_=0							):


								if str(UpperCamelCase__							).startswith("""mps"""							):
												lowerCamelCase			=torch.manual_seed(UpperCamelCase__							)
								else:
												lowerCamelCase			=torch.Generator(device=UpperCamelCase__							).manual_seed(UpperCamelCase__							)

								lowerCamelCase			=2
								lowerCamelCase			=randn_tensor(
								    (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)				,	generator=UpperCamelCase__				,	device=torch.device(UpperCamelCase__							)				,	)
								lowerCamelCase			=floats_tensor(control_image.shape				,	rng=random.Random(UpperCamelCase__							)							).to(UpperCamelCase__							)
								lowerCamelCase			=image.cpu().permute(0				,	2				,	3				,	1							)[0]
								lowerCamelCase			=Image.fromarray(np.uinta(UpperCamelCase__							)							).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 _snake_case    (						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 _snake_case    (						self							):
								self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3							)


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




class  __A       (  __snake_case      ,   __snake_case      ,   unittest.TestCase      ):
				__A								=			StableDiffusionControlNetImgaImgPipeline
				__A								=			TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
				__A								=			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
				__A								=			frozenset([]      )  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess


				def _snake_case    (						self							):
								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(UpperCAmelCase_							):
												if isinstance(UpperCamelCase__				,	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(UpperCamelCase__							)

								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(UpperCamelCase__							)

								torch.manual_seed(0							)
								lowerCamelCase			=DDIMScheduler(
								    beta_start=0.0_0_0_8_5				,	beta_end=0.0_1_2				,	beta_schedule="""scaled_linear"""				,	clip_sample=UpperCamelCase__				,	set_alpha_to_one=UpperCamelCase__				,	)
								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(UpperCamelCase__							)
								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 _snake_case    (						self				,	UpperCAmelCase_				,	UpperCAmelCase_=0							):


								if str(UpperCamelCase__							).startswith("""mps"""							):
												lowerCamelCase			=torch.manual_seed(UpperCamelCase__							)
								else:
												lowerCamelCase			=torch.Generator(device=UpperCamelCase__							).manual_seed(UpperCamelCase__							)

								lowerCamelCase			=2

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

								lowerCamelCase			=floats_tensor(control_image[0].shape				,	rng=random.Random(UpperCamelCase__							)							).to(UpperCamelCase__							)
								lowerCamelCase			=image.cpu().permute(0				,	2				,	3				,	1							)[0]
								lowerCamelCase			=Image.fromarray(np.uinta(UpperCamelCase__							)							).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 _snake_case    (						self							):
								lowerCamelCase			=self.get_dummy_components()
								lowerCamelCase			=self.pipeline_class(**UpperCamelCase__							)
								pipe.to(UpperCamelCase__							)

								lowerCamelCase			=1_0.0
								lowerCamelCase			=4

								lowerCamelCase			=self.get_dummy_inputs(UpperCamelCase__							)
								lowerCamelCase			=steps
								lowerCamelCase			=scale
								lowerCamelCase			=pipe(**UpperCamelCase__							)[0]

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

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

								lowerCamelCase			=self.get_dummy_inputs(UpperCamelCase__							)
								lowerCamelCase			=steps
								lowerCamelCase			=scale
								lowerCamelCase			=pipe(**UpperCamelCase__				,	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 _snake_case    (						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 _snake_case    (						self							):
								self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3							)


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


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




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


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


				def _snake_case    (						self							):
								lowerCamelCase			=ControlNetModel.from_pretrained("""lllyasviel/sd-controlnet-canny"""							)

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

								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(
								    UpperCamelCase__				,	UpperCamelCase__				,	control_image=UpperCamelCase__				,	generator=UpperCamelCase__				,	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












# this script reports modified .py files under the desired list of top-level sub-dirs passed as a list of arguments, e.g.:
#   python ./utils/get_modified_files.py utils src tests examples
#
# it uses git to find the forking point and which files were modified - i.e. files not under git won't be considered
# since the output of this script is fed into Makefile commands it doesn't print a newline after the results

import re
import subprocess
import sys


UpperCAmelCase__  :     Union[str, Any]   			=subprocess.check_output('''git merge-base main HEAD'''.split()).decode('''utf-8''')
UpperCAmelCase__  :     List[str]   			=(
    subprocess.check_output(F"git diff --diff-filter=d --name-only {fork_point_sha}".split()).decode('''utf-8''').split()
)

UpperCAmelCase__  :     Tuple   			='''|'''.join(sys.argv[1:])
UpperCAmelCase__  :     List[str]   			=re.compile(rF"^({joined_dirs}).*?\.py$")

UpperCAmelCase__  :     Any   			=[x for x in modified_files if regex.match(x)]
print(''' '''.join(relevant_modified_files), end='''''')





