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

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block


@dataclass
class   __lowerCAmelCase		(		lowerCAmelCase_ ):



      """simple docstring"""







      A__		:     torch.FloatTensor






class   __lowerCAmelCase		(		lowerCAmelCase_	,	lowerCAmelCase_ ):



      """simple docstring"""







      @register_to_config
      def __init__( self						:    Any			,    _snake_case						:    Optional[Union[int, Tuple[int, int]]] = None			,    _snake_case						:    int = 3			,    _snake_case						:    int = 3			,    _snake_case						:    bool = False			,    _snake_case						:    str = "positional"			,    _snake_case						:    int = 0			,    _snake_case						:    bool = True			,    _snake_case						:    Tuple[str] = ("DownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D", "AttnDownBlock2D")			,    _snake_case						:    Tuple[str] = ("AttnUpBlock2D", "AttnUpBlock2D", "AttnUpBlock2D", "UpBlock2D")			,    _snake_case						:    Tuple[int] = (224, 448, 672, 896)			,    _snake_case						:    int = 2			,    _snake_case						:    float = 1			,    _snake_case						:    int = 1			,    _snake_case						:    str = "conv"			,    _snake_case						:    str = "conv"			,    _snake_case						:    str = "silu"			,    _snake_case						:    Optional[int] = 8			,    _snake_case						:    int = 32			,    _snake_case						:    float = 1E-5			,    _snake_case						:    str = "default"			,    _snake_case						:    bool = True			,    _snake_case						:    Optional[str] = None			,    _snake_case						:    Optional[int] = None			,    ):
          super().__init__()

          __lowercase						:     Dict		      =    sample_size
          __lowercase						:     List[Any]		      =    block_out_channels[0] * 4

          # Check inputs
          if len(_snake_case			) != len(_snake_case			):
              raise ValueError(
                  F'Must provide the same number of `down_block_types` as `up_block_types`. `down_block_types`: {down_block_types}. `up_block_types`: {up_block_types}.'			)

          if len(_snake_case			) != len(_snake_case			):
              raise ValueError(
                  F'Must provide the same number of `block_out_channels` as `down_block_types`. `block_out_channels`: {block_out_channels}. `down_block_types`: {down_block_types}.'			)

          # input
          __lowercase						:     Optional[Any]		      =    nn.Convad(_snake_case			,    block_out_channels[0]			,    kernel_size=3			,    padding=(1, 1)			)

          # time
          if time_embedding_type == "fourier":
              __lowercase						:     List[str]		      =    GaussianFourierProjection(embedding_size=block_out_channels[0]			,    scale=16			)
              __lowercase						:     str		      =    2 * block_out_channels[0]
          elif time_embedding_type == "positional":
              __lowercase						:     Dict		      =    Timesteps(block_out_channels[0]			,    _snake_case			,    _snake_case			)
              __lowercase						:     Union[str, Any]		      =    block_out_channels[0]

          __lowercase						:     Optional[int]		      =    TimestepEmbedding(_snake_case			,    _snake_case			)

          # class embedding
          if class_embed_type is None and num_class_embeds is not None:
              __lowercase						:     Any		      =    nn.Embedding(_snake_case			,    _snake_case			)
          elif class_embed_type == "timestep":
              __lowercase						:     int		      =    TimestepEmbedding(_snake_case			,    _snake_case			)
          elif class_embed_type == "identity":
              __lowercase						:     Union[str, Any]		      =    nn.Identity(_snake_case			,    _snake_case			)
          else:
              __lowercase						:     str		      =    None

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

          # down
          __lowercase						:     Tuple		      =    block_out_channels[0]
          for i, down_block_type in enumerate(_snake_case			):
              __lowercase						:     List[str]		      =    output_channel
              __lowercase						:     List[str]		      =    block_out_channels[i]
              __lowercase						:     Any		      =    i == len(_snake_case			) - 1

              __lowercase						:     str		      =    get_down_block(
                  _snake_case			,    num_layers=_snake_case			,    in_channels=_snake_case			,    out_channels=_snake_case			,    temb_channels=_snake_case			,    add_downsample=not is_final_block			,    resnet_eps=_snake_case			,    resnet_act_fn=_snake_case			,    resnet_groups=_snake_case			,    attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel			,    downsample_padding=_snake_case			,    resnet_time_scale_shift=_snake_case			,    downsample_type=_snake_case			,    )
              self.down_blocks.append(_snake_case			)

          # mid
          __lowercase						:     List[Any]		      =    UNetMidBlockaD(
              in_channels=block_out_channels[-1]			,    temb_channels=_snake_case			,    resnet_eps=_snake_case			,    resnet_act_fn=_snake_case			,    output_scale_factor=_snake_case			,    resnet_time_scale_shift=_snake_case			,    attention_head_dim=attention_head_dim if attention_head_dim is not None else block_out_channels[-1]			,    resnet_groups=_snake_case			,    add_attention=_snake_case			,    )

          # up
          __lowercase						:     List[str]		      =    list(reversed(_snake_case			)			)
          __lowercase						:     str		      =    reversed_block_out_channels[0]
          for i, up_block_type in enumerate(_snake_case			):
              __lowercase						:     List[str]		      =    output_channel
              __lowercase						:     Tuple		      =    reversed_block_out_channels[i]
              __lowercase						:     Union[str, Any]		      =    reversed_block_out_channels[min(i + 1			,    len(_snake_case			) - 1			)]

              __lowercase						:     List[Any]		      =    i == len(_snake_case			) - 1

              __lowercase						:     Any		      =    get_up_block(
                  _snake_case			,    num_layers=layers_per_block + 1			,    in_channels=_snake_case			,    out_channels=_snake_case			,    prev_output_channel=_snake_case			,    temb_channels=_snake_case			,    add_upsample=not is_final_block			,    resnet_eps=_snake_case			,    resnet_act_fn=_snake_case			,    resnet_groups=_snake_case			,    attention_head_dim=attention_head_dim if attention_head_dim is not None else output_channel			,    resnet_time_scale_shift=_snake_case			,    upsample_type=_snake_case			,    )
              self.up_blocks.append(_snake_case			)
              __lowercase						:     str		      =    output_channel

          # out
          __lowercase						:     Tuple		      =    norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4			,    32			)
          __lowercase						:     int		      =    nn.GroupNorm(num_channels=block_out_channels[0]			,    num_groups=_snake_case			,    eps=_snake_case			)
          __lowercase						:     Union[str, Any]		      =    nn.SiLU()
          __lowercase						:     Dict		      =    nn.Convad(block_out_channels[0]			,    _snake_case			,    kernel_size=3			,    padding=1			)




      def 						snake_case_     ( self						:    Any			,    _snake_case						:    torch.FloatTensor			,    _snake_case						:    Union[torch.Tensor, float, int]			,    _snake_case						:    Optional[torch.Tensor] = None			,    _snake_case						:    bool = True			,    ):
          # 0. center input if necessary
          if self.config.center_input_sample:
              __lowercase						:     Optional[Any]		      =    2 * sample - 1.0

          # 1. time
          __lowercase						:     int		      =    timestep
          if not torch.is_tensor(_snake_case			):
              __lowercase						:     Any		      =    torch.tensor([timesteps]			,    dtype=torch.long			,    device=sample.device			)
          elif torch.is_tensor(_snake_case			) and len(timesteps.shape			) == 0:
              __lowercase						:     Tuple		      =    timesteps[None].to(sample.device			)

          # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
          __lowercase						:     int		      =    timesteps * torch.ones(sample.shape[0]			,    dtype=timesteps.dtype			,    device=timesteps.device			)

          __lowercase						:     Optional[Any]		      =    self.time_proj(_snake_case			)

          # timesteps does not contain any weights and will always return f32 tensors
          # but time_embedding might actually be running in fp16. so we need to cast here.
          # there might be better ways to encapsulate this.
          __lowercase						:     Optional[Any]		      =    t_emb.to(dtype=self.dtype			)
          __lowercase						:     Optional[Any]		      =    self.time_embedding(_snake_case			)

          if self.class_embedding is not None:
              if class_labels is None:
                  raise ValueError('''class_labels should be provided when doing class conditioning'''			)

              if self.config.class_embed_type == "timestep":
                  __lowercase						:     List[str]		      =    self.time_proj(_snake_case			)

              __lowercase						:     int		      =    self.class_embedding(_snake_case			).to(dtype=self.dtype			)
              __lowercase						:     Tuple		      =    emb + class_emb

          # 2. pre-process
          __lowercase						:     List[Any]		      =    sample
          __lowercase						:     List[Any]		      =    self.conv_in(_snake_case			)

          # 3. down
          __lowercase						:     Optional[int]		      =    (sample,)
          for downsample_block in self.down_blocks:
              if hasattr(_snake_case			,    '''skip_conv'''			):
                  __lowercase      , __lowercase      , __lowercase						:     str		      =    downsample_block(
                      hidden_states=_snake_case			,    temb=_snake_case			,    skip_sample=_snake_case			)
              else:
                  __lowercase      , __lowercase						:     Dict		      =    downsample_block(hidden_states=_snake_case			,    temb=_snake_case			)

              down_block_res_samples += res_samples

          # 4. mid
          __lowercase						:     Optional[int]		      =    self.mid_block(_snake_case			,    _snake_case			)

          # 5. up
          __lowercase						:     Optional[int]		      =    None
          for upsample_block in self.up_blocks:
              __lowercase						:     Dict		      =    down_block_res_samples[-len(upsample_block.resnets			) :]
              __lowercase						:     Tuple		      =    down_block_res_samples[: -len(upsample_block.resnets			)]

              if hasattr(_snake_case			,    '''skip_conv'''			):
                  __lowercase      , __lowercase						:     str		      =    upsample_block(_snake_case			,    _snake_case			,    _snake_case			,    _snake_case			)
              else:
                  __lowercase						:     List[Any]		      =    upsample_block(_snake_case			,    _snake_case			,    _snake_case			)

        # 6. post-process
          __lowercase						:     str		      =    self.conv_norm_out(_snake_case			)
          __lowercase						:     List[Any]		      =    self.conv_act(_snake_case			)
          __lowercase						:     Optional[Any]		      =    self.conv_out(_snake_case			)

          if skip_sample is not None:
              sample += skip_sample

          if self.config.time_embedding_type == "fourier":
              __lowercase						:     Dict		      =    timesteps.reshape((sample.shape[0], *([1] * len(sample.shape[1:]			)))			)
              __lowercase						:     Optional[int]		      =    sample / timesteps

          if not return_dict:
              return (sample,)

          return UNetaDOutput(sample=_snake_case			)




