
"""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 				lowerCAmelCase_				(      _lowercase		,					_lowercase		,					_lowercase		,					unittest.TestCase    ):



						'''simple docstring'''


						_lowerCamelCase:    Dict      =    StableDiffusionControlNetImgaImgPipeline
						_lowerCamelCase:    Optional[Any]      =    TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
						_lowerCamelCase:    List[str]      =    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
						_lowerCamelCase:    Optional[int]      =    IMAGE_TO_IMAGE_IMAGE_PARAMS.union({'''control_image'''}    )
						_lowerCamelCase:    Optional[Any]      =    IMAGE_TO_IMAGE_IMAGE_PARAMS


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Dict      )     ->   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      )
							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_00_85  ,beta_end=0.0_12  ,beta_schedule='scaled_linear'  ,clip_sample=A_  ,set_alpha_to_one=A_  ,)
							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=1000  ,)
							A						   =   CLIPTextModel(A_      )
							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 						_SCREAMING_SNAKE_CASE      ( self     :      str  ,A_     :      Any  ,A_     :      Union[str, Any]=0      )     ->   int:
							if str(A_      ).startswith('mps'      ):
								A						   =   torch.manual_seed(A_      )
							else:
								A						   =   torch.Generator(device=A_      ).manual_seed(A_      )

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




						def 						_SCREAMING_SNAKE_CASE      ( self     :      Tuple      )     ->   Optional[int]:
							self._test_inference_batch_single_identical(expected_max_diff=2e-3      )







class 				lowerCAmelCase_				(      _lowercase		,					_lowercase		,					unittest.TestCase    ):



						'''simple docstring'''


						_lowerCamelCase:    Tuple      =    StableDiffusionControlNetImgaImgPipeline
						_lowerCamelCase:    Dict      =    TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
						_lowerCamelCase:    int      =    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
						_lowerCamelCase:    Optional[int]      =    frozenset([]    )  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Any      )     ->   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      )

							def init_weights(A_     :      Tuple      ):
								if isinstance(A_  ,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(A_      )

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

							torch.manual_seed(0      )
							A						   =   DDIMScheduler(
							    beta_start=0.0_00_85  ,beta_end=0.0_12  ,beta_schedule='scaled_linear'  ,clip_sample=A_  ,set_alpha_to_one=A_  ,)
							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=1000  ,)
							A						   =   CLIPTextModel(A_      )
							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 						_SCREAMING_SNAKE_CASE      ( self     :      List[Any]  ,A_     :      Tuple  ,A_     :      Optional[int]=0      )     ->   str:
							if str(A_      ).startswith('mps'      ):
								A						   =   torch.manual_seed(A_      )
							else:
								A						   =   torch.Generator(device=A_      ).manual_seed(A_      )

							A						   =   2

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

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

							A						   =   10.0
							A						   =   4

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

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

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

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


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Union[str, Any]      )     ->   Optional[int]:
							self._test_inference_batch_single_identical(expected_max_diff=2e-3      )




						def 						_SCREAMING_SNAKE_CASE      ( self     :      Union[str, Any]      )     ->   Any:
							A						   =   self.get_dummy_components()
							A						   =   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 				lowerCAmelCase_				(      unittest.TestCase    ):



						'''simple docstring'''


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




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

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

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


from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_flax_available,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
    is_vision_available,
)


_lowercase							   =				{
    '''configuration_clip''': [
        '''CLIP_PRETRAINED_CONFIG_ARCHIVE_MAP''',
        '''CLIPConfig''',
        '''CLIPOnnxConfig''',
        '''CLIPTextConfig''',
        '''CLIPVisionConfig''',
    ],
    '''processing_clip''': ['''CLIPProcessor'''],
    '''tokenization_clip''': ['''CLIPTokenizer'''],
}

try:
		if not is_tokenizers_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_lowercase							   =				['''CLIPTokenizerFast''']

try:
		if not is_vision_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_lowercase							   =				['''CLIPFeatureExtractor''']
		_lowercase							   =				['''CLIPImageProcessor''']

try:
		if not is_torch_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_lowercase							   =				[
		    '''CLIP_PRETRAINED_MODEL_ARCHIVE_LIST''',
		    '''CLIPModel''',
		    '''CLIPPreTrainedModel''',
		    '''CLIPTextModel''',
		    '''CLIPTextModelWithProjection''',
		    '''CLIPVisionModel''',
		    '''CLIPVisionModelWithProjection''',
		]

try:
		if not is_tf_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_lowercase							   =				[
		    '''TF_CLIP_PRETRAINED_MODEL_ARCHIVE_LIST''',
		    '''TFCLIPModel''',
		    '''TFCLIPPreTrainedModel''',
		    '''TFCLIPTextModel''',
		    '''TFCLIPVisionModel''',
		]

try:
		if not is_flax_available():
				raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
		pass
else:
		_lowercase							   =				[
		    '''FlaxCLIPModel''',
		    '''FlaxCLIPPreTrainedModel''',
		    '''FlaxCLIPTextModel''',
		    '''FlaxCLIPTextPreTrainedModel''',
		    '''FlaxCLIPVisionModel''',
		    '''FlaxCLIPVisionPreTrainedModel''',
		]


if TYPE_CHECKING:
		from .configuration_clip import (
		    CLIP_PRETRAINED_CONFIG_ARCHIVE_MAP,
		    CLIPConfig,
		    CLIPOnnxConfig,
		    CLIPTextConfig,
		    CLIPVisionConfig,
		)
		from .processing_clip import CLIPProcessor
		from .tokenization_clip import CLIPTokenizer

		try:
				if not is_tokenizers_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .tokenization_clip_fast import CLIPTokenizerFast

		try:
				if not is_vision_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .feature_extraction_clip import CLIPFeatureExtractor
				from .image_processing_clip import CLIPImageProcessor

		try:
				if not is_torch_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_clip import (
				    CLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
				    CLIPModel,
				    CLIPPreTrainedModel,
				    CLIPTextModel,
				    CLIPTextModelWithProjection,
				    CLIPVisionModel,
				    CLIPVisionModelWithProjection,
				)

		try:
				if not is_tf_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_tf_clip import (
				    TF_CLIP_PRETRAINED_MODEL_ARCHIVE_LIST,
				    TFCLIPModel,
				    TFCLIPPreTrainedModel,
				    TFCLIPTextModel,
				    TFCLIPVisionModel,
				)

		try:
				if not is_flax_available():
						raise OptionalDependencyNotAvailable()
		except OptionalDependencyNotAvailable:
				pass
		else:
				from .modeling_flax_clip import (
				    FlaxCLIPModel,
				    FlaxCLIPPreTrainedModel,
				    FlaxCLIPTextModel,
				    FlaxCLIPTextPreTrainedModel,
				    FlaxCLIPVisionModel,
				    FlaxCLIPVisionPreTrainedModel,
				)


else:
		import sys

		_lowercase							   =				_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)