

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_					  =   logging.get_logger(__name__)



def   UpperCamelCase(   lowercase_     ,   lowercase_ = True				)      ->							List[Any]:





  '''simple docstring'''





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

    snake_case_					= LoRAAttnProcessor(hidden_size=_lowercase     ,   cross_attention_dim=_lowercase				)
    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 					__lowerCamelCase							(      _lowerCAmelCase    ,				_lowerCAmelCase    ,				unittest.TestCase      ):
      lowerCamelCase_      : Optional[Any]  			=   UNetaDConditionModel
      lowerCamelCase_      : Any  			=   """sample"""







      @property
      def       lowerCAmelCase_       (	self  )		->			Optional[Any]:
        snake_case_					= 4
        snake_case_					= 4
        snake_case_					= 4
        snake_case_					= (32, 32)

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

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







      @property
      def       lowerCAmelCase_       (	self  )		->			int:
        return (4, 4, 32, 32)







      @property
      def       lowerCAmelCase_       (	self  )		->			List[str]:
        return (4, 4, 32, 32)







      def       lowerCAmelCase_       (	self  )		->			List[str]:
        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,
        }
        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       lowerCAmelCase_       (	self  )		->			List[str]:
        snake_case_					= self.prepare_init_args_and_inputs_for_common()
        snake_case_					= self.model_class(**lowerCamelCase  )

        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  )		->			Tuple:
        snake_case_					= self.prepare_init_args_and_inputs_for_common()

        snake_case_					= 32

        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )
        model.eval()

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  )

          if isinstance(lowerCamelCase				,    lowerCamelCase  ):
            snake_case_					= output.sample

        self.assertIsNotNone(lowerCamelCase  )
        snake_case_					= inputs_dict["""sample"""].shape
        self.assertEqual(output.shape				,    lowerCamelCase				,    """Input and output shapes do not match"""  )







      def       lowerCAmelCase_       (	self  )		->			Optional[Any]:
        snake_case_					= self.prepare_init_args_and_inputs_for_common()
        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )
        model.eval()

        with torch.no_grad():
          # Warmup pass when using mps (see #372)
          if torch_device == "mps" and isinstance(lowerCamelCase				,    lowerCamelCase  ):
            model(**self.dummy_input  )

          snake_case_					= model(**lowerCamelCase  )
          if isinstance(lowerCamelCase				,    lowerCamelCase  ):
            snake_case_					= first.sample

          snake_case_					= model(**lowerCamelCase  )
          if isinstance(lowerCamelCase				,    lowerCamelCase  ):
            snake_case_					= second.sample

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







      def       lowerCAmelCase_       (	self  )		->			int:
        snake_case_					= self.prepare_init_args_and_inputs_for_common()

        snake_case_					= 8

        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )
        model.eval()

        model.set_attention_slice("""auto"""  )
        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  )
        assert output is not None

        model.set_attention_slice("""max"""  )
        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  )
        assert output is not None

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







      def       lowerCAmelCase_       (	self  )		->			Optional[Any]:
        snake_case_					= self.prepare_init_args_and_inputs_for_common()

        snake_case_					= 8

        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  ).sample

        snake_case_					= create_lora_layers(lowerCamelCase  )

        # make sure we can set a list of attention processors
        model.set_attn_processor(lowerCamelCase  )
        model.to(lowerCamelCase  )

        # test that attn processors can be set to itself
        model.set_attn_processor(model.attn_processors  )

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase				,    cross_attention_kwargs={"""scale""": 0.0}  ).sample
          snake_case_					= model(**lowerCamelCase				,    cross_attention_kwargs={"""scale""": 0.5}  ).sample
          snake_case_					= model(**lowerCamelCase				,    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]:
        snake_case_					= self.prepare_init_args_and_inputs_for_common()

        snake_case_					= 8

        torch.manual_seed(0  )
        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  ).sample

        snake_case_					= create_lora_layers(lowerCamelCase  )
        model.set_attn_processor(lowerCamelCase  )

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase				,    cross_attention_kwargs={"""scale""": 0.5}  ).sample

        with tempfile.TemporaryDirectory() as tmpdirname:
          model.save_attn_procs(lowerCamelCase  )
          self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase				,    """pytorch_lora_weights.bin"""  )  )  )
          torch.manual_seed(0  )
          snake_case_					= self.model_class(**lowerCamelCase  )
          new_model.to(lowerCamelCase  )
          new_model.load_attn_procs(lowerCamelCase  )

        with torch.no_grad():
          snake_case_					= new_model(**lowerCamelCase				,    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  )		->			Union[str, Any]:
        snake_case_					= self.prepare_init_args_and_inputs_for_common()

        snake_case_					= 8

        torch.manual_seed(0  )
        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  ).sample

        snake_case_					= create_lora_layers(lowerCamelCase  )
        model.set_attn_processor(lowerCamelCase  )

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase				,    cross_attention_kwargs={"""scale""": 0.5}  ).sample

        with tempfile.TemporaryDirectory() as tmpdirname:
          model.save_attn_procs(lowerCamelCase				,    safe_serialization=lowerCamelCase  )
          self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase				,    """pytorch_lora_weights.safetensors"""  )  )  )
          torch.manual_seed(0  )
          snake_case_					= self.model_class(**lowerCamelCase  )
          new_model.to(lowerCamelCase  )
          new_model.load_attn_procs(lowerCamelCase  )

        with torch.no_grad():
          snake_case_					= new_model(**lowerCamelCase				,    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
        snake_case_					= self.prepare_init_args_and_inputs_for_common()

        snake_case_					= 8

        torch.manual_seed(0  )
        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )

        snake_case_					= create_lora_layers(lowerCamelCase				,    mock_weights=lowerCamelCase  )
        model.set_attn_processor(lowerCamelCase  )
        # Saving as torch, properly reloads with directly filename
        with tempfile.TemporaryDirectory() as tmpdirname:
          model.save_attn_procs(lowerCamelCase  )
          self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase				,    """pytorch_lora_weights.bin"""  )  )  )
          torch.manual_seed(0  )
          snake_case_					= self.model_class(**lowerCamelCase  )
          new_model.to(lowerCamelCase  )
          new_model.load_attn_procs(lowerCamelCase				,    weight_name="""pytorch_lora_weights.bin"""  )







      def       lowerCAmelCase_       (	self  )		->			Tuple:
        snake_case_					= self.prepare_init_args_and_inputs_for_common()

        snake_case_					= 8

        torch.manual_seed(0  )
        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )

        snake_case_					= create_lora_layers(lowerCamelCase				,    mock_weights=lowerCamelCase  )
        model.set_attn_processor(lowerCamelCase  )
        # Saving as torch, properly reloads with directly filename
        with tempfile.TemporaryDirectory() as tmpdirname:
          model.save_attn_procs(lowerCamelCase  )
          self.assertTrue(os.path.isfile(os.path.join(lowerCamelCase				,    """pytorch_lora_weights.bin"""  )  )  )
          torch.manual_seed(0  )
          snake_case_					= self.model_class(**lowerCamelCase  )
          new_model.to(lowerCamelCase  )
          with self.assertRaises(lowerCamelCase  ) as e:
            new_model.load_attn_procs(lowerCamelCase				,    use_safetensors=lowerCamelCase  )
          self.assertIn("""Error no file named pytorch_lora_weights.safetensors"""				,    str(e.exception  )  )







      def       lowerCAmelCase_       (	self  )		->			Dict:
        snake_case_					= self.prepare_init_args_and_inputs_for_common()

        snake_case_					= 8

        torch.manual_seed(0  )
        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  ).sample

        snake_case_					= create_lora_layers(lowerCamelCase  )
        model.set_attn_processor(lowerCamelCase  )

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase				,    cross_attention_kwargs={"""scale""": 0.0}  ).sample

        model.set_attn_processor(AttnProcessor()  )

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  ).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  )		->			Union[str, Any]:
        # enable deterministic behavior for gradient checkpointing
        snake_case_					= self.prepare_init_args_and_inputs_for_common()

        snake_case_					= 4

        torch.manual_seed(0  )
        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )
        snake_case_					= create_lora_layers(lowerCamelCase  )
        model.set_attn_processor(lowerCamelCase  )

        # default
        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  ).sample

          model.enable_xformers_memory_efficient_attention()
          snake_case_					= model(**lowerCamelCase  ).sample

          model.disable_xformers_memory_efficient_attention()
          snake_case_					= model(**lowerCamelCase  ).sample

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







      def       lowerCAmelCase_       (	self  )		->			str:
        snake_case_					= self.prepare_init_args_and_inputs_for_common()
        snake_case_					= 32

        snake_case_					= self.model_class(**lowerCamelCase  )
        model.to(lowerCamelCase  )
        model.eval()

        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  )[0]

        model.enable_forward_chunking()
        with torch.no_grad():
          snake_case_					= model(**lowerCamelCase  )[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"""
import doctest

import projectq
from projectq.ops import H, Measure
def   A_    (     _lowercase      ):







						'''simple docstring'''


						snake_case_					:Union[str, Any]		=     quantum_engine.allocate_qubit()
						H | qubit
						Measure | qubit
						return int(_lowercase      )


if __name__ == "__main__":
						doctest.testmod()

						# initialises a new quantum backend
						__a           =							projectq.MainEngine()

						# Generate a list of 10 random numbers
						__a           =							[get_random_number(quantum_engine) for _ in range(10)]

						# Flushes the quantum engine from memory
						quantum_engine.flush()

						print("Random numbers", random_numbers_list)
