
"""simple docstring"""







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 			A__	(     _lowerCamelCase	,					_lowerCamelCase	,					_lowerCamelCase	,					unittest.TestCase):
		A_						:	str           =     StableDiffusionControlNetImgaImgPipeline
		A_						:	Optional[int]           =     TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'height', 'width'}
		A_						:	Tuple           =     TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
		A_						:	Union[str, Any]           =     IMAGE_TO_IMAGE_IMAGE_PARAMS.union({'control_image'})
		A_						:	Union[str, Any]           =     IMAGE_TO_IMAGE_IMAGE_PARAMS

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

			__lowerCAmelCase      :						Union[str, Any]						      =				{
			    'unet': unet,
			    'controlnet': controlnet,
			    'scheduler': scheduler,
			    'vae': vae,
			    'text_encoder': text_encoder,
			    'tokenizer': tokenizer,
			    'safety_checker': None,
			    'feature_extractor': None,
			}
			return components

		def 							__lowerCamelCase							(				self	,		_SCREAMING_SNAKE_CASE	,		_SCREAMING_SNAKE_CASE=0			):
			if str(_SCREAMING_SNAKE_CASE			).startswith('mps'			):
				__lowerCAmelCase      :						Optional[int]						      =				torch.manual_seed(_SCREAMING_SNAKE_CASE			)
			else:
				__lowerCAmelCase      :						List[Any]						      =				torch.Generator(device=_SCREAMING_SNAKE_CASE			).manual_seed(_SCREAMING_SNAKE_CASE			)

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







		def 							__lowerCamelCase							(				self			):
			self._test_inference_batch_single_identical(expected_max_diff=2E-3			)




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

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

			__lowerCAmelCase      :						int						      =				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(_SCREAMING_SNAKE_CASE			)

			torch.manual_seed(0			)
			__lowerCAmelCase      :						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(_SCREAMING_SNAKE_CASE			)

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

			__lowerCAmelCase      :						str						      =				MultiControlNetModel([controlneta, controlneta]			)

			__lowerCAmelCase      :						Union[str, Any]						      =				{
			    'unet': unet,
			    'controlnet': controlnet,
			    'scheduler': scheduler,
			    'vae': vae,
			    'text_encoder': text_encoder,
			    'tokenizer': tokenizer,
			    'safety_checker': None,
			    'feature_extractor': None,
			}
			return components

		def 							__lowerCamelCase							(				self	,		_SCREAMING_SNAKE_CASE	,		_SCREAMING_SNAKE_CASE=0			):
			if str(_SCREAMING_SNAKE_CASE			).startswith('mps'			):
				__lowerCAmelCase      :						Dict						      =				torch.manual_seed(_SCREAMING_SNAKE_CASE			)
			else:
				__lowerCAmelCase      :						Optional[int]						      =				torch.Generator(device=_SCREAMING_SNAKE_CASE			).manual_seed(_SCREAMING_SNAKE_CASE			)

			__lowerCAmelCase      :						List[Any]						      =				2

			__lowerCAmelCase      :						Optional[Any]						      =				[
			    randn_tensor(
			        (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)	,		generator=_SCREAMING_SNAKE_CASE	,		device=torch.device(_SCREAMING_SNAKE_CASE			)	,		),
			    randn_tensor(
			        (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)	,		generator=_SCREAMING_SNAKE_CASE	,		device=torch.device(_SCREAMING_SNAKE_CASE			)	,		),
			]

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

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

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

			__lowerCAmelCase      :						Tuple						      =				self.get_dummy_inputs(_SCREAMING_SNAKE_CASE			)
			__lowerCAmelCase      :						Tuple						      =				steps
			__lowerCAmelCase      :						List[Any]						      =				scale
			__lowerCAmelCase      :						Dict						      =				pipe(**_SCREAMING_SNAKE_CASE	,		control_guidance_start=0.1	,		control_guidance_end=0.2			)[0]

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

			__lowerCAmelCase      :						str						      =				self.get_dummy_inputs(_SCREAMING_SNAKE_CASE			)
			__lowerCAmelCase      :						Tuple						      =				steps
			__lowerCAmelCase      :						Union[str, Any]						      =				scale
			__lowerCAmelCase      :						Union[str, Any]						      =				pipe(**_SCREAMING_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 							__lowerCamelCase							(				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 							__lowerCamelCase							(				self			):
			self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2E-3			)

		def 							__lowerCamelCase							(				self			):
			self._test_inference_batch_single_identical(expected_max_diff=2E-3			)







		def 							__lowerCamelCase							(				self			):
			__lowerCAmelCase      :						Optional[Any]						      =				self.get_dummy_components()
			__lowerCAmelCase      :						Optional[Any]						      =				self.pipeline_class(**_SCREAMING_SNAKE_CASE			)
			pipe.to(_SCREAMING_SNAKE_CASE			)
			pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE			)
			with tempfile.TemporaryDirectory() as tmpdir:
				try:
					# save_pretrained is not implemented for Multi-ControlNet
					pipe.save_pretrained(_SCREAMING_SNAKE_CASE			)
				except NotImplementedError:
					pass




