


'''simple docstring'''






import json
import os
from functools import lru_cache
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       =							logging.get_logger(__name__)


__UpperCAmelCase       =							{
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
}

__UpperCAmelCase       =							{
    '''vocab_file''': {'''facebook/blenderbot-3B''': '''https://huggingface.co/facebook/blenderbot-3B/resolve/main/vocab.json'''},
    '''merges_file''': {'''facebook/blenderbot-3B''': '''https://huggingface.co/facebook/blenderbot-3B/resolve/main/merges.txt'''},
    '''tokenizer_config_file''': {
        '''facebook/blenderbot-3B''': '''https://huggingface.co/facebook/blenderbot-3B/resolve/main/tokenizer_config.json'''
    },
}

__UpperCAmelCase       =							{'''facebook/blenderbot-3B''': 128}


@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def  _snake_case       (					)						->   Any:
							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(lowerCAmelCase__	)
																					cs.append(2**8 + n	)
																					n += 1
							lowerCAmelCase__	       =     [chr(lowerCAmelCase__	) for n in cs]
							return dict(zip(lowerCAmelCase__			,   lowerCAmelCase__	)	)


def  _snake_case       (					A	)						->   Any:
							lowerCAmelCase__	       =     set()
							lowerCAmelCase__	       =     word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)	)
														lowerCAmelCase__	       =     char
							return pairs






class 		a__    (							__lowercase  ):





							'''simple docstring'''






							lowercase__      :       int     						=				VOCAB_FILES_NAMES
							lowercase__      :       Any     						=				PRETRAINED_VOCAB_FILES_MAP
							lowercase__      :       Optional[Any]     						=				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							lowercase__      :       Union[str, Any]     						=				["input_ids", "attention_mask"]







							def __init__(		self ,   lowerCamelCase_ ,   lowerCamelCase_ ,   lowerCamelCase_="replace" ,   lowerCamelCase_="<s>" ,   lowerCamelCase_="</s>" ,   lowerCamelCase_="</s>" ,   lowerCamelCase_="<s>" ,   lowerCamelCase_="<unk>" ,   lowerCamelCase_="<pad>" ,   lowerCamelCase_="<mask>" ,   lowerCamelCase_=False ,   **lowerCamelCase_ ,   )	->							Dict:
														lowerCAmelCase__	       =     AddedToken(snake_case_ ,   lstrip=snake_case_ ,   rstrip=snake_case_       ) if isinstance(snake_case_ ,   snake_case_       ) else bos_token
														lowerCAmelCase__	       =     AddedToken(snake_case_ ,   lstrip=snake_case_ ,   rstrip=snake_case_       ) if isinstance(snake_case_ ,   snake_case_       ) else eos_token
														lowerCAmelCase__	       =     AddedToken(snake_case_ ,   lstrip=snake_case_ ,   rstrip=snake_case_       ) if isinstance(snake_case_ ,   snake_case_       ) else sep_token
														lowerCAmelCase__	       =     AddedToken(snake_case_ ,   lstrip=snake_case_ ,   rstrip=snake_case_       ) if isinstance(snake_case_ ,   snake_case_       ) else cls_token
														lowerCAmelCase__	       =     AddedToken(snake_case_ ,   lstrip=snake_case_ ,   rstrip=snake_case_       ) if isinstance(snake_case_ ,   snake_case_       ) else unk_token
														lowerCAmelCase__	       =     AddedToken(snake_case_ ,   lstrip=snake_case_ ,   rstrip=snake_case_       ) if isinstance(snake_case_ ,   snake_case_       ) else pad_token

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

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

														with open(snake_case_ ,   encoding='''utf-8'''       ) as vocab_handle:
																					lowerCAmelCase__	       =     json.load(snake_case_       )
														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(snake_case_ ,   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(snake_case_ ,   range(len(snake_case_       )       )       )       )
														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
							# Copied from transformers.models.roberta.tokenization_roberta.RobertaTokenizer.vocab_size with Roberta->Blenderbot, RoBERTa->Blenderbot
							def   __SCREAMING_SNAKE_CASE							(		self       )	->							List[str]:
														return len(self.encoder       )







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







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_       )	->							Tuple:
														if token in self.cache:
																					return self.cache[token]
														lowerCAmelCase__	       =     tuple(snake_case_       )
														lowerCAmelCase__	       =     get_pairs(snake_case_       )

														if not pairs:
																					return token

														while True:
																					lowerCAmelCase__	       =     min(snake_case_ ,   key=lambda lowerCamelCase_       : self.bpe_ranks.get(snake_case_ ,   float('''inf'''       )       )       )
																					if bigram not in self.bpe_ranks:
																												break
																					lowerCAmelCase__	       =     bigram
																					lowerCAmelCase__	       =     []
																					lowerCAmelCase__	       =     0
																					while i < len(snake_case_       ):
																												try:
																																			lowerCAmelCase__	       =     word.index(snake_case_ ,   snake_case_       )
																												except ValueError:
																																			new_word.extend(word[i:]       )
																																			break
																												else:
																																			new_word.extend(word[i:j]       )
																																			lowerCAmelCase__	       =     j

																												if word[i] == first and i < len(snake_case_       ) - 1 and word[i + 1] == second:
																																			new_word.append(first + second       )
																																			i += 2
																												else:
																																			new_word.append(word[i]       )
																																			i += 1
																					lowerCAmelCase__	       =     tuple(snake_case_       )
																					lowerCAmelCase__	       =     new_word
																					if len(snake_case_       ) == 1:
																												break
																					else:
																												lowerCAmelCase__	       =     get_pairs(snake_case_       )
														lowerCAmelCase__	       =     ''' '''.join(snake_case_       )
														lowerCAmelCase__	       =     word
														return word







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_       )	->							Any:
														lowerCAmelCase__	       =     []
														for token in re.findall(self.pat ,   snake_case_       ):
																					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(snake_case_       ).split(''' '''       )       )
														return bpe_tokens







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_       )	->							int:
														return self.encoder.get(snake_case_ ,   self.encoder.get(self.unk_token       )       )







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_       )	->							Optional[Any]:
														return self.decoder.get(snake_case_       )







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_       )	->							str:
														lowerCAmelCase__	       =     ''''''.join(snake_case_       )
														lowerCAmelCase__	       =     bytearray([self.byte_decoder[c] for c in text]       ).decode('''utf-8''' ,   errors=self.errors       )
														return text







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_ = None       )	->							Any:
														if not os.path.isdir(snake_case_       ):
																					logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""       )
																					return
														lowerCAmelCase__	       =     os.path.join(
														    snake_case_ ,   (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']       )
														lowerCAmelCase__	       =     os.path.join(
														    snake_case_ ,   (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']       )

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

														lowerCAmelCase__	       =     0
														with open(snake_case_ ,   '''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!'''       )
																																			lowerCAmelCase__	       =     token_index
																												writer.write(''' '''.join(snake_case_       ) + '''\n'''       )
																												index += 1

														return vocab_file, merge_file







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_ = None ,   lowerCamelCase_ = False       )	->							List[str]:
														if already_has_special_tokens:
																					return super().get_special_tokens_mask(
																					    token_ids_a=snake_case_ ,   token_ids_a=snake_case_ ,   already_has_special_tokens=snake_case_       )

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







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_ = None       )	->							Tuple:
														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   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_=False ,   **lowerCamelCase_       )	->							Any:
														lowerCAmelCase__	       =     kwargs.pop('''add_prefix_space''' ,   self.add_prefix_space       )
														if (is_split_into_words or add_prefix_space) and (len(snake_case_       ) > 0 and not text[0].isspace()):
																					lowerCAmelCase__	       =     ''' ''' + text
														return (text, kwargs)







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_ = None       )	->							Union[str, Any]:
														return token_ids_a + [self.eos_token_id]







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_       )	->							Union[str, Any]:
														lowerCAmelCase__	       =     []
														for is_user, text in conversation.iter_texts():
																					if is_user:
																												# We need to space prefix as it's being done within blenderbot
																												inputs.append(''' ''' + text       )
																					else:
																												# Generated responses should contain them already.
																												inputs.append(snake_case_       )

														lowerCAmelCase__	       =     '''  '''.join(snake_case_       )
														lowerCAmelCase__	       =     self.encode(snake_case_       )
														if len(snake_case_       ) > self.model_max_length:
																					lowerCAmelCase__	       =     input_ids[-self.model_max_length :]
																					logger.warning(F"""Trimmed input from conversation as it was longer than {self.model_max_length} tokens."""       )
														return input_ids








