
'''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  __A				( __SCREAMING_SNAKE_CASE			):
 lowerCamelCase				       =42







class  __A				( nn.Module			):
 def __init__(   self			:    Optional[int]				,   lowerCamelCase			:    Optional[int]=3				,   lowerCamelCase			:    Optional[int]=3				,   lowerCamelCase			:    Optional[Any]=("DownEncoderBlock2D",)				,   lowerCamelCase			:    Tuple=(64,)				,   lowerCamelCase			:    Optional[int]=2				,   lowerCamelCase			:    List[Any]=32				,   lowerCamelCase			:    Dict="silu"				,   lowerCamelCase			:    int=True				,   ):



   """simple docstring"""





   super().__init__()
   __A		:     Optional[Any]												=		layers_per_block

   __A		:     List[str]												=		torch.nn.Convad(
       __snake_case				,   block_out_channels[0]				,   kernel_size=3				,   stride=1				,   padding=1				,   )

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

   # down
   __A		:     Tuple												=		block_out_channels[0]
   for i, down_block_type in enumerate(__snake_case       ):
     __A		:     Union[str, Any]												=		output_channel
     __A		:     Tuple												=		block_out_channels[i]
     __A		:     Dict												=		i == len(__snake_case       ) - 1

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

   # mid
   __A		:     List[Any]												=		UNetMidBlockaD(
       in_channels=block_out_channels[-1]				,   resnet_eps=1e-6				,   resnet_act_fn=__snake_case				,   output_scale_factor=1				,   resnet_time_scale_shift="""default"""				,   attention_head_dim=block_out_channels[-1]				,   resnet_groups=__snake_case				,   temb_channels=__snake_case				,   )

   # out
   __A		:     Optional[int]												=		nn.GroupNorm(num_channels=block_out_channels[-1]				,   num_groups=__snake_case				,   eps=1e-6       )
   __A		:     Union[str, Any]												=		nn.SiLU()

   __A		:     Any												=		2 * out_channels if double_z else out_channels
   __A		:     Optional[int]												=		nn.Convad(block_out_channels[-1]				,   __snake_case				,   3				,   padding=1       )

   __A		:     str												=		False






 def 		lowercase_(   self			:    str				,   lowerCamelCase			:    Optional[Any]       ):



   """simple docstring"""





   __A		:     Optional[int]												=		x
   __A		:     Optional[Any]												=		self.conv_in(__snake_case       )

   if self.training and self.gradient_checkpointing:

     def create_custom_forward(lowerCamelCase			:    Optional[int]       ):
       def custom_forward(*lowerCamelCase			:    Union[str, Any]       ):
         return module(*__snake_case       )

       return custom_forward

     # down
     if is_torch_version(""">="""				,   """1.11.0"""       ):
       for down_block in self.down_blocks:
         __A		:     List[str]												=		torch.utils.checkpoint.checkpoint(
             create_custom_forward(__snake_case       )				,   __snake_case				,   use_reentrant=__snake_case       )
       # middle
       __A		:     int												=		torch.utils.checkpoint.checkpoint(
           create_custom_forward(self.mid_block       )				,   __snake_case				,   use_reentrant=__snake_case       )
     else:
       for down_block in self.down_blocks:
         __A		:     Optional[int]												=		torch.utils.checkpoint.checkpoint(create_custom_forward(__snake_case       )				,   __snake_case       )
       # middle
       __A		:     Tuple												=		torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block       )				,   __snake_case       )

   else:
     # down
     for down_block in self.down_blocks:
       __A		:     Optional[Any]												=		down_block(__snake_case       )

     # middle
     __A		:     List[str]												=		self.mid_block(__snake_case       )

   # post-process
   __A		:     Tuple												=		self.conv_norm_out(__snake_case       )
   __A		:     List[str]												=		self.conv_act(__snake_case       )
   __A		:     Tuple												=		self.conv_out(__snake_case       )

   return sample







class  __A				( nn.Module			):
 def __init__(   self			:    List[str]				,   lowerCamelCase			:    List[Any]=3				,   lowerCamelCase			:    Tuple=3				,   lowerCamelCase			:    Optional[int]=("UpDecoderBlock2D",)				,   lowerCamelCase			:    Tuple=(64,)				,   lowerCamelCase			:    str=2				,   lowerCamelCase			:    Optional[int]=32				,   lowerCamelCase			:    List[str]="silu"				,   lowerCamelCase			:    List[Any]="group"				,   ):



   """simple docstring"""





   super().__init__()
   __A		:     str												=		layers_per_block

   __A		:     Tuple												=		nn.Convad(
       __snake_case				,   block_out_channels[-1]				,   kernel_size=3				,   stride=1				,   padding=1				,   )

   __A		:     Union[str, Any]												=		None
   __A		:     Tuple												=		nn.ModuleList([]       )

   __A		:     str												=		in_channels if norm_type == '''spatial''' else None

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

   # up
   __A		:     Union[str, Any]												=		list(reversed(__snake_case       )       )
   __A		:     Optional[Any]												=		reversed_block_out_channels[0]
   for i, up_block_type in enumerate(__snake_case       ):
     __A		:     Optional[Any]												=		output_channel
     __A		:     str												=		reversed_block_out_channels[i]

     __A		:     int												=		i == len(__snake_case       ) - 1

     __A		:     List[Any]												=		get_up_block(
         __snake_case				,   num_layers=self.layers_per_block + 1				,   in_channels=__snake_case				,   out_channels=__snake_case				,   prev_output_channel=__snake_case				,   add_upsample=not is_final_block				,   resnet_eps=1e-6				,   resnet_act_fn=__snake_case				,   resnet_groups=__snake_case				,   attention_head_dim=__snake_case				,   temb_channels=__snake_case				,   resnet_time_scale_shift=__snake_case				,   )
     self.up_blocks.append(__snake_case       )
     __A		:     Optional[Any]												=		output_channel

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

   __A		:     str												=		False






 def 		lowercase_(   self			:    str				,   lowerCamelCase			:    Union[str, Any]				,   lowerCamelCase			:    Optional[Any]=None       ):



   """simple docstring"""





   __A		:     str												=		z
   __A		:     List[Any]												=		self.conv_in(__snake_case       )

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

     def create_custom_forward(lowerCamelCase			:    Any       ):
       def custom_forward(*lowerCamelCase			:    Optional[int]       ):
         return module(*__snake_case       )

       return custom_forward

     if is_torch_version(""">="""				,   """1.11.0"""       ):
       # middle
       __A		:     int												=		torch.utils.checkpoint.checkpoint(
           create_custom_forward(self.mid_block       )				,   __snake_case				,   __snake_case				,   use_reentrant=__snake_case       )
       __A		:     Optional[int]												=		sample.to(__snake_case       )

       # up
       for up_block in self.up_blocks:
         __A		:     List[Any]												=		torch.utils.checkpoint.checkpoint(
             create_custom_forward(__snake_case       )				,   __snake_case				,   __snake_case				,   use_reentrant=__snake_case       )
     else:
       # middle
       __A		:     str												=		torch.utils.checkpoint.checkpoint(
           create_custom_forward(self.mid_block       )				,   __snake_case				,   __snake_case       )
       __A		:     Tuple												=		sample.to(__snake_case       )

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

     # up
     for up_block in self.up_blocks:
       __A		:     Optional[Any]												=		up_block(__snake_case				,   __snake_case       )

        # post-process
   if latent_embeds is None:
     __A		:     Optional[int]												=		self.conv_norm_out(__snake_case       )
   else:
     __A		:     List[Any]												=		self.conv_norm_out(__snake_case				,   __snake_case       )
   __A		:     Tuple												=		self.conv_act(__snake_case       )
   __A		:     Union[str, Any]												=		self.conv_out(__snake_case       )

   return sample







class  __A				( nn.Module			):
 def __init__(   self			:    Optional[int]				,   lowerCamelCase			:    Tuple				,   lowerCamelCase			:    Union[str, Any]				,   lowerCamelCase			:    int				,   lowerCamelCase			:    Optional[Any]=None				,   lowerCamelCase			:    Optional[Any]="random"				,   lowerCamelCase			:    Any=False				,   lowerCamelCase			:    int=True       ):



   """simple docstring"""





   super().__init__()
   __A		:     List[str]												=		n_e
   __A		:     Dict												=		vq_embed_dim
   __A		:     int												=		beta
   __A		:     Optional[int]												=		legacy

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

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

   __A		:     Optional[Any]												=		sane_index_shape
 def 		lowercase_(   self			:    Optional[Any]				,   lowerCamelCase			:    Optional[Any]       ):



   """simple docstring"""





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



   """simple docstring"""





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



   """simple docstring"""





   __A		:     Any												=		z.permute(0				,   2				,   3				,   1       ).contiguous()
   __A		:     Optional[int]												=		z.view(-1				,   self.vq_embed_dim       )

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

   __A		:     List[str]												=		self.embedding(__snake_case       ).view(z.shape       )
   __A		:     Optional[int]												=		None
   __A		:     Tuple												=		None

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

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

   # reshape back to match original input shape
   __A		:     Optional[int]												=		z_q.permute(0				,   3				,   1				,   2       ).contiguous()

   if self.remap is not None:
     __A		:     Dict												=		min_encoding_indices.reshape(z.shape[0]				,   -1       )  # add batch axis
     __A		:     int												=		self.remap_to_used(__snake_case       )
     __A		:     Union[str, Any]												=		min_encoding_indices.reshape(-1				,   1       )  # flatten

   if self.sane_index_shape:
     __A		:     Tuple												=		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			:    str				,   lowerCamelCase			:    List[Any]				,   lowerCamelCase			:    Union[str, Any]       ):



   """simple docstring"""





   if self.remap is not None:
     __A		:     Optional[int]												=		indices.reshape(shape[0]				,   -1       )  # add batch axis
     __A		:     Tuple												=		self.unmap_to_all(__snake_case       )
     __A		:     List[Any]												=		indices.reshape(-1       )  # flatten again

   # get quantized latent vectors
   __A		:     Union[str, Any]												=		self.embedding(__snake_case       )

   if shape is not None:
     __A		:     Optional[Any]												=		z_q.view(__snake_case       )
     # reshape back to match original input shape
     __A		:     Optional[int]												=		z_q.permute(0				,   3				,   1				,   2       ).contiguous()

   return z_q







class  __A				( __SCREAMING_SNAKE_CASE			):
 def __init__(   self			:    Union[str, Any]				,   lowerCamelCase			:    Optional[int]				,   lowerCamelCase			:    str=False       ):



   """simple docstring"""





   __A		:     Any												=		parameters
   __A		:     List[str]												=		torch.chunk(__snake_case				,   2				,   dim=1       )
   __A		:     List[Any]												=		torch.clamp(self.logvar				,   -30.0				,   20.0       )
   __A		:     str												=		deterministic
   __A		:     Dict												=		torch.exp(0.5 * self.logvar       )
   __A		:     Optional[int]												=		torch.exp(self.logvar       )
   if self.deterministic:
     __A		:     Optional[int]												=		torch.zeros_like(
         self.mean				,   device=self.parameters.device				,   dtype=self.parameters.dtype       )
 def 		lowercase_(   self			:    str				,   lowerCamelCase			:    Optional[torch.Generator] = None       ):



   """simple docstring"""





   __A		:     Tuple												=		randn_tensor(
       self.mean.shape				,   generator=__snake_case				,   device=self.parameters.device				,   dtype=self.parameters.dtype       )
   __A		:     str												=		self.mean + self.std * sample
   return x
 def 		lowercase_(   self			:    List[str]				,   lowerCamelCase			:    Optional[Any]=None       ):



   """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[str]				,   lowerCamelCase			:    Union[str, Any]				,   lowerCamelCase			:    Optional[Any]=[1, 2, 3]       ):



   """simple docstring"""





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






 def 		lowercase_(   self			:    str       ):



   """simple docstring"""





   return self.mean





'''simple docstring'''
def        A_     (						__SCREAMING_SNAKE_CASE						:  list			,     __SCREAMING_SNAKE_CASE						:  list			,     __SCREAMING_SNAKE_CASE						:  int			,     __SCREAMING_SNAKE_CASE						:  int			,     __SCREAMING_SNAKE_CASE						:  int )				->       int:





  """simple docstring"""

  if index == number_of_items:
    return 0
  __A		:     List[Any]												=		0
  __A		:     Union[str, Any]												=		0
  __A		:     List[Any]												=		knapsack(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     index + 1 )
  if weights[index] <= max_weight:
    __A		:     Tuple												=		values[index] + knapsack(
        __SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     max_weight - weights[index]			,     index + 1 )
  return max(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE )


if __name__ == "__main__":
     import doctest

     doctest.testmod()



