




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

     def 					a__	(  self				)						->							Any:
           # A mock response for an HTTP head request to emulate server down
           _A							:     Optional[int]			=						mock.Mock()
           _A							:     Optional[Any]			=						500
           _A							:     Dict			=						{}
           _A							:     Union[str, Any]			=						HTTPError
           _A							:     List[Any]			=						{}

           # Download this model to make sure it's in the cache.
           _A							:     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=_a				) as mock_head:
                 _A							:     List[Any]			=						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				)						->							Optional[Any]:
           # A mock response for an HTTP head request to emulate server down
           _A							:     str			=						mock.Mock()
           _A							:     Any			=						500
           _A							:     Optional[int]			=						{}
           _A							:     List[str]			=						HTTPError
           _A							:     int			=						{}

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

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

                 _A							:     str			=						AlbertTokenizer.from_pretrained(_a				)
           finally:
                 os.remove(_a				)

           # 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"""		,     _a				)
                 _A							:     Optional[int]			=						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 					a__	(  self				)						->							Union[str, Any]:
           # This test is for deprecated behavior and can be removed in v5
           _A							:     Dict			=						AlbertTokenizer.from_pretrained("""https://huggingface.co/albert-base-v1/resolve/main/spiece.model"""				)







@is_staging_test
class 						lowercase							(   unittest.TestCase		):
     _a							 =						["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]

     @classmethod
     def 					a__	(  cls				)						->							Tuple:
           _A							:     int			=						TOKEN
           HfFolder.save_token(_a				)

     @classmethod
     def 					a__	(  cls				)						->							List[Any]:
           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				)						->							Dict:
           with tempfile.TemporaryDirectory() as tmp_dir:
                 _A							:     Optional[Any]			=						os.path.join(_a		,     """vocab.txt"""				)
                 with open(_a		,     """w"""		,     encoding="""utf-8"""				) as vocab_writer:
                       vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]				)				)
                 _A							:     Optional[Any]			=						BertTokenizer(_a				)

           tokenizer.push_to_hub("""test-tokenizer"""		,     use_auth_token=self._token				)
           _A							:     List[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(_a		,     repo_id="""test-tokenizer"""		,     push_to_hub=_a		,     use_auth_token=self._token				)

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

     def 					a__	(  self				)						->							Optional[int]:
           with tempfile.TemporaryDirectory() as tmp_dir:
                 _A							:     Optional[int]			=						os.path.join(_a		,     """vocab.txt"""				)
                 with open(_a		,     """w"""		,     encoding="""utf-8"""				) as vocab_writer:
                       vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]				)				)
                 _A							:     Tuple			=						BertTokenizer(_a				)

           tokenizer.push_to_hub("""valid_org/test-tokenizer-org"""		,     use_auth_token=self._token				)
           _A							:     List[Any]			=						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(
                     _a		,     repo_id="""valid_org/test-tokenizer-org"""		,     push_to_hub=_a		,     use_auth_token=self._token				)

           _A							:     Dict			=						BertTokenizer.from_pretrained("""valid_org/test-tokenizer-org"""				)
           self.assertDictEqual(new_tokenizer.vocab		,     tokenizer.vocab				)



     @require_tokenizers
     def 					a__	(  self				)						->							str:
           CustomTokenizer.register_for_auto_class()
           with tempfile.TemporaryDirectory() as tmp_dir:
                 _A							:     Union[str, Any]			=						os.path.join(_a		,     """vocab.txt"""				)
                 with open(_a		,     """w"""		,     encoding="""utf-8"""				) as vocab_writer:
                       vocab_writer.write("""""".join([x + """\n""" for x in self.vocab_tokens]				)				)
                 _A							:     str			=						CustomTokenizer(_a				)

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

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

                 _A							:     Dict			=						BertTokenizerFast.from_pretrained(_a				)
                 bert_tokenizer.save_pretrained(_a				)
                 _A							:     Dict			=						CustomTokenizerFast.from_pretrained(_a				)

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

           _A							:     Tuple			=						AutoTokenizer.from_pretrained(F'''{USER}/test-dynamic-tokenizer'''		,     trust_remote_code=_a				)
           # 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"""				)
           _A							:     Tuple			=						AutoTokenizer.from_pretrained(
               F'''{USER}/test-dynamic-tokenizer'''		,     use_fast=_a		,     trust_remote_code=_a				)
           # 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 						lowercase							(   unittest.TestCase		):

     def 					a__	(  self				)						->							List[Any]:
           _A							:     Optional[Any]			=						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				)						->							Union[str, Any]:
           _A							:     List[Any]			=						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				)						->							Dict:
           _A							:     List[Any]			=						Trie()
           trie.add("""A"""				)
           self.assertEqual(trie.split("""ABC"""				)		,     ["""A""", """BC"""]				)
           self.assertEqual(trie.split("""BCA"""				)		,     ["""BC""", """A"""]				)

     def 					a__	(  self				)						->							Dict:
           _A							:     Dict			=						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				)						->							List[Any]:
           _A							:     Optional[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				)						->							List[Any]:
           _A							:     Dict			=						Trie()
           trie.add("""AB"""				)
           trie.add("""B"""				)
           trie.add("""C"""				)
           self.assertEqual(trie.split("""ABC"""				)		,     ["""AB""", """C"""]				)

     def 					a__	(  self				)						->							int:
           _A							:     Any			=						Trie()
           trie.add("""ABC"""				)
           trie.add("""B"""				)
           trie.add("""CD"""				)
           self.assertEqual(trie.split("""ABCD"""				)		,     ["""ABC""", """D"""]				)



     def 					a__	(  self				)						->							List[Any]:
           # Even if the offsets are wrong, we necessarily output correct string
           # parts.
           _A							:     Tuple			=						Trie()
           _A							:     int			=						trie.cut_text("""ABC"""		,     [0, 0, 2, 1, 2, 3]				)
           self.assertEqual(_a		,     ["""AB""", """C"""]				)



