
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     UpperCAmelCase						(		A_					):
   A__   :						torch.FloatTensor






class     UpperCAmelCase						(		nn.Module					):



   def __init__(self						:   Dict       ,		snake_case__						:   Union[str, Any]=3       ,		snake_case__						:   Dict=3       ,		snake_case__						:   str=("DownEncoderBlock2D",)       ,		snake_case__						:   List[str]=(64,)       ,		snake_case__						:   str=2       ,		snake_case__						:   Dict=32       ,		snake_case__						:   Optional[int]="silu"       ,		snake_case__						:   Any=True       ,		)						->      Union[str, Any]:


          '''simple docstring'''


          super().__init__()
          snake_case			:  Dict     							= layers_per_block

          snake_case			:  Any     							= torch.nn.Convad(
              snake_case__       ,		block_out_channels[0]       ,		kernel_size=3       ,		stride=1       ,		padding=1       ,		)

          snake_case			:  Union[str, Any]     							= None
          snake_case			:  List[str]     							= nn.ModuleList([]					)

          # down
          snake_case			:  Optional[int]     							= block_out_channels[0]
          for i, down_block_type in enumerate(snake_case__					):
                 snake_case			:  Optional[Any]     							= output_channel
                 snake_case			:  int     							= block_out_channels[i]
                 snake_case			:  Union[str, Any]     							= i == len(snake_case__					) - 1

                 snake_case			:  Optional[Any]     							= 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
          snake_case			:  Optional[int]     							= 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
          snake_case			:  Union[str, Any]     							= nn.GroupNorm(num_channels=block_out_channels[-1]       ,		num_groups=snake_case__       ,		eps=1e-6					)
          snake_case			:  Union[str, Any]     							= nn.SiLU()

          snake_case			:  List[Any]     							= 2 * out_channels if double_z else out_channels
          snake_case			:  Any     							= nn.Convad(block_out_channels[-1]       ,		snake_case__       ,		3       ,		padding=1					)

          snake_case			:  Tuple     							= False



   def 							_SCREAMING_SNAKE_CASE  (self						:   Tuple       ,		snake_case__						:   Optional[Any]					)						->      List[Any]:


          '''simple docstring'''


          snake_case			:  int     							= x
          snake_case			:  Optional[Any]     							= self.conv_in(snake_case__					)

          if self.training and self.gradient_checkpointing:

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

                        return custom_forward

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

          else:
                 # down
                 for down_block in self.down_blocks:
                        snake_case			:  int     							= down_block(snake_case__					)

                 # middle
                 snake_case			:  int     							= self.mid_block(snake_case__					)

          # post-process
          snake_case			:  Tuple     							= self.conv_norm_out(snake_case__					)
          snake_case			:  Optional[Any]     							= self.conv_act(snake_case__					)
          snake_case			:  Tuple     							= self.conv_out(snake_case__					)

          return sample






