



import importlib
import os
from dataclasses import dataclass
from enum import Enum
from typing import Any, Dict, Optional, Union

import torch

from ..utils import BaseOutput


_lowerCamelCase    :		List[str]		      =    """scheduler_config.json"""


class 						lowercase    (							SCREAMING_SNAKE_CASE_):
						'''simple docstring'''


						UpperCAmelCase			:		Union[str, Any]		=		1
						UpperCAmelCase			:		Optional[int]		=		2
						UpperCAmelCase			:		Union[str, Any]		=		3
						UpperCAmelCase			:		List[str]		=		4
						UpperCAmelCase			:		Optional[int]		=		5
						UpperCAmelCase			:		Optional[Any]		=		6
						UpperCAmelCase			:		List[str]		=		7
						UpperCAmelCase			:		Optional[int]		=		8
						UpperCAmelCase			:		List[str]		=		9
						UpperCAmelCase			:		Optional[int]		=		10
						UpperCAmelCase			:		List[str]		=		11
						UpperCAmelCase			:		Any		=		12
						UpperCAmelCase			:		List[str]		=		13
						UpperCAmelCase			:		int		=		14


@dataclass
class 						lowercase    (							SCREAMING_SNAKE_CASE_):
						'''simple docstring'''


						UpperCAmelCase			:		torch.FloatTensor


class 						lowercase    :
						'''simple docstring'''


						UpperCAmelCase			:		Union[str, Any]		=		SCHEDULER_CONFIG_NAME
						UpperCAmelCase			:		List[str]		=		[]
						UpperCAmelCase			:		Optional[Any]		=		True







						@classmethod
						def       lowerCamelCase_				(  cls    :				Any       ,	snake_case    :				Dict[str, Any] = None       ,	snake_case    :				Optional[str] = None       ,	snake_case    :				Optional[int]=False       ,	**snake_case    :				Optional[Any]       ,	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE					,  SCREAMING_SNAKE_CASE					,  SCREAMING_SNAKE_CASE							:     List[str]						=				cls.load_config(
											    pretrained_model_name_or_path=snake_case       ,	subfolder=snake_case       ,	return_unused_kwargs=snake_case       ,	return_commit_hash=snake_case       ,	**snake_case       ,	)
											return cls.from_config(snake_case       ,	return_unused_kwargs=snake_case       ,	**snake_case	)







						def       lowerCamelCase_				(  self    :				str       ,	snake_case    :				Union[str, os.PathLike]       ,	snake_case    :				bool = False       ,	**snake_case    :				Dict	):



											'''simple docstring'''


											self.save_config(save_directory=snake_case       ,	push_to_hub=snake_case       ,	**snake_case	)







						@property
						def       lowerCamelCase_				(  self    :				str	):



											'''simple docstring'''


											return self._get_compatibles()







						@classmethod
						def       lowerCamelCase_				(  cls    :				List[Any]	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     Tuple						=				list(set([cls.__name__] + cls._compatibles	)	)
											SCREAMING_SNAKE_CASE							:     str						=				importlib.import_module(__name__.split('.'	)[0]	)
											SCREAMING_SNAKE_CASE							:     Tuple						=				[
											    getattr(snake_case       ,	snake_case	) for c in compatible_classes_str if hasattr(snake_case       ,	snake_case	)
											]
											return compatible_classes




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

_lowerCamelCase    :		Any		      =    logging.get_logger(__name__)


_lowerCamelCase    :		Union[str, Any]		      =    {
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
    """tokenizer_config_file""": """tokenizer_config.json""",
}

_lowerCamelCase    :		Optional[Any]		      =    {
    """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"""
    },
}

_lowerCamelCase    :		Optional[int]		      =    {"""facebook/blenderbot-3B""": 128}

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

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



class 						lowercase    (							SCREAMING_SNAKE_CASE_):
						'''simple docstring'''


						UpperCAmelCase			:		Optional[int]		=		VOCAB_FILES_NAMES
						UpperCAmelCase			:		Any		=		PRETRAINED_VOCAB_FILES_MAP
						UpperCAmelCase			:		str		=		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						UpperCAmelCase			:		Dict		=		['input_ids', 'attention_mask']







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



											'''simple docstring'''


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

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



											'''simple docstring'''


											return len(self.encoder	)







						def       lowerCamelCase_				(  self    :				Optional[Any]	):



											'''simple docstring'''


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







						def       lowerCamelCase_				(  self    :				List[str]       ,	snake_case    :				int	):



											'''simple docstring'''


											if token in self.cache:
																return self.cache[token]
											SCREAMING_SNAKE_CASE							:     int						=				tuple(snake_case	)
											SCREAMING_SNAKE_CASE							:     Dict						=				get_pairs(snake_case	)

											if not pairs:
																return token

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







						def       lowerCamelCase_				(  self    :				Any       ,	snake_case    :				Tuple	):



											'''simple docstring'''


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







						def       lowerCamelCase_				(  self    :				Any       ,	snake_case    :				int	):



											'''simple docstring'''


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







						def       lowerCamelCase_				(  self    :				List[Any]       ,	snake_case    :				Dict	):



											'''simple docstring'''


											return self.decoder.get(snake_case	)







						def       lowerCamelCase_				(  self    :				List[Any]       ,	snake_case    :				int	):



											'''simple docstring'''


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







						def       lowerCamelCase_				(  self    :				Union[str, Any]       ,	snake_case    :				str       ,	snake_case    :				Optional[str] = None	):



											'''simple docstring'''


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

											SCREAMING_SNAKE_CASE							:     str						=				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 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!'	)
																										SCREAMING_SNAKE_CASE							:     List[str]						=				token_index
																					writer.write(' '.join(snake_case	) + '\n'	)
																					index += 1

											return vocab_file, merge_file







						def       lowerCamelCase_				(  self    :				Dict       ,	snake_case    :				List[int]       ,	snake_case    :				Optional[List[int]] = None       ,	snake_case    :				bool = False	):



											'''simple docstring'''


											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       lowerCamelCase_				(  self    :				int       ,	snake_case    :				List[int]       ,	snake_case    :				Optional[List[int]] = None	):



											'''simple docstring'''


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



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     Optional[int]						=				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()):
																SCREAMING_SNAKE_CASE							:     Union[str, Any]						=				' ' + text
											return (text, kwargs)







						def       lowerCamelCase_				(  self    :				List[str]       ,	snake_case    :				List[int]       ,	snake_case    :				Optional[List[int]] = None	):



											'''simple docstring'''


											return token_ids_a + [self.eos_token_id]







						def       lowerCamelCase_				(  self    :				List[str]       ,	snake_case    :				"Conversation"	):



											'''simple docstring'''


											SCREAMING_SNAKE_CASE							:     Tuple						=				[]
											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	)

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