






import inspect
import os
from argparse import ArgumentParser, Namespace
from importlib import import_module

import huggingface_hub
import numpy as np
from packaging import version

from .. import (
    FEATURE_EXTRACTOR_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoImageProcessor,
    AutoProcessor,
    AutoTokenizer,
    is_datasets_available,
    is_tf_available,
    is_torch_available,
)
from ..utils import TF2_WEIGHTS_INDEX_NAME, TF2_WEIGHTS_NAME, logging
from . import BaseTransformersCLICommand


if is_tf_available():
		import tensorflow as tf

		tf.config.experimental.enable_tensor_float_aa_execution(False)

if is_torch_available():
		import torch

if is_datasets_available():
		from datasets import load_dataset


__UpperCAmelCase										=		5E-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors




def   lowercase__   (  __snake_case					:							Namespace						):


					'''simple docstring'''


					return PTtoTFCommand(
					    args.model_name  , args.local_dir  , args.max_error  , args.new_weights  , args.no_pr  , args.push  , args.extra_commit_description  , args.override_model_class  , )






class 						lowerCamelCase   (_snake_case    ):







		'''simple docstring'''



		@staticmethod
		def       __UpperCAmelCase    (							_UpperCamelCase     )			->							Union[str, Any]:
							UpperCAmelCase_					:   List[str]   =		parser.add_parser(
							    'pt-to-tf'       ,	help=(
							        'CLI tool to run convert a transformers model from a PyTorch checkpoint to a TensorFlow checkpoint.'
							        ' Can also be used to validate existing weights without opening PRs, with --no-pr.'
							    )       ,	)
							train_parser.add_argument(
							    '--model-name'       ,	type=_UpperCamelCase       ,	required=_UpperCamelCase       ,	help='The model name, including owner/organization, as seen on the hub.'       ,	)
							train_parser.add_argument(
							    '--local-dir'       ,	type=_UpperCamelCase       ,	default=''       ,	help='Optional local directory of the model repository. Defaults to /tmp/{model_name}'       ,	)
							train_parser.add_argument(
							    '--max-error'       ,	type=_UpperCamelCase       ,	default=_UpperCamelCase       ,	help=(
							        f"Maximum error tolerance. Defaults to {MAX_ERROR}. This flag should be avoided, use at your own risk."
							    )       ,	)
							train_parser.add_argument(
							    '--new-weights'       ,	action='store_true'       ,	help='Optional flag to create new TensorFlow weights, even if they already exist.'       ,	)
							train_parser.add_argument(
							    '--no-pr'       ,	action='store_true'       ,	help='Optional flag to NOT open a PR with converted weights.'     )
							train_parser.add_argument(
							    '--push'       ,	action='store_true'       ,	help='Optional flag to push the weights directly to `main` (requires permissions)'       ,	)
							train_parser.add_argument(
							    '--extra-commit-description'       ,	type=_UpperCamelCase       ,	default=''       ,	help='Optional additional commit description to use when opening a PR (e.g. to tag the owner).'       ,	)
							train_parser.add_argument(
							    '--override-model-class'       ,	type=_UpperCamelCase       ,	default=_UpperCamelCase       ,	help='If you think you know better than the auto-detector, you can specify the model class here. '
							    'Can be either an AutoModel class or a specific model class like BertForSequenceClassification.'       ,	)
							train_parser.set_defaults(func=_UpperCamelCase     )



		@staticmethod
		def       __UpperCAmelCase    (							_UpperCamelCase       ,	_UpperCamelCase     )			->							str:
							UpperCAmelCase_					:   str   =		set(pt_outputs.keys()     )
							UpperCAmelCase_					:   int   =		set(tf_outputs.keys()     )
							if pt_out_attrs != tf_out_attrs:
												raise ValueError(
												    f"The model outputs have different attributes, aborting. (Pytorch: {pt_out_attrs}, TensorFlow:"
												    f" {tf_out_attrs})"     )

							# 2. For each output attribute, computes the difference
							def _find_pt_tf_differences(_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase=""     ):
												# If the current attribute is a tensor, it is a leaf and we make the comparison. Otherwise, we will dig in
												# recursivelly, keeping the name of the attribute.
												if isinstance(_UpperCamelCase       ,	torch.Tensor     ):
																	UpperCAmelCase_					:   List[Any]   =		np.max(np.abs(pt_out.numpy() - tf_out.numpy()     )     )
																	UpperCAmelCase_					:   str   =		tensor_difference
												else:
																	UpperCAmelCase_					:   Dict   =		attr_name
																	for i, pt_item in enumerate(_UpperCamelCase     ):
																						# If it is a named attribute, we keep the name. Otherwise, just its index.
																						if isinstance(_UpperCamelCase       ,	_UpperCamelCase     ):
																											UpperCAmelCase_					:   Tuple   =		root_name + pt_item
																											UpperCAmelCase_					:   Dict   =		tf_out[pt_item]
																											UpperCAmelCase_					:   List[str]   =		pt_out[pt_item]
																						else:
																											UpperCAmelCase_					:   Optional[Any]   =		root_name + f"[{i}]"
																											UpperCAmelCase_					:   List[str]   =		tf_out[i]
																						UpperCAmelCase_					:   List[Any]   =		_find_pt_tf_differences(_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase     )

												return differences

							return _find_pt_tf_differences(_UpperCamelCase       ,	_UpperCamelCase       ,	{}     )



		def __init__(							self       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	*_UpperCamelCase       ,	)			->							str:
							UpperCAmelCase_					:   str   =		logging.get_logger('transformers-cli/pt_to_tf'     )
							UpperCAmelCase_					:   Any   =		model_name
							UpperCAmelCase_					:   Union[str, Any]   =		local_dir if local_dir else os.path.join('/tmp'       ,	_UpperCamelCase     )
							UpperCAmelCase_					:   Dict   =		max_error
							UpperCAmelCase_					:   str   =		new_weights
							UpperCAmelCase_					:   Optional[Any]   =		no_pr
							UpperCAmelCase_					:   Any   =		push
							UpperCAmelCase_					:   int   =		extra_commit_description
							UpperCAmelCase_					:   str   =		override_model_class



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase     )			->							List[str]:

							def _get_audio_input():
												UpperCAmelCase_					:   Optional[int]   =		load_dataset('hf-internal-testing/librispeech_asr_dummy'       ,	'clean'       ,	split='validation'     )
												UpperCAmelCase_					:   Union[str, Any]   =		ds.sort('id'     ).select(range(2     )     )[:2]['audio']
												UpperCAmelCase_					:   Optional[Any]   =		[x['array'] for x in speech_samples]
												return raw_samples

							UpperCAmelCase_					:   Tuple   =		type(pt_model.config     )
							if model_config_class in PROCESSOR_MAPPING:
												UpperCAmelCase_					:   List[Any]   =		AutoProcessor.from_pretrained(self._local_dir     )
												if model_config_class in TOKENIZER_MAPPING and processor.tokenizer.pad_token is None:
																	UpperCAmelCase_					:   Dict   =		processor.tokenizer.eos_token
							elif model_config_class in IMAGE_PROCESSOR_MAPPING:
												UpperCAmelCase_					:   List[str]   =		AutoImageProcessor.from_pretrained(self._local_dir     )
							elif model_config_class in FEATURE_EXTRACTOR_MAPPING:
												UpperCAmelCase_					:   Tuple   =		AutoFeatureExtractor.from_pretrained(self._local_dir     )
							elif model_config_class in TOKENIZER_MAPPING:
												UpperCAmelCase_					:   int   =		AutoTokenizer.from_pretrained(self._local_dir     )
												if processor.pad_token is None:
																	UpperCAmelCase_					:   Tuple   =		processor.eos_token
							else:
												raise ValueError(f"Unknown data processing type (model config type: {model_config_class})"     )

							UpperCAmelCase_					:   List[Any]   =		set(inspect.signature(pt_model.forward     ).parameters.keys()     )
							UpperCAmelCase_					:   List[Any]   =		{}
							if "input_ids" in model_forward_signature:
												processor_inputs.update(
												    {
												        'text': ['Hi there!', 'I am a batch with more than one row and different input lengths.'],
												        'padding': True,
												        'truncation': True,
												    }     )
							if "pixel_values" in model_forward_signature:
												UpperCAmelCase_					:   Union[str, Any]   =		load_dataset('cifar10'       ,	'plain_text'       ,	split='test'     )[:2]['img']
												processor_inputs.update({'images': sample_images}     )
							if "input_features" in model_forward_signature:
												UpperCAmelCase_					:   str   =		inspect.signature(processor.feature_extractor     ).parameters
												# Pad to the largest input length by default but take feature extractor default
												# padding value if it exists e.g. "max_length" and is not False or None
												if "padding" in feature_extractor_signature:
																	UpperCAmelCase_					:   List[Any]   =		feature_extractor_signature['padding'].default
																	if default_strategy is not False and default_strategy is not None:
																						UpperCAmelCase_					:   Optional[Any]   =		default_strategy
																	else:
																						UpperCAmelCase_					:   Any   =		True
												else:
																	UpperCAmelCase_					:   Optional[Any]   =		True
												processor_inputs.update({'audio': _get_audio_input(), 'padding': padding_strategy}     )
							if "input_values" in model_forward_signature:  # Wav2Vec2 audio input
												processor_inputs.update({'audio': _get_audio_input(), 'padding': True}     )
							UpperCAmelCase_					:   Optional[Any]   =		processor(**_UpperCamelCase       ,	return_tensors='pt'     )
							UpperCAmelCase_					:   Union[str, Any]   =		processor(**_UpperCamelCase       ,	return_tensors='tf'     )

							# Extra input requirements, in addition to the input modality
							if (
							    config.is_encoder_decoder
							    or (hasattr(_UpperCamelCase       ,	'encoder'     ) and hasattr(_UpperCamelCase       ,	'decoder'     ))
							    or "decoder_input_ids" in tf_dummy_inputs
							):
												UpperCAmelCase_					:   Optional[Any]   =		np.asarray([[1], [1]]       ,	dtype=_UpperCamelCase     ) * (pt_model.config.decoder_start_token_id or 0)
												pt_input.update({'decoder_input_ids': torch.tensor(_UpperCamelCase     )}     )
												tf_input.update({'decoder_input_ids': tf.convert_to_tensor(_UpperCamelCase     )}     )

							return pt_input, tf_input



		def       __UpperCAmelCase    (							self     )			->							Any:
							# hub version 0.9.0 introduced the possibility of programmatically opening PRs with normal write tokens.
							if version.parse(huggingface_hub.__version__     ) < version.parse('0.9.0'     ):
												raise ImportError(
												    'The huggingface_hub version must be >= 0.9.0 to use this command. Please update your huggingface_hub'
												    ' installation.'     )
							else:
												from huggingface_hub import Repository, create_commit
												from huggingface_hub._commit_api import CommitOperationAdd

							# Fetch remote data
							UpperCAmelCase_					:   List[Any]   =		Repository(local_dir=self._local_dir       ,	clone_from=self._model_name     )

							# Load config and get the appropriate architecture -- the latter is needed to convert the head's weights
							UpperCAmelCase_					:   Optional[Any]   =		AutoConfig.from_pretrained(self._local_dir     )
							UpperCAmelCase_					:   int   =		config.architectures
							if self._override_model_class is not None:
												if self._override_model_class.startswith('TF'     ):
																	UpperCAmelCase_					:   Any   =		[self._override_model_class[2:]]
												else:
																	UpperCAmelCase_					:   Optional[int]   =		[self._override_model_class]
												try:
																	UpperCAmelCase_					:   Tuple   =		getattr(import_module('transformers'     )       ,	architectures[0]     )
												except AttributeError:
																	raise ValueError(f"Model class {self._override_model_class} not found in transformers."     )
												try:
																	UpperCAmelCase_					:   Dict   =		getattr(import_module('transformers'     )       ,	'TF' + architectures[0]     )
												except AttributeError:
																	raise ValueError(f"TF model class TF{self._override_model_class} not found in transformers."     )
							elif architectures is None:  # No architecture defined -- use auto classes
												UpperCAmelCase_					:   str   =		getattr(import_module('transformers'     )       ,	'AutoModel'     )
												UpperCAmelCase_					:   Union[str, Any]   =		getattr(import_module('transformers'     )       ,	'TFAutoModel'     )
												self._logger.warning('No detected architecture, using AutoModel/TFAutoModel'     )
							else:  # Architecture defined -- use it
												if len(_UpperCamelCase     ) > 1:
																	raise ValueError(f"More than one architecture was found, aborting. (architectures = {architectures})"     )
												self._logger.warning(f"Detected architecture: {architectures[0]}"     )
												UpperCAmelCase_					:   Dict   =		getattr(import_module('transformers'     )       ,	architectures[0]     )
												try:
																	UpperCAmelCase_					:   Dict   =		getattr(import_module('transformers'     )       ,	'TF' + architectures[0]     )
												except AttributeError:
																	raise AttributeError(f"The TensorFlow equivalent of {architectures[0]} doesn't exist in transformers."     )

        # Check the TF dummy inputs to see what keys we need in the forward pass
							UpperCAmelCase_					:   Optional[Any]   =		tf_class.from_config(_UpperCamelCase     )
							UpperCAmelCase_					:   str   =		tf_from_pt_model.dummy_inputs

							del tf_from_pt_model  # Try to keep only one model in memory at a time

							# Load the model and get some basic inputs
							UpperCAmelCase_					:   Optional[int]   =		pt_class.from_pretrained(self._local_dir     )
							pt_model.eval()

							UpperCAmelCase_ ,						UpperCAmelCase_					:   str   =		self.get_inputs(_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase     )

							with torch.no_grad():
												UpperCAmelCase_					:   Tuple   =		pt_model(**_UpperCamelCase       ,	output_hidden_states=_UpperCamelCase     )
							del pt_model  # will no longer be used, and may have a large memory footprint

							UpperCAmelCase_					:   List[str]   =		tf_class.from_pretrained(self._local_dir       ,	from_pt=_UpperCamelCase     )
							UpperCAmelCase_					:   Any   =		tf_from_pt_model(**_UpperCamelCase       ,	output_hidden_states=_UpperCamelCase       ,	training=_UpperCamelCase     )

							# Confirms that cross loading PT weights into TF worked.
							UpperCAmelCase_					:   Optional[Any]   =		self.find_pt_tf_differences(_UpperCamelCase       ,	_UpperCamelCase     )
							UpperCAmelCase_					:   List[Any]   =		{k: v for k, v in crossload_differences.items() if 'hidden' not in k}
							UpperCAmelCase_					:   Tuple   =		{k: v for k, v in crossload_differences.items() if 'hidden' in k}
							if len(_UpperCamelCase     ) == 0 and architectures is not None:
												raise ValueError(
												    f"Something went wrong -- the config file has architectures ({architectures}), but no model head"
												    ' output was found. All outputs start with \'hidden\''     )
							UpperCAmelCase_					:   int   =		max(output_differences.values()     ) if output_differences else 0.0
							UpperCAmelCase_					:   str   =		max(hidden_differences.values()     )
							if max_crossload_output_diff > self._max_error or max_crossload_hidden_diff > self._max_error:
												raise ValueError(
												    'The cross-loaded TensorFlow model has different outputs, something went wrong!\n'
												    + f"\nList of maximum output differences above the threshold ({self._max_error}):\n"
												    + '\n'.join([f"{k}: {v:.3e}" for k, v in output_differences.items() if v > self._max_error]     )
												    + f"\n\nList of maximum hidden layer differences above the threshold ({self._max_error}):\n"
												    + '\n'.join([f"{k}: {v:.3e}" for k, v in hidden_differences.items() if v > self._max_error]     )     )

							# Save the weights in a TF format (if needed) and confirms that the results are still good
							UpperCAmelCase_					:   Optional[Any]   =		os.path.join(self._local_dir       ,	_UpperCamelCase     )
							UpperCAmelCase_					:   List[str]   =		os.path.join(self._local_dir       ,	_UpperCamelCase     )
							if (not os.path.exists(_UpperCamelCase     ) and not os.path.exists(_UpperCamelCase     )) or self._new_weights:
												tf_from_pt_model.save_pretrained(self._local_dir     )
							del tf_from_pt_model  # will no longer be used, and may have a large memory footprint

							UpperCAmelCase_					:   str   =		tf_class.from_pretrained(self._local_dir     )
							UpperCAmelCase_					:   Dict   =		tf_model(**_UpperCamelCase       ,	output_hidden_states=_UpperCamelCase     )

							UpperCAmelCase_					:   List[Any]   =		self.find_pt_tf_differences(_UpperCamelCase       ,	_UpperCamelCase     )
							UpperCAmelCase_					:   Optional[Any]   =		{k: v for k, v in conversion_differences.items() if 'hidden' not in k}
							UpperCAmelCase_					:   int   =		{k: v for k, v in conversion_differences.items() if 'hidden' in k}
							if len(_UpperCamelCase     ) == 0 and architectures is not None:
												raise ValueError(
												    f"Something went wrong -- the config file has architectures ({architectures}), but no model head"
												    ' output was found. All outputs start with \'hidden\''     )
							UpperCAmelCase_					:   Tuple   =		max(output_differences.values()     ) if output_differences else 0.0
							UpperCAmelCase_					:   Optional[int]   =		max(hidden_differences.values()     )
							if max_conversion_output_diff > self._max_error or max_conversion_hidden_diff > self._max_error:
												raise ValueError(
												    'The converted TensorFlow model has different outputs, something went wrong!\n'
												    + f"\nList of maximum output differences above the threshold ({self._max_error}):\n"
												    + '\n'.join([f"{k}: {v:.3e}" for k, v in output_differences.items() if v > self._max_error]     )
												    + f"\n\nList of maximum hidden layer differences above the threshold ({self._max_error}):\n"
												    + '\n'.join([f"{k}: {v:.3e}" for k, v in hidden_differences.items() if v > self._max_error]     )     )

							UpperCAmelCase_					:   Optional[Any]   =		'Update TF weights' if self._new_weights else 'Add TF weights'
							if self._push:
												repo.git_add(auto_lfs_track=_UpperCamelCase     )
												repo.git_commit(_UpperCamelCase     )
												repo.git_push(blocking=_UpperCamelCase     )  # this prints a progress bar with the upload
												self._logger.warning(f"TF weights pushed into {self._model_name}"     )
							elif not self._no_pr:
												self._logger.warning('Uploading the weights into a new PR...'     )
												UpperCAmelCase_					:   Any   =		(
												    'Model converted by the [`transformers`\' `pt_to_tf`'
												    ' CLI](https://github.com/huggingface/transformers/blob/main/src/transformers/commands/pt_to_tf.py). '
												    'All converted model outputs and hidden layers were validated against its PyTorch counterpart.\n\n'
												    f"Maximum crossload output difference={max_crossload_output_diff:.3e}; "
												    f"Maximum crossload hidden layer difference={max_crossload_hidden_diff:.3e};\n"
												    f"Maximum conversion output difference={max_conversion_output_diff:.3e}; "
												    f"Maximum conversion hidden layer difference={max_conversion_hidden_diff:.3e};\n"
												)
												if self._max_error > MAX_ERROR:
																	commit_descrition += (
																	    f"\n\nCAUTION: The maximum admissible error was manually increased to {self._max_error}!"
																	)
												if self._extra_commit_description:
																	commit_descrition += "\n\n" + self._extra_commit_description

												# sharded model -> adds all related files (index and .h5 shards)
												if os.path.exists(_UpperCamelCase     ):
																	UpperCAmelCase_					:   Dict   =		[
																	    CommitOperationAdd(path_in_repo=_UpperCamelCase       ,	path_or_fileobj=_UpperCamelCase     )
																	]
																	for shard_path in tf.io.gfile.glob(self._local_dir + '/tf_model-*.h5'     ):
																						operations += [
																						    CommitOperationAdd(path_in_repo=os.path.basename(_UpperCamelCase     )       ,	path_or_fileobj=_UpperCamelCase     )
																						]
												else:
																	UpperCAmelCase_					:   Any   =		[CommitOperationAdd(path_in_repo=_UpperCamelCase       ,	path_or_fileobj=_UpperCamelCase     )]

												UpperCAmelCase_					:   Optional[Any]   =		create_commit(
												    repo_id=self._model_name       ,	operations=_UpperCamelCase       ,	commit_message=_UpperCamelCase       ,	commit_description=_UpperCamelCase       ,	repo_type='model'       ,	create_pr=_UpperCamelCase       ,	).pr_url
												self._logger.warning(f"PR open in {hub_pr_url}"     )














