


from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block



@dataclass
class 							A   (	UpperCAmelCase__							):
					'''simple docstring'''






					A__      =			42






class 							A   (	nn.Module							):
					'''simple docstring'''






					def __init__(self   :						Optional[int]				, _UpperCAmelCase   :						Any=3				, _UpperCAmelCase   :						Any=3				, _UpperCAmelCase   :						Any=("DownEncoderBlock2D",)				, _UpperCAmelCase   :						List[str]=(64,)				, _UpperCAmelCase   :						Optional[Any]=2				, _UpperCAmelCase   :						List[Any]=32				, _UpperCAmelCase   :						Optional[Any]="silu"				, _UpperCAmelCase   :						List[str]=True				, )						->       Optional[int]:




							"""simple docstring"""


							super().__init__()
							lowercase__					      =		layers_per_block

							lowercase__					      =		torch.nn.Convad(
							    _UpperCAmelCase				, block_out_channels[0]				, kernel_size=3				, stride=1				, padding=1				, )

							lowercase__					      =		None
							lowercase__					      =		nn.ModuleList([]		)

							# down
							lowercase__					      =		block_out_channels[0]
							for i, down_block_type in enumerate(_UpperCAmelCase		):
									lowercase__					      =		output_channel
									lowercase__					      =		block_out_channels[i]
									lowercase__					      =		i == len(_UpperCAmelCase		) - 1

									lowercase__					      =		get_down_block(
									    _UpperCAmelCase				, num_layers=self.layers_per_block				, in_channels=_UpperCAmelCase				, out_channels=_UpperCAmelCase				, add_downsample=not is_final_block				, resnet_eps=1E-6				, downsample_padding=0				, resnet_act_fn=_UpperCAmelCase				, resnet_groups=_UpperCAmelCase				, attention_head_dim=_UpperCAmelCase				, temb_channels=_UpperCAmelCase				, )
									self.down_blocks.append(_UpperCAmelCase		)

							# mid
							lowercase__					      =		UNetMidBlockaD(
							    in_channels=block_out_channels[-1]				, resnet_eps=1E-6				, resnet_act_fn=_UpperCAmelCase				, output_scale_factor=1				, resnet_time_scale_shift="""default"""				, attention_head_dim=block_out_channels[-1]				, resnet_groups=_UpperCAmelCase				, temb_channels=_UpperCAmelCase				, )

							# out
							lowercase__					      =		nn.GroupNorm(num_channels=block_out_channels[-1]				, num_groups=_UpperCAmelCase				, eps=1E-6		)
							lowercase__					      =		nn.SiLU()

							lowercase__					      =		2 * out_channels if double_z else out_channels
							lowercase__					      =		nn.Convad(block_out_channels[-1]				, _UpperCAmelCase				, 3				, padding=1		)

							lowercase__					      =		False


					def   lowerCamelCase__					(self   :						Optional[int]				, _UpperCAmelCase   :						Union[str, Any]		)						->       str:




							"""simple docstring"""


							lowercase__					      =		x
							lowercase__					      =		self.conv_in(_UpperCAmelCase		)

							if self.training and self.gradient_checkpointing:

									def create_custom_forward(_UpperCAmelCase   :						List[Any]		):
											def custom_forward(*_UpperCAmelCase   :						List[Any]		):
													return module(*_UpperCAmelCase		)

											return custom_forward

									# down
									if is_torch_version(""">="""				, """1.11.0"""		):
											for down_block in self.down_blocks:
													lowercase__					      =		torch.utils.checkpoint.checkpoint(
													    create_custom_forward(_UpperCAmelCase		)				, _UpperCAmelCase				, use_reentrant=_UpperCAmelCase		)
											# middle
											lowercase__					      =		torch.utils.checkpoint.checkpoint(
											    create_custom_forward(self.mid_block		)				, _UpperCAmelCase				, use_reentrant=_UpperCAmelCase		)
									else:
											for down_block in self.down_blocks:
													lowercase__					      =		torch.utils.checkpoint.checkpoint(create_custom_forward(_UpperCAmelCase		)				, _UpperCAmelCase		)
											# middle
											lowercase__					      =		torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block		)				, _UpperCAmelCase		)

							else:
									# down
									for down_block in self.down_blocks:
											lowercase__					      =		down_block(_UpperCAmelCase		)

									# middle
									lowercase__					      =		self.mid_block(_UpperCAmelCase		)

							# post-process
							lowercase__					      =		self.conv_norm_out(_UpperCAmelCase		)
							lowercase__					      =		self.conv_act(_UpperCAmelCase		)
							lowercase__					      =		self.conv_out(_UpperCAmelCase		)

							return sample






