

'''simple docstring'''






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

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


lowerCAmelCase__											=	logging.get_logger(__name__)


lowerCAmelCase__											=	{'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt'''}

# See all LED models at https://huggingface.co/models?filter=LED
lowerCAmelCase__											=	{
    '''vocab_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json''',
    },
    '''merges_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt''',
    },
    '''tokenizer_file''': {
        '''allenai/led-base-16384''': '''https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json''',
    },
}

lowerCAmelCase__											=	{
    '''allenai/led-base-16384''': 1_6384,
}


@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def 		_A     (    ):






 """simple docstring"""



 __lowercase							  = (
     list(range(ord('''!'''      )     ,				ord('''~'''      ) + 1      )      ) + list(range(ord('''¡'''      )     ,				ord('''¬'''      ) + 1      )      ) + list(range(ord('''®'''      )     ,				ord('''ÿ'''      ) + 1      )      )
 )
 __lowercase							  = bs[:]
 __lowercase							  = 0
 for b in range(2**8      ):
  if b not in bs:
   bs.append(A__      )
   cs.append(2**8 + n      )
   n += 1
 __lowercase							  = [chr(A__      ) for n in cs]
 return dict(zip(A__     ,				A__      )      )


def 		_A     (    A__      ):






 """simple docstring"""



 __lowercase							  = set()
 __lowercase							  = word[0]
 for char in word[1:]:
  pairs.add((prev_char, char)      )
  __lowercase							  = char
 return pairs