from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
      import PIL


lowercase__       =logging.get_logger(__name__)



class 		UpperCamelCase__  (       __lowercase    ):
   _SCREAMING_SNAKE_CASE      :							Optional[Any]	       =      ["pixel_values"]
   def __init__(self :					str					,				snake_case_ :					bool = True					,				snake_case_ :					Dict[str, int] = None					,				snake_case_ :					float = None					,				snake_case_ :					PILImageResampling = PILImageResampling.BILINEAR					,				snake_case_ :					bool = True					,				snake_case_ :					Union[int, float] = 1 / 2_5_5					,				snake_case_ :					bool = True					,				snake_case_ :					Optional[Union[float, List[float]]] = None					,				snake_case_ :					Optional[Union[float, List[float]]] = None					,				**snake_case_ :					Tuple					,				):
    super().__init__(**snake_case_   )
    __a :  Optional[Any]						=  size if size is not None else {'''shortest_edge''': 3_8_4}
    __a :  str						=  get_size_dict(snake_case_					,				default_to_square=snake_case_   )

    __a :  Tuple						=  do_resize
    __a :  List[str]						=  size
    # Default value set here for backwards compatibility where the value in config is None
    __a :  List[str]						=  crop_pct if crop_pct is not None else 2_2_4 / 2_5_6
    __a :  Tuple						=  resample
    __a :  Any						=  do_rescale
    __a :  Optional[Any]						=  rescale_factor
    __a :  int						=  do_normalize
    __a :  List[str]						=  image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
    __a :  Union[str, Any]						=  image_std if image_std is not None else IMAGENET_STANDARD_STD
   def       lowerCAmelCase			(self :					int					,				snake_case_ :					np.ndarray					,				snake_case_ :					Dict[str, int]					,				snake_case_ :					float					,				snake_case_ :					PILImageResampling = PILImageResampling.BICUBIC					,				snake_case_ :					Optional[Union[str, ChannelDimension]] = None					,				**snake_case_ :					str					,				):
    __a :  str						=  get_size_dict(snake_case_					,				default_to_square=snake_case_   )
    if "shortest_edge" not in size:
     raise ValueError(f"Size dictionary must contain 'shortest_edge' key. Got {size.keys()}"   )
    __a :  Any						=  size['''shortest_edge''']

    if shortest_edge < 3_8_4:
     # maintain same ratio, resizing shortest edge to shortest_edge/crop_pct
     __a :  Union[str, Any]						=  int(shortest_edge / crop_pct   )
     __a :  Tuple						=  get_resize_output_image_size(snake_case_					,				size=snake_case_					,				default_to_square=snake_case_   )
     __a :  List[str]						=  resize(image=snake_case_					,				size=snake_case_					,				resample=snake_case_					,				data_format=snake_case_					,				**snake_case_   )
     # then crop to (shortest_edge, shortest_edge)
     return center_crop(image=snake_case_					,				size=(shortest_edge, shortest_edge)					,				data_format=snake_case_					,				**snake_case_   )
    else:
     # warping (no cropping) when evaluated at 384 or larger
     return resize(
         snake_case_					,				size=(shortest_edge, shortest_edge)					,				resample=snake_case_					,				data_format=snake_case_					,				**snake_case_   )
   def       lowerCAmelCase			(self :					Optional[int]					,				snake_case_ :					np.ndarray					,				snake_case_ :					Union[int, float]					,				snake_case_ :					Optional[Union[str, ChannelDimension]] = None					,				**snake_case_ :					Union[str, Any]					,				):
    return rescale(snake_case_					,				scale=snake_case_					,				data_format=snake_case_					,				**snake_case_   )
   def       lowerCAmelCase			(self :					Union[str, Any]					,				snake_case_ :					np.ndarray					,				snake_case_ :					Union[float, List[float]]					,				snake_case_ :					Union[float, List[float]]					,				snake_case_ :					Optional[Union[str, ChannelDimension]] = None					,				**snake_case_ :					Union[str, Any]					,				):
    return normalize(snake_case_					,				mean=snake_case_					,				std=snake_case_					,				data_format=snake_case_					,				**snake_case_   )







   def       lowerCAmelCase			(self :					Dict					,				snake_case_ :					ImageInput					,				snake_case_ :					bool = None					,				snake_case_ :					Dict[str, int] = None					,				snake_case_ :					float = None					,				snake_case_ :					PILImageResampling = None					,				snake_case_ :					bool = None					,				snake_case_ :					float = None					,				snake_case_ :					bool = None					,				snake_case_ :					Optional[Union[float, List[float]]] = None					,				snake_case_ :					Optional[Union[float, List[float]]] = None					,				snake_case_ :					Optional[Union[str, TensorType]] = None					,				snake_case_ :					ChannelDimension = ChannelDimension.FIRST					,				**snake_case_ :					Dict					,				):
    __a :  List[Any]						=  do_resize if do_resize is not None else self.do_resize
    __a :  Dict						=  crop_pct if crop_pct is not None else self.crop_pct
    __a :  Tuple						=  resample if resample is not None else self.resample
    __a :  Optional[int]						=  do_rescale if do_rescale is not None else self.do_rescale
    __a :  Optional[int]						=  rescale_factor if rescale_factor is not None else self.rescale_factor
    __a :  int						=  do_normalize if do_normalize is not None else self.do_normalize
    __a :  Optional[int]						=  image_mean if image_mean is not None else self.image_mean
    __a :  Optional[int]						=  image_std if image_std is not None else self.image_std

    __a :  Dict						=  size if size is not None else self.size
    __a :  List[str]						=  get_size_dict(snake_case_					,				default_to_square=snake_case_   )

    __a :  List[str]						=  make_list_of_images(snake_case_   )

    if not valid_images(snake_case_   ):
     raise ValueError(
         '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
         '''torch.Tensor, tf.Tensor or jax.ndarray.'''   )

    if do_resize and size is None or resample is None:
     raise ValueError('''Size and resample must be specified if do_resize is True.'''   )

    if do_resize and size["shortest_edge"] < 3_8_4 and crop_pct is None:
     raise ValueError('''crop_pct must be specified if size < 384.'''   )

    if do_rescale and rescale_factor is None:
     raise ValueError('''Rescale factor must be specified if do_rescale is True.'''   )

    if do_normalize and (image_mean is None or image_std is None):
     raise ValueError('''Image mean and std must be specified if do_normalize is True.'''   )

    # All transformations expect numpy arrays.
    __a :  List[Any]						=  [to_numpy_array(snake_case_   ) for image in images]

    if do_resize:
     __a :  List[str]						=  [self.resize(image=snake_case_					,				size=snake_case_					,				crop_pct=snake_case_					,				resample=snake_case_   ) for image in images]

    if do_rescale:
     __a :  List[Any]						=  [self.rescale(image=snake_case_					,				scale=snake_case_   ) for image in images]

    if do_normalize:
     __a :  Optional[int]						=  [self.normalize(image=snake_case_					,				mean=snake_case_					,				std=snake_case_   ) for image in images]

    __a :  str						=  [to_channel_dimension_format(snake_case_					,				snake_case_   ) for image in images]

    __a :  Tuple						=  {'''pixel_values''': images}
    return BatchFeature(data=snake_case_					,				tensor_type=snake_case_   )
