






import os
from typing import List, Optional, Union

from ...image_processing_utils import BatchFeature
from ...image_utils import ImageInput
from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import PaddingStrategy, PreTokenizedInput, TextInput, TruncationStrategy
from ...utils import TensorType
from ..auto import AutoTokenizer




class     _lowerCAmelCase(	UpperCAmelCase_ ):





							"""simple docstring"""







							a    :		Optional[Any]									=['''image_processor''', '''tokenizer''']
							a    :		Union[str, Any]									='''BlipImageProcessor'''
							a    :		Union[str, Any]									='''AutoTokenizer'''





							def __init__( self  ,      _lowerCamelCase  ,      _lowerCamelCase  ,      _lowerCamelCase							):
									super().__init__(_lowerCamelCase  ,      _lowerCamelCase							)

									# add QFormer tokenizer
									UpperCamelCase_: Tuple   				=    qformer_tokenizer





							def __call__( self  ,      _lowerCamelCase = None  ,      _lowerCamelCase = None  ,      _lowerCamelCase = True  ,      _lowerCamelCase = False  ,      _lowerCamelCase = None  ,      _lowerCamelCase = None  ,      _lowerCamelCase = 0  ,      _lowerCamelCase = None  ,      _lowerCamelCase = None  ,      _lowerCamelCase = False  ,      _lowerCamelCase = False  ,      _lowerCamelCase = False  ,      _lowerCamelCase = False  ,      _lowerCamelCase = False  ,      _lowerCamelCase = True  ,      _lowerCamelCase = None  ,      **_lowerCamelCase  ,      ):
									if images is None and text is None:
											raise ValueError('You have to specify at least images or text.'							)

									UpperCamelCase_: Tuple   				=    BatchFeature()

									if text is not None:
											UpperCamelCase_: Optional[int]   				=    self.tokenizer(
											    text=_lowerCamelCase  ,      add_special_tokens=_lowerCamelCase  ,      padding=_lowerCamelCase  ,      truncation=_lowerCamelCase  ,      max_length=_lowerCamelCase  ,      stride=_lowerCamelCase  ,      pad_to_multiple_of=_lowerCamelCase  ,      return_attention_mask=_lowerCamelCase  ,      return_overflowing_tokens=_lowerCamelCase  ,      return_special_tokens_mask=_lowerCamelCase  ,      return_offsets_mapping=_lowerCamelCase  ,      return_token_type_ids=_lowerCamelCase  ,      return_length=_lowerCamelCase  ,      verbose=_lowerCamelCase  ,      return_tensors=_lowerCamelCase  ,      **_lowerCamelCase  ,      )
											encoding.update(_lowerCamelCase							)
											UpperCamelCase_: Union[str, Any]   				=    self.qformer_tokenizer(
											    text=_lowerCamelCase  ,      add_special_tokens=_lowerCamelCase  ,      padding=_lowerCamelCase  ,      truncation=_lowerCamelCase  ,      max_length=_lowerCamelCase  ,      stride=_lowerCamelCase  ,      pad_to_multiple_of=_lowerCamelCase  ,      return_attention_mask=_lowerCamelCase  ,      return_overflowing_tokens=_lowerCamelCase  ,      return_special_tokens_mask=_lowerCamelCase  ,      return_offsets_mapping=_lowerCamelCase  ,      return_token_type_ids=_lowerCamelCase  ,      return_length=_lowerCamelCase  ,      verbose=_lowerCamelCase  ,      return_tensors=_lowerCamelCase  ,      **_lowerCamelCase  ,      )
											UpperCamelCase_: Dict   				=    qformer_text_encoding.pop('input_ids'							)
											UpperCamelCase_: int   				=    qformer_text_encoding.pop('attention_mask'							)

									if images is not None:
											UpperCamelCase_: int   				=    self.image_processor(_lowerCamelCase  ,      return_tensors=_lowerCamelCase							)
											encoding.update(_lowerCamelCase							)

									return encoding





							def       _a   ( self  ,      *_lowerCamelCase  ,      **_lowerCamelCase							):
									return self.tokenizer.batch_decode(*_lowerCamelCase  ,      **_lowerCamelCase							)





							def       _a   ( self  ,      *_lowerCamelCase  ,      **_lowerCamelCase							):
									return self.tokenizer.decode(*_lowerCamelCase  ,      **_lowerCamelCase							)





							@property
							# Copied from transformers.models.blip.processing_blip.BlipProcessor.model_input_names
							def       _a   ( self							):
									UpperCamelCase_: Optional[int]   				=    self.tokenizer.model_input_names
									UpperCamelCase_: Union[str, Any]   				=    self.image_processor.model_input_names
									return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names							)							)





							def       _a   ( self  ,      _lowerCamelCase  ,      **_lowerCamelCase							):
									if os.path.isfile(_lowerCamelCase							):
											raise ValueError(f'''Provided path ({save_directory}) should be a directory, not a file'''							)
									os.makedirs(_lowerCamelCase  ,      exist_ok=_lowerCamelCase							)
									UpperCamelCase_: List[str]   				=    os.path.join(_lowerCamelCase  ,      'qformer_tokenizer'							)
									self.qformer_tokenizer.save_pretrained(_lowerCamelCase							)
									return super().save_pretrained(_lowerCamelCase  ,      **_lowerCamelCase							)





							@classmethod
							def       _a   ( cls  ,      _lowerCamelCase  ,      **_lowerCamelCase							):
									UpperCamelCase_: Tuple   				=    AutoTokenizer.from_pretrained(_lowerCamelCase  ,      subfolder='qformer_tokenizer'							)
									UpperCamelCase_: int   				=    cls._get_arguments_from_pretrained(_lowerCamelCase  ,      **_lowerCamelCase							)
									args.append(_lowerCamelCase							)
									return cls(*_lowerCamelCase							)







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

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

