






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





							"""simple docstring"""

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

							def       __A (  self:   Optional[int]   )    ->				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   )
												_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.00_085		,     beta_end=0.012		,     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=10_00		,     )
												_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       __A (  self:   Optional[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   )		,     )
												_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       __A (  self:   List[Any]   )    ->				List[Any]:
												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       __A (  self:   str   )    ->				int:
												self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3   )



							def       __A (  self:   str   )    ->				Optional[Any]:
												self._test_inference_batch_single_identical(expected_max_diff=2e-3   )





class 						SCREAMING_SNAKE_CASE				( snake_case  ,	snake_case  ,	unittest.TestCase					):





							"""simple docstring"""

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

							def       __A (  self:   int   )    ->				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:   List[str]   ):
																	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.00_085		,     beta_end=0.012		,     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=10_00		,     )
												_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       __A (  self:   Optional[int]		,     __A:   List[Any]		,     __A:   Optional[Any]=0   )    ->				Optional[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   )		,     ),
												    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       __A (  self:   Optional[int]   )    ->				int:
												_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       __A (  self:   Union[str, Any]   )    ->				Any:
												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       __A (  self:   Optional[int]   )    ->				Optional[int]:
												self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3   )

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



							def       __A (  self:   str   )    ->				Optional[int]:
												_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 						SCREAMING_SNAKE_CASE				( unittest.TestCase					):





							"""simple docstring"""

							def       __A (  self:   Union[str, Any]   )    ->				str:
												super().tearDown()
												gc.collect()
												torch.cuda.empty_cache()



							def       __A (  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((5_12, 5_12)   )
												_A								= load_image(
												    '''https://huggingface.co/lllyasviel/sd-controlnet-canny/resolve/main/images/bird.png'''   ).resize((5_12, 5_12)   )

												_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 == (5_12, 5_12, 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











import os
import warnings
from typing import List, Optional

from ...tokenization_utils_base import BatchEncoding
from ...utils import logging
from .configuration_rag import RagConfig


__A      =				logging.get_logger(__name__)





class 						SCREAMING_SNAKE_CASE				:





							"""simple docstring"""

							def __init__(  self:   int		,     __A:   Optional[int]		,     __A:   Optional[Any]   )    ->				str:
												_A								= question_encoder
												_A								= generator
												_A								= self.question_encoder

							def       __A (  self:   Optional[int]		,     __A:   Union[str, Any]   )    ->				Dict:
												if os.path.isfile(__A   ):
																	raise ValueError(f"""Provided path ({save_directory}) should be a directory, not a file"""   )
												os.makedirs(__A		,     exist_ok=__A   )
												_A								= os.path.join(__A		,     '''question_encoder_tokenizer'''   )
												_A								= os.path.join(__A		,     '''generator_tokenizer'''   )
												self.question_encoder.save_pretrained(__A   )
												self.generator.save_pretrained(__A   )

							@classmethod
							def       __A (  cls:   Optional[Any]		,     __A:   List[str]		,     **__A:   int   )    ->				Any:
												# dynamically import AutoTokenizer
												from ..auto.tokenization_auto import AutoTokenizer

												_A								= kwargs.pop('''config'''		,     __A   )

												if config is None:
																	_A								= RagConfig.from_pretrained(__A   )

												_A								= AutoTokenizer.from_pretrained(
												    __A		,     config=config.question_encoder		,     subfolder='''question_encoder_tokenizer'''   )
												_A								= AutoTokenizer.from_pretrained(
												    __A		,     config=config.generator		,     subfolder='''generator_tokenizer'''   )
												return cls(question_encoder=__A		,     generator=__A   )

							def __call__(  self:   int		,     *__A:   Optional[int]		,     **__A:   List[str]   )    ->				int:
												return self.current_tokenizer(*__A		,     **__A   )

							def       __A (  self:   Dict		,     *__A:   List[str]		,     **__A:   List[str]   )    ->				Dict:
												return self.generator.batch_decode(*__A		,     **__A   )

							def       __A (  self:   Union[str, Any]		,     *__A:   Tuple		,     **__A:   List[str]   )    ->				Tuple:
												return self.generator.decode(*__A		,     **__A   )

							def       __A (  self:   Dict   )    ->				List[str]:
												_A								= self.question_encoder

							def       __A (  self:   Union[str, Any]   )    ->				int:
												_A								= self.generator



							def       __A (  self:   Dict		,     __A:   List[str]		,     __A:   Optional[List[str]] = None		,     __A:   Optional[int] = None		,     __A:   Optional[int] = None		,     __A:   str = "longest"		,     __A:   str = None		,     __A:   bool = True		,     **__A:   Tuple		,     )    ->				BatchEncoding:
												warnings.warn(
												    '''`prepare_seq2seq_batch` is deprecated and will be removed in version 5 of 🤗 Transformers. Use the '''
												    '''regular `__call__` method to prepare your inputs and the tokenizer under the `with_target_tokenizer` '''
												    '''context manager to prepare your targets. See the documentation of your specific tokenizer for more '''
												    '''details'''		,     __A		,     )
												if max_length is None:
																	_A								= self.current_tokenizer.model_max_length
												_A								= self(
												    __A		,     add_special_tokens=__A		,     return_tensors=__A		,     max_length=__A		,     padding=__A		,     truncation=__A		,     **__A		,     )
												if tgt_texts is None:
																	return model_inputs
												# Process tgt_texts
												if max_target_length is None:
																	_A								= self.current_tokenizer.model_max_length
												_A								= self(
												    text_target=__A		,     add_special_tokens=__A		,     return_tensors=__A		,     padding=__A		,     max_length=__A		,     truncation=__A		,     **__A		,     )
												_A								= labels['''input_ids''']
												return model_inputs



