



from dataclasses import dataclass
from typing import Dict, Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, apply_forward_hook
from .attention_processor import AttentionProcessor, AttnProcessor
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, DiagonalGaussianDistribution, Encoder







@dataclass
class 	UpperCAmelCase	(		UpperCamelCase__):


 '''simple docstring'''




 __magic_name__					: Dict  =      42







class 	UpperCAmelCase	(		UpperCamelCase__					,	UpperCamelCase__):


 '''simple docstring'''




 __magic_name__					: Any  =      True




 @register_to_config
 def __init__(   self							,  lowerCAmelCase_ = 3							,  lowerCAmelCase_ = 3							,  lowerCAmelCase_ = ("DownEncoderBlock2D",)							,  lowerCAmelCase_ = ("UpDecoderBlock2D",)							,  lowerCAmelCase_ = (6_4,)							,  lowerCAmelCase_ = 1							,  lowerCAmelCase_ = "silu"							,  lowerCAmelCase_ = 4							,  lowerCAmelCase_ = 3_2							,  lowerCAmelCase_ = 3_2							,  lowerCAmelCase_ = 0.1_8_2_1_5							,  ) ->      Any:



      """simple docstring"""







      super().__init__()

      # pass init params to Encoder
      a_       		=Encoder(
          in_channels=_a							,  out_channels=_a							,  down_block_types=_a							,  block_out_channels=_a							,  layers_per_block=_a							,  act_fn=_a							,  norm_num_groups=_a							,  double_z=_a							,  )

      # pass init params to Decoder
      a_       		=Decoder(
          in_channels=_a							,  out_channels=_a							,  up_block_types=_a							,  block_out_channels=_a							,  layers_per_block=_a							,  norm_num_groups=_a							,  act_fn=_a							,  )

      a_       		=nn.Convad(2 * latent_channels							,  2 * latent_channels							,  1)
      a_       		=nn.Convad(_a							,  _a							,  1)

      a_       		=False
      a_       		=False

      # only relevant if vae tiling is enabled
      a_       		=self.config.sample_size
      a_       		=(
          self.config.sample_size[0]
          if isinstance(self.config.sample_size							,  (list, tuple))
          else self.config.sample_size
      )
      a_       		=int(sample_size / (2 ** (len(self.config.block_out_channels) - 1)))
      a_       		=0.2_5




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_=False) ->      List[str]:



      """simple docstring"""







      if isinstance(_a							,  (Encoder, Decoder)):
           a_       		=value




 def       lowercase_  (   self							,  lowerCAmelCase_ = True) ->      Any:



      """simple docstring"""







      a_       		=use_tiling




 def       lowercase_  (   self) ->      List[Any]:



      """simple docstring"""







      self.enable_tiling(_a)




 def       lowercase_  (   self) ->      Dict:



      """simple docstring"""







      a_       		=True




 def       lowercase_  (   self) ->      List[str]:



      """simple docstring"""







      a_       		=False




 @property
 # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
 def       lowercase_  (   self) ->      Dict[str, AttentionProcessor]:



      """simple docstring"""







      a_       		={}

      def fn_recursive_add_processors(lowerCAmelCase_							,  lowerCAmelCase_							,  lowerCAmelCase_):
           if hasattr(_a							,  "set_processor"):
                a_       		=module.processor

           for sub_name, child in module.named_children():
                fn_recursive_add_processors(f"""{name}.{sub_name}"""							,  _a							,  _a)

           return processors

      for name, module in self.named_children():
           fn_recursive_add_processors(_a							,  _a							,  _a)

      return processors




 def       lowercase_  (   self							,  lowerCAmelCase_) ->      Union[str, Any]:



      """simple docstring"""







      a_       		=len(self.attn_processors.keys())

      if isinstance(_a							,  _a) and len(_a) != count:
           raise ValueError(
               f"""A dict of processors was passed, but the number of processors {len(_a)} does not match the"""
               f""" number of attention layers: {count}. Please make sure to pass {count} processor classes.""")

      def fn_recursive_attn_processor(lowerCAmelCase_							,  lowerCAmelCase_							,  lowerCAmelCase_):
           if hasattr(_a							,  "set_processor"):
                if not isinstance(_a							,  _a):
                     module.set_processor(_a)
                else:
                     module.set_processor(processor.pop(f"""{name}.processor"""))

           for sub_name, child in module.named_children():
                fn_recursive_attn_processor(f"""{name}.{sub_name}"""							,  _a							,  _a)

      for name, module in self.named_children():
           fn_recursive_attn_processor(_a							,  _a							,  _a)




 def       lowercase_  (   self) ->      Union[str, Any]:



      """simple docstring"""







      self.set_attn_processor(AttnProcessor())




 @apply_forward_hook
 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = True) ->      AutoencoderKLOutput:



      """simple docstring"""







      if self.use_tiling and (x.shape[-1] > self.tile_sample_min_size or x.shape[-2] > self.tile_sample_min_size):
           return self.tiled_encode(_a							,  return_dict=_a)

      if self.use_slicing and x.shape[0] > 1:
           a_       		=[self.encoder(_a) for x_slice in x.split(1)]
           a_       		=torch.cat(_a)
      else:
           a_       		=self.encoder(_a)

      a_       		=self.quant_conv(_a)
      a_       		=DiagonalGaussianDistribution(_a)

      if not return_dict:
           return (posterior,)

      return AutoencoderKLOutput(latent_dist=_a)




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = True) ->      Union[DecoderOutput, torch.FloatTensor]:



      """simple docstring"""







      if self.use_tiling and (z.shape[-1] > self.tile_latent_min_size or z.shape[-2] > self.tile_latent_min_size):
           return self.tiled_decode(_a							,  return_dict=_a)

      a_       		=self.post_quant_conv(_a)
      a_       		=self.decoder(_a)

      if not return_dict:
           return (dec,)

      return DecoderOutput(sample=_a)




 @apply_forward_hook
 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = True) ->      Union[DecoderOutput, torch.FloatTensor]:



      """simple docstring"""







      if self.use_slicing and z.shape[0] > 1:
           a_       		=[self._decode(_a).sample for z_slice in z.split(1)]
           a_       		=torch.cat(_a)
      else:
           a_       		=self._decode(_a).sample

      if not return_dict:
           return (decoded,)

      return DecoderOutput(sample=_a)




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_							,  lowerCAmelCase_) ->      List[str]:



      """simple docstring"""







      a_       		=min(a.shape[2]							,  b.shape[2]							,  _a)
      for y in range(_a):
           a_       		=a[:, :, -blend_extent + y, :] * (1 - y / blend_extent) + b[:, :, y, :] * (y / blend_extent)
      return b




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_							,  lowerCAmelCase_) ->      Tuple:



      """simple docstring"""







      a_       		=min(a.shape[3]							,  b.shape[3]							,  _a)
      for x in range(_a):
           a_       		=a[:, :, :, -blend_extent + x] * (1 - x / blend_extent) + b[:, :, :, x] * (x / blend_extent)
      return b




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = True) ->      AutoencoderKLOutput:



      """simple docstring"""







      a_       		=int(self.tile_sample_min_size * (1 - self.tile_overlap_factor))
      a_       		=int(self.tile_latent_min_size * self.tile_overlap_factor)
      a_       		=self.tile_latent_min_size - blend_extent

      # Split the image into 512x512 tiles and encode them separately.
      a_       		=[]
      for i in range(0							,  x.shape[2]							,  _a):
           a_       		=[]
           for j in range(0							,  x.shape[3]							,  _a):
                a_       		=x[:, :, i : i + self.tile_sample_min_size, j : j + self.tile_sample_min_size]
                a_       		=self.encoder(_a)
                a_       		=self.quant_conv(_a)
                row.append(_a)
           rows.append(_a)
      a_       		=[]
      for i, row in enumerate(_a):
           a_       		=[]
           for j, tile in enumerate(_a):
                # blend the above tile and the left tile
                # to the current tile and add the current tile to the result row
                if i > 0:
                     a_       		=self.blend_v(rows[i - 1][j]							,  _a							,  _a)
                if j > 0:
                     a_       		=self.blend_h(row[j - 1]							,  _a							,  _a)
                result_row.append(tile[:, :, :row_limit, :row_limit])
           result_rows.append(torch.cat(_a							,  dim=3))

      a_       		=torch.cat(_a							,  dim=2)
      a_       		=DiagonalGaussianDistribution(_a)

      if not return_dict:
           return (posterior,)

      return AutoencoderKLOutput(latent_dist=_a)




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = True) ->      Union[DecoderOutput, torch.FloatTensor]:



      """simple docstring"""







      a_       		=int(self.tile_latent_min_size * (1 - self.tile_overlap_factor))
      a_       		=int(self.tile_sample_min_size * self.tile_overlap_factor)
      a_       		=self.tile_sample_min_size - blend_extent

      # Split z into overlapping 64x64 tiles and decode them separately.
      # The tiles have an overlap to avoid seams between tiles.
      a_       		=[]
      for i in range(0							,  z.shape[2]							,  _a):
           a_       		=[]
           for j in range(0							,  z.shape[3]							,  _a):
                a_       		=z[:, :, i : i + self.tile_latent_min_size, j : j + self.tile_latent_min_size]
                a_       		=self.post_quant_conv(_a)
                a_       		=self.decoder(_a)
                row.append(_a)
           rows.append(_a)
      a_       		=[]
      for i, row in enumerate(_a):
           a_       		=[]
           for j, tile in enumerate(_a):
                # blend the above tile and the left tile
                # to the current tile and add the current tile to the result row
                if i > 0:
                     a_       		=self.blend_v(rows[i - 1][j]							,  _a							,  _a)
                if j > 0:
                     a_       		=self.blend_h(row[j - 1]							,  _a							,  _a)
                result_row.append(tile[:, :, :row_limit, :row_limit])
           result_rows.append(torch.cat(_a							,  dim=3))

      a_       		=torch.cat(_a							,  dim=2)
      if not return_dict:
           return (dec,)

      return DecoderOutput(sample=_a)




 def       lowercase_  (   self							,  lowerCAmelCase_							,  lowerCAmelCase_ = False							,  lowerCAmelCase_ = True							,  lowerCAmelCase_ = None							,  ) ->      Union[DecoderOutput, torch.FloatTensor]:



      """simple docstring"""







      a_       		=sample
      a_       		=self.encode(_a).latent_dist
      if sample_posterior:
           a_       		=posterior.sample(generator=_a)
      else:
           a_       		=posterior.mode()
      a_       		=self.decode(_a).sample

      if not return_dict:
           return (dec,)

      return DecoderOutput(sample=_a)







