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        lowercase_   (					__snake_case		):
							_lowerCamelCase						=		42




class        lowercase_   (					nn.Module		):



							def __init__(						self				,    lowercase_=3				,    lowercase_=3				,    lowercase_=("DownEncoderBlock2D",)				,    lowercase_=(64,)				,    lowercase_=2				,    lowercase_=32				,    lowercase_="silu"				,    lowercase_=True				,    ):
													super().__init__()
													_snake_case						:					List[Any]    =   layers_per_block

													_snake_case						:					Tuple    =   torch.nn.Convad(
													    __UpperCAmelCase				,    block_out_channels[0]				,    kernel_size=3				,    stride=1				,    padding=1				,    )

													_snake_case						:					Optional[int]    =   None
													_snake_case						:					Optional[int]    =   nn.ModuleList([]    )

													# down
													_snake_case						:					Tuple    =   block_out_channels[0]
													for i, down_block_type in enumerate(__UpperCAmelCase    ):
																			_snake_case						:					Dict    =   output_channel
																			_snake_case						:					str    =   block_out_channels[i]
																			_snake_case						:					Optional[int]    =   i == len(__UpperCAmelCase    ) - 1

																			_snake_case						:					List[Any]    =   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
													_snake_case						:					Tuple    =   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
													_snake_case						:					List[Any]    =   nn.GroupNorm(num_channels=block_out_channels[-1]				,    num_groups=__UpperCAmelCase				,    eps=1e-6    )
													_snake_case						:					Union[str, Any]    =   nn.SiLU()

													_snake_case						:					Dict    =   2 * out_channels if double_z else out_channels
													_snake_case						:					Optional[int]    =   nn.Convad(block_out_channels[-1]				,    __UpperCAmelCase				,    3				,    padding=1    )

													_snake_case						:					Tuple    =   False



							def 	UpperCamelCase	(						self				,    lowercase_    ):
													_snake_case						:					Optional[Any]    =   x
													_snake_case						:					Any    =   self.conv_in(__UpperCAmelCase    )

													if self.training and self.gradient_checkpointing:

																			def create_custom_forward(lowercase_    ):
																									def custom_forward(*lowercase_    ):
																															return module(*__UpperCAmelCase    )

																									return custom_forward

																			# down
																			if is_torch_version(">="				,    "1.11.0"    ):
																									for down_block in self.down_blocks:
																															_snake_case						:					int    =   torch.utils.checkpoint.checkpoint(
																															    create_custom_forward(__UpperCAmelCase    )				,    __UpperCAmelCase				,    use_reentrant=__UpperCAmelCase    )
																									# middle
																									_snake_case						:					Dict    =   torch.utils.checkpoint.checkpoint(
																									    create_custom_forward(self.mid_block    )				,    __UpperCAmelCase				,    use_reentrant=__UpperCAmelCase    )
																			else:
																									for down_block in self.down_blocks:
																															_snake_case						:					Optional[Any]    =   torch.utils.checkpoint.checkpoint(create_custom_forward(__UpperCAmelCase    )				,    __UpperCAmelCase    )
																									# middle
																									_snake_case						:					str    =   torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block    )				,    __UpperCAmelCase    )

													else:
																			# down
																			for down_block in self.down_blocks:
																									_snake_case						:					Dict    =   down_block(__UpperCAmelCase    )

																			# middle
																			_snake_case						:					List[str]    =   self.mid_block(__UpperCAmelCase    )

													# post-process
													_snake_case						:					Dict    =   self.conv_norm_out(__UpperCAmelCase    )
													_snake_case						:					Dict    =   self.conv_act(__UpperCAmelCase    )
													_snake_case						:					List[str]    =   self.conv_out(__UpperCAmelCase    )

													return sample




