import gc
import inspect
import unittest

import torch
from parameterized import parameterized

from diffusers import PriorTransformer
from diffusers.utils import floats_tensor, slow, torch_all_close, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin


enable_full_determinism()



class   snake_case_       (		__A  ,unittest.TestCase       ):
   __A	:		Any         =				PriorTransformer
   __A	:		Union[str, Any]         =				"hidden_states"






   @property
   def       __UpperCamelCase     (      self				:     Dict  )					->						str:
       lowercase__ :   Optional[Any]    					=				4
       lowercase__ :   Any    					=				8
       lowercase__ :   str    					=				7

       lowercase__ :   Dict    					=				floats_tensor((batch_size, embedding_dim)  ).to(lowercase_  )

       lowercase__ :   List[str]    					=				floats_tensor((batch_size, embedding_dim)  ).to(lowercase_  )
       lowercase__ :   Dict    					=				floats_tensor((batch_size, num_embeddings, embedding_dim)  ).to(lowercase_  )

       return {
           "hidden_states": hidden_states,
           "timestep": 2,
           "proj_embedding": proj_embedding,
           "encoder_hidden_states": encoder_hidden_states,
       }






   def       __UpperCamelCase     (      self				:     Dict	,						lowercase_				:     Optional[int]=0  )					->						int:
       torch.manual_seed(lowercase_  )
       lowercase__ :   Tuple    					=				4
       lowercase__ :   Optional[int]    					=				8
       lowercase__ :   int    					=				7

       lowercase__ :   Union[str, Any]    					=				torch.randn((batch_size, embedding_dim)  ).to(lowercase_  )

       lowercase__ :   Any    					=				torch.randn((batch_size, embedding_dim)  ).to(lowercase_  )
       lowercase__ :   List[Any]    					=				torch.randn((batch_size, num_embeddings, embedding_dim)  ).to(lowercase_  )

       return {
           "hidden_states": hidden_states,
           "timestep": 2,
           "proj_embedding": proj_embedding,
           "encoder_hidden_states": encoder_hidden_states,
       }






   @property
   def       __UpperCamelCase     (      self				:     List[str]  )					->						Optional[int]:
       return (4, 8)






   @property
   def       __UpperCamelCase     (      self				:     int  )					->						int:
       return (4, 8)






   def       __UpperCamelCase     (      self				:     List[Any]  )					->						Optional[Any]:
       lowercase__ :   Union[str, Any]    					=				{
           "num_attention_heads": 2,
           "attention_head_dim": 4,
           "num_layers": 2,
           "embedding_dim": 8,
           "num_embeddings": 7,
           "additional_embeddings": 4,
       }
       lowercase__ :   Tuple    					=				self.dummy_input
       return init_dict, inputs_dict






   def       __UpperCamelCase     (      self				:     Optional[int]  )					->						List[str]:
       lowercase__					,   lowercase__ :   Any    					=				PriorTransformer.from_pretrained(
           "hf-internal-testing/prior-dummy"	,						output_loading_info=lowercase_  )
       self.assertIsNotNone(lowercase_  )
       self.assertEqual(len(loading_info["missing_keys"]  )	,						0  )

       model.to(lowercase_  )
       lowercase__ :   Dict    					=				model(**self.dummy_input  )[0]

       assert hidden_states is not None, "Make sure output is not None"






   def       __UpperCamelCase     (      self				:     Optional[Any]  )					->						Union[str, Any]:
       lowercase__					,   lowercase__ :   str    					=				self.prepare_init_args_and_inputs_for_common()

       lowercase__ :   Any    					=				self.model_class(**lowercase_  )
       lowercase__ :   Tuple    					=				inspect.signature(model.forward  )
       # signature.parameters is an OrderedDict => so arg_names order is deterministic
       lowercase__ :   Optional[int]    					=				[*signature.parameters.keys()]

       lowercase__ :   Optional[Any]    					=				["hidden_states", "timestep"]
       self.assertListEqual(arg_names[:2]	,						lowercase_  )







   def       __UpperCamelCase     (      self				:     Optional[int]  )					->						Optional[int]:
       lowercase__ :   str    					=				PriorTransformer.from_pretrained("hf-internal-testing/prior-dummy"  )
       lowercase__ :   Tuple    					=				model.to(lowercase_  )

       if hasattr(lowercase_	,						"set_default_attn_processor"  ):
           model.set_default_attn_processor()

       lowercase__ :   Tuple    					=				self.get_dummy_seed_input()

       with torch.no_grad():
           lowercase__ :   Union[str, Any]    					=				model(**lowercase_  )[0]

       lowercase__ :   int    					=				output[0, :5].flatten().cpu()
       print(lowercase_  )

       # Since the VAE Gaussian prior's generator is seeded on the appropriate device,
       # the expected output slices are not the same for CPU and GPU.
       lowercase__ :   Dict    					=				torch.tensor([-1.34_36, -0.28_70, 0.75_38, 0.43_68, -0.02_39]  )
       self.assertTrue(torch_all_close(lowercase_	,						lowercase_	,						rtol=1E-2  )  )



