





import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

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


lowercase				:	str      						=  logging.get_logger(__name__)


lowercase				:	Any      						=  {"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt"""}

lowercase				:	List[Any]      						=  {
    """vocab_file""": {
        """allenai/longformer-base-4096""": """https://huggingface.co/allenai/longformer-base-4096/resolve/main/vocab.json""",
        """allenai/longformer-large-4096""": (
            """https://huggingface.co/allenai/longformer-large-4096/resolve/main/vocab.json"""
        ),
        """allenai/longformer-large-4096-finetuned-triviaqa""": (
            """https://huggingface.co/allenai/longformer-large-4096-finetuned-triviaqa/resolve/main/vocab.json"""
        ),
        """allenai/longformer-base-4096-extra.pos.embd.only""": (
            """https://huggingface.co/allenai/longformer-base-4096-extra.pos.embd.only/resolve/main/vocab.json"""
        ),
        """allenai/longformer-large-4096-extra.pos.embd.only""": (
            """https://huggingface.co/allenai/longformer-large-4096-extra.pos.embd.only/resolve/main/vocab.json"""
        ),
    },
    """merges_file""": {
        """allenai/longformer-base-4096""": """https://huggingface.co/allenai/longformer-base-4096/resolve/main/merges.txt""",
        """allenai/longformer-large-4096""": (
            """https://huggingface.co/allenai/longformer-large-4096/resolve/main/merges.txt"""
        ),
        """allenai/longformer-large-4096-finetuned-triviaqa""": (
            """https://huggingface.co/allenai/longformer-large-4096-finetuned-triviaqa/resolve/main/merges.txt"""
        ),
        """allenai/longformer-base-4096-extra.pos.embd.only""": (
            """https://huggingface.co/allenai/longformer-base-4096-extra.pos.embd.only/resolve/main/merges.txt"""
        ),
        """allenai/longformer-large-4096-extra.pos.embd.only""": (
            """https://huggingface.co/allenai/longformer-large-4096-extra.pos.embd.only/resolve/main/merges.txt"""
        ),
    },
}

lowercase				:	List[str]      						=  {
    """allenai/longformer-base-4096""": 4_0_9_6,
    """allenai/longformer-large-4096""": 4_0_9_6,
    """allenai/longformer-large-4096-finetuned-triviaqa""": 4_0_9_6,
    """allenai/longformer-base-4096-extra.pos.embd.only""": 4_0_9_6,
    """allenai/longformer-large-4096-extra.pos.embd.only""": 4_0_9_6,
}



@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def A_     (		)  ->		Dict:
       a__						:     int			=    (
           list(range(ord('!'    )		,				ord('~'    ) + 1    )    ) + list(range(ord('¡'    )		,				ord('¬'    ) + 1    )    ) + list(range(ord('®'    )		,				ord('ÿ'    ) + 1    )    )
       )
       a__						:     Dict			=    bs[:]
       a__						:     str			=    0
       for b in range(2**8    ):
              if b not in bs:
                     bs.append(A__    )
                     cs.append(2**8 + n    )
                     n += 1
       a__						:     int			=    [chr(A__    ) for n in cs]
       return dict(zip(A__		,				A__    )    )



def A_     (		A__    )  ->		Union[str, Any]:
       a__						:     Any			=    set()
       a__						:     List[str]			=    word[0]
       for char in word[1:]:
              pairs.add((prev_char, char)    )
              a__						:     Optional[Any]			=    char
       return pairs