'''simple docstring'''






import os
from typing import Optional

import fsspec
from fsspec.archive import AbstractArchiveFileSystem
from fsspec.utils import DEFAULT_BLOCK_SIZE




class 		a__    (							a__  ):





							'''simple docstring'''






							lowercase__      :       Dict     						=				""
							lowercase__      :       str     						=				(
							    None  # protocol passed in prefix to the url. ex: "gzip", for gzip://file.txt::http://foo.bar/file.txt.gz
							)
							lowercase__      :       str     						=				None  # compression type in fsspec. ex: "gzip"
							lowercase__      :       str     						=				None  # extension of the filename to strip. ex: "".gz" to get file.txt from file.txt.gz







							def __init__(		self ,   lowerCamelCase_ = "" ,   lowerCamelCase_ = None ,   lowerCamelCase_ = None ,   **lowerCamelCase_       )	->							Any:
														super().__init__(self ,   **lowerCamelCase_       )
														# always open as "rb" since fsspec can then use the TextIOWrapper to make it work for "r" mode
														lowerCAmelCase__	       =     fsspec.open(
														    lowerCamelCase_ ,   mode='''rb''' ,   protocol=lowerCamelCase_ ,   compression=self.compression ,   client_kwargs={
														        '''requote_redirect_url''': False,  # see https://github.com/huggingface/datasets/pull/5459
														        '''trust_env''': True,  # Enable reading proxy env variables.
														        **(target_options or {}).pop('''client_kwargs''' ,   {}       ),  # To avoid issues if it was already passed.
														    } ,   **(target_options or {}) ,   )
														lowerCAmelCase__	       =     os.path.basename(self.file.path.split('''::'''       )[0]       )
														lowerCAmelCase__	       =     (
														    self.compressed_name[: self.compressed_name.rindex('''.'''       )]
														    if '''.''' in self.compressed_name
														    else self.compressed_name
														)
														lowerCAmelCase__	       =     None







							@classmethod
							def   __SCREAMING_SNAKE_CASE							(		cls ,   lowerCamelCase_       )	->							Any:
														# compressed file paths are always relative to the archive root
														return super()._strip_protocol(lowerCamelCase_       ).lstrip('''/'''       )







							def   __SCREAMING_SNAKE_CASE							(		self       )	->							Optional[int]:
														if self.dir_cache is None:
																					lowerCAmelCase__	       =     {**self.file.fs.info(self.file.path       ), '''name''': self.uncompressed_name}
																					lowerCAmelCase__	       =     {f['''name''']: f}







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_       )	->							List[Any]:
														return self.file.open().read()







							def   __SCREAMING_SNAKE_CASE							(		self ,   lowerCamelCase_ ,   lowerCamelCase_ = "rb" ,   lowerCamelCase_=None ,   lowerCamelCase_=True ,   lowerCamelCase_=None ,   **lowerCamelCase_ ,   )	->							List[str]:
														lowerCAmelCase__	       =     self._strip_protocol(lowerCamelCase_       )
														if mode != "rb":
																					raise ValueError(F"""Tried to read with mode {mode} on file {self.file.path} opened with mode 'rb'"""       )
														return self.file.open()




