



"""simple docstring"""
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    SCREAMING_SNAKE_CASE__						(      UpperCAmelCase__   ,    unittest.TestCase				):
   __lowerCAmelCase						:     int		=		PriorTransformer
   __lowerCAmelCase						:     Dict		=		'hidden_states'

   @property
   def 			SCREAMING_SNAKE_CASE  (     self		)						->   List[Any]:





     '''simple docstring'''




     UpperCAmelCase			:      Tuple			=			4
     UpperCAmelCase			:      Optional[Any]			=			8
     UpperCAmelCase			:      Union[str, Any]			=			7

     UpperCAmelCase			:      Any			=			floats_tensor((batch_size, embedding_dim)		).to(_SCREAMING_SNAKE_CASE		)

     UpperCAmelCase			:      List[Any]			=			floats_tensor((batch_size, embedding_dim)		).to(_SCREAMING_SNAKE_CASE		)
     UpperCAmelCase			:      List[Any]			=			floats_tensor((batch_size, num_embeddings, embedding_dim)		).to(_SCREAMING_SNAKE_CASE		)

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

   def 			SCREAMING_SNAKE_CASE  (     self   ,  _SCREAMING_SNAKE_CASE=0		)						->   Any:





     '''simple docstring'''




     torch.manual_seed(_SCREAMING_SNAKE_CASE		)
     UpperCAmelCase			:      Any			=			4
     UpperCAmelCase			:      Any			=			8
     UpperCAmelCase			:      List[Any]			=			7

     UpperCAmelCase			:      Any			=			torch.randn((batch_size, embedding_dim)		).to(_SCREAMING_SNAKE_CASE		)

     UpperCAmelCase			:      List[Any]			=			torch.randn((batch_size, embedding_dim)		).to(_SCREAMING_SNAKE_CASE		)
     UpperCAmelCase			:      str			=			torch.randn((batch_size, num_embeddings, embedding_dim)		).to(_SCREAMING_SNAKE_CASE		)

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

   @property
   def 			SCREAMING_SNAKE_CASE  (     self		)						->   List[Any]:





     '''simple docstring'''




     return (4, 8)

   @property
   def 			SCREAMING_SNAKE_CASE  (     self		)						->   Dict:





     '''simple docstring'''




     return (4, 8)

   def 			SCREAMING_SNAKE_CASE  (     self		)						->   int:





     '''simple docstring'''




     UpperCAmelCase			:      Optional[int]			=			{
         """num_attention_heads""": 2,
         """attention_head_dim""": 4,
         """num_layers""": 2,
         """embedding_dim""": 8,
         """num_embeddings""": 7,
         """additional_embeddings""": 4,
     }
     UpperCAmelCase			:      Tuple			=			self.dummy_input
     return init_dict, inputs_dict

   def 			SCREAMING_SNAKE_CASE  (     self		)						->   Dict:





     '''simple docstring'''




     UpperCAmelCase			:      Optional[Any]			=			PriorTransformer.from_pretrained(
         """hf-internal-testing/prior-dummy"""   ,  output_loading_info=_SCREAMING_SNAKE_CASE		)
     self.assertIsNotNone(_SCREAMING_SNAKE_CASE		)
     self.assertEqual(len(loading_info["""missing_keys"""]		)   ,  0		)

     model.to(_SCREAMING_SNAKE_CASE		)
     UpperCAmelCase			:      Dict			=			model(**self.dummy_input		)[0]

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

   def 			SCREAMING_SNAKE_CASE  (     self		)						->   Any:





     '''simple docstring'''




     UpperCAmelCase			:      Tuple			=			self.prepare_init_args_and_inputs_for_common()

     UpperCAmelCase			:      str			=			self.model_class(**_SCREAMING_SNAKE_CASE		)
     UpperCAmelCase			:      List[str]			=			inspect.signature(model.forward		)
     # signature.parameters is an OrderedDict => so arg_names order is deterministic
     UpperCAmelCase			:      Union[str, Any]			=			[*signature.parameters.keys()]

     UpperCAmelCase			:      int			=			["""hidden_states""", """timestep"""]
     self.assertListEqual(arg_names[:2]   ,  _SCREAMING_SNAKE_CASE		)




   def 			SCREAMING_SNAKE_CASE  (     self		)						->   Any:





     '''simple docstring'''




     UpperCAmelCase			:      Optional[Any]			=			PriorTransformer.from_pretrained("""hf-internal-testing/prior-dummy"""		)
     UpperCAmelCase			:      Dict			=			model.to(_SCREAMING_SNAKE_CASE		)

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

     UpperCAmelCase			:      str			=			self.get_dummy_seed_input()

     with torch.no_grad():
       UpperCAmelCase			:      str			=			model(**_SCREAMING_SNAKE_CASE		)[0]

     UpperCAmelCase			:      Dict			=			output[0, :5].flatten().cpu()
     print(_SCREAMING_SNAKE_CASE		)

     # 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.
     UpperCAmelCase			:      Dict			=			torch.tensor([-1.3436, -0.2870, 0.7538, 0.4368, -0.0239]		)
     self.assertTrue(torch_all_close(_SCREAMING_SNAKE_CASE   ,  _SCREAMING_SNAKE_CASE   ,  rtol=1E-2		)		)



