



'''simple docstring'''






import json
import os
import pickle
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers import is_faiss_available
from transformers.models.bart.configuration_bart import BartConfig
from transformers.models.bart.tokenization_bart import BartTokenizer
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
from transformers.models.dpr.configuration_dpr import DPRConfig
from transformers.models.dpr.tokenization_dpr import DPRContextEncoderTokenizer, DPRQuestionEncoderTokenizer
from transformers.models.rag.configuration_rag import RagConfig
from transformers.models.rag.retrieval_rag import CustomHFIndex, RagRetriever
from transformers.models.roberta.tokenization_roberta import VOCAB_FILES_NAMES as BART_VOCAB_FILES_NAMES
from transformers.testing_utils import require_faiss, require_sentencepiece, require_tokenizers, require_torch


if is_faiss_available():
						import faiss

@require_faiss
class 				SCREAMING_SNAKE_CASE(							A__			):







	"""simple docstring"""






	def      A	(						self			:     Tuple		)			->		Any:
								UpperCAmelCase :       Any 							=		tempfile.mkdtemp()
								UpperCAmelCase :       Any 							=		8

								# DPR tok
								UpperCAmelCase :       Optional[int] 							=		[
								    '''[UNK]''',
								    '''[CLS]''',
								    '''[SEP]''',
								    '''[PAD]''',
								    '''[MASK]''',
								    '''want''',
								    '''##want''',
								    '''##ed''',
								    '''wa''',
								    '''un''',
								    '''runn''',
								    '''##ing''',
								    ''',''',
								    '''low''',
								    '''lowest''',
								]
								UpperCAmelCase :       Union[str, Any] 							=		os.path.join(self.tmpdirname		,			'''dpr_tokenizer'''		)
								os.makedirs(_snake_case		,			exist_ok=_snake_case		)
								UpperCAmelCase :       List[str] 							=		os.path.join(_snake_case		,			DPR_VOCAB_FILES_NAMES['''vocab_file''']		)
								with open(self.vocab_file		,			'''w'''		,			encoding='''utf-8'''		) as vocab_writer:
															vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]		)		)

								# BART tok
								UpperCAmelCase :       Dict 							=		[
								    '''l''',
								    '''o''',
								    '''w''',
								    '''e''',
								    '''r''',
								    '''s''',
								    '''t''',
								    '''i''',
								    '''d''',
								    '''n''',
								    '''\u0120''',
								    '''\u0120l''',
								    '''\u0120n''',
								    '''\u0120lo''',
								    '''\u0120low''',
								    '''er''',
								    '''\u0120lowest''',
								    '''\u0120newer''',
								    '''\u0120wider''',
								    '''<unk>''',
								]
								UpperCAmelCase :       Optional[Any] 							=		dict(zip(_snake_case		,			range(len(_snake_case		)		)		)		)
								UpperCAmelCase :       Any 							=		['''#version: 0.2''', '''\u0120 l''', '''\u0120l o''', '''\u0120lo w''', '''e r''', '''''']
								UpperCAmelCase :       Any 							=		{'''unk_token''': '''<unk>'''}

								UpperCAmelCase :       List[str] 							=		os.path.join(self.tmpdirname		,			'''bart_tokenizer'''		)
								os.makedirs(_snake_case		,			exist_ok=_snake_case		)
								UpperCAmelCase :       str 							=		os.path.join(_snake_case		,			BART_VOCAB_FILES_NAMES['''vocab_file''']		)
								UpperCAmelCase :       Optional[int] 							=		os.path.join(_snake_case		,			BART_VOCAB_FILES_NAMES['''merges_file''']		)
								with open(self.vocab_file		,			'''w'''		,			encoding='''utf-8'''		) as fp:
															fp.write(json.dumps(_snake_case		) + '''\n'''		)
								with open(self.merges_file		,			'''w'''		,			encoding='''utf-8'''		) as fp:
															fp.write('''\n'''.join(_snake_case		)		)







	def      A	(						self			:     str		)			->		Dict:
								return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname		,			'''dpr_tokenizer'''		)		)







	def      A	(						self			:     int		)			->		Any:
								return DPRContextEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname		,			'''dpr_tokenizer'''		)		)







	def      A	(						self			:     Dict		)			->		Union[str, Any]:
								return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname		,			'''bart_tokenizer'''		)		)







	def      A	(						self			:     Dict		)			->		List[Any]:
								shutil.rmtree(self.tmpdirname		)







	def      A	(						self			:     int		)			->		Union[str, Any]:
								UpperCAmelCase :       Union[str, Any] 							=		Dataset.from_dict(
								    {
								        '''id''': ['''0''', '''1'''],
								        '''text''': ['''foo''', '''bar'''],
								        '''title''': ['''Foo''', '''Bar'''],
								        '''embeddings''': [np.ones(self.retrieval_vector_size		), 2 * np.ones(self.retrieval_vector_size		)],
								    }		)
								dataset.add_faiss_index('''embeddings'''		,			string_factory='''Flat'''		,			metric_type=faiss.METRIC_INNER_PRODUCT		)
								return dataset







	def      A	(						self			:     Optional[Any]		)			->		Tuple:
								UpperCAmelCase :       Any 							=		self.get_dummy_dataset()
								UpperCAmelCase :       Optional[Any] 							=		RagConfig(
								    retrieval_vector_size=self.retrieval_vector_size		,			question_encoder=DPRConfig().to_dict()		,			generator=BartConfig().to_dict()		,			)
								with patch('''transformers.models.rag.retrieval_rag.load_dataset'''		) as mock_load_dataset:
															UpperCAmelCase :       Optional[Any] 							=		dataset
															UpperCAmelCase :       Optional[Any] 							=		RagRetriever(
															    _snake_case		,			question_encoder_tokenizer=self.get_dpr_tokenizer()		,			generator_tokenizer=self.get_bart_tokenizer()		,			)
								return retriever







	def      A	(						self			:     List[str]		,			__snake_case			:     bool		)			->		Dict:
								UpperCAmelCase :       List[Any] 							=		self.get_dummy_dataset()
								UpperCAmelCase :       Optional[int] 							=		RagConfig(
								    retrieval_vector_size=self.retrieval_vector_size		,			question_encoder=DPRConfig().to_dict()		,			generator=BartConfig().to_dict()		,			index_name='''custom'''		,			)
								if from_disk:
															UpperCAmelCase :       Union[str, Any] 							=		os.path.join(self.tmpdirname		,			'''dataset'''		)
															UpperCAmelCase :       Tuple 							=		os.path.join(self.tmpdirname		,			'''index.faiss'''		)
															dataset.get_index('''embeddings'''		).save(os.path.join(self.tmpdirname		,			'''index.faiss'''		)		)
															dataset.drop_index('''embeddings'''		)
															dataset.save_to_disk(os.path.join(self.tmpdirname		,			'''dataset'''		)		)
															del dataset
															UpperCAmelCase :       List[str] 							=		RagRetriever(
															    _snake_case		,			question_encoder_tokenizer=self.get_dpr_tokenizer()		,			generator_tokenizer=self.get_bart_tokenizer()		,			)
								else:
															UpperCAmelCase :       Optional[Any] 							=		RagRetriever(
															    _snake_case		,			question_encoder_tokenizer=self.get_dpr_tokenizer()		,			generator_tokenizer=self.get_bart_tokenizer()		,			index=CustomHFIndex(config.retrieval_vector_size		,			_snake_case		)		,			)
								return retriever







	def      A	(						self			:     Optional[int]		)			->		Any:
								UpperCAmelCase :       Any 							=		Dataset.from_dict(
								    {
								        '''id''': ['''0''', '''1'''],
								        '''text''': ['''foo''', '''bar'''],
								        '''title''': ['''Foo''', '''Bar'''],
								        '''embeddings''': [np.ones(self.retrieval_vector_size + 1		), 2 * np.ones(self.retrieval_vector_size + 1		)],
								    }		)
								dataset.add_faiss_index('''embeddings'''		,			string_factory='''Flat'''		,			metric_type=faiss.METRIC_INNER_PRODUCT		)

								UpperCAmelCase :       Tuple 							=		os.path.join(self.tmpdirname		,			'''hf_bert_base.hnswSQ8_correct_phi_128.c_index'''		)
								dataset.save_faiss_index('''embeddings'''		,			index_file_name + '''.index.dpr'''		)
								pickle.dump(dataset['''id''']		,			open(index_file_name + '''.index_meta.dpr'''		,			'''wb'''		)		)

								UpperCAmelCase :       int 							=		os.path.join(self.tmpdirname		,			'''psgs_w100.tsv.pkl'''		)
								UpperCAmelCase :       Optional[int] 							=		{sample['''id''']: [sample['''text'''], sample['''title''']] for sample in dataset}
								pickle.dump(_snake_case		,			open(_snake_case		,			'''wb'''		)		)

								UpperCAmelCase :       str 							=		RagConfig(
								    retrieval_vector_size=self.retrieval_vector_size		,			question_encoder=DPRConfig().to_dict()		,			generator=BartConfig().to_dict()		,			index_name='''legacy'''		,			index_path=self.tmpdirname		,			)
								UpperCAmelCase :       int 							=		RagRetriever(
								    _snake_case		,			question_encoder_tokenizer=self.get_dpr_tokenizer()		,			generator_tokenizer=self.get_bart_tokenizer()		)
								return retriever







	def      A	(						self			:     Optional[Any]		)			->		Tuple:
								UpperCAmelCase :       List[str] 							=		1
								UpperCAmelCase :       Tuple 							=		self.get_dummy_canonical_hf_index_retriever()
								UpperCAmelCase :       Tuple 							=		np.array(
								    [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]		,			dtype=np.floataa		)
								UpperCAmelCase							,		UpperCAmelCase							,		UpperCAmelCase :       List[str] 							=		retriever.retrieve(_snake_case		,			n_docs=_snake_case		)
								self.assertEqual(retrieved_doc_embeds.shape		,			(2, n_docs, self.retrieval_vector_size)		)
								self.assertEqual(len(_snake_case		)		,			2		)
								self.assertEqual(sorted(doc_dicts[0]		)		,			['''embeddings''', '''id''', '''text''', '''title''']		)
								self.assertEqual(len(doc_dicts[0]['''id''']		)		,			_snake_case		)
								self.assertEqual(doc_dicts[0]['''id'''][0]		,			'''1'''		)  # max inner product is reached with second doc
								self.assertEqual(doc_dicts[1]['''id'''][0]		,			'''0'''		)  # max inner product is reached with first doc
								self.assertListEqual(doc_ids.tolist()		,			[[1], [0]]		)







	def      A	(						self			:     List[Any]		)			->		str:
								UpperCAmelCase :       Optional[int] 							=		self.get_dummy_canonical_hf_index_retriever()
								with tempfile.TemporaryDirectory() as tmp_dirname:
															with patch('''transformers.models.rag.retrieval_rag.load_dataset'''		) as mock_load_dataset:
																						UpperCAmelCase :       str 							=		self.get_dummy_dataset()
																						retriever.save_pretrained(_snake_case		)
																						UpperCAmelCase :       Dict 							=		RagRetriever.from_pretrained(_snake_case		)
																						self.assertIsInstance(_snake_case		,			_snake_case		)
																						UpperCAmelCase :       int 							=		np.array(
																						    [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]		,			dtype=np.floataa		)
																						UpperCAmelCase :       Optional[int] 							=		retriever.retrieve(_snake_case		,			n_docs=1		)
																						self.assertTrue(out is not None		)







	def      A	(						self			:     Optional[int]		)			->		Union[str, Any]:
								UpperCAmelCase :       List[str] 							=		1
								UpperCAmelCase :       Optional[Any] 							=		self.get_dummy_custom_hf_index_retriever(from_disk=_snake_case		)
								UpperCAmelCase :       Any 							=		np.array(
								    [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]		,			dtype=np.floataa		)
								UpperCAmelCase							,		UpperCAmelCase							,		UpperCAmelCase :       Optional[int] 							=		retriever.retrieve(_snake_case		,			n_docs=_snake_case		)
								self.assertEqual(retrieved_doc_embeds.shape		,			(2, n_docs, self.retrieval_vector_size)		)
								self.assertEqual(len(_snake_case		)		,			2		)
								self.assertEqual(sorted(doc_dicts[0]		)		,			['''embeddings''', '''id''', '''text''', '''title''']		)
								self.assertEqual(len(doc_dicts[0]['''id''']		)		,			_snake_case		)
								self.assertEqual(doc_dicts[0]['''id'''][0]		,			'''1'''		)  # max inner product is reached with second doc
								self.assertEqual(doc_dicts[1]['''id'''][0]		,			'''0'''		)  # max inner product is reached with first doc
								self.assertListEqual(doc_ids.tolist()		,			[[1], [0]]		)







	def      A	(						self			:     Dict		)			->		Optional[Any]:
								UpperCAmelCase :       int 							=		self.get_dummy_custom_hf_index_retriever(from_disk=_snake_case		)
								with tempfile.TemporaryDirectory() as tmp_dirname:
															retriever.save_pretrained(_snake_case		)
															UpperCAmelCase :       List[str] 							=		RagRetriever.from_pretrained(_snake_case		)
															self.assertIsInstance(_snake_case		,			_snake_case		)
															UpperCAmelCase :       Dict 							=		np.array(
															    [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]		,			dtype=np.floataa		)
															UpperCAmelCase :       str 							=		retriever.retrieve(_snake_case		,			n_docs=1		)
															self.assertTrue(out is not None		)







	def      A	(						self			:     List[str]		)			->		List[str]:
								UpperCAmelCase :       Dict 							=		1
								UpperCAmelCase :       Union[str, Any] 							=		self.get_dummy_custom_hf_index_retriever(from_disk=_snake_case		)
								UpperCAmelCase :       Optional[int] 							=		np.array(
								    [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]		,			dtype=np.floataa		)
								UpperCAmelCase							,		UpperCAmelCase							,		UpperCAmelCase :       str 							=		retriever.retrieve(_snake_case		,			n_docs=_snake_case		)
								self.assertEqual(retrieved_doc_embeds.shape		,			(2, n_docs, self.retrieval_vector_size)		)
								self.assertEqual(len(_snake_case		)		,			2		)
								self.assertEqual(sorted(doc_dicts[0]		)		,			['''embeddings''', '''id''', '''text''', '''title''']		)
								self.assertEqual(len(doc_dicts[0]['''id''']		)		,			_snake_case		)
								self.assertEqual(doc_dicts[0]['''id'''][0]		,			'''1'''		)  # max inner product is reached with second doc
								self.assertEqual(doc_dicts[1]['''id'''][0]		,			'''0'''		)  # max inner product is reached with first doc
								self.assertListEqual(doc_ids.tolist()		,			[[1], [0]]		)







	def      A	(						self			:     Optional[int]		)			->		List[Any]:
								UpperCAmelCase :       Tuple 							=		self.get_dummy_custom_hf_index_retriever(from_disk=_snake_case		)
								with tempfile.TemporaryDirectory() as tmp_dirname:
															retriever.save_pretrained(_snake_case		)
															UpperCAmelCase :       List[Any] 							=		RagRetriever.from_pretrained(_snake_case		)
															self.assertIsInstance(_snake_case		,			_snake_case		)
															UpperCAmelCase :       Dict 							=		np.array(
															    [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]		,			dtype=np.floataa		)
															UpperCAmelCase :       Optional[Any] 							=		retriever.retrieve(_snake_case		,			n_docs=1		)
															self.assertTrue(out is not None		)







	def      A	(						self			:     Optional[Any]		)			->		List[str]:
								UpperCAmelCase :       Tuple 							=		1
								UpperCAmelCase :       Any 							=		self.get_dummy_legacy_index_retriever()
								UpperCAmelCase :       List[str] 							=		np.array(
								    [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]		,			dtype=np.floataa		)
								UpperCAmelCase							,		UpperCAmelCase							,		UpperCAmelCase :       Optional[int] 							=		retriever.retrieve(_snake_case		,			n_docs=_snake_case		)
								self.assertEqual(retrieved_doc_embeds.shape		,			(2, n_docs, self.retrieval_vector_size)		)
								self.assertEqual(len(_snake_case		)		,			2		)
								self.assertEqual(sorted(doc_dicts[0]		)		,			['''text''', '''title''']		)
								self.assertEqual(len(doc_dicts[0]['''text''']		)		,			_snake_case		)
								self.assertEqual(doc_dicts[0]['''text'''][0]		,			'''bar'''		)  # max inner product is reached with second doc
								self.assertEqual(doc_dicts[1]['''text'''][0]		,			'''foo'''		)  # max inner product is reached with first doc
								self.assertListEqual(doc_ids.tolist()		,			[[1], [0]]		)







	def      A	(						self			:     List[Any]		)			->		Tuple:
								UpperCAmelCase :       Dict 							=		self.get_dummy_legacy_index_retriever()
								with tempfile.TemporaryDirectory() as tmp_dirname:
															retriever.save_pretrained(_snake_case		)
															UpperCAmelCase :       List[Any] 							=		RagRetriever.from_pretrained(_snake_case		)
															self.assertIsInstance(_snake_case		,			_snake_case		)
															UpperCAmelCase :       Tuple 							=		np.array(
															    [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]		,			dtype=np.floataa		)
															UpperCAmelCase :       Dict 							=		retriever.retrieve(_snake_case		,			n_docs=1		)
															self.assertTrue(out is not None		)







	@require_torch
	@require_tokenizers
	@require_sentencepiece
	def      A	(						self			:     Union[str, Any]		)			->		List[Any]:
								import torch

								UpperCAmelCase :       str 							=		1
								UpperCAmelCase :       int 							=		self.get_dummy_canonical_hf_index_retriever()
								UpperCAmelCase :       Optional[int] 							=		[[5, 7], [10, 11]]
								UpperCAmelCase :       Union[str, Any] 							=		np.array(
								    [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]		,			dtype=np.floataa		)
								UpperCAmelCase :       List[Any] 							=		retriever(_snake_case		,			_snake_case		,			prefix=retriever.config.generator.prefix		,			n_docs=_snake_case		)
								UpperCAmelCase							,		UpperCAmelCase							,		UpperCAmelCase :       Any 							=		(
								    out['''context_input_ids'''],
								    out['''context_attention_mask'''],
								    out['''retrieved_doc_embeds'''],
								)
								self.assertEqual(retrieved_doc_embeds.shape		,			(2, n_docs, self.retrieval_vector_size)		)
								self.assertIsInstance(_snake_case		,			_snake_case		)
								self.assertIsInstance(_snake_case		,			_snake_case		)
								self.assertIsInstance(_snake_case		,			np.ndarray		)

								UpperCAmelCase :       Optional[Any] 							=		retriever(
								    _snake_case		,			_snake_case		,			prefix=retriever.config.generator.prefix		,			n_docs=_snake_case		,			return_tensors='''pt'''		,			)
								UpperCAmelCase							,		UpperCAmelCase							,		UpperCAmelCase							,		UpperCAmelCase :       Dict 							=		(  # noqa: F841
								    out['''context_input_ids'''],
								    out['''context_attention_mask'''],
								    out['''retrieved_doc_embeds'''],
								    out['''doc_ids'''],
								)
								self.assertEqual(retrieved_doc_embeds.shape		,			(2, n_docs, self.retrieval_vector_size)		)
								self.assertIsInstance(_snake_case		,			torch.Tensor		)
								self.assertIsInstance(_snake_case		,			torch.Tensor		)
								self.assertIsInstance(_snake_case		,			torch.Tensor		)







	@require_torch
	@require_tokenizers
	@require_sentencepiece
	def      A	(						self			:     Tuple		)			->		int:
								UpperCAmelCase :       Dict 							=		self.get_dpr_ctx_encoder_tokenizer()
								UpperCAmelCase :       int 							=		1
								UpperCAmelCase :       Union[str, Any] 							=		self.get_dummy_custom_hf_index_retriever(from_disk=_snake_case		)
								retriever.set_ctx_encoder_tokenizer(_snake_case		)

								UpperCAmelCase :       str 							=		[[5, 7], [10, 11]]
								UpperCAmelCase :       Dict 							=		np.array(
								    [np.ones(self.retrieval_vector_size		), -np.ones(self.retrieval_vector_size		)]		,			dtype=np.floataa		)
								UpperCAmelCase :       Optional[int] 							=		retriever(_snake_case		,			_snake_case		,			prefix=retriever.config.generator.prefix		,			n_docs=_snake_case		)

								self.assertEqual(
								    len(_snake_case		)		,			6		)  # check whether the retriever output consist of 6 attributes including tokenized docs
								self.assertEqual(
								    all(k in out for k in ('''tokenized_doc_ids''', '''tokenized_doc_attention_mask''')		)		,			_snake_case		)  # check for doc token related keys in dictionary.











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







   def lowerCamelCase						(      self		:						int):

        """simple docstring"""


        torch.manual_seed(0)
        UpperCAmelCase_          =							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)
        UpperCAmelCase_          =							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)
        UpperCAmelCase_          =							DDIMScheduler(
            beta_start=0.0_0_0_8_5					,						beta_end=0.0_1_2					,						beta_schedule='''scaled_linear'''					,						clip_sample=_snake_case					,						set_alpha_to_one=_snake_case					,						)
        torch.manual_seed(0)
        UpperCAmelCase_          =							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)
        UpperCAmelCase_          =							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					,						)
        UpperCAmelCase_          =							CLIPTextModel(_snake_case)
        UpperCAmelCase_          =							CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip''')

        UpperCAmelCase_          =							{
            '''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		:						Union[str, Any]					,						_snake_case		:						Any					,						_snake_case		:						Dict=0):

        """simple docstring"""


        if str(_snake_case).startswith('''mps'''):
             UpperCAmelCase_          =							torch.manual_seed(_snake_case)
        else:
             UpperCAmelCase_          =							torch.Generator(device=_snake_case).manual_seed(_snake_case)

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

        """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 lowerCamelCase						(      self		:						Any):

        """simple docstring"""


        self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3)







   def lowerCamelCase						(      self		:						Optional[Any]):

        """simple docstring"""


        self._test_inference_batch_single_identical(expected_max_diff=2e-3)







class __snake_case			(			a ,  a ,  unittest.TestCase		):
   UpperCAmelCase__  :    str          =  StableDiffusionControlNetImgaImgPipeline
   UpperCAmelCase__  :    List[str]          =  TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width'''}
   UpperCAmelCase__  :    Optional[Any]          =  TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
   UpperCAmelCase__  :    str          =  frozenset([]		)  # TO_DO: add image_params once refactored VaeImageProcessor.preprocess







   def lowerCamelCase						(      self		:						str):

        """simple docstring"""


        torch.manual_seed(0)
        UpperCAmelCase_          =							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(_snake_case		:						Optional[int]):
             if isinstance(_snake_case					,						torch.nn.Convad):
                  torch.nn.init.normal(m.weight)
                  m.bias.data.fill_(1.0)

        UpperCAmelCase_          =							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(_snake_case)

        torch.manual_seed(0)
        UpperCAmelCase_          =							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(_snake_case)

        torch.manual_seed(0)
        UpperCAmelCase_          =							DDIMScheduler(
            beta_start=0.0_0_0_8_5					,						beta_end=0.0_1_2					,						beta_schedule='''scaled_linear'''					,						clip_sample=_snake_case					,						set_alpha_to_one=_snake_case					,						)
        torch.manual_seed(0)
        UpperCAmelCase_          =							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)
        UpperCAmelCase_          =							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					,						)
        UpperCAmelCase_          =							CLIPTextModel(_snake_case)
        UpperCAmelCase_          =							CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip''')

        UpperCAmelCase_          =							MultiControlNetModel([controlneta, controlneta])

        UpperCAmelCase_          =							{
            '''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		:						int					,						_snake_case		:						Union[str, Any]					,						_snake_case		:						str=0):

        """simple docstring"""


        if str(_snake_case).startswith('''mps'''):
             UpperCAmelCase_          =							torch.manual_seed(_snake_case)
        else:
             UpperCAmelCase_          =							torch.Generator(device=_snake_case).manual_seed(_snake_case)

        UpperCAmelCase_          =							2

        UpperCAmelCase_          =							[
            randn_tensor(
                (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)					,						generator=_snake_case					,						device=torch.device(_snake_case)					,						),
            randn_tensor(
                (1, 3, 32 * controlnet_embedder_scale_factor, 32 * controlnet_embedder_scale_factor)					,						generator=_snake_case					,						device=torch.device(_snake_case)					,						),
        ]

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

        """simple docstring"""


        UpperCAmelCase_          =							self.get_dummy_components()
        UpperCAmelCase_          =							self.pipeline_class(**_snake_case)
        pipe.to(_snake_case)

        UpperCAmelCase_          =							1_0.0
        UpperCAmelCase_          =							4

        UpperCAmelCase_          =							self.get_dummy_inputs(_snake_case)
        UpperCAmelCase_          =							steps
        UpperCAmelCase_          =							scale
        UpperCAmelCase_          =							pipe(**_snake_case)[0]

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

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

        UpperCAmelCase_          =							self.get_dummy_inputs(_snake_case)
        UpperCAmelCase_          =							steps
        UpperCAmelCase_          =							scale
        UpperCAmelCase_          =							pipe(**_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		:						Dict):

        """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 lowerCamelCase						(      self		:						int):

        """simple docstring"""


        self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=2e-3)







   def lowerCamelCase						(      self		:						int):

        """simple docstring"""


        self._test_inference_batch_single_identical(expected_max_diff=2e-3)







   def lowerCamelCase						(      self		:						Optional[int]):

        """simple docstring"""


        UpperCAmelCase_          =							self.get_dummy_components()
        UpperCAmelCase_          =							self.pipeline_class(**_snake_case)
        pipe.to(_snake_case)
        pipe.set_progress_bar_config(disable=_snake_case)
        with tempfile.TemporaryDirectory() as tmpdir:
             try:
                  # save_pretrained is not implemented for Multi-ControlNet
                  pipe.save_pretrained(_snake_case)
             except NotImplementedError:
                  pass







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







   def lowerCamelCase						(      self		:						Optional[int]):

        """simple docstring"""


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







   def lowerCamelCase						(      self		:						Optional[int]):

        """simple docstring"""


        UpperCAmelCase_          =							ControlNetModel.from_pretrained('''lllyasviel/sd-controlnet-canny''')

        UpperCAmelCase_          =							StableDiffusionControlNetImgaImgPipeline.from_pretrained(
            '''runwayml/stable-diffusion-v1-5'''					,						safety_checker=_snake_case					,						controlnet=_snake_case)
        pipe.enable_model_cpu_offload()
        pipe.set_progress_bar_config(disable=_snake_case)

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

        UpperCAmelCase_          =							pipe(
            _snake_case					,						_snake_case					,						control_image=_snake_case					,						generator=_snake_case					,						output_type='''np'''					,						num_inference_steps=50					,						strength=0.6					,						)

        UpperCAmelCase_          =							output.images[0]

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

        UpperCAmelCase_          =							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






