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

lowerCAmelCase__    :    Optional[Any]			=      logging.get_logger(__name__)



def     a_		(   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=A__   ,     cross_attention_dim=A__			)
								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 							snake_case			(					A_ ,       A_ ,       unittest.TestCase							):




					"""simple docstring"""

					snake_case__							       =					UNetaDConditionModel
					snake_case__							       =					"sample"







					@property
					def 						__lowerCAmelCase		(							self  :      str     ):
									UpperCAmelCase__			 =				4
									UpperCAmelCase__			 =				4
									UpperCAmelCase__			 =				4
									UpperCAmelCase__			 =				(32, 32)

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

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







					@property
					def 						__lowerCAmelCase		(							self  :      Optional[int]     ):
									return (4, 4, 32, 32)







					@property
					def 						__lowerCAmelCase		(							self  :      Any     ):
									return (4, 4, 32, 32)







					def 						__lowerCAmelCase		(							self  :      Any     ):
									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  :      List[str]     ):
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()
									UpperCAmelCase__			 =				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]     ):
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()

									UpperCAmelCase__			 =				32

									UpperCAmelCase__			 =				self.model_class(**_A     )
									model.to(_A     )
									model.eval()

									with torch.no_grad():
													UpperCAmelCase__			 =				model(**_A     )

													if isinstance(_A       ,_A     ):
																	UpperCAmelCase__			 =				output.sample

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







					def 						__lowerCAmelCase		(							self  :      int     ):
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()
									UpperCAmelCase__			 =				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__			 =				model(**_A     )
													if isinstance(_A       ,_A     ):
																	UpperCAmelCase__			 =				first.sample

													UpperCAmelCase__			 =				model(**_A     )
													if isinstance(_A       ,_A     ):
																	UpperCAmelCase__			 =				second.sample

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







					def 						__lowerCAmelCase		(							self  :      int     ):
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()

									UpperCAmelCase__			 =				8

									UpperCAmelCase__			 =				self.model_class(**_A     )
									model.to(_A     )
									model.eval()

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

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

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







					def 						__lowerCAmelCase		(							self  :      int     ):
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()

									UpperCAmelCase__			 =				8

									UpperCAmelCase__			 =				self.model_class(**_A     )
									model.to(_A     )

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

									UpperCAmelCase__			 =				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__			 =				model(**_A       ,cross_attention_kwargs={'scale': 0.0}     ).sample
													UpperCAmelCase__			 =				model(**_A       ,cross_attention_kwargs={'scale': 0.5}     ).sample
													UpperCAmelCase__			 =				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  :      int     ):
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()

									UpperCAmelCase__			 =				8

									torch.manual_seed(0     )
									UpperCAmelCase__			 =				self.model_class(**_A     )
									model.to(_A     )

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

									UpperCAmelCase__			 =				create_lora_layers(_A     )
									model.set_attn_processor(_A     )

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

									with torch.no_grad():
													UpperCAmelCase__			 =				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  :      Optional[Any]     ):
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()

									UpperCAmelCase__			 =				8

									torch.manual_seed(0     )
									UpperCAmelCase__			 =				self.model_class(**_A     )
									model.to(_A     )

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

									UpperCAmelCase__			 =				create_lora_layers(_A     )
									model.set_attn_processor(_A     )

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

									with torch.no_grad():
													UpperCAmelCase__			 =				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  :      Optional[Any]     ):
									# enable deterministic behavior for gradient checkpointing
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()

									UpperCAmelCase__			 =				8

									torch.manual_seed(0     )
									UpperCAmelCase__			 =				self.model_class(**_A     )
									model.to(_A     )

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







					def 						__lowerCAmelCase		(							self  :      Optional[Any]     ):
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()

									UpperCAmelCase__			 =				8

									torch.manual_seed(0     )
									UpperCAmelCase__			 =				self.model_class(**_A     )
									model.to(_A     )

									UpperCAmelCase__			 =				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__			 =				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  :      Any     ):
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()

									UpperCAmelCase__			 =				8

									torch.manual_seed(0     )
									UpperCAmelCase__			 =				self.model_class(**_A     )
									model.to(_A     )

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

									UpperCAmelCase__			 =				create_lora_layers(_A     )
									model.set_attn_processor(_A     )

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

									model.set_attn_processor(AttnProcessor()     )

									with torch.no_grad():
													UpperCAmelCase__			 =				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  :      str     ):
									# enable deterministic behavior for gradient checkpointing
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()

									UpperCAmelCase__			 =				4

									torch.manual_seed(0     )
									UpperCAmelCase__			 =				self.model_class(**_A     )
									model.to(_A     )
									UpperCAmelCase__			 =				create_lora_layers(_A     )
									model.set_attn_processor(_A     )

									# default
									with torch.no_grad():
													UpperCAmelCase__			 =				model(**_A     ).sample

													model.enable_xformers_memory_efficient_attention()
													UpperCAmelCase__			 =				model(**_A     ).sample

													model.disable_xformers_memory_efficient_attention()
													UpperCAmelCase__			 =				model(**_A     ).sample

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







					def 						__lowerCAmelCase		(							self  :      Tuple     ):
									UpperCAmelCase__ ,      UpperCAmelCase__			 =				self.prepare_init_args_and_inputs_for_common()
									UpperCAmelCase__			 =				32

									UpperCAmelCase__			 =				self.model_class(**_A     )
									model.to(_A     )
									model.eval()

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

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





