





import json
import os
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

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


if TYPE_CHECKING:
       from transformers.pipelines.conversational import Conversation

UpperCAmelCase			:			Optional[Any]	  = logging.get_logger(__name__)

UpperCAmelCase			:			Optional[int]	  = {"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt"""}

UpperCAmelCase			:			Any	  = {
    """vocab_file""": {
        """microsoft/deberta-base""": """https://huggingface.co/microsoft/deberta-base/resolve/main/vocab.json""",
        """microsoft/deberta-large""": """https://huggingface.co/microsoft/deberta-large/resolve/main/vocab.json""",
        """microsoft/deberta-xlarge""": """https://huggingface.co/microsoft/deberta-xlarge/resolve/main/vocab.json""",
        """microsoft/deberta-base-mnli""": """https://huggingface.co/microsoft/deberta-base-mnli/resolve/main/vocab.json""",
        """microsoft/deberta-large-mnli""": """https://huggingface.co/microsoft/deberta-large-mnli/resolve/main/vocab.json""",
        """microsoft/deberta-xlarge-mnli""": (
            """https://huggingface.co/microsoft/deberta-xlarge-mnli/resolve/main/vocab.json"""
        ),
    },
    """merges_file""": {
        """microsoft/deberta-base""": """https://huggingface.co/microsoft/deberta-base/resolve/main/merges.txt""",
        """microsoft/deberta-large""": """https://huggingface.co/microsoft/deberta-large/resolve/main/merges.txt""",
        """microsoft/deberta-xlarge""": """https://huggingface.co/microsoft/deberta-xlarge/resolve/main/merges.txt""",
        """microsoft/deberta-base-mnli""": """https://huggingface.co/microsoft/deberta-base-mnli/resolve/main/merges.txt""",
        """microsoft/deberta-large-mnli""": """https://huggingface.co/microsoft/deberta-large-mnli/resolve/main/merges.txt""",
        """microsoft/deberta-xlarge-mnli""": (
            """https://huggingface.co/microsoft/deberta-xlarge-mnli/resolve/main/merges.txt"""
        ),
    },
}

UpperCAmelCase			:			Dict	  = {
    """microsoft/deberta-base""": 512,
    """microsoft/deberta-large""": 512,
    """microsoft/deberta-xlarge""": 512,
    """microsoft/deberta-base-mnli""": 512,
    """microsoft/deberta-large-mnli""": 512,
    """microsoft/deberta-xlarge-mnli""": 512,
}

UpperCAmelCase			:			int	  = {
    """microsoft/deberta-base""": {"""do_lower_case""": False},
    """microsoft/deberta-large""": {"""do_lower_case""": False},
}
def   _A (	):





  """simple docstring"""



  a__			:      List[str]            =(
      list(range(ord("!"		) , ord("~"		) + 1		)		) + list(range(ord("¡"		) , ord("¬"		) + 1		)		) + list(range(ord("®"		) , ord("ÿ"		) + 1		)		)
  )
  a__			:      List[str]            =bs[:]
  a__			:      Dict            =0
  for b in range(2**8		):
    if b not in bs:
      bs.append(SCREAMING_SNAKE_CASE		)
      cs.append(2**8 + n		)
      n += 1
  a__			:      Tuple            =[chr(SCREAMING_SNAKE_CASE		) for n in cs]
  return dict(zip(SCREAMING_SNAKE_CASE , SCREAMING_SNAKE_CASE		)		)
def   _A (	SCREAMING_SNAKE_CASE  :     Union[str, Any]		):





  """simple docstring"""



  a__			:      str            =set()
  a__			:      str            =word[0]
  for char in word[1:]:
    pairs.add((prev_char, char)		)
    a__			:      Optional[Any]            =char
  return pairs