class 							A   (	nn.Module							):
					'''simple docstring'''






					def __init__(self   :						Optional[int]				, _UpperCAmelCase   :						Optional[Any]=3				, _UpperCAmelCase   :						List[str]=3				, _UpperCAmelCase   :						Union[str, Any]=("UpDecoderBlock2D",)				, _UpperCAmelCase   :						Tuple=(64,)				, _UpperCAmelCase   :						Union[str, Any]=2				, _UpperCAmelCase   :						Dict=32				, _UpperCAmelCase   :						Union[str, Any]="silu"				, _UpperCAmelCase   :						Optional[Any]="group"				, )						->       Any:




							"""simple docstring"""


							super().__init__()
							lowercase__					      =		layers_per_block

							lowercase__					      =		nn.Convad(
							    _UpperCAmelCase				, block_out_channels[-1]				, kernel_size=3				, stride=1				, padding=1				, )

							lowercase__					      =		None
							lowercase__					      =		nn.ModuleList([]		)

							lowercase__					      =		in_channels if norm_type == """spatial""" else None

							# mid
							lowercase__					      =		UNetMidBlockaD(
							    in_channels=block_out_channels[-1]				, resnet_eps=1E-6				, resnet_act_fn=_UpperCAmelCase				, output_scale_factor=1				, resnet_time_scale_shift="""default""" if norm_type == """group""" else norm_type				, attention_head_dim=block_out_channels[-1]				, resnet_groups=_UpperCAmelCase				, temb_channels=_UpperCAmelCase				, )

							# up
							lowercase__					      =		list(reversed(_UpperCAmelCase		)		)
							lowercase__					      =		reversed_block_out_channels[0]
							for i, up_block_type in enumerate(_UpperCAmelCase		):
									lowercase__					      =		output_channel
									lowercase__					      =		reversed_block_out_channels[i]

									lowercase__					      =		i == len(_UpperCAmelCase		) - 1

									lowercase__					      =		get_up_block(
									    _UpperCAmelCase				, num_layers=self.layers_per_block + 1				, in_channels=_UpperCAmelCase				, out_channels=_UpperCAmelCase				, prev_output_channel=_UpperCAmelCase				, add_upsample=not is_final_block				, resnet_eps=1E-6				, resnet_act_fn=_UpperCAmelCase				, resnet_groups=_UpperCAmelCase				, attention_head_dim=_UpperCAmelCase				, temb_channels=_UpperCAmelCase				, resnet_time_scale_shift=_UpperCAmelCase				, )
									self.up_blocks.append(_UpperCAmelCase		)
									lowercase__					      =		output_channel

							# out
							if norm_type == "spatial":
									lowercase__					      =		SpatialNorm(block_out_channels[0]				, _UpperCAmelCase		)
							else:
									lowercase__					      =		nn.GroupNorm(num_channels=block_out_channels[0]				, num_groups=_UpperCAmelCase				, eps=1E-6		)
							lowercase__					      =		nn.SiLU()
							lowercase__					      =		nn.Convad(block_out_channels[0]				, _UpperCAmelCase				, 3				, padding=1		)

							lowercase__					      =		False


					def   lowerCamelCase__					(self   :						str				, _UpperCAmelCase   :						Tuple				, _UpperCAmelCase   :						List[str]=None		)						->       Tuple:




							"""simple docstring"""


							lowercase__					      =		z
							lowercase__					      =		self.conv_in(_UpperCAmelCase		)

							lowercase__					      =		next(iter(self.up_blocks.parameters()		)		).dtype
							if self.training and self.gradient_checkpointing:

									def create_custom_forward(_UpperCAmelCase   :						Union[str, Any]		):
											def custom_forward(*_UpperCAmelCase   :						Dict		):
													return module(*_UpperCAmelCase		)

											return custom_forward

									if is_torch_version(""">="""				, """1.11.0"""		):
											# middle
											lowercase__					      =		torch.utils.checkpoint.checkpoint(
											    create_custom_forward(self.mid_block		)				, _UpperCAmelCase				, _UpperCAmelCase				, use_reentrant=_UpperCAmelCase		)
											lowercase__					      =		sample.to(_UpperCAmelCase		)

											# up
											for up_block in self.up_blocks:
													lowercase__					      =		torch.utils.checkpoint.checkpoint(
													    create_custom_forward(_UpperCAmelCase		)				, _UpperCAmelCase				, _UpperCAmelCase				, use_reentrant=_UpperCAmelCase		)
									else:
											# middle
											lowercase__					      =		torch.utils.checkpoint.checkpoint(
											    create_custom_forward(self.mid_block		)				, _UpperCAmelCase				, _UpperCAmelCase		)
											lowercase__					      =		sample.to(_UpperCAmelCase		)

											# up
											for up_block in self.up_blocks:
													lowercase__					      =		torch.utils.checkpoint.checkpoint(create_custom_forward(_UpperCAmelCase		)				, _UpperCAmelCase				, _UpperCAmelCase		)
							else:
									# middle
									lowercase__					      =		self.mid_block(_UpperCAmelCase				, _UpperCAmelCase		)
									lowercase__					      =		sample.to(_UpperCAmelCase		)

									# up
									for up_block in self.up_blocks:
											lowercase__					      =		up_block(_UpperCAmelCase				, _UpperCAmelCase		)

        # post-process
							if latent_embeds is None:
									lowercase__					      =		self.conv_norm_out(_UpperCAmelCase		)
							else:
									lowercase__					      =		self.conv_norm_out(_UpperCAmelCase				, _UpperCAmelCase		)
							lowercase__					      =		self.conv_act(_UpperCAmelCase		)
							lowercase__					      =		self.conv_out(_UpperCAmelCase		)

							return sample






