


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()

snake_case_  :	Union[str, Any]							  =							logging.get_logger(__name__)







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

								_SCREAMING_SNAKE_CASE	     =LoRAAttnProcessor(hidden_size=a__		,cross_attention_dim=a__)
								_SCREAMING_SNAKE_CASE	     =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   A__							(						UpperCamelCase__      ,   UpperCamelCase__      ,   unittest.TestCase     ):
						UpperCAmelCase									=  UNetaDConditionModel
						UpperCAmelCase									=  "sample"




						@property
						def 							__UpperCamelCase (				self    :					List[str]     )							->   List[str]:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE	     =4
										_SCREAMING_SNAKE_CASE	     =4
										_SCREAMING_SNAKE_CASE	     =4
										_SCREAMING_SNAKE_CASE	     =(32, 32)

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

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




						@property
						def 							__UpperCamelCase (				self    :					Tuple     )							->   Optional[Any]:


										"""simple docstring"""





										return (4, 4, 32, 32)




						@property
						def 							__UpperCamelCase (				self    :					Union[str, Any]     )							->   Any:


										"""simple docstring"""





										return (4, 4, 32, 32)




						def 							__UpperCamelCase (				self    :					Tuple     )							->   Dict:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE	     ={
										    '''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,
										}
										_SCREAMING_SNAKE_CASE	     =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 							__UpperCamelCase (				self    :					Dict     )							->   Optional[int]:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()
										_SCREAMING_SNAKE_CASE	     =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 							__UpperCamelCase (				self    :					Union[str, Any]     )							->   Dict:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()

										_SCREAMING_SNAKE_CASE	     =32

										_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
										model.to(_a     )
										model.eval()

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =model(**_a     )

														if isinstance(_a ,			_a     ):
																		_SCREAMING_SNAKE_CASE	     =output.sample

										self.assertIsNotNone(_a     )
										_SCREAMING_SNAKE_CASE	     =inputs_dict['''sample'''].shape
										self.assertEqual(output.shape ,			_a ,			'''Input and output shapes do not match'''     )




						def 							__UpperCamelCase (				self    :					Optional[int]     )							->   int:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()
										_SCREAMING_SNAKE_CASE	     =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     )

														_SCREAMING_SNAKE_CASE	     =model(**_a     )
														if isinstance(_a ,			_a     ):
																		_SCREAMING_SNAKE_CASE	     =first.sample

														_SCREAMING_SNAKE_CASE	     =model(**_a     )
														if isinstance(_a ,			_a     ):
																		_SCREAMING_SNAKE_CASE	     =second.sample

										_SCREAMING_SNAKE_CASE	     =first.cpu().numpy()
										_SCREAMING_SNAKE_CASE	     =second.cpu().numpy()
										_SCREAMING_SNAKE_CASE	     =out_a[~np.isnan(_a     )]
										_SCREAMING_SNAKE_CASE	     =out_a[~np.isnan(_a     )]
										_SCREAMING_SNAKE_CASE	     =np.amax(np.abs(out_a - out_a     )     )
										self.assertLessEqual(_a ,			1E-5     )




						def 							__UpperCamelCase (				self    :					List[Any]     )							->   Dict:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()

										_SCREAMING_SNAKE_CASE	     =8

										_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
										model.to(_a     )
										model.eval()

										model.set_attention_slice('''auto'''     )
										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =model(**_a     )
										assert output is not None

										model.set_attention_slice('''max'''     )
										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =model(**_a     )
										assert output is not None

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




						def 							__UpperCamelCase (				self    :					Dict     )							->   Dict:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()

										_SCREAMING_SNAKE_CASE	     =8

										_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
										model.to(_a     )

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =model(**_a     ).sample

										_SCREAMING_SNAKE_CASE	     =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():
														_SCREAMING_SNAKE_CASE	     =model(**_a ,			cross_attention_kwargs={'''scale''': 0.0}     ).sample
														_SCREAMING_SNAKE_CASE	     =model(**_a ,			cross_attention_kwargs={'''scale''': 0.5}     ).sample
														_SCREAMING_SNAKE_CASE	     =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 							__UpperCamelCase (				self    :					Tuple     )							->   Tuple:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()

										_SCREAMING_SNAKE_CASE	     =8

										torch.manual_seed(0     )
										_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
										model.to(_a     )

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =model(**_a     ).sample

										_SCREAMING_SNAKE_CASE	     =create_lora_layers(_a     )
										model.set_attn_processor(_a     )

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =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     )
														_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
														new_model.to(_a     )
														new_model.load_attn_procs(_a     )

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =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 							__UpperCamelCase (				self    :					Optional[Any]     )							->   str:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()

										_SCREAMING_SNAKE_CASE	     =8

										torch.manual_seed(0     )
										_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
										model.to(_a     )

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =model(**_a     ).sample

										_SCREAMING_SNAKE_CASE	     =create_lora_layers(_a     )
										model.set_attn_processor(_a     )

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =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     )
														_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
														new_model.to(_a     )
														new_model.load_attn_procs(_a     )

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =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 							__UpperCamelCase (				self    :					Union[str, Any]     )							->   Optional[int]:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()

										_SCREAMING_SNAKE_CASE	     =8

										torch.manual_seed(0     )
										_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
										model.to(_a     )

										_SCREAMING_SNAKE_CASE	     =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     )
														_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
														new_model.to(_a     )
														new_model.load_attn_procs(_a ,			weight_name='''pytorch_lora_weights.bin'''     )




						def 							__UpperCamelCase (				self    :					int     )							->   Union[str, Any]:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()

										_SCREAMING_SNAKE_CASE	     =8

										torch.manual_seed(0     )
										_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
										model.to(_a     )

										_SCREAMING_SNAKE_CASE	     =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     )
														_SCREAMING_SNAKE_CASE	     =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 							__UpperCamelCase (				self    :					Optional[int]     )							->   Any:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()

										_SCREAMING_SNAKE_CASE	     =8

										torch.manual_seed(0     )
										_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
										model.to(_a     )

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =model(**_a     ).sample

										_SCREAMING_SNAKE_CASE	     =create_lora_layers(_a     )
										model.set_attn_processor(_a     )

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =model(**_a ,			cross_attention_kwargs={'''scale''': 0.0}     ).sample

										model.set_attn_processor(AttnProcessor()     )

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =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 							__UpperCamelCase (				self    :					Optional[Any]     )							->   Dict:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE     , _SCREAMING_SNAKE_CASE	     =self.prepare_init_args_and_inputs_for_common()

										_SCREAMING_SNAKE_CASE	     =4

										torch.manual_seed(0     )
										_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
										model.to(_a     )
										_SCREAMING_SNAKE_CASE	     =create_lora_layers(_a     )
										model.set_attn_processor(_a     )

										# default
										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =model(**_a     ).sample

														model.enable_xformers_memory_efficient_attention()
														_SCREAMING_SNAKE_CASE	     =model(**_a     ).sample

														model.disable_xformers_memory_efficient_attention()
														_SCREAMING_SNAKE_CASE	     =model(**_a     ).sample

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




						def 							__UpperCamelCase (				self    :					Union[str, Any]     )							->   Union[str, Any]:


										"""simple docstring"""





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

										_SCREAMING_SNAKE_CASE	     =self.model_class(**_a     )
										model.to(_a     )
										model.eval()

										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =model(**_a     )[0]

										model.enable_forward_chunking()
										with torch.no_grad():
														_SCREAMING_SNAKE_CASE	     =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.



# limitations under the License.


from typing import Optional, Tuple, Union

import torch

from diffusers.pipeline_utils import DiffusionPipeline, ImagePipelineOutput

class   A__							(						UpperCamelCase__     ):




						def __init__(				self    :					Any ,			_a    :					Optional[Any] ,			_a    :					List[str]     )							->   Tuple:


										"""simple docstring"""





										super().__init__()
										self.register_modules(unet=_a ,			scheduler=_a     )




						@torch.no_grad()
						def __call__(				self    :					Any ,			_a    :					int = 1 ,			_a    :					Optional[torch.Generator] = None ,			_a    :					int = 50 ,			_a    :					Optional[str] = "pil" ,			_a    :					bool = True ,			**_a    :					Dict ,			)							->   Union[ImagePipelineOutput, Tuple]:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE	     =torch.randn(
										    (batch_size, self.unet.config.in_channels, self.unet.config.sample_size, self.unet.config.sample_size) ,			generator=_a ,			)
										_SCREAMING_SNAKE_CASE	     =image.to(self.device     )

										# set step values
										self.scheduler.set_timesteps(_a     )

										for t in self.progress_bar(self.scheduler.timesteps     ):
														# 1. predict noise model_output
														_SCREAMING_SNAKE_CASE	     =self.unet(_a ,			_a     ).sample

														# 2. predict previous mean of image x_t-1 and add variance depending on eta
														# eta corresponds to η in paper and should be between [0, 1]
														# do x_t -> x_t-1
														_SCREAMING_SNAKE_CASE	     =self.scheduler.step(_a ,			_a ,			_a     ).prev_sample

										_SCREAMING_SNAKE_CASE	     =(image / 2 + 0.5).clamp(0 ,			1     )
										_SCREAMING_SNAKE_CASE	     =image.cpu().permute(0 ,			2 ,			3 ,			1     ).numpy()
										if output_type == "pil":
														_SCREAMING_SNAKE_CASE	     =self.numpy_to_pil(_a     )

										if not return_dict:
														return (image,), "This is a local test"

										return ImagePipelineOutput(images=_a     ), "This is a local test"