@slow
class   snake_case_       (		unittest.TestCase       ):






   def       __UpperCamelCase     (      self				:     Optional[Any]	,						lowercase_				:     Any=1	,						lowercase_				:     Any=7_68	,						lowercase_				:     Any=77	,						lowercase_				:     Tuple=0  )					->						int:
       torch.manual_seed(lowercase_  )
       lowercase__ :   int    					=				batch_size
       lowercase__ :   Any    					=				embedding_dim
       lowercase__ :   Dict    					=				num_embeddings

       lowercase__ :   str    					=				torch.randn((batch_size, embedding_dim)  ).to(lowercase_  )

       lowercase__ :   Optional[int]    					=				torch.randn((batch_size, embedding_dim)  ).to(lowercase_  )
       lowercase__ :   str    					=				torch.randn((batch_size, num_embeddings, embedding_dim)  ).to(lowercase_  )

       return {
           "hidden_states": hidden_states,
           "timestep": 2,
           "proj_embedding": proj_embedding,
           "encoder_hidden_states": encoder_hidden_states,
       }






   def       __UpperCamelCase     (      self				:     Any  )					->						Tuple:
       # clean up the VRAM after each test
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()







   @parameterized.expand(
       [
           # fmt: off
           [13, [-0.58_61,  0.12_83, -0.09_31,  0.08_82,  0.44_76,  0.13_29, -0.04_98,  0.06_40]],
           [37, [-0.49_13,  0.01_10, -0.04_83,  0.05_41,  0.49_54, -0.01_70,  0.03_54,  0.16_51]],
           # fmt: on
       ]  )
   def       __UpperCamelCase     (      self				:     Dict	,						lowercase_				:     Dict	,						lowercase_				:     Optional[int]  )					->						List[Any]:
       lowercase__ :   Tuple    					=				PriorTransformer.from_pretrained("kandinsky-community/kandinsky-2-1-prior"	,						subfolder="prior"  )
       model.to(lowercase_  )
       lowercase__ :   Any    					=				self.get_dummy_seed_input(seed=lowercase_  )

       with torch.no_grad():
           lowercase__ :   int    					=				model(**lowercase_  )[0]

       assert list(sample.shape  ) == [1, 7_68]

       lowercase__ :   List[Any]    					=				sample[0, :8].flatten().cpu()
       print(lowercase_  )
       lowercase__ :   List[Any]    					=				torch.tensor(lowercase_  )

       assert torch_all_close(lowercase_	,						lowercase_	,						atol=1E-3  )







"""simple docstring"""





import gc
import unittest

from diffusers import FlaxDPMSolverMultistepScheduler, FlaxStableDiffusionPipeline
from diffusers.utils import is_flax_available, slow
from diffusers.utils.testing_utils import require_flax


if is_flax_available():
 import jax
 import jax.numpy as jnp
 from flax.jax_utils import replicate
 from flax.training.common_utils import shard