class 			lowercase_	(lowerCamelCase__      ):




  """simple docstring"""




  SCREAMING_SNAKE_CASE    :  int							=			VOCAB_FILES_NAMES
  SCREAMING_SNAKE_CASE    :  Optional[int]							=			PRETRAINED_VOCAB_FILES_MAP
  SCREAMING_SNAKE_CASE    :  Union[str, Any]							=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  SCREAMING_SNAKE_CASE    :  Union[str, Any]							=			['input_ids', 'attention_mask']






  def __init__(      self   :     Dict					,lowercase__   :     Tuple					,lowercase__   :     Dict					,lowercase__   :     Union[str, Any]="replace"					,lowercase__   :     Any="<s>"					,lowercase__   :     Dict="</s>"					,lowercase__   :     Union[str, Any]="</s>"					,lowercase__   :     Dict="<s>"					,lowercase__   :     Dict="<unk>"					,lowercase__   :     Union[str, Any]="<pad>"					,lowercase__   :     Tuple="<mask>"					,lowercase__   :     List[str]=False					,**lowercase__   :     List[Any]					,):
   __lowercase							  = AddedToken(lowercase__					,lstrip=lowercase__					,rstrip=lowercase__			) if isinstance(lowercase__					,lowercase__			) else bos_token
   __lowercase							  = AddedToken(lowercase__					,lstrip=lowercase__					,rstrip=lowercase__			) if isinstance(lowercase__					,lowercase__			) else eos_token
   __lowercase							  = AddedToken(lowercase__					,lstrip=lowercase__					,rstrip=lowercase__			) if isinstance(lowercase__					,lowercase__			) else sep_token
   __lowercase							  = AddedToken(lowercase__					,lstrip=lowercase__					,rstrip=lowercase__			) if isinstance(lowercase__					,lowercase__			) else cls_token
   __lowercase							  = AddedToken(lowercase__					,lstrip=lowercase__					,rstrip=lowercase__			) if isinstance(lowercase__					,lowercase__			) else unk_token
   __lowercase							  = 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
   __lowercase							  = 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:
    __lowercase							  = json.load(lowercase__			)
   __lowercase							  = {v: k for k, v in self.encoder.items()}
   __lowercase							  = errors  # how to handle errors in decoding
   __lowercase							  = bytes_to_unicode()
   __lowercase							  = {v: k for k, v in self.byte_encoder.items()}
   with open(lowercase__					,encoding='''utf-8'''			) as merges_handle:
    __lowercase							  = merges_handle.read().split('''\n'''			)[1:-1]
   __lowercase							  = [tuple(merge.split()			) for merge in bpe_merges]
   __lowercase							  = dict(zip(lowercase__					,range(len(lowercase__			)			)			)			)
   __lowercase							  = {}
   __lowercase							  = add_prefix_space

   # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
   __lowercase							  = 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.bart.tokenization_bart.BartTokenizer.vocab_size
  def 					SCREAMING_SNAKE_CASE      (      self   :     Union[str, Any]			):
   return len(self.encoder			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     List[str]			):
   return dict(self.encoder					,**self.added_tokens_encoder			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[Any]					,lowercase__   :     Optional[Any]			):
   if token in self.cache:
    return self.cache[token]
   __lowercase							  = tuple(lowercase__			)
   __lowercase							  = get_pairs(lowercase__			)

   if not pairs:
    return token

   while True:
    __lowercase							  = min(lowercase__					,key=lambda lowercase__			: self.bpe_ranks.get(lowercase__					,float('''inf'''			)			)			)
    if bigram not in self.bpe_ranks:
     break
    __lowercase  ,	__lowercase							  = bigram
    __lowercase							  = []
    __lowercase							  = 0
    while i < len(lowercase__			):
     try:
      __lowercase							  = word.index(lowercase__					,lowercase__			)
     except ValueError:
      new_word.extend(word[i:]			)
      break
     else:
      new_word.extend(word[i:j]			)
      __lowercase							  = 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
    __lowercase							  = tuple(lowercase__			)
    __lowercase							  = new_word
    if len(lowercase__			) == 1:
     break
    else:
     __lowercase							  = get_pairs(lowercase__			)
   __lowercase							  = ''' '''.join(lowercase__			)
   __lowercase							  = word
   return word






  def 					SCREAMING_SNAKE_CASE      (      self   :     List[Any]					,lowercase__   :     List[Any]			):
   __lowercase							  = []
   for token in re.findall(self.pat					,lowercase__			):
    __lowercase							  = ''''''.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 					SCREAMING_SNAKE_CASE      (      self   :     str					,lowercase__   :     Tuple			):
   return self.encoder.get(lowercase__					,self.encoder.get(self.unk_token			)			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Union[str, Any]					,lowercase__   :     Any			):
   return self.decoder.get(lowercase__			)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Optional[int]					,lowercase__   :     Optional[Any]			):
   __lowercase							  = ''''''.join(lowercase__			)
   __lowercase							  = bytearray([self.byte_decoder[c] for c in text]			).decode('''utf-8'''					,errors=self.errors			)
   return text






  def 					SCREAMING_SNAKE_CASE      (      self   :     List[Any]					,lowercase__   :     str					,lowercase__   :     Optional[str] = None			):
   if not os.path.isdir(lowercase__			):
    logger.error(F"Vocabulary path ({save_directory}) should be a directory"			)
    return
   __lowercase							  = os.path.join(
       lowercase__					,(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']			)
   __lowercase							  = 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'''			)

   __lowercase							  = 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!'''			)
      __lowercase							  = token_index
     writer.write(''' '''.join(lowercase__			) + '''\n'''			)
     index += 1

   return vocab_file, merge_file






  def 					SCREAMING_SNAKE_CASE      (      self   :     Any					,lowercase__   :     List[int]					,lowercase__   :     Optional[List[int]] = None			):
   if token_ids_a is None:
    return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
   __lowercase							  = [self.cls_token_id]
   __lowercase							  = [self.sep_token_id]
   return cls + token_ids_a + sep + sep + token_ids_a + sep






  def 					SCREAMING_SNAKE_CASE      (      self   :     Union[str, Any]					,lowercase__   :     List[int]					,lowercase__   :     Optional[List[int]] = None					,lowercase__   :     bool = False			):
   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 					SCREAMING_SNAKE_CASE      (      self   :     Union[str, Any]					,lowercase__   :     List[int]					,lowercase__   :     Optional[List[int]] = None			):
   __lowercase							  = [self.sep_token_id]
   __lowercase							  = [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 					SCREAMING_SNAKE_CASE      (      self   :     int					,lowercase__   :     Optional[int]					,lowercase__   :     Optional[Any]=False					,**lowercase__   :     Optional[int]			):
   __lowercase							  = 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()):
    __lowercase							  = ''' ''' + text
   return (text, kwargs)






  def 					SCREAMING_SNAKE_CASE      (      self   :     Any					,lowercase__   :     Union[Dict[str, EncodedInput], BatchEncoding]					,lowercase__   :     Optional[int] = None					,lowercase__   :     PaddingStrategy = PaddingStrategy.DO_NOT_PAD					,lowercase__   :     Optional[int] = None					,lowercase__   :     Optional[bool] = None					,):
   __lowercase							  = super()._pad(
       encoded_inputs=lowercase__					,max_length=lowercase__					,padding_strategy=lowercase__					,pad_to_multiple_of=lowercase__					,return_attention_mask=lowercase__					,)

   # Load from model defaults
   if return_attention_mask is None:
    __lowercase							  = '''attention_mask''' in self.model_input_names

   if return_attention_mask and "global_attention_mask" in encoded_inputs:
    __lowercase							  = encoded_inputs[self.model_input_names[0]]
    # `global_attention_mask` need to have the same length as other (sequential) inputs.
    __lowercase							  = len(encoded_inputs['''global_attention_mask''']			) != len(lowercase__			)

    if needs_to_be_padded:
     __lowercase							  = len(lowercase__			) - len(encoded_inputs['''global_attention_mask''']			)

     if self.padding_side == "right":
      # Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
      __lowercase							  = (
          encoded_inputs['''global_attention_mask'''] + [-1] * difference
      )
     elif self.padding_side == "left":
      __lowercase							  = [-1] * difference + encoded_inputs[
          '''global_attention_mask'''
      ]
     else:
      raise ValueError('''Invalid padding strategy:''' + str(self.padding_side			)			)

   return encoded_inputs




'''simple docstring'''






import pytest
import requests

from datasets.utils.file_utils import http_head

from .utils import OfflineSimulationMode, RequestWouldHangIndefinitelyError, offline


@pytest.mark.integration
def 		_A     (    ):






 """simple docstring"""



 with offline(OfflineSimulationMode.CONNECTION_TIMES_OUT      ):
  with pytest.raises(A__      ):
   requests.request('''GET'''     ,				'''https://huggingface.co'''      )
  with pytest.raises(requests.exceptions.ConnectTimeout      ):
   requests.request('''GET'''     ,				'''https://huggingface.co'''     ,				timeout=1.0      )


@pytest.mark.integration
def 		_A     (    ):






 """simple docstring"""



 with offline(OfflineSimulationMode.CONNECTION_FAILS      ):
  with pytest.raises(requests.exceptions.ConnectionError      ):
   requests.request('''GET'''     ,				'''https://huggingface.co'''      )






def 		_A     (    ):






 """simple docstring"""



 with offline(OfflineSimulationMode.HF_DATASETS_OFFLINE_SET_TO_1      ):
  with pytest.raises(A__      ):
   http_head('''https://huggingface.co'''      )

