

# 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       ( lowercase_			,						lowercase_			,						lowercase_			,						unittest.TestCase    ):
				__SCREAMING_SNAKE_CASE      :				List[Any]     =    StableDiffusionControlNetImgaImgPipeline
				__SCREAMING_SNAKE_CASE      :				Any     =    TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
				__SCREAMING_SNAKE_CASE      :				Any     =    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
				__SCREAMING_SNAKE_CASE      :				Tuple     =    IMAGE_TO_IMAGE_IMAGE_PARAMS.union({'''control_image'''}    )
				__SCREAMING_SNAKE_CASE      :				Tuple     =    IMAGE_TO_IMAGE_IMAGE_PARAMS




				def 					a							(      self ):
											torch.manual_seed(0 )
											snake_case_           =					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 )
											snake_case_           =					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 )
											snake_case_           =					DDIMScheduler(
											    beta_start=0.0_00_85   ,     beta_end=0.0_12   ,     beta_schedule='scaled_linear'   ,     clip_sample=snake_case   ,     set_alpha_to_one=snake_case   ,     )
											torch.manual_seed(0 )
											snake_case_           =					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 )
											snake_case_           =					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   ,     )
											snake_case_           =					CLIPTextModel(snake_case )
											snake_case_           =					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

											snake_case_           =					{
											    'unet': unet,
											    'controlnet': controlnet,
											    'scheduler': scheduler,
											    'vae': vae,
											    'text_encoder': text_encoder,
											    'tokenizer': tokenizer,
											    'safety_checker': None,
											    'feature_extractor': None,
											}
											return components




				def 					a							(      self   ,     snake_case   ,     snake_case=0 ):
											if str(snake_case ).startswith('mps' ):
																		snake_case_           =					torch.manual_seed(snake_case )
											else:
																		snake_case_           =					torch.Generator(device=snake_case ).manual_seed(snake_case )

											snake_case_           =					2
											snake_case_           =					randn_tensor(
											    (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)   ,     generator=snake_case   ,     device=torch.device(snake_case )   ,     )
											snake_case_           =					floats_tensor(control_image.shape   ,     rng=random.Random(snake_case ) ).to(snake_case )
											snake_case_           =					image.cpu().permute(0   ,     2   ,     3   ,     1 )[0]
											snake_case_           =					Image.fromarray(np.uinta(snake_case ) ).convert('RGB' ).resize((64, 64) )
											snake_case_           =					{
											    '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 					a							(      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 					a							(      self ):
											self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3 )




				def 					a							(      self ):
											self._test_inference_batch_single_identical(expected_max_diff=2e-3 )







class 				lowercase       ( lowercase_			,						lowercase_			,						unittest.TestCase    ):
				__SCREAMING_SNAKE_CASE      :				Optional[int]     =    StableDiffusionControlNetImgaImgPipeline
				__SCREAMING_SNAKE_CASE      :				str     =    TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
				__SCREAMING_SNAKE_CASE      :				Dict     =    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
				__SCREAMING_SNAKE_CASE      :				Optional[Any]     =    frozenset([]    )  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess




				def 					a							(      self ):
											torch.manual_seed(0 )
											snake_case_           =					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(snake_case ):
																		if isinstance(snake_case   ,     torch.nn.Convad ):
																									torch.nn.init.normal(m.weight )
																									m.bias.data.fill_(1.0 )

											snake_case_           =					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 )
											snake_case_           =					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 )
											snake_case_           =					DDIMScheduler(
											    beta_start=0.0_00_85   ,     beta_end=0.0_12   ,     beta_schedule='scaled_linear'   ,     clip_sample=snake_case   ,     set_alpha_to_one=snake_case   ,     )
											torch.manual_seed(0 )
											snake_case_           =					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 )
											snake_case_           =					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   ,     )
											snake_case_           =					CLIPTextModel(snake_case )
											snake_case_           =					CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip' )

											snake_case_           =					MultiControlNetModel([controlneta, controlneta] )

											snake_case_           =					{
											    'unet': unet,
											    'controlnet': controlnet,
											    'scheduler': scheduler,
											    'vae': vae,
											    'text_encoder': text_encoder,
											    'tokenizer': tokenizer,
											    'safety_checker': None,
											    'feature_extractor': None,
											}
											return components




				def 					a							(      self   ,     snake_case   ,     snake_case=0 ):
											if str(snake_case ).startswith('mps' ):
																		snake_case_           =					torch.manual_seed(snake_case )
											else:
																		snake_case_           =					torch.Generator(device=snake_case ).manual_seed(snake_case )

											snake_case_           =					2

											snake_case_           =					[
											    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 )   ,     ),
											]

											snake_case_           =					floats_tensor(control_image[0].shape   ,     rng=random.Random(snake_case ) ).to(snake_case )
											snake_case_           =					image.cpu().permute(0   ,     2   ,     3   ,     1 )[0]
											snake_case_           =					Image.fromarray(np.uinta(snake_case ) ).convert('RGB' ).resize((64, 64) )
											snake_case_           =					{
											    '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 					a							(      self ):
											snake_case_           =					self.get_dummy_components()
											snake_case_           =					self.pipeline_class(**snake_case )
											pipe.to(snake_case )

											snake_case_           =					10.0
											snake_case_           =					4

											snake_case_           =					self.get_dummy_inputs(snake_case )
											snake_case_           =					steps
											snake_case_           =					scale
											snake_case_           =					pipe(**snake_case )[0]

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

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

											snake_case_           =					self.get_dummy_inputs(snake_case )
											snake_case_           =					steps
											snake_case_           =					scale
											snake_case_           =					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 					a							(      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 					a							(      self ):
											self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3 )




				def 					a							(      self ):
											self._test_inference_batch_single_identical(expected_max_diff=2e-3 )




				def 					a							(      self ):
											snake_case_           =					self.get_dummy_components()
											snake_case_           =					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    ):




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




				def 					a							(      self ):
											snake_case_           =					ControlNetModel.from_pretrained('lllyasviel/sd-controlnet-canny' )

											snake_case_           =					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 )

											snake_case_           =					torch.Generator(device='cpu' ).manual_seed(0 )
											snake_case_           =					'evil space-punk bird'
											snake_case_           =					load_image(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png' ).resize((512, 512) )
											snake_case_           =					load_image(
											    'https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png' ).resize((512, 512) )

											snake_case_           =					pipe(
											    snake_case   ,     snake_case   ,     control_image=snake_case   ,     generator=snake_case   ,     output_type='np'   ,     num_inference_steps=50   ,     strength=0.6   ,     )

											snake_case_           =					output.images[0]

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

											snake_case_           =					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







# NOTE: This file is deprecated and will be removed in a future version.
# It only exists so that temporarely `from diffusers.pipelines import DiffusionPipeline` works

from ...utils import deprecate
from ..controlnet.pipeline_flax_controlnet import FlaxStableDiffusionControlNetPipeline  # noqa: F401


deprecate(
    """stable diffusion controlnet""",
    """0.22.0""",
    """Importing `FlaxStableDiffusionControlNetPipeline` from diffusers.pipelines.stable_diffusion.flax_pipeline_stable_diffusion_controlnet is deprecated. Please import `from diffusers import FlaxStableDiffusionControlNetPipeline` instead.""",
    standard_warn=False,
    stacklevel=3,
)




