



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 						lowercase		(					__UpperCamelCase					,       __UpperCamelCase					,       __UpperCamelCase					,       unittest.TestCase):






	"""simple docstring"""


	a__ :   int						  =				StableDiffusionControlNetImgaImgPipeline
	a__ :   int						  =				TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width"}
	a__ :   Union[str, Any]						  =				TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
	a__ :   Optional[Any]						  =				IMAGE_TO_IMAGE_IMAGE_PARAMS.union({"control_image"})
	a__ :   Optional[Any]						  =				IMAGE_TO_IMAGE_IMAGE_PARAMS




	def     _SCREAMING_SNAKE_CASE						(      self      :		Dict	)     ->					List[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	)
							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=snake_case__		,    set_alpha_to_one=snake_case__		,    )
							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-05		,    num_attention_heads=4		,    num_hidden_layers=5		,    pad_token_id=1		,    vocab_size=1_000		,    )
							UpperCAmelCase_=  CLIPTextModel(snake_case__	)
							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     _SCREAMING_SNAKE_CASE						(      self      :		Optional[int]		,    __UpperCAmelCase      :		Optional[Any]		,    __UpperCAmelCase      :		List[str]=0	)     ->					Union[str, Any]:
							if str(snake_case__	).startswith("""mps"""	):
													UpperCAmelCase_=  torch.manual_seed(snake_case__	)
							else:
													UpperCAmelCase_=  torch.Generator(device=snake_case__	).manual_seed(snake_case__	)

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




	def     _SCREAMING_SNAKE_CASE						(      self      :		Tuple	)     ->					Dict:
							self._test_inference_batch_single_identical(expected_max_diff=2E-3	)