import logging

# standard libraries
import sys

import numpy as np
import pytest  # type: ignore

# Custom/local libraries
from matrix import matrix_operation as matop

_A       =					[[12, 10], [3, 9]]
_A       =					[[3, 4], [7, 4]]
_A       =					[[3, 0, 2], [2, 0, -2], [0, 1, 1]]
_A       =					[[3, 0, -2], [2, 0, 2], [0, 1, 1]]
_A       =					[[3, 0, 2], [2, 0, -2], [0, 1, 1], [2, 0, -2]]
_A       =					[1]
_A       =					[2]

_A       =					logging.getLogger()
_A       =					logging.DEBUG
_A       =					logging.StreamHandler(sys.stdout)
logger.addHandler(stream_handler)



@pytest.mark.mat_ops
@pytest.mark.parametrize(
    ("mat1", "mat2")					,						[(mat_a, mat_b), (mat_c, mat_d), (mat_d, mat_e), (mat_f, mat_h)]						)
def 		lowercase_     (  A__					,						A__						)				->				Dict:
    """simple docstring"""

    if (np.array(A__						)).shape < (2, 2) or (np.array(A__						)).shape < (2, 2):
        with pytest.raises(A__						):
            logger.info(F'\n\t{test_addition.__name__} returned integer'						)
            matop.add(A__					,						A__						)
    elif (np.array(A__						)).shape == (np.array(A__						)).shape:
        logger.info(F'\n\t{test_addition.__name__} with same matrix dims'						)
        snake_case						=   (np.array(A__						) + np.array(A__						)).tolist()
        snake_case						=   matop.add(A__					,						A__						)
        assert theo == act
    else:
        with pytest.raises(A__						):
            logger.info(F'\n\t{test_addition.__name__} with different matrix dims'						)
            matop.add(A__					,						A__						)



@pytest.mark.mat_ops
@pytest.mark.parametrize(
    ("mat1", "mat2")					,						[(mat_a, mat_b), (mat_c, mat_d), (mat_d, mat_e), (mat_f, mat_h)]						)
def 		lowercase_     (  A__					,						A__						)				->				List[Any]:
    """simple docstring"""

    if (np.array(A__						)).shape < (2, 2) or (np.array(A__						)).shape < (2, 2):
        with pytest.raises(A__						):
            logger.info(F'\n\t{test_subtraction.__name__} returned integer'						)
            matop.subtract(A__					,						A__						)
    elif (np.array(A__						)).shape == (np.array(A__						)).shape:
        logger.info(F'\n\t{test_subtraction.__name__} with same matrix dims'						)
        snake_case						=   (np.array(A__						) - np.array(A__						)).tolist()
        snake_case						=   matop.subtract(A__					,						A__						)
        assert theo == act
    else:
        with pytest.raises(A__						):
            logger.info(F'\n\t{test_subtraction.__name__} with different matrix dims'						)
            assert matop.subtract(A__					,						A__						)



@pytest.mark.mat_ops
@pytest.mark.parametrize(
    ("mat1", "mat2")					,						[(mat_a, mat_b), (mat_c, mat_d), (mat_d, mat_e), (mat_f, mat_h)]						)
def 		lowercase_     (  A__					,						A__						)				->				Union[str, Any]:
    """simple docstring"""

    if (np.array(A__						)).shape < (2, 2) or (np.array(A__						)).shape < (2, 2):
        logger.info(F'\n\t{test_multiplication.__name__} returned integer'						)
        with pytest.raises(A__						):
            matop.add(A__					,						A__						)
    elif (np.array(A__						)).shape == (np.array(A__						)).shape:
        logger.info(F'\n\t{test_multiplication.__name__} meets dim requirements'						)
        snake_case						=   (np.matmul(A__					,						A__						)).tolist()
        snake_case						=   matop.multiply(A__					,						A__						)
        assert theo == act
    else:
        with pytest.raises(A__						):
            logger.info(
                F'\n\t{test_multiplication.__name__} does not meet dim requirements'						)
            assert matop.subtract(A__					,						A__						)



@pytest.mark.mat_ops
def 		lowercase_     (  )				->				int:
    """simple docstring"""

    snake_case						=   (3.5 * np.array(A__						)).tolist()
    snake_case						=   matop.scalar_multiply(A__					,						3.5						)
    assert theo == act



@pytest.mark.mat_ops
def 		lowercase_     (  )				->				Any:
    """simple docstring"""

    snake_case						=   (np.identity(5						)).tolist()
    snake_case						=   matop.identity(5						)
    assert theo == act





@pytest.mark.mat_ops
@pytest.mark.parametrize("mat"					,						[mat_a, mat_b, mat_c, mat_d, mat_e, mat_f]						)
def 		lowercase_     (  A__						)				->				List[str]:
    """simple docstring"""

    if (np.array(A__						)).shape < (2, 2):
        with pytest.raises(A__						):
            logger.info(F'\n\t{test_transpose.__name__} returned integer'						)
            matop.transpose(A__						)
    else:
        snake_case						=   (np.transpose(A__						)).tolist()
        snake_case						=   matop.transpose(A__					,						return_map=A__						)
        assert theo == act