@slow
class    SCREAMING_SNAKE_CASE__						(      unittest.TestCase				):

   def 			SCREAMING_SNAKE_CASE  (     self   ,  _SCREAMING_SNAKE_CASE=1   ,  _SCREAMING_SNAKE_CASE=768   ,  _SCREAMING_SNAKE_CASE=77   ,  _SCREAMING_SNAKE_CASE=0		)						->   int:





     '''simple docstring'''




     torch.manual_seed(_SCREAMING_SNAKE_CASE		)
     UpperCAmelCase			:      Optional[Any]			=			batch_size
     UpperCAmelCase			:      int			=			embedding_dim
     UpperCAmelCase			:      Tuple			=			num_embeddings

     UpperCAmelCase			:      Optional[Any]			=			torch.randn((batch_size, embedding_dim)		).to(_SCREAMING_SNAKE_CASE		)

     UpperCAmelCase			:      str			=			torch.randn((batch_size, embedding_dim)		).to(_SCREAMING_SNAKE_CASE		)
     UpperCAmelCase			:      Dict			=			torch.randn((batch_size, num_embeddings, embedding_dim)		).to(_SCREAMING_SNAKE_CASE		)

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

   def 			SCREAMING_SNAKE_CASE  (     self		)						->   int:





     '''simple docstring'''




     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()




   @parameterized.expand(
       [
           # fmt: off
           [13, [-0.5861,  0.1283, -0.0931,  0.0882,  0.4476,  0.1329, -0.0498,  0.0640]],
           [37, [-0.4913,  0.0110, -0.0483,  0.0541,  0.4954, -0.0170,  0.0354,  0.1651]],
           # fmt: on
       ]		)
   def 			SCREAMING_SNAKE_CASE  (     self   ,  _SCREAMING_SNAKE_CASE   ,  _SCREAMING_SNAKE_CASE		)						->   Any:





     '''simple docstring'''




     UpperCAmelCase			:      Dict			=			PriorTransformer.from_pretrained("""kandinsky-community/kandinsky-2-1-prior"""   ,  subfolder="""prior"""		)
     model.to(_SCREAMING_SNAKE_CASE		)
     UpperCAmelCase			:      Any			=			self.get_dummy_seed_input(seed=_SCREAMING_SNAKE_CASE		)

     with torch.no_grad():
       UpperCAmelCase			:      Any			=			model(**_SCREAMING_SNAKE_CASE		)[0]

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

     UpperCAmelCase			:      int			=			sample[0, :8].flatten().cpu()
     print(_SCREAMING_SNAKE_CASE		)
     UpperCAmelCase			:      Optional[int]			=			torch.tensor(_SCREAMING_SNAKE_CASE		)

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





"""simple docstring"""
import os

import pytest
import yaml

from datasets.features.features import Features, Value
from datasets.info import DatasetInfo, DatasetInfosDict

@pytest.mark.parametrize(
    """files"""    ,	[
        ["""full:README.md""", """dataset_infos.json"""],
        ["""empty:README.md""", """dataset_infos.json"""],
        ["""dataset_infos.json"""],
        ["""full:README.md"""],
    ]    ,	)
def    _snake_case		(		UpperCamelCase		:	str    ,	UpperCamelCase		:	Optional[Any]		):
 UpperCAmelCase			:      List[Any]			=			tmp_path_factory.mktemp("""dset_infos_dir"""		)
 if "full:README.md" in files:
  with open(dataset_infos_dir / """README.md"""    ,	"""w"""		) as f:
   f.write("""---\ndataset_info:\n  dataset_size: 42\n---"""		)
 if "empty:README.md" in files:
  with open(dataset_infos_dir / """README.md"""    ,	"""w"""		) as f:
   f.write(""""""		)
    # we want to support dataset_infos.json for backward compatibility
 if "dataset_infos.json" in files:
  with open(dataset_infos_dir / """dataset_infos.json"""    ,	"""w"""		) as f:
   f.write("""{\"default\": {\"dataset_size\": 42}}"""		)
 UpperCAmelCase			:      str			=			DatasetInfosDict.from_directory(UpperCamelCase		)
 assert dataset_infos
 assert dataset_infos["default"].dataset_size == 42

