





import os
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


__UpperCAmelCase	:       str    	=							logging.get_logger(__name__)
__UpperCAmelCase	:       List[str]    	=							{"vocab_file": "spiece.model"}

__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",
    }
}

__UpperCAmelCase	:       Union[str, Any]    	=							{
    "albert-base-v1": 512,
    "albert-large-v1": 512,
    "albert-xlarge-v1": 512,
    "albert-xxlarge-v1": 512,
    "albert-base-v2": 512,
    "albert-large-v2": 512,
    "albert-xlarge-v2": 512,
    "albert-xxlarge-v2": 512,
}

__UpperCAmelCase	:       Tuple    	=							"▁"







class    __snake_case	(							__lowerCamelCase      ):





       '''simple docstring'''







       lowerCAmelCase__						=					VOCAB_FILES_NAMES
       lowerCAmelCase__						=					PRETRAINED_VOCAB_FILES_MAP
       lowerCAmelCase__						=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES




       def __init__(  self			:   Union[str, Any]							,    A			:   Tuple							,    A			:   List[str]=True							,    A			:   List[str]=True							,    A			:   List[str]=False							,    A			:   Tuple="[CLS]"							,    A			:   Optional[Any]="[SEP]"							,    A			:   Dict="<unk>"							,    A			:   Optional[Any]="[SEP]"							,    A			:   Dict="<pad>"							,    A			:   str="[CLS]"							,    A			:   List[str]="[MASK]"							,    A			:   Optional[Dict[str, Any]] = None							,    **A			:   Tuple							,    ):
         # 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.
         __snake_case:						List[Any]     =						(
             AddedToken(A							,    lstrip=A							,    rstrip=A							,    normalized=A		)
             if isinstance(A							,    A		)
             else mask_token
         )

         __snake_case:						List[Any]     =						{} if sp_model_kwargs is None else sp_model_kwargs

         super().__init__(
             do_lower_case=A							,    remove_space=A							,    keep_accents=A							,    bos_token=A							,    eos_token=A							,    unk_token=A							,    sep_token=A							,    pad_token=A							,    cls_token=A							,    mask_token=A							,    sp_model_kwargs=self.sp_model_kwargs							,    **A							,    )

         __snake_case:						Optional[Any]     =						do_lower_case
         __snake_case:						Tuple     =						remove_space
         __snake_case:						Tuple     =						keep_accents
         __snake_case:						int     =						vocab_file

         __snake_case:						Optional[Any]     =						spm.SentencePieceProcessor(**self.sp_model_kwargs		)
         self.sp_model.Load(A		)




       @property
       def 							UpperCAmelCase__			(  self			:   List[Any]		):
         return len(self.sp_model		)




       def 							UpperCAmelCase__			(  self			:   List[Any]		):
         __snake_case:						Optional[int]     =						{self.convert_ids_to_tokens(A		): i for i in range(self.vocab_size		)}
         vocab.update(self.added_tokens_encoder		)
         return vocab




       def __getstate__(  self			:   str		):
         __snake_case:						List[str]     =						self.__dict__.copy()
         __snake_case:						Any     =						None
         return state




       def __setstate__(  self			:   Optional[Any]							,    A			:   str		):
         __snake_case:						str     =						d

         # for backward compatibility
         if not hasattr(self							,    """sp_model_kwargs"""		):
           __snake_case:						int     =						{}

         __snake_case:						str     =						spm.SentencePieceProcessor(**self.sp_model_kwargs		)
         self.sp_model.Load(self.vocab_file		)




       def 							UpperCAmelCase__			(  self			:   str							,    A			:   Optional[Any]		):
         if self.remove_space:
           __snake_case:						List[str]     =						""" """.join(inputs.strip().split()		)
         else:
           __snake_case:						Union[str, Any]     =						inputs
         __snake_case:						Union[str, Any]     =						outputs.replace("""``"""							,    """\""""		).replace("""''"""							,    """\""""		)

         if not self.keep_accents:
           __snake_case:						str     =						unicodedata.normalize("""NFKD"""							,    A		)
           __snake_case:						List[str]     =						"""""".join([c for c in outputs if not unicodedata.combining(A		)]		)
         if self.do_lower_case:
           __snake_case:						Dict     =						outputs.lower()

         return outputs




       def 							UpperCAmelCase__			(  self			:   Tuple							,    A			:   str		):
         __snake_case:						List[str]     =						self.preprocess_text(A		)
         __snake_case:						Any     =						self.sp_model.encode(A							,    out_type=A		)
         __snake_case:						List[Any]     =						[]
         for piece in pieces:
           if len(A		) > 1 and piece[-1] == str(""","""		) and piece[-2].isdigit():
             __snake_case:						str     =						self.sp_model.EncodeAsPieces(piece[:-1].replace(A							,    """"""		)		)
             if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
               if len(cur_pieces[0]		) == 1:
                 __snake_case:						Tuple     =						cur_pieces[1:]
               else:
                 __snake_case:						Dict     =						cur_pieces[0][1:]
             cur_pieces.append(piece[-1]		)
             new_pieces.extend(A		)
           else:
             new_pieces.append(A		)

         return new_pieces




       def 							UpperCAmelCase__			(  self			:   int							,    A			:   Optional[Any]		):
         return self.sp_model.PieceToId(A		)




       def 							UpperCAmelCase__			(  self			:   Dict							,    A			:   str		):
         return self.sp_model.IdToPiece(A		)




       def 							UpperCAmelCase__			(  self			:   str							,    A			:   List[str]		):
         __snake_case:						Optional[Any]     =						[]
         __snake_case:						int     =						""""""
         __snake_case:						int     =						False
         for token in tokens:
           # make sure that special tokens are not decoded using sentencepiece model
           if token in self.all_special_tokens:
             if not prev_is_special:
               out_string += " "
             out_string += self.sp_model.decode(A		) + token
             __snake_case:						List[Any]     =						True
             __snake_case:						Dict     =						[]
           else:
             current_sub_tokens.append(A		)
             __snake_case:						List[Any]     =						False
         out_string += self.sp_model.decode(A		)
         return out_string.strip()




       def 							UpperCAmelCase__			(  self			:   List[str]							,    A			:   List[int]							,    A			:   Optional[List[int]] = None		):
         __snake_case:						Any     =						[self.sep_token_id]
         __snake_case:						Optional[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 							UpperCAmelCase__			(  self			:   Union[str, Any]							,    A			:   List[int]							,    A			:   Optional[List[int]] = None							,    A			:   bool = False		):

         if already_has_special_tokens:
           return super().get_special_tokens_mask(
               token_ids_a=A							,    token_ids_a=A							,    already_has_special_tokens=A		)

         if token_ids_a is not None:
           return [1] + ([0] * len(A		)) + [1] + ([0] * len(A		)) + [1]
         return [1] + ([0] * len(A		)) + [1]




       def 							UpperCAmelCase__			(  self			:   Any							,    A			:   List[int]							,    A			:   Optional[List[int]] = None		):
         __snake_case:						Dict     =						[self.sep_token_id]
         __snake_case:						Optional[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 							UpperCAmelCase__			(  self			:   Optional[int]							,    A			:   str							,    A			:   Optional[str] = None		):
         if not os.path.isdir(A		):
           logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''		)
           return
         __snake_case:						Tuple     =						os.path.join(
             A							,    (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]		)

         if os.path.abspath(self.vocab_file		) != os.path.abspath(A		) and os.path.isfile(self.vocab_file		):
           copyfile(self.vocab_file							,    A		)
         elif not os.path.isfile(self.vocab_file		):
           with open(A							,    """wb"""		) as fi:
             __snake_case:						str     =						self.sp_model.serialized_model_proto()
             fi.write(A		)

         return (out_vocab_file,)













import copy
import tempfile
import unittest

from huggingface_hub import HfFolder, delete_repo
from parameterized import parameterized
from requests.exceptions import HTTPError

from transformers import AutoConfig, GenerationConfig
from transformers.testing_utils import TOKEN, USER, is_staging_test







class    __snake_case	(							unittest.TestCase      ):





       '''simple docstring'''







       @parameterized.expand([(None,), ("""foo.json""",)]		)
       def 							UpperCAmelCase__			(  self			:   List[str]							,    A			:   Optional[Any]		):
         __snake_case:						Any			  =						GenerationConfig(
             do_sample=A							,    temperature=0.7							,    length_penalty=1.0							,    bad_words_ids=[[1, 2, 3], [4, 5]]							,    )
         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(A							,    config_name=A		)
           __snake_case:						Optional[int]			  =						GenerationConfig.from_pretrained(A							,    config_name=A		)

         # Checks parameters that were specified
         self.assertEqual(loaded_config.do_sample							,    A		)
         self.assertEqual(loaded_config.temperature							,    0.7		)
         self.assertEqual(loaded_config.length_penalty							,    1.0		)
         self.assertEqual(loaded_config.bad_words_ids							,    [[1, 2, 3], [4, 5]]		)

         # Checks parameters that were not specified (defaults)
         self.assertEqual(loaded_config.top_k							,    50		)
         self.assertEqual(loaded_config.max_length							,    20		)
         self.assertEqual(loaded_config.max_time							,    A		)




       def 							UpperCAmelCase__			(  self			:   Dict		):
         __snake_case:						str			  =						AutoConfig.from_pretrained("""gpt2"""		)
         __snake_case:						Any			  =						GenerationConfig.from_model_config(A		)
         __snake_case:						str			  =						GenerationConfig()

         # The generation config has loaded a few non-default parameters from the model config
         self.assertNotEqual(A							,    A		)

         # One of those parameters is eos_token_id -- check if it matches
         self.assertNotEqual(generation_config_from_model.eos_token_id							,    default_generation_config.eos_token_id		)
         self.assertEqual(generation_config_from_model.eos_token_id							,    model_config.eos_token_id		)




       def 							UpperCAmelCase__			(  self			:   str		):
         __snake_case:						List[str]			  =						GenerationConfig()
         __snake_case:						Tuple			  =						{
             """max_new_tokens""": 1_024,
             """foo""": """bar""",
         }
         __snake_case:						List[str]			  =						copy.deepcopy(A		)
         __snake_case:						Optional[int]			  =						generation_config.update(**A		)

         # update_kwargs was not modified (no side effects)
         self.assertEqual(A							,    A		)

         # update_kwargs was used to update the config on valid attributes
         self.assertEqual(generation_config.max_new_tokens							,    1_024		)

         # `.update()` returns a dictionary of unused kwargs
         self.assertEqual(A							,    {"""foo""": """bar"""}		)




       def 							UpperCAmelCase__			(  self			:   Tuple		):
         __snake_case:						List[str]			  =						GenerationConfig()
         __snake_case:						Optional[int]			  =						"""bar"""

         with tempfile.TemporaryDirectory("""test-generation-config"""		) as tmp_dir:
           generation_config.save_pretrained(A		)

           __snake_case:						Any			  =						GenerationConfig.from_pretrained(A		)
         # update_kwargs was used to update the config on valid attributes
         self.assertEqual(new_config.foo							,    """bar"""		)

         __snake_case:						int			  =						GenerationConfig.from_model_config(A		)
         assert not hasattr(A							,    """foo"""		)  # no new kwargs should be initialized if from config






       def 							UpperCAmelCase__			(  self			:   Dict		):
         __snake_case:						Dict			  =						GenerationConfig()
         self.assertEqual(default_config.temperature							,    1.0		)
         self.assertEqual(default_config.do_sample							,    A		)
         self.assertEqual(default_config.num_beams							,    1		)

         __snake_case:						Union[str, Any]			  =						GenerationConfig(
             do_sample=A							,    temperature=0.7							,    length_penalty=1.0							,    bad_words_ids=[[1, 2, 3], [4, 5]]							,    )
         self.assertEqual(config.temperature							,    0.7		)
         self.assertEqual(config.do_sample							,    A		)
         self.assertEqual(config.num_beams							,    1		)

         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(A		)
           __snake_case:						Tuple			  =						GenerationConfig.from_pretrained(A							,    temperature=1.0		)

         self.assertEqual(loaded_config.temperature							,    1.0		)
         self.assertEqual(loaded_config.do_sample							,    A		)
         self.assertEqual(loaded_config.num_beams							,    1		)  # default value







@is_staging_test
class    __snake_case	(							unittest.TestCase      ):





       '''simple docstring'''







       @classmethod
       def 							UpperCAmelCase__			(  cls			:   List[str]		):
         __snake_case:						Optional[int]			  =						TOKEN
         HfFolder.save_token(A		)




       @classmethod
       def 							UpperCAmelCase__			(  cls			:   List[Any]		):
         try:
           delete_repo(token=cls._token							,    repo_id="""test-generation-config"""		)
         except HTTPError:
           pass

         try:
           delete_repo(token=cls._token							,    repo_id="""valid_org/test-generation-config-org"""		)
         except HTTPError:
           pass




       def 							UpperCAmelCase__			(  self			:   Tuple		):
         __snake_case:						Optional[int]			  =						GenerationConfig(
             do_sample=A							,    temperature=0.7							,    length_penalty=1.0							,    )
         config.push_to_hub("""test-generation-config"""							,    use_auth_token=self._token		)

         __snake_case:						str			  =						GenerationConfig.from_pretrained(f'''{USER}/test-generation-config'''		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(A							,    getattr(A							,    A		)		)

        # Reset repo
         delete_repo(token=self._token							,    repo_id="""test-generation-config"""		)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(
               A							,    repo_id="""test-generation-config"""							,    push_to_hub=A							,    use_auth_token=self._token		)

         __snake_case:						Optional[Any]			  =						GenerationConfig.from_pretrained(f'''{USER}/test-generation-config'''		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(A							,    getattr(A							,    A		)		)






       def 							UpperCAmelCase__			(  self			:   List[Any]		):
         __snake_case:						Union[str, Any]			  =						GenerationConfig(
             do_sample=A							,    temperature=0.7							,    length_penalty=1.0							,    )
         config.push_to_hub("""valid_org/test-generation-config-org"""							,    use_auth_token=self._token		)

         __snake_case:						int			  =						GenerationConfig.from_pretrained("""valid_org/test-generation-config-org"""		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(A							,    getattr(A							,    A		)		)

        # Reset repo
         delete_repo(token=self._token							,    repo_id="""valid_org/test-generation-config-org"""		)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           config.save_pretrained(
               A							,    repo_id="""valid_org/test-generation-config-org"""							,    push_to_hub=A							,    use_auth_token=self._token		)

         __snake_case:						Optional[int]			  =						GenerationConfig.from_pretrained("""valid_org/test-generation-config-org"""		)
         for k, v in config.to_dict().items():
           if k != "transformers_version":
             self.assertEqual(A							,    getattr(A							,    A		)		)






