




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 __snake_case			(			unittest.TestCase		):







   def lowerCamelCase						(      self		:						Optional[int]):

        """simple docstring"""


        UpperCAmelCase_          =							mock.Mock()
        UpperCAmelCase_          =							500
        UpperCAmelCase_          =							{}
        UpperCAmelCase_          =							HTTPError
        UpperCAmelCase_          =							{}

        # Download this model to make sure it's in the cache.
        UpperCAmelCase_          =							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=_snake_case) as mock_head:
             UpperCAmelCase_          =							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 lowerCamelCase						(      self		:						Any):

        """simple docstring"""


        UpperCAmelCase_          =							mock.Mock()
        UpperCAmelCase_          =							500
        UpperCAmelCase_          =							{}
        UpperCAmelCase_          =							HTTPError
        UpperCAmelCase_          =							{}

        # Download this model to make sure it's in the cache.
        UpperCAmelCase_          =							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=_snake_case) as mock_head:
             UpperCAmelCase_          =							GPTaTokenizerFast.from_pretrained('''gpt2''')
             # This check we did call the fake head request
             mock_head.assert_called()







   def lowerCamelCase						(      self		:						Tuple):

        """simple docstring"""


        try:
             UpperCAmelCase_          =							tempfile.mktemp()
             with open(_snake_case					,						'''wb''') as f:
                  http_get('''https://huggingface.co/albert-base-v1/resolve/main/spiece.model'''					,						_snake_case)

             UpperCAmelCase_          =							AlbertTokenizer.from_pretrained(_snake_case)
        finally:
             os.remove(_snake_case)

        # 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'''					,						_snake_case)
             UpperCAmelCase_          =							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					,						1000)
             # Tokenizer should depend on the remote checkpoint, not the local tokenizer.json file.

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







   def lowerCamelCase						(      self		:						Optional[int]):

        """simple docstring"""


        UpperCAmelCase_          =							AlbertTokenizer.from_pretrained('''https://huggingface.co/albert-base-v1/resolve/main/spiece.model''')