class 						lowercase		(					__UpperCamelCase					,       __UpperCamelCase					,       unittest.TestCase):






	"""simple docstring"""


	a__ :   Any						  =				StableDiffusionControlNetImgaImgPipeline
	a__ :   str						  =				TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width"}
	a__ :   Union[str, Any]						  =				TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
	a__ :   Optional[int]						  =				frozenset([])  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess




	def     _SCREAMING_SNAKE_CASE						(      self      :		str	)     ->					Optional[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(__UpperCAmelCase      :		List[Any]	):
													if isinstance(snake_case__		,    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(snake_case__	)

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

							torch.manual_seed(0	)
							UpperCAmelCase_=  DDIMScheduler(
							    beta_start=0.00_085		,    beta_end=0.012		,    beta_schedule="""scaled_linear"""		,    clip_sample=snake_case__		,    set_alpha_to_one=snake_case__		,    )
							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-05		,    num_attention_heads=4		,    num_hidden_layers=5		,    pad_token_id=1		,    vocab_size=1_000		,    )
							UpperCAmelCase_=  CLIPTextModel(snake_case__	)
							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     _SCREAMING_SNAKE_CASE						(      self      :		str		,    __UpperCAmelCase      :		Any		,    __UpperCAmelCase      :		Any=0	)     ->					List[str]:
							if str(snake_case__	).startswith("""mps"""	):
													UpperCAmelCase_=  torch.manual_seed(snake_case__	)
							else:
													UpperCAmelCase_=  torch.Generator(device=snake_case__	).manual_seed(snake_case__	)

							UpperCAmelCase_=  2

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

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

							UpperCAmelCase_=  10.0
							UpperCAmelCase_=  4

							UpperCAmelCase_=  self.get_dummy_inputs(snake_case__	)
							UpperCAmelCase_=  steps
							UpperCAmelCase_=  scale
							UpperCAmelCase_=  pipe(**snake_case__	)[0]

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

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

							UpperCAmelCase_=  self.get_dummy_inputs(snake_case__	)
							UpperCAmelCase_=  steps
							UpperCAmelCase_=  scale
							UpperCAmelCase_=  pipe(**snake_case__		,    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     _SCREAMING_SNAKE_CASE						(      self      :		Dict	)     ->					str:
							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     _SCREAMING_SNAKE_CASE						(      self      :		Any	)     ->					Union[str, Any]:
							self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3	)




	def     _SCREAMING_SNAKE_CASE						(      self      :		List[str]	)     ->					Tuple:
							self._test_inference_batch_single_identical(expected_max_diff=2E-3	)




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






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






	"""simple docstring"""


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




	def     _SCREAMING_SNAKE_CASE						(      self      :		Any	)     ->					int:
							UpperCAmelCase_=  ControlNetModel.from_pretrained("""lllyasviel/sd-controlnet-canny"""	)

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

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





"""simple docstring"""




import time
from contextlib import contextmanager
from pathlib import Path

import pytest
import requests
from huggingface_hub.hf_api import HfApi, HfFolder


__lowerCAmelCase     :				str    				=    "__DUMMY_TRANSFORMERS_USER__"
__lowerCAmelCase     :				Dict    				=    "Dummy User"
__lowerCAmelCase     :				Optional[Any]    				=    "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"

__lowerCAmelCase     :				str    				=    "https://hub-ci.huggingface.co"
__lowerCAmelCase     :				int    				=    CI_HUB_ENDPOINT + "/datasets/{repo_id}/resolve/{revision}/{path}"
__lowerCAmelCase     :				Dict    				=    CI_HUB_ENDPOINT + "/{repo_id}/resolve/{revision}/{filename}"
__lowerCAmelCase     :				Optional[Any]    				=    Path("~/.huggingface/hub_ci_token").expanduser()







@pytest.fixture
def 				_UpperCAmelCase		(							lowerCamelCase__  ):





		"""simple docstring"""




		monkeypatch.setattr(
		    """huggingface_hub.file_download.HUGGINGFACE_CO_URL_TEMPLATE"""      ,	lowerCamelCase__  )







@pytest.fixture
def 				_UpperCAmelCase		(							lowerCamelCase__  ):





		"""simple docstring"""




		monkeypatch.setattr("""datasets.config.HF_ENDPOINT"""      ,	lowerCamelCase__  )
		monkeypatch.setattr("""datasets.config.HUB_DATASETS_URL"""      ,	lowerCamelCase__  )







@pytest.fixture
def 				_UpperCAmelCase		(							lowerCamelCase__  ):





		"""simple docstring"""




		monkeypatch.setattr("""huggingface_hub.hf_api.HfFolder.path_token"""      ,	lowerCamelCase__  )







@pytest.fixture
def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		HfFolder.save_token(lowerCamelCase__  )
		yield
		HfFolder.delete_token()







@pytest.fixture(scope="""session"""  )
def 				_UpperCAmelCase		(							):





		"""simple docstring"""




		return HfApi(endpoint=lowerCamelCase__  )







@pytest.fixture(scope="""session"""  )
def 				_UpperCAmelCase		(							lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = HfFolder.get_token()
		HfFolder.save_token(lowerCamelCase__  )
		yield CI_HUB_USER_TOKEN
		if previous_token is not None:
				HfFolder.save_token(lowerCamelCase__  )







@pytest.fixture
def 				_UpperCAmelCase		(							lowerCamelCase__  ):





		"""simple docstring"""




		def _cleanup_repo(lowerCamelCase__  ):
				hf_api.delete_repo(lowerCamelCase__      ,	token=lowerCamelCase__      ,	repo_type="""dataset"""  )

		return _cleanup_repo







@pytest.fixture
def 				_UpperCAmelCase		(							lowerCamelCase__  ):





		"""simple docstring"""




		@contextmanager
		def _temporary_repo(lowerCamelCase__  ):
				try:
						yield repo_id
				finally:
						cleanup_repo(lowerCamelCase__  )

		return _temporary_repo







@pytest.fixture(scope="""session"""  )
def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = f"""repo_txt_data-{int(time.time() * 10e3  )}"""
		lowerCAmelCase__   = f"""{CI_HUB_USER}/{repo_name}"""
		hf_api.create_repo(lowerCamelCase__      ,	token=lowerCamelCase__      ,	repo_type="""dataset"""      ,	private=lowerCamelCase__  )
		hf_api.upload_file(
		    token=lowerCamelCase__      ,	path_or_fileobj=str(lowerCamelCase__  )      ,	path_in_repo="""data/text_data.txt"""      ,	repo_id=lowerCamelCase__      ,	repo_type="""dataset"""      ,	)
		yield repo_id
		try:
				hf_api.delete_repo(lowerCamelCase__      ,	token=lowerCamelCase__      ,	repo_type="""dataset"""  )
		except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
				pass







@pytest.fixture()
def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		return hf_private_dataset_repo_txt_data_







@pytest.fixture(scope="""session"""  )
def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = f"""repo_zipped_txt_data-{int(time.time() * 10e3  )}"""
		lowerCAmelCase__   = f"""{CI_HUB_USER}/{repo_name}"""
		hf_api.create_repo(lowerCamelCase__      ,	token=lowerCamelCase__      ,	repo_type="""dataset"""      ,	private=lowerCamelCase__  )
		hf_api.upload_file(
		    token=lowerCamelCase__      ,	path_or_fileobj=str(lowerCamelCase__  )      ,	path_in_repo="""data.zip"""      ,	repo_id=lowerCamelCase__      ,	repo_type="""dataset"""      ,	)
		yield repo_id
		try:
				hf_api.delete_repo(lowerCamelCase__      ,	token=lowerCamelCase__      ,	repo_type="""dataset"""  )
		except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
				pass







@pytest.fixture()
def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		return hf_private_dataset_repo_zipped_txt_data_







@pytest.fixture(scope="""session"""  )
def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		lowerCAmelCase__   = f"""repo_zipped_img_data-{int(time.time() * 10e3  )}"""
		lowerCAmelCase__   = f"""{CI_HUB_USER}/{repo_name}"""
		hf_api.create_repo(lowerCamelCase__      ,	token=lowerCamelCase__      ,	repo_type="""dataset"""      ,	private=lowerCamelCase__  )
		hf_api.upload_file(
		    token=lowerCamelCase__      ,	path_or_fileobj=str(lowerCamelCase__  )      ,	path_in_repo="""data.zip"""      ,	repo_id=lowerCamelCase__      ,	repo_type="""dataset"""      ,	)
		yield repo_id
		try:
				hf_api.delete_repo(lowerCamelCase__      ,	token=lowerCamelCase__      ,	repo_type="""dataset"""  )
		except (requests.exceptions.HTTPError, ValueError):  # catch http error and token invalid error
				pass







@pytest.fixture()
def 				_UpperCAmelCase		(							lowerCamelCase__      ,	lowerCamelCase__      ,	lowerCamelCase__  ):





		"""simple docstring"""




		return hf_private_dataset_repo_zipped_img_data_