from transformers import (
    AlbertTokenizer,
    AutoTokenizer,
    BertTokenizer,
    BertTokenizerFast,
    GPTaTokenizerFast,
    is_tokenizers_available,
)
from transformers.testing_utils import TOKEN, USER, is_staging_test, require_tokenizers
from transformers.tokenization_utils import Trie


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

from test_module.custom_tokenization import CustomTokenizer  # noqa E402


if is_tokenizers_available():
   from test_module.custom_tokenization_fast import CustomTokenizerFast




class     _lowerCAmelCase(	unittest.TestCase ):





       """simple docstring"""







       def       _a   ( self							):
         # A mock response for an HTTP head request to emulate server down
         UpperCamelCase_: Any   				=    mock.Mock()
         UpperCamelCase_: Dict   				=    5_0_0
         UpperCamelCase_: Any   				=    {}
         UpperCamelCase_: Tuple   				=    HTTPError
         UpperCamelCase_: List[str]   				=    {}

         # Download this model to make sure it's in the cache.
         UpperCamelCase_: int   				=    BertTokenizer.from_pretrained('hf-internal-testing/tiny-random-bert'							)

         # Under the mock environment we get a 500 error when trying to reach the tokenizer.
         with mock.patch('requests.Session.request'  ,      return_value=_lowerCamelCase							) as mock_head:
           UpperCamelCase_: Optional[int]   				=    BertTokenizer.from_pretrained('hf-internal-testing/tiny-random-bert'							)
           # This check we did call the fake head request
           mock_head.assert_called()





       @require_tokenizers
       def       _a   ( self							):
         # A mock response for an HTTP head request to emulate server down
         UpperCamelCase_: Union[str, Any]   				=    mock.Mock()
         UpperCamelCase_: Union[str, Any]   				=    5_0_0
         UpperCamelCase_: str   				=    {}
         UpperCamelCase_: List[str]   				=    HTTPError
         UpperCamelCase_: Optional[int]   				=    {}

         # Download this model to make sure it's in the cache.
         UpperCamelCase_: List[str]   				=    GPTaTokenizerFast.from_pretrained('gpt2'							)

         # Under the mock environment we get a 500 error when trying to reach the tokenizer.
         with mock.patch('requests.Session.request'  ,      return_value=_lowerCamelCase							) as mock_head:
           UpperCamelCase_: str   				=    GPTaTokenizerFast.from_pretrained('gpt2'							)
           # This check we did call the fake head request
           mock_head.assert_called()





       def       _a   ( self							):
         # This test is for deprecated behavior and can be removed in v5
         try:
           UpperCamelCase_: Optional[int]   				=    tempfile.mktemp()
           with open(_lowerCamelCase  ,      'wb'							) as f:
             http_get('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'  ,      _lowerCamelCase							)

           UpperCamelCase_: Tuple   				=    AlbertTokenizer.from_pretrained(_lowerCamelCase							)
         finally:
           os.remove(_lowerCamelCase							)

         # Supporting this legacy load introduced a weird bug where the tokenizer would load local files if they are in
         # the current folder and have the right name.
         if os.path.isfile('tokenizer.json'							):
           # We skip the test if the user has a `tokenizer.json` in this folder to avoid deleting it.
           return
         try:
           with open('tokenizer.json'  ,      'wb'							) as f:
             http_get('https://huggingface.co/hf-internal-testing/tiny-random-bert/blob/main/tokenizer.json'  ,      _lowerCamelCase							)
           UpperCamelCase_: List[str]   				=    AutoTokenizer.from_pretrained('hf-internal-testing/tiny-random-gpt2'							)
           # The tiny random BERT has a vocab size of 1024, tiny gpt2 as a vocab size of 1000
           self.assertEqual(tokenizer.vocab_size  ,      1_0_0_0							)
           # Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

         finally:
           os.remove('tokenizer.json'							)





       def       _a   ( self							):
         # This test is for deprecated behavior and can be removed in v5
         UpperCamelCase_: Any   				=    AlbertTokenizer.from_pretrained('https://huggingface.co/albert-base-v1/resolve/main/spiece.model'							)







