

'''simple docstring'''




import os
import warnings
from pathlib import Path
from typing import Optional, Union

from .dynamic_module_utils import custom_object_save
from .tokenization_utils_base import PreTrainedTokenizerBase
from .utils import PushToHubMixin, copy_func, direct_transformers_import, logging


_a			:	str								=  logging.get_logger(__name__)

# Dynamically import the Transformers module to grab the attribute classes of the processor form their names.
_a			:	Optional[Any]								=  direct_transformers_import(Path(__file__).parent)


_a			:	Optional[int]								=  {
    'AutoTokenizer': 'PreTrainedTokenizerBase',
    'AutoFeatureExtractor': 'FeatureExtractionMixin',
    'AutoImageProcessor': 'ImageProcessingMixin',
}



class     lowercase_		(						a	):

  '''simple docstring'''





  __lowerCAmelCase   :	int      					=				["feature_extractor", "tokenizer"]
  # Names need to be attr_class for attr in attributes
  __lowerCAmelCase   :	Optional[Any]      					=				None
  __lowerCAmelCase   :	int      					=				None
  __lowerCAmelCase   :	List[Any]      					=				None
  def __init__(       self				,	*a_				,	**a_       )							->			int:


    """simple docstring"""
    # Sanitize args and kwargs
    for key in kwargs:
      if key not in self.attributes:
        raise TypeError(F'''Unexpected keyword argument {key}.'''       )
    for arg, attribute_name in zip(a_				,	self.attributes       ):
      if attribute_name in kwargs:
        raise TypeError(F'''Got multiple values for argument {attribute_name}.'''       )
      else:
        UpperCAmelCase  							=		arg

    if len(a_       ) != len(self.attributes       ):
      raise ValueError(
          F'''This processor requires {len(self.attributes       )} arguments: {', '.join(self.attributes       )}. Got '''
          F'''{len(a_       )} arguments instead.'''       )

    # Check each arg is of the proper class (this will also catch a user initializing in the wrong order)
    for attribute_name, arg in kwargs.items():
      UpperCAmelCase  							=		getattr(self				,	F'''{attribute_name}_class'''       )
      # Nothing is ever going to be an instance of "AutoXxx", in that case we check the base class.
      UpperCAmelCase  							=		AUTO_TO_BASE_CLASS_MAPPING.get(a_				,	a_       )
      if isinstance(a_				,	a_       ):
        UpperCAmelCase  							=		tuple(getattr(a_				,	a_       ) for n in class_name if n is not None       )
      else:
        UpperCAmelCase  							=		getattr(a_				,	a_       )

      if not isinstance(a_				,	a_       ):
        raise ValueError(
            F'''Received a {type(a_       ).__name__} for argument {attribute_name}, but a {class_name} was expected.'''       )

      setattr(self				,	a_				,	a_       )
  def __repr__(       self       )							->			List[str]:


    """simple docstring"""
    UpperCAmelCase  							=		[F'''- {name}: {repr(getattr(self				,	a_       )       )}''' for name in self.attributes]
    UpperCAmelCase  							=		'\n'.join(a_       )
    return F'''{self.__class__.__name__}:\n{attributes_repr}'''
  def   snake_case_     (       self				,	a_				,	a_ = False				,	**a_       )							->			str:


    """simple docstring"""
    os.makedirs(a_				,	exist_ok=a_       )

    if push_to_hub:
      UpperCAmelCase  							=		kwargs.pop('commit_message'				,	a_       )
      UpperCAmelCase  							=		kwargs.pop('repo_id'				,	save_directory.split(os.path.sep       )[-1]       )
      UpperCAmelCase  							=		self._create_repo(a_				,	**a_       )
      UpperCAmelCase  							=		self._get_files_timestamps(a_       )
    # If we have a custom config, we copy the file defining it in the folder and set the attributes so it can be
    # loaded from the Hub.
    if self._auto_class is not None:
      UpperCAmelCase  							=		[getattr(self				,	a_       ) for attribute_name in self.attributes]
      UpperCAmelCase  							=		[(a.init_kwargs if isinstance(a_				,	a_       ) else a) for a in attrs]
      custom_object_save(self				,	a_				,	config=a_       )

    for attribute_name in self.attributes:
      UpperCAmelCase  							=		getattr(self				,	a_       )
      # Include the processor class in the attribute config so this processor can then be reloaded with the
      # `AutoProcessor` API.
      if hasattr(a_				,	'_set_processor_class'       ):
        attribute._set_processor_class(self.__class__.__name__       )
      attribute.save_pretrained(a_       )

    if self._auto_class is not None:
      # We added an attribute to the init_kwargs of the tokenizers, which needs to be cleaned up.
      for attribute_name in self.attributes:
        UpperCAmelCase  							=		getattr(self				,	a_       )
        if isinstance(a_				,	a_       ):
          del attribute.init_kwargs["auto_map"]

    if push_to_hub:
      self._upload_modified_files(
          a_				,	a_				,	a_				,	commit_message=a_				,	token=kwargs.get('use_auth_token'       )				,	)
  @classmethod
  def   snake_case_     (       cls				,	a_				,	a_ = None				,	a_ = False				,	a_ = False				,	a_ = None				,	a_ = "main"				,	**a_				,	)							->			Union[str, Any]:


    """simple docstring"""
    UpperCAmelCase  							=		cache_dir
    UpperCAmelCase  							=		force_download
    UpperCAmelCase  							=		local_files_only
    UpperCAmelCase  							=		revision

    UpperCAmelCase  							=		kwargs.pop('use_auth_token'				,	a_       )
    if use_auth_token is not None:
      warnings.warn(
          'The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers.'				,	a_       )
      if token is not None:
        raise ValueError(
            '`token` and `use_auth_token` are both specified. Please set only the argument `token`.'       )
      UpperCAmelCase  							=		use_auth_token

    if token is not None:
      # change to `token` in a follow-up PR
      UpperCAmelCase  							=		token

    UpperCAmelCase  							=		cls._get_arguments_from_pretrained(a_				,	**a_       )
    return cls(*a_       )
  @classmethod
  def   snake_case_     (       cls				,	a_="AutoProcessor"       )							->			Tuple:


    """simple docstring"""
    if not isinstance(a_				,	a_       ):
      UpperCAmelCase  							=		auto_class.__name__

    import transformers.models.auto as auto_module

    if not hasattr(a_				,	a_       ):
      raise ValueError(F'''{auto_class} is not a valid auto class.'''       )

    UpperCAmelCase  							=		auto_class
  @classmethod
  def   snake_case_     (       cls				,	a_				,	**a_       )							->			Any:


    """simple docstring"""
    UpperCAmelCase  							=		[]
    for attribute_name in cls.attributes:
      UpperCAmelCase  							=		getattr(cls				,	F'''{attribute_name}_class'''       )
      if isinstance(a_				,	a_       ):
        UpperCAmelCase  							=		tuple(getattr(a_				,	a_       ) if n is not None else None for n in class_name       )
        UpperCAmelCase  							=		kwargs.get('use_fast'				,	a_       )
        if use_fast and classes[1] is not None:
          UpperCAmelCase  							=		classes[1]
        else:
          UpperCAmelCase  							=		classes[0]
      else:
        UpperCAmelCase  							=		getattr(a_				,	a_       )

      args.append(attribute_class.from_pretrained(a_				,	**a_       )       )
    return args





  @property
  def   snake_case_     (       self       )							->			List[Any]:


    """simple docstring"""
    UpperCAmelCase  							=		getattr(self				,	self.attributes[0]       )
    return getattr(a_				,	'model_input_names'				,	a_       )




