
'''simple docstring'''







import tensorflow as tf

from ...tf_utils import shape_list


class 						__UpperCAmelCase      (					tf.keras.layers.Layer						):


							def __init__(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_=1    ,   lowerCAmelCase_=False    ,   **lowerCAmelCase_  ):





								"""simple docstring"""






								super().__init__(**lowerCAmelCase_  )

								_snake_case   						=		vocab_size
								_snake_case   						=		d_embed
								_snake_case   						=		d_proj

								_snake_case   						=		cutoffs + [vocab_size]
								_snake_case   						=		[0] + self.cutoffs
								_snake_case   						=		div_val

								_snake_case   						=		self.cutoffs[0]
								_snake_case   						=		len(self.cutoffs  ) - 1
								_snake_case   						=		self.shortlist_size + self.n_clusters
								_snake_case   						=		keep_order

								_snake_case   						=		[]
								_snake_case   						=		[]


							def  lowerCamelCase				(       self    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								if self.n_clusters > 0:
									_snake_case   						=		self.add_weight(
									    shape=(self.n_clusters, self.d_embed)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name='cluster_weight'  )
									_snake_case   						=		self.add_weight(
									    shape=(self.n_clusters,)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name='cluster_bias'  )

								if self.div_val == 1:
									for i in range(len(self.cutoffs  )  ):
										if self.d_proj != self.d_embed:
											_snake_case   						=		self.add_weight(
											    shape=(self.d_embed, self.d_proj)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_projs_._{i}'    ,   )
											self.out_projs.append(lowerCAmelCase_  )
										else:
											self.out_projs.append(lowerCAmelCase_  )
										_snake_case   						=		self.add_weight(
										    shape=(self.vocab_size, self.d_embed)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_layers_._{i}_._weight'    ,   )
										_snake_case   						=		self.add_weight(
										    shape=(self.vocab_size,)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_layers_._{i}_._bias'    ,   )
										self.out_layers.append((weight, bias)  )
								else:
									for i in range(len(self.cutoffs  )  ):
										_snake_case						, _snake_case   						=		self.cutoff_ends[i], self.cutoff_ends[i + 1]
										_snake_case   						=		self.d_embed // (self.div_val**i)

										_snake_case   						=		self.add_weight(
										    shape=(d_emb_i, self.d_proj)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_projs_._{i}'  )
										self.out_projs.append(lowerCAmelCase_  )
										_snake_case   						=		self.add_weight(
										    shape=(r_idx - l_idx, d_emb_i)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_layers_._{i}_._weight'    ,   )
										_snake_case   						=		self.add_weight(
										    shape=(r_idx - l_idx,)    ,   initializer='zeros'    ,   trainable=lowerCAmelCase_    ,   name=F'out_layers_._{i}_._bias'    ,   )
										self.out_layers.append((weight, bias)  )
								super().build(lowerCAmelCase_  )


							@staticmethod
							def  lowerCamelCase				(       lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_=None  ):





								"""simple docstring"""






								_snake_case   						=		x
								if proj is not None:
									_snake_case   						=		tf.einsum('ibd,ed->ibe'    ,   lowerCAmelCase_    ,   lowerCAmelCase_  )
								return tf.einsum('ibd,nd->ibn'    ,   lowerCAmelCase_    ,   lowerCAmelCase_  ) + b


							@staticmethod
							def  lowerCamelCase				(       lowerCAmelCase_    ,   lowerCAmelCase_  ):





								"""simple docstring"""






								_snake_case   						=		shape_list(lowerCAmelCase_  )
								_snake_case   						=		tf.range(lp_size[0]    ,   dtype=target.dtype  )
								_snake_case   						=		tf.stack([r, target]    ,   1  )
								return tf.gather_nd(lowerCAmelCase_    ,   lowerCAmelCase_  )




							def  lowerCamelCase				(       self    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_=True    ,   lowerCAmelCase_=False  ):





								"""simple docstring"""






								_snake_case   						=		0
								if self.n_clusters == 0:
									_snake_case   						=		self._logit(lowerCAmelCase_    ,   self.out_layers[0][0]    ,   self.out_layers[0][1]    ,   self.out_projs[0]  )
									if target is not None:
										_snake_case   						=		tf.nn.sparse_softmax_cross_entropy_with_logits(labels=lowerCAmelCase_    ,   logits=lowerCAmelCase_  )
									_snake_case   						=		tf.nn.log_softmax(lowerCAmelCase_    ,   axis=-1  )
								else:
									_snake_case   						=		shape_list(lowerCAmelCase_  )
									_snake_case   						=		[]
									_snake_case   						=		tf.zeros(hidden_sizes[:2]  )
									for i in range(len(self.cutoffs  )  ):
										_snake_case						, _snake_case   						=		self.cutoff_ends[i], self.cutoff_ends[i + 1]
										if target is not None:
											_snake_case   						=		(target >= l_idx) & (target < r_idx)
											_snake_case   						=		tf.where(lowerCAmelCase_  )
											_snake_case   						=		tf.boolean_mask(lowerCAmelCase_    ,   lowerCAmelCase_  ) - l_idx

										if self.div_val == 1:
											_snake_case   						=		self.out_layers[0][0][l_idx:r_idx]
											_snake_case   						=		self.out_layers[0][1][l_idx:r_idx]
										else:
											_snake_case   						=		self.out_layers[i][0]
											_snake_case   						=		self.out_layers[i][1]

										if i == 0:
											_snake_case   						=		tf.concat([cur_W, self.cluster_weight]    ,   0  )
											_snake_case   						=		tf.concat([cur_b, self.cluster_bias]    ,   0  )

											_snake_case   						=		self._logit(lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   self.out_projs[0]  )
											_snake_case   						=		tf.nn.log_softmax(lowerCAmelCase_  )
											out.append(head_logprob[..., : self.cutoffs[0]]  )
											if target is not None:
												_snake_case   						=		tf.boolean_mask(lowerCAmelCase_    ,   lowerCAmelCase_  )
												_snake_case   						=		self._gather_logprob(lowerCAmelCase_    ,   lowerCAmelCase_  )
										else:
											_snake_case   						=		self._logit(lowerCAmelCase_    ,   lowerCAmelCase_    ,   lowerCAmelCase_    ,   self.out_projs[i]  )
											_snake_case   						=		tf.nn.log_softmax(lowerCAmelCase_  )
											_snake_case   						=		self.cutoffs[0] + i - 1  # No probability for the head cluster
											_snake_case   						=		head_logprob[..., cluster_prob_idx, None] + tail_logprob
											out.append(lowerCAmelCase_  )
											if target is not None:
												_snake_case   						=		tf.boolean_mask(lowerCAmelCase_    ,   lowerCAmelCase_  )
												_snake_case   						=		tf.boolean_mask(lowerCAmelCase_    ,   lowerCAmelCase_  )
												_snake_case   						=		self._gather_logprob(lowerCAmelCase_    ,   lowerCAmelCase_  )
												cur_logprob += cur_head_logprob[:, self.cutoff_ends[1] + i - 1]
										if target is not None:
											loss += tf.scatter_nd(lowerCAmelCase_    ,   -cur_logprob    ,   shape_list(lowerCAmelCase_  )  )
									_snake_case   						=		tf.concat(lowerCAmelCase_    ,   axis=-1  )

								if target is not None:
									if return_mean:
										_snake_case   						=		tf.reduce_mean(lowerCAmelCase_  )
									# Add the training-time loss value to the layer using `self.add_loss()`.
									self.add_loss(lowerCAmelCase_  )

									# Log the loss as a metric (we could log arbitrary metrics,
									# including different metrics for training and inference.
									self.add_metric(lowerCAmelCase_    ,   name=self.name    ,   aggregation='mean' if return_mean else ''  )

								return out



'''simple docstring'''







import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionTextToImagePipeline
from diffusers.utils.testing_utils import nightly, require_torch_gpu, torch_device


lowercase    :							Optional[Any]      =					False


class 						__UpperCAmelCase      (					unittest.TestCase						):
							pass


@nightly
@require_torch_gpu
class 						__UpperCAmelCase      (					unittest.TestCase						):


							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






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


							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		VersatileDiffusionTextToImagePipeline.from_pretrained('shi-labs/versatile-diffusion'  )
								# remove text_unet
								pipe.remove_unused_weights()
								pipe.to(lowerCAmelCase_  )
								pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

								_snake_case   						=		'A painting of a squirrel eating a burger '
								_snake_case   						=		torch.manual_seed(0  )
								_snake_case   						=		pipe(
								    prompt=lowerCAmelCase_    ,   generator=lowerCAmelCase_    ,   guidance_scale=7.5    ,   num_inference_steps=2    ,   output_type='numpy'  ).images

								with tempfile.TemporaryDirectory() as tmpdirname:
									pipe.save_pretrained(lowerCAmelCase_  )
									_snake_case   						=		VersatileDiffusionTextToImagePipeline.from_pretrained(lowerCAmelCase_  )
								pipe.to(lowerCAmelCase_  )
								pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

								_snake_case   						=		generator.manual_seed(0  )
								_snake_case   						=		pipe(
								    prompt=lowerCAmelCase_    ,   generator=lowerCAmelCase_    ,   guidance_scale=7.5    ,   num_inference_steps=2    ,   output_type='numpy'  ).images

								assert np.abs(image - new_image  ).sum() < 1E-5, "Models don't have the same forward pass"




							def  lowerCamelCase				(       self  ):





								"""simple docstring"""






								_snake_case   						=		VersatileDiffusionTextToImagePipeline.from_pretrained(
								    'shi-labs/versatile-diffusion'    ,   torch_dtype=torch.floataa  )
								pipe.to(lowerCAmelCase_  )
								pipe.set_progress_bar_config(disable=lowerCAmelCase_  )

								_snake_case   						=		'A painting of a squirrel eating a burger '
								_snake_case   						=		torch.manual_seed(0  )
								_snake_case   						=		pipe(
								    prompt=lowerCAmelCase_    ,   generator=lowerCAmelCase_    ,   guidance_scale=7.5    ,   num_inference_steps=50    ,   output_type='numpy'  ).images

								_snake_case   						=		image[0, 2_53:2_56, 2_53:2_56, -1]

								assert image.shape == (1, 5_12, 5_12, 3)
								_snake_case   						=		np.array([0.3367, 0.3169, 0.2656, 0.3870, 0.4790, 0.3796, 0.4009, 0.4878, 0.4778]  )

								assert np.abs(image_slice.flatten() - expected_slice  ).max() < 1E-2