class        lowercase_   (					nn.Module		):



							def __init__(						self				,    lowercase_=3				,    lowercase_=3				,    lowercase_=("UpDecoderBlock2D",)				,    lowercase_=(64,)				,    lowercase_=2				,    lowercase_=32				,    lowercase_="silu"				,    lowercase_="group"				,    ):
													super().__init__()
													_snake_case						:					Optional[int]    =   layers_per_block

													_snake_case						:					Any    =   nn.Convad(
													    __UpperCAmelCase				,    block_out_channels[-1]				,    kernel_size=3				,    stride=1				,    padding=1				,    )

													_snake_case						:					List[str]    =   None
													_snake_case						:					Optional[Any]    =   nn.ModuleList([]    )

													_snake_case						:					str    =   in_channels if norm_type == "spatial" else None

													# mid
													_snake_case						:					Optional[int]    =   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
													_snake_case						:					int    =   list(reversed(__UpperCAmelCase    )    )
													_snake_case						:					str    =   reversed_block_out_channels[0]
													for i, up_block_type in enumerate(__UpperCAmelCase    ):
																			_snake_case						:					Any    =   output_channel
																			_snake_case						:					List[Any]    =   reversed_block_out_channels[i]

																			_snake_case						:					Union[str, Any]    =   i == len(__UpperCAmelCase    ) - 1

																			_snake_case						:					List[Any]    =   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    )
																			_snake_case						:					List[str]    =   output_channel

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

													_snake_case						:					Optional[int]    =   False



							def 	UpperCamelCase	(						self				,    lowercase_				,    lowercase_=None    ):
													_snake_case						:					Any    =   z
													_snake_case						:					Any    =   self.conv_in(__UpperCAmelCase    )

													_snake_case						:					List[str]    =   next(iter(self.up_blocks.parameters()    )    ).dtype
													if self.training and self.gradient_checkpointing:

																			def create_custom_forward(lowercase_    ):
																									def custom_forward(*lowercase_    ):
																															return module(*__UpperCAmelCase    )

																									return custom_forward

																			if is_torch_version(">="				,    "1.11.0"    ):
																									# middle
																									_snake_case						:					str    =   torch.utils.checkpoint.checkpoint(
																									    create_custom_forward(self.mid_block    )				,    __UpperCAmelCase				,    __UpperCAmelCase				,    use_reentrant=__UpperCAmelCase    )
																									_snake_case						:					Union[str, Any]    =   sample.to(__UpperCAmelCase    )

																									# up
																									for up_block in self.up_blocks:
																															_snake_case						:					Optional[Any]    =   torch.utils.checkpoint.checkpoint(
																															    create_custom_forward(__UpperCAmelCase    )				,    __UpperCAmelCase				,    __UpperCAmelCase				,    use_reentrant=__UpperCAmelCase    )
																			else:
																									# middle
																									_snake_case						:					Dict    =   torch.utils.checkpoint.checkpoint(
																									    create_custom_forward(self.mid_block    )				,    __UpperCAmelCase				,    __UpperCAmelCase    )
																									_snake_case						:					Tuple    =   sample.to(__UpperCAmelCase    )

																									# up
																									for up_block in self.up_blocks:
																															_snake_case						:					Optional[int]    =   torch.utils.checkpoint.checkpoint(create_custom_forward(__UpperCAmelCase    )				,    __UpperCAmelCase				,    __UpperCAmelCase    )
													else:
																			# middle
																			_snake_case						:					Optional[int]    =   self.mid_block(__UpperCAmelCase				,    __UpperCAmelCase    )
																			_snake_case						:					Tuple    =   sample.to(__UpperCAmelCase    )

																			# up
																			for up_block in self.up_blocks:
																									_snake_case						:					Optional[int]    =   up_block(__UpperCAmelCase				,    __UpperCAmelCase    )

																			# post-process
													if latent_embeds is None:
																			_snake_case						:					Union[str, Any]    =   self.conv_norm_out(__UpperCAmelCase    )
													else:
																			_snake_case						:					Any    =   self.conv_norm_out(__UpperCAmelCase				,    __UpperCAmelCase    )
													_snake_case						:					Union[str, Any]    =   self.conv_act(__UpperCAmelCase    )
													_snake_case						:					Union[str, Any]    =   self.conv_out(__UpperCAmelCase    )

													return sample




class        lowercase_   (					nn.Module		):



							def __init__(						self				,    lowercase_				,    lowercase_				,    lowercase_				,    lowercase_=None				,    lowercase_="random"				,    lowercase_=False				,    lowercase_=True    ):
													super().__init__()
													_snake_case						:					str    =   n_e
													_snake_case						:					Dict    =   vq_embed_dim
													_snake_case						:					Optional[Any]    =   beta
													_snake_case						:					Union[str, Any]    =   legacy

													_snake_case						:					List[Any]    =   nn.Embedding(self.n_e				,    self.vq_embed_dim    )
													self.embedding.weight.data.uniform_(-1.0 / self.n_e				,    1.0 / self.n_e    )

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

													_snake_case						:					Any    =   sane_index_shape



							def 	UpperCamelCase	(						self				,    lowercase_    ):
													_snake_case						:					Optional[Any]    =   inds.shape
													assert len(__UpperCAmelCase    ) > 1
													_snake_case						:					Optional[Any]    =   inds.reshape(ishape[0]				,    -1    )
													_snake_case						:					List[str]    =   self.used.to(__UpperCAmelCase    )
													_snake_case						:					Optional[int]    =   (inds[:, :, None] == used[None, None, ...]).long()
													_snake_case						:					Any    =   match.argmax(-1    )
													_snake_case						:					str    =   match.sum(2    ) < 1
													if self.unknown_index == "random":
																			_snake_case						:					int    =   torch.randint(0				,    self.re_embed				,    size=new[unknown].shape    ).to(device=new.device    )
													else:
																			_snake_case						:					Optional[Any]    =   self.unknown_index
													return new.reshape(__UpperCAmelCase    )



							def 	UpperCamelCase	(						self				,    lowercase_    ):
													_snake_case						:					List[Any]    =   inds.shape
													assert len(__UpperCAmelCase    ) > 1
													_snake_case						:					int    =   inds.reshape(ishape[0]				,    -1    )
													_snake_case						:					int    =   self.used.to(__UpperCAmelCase    )
													if self.re_embed > self.used.shape[0]:  # extra token
																			_snake_case						:					Union[str, Any]    =   0  # simply set to zero
													_snake_case						:					Optional[int]    =   torch.gather(used[None, :][inds.shape[0] * [0], :]				,    1				,    __UpperCAmelCase    )
													return back.reshape(__UpperCAmelCase    )



							def 	UpperCamelCase	(						self				,    lowercase_    ):
													_snake_case						:					Optional[Any]    =   z.permute(0				,    2				,    3				,    1    ).contiguous()
													_snake_case						:					Tuple    =   z.view(-1				,    self.vq_embed_dim    )

													# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
													_snake_case						:					List[str]    =   torch.argmin(torch.cdist(__UpperCAmelCase				,    self.embedding.weight    )				,    dim=1    )

													_snake_case						:					List[Any]    =   self.embedding(__UpperCAmelCase    ).view(z.shape    )
													_snake_case						:					Dict    =   None
													_snake_case						:					str    =   None

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

													# preserve gradients
													_snake_case						:					List[str]    =   z + (z_q - z).detach()

													# reshape back to match original input shape
													_snake_case						:					Union[str, Any]    =   z_q.permute(0				,    3				,    1				,    2    ).contiguous()

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

													if self.sane_index_shape:
																			_snake_case						:					Dict    =   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 	UpperCamelCase	(						self				,    lowercase_				,    lowercase_    ):


													if self.remap is not None:
																			_snake_case						:					Any    =   indices.reshape(shape[0]				,    -1    )  # add batch axis
																			_snake_case						:					Any    =   self.unmap_to_all(__UpperCAmelCase    )
																			_snake_case						:					List[str]    =   indices.reshape(-1    )  # flatten again

													# get quantized latent vectors
													_snake_case						:					Optional[int]    =   self.embedding(__UpperCAmelCase    )

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

													return z_q