@slow
@require_flax
class 						snake_case_(				unittest.TestCase	):






     def    lowerCamelCase__   (		self     :    int ):
           # clean up the VRAM after each test
           super().tearDown()
           gc.collect()






     def    lowerCamelCase__   (		self     :    Optional[Any] ):
           lowerCAmelCase,					lowerCAmelCase    :		Optional[int]				=      FlaxStableDiffusionPipeline.from_pretrained(
               '''stabilityai/stable-diffusion-2'''    , revision='''bf16'''    , dtype=jnp.bfloataa    , )

           lowerCAmelCase    :		Optional[int]				=      '''A painting of a squirrel eating a burger'''
           lowerCAmelCase    :		List[str]				=      jax.device_count()
           lowerCAmelCase    :		Optional[int]				=      num_samples * [prompt]
           lowerCAmelCase    :		Any				=      sd_pipe.prepare_inputs(UpperCamelCase_ )

           lowerCAmelCase    :		Optional[int]				=      replicate(UpperCamelCase_ )
           lowerCAmelCase    :		Union[str, Any]				=      shard(UpperCamelCase_ )

           lowerCAmelCase    :		Optional[int]				=      jax.random.PRNGKey(0 )
           lowerCAmelCase    :		Optional[Any]				=      jax.random.split(UpperCamelCase_    , jax.device_count() )

           lowerCAmelCase    :		str				=      sd_pipe(UpperCamelCase_    , UpperCamelCase_    , UpperCamelCase_    , num_inference_steps=2_5    , jit=UpperCamelCase_ )[0]
           assert images.shape == (jax.device_count(), 1, 7_6_8, 7_6_8, 3)

           lowerCAmelCase    :		str				=      images.reshape((images.shape[0] * images.shape[1],) + images.shape[-3:] )
           lowerCAmelCase    :		List[str]				=      images[0, 2_5_3:2_5_6, 2_5_3:2_5_6, -1]

           lowerCAmelCase    :		Dict				=      jnp.asarray(jax.device_get(image_slice.flatten() ) )
           lowerCAmelCase    :		List[str]				=      jnp.array([0.4_238, 0.4_414, 0.4_395, 0.4_453, 0.4_629, 0.4_590, 0.4_531, 0.45_508, 0.4_512] )
           print(F'''output_slice: {output_slice}''' )
           assert jnp.abs(output_slice - expected_slice ).max() < 1E-2






     def    lowerCamelCase__   (		self     :    Union[str, Any] ):
           lowerCAmelCase    :		Union[str, Any]				=      '''stabilityai/stable-diffusion-2'''
           lowerCAmelCase,					lowerCAmelCase    :		Dict				=      FlaxDPMSolverMultistepScheduler.from_pretrained(UpperCamelCase_    , subfolder='''scheduler''' )
           lowerCAmelCase,					lowerCAmelCase    :		int				=      FlaxStableDiffusionPipeline.from_pretrained(
               UpperCamelCase_    , scheduler=UpperCamelCase_    , revision='''bf16'''    , dtype=jnp.bfloataa    , )
           lowerCAmelCase    :		List[Any]				=      scheduler_params

           lowerCAmelCase    :		List[Any]				=      '''A painting of a squirrel eating a burger'''
           lowerCAmelCase    :		Any				=      jax.device_count()
           lowerCAmelCase    :		int				=      num_samples * [prompt]
           lowerCAmelCase    :		int				=      sd_pipe.prepare_inputs(UpperCamelCase_ )

           lowerCAmelCase    :		Dict				=      replicate(UpperCamelCase_ )
           lowerCAmelCase    :		Tuple				=      shard(UpperCamelCase_ )

           lowerCAmelCase    :		int				=      jax.random.PRNGKey(0 )
           lowerCAmelCase    :		Optional[int]				=      jax.random.split(UpperCamelCase_    , jax.device_count() )

           lowerCAmelCase    :		Tuple				=      sd_pipe(UpperCamelCase_    , UpperCamelCase_    , UpperCamelCase_    , num_inference_steps=2_5    , jit=UpperCamelCase_ )[0]
           assert images.shape == (jax.device_count(), 1, 7_6_8, 7_6_8, 3)

           lowerCAmelCase    :		Any				=      images.reshape((images.shape[0] * images.shape[1],) + images.shape[-3:] )
           lowerCAmelCase    :		str				=      images[0, 2_5_3:2_5_6, 2_5_3:2_5_6, -1]

           lowerCAmelCase    :		Optional[int]				=      jnp.asarray(jax.device_get(image_slice.flatten() ) )
           lowerCAmelCase    :		Tuple				=      jnp.array([0.4_336, 0.42_969, 0.4_453, 0.4_199, 0.4_297, 0.4_531, 0.4_434, 0.4_434, 0.4_297] )
           print(F'''output_slice: {output_slice}''' )
           assert jnp.abs(output_slice - expected_slice ).max() < 1E-2