class        __lowerCAmelCase (					UpperCamelCase__):
      _lowercase       :      Any        =    VOCAB_FILES_NAMES
      _lowercase       :      Union[str, Any]        =    PRETRAINED_VOCAB_FILES_MAP
      _lowercase       :      List[Any]        =    PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _lowercase       :      Any        =    ["""input_ids""", """attention_mask""", """token_type_ids"""]







      def __init__(					self	,   lowerCAmelCase__	,   lowerCAmelCase__	,   lowerCAmelCase__="replace"	,   lowerCAmelCase__="[CLS]"	,   lowerCAmelCase__="[SEP]"	,   lowerCAmelCase__="[SEP]"	,   lowerCAmelCase__="[CLS]"	,   lowerCAmelCase__="[UNK]"	,   lowerCAmelCase__="[PAD]"	,   lowerCAmelCase__="[MASK]"	,   lowerCAmelCase__=False	,   lowerCAmelCase__=False	,   **lowerCAmelCase__	,   )     ->		Any:



        '''simple docstring'''




        a__			:      List[str]            =AddedToken(lowerCAmelCase__	,   lstrip=lowerCAmelCase__	,   rstrip=lowerCAmelCase__							) if isinstance(lowerCAmelCase__	,   lowerCAmelCase__							) else bos_token
        a__			:      List[str]            =AddedToken(lowerCAmelCase__	,   lstrip=lowerCAmelCase__	,   rstrip=lowerCAmelCase__							) if isinstance(lowerCAmelCase__	,   lowerCAmelCase__							) else eos_token
        a__			:      Union[str, Any]            =AddedToken(lowerCAmelCase__	,   lstrip=lowerCAmelCase__	,   rstrip=lowerCAmelCase__							) if isinstance(lowerCAmelCase__	,   lowerCAmelCase__							) else sep_token
        a__			:      List[str]            =AddedToken(lowerCAmelCase__	,   lstrip=lowerCAmelCase__	,   rstrip=lowerCAmelCase__							) if isinstance(lowerCAmelCase__	,   lowerCAmelCase__							) else cls_token
        a__			:      int            =AddedToken(lowerCAmelCase__	,   lstrip=lowerCAmelCase__	,   rstrip=lowerCAmelCase__							) if isinstance(lowerCAmelCase__	,   lowerCAmelCase__							) else unk_token
        a__			:      Optional[Any]            =AddedToken(lowerCAmelCase__	,   lstrip=lowerCAmelCase__	,   rstrip=lowerCAmelCase__							) if isinstance(lowerCAmelCase__	,   lowerCAmelCase__							) else pad_token

        # Mask token behave like a normal word, i.e. include the space before it
        a__			:      Dict            =AddedToken(lowerCAmelCase__	,   lstrip=lowerCAmelCase__	,   rstrip=lowerCAmelCase__							) if isinstance(lowerCAmelCase__	,   lowerCAmelCase__							) else mask_token

        super().__init__(
            errors=lowerCAmelCase__	,   bos_token=lowerCAmelCase__	,   eos_token=lowerCAmelCase__	,   unk_token=lowerCAmelCase__	,   sep_token=lowerCAmelCase__	,   cls_token=lowerCAmelCase__	,   pad_token=lowerCAmelCase__	,   mask_token=lowerCAmelCase__	,   add_prefix_space=lowerCAmelCase__	,   add_bos_token=lowerCAmelCase__	,   **lowerCAmelCase__	,   )
        a__			:      Optional[int]            =add_bos_token

        with open(lowerCAmelCase__	,   encoding="utf-8"							) as vocab_handle:
          a__			:      List[Any]            =json.load(lowerCAmelCase__							)
        a__			:      List[str]            ={v: k for k, v in self.encoder.items()}
        a__			:      List[Any]            =errors  # how to handle errors in decoding
        a__			:      Optional[int]            =bytes_to_unicode()
        a__			:      Dict            ={v: k for k, v in self.byte_encoder.items()}
        with open(lowerCAmelCase__	,   encoding="utf-8"							) as merges_handle:
          a__			:      Any            =merges_handle.read().split("\n"							)[1:-1]
        a__			:      Union[str, Any]            =[tuple(merge.split()							) for merge in bpe_merges]
        a__			:      Optional[Any]            =dict(zip(lowerCAmelCase__	,   range(len(lowerCAmelCase__							)							)							)							)
        a__			:      List[str]            ={}
        a__			:      List[str]            =add_prefix_space

        # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
        a__			:      Optional[int]            =re.compile(r"'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"							)







      @property
      # Copied from transformers.models.gpt2.tokenization_gpt2.GPT2Tokenizer.vocab_size
      def      _lowercase						(					self							)     ->		Any:



        '''simple docstring'''




        return len(self.encoder							)







      def      _lowercase						(					self							)     ->		Optional[int]:



        '''simple docstring'''




        return dict(self.encoder	,   **self.added_tokens_encoder							)







      def      _lowercase						(					self	,   lowerCAmelCase__							)     ->		List[str]:



        '''simple docstring'''




        if token in self.cache:
          return self.cache[token]
        a__			:      Optional[Any]            =tuple(lowerCAmelCase__							)
        a__			:      str            =get_pairs(lowerCAmelCase__							)

        if not pairs:
          return token

        while True:
          a__			:      List[str]            =min(lowerCAmelCase__	,   key=lambda lowerCAmelCase__							: self.bpe_ranks.get(lowerCAmelCase__	,   float("inf"							)							)							)
          if bigram not in self.bpe_ranks:
            break
          a__			:      Optional[int]            =bigram
          a__			:      Tuple            =[]
          a__			:      Optional[int]            =0
          while i < len(lowerCAmelCase__							):
            try:
              a__			:      Optional[Any]            =word.index(lowerCAmelCase__	,   lowerCAmelCase__							)
            except ValueError:
              new_word.extend(word[i:]							)
              break
            else:
              new_word.extend(word[i:j]							)
              a__			:      Optional[int]            =j

            if word[i] == first and i < len(lowerCAmelCase__							) - 1 and word[i + 1] == second:
              new_word.append(first + second							)
              i += 2
            else:
              new_word.append(word[i]							)
              i += 1
          a__			:      Optional[int]            =tuple(lowerCAmelCase__							)
          a__			:      Any            =new_word
          if len(lowerCAmelCase__							) == 1:
            break
          else:
            a__			:      Optional[int]            =get_pairs(lowerCAmelCase__							)
        a__			:      Union[str, Any]            =" ".join(lowerCAmelCase__							)
        a__			:      Union[str, Any]            =word
        return word







      def      _lowercase						(					self	,   lowerCAmelCase__	,   lowerCAmelCase__ = None							)     ->		List[int]:



        '''simple docstring'''




        if token_ids_a is None:
          return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
        a__			:      Union[str, Any]            =[self.cls_token_id]
        a__			:      List[Any]            =[self.sep_token_id]
        return cls + token_ids_a + sep + token_ids_a + sep







      def      _lowercase						(					self	,   lowerCAmelCase__	,   lowerCAmelCase__ = None	,   lowerCAmelCase__ = False							)     ->		List[int]:



        '''simple docstring'''




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

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







      def      _lowercase						(					self	,   lowerCAmelCase__	,   lowerCAmelCase__ = None							)     ->		List[int]:



        '''simple docstring'''




        a__			:      Dict            =[self.sep_token_id]
        a__			:      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      _lowercase						(					self	,   lowerCAmelCase__							)     ->		Optional[int]:



        '''simple docstring'''




        a__			:      Tuple            =[]
        for token in re.findall(self.pat	,   lowerCAmelCase__							):
          a__			:      Any            ="".join(
              self.byte_encoder[b] for b in token.encode("utf-8"							)							)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
          bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowerCAmelCase__							).split(" "							)							)
        return bpe_tokens







      def      _lowercase						(					self	,   lowerCAmelCase__							)     ->		int:



        '''simple docstring'''




        return self.encoder.get(lowerCAmelCase__	,   self.encoder.get(self.unk_token							)							)







      def      _lowercase						(					self	,   lowerCAmelCase__							)     ->		int:



        '''simple docstring'''




        return self.decoder.get(lowerCAmelCase__							)







      def      _lowercase						(					self	,   lowerCAmelCase__							)     ->		str:



        '''simple docstring'''




        a__			:      Optional[int]            ="".join(lowerCAmelCase__							)
        a__			:      Tuple            =bytearray([self.byte_decoder[c] for c in text]							).decode("utf-8"	,   errors=self.errors							)
        return text







      def      _lowercase						(					self	,   lowerCAmelCase__	,   lowerCAmelCase__ = None							)     ->		Tuple[str]:



        '''simple docstring'''




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

        with open(lowerCAmelCase__	,   "w"	,   encoding="utf-8"							) as f:
          f.write(json.dumps(self.encoder	,   indent=2	,   sort_keys=lowerCAmelCase__	,   ensure_ascii=lowerCAmelCase__							) + "\n"							)

        a__			:      List[str]            =0
        with open(lowerCAmelCase__	,   "w"	,   encoding="utf-8"							) as writer:
          writer.write("#version: 0.2\n"							)
          for bpe_tokens, token_index in sorted(self.bpe_ranks.items()	,   key=lambda lowerCAmelCase__							: kv[1]							):
            if index != token_index:
              logger.warning(
                  F'''Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'''
                  " Please check that the tokenizer is not corrupted!"							)
              a__			:      Union[str, Any]            =token_index
            writer.write(" ".join(lowerCAmelCase__							) + "\n"							)
            index += 1

        return vocab_file, merge_file







      def      _lowercase						(					self	,   lowerCAmelCase__	,   lowerCAmelCase__=False	,   **lowerCAmelCase__							)     ->		Dict:



        '''simple docstring'''




        a__			:      List[Any]            =kwargs.pop("add_prefix_space"	,   self.add_prefix_space							)
        if (is_split_into_words or add_prefix_space) and (len(lowerCAmelCase__							) > 0 and not text[0].isspace()):
          a__			:      Tuple            =" " + text
        return (text, kwargs)







      def      _lowercase						(					self	,   lowerCAmelCase__							)     ->		List[int]:



        '''simple docstring'''




        a__			:      Dict            =[]
        for is_user, text in conversation.iter_texts():
          input_ids.extend(self.encode(lowerCAmelCase__	,   add_special_tokens=lowerCAmelCase__							) + [self.eos_token_id]							)
        if len(lowerCAmelCase__							) > self.model_max_length:
          a__			:      Tuple            =input_ids[-self.model_max_length :]
        return input_ids













