

'''simple docstring'''



import tensorflow as tf

from ...tf_utils import shape_list



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



			'''simple docstring'''






			def __init__(				self	:						Any		,      lowerCAmelCase_	:						List[str]		,      lowerCAmelCase_	:						Tuple		,      lowerCAmelCase_	:						int		,      lowerCAmelCase_	:						Any		,      lowerCAmelCase_	:						Dict=1		,      lowerCAmelCase_	:						Union[str, Any]=False		,      **lowerCAmelCase_	:						Tuple			)      ->		List[Any]:





									'''simple docstring'''
									super().__init__(**lowerCAmelCase_			)

									A__					:       List[Any]     =vocab_size
									A__					:       Tuple     =d_embed
									A__					:       str     =d_proj

									A__					:       List[Any]     =cutoffs + [vocab_size]
									A__					:       List[Any]     =[0] + self.cutoffs
									A__					:       int     =div_val

									A__					:       List[str]     =self.cutoffs[0]
									A__					:       Union[str, Any]     =len(self.cutoffs			) - 1
									A__					:       List[str]     =self.shortlist_size + self.n_clusters
									A__					:       List[Any]     =keep_order

									A__					:       Optional[Any]     =[]
									A__					:       int     =[]


			def lowercase__						(				self	:						int		,      lowerCAmelCase_	:						str			)      ->		int:





									'''simple docstring'''
									if self.n_clusters > 0:
															A__					:       Union[str, Any]     =self.add_weight(
															    shape=(self.n_clusters, self.d_embed)		,      initializer="""zeros"""		,      trainable=lowerCAmelCase_		,      name="""cluster_weight"""			)
															A__					:       Optional[int]     =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:
																											A__					:       List[Any]     =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_			)
																					A__					:       Optional[Any]     =self.add_weight(
																					    shape=(self.vocab_size, self.d_embed)		,      initializer="""zeros"""		,      trainable=lowerCAmelCase_		,      name=f"out_layers_._{i}_._weight"		,      )
																					A__					:       int     =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			)			):
																					A__   ,						A__					:       Tuple     =self.cutoff_ends[i], self.cutoff_ends[i + 1]
																					A__					:       List[str]     =self.d_embed // (self.div_val**i)

																					A__					:       Dict     =self.add_weight(
																					    shape=(d_emb_i, self.d_proj)		,      initializer="""zeros"""		,      trainable=lowerCAmelCase_		,      name=f"out_projs_._{i}"			)
																					self.out_projs.append(lowerCAmelCase_			)
																					A__					:       Union[str, Any]     =self.add_weight(
																					    shape=(r_idx - l_idx, d_emb_i)		,      initializer="""zeros"""		,      trainable=lowerCAmelCase_		,      name=f"out_layers_._{i}_._weight"		,      )
																					A__					:       str     =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 lowercase__						(				lowerCAmelCase_	:						Dict		,      lowerCAmelCase_	:						str		,      lowerCAmelCase_	:						Any		,      lowerCAmelCase_	:						int=None			)      ->		List[Any]:





									'''simple docstring'''
									A__					:       Dict     =x
									if proj is not None:
															A__					:       Union[str, Any]     =tf.einsum("""ibd,ed->ibe"""		,      lowerCAmelCase_		,      lowerCAmelCase_			)
									return tf.einsum("""ibd,nd->ibn"""		,      lowerCAmelCase_		,      lowerCAmelCase_			) + b


			@staticmethod
			def lowercase__						(				lowerCAmelCase_	:						Dict		,      lowerCAmelCase_	:						Union[str, Any]			)      ->		Dict:





									'''simple docstring'''
									A__					:       List[Any]     =shape_list(lowerCAmelCase_			)
									A__					:       Tuple     =tf.range(lp_size[0]		,      dtype=target.dtype			)
									A__					:       List[Any]     =tf.stack([r, target]		,      1			)
									return tf.gather_nd(lowerCAmelCase_		,      lowerCAmelCase_			)



			def lowercase__						(				self	:						int		,      lowerCAmelCase_	:						Dict		,      lowerCAmelCase_	:						List[str]		,      lowerCAmelCase_	:						Optional[Any]=True		,      lowerCAmelCase_	:						Dict=False			)      ->		List[Any]:





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

																					if self.div_val == 1:
																											A__					:       int     =self.out_layers[0][0][l_idx:r_idx]
																											A__					:       Dict     =self.out_layers[0][1][l_idx:r_idx]
																					else:
																											A__					:       int     =self.out_layers[i][0]
																											A__					:       Union[str, Any]     =self.out_layers[i][1]

																					if i == 0:
																											A__					:       Optional[Any]     =tf.concat([cur_W, self.cluster_weight]		,      0			)
																											A__					:       List[Any]     =tf.concat([cur_b, self.cluster_bias]		,      0			)

																											A__					:       Dict     =self._logit(lowerCAmelCase_		,      lowerCAmelCase_		,      lowerCAmelCase_		,      self.out_projs[0]			)
																											A__					:       Tuple     =tf.nn.log_softmax(lowerCAmelCase_			)
																											out.append(head_logprob[..., : self.cutoffs[0]]			)
																											if target is not None:
																																	A__					:       List[str]     =tf.boolean_mask(lowerCAmelCase_		,      lowerCAmelCase_			)
																																	A__					:       Any     =self._gather_logprob(lowerCAmelCase_		,      lowerCAmelCase_			)
																					else:
																											A__					:       Dict     =self._logit(lowerCAmelCase_		,      lowerCAmelCase_		,      lowerCAmelCase_		,      self.out_projs[i]			)
																											A__					:       Optional[Any]     =tf.nn.log_softmax(lowerCAmelCase_			)
																											A__					:       Optional[Any]     =self.cutoffs[0] + i - 1  # No probability for the head cluster
																											A__					:       List[Any]     =head_logprob[..., cluster_prob_idx, None] + tail_logprob
																											out.append(lowerCAmelCase_			)
																											if target is not None:
																																	A__					:       Union[str, Any]     =tf.boolean_mask(lowerCAmelCase_		,      lowerCAmelCase_			)
																																	A__					:       Union[str, Any]     =tf.boolean_mask(lowerCAmelCase_		,      lowerCAmelCase_			)
																																	A__					:       List[Any]     =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_			)			)
															A__					:       Optional[Any]     =tf.concat(lowerCAmelCase_		,      axis=-1			)

									if target is not None:
															if return_mean:
																					A__					:       Dict     =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 tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np