class 							A   (	nn.Module							):
					'''simple docstring'''






					def __init__(self   :						List[Any]				, _UpperCAmelCase   :						List[Any]				, _UpperCAmelCase   :						str				, _UpperCAmelCase   :						Optional[Any]				, _UpperCAmelCase   :						List[Any]=None				, _UpperCAmelCase   :						Dict="random"				, _UpperCAmelCase   :						str=False				, _UpperCAmelCase   :						int=True		)						->       Optional[Any]:




							"""simple docstring"""


							super().__init__()
							lowercase__					      =		n_e
							lowercase__					      =		vq_embed_dim
							lowercase__					      =		beta
							lowercase__					      =		legacy

							lowercase__					      =		nn.Embedding(self.n_e				, self.vq_embed_dim		)
							self.embedding.weight.data.uniform_(-1.0 / self.n_e				, 1.0 / self.n_e		)

							lowercase__					      =		remap
							if self.remap is not None:
									self.register_buffer("""used"""				, torch.tensor(np.load(self.remap		)		)		)
									lowercase__					      =		self.used.shape[0]
									lowercase__					      =		unknown_index  # "random" or "extra" or integer
									if self.unknown_index == "extra":
											lowercase__					      =		self.re_embed
											lowercase__					      =		self.re_embed + 1
									print(
									    f'''Remapping {self.n_e} indices to {self.re_embed} indices. '''
									    f'''Using {self.unknown_index} for unknown indices.'''		)
							else:
									lowercase__					      =		n_e

							lowercase__					      =		sane_index_shape

					def   lowerCamelCase__					(self   :						Any				, _UpperCAmelCase   :						Tuple		)						->       Optional[int]:




							"""simple docstring"""


							lowercase__					      =		inds.shape
							assert len(_UpperCAmelCase		) > 1
							lowercase__					      =		inds.reshape(ishape[0]				, -1		)
							lowercase__					      =		self.used.to(_UpperCAmelCase		)
							lowercase__					      =		(inds[:, :, None] == used[None, None, ...]).long()
							lowercase__					      =		match.argmax(-1		)
							lowercase__					      =		match.sum(2		) < 1
							if self.unknown_index == "random":
									lowercase__					      =		torch.randint(0				, self.re_embed				, size=new[unknown].shape		).to(device=new.device		)
							else:
									lowercase__					      =		self.unknown_index
							return new.reshape(_UpperCAmelCase		)

					def   lowerCamelCase__					(self   :						Union[str, Any]				, _UpperCAmelCase   :						Union[str, Any]		)						->       Any:




							"""simple docstring"""


							lowercase__					      =		inds.shape
							assert len(_UpperCAmelCase		) > 1
							lowercase__					      =		inds.reshape(ishape[0]				, -1		)
							lowercase__					      =		self.used.to(_UpperCAmelCase		)
							if self.re_embed > self.used.shape[0]:  # extra token
									lowercase__					      =		0  # simply set to zero
							lowercase__					      =		torch.gather(used[None, :][inds.shape[0] * [0], :]				, 1				, _UpperCAmelCase		)
							return back.reshape(_UpperCAmelCase		)

					def   lowerCamelCase__					(self   :						int				, _UpperCAmelCase   :						Tuple		)						->       Optional[Any]:




							"""simple docstring"""


							lowercase__					      =		z.permute(0				, 2				, 3				, 1		).contiguous()
							lowercase__					      =		z.view(-1				, self.vq_embed_dim		)

							# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
							lowercase__					      =		torch.argmin(torch.cdist(_UpperCAmelCase				, self.embedding.weight		)				, dim=1		)

							lowercase__					      =		self.embedding(_UpperCAmelCase		).view(z.shape		)
							lowercase__					      =		None
							lowercase__					      =		None

							# compute loss for embedding
							if not self.legacy:
									lowercase__					      =		self.beta * torch.mean((z_q.detach() - z) ** 2		) + torch.mean((z_q - z.detach()) ** 2		)
							else:
									lowercase__					      =		torch.mean((z_q.detach() - z) ** 2		) + self.beta * torch.mean((z_q - z.detach()) ** 2		)

							# preserve gradients
							lowercase__					      =		z + (z_q - z).detach()

							# reshape back to match original input shape
							lowercase__					      =		z_q.permute(0				, 3				, 1				, 2		).contiguous()

							if self.remap is not None:
									lowercase__					      =		min_encoding_indices.reshape(z.shape[0]				, -1		)  # add batch axis
									lowercase__					      =		self.remap_to_used(_UpperCAmelCase		)
									lowercase__					      =		min_encoding_indices.reshape(-1				, 1		)  # flatten

							if self.sane_index_shape:
									lowercase__					      =		min_encoding_indices.reshape(z_q.shape[0]				, z_q.shape[2]				, z_q.shape[3]		)

							return z_q, loss, (perplexity, min_encodings, min_encoding_indices)


					def   lowerCamelCase__					(self   :						List[Any]				, _UpperCAmelCase   :						List[str]				, _UpperCAmelCase   :						Any		)						->       Any:




							"""simple docstring"""


							if self.remap is not None:
									lowercase__					      =		indices.reshape(shape[0]				, -1		)  # add batch axis
									lowercase__					      =		self.unmap_to_all(_UpperCAmelCase		)
									lowercase__					      =		indices.reshape(-1		)  # flatten again

							# get quantized latent vectors
							lowercase__					      =		self.embedding(_UpperCAmelCase		)

							if shape is not None:
									lowercase__					      =		z_q.view(_UpperCAmelCase		)
									# reshape back to match original input shape
									lowercase__					      =		z_q.permute(0				, 3				, 1				, 2		).contiguous()

							return z_q






