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

SCREAMING_SNAKE_CASE   :		Any					 =       logging.get_logger(__name__)



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

								A_       	=  LoRAAttnProcessor(hidden_size=__UpperCamelCase ,						cross_attention_dim=__UpperCamelCase     )
								A_       	=  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       ):
					__magic_name__						:							Optional[Any]     	=	UNetaDConditionModel
					__magic_name__						:							List[str]     	=	'''sample'''


					@property
					def        lowerCAmelCase_    (	self						)		->							List[Any]:





									'''simple docstring'''
									A_       	=  4
									A_       	=  4
									A_       	=  4
									A_       	=  (3_2, 3_2)

									A_       	=  floats_tensor((batch_size, num_channels, num_frames) + sizes						).to(a__						)
									A_       	=  torch.tensor([1_0]						).to(a__						)
									A_       	=  floats_tensor((batch_size, 4, 3_2)						).to(a__						)

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


					@property
					def        lowerCAmelCase_    (	self						)		->							Optional[int]:





									'''simple docstring'''
									return (4, 4, 3_2, 3_2)


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





									'''simple docstring'''
									return (4, 4, 3_2, 3_2)


					def        lowerCAmelCase_    (	self						)		->							List[Any]:





									'''simple docstring'''
									A_       	=  {
									    '''block_out_channels''': (3_2, 6_4),
									    '''down_block_types''': (
									        '''CrossAttnDownBlock3D''',
									        '''DownBlock3D''',
									    ),
									    '''up_block_types''': ('''UpBlock3D''', '''CrossAttnUpBlock3D'''),
									    '''cross_attention_dim''': 3_2,
									    '''attention_head_dim''': 8,
									    '''out_channels''': 4,
									    '''in_channels''': 4,
									    '''layers_per_block''': 1,
									    '''sample_size''': 3_2,
									}
									A_       	=  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						)		->							Tuple:





									'''simple docstring'''
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()
									A_       	=  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        lowerCAmelCase_    (	self						)		->							Union[str, Any]:





									'''simple docstring'''
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()

									A_       	=  3_2

									A_       	=  self.model_class(**a__						)
									model.to(a__						)
									model.eval()

									with torch.no_grad():
													A_       	=  model(**a__						)

													if isinstance(a__       ,       a__						):
																	A_       	=  output.sample

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


					def        lowerCAmelCase_    (	self						)		->							str:





									'''simple docstring'''
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()
									A_       	=  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						)

													A_       	=  model(**a__						)
													if isinstance(a__       ,       a__						):
																	A_       	=  first.sample

													A_       	=  model(**a__						)
													if isinstance(a__       ,       a__						):
																	A_       	=  second.sample

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


					def        lowerCAmelCase_    (	self						)		->							Any:





									'''simple docstring'''
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()

									A_       	=  8

									A_       	=  self.model_class(**a__						)
									model.to(a__						)
									model.eval()

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

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

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


					def        lowerCAmelCase_    (	self						)		->							int:





									'''simple docstring'''
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()

									A_       	=  8

									A_       	=  self.model_class(**a__						)
									model.to(a__						)

									with torch.no_grad():
													A_       	=  model(**a__						).sample

									A_       	=  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():
													A_       	=  model(**a__       ,       cross_attention_kwargs={'''scale''': 0.0}						).sample
													A_       	=  model(**a__       ,       cross_attention_kwargs={'''scale''': 0.5}						).sample
													A_       	=  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        lowerCAmelCase_    (	self						)		->							Optional[Any]:





									'''simple docstring'''
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()

									A_       	=  8

									torch.manual_seed(0						)
									A_       	=  self.model_class(**a__						)
									model.to(a__						)

									with torch.no_grad():
													A_       	=  model(**a__						).sample

									A_       	=  create_lora_layers(a__						)
									model.set_attn_processor(a__						)

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

									with torch.no_grad():
													A_       	=  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        lowerCAmelCase_    (	self						)		->							Dict:





									'''simple docstring'''
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()

									A_       	=  8

									torch.manual_seed(0						)
									A_       	=  self.model_class(**a__						)
									model.to(a__						)

									with torch.no_grad():
													A_       	=  model(**a__						).sample

									A_       	=  create_lora_layers(a__						)
									model.set_attn_processor(a__						)

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

									with torch.no_grad():
													A_       	=  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        lowerCAmelCase_    (	self						)		->							str:





									'''simple docstring'''
									# enable deterministic behavior for gradient checkpointing
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()

									A_       	=  8

									torch.manual_seed(0						)
									A_       	=  self.model_class(**a__						)
									model.to(a__						)

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


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





									'''simple docstring'''
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()

									A_       	=  8

									torch.manual_seed(0						)
									A_       	=  self.model_class(**a__						)
									model.to(a__						)

									A_       	=  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						)
													A_       	=  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        lowerCAmelCase_    (	self						)		->							Tuple:





									'''simple docstring'''
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()

									A_       	=  8

									torch.manual_seed(0						)
									A_       	=  self.model_class(**a__						)
									model.to(a__						)

									with torch.no_grad():
													A_       	=  model(**a__						).sample

									A_       	=  create_lora_layers(a__						)
									model.set_attn_processor(a__						)

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

									model.set_attn_processor(AttnProcessor()						)

									with torch.no_grad():
													A_       	=  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        lowerCAmelCase_    (	self						)		->							Any:





									'''simple docstring'''
									# enable deterministic behavior for gradient checkpointing
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()

									A_       	=  4

									torch.manual_seed(0						)
									A_       	=  self.model_class(**a__						)
									model.to(a__						)
									A_       	=  create_lora_layers(a__						)
									model.set_attn_processor(a__						)

									# default
									with torch.no_grad():
													A_       	=  model(**a__						).sample

													model.enable_xformers_memory_efficient_attention()
													A_       	=  model(**a__						).sample

													model.disable_xformers_memory_efficient_attention()
													A_       	=  model(**a__						).sample

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


					def        lowerCAmelCase_    (	self						)		->							Any:





									'''simple docstring'''
									A_	,					A_       	=  self.prepare_init_args_and_inputs_for_common()
									A_       	=  3_2

									A_       	=  self.model_class(**a__						)
									model.to(a__						)
									model.eval()

									with torch.no_grad():
													A_       	=  model(**a__						)[0]

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

