




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_				( _SCREAMING_SNAKE_CASE   ):


 '''simple docstring'''







 lowercase_							       =       42





class 			lowerCamelCase_				( nn.Module   ):


 '''simple docstring'''







 def __init__(				self    :						Dict							,							_lowerCAmelCase    :						Union[str, Any]=3							,							_lowerCAmelCase    :						Optional[int]=3							,							_lowerCAmelCase    :						Union[str, Any]=("DownEncoderBlock2D",)							,							_lowerCAmelCase    :						Any=(64,)							,							_lowerCAmelCase    :						Optional[int]=2							,							_lowerCAmelCase    :						Tuple=32							,							_lowerCAmelCase    :						Any="silu"							,							_lowerCAmelCase    :						Any=True							,							):
       super().__init__()
       SCREAMING_SNAKE_CASE_      		=     layers_per_block

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

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

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

             SCREAMING_SNAKE_CASE_      		=     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
       SCREAMING_SNAKE_CASE_      		=     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
       SCREAMING_SNAKE_CASE_      		=     nn.GroupNorm(num_channels=block_out_channels[-1]							,							num_groups=_lowerCAmelCase							,							eps=1E-6		)
       SCREAMING_SNAKE_CASE_      		=     nn.SiLU()

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

       SCREAMING_SNAKE_CASE_      		=     False




 def  lowerCAmelCase_    (				self    :						Tuple							,							_lowerCAmelCase    :						int		):
       SCREAMING_SNAKE_CASE_      		=     x
       SCREAMING_SNAKE_CASE_      		=     self.conv_in(_lowerCAmelCase		)

       if self.training and self.gradient_checkpointing:

             def create_custom_forward(_lowerCAmelCase    :						Dict		):
                   def custom_forward(*_lowerCAmelCase    :						Optional[Any]		):
                         return module(*_lowerCAmelCase		)

                   return custom_forward

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

       else:
             # down
             for down_block in self.down_blocks:
                   SCREAMING_SNAKE_CASE_      		=     down_block(_lowerCAmelCase		)

             # middle
             SCREAMING_SNAKE_CASE_      		=     self.mid_block(_lowerCAmelCase		)

       # post-process
       SCREAMING_SNAKE_CASE_      		=     self.conv_norm_out(_lowerCAmelCase		)
       SCREAMING_SNAKE_CASE_      		=     self.conv_act(_lowerCAmelCase		)
       SCREAMING_SNAKE_CASE_      		=     self.conv_out(_lowerCAmelCase		)

       return sample





class 			lowerCamelCase_				( nn.Module   ):


 '''simple docstring'''







 def __init__(				self    :						Any							,							_lowerCAmelCase    :						str=3							,							_lowerCAmelCase    :						int=3							,							_lowerCAmelCase    :						List[Any]=("UpDecoderBlock2D",)							,							_lowerCAmelCase    :						List[str]=(64,)							,							_lowerCAmelCase    :						Optional[Any]=2							,							_lowerCAmelCase    :						Tuple=32							,							_lowerCAmelCase    :						int="silu"							,							_lowerCAmelCase    :						Any="group"							,							):
       super().__init__()
       SCREAMING_SNAKE_CASE_      		=     layers_per_block

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

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

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

       # mid
       SCREAMING_SNAKE_CASE_      		=     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
       SCREAMING_SNAKE_CASE_      		=     list(reversed(_lowerCAmelCase		)		)
       SCREAMING_SNAKE_CASE_      		=     reversed_block_out_channels[0]
       for i, up_block_type in enumerate(_lowerCAmelCase		):
             SCREAMING_SNAKE_CASE_      		=     output_channel
             SCREAMING_SNAKE_CASE_      		=     reversed_block_out_channels[i]

             SCREAMING_SNAKE_CASE_      		=     i == len(_lowerCAmelCase		) - 1

             SCREAMING_SNAKE_CASE_      		=     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		)
             SCREAMING_SNAKE_CASE_      		=     output_channel

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

       SCREAMING_SNAKE_CASE_      		=     False




 def  lowerCAmelCase_    (				self    :						Dict							,							_lowerCAmelCase    :						int							,							_lowerCAmelCase    :						List[Any]=None		):
       SCREAMING_SNAKE_CASE_      		=     z
       SCREAMING_SNAKE_CASE_      		=     self.conv_in(_lowerCAmelCase		)

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

             def create_custom_forward(_lowerCAmelCase    :						Dict		):
                   def custom_forward(*_lowerCAmelCase    :						int		):
                         return module(*_lowerCAmelCase		)

                   return custom_forward

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

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

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

             # up
             for up_block in self.up_blocks:
                   SCREAMING_SNAKE_CASE_      		=     up_block(_lowerCAmelCase							,							_lowerCAmelCase		)

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

       return sample





