



# 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        lowerCAmelCase__					(  _lowerCamelCase ,				_lowerCamelCase ,				_lowerCamelCase ,				unittest.TestCase       ):
				A_   :     Optional[Any]								=     StableDiffusionControlNetImgaImgPipeline
				A_   :     Union[str, Any]								=     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_   :     str								=     IMAGE_TO_IMAGE_IMAGE_PARAMS


				def __UpperCamelCase						(							self      :  Optional[Any]       )		->     List[str]:
											torch.manual_seed(0       )
											A      						=    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       )
											A      						=    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       )
											A      						=    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       )
											A      						=    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       )
											A      						=    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						, )
											A      						=    CLIPTextModel(__UpperCamelCase       )
											A      						=    CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'       )

											A      						=    {
											    '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      :  str						, __UpperCamelCase      :  Dict						, __UpperCamelCase      :  Union[str, Any]=0       )		->     Union[str, Any]:
											if str(__UpperCamelCase       ).startswith('mps'       ):
																		A      						=    torch.manual_seed(__UpperCamelCase       )
											else:
																		A      						=    torch.Generator(device=__UpperCamelCase       ).manual_seed(__UpperCamelCase       )

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


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





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


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

											A      						=    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       )
											A      						=    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       )
											A      						=    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       )
											A      						=    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       )
											A      						=    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						, )
											A      						=    CLIPTextModel(__UpperCamelCase       )
											A      						=    CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'       )

											A      						=    MultiControlNetModel([controlneta, controlneta]       )

											A      						=    {
											    '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      :  Union[str, Any]						, __UpperCamelCase      :  List[Any]						, __UpperCamelCase      :  Optional[int]=0       )		->     Optional[Any]:
											if str(__UpperCamelCase       ).startswith('mps'       ):
																		A      						=    torch.manual_seed(__UpperCamelCase       )
											else:
																		A      						=    torch.Generator(device=__UpperCamelCase       ).manual_seed(__UpperCamelCase       )

											A      						=    2

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

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

											A      						=    1_0.0
											A      						=    4

											A      						=    self.get_dummy_inputs(__UpperCamelCase       )
											A      						=    steps
											A      						=    scale
											A      						=    pipe(**__UpperCamelCase       )[0]

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

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

											A      						=    self.get_dummy_inputs(__UpperCamelCase       )
											A      						=    steps
											A      						=    scale
											A      						=    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 __UpperCamelCase						(							self      :  Dict       )		->     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      :  Optional[Any]       )		->     List[str]:
											self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3       )


				def __UpperCamelCase						(							self      :  Any       )		->     List[Any]:
											self._test_inference_batch_single_identical(expected_max_diff=2e-3       )


				def __UpperCamelCase						(							self      :  Optional[int]       )		->     int:
											A      						=    self.get_dummy_components()
											A      						=    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        lowerCAmelCase__					(  unittest.TestCase       ):


				def __UpperCamelCase						(							self      :  Optional[Any]       )		->     List[str]:
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()


				def __UpperCamelCase						(							self      :  List[Any]       )		->     str:
											A      						=    ControlNetModel.from_pretrained('lllyasviel/sd-controlnet-canny'       )

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

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

											A      						=    pipe(
											    __UpperCamelCase						, __UpperCamelCase						, control_image=__UpperCamelCase						, generator=__UpperCamelCase						, output_type='np'						, num_inference_steps=50						, strength=0.6						, )

											A      						=    output.images[0]

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

											A      						=    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"""
# limitations under the License.

# 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 .pipelines import DiffusionPipeline, ImagePipelineOutput  # noqa: F401
from .utils import deprecate


deprecate(
    """pipelines_utils""",
    """0.22.0""",
    """Importing `DiffusionPipeline` or `ImagePipelineOutput` from diffusers.pipeline_utils is deprecated. Please import from diffusers.pipelines.pipeline_utils instead.""",
    standard_warn=False,
    stacklevel=3,
)