from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import center_crop, normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    IMAGENET_DEFAULT_MEAN,
    IMAGENET_DEFAULT_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    get_image_size,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
						import PIL


SCREAMING_SNAKE_CASE   :		Optional[int]					 =       logging.get_logger(__name__)






class 					__lowercase		(					A       ):
					__magic_name__						:							int     	=	['''pixel_values''']


					def __init__(	self       ,       a__ = True       ,       a__ = None       ,       a__ = True       ,       a__ = None       ,       a__ = PILImageResampling.BICUBIC       ,       a__ = True       ,       a__ = 1 / 2_5_5       ,       a__ = True       ,       a__ = None       ,       a__ = None       ,       **a__       ,       )		->							None:





									'''simple docstring'''
									super().__init__(**a__						)
									A_       	=  crop_size if crop_size is not None else {'''height''': 2_5_6, '''width''': 2_5_6}
									A_       	=  get_size_dict(a__       ,       param_name='''crop_size'''						)
									A_       	=  size if size is not None else {'''height''': 2_2_4, '''width''': 2_2_4}
									A_       	=  get_size_dict(a__						)

									A_       	=  do_center_crop
									A_       	=  crop_size
									A_       	=  do_resize
									A_       	=  size
									A_       	=  resample
									A_       	=  do_rescale
									A_       	=  rescale_factor
									A_       	=  do_normalize
									A_       	=  image_mean if image_mean is not None else IMAGENET_DEFAULT_MEAN
									A_       	=  image_std if image_std is not None else IMAGENET_DEFAULT_STD


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__       ,       a__ = None       ,       a__ = None       ,       **a__       ,       )		->							np.ndarray:





									'''simple docstring'''
									A_       	=  self.size if size is None else size
									A_       	=  get_size_dict(a__						)
									A_       	=  get_size_dict(a__       ,       param_name='''crop_size'''						)

									A_	,					A_       	=  get_image_size(a__						)
									A_       	=  min(a__       ,       a__						)
									A_       	=  (size['''height'''] / crop_size['''height''']) * min_dim
									A_       	=  (size['''width'''] / crop_size['''width''']) * min_dim
									return center_crop(a__       ,       size=(cropped_height, cropped_width)       ,       data_format=a__       ,       **a__						)


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__       ,       a__ = PIL.Image.BICUBIC       ,       a__ = None       ,       **a__       ,       )		->							np.ndarray:





									'''simple docstring'''
									A_       	=  get_size_dict(a__						)
									if "height" not in size or "width" not in size:
													raise ValueError(F"The size dictionary must contain the keys 'height' and 'width'. Got {size.keys()}"						)
									return resize(
									    a__       ,       size=(size['''height'''], size['''width'''])       ,       resample=a__       ,       data_format=a__       ,       **a__						)


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__       ,       a__ = None       ,       **a__       ,       )		->							Union[str, Any]:





									'''simple docstring'''
									return rescale(a__       ,       scale=a__       ,       data_format=a__       ,       **a__						)


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__       ,       a__       ,       a__ = None       ,       **a__       ,       )		->							np.ndarray:





									'''simple docstring'''
									return normalize(a__       ,       mean=a__       ,       std=a__       ,       data_format=a__       ,       **a__						)


					def        lowerCAmelCase_    (	self       ,       a__       ,       a__ = None       ,       a__ = None       ,       a__ = None       ,       a__ = None       ,       a__ = None       ,       a__ = None       ,       a__ = None       ,       a__ = None       ,       a__ = None       ,       a__ = None       ,       a__ = None       ,       a__ = ChannelDimension.FIRST       ,       **a__       ,       )		->							PIL.Image.Image:





									'''simple docstring'''
									A_       	=  do_center_crop if do_center_crop is not None else self.do_center_crop
									A_       	=  crop_size if crop_size is not None else self.crop_size
									A_       	=  get_size_dict(a__       ,       param_name='''crop_size'''						)
									A_       	=  do_resize if do_resize is not None else self.do_resize
									A_       	=  size if size is not None else self.size
									A_       	=  get_size_dict(a__						)
									A_       	=  resample if resample is not None else self.resample
									A_       	=  do_rescale if do_rescale is not None else self.do_rescale
									A_       	=  rescale_factor if rescale_factor is not None else self.rescale_factor
									A_       	=  do_normalize if do_normalize is not None else self.do_normalize
									A_       	=  image_mean if image_mean is not None else self.image_mean
									A_       	=  image_std if image_std is not None else self.image_std

									A_       	=  make_list_of_images(a__						)

									if not valid_images(a__						):
													raise ValueError(
													    '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
													    '''torch.Tensor, tf.Tensor or jax.ndarray.'''						)

									if do_center_crop and crop_size is None:
													raise ValueError('''If `do_center_crop` is set to `True`, `crop_size` must be provided.'''						)

									if do_resize and size is None:
													raise ValueError('''Size must be specified if do_resize is True.'''						)

									if do_rescale and rescale_factor is None:
													raise ValueError('''Rescale factor must be specified if do_rescale is True.'''						)

									if do_normalize and (image_mean is None or image_std is None):
													raise ValueError('''Image mean and image standard deviation must be specified if do_normalize is True.'''						)

									# All transformations expect numpy arrays.
									A_       	=  [to_numpy_array(a__						) for image in images]

									if do_center_crop:
													A_       	=  [self.center_crop(a__       ,       a__       ,       size=a__						) for image in images]

									if do_resize:
													A_       	=  [self.resize(image=a__       ,       size=a__       ,       resample=a__						) for image in images]

									if do_rescale:
													A_       	=  [self.rescale(image=a__       ,       scale=a__						) for image in images]

									if do_normalize:
													A_       	=  [self.normalize(image=a__       ,       mean=a__       ,       std=a__						) for image in images]

									A_       	=  [to_channel_dimension_format(a__       ,       a__						) for image in images]

									A_       	=  {'''pixel_values''': images}
									return BatchFeature(data=a__       ,       tensor_type=a__						)