





'''simple docstring'''






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


A__				:   Optional[int]      =logging.get_logger(__name__)


A__				:   List[str]      ={'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt'''}

A__				:   int      ={
    '''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'''
        ),
    },
}

A__				:   Optional[Any]      ={
    '''allenai/longformer-base-4096''': 40_96,
    '''allenai/longformer-large-4096''': 40_96,
    '''allenai/longformer-large-4096-finetuned-triviaqa''': 40_96,
    '''allenai/longformer-base-4096-extra.pos.embd.only''': 40_96,
    '''allenai/longformer-large-4096-extra.pos.embd.only''': 40_96,
}
@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def    UpperCamelCase__      (		):





      """simple docstring"""


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





      """simple docstring"""


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





class  UpperCAmelCase     (			a__   ):
      _lowercase:			Optional[Any]						     =   VOCAB_FILES_NAMES
      _lowercase:			Dict						     =   PRETRAINED_VOCAB_FILES_MAP
      _lowercase:			Any						     =   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _lowercase:			Optional[Any]						     =   ["""input_ids""", """attention_mask"""]




      def __init__(			self			: Optional[int]			,			__snake_case			: int			,			__snake_case			: List[str]			,			__snake_case			: Dict="replace"			,			__snake_case			: str="<s>"			,			__snake_case			: Union[str, Any]="</s>"			,			__snake_case			: Optional[Any]="</s>"			,			__snake_case			: Tuple="<s>"			,			__snake_case			: Dict="<unk>"			,			__snake_case			: int="<pad>"			,			__snake_case			: int="<mask>"			,			__snake_case			: int=False			,			**__snake_case			: str			,			)				->      str:
            _lowerCAmelCase							     =							AddedToken(lowerCAmelCase__			,			lstrip=lowerCAmelCase__			,			rstrip=lowerCAmelCase__  ) if isinstance(lowerCAmelCase__			,			lowerCAmelCase__  ) else bos_token
            _lowerCAmelCase							     =							AddedToken(lowerCAmelCase__			,			lstrip=lowerCAmelCase__			,			rstrip=lowerCAmelCase__  ) if isinstance(lowerCAmelCase__			,			lowerCAmelCase__  ) else eos_token
            _lowerCAmelCase							     =							AddedToken(lowerCAmelCase__			,			lstrip=lowerCAmelCase__			,			rstrip=lowerCAmelCase__  ) if isinstance(lowerCAmelCase__			,			lowerCAmelCase__  ) else sep_token
            _lowerCAmelCase							     =							AddedToken(lowerCAmelCase__			,			lstrip=lowerCAmelCase__			,			rstrip=lowerCAmelCase__  ) if isinstance(lowerCAmelCase__			,			lowerCAmelCase__  ) else cls_token
            _lowerCAmelCase							     =							AddedToken(lowerCAmelCase__			,			lstrip=lowerCAmelCase__			,			rstrip=lowerCAmelCase__  ) if isinstance(lowerCAmelCase__			,			lowerCAmelCase__  ) else unk_token
            _lowerCAmelCase							     =							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
            _lowerCAmelCase							     =							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__			,			**lowerCAmelCase__			,			)

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

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




      def 				lowercase__    (			self			: Tuple  )				->      str:
            return dict(self.encoder			,			**self.added_tokens_encoder  )




      def 				lowercase__    (			self			: Union[str, Any]			,			__snake_case			: str  )				->      List[str]:
            if token in self.cache:
                  return self.cache[token]
            _lowerCAmelCase							     =							tuple(lowerCAmelCase__  )
            _lowerCAmelCase							     =							get_pairs(lowerCAmelCase__  )

            if not pairs:
                  return token

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




      def 				lowercase__    (			self			: Any			,			__snake_case			: Optional[int]  )				->      Tuple:
            _lowerCAmelCase							     =							[]
            for token in re.findall(self.pat			,			lowerCAmelCase__  ):
                  _lowerCAmelCase							     =							"".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			: Tuple			,			__snake_case			: List[Any]  )				->      Tuple:
            return self.encoder.get(lowerCAmelCase__			,			self.encoder.get(self.unk_token  )  )




      def 				lowercase__    (			self			: Any			,			__snake_case			: Optional[int]  )				->      Dict:
            return self.decoder.get(lowerCAmelCase__  )




      def 				lowercase__    (			self			: Dict			,			__snake_case			: Optional[Any]  )				->      Union[str, Any]:
            _lowerCAmelCase							     =							"".join(lowerCAmelCase__  )
            _lowerCAmelCase							     =							bytearray([self.byte_decoder[c] for c in text]  ).decode("""utf-8"""			,			errors=self.errors  )
            return text




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

            _lowerCAmelCase							     =							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 __snake_case  : 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!"""  )
                              _lowerCAmelCase							     =							token_index
                        writer.write(""" """.join(lowerCAmelCase__  ) + """\n"""  )
                        index += 1

            return vocab_file, merge_file




      def 				lowercase__    (			self			: Optional[Any]			,			__snake_case			: List[str]			,			__snake_case			: str = None  )				->      List[int]:
            if token_ids_a is None:
                  return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
            _lowerCAmelCase							     =							[self.cls_token_id]
            _lowerCAmelCase							     =							[self.sep_token_id]
            return cls + token_ids_a + sep + sep + token_ids_a + sep




      def 				lowercase__    (			self			: Dict			,			__snake_case			: Optional[int]			,			__snake_case			: str = None			,			__snake_case			: Optional[int] = False  )				->      List[int]:
            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, 1] + ([0] * len(lowerCAmelCase__  )) + [1]




      def 				lowercase__    (			self			: List[Any]			,			__snake_case			: List[Any]			,			__snake_case			: Dict = None  )				->      List[int]:
            _lowerCAmelCase							     =							[self.sep_token_id]
            _lowerCAmelCase							     =							[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			: Dict			,			__snake_case			: List[str]			,			__snake_case			: Optional[int]=False			,			**__snake_case			: str  )				->      Optional[Any]:
            _lowerCAmelCase							     =							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()):
                  _lowerCAmelCase							     =							" " + text
            return (text, kwargs)













"""simple docstring"""




from unittest.mock import Mock, patch

from file_transfer.send_file import send_file






@patch("socket.socket"					)
@patch("builtins.open"					)
def       _SCREAMING_SNAKE_CASE		(	_lowercase						:    List[Any]    ,       _lowercase						:    int					) ->str:






	'''simple docstring'''






	a      :		Optional[Any]	=				Mock()
	a      :		Dict	=				conn, Mock()
	a      :		Union[str, Any]	=				iter([1, None]					)
	a      :		Optional[int]	=				lambda _lowercase					: next(_lowercase					)

	# ===== invoke =====
	send_file(filename="mytext.txt"    ,       testing=_lowercase					)

	# ===== ensurance =====
	sock.assert_called_once()
	sock.return_value.bind.assert_called_once()
	sock.return_value.listen.assert_called_once()
	sock.return_value.accept.assert_called_once()
	conn.recv.assert_called_once()

	file.return_value.__enter__.assert_called_once()
	file.return_value.__enter__.return_value.read.assert_called()

	conn.send.assert_called_once()
	conn.close.assert_called_once()
	sock.return_value.shutdown.assert_called_once()
	sock.return_value.close.assert_called_once()


