



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				:							Tuple								=logging.get_logger(__name__)



def 				_lowerCAmelCase      (_lowerCAmelCase				,		_lowerCAmelCase = True):
					UpperCamelCase_											=				{}
					for name in model.attn_processors.keys():
										UpperCamelCase_											=				name.endswith("attn2.processor") and not (
										    name.startswith("transformer_in") or "temp_attentions" in name.split(".")
										)
										UpperCamelCase_											=				model.config.cross_attention_dim if has_cross_attention else None
										if name.startswith("mid_block"):
															UpperCamelCase_											=				model.config.block_out_channels[-1]
										elif name.startswith("up_blocks"):
															UpperCamelCase_											=				int(name[len("up_blocks.")])
															UpperCamelCase_											=				list(reversed(model.config.block_out_channels))[block_id]
										elif name.startswith("down_blocks"):
															UpperCamelCase_											=				int(name[len("down_blocks.")])
															UpperCamelCase_											=				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_											=				8 * model.config.attention_head_dim

										UpperCamelCase_											=				LoRAAttnProcessor(hidden_size=_lowerCAmelCase				,		cross_attention_dim=_lowerCAmelCase)
										UpperCamelCase_											=				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  _lowercase				(a_ ,				a_ ,				unittest.TestCase ):
				'''simple docstring'''







				lowercase__				  =  UNetaDConditionModel
				lowercase__				  =  """sample"""

				@property
				def  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_											=				4
									UpperCamelCase_											=				4
									UpperCamelCase_											=				4
									UpperCamelCase_											=				(32, 32)

									UpperCamelCase_											=				floats_tensor((batch_size, num_channels, num_frames) + sizes      ).to(snake_case__      )
									UpperCamelCase_											=				torch.tensor([10]      ).to(snake_case__      )
									UpperCamelCase_											=				floats_tensor((batch_size, 4, 32)      ).to(snake_case__      )

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

				@property
				def  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									return (4, 4, 32, 32)

				@property
				def  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									return (4, 4, 32, 32)

				def  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_											=				{
									    "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_											=				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  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()
									UpperCamelCase_											=				self.model_class(**snake_case__      )

									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  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()

									UpperCamelCase_											=				32

									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )
									model.eval()

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      )

														if isinstance(snake_case__       ,	snake_case__      ):
																			UpperCamelCase_											=				output.sample

									self.assertIsNotNone(snake_case__      )
									UpperCamelCase_											=				inputs_dict["sample"].shape
									self.assertEqual(output.shape       ,	snake_case__       ,	"Input and output shapes do not match"      )

				def  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()
									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )
									model.eval()

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

														UpperCamelCase_											=				model(**snake_case__      )
														if isinstance(snake_case__       ,	snake_case__      ):
																			UpperCamelCase_											=				first.sample

														UpperCamelCase_											=				model(**snake_case__      )
														if isinstance(snake_case__       ,	snake_case__      ):
																			UpperCamelCase_											=				second.sample

									UpperCamelCase_											=				first.cpu().numpy()
									UpperCamelCase_											=				second.cpu().numpy()
									UpperCamelCase_											=				out_a[~np.isnan(snake_case__      )]
									UpperCamelCase_											=				out_a[~np.isnan(snake_case__      )]
									UpperCamelCase_											=				np.amax(np.abs(out_a - out_a      )      )
									self.assertLessEqual(snake_case__       ,	1e-5      )

				def  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()

									UpperCamelCase_											=				8

									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )
									model.eval()

									model.set_attention_slice("auto"      )
									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      )
									assert output is not None

									model.set_attention_slice("max"      )
									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      )
									assert output is not None

									model.set_attention_slice(2      )
									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      )
									assert output is not None

				def  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()

									UpperCamelCase_											=				8

									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      ).sample

									UpperCamelCase_											=				create_lora_layers(snake_case__      )

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

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

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__       ,	cross_attention_kwargs={"scale": 0.0}      ).sample
														UpperCamelCase_											=				model(**snake_case__       ,	cross_attention_kwargs={"scale": 0.5}      ).sample
														UpperCamelCase_											=				model(**snake_case__       ,	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  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()

									UpperCamelCase_											=				8

									torch.manual_seed(0      )
									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      ).sample

									UpperCamelCase_											=				create_lora_layers(snake_case__      )
									model.set_attn_processor(snake_case__      )

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__       ,	cross_attention_kwargs={"scale": 0.5}      ).sample

									with tempfile.TemporaryDirectory() as tmpdirname:
														model.save_attn_procs(snake_case__      )
														self.assertTrue(os.path.isfile(os.path.join(snake_case__       ,	"pytorch_lora_weights.bin"      )      )      )
														torch.manual_seed(0      )
														UpperCamelCase_											=				self.model_class(**snake_case__      )
														new_model.to(snake_case__      )
														new_model.load_attn_procs(snake_case__      )

									with torch.no_grad():
														UpperCamelCase_											=				new_model(**snake_case__       ,	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  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()

									UpperCamelCase_											=				8

									torch.manual_seed(0      )
									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      ).sample

									UpperCamelCase_											=				create_lora_layers(snake_case__      )
									model.set_attn_processor(snake_case__      )

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__       ,	cross_attention_kwargs={"scale": 0.5}      ).sample

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

									with torch.no_grad():
														UpperCamelCase_											=				new_model(**snake_case__       ,	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  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()

									UpperCamelCase_											=				8

									torch.manual_seed(0      )
									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )

									UpperCamelCase_											=				create_lora_layers(snake_case__       ,	mock_weights=snake_case__      )
									model.set_attn_processor(snake_case__      )
									# Saving as torch, properly reloads with directly filename
									with tempfile.TemporaryDirectory() as tmpdirname:
														model.save_attn_procs(snake_case__      )
														self.assertTrue(os.path.isfile(os.path.join(snake_case__       ,	"pytorch_lora_weights.bin"      )      )      )
														torch.manual_seed(0      )
														UpperCamelCase_											=				self.model_class(**snake_case__      )
														new_model.to(snake_case__      )
														new_model.load_attn_procs(snake_case__       ,	weight_name="pytorch_lora_weights.bin"      )

				def  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()

									UpperCamelCase_											=				8

									torch.manual_seed(0      )
									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )

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

				def  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()

									UpperCamelCase_											=				8

									torch.manual_seed(0      )
									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      ).sample

									UpperCamelCase_											=				create_lora_layers(snake_case__      )
									model.set_attn_processor(snake_case__      )

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__       ,	cross_attention_kwargs={"scale": 0.0}      ).sample

									model.set_attn_processor(AttnProcessor()      )

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      ).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  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()

									UpperCamelCase_											=				4

									torch.manual_seed(0      )
									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )
									UpperCamelCase_											=				create_lora_layers(snake_case__      )
									model.set_attn_processor(snake_case__      )

									# default
									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      ).sample

														model.enable_xformers_memory_efficient_attention()
														UpperCamelCase_											=				model(**snake_case__      ).sample

														model.disable_xformers_memory_efficient_attention()
														UpperCamelCase_											=				model(**snake_case__      ).sample

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

				def  _lowerCamelCase   (	self      ):

									'''simple docstring'''


									UpperCamelCase_					,      UpperCamelCase_											=				self.prepare_init_args_and_inputs_for_common()
									UpperCamelCase_											=				32

									UpperCamelCase_											=				self.model_class(**snake_case__      )
									model.to(snake_case__      )
									model.eval()

									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      )[0]

									model.enable_forward_chunking()
									with torch.no_grad():
														UpperCamelCase_											=				model(**snake_case__      )[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.





def 				_lowerCAmelCase      (_lowerCAmelCase				,		_lowerCAmelCase):
					UpperCamelCase_											=				0
					for ele in arr:
										total_sum += product_sum(_lowerCAmelCase				,		depth + 1) if isinstance(_lowerCAmelCase				,		_lowerCAmelCase) else ele
					return total_sum * depth



def 				_lowerCAmelCase      (_lowerCAmelCase):
					return product_sum(_lowerCAmelCase				,		1)


if __name__ == "__main__":
	import doctest

	doctest.testmod()
