code: 
  import glob
  import os
  from typing import Dict, List, Union
  
  import torch
  
  from diffusers.utils import is_safetensors_available
  
  
  if is_safetensors_available():
  				import safetensors.torch
  
  from huggingface_hub import snapshot_download
  
  from diffusers import DiffusionPipeline, __version__
  from diffusers.schedulers.scheduling_utils import SCHEDULER_CONFIG_NAME
  from diffusers.utils import CONFIG_NAME, DIFFUSERS_CACHE, ONNX_WEIGHTS_NAME, WEIGHTS_NAME
  
  
  
  
  
  class    _snake_case      (  A_       ):
  
  
  
  
  
  
  
  			def __init__(       self	):
  				self.register_to_config()
  				super().__init__()
  
  
  
  
  
  
  
  			def     SCREAMING_SNAKE_CASE					(       self		,     _a		,     _a	):
  				if dicta == dicta:
  					return True
  				else:
  					__magic_name__     ,					__magic_name__	:	Tuple			     =  self._remove_meta_keys(_a	)
  					__magic_name__     ,					__magic_name__	:	int			     =  self._remove_meta_keys(_a	)
  					if configa == configa:
  						print(f'''Warning !: Mismatch in keys {meta_keysa} and {meta_keysa}.'''	)
  						return True
  				return False
  
  
  
  
  
  
  
  			def     SCREAMING_SNAKE_CASE					(       self		,     _a	):
  				__magic_name__	:	Optional[int]			     =  []
  				__magic_name__	:	Tuple			     =  config_dict.copy()
  				for key in config_dict.keys():
  					if key.startswith("_"	):
  						temp_dict.pop(_a	)
  						meta_keys.append(_a	)
  				return (temp_dict, meta_keys)
  
  
  
  
  
  
  
  			@torch.no_grad()
  			def     SCREAMING_SNAKE_CASE					(       self		,     _a		,     **_a	):
  				__magic_name__	:	Union[str, Any]			     =  kwargs.pop("cache_dir"		,     _a	)
  				__magic_name__	:	List[Any]			     =  kwargs.pop("resume_download"		,     _a	)
  				__magic_name__	:	Optional[int]			     =  kwargs.pop("force_download"		,     _a	)
  				__magic_name__	:	Dict			     =  kwargs.pop("proxies"		,     _a	)
  				__magic_name__	:	List[str]			     =  kwargs.pop("local_files_only"		,     _a	)
  				__magic_name__	:	Optional[int]			     =  kwargs.pop("use_auth_token"		,     _a	)
  				__magic_name__	:	int			     =  kwargs.pop("revision"		,     _a	)
  				__magic_name__	:	str			     =  kwargs.pop("torch_dtype"		,     _a	)
  				__magic_name__	:	str			     =  kwargs.pop("device_map"		,     _a	)
  
  				__magic_name__	:	int			     =  kwargs.pop("alpha"		,     0.5	)
  				__magic_name__	:	int			     =  kwargs.pop("interp"		,     _a	)
  
  				print("Received list"		,     _a	)
  				print(f'''Combining with alpha={alpha}, interpolation mode={interp}'''	)
  
  				__magic_name__	:	Union[str, Any]			     =  len(_a	)
  				# Ignore result from model_index_json comparision of the two checkpoints
  				__magic_name__	:	Dict			     =  kwargs.pop("force"		,     _a	)
  
  				# If less than 2 checkpoints, nothing to merge. If more than 3, not supported for now.
  				if checkpoint_count > 3 or checkpoint_count < 2:
  					raise ValueError(
  					    "Received incorrect number of checkpoints to merge. Ensure that either 2 or 3 checkpoints are being"
  					    " passed."	)
  
  				print("Received the right number of checkpoints"	)
  				# chkpt0, chkpt1 = pretrained_model_name_or_path_list[0:2]
  				# chkpt2 = pretrained_model_name_or_path_list[2] if checkpoint_count == 3 else None
  
  				# Validate that the checkpoints can be merged
  				# Step 1: Load the model config and compare the checkpoints. We'll compare the model_index.json first while ignoring the keys starting with '_'
  				__magic_name__	:	List[str]			     =  []
  				for pretrained_model_name_or_path in pretrained_model_name_or_path_list:
  					__magic_name__	:	Tuple			     =  DiffusionPipeline.load_config(
  					    _a		,     cache_dir=_a		,     resume_download=_a		,     force_download=_a		,     proxies=_a		,     local_files_only=_a		,     use_auth_token=_a		,     revision=_a		,     )
  					config_dicts.append(_a	)
  
  				__magic_name__	:	Any			     =  True
  				for idx in range(1		,     len(_a	)	):
  					comparison_result &= self._compare_model_configs(config_dicts[idx - 1]		,     config_dicts[idx]	)
  					if not force and comparison_result is False:
  						raise ValueError("Incompatible checkpoints. Please check model_index.json for the models."	)
  						print(config_dicts[0]		,     config_dicts[1]	)
  				print("Compatible model_index.json files found"	)
  				# Step 2: Basic Validation has succeeded. Let's download the models and save them into our local files.
  				__magic_name__	:	Tuple			     =  []
  				for pretrained_model_name_or_path, config_dict in zip(_a		,     _a	):
  					__magic_name__	:	str			     =  [k for k in config_dict.keys() if not k.startswith("_"	)]
  					__magic_name__	:	Tuple			     =  [os.path.join(_a		,     "*"	) for k in folder_names]
  					allow_patterns += [
  					    WEIGHTS_NAME,
  					    SCHEDULER_CONFIG_NAME,
  					    CONFIG_NAME,
  					    ONNX_WEIGHTS_NAME,
  					    DiffusionPipeline.config_name,
  					]
  					__magic_name__	:	List[Any]			     =  config_dict.get("_class_name"	)
  					__magic_name__	:	Optional[int]			     =  {"diffusers": __version__, "pipeline_class": requested_pipeline_class}
  
  					__magic_name__	:	Any			     =  (
  					    pretrained_model_name_or_path
  					    if os.path.isdir(_a	)
  					    else snapshot_download(
  					        _a		,     cache_dir=_a		,     resume_download=_a		,     proxies=_a		,     local_files_only=_a		,     use_auth_token=_a		,     revision=_a		,     allow_patterns=_a		,     user_agent=_a		,     )
  					)
  					print("Cached Folder"		,     _a	)
  					cached_folders.append(_a	)
  
  				# Step 3:-
  				# Load the first checkpoint as a diffusion pipeline and modify its module state_dict in place
  				__magic_name__	:	Any			     =  DiffusionPipeline.from_pretrained(
  				    cached_folders[0]		,     torch_dtype=_a		,     device_map=_a	)
  				final_pipe.to(self.device	)
  
  				__magic_name__	:	Any			     =  None
  				if len(_a	) > 2:
  					__magic_name__	:	Tuple			     =  os.path.join(cached_folders[2]	)
  
  				if interp == "sigmoid":
  					__magic_name__	:	Dict			     =  CheckpointMergerPipeline.sigmoid
  				elif interp == "inv_sigmoid":
  					__magic_name__	:	Optional[int]			     =  CheckpointMergerPipeline.inv_sigmoid
  				elif interp == "add_diff":
  					__magic_name__	:	str			     =  CheckpointMergerPipeline.add_difference
  				else:
  					__magic_name__	:	Union[str, Any]			     =  CheckpointMergerPipeline.weighted_sum
  
  				# Find each module's state dict.
  				for attr in final_pipe.config.keys():
  					if not attr.startswith("_"	):
  						__magic_name__	:	Union[str, Any]			     =  os.path.join(cached_folders[1]		,     _a	)
  						if os.path.exists(_a	):
  							__magic_name__	:	Dict			     =  [
  							    *glob.glob(os.path.join(_a		,     "*.safetensors"	)	),
  							    *glob.glob(os.path.join(_a		,     "*.bin"	)	),
  							]
  							__magic_name__	:	Union[str, Any]			     =  files[0] if len(_a	) > 0 else None
  						if len(_a	) < 3:
  							__magic_name__	:	Tuple			     =  None
  						else:
  							__magic_name__	:	Optional[Any]			     =  os.path.join(cached_folders[2]		,     _a	)
  							if os.path.exists(_a	):
  								__magic_name__	:	Dict			     =  [
  								    *glob.glob(os.path.join(_a		,     "*.safetensors"	)	),
  								    *glob.glob(os.path.join(_a		,     "*.bin"	)	),
  								]
  								__magic_name__	:	List[str]			     =  files[0] if len(_a	) > 0 else None
                  # For an attr if both checkpoint_path_1 and 2 are None, ignore.
                  # If atleast one is present, deal with it according to interp method, of course only if the state_dict keys match.
  						if checkpoint_path_a is None and checkpoint_path_a is None:
  							print(f'''Skipping {attr}: not present in 2nd or 3d model'''	)
  							continue
  						try:
  							__magic_name__	:	str			     =  getattr(_a		,     _a	)
  							if isinstance(_a		,     _a	):  # ignore requires_safety_checker boolean
  								continue
  							__magic_name__	:	List[str]			     =  getattr(_a		,     "state_dict"	)
  							__magic_name__	:	Tuple			     =  theta_a()
  
  							__magic_name__	:	Union[str, Any]			     =  getattr(_a		,     "load_state_dict"	)
  							__magic_name__	:	Optional[Any]			     =  (
  							    safetensors.torch.load_file(_a	)
  							    if (is_safetensors_available() and checkpoint_path_a.endswith(".safetensors"	))
  							    else torch.load(_a		,     map_location="cpu"	)
  							)
  							__magic_name__	:	Optional[Any]			     =  None
  							if checkpoint_path_a:
  								__magic_name__	:	Dict			     =  (
  								    safetensors.torch.load_file(_a	)
  								    if (is_safetensors_available() and checkpoint_path_a.endswith(".safetensors"	))
  								    else torch.load(_a		,     map_location="cpu"	)
  								)
  
  							if not theta_a.keys() == theta_a.keys():
  								print(f'''Skipping {attr}: key mismatch'''	)
  								continue
  							if theta_a and not theta_a.keys() == theta_a.keys():
  								print(f'''Skipping {attr}:y mismatch'''	)
  						except Exception as e:
  							print(f'''Skipping {attr} do to an unexpected error: {str(_a	)}'''	)
  							continue
  						print(f'''MERGING {attr}'''	)
  
  						for key in theta_a.keys():
  							if theta_a:
  								__magic_name__	:	str			     =  theta_func(theta_a[key]		,     theta_a[key]		,     theta_a[key]		,     _a	)
  							else:
  								__magic_name__	:	int			     =  theta_func(theta_a[key]		,     theta_a[key]		,     _a		,     _a	)
  
  						del theta_a
  						del theta_a
  						update_theta_a(_a	)
  
  						del theta_a
  				return final_pipe
  
  
  
  
  
  
  
  			@staticmethod
  			def     SCREAMING_SNAKE_CASE					(       _a		,     _a		,     _a		,     _a	):
  				return ((1 - alpha) * thetaa) + (alpha * thetaa)
  
  
  
  
  
  
  
  			@staticmethod
  			def     SCREAMING_SNAKE_CASE					(       _a		,     _a		,     _a		,     _a	):
  				__magic_name__	:	Union[str, Any]			     =  alpha * alpha * (3 - (2 * alpha))
  				return thetaa + ((thetaa - thetaa) * alpha)
  
  
  
  
  
  
  
  			@staticmethod
  			def     SCREAMING_SNAKE_CASE					(       _a		,     _a		,     _a		,     _a	):
  				import math
  
  				__magic_name__	:	Union[str, Any]			     =  0.5 - math.sin(math.asin(1.0 - 2.0 * alpha	) / 3.0	)
  				return thetaa + ((thetaa - thetaa) * alpha)
  
  
  
  
  
  
  
  			@staticmethod
  			def     SCREAMING_SNAKE_CASE					(       _a		,     _a		,     _a		,     _a	):
  				return thetaa + (thetaa - thetaa) * (1.0 - alpha)
  
  
  
  