'''simple docstring'''


from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowercase							       =			logging.get_logger(__name__)

lowercase							       =			{
    '''google/switch-base-8''': '''https://huggingface.co/google/switch-base-8/blob/main/config.json''',
}







class 	UpperCAmelCase	(		__a):


 '''simple docstring'''




 __magic_name__					: int  =      "switch_transformers"
 __magic_name__					: List[Any]  =      ["past_key_values"]
 __magic_name__					: Union[str, Any]  =      {"hidden_size": "d_model", "num_attention_heads": "num_heads", "num_hidden_layers": "num_layers"}




 def __init__(   self							,  lowerCAmelCase_=3_2_1_2_8							,  lowerCAmelCase_=7_6_8							,  lowerCAmelCase_=6_4							,  lowerCAmelCase_=2_0_4_8							,  lowerCAmelCase_=6_4							,  lowerCAmelCase_=1_2							,  lowerCAmelCase_=3							,  lowerCAmelCase_=1_2							,  lowerCAmelCase_=3							,  lowerCAmelCase_=1_2							,  lowerCAmelCase_=8							,  lowerCAmelCase_=False							,  lowerCAmelCase_=0.0_1							,  lowerCAmelCase_="float32"							,  lowerCAmelCase_=False							,  lowerCAmelCase_=3_2							,  lowerCAmelCase_=1_2_8							,  lowerCAmelCase_=0.1							,  lowerCAmelCase_=1e-6							,  lowerCAmelCase_=0.0_0_1							,  lowerCAmelCase_=0.0_0_1							,  lowerCAmelCase_=1.0							,  lowerCAmelCase_="relu"							,  lowerCAmelCase_=True							,  lowerCAmelCase_=False							,  lowerCAmelCase_=True							,  lowerCAmelCase_=0							,  lowerCAmelCase_=1							,  **lowerCAmelCase_							,  ) ->      Optional[int]:



      """simple docstring"""







      a_       		=vocab_size
      a_       		=d_model
      a_       		=d_kv
      a_       		=d_ff

      a_       		=num_sparse_encoder_layers

      a_       		=num_layers
      a_       		=(
          num_decoder_layers if num_decoder_layers is not None else self.num_layers
      )  # default = symmetry
      a_       		=num_sparse_decoder_layers

      # This tells us, each how many encoder layer we'll have to set a sparse layer.
      if self.num_sparse_encoder_layers > 0:
           a_       		=self.num_layers // self.num_sparse_encoder_layers
      else:
           a_       		=self.num_layers  # HACK: this will create 0 sparse layers

      # This tells us, each how many encoder layer we'll have to set a sparse layer.
      if self.num_sparse_decoder_layers > 0:
           a_       		=self.num_decoder_layers // self.num_sparse_decoder_layers
      else:
           a_       		=self.num_decoder_layers  # HACK: this will create 0 sparse layers

      a_       		=num_heads
      a_       		=num_experts
      a_       		=expert_capacity
      a_       		=router_bias
      a_       		=router_jitter_noise
      if router_dtype not in ["float32", "float16", "bfloat16"]:
           raise ValueError(f"""`router_dtype` must be one of 'float32', 'float16' or 'bfloat16', got {router_dtype}""")
      a_       		=router_dtype

      a_       		=router_ignore_padding_tokens
      a_       		=relative_attention_num_buckets
      a_       		=relative_attention_max_distance

      a_       		=dropout_rate
      a_       		=layer_norm_epsilon
      a_       		=initializer_factor
      a_       		=feed_forward_proj
      a_       		=use_cache
      a_       		=add_router_probs

      a_       		=router_z_loss_coef
      a_       		=router_aux_loss_coef

      a_       		=self.feed_forward_proj.split("-")
      a_       		=act_info[-1]
      a_       		=act_info[0] == "gated"

      if len(lowerCAmelCase_) > 1 and act_info[0] != "gated" or len(lowerCAmelCase_) > 2:
           raise ValueError(
               f"""`feed_forward_proj`: {feed_forward_proj} is not a valid activation function of the dense layer."""
               "Please make sure `feed_forward_proj` is of the format `gated-{ACT_FN}` or `{ACT_FN}`, e.g. "
               "'gated-gelu' or 'relu'")

      # for backwards compatibility
      if feed_forward_proj == "gated-gelu":
           a_       		="gelu_new"

      super().__init__(
          pad_token_id=lowerCAmelCase_							,  eos_token_id=lowerCAmelCase_							,  is_encoder_decoder=lowerCAmelCase_							,  **lowerCAmelCase_							,  )


