



"""simple docstring"""
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 							lowerCamelCase_(   A__    ):





					'''simple docstring'''



					lowercase__   :       torch.FloatTensor


class 							lowerCamelCase_(   nn.Module    ):





					'''simple docstring'''



					def __init__(    self				,       lowerCamelCase__=3				,       lowerCamelCase__=3				,       lowerCamelCase__=("DownEncoderBlock2D",)				,       lowerCamelCase__=(6_4,)				,       lowerCamelCase__=2				,       lowerCamelCase__=3_2				,       lowerCamelCase__="silu"				,       lowerCamelCase__=True				,       ):
									super().__init__()
									_lowerCamelCase        =				layers_per_block

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

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

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

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

									# mid
									_lowerCamelCase        =				UNetMidBlockaD(
									    in_channels=block_out_channels[-1]				,       resnet_eps=1e-6				,       resnet_act_fn=lowerCamelCase__				,       output_scale_factor=1				,       resnet_time_scale_shift='''default'''				,       attention_head_dim=block_out_channels[-1]				,       resnet_groups=lowerCamelCase__				,       temb_channels=lowerCamelCase__				,       )

									# out
									_lowerCamelCase        =				nn.GroupNorm(num_channels=block_out_channels[-1]				,       num_groups=lowerCamelCase__				,       eps=1e-6   )
									_lowerCamelCase        =				nn.SiLU()

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

									_lowerCamelCase        =				False





					def 			snake_case__      (    self				,       lowerCamelCase__   ):
									_lowerCamelCase        =				x
									_lowerCamelCase        =				self.conv_in(lowerCamelCase__   )

									if self.training and self.gradient_checkpointing:

													def create_custom_forward(lowerCamelCase__   ):
																	def custom_forward(*lowerCamelCase__   ):
																					return module(*lowerCamelCase__   )

																	return custom_forward

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

									else:
													# down
													for down_block in self.down_blocks:
																	_lowerCamelCase        =				down_block(lowerCamelCase__   )

													# middle
													_lowerCamelCase        =				self.mid_block(lowerCamelCase__   )

									# post-process
									_lowerCamelCase        =				self.conv_norm_out(lowerCamelCase__   )
									_lowerCamelCase        =				self.conv_act(lowerCamelCase__   )
									_lowerCamelCase        =				self.conv_out(lowerCamelCase__   )

									return sample


class 							lowerCamelCase_(   nn.Module    ):





					'''simple docstring'''



					def __init__(    self				,       lowerCamelCase__=3				,       lowerCamelCase__=3				,       lowerCamelCase__=("UpDecoderBlock2D",)				,       lowerCamelCase__=(6_4,)				,       lowerCamelCase__=2				,       lowerCamelCase__=3_2				,       lowerCamelCase__="silu"				,       lowerCamelCase__="group"				,       ):
									super().__init__()
									_lowerCamelCase        =				layers_per_block

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

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

									_lowerCamelCase        =				in_channels if norm_type == '''spatial''' else None

									# mid
									_lowerCamelCase        =				UNetMidBlockaD(
									    in_channels=block_out_channels[-1]				,       resnet_eps=1e-6				,       resnet_act_fn=lowerCamelCase__				,       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=lowerCamelCase__				,       temb_channels=lowerCamelCase__				,       )

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

													_lowerCamelCase        =				i == len(lowerCamelCase__   ) - 1

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

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

									_lowerCamelCase        =				False





					def 			snake_case__      (    self				,       lowerCamelCase__				,       lowerCamelCase__=None   ):
									_lowerCamelCase        =				z
									_lowerCamelCase        =				self.conv_in(lowerCamelCase__   )

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

													def create_custom_forward(lowerCamelCase__   ):
																	def custom_forward(*lowerCamelCase__   ):
																					return module(*lowerCamelCase__   )

																	return custom_forward

													if is_torch_version('''>='''				,       '''1.11.0'''   ):
																	# middle
																	_lowerCamelCase        =				torch.utils.checkpoint.checkpoint(
																	    create_custom_forward(self.mid_block   )				,       lowerCamelCase__				,       lowerCamelCase__				,       use_reentrant=lowerCamelCase__   )
																	_lowerCamelCase        =				sample.to(lowerCamelCase__   )

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

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

													# up
													for up_block in self.up_blocks:
																	_lowerCamelCase        =				up_block(lowerCamelCase__				,       lowerCamelCase__   )

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

									return sample


class 							lowerCamelCase_(   nn.Module    ):





					'''simple docstring'''



					def __init__(    self				,       lowerCamelCase__				,       lowerCamelCase__				,       lowerCamelCase__				,       lowerCamelCase__=None				,       lowerCamelCase__="random"				,       lowerCamelCase__=False				,       lowerCamelCase__=True   ):
									super().__init__()
									_lowerCamelCase        =				n_e
									_lowerCamelCase        =				vq_embed_dim
									_lowerCamelCase        =				beta
									_lowerCamelCase        =				legacy

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

									_lowerCamelCase        =				remap
									if self.remap is not None:
													self.register_buffer('''used'''				,       torch.tensor(np.load(self.remap   )   )   )
													_lowerCamelCase        =				self.used.shape[0]
													_lowerCamelCase        =				unknown_index  # "random" or "extra" or integer
													if self.unknown_index == "extra":
																	_lowerCamelCase        =				self.re_embed
																	_lowerCamelCase        =				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:
													_lowerCamelCase        =				n_e

									_lowerCamelCase        =				sane_index_shape

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

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

					def 			snake_case__      (    self				,       lowerCamelCase__   ):
									# reshape z -> (batch, height, width, channel) and flatten
									_lowerCamelCase        =				z.permute(0				,       2				,       3				,       1   ).contiguous()
									_lowerCamelCase        =				z.view(-1				,       self.vq_embed_dim   )

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

									_lowerCamelCase        =				self.embedding(lowerCamelCase__   ).view(z.shape   )
									_lowerCamelCase        =				None
									_lowerCamelCase        =				None

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

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

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

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

									if self.sane_index_shape:
													_lowerCamelCase        =				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 			snake_case__      (    self				,       lowerCamelCase__				,       lowerCamelCase__   ):
									# shape specifying (batch, height, width, channel)
									if self.remap is not None:
													_lowerCamelCase        =				indices.reshape(shape[0]				,       -1   )  # add batch axis
													_lowerCamelCase        =				self.unmap_to_all(lowerCamelCase__   )
													_lowerCamelCase        =				indices.reshape(-1   )  # flatten again

									# get quantized latent vectors
									_lowerCamelCase        =				self.embedding(lowerCamelCase__   )

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

									return z_q