@is_staging_test
class     _lowerCAmelCase(	unittest.TestCase ):





       """simple docstring"""







       a    :		Dict									=['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']





       @classmethod
       def       _a   ( cls							):
         UpperCamelCase_: Optional[int]   				=    TOKEN
         HfFolder.save_token(_lowerCamelCase							)





       @classmethod
       def       _a   ( cls							):
         try:
           delete_repo(token=cls._token  ,      repo_id='test-tokenizer'							)
         except HTTPError:
           pass

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

         try:
           delete_repo(token=cls._token  ,      repo_id='test-dynamic-tokenizer'							)
         except HTTPError:
           pass





       def       _a   ( self							):
         with tempfile.TemporaryDirectory() as tmp_dir:
           UpperCamelCase_: List[Any]   				=    os.path.join(_lowerCamelCase  ,      'vocab.txt'							)
           with open(_lowerCamelCase  ,      'w'  ,      encoding='utf-8'							) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]							)							)
           UpperCamelCase_: int   				=    BertTokenizer(_lowerCamelCase							)

         tokenizer.push_to_hub('test-tokenizer'  ,      use_auth_token=self._token							)
         UpperCamelCase_: Union[str, Any]   				=    BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''							)
         self.assertDictEqual(new_tokenizer.vocab  ,      tokenizer.vocab							)

         # Reset repo
         delete_repo(token=self._token  ,      repo_id='test-tokenizer'							)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           tokenizer.save_pretrained(_lowerCamelCase  ,      repo_id='test-tokenizer'  ,      push_to_hub=_lowerCamelCase  ,      use_auth_token=self._token							)

         UpperCamelCase_: List[str]   				=    BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''							)
         self.assertDictEqual(new_tokenizer.vocab  ,      tokenizer.vocab							)





       def       _a   ( self							):
         with tempfile.TemporaryDirectory() as tmp_dir:
           UpperCamelCase_: Optional[Any]   				=    os.path.join(_lowerCamelCase  ,      'vocab.txt'							)
           with open(_lowerCamelCase  ,      'w'  ,      encoding='utf-8'							) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]							)							)
           UpperCamelCase_: Union[str, Any]   				=    BertTokenizer(_lowerCamelCase							)

         tokenizer.push_to_hub('valid_org/test-tokenizer-org'  ,      use_auth_token=self._token							)
         UpperCamelCase_: Dict   				=    BertTokenizer.from_pretrained('valid_org/test-tokenizer-org'							)
         self.assertDictEqual(new_tokenizer.vocab  ,      tokenizer.vocab							)

         # Reset repo
         delete_repo(token=self._token  ,      repo_id='valid_org/test-tokenizer-org'							)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
           tokenizer.save_pretrained(
               _lowerCamelCase  ,      repo_id='valid_org/test-tokenizer-org'  ,      push_to_hub=_lowerCamelCase  ,      use_auth_token=self._token							)

         UpperCamelCase_: Optional[Any]   				=    BertTokenizer.from_pretrained('valid_org/test-tokenizer-org'							)
         self.assertDictEqual(new_tokenizer.vocab  ,      tokenizer.vocab							)





       @require_tokenizers
       def       _a   ( self							):
         CustomTokenizer.register_for_auto_class()
         with tempfile.TemporaryDirectory() as tmp_dir:
           UpperCamelCase_: Optional[Any]   				=    os.path.join(_lowerCamelCase  ,      'vocab.txt'							)
           with open(_lowerCamelCase  ,      'w'  ,      encoding='utf-8'							) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]							)							)
           UpperCamelCase_: Optional[int]   				=    CustomTokenizer(_lowerCamelCase							)

         # No fast custom tokenizer
         tokenizer.push_to_hub('test-dynamic-tokenizer'  ,      use_auth_token=self._token							)

         UpperCamelCase_: str   				=    AutoTokenizer.from_pretrained(f'''{USER}/test-dynamic-tokenizer'''  ,      trust_remote_code=_lowerCamelCase							)
         # Can't make an isinstance check because the new_model.config is from the CustomTokenizer class of a dynamic module
         self.assertEqual(tokenizer.__class__.__name__  ,      'CustomTokenizer'							)

         # Fast and slow custom tokenizer
         CustomTokenizerFast.register_for_auto_class()
         with tempfile.TemporaryDirectory() as tmp_dir:
           UpperCamelCase_: Optional[Any]   				=    os.path.join(_lowerCamelCase  ,      'vocab.txt'							)
           with open(_lowerCamelCase  ,      'w'  ,      encoding='utf-8'							) as vocab_writer:
             vocab_writer.write(''.join([x + '\n' for x in self.vocab_tokens]							)							)

           UpperCamelCase_: Dict   				=    BertTokenizerFast.from_pretrained(_lowerCamelCase							)
           bert_tokenizer.save_pretrained(_lowerCamelCase							)
           UpperCamelCase_: List[str]   				=    CustomTokenizerFast.from_pretrained(_lowerCamelCase							)

         tokenizer.push_to_hub('test-dynamic-tokenizer'  ,      use_auth_token=self._token							)

         UpperCamelCase_: Tuple   				=    AutoTokenizer.from_pretrained(f'''{USER}/test-dynamic-tokenizer'''  ,      trust_remote_code=_lowerCamelCase							)
         # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
         self.assertEqual(tokenizer.__class__.__name__  ,      'CustomTokenizerFast'							)
         UpperCamelCase_: int   				=    AutoTokenizer.from_pretrained(
             f'''{USER}/test-dynamic-tokenizer'''  ,      use_fast=_lowerCamelCase  ,      trust_remote_code=_lowerCamelCase							)
         # Can't make an isinstance check because the new_model.config is from the FakeConfig class of a dynamic module
         self.assertEqual(tokenizer.__class__.__name__  ,      'CustomTokenizer'							)