class lowerCamelCase			:



			'''simple docstring'''






			def lowercase__						(				self	:						Any			)      ->		Tuple:





									'''simple docstring'''
									torch.manual_seed(0			)
									A__					:       str     =TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""			)

									torch.manual_seed(0			)
									A__					:       int     =AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""			)

									torch.manual_seed(0			)
									A__					:       Union[str, Any]     =UNetaDConditionModel(
									    sample_size=32		,      layers_per_block=1		,      block_out_channels=[32, 64]		,      down_block_types=[
									        """ResnetDownsampleBlock2D""",
									        """SimpleCrossAttnDownBlock2D""",
									    ]		,      mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""		,      up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]		,      in_channels=3		,      out_channels=6		,      cross_attention_dim=32		,      encoder_hid_dim=32		,      attention_head_dim=8		,      addition_embed_type="""text"""		,      addition_embed_type_num_heads=2		,      cross_attention_norm="""group_norm"""		,      resnet_time_scale_shift="""scale_shift"""		,      act_fn="""gelu"""		,      )
									unet.set_attn_processor(AttnAddedKVProcessor()			)  # For reproducibility tests

									torch.manual_seed(0			)
									A__					:       Dict     =DDPMScheduler(
									    num_train_timesteps=10_00		,      beta_schedule="""squaredcos_cap_v2"""		,      beta_start=0.0001		,      beta_end=0.02		,      thresholding=lowerCAmelCase_		,      dynamic_thresholding_ratio=0.95		,      sample_max_value=1.0		,      prediction_type="""epsilon"""		,      variance_type="""learned_range"""		,      )

									torch.manual_seed(0			)
									A__					:       Union[str, Any]     =IFWatermarker()

									return {
									    "text_encoder": text_encoder,
									    "tokenizer": tokenizer,
									    "unet": unet,
									    "scheduler": scheduler,
									    "watermarker": watermarker,
									    "safety_checker": None,
									    "feature_extractor": None,
									}


			def lowercase__						(				self	:						Union[str, Any]			)      ->		int:





									'''simple docstring'''
									torch.manual_seed(0			)
									A__					:       List[Any]     =TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""			)

									torch.manual_seed(0			)
									A__					:       str     =AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""			)

									torch.manual_seed(0			)
									A__					:       Dict     =UNetaDConditionModel(
									    sample_size=32		,      layers_per_block=[1, 2]		,      block_out_channels=[32, 64]		,      down_block_types=[
									        """ResnetDownsampleBlock2D""",
									        """SimpleCrossAttnDownBlock2D""",
									    ]		,      mid_block_type="""UNetMidBlock2DSimpleCrossAttn"""		,      up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""]		,      in_channels=6		,      out_channels=6		,      cross_attention_dim=32		,      encoder_hid_dim=32		,      attention_head_dim=8		,      addition_embed_type="""text"""		,      addition_embed_type_num_heads=2		,      cross_attention_norm="""group_norm"""		,      resnet_time_scale_shift="""scale_shift"""		,      act_fn="""gelu"""		,      class_embed_type="""timestep"""		,      mid_block_scale_factor=1.414		,      time_embedding_act_fn="""gelu"""		,      time_embedding_dim=32		,      )
									unet.set_attn_processor(AttnAddedKVProcessor()			)  # For reproducibility tests

									torch.manual_seed(0			)
									A__					:       Optional[int]     =DDPMScheduler(
									    num_train_timesteps=10_00		,      beta_schedule="""squaredcos_cap_v2"""		,      beta_start=0.0001		,      beta_end=0.02		,      thresholding=lowerCAmelCase_		,      dynamic_thresholding_ratio=0.95		,      sample_max_value=1.0		,      prediction_type="""epsilon"""		,      variance_type="""learned_range"""		,      )

									torch.manual_seed(0			)
									A__					:       int     =DDPMScheduler(
									    num_train_timesteps=10_00		,      beta_schedule="""squaredcos_cap_v2"""		,      beta_start=0.0001		,      beta_end=0.02		,      )

									torch.manual_seed(0			)
									A__					:       List[str]     =IFWatermarker()

									return {
									    "text_encoder": text_encoder,
									    "tokenizer": tokenizer,
									    "unet": unet,
									    "scheduler": scheduler,
									    "image_noising_scheduler": image_noising_scheduler,
									    "watermarker": watermarker,
									    "safety_checker": None,
									    "feature_extractor": None,
									}


			def lowercase__						(				self	:						str			)      ->		Tuple:





									'''simple docstring'''
									A__					:       Tuple     =self.get_dummy_components()
									A__					:       str     =self.pipeline_class(**lowerCAmelCase_			)
									pipe.to(lowerCAmelCase_			)
									pipe.set_progress_bar_config(disable=lowerCAmelCase_			)

									A__					:       Optional[int]     =self.get_dummy_inputs(lowerCAmelCase_			)

									A__					:       str     =inputs["""prompt"""]
									A__					:       Optional[int]     =inputs["""generator"""]
									A__					:       Optional[Any]     =inputs["""num_inference_steps"""]
									A__					:       Union[str, Any]     =inputs["""output_type"""]

									if "image" in inputs:
															A__					:       Union[str, Any]     =inputs["""image"""]
									else:
															A__					:       List[Any]     =None

									if "mask_image" in inputs:
															A__					:       Union[str, Any]     =inputs["""mask_image"""]
									else:
															A__					:       Tuple     =None

									if "original_image" in inputs:
															A__					:       Optional[Any]     =inputs["""original_image"""]
									else:
															A__					:       Tuple     =None

									A__   ,						A__					:       Optional[Any]     =pipe.encode_prompt(lowerCAmelCase_			)

									# inputs with prompt converted to embeddings
									A__					:       Optional[int]     ={
									    """prompt_embeds""": prompt_embeds,
									    """negative_prompt_embeds""": negative_prompt_embeds,
									    """generator""": generator,
									    """num_inference_steps""": num_inference_steps,
									    """output_type""": output_type,
									}

									if image is not None:
															A__					:       int     =image

									if mask_image is not None:
															A__					:       Tuple     =mask_image

									if original_image is not None:
															A__					:       Optional[int]     =original_image

									# set all optional components to None
									for optional_component in pipe._optional_components:
															setattr(lowerCAmelCase_		,      lowerCAmelCase_		,      lowerCAmelCase_			)

									A__					:       Any     =pipe(**lowerCAmelCase_			)[0]

									with tempfile.TemporaryDirectory() as tmpdir:
															pipe.save_pretrained(lowerCAmelCase_			)
															A__					:       int     =self.pipeline_class.from_pretrained(lowerCAmelCase_			)
															pipe_loaded.to(lowerCAmelCase_			)
															pipe_loaded.set_progress_bar_config(disable=lowerCAmelCase_			)

									pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()			)  # For reproducibility tests

									for optional_component in pipe._optional_components:
															self.assertTrue(
															    getattr(lowerCAmelCase_		,      lowerCAmelCase_			) is None		,      f"`{optional_component}` did not stay set to None after loading."		,      )

									A__					:       Dict     =self.get_dummy_inputs(lowerCAmelCase_			)

									A__					:       int     =inputs["""generator"""]
									A__					:       str     =inputs["""num_inference_steps"""]
									A__					:       Optional[Any]     =inputs["""output_type"""]

									# inputs with prompt converted to embeddings
									A__					:       List[Any]     ={
									    """prompt_embeds""": prompt_embeds,
									    """negative_prompt_embeds""": negative_prompt_embeds,
									    """generator""": generator,
									    """num_inference_steps""": num_inference_steps,
									    """output_type""": output_type,
									}

									if image is not None:
															A__					:       int     =image

									if mask_image is not None:
															A__					:       int     =mask_image

									if original_image is not None:
															A__					:       Optional[int]     =original_image

									A__					:       List[str]     =pipe_loaded(**lowerCAmelCase_			)[0]

									A__					:       Union[str, Any]     =np.abs(to_np(lowerCAmelCase_			) - to_np(lowerCAmelCase_			)			).max()
									self.assertLess(lowerCAmelCase_		,      1e-4			)



			def lowercase__						(				self	:						List[str]			)      ->		Dict:





									'''simple docstring'''
									A__					:       Union[str, Any]     =self.get_dummy_components()
									A__					:       int     =self.pipeline_class(**lowerCAmelCase_			)
									pipe.to(lowerCAmelCase_			)
									pipe.set_progress_bar_config(disable=lowerCAmelCase_			)

									A__					:       int     =self.get_dummy_inputs(lowerCAmelCase_			)
									A__					:       List[Any]     =pipe(**lowerCAmelCase_			)[0]

									with tempfile.TemporaryDirectory() as tmpdir:
															pipe.save_pretrained(lowerCAmelCase_			)
															A__					:       List[Any]     =self.pipeline_class.from_pretrained(lowerCAmelCase_			)
															pipe_loaded.to(lowerCAmelCase_			)
															pipe_loaded.set_progress_bar_config(disable=lowerCAmelCase_			)

									pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()			)  # For reproducibility tests

									A__					:       int     =self.get_dummy_inputs(lowerCAmelCase_			)
									A__					:       Tuple     =pipe_loaded(**lowerCAmelCase_			)[0]

									A__					:       Tuple     =np.abs(to_np(lowerCAmelCase_			) - to_np(lowerCAmelCase_			)			).max()
									self.assertLess(lowerCAmelCase_		,      1e-4			)