class     UpperCAmelCase						(		nn.Module					):



   def __init__(self						:   List[Any]       ,		snake_case__						:   List[Any]=3       ,		snake_case__						:   Union[str, Any]=3       ,		snake_case__						:   List[Any]=("UpDecoderBlock2D",)       ,		snake_case__						:   int=(64,)       ,		snake_case__						:   List[Any]=2       ,		snake_case__						:   Tuple=32       ,		snake_case__						:   str="silu"       ,		snake_case__						:   Any="group"       ,		)						->      Any:


          '''simple docstring'''


          super().__init__()
          snake_case			:  Dict     							= layers_per_block

          snake_case			:  Union[str, Any]     							= nn.Convad(
              snake_case__       ,		block_out_channels[-1]       ,		kernel_size=3       ,		stride=1       ,		padding=1       ,		)

          snake_case			:  List[Any]     							= None
          snake_case			:  Dict     							= nn.ModuleList([]					)

          snake_case			:  Tuple     							= in_channels if norm_type == "spatial" else None

          # mid
          snake_case			:  Optional[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" if norm_type == "group" else norm_type       ,		attention_head_dim=block_out_channels[-1]       ,		resnet_groups=snake_case__       ,		temb_channels=snake_case__       ,		)

          # up
          snake_case			:  Any     							= list(reversed(snake_case__					)					)
          snake_case			:  Optional[Any]     							= reversed_block_out_channels[0]
          for i, up_block_type in enumerate(snake_case__					):
                 snake_case			:  Dict     							= output_channel
                 snake_case			:  str     							= reversed_block_out_channels[i]

                 snake_case			:  Any     							= i == len(snake_case__					) - 1

                 snake_case			:  int     							= 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__					)
                 snake_case			:  Union[str, Any]     							= output_channel

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

          snake_case			:  List[str]     							= False



   def 							_SCREAMING_SNAKE_CASE  (self						:   List[Any]       ,		snake_case__						:   List[Any]       ,		snake_case__						:   Dict=None					)						->      List[Any]:


          '''simple docstring'''


          snake_case			:  List[str]     							= z
          snake_case			:  Union[str, Any]     							= self.conv_in(snake_case__					)

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

                 def create_custom_forward(snake_case__						:   str					):
                        def custom_forward(*snake_case__						:   Optional[Any]					):
                               return module(*snake_case__					)

                        return custom_forward

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

                        # up
                        for up_block in self.up_blocks:
                               snake_case			:  Optional[Any]     							= torch.utils.checkpoint.checkpoint(
                                   create_custom_forward(snake_case__					)       ,		snake_case__       ,		snake_case__       ,		use_reentrant=snake_case__					)
                 else:
                        # middle
                        snake_case			:  List[Any]     							= torch.utils.checkpoint.checkpoint(
                            create_custom_forward(self.mid_block					)       ,		snake_case__       ,		snake_case__					)
                        snake_case			:  Optional[Any]     							= sample.to(snake_case__					)

                        # up
                        for up_block in self.up_blocks:
                               snake_case			:  Tuple     							= torch.utils.checkpoint.checkpoint(create_custom_forward(snake_case__					)       ,		snake_case__       ,		snake_case__					)
          else:
                 # middle
                 snake_case			:  Tuple     							= self.mid_block(snake_case__       ,		snake_case__					)
                 snake_case			:  int     							= sample.to(snake_case__					)

                 # up
                 for up_block in self.up_blocks:
                        snake_case			:  List[str]     							= up_block(snake_case__       ,		snake_case__					)

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

          return sample






class     UpperCAmelCase						(		nn.Module					):



   def __init__(self						:   Union[str, Any]       ,		snake_case__						:   List[Any]       ,		snake_case__						:   Union[str, Any]       ,		snake_case__						:   Optional[Any]       ,		snake_case__						:   List[str]=None       ,		snake_case__						:   Tuple="random"       ,		snake_case__						:   Dict=False       ,		snake_case__						:   Any=True					)						->      Union[str, Any]:


          '''simple docstring'''


          super().__init__()
          snake_case			:  Optional[int]     							= n_e
          snake_case			:  Union[str, Any]     							= vq_embed_dim
          snake_case			:  Dict     							= beta
          snake_case			:  Union[str, Any]     							= legacy

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

          snake_case			:  Optional[Any]     							= sane_index_shape



   def 							_SCREAMING_SNAKE_CASE  (self						:   List[str]       ,		snake_case__						:   str					)						->      List[str]:


          '''simple docstring'''


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



   def 							_SCREAMING_SNAKE_CASE  (self						:   Tuple       ,		snake_case__						:   Optional[Any]					)						->      Optional[Any]:


          '''simple docstring'''


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



   def 							_SCREAMING_SNAKE_CASE  (self						:   str       ,		snake_case__						:   Tuple					)						->      Optional[int]:


          '''simple docstring'''


          snake_case			:  str     							= z.permute(0       ,		2       ,		3       ,		1					).contiguous()
          snake_case			:  Union[str, Any]     							= 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(snake_case__       ,		self.embedding.weight					)       ,		dim=1					)

          snake_case			:  Any     							= self.embedding(snake_case__					).view(z.shape					)
          snake_case			:  int     							= None
          snake_case			:  Dict     							= None

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

          # preserve gradients
          snake_case			:  Optional[int]     							= z + (z_q - z).detach()

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

          if self.remap is not None:
                 snake_case			:  Tuple     							= min_encoding_indices.reshape(z.shape[0]       ,		-1					)  # add batch axis
                 snake_case			:  Tuple     							= self.remap_to_used(snake_case__					)
                 snake_case			:  Dict     							= 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 							_SCREAMING_SNAKE_CASE  (self						:   int       ,		snake_case__						:   Tuple       ,		snake_case__						:   Any					)						->      Optional[Any]:


          '''simple docstring'''


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

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

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

          return z_q






