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_					):
   __UpperCAmelCase			:			torch.FloatTensor





class 					A							(    nn.Module					):


   def __init__(self					:						Dict	,  __UpperCAmelCase					:						str=3	,  __UpperCAmelCase					:						Dict=3	,  __UpperCAmelCase					:						Any=("DownEncoderBlock2D",)	,  __UpperCAmelCase					:						List[str]=(6_4,)	,  __UpperCAmelCase					:						int=2	,  __UpperCAmelCase					:						Dict=3_2	,  __UpperCAmelCase					:						Any="silu"	,  __UpperCAmelCase					:						Optional[int]=True	,  )						->    Tuple:
         """simple docstring"""



         super().__init__()
         UpperCAmelCase__								=  layers_per_block

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

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

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

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

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

         UpperCAmelCase__								=  False





   def       lowercase_				(self					:						Union[str, Any]	,  __UpperCAmelCase					:						Any		)						->    Tuple:
         """simple docstring"""



         UpperCAmelCase__								=  x
         UpperCAmelCase__								=  self.conv_in(__UpperCAmelCase		)

         if self.training and self.gradient_checkpointing:

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

                     return custom_forward

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

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

               # middle
               UpperCAmelCase__								=  self.mid_block(__UpperCAmelCase		)

         # post-process
         UpperCAmelCase__								=  self.conv_norm_out(__UpperCAmelCase		)
         UpperCAmelCase__								=  self.conv_act(__UpperCAmelCase		)
         UpperCAmelCase__								=  self.conv_out(__UpperCAmelCase		)

         return sample





class 					A							(    nn.Module					):


   def __init__(self					:						Optional[Any]	,  __UpperCAmelCase					:						Union[str, Any]=3	,  __UpperCAmelCase					:						str=3	,  __UpperCAmelCase					:						Dict=("UpDecoderBlock2D",)	,  __UpperCAmelCase					:						List[str]=(6_4,)	,  __UpperCAmelCase					:						List[str]=2	,  __UpperCAmelCase					:						Optional[int]=3_2	,  __UpperCAmelCase					:						str="silu"	,  __UpperCAmelCase					:						str="group"	,  )						->    Union[str, Any]:
         """simple docstring"""



         super().__init__()
         UpperCAmelCase__								=  layers_per_block

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

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

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

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

               UpperCAmelCase__								=  i == len(__UpperCAmelCase		) - 1

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

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

         UpperCAmelCase__								=  False





   def       lowercase_				(self					:						List[Any]	,  __UpperCAmelCase					:						Union[str, Any]	,  __UpperCAmelCase					:						Optional[Any]=None		)						->    Optional[int]:
         """simple docstring"""



         UpperCAmelCase__								=  z
         UpperCAmelCase__								=  self.conv_in(__UpperCAmelCase		)

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

               def create_custom_forward(__UpperCAmelCase					:						List[str]		):
                     def custom_forward(*__UpperCAmelCase					:						str		):
                           return module(*__UpperCAmelCase		)

                     return custom_forward

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

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

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

               # up
               for up_block in self.up_blocks:
                     UpperCAmelCase__								=  up_block(__UpperCAmelCase	,  __UpperCAmelCase		)

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

         return sample





class 					A							(    nn.Module					):


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



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

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

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

         UpperCAmelCase__								=  sane_index_shape


   def       lowercase_				(self					:						Union[str, Any]	,  __UpperCAmelCase					:						List[Any]		)						->    Dict:
         """simple docstring"""



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


   def       lowercase_				(self					:						Dict	,  __UpperCAmelCase					:						Dict		)						->    Tuple:
         """simple docstring"""



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


   def       lowercase_				(self					:						Optional[Any]	,  __UpperCAmelCase					:						List[str]		)						->    Tuple:
         """simple docstring"""



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

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

         UpperCAmelCase__								=  self.embedding(__UpperCAmelCase		).view(z.shape		)
         UpperCAmelCase__								=  None
         UpperCAmelCase__								=  None

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

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

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

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

         if self.sane_index_shape:
               UpperCAmelCase__								=  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       lowercase_				(self					:						List[Any]	,  __UpperCAmelCase					:						Union[str, Any]	,  __UpperCAmelCase					:						Any		)						->    Optional[int]:
         """simple docstring"""



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

         # get quantized latent vectors
         UpperCAmelCase__								=  self.embedding(__UpperCAmelCase		)

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

         return z_q