class 		a__    (							a__  ):





							'''simple docstring'''






							lowercase__      :       Optional[Any]     						=				"bz2"
							lowercase__      :       str     						=				"bz2"
							lowercase__      :       Optional[int]     						=				".bz2"




class 		a__    (							a__  ):





							'''simple docstring'''






							lowercase__      :       Any     						=				"gzip"
							lowercase__      :       int     						=				"gzip"
							lowercase__      :       int     						=				".gz"




class 		a__    (							a__  ):





							'''simple docstring'''






							lowercase__      :       Tuple     						=				"lz4"
							lowercase__      :       Optional[Any]     						=				"lz4"
							lowercase__      :       int     						=				".lz4"




class 		a__    (							a__  ):





							'''simple docstring'''






							lowercase__      :       Optional[int]     						=				"xz"
							lowercase__      :       str     						=				"xz"
							lowercase__      :       List[Any]     						=				".xz"




class 		a__    (							a__  ):





							'''simple docstring'''






							lowercase__      :       Union[str, Any]     						=				"zstd"
							lowercase__      :       Union[str, Any]     						=				"zstd"
							lowercase__      :       Dict     						=				".zst"







							def __init__(		self ,   lowerCamelCase_ ,   lowerCamelCase_ = "rb" ,   lowerCamelCase_ = None ,   lowerCamelCase_ = None ,   lowerCamelCase_ = DEFAULT_BLOCK_SIZE ,   **lowerCamelCase_ ,   )	->							List[Any]:
														super().__init__(
														    fo=lowerCamelCase_ ,   mode=lowerCamelCase_ ,   target_protocol=lowerCamelCase_ ,   target_options=lowerCamelCase_ ,   block_size=lowerCamelCase_ ,   **lowerCamelCase_ ,   )
														# We need to wrap the zstd decompressor to avoid this error in fsspec==2021.7.0 and zstandard==0.15.2:
														#
														# File "/Users/user/.virtualenvs/hf-datasets/lib/python3.7/site-packages/fsspec/core.py", line 145, in open
														#     out.close = close
														# AttributeError: 'zstd.ZstdDecompressionReader' object attribute 'close' is read-only
														#
														# see https://github.com/intake/filesystem_spec/issues/725
														lowerCAmelCase__	       =     self.file.__enter__




														class 		a__    :





																					'''simple docstring'''






																					def __init__(		self ,   lowerCamelCase_       )	->							List[str]:
																												lowerCAmelCase__	       =     file_







																					def __enter__(		self       )	->							Tuple:
																												self._file.__enter__()
																												return self







																					def __exit__(		self ,   *lowerCamelCase_ ,   **lowerCamelCase_       )	->							str:
																												self._file.__exit__(*lowerCamelCase_ ,   **lowerCamelCase_       )







																					def __iter__(		self       )	->							Any:
																												return iter(self._file       )







																					def   __SCREAMING_SNAKE_CASE							(		self       )	->							Union[str, Any]:
																												return next(self._file       )







																					def __getattr__(		self ,   lowerCamelCase_       )	->							str:
																												return getattr(self._file ,   lowerCamelCase_       )

														def fixed_enter(*lowerCamelCase_ ,   **lowerCamelCase_       ):
																					return WrappedFile(_enter(*lowerCamelCase_ ,   **lowerCamelCase_       )       )

														lowerCAmelCase__	       =     fixed_enter