


from typing import List, Optional, Union

import numpy as np
import tensorflow as tf

from .utils import logging


A				:       Union[str, Any]      =  logging.get_logger(__name__)





def 						_lowerCAmelCase							(     _lowerCAmelCase			)    ->						List[int]:


						'''simple docstring'''






						if isinstance(_lowerCAmelCase   ,						np.ndarray			):
												return list(tensor.shape			)

						__snake_case				=		tf.shape(_lowerCAmelCase			)

						if tensor.shape == tf.TensorShape(_lowerCAmelCase			):
												return dynamic

						__snake_case				=		tensor.shape.as_list()

						return [dynamic[i] if s is None else s for i, s in enumerate(_lowerCAmelCase			)]





def 						_lowerCAmelCase							(     _lowerCAmelCase   ,						_lowerCAmelCase = None   ,						_lowerCAmelCase = None			)    ->						tf.Tensor:


						'''simple docstring'''






						return tf.nn.softmax(logits=logits + 1E-9   ,						axis=_lowerCAmelCase   ,						name=_lowerCAmelCase			)





def 						_lowerCAmelCase							(     _lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase=1E-5   ,						_lowerCAmelCase=-1			)    ->						str:


						'''simple docstring'''






						if weight.shape.rank != 1 or bias.shape.rank != 1 or not isinstance(_lowerCAmelCase   ,						_lowerCAmelCase			):
												raise NotImplementedError("Only 1D weight and bias tensors are supported for now, with only a single axis."			)

						# Get mean and variance on the axis to be normalized
						__snake_case		,       __snake_case				=		tf.nn.moments(_lowerCAmelCase   ,						axes=[axis]   ,						keepdims=_lowerCAmelCase			)

						if axis != -1:
												# Reshape scale and weight to have the same rank as inputs, but with 1 dimensions
												# on every dimension except axis
												__snake_case				=		[1] * inputs.shape.rank
												__snake_case				=		shape_list(_lowerCAmelCase			)[axis]
												__snake_case				=		tf.reshape(_lowerCAmelCase   ,						_lowerCAmelCase			)
												__snake_case				=		tf.reshape(_lowerCAmelCase   ,						_lowerCAmelCase			)

						# Compute layer normalization using the batch_normalization
						# function.
						__snake_case				=		tf.nn.batch_normalization(
						    _lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase   ,						offset=_lowerCAmelCase   ,						scale=_lowerCAmelCase   ,						variance_epsilon=_lowerCAmelCase   ,						)
						return outputs





def 						_lowerCAmelCase							(     _lowerCAmelCase   ,						_lowerCAmelCase=0   ,						_lowerCAmelCase=-1			)    ->						str:


						'''simple docstring'''






						if end_dim < 0:
												end_dim += input.shape.rank
						if start_dim < 0:
												start_dim += input.shape.rank

						if start_dim == end_dim:
												return input

						__snake_case				=		tf.shape(_lowerCAmelCase			)
						__snake_case				=		tf.math.reduce_prod(in_shape[start_dim : end_dim + 1]			)
						__snake_case				=		tf.concat([in_shape[:start_dim], [flattened_dim], in_shape[end_dim + 1 :]]   ,						axis=0			)
						return tf.reshape(_lowerCAmelCase   ,						_lowerCAmelCase			)





def 						_lowerCAmelCase							(     _lowerCAmelCase			)    ->						tf.Tensor:


						'''simple docstring'''






						if not isinstance(_lowerCAmelCase   ,						tf.Tensor			):
												__snake_case				=		tf.convert_to_tensor(_lowerCAmelCase			)  # Catches stray NumPy inputs
						if encoder_attention_mask.shape.rank == 3:
												__snake_case				=		encoder_attention_mask[:, None, :, :]
						if encoder_attention_mask.shape.rank == 2:
												__snake_case				=		encoder_attention_mask[:, None, None, :]
						# T5 has a mask that can compare sequence ids, we can simulate this here with this transposition
						# Cf. https://github.com/tensorflow/mesh/blob/8d2465e9bc93129b913b5ccc6a59aa97abd96ec6/mesh_tensorflow
						# /transformer/transformer_layers.py#L270
						# encoder_extended_attention_mask = (encoder_extended_attention_mask ==
						# encoder_extended_attention_mask.transpose(-1, -2))
						__snake_case				=		(
						    tf.cast(1   ,						encoder_attention_mask.dtype			) - encoder_extended_attention_mask
						) * encoder_extended_attention_mask.dtype.min

						return encoder_extended_attention_mask