import importlib
import sys
from argparse import REMAINDER, ArgumentParser
from pathlib import Path

import torch_xla.distributed.xla_multiprocessing as xmp
def   _A (	):





  """simple docstring"""



  a__			:      Optional[int]            =ArgumentParser(
      description=(
          "PyTorch TPU distributed training launch helper utility that will spawn up multiple distributed processes"
      )		)

  # Optional arguments for the launch helper
  parser.add_argument("--num_cores" , type=SCREAMING_SNAKE_CASE , default=1 , help="Number of TPU cores to use (1 or 8)."		)

  # positional
  parser.add_argument(
      "training_script" , type=SCREAMING_SNAKE_CASE , help=(
          "The full path to the single TPU training "
          "program/script to be launched in parallel, "
          "followed by all the arguments for the "
          "training script"
      ) , )

  # rest from the training program
  parser.add_argument("training_script_args" , nargs=SCREAMING_SNAKE_CASE		)

  return parser.parse_args()
def   _A (	):





  """simple docstring"""



  a__			:      Optional[Any]            =parse_args()

  # Import training_script as a module.
  a__			:      Tuple            =Path(args.training_script		)
  sys.path.append(str(script_fpath.parent.resolve()		)		)
  a__			:      Tuple            =script_fpath.stem
  a__			:      Any            =importlib.import_module(SCREAMING_SNAKE_CASE		)

  # Patch sys.argv
  a__			:      List[str]            =[args.training_script] + args.training_script_args + ["--tpu_num_cores", str(args.num_cores		)]

  xmp.spawn(mod._mp_fn , args=() , nprocs=args.num_cores		)


if __name__ == "__main__":
     main()






