



import tempfile
import unittest

import numpy as np
import torch

from diffusers import DiffusionPipeline
from diffusers.models.attention_processor import Attention, AttnAddedKVProcessor




class 				lowerCAmelCase__	(						unittest.TestCase    ):







					def 			_lowercase      (							self   :  Union[str, Any]      ,		_A   :  bool = False):
											A__						:				List[str]         =			10

											if only_cross_attention:
																	A__						:				Dict         =			12
											else:
																	# when only cross attention is not set, the cross attention dim must be the same as the query dim
																	A__						:				str         =			query_dim

											return {
											    "query_dim": query_dim,
											    "cross_attention_dim": cross_attention_dim,
											    "heads": 2,
											    "dim_head": 4,
											    "added_kv_proj_dim": 6,
											    "norm_num_groups": 1,
											    "only_cross_attention": only_cross_attention,
											    "processor": AttnAddedKVProcessor(),
											}







					def 			_lowercase      (							self   :  Optional[Any]      ,		_A   :  Optional[Any]      ,		_A   :  Dict):
											A__						:				Tuple         =			2

											A__						:				List[Any]         =			torch.rand(UpperCamelCase_      ,		UpperCamelCase_      ,		3      ,		2)
											A__						:				Any         =			torch.rand(UpperCamelCase_      ,		4      ,		UpperCamelCase_)
											A__						:				Union[str, Any]         =			None

											return {
											    "hidden_states": hidden_states,
											    "encoder_hidden_states": encoder_hidden_states,
											    "attention_mask": attention_mask,
											}







					def 			_lowercase      (							self   :  Union[str, Any]):
											# self and cross attention

											torch.manual_seed(0)

											A__						:				Dict         =			self.get_constructor_arguments(only_cross_attention=UpperCamelCase_)
											A__						:				Optional[Any]         =			Attention(**UpperCamelCase_)

											self.assertTrue(attn.to_k is not None)
											self.assertTrue(attn.to_v is not None)

											A__						:				Tuple         =			self.get_forward_arguments(
											    query_dim=constructor_args["query_dim"]      ,		added_kv_proj_dim=constructor_args["added_kv_proj_dim"])

											A__						:				Any         =			attn(**UpperCamelCase_)

											# only self attention

											torch.manual_seed(0)

											A__						:				Optional[int]         =			self.get_constructor_arguments(only_cross_attention=UpperCamelCase_)
											A__						:				Tuple         =			Attention(**UpperCamelCase_)

											self.assertTrue(attn.to_k is None)
											self.assertTrue(attn.to_v is None)

											A__						:				Optional[Any]         =			self.get_forward_arguments(
											    query_dim=constructor_args["query_dim"]      ,		added_kv_proj_dim=constructor_args["added_kv_proj_dim"])

											A__						:				List[str]         =			attn(**UpperCamelCase_)

											self.assertTrue((only_cross_attn_out != self_and_cross_attn_out).all())




class 				lowerCAmelCase__	(						unittest.TestCase    ):







					def 			_lowercase      (							self   :  Any):
											A__						:				Dict         =			DiffusionPipeline.from_pretrained("hf-internal-testing/tiny-stable-diffusion-pipe"      ,		safety_checker=UpperCamelCase_)

											A__						:				Optional[int]         =			pipe(
											    "foo"      ,		num_inference_steps=2      ,		generator=torch.Generator("cpu").manual_seed(0)      ,		output_type="np"      ,		).images

											# the initial conversion succeeds
											A__						:				int         =			DiffusionPipeline.from_pretrained(
											    "hf-internal-testing/tiny-stable-diffusion-pipe"      ,		device_map="sequential"      ,		safety_checker=UpperCamelCase_)

											A__						:				List[Any]         =			pipe(
											    "foo"      ,		num_inference_steps=2      ,		generator=torch.Generator("cpu").manual_seed(0)      ,		output_type="np"      ,		).images

											with tempfile.TemporaryDirectory() as tmpdir:
																	# save the converted model
																	pipe.save_pretrained(UpperCamelCase_)

																	# can also load the converted weights
																	A__						:				Tuple         =			DiffusionPipeline.from_pretrained(UpperCamelCase_      ,		device_map="sequential"      ,		safety_checker=UpperCamelCase_)

											A__						:				Tuple         =			pipe(
											    "foo"      ,		num_inference_steps=2      ,		generator=torch.Generator("cpu").manual_seed(0)      ,		output_type="np"      ,		).images

											self.assertTrue(np.allclose(UpperCamelCase_      ,		UpperCamelCase_      ,		atol=1e-5))
											self.assertTrue(np.allclose(UpperCamelCase_      ,		UpperCamelCase_      ,		atol=1e-5))





from typing import Callable




def    lowerCamelCase__			(		A__			:   Callable      ):





       '''simple docstring'''



       def wrapper(A__			:   int      ):
              __lowerCamelCase  =		function_with_docstring.__doc__
              return target_function

       return wrapper

