


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

__SCREAMING_SNAKE_CASE				:     Dict		=     logging.get_logger(__name__)







def     _a					(   _SCREAMING_SNAKE_CASE					,			_SCREAMING_SNAKE_CASE = True					)     ->					Dict:
							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=_SCREAMING_SNAKE_CASE					,			cross_attention_dim=_SCREAMING_SNAKE_CASE					)
														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   __A  (snake_case__    ,		snake_case__    ,		unittest.TestCase):





							'''simple docstring'''

							__lowercase:		List[Any]				   =     UNetaDConditionModel
							__lowercase:		Union[str, Any]				   =     """sample"""



							@property
							def 		lowerCAmelCase							(					self				:   Dict							)       ->Optional[Any]:



														"""simple docstring"""



														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(UpperCAmelCase_							)
														snake_case_     							=		torch.tensor([10]							).to(UpperCAmelCase_							)
														snake_case_     							=		floats_tensor((batch_size, 4, 32)							).to(UpperCAmelCase_							)

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



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



														"""simple docstring"""



														return (4, 4, 32, 32)



							@property
							def 		lowerCAmelCase							(					self				:   str							)       ->Optional[int]:



														"""simple docstring"""



														return (4, 4, 32, 32)



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



														"""simple docstring"""



														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]							)       ->List[Any]:



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()
														snake_case_     							=		self.model_class(**UpperCAmelCase_							)

														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				:   List[Any]							)       ->Optional[Any]:



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()

														snake_case_     							=		32

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

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

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

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



							def 		lowerCAmelCase							(					self				:   List[str]							)       ->Tuple:



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()
														snake_case_     							=		self.model_class(**UpperCAmelCase_							)
														model.to(UpperCAmelCase_							)
														model.eval()

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

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

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

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



							def 		lowerCAmelCase							(					self				:   str							)       ->Any:



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()

														snake_case_     							=		8

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

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

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

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



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



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()

														snake_case_     							=		8

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

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

														snake_case_     							=		create_lora_layers(UpperCAmelCase_							)

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

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

														with torch.no_grad():
																					snake_case_     							=		model(**UpperCAmelCase_		,							cross_attention_kwargs={"""scale""": 0.0}							).sample
																					snake_case_     							=		model(**UpperCAmelCase_		,							cross_attention_kwargs={"""scale""": 0.5}							).sample
																					snake_case_     							=		model(**UpperCAmelCase_		,							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				:   Tuple							)       ->List[Any]:



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()

														snake_case_     							=		8

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

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

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

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

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

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



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()

														snake_case_     							=		8

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

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

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

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

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

														with torch.no_grad():
																					snake_case_     							=		new_model(**UpperCAmelCase_		,							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[int]							)       ->Optional[Any]:



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()

														snake_case_     							=		8

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

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



							def 		lowerCAmelCase							(					self				:   Dict							)       ->Dict:



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()

														snake_case_     							=		8

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

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



							def 		lowerCAmelCase							(					self				:   Optional[int]							)       ->List[Any]:



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()

														snake_case_     							=		8

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

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

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

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

														model.set_attn_processor(AttnProcessor()							)

														with torch.no_grad():
																					snake_case_     							=		model(**UpperCAmelCase_							).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				:   Tuple							)       ->Dict:



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()

														snake_case_     							=		4

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

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

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

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

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



							def 		lowerCAmelCase							(					self				:   List[Any]							)       ->Optional[Any]:



														"""simple docstring"""



														snake_case_							,    snake_case_     							=		self.prepare_init_args_and_inputs_for_common()
														snake_case_     							=		32

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

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

														model.enable_forward_chunking()
														with torch.no_grad():
																					snake_case_     							=		model(**UpperCAmelCase_							)[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 __future__ import annotations

from collections import deque
from enum import Enum
from math import atana, degrees
from sys import maxsize





class   __A  (snake_case__):





							'''simple docstring'''

							__lowercase:		Dict				   =     1
							__lowercase:		List[str]				   =     2
							__lowercase:		int				   =     3



							def __repr__(					self				:   Optional[Any]							)       ->Optional[int]:



														"""simple docstring"""



														return F"""{self.__class__.__name__}.{self.name}"""







def     _a					(   _SCREAMING_SNAKE_CASE					,			_SCREAMING_SNAKE_CASE					,			_SCREAMING_SNAKE_CASE					)     ->					float:
							snake_case_							,    snake_case_     							=		point
							return degrees(atana(y - miny					,			x - minx					)					)







def     _a					(   _SCREAMING_SNAKE_CASE					,			_SCREAMING_SNAKE_CASE					,			_SCREAMING_SNAKE_CASE					)     ->					Direction:
							snake_case_							,    snake_case_     							=		starting
							snake_case_							,    snake_case_     							=		via
							snake_case_							,    snake_case_     							=		target
							snake_case_     							=		degrees(atana(ya - ya					,			xa - xa					)					)
							via_angle %= 360
							snake_case_     							=		degrees(atana(ya - ya					,			xa - xa					)					)
							target_angle %= 360
							# t-
							#  \ \
							#   \ v
							#    \|
							#     s
							# via_angle is always lower than target_angle, if direction is left.
							# If they are same, it means they are on a same line of convex hull.
							if target_angle > via_angle:
														return Direction.left
							elif target_angle == via_angle:
														return Direction.straight
							else:
														return Direction.right







def     _a					(   _SCREAMING_SNAKE_CASE					)     ->					list[tuple[int, int]]:

							if len(_SCREAMING_SNAKE_CASE					) <= 2:
														# There is no convex hull
														raise ValueError("""graham_scan: argument must contain more than 3 points."""					)
							if len(_SCREAMING_SNAKE_CASE					) == 3:
														return points
							# find the lowest and the most left point
							snake_case_     							=		0
							snake_case_							,    snake_case_     							=		maxsize, maxsize
							for i, point in enumerate(_SCREAMING_SNAKE_CASE					):
														snake_case_     							=		point[0]
														snake_case_     							=		point[1]
														if y < miny:
																					snake_case_     							=		y
																					snake_case_     							=		x
																					snake_case_     							=		i
														if y == miny and x < minx:
																					snake_case_     							=		x
																					snake_case_     							=		i

    # remove the lowest and the most left point from points for preparing for sort
							points.pop(_SCREAMING_SNAKE_CASE					)

							snake_case_     							=		sorted(_SCREAMING_SNAKE_CASE					,			key=lambda _SCREAMING_SNAKE_CASE					: angle_comparer(_SCREAMING_SNAKE_CASE					,			_SCREAMING_SNAKE_CASE					,			_SCREAMING_SNAKE_CASE					)					)
							# This insert actually costs complexity,
							# and you should instead add (minx, miny) into stack later.
							# I'm using insert just for easy understanding.
							sorted_points.insert(0					,			(minx, miny)					)

							snake_case_     							=		deque()
							stack.append(sorted_points[0]					)
							stack.append(sorted_points[1]					)
							stack.append(sorted_points[2]					)
							# In any ways, the first 3 points line are towards left.
							# Because we sort them the angle from minx, miny.
							snake_case_     							=		Direction.left

							for i in range(3					,			len(_SCREAMING_SNAKE_CASE					)					):
														while True:
																					snake_case_     							=		stack[-2]
																					snake_case_     							=		stack[-1]
																					snake_case_     							=		sorted_points[i]
																					snake_case_     							=		check_direction(_SCREAMING_SNAKE_CASE					,			_SCREAMING_SNAKE_CASE					,			_SCREAMING_SNAKE_CASE					)

																					if next_direction == Direction.left:
																												snake_case_     							=		Direction.left
																												break
																					if next_direction == Direction.straight:
																												if current_direction == Direction.left:
																																			# We keep current_direction as left.
																																			# Because if the straight line keeps as straight,
																																			# we want to know if this straight line is towards left.
																																			break
																												elif current_direction == Direction.right:
																																			# If the straight line is towards right,
																																			# every previous points on those straigh line is not convex hull.
																																			stack.pop()
																					if next_direction == Direction.right:
																												stack.pop()
														stack.append(sorted_points[i]					)
							return list(_SCREAMING_SNAKE_CASE					)