import json
import os
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'}

__UpperCAmelCase										=		{
    'vocab_file': {
        'microsoft/deberta-base': 'https://huggingface.co/microsoft/deberta-base/resolve/main/vocab.json',
        'microsoft/deberta-large': 'https://huggingface.co/microsoft/deberta-large/resolve/main/vocab.json',
        'microsoft/deberta-xlarge': 'https://huggingface.co/microsoft/deberta-xlarge/resolve/main/vocab.json',
        'microsoft/deberta-base-mnli': 'https://huggingface.co/microsoft/deberta-base-mnli/resolve/main/vocab.json',
        'microsoft/deberta-large-mnli': 'https://huggingface.co/microsoft/deberta-large-mnli/resolve/main/vocab.json',
        'microsoft/deberta-xlarge-mnli': (
            'https://huggingface.co/microsoft/deberta-xlarge-mnli/resolve/main/vocab.json'
        ),
    },
    'merges_file': {
        'microsoft/deberta-base': 'https://huggingface.co/microsoft/deberta-base/resolve/main/merges.txt',
        'microsoft/deberta-large': 'https://huggingface.co/microsoft/deberta-large/resolve/main/merges.txt',
        'microsoft/deberta-xlarge': 'https://huggingface.co/microsoft/deberta-xlarge/resolve/main/merges.txt',
        'microsoft/deberta-base-mnli': 'https://huggingface.co/microsoft/deberta-base-mnli/resolve/main/merges.txt',
        'microsoft/deberta-large-mnli': 'https://huggingface.co/microsoft/deberta-large-mnli/resolve/main/merges.txt',
        'microsoft/deberta-xlarge-mnli': (
            'https://huggingface.co/microsoft/deberta-xlarge-mnli/resolve/main/merges.txt'
        ),
    },
}