@is_staging_test
class __snake_case			(			unittest.TestCase		):
   UpperCAmelCase__  :    str          =  ['''[UNK]''', '''[CLS]''', '''[SEP]''', '''[PAD]''', '''[MASK]''', '''bla''', '''blou''']







   @classmethod
   def lowerCamelCase						(      cls		:						Union[str, Any]):

        """simple docstring"""


        UpperCAmelCase_          =							TOKEN
        HfFolder.save_token(_snake_case)







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

        """simple docstring"""


        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 lowerCamelCase						(      self		:						Tuple):

        """simple docstring"""


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

        tokenizer.push_to_hub('''test-tokenizer'''					,						use_auth_token=self._token)
        UpperCAmelCase_          =							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(_snake_case					,						repo_id='''test-tokenizer'''					,						push_to_hub=_snake_case					,						use_auth_token=self._token)

        UpperCAmelCase_          =							BertTokenizer.from_pretrained(F"""{USER}/test-tokenizer""")
        self.assertDictEqual(new_tokenizer.vocab					,						tokenizer.vocab)







   def lowerCamelCase						(      self		:						List[Any]):

        """simple docstring"""


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

        tokenizer.push_to_hub('''valid_org/test-tokenizer-org'''					,						use_auth_token=self._token)
        UpperCAmelCase_          =							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(
                 _snake_case					,						repo_id='''valid_org/test-tokenizer-org'''					,						push_to_hub=_snake_case					,						use_auth_token=self._token)

        UpperCAmelCase_          =							BertTokenizer.from_pretrained('''valid_org/test-tokenizer-org''')
        self.assertDictEqual(new_tokenizer.vocab					,						tokenizer.vocab)







   @require_tokenizers
   def lowerCamelCase						(      self		:						int):

        """simple docstring"""


        CustomTokenizer.register_for_auto_class()
        with tempfile.TemporaryDirectory() as tmp_dir:
             UpperCAmelCase_          =							os.path.join(_snake_case					,						'''vocab.txt''')
             with open(_snake_case					,						'''w'''					,						encoding='''utf-8''') as vocab_writer:
                  vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]))
             UpperCAmelCase_          =							CustomTokenizer(_snake_case)

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

        UpperCAmelCase_          =							AutoTokenizer.from_pretrained(F"""{USER}/test-dynamic-tokenizer"""					,						trust_remote_code=_snake_case)
        # 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_          =							os.path.join(_snake_case					,						'''vocab.txt''')
             with open(_snake_case					,						'''w'''					,						encoding='''utf-8''') as vocab_writer:
                  vocab_writer.write(''''''.join([x + '''\n''' for x in self.vocab_tokens]))

             UpperCAmelCase_          =							BertTokenizerFast.from_pretrained(_snake_case)
             bert_tokenizer.save_pretrained(_snake_case)
             UpperCAmelCase_          =							CustomTokenizerFast.from_pretrained(_snake_case)

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

        UpperCAmelCase_          =							AutoTokenizer.from_pretrained(F"""{USER}/test-dynamic-tokenizer"""					,						trust_remote_code=_snake_case)
        # 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_          =							AutoTokenizer.from_pretrained(
            F"""{USER}/test-dynamic-tokenizer"""					,						use_fast=_snake_case					,						trust_remote_code=_snake_case)
        # 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 __snake_case			(			unittest.TestCase		):







   def lowerCamelCase						(      self		:						Any):

        """simple docstring"""


        UpperCAmelCase_          =							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 lowerCamelCase						(      self		:						str):

        """simple docstring"""


        UpperCAmelCase_          =							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 lowerCamelCase						(      self		:						Tuple):

        """simple docstring"""


        UpperCAmelCase_          =							Trie()
        trie.add('''A''')
        self.assertEqual(trie.split('''ABC''')					,						['''A''', '''BC'''])
        self.assertEqual(trie.split('''BCA''')					,						['''BC''', '''A'''])







   def lowerCamelCase						(      self		:						str):

        """simple docstring"""


        UpperCAmelCase_          =							Trie()
        trie.add('''TOKEN]''')
        trie.add('''[SPECIAL_TOKEN]''')
        self.assertEqual(trie.split('''This is something [SPECIAL_TOKEN]''')					,						['''This is something ''', '''[SPECIAL_TOKEN]'''])







   def lowerCamelCase						(      self		:						str):

        """simple docstring"""


        UpperCAmelCase_          =							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 lowerCamelCase						(      self		:						Tuple):

        """simple docstring"""


        UpperCAmelCase_          =							Trie()
        trie.add('''AB''')
        trie.add('''B''')
        trie.add('''C''')
        self.assertEqual(trie.split('''ABC''')					,						['''AB''', '''C'''])







   def lowerCamelCase						(      self		:						Tuple):

        """simple docstring"""


        UpperCAmelCase_          =							Trie()
        trie.add('''ABC''')
        trie.add('''B''')
        trie.add('''CD''')
        self.assertEqual(trie.split('''ABCD''')					,						['''ABC''', '''D'''])







   def lowerCamelCase						(      self		:						Tuple):

        """simple docstring"""


        UpperCAmelCase_          =							Trie()
        UpperCAmelCase_          =							trie.cut_text('''ABC'''					,						[0, 0, 2, 1, 2, 3])
        self.assertEqual(_snake_case					,						['''AB''', '''C'''])











from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
				from PIL import Image

				from ..image_utils import load_image

if is_tf_available():
				from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_VISION_2_SEQ_MAPPING

if is_torch_available():
				import torch

				from ..models.auto.modeling_auto import MODEL_FOR_VISION_2_SEQ_MAPPING

_lowercase: Union[str, Any]      			=		logging.get_logger(__name__)



