


import tempfile
import unittest

import numpy as np
import torch

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


class 	a__						(    unittest.TestCase     ):

       def     UpperCAmelCase( self   :  List[str]   ,				lowerCamelCase_   :  bool = False     ):
            a_     :  List[Any]				      =     1_0

            if only_cross_attention:
                 a_     :  Any				      =     1_2
            else:
                 # when only cross attention is not set, the cross attention dim must be the same as the query dim
                 a_     :  Union[str, 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     UpperCAmelCase( self   :  Union[str, Any]   ,				lowerCamelCase_   :  str   ,				lowerCamelCase_   :  str     ):
            a_     :  str				      =     2

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

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

       def     UpperCAmelCase( self   :  List[Any]     ):
            # self and cross attention

            torch.manual_seed(0     )

            a_     :  List[str]				      =     self.get_constructor_arguments(only_cross_attention=lowerCamelCase_     )
            a_     :  List[Any]				      =     Attention(**lowerCamelCase_     )

            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_     :  Dict				      =     attn(**lowerCamelCase_     )

            # only self attention

            torch.manual_seed(0     )

            a_     :  Tuple				      =     self.get_constructor_arguments(only_cross_attention=lowerCamelCase_     )
            a_     :  int				      =     Attention(**lowerCamelCase_     )

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

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

            a_     :  Union[str, Any]				      =     attn(**lowerCamelCase_     )

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


class 	a__						(    unittest.TestCase     ):

       def     UpperCAmelCase( self   :  str     ):
            a_     :  List[str]				      =     DiffusionPipeline.from_pretrained("""hf-internal-testing/tiny-stable-diffusion-pipe"""   ,				safety_checker=lowerCamelCase_     )

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

            # the initial conversion succeeds
            a_     :  Optional[int]				      =     DiffusionPipeline.from_pretrained(
                """hf-internal-testing/tiny-stable-diffusion-pipe"""   ,				device_map="""sequential"""   ,				safety_checker=lowerCamelCase_     )

            a_     :  List[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(lowerCamelCase_     )

                 # can also load the converted weights
                 a_     :  List[str]				      =     DiffusionPipeline.from_pretrained(lowerCamelCase_   ,				device_map="""sequential"""   ,				safety_checker=lowerCamelCase_     )

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

            self.assertTrue(np.allclose(lowerCamelCase_   ,				lowerCamelCase_   ,				atol=1E-5     )     )
            self.assertTrue(np.allclose(lowerCamelCase_   ,				lowerCamelCase_   ,				atol=1E-5     )     )







class 	a__						:

       def __init__( self   :  int   ,				lowerCamelCase_   :  int     ):
            a_     :  Dict				      =     data
            a_     :  int
            a_     :  Node



def 			_a      (    __UpperCamelCase			):
     a_     :  int				      =     0
     a_     :  Optional[Any]				      =     x



def 			_a      (    __UpperCamelCase	,	__UpperCamelCase			):
     a_      ,      a_     :  Dict				      =     find_set(__UpperCamelCase			), find_set(__UpperCamelCase			)
     if x == y:
          return

     elif x.rank > y.rank:
          a_     :  Tuple				      =     x
     else:
          a_     :  Optional[Any]				      =     y
          if x.rank == y.rank:
               y.rank += 1



def 			_a      (    __UpperCamelCase			):
     if x != x.parent:
          a_     :  Tuple				      =     find_set(x.parent			)
     return x.parent



def 			_a      (    __UpperCamelCase			):
     a_     :  List[Any]				      =     ({0, 1, 2}, {3, 4, 5})
     for s in sets:
          if node.data in s:
               return s
     a_     :  Union[str, Any]				      =     F'''{node.data} is not in {sets}'''
     raise ValueError(__UpperCamelCase			)



def 			_a      (    ):
     a_     :  int				      =     [Node(__UpperCamelCase			) for i in range(6			)]
     for v in vertex:
          make_set(__UpperCamelCase			)

     union_set(vertex[0]	,	vertex[1]			)
     union_set(vertex[1]	,	vertex[2]			)
     union_set(vertex[3]	,	vertex[4]			)
     union_set(vertex[3]	,	vertex[5]			)

     for nodea in vertex:
          for nodea in vertex:
               if find_python_set(__UpperCamelCase			).isdisjoint(find_python_set(__UpperCamelCase			)			):
                    assert find_set(__UpperCamelCase			) != find_set(__UpperCamelCase			)
               else:
                    assert find_set(__UpperCamelCase			) == find_set(__UpperCamelCase			)


if __name__ == "__main__":
       test_disjoint_set()



