


'''simple docstring'''







import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoImageProcessor, ViTImageProcessor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / 'utils'))

from test_module.custom_image_processing import CustomImageProcessor  # noqa E402


SCREAMING_SNAKE_CASE_      =			get_tests_dir('fixtures')


class  lowerCAmelCase_						(       unittest.TestCase      ):





  """simple docstring"""



  def __a						(					self     :							Dict						):





       '''simple docstring'''

       __a       =   mock.Mock()
       __a       =   5_0_0
       __a       =   {}
       __a       =   HTTPError
       __a       =   {}

       # Download this model to make sure it's in the cache.
       __a       =   ViTImageProcessor.from_pretrained("""hf-internal-testing/tiny-random-vit"""						)
       # Under the mock environment we get a 500 error when trying to reach the model.
       with mock.patch("""requests.Session.request"""				,							return_value=SCREAMING_SNAKE_CASE__						) as mock_head:
            __a       =   ViTImageProcessor.from_pretrained("""hf-internal-testing/tiny-random-vit"""						)
            # This check we did call the fake head request
            mock_head.assert_called()

  def __a						(					self     :							List[str]						):





       '''simple docstring'''

       __a       =   ViTImageProcessor.from_pretrained(
           """https://huggingface.co/hf-internal-testing/tiny-random-vit/resolve/main/preprocessor_config.json"""						)


  def __a						(					self     :							Any						):





       '''simple docstring'''

       with self.assertRaises(SCREAMING_SNAKE_CASE__						):
            # config is in subfolder, the following should not work without specifying the subfolder
            __a       =   AutoImageProcessor.from_pretrained("""hf-internal-testing/stable-diffusion-all-variants"""						)

       __a       =   AutoImageProcessor.from_pretrained(
           """hf-internal-testing/stable-diffusion-all-variants"""				,							subfolder="""feature_extractor"""						)

       self.assertIsNotNone(SCREAMING_SNAKE_CASE__						)


@is_staging_test
class  lowerCAmelCase_						(       unittest.TestCase      ):





  """simple docstring"""



  @classmethod
  def __a						(					cls     :							str						):





       '''simple docstring'''

       __a       =   TOKEN
       HfFolder.save_token(SCREAMING_SNAKE_CASE__						)

  @classmethod
  def __a						(					cls     :							List[str]						):





       '''simple docstring'''

       try:
            delete_repo(token=cls._token				,							repo_id="""test-image-processor"""						)
       except HTTPError:
            pass

       try:
            delete_repo(token=cls._token				,							repo_id="""valid_org/test-image-processor-org"""						)
       except HTTPError:
            pass

       try:
            delete_repo(token=cls._token				,							repo_id="""test-dynamic-image-processor"""						)
       except HTTPError:
            pass

  def __a						(					self     :							Tuple						):





       '''simple docstring'''

       __a       =   ViTImageProcessor.from_pretrained(SCREAMING_SNAKE_CASE__						)
       image_processor.push_to_hub("""test-image-processor"""				,							use_auth_token=self._token						)

       __a       =   ViTImageProcessor.from_pretrained(f'''{USER}/test-image-processor'''						)
       for k, v in image_processor.__dict__.items():
            self.assertEqual(SCREAMING_SNAKE_CASE__				,							getattr(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__						)						)

       # Reset repo
       delete_repo(token=self._token				,							repo_id="""test-image-processor"""						)

       # Push to hub via save_pretrained
       with tempfile.TemporaryDirectory() as tmp_dir:
            image_processor.save_pretrained(
                SCREAMING_SNAKE_CASE__				,							repo_id="""test-image-processor"""				,							push_to_hub=SCREAMING_SNAKE_CASE__				,							use_auth_token=self._token						)

       __a       =   ViTImageProcessor.from_pretrained(f'''{USER}/test-image-processor'''						)
       for k, v in image_processor.__dict__.items():
            self.assertEqual(SCREAMING_SNAKE_CASE__				,							getattr(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__						)						)

  def __a						(					self     :							str						):





       '''simple docstring'''

       __a       =   ViTImageProcessor.from_pretrained(SCREAMING_SNAKE_CASE__						)
       image_processor.push_to_hub("""valid_org/test-image-processor"""				,							use_auth_token=self._token						)

       __a       =   ViTImageProcessor.from_pretrained("""valid_org/test-image-processor"""						)
       for k, v in image_processor.__dict__.items():
            self.assertEqual(SCREAMING_SNAKE_CASE__				,							getattr(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__						)						)

       # Reset repo
       delete_repo(token=self._token				,							repo_id="""valid_org/test-image-processor"""						)

       # Push to hub via save_pretrained
       with tempfile.TemporaryDirectory() as tmp_dir:
            image_processor.save_pretrained(
                SCREAMING_SNAKE_CASE__				,							repo_id="""valid_org/test-image-processor-org"""				,							push_to_hub=SCREAMING_SNAKE_CASE__				,							use_auth_token=self._token						)

       __a       =   ViTImageProcessor.from_pretrained("""valid_org/test-image-processor-org"""						)
       for k, v in image_processor.__dict__.items():
            self.assertEqual(SCREAMING_SNAKE_CASE__				,							getattr(SCREAMING_SNAKE_CASE__				,							SCREAMING_SNAKE_CASE__						)						)


  def __a						(					self     :							List[str]						):





       '''simple docstring'''

       CustomImageProcessor.register_for_auto_class()
       __a       =   CustomImageProcessor.from_pretrained(SCREAMING_SNAKE_CASE__						)

       image_processor.push_to_hub("""test-dynamic-image-processor"""				,							use_auth_token=self._token						)

       # This has added the proper auto_map field to the config
       self.assertDictEqual(
           image_processor.auto_map				,							{"""AutoImageProcessor""": """custom_image_processing.CustomImageProcessor"""}				,							)

       __a       =   AutoImageProcessor.from_pretrained(
           f'''{USER}/test-dynamic-image-processor'''				,							trust_remote_code=SCREAMING_SNAKE_CASE__						)
       # Can't make an isinstance check because the new_image_processor is from the CustomImageProcessor class of a dynamic module
       self.assertEqual(new_image_processor.__class__.__name__				,							"""CustomImageProcessor"""						)