_a			:	List[Any]								=  copy_func(ProcessorMixin.push_to_hub)
if ProcessorMixin.push_to_hub.__doc__ is not None:
  _a			:	Optional[Any]								=  ProcessorMixin.push_to_hub.__doc__.format(
      object='processor', object_class='AutoProcessor', object_files='processor files'
  )





'''simple docstring'''




import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


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


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

# See all MVP models at https://huggingface.co/models?filter=mvp
_a			:	Tuple								=  {
    'vocab_file': {
        'RUCAIBox/mvp': 'https://huggingface.co/RUCAIBox/mvp/resolve/main/vocab.json',
    },
    'added_tokens.json': {
        'RUCAIBox/mvp': 'https://huggingface.co/RUCAIBox/mvp/resolve/main/added_tokens.json',
    },
    'merges_file': {
        'RUCAIBox/mvp': 'https://huggingface.co/RUCAIBox/mvp/resolve/main/merges.txt',
    },
}

_a			:	str								=  {
    'RUCAIBox/mvp': 1_024,
}


@lru_cache()
def  lowerCamelCase__						(  ):
  UpperCAmelCase  							=		(
      list(range(ord('!'  )    ,				ord('~'  ) + 1  )  ) + list(range(ord('¡'  )    ,				ord('¬'  ) + 1  )  ) + list(range(ord('®'  )    ,				ord('ÿ'  ) + 1  )  )
  )
  UpperCAmelCase  							=		bs[:]
  UpperCAmelCase  							=		0
  for b in range(2**8  ):
    if b not in bs:
      bs.append(SCREAMING_SNAKE_CASE  )
      cs.append(2**8 + n  )
      n += 1
  UpperCAmelCase  							=		[chr(SCREAMING_SNAKE_CASE  ) for n in cs]
  return dict(zip(SCREAMING_SNAKE_CASE    ,				SCREAMING_SNAKE_CASE  )  )


