



import torch
import torch.nn as nn
from transformers.modeling_utils import ModuleUtilsMixin
from transformers.models.ta.modeling_ta import (
    TaBlock,
    TaConfig,
    TaLayerNorm,
)

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin





class 		__SCREAMING_SNAKE_CASE      (				lowercase__					,			lowercase__					,			lowercase__      ):






				"""simple docstring"""

				@register_to_config
				def __init__(				self						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__ = False						,	):






					"""simple docstring"""

					super().__init__()

					_snake_case					:							List[Any]					     =   nn.Linear(SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	bias=SCREAMING_SNAKE_CASE__      )

					_snake_case					:							List[Any]					     =   nn.Embedding(SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__      )
					_snake_case					:							Dict					     =   False

					_snake_case					:							Tuple					     =   nn.Dropout(p=SCREAMING_SNAKE_CASE__      )

					_snake_case					:							Dict					     =   TaConfig(
					    d_model=SCREAMING_SNAKE_CASE__						,	num_heads=SCREAMING_SNAKE_CASE__						,	d_kv=SCREAMING_SNAKE_CASE__						,	d_ff=SCREAMING_SNAKE_CASE__						,	feed_forward_proj=SCREAMING_SNAKE_CASE__						,	dropout_rate=SCREAMING_SNAKE_CASE__						,	is_decoder=SCREAMING_SNAKE_CASE__						,	is_encoder_decoder=SCREAMING_SNAKE_CASE__						,	)
					_snake_case					:							Union[str, Any]					     =   nn.ModuleList()
					for lyr_num in range(SCREAMING_SNAKE_CASE__      ):
						_snake_case					:							Optional[int]					     =   TaBlock(SCREAMING_SNAKE_CASE__      )
						self.encoders.append(SCREAMING_SNAKE_CASE__      )

					_snake_case					:							Optional[Any]					     =   TaLayerNorm(SCREAMING_SNAKE_CASE__      )
					_snake_case					:							Union[str, Any]					     =   nn.Dropout(p=SCREAMING_SNAKE_CASE__      )







				def 		__lowerCamelCase(				self						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__      ):






					"""simple docstring"""

					_snake_case					:							Dict					     =   self.input_proj(SCREAMING_SNAKE_CASE__      )

					# terminal relative positional encodings
					_snake_case					:							List[Any]					     =   encoder_inputs.shape[1]
					_snake_case					:							Optional[Any]					     =   torch.arange(SCREAMING_SNAKE_CASE__						,	device=encoder_inputs.device      )

					_snake_case					:							Union[str, Any]					     =   encoder_inputs_mask.sum(-1      )
					_snake_case					:							Tuple					     =   torch.roll(input_positions.unsqueeze(0      )						,	tuple(seq_lens.tolist()      )						,	dims=0      )
					x += self.position_encoding(SCREAMING_SNAKE_CASE__      )

					_snake_case					:							Any					     =   self.dropout_pre(SCREAMING_SNAKE_CASE__      )

					# inverted the attention mask
					_snake_case					:							Dict					     =   encoder_inputs.size()
					_snake_case					:							int					     =   self.get_extended_attention_mask(SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__      )

					for lyr in self.encoders:
						_snake_case					:							List[str]					     =   lyr(SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__      )[0]
					_snake_case					:							Dict					     =   self.layer_norm(SCREAMING_SNAKE_CASE__      )

					return self.dropout_post(SCREAMING_SNAKE_CASE__      ), encoder_inputs_mask










import tempfile
import unittest

import numpy as np
import torch

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





class 		__SCREAMING_SNAKE_CASE      (				unittest.TestCase      ):






				"""simple docstring"""

				def 		__lowerCamelCase(				self						,	SCREAMING_SNAKE_CASE__ = False      ):






					"""simple docstring"""

					_snake_case					:							Any					     =   10

					if only_cross_attention:
						_snake_case					:							Tuple					     =   12
					else:
						# when only cross attention is not set, the cross attention dim must be the same as the query dim
						_snake_case					:							Any					     =   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 		__lowerCamelCase(				self						,	SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__      ):






					"""simple docstring"""

					_snake_case					:							Any					     =   2

					_snake_case					:							List[str]					     =   torch.rand(SCREAMING_SNAKE_CASE__						,	SCREAMING_SNAKE_CASE__						,	3						,	2      )
					_snake_case					:							List[str]					     =   torch.rand(SCREAMING_SNAKE_CASE__						,	4						,	SCREAMING_SNAKE_CASE__      )
					_snake_case					:							Optional[int]					     =   None

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







				def 		__lowerCamelCase(				self      ):






					"""simple docstring"""

					torch.manual_seed(0      )

					_snake_case					:							Optional[int]					     =   self.get_constructor_arguments(only_cross_attention=SCREAMING_SNAKE_CASE__      )
					_snake_case					:							List[str]					     =   Attention(**SCREAMING_SNAKE_CASE__      )

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

					_snake_case					:							Dict					     =   self.get_forward_arguments(
					    query_dim=constructor_args["""query_dim"""]						,	added_kv_proj_dim=constructor_args["""added_kv_proj_dim"""]      )

					_snake_case					:							Any					     =   attn(**SCREAMING_SNAKE_CASE__      )

					# only self attention

					torch.manual_seed(0      )

					_snake_case					:							Any					     =   self.get_constructor_arguments(only_cross_attention=SCREAMING_SNAKE_CASE__      )
					_snake_case					:							List[str]					     =   Attention(**SCREAMING_SNAKE_CASE__      )

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

					_snake_case					:							str					     =   self.get_forward_arguments(
					    query_dim=constructor_args["""query_dim"""]						,	added_kv_proj_dim=constructor_args["""added_kv_proj_dim"""]      )

					_snake_case					:							Any					     =   attn(**SCREAMING_SNAKE_CASE__      )

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





class 		__SCREAMING_SNAKE_CASE      (				unittest.TestCase      ):






				"""simple docstring"""

				def 		__lowerCamelCase(				self      ):






					"""simple docstring"""

					_snake_case					:							str					     =   DiffusionPipeline.from_pretrained("""hf-internal-testing/tiny-stable-diffusion-pipe"""						,	safety_checker=SCREAMING_SNAKE_CASE__      )

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

					# the initial conversion succeeds
					_snake_case					:							List[str]					     =   DiffusionPipeline.from_pretrained(
					    """hf-internal-testing/tiny-stable-diffusion-pipe"""						,	device_map="""sequential"""						,	safety_checker=SCREAMING_SNAKE_CASE__      )

					_snake_case					:							Union[str, 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(SCREAMING_SNAKE_CASE__      )

						# can also load the converted weights
						_snake_case					:							str					     =   DiffusionPipeline.from_pretrained(SCREAMING_SNAKE_CASE__						,	device_map="""sequential"""						,	safety_checker=SCREAMING_SNAKE_CASE__      )

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

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