'''simple docstring'''







import os
from shutil import copyfile
from typing import List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


SCREAMING_SNAKE_CASE_      =			logging.get_logger(__name__)

SCREAMING_SNAKE_CASE_      =			{'vocab_file': 'sentencepiece.model'}

SCREAMING_SNAKE_CASE_      =			{
    'vocab_file': {
        'google/rembert': 'https://huggingface.co/google/rembert/resolve/main/sentencepiece.model',
    },
}

SCREAMING_SNAKE_CASE_      =			{
    'google/rembert': 2_56,
}


class  lowerCAmelCase_						(       snake_case__      ):





  """simple docstring"""



  a_       :Tuple												=VOCAB_FILES_NAMES
  a_       :Tuple												=PRETRAINED_VOCAB_FILES_MAP
  a_       :Optional[int]												=PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES

  def __init__(					self     :							Union[str, Any]				,							SCREAMING_SNAKE_CASE__     :							str				,							SCREAMING_SNAKE_CASE__     :							Tuple=False				,							SCREAMING_SNAKE_CASE__     :							Optional[Any]=True				,							SCREAMING_SNAKE_CASE__     :							Union[str, Any]=True				,							SCREAMING_SNAKE_CASE__     :							int="[CLS]"				,							SCREAMING_SNAKE_CASE__     :							Optional[int]="[SEP]"				,							SCREAMING_SNAKE_CASE__     :							Optional[Any]="[UNK]"				,							SCREAMING_SNAKE_CASE__     :							Any="[SEP]"				,							SCREAMING_SNAKE_CASE__     :							Dict="[PAD]"				,							SCREAMING_SNAKE_CASE__     :							Dict="[CLS]"				,							SCREAMING_SNAKE_CASE__     :							Dict="[MASK]"				,							**SCREAMING_SNAKE_CASE__     :							List[str]				,							):





       '''simple docstring'''

       super().__init__(
           do_lower_case=SCREAMING_SNAKE_CASE__				,							remove_space=SCREAMING_SNAKE_CASE__				,							keep_accents=SCREAMING_SNAKE_CASE__				,							bos_token=SCREAMING_SNAKE_CASE__				,							eos_token=SCREAMING_SNAKE_CASE__				,							unk_token=SCREAMING_SNAKE_CASE__				,							sep_token=SCREAMING_SNAKE_CASE__				,							pad_token=SCREAMING_SNAKE_CASE__				,							cls_token=SCREAMING_SNAKE_CASE__				,							mask_token=SCREAMING_SNAKE_CASE__				,							**SCREAMING_SNAKE_CASE__				,							)

       __a       =   do_lower_case
       __a       =   remove_space
       __a       =   keep_accents
       __a       =   vocab_file

       __a       =   spm.SentencePieceProcessor()
       self.sp_model.Load(SCREAMING_SNAKE_CASE__						)

  @property
  def __a						(					self     :							List[str]						):





       '''simple docstring'''

       return len(self.sp_model						)

  def __a						(					self     :							str						):





       '''simple docstring'''

       __a       =   {self.convert_ids_to_tokens(SCREAMING_SNAKE_CASE__						): i for i in range(self.vocab_size						)}
       vocab.update(self.added_tokens_encoder						)
       return vocab

  def __getstate__(					self     :							List[Any]						):





       '''simple docstring'''

       __a       =   self.__dict__.copy()
       __a       =   None
       return state

  def __setstate__(					self     :							Dict				,							SCREAMING_SNAKE_CASE__     :							Optional[int]						):





       '''simple docstring'''

       __a       =   d
       __a       =   spm.SentencePieceProcessor()
       self.sp_model.Load(self.vocab_file						)

  def __a						(					self     :							List[Any]				,							SCREAMING_SNAKE_CASE__     :							Tuple				,							SCREAMING_SNAKE_CASE__     :							List[str]=False						):





       '''simple docstring'''

       __a       =   self.sp_model.EncodeAsPieces(SCREAMING_SNAKE_CASE__						)
       return pieces

  def __a						(					self     :							Optional[Any]				,							SCREAMING_SNAKE_CASE__     :							int						):





       '''simple docstring'''

       return self.sp_model.PieceToId(SCREAMING_SNAKE_CASE__						)

  def __a						(					self     :							List[Any]				,							SCREAMING_SNAKE_CASE__     :							List[Any]						):





       '''simple docstring'''

       return self.sp_model.IdToPiece(SCREAMING_SNAKE_CASE__						)

  def __a						(					self     :							List[str]				,							SCREAMING_SNAKE_CASE__     :							str						):





       '''simple docstring'''

       __a       =   self.sp_model.decode_pieces(SCREAMING_SNAKE_CASE__						)
       return out_string

  def __a						(					self     :							Tuple				,							SCREAMING_SNAKE_CASE__     :							List[int]				,							SCREAMING_SNAKE_CASE__     :							Optional[List[int]] = None						):





       '''simple docstring'''

       __a       =   [self.sep_token_id]
       __a       =   [self.cls_token_id]
       if token_ids_a is None:
            return cls + token_ids_a + sep
       return cls + token_ids_a + sep + token_ids_a + sep

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





       '''simple docstring'''

       if already_has_special_tokens:
            if token_ids_a is not None:
                 raise ValueError(
                     """You should not supply a second sequence if the provided sequence of """
                     """ids is already formatted with special tokens for the model."""						)
            return [1 if x in [self.sep_token_id, self.cls_token_id] else 0 for x in token_ids_a]

       if token_ids_a is not None:
            return [1] + ([0] * len(SCREAMING_SNAKE_CASE__						)) + [1] + ([0] * len(SCREAMING_SNAKE_CASE__						)) + [1]
       return [1] + ([0] * len(SCREAMING_SNAKE_CASE__						)) + [1]

  def __a						(					self     :							Union[str, Any]				,							SCREAMING_SNAKE_CASE__     :							List[int]				,							SCREAMING_SNAKE_CASE__     :							Optional[List[int]] = None						):





       '''simple docstring'''

       __a       =   [self.sep_token_id]
       __a       =   [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 __a						(					self     :							int				,							SCREAMING_SNAKE_CASE__     :							str				,							SCREAMING_SNAKE_CASE__     :							Optional[str] = None						):





       '''simple docstring'''

       if not os.path.isdir(SCREAMING_SNAKE_CASE__						):
            logger.error("""Vocabulary path ({}) should be a directory""".format(SCREAMING_SNAKE_CASE__						)						)
            return
       __a       =   os.path.join(
           SCREAMING_SNAKE_CASE__				,							(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]						)

       if os.path.abspath(self.vocab_file						) != os.path.abspath(SCREAMING_SNAKE_CASE__						):
            copyfile(self.vocab_file				,							SCREAMING_SNAKE_CASE__						)

       return (out_vocab_file,)