class     _lowerCAmelCase(	unittest.TestCase ):





       """simple docstring"""







       def       _a   ( self							):
         UpperCamelCase_: Dict   				=    Trie()
         trie.add('Hello 友達'							)
         self.assertEqual(trie.data  ,      {'H': {'e': {'l': {'l': {'o': {' ': {'友': {'達': {'': 1}}}}}}}}}							)
         trie.add('Hello'							)
         trie.data
         self.assertEqual(trie.data  ,      {'H': {'e': {'l': {'l': {'o': {'': 1, ' ': {'友': {'達': {'': 1}}}}}}}}}							)





       def       _a   ( self							):
         UpperCamelCase_: Optional[int]   				=    Trie()
         self.assertEqual(trie.split('[CLS] This is a extra_id_100'							)  ,      ['[CLS] This is a extra_id_100']							)
         trie.add('[CLS]'							)
         trie.add('extra_id_1'							)
         trie.add('extra_id_100'							)
         self.assertEqual(trie.split('[CLS] This is a extra_id_100'							)  ,      ['[CLS]', ' This is a ', 'extra_id_100']							)





       def       _a   ( self							):
         UpperCamelCase_: int   				=    Trie()
         trie.add('A'							)
         self.assertEqual(trie.split('ABC'							)  ,      ['A', 'BC']							)
         self.assertEqual(trie.split('BCA'							)  ,      ['BC', 'A']							)





       def       _a   ( self							):
         UpperCamelCase_: str   				=    Trie()
         trie.add('TOKEN]'							)
         trie.add('[SPECIAL_TOKEN]'							)
         self.assertEqual(trie.split('This is something [SPECIAL_TOKEN]'							)  ,      ['This is something ', '[SPECIAL_TOKEN]']							)





       def       _a   ( self							):
         UpperCamelCase_: Union[str, Any]   				=    Trie()
         trie.add('A'							)
         trie.add('P'							)
         trie.add('[SPECIAL_TOKEN]'							)
         self.assertEqual(trie.split('This is something [SPECIAL_TOKEN]'							)  ,      ['This is something ', '[SPECIAL_TOKEN]']							)





       def       _a   ( self							):
         UpperCamelCase_: List[str]   				=    Trie()
         trie.add('AB'							)
         trie.add('B'							)
         trie.add('C'							)
         self.assertEqual(trie.split('ABC'							)  ,      ['AB', 'C']							)





       def       _a   ( self							):
         UpperCamelCase_: List[str]   				=    Trie()
         trie.add('ABC'							)
         trie.add('B'							)
         trie.add('CD'							)
         self.assertEqual(trie.split('ABCD'							)  ,      ['ABC', 'D']							)





       def       _a   ( self							):
         # Even if the offsets are wrong, we necessarily output correct string
         # parts.
         UpperCamelCase_: Union[str, Any]   				=    Trie()
         UpperCamelCase_: Any   				=    trie.cut_text('ABC'  ,      [0, 0, 2, 1, 2, 3]							)
         self.assertEqual(_lowerCamelCase  ,      ['AB', 'C']							)