class 			A__   (							__UpperCAmelCase     ):




   """simple docstring"""



   __A       :    List[Any]	  =  VOCAB_FILES_NAMES
   __A       :    Tuple	  =  PRETRAINED_VOCAB_FILES_MAP
   __A       :    Any	  =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
   __A       :    List[Any]	  =  ['''input_ids''', '''attention_mask''']

   def __init__( self ,  lowercase ,  lowercase ,  lowercase="replace" ,  lowercase="<s>" ,  lowercase="</s>" ,  lowercase="</s>" ,  lowercase="<s>" ,  lowercase="<unk>" ,  lowercase="<pad>" ,  lowercase="<mask>" ,  lowercase=False ,  **lowercase ,  )      ->						Union[str, Any]:







          '''simple docstring'''

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

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

          super().__init__(
              errors=lowercase ,  bos_token=lowercase ,  eos_token=lowercase ,  unk_token=lowercase ,  sep_token=lowercase ,  cls_token=lowercase ,  pad_token=lowercase ,  mask_token=lowercase ,  add_prefix_space=lowercase ,  **lowercase ,  )

          with open(lowercase ,  encoding='utf-8') as vocab_handle:
                 a__						:     int			=    json.load(lowercase)
          a__						:     List[str]			=    {v: k for k, v in self.encoder.items()}
          a__						:     List[Any]			=    errors  # how to handle errors in decoding
          a__						:     Tuple			=    bytes_to_unicode()
          a__						:     str			=    {v: k for k, v in self.byte_encoder.items()}
          with open(lowercase ,  encoding='utf-8') as merges_handle:
                 a__						:     List[str]			=    merges_handle.read().split('\n')[1:-1]
          a__						:     List[str]			=    [tuple(merge.split()) for merge in bpe_merges]
          a__						:     Optional[int]			=    dict(zip(lowercase ,  range(len(lowercase))))
          a__						:     List[str]			=    {}
          a__						:     List[Any]			=    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
   def 							__lowercase		( self)      ->						List[str]:







          '''simple docstring'''

          return len(self.encoder)

   def 							__lowercase		( self)      ->						List[str]:







          '''simple docstring'''

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

   def 							__lowercase		( self ,  lowercase)      ->						Optional[Any]:







          '''simple docstring'''

          if token in self.cache:
                 return self.cache[token]
          a__						:     Any			=    tuple(lowercase)
          a__						:     Union[str, Any]			=    get_pairs(lowercase)

          if not pairs:
                 return token

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

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

   def 							__lowercase		( self ,  lowercase)      ->						Tuple:







          '''simple docstring'''

          a__						:     Union[str, Any]			=    []
          for token in re.findall(self.pat ,  lowercase):
                 a__						:     Union[str, 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(lowercase).split(' '))
          return bpe_tokens

   def 							__lowercase		( self ,  lowercase)      ->						List[Any]:







          '''simple docstring'''

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

   def 							__lowercase		( self ,  lowercase)      ->						Union[str, Any]:







          '''simple docstring'''

          return self.decoder.get(lowercase)

   def 							__lowercase		( self ,  lowercase)      ->						Dict:







          '''simple docstring'''

          a__						:     int			=    ''.join(lowercase)
          a__						:     Union[str, Any]			=    bytearray([self.byte_decoder[c] for c in text]).decode('utf-8' ,  errors=self.errors)
          return text

   def 							__lowercase		( self ,  lowercase ,  lowercase = None)      ->						Tuple[str]:







          '''simple docstring'''

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

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

          a__						:     str			=    0
          with open(lowercase ,  '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 lowercase: 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__						:     Dict			=    token_index
                        writer.write(' '.join(lowercase) + '\n')
                        index += 1

          return vocab_file, merge_file

   def 							__lowercase		( self ,  lowercase ,  lowercase = None)      ->						List[int]:







          '''simple docstring'''

          if token_ids_a is None:
                 return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
          a__						:     Dict			=    [self.cls_token_id]
          a__						:     Optional[Any]			=    [self.sep_token_id]
          return cls + token_ids_a + sep + sep + token_ids_a + sep

   def 							__lowercase		( self ,  lowercase ,  lowercase = None ,  lowercase = False)      ->						List[int]:







          '''simple docstring'''

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

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

   def 							__lowercase		( self ,  lowercase ,  lowercase = None)      ->						List[int]:







          '''simple docstring'''

          a__						:     Dict			=    [self.sep_token_id]
          a__						:     Dict			=    [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 + sep + token_ids_a + sep) * [0]

   def 							__lowercase		( self ,  lowercase ,  lowercase=False ,  **lowercase)      ->						Union[str, Any]:







          '''simple docstring'''

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









import os
import socket
from contextlib import contextmanager

import torch

from ..commands.config.default import write_basic_config  # noqa: F401
from ..state import PartialState
from .dataclasses import DistributedType
from .imports import is_deepspeed_available, is_tpu_available
from .transformer_engine import convert_model
from .versions import is_torch_version


if is_deepspeed_available():
    from deepspeed import DeepSpeedEngine

if is_tpu_available(check_device=False):
    import torch_xla.core.xla_model as xm



def A_     (		A__    )  ->		Optional[int]:
       if is_torch_version('<'		,				'2.0.0'    ) or not hasattr(A__		,				'_dynamo'    ):
              return False
       return isinstance(A__		,				torch._dynamo.eval_frame.OptimizedModule    )



def A_     (		A__		,				A__ = True    )  ->		int:
       a__						:     Optional[Any]			=    (torch.nn.parallel.DistributedDataParallel, torch.nn.DataParallel)

       a__						:     Union[str, Any]			=    is_compiled_module(A__    )
       if is_compiled:
              a__						:     List[str]			=    model
              a__						:     Dict			=    model._orig_mod

       if is_deepspeed_available():
              options += (DeepSpeedEngine,)

       while isinstance(A__		,				A__    ):
              a__						:     str			=    model.module

       if not keep_fpaa_wrapper:
              a__						:     Union[str, Any]			=    getattr(A__		,				'forward'    )
              a__						:     List[Any]			=    model.__dict__.pop('_original_forward'		,				A__    )
              if original_forward is not None:
                     while hasattr(A__		,				'__wrapped__'    ):
                            a__						:     int			=    forward.__wrapped__
                            if forward == original_forward:
                                   break
                     a__						:     List[Any]			=    forward
              if getattr(A__		,				'_converted_to_transformer_engine'		,				A__    ):
                     convert_model(A__		,				to_transformer_engine=A__    )

       if is_compiled:
              a__						:     List[str]			=    model
              a__						:     Any			=    compiled_model

       return model



def A_     (		)  ->		int:
       PartialState().wait_for_everyone()



def A_     (		A__		,				A__    )  ->		Dict:
       if PartialState().distributed_type == DistributedType.TPU:
              xm.save(A__		,				A__    )
       elif PartialState().local_process_index == 0:
              torch.save(A__		,				A__    )



@contextmanager
def A_     (		**A__    )  ->		Any:
       for key, value in kwargs.items():
              a__						:     Optional[int]			=    str(A__    )

       yield

       for key in kwargs:
              if key.upper() in os.environ:
                     del os.environ[key.upper()]



def A_     (		A__    )  ->		List[str]:
       if not hasattr(A__		,				'__qualname__'    ) and not hasattr(A__		,				'__name__'    ):
              a__						:     Dict			=    getattr(A__		,				'__class__'		,				A__    )
       if hasattr(A__		,				'__qualname__'    ):
              return obj.__qualname__
       if hasattr(A__		,				'__name__'    ):
              return obj.__name__
       return str(A__    )



def A_     (		A__		,				A__    )  ->		Dict:
       for key, value in source.items():
              if isinstance(A__		,				A__    ):
                     a__						:     Optional[Any]			=    destination.setdefault(A__		,				{}    )
                     merge_dicts(A__		,				A__    )
              else:
                     a__						:     Optional[int]			=    value

       return destination



def A_     (		A__ = None    )  ->		bool:
       if port is None:
              a__						:     List[Any]			=    2_9500
       with socket.socket(socket.AF_INET		,				socket.SOCK_STREAM    ) as s:
              return s.connect_ex(('localhost', port)    ) == 0