class 					A							(    UpperCAmelCase_					):


   def __init__(self					:						List[Any]	,  __UpperCAmelCase					:						Union[str, Any]	,  __UpperCAmelCase					:						List[str]=False		)						->    int:
         """simple docstring"""



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


   def       lowercase_				(self					:						List[Any]	,  __UpperCAmelCase					:						Optional[torch.Generator] = None		)						->    torch.FloatTensor:
         """simple docstring"""



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


   def       lowercase_				(self					:						Optional[int]	,  __UpperCAmelCase					:						int=None		)						->    Dict:
         """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       lowercase_				(self					:						List[Any]	,  __UpperCAmelCase					:						Any	,  __UpperCAmelCase					:						str=[1, 2, 3]		)						->    Any:
         """simple docstring"""



         if self.deterministic:
               return torch.Tensor([0.0]		)
         UpperCAmelCase__								=  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       lowercase_				(self					:						Union[str, Any]		)						->    Optional[Any]:
         """simple docstring"""



         return self.mean




# Lint as: python3
import itertools
import os
import re


UpperCamelCase__    							= re.compile(R'([A-Z]+)([A-Z][a-z])')
UpperCamelCase__    							= re.compile(R'([a-z\d])([A-Z])')

UpperCamelCase__    							= re.compile(R'(?<!_)_(?!_)')
UpperCamelCase__    							= re.compile(R'(_{2,})')

UpperCamelCase__    							= R'^\w+(\.\w+)*$'

UpperCamelCase__    							= R'<>:/\|?*'




def        lowerCAmelCase_			(					__A							)					->    Dict:







      '''simple docstring'''


      UpperCAmelCase__								=  _uppercase_uppercase_re.sub(r"\1_\2",  __A							)
      UpperCAmelCase__								=  _lowercase_uppercase_re.sub(r"\1_\2",  __A							)
      return name.lower()




def        lowerCAmelCase_			(					__A							)					->    Optional[Any]:







      '''simple docstring'''


      UpperCAmelCase__								=  _single_underscore_re.split(__A							)
      UpperCAmelCase__								=  [_multiple_underscores_re.split(__A							) for n in name]
      return "".join(n.capitalize() for n in itertools.chain.from_iterable(__A							) if n != ""							)




def        lowerCAmelCase_			(					__A							)					->    List[str]:







      '''simple docstring'''


      if os.path.basename(__A							) != name:
            raise ValueError(f"""Should be a dataset name, not a path: {name}"""							)
      return camelcase_to_snakecase(__A							)




def        lowerCAmelCase_			(					__A,  __A							)					->    Optional[int]:







      '''simple docstring'''


      if os.path.basename(__A							) != name:
            raise ValueError(f"""Should be a dataset name, not a path: {name}"""							)
      if not re.match(_split_re,  __A							):
            raise ValueError(f"""Split name should match '{_split_re}'' but got '{split}'."""							)
      return f"""{filename_prefix_for_name(__A							)}-{split}"""




def        lowerCAmelCase_			(					__A,  __A,  __A,  __A=None							)					->    str:







      '''simple docstring'''


      UpperCAmelCase__								=  filename_prefix_for_split(__A,  __A							)
      if filetype_suffix:
            prefix += f""".{filetype_suffix}"""
      UpperCAmelCase__								=  os.path.join(__A,  __A							)
      return f"""{filepath}*"""







def        lowerCAmelCase_			(					__A,  __A,  __A,  __A=None,  __A=None							)					->    Any:







      '''simple docstring'''


      UpperCAmelCase__								=  filename_prefix_for_split(__A,  __A							)
      UpperCAmelCase__								=  os.path.join(__A,  __A							)

      if shard_lengths:
            UpperCAmelCase__								=  len(__A							)
            UpperCAmelCase__								=  [f"""{prefix}-{shard_id:05d}-of-{num_shards:05d}""" for shard_id in range(__A							)]
            if filetype_suffix:
                  UpperCAmelCase__								=  [filename + f""".{filetype_suffix}""" for filename in filenames]
            return filenames
      else:
            UpperCAmelCase__								=  prefix
            if filetype_suffix:
                  filename += f""".{filetype_suffix}"""
            return [filename]