import os
from typing import List, Union

import tensorflow as tf
from tensorflow_text import BertTokenizer as BertTokenizerLayer
from tensorflow_text import FastBertTokenizer, ShrinkLongestTrimmer, case_fold_utfa, combine_segments, pad_model_inputs

from .tokenization_bert import BertTokenizer


class   __lowerCAmelCase		(		tf.keras.layers.Layer ):



      """simple docstring"""







      def __init__( self						:    int			,    _snake_case						:    List			,    _snake_case						:    bool			,    _snake_case						:    int = None			,    _snake_case						:    int = None			,    _snake_case						:    int = None			,    _snake_case						:    str = "longest"			,    _snake_case						:    bool = True			,    _snake_case						:    int = 512			,    _snake_case						:    int = None			,    _snake_case						:    bool = True			,    _snake_case						:    bool = True			,    _snake_case						:    bool = True			,    **_snake_case						:    int			,    ):
          super().__init__()
          if use_fast_bert_tokenizer:
              __lowercase						:     Optional[int]		      =    FastBertTokenizer(
                  _snake_case			,    token_out_type=tf.intaa			,    lower_case_nfd_strip_accents=_snake_case			,    **_snake_case			)
          else:
              __lowercase						:     List[str]		      =    tf.lookup.StaticVocabularyTable(
                  tf.lookup.KeyValueTensorInitializer(
                      keys=_snake_case			,    key_dtype=tf.string			,    values=tf.range(tf.size(_snake_case			,    out_type=tf.intaa			)			,    dtype=tf.intaa			)			,    value_dtype=tf.intaa			,    )			,    num_oov_buckets=1			,    )
              __lowercase						:     List[Any]		      =    BertTokenizerLayer(
                  _snake_case			,    token_out_type=tf.intaa			,    lower_case=_snake_case			,    **_snake_case			)

          __lowercase						:     Dict		      =    vocab_list
          __lowercase						:     str		      =    do_lower_case
          __lowercase						:     Optional[int]		      =    cls_token_id or vocab_list.index('''[CLS]'''			)
          __lowercase						:     Dict		      =    sep_token_id or vocab_list.index('''[SEP]'''			)
          __lowercase						:     List[str]		      =    pad_token_id or vocab_list.index('''[PAD]'''			)
          __lowercase						:     str		      =    ShrinkLongestTrimmer(max_length - 3			,    axis=1			)  # Allow room for special tokens
          __lowercase						:     Optional[Any]		      =    max_length
          __lowercase						:     Tuple		      =    padding
          __lowercase						:     str		      =    truncation
          __lowercase						:     str		      =    pad_to_multiple_of
          __lowercase						:     int		      =    return_token_type_ids
          __lowercase						:     List[str]		      =    return_attention_mask




      @classmethod
      def 						snake_case_     ( cls						:    Optional[Any]			,    _snake_case						:    "PreTrainedTokenizerBase"			,    **_snake_case						:    Dict			):  # noqa: F821
          __lowercase						:     Tuple		      =    kwargs.pop('''do_lower_case'''			,    _snake_case			)
          __lowercase						:     Union[str, Any]		      =    tokenizer.do_lower_case if do_lower_case is None else do_lower_case
          __lowercase						:     Optional[Any]		      =    kwargs.pop('''cls_token_id'''			,    _snake_case			)
          __lowercase						:     str		      =    tokenizer.cls_token_id if cls_token_id is None else cls_token_id
          __lowercase						:     Tuple		      =    kwargs.pop('''sep_token_id'''			,    _snake_case			)
          __lowercase						:     Any		      =    tokenizer.sep_token_id if sep_token_id is None else sep_token_id
          __lowercase						:     Dict		      =    kwargs.pop('''pad_token_id'''			,    _snake_case			)
          __lowercase						:     List[Any]		      =    tokenizer.pad_token_id if pad_token_id is None else pad_token_id

          __lowercase						:     Optional[int]		      =    tokenizer.get_vocab()
          __lowercase						:     Tuple		      =    sorted([(wordpiece, idx) for wordpiece, idx in vocab.items()]			,    key=lambda _snake_case			: x[1]			)
          __lowercase						:     Optional[int]		      =    [entry[0] for entry in vocab]
          return cls(
              vocab_list=_snake_case			,    do_lower_case=_snake_case			,    cls_token_id=_snake_case			,    sep_token_id=_snake_case			,    pad_token_id=_snake_case			,    **_snake_case			,    )




      @classmethod
      def 						snake_case_     ( cls						:    Any			,    _snake_case						:    Union[str, os.PathLike]			,    *_snake_case						:    List[str]			,    **_snake_case						:    Tuple			):
          try:
              __lowercase						:     List[Any]		      =    BertTokenizer.from_pretrained(_snake_case			,    *_snake_case			,    **_snake_case			)
          except:  # noqa: E722
              from .tokenization_bert_fast import BertTokenizerFast

              __lowercase						:     List[str]		      =    BertTokenizerFast.from_pretrained(_snake_case			,    *_snake_case			,    **_snake_case			)
          return cls.from_tokenizer(_snake_case			,    **_snake_case			)




      def 						snake_case_     ( self						:    Optional[int]			,    _snake_case						:    Union[str, Any]			):
          if self.do_lower_case:
              __lowercase						:     int		      =    case_fold_utfa(_snake_case			)
          __lowercase						:     Optional[int]		      =    self.tf_tokenizer.tokenize(_snake_case			)
          return tokens.merge_dims(1			,    -1			)




      def 						snake_case_     ( self						:    Tuple			,    _snake_case						:    Optional[int]			,    _snake_case						:    Any=None			,    _snake_case						:    List[Any]=None			,    _snake_case						:    Any=None			,    _snake_case						:    Dict=None			,    _snake_case						:    List[Any]=None			,    _snake_case						:    str=None			,    _snake_case						:    List[Any]=None			,    ):
          if padding is None:
              __lowercase						:     List[Any]		      =    self.padding
          if padding not in ("longest", "max_length"):
              raise ValueError('''Padding must be either \'longest\' or \'max_length\'!'''			)
          if max_length is not None and text_pair is not None:
              # Because we have to instantiate a Trimmer to do it properly
              raise ValueError('''max_length cannot be overridden at call time when truncating paired texts!'''			)
          if max_length is None:
              __lowercase						:     Dict		      =    self.max_length
          if truncation is None:
              __lowercase						:     int		      =    self.truncation
          if pad_to_multiple_of is None:
              __lowercase						:     Tuple		      =    self.pad_to_multiple_of
          if return_token_type_ids is None:
              __lowercase						:     int		      =    self.return_token_type_ids
          if return_attention_mask is None:
              __lowercase						:     Optional[int]		      =    self.return_attention_mask
          if not isinstance(_snake_case			,    tf.Tensor			):
              __lowercase						:     Dict		      =    tf.convert_to_tensor(_snake_case			)
          if text_pair is not None and not isinstance(_snake_case			,    tf.Tensor			):
              __lowercase						:     Optional[Any]		      =    tf.convert_to_tensor(_snake_case			)
          if text_pair is not None:
              if text.shape.rank > 1:
                  raise ValueError('''text argument should not be multidimensional when a text pair is supplied!'''			)
              if text_pair.shape.rank > 1:
                  raise ValueError('''text_pair should not be multidimensional!'''			)
          if text.shape.rank == 2:
              __lowercase      , __lowercase						:     int		      =    text[:, 0], text[:, 1]
          __lowercase						:     int		      =    self.unpaired_tokenize(_snake_case			)
          if text_pair is None:  # Unpaired text
              if truncation:
                  __lowercase						:     int		      =    text[:, : max_length - 2]  # Allow room for special tokens
              __lowercase      , __lowercase						:     Dict		      =    combine_segments(
                  (text,)			,    start_of_sequence_id=self.cls_token_id			,    end_of_segment_id=self.sep_token_id			)
          else:  # Paired text
              __lowercase						:     int		      =    self.unpaired_tokenize(_snake_case			)
              if truncation:
                  __lowercase      , __lowercase						:     Union[str, Any]		      =    self.paired_trimmer.trim([text, text_pair]			)
              __lowercase      , __lowercase						:     int		      =    combine_segments(
                  (text, text_pair)			,    start_of_sequence_id=self.cls_token_id			,    end_of_segment_id=self.sep_token_id			)
          if padding == "longest":
              __lowercase						:     List[Any]		      =    input_ids.bounding_shape(axis=1			)
              if pad_to_multiple_of is not None:
                  # No ceiling division in tensorflow, so we negate floordiv instead
                  __lowercase						:     List[Any]		      =    pad_to_multiple_of * (-tf.math.floordiv(-pad_length			,    _snake_case			))
          else:
              __lowercase						:     str		      =    max_length

          __lowercase      , __lowercase						:     Optional[int]		      =    pad_model_inputs(_snake_case			,    max_seq_length=_snake_case			,    pad_value=self.pad_token_id			)
          __lowercase						:     Dict		      =    {'''input_ids''': input_ids}
          if return_attention_mask:
              __lowercase						:     Any		      =    attention_mask
          if return_token_type_ids:
              __lowercase      , __lowercase						:     List[str]		      =    pad_model_inputs(
                  _snake_case			,    max_seq_length=_snake_case			,    pad_value=self.pad_token_id			)
              __lowercase						:     Union[str, Any]		      =    token_type_ids
          return output




      def 						snake_case_     ( self						:    str			):
          return {
              "vocab_list": self.vocab_list,
              "do_lower_case": self.do_lower_case,
              "cls_token_id": self.cls_token_id,
              "sep_token_id": self.sep_token_id,
              "pad_token_id": self.pad_token_id,
          }


