





import os
from shutil import copyfile
from typing import List, Optional, Tuple

from ...tokenization_utils import AddedToken
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
    from .tokenization_albert import AlbertTokenizer
else:
    _UpperCAmelCase				:       str	   =				None

_UpperCAmelCase				:       List[Any]	   =				logging.get_logger(__name__)
_UpperCAmelCase				:       List[str]	   =				{"""vocab_file""": """spiece.model""", """tokenizer_file""": """tokenizer.json"""}

_UpperCAmelCase				:       Optional[int]	   =				{
    """vocab_file""": {
        """albert-base-v1""": """https://huggingface.co/albert-base-v1/resolve/main/spiece.model""",
        """albert-large-v1""": """https://huggingface.co/albert-large-v1/resolve/main/spiece.model""",
        """albert-xlarge-v1""": """https://huggingface.co/albert-xlarge-v1/resolve/main/spiece.model""",
        """albert-xxlarge-v1""": """https://huggingface.co/albert-xxlarge-v1/resolve/main/spiece.model""",
        """albert-base-v2""": """https://huggingface.co/albert-base-v2/resolve/main/spiece.model""",
        """albert-large-v2""": """https://huggingface.co/albert-large-v2/resolve/main/spiece.model""",
        """albert-xlarge-v2""": """https://huggingface.co/albert-xlarge-v2/resolve/main/spiece.model""",
        """albert-xxlarge-v2""": """https://huggingface.co/albert-xxlarge-v2/resolve/main/spiece.model""",
    },
    """tokenizer_file""": {
        """albert-base-v1""": """https://huggingface.co/albert-base-v1/resolve/main/tokenizer.json""",
        """albert-large-v1""": """https://huggingface.co/albert-large-v1/resolve/main/tokenizer.json""",
        """albert-xlarge-v1""": """https://huggingface.co/albert-xlarge-v1/resolve/main/tokenizer.json""",
        """albert-xxlarge-v1""": """https://huggingface.co/albert-xxlarge-v1/resolve/main/tokenizer.json""",
        """albert-base-v2""": """https://huggingface.co/albert-base-v2/resolve/main/tokenizer.json""",
        """albert-large-v2""": """https://huggingface.co/albert-large-v2/resolve/main/tokenizer.json""",
        """albert-xlarge-v2""": """https://huggingface.co/albert-xlarge-v2/resolve/main/tokenizer.json""",
        """albert-xxlarge-v2""": """https://huggingface.co/albert-xxlarge-v2/resolve/main/tokenizer.json""",
    },
}

_UpperCAmelCase				:       Optional[Any]	   =				{
    """albert-base-v1""": 5_12,
    """albert-large-v1""": 5_12,
    """albert-xlarge-v1""": 5_12,
    """albert-xxlarge-v1""": 5_12,
    """albert-base-v2""": 5_12,
    """albert-large-v2""": 5_12,
    """albert-xlarge-v2""": 5_12,
    """albert-xxlarge-v2""": 5_12,
}

_UpperCAmelCase				:       Optional[int]	   =				"""▁"""

class    lowerCAmelCase							(					__UpperCamelCase			):
 UpperCAmelCase__													=		VOCAB_FILES_NAMES
 UpperCAmelCase__													=		PRETRAINED_VOCAB_FILES_MAP
 UpperCAmelCase__													=		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 UpperCAmelCase__													=		AlbertTokenizer
 def __init__(      self  :      Optional[Any]			,     UpperCAmelCase  :      Tuple=None			,     UpperCAmelCase  :      Optional[Any]=None			,     UpperCAmelCase  :      Optional[Any]=True			,     UpperCAmelCase  :      Optional[Any]=True			,     UpperCAmelCase  :      List[Any]=False			,     UpperCAmelCase  :      Tuple="[CLS]"			,     UpperCAmelCase  :      str="[SEP]"			,     UpperCAmelCase  :      Any="<unk>"			,     UpperCAmelCase  :      Any="[SEP]"			,     UpperCAmelCase  :      List[Any]="<pad>"			,     UpperCAmelCase  :      List[str]="[CLS]"			,     UpperCAmelCase  :      Optional[int]="[MASK]"			,     **UpperCAmelCase  :      int			,     )			->	Dict:
     # Mask token behave like a normal word, i.e. include the space before it and
     # is included in the raw text, there should be a match in a non-normalized sentence.
     lowerCamelCase__    :      Any					 =		(
         AddedToken(UpperCAmelCase			,     lstrip=UpperCAmelCase			,     rstrip=UpperCAmelCase			,     normalized=UpperCAmelCase	)
         if isinstance(UpperCAmelCase			,     UpperCAmelCase	)
         else mask_token
     )

     super().__init__(
         UpperCAmelCase			,     tokenizer_file=UpperCAmelCase			,     do_lower_case=UpperCAmelCase			,     remove_space=UpperCAmelCase			,     keep_accents=UpperCAmelCase			,     bos_token=UpperCAmelCase			,     eos_token=UpperCAmelCase			,     unk_token=UpperCAmelCase			,     sep_token=UpperCAmelCase			,     pad_token=UpperCAmelCase			,     cls_token=UpperCAmelCase			,     mask_token=UpperCAmelCase			,     **UpperCAmelCase			,     )

     lowerCamelCase__    :      Union[str, Any]					 =		do_lower_case
     lowerCamelCase__    :      List[Any]					 =		remove_space
     lowerCamelCase__    :      Optional[Any]					 =		keep_accents
     lowerCamelCase__    :      Tuple					 =		vocab_file
     lowerCamelCase__    :      Tuple					 =		False if not self.vocab_file else True
 def      A_	(      self  :      List[str]			,     UpperCAmelCase  :      List[int]			,     UpperCAmelCase  :      Optional[List[int]] = None	)			->	List[int]:
     lowerCamelCase__    :      Optional[Any]					 =		[self.sep_token_id]
     lowerCamelCase__    :      List[Any]					 =		[self.cls_token_id]
     if token_ids_a is None:
         return cls + token_ids_a + sep
     return cls + token_ids_a + sep + token_ids_a + sep
 def      A_	(      self  :      List[Any]			,     UpperCAmelCase  :      List[int]			,     UpperCAmelCase  :      Optional[List[int]] = None	)			->	List[int]:
     lowerCamelCase__    :      List[str]					 =		[self.sep_token_id]
     lowerCamelCase__    :      int					 =		[self.cls_token_id]

     if token_ids_a is None:
         return len(cls + token_ids_a + sep	) * [0]
     return len(cls + token_ids_a + sep	) * [0] + len(token_ids_a + sep	) * [1]







 def      A_	(      self  :      Any			,     UpperCAmelCase  :      str			,     UpperCAmelCase  :      Optional[str] = None	)			->	Tuple[str]:
     if not self.can_save_slow_tokenizer:
         raise ValueError(
             'Your fast tokenizer does not have the necessary information to save the vocabulary for a slow '
             'tokenizer.'	)

     if not os.path.isdir(UpperCAmelCase	):
         logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""	)
         return
     lowerCamelCase__    :      List[str]					 =		os.path.join(
         UpperCAmelCase			,     (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']	)

     if os.path.abspath(self.vocab_file	) != os.path.abspath(UpperCAmelCase	):
         copyfile(self.vocab_file			,     UpperCAmelCase	)

     return (out_vocab_file,)












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 get_down_block, get_mid_block, get_out_block, get_up_block

@dataclass
class    lowerCAmelCase							(					__UpperCamelCase			):
 UpperCAmelCase__													=		42


class    lowerCAmelCase							(					__UpperCamelCase,  __UpperCamelCase			):
 @register_to_config
 def __init__(      self  :      List[str]			,     UpperCAmelCase  :      int = 65536			,     UpperCAmelCase  :      Optional[int] = None			,     UpperCAmelCase  :      int = 2			,     UpperCAmelCase  :      int = 2			,     UpperCAmelCase  :      int = 0			,     UpperCAmelCase  :      str = "fourier"			,     UpperCAmelCase  :      bool = True			,     UpperCAmelCase  :      bool = False			,     UpperCAmelCase  :      float = 0.0			,     UpperCAmelCase  :      Tuple[str] = ("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")			,     UpperCAmelCase  :      Tuple[str] = ("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")			,     UpperCAmelCase  :      Tuple[str] = "UNetMidBlock1D"			,     UpperCAmelCase  :      str = None			,     UpperCAmelCase  :      Tuple[int] = (32, 32, 64)			,     UpperCAmelCase  :      str = None			,     UpperCAmelCase  :      int = 8			,     UpperCAmelCase  :      int = 1			,     UpperCAmelCase  :      bool = False			,     )			->	List[Any]:
     super().__init__()
     lowerCamelCase__    :      Optional[int]					 =		sample_size

     # time
     if time_embedding_type == "fourier":
         lowerCamelCase__    :      Optional[Any]					 =		GaussianFourierProjection(
             embedding_size=8			,     set_W_to_weight=UpperCAmelCase			,     log=UpperCAmelCase			,     flip_sin_to_cos=UpperCAmelCase	)
         lowerCamelCase__    :      Any					 =		2 * block_out_channels[0]
     elif time_embedding_type == "positional":
         lowerCamelCase__    :      List[Any]					 =		Timesteps(
             block_out_channels[0]			,     flip_sin_to_cos=UpperCAmelCase			,     downscale_freq_shift=UpperCAmelCase	)
         lowerCamelCase__    :      Dict					 =		block_out_channels[0]

     if use_timestep_embedding:
         lowerCamelCase__    :      str					 =		block_out_channels[0] * 4
         lowerCamelCase__    :      List[Any]					 =		TimestepEmbedding(
             in_channels=UpperCAmelCase			,     time_embed_dim=UpperCAmelCase			,     act_fn=UpperCAmelCase			,     out_dim=block_out_channels[0]			,     )

     lowerCamelCase__    :      Any					 =		nn.ModuleList([]	)
     lowerCamelCase__    :      Tuple					 =		None
     lowerCamelCase__    :      List[str]					 =		nn.ModuleList([]	)
     lowerCamelCase__    :      Optional[int]					 =		None

     # down
     lowerCamelCase__    :      Optional[int]					 =		in_channels
     for i, down_block_type in enumerate(UpperCAmelCase	):
         lowerCamelCase__    :      Union[str, Any]					 =		output_channel
         lowerCamelCase__    :      Tuple					 =		block_out_channels[i]

         if i == 0:
             input_channel += extra_in_channels

         lowerCamelCase__    :      Union[str, Any]					 =		i == len(UpperCAmelCase	) - 1

         lowerCamelCase__    :      Optional[int]					 =		get_down_block(
             UpperCAmelCase			,     num_layers=UpperCAmelCase			,     in_channels=UpperCAmelCase			,     out_channels=UpperCAmelCase			,     temb_channels=block_out_channels[0]			,     add_downsample=not is_final_block or downsample_each_block			,     )
         self.down_blocks.append(UpperCAmelCase	)

     # mid
     lowerCamelCase__    :      Optional[int]					 =		get_mid_block(
         UpperCAmelCase			,     in_channels=block_out_channels[-1]			,     mid_channels=block_out_channels[-1]			,     out_channels=block_out_channels[-1]			,     embed_dim=block_out_channels[0]			,     num_layers=UpperCAmelCase			,     add_downsample=UpperCAmelCase			,     )

     # up
     lowerCamelCase__    :      Optional[int]					 =		list(reversed(UpperCAmelCase	)	)
     lowerCamelCase__    :      Optional[int]					 =		reversed_block_out_channels[0]
     if out_block_type is None:
         lowerCamelCase__    :      List[str]					 =		out_channels
     else:
         lowerCamelCase__    :      Any					 =		block_out_channels[0]

     for i, up_block_type in enumerate(UpperCAmelCase	):
         lowerCamelCase__    :      Tuple					 =		output_channel
         lowerCamelCase__    :      Union[str, Any]					 =		(
             reversed_block_out_channels[i + 1] if i < len(UpperCAmelCase	) - 1 else final_upsample_channels
         )

         lowerCamelCase__    :      List[str]					 =		i == len(UpperCAmelCase	) - 1

         lowerCamelCase__    :      Dict					 =		get_up_block(
             UpperCAmelCase			,     num_layers=UpperCAmelCase			,     in_channels=UpperCAmelCase			,     out_channels=UpperCAmelCase			,     temb_channels=block_out_channels[0]			,     add_upsample=not is_final_block			,     )
         self.up_blocks.append(UpperCAmelCase	)
         lowerCamelCase__    :      int					 =		output_channel

     # out
     lowerCamelCase__    :      int					 =		norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4			,     32	)
     lowerCamelCase__    :      List[Any]					 =		get_out_block(
         out_block_type=UpperCAmelCase			,     num_groups_out=UpperCAmelCase			,     embed_dim=block_out_channels[0]			,     out_channels=UpperCAmelCase			,     act_fn=UpperCAmelCase			,     fc_dim=block_out_channels[-1] // 4			,     )







 def      A_	(      self  :      List[Any]			,     UpperCAmelCase  :      torch.FloatTensor			,     UpperCAmelCase  :      Union[torch.Tensor, float, int]			,     UpperCAmelCase  :      bool = True			,     )			->	Union[UNetaDOutput, Tuple]:
     lowerCamelCase__    :      Optional[Any]					 =		timestep
     if not torch.is_tensor(UpperCAmelCase	):
         lowerCamelCase__    :      Optional[int]					 =		torch.tensor([timesteps]			,     dtype=torch.long			,     device=sample.device	)
     elif torch.is_tensor(UpperCAmelCase	) and len(timesteps.shape	) == 0:
         lowerCamelCase__    :      List[str]					 =		timesteps[None].to(sample.device	)

     lowerCamelCase__    :      Optional[int]					 =		self.time_proj(UpperCAmelCase	)
     if self.config.use_timestep_embedding:
         lowerCamelCase__    :      str					 =		self.time_mlp(UpperCAmelCase	)
     else:
         lowerCamelCase__    :      List[str]					 =		timestep_embed[..., None]
         lowerCamelCase__    :      str					 =		timestep_embed.repeat([1, 1, sample.shape[2]]	).to(sample.dtype	)
         lowerCamelCase__    :      str					 =		timestep_embed.broadcast_to((sample.shape[:1] + timestep_embed.shape[1:])	)

     # 2. down
     lowerCamelCase__    :      str					 =		()
     for downsample_block in self.down_blocks:
         lowerCamelCase__ ,							lowerCamelCase__    :      Union[str, Any]					 =		downsample_block(hidden_states=UpperCAmelCase			,     temb=UpperCAmelCase	)
         down_block_res_samples += res_samples

     # 3. mid
     if self.mid_block:
         lowerCamelCase__    :      Optional[Any]					 =		self.mid_block(UpperCAmelCase			,     UpperCAmelCase	)

     # 4. up
     for i, upsample_block in enumerate(self.up_blocks	):
         lowerCamelCase__    :      Dict					 =		down_block_res_samples[-1:]
         lowerCamelCase__    :      Optional[Any]					 =		down_block_res_samples[:-1]
         lowerCamelCase__    :      Any					 =		upsample_block(UpperCAmelCase			,     res_hidden_states_tuple=UpperCAmelCase			,     temb=UpperCAmelCase	)

     # 5. post-process
     if self.out_block:
         lowerCamelCase__    :      Any					 =		self.out_block(UpperCAmelCase			,     UpperCAmelCase	)

     if not return_dict:
         return (sample,)

     return UNetaDOutput(sample=UpperCAmelCase	)