class 							A   (	UpperCAmelCase__							):
					'''simple docstring'''






					def __init__(self   :						List[Any]				, _UpperCAmelCase   :						Optional[Any]				, _UpperCAmelCase   :						int=False		)						->       Union[str, Any]:




							"""simple docstring"""


							lowercase__					      =		parameters
							lowercase__      ,     lowercase__					      =		torch.chunk(_UpperCAmelCase				, 2				, dim=1		)
							lowercase__					      =		torch.clamp(self.logvar				, -30.0				, 20.0		)
							lowercase__					      =		deterministic
							lowercase__					      =		torch.exp(0.5 * self.logvar		)
							lowercase__					      =		torch.exp(self.logvar		)
							if self.deterministic:
									lowercase__					      =		lowercase__					      =		torch.zeros_like(
									    self.mean				, device=self.parameters.device				, dtype=self.parameters.dtype		)

					def   lowerCamelCase__					(self   :						Any				, _UpperCAmelCase   :						Optional[torch.Generator] = None		)						->       torch.FloatTensor:




							"""simple docstring"""


							lowercase__					      =		randn_tensor(
							    self.mean.shape				, generator=_UpperCAmelCase				, device=self.parameters.device				, dtype=self.parameters.dtype		)
							lowercase__					      =		self.mean + self.std * sample
							return x

					def   lowerCamelCase__					(self   :						int				, _UpperCAmelCase   :						Any=None		)						->       List[str]:




							"""simple docstring"""


							if self.deterministic:
									return torch.Tensor([0.0]		)
							else:
									if other is None:
											return 0.5 * torch.sum(torch.pow(self.mean				, 2		) + self.var - 1.0 - self.logvar				, dim=[1, 2, 3]		)
									else:
											return 0.5 * torch.sum(
											    torch.pow(self.mean - other.mean				, 2		) / other.var
											    + self.var / other.var
											    - 1.0
											    - self.logvar
											    + other.logvar				, dim=[1, 2, 3]				, )

					def   lowerCamelCase__					(self   :						str				, _UpperCAmelCase   :						Tuple				, _UpperCAmelCase   :						str=[1, 2, 3]		)						->       List[Any]:




							"""simple docstring"""


							if self.deterministic:
									return torch.Tensor([0.0]		)
							lowercase__					      =		np.log(2.0 * np.pi		)
							return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean				, 2		) / self.var				, dim=_UpperCAmelCase		)


					def   lowerCamelCase__					(self   :						List[Any]		)						->       List[str]:




							"""simple docstring"""


							return self.mean





