'''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	: List[Any]    		=  logging.get_logger(__name__)


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

__UpperCamelCase	: Optional[int]    		=  {
    '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	: str    		=  {'facebook/blenderbot-3B': 128}
@lru_cache()
# Copied from transformers.models.roberta.tokenization_roberta.bytes_to_unicode
def    A			(					):
						SCREAMING_SNAKE_CASE							:					Optional[Any]           =							(
						    list(range(ord('''!'''		)	,						ord('''~'''		) + 1		)		) + list(range(ord('''¡'''		)	,						ord('''¬'''		) + 1		)		) + list(range(ord('''®'''		)	,						ord('''ÿ'''		) + 1		)		)
						)
						SCREAMING_SNAKE_CASE							:					Optional[Any]           =							bs[:]
						SCREAMING_SNAKE_CASE							:					int           =							0
						for b in range(2**8		):
												if b not in bs:
																		bs.append(_lowercase		)
																		cs.append(2**8 + n		)
																		n += 1
						SCREAMING_SNAKE_CASE							:					Optional[Any]           =							[chr(_lowercase		) for n in cs]
						return dict(zip(_lowercase	,						_lowercase		)		)
def    A			(					_lowercase		):
						SCREAMING_SNAKE_CASE							:					Optional[Any]           =							set()
						SCREAMING_SNAKE_CASE							:					int           =							word[0]
						for char in word[1:]:
												pairs.add((prev_char, char)		)
												SCREAMING_SNAKE_CASE							:					int           =							char
						return pairs





class       lowercase__		(     UpperCamelCase_):
						UpperCamelCase_  =	VOCAB_FILES_NAMES
						UpperCamelCase_  =	PRETRAINED_VOCAB_FILES_MAP
						UpperCamelCase_  =	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						UpperCamelCase_  =	["""input_ids""", """attention_mask"""]


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





												'''simple docstring'''





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

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

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

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

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





												'''simple docstring'''





												return len(self.encoder		)


						def  __A ( self			:       Dict		):





												'''simple docstring'''





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


						def  __A ( self			:       int					,	UpperCamelCase__			:       str		):





												'''simple docstring'''





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

												if not pairs:
																		return token

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

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


						def  __A ( self			:       List[Any]					,	UpperCamelCase__			:       Optional[Any]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					List[Any]           =							[]
												for token in re.findall(self.pat					,	UpperCamelCase__		):
																		SCREAMING_SNAKE_CASE							:					Dict           =							''''''.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(UpperCamelCase__		).split(''' '''		)		)
												return bpe_tokens


						def  __A ( self			:       Dict					,	UpperCamelCase__			:       Union[str, Any]		):





												'''simple docstring'''





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


						def  __A ( self			:       Optional[int]					,	UpperCamelCase__			:       str		):





												'''simple docstring'''





												return self.decoder.get(UpperCamelCase__		)


						def  __A ( self			:       Optional[int]					,	UpperCamelCase__			:       int		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Optional[Any]           =							''''''.join(UpperCamelCase__		)
												SCREAMING_SNAKE_CASE							:					List[str]           =							bytearray([self.byte_decoder[c] for c in text]		).decode('''utf-8'''					,	errors=self.errors		)
												return text


						def  __A ( self			:       List[str]					,	UpperCamelCase__			:       str					,	UpperCamelCase__			:       Optional[str] = None		):





												'''simple docstring'''





												if not os.path.isdir(UpperCamelCase__		):
																		logger.error(f"""Vocabulary path ({save_directory}) should be a directory"""		)
																		return
												SCREAMING_SNAKE_CASE							:					Dict           =							os.path.join(
												    UpperCamelCase__					,	(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']		)
												SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							os.path.join(
												    UpperCamelCase__					,	(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']		)

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

												SCREAMING_SNAKE_CASE							:					Optional[int]           =							0
												with open(UpperCamelCase__					,	'''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 UpperCamelCase__		: 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!'''		)
																														SCREAMING_SNAKE_CASE							:					List[str]           =							token_index
																								writer.write(''' '''.join(UpperCamelCase__		) + '''\n'''		)
																								index += 1

												return vocab_file, merge_file


						def  __A ( self			:       Any					,	UpperCamelCase__			:       List[int]					,	UpperCamelCase__			:       Optional[List[int]] = None					,	UpperCamelCase__			:       bool = False		):





												'''simple docstring'''





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

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


						def  __A ( self			:       Tuple					,	UpperCamelCase__			:       List[int]					,	UpperCamelCase__			:       Optional[List[int]] = None		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Optional[int]           =							[self.sep_token_id]
												SCREAMING_SNAKE_CASE							:					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  __A ( self			:       Dict					,	UpperCamelCase__			:       Optional[int]					,	UpperCamelCase__			:       int=False					,	**UpperCamelCase__			:       List[str]		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					Any           =							kwargs.pop('''add_prefix_space'''					,	self.add_prefix_space		)
												if (is_split_into_words or add_prefix_space) and (len(UpperCamelCase__		) > 0 and not text[0].isspace()):
																		SCREAMING_SNAKE_CASE							:					str           =							''' ''' + text
												return (text, kwargs)


						def  __A ( self			:       Optional[Any]					,	UpperCamelCase__			:       List[int]					,	UpperCamelCase__			:       Optional[List[int]] = None		):





												'''simple docstring'''





												return token_ids_a + [self.eos_token_id]



						def  __A ( self			:       List[Any]					,	UpperCamelCase__			:       "Conversation"		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:					str           =							[]
												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(UpperCamelCase__		)

												SCREAMING_SNAKE_CASE							:					str           =							'''  '''.join(UpperCamelCase__		)
												SCREAMING_SNAKE_CASE							:					Dict           =							self.encode(UpperCamelCase__		)
												if len(UpperCamelCase__		) > self.model_max_length:
																		SCREAMING_SNAKE_CASE							:					Union[str, Any]           =							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






from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tf_available, is_torch_available


__UpperCamelCase	: Tuple    		=  {
    'configuration_ctrl': ['CTRL_PRETRAINED_CONFIG_ARCHIVE_MAP', 'CTRLConfig'],
    'tokenization_ctrl': ['CTRLTokenizer'],
}

try:
					if not is_torch_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					__UpperCamelCase	: Dict    		=  [
					    'CTRL_PRETRAINED_MODEL_ARCHIVE_LIST',
					    'CTRLForSequenceClassification',
					    'CTRLLMHeadModel',
					    'CTRLModel',
					    'CTRLPreTrainedModel',
					]

try:
					if not is_tf_available():
										raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
					pass
else:
					__UpperCamelCase	: Any    		=  [
					    'TF_CTRL_PRETRAINED_MODEL_ARCHIVE_LIST',
					    'TFCTRLForSequenceClassification',
					    'TFCTRLLMHeadModel',
					    'TFCTRLModel',
					    'TFCTRLPreTrainedModel',
					]


if TYPE_CHECKING:
					from .configuration_ctrl import CTRL_PRETRAINED_CONFIG_ARCHIVE_MAP, CTRLConfig
					from .tokenization_ctrl import CTRLTokenizer

					try:
										if not is_torch_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_ctrl import (
										    CTRL_PRETRAINED_MODEL_ARCHIVE_LIST,
										    CTRLForSequenceClassification,
										    CTRLLMHeadModel,
										    CTRLModel,
										    CTRLPreTrainedModel,
										)

					try:
										if not is_tf_available():
															raise OptionalDependencyNotAvailable()
					except OptionalDependencyNotAvailable:
										pass
					else:
										from .modeling_tf_ctrl import (
										    TF_CTRL_PRETRAINED_MODEL_ARCHIVE_LIST,
										    TFCTRLForSequenceClassification,
										    TFCTRLLMHeadModel,
										    TFCTRLModel,
										    TFCTRLPreTrainedModel,
										)

else:
					import sys

					__UpperCamelCase	: Optional[Any]    		=  _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)