@add_end_docstrings(lowerCAmelCase	)
class 	_lowercase							(   lowerCAmelCase	):





	"""simple docstring"""




	def __init__(self		,       *lowerCamelCase_		,       **lowerCamelCase_		):
						"""simple docstring"""

						super().__init__(*lowerCamelCase_		,       **lowerCamelCase_		)
						requires_backends(self		,       "vision"		)
						self.check_model_type(
						    TF_MODEL_FOR_VISION_2_SEQ_MAPPING if self.framework == "tf" else MODEL_FOR_VISION_2_SEQ_MAPPING		)




	def      UpperCamelCase_       (self		,       lowerCamelCase_=None		,       lowerCamelCase_=None		,       lowerCamelCase_=None		):
						"""simple docstring"""

						a					     =   {}
						a					     =   {}

						if prompt is not None:
											a					     =   prompt

						if generate_kwargs is not None:
											a					     =   generate_kwargs
						if max_new_tokens is not None:
											if "generate_kwargs" not in forward_kwargs:
																a					     =   {}
											if "max_new_tokens" in forward_kwargs["generate_kwargs"]:
																raise ValueError(
																    "'max_new_tokens' is defined twice, once in 'generate_kwargs' and once as a direct parameter,"
																    " please use only one"		)
											a					     =   max_new_tokens
						return preprocess_params, forward_kwargs, {}




	def __call__(self		,       lowerCamelCase_		,       **lowerCamelCase_		):
						"""simple docstring"""

						return super().__call__(lowerCamelCase_		,       **lowerCamelCase_		)




	def      UpperCamelCase_       (self		,       lowerCamelCase_		,       lowerCamelCase_=None		):
						"""simple docstring"""

						a					     =   load_image(lowerCamelCase_		)

						if prompt is not None:
											if not isinstance(lowerCamelCase_		,       lowerCamelCase_		):
																raise ValueError(
																    F'''Received an invalid text input, got - {type(lowerCamelCase_		)} - but expected a single string. '''
																    "Note also that one single text can be provided for conditional image to text generation."		)

											a					     =   self.model.config.model_type

											if model_type == "git":
																a					     =   self.image_processor(images=lowerCamelCase_		,       return_tensors=self.framework		)
																a					     =   self.tokenizer(text=lowerCamelCase_		,       add_special_tokens=lowerCamelCase_		).input_ids
																a					     =   [self.tokenizer.cls_token_id] + input_ids
																a					     =   torch.tensor(lowerCamelCase_		).unsqueeze(0		)
																model_inputs.update({"input_ids": input_ids}		)

											elif model_type == "pix2struct":
																a					     =   self.image_processor(images=lowerCamelCase_		,       header_text=lowerCamelCase_		,       return_tensors=self.framework		)

											elif model_type != "vision-encoder-decoder":
																# vision-encoder-decoder does not support conditional generation
																a					     =   self.image_processor(images=lowerCamelCase_		,       return_tensors=self.framework		)
																a					     =   self.tokenizer(lowerCamelCase_		,       return_tensors=self.framework		)
																model_inputs.update(lowerCamelCase_		)

											else:
																raise ValueError(F'''Model type {model_type} does not support conditional text generation'''		)

						else:
											a					     =   self.image_processor(images=lowerCamelCase_		,       return_tensors=self.framework		)

						if self.model.config.model_type == "git" and prompt is None:
											a					     =   None

						return model_inputs




	def      UpperCamelCase_       (self		,       lowerCamelCase_		,       lowerCamelCase_=None		):
						"""simple docstring"""

						if (
						    "input_ids" in model_inputs
						    and isinstance(model_inputs["input_ids"]		,       lowerCamelCase_		)
						    and all(x is None for x in model_inputs["input_ids"]		)
						):
											a					     =   None

						if generate_kwargs is None:
											a					     =   {}
						# FIXME: We need to pop here due to a difference in how `generation.py` and `generation.tf_utils.py`
						#  parse inputs. In the Tensorflow version, `generate` raises an error if we don't use `input_ids` whereas
						#  the PyTorch version matches it with `self.model.main_input_name` or `self.model.encoder.main_input_name`
						#  in the `_prepare_model_inputs` method.
						a					     =   model_inputs.pop(self.model.main_input_name		)
						a					     =   self.model.generate(lowerCamelCase_		,       **lowerCamelCase_		,       **lowerCamelCase_		)
						return model_outputs




	def      UpperCamelCase_       (self		,       lowerCamelCase_		):
						"""simple docstring"""

						a					     =   []
						for output_ids in model_outputs:
											a					     =   {
											    "generated_text": self.tokenizer.decode(
											        lowerCamelCase_		,       skip_special_tokens=lowerCamelCase_		,       )
											}
											records.append(lowerCamelCase_		)
						return records
