



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 	_lowerCamelCase      (			self    :	Any   ,							_a    :	bool = False      ):
													a__:							Union[str, Any]									=1_0

													if only_cross_attention:
																				a__:							List[str]									=1_2
													else:
																				# when only cross attention is not set, the cross attention dim must be the same as the query dim
																				a__:							Optional[int]									=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    :	Optional[Any]   ,							_a    :	List[Any]   ,							_a    :	Optional[int]      ):
													a__:							Optional[int]									=2

													a__:							Dict									=torch.rand(_UpperCamelCase   ,							_UpperCamelCase   ,							3   ,							2      )
													a__:							Optional[int]									=torch.rand(_UpperCamelCase   ,							4   ,							_UpperCamelCase      )
													a__:							Tuple									=None

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







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

													torch.manual_seed(0      )

													a__:							List[Any]									=self.get_constructor_arguments(only_cross_attention=_UpperCamelCase      )
													a__:							Dict									=Attention(**_UpperCamelCase      )

													self.assertTrue(attn.to_k is not None      )
													self.assertTrue(attn.to_v is not 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[Any]									=attn(**_UpperCamelCase      )

													# only self attention

													torch.manual_seed(0      )

													a__:							Union[str, Any]									=self.get_constructor_arguments(only_cross_attention=_UpperCamelCase      )
													a__:							List[Any]									=Attention(**_UpperCamelCase      )

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

													a__:							Optional[int]									=self.get_forward_arguments(
													    query_dim=constructor_args["query_dim"]   ,							added_kv_proj_dim=constructor_args["added_kv_proj_dim"]      )

													a__:							Optional[int]									=attn(**_UpperCamelCase      )

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


class 						lowerCamelCase__			(		unittest.TestCase ):


						def 	_lowerCamelCase      (			self    :	Union[str, Any]      ):
													a__:							Union[str, Any]									=DiffusionPipeline.from_pretrained("hf-internal-testing/tiny-stable-diffusion-pipe"   ,							safety_checker=_UpperCamelCase      )

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

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

													a__:							str									=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__:							int									=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 TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


snake_case__   :				Tuple						=	{
    'configuration_realm': ['REALM_PRETRAINED_CONFIG_ARCHIVE_MAP', 'RealmConfig'],
    'tokenization_realm': ['RealmTokenizer'],
}

try:
       if not is_tokenizers_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       snake_case__   :				Tuple						=	['RealmTokenizerFast']

try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       snake_case__   :				str						=	[
           'REALM_PRETRAINED_MODEL_ARCHIVE_LIST',
           'RealmEmbedder',
           'RealmForOpenQA',
           'RealmKnowledgeAugEncoder',
           'RealmPreTrainedModel',
           'RealmReader',
           'RealmScorer',
           'load_tf_weights_in_realm',
       ]
       snake_case__   :				Any						=	['RealmRetriever']


if TYPE_CHECKING:
       from .configuration_realm import REALM_PRETRAINED_CONFIG_ARCHIVE_MAP, RealmConfig
       from .tokenization_realm import RealmTokenizer

       try:
              if not is_tokenizers_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .tokenization_realm import RealmTokenizerFast

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_realm import (
                  REALM_PRETRAINED_MODEL_ARCHIVE_LIST,
                  RealmEmbedder,
                  RealmForOpenQA,
                  RealmKnowledgeAugEncoder,
                  RealmPreTrainedModel,
                  RealmReader,
                  RealmScorer,
                  load_tf_weights_in_realm,
              )
              from .retrieval_realm import RealmRetriever


else:
       import sys

       snake_case__   :				Any						=	_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)