from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_sentencepiece, require_tf, require_tokenizers, slow


if is_tf_available():
	import numpy as np
	import tensorflow as tf

	from transformers import TFXLMRobertaModel



@require_tf
@require_sentencepiece
@require_tokenizers
class 							A   (	unittest.TestCase							):
					'''simple docstring'''






					@slow
					def   lowerCamelCase__					(self   :						Any		)						->       Union[str, Any]:




							"""simple docstring"""


							lowercase__					      =		TFXLMRobertaModel.from_pretrained("""jplu/tf-xlm-roberta-base"""		)

							lowercase__					      =		{
							    """input_ids""": tf.convert_to_tensor([[0, 2646, 1_0269, 83, 9_9942, 2]]				, dtype=tf.intaa		),  # "My dog is cute"
							    """attention_mask""": tf.convert_to_tensor([[1, 1, 1, 1, 1, 1]]				, dtype=tf.intaa		),
							}

							lowercase__					      =		model(_UpperCAmelCase		)["""last_hidden_state"""]
							lowercase__					      =		tf.TensorShape((1, 6, 768)		)
							self.assertEqual(output.shape				, _UpperCAmelCase		)
							# compare the actual values for a slice.
							lowercase__					      =		tf.convert_to_tensor(
							    [
							        [
							            [0.0_681_762, 0.10_894_451, 0.06_772_504],
							            [-0.06_423_668, 0.02_366_615, 0.04_329_344],
							            [-0.06_057_295, 0.09_974_135, -0.00_070_584],
							        ]
							    ]				, dtype=tf.floataa				, )

							self.assertTrue(np.allclose(output[:, :3, :3].numpy()				, expected_slice.numpy()				, atol=1E-4		)		)

