

# 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 				SCREAMING_SNAKE_CASE__ (__snake_case		,	__snake_case		,	__snake_case		,	unittest.TestCase					):
    __lowerCamelCase						:	List[Any]														=					StableDiffusionControlNetImgaImgPipeline
    __lowerCamelCase						:	Optional[Any]														=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
    __lowerCamelCase						:	Dict														=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
    __lowerCamelCase						:	List[str]														=					IMAGE_TO_IMAGE_IMAGE_PARAMS.union({"""control_image"""}					)
    __lowerCamelCase						:	List[Any]														=					IMAGE_TO_IMAGE_IMAGE_PARAMS





    def 				snake_case_    (				self):
         torch.manual_seed(0)
         lowercase__   :       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)
         lowercase__   :       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)						,       )
         torch.manual_seed(0)
         lowercase__   :       Dict				=	DDIMScheduler(
             beta_start=0.00_085						,       beta_end=0.012						,       beta_schedule='scaled_linear'						,       clip_sample=a						,       set_alpha_to_one=a						,       )
         torch.manual_seed(0)
         lowercase__   :       List[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)
         lowercase__   :       Tuple				=	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						,       )
         lowercase__   :       Tuple				=	CLIPTextModel(a)
         lowercase__   :       Optional[int]				=	CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

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

         lowercase__   :       Optional[Any]				=	2
         lowercase__   :       Optional[int]				=	randn_tensor(
             (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)						,       generator=a						,       device=torch.device(a)						,       )
         lowercase__   :       Tuple				=	floats_tensor(control_image.shape						,       rng=random.Random(a)).to(a)
         lowercase__   :       Any				=	image.cpu().permute(0						,       2						,       3						,       1)[0]
         lowercase__   :       List[str]				=	Image.fromarray(np.uinta(a)).convert('RGB').resize((64, 64))
         lowercase__   :       Dict				=	{
             '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 				SCREAMING_SNAKE_CASE__ (__snake_case		,	__snake_case		,	unittest.TestCase					):
    __lowerCamelCase						:	Dict														=					StableDiffusionControlNetImgaImgPipeline
    __lowerCamelCase						:	str														=					TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
    __lowerCamelCase						:	int														=					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
    __lowerCamelCase						:	str														=					frozenset([]					)  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess





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

         lowercase__   :       Dict				=	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)
         lowercase__   :       Optional[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)
         lowercase__   :       List[str]				=	DDIMScheduler(
             beta_start=0.00_085						,       beta_end=0.012						,       beta_schedule='scaled_linear'						,       clip_sample=a						,       set_alpha_to_one=a						,       )
         torch.manual_seed(0)
         lowercase__   :       Optional[int]				=	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)
         lowercase__   :       Union[str, Any]				=	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						,       )
         lowercase__   :       Optional[Any]				=	CLIPTextModel(a)
         lowercase__   :       Optional[Any]				=	CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

         lowercase__   :       Any				=	MultiControlNetModel([controlneta, controlneta])

         lowercase__   :       Any				=	{
             '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						,       a						,       a=0):
         if str(a).startswith('mps'):
              lowercase__   :       int				=	torch.manual_seed(a)
         else:
              lowercase__   :       Any				=	torch.Generator(device=a).manual_seed(a)

         lowercase__   :       Tuple				=	2

         lowercase__   :       Optional[int]				=	[
             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)						,       ),
         ]

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

         lowercase__   :       int				=	10.0
         lowercase__   :       Any				=	4

         lowercase__   :       List[Any]				=	self.get_dummy_inputs(a)
         lowercase__   :       Optional[Any]				=	steps
         lowercase__   :       Optional[Any]				=	scale
         lowercase__   :       Dict				=	pipe(**a)[0]

         lowercase__   :       Optional[int]				=	self.get_dummy_inputs(a)
         lowercase__   :       Tuple				=	steps
         lowercase__   :       Any				=	scale
         lowercase__   :       Union[str, Any]				=	pipe(**a						,       control_guidance_start=0.1						,       control_guidance_end=0.2)[0]

         lowercase__   :       int				=	self.get_dummy_inputs(a)
         lowercase__   :       Optional[int]				=	steps
         lowercase__   :       Dict				=	scale
         lowercase__   :       Tuple				=	pipe(**a						,       control_guidance_start=[0.1, 0.3]						,       control_guidance_end=[0.2, 0.7])[0]

         lowercase__   :       Union[str, Any]				=	self.get_dummy_inputs(a)
         lowercase__   :       Union[str, Any]				=	steps
         lowercase__   :       Dict				=	scale
         lowercase__   :       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 				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):
         lowercase__   :       Union[str, Any]				=	self.get_dummy_components()
         lowercase__   :       List[Any]				=	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 				SCREAMING_SNAKE_CASE__ (unittest.TestCase					):





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





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

         lowercase__   :       Dict				=	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)

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

         lowercase__   :       Dict				=	pipe(
             a						,       a						,       control_image=a						,       generator=a						,       output_type='np'						,       num_inference_steps=50						,       strength=0.6						,       )

         lowercase__   :       List[Any]				=	output.images[0]

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

         lowercase__   :       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