__UpperCAmelCase										=		{
    'microsoft/deberta-base': 512,
    'microsoft/deberta-large': 512,
    'microsoft/deberta-xlarge': 512,
    'microsoft/deberta-base-mnli': 512,
    'microsoft/deberta-large-mnli': 512,
    'microsoft/deberta-xlarge-mnli': 512,
}

__UpperCAmelCase										=		{
    'microsoft/deberta-base': {'do_lower_case': False},
    'microsoft/deberta-large': {'do_lower_case': False},
}




def   lowercase__   (  ):


					'''simple docstring'''


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




def   lowercase__   (  __snake_case					:							Optional[int]						):


					'''simple docstring'''


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






class 						lowerCamelCase   (_snake_case    ):







		'''simple docstring'''



		_snake_case					:					Tuple 		=						VOCAB_FILES_NAMES
		_snake_case					:					Any 		=						PRETRAINED_VOCAB_FILES_MAP
		_snake_case					:					Union[str, Any] 		=						PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		_snake_case					:					Tuple 		=						['''input_ids''', '''attention_mask''', '''token_type_ids''']



		def __init__(							self       ,	_UpperCamelCase       ,	_UpperCamelCase       ,	_UpperCamelCase="replace"       ,	_UpperCamelCase="[CLS]"       ,	_UpperCamelCase="[SEP]"       ,	_UpperCamelCase="[SEP]"       ,	_UpperCamelCase="[CLS]"       ,	_UpperCamelCase="[UNK]"       ,	_UpperCamelCase="[PAD]"       ,	_UpperCamelCase="[MASK]"       ,	_UpperCamelCase=False       ,	_UpperCamelCase=False       ,	**_UpperCamelCase       ,	)			->							Any:
							UpperCAmelCase_					:   List[str]   =		AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase     ) if isinstance(_UpperCamelCase       ,	_UpperCamelCase     ) else bos_token
							UpperCAmelCase_					:   List[str]   =		AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase     ) if isinstance(_UpperCamelCase       ,	_UpperCamelCase     ) else eos_token
							UpperCAmelCase_					:   Dict   =		AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase     ) if isinstance(_UpperCamelCase       ,	_UpperCamelCase     ) else sep_token
							UpperCAmelCase_					:   Optional[int]   =		AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase     ) if isinstance(_UpperCamelCase       ,	_UpperCamelCase     ) else cls_token
							UpperCAmelCase_					:   Any   =		AddedToken(_UpperCamelCase       ,	lstrip=_UpperCamelCase       ,	rstrip=_UpperCamelCase     ) if isinstance(_UpperCamelCase       ,	_UpperCamelCase     ) else unk_token
							UpperCAmelCase_					:   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
							UpperCAmelCase_					:   Optional[Any]   =		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       ,	add_bos_token=_UpperCamelCase       ,	**_UpperCamelCase       ,	)
							UpperCAmelCase_					:   List[str]   =		add_bos_token

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

							# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
							UpperCAmelCase_					:   Optional[Any]   =		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.gpt2.tokenization_gpt2.GPT2Tokenizer.vocab_size
		def       __UpperCAmelCase    (							self     )			->							Any:
							return len(self.encoder     )



		def       __UpperCAmelCase    (							self     )			->							Optional[int]:
							return dict(self.encoder       ,	**self.added_tokens_encoder     )



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase     )			->							List[Any]:
							if token in self.cache:
												return self.cache[token]
							UpperCAmelCase_					:   Any   =		tuple(_UpperCamelCase     )
							UpperCAmelCase_					:   Tuple   =		get_pairs(_UpperCamelCase     )

							if not pairs:
												return token

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



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase       ,	_UpperCamelCase = None     )			->							List[int]:
							if token_ids_a is None:
												return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
							UpperCAmelCase_					:   Optional[Any]   =		[self.cls_token_id]
							UpperCAmelCase_					:   Union[str, Any]   =		[self.sep_token_id]
							return cls + token_ids_a + sep + token_ids_a + sep



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase       ,	_UpperCamelCase = None       ,	_UpperCamelCase = False     )			->							List[int]:
							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] + ([0] * len(_UpperCamelCase     )) + [1]



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase       ,	_UpperCamelCase = None     )			->							List[int]:
							UpperCAmelCase_					:   Optional[Any]   =		[self.sep_token_id]
							UpperCAmelCase_					:   Union[str, Any]   =		[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     ) * [0] + len(token_ids_a + sep     ) * [1]



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase     )			->							List[str]:
							UpperCAmelCase_					:   Optional[Any]   =		[]
							for token in re.findall(self.pat       ,	_UpperCamelCase     ):
												UpperCAmelCase_					:   int   =		''.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       __UpperCAmelCase    (							self       ,	_UpperCamelCase     )			->							Optional[Any]:
							return self.encoder.get(_UpperCamelCase       ,	self.encoder.get(self.unk_token     )     )



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase     )			->							int:
							return self.decoder.get(_UpperCamelCase     )



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase     )			->							Optional[int]:
							UpperCAmelCase_					:   Tuple   =		''.join(_UpperCamelCase     )
							UpperCAmelCase_					:   Any   =		bytearray([self.byte_decoder[c] for c in text]     ).decode('utf-8'       ,	errors=self.errors     )
							return text



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase       ,	_UpperCamelCase = None     )			->							Tuple[str]:
							if not os.path.isdir(_UpperCamelCase     ):
												logger.error(f"Vocabulary path ({save_directory}) should be a directory"     )
												return
							UpperCAmelCase_					:   Dict   =		os.path.join(
							    _UpperCamelCase       ,	(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']     )
							UpperCAmelCase_					:   List[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'     )

							UpperCAmelCase_					:   List[Any]   =		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!'     )
																						UpperCAmelCase_					:   Tuple   =		token_index
																	writer.write(' '.join(_UpperCamelCase     ) + '\n'     )
																	index += 1

							return vocab_file, merge_file



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase       ,	_UpperCamelCase=False       ,	**_UpperCamelCase     )			->							Optional[int]:
							UpperCAmelCase_					:   str   =		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()):
												UpperCAmelCase_					:   Optional[Any]   =		' ' + text
							return (text, kwargs)



		def       __UpperCAmelCase    (							self       ,	_UpperCamelCase     )			->							List[int]:
							UpperCAmelCase_					:   Union[str, Any]   =		[]
							for is_user, text in conversation.iter_texts():
												input_ids.extend(self.encode(_UpperCamelCase       ,	add_special_tokens=_UpperCamelCase     ) + [self.eos_token_id]     )
							if len(_UpperCamelCase     ) > self.model_max_length:
												UpperCAmelCase_					:   Union[str, Any]   =		input_ids[-self.model_max_length :]
							return input_ids






