
"""simple docstring"""

# 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 	_lowercase       (    __UpperCAmelCase    ,	__UpperCAmelCase    ,	__UpperCAmelCase    ,	unittest.TestCase		):
	lowercase_				=     StableDiffusionControlNetImgaImgPipeline
	lowercase_				=     TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
	lowercase_				=     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
	lowercase_				=     IMAGE_TO_IMAGE_IMAGE_PARAMS.union({'control_image'}		)
	lowercase_				=     IMAGE_TO_IMAGE_IMAGE_PARAMS




	def       _UpperCamelCase   (			self )		->							Optional[Any]:
			torch.manual_seed(0 )
			lowerCamelCase					:      str          =   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					:      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)		,			)
			torch.manual_seed(0 )
			lowerCamelCase					:      List[Any]          =   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					:      Tuple          =   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[Any]          =   CLIPTextConfig(
			    bos_token_id=0		,			eos_token_id=2		,			hidden_size=32		,			intermediate_size=37		,			layer_norm_eps=1E-0_5		,			num_attention_heads=4		,			num_hidden_layers=5		,			pad_token_id=1		,			vocab_size=1000		,			)
			lowerCamelCase					:      Any          =   CLIPTextModel(_a )
			lowerCamelCase					:      int          =   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		,			UpperCAmelCase_		,			UpperCAmelCase_=0 )		->							str:
			if str(_a ).startswith('mps' ):
					lowerCamelCase					:      Any          =   torch.manual_seed(_a )
			else:
					lowerCamelCase					:      Optional[int]          =   torch.Generator(device=_a ).manual_seed(_a )

			lowerCamelCase					:      List[str]          =   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					:      Any          =   floats_tensor(control_image.shape		,			rng=random.Random(_a ) ).to(_a )
			lowerCamelCase					:      Optional[Any]          =   image.cpu().permute(0		,			2		,			3		,			1 )[0]
			lowerCamelCase					:      int          =   Image.fromarray(np.uinta(_a ) ).convert('RGB' ).resize((64, 64) )
			lowerCamelCase					:      List[str]          =   {
			    "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       _UpperCamelCase   (			self )		->							Union[str, Any]:
			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 )		->							Tuple:
			self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3 )




	def       _UpperCamelCase   (			self )		->							Optional[Any]:
			self._test_inference_batch_single_identical(expected_max_diff=2E-3 )






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




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

			lowerCamelCase					:      List[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(_a )

			torch.manual_seed(0 )
			lowerCamelCase					:      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(_a )

			torch.manual_seed(0 )
			lowerCamelCase					:      str          =   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					:      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					:      Union[str, Any]          =   CLIPTextConfig(
			    bos_token_id=0		,			eos_token_id=2		,			hidden_size=32		,			intermediate_size=37		,			layer_norm_eps=1E-0_5		,			num_attention_heads=4		,			num_hidden_layers=5		,			pad_token_id=1		,			vocab_size=1000		,			)
			lowerCamelCase					:      Union[str, Any]          =   CLIPTextModel(_a )
			lowerCamelCase					:      Any          =   CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

			lowerCamelCase					:      Union[str, Any]          =   MultiControlNetModel([controlneta, controlneta] )

			lowerCamelCase					:      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		,			UpperCAmelCase_		,			UpperCAmelCase_=0 )		->							Optional[int]:
			if str(_a ).startswith('mps' ):
					lowerCamelCase					:      List[str]          =   torch.manual_seed(_a )
			else:
					lowerCamelCase					:      str          =   torch.Generator(device=_a ).manual_seed(_a )

			lowerCamelCase					:      Dict          =   2

			lowerCamelCase					:      List[Any]          =   [
			    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					:      Union[str, Any]          =   floats_tensor(control_image[0].shape		,			rng=random.Random(_a ) ).to(_a )
			lowerCamelCase					:      Union[str, Any]          =   image.cpu().permute(0		,			2		,			3		,			1 )[0]
			lowerCamelCase					:      Any          =   Image.fromarray(np.uinta(_a ) ).convert('RGB' ).resize((64, 64) )
			lowerCamelCase					:      List[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,
			    "control_image": control_image,
			}

			return inputs




	def       _UpperCamelCase   (			self )		->							Any:
			lowerCamelCase					:      Dict          =   self.get_dummy_components()
			lowerCamelCase					:      int          =   self.pipeline_class(**_a )
			pipe.to(_a )

			lowerCamelCase					:      Dict          =   10.0
			lowerCamelCase					:      str          =   4

			lowerCamelCase					:      List[Any]          =   self.get_dummy_inputs(_a )
			lowerCamelCase					:      Any          =   steps
			lowerCamelCase					:      Union[str, Any]          =   scale
			lowerCamelCase					:      Dict          =   pipe(**_a )[0]

			lowerCamelCase					:      Union[str, Any]          =   self.get_dummy_inputs(_a )
			lowerCamelCase					:      Any          =   steps
			lowerCamelCase					:      str          =   scale
			lowerCamelCase					:      str          =   pipe(**_a		,			control_guidance_start=0.1		,			control_guidance_end=0.2 )[0]

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

			lowerCamelCase					:      Optional[Any]          =   self.get_dummy_inputs(_a )
			lowerCamelCase					:      Optional[Any]          =   steps
			lowerCamelCase					:      Optional[Any]          =   scale
			lowerCamelCase					:      Tuple          =   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 )		->							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       _UpperCamelCase   (			self )		->							List[Any]:
			self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3 )




	def       _UpperCamelCase   (			self )		->							Optional[Any]:
			self._test_inference_batch_single_identical(expected_max_diff=2E-3 )




	def       _UpperCamelCase   (			self )		->							Optional[Any]:
			lowerCamelCase					:      List[Any]          =   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 	_lowercase       (    unittest.TestCase		):




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




	def       _UpperCamelCase   (			self )		->							Union[str, Any]:
			lowerCamelCase					:      Tuple          =   ControlNetModel.from_pretrained('lllyasviel/sd-controlnet-canny' )

			lowerCamelCase					:      List[str]          =   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					:      List[Any]          =   torch.Generator(device='cpu' ).manual_seed(0 )
			lowerCamelCase					:      Tuple          =   "evil space-punk bird"
			lowerCamelCase					:      Optional[int]          =   load_image(
			    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png' ).resize((512, 512) )
			lowerCamelCase					:      Any          =   load_image(
			    'https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png' ).resize((512, 512) )

			lowerCamelCase					:      Any          =   pipe(
			    _a		,			_a		,			control_image=_a		,			generator=_a		,			output_type='np'		,			num_inference_steps=50		,			strength=0.6		,			)

			lowerCamelCase					:      Optional[Any]          =   output.images[0]

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

			lowerCamelCase					:      Optional[int]          =   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 copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging


_A						=			logging.get_logger(__name__)






class 	_lowercase       (    __UpperCAmelCase		):
	lowercase_				=     'encoder-decoder'
	lowercase_				=     True




	def __init__(			self		,			**UpperCAmelCase_ )		->							str:
			super().__init__(**UpperCAmelCase_ )
			assert (
			    "encoder" in kwargs and "decoder" in kwargs
			), "Config has to be initialized with encoder and decoder config"
			lowerCamelCase					:      List[Any]          =   kwargs.pop('encoder' )
			lowerCamelCase					:      Optional[int]          =   encoder_config.pop('model_type' )
			lowerCamelCase					:      str          =   kwargs.pop('decoder' )
			lowerCamelCase					:      Dict          =   decoder_config.pop('model_type' )

			from ..auto.configuration_auto import AutoConfig

			lowerCamelCase					:      int          =   AutoConfig.for_model(UpperCAmelCase_		,			**UpperCAmelCase_ )
			lowerCamelCase					:      List[str]          =   AutoConfig.for_model(UpperCAmelCase_		,			**UpperCAmelCase_ )
			lowerCamelCase					:      List[str]          =   True




	@classmethod
	def       _UpperCamelCase   (			cls		,			UpperCAmelCase_		,			UpperCAmelCase_		,			**UpperCAmelCase_ )		->							PretrainedConfig:
			logger.info('Set `config.is_decoder=True` and `config.add_cross_attention=True` for decoder_config' )
			lowerCamelCase					:      str          =   True
			lowerCamelCase					:      Optional[Any]          =   True

			return cls(encoder=encoder_config.to_dict()		,			decoder=decoder_config.to_dict()		,			**UpperCAmelCase_ )




	def       _UpperCamelCase   (			self )		->							Union[str, Any]:
			lowerCamelCase					:      Union[str, Any]          =   copy.deepcopy(self.__dict__ )
			lowerCamelCase					:      Union[str, Any]          =   self.encoder.to_dict()
			lowerCamelCase					:      List[Any]          =   self.decoder.to_dict()
			lowerCamelCase					:      Tuple          =   self.__class__.model_type
			return output