class     UpperCAmelCase						(		A_					):



   def __init__(self						:   Tuple       ,		snake_case__						:   Union[str, Any]       ,		snake_case__						:   Dict=False					)						->      str:


          '''simple docstring'''


          snake_case			:  Optional[int]     							= parameters
          snake_case							,   snake_case			:  str     							= torch.chunk(snake_case__       ,		2       ,		dim=1					)
          snake_case			:  Tuple     							= torch.clamp(self.logvar       ,		-30.0       ,		20.0					)
          snake_case			:  Optional[Any]     							= deterministic
          snake_case			:  Any     							= torch.exp(0.5 * self.logvar					)
          snake_case			:  Optional[int]     							= torch.exp(self.logvar					)
          if self.deterministic:
                 snake_case			:  Dict     							= torch.zeros_like(
                     self.mean       ,		device=self.parameters.device       ,		dtype=self.parameters.dtype					)



   def 							_SCREAMING_SNAKE_CASE  (self						:   Dict       ,		snake_case__						:   Optional[torch.Generator] = None					)						->      torch.FloatTensor:


          '''simple docstring'''


          snake_case			:  Tuple     							= randn_tensor(
              self.mean.shape       ,		generator=snake_case__       ,		device=self.parameters.device       ,		dtype=self.parameters.dtype					)
          snake_case			:  List[str]     							= self.mean + self.std * sample
          return x



   def 							_SCREAMING_SNAKE_CASE  (self						:   Optional[Any]       ,		snake_case__						:   Union[str, Any]=None					)						->      List[Any]:


          '''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 							_SCREAMING_SNAKE_CASE  (self						:   int       ,		snake_case__						:   Any       ,		snake_case__						:   Optional[Any]=[1, 2, 3]					)						->      Optional[int]:


          '''simple docstring'''


          if self.deterministic:
                 return torch.Tensor([0.0]					)
          snake_case			:  Union[str, Any]     							= 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 							_SCREAMING_SNAKE_CASE  (self						:   int					)						->      str:


          '''simple docstring'''


          return self.mean







from math import factorial


def      UpperCamelCase   (       __lowerCamelCase	:					int							,   __lowerCamelCase	:					int							,   __lowerCamelCase	:					float       ):
       if successes > trials:
              raise ValueError("successes must be lower or equal to trials"       )
       if trials < 0 or successes < 0:
              raise ValueError("the function is defined for non-negative integers"       )
       if not isinstance(__lowerCamelCase							,   __lowerCamelCase       ) or not isinstance(__lowerCamelCase							,   __lowerCamelCase       ):
              raise ValueError("the function is defined for non-negative integers"       )
       if not 0 < prob < 1:
              raise ValueError("prob has to be in range of 1 - 0"       )
       snake_case			:  Tuple     							= (prob**successes) * ((1 - prob) ** (trials - successes))
       # Calculate the binomial coefficient: n! / k!(n-k)!
       snake_case			:  str     							= float(factorial(__lowerCamelCase       )       )
       coefficient /= factorial(__lowerCamelCase       ) * factorial(trials - successes       )
       return probability * coefficient


if __name__ == "__main__":
   from doctest import testmod

   testmod()
   print("""Probability of 2 successes out of 4 trails""")
   print("""with probability of 0.75 is:""", end=""" """)
   print(binomial_distribution(2, 4, 0.75))





