


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

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






def  _UpperCAmelCase						( __lowerCamelCase			,  __lowerCamelCase = True							)		->Tuple:


 '''simple docstring'''
 snake_case :			List[str]							=   {}
 for name in model.attn_processors.keys():
  snake_case :			Optional[Any]							=   name.endswith('''attn2.processor'''							) and not (
      name.startswith('''transformer_in'''							) or """temp_attentions""" in name.split('''.'''							)
  )
  snake_case :			Tuple							=   model.config.cross_attention_dim if has_cross_attention else None
  if name.startswith('''mid_block'''							):
   snake_case :			Tuple							=   model.config.block_out_channels[-1]
  elif name.startswith('''up_blocks'''							):
   snake_case :			str							=   int(name[len('''up_blocks.'''							)]							)
   snake_case :			Tuple							=   list(reversed(model.config.block_out_channels							)							)[block_id]
  elif name.startswith('''down_blocks'''							):
   snake_case :			List[str]							=   int(name[len('''down_blocks.'''							)]							)
   snake_case :			Tuple							=   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
   snake_case :			List[str]							=   8 * model.config.attention_head_dim

  snake_case :			Optional[Any]							=   LoRAAttnProcessor(hidden_size=UpperCAmelCase__			,  cross_attention_dim=UpperCAmelCase__							)
  snake_case :			Optional[int]							=   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   lowerCamelCase						(	SCREAMING_SNAKE_CASE,      SCREAMING_SNAKE_CASE,      unittest.TestCase  ):
       _UpperCamelCase     		=   UNetaDConditionModel
       _UpperCamelCase     		=   """sample"""

       @property
       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        snake_case :			Dict							=   4
        snake_case :			str							=   4
        snake_case :			str							=   4
        snake_case :			Union[str, Any]							=   (32, 32)

        snake_case :			List[Any]							=   floats_tensor((batch_size, num_channels, num_frames) + sizes       ).to(a__       )
        snake_case :			Any							=   torch.tensor([10]       ).to(a__       )
        snake_case :			Optional[Any]							=   floats_tensor((batch_size, 4, 32)       ).to(a__       )

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

       @property
       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        return (4, 4, 32, 32)

       @property
       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        return (4, 4, 32, 32)

       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        snake_case :			Tuple							=   {
            """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,
        }
        snake_case :			Optional[Any]							=   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       ):






        '''simple docstring'''






        snake_case :			Optional[Any]							=   self.prepare_init_args_and_inputs_for_common()
        snake_case :			Dict							=   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       ):






        '''simple docstring'''






        snake_case :			Optional[int]							=   self.prepare_init_args_and_inputs_for_common()

        snake_case :			Tuple							=   32

        snake_case :			List[Any]							=   self.model_class(**a__       )
        model.to(a__       )
        model.eval()

        with torch.no_grad():
         snake_case :			Dict							=   model(**a__       )

         if isinstance(a__						,    a__       ):
          snake_case :			int							=   output.sample

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

       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        snake_case :			Optional[Any]							=   self.prepare_init_args_and_inputs_for_common()
        snake_case :			Any							=   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       )

         snake_case :			List[str]							=   model(**a__       )
         if isinstance(a__						,    a__       ):
          snake_case :			List[Any]							=   first.sample

         snake_case :			List[str]							=   model(**a__       )
         if isinstance(a__						,    a__       ):
          snake_case :			str							=   second.sample

        snake_case :			Optional[Any]							=   first.cpu().numpy()
        snake_case :			Tuple							=   second.cpu().numpy()
        snake_case :			List[Any]							=   out_a[~np.isnan(a__       )]
        snake_case :			Dict							=   out_a[~np.isnan(a__       )]
        snake_case :			str							=   np.amax(np.abs(out_a - out_a       )       )
        self.assertLessEqual(a__						,    1e-5       )

       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        snake_case :			List[Any]							=   self.prepare_init_args_and_inputs_for_common()

        snake_case :			List[str]							=   8

        snake_case :			Optional[Any]							=   self.model_class(**a__       )
        model.to(a__       )
        model.eval()

        model.set_attention_slice('''auto'''       )
        with torch.no_grad():
         snake_case :			List[str]							=   model(**a__       )
        assert output is not None

        model.set_attention_slice('''max'''       )
        with torch.no_grad():
         snake_case :			Union[str, Any]							=   model(**a__       )
        assert output is not None

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

       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        snake_case :			Dict							=   self.prepare_init_args_and_inputs_for_common()

        snake_case :			List[Any]							=   8

        snake_case :			Union[str, Any]							=   self.model_class(**a__       )
        model.to(a__       )

        with torch.no_grad():
         snake_case :			Optional[int]							=   model(**a__       ).sample

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






        '''simple docstring'''






        snake_case :			Dict							=   self.prepare_init_args_and_inputs_for_common()

        snake_case :			Optional[Any]							=   8

        torch.manual_seed(0       )
        snake_case :			str							=   self.model_class(**a__       )
        model.to(a__       )

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

        snake_case :			str							=   create_lora_layers(a__       )
        model.set_attn_processor(a__       )

        with torch.no_grad():
         snake_case :			Union[str, Any]							=   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       )
         snake_case :			Union[str, Any]							=   self.model_class(**a__       )
         new_model.to(a__       )
         new_model.load_attn_procs(a__       )

        with torch.no_grad():
         snake_case :			List[Any]							=   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       ):






        '''simple docstring'''






        snake_case :			Optional[Any]							=   self.prepare_init_args_and_inputs_for_common()

        snake_case :			Optional[Any]							=   8

        torch.manual_seed(0       )
        snake_case :			Any							=   self.model_class(**a__       )
        model.to(a__       )

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

        snake_case :			Union[str, Any]							=   create_lora_layers(a__       )
        model.set_attn_processor(a__       )

        with torch.no_grad():
         snake_case :			Tuple							=   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       )
         snake_case :			Optional[Any]							=   self.model_class(**a__       )
         new_model.to(a__       )
         new_model.load_attn_procs(a__       )

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






        '''simple docstring'''






        snake_case :			List[Any]							=   self.prepare_init_args_and_inputs_for_common()

        snake_case :			Any							=   8

        torch.manual_seed(0       )
        snake_case :			str							=   self.model_class(**a__       )
        model.to(a__       )

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

       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        snake_case :			Union[str, Any]							=   self.prepare_init_args_and_inputs_for_common()

        snake_case :			Optional[Any]							=   8

        torch.manual_seed(0       )
        snake_case :			Any							=   self.model_class(**a__       )
        model.to(a__       )

        snake_case :			Dict							=   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       )
         snake_case :			Optional[Any]							=   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       ):






        '''simple docstring'''






        snake_case :			Dict							=   self.prepare_init_args_and_inputs_for_common()

        snake_case :			Tuple							=   8

        torch.manual_seed(0       )
        snake_case :			Optional[int]							=   self.model_class(**a__       )
        model.to(a__       )

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

        snake_case :			Union[str, Any]							=   create_lora_layers(a__       )
        model.set_attn_processor(a__       )

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

        model.set_attn_processor(AttnProcessor()       )

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






        '''simple docstring'''






        snake_case :			List[str]							=   self.prepare_init_args_and_inputs_for_common()

        snake_case :			List[Any]							=   4

        torch.manual_seed(0       )
        snake_case :			Tuple							=   self.model_class(**a__       )
        model.to(a__       )
        snake_case :			List[str]							=   create_lora_layers(a__       )
        model.set_attn_processor(a__       )

        # default
        with torch.no_grad():
         snake_case :			Optional[Any]							=   model(**a__       ).sample

         model.enable_xformers_memory_efficient_attention()
         snake_case :			List[str]							=   model(**a__       ).sample

         model.disable_xformers_memory_efficient_attention()
         snake_case :			Union[str, Any]							=   model(**a__       ).sample

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



       def 					UpperCAmelCase					(      self       ):






        '''simple docstring'''






        snake_case :			Optional[Any]							=   self.prepare_init_args_and_inputs_for_common()
        snake_case :			List[Any]							=   32

        snake_case :			Any							=   self.model_class(**a__       )
        model.to(a__       )
        model.eval()

        with torch.no_grad():
         snake_case :			List[Any]							=   model(**a__       )[0]

        model.enable_forward_chunking()
        with torch.no_grad():
         snake_case :			Union[str, Any]							=   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.














"""simple docstring"""



from itertools import combinations, product




def   _lowerCAmelCase		(				UpperCAmelCase__							:       int,    UpperCAmelCase__							:       int,    UpperCAmelCase__							:       int,    UpperCAmelCase__							:       int   )		->bool:
						if xa == ya == 0 or xa == ya == 0:
												return False
						A__    :	List[str]			=	xa * xa + ya * ya
						A__    :	List[Any]			=	xa * xa + ya * ya
						A__    :	Optional[int]			=	(xa - xa) * (xa - xa) + (ya - ya) * (ya - ya)
						return (
						    a_square + b_square == c_square
						    or a_square + c_square == b_square
						    or b_square + c_square == a_square
						)




def   _lowerCAmelCase		(				UpperCAmelCase__							:       int = 5_0   )		->int:
						return sum(
						    1
						    for pta, pta in combinations(product(range(limit + 1   ),    repeat=2   ),    2   )
						    if is_right(*UpperCAmelCase__,    *UpperCAmelCase__   )   )


if __name__ == "__main__":
						print(F'{solution() = }')






