




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_						:					List[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 		__UpperCAmelCase(		lowercase__       ,	lowercase__       ,	unittest.TestCase ):





     """simple docstring"""



     __UpperCAmelCase	: Optional[Any]						=    UNetaDConditionModel
     __UpperCAmelCase	: List[Any]						=    """sample"""

     @property
     def 				__SCREAMING_SNAKE_CASE		(      self :       str    )			->			int:






         """simple docstring"""
         a__      =       4
         a__      =       4
         a__      =       4
         a__      =       (32, 32)

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

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

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






         """simple docstring"""
         return (4, 4, 32, 32)

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






         """simple docstring"""
         return (4, 4, 32, 32)

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






         """simple docstring"""
         a__      =       {
             """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,
         }
         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 				__SCREAMING_SNAKE_CASE		(      self :       List[str]    )			->			Union[str, Any]:






         """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 				__SCREAMING_SNAKE_CASE		(      self :       Any    )			->			Any:






         """simple docstring"""
         a__ ,		a__      =       self.prepare_init_args_and_inputs_for_common()

         a__      =       32

         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 				__SCREAMING_SNAKE_CASE		(      self :       List[str]    )			->			Dict:






         """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 				__SCREAMING_SNAKE_CASE		(      self :       Optional[Any]    )			->			List[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 				__SCREAMING_SNAKE_CASE		(      self :       Optional[int]    )			->			Dict:






         """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 				__SCREAMING_SNAKE_CASE		(      self :       Any    )			->			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 				__SCREAMING_SNAKE_CASE		(      self :       Any    )			->			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						,				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 				__SCREAMING_SNAKE_CASE		(      self :       int    )			->			Union[str, Any]:






         """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 				__SCREAMING_SNAKE_CASE		(      self :       str    )			->			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    )

         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 				__SCREAMING_SNAKE_CASE		(      self :       Tuple    )			->			List[str]:






         """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 				__SCREAMING_SNAKE_CASE		(      self :       str    )			->			int:






         """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 				__SCREAMING_SNAKE_CASE		(      self :       str    )			->			Any:






         """simple docstring"""
         a__ ,		a__      =       self.prepare_init_args_and_inputs_for_common()
         a__      =       32

         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.












import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
 import torch

if is_vision_available():
 from PIL import Image

 from transformers import SwinaSRImageProcessor
 from transformers.image_transforms import get_image_size







class 		__UpperCAmelCase(		unittest.TestCase ):





     """simple docstring"""



     def __init__(      self :       Optional[int]						,				A :       Dict						,				A :       Union[str, Any]=7						,				A :       List[str]=3						,				A :       Optional[int]=18						,				A :       Optional[int]=30						,				A :       Optional[Any]=4_00						,				A :       Optional[Any]=True						,				A :       Tuple=1 / 2_55						,				A :       Tuple=True						,				A :       Optional[int]=8						,				)			->			Any:






         """simple docstring"""
         a__      =       parent
         a__      =       batch_size
         a__      =       num_channels
         a__      =       image_size
         a__      =       min_resolution
         a__      =       max_resolution
         a__      =       do_rescale
         a__      =       rescale_factor
         a__      =       do_pad
         a__      =       pad_size

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






         """simple docstring"""
         return {
             "do_rescale": self.do_rescale,
             "rescale_factor": self.rescale_factor,
             "do_pad": self.do_pad,
             "pad_size": self.pad_size,
         }






     def 				__SCREAMING_SNAKE_CASE		(      self :       int						,				A :       int=False						,				A :       Any=False						,				A :       List[Any]=False    )			->			Optional[Any]:






         """simple docstring"""
         assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

         if equal_resolution:
             a__      =       []
             for i in range(self.batch_size    ):
                 image_inputs.append(
                     np.random.randint(
                         2_55						,				size=(self.num_channels, self.max_resolution, self.max_resolution)						,				dtype=np.uinta    )    )
         else:
             a__      =       []
             for i in range(self.batch_size    ):
                 a__ ,		a__      =       np.random.choice(np.arange(self.min_resolution						,				self.max_resolution    )						,				2    )
                 image_inputs.append(np.random.randint(2_55						,				size=(self.num_channels, width, height)						,				dtype=np.uinta    )    )

         if not numpify and not torchify:
             # PIL expects the channel dimension as last dimension
             a__      =       [Image.fromarray(np.moveaxis(A						,				0						,				-1    )    ) for x in image_inputs]

         if torchify:
             a__      =       [torch.from_numpy(A    ) for x in image_inputs]

         return image_inputs







@require_torch
@require_vision
class 		__UpperCAmelCase(		lowercase__       ,	unittest.TestCase ):





     """simple docstring"""



     __UpperCAmelCase	: Dict						=    SwinaSRImageProcessor if is_vision_available() else None

     def 				__SCREAMING_SNAKE_CASE		(      self :       int    )			->			Dict:






         """simple docstring"""
         a__      =       SwinaSRImageProcessingTester(self    )

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






         """simple docstring"""
         return self.image_processor_tester.prepare_image_processor_dict()

     def 				__SCREAMING_SNAKE_CASE		(      self :       Tuple    )			->			int:






         """simple docstring"""
         a__      =       self.image_processing_class(**self.image_processor_dict    )
         self.assertTrue(hasattr(A						,				"""do_rescale"""    )    )
         self.assertTrue(hasattr(A						,				"""rescale_factor"""    )    )
         self.assertTrue(hasattr(A						,				"""do_pad"""    )    )
         self.assertTrue(hasattr(A						,				"""pad_size"""    )    )

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






         """simple docstring"""
         pass

     def 				__SCREAMING_SNAKE_CASE		(      self :       int						,				A :       Optional[Any]    )			->			List[str]:






         """simple docstring"""
         a__ ,		a__      =       get_image_size(A    )
         a__      =       self.image_processor_tester.pad_size

         a__      =       (old_height // size + 1) * size - old_height
         a__      =       (old_width // size + 1) * size - old_width
         return old_height + pad_height, old_width + pad_width

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






         """simple docstring"""
         # Initialize image_processor
         a__      =       self.image_processing_class(**self.image_processor_dict    )
         # create random PIL images
         a__      =       self.image_processor_tester.prepare_inputs(equal_resolution=A    )
         for image in image_inputs:
             self.assertIsInstance(A						,				Image.Image    )

         # Test not batched input
         a__      =       image_processor(image_inputs[0]						,				return_tensors="""pt"""    ).pixel_values
         a__ ,		a__      =       self.calculate_expected_size(np.array(image_inputs[0]    )    )
         self.assertEqual(
             encoded_images.shape						,				(
                 1,
                 self.image_processor_tester.num_channels,
                 expected_height,
                 expected_width,
             )						,				)

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






         """simple docstring"""
         # Initialize image_processor
         a__      =       self.image_processing_class(**self.image_processor_dict    )
         # create random numpy tensors
         a__      =       self.image_processor_tester.prepare_inputs(equal_resolution=A						,				numpify=A    )
         for image in image_inputs:
             self.assertIsInstance(A						,				np.ndarray    )

         # Test not batched input
         a__      =       image_processor(image_inputs[0]						,				return_tensors="""pt"""    ).pixel_values
         a__ ,		a__      =       self.calculate_expected_size(image_inputs[0]    )
         self.assertEqual(
             encoded_images.shape						,				(
                 1,
                 self.image_processor_tester.num_channels,
                 expected_height,
                 expected_width,
             )						,				)






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






         """simple docstring"""
         # Initialize image_processor
         a__      =       self.image_processing_class(**self.image_processor_dict    )
         # create random PyTorch tensors
         a__      =       self.image_processor_tester.prepare_inputs(equal_resolution=A						,				torchify=A    )
         for image in image_inputs:
             self.assertIsInstance(A						,				torch.Tensor    )

         # Test not batched input
         a__      =       image_processor(image_inputs[0]						,				return_tensors="""pt"""    ).pixel_values
         a__ ,		a__      =       self.calculate_expected_size(image_inputs[0]    )
         self.assertEqual(
             encoded_images.shape						,				(
                 1,
                 self.image_processor_tester.num_channels,
                 expected_height,
                 expected_width,
             )						,				)