def 						_lowerCAmelCase							(     _lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase = "input_ids"			)    ->						None:


						'''simple docstring'''






						tf.debugging.assert_less(
						    _lowerCAmelCase   ,						tf.cast(_lowerCAmelCase   ,						dtype=tensor.dtype			)   ,						message=(
						        F'''The maximum value of {tensor_name} ({tf.math.reduce_max(_lowerCAmelCase			)}) must be smaller than the embedding '''
						        F'''layer\'s input dimension ({embed_dim}). The likely cause is some problem at tokenization time.'''
						    )   ,						)





def 						_lowerCAmelCase							(     _lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase			)    ->						List[Any]:


						'''simple docstring'''






						__snake_case				=		6_4512
						# Check that no item in `data` is larger than `HDF5_OBJECT_HEADER_LIMIT`
						# because in that case even chunking the array would not make the saving
						# possible.
						__snake_case				=		[x for x in data if len(_lowerCAmelCase			) > HDF5_OBJECT_HEADER_LIMIT]

						# Expecting this to never be true.
						if bad_attributes:
												raise RuntimeError(
												    "The following attributes cannot be saved to HDF5 file because "
												    F'''they are larger than {HDF5_OBJECT_HEADER_LIMIT} '''
												    F'''bytes: {bad_attributes}'''			)

						__snake_case				=		np.asarray(_lowerCAmelCase			)

						__snake_case				=		1
						__snake_case				=		np.array_split(_lowerCAmelCase   ,						_lowerCAmelCase			)

						# This will never loop forever thanks to the test above.
						while any(x.nbytes > HDF5_OBJECT_HEADER_LIMIT for x in chunked_data			):
												num_chunks += 1
												__snake_case				=		np.array_split(_lowerCAmelCase   ,						_lowerCAmelCase			)

						if num_chunks > 1:
												for chunk_id, chunk_data in enumerate(_lowerCAmelCase			):
																		__snake_case				=		chunk_data
						else:
												__snake_case				=		data





def 						_lowerCAmelCase							(     _lowerCAmelCase   ,						_lowerCAmelCase			)    ->						Dict:


						'''simple docstring'''






						if name in group.attrs:
												__snake_case				=		[n.decode("utf8"			) if hasattr(_lowerCAmelCase   ,						"decode"			) else n for n in group.attrs[name]]
						else:
												__snake_case				=		[]
												__snake_case				=		0
												while "%s%d" % (name, chunk_id) in group.attrs:
																		data.extend(
																		    [n.decode("utf8"			) if hasattr(_lowerCAmelCase   ,						"decode"			) else n for n in group.attrs["%s%d" % (name, chunk_id)]]			)
																		chunk_id += 1
						return data