import warnings
from typing import List, Optional, Union

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding, PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType


class 				SCREAMING_SNAKE_CASE__ (__snake_case					):
    __lowerCamelCase						:	Tuple														=					["""image_processor""", """tokenizer"""]
    __lowerCamelCase						:	int														=					"""ViltImageProcessor"""
    __lowerCamelCase						:	Union[str, Any]														=					("""BertTokenizer""", """BertTokenizerFast""")





    def __init__(				self						,       a=None						,       a=None						,       **a):
         lowercase__   :       Any				=	None
         if "feature_extractor" in kwargs:
              warnings.warn(
                  'The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'
                  ' instead.'						,       a						,       )
              lowercase__   :       Union[str, Any]				=	kwargs.pop('feature_extractor')

         lowercase__   :       List[Any]				=	image_processor if image_processor is not None else feature_extractor
         if image_processor is None:
              raise ValueError('You need to specify an `image_processor`.')
         if tokenizer is None:
              raise ValueError('You need to specify a `tokenizer`.')

         super().__init__(a						,       a)
         lowercase__   :       str				=	self.image_processor





    def __call__(				self						,       a						,       a = None						,       a = True						,       a = False						,       a = None						,       a = None						,       a = 0						,       a = None						,       a = None						,       a = None						,       a = False						,       a = False						,       a = False						,       a = False						,       a = True						,       a = None						,       **a						,       ):
         lowercase__   :       Optional[Any]				=	self.tokenizer(
             text=a						,       add_special_tokens=a						,       padding=a						,       truncation=a						,       max_length=a						,       stride=a						,       pad_to_multiple_of=a						,       return_token_type_ids=a						,       return_attention_mask=a						,       return_overflowing_tokens=a						,       return_special_tokens_mask=a						,       return_offsets_mapping=a						,       return_length=a						,       verbose=a						,       return_tensors=a						,       **a						,       )
         # add pixel_values + pixel_mask
         lowercase__   :       str				=	self.image_processor(a						,       return_tensors=a)
         encoding.update(a)

         return encoding





    def 				snake_case_    (				self						,       *a						,       **a):
         return self.tokenizer.batch_decode(*a						,       **a)





    def 				snake_case_    (				self						,       *a						,       **a):
         return self.tokenizer.decode(*a						,       **a)





    @property
    def 				snake_case_    (				self):
         lowercase__   :       int				=	self.tokenizer.model_input_names
         lowercase__   :       List[str]				=	self.image_processor.model_input_names
         return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names))





    @property
    def 				snake_case_    (				self):
         warnings.warn(
             '`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'						,       a						,       )
         return self.image_processor_class





    @property
    def 				snake_case_    (				self):
         warnings.warn(
             '`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'						,       a						,       )
         return self.image_processor