class 							lowerCamelCase_(   A__    ):





					'''simple docstring'''



					def __init__(    self				,       lowerCamelCase__				,       lowerCamelCase__=False   ):
									_lowerCamelCase        =				parameters
									_lowerCamelCase  ,     _lowerCamelCase        =				torch.chunk(lowerCamelCase__				,       2				,       dim=1   )
									_lowerCamelCase        =				torch.clamp(self.logvar				,       -3_0.0				,       2_0.0   )
									_lowerCamelCase        =				deterministic
									_lowerCamelCase        =				torch.exp(0.5 * self.logvar   )
									_lowerCamelCase        =				torch.exp(self.logvar   )
									if self.deterministic:
													_lowerCamelCase        =				_lowerCamelCase        =				torch.zeros_like(
													    self.mean				,       device=self.parameters.device				,       dtype=self.parameters.dtype   )

					def 			snake_case__      (    self				,       lowerCamelCase__ = None   ):
									# make sure sample is on the same device as the parameters and has same dtype
									_lowerCamelCase        =				randn_tensor(
									    self.mean.shape				,       generator=lowerCamelCase__				,       device=self.parameters.device				,       dtype=self.parameters.dtype   )
									_lowerCamelCase        =				self.mean + self.std * sample
									return x

					def 			snake_case__      (    self				,       lowerCamelCase__=None   ):
									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 			snake_case__      (    self				,       lowerCamelCase__				,       lowerCamelCase__=[1, 2, 3]   ):
									if self.deterministic:
													return torch.Tensor([0.0]   )
									_lowerCamelCase        =				np.log(2.0 * np.pi   )
									return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean				,       2   ) / self.var				,       dim=lowerCamelCase__   )





					def 			snake_case__      (    self   ):
									return self.mean





"""simple docstring"""
from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tf_available, is_torch_available


__SCREAMING_SNAKE_CASE  :						List[Any]    							=      {
    '''configuration_xlm''': ['''XLM_PRETRAINED_CONFIG_ARCHIVE_MAP''', '''XLMConfig''', '''XLMOnnxConfig'''],
    '''tokenization_xlm''': ['''XLMTokenizer'''],
}

try:
			if not is_torch_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			__SCREAMING_SNAKE_CASE  :						int    							=      [
			    '''XLM_PRETRAINED_MODEL_ARCHIVE_LIST''',
			    '''XLMForMultipleChoice''',
			    '''XLMForQuestionAnswering''',
			    '''XLMForQuestionAnsweringSimple''',
			    '''XLMForSequenceClassification''',
			    '''XLMForTokenClassification''',
			    '''XLMModel''',
			    '''XLMPreTrainedModel''',
			    '''XLMWithLMHeadModel''',
			]

try:
			if not is_tf_available():
						raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
			pass
else:
			__SCREAMING_SNAKE_CASE  :						Union[str, Any]    							=      [
			    '''TF_XLM_PRETRAINED_MODEL_ARCHIVE_LIST''',
			    '''TFXLMForMultipleChoice''',
			    '''TFXLMForQuestionAnsweringSimple''',
			    '''TFXLMForSequenceClassification''',
			    '''TFXLMForTokenClassification''',
			    '''TFXLMMainLayer''',
			    '''TFXLMModel''',
			    '''TFXLMPreTrainedModel''',
			    '''TFXLMWithLMHeadModel''',
			]


if TYPE_CHECKING:
			from .configuration_xlm import XLM_PRETRAINED_CONFIG_ARCHIVE_MAP, XLMConfig, XLMOnnxConfig
			from .tokenization_xlm import XLMTokenizer

			try:
						if not is_torch_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_xlm import (
						    XLM_PRETRAINED_MODEL_ARCHIVE_LIST,
						    XLMForMultipleChoice,
						    XLMForQuestionAnswering,
						    XLMForQuestionAnsweringSimple,
						    XLMForSequenceClassification,
						    XLMForTokenClassification,
						    XLMModel,
						    XLMPreTrainedModel,
						    XLMWithLMHeadModel,
						)

			try:
						if not is_tf_available():
									raise OptionalDependencyNotAvailable()
			except OptionalDependencyNotAvailable:
						pass
			else:
						from .modeling_tf_xlm import (
						    TF_XLM_PRETRAINED_MODEL_ARCHIVE_LIST,
						    TFXLMForMultipleChoice,
						    TFXLMForQuestionAnsweringSimple,
						    TFXLMForSequenceClassification,
						    TFXLMForTokenClassification,
						    TFXLMMainLayer,
						    TFXLMModel,
						    TFXLMPreTrainedModel,
						    TFXLMWithLMHeadModel,
						)

else:
			import sys

			__SCREAMING_SNAKE_CASE  :						Tuple    							=      _LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)