def 						_lowerCAmelCase							(     _lowerCAmelCase			)    ->						List[Any]:


						'''simple docstring'''






						def _expand_single_ad_tensor(_lowerCAmelCase			):
												if isinstance(_lowerCAmelCase   ,						tf.Tensor			) and t.shape.rank == 1:
																		return tf.expand_dims(_lowerCAmelCase   ,						axis=-1			)
												return t

						return tf.nest.map_structure(_expand_single_ad_tensor   ,						_lowerCAmelCase			)









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      UpperCamelCase(	_a  ,				unittest.TestCase ):
					snake_case_		:					Any    					=						PriorTransformer
					snake_case_		:					List[str]    					=						"""hidden_states"""


					@property
					def 				SCREAMING_SNAKE_CASE_	(    self						:		str    )      ->						str:




											'''simple docstring'''

											__snake_case				=		4
											__snake_case				=		8
											__snake_case				=		7

											__snake_case				=		floats_tensor((batch_size, embedding_dim)    ).to(SCREAMING_SNAKE_CASE    )

											__snake_case				=		floats_tensor((batch_size, embedding_dim)    ).to(SCREAMING_SNAKE_CASE    )
											__snake_case				=		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						:		str  ,     SCREAMING_SNAKE_CASE						:		Optional[Any]=0    )      ->						Any:




											'''simple docstring'''

											torch.manual_seed(SCREAMING_SNAKE_CASE    )
											__snake_case				=		4
											__snake_case				=		8
											__snake_case				=		7

											__snake_case				=		torch.randn((batch_size, embedding_dim)    ).to(SCREAMING_SNAKE_CASE    )

											__snake_case				=		torch.randn((batch_size, embedding_dim)    ).to(SCREAMING_SNAKE_CASE    )
											__snake_case				=		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						:		str    )      ->						str:




											'''simple docstring'''

											return (4, 8)


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




											'''simple docstring'''

											return (4, 8)


					def 				SCREAMING_SNAKE_CASE_	(    self						:		str    )      ->						Dict:




											'''simple docstring'''

											__snake_case				=		{
											    "num_attention_heads": 2,
											    "attention_head_dim": 4,
											    "num_layers": 2,
											    "embedding_dim": 8,
											    "num_embeddings": 7,
											    "additional_embeddings": 4,
											}
											__snake_case				=		self.dummy_input
											return init_dict, inputs_dict


					def 				SCREAMING_SNAKE_CASE_	(    self						:		Any    )      ->						Dict:




											'''simple docstring'''

											__snake_case		,       __snake_case				=		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    )
											__snake_case				=		model(**self.dummy_input    )[0]

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


					def 				SCREAMING_SNAKE_CASE_	(    self						:		Dict    )      ->						Optional[Any]:




											'''simple docstring'''

											__snake_case		,       __snake_case				=		self.prepare_init_args_and_inputs_for_common()

											__snake_case				=		self.model_class(**SCREAMING_SNAKE_CASE    )
											__snake_case				=		inspect.signature(model.forward    )
											# signature.parameters is an OrderedDict => so arg_names order is deterministic
											__snake_case				=		[*signature.parameters.keys()]

											__snake_case				=		["hidden_states", "timestep"]
											self.assertListEqual(arg_names[:2]  ,     SCREAMING_SNAKE_CASE    )






					def 				SCREAMING_SNAKE_CASE_	(    self						:		Optional[int]    )      ->						List[str]:




											'''simple docstring'''

											__snake_case				=		PriorTransformer.from_pretrained("hf-internal-testing/prior-dummy"    )
											__snake_case				=		model.to(SCREAMING_SNAKE_CASE    )

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

											__snake_case				=		self.get_dummy_seed_input()

											with torch.no_grad():
																	__snake_case				=		model(**SCREAMING_SNAKE_CASE    )[0]

											__snake_case				=		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.
											__snake_case				=		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      UpperCamelCase(	unittest.TestCase ):


					def 				SCREAMING_SNAKE_CASE_	(    self						:		List[Any]  ,     SCREAMING_SNAKE_CASE						:		int=1  ,     SCREAMING_SNAKE_CASE						:		str=7_6_8  ,     SCREAMING_SNAKE_CASE						:		Dict=7_7  ,     SCREAMING_SNAKE_CASE						:		int=0    )      ->						List[str]:




											'''simple docstring'''

											torch.manual_seed(SCREAMING_SNAKE_CASE    )
											__snake_case				=		batch_size
											__snake_case				=		embedding_dim
											__snake_case				=		num_embeddings

											__snake_case				=		torch.randn((batch_size, embedding_dim)    ).to(SCREAMING_SNAKE_CASE    )

											__snake_case				=		torch.randn((batch_size, embedding_dim)    ).to(SCREAMING_SNAKE_CASE    )
											__snake_case				=		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						:		Optional[int]    )      ->						Union[str, Any]:




											'''simple docstring'''

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






					@parameterized.expand(
					    [
					        # fmt: off
					        [1_3, [-0.5861,  0.1283, -0.0931,  0.0882,  0.4476,  0.1329, -0.0498,  0.0640]],
					        [3_7, [-0.4913,  0.0110, -0.0483,  0.0541,  0.4954, -0.0170,  0.0354,  0.1651]],
					        # fmt: on
					    ]    )
					def 				SCREAMING_SNAKE_CASE_	(    self						:		Union[str, Any]  ,     SCREAMING_SNAKE_CASE						:		int  ,     SCREAMING_SNAKE_CASE						:		Tuple    )      ->						List[str]:




											'''simple docstring'''

											__snake_case				=		PriorTransformer.from_pretrained("kandinsky-community/kandinsky-2-1-prior"  ,     subfolder="prior"    )
											model.to(SCREAMING_SNAKE_CASE    )
											__snake_case				=		self.get_dummy_seed_input(seed=SCREAMING_SNAKE_CASE    )

											with torch.no_grad():
																	__snake_case				=		model(**SCREAMING_SNAKE_CASE    )[0]

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

											__snake_case				=		sample[0, :8].flatten().cpu()
											print(SCREAMING_SNAKE_CASE    )
											__snake_case				=		torch.tensor(SCREAMING_SNAKE_CASE    )

											assert torch_all_close(SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE  ,     atol=1e-3    )