@pytest.mark.parametrize(
    """dataset_info"""    ,	[
        DatasetInfo(),
        DatasetInfo(
            description="""foo"""    ,	features=Features({"""a""": Value("""int32"""		)}		)    ,	builder_name="""builder"""    ,	config_name="""config"""    ,	version="""1.0.0"""    ,	splits=[{"""name""": """train"""}]    ,	download_size=42    ,	),
    ]    ,	)
def    _snake_case		(		UpperCamelCase		:	str    ,	UpperCamelCase		:	DatasetInfo		):
 UpperCAmelCase			:      List[Any]			=			str(UpperCamelCase		)
 dataset_info.write_to_directory(UpperCamelCase		)
 UpperCAmelCase			:      List[str]			=			DatasetInfo.from_directory(UpperCamelCase		)
 assert dataset_info == reloaded
 assert os.path.exists(os.path.join(UpperCamelCase    ,	"""dataset_info.json"""		)		)

def    _snake_case		(		):
 UpperCAmelCase			:      List[str]			=			DatasetInfo(
     description="""foo"""    ,	citation="""bar"""    ,	homepage="""https://foo.bar"""    ,	license="""CC0"""    ,	features=Features({"""a""": Value("""int32"""		)}		)    ,	post_processed={}    ,	supervised_keys=()    ,	task_templates=[]    ,	builder_name="""builder"""    ,	config_name="""config"""    ,	version="""1.0.0"""    ,	splits=[{"""name""": """train""", """num_examples""": 42}]    ,	download_checksums={}    ,	download_size=1337    ,	post_processing_size=442    ,	dataset_size=1234    ,	size_in_bytes=1337 + 442 + 1234    ,	)
 UpperCAmelCase			:      str			=			dataset_info._to_yaml_dict()
 assert sorted(UpperCamelCase		) == sorted(DatasetInfo._INCLUDED_INFO_IN_YAML		)
 for key in DatasetInfo._INCLUDED_INFO_IN_YAML:
  assert key in dataset_info_yaml_dict
  assert isinstance(dataset_info_yaml_dict[key]    ,	(list, dict, int, str)		)
 UpperCAmelCase			:      Union[str, Any]			=			yaml.safe_dump(UpperCamelCase		)
 UpperCAmelCase			:      List[str]			=			yaml.safe_load(UpperCamelCase		)
 assert dataset_info_yaml_dict == reloaded

def    _snake_case		(		):
 UpperCAmelCase			:      List[str]			=			DatasetInfo()
 UpperCAmelCase			:      Union[str, Any]			=			dataset_info._to_yaml_dict()
 assert dataset_info_yaml_dict == {}

@pytest.mark.parametrize(
    """dataset_infos_dict"""    ,	[
        DatasetInfosDict(),
        DatasetInfosDict({"""default""": DatasetInfo()}		),
        DatasetInfosDict({"""my_config_name""": DatasetInfo()}		),
        DatasetInfosDict(
            {
                """default""": DatasetInfo(
                    description="""foo"""    ,	features=Features({"""a""": Value("""int32"""		)}		)    ,	builder_name="""builder"""    ,	config_name="""config"""    ,	version="""1.0.0"""    ,	splits=[{"""name""": """train"""}]    ,	download_size=42    ,	)
            }		),
        DatasetInfosDict(
            {
                """v1""": DatasetInfo(dataset_size=42		),
                """v2""": DatasetInfo(dataset_size=1337		),
            }		),
    ]    ,	)
def    _snake_case		(		UpperCamelCase		:	Optional[int]    ,	UpperCamelCase		:	DatasetInfosDict		):
 UpperCAmelCase			:      List[str]			=			str(UpperCamelCase		)
 dataset_infos_dict.write_to_directory(UpperCamelCase		)
 UpperCAmelCase			:      List[Any]			=			DatasetInfosDict.from_directory(UpperCamelCase		)

 # the config_name of the dataset_infos_dict take over the attribute
 for config_name, dataset_info in dataset_infos_dict.items():
  UpperCAmelCase			:      Union[str, Any]			=			config_name
  # the yaml representation doesn't include fields like description or citation
  # so we just test that we can recover what we can from the yaml
  UpperCAmelCase			:      Optional[Any]			=			DatasetInfo._from_yaml_dict(dataset_info._to_yaml_dict()		)
 assert dataset_infos_dict == reloaded

 if dataset_infos_dict:
  assert os.path.exists(os.path.join(UpperCamelCase    ,	"""README.md"""		)		)