def  lowerCamelCase__						(  SCREAMING_SNAKE_CASE :				List[Any]  ):
  UpperCAmelCase  							=		set()
  UpperCAmelCase  							=		word[0]
  for char in word[1:]:
    pairs.add((prev_char, char)  )
    UpperCAmelCase  							=		char
  return pairs



class     lowercase_		(						a	):

  '''simple docstring'''





  __lowerCAmelCase   :	str      					=				VOCAB_FILES_NAMES
  __lowerCAmelCase   :	List[Any]      					=				PRETRAINED_VOCAB_FILES_MAP
  __lowerCAmelCase   :	Dict      					=				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  __lowerCAmelCase   :	Optional[Any]      					=				["input_ids", "attention_mask"]
  def __init__(       self				,	a_				,	a_				,	a_="replace"				,	a_="<s>"				,	a_="</s>"				,	a_="</s>"				,	a_="<s>"				,	a_="<unk>"				,	a_="<pad>"				,	a_="<mask>"				,	a_=False				,	**a_				,	)							->			Optional[Any]:


    """simple docstring"""
    UpperCAmelCase  							=		AddedToken(a_				,	lstrip=a_				,	rstrip=a_       ) if isinstance(a_				,	a_       ) else bos_token
    UpperCAmelCase  							=		AddedToken(a_				,	lstrip=a_				,	rstrip=a_       ) if isinstance(a_				,	a_       ) else eos_token
    UpperCAmelCase  							=		AddedToken(a_				,	lstrip=a_				,	rstrip=a_       ) if isinstance(a_				,	a_       ) else sep_token
    UpperCAmelCase  							=		AddedToken(a_				,	lstrip=a_				,	rstrip=a_       ) if isinstance(a_				,	a_       ) else cls_token
    UpperCAmelCase  							=		AddedToken(a_				,	lstrip=a_				,	rstrip=a_       ) if isinstance(a_				,	a_       ) else unk_token
    UpperCAmelCase  							=		AddedToken(a_				,	lstrip=a_				,	rstrip=a_       ) if isinstance(a_				,	a_       ) else pad_token

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

    super().__init__(
        errors=a_				,	bos_token=a_				,	eos_token=a_				,	unk_token=a_				,	sep_token=a_				,	cls_token=a_				,	pad_token=a_				,	mask_token=a_				,	add_prefix_space=a_				,	**a_				,	)
    with open(a_				,	encoding='utf-8'       ) as vocab_handle:
      UpperCAmelCase  							=		json.load(a_       )
    UpperCAmelCase  							=		{v: k for k, v in self.encoder.items()}
    UpperCAmelCase  							=		errors  # how to handle errors in decoding
    UpperCAmelCase  							=		bytes_to_unicode()
    UpperCAmelCase  							=		{v: k for k, v in self.byte_encoder.items()}
    with open(a_				,	encoding='utf-8'       ) as merges_handle:
      UpperCAmelCase  							=		merges_handle.read().split('\n'       )[1:-1]
    UpperCAmelCase  							=		[tuple(merge.split()       ) for merge in bpe_merges]
    UpperCAmelCase  							=		dict(zip(a_				,	range(len(a_       )       )       )       )
    UpperCAmelCase  							=		{}
    UpperCAmelCase  							=		add_prefix_space

    # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
    UpperCAmelCase  							=		re.compile(r'\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'       )
  @property
  def   snake_case_     (       self       )							->			Any:


    """simple docstring"""
    return len(self.encoder       )
  def   snake_case_     (       self       )							->			Union[str, Any]:


    """simple docstring"""
    return dict(self.encoder				,	**self.added_tokens_encoder       )
  def   snake_case_     (       self				,	a_       )							->			int:


    """simple docstring"""
    if token in self.cache:
      return self.cache[token]
    UpperCAmelCase  							=		tuple(a_       )
    UpperCAmelCase  							=		get_pairs(a_       )

    if not pairs:
      return token

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

        if word[i] == first and i < len(a_       ) - 1 and word[i + 1] == second:
          new_word.append(first + second       )
          i += 2
        else:
          new_word.append(word[i]       )
          i += 1
      UpperCAmelCase  							=		tuple(a_       )
      UpperCAmelCase  							=		new_word
      if len(a_       ) == 1:
        break
      else:
        UpperCAmelCase  							=		get_pairs(a_       )
    UpperCAmelCase  							=		' '.join(a_       )
    UpperCAmelCase  							=		word
    return word
  def   snake_case_     (       self				,	a_       )							->			int:


    """simple docstring"""
    UpperCAmelCase  							=		[]
    for token in re.findall(self.pat				,	a_       ):
      UpperCAmelCase  							=		''.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(a_       ).split(' '       )       )
    return bpe_tokens
  def   snake_case_     (       self				,	a_       )							->			Optional[int]:


    """simple docstring"""
    return self.encoder.get(a_				,	self.encoder.get(self.unk_token       )       )
  def   snake_case_     (       self				,	a_       )							->			Tuple:


    """simple docstring"""
    return self.decoder.get(a_       )
  def   snake_case_     (       self				,	a_       )							->			List[Any]:


    """simple docstring"""
    UpperCAmelCase  							=		''.join(a_       )
    UpperCAmelCase  							=		bytearray([self.byte_decoder[c] for c in text]       ).decode('utf-8'				,	errors=self.errors       )
    return text
  def   snake_case_     (       self				,	a_				,	a_ = None       )							->			Tuple[str]:


    """simple docstring"""
    if not os.path.isdir(a_       ):
      logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''       )
      return
    UpperCAmelCase  							=		os.path.join(
        a_				,	(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']       )
    UpperCAmelCase  							=		os.path.join(
        a_				,	(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file']       )

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

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

    return vocab_file, merge_file
  def   snake_case_     (       self				,	a_				,	a_ = None       )							->			List[int]:


    """simple docstring"""
    if token_ids_a is None:
      return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
    UpperCAmelCase  							=		[self.cls_token_id]
    UpperCAmelCase  							=		[self.sep_token_id]
    return cls + token_ids_a + sep + sep + token_ids_a + sep
  def   snake_case_     (       self				,	a_				,	a_ = None				,	a_ = False       )							->			List[int]:


    """simple docstring"""
    if already_has_special_tokens:
      return super().get_special_tokens_mask(
          token_ids_a=a_				,	token_ids_a=a_				,	already_has_special_tokens=a_       )

    if token_ids_a is None:
      return [1] + ([0] * len(a_       )) + [1]
    return [1] + ([0] * len(a_       )) + [1, 1] + ([0] * len(a_       )) + [1]
  def   snake_case_     (       self				,	a_				,	a_ = None       )							->			List[int]:


    """simple docstring"""
    UpperCAmelCase  							=		[self.sep_token_id]
    UpperCAmelCase  							=		[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   snake_case_     (       self				,	a_				,	a_=False				,	**a_       )							->			Optional[Any]:


    """simple docstring"""
    UpperCAmelCase  							=		kwargs.pop('add_prefix_space'				,	self.add_prefix_space       )
    if (is_split_into_words or add_prefix_space) and (len(a_       ) > 0 and not text[0].isspace()):
      UpperCAmelCase  							=		' ' + text
    return (text, kwargs)