style_context: 
  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()
  
  snake_case				:	int							   =						logging.get_logger(__name__)
  
  
  
  
  
  
  def 						lowerCAmelCase_		(      _snake_case   :						List[Any]   ,     _snake_case   :						bool = True     )       ->	Tuple:
  
  
  
  
  
  
  
  	'''simple docstring'''
  
  
  	__magic_name__	:	List[Any]			     =  {}
  	for name in model.attn_processors.keys():
  		__magic_name__	:	List[Any]			     =  name.endswith("attn2.processor"     ) and not (
  		    name.startswith("transformer_in"     ) or "temp_attentions" in name.split("."     )
  		)
  		__magic_name__	:	List[str]			     =  model.config.cross_attention_dim if has_cross_attention else None
  		if name.startswith("mid_block"     ):
  			__magic_name__	:	Tuple			     =  model.config.block_out_channels[-1]
  		elif name.startswith("up_blocks"     ):
  			__magic_name__	:	Tuple			     =  int(name[len("up_blocks."     )]     )
  			__magic_name__	:	Optional[Any]			     =  list(reversed(model.config.block_out_channels     )     )[block_id]
  		elif name.startswith("down_blocks"     ):
  			__magic_name__	:	Optional[int]			     =  int(name[len("down_blocks."     )]     )
  			__magic_name__	:	str			     =  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
  			__magic_name__	:	List[Any]			     =  8 * model.config.attention_head_dim
  
  		__magic_name__	:	Union[str, Any]			     =  LoRAAttnProcessor(hidden_size=_snake_case   ,     cross_attention_dim=_snake_case     )
  		__magic_name__	:	str			     =  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      (  snake_case       ,		snake_case       ,		unittest.TestCase       ):
  			UpperCamelCase__					   =					UNetaDConditionModel
  			UpperCamelCase__					   =					'sample'
  
  
  
  
  
  
  
  			@property
  			def     SCREAMING_SNAKE_CASE					(       self	):
  				__magic_name__	:	Optional[Any]			     =  4
  				__magic_name__	:	str			     =  4
  				__magic_name__	:	List[str]			     =  4
  				__magic_name__	:	Optional[int]			     =  (32, 32)
  
  				__magic_name__	:	Optional[Any]			     =  floats_tensor((batch_size, num_channels, num_frames) + sizes	).to(_a	)
  				__magic_name__	:	Union[str, Any]			     =  torch.tensor([10]	).to(_a	)
  				__magic_name__	:	List[str]			     =  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	):
  				return (4, 4, 32, 32)
  
  
  
  
  
  
  
  			@property
  			def     SCREAMING_SNAKE_CASE					(       self	):
  				return (4, 4, 32, 32)
  
  
  
  
  
  
  
  			def     SCREAMING_SNAKE_CASE					(       self	):
  				__magic_name__	:	Optional[Any]			     =  {
  				    "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,
  				}
  				__magic_name__	:	List[str]			     =  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	):
  				__magic_name__     ,					__magic_name__	:	Any			     =  self.prepare_init_args_and_inputs_for_common()
  				__magic_name__	:	Union[str, Any]			     =  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	):
  				__magic_name__     ,					__magic_name__	:	Tuple			     =  self.prepare_init_args_and_inputs_for_common()
  
  				__magic_name__	:	Any			     =  32
  
  				__magic_name__	:	Dict			     =  self.model_class(**_a	)
  				model.to(_a	)
  				model.eval()
  
  				with torch.no_grad():
  					__magic_name__	:	Any			     =  model(**_a	)
  
  					if isinstance(_a		,     _a	):
  						__magic_name__	:	Any			     =  output.sample
  
  				self.assertIsNotNone(_a	)
  				__magic_name__	:	Union[str, Any]			     =  inputs_dict["sample"].shape
  				self.assertEqual(output.shape		,     _a		,     "Input and output shapes do not match"	)
  
  
  
  
  
  
  
  			def     SCREAMING_SNAKE_CASE					(       self	):
  				__magic_name__     ,					__magic_name__	:	Optional[int]			     =  self.prepare_init_args_and_inputs_for_common()
  				__magic_name__	:	int			     =  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	)
  
  					__magic_name__	:	str			     =  model(**_a	)
  					if isinstance(_a		,     _a	):
  						__magic_name__	:	str			     =  first.sample
  
  					__magic_name__	:	Tuple			     =  model(**_a	)
  					if isinstance(_a		,     _a	):
  						__magic_name__	:	str			     =  second.sample
  
  				__magic_name__	:	Optional[int]			     =  first.cpu().numpy()
  				__magic_name__	:	Any			     =  second.cpu().numpy()
  				__magic_name__	:	List[Any]			     =  out_a[~np.isnan(_a	)]
  				__magic_name__	:	Any			     =  out_a[~np.isnan(_a	)]
  				__magic_name__	:	Tuple			     =  np.amax(np.abs(out_a - out_a	)	)
  				self.assertLessEqual(_a		,     1e-5	)
  
  
  
  
  
  
  
  			def     SCREAMING_SNAKE_CASE					(       self	):
  				__magic_name__     ,					__magic_name__	:	Optional[int]			     =  self.prepare_init_args_and_inputs_for_common()
  
  				__magic_name__	:	Union[str, Any]			     =  8
  
  				__magic_name__	:	List[str]			     =  self.model_class(**_a	)
  				model.to(_a	)
  				model.eval()
  
  				model.set_attention_slice("auto"	)
  				with torch.no_grad():
  					__magic_name__	:	List[Any]			     =  model(**_a	)
  				assert output is not None
  
  				model.set_attention_slice("max"	)
  				with torch.no_grad():
  					__magic_name__	:	Tuple			     =  model(**_a	)
  				assert output is not None
  
  				model.set_attention_slice(2	)
  				with torch.no_grad():
  					__magic_name__	:	Union[str, Any]			     =  model(**_a	)
  				assert output is not None
  
  
  
  
  
  
  
  			def     SCREAMING_SNAKE_CASE					(       self	):
  				__magic_name__     ,					__magic_name__	:	str			     =  self.prepare_init_args_and_inputs_for_common()
  
  				__magic_name__	:	List[Any]			     =  8
  
  				__magic_name__	:	Dict			     =  self.model_class(**_a	)
  				model.to(_a	)
  
  				with torch.no_grad():
  					__magic_name__	:	Union[str, Any]			     =  model(**_a	).sample
  
  				__magic_name__	:	int			     =  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():
  					__magic_name__	:	int			     =  model(**_a		,     cross_attention_kwargs={"scale": 0.0}	).sample
  					__magic_name__	:	Optional[Any]			     =  model(**_a		,     cross_attention_kwargs={"scale": 0.5}	).sample
  					__magic_name__	:	Union[str, 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     SCREAMING_SNAKE_CASE					(       self	):
  				__magic_name__     ,					__magic_name__	:	Union[str, Any]			     =  self.prepare_init_args_and_inputs_for_common()
  
  				__magic_name__	:	Optional[int]			     =  8
  
  				torch.manual_seed(0	)
  				__magic_name__	:	Tuple			     =  self.model_class(**_a	)
  				model.to(_a	)
  
  				with torch.no_grad():
  					__magic_name__	:	Tuple			     =  model(**_a	).sample
  
  				__magic_name__	:	Union[str, Any]			     =  create_lora_layers(_a	)
  				model.set_attn_processor(_a	)
  
  				with torch.no_grad():
  					__magic_name__	:	Optional[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	)
  					__magic_name__	:	Union[str, Any]			     =  self.model_class(**_a	)
  					new_model.to(_a	)
  					new_model.load_attn_procs(_a	)
  
  				with torch.no_grad():
  					__magic_name__	:	int			     =  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	):
  				__magic_name__     ,					__magic_name__	:	Optional[Any]			     =  self.prepare_init_args_and_inputs_for_common()
  
  				__magic_name__	:	Optional[int]			     =  8
  
  				torch.manual_seed(0	)
  				__magic_name__	:	Optional[int]			     =  self.model_class(**_a	)
  				model.to(_a	)
  
  				with torch.no_grad():
  					__magic_name__	:	str			     =  model(**_a	).sample
  
  				__magic_name__	:	Union[str, Any]			     =  create_lora_layers(_a	)
  				model.set_attn_processor(_a	)
  
  				with torch.no_grad():
  					__magic_name__	:	Dict			     =  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	)
  					__magic_name__	:	List[str]			     =  self.model_class(**_a	)
  					new_model.to(_a	)
  					new_model.load_attn_procs(_a	)
  
  				with torch.no_grad():
  					__magic_name__	:	List[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     SCREAMING_SNAKE_CASE					(       self	):
  				# enable deterministic behavior for gradient checkpointing
  				__magic_name__     ,					__magic_name__	:	Union[str, Any]			     =  self.prepare_init_args_and_inputs_for_common()
  
  				__magic_name__	:	List[str]			     =  8
  
  				torch.manual_seed(0	)
  				__magic_name__	:	Tuple			     =  self.model_class(**_a	)
  				model.to(_a	)
  
  				__magic_name__	:	Any			     =  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	)
  					__magic_name__	:	List[Any]			     =  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	):
  				__magic_name__     ,					__magic_name__	:	str			     =  self.prepare_init_args_and_inputs_for_common()
  
  				__magic_name__	:	List[str]			     =  8
  
  				torch.manual_seed(0	)
  				__magic_name__	:	Union[str, Any]			     =  self.model_class(**_a	)
  				model.to(_a	)
  
  				__magic_name__	:	str			     =  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	)
  					__magic_name__	:	List[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     SCREAMING_SNAKE_CASE					(       self	):
  				__magic_name__     ,					__magic_name__	:	Optional[Any]			     =  self.prepare_init_args_and_inputs_for_common()
  
  				__magic_name__	:	Any			     =  8
  
  				torch.manual_seed(0	)
  				__magic_name__	:	List[str]			     =  self.model_class(**_a	)
  				model.to(_a	)
  
  				with torch.no_grad():
  					__magic_name__	:	Any			     =  model(**_a	).sample
  
  				__magic_name__	:	int			     =  create_lora_layers(_a	)
  				model.set_attn_processor(_a	)
  
  				with torch.no_grad():
  					__magic_name__	:	Dict			     =  model(**_a		,     cross_attention_kwargs={"scale": 0.0}	).sample
  
  				model.set_attn_processor(AttnProcessor()	)
  
  				with torch.no_grad():
  					__magic_name__	:	Optional[Any]			     =  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	):
  				# enable deterministic behavior for gradient checkpointing
  				__magic_name__     ,					__magic_name__	:	int			     =  self.prepare_init_args_and_inputs_for_common()
  
  				__magic_name__	:	Optional[int]			     =  4
  
  				torch.manual_seed(0	)
  				__magic_name__	:	str			     =  self.model_class(**_a	)
  				model.to(_a	)
  				__magic_name__	:	Any			     =  create_lora_layers(_a	)
  				model.set_attn_processor(_a	)
  
  				# default
  				with torch.no_grad():
  					__magic_name__	:	int			     =  model(**_a	).sample
  
  					model.enable_xformers_memory_efficient_attention()
  					__magic_name__	:	str			     =  model(**_a	).sample
  
  					model.disable_xformers_memory_efficient_attention()
  					__magic_name__	:	Optional[Any]			     =  model(**_a	).sample
  
  				assert (sample - on_sample).abs().max() < 1e-4
  				assert (sample - off_sample).abs().max() < 1e-4
  
  
  
  
  
  
  
  			def     SCREAMING_SNAKE_CASE					(       self	):
  				__magic_name__     ,					__magic_name__	:	int			     =  self.prepare_init_args_and_inputs_for_common()
  				__magic_name__	:	str			     =  32
  
  				__magic_name__	:	int			     =  self.model_class(**_a	)
  				model.to(_a	)
  				model.eval()
  
  				with torch.no_grad():
  					__magic_name__	:	Optional[Any]			     =  model(**_a	)[0]
  
  				model.enable_forward_chunking()
  				with torch.no_grad():
  					__magic_name__	:	int			     =  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.
  