






'''simple docstring'''





from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding


class 				snake_case__     (			UpperCamelCase):
					a_       =			["image_processor", "tokenizer"]
					a_       =			"EfficientNetImageProcessor"
					a_       =			("BertTokenizer", "BertTokenizerFast")




					def __init__(       self       :						Optional[int]							,				_A       :						str							,				_A       :						List[Any]     )  ->    List[str]:
								super().__init__(_A							,				_A     )




					def __call__(       self       :						Optional[int]							,				_A       :						Optional[Any]=None							,				_A       :						str=None							,				_A       :						Dict="max_length"							,				_A       :						str=64							,				_A       :						str=None							,				**_A       :						List[str]     )  ->    Tuple:
								if text is None and images is None:
											raise ValueError('''You have to specify either text or images. Both cannot be none.'''     )

								if text is not None:
											UpperCAmelCase_     :   Any       =   self.tokenizer(
											    _A							,				padding=_A							,				max_length=_A							,				return_tensors=_A							,				**_A     )

								if images is not None:
											UpperCAmelCase_     :   Tuple       =   self.image_processor(_A							,				return_tensors=_A							,				**_A     )

								if text is not None and images is not None:
											UpperCAmelCase_     :   List[str]       =   image_features.pixel_values
											return encoding
								elif text is not None:
											return encoding
								else:
											return BatchEncoding(data=dict(**_A     )							,				tensor_type=_A     )




					def 						A     (       self       :						str							,				*_A       :						Dict							,				**_A       :						int     )  ->    Union[str, Any]:
								return self.tokenizer.batch_decode(*_A							,				**_A     )




					def 						A     (       self       :						str							,				*_A       :						Tuple							,				**_A       :						Tuple     )  ->    Optional[int]:
								return self.tokenizer.decode(*_A							,				**_A     )





					@property
					def 						A     (       self       :						Dict     )  ->    str:
								UpperCAmelCase_     :   List[str]       =   self.tokenizer.model_input_names
								UpperCAmelCase_     :   str       =   self.image_processor.model_input_names
								return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names     )     )










'''simple docstring'''





import os
import tempfile
import unittest

import numpy as np
import torch

from diffusers.models import ModelMixin, UNetaDConditionModel
from diffusers.models.attention_processor import AttnProcessor, LoRAAttnProcessor
from diffusers.utils import (
    floats_tensor,
    logging,
    skip_mps,
    torch_device,
)
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


enable_full_determinism()

_UpperCamelCase		:		Dict								=			logging.get_logger(__name__)
def __UpperCAmelCase (			A			:	List[Any]	,   A			:	bool = True						)  -> List[Any]:
			UpperCAmelCase_     :   Any       =   {}
			for name in model.attn_processors.keys():
						UpperCAmelCase_     :   List[Any]       =   name.endswith('''attn2.processor'''						) and not (
						    name.startswith('''transformer_in'''						) or '''temp_attentions''' in name.split('''.'''						)
						)
						UpperCAmelCase_     :   Optional[Any]       =   model.config.cross_attention_dim if has_cross_attention else None
						if name.startswith('''mid_block'''						):
									UpperCAmelCase_     :   Dict       =   model.config.block_out_channels[-1]
						elif name.startswith('''up_blocks'''						):
									UpperCAmelCase_     :   Any       =   int(name[len('''up_blocks.'''						)]						)
									UpperCAmelCase_     :   int       =   list(reversed(model.config.block_out_channels						)						)[block_id]
						elif name.startswith('''down_blocks'''						):
									UpperCAmelCase_     :   Dict       =   int(name[len('''down_blocks.'''						)]						)
									UpperCAmelCase_     :   Any       =   model.config.block_out_channels[block_id]
						elif name.startswith('''transformer_in'''						):
									# Note that the `8 * ...` comes from: https://github.com/huggingface/diffusers/blob/7139f0e874f10b2463caa8cbd585762a309d12d6/src/diffusers/models/unet_3d_condition.py#L148
									UpperCAmelCase_     :   int       =   8 * model.config.attention_head_dim

						UpperCAmelCase_     :   Optional[int]       =   LoRAAttnProcessor(hidden_size=A	,   cross_attention_dim=A						)
						UpperCAmelCase_     :   Tuple       =   lora_attn_procs[name].to(model.device						)

						if mock_weights:
									# add 1 to weights to mock trained weights
									with torch.no_grad():
												lora_attn_procs[name].to_q_lora.up.weight += 1
												lora_attn_procs[name].to_k_lora.up.weight += 1
												lora_attn_procs[name].to_v_lora.up.weight += 1
												lora_attn_procs[name].to_out_lora.up.weight += 1

			return lora_attn_procs







@skip_mps
class 				snake_case__     (			UpperCamelCase      ,							UpperCamelCase      ,							unittest.TestCase):
					a_       =			UNetaDConditionModel
					a_       =			"sample"




					@property
					def 						A     (       self       :						Tuple     )  ->    int:
								UpperCAmelCase_     :   Optional[Any]       =   4
								UpperCAmelCase_     :   int       =   4
								UpperCAmelCase_     :   List[Any]       =   4
								UpperCAmelCase_     :   int       =   (32, 32)

								UpperCAmelCase_     :   Optional[int]       =   floats_tensor((batch_size, num_channels, num_frames) + sizes     ).to(_A     )
								UpperCAmelCase_     :   Any       =   torch.tensor([10]     ).to(_A     )
								UpperCAmelCase_     :   Optional[Any]       =   floats_tensor((batch_size, 4, 32)     ).to(_A     )

								return {"sample": noise, "timestep": time_step, "encoder_hidden_states": encoder_hidden_states}




					@property
					def 						A     (       self       :						Dict     )  ->    Dict:
								return (4, 4, 32, 32)




					@property
					def 						A     (       self       :						Union[str, Any]     )  ->    Union[str, Any]:
								return (4, 4, 32, 32)




					def 						A     (       self       :						Dict     )  ->    List[Any]:
								UpperCAmelCase_     :   List[str]       =   {
								    '''block_out_channels''': (32, 64),
								    '''down_block_types''': (
								        '''CrossAttnDownBlock3D''',
								        '''DownBlock3D''',
								    ),
								    '''up_block_types''': ('''UpBlock3D''', '''CrossAttnUpBlock3D'''),
								    '''cross_attention_dim''': 32,
								    '''attention_head_dim''': 8,
								    '''out_channels''': 4,
								    '''in_channels''': 4,
								    '''layers_per_block''': 1,
								    '''sample_size''': 32,
								}
								UpperCAmelCase_     :   Union[str, Any]       =   self.dummy_input
								return init_dict, inputs_dict




					@unittest.skipIf(
					    torch_device != '''cuda''' or not is_xformers_available()							,				reason='''XFormers attention is only available with CUDA and `xformers` installed'''							,				)
					def 						A     (       self       :						Union[str, Any]     )  ->    Optional[int]:
								UpperCAmelCase_  ,     UpperCAmelCase_     :   Optional[int]       =   self.prepare_init_args_and_inputs_for_common()
								UpperCAmelCase_     :   List[str]       =   self.model_class(**_A     )

								model.enable_xformers_memory_efficient_attention()

								assert (
								    model.mid_block.attentions[0].transformer_blocks[0].attna.processor.__class__.__name__
								    == "XFormersAttnProcessor"
								), "xformers is not enabled"




					def 						A     (       self       :						Optional[Any]     )  ->    List[str]:
								UpperCAmelCase_  ,     UpperCAmelCase_     :   Tuple       =   self.prepare_init_args_and_inputs_for_common()

								UpperCAmelCase_     :   str       =   32

								UpperCAmelCase_     :   Union[str, Any]       =   self.model_class(**_A     )
								model.to(_A     )
								model.eval()

								with torch.no_grad():
											UpperCAmelCase_     :   Any       =   model(**_A     )

											if isinstance(_A							,				_A     ):
														UpperCAmelCase_     :   Optional[Any]       =   output.sample

								self.assertIsNotNone(_A     )
								UpperCAmelCase_     :   int       =   inputs_dict['''sample'''].shape
								self.assertEqual(output.shape							,				_A							,				'''Input and output shapes do not match'''     )




					def 						A     (       self       :						Union[str, Any]     )  ->    Optional[int]:
								UpperCAmelCase_  ,     UpperCAmelCase_     :   Optional[Any]       =   self.prepare_init_args_and_inputs_for_common()
								UpperCAmelCase_     :   Optional[Any]       =   self.model_class(**_A     )
								model.to(_A     )
								model.eval()

								with torch.no_grad():
											# Warmup pass when using mps (see #372)
											if torch_device == "mps" and isinstance(_A							,				_A     ):
														model(**self.dummy_input     )

											UpperCAmelCase_     :   List[Any]       =   model(**_A     )
											if isinstance(_A							,				_A     ):
														UpperCAmelCase_     :   List[Any]       =   first.sample

											UpperCAmelCase_     :   Tuple       =   model(**_A     )
											if isinstance(_A							,				_A     ):
														UpperCAmelCase_     :   Union[str, Any]       =   second.sample

								UpperCAmelCase_     :   Tuple       =   first.cpu().numpy()
								UpperCAmelCase_     :   Tuple       =   second.cpu().numpy()
								UpperCAmelCase_     :   Tuple       =   out_a[~np.isnan(_A     )]
								UpperCAmelCase_     :   List[str]       =   out_a[~np.isnan(_A     )]
								UpperCAmelCase_     :   Optional[int]       =   np.amax(np.abs(out_a - out_a     )     )
								self.assertLessEqual(_A							,				1e-5     )




					def 						A     (       self       :						str     )  ->    str:
								UpperCAmelCase_  ,     UpperCAmelCase_     :   Optional[Any]       =   self.prepare_init_args_and_inputs_for_common()

								UpperCAmelCase_     :   Union[str, Any]       =   8

								UpperCAmelCase_     :   Tuple       =   self.model_class(**_A     )
								model.to(_A     )
								model.eval()

								model.set_attention_slice('''auto'''     )
								with torch.no_grad():
											UpperCAmelCase_     :   Dict       =   model(**_A     )
								assert output is not None

								model.set_attention_slice('''max'''     )
								with torch.no_grad():
											UpperCAmelCase_     :   Union[str, Any]       =   model(**_A     )
								assert output is not None

								model.set_attention_slice(2     )
								with torch.no_grad():
											UpperCAmelCase_     :   int       =   model(**_A     )
								assert output is not None




					def 						A     (       self       :						List[Any]     )  ->    Any:
								UpperCAmelCase_  ,     UpperCAmelCase_     :   List[Any]       =   self.prepare_init_args_and_inputs_for_common()

								UpperCAmelCase_     :   Dict       =   8

								UpperCAmelCase_     :   List[Any]       =   self.model_class(**_A     )
								model.to(_A     )

								with torch.no_grad():
											UpperCAmelCase_     :   int       =   model(**_A     ).sample

								UpperCAmelCase_     :   Any       =   create_lora_layers(_A     )

								# make sure we can set a list of attention processors
								model.set_attn_processor(_A     )
								model.to(_A     )

								# test that attn processors can be set to itself
								model.set_attn_processor(model.attn_processors     )

								with torch.no_grad():
											UpperCAmelCase_     :   int       =   model(**_A							,				cross_attention_kwargs={'''scale''': 0.0}     ).sample
											UpperCAmelCase_     :   int       =   model(**_A							,				cross_attention_kwargs={'''scale''': 0.5}     ).sample
											UpperCAmelCase_     :   int       =   model(**_A							,				cross_attention_kwargs={'''scale''': 0.5}     ).sample

								assert (samplea - samplea).abs().max() < 3e-3
								assert (samplea - samplea).abs().max() < 3e-3

								# sample 2 and sample 3 should be different
								assert (samplea - samplea).abs().max() > 3e-3




					def 						A     (       self       :						Union[str, Any]     )  ->    Tuple:
								UpperCAmelCase_  ,     UpperCAmelCase_     :   List[str]       =   self.prepare_init_args_and_inputs_for_common()

								UpperCAmelCase_     :   str       =   8

								torch.manual_seed(0     )
								UpperCAmelCase_     :   Optional[int]       =   self.model_class(**_A     )
								model.to(_A     )

								with torch.no_grad():
											UpperCAmelCase_     :   Union[str, Any]       =   model(**_A     ).sample

								UpperCAmelCase_     :   int       =   create_lora_layers(_A     )
								model.set_attn_processor(_A     )

								with torch.no_grad():
											UpperCAmelCase_     :   Optional[int]       =   model(**_A							,				cross_attention_kwargs={'''scale''': 0.5}     ).sample

								with tempfile.TemporaryDirectory() as tmpdirname:
											model.save_attn_procs(_A     )
											self.assertTrue(os.path.isfile(os.path.join(_A							,				'''pytorch_lora_weights.bin'''     )     )     )
											torch.manual_seed(0     )
											UpperCAmelCase_     :   Optional[Any]       =   self.model_class(**_A     )
											new_model.to(_A     )
											new_model.load_attn_procs(_A     )

								with torch.no_grad():
											UpperCAmelCase_     :   Dict       =   new_model(**_A							,				cross_attention_kwargs={'''scale''': 0.5}     ).sample

								assert (sample - new_sample).abs().max() < 1e-3

								# LoRA and no LoRA should NOT be the same
								assert (sample - old_sample).abs().max() > 1e-4




					def 						A     (       self       :						Tuple     )  ->    Union[str, Any]:
								UpperCAmelCase_  ,     UpperCAmelCase_     :   Dict       =   self.prepare_init_args_and_inputs_for_common()

								UpperCAmelCase_     :   List[str]       =   8

								torch.manual_seed(0     )
								UpperCAmelCase_     :   str       =   self.model_class(**_A     )
								model.to(_A     )

								with torch.no_grad():
											UpperCAmelCase_     :   Dict       =   model(**_A     ).sample

								UpperCAmelCase_     :   Tuple       =   create_lora_layers(_A     )
								model.set_attn_processor(_A     )

								with torch.no_grad():
											UpperCAmelCase_     :   Dict       =   model(**_A							,				cross_attention_kwargs={'''scale''': 0.5}     ).sample

								with tempfile.TemporaryDirectory() as tmpdirname:
											model.save_attn_procs(_A							,				safe_serialization=_A     )
											self.assertTrue(os.path.isfile(os.path.join(_A							,				'''pytorch_lora_weights.safetensors'''     )     )     )
											torch.manual_seed(0     )
											UpperCAmelCase_     :   Dict       =   self.model_class(**_A     )
											new_model.to(_A     )
											new_model.load_attn_procs(_A     )

								with torch.no_grad():
											UpperCAmelCase_     :   Union[str, Any]       =   new_model(**_A							,				cross_attention_kwargs={'''scale''': 0.5}     ).sample

								assert (sample - new_sample).abs().max() < 3e-3

								# LoRA and no LoRA should NOT be the same
								assert (sample - old_sample).abs().max() > 1e-4




					def 						A     (       self       :						Union[str, Any]     )  ->    Dict:
								# enable deterministic behavior for gradient checkpointing
								UpperCAmelCase_  ,     UpperCAmelCase_     :   List[Any]       =   self.prepare_init_args_and_inputs_for_common()

								UpperCAmelCase_     :   Any       =   8

								torch.manual_seed(0     )
								UpperCAmelCase_     :   Optional[Any]       =   self.model_class(**_A     )
								model.to(_A     )

								UpperCAmelCase_     :   str       =   create_lora_layers(_A							,				mock_weights=_A     )
								model.set_attn_processor(_A     )
								# Saving as torch, properly reloads with directly filename
								with tempfile.TemporaryDirectory() as tmpdirname:
											model.save_attn_procs(_A     )
											self.assertTrue(os.path.isfile(os.path.join(_A							,				'''pytorch_lora_weights.bin'''     )     )     )
											torch.manual_seed(0     )
											UpperCAmelCase_     :   Union[str, Any]       =   self.model_class(**_A     )
											new_model.to(_A     )
											new_model.load_attn_procs(_A							,				weight_name='''pytorch_lora_weights.bin'''     )




					def 						A     (       self       :						Optional[Any]     )  ->    Optional[int]:
								UpperCAmelCase_  ,     UpperCAmelCase_     :   Optional[int]       =   self.prepare_init_args_and_inputs_for_common()

								UpperCAmelCase_     :   List[Any]       =   8

								torch.manual_seed(0     )
								UpperCAmelCase_     :   Dict       =   self.model_class(**_A     )
								model.to(_A     )

								UpperCAmelCase_     :   Tuple       =   create_lora_layers(_A							,				mock_weights=_A     )
								model.set_attn_processor(_A     )
								# Saving as torch, properly reloads with directly filename
								with tempfile.TemporaryDirectory() as tmpdirname:
											model.save_attn_procs(_A     )
											self.assertTrue(os.path.isfile(os.path.join(_A							,				'''pytorch_lora_weights.bin'''     )     )     )
											torch.manual_seed(0     )
											UpperCAmelCase_     :   Optional[Any]       =   self.model_class(**_A     )
											new_model.to(_A     )
											with self.assertRaises(_A     ) as e:
														new_model.load_attn_procs(_A							,				use_safetensors=_A     )
											self.assertIn('''Error no file named pytorch_lora_weights.safetensors'''							,				str(e.exception     )     )




					def 						A     (       self       :						int     )  ->    Tuple:
								UpperCAmelCase_  ,     UpperCAmelCase_     :   Any       =   self.prepare_init_args_and_inputs_for_common()

								UpperCAmelCase_     :   Dict       =   8

								torch.manual_seed(0     )
								UpperCAmelCase_     :   Optional[Any]       =   self.model_class(**_A     )
								model.to(_A     )

								with torch.no_grad():
											UpperCAmelCase_     :   str       =   model(**_A     ).sample

								UpperCAmelCase_     :   str       =   create_lora_layers(_A     )
								model.set_attn_processor(_A     )

								with torch.no_grad():
											UpperCAmelCase_     :   Optional[int]       =   model(**_A							,				cross_attention_kwargs={'''scale''': 0.0}     ).sample

								model.set_attn_processor(AttnProcessor()     )

								with torch.no_grad():
											UpperCAmelCase_     :   int       =   model(**_A     ).sample

								assert (sample - new_sample).abs().max() < 1e-4
								assert (sample - old_sample).abs().max() < 3e-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     )  ->    Optional[Any]:
								# enable deterministic behavior for gradient checkpointing
								UpperCAmelCase_  ,     UpperCAmelCase_     :   Any       =   self.prepare_init_args_and_inputs_for_common()

								UpperCAmelCase_     :   Dict       =   4

								torch.manual_seed(0     )
								UpperCAmelCase_     :   List[Any]       =   self.model_class(**_A     )
								model.to(_A     )
								UpperCAmelCase_     :   List[str]       =   create_lora_layers(_A     )
								model.set_attn_processor(_A     )

								# default
								with torch.no_grad():
											UpperCAmelCase_     :   List[str]       =   model(**_A     ).sample

											model.enable_xformers_memory_efficient_attention()
											UpperCAmelCase_     :   List[str]       =   model(**_A     ).sample

											model.disable_xformers_memory_efficient_attention()
											UpperCAmelCase_     :   Any       =   model(**_A     ).sample

								assert (sample - on_sample).abs().max() < 1e-4
								assert (sample - off_sample).abs().max() < 1e-4





					def 						A     (       self       :						Tuple     )  ->    Union[str, Any]:
								UpperCAmelCase_  ,     UpperCAmelCase_     :   Optional[int]       =   self.prepare_init_args_and_inputs_for_common()
								UpperCAmelCase_     :   int       =   32

								UpperCAmelCase_     :   Any       =   self.model_class(**_A     )
								model.to(_A     )
								model.eval()

								with torch.no_grad():
											UpperCAmelCase_     :   Tuple       =   model(**_A     )[0]

								model.enable_forward_chunking()
								with torch.no_grad():
											UpperCAmelCase_     :   Optional[Any]       =   model(**_A     )[0]

								self.assertEqual(output.shape							,				output_a.shape							,				'''Shape doesn\'t match'''     )
								assert np.abs(output.cpu() - output_a.cpu()     ).max() < 1e-2


# (todo: sayakpaul) implement SLOW tests.