class        lowercase_   (					__snake_case		):



							def __init__(						self				,    lowercase_				,    lowercase_=False    ):
													_snake_case						:					Union[str, Any]    =   parameters
													_snake_case    ,_snake_case						:					Any    =   torch.chunk(__UpperCAmelCase				,    2				,    dim=1    )
													_snake_case						:					List[str]    =   torch.clamp(self.logvar				,    -30.0				,    20.0    )
													_snake_case						:					Dict    =   deterministic
													_snake_case						:					Dict    =   torch.exp(0.5 * self.logvar    )
													_snake_case						:					Optional[Any]    =   torch.exp(self.logvar    )
													if self.deterministic:
																			_snake_case						:					Union[str, Any]    =   torch.zeros_like(
																			    self.mean				,    device=self.parameters.device				,    dtype=self.parameters.dtype    )



							def 	UpperCamelCase	(						self				,    lowercase_ = None    ):
													_snake_case						:					str    =   randn_tensor(
													    self.mean.shape				,    generator=__UpperCAmelCase				,    device=self.parameters.device				,    dtype=self.parameters.dtype    )
													_snake_case						:					Any    =   self.mean + self.std * sample
													return x



							def 	UpperCamelCase	(						self				,    lowercase_=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 	UpperCamelCase	(						self				,    lowercase_				,    lowercase_=[1, 2, 3]    ):


													if self.deterministic:
																			return torch.Tensor([0.0]    )
													_snake_case						:					Dict    =   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 	UpperCamelCase	(						self    ):
													return self.mean






# Copyright 2023 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

# rely on isort to merge the imports
from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tokenizers_available, is_torch_available


__SCREAMING_SNAKE_CASE						:   Any         =			{'configuration_mra': ['MRA_PRETRAINED_CONFIG_ARCHIVE_MAP', 'MraConfig']}

try:
				if not is_torch_available():
								raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
				pass
else:
				__SCREAMING_SNAKE_CASE						:   Optional[Any]         =			[
				    'MRA_PRETRAINED_MODEL_ARCHIVE_LIST',
				    'MraForMaskedLM',
				    'MraForMultipleChoice',
				    'MraForQuestionAnswering',
				    'MraForSequenceClassification',
				    'MraForTokenClassification',
				    'MraLayer',
				    'MraModel',
				    'MraPreTrainedModel',
				]


if TYPE_CHECKING:
				from .configuration_mra import MRA_PRETRAINED_CONFIG_ARCHIVE_MAP, MraConfig

				try:
								if not is_torch_available():
												raise OptionalDependencyNotAvailable()
				except OptionalDependencyNotAvailable:
								pass
				else:
								from .modeling_mra import (
								    MRA_PRETRAINED_MODEL_ARCHIVE_LIST,
								    MraForMaskedLM,
								    MraForMultipleChoice,
								    MraForQuestionAnswering,
								    MraForSequenceClassification,
								    MraForTokenClassification,
								    MraLayer,
								    MraModel,
								    MraPreTrainedModel,
								)
else:
				import sys

				__SCREAMING_SNAKE_CASE						:   str         =			_LazyModule(__name__, globals()['__file__'], _import_structure)