@slow
@require_torch_gpu
class 			A__	(     unittest.TestCase):

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







		def 							__lowerCamelCase							(				self			):
			__lowerCAmelCase      :						int						      =				ControlNetModel.from_pretrained('lllyasviel/sd-controlnet-canny'			)

			__lowerCAmelCase      :						Optional[Any]						      =				StableDiffusionControlNetImgaImgPipeline.from_pretrained(
			    'runwayml/stable-diffusion-v1-5'	,		safety_checker=_SCREAMING_SNAKE_CASE	,		controlnet=_SCREAMING_SNAKE_CASE			)
			pipe.enable_model_cpu_offload()
			pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE			)

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

			__lowerCAmelCase      :						Union[str, Any]						      =				pipe(
			    _SCREAMING_SNAKE_CASE	,		_SCREAMING_SNAKE_CASE	,		control_image=_SCREAMING_SNAKE_CASE	,		generator=_SCREAMING_SNAKE_CASE	,		output_type='np'	,		num_inference_steps=50	,		strength=0.6	,		)

			__lowerCAmelCase      :						Dict						      =				output.images[0]

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

			__lowerCAmelCase      :						Dict						      =				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 copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import ClassLabel, Features, Value
from .base import TaskTemplate




@dataclass(frozen=_lowerCamelCase)
class 			A__	(     _lowerCamelCase):
		# `task` is not a ClassVar since we want it to be part of the `asdict` output for JSON serialization
		A_						:	str           =     field(default='text-classification'	,					metadata={'include_in_asdict_even_if_is_default': True})
		A_						:	ClassVar[Features]           =     Features({'text': Value('string')})
		A_						:	ClassVar[Features]           =     Features({'labels': ClassLabel})
		A_						:	str           =     "text"
		A_						:	str           =     "labels"

		def 							__lowerCamelCase							(				self	,		_SCREAMING_SNAKE_CASE			):
			if self.label_column not in features:
				raise ValueError(f"Column {self.label_column} is not present in features."			)
			if not isinstance(features[self.label_column]	,		_SCREAMING_SNAKE_CASE			):
				raise ValueError(f"Column {self.label_column} is not a ClassLabel."			)
			__lowerCAmelCase      :						Any						      =				copy.deepcopy(self			)
			__lowerCAmelCase      :						Dict						      =				self.label_schema.copy()
			__lowerCAmelCase      :						List[Any]						      =				features[self.label_column]
			__lowerCAmelCase      :						Dict						      =				label_schema
			return task_template







		@property
		def 							__lowerCamelCase							(				self			):
			return {
			    self.text_column: "text",
			    self.label_column: "labels",
			}