


"""simple docstring"""



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 SPIECE_UNDERLINE, logging


__snake_case   			=    logging.get_logger(__name__)

__snake_case   			=    {"""vocab_file""": """spiece.model"""}

__snake_case   			=    {
    """vocab_file""": {
        """xlnet-base-cased""": """https://huggingface.co/xlnet-base-cased/resolve/main/spiece.model""",
        """xlnet-large-cased""": """https://huggingface.co/xlnet-large-cased/resolve/main/spiece.model""",
    }
}

__snake_case   			=    {
    """xlnet-base-cased""": None,
    """xlnet-large-cased""": None,
}

# Segments (not really needed)
__snake_case   			=    0
__snake_case   			=    1
__snake_case   			=    2
__snake_case   			=    3
__snake_case   			=    4







class _lowerCAmelCase				(      snake_case_	):
   __UpperCAmelCase :       Tuple							  =  VOCAB_FILES_NAMES
   __UpperCAmelCase :       Union[str, Any]							  =  PRETRAINED_VOCAB_FILES_MAP
   __UpperCAmelCase :       Optional[Any]							  =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   __UpperCAmelCase :       str							  =  '''left'''







   def __init__(	self						,	UpperCamelCase__						,	UpperCamelCase__=False						,	UpperCamelCase__=True						,	UpperCamelCase__=False						,	UpperCamelCase__="<s>"						,	UpperCamelCase__="</s>"						,	UpperCamelCase__="<unk>"						,	UpperCamelCase__="<sep>"						,	UpperCamelCase__="<pad>"						,	UpperCamelCase__="<cls>"						,	UpperCamelCase__="<mask>"						,	UpperCamelCase__=["<eop>", "<eod>"]						,	UpperCamelCase__ = None						,	**UpperCamelCase__						,	)	->		None:

       '''simple docstring'''




       snake_case    :      Tuple      					=    AddedToken(UpperCamelCase__						,	lstrip=UpperCamelCase__						,	rstrip=UpperCamelCase__						) if isinstance(UpperCamelCase__						,	UpperCamelCase__						) else mask_token

       snake_case    :      str      					=    {} if sp_model_kwargs is None else sp_model_kwargs

       super().__init__(
           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__						,	additional_special_tokens=UpperCamelCase__						,	sp_model_kwargs=self.sp_model_kwargs						,	**UpperCamelCase__						,	)

       snake_case    :      Optional[int]      					=    3

       snake_case    :      Union[str, Any]      					=    do_lower_case
       snake_case    :      str      					=    remove_space
       snake_case    :      List[Any]      					=    keep_accents
       snake_case    :      Any      					=    vocab_file

       snake_case    :      Optional[int]      					=    spm.SentencePieceProcessor(**self.sp_model_kwargs						)
       self.sp_model.Load(UpperCamelCase__						)







   @property
   def  lowerCamelCase		(	self						)	->		Tuple:

       '''simple docstring'''




       return len(self.sp_model						)







   def  lowerCamelCase		(	self						)	->		str:

       '''simple docstring'''




       snake_case    :      Any      					=    {self.convert_ids_to_tokens(UpperCamelCase__						): i for i in range(self.vocab_size						)}
       vocab.update(self.added_tokens_encoder						)
       return vocab







   def __getstate__(	self						)	->		Any:

       '''simple docstring'''




       snake_case    :      Tuple      					=    self.__dict__.copy()
       snake_case    :      Dict      					=    None
       return state







   def __setstate__(	self						,	UpperCamelCase__						)	->		str:

       '''simple docstring'''




       snake_case    :      Dict      					=    d

       # for backward compatibility
       if not hasattr(self						,	"sp_model_kwargs"						):
           snake_case    :      Optional[Any]      					=    {}

       snake_case    :      List[Any]      					=    spm.SentencePieceProcessor(**self.sp_model_kwargs						)
       self.sp_model.Load(self.vocab_file						)







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		Dict:

       '''simple docstring'''




       if self.remove_space:
           snake_case    :      int      					=    " ".join(inputs.strip().split()						)
       else:
           snake_case    :      Any      					=    inputs
       snake_case    :      List[Any]      					=    outputs.replace("``"						,	"\""						).replace("''"						,	"\""						)

       if not self.keep_accents:
           snake_case    :      str      					=    unicodedata.normalize("NFKD"						,	UpperCamelCase__						)
           snake_case    :      Any      					=    "".join([c for c in outputs if not unicodedata.combining(UpperCamelCase__						)]						)
       if self.do_lower_case:
           snake_case    :      int      					=    outputs.lower()

       return outputs







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		List[str]:

       '''simple docstring'''




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

       return new_pieces







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		List[str]:

       '''simple docstring'''




       return self.sp_model.PieceToId(UpperCamelCase__						)







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		Any:

       '''simple docstring'''




       return self.sp_model.IdToPiece(UpperCamelCase__						)







   def  lowerCamelCase		(	self						,	UpperCamelCase__						)	->		Optional[int]:

       '''simple docstring'''




       snake_case    :      Any      					=    "".join(UpperCamelCase__						).replace(UpperCamelCase__						,	" "						).strip()
       return out_string







   def  lowerCamelCase		(	self						,	UpperCamelCase__						,	UpperCamelCase__ = False						,	UpperCamelCase__ = None						,	UpperCamelCase__ = True						,	**UpperCamelCase__						,	)	->		str:

       '''simple docstring'''




       snake_case    :      str      					=    kwargs.pop("use_source_tokenizer"						,	UpperCamelCase__						)

       snake_case    :      Union[str, Any]      					=    self.convert_ids_to_tokens(UpperCamelCase__						,	skip_special_tokens=UpperCamelCase__						)

       # To avoid mixing byte-level and unicode for byte-level BPT
       # we need to build string separately for added tokens and byte-level tokens
       # cf. https://github.com/huggingface/transformers/issues/1133
       snake_case    :      List[str]      					=    []
       snake_case    :      Dict      					=    []
       for token in filtered_tokens:
           if skip_special_tokens and token in self.all_special_ids:
               continue
           if token in self.added_tokens_encoder:
               if current_sub_text:
                   sub_texts.append(self.convert_tokens_to_string(UpperCamelCase__						)						)
                   snake_case    :      str      					=    []
               sub_texts.append(UpperCamelCase__						)
           else:
               current_sub_text.append(UpperCamelCase__						)
       if current_sub_text:
           sub_texts.append(self.convert_tokens_to_string(UpperCamelCase__						)						)

       # Mimic the behavior of the Rust tokenizer:
       # By default, there are no spaces between special tokens
       snake_case    :      List[Any]      					=    "".join(UpperCamelCase__						)

       snake_case    :      Tuple      					=    (
           clean_up_tokenization_spaces
           if clean_up_tokenization_spaces is not None
           else self.clean_up_tokenization_spaces
       )
       if clean_up_tokenization_spaces:
           snake_case    :      int      					=    self.clean_up_tokenization(UpperCamelCase__						)
           return clean_text
       else:
           return text







   def  lowerCamelCase		(	self						,	UpperCamelCase__						,	UpperCamelCase__ = None						)	->		List[int]:

       '''simple docstring'''




       snake_case    :      List[Any]      					=    [self.sep_token_id]
       snake_case    :      Tuple      					=    [self.cls_token_id]
       if token_ids_a is None:
           return token_ids_a + sep + cls
       return token_ids_a + sep + token_ids_a + sep + cls







   def  lowerCamelCase		(	self						,	UpperCamelCase__						,	UpperCamelCase__ = None						,	UpperCamelCase__ = False						)	->		List[int]:

       '''simple docstring'''




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

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







   def  lowerCamelCase		(	self						,	UpperCamelCase__						,	UpperCamelCase__ = None						)	->		List[int]:

       '''simple docstring'''




       snake_case    :      str      					=    [self.sep_token_id]
       snake_case    :      str      					=    [2]

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







   def  lowerCamelCase		(	self						,	UpperCamelCase__						,	UpperCamelCase__ = None						)	->		Tuple[str]:

       '''simple docstring'''




       if not os.path.isdir(UpperCamelCase__						):
           logger.error(F'Vocabulary path ({save_directory}) should be a directory'						)
           return
       snake_case    :      Dict      					=    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__						) and os.path.isfile(self.vocab_file						):
           copyfile(self.vocab_file						,	UpperCamelCase__						)
       elif not os.path.isfile(self.vocab_file						):
           with open(UpperCamelCase__						,	"wb"						) as fi:
               snake_case    :      Dict      					=    self.sp_model.serialized_model_proto()
               fi.write(UpperCamelCase__						)

       return (out_vocab_file,)









"""simple docstring"""



import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


__snake_case   			=    logging.get_logger(__name__)







def   __lowerCAmelCase						(						)	->						str:


    """simple docstring"""





    snake_case    :      Dict      					=    os.getenv("SM_HP_MP_PARAMETERS"						,							"{}" )
    try:
        # Parse it and check the field "partitions" is included, it is required for model parallel.
        snake_case    :      Optional[int]      					=    json.loads(lowercase )
        if "partitions" not in smp_options:
            return False
    except json.JSONDecodeError:
        return False

    # Get the sagemaker specific framework parameters from mpi_options variable.
    snake_case    :      Optional[int]      					=    os.getenv("SM_FRAMEWORK_PARAMS"						,							"{}" )
    try:
        # Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
        snake_case    :      Any      					=    json.loads(lowercase )
        if not mpi_options.get("sagemaker_mpi_enabled"						,							lowercase ):
            return False
    except json.JSONDecodeError:
        return False
    # Lastly, check if the `smdistributed` module is present.
    return importlib.util.find_spec("smdistributed" ) is not None


if is_sagemaker_model_parallel_available():
  import smdistributed.modelparallel.torch as smp

  smp.init()







@dataclass
class _lowerCAmelCase				(      snake_case_	):
   __UpperCAmelCase :       str							  =  field(
       default=''''''    ,							metadata={'''help''': '''Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer'''}    ,							)







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

       '''simple docstring'''




       super().__post_init__()
       warnings.warn(
           "`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use "
           "`TrainingArguments` instead."						,	UpperCamelCase__						,	)







   @cached_property
   def  lowerCamelCase		(	self						)	->		"torch.device":

       '''simple docstring'''




       logger.info("PyTorch: setting up devices"						)
       if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
           logger.warning(
               "torch.distributed process group is initialized, but local_rank == -1. "
               "In order to use Torch DDP, launch your script with `python -m torch.distributed.launch"						)
       if self.no_cuda:
           snake_case    :      Optional[Any]      					=    torch.device("cpu"						)
           snake_case    :      List[Any]      					=    0
       elif is_sagemaker_model_parallel_available():
           snake_case    :      Tuple      					=    smp.local_rank()
           snake_case    :      int      					=    torch.device("cuda"						,	UpperCamelCase__						)
           snake_case    :      Dict      					=    1
       elif is_sagemaker_dp_enabled():
           import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

           torch.distributed.init_process_group(backend="smddp"						,	timeout=self.ddp_timeout_delta						)
           snake_case    :      Any      					=    int(os.getenv("SMDATAPARALLEL_LOCAL_RANK"						)						)
           snake_case    :      Optional[Any]      					=    torch.device("cuda"						,	self.local_rank						)
           snake_case    :      str      					=    1
       elif self.local_rank == -1:
           # if n_gpu is > 1 we'll use nn.DataParallel.
           # If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
           # Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
           # trigger an error that a device index is missing. Index 0 takes into account the
           # GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
           # will use the first GPU in that env, i.e. GPU#1
           snake_case    :      List[str]      					=    torch.device("cuda:0" if torch.cuda.is_available() else "cpu"						)
           # Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
           # the default value.
           snake_case    :      Optional[Any]      					=    torch.cuda.device_count()
       else:
           # Here, we'll use torch.distributed.
           # Initializes the distributed backend which will take care of synchronizing nodes/GPUs
           if not torch.distributed.is_initialized():
               torch.distributed.init_process_group(backend="nccl"						,	timeout=self.ddp_timeout_delta						)
           snake_case    :      Any      					=    torch.device("cuda"						,	self.local_rank						)
           snake_case    :      Dict      					=    1

       if device.type == "cuda":
           torch.cuda.set_device(UpperCamelCase__						)

       return device







   @property
   def  lowerCamelCase		(	self						)	->		Optional[int]:

       '''simple docstring'''




       if is_sagemaker_model_parallel_available():
           return smp.dp_size()

       return super().world_size







   @property
   def  lowerCamelCase		(	self						)	->		List[str]:

       '''simple docstring'''




       return not is_sagemaker_model_parallel_available()







   @property
   def  lowerCamelCase		(	self						)	->		str:

       '''simple docstring'''




       return False