class 			lowerCamelCase_				( nn.Module   ):


 '''simple docstring'''







 def __init__(				self    :						str							,							_lowerCAmelCase    :						Optional[Any]							,							_lowerCAmelCase    :						Dict							,							_lowerCAmelCase    :						List[str]							,							_lowerCAmelCase    :						Optional[int]=None							,							_lowerCAmelCase    :						List[str]="random"							,							_lowerCAmelCase    :						Optional[int]=False							,							_lowerCAmelCase    :						Dict=True		):
       super().__init__()
       SCREAMING_SNAKE_CASE_      		=     n_e
       SCREAMING_SNAKE_CASE_      		=     vq_embed_dim
       SCREAMING_SNAKE_CASE_      		=     beta
       SCREAMING_SNAKE_CASE_      		=     legacy

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

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

       SCREAMING_SNAKE_CASE_      		=     sane_index_shape



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



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



 def  lowerCAmelCase_    (				self    :						Tuple							,							_lowerCAmelCase    :						Optional[int]		):
       # reshape z -> (batch, height, width, channel) and flatten
       SCREAMING_SNAKE_CASE_      		=     z.permute(0							,							2							,							3							,							1		).contiguous()
       SCREAMING_SNAKE_CASE_      		=     z.view(-1							,							self.vq_embed_dim		)

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

       SCREAMING_SNAKE_CASE_      		=     self.embedding(_lowerCAmelCase		).view(z.shape		)
       SCREAMING_SNAKE_CASE_      		=     None
       SCREAMING_SNAKE_CASE_      		=     None

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

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

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

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

       if self.sane_index_shape:
             SCREAMING_SNAKE_CASE_      		=     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    :						Union[str, Any]							,							_lowerCAmelCase    :						Dict							,							_lowerCAmelCase    :						int		):
       # shape specifying (batch, height, width, channel)
       if self.remap is not None:
             SCREAMING_SNAKE_CASE_      		=     indices.reshape(shape[0]							,							-1		)  # add batch axis
             SCREAMING_SNAKE_CASE_      		=     self.unmap_to_all(_lowerCAmelCase		)
             SCREAMING_SNAKE_CASE_      		=     indices.reshape(-1		)  # flatten again

       # get quantized latent vectors
       SCREAMING_SNAKE_CASE_      		=     self.embedding(_lowerCAmelCase		)

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

       return z_q





class 			lowerCamelCase_				( _SCREAMING_SNAKE_CASE   ):


 '''simple docstring'''







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



 def  lowerCAmelCase_    (				self    :						Optional[int]							,							_lowerCAmelCase    :						Optional[torch.Generator] = None		):
       # make sure sample is on the same device as the parameters and has same dtype
       SCREAMING_SNAKE_CASE_      		=     randn_tensor(
           self.mean.shape							,							generator=_lowerCAmelCase							,							device=self.parameters.device							,							dtype=self.parameters.dtype		)
       SCREAMING_SNAKE_CASE_      		=     self.mean + self.std * sample
       return x



 def  lowerCAmelCase_    (				self    :						List[str]							,							_lowerCAmelCase    :						Union[str, Any]=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  lowerCAmelCase_    (				self    :						Dict							,							_lowerCAmelCase    :						Any							,							_lowerCAmelCase    :						Optional[int]=[1, 2, 3]		):
       if self.deterministic:
             return torch.Tensor([0.0]		)
       SCREAMING_SNAKE_CASE_      		=     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  lowerCAmelCase_    (				self    :						Tuple		):
       return self.mean





from typing import TYPE_CHECKING

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


lowerCamelCase__					:  Optional[int]      							=				{'configuration_mmbt': ['MMBTConfig']}

try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   pass
else:
   lowerCamelCase__					:  Any      							=				['MMBTForClassification', 'MMBTModel', 'ModalEmbeddings']


if TYPE_CHECKING:
   from .configuration_mmbt import MMBTConfig

   try:
      if not is_torch_available():
         raise OptionalDependencyNotAvailable()
   except OptionalDependencyNotAvailable:
      pass
   else:
      from .modeling_mmbt import MMBTForClassification, MMBTModel, ModalEmbeddings

else:
   import sys

   lowerCamelCase__					:  str      							=				_LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)