


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






		'''simple docstring'''

		lowerCamelCase__				=StableDiffusionControlNetImgaImgPipeline
		lowerCamelCase__				=TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
		lowerCamelCase__				=TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
		lowerCamelCase__				=IMAGE_TO_IMAGE_IMAGE_PARAMS.union({'control_image'}				)
		lowerCamelCase__				=IMAGE_TO_IMAGE_IMAGE_PARAMS



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			torch.manual_seed(0					)
			__snake_case	:						List[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					)
			__snake_case	:						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)				,       )
			torch.manual_seed(0					)
			__snake_case	:						Any						=				DDIMScheduler(
			    beta_start=0.0_0085				,       beta_end=0.012				,       beta_schedule='''scaled_linear'''				,       clip_sample=a_				,       set_alpha_to_one=a_				,       )
			torch.manual_seed(0					)
			__snake_case	:						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					)
			__snake_case	:						List[str]						=				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=10_00				,       )
			__snake_case	:						Any						=				CLIPTextModel(a_					)
			__snake_case	:						List[str]						=				CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

			__snake_case	:						Tuple						=				{
			    '''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				,       a_				,       a_=0					):





			'''simple docstring'''

			if str(a_					).startswith('''mps'''					):
				__snake_case	:						int						=				torch.manual_seed(a_					)
			else:
				__snake_case	:						int						=				torch.Generator(device=a_					).manual_seed(a_					)

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





			'''simple docstring'''

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





			'''simple docstring'''

			self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3					)



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			self._test_inference_batch_single_identical(expected_max_diff=2E-3					)



class 			_UpperCAmelCase (							__snake_case,				__snake_case,				unittest.TestCase				):






		'''simple docstring'''

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



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			torch.manual_seed(0					)
			__snake_case	:						Optional[int]						=				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					)

			__snake_case	:						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					)
			__snake_case	:						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					)
			__snake_case	:						List[Any]						=				DDIMScheduler(
			    beta_start=0.0_0085				,       beta_end=0.012				,       beta_schedule='''scaled_linear'''				,       clip_sample=a_				,       set_alpha_to_one=a_				,       )
			torch.manual_seed(0					)
			__snake_case	:						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					)
			__snake_case	:						Optional[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=10_00				,       )
			__snake_case	:						Dict						=				CLIPTextModel(a_					)
			__snake_case	:						List[str]						=				CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

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

			__snake_case	:						Any						=				{
			    '''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				,       a_				,       a_=0					):





			'''simple docstring'''

			if str(a_					).startswith('''mps'''					):
				__snake_case	:						Dict						=				torch.manual_seed(a_					)
			else:
				__snake_case	:						Optional[int]						=				torch.Generator(device=a_					).manual_seed(a_					)

			__snake_case	:						Union[str, Any]						=				2

			__snake_case	:						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_					)				,       ),
			]

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





			'''simple docstring'''

			__snake_case	:						str						=				self.get_dummy_components()
			__snake_case	:						List[Any]						=				self.pipeline_class(**a_					)
			pipe.to(a_					)

			__snake_case	:						List[Any]						=				10.0
			__snake_case	:						List[str]						=				4

			__snake_case	:						Optional[Any]						=				self.get_dummy_inputs(a_					)
			__snake_case	:						int						=				steps
			__snake_case	:						Union[str, Any]						=				scale
			__snake_case	:						str						=				pipe(**a_					)[0]

			__snake_case	:						int						=				self.get_dummy_inputs(a_					)
			__snake_case	:						Tuple						=				steps
			__snake_case	:						Any						=				scale
			__snake_case	:						str						=				pipe(**a_				,       control_guidance_start=0.1				,       control_guidance_end=0.2					)[0]

			__snake_case	:						str						=				self.get_dummy_inputs(a_					)
			__snake_case	:						Union[str, Any]						=				steps
			__snake_case	:						Tuple						=				scale
			__snake_case	:						List[str]						=				pipe(**a_				,       control_guidance_start=[0.1, 0.3]				,       control_guidance_end=[0.2, 0.7]					)[0]

			__snake_case	:						int						=				self.get_dummy_inputs(a_					)
			__snake_case	:						Tuple						=				steps
			__snake_case	:						List[Any]						=				scale
			__snake_case	:						str						=				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					):





			'''simple docstring'''

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





			'''simple docstring'''

			self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3					)



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			self._test_inference_batch_single_identical(expected_max_diff=2E-3					)



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			__snake_case	:						str						=				self.get_dummy_components()
			__snake_case	:						Union[str, 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 			_UpperCAmelCase (							unittest.TestCase				):






		'''simple docstring'''

		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			super().tearDown()
			gc.collect()
			torch.cuda.empty_cache()



		def 				SCREAMING_SNAKE_CASE  (self					):





			'''simple docstring'''

			__snake_case	:						Any						=				ControlNetModel.from_pretrained('''lllyasviel/sd-controlnet-canny'''					)

			__snake_case	:						Union[str, Any]						=				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_					)

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

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

			__snake_case	:						List[str]						=				output.images[0]

			assert image.shape == (5_12, 5_12, 3)

			__snake_case	:						Union[str, Any]						=				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 numpy as np






def 		lowercase							(     _snake_case						:      int     ,   _snake_case						:      Optional[Any]     ,   _snake_case						:      Optional[int]     ,   _snake_case						:      int     ,   _snake_case						:      Union[str, Any]     )     ->Dict:
	"""simple docstring"""





	__snake_case	:						Union[str, Any]						=				int(np.ceil((x_end - xa) / h     )     )
	__snake_case	:						Dict						=				np.zeros((n + 1,)     )
	__snake_case	:						List[Any]						=				ya
	__snake_case	:						int						=				xa

	for k in range(_snake_case     ):
		__snake_case	:						Any						=				f(_snake_case     ,   y[k]     )
		__snake_case	:						List[Any]						=				f(x + 0.5 * h     ,   y[k] + 0.5 * h * ka     )
		__snake_case	:						int						=				f(x + 0.5 * h     ,   y[k] + 0.5 * h * ka     )
		__snake_case	:						Optional[int]						=				f(x + h     ,   y[k] + h * ka     )
		__snake_case	:						Optional[int]						=				y[k] + (1 / 6) * h * (ka + 2 * ka + 2 * ka + ka)
		x += h

	return y


if __name__ == "__main__":
			import doctest

			doctest.testmod()





