






'''simple docstring'''





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      ):


    '''simple docstring'''

    def    a__				(					self     :     List[Any]     )							->							Optional[int]:

     '''simple docstring'''




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

     # Download this model to make sure it's in the cache.
     lowerCamelCase__  =     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:
      lowerCamelCase__  =     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     :     Dict     )							->							Tuple:

     '''simple docstring'''




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

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

    def    a__				(					self     :     Any     )							->							int:

     '''simple docstring'''




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

      lowerCamelCase__  =     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     )
      lowerCamelCase__  =     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     :     List[str]     )							->							int:

     '''simple docstring'''




     lowerCamelCase__  =     AlbertTokenizer.from_pretrained("https://huggingface.co/albert-base-v1/resolve/main/spiece.model"     )
@is_staging_test
class 				lowercase       (							unittest.TestCase      ):


    '''simple docstring'''

    lowerCAmelCase__      				=							["[UNK]", "[CLS]", "[SEP]", "[PAD]", "[MASK]", "bla", "blou"]

    @classmethod
    def    a__				(					cls     :     str     )							->							List[str]:

     '''simple docstring'''




     lowerCamelCase__  =     TOKEN
     HfFolder.save_token(__lowerCamelCase     )

    @classmethod
    def    a__				(					cls     :     List[str]     )							->							List[Any]:

     '''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    a__				(					self     :     Union[str, Any]     )							->							List[str]:

     '''simple docstring'''




     with tempfile.TemporaryDirectory() as tmp_dir:
      lowerCamelCase__  =     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]     )     )
      lowerCamelCase__  =     BertTokenizer(__lowerCamelCase     )

     tokenizer.push_to_hub("test-tokenizer"				,     use_auth_token=self._token     )
     lowerCamelCase__  =     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     )

     lowerCamelCase__  =     BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''     )
     self.assertDictEqual(new_tokenizer.vocab				,     tokenizer.vocab     )

    def    a__				(					self     :     Optional[Any]     )							->							List[str]:

     '''simple docstring'''




     with tempfile.TemporaryDirectory() as tmp_dir:
      lowerCamelCase__  =     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]     )     )
      lowerCamelCase__  =     BertTokenizer(__lowerCamelCase     )

     tokenizer.push_to_hub("valid_org/test-tokenizer-org"				,     use_auth_token=self._token     )
     lowerCamelCase__  =     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     )

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




    @require_tokenizers
    def    a__				(					self     :     Any     )							->							List[Any]:

     '''simple docstring'''




     CustomTokenizer.register_for_auto_class()
     with tempfile.TemporaryDirectory() as tmp_dir:
      lowerCamelCase__  =     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]     )     )
      lowerCamelCase__  =     CustomTokenizer(__lowerCamelCase     )

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

     lowerCamelCase__  =     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:
      lowerCamelCase__  =     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]     )     )

      lowerCamelCase__  =     BertTokenizerFast.from_pretrained(__lowerCamelCase     )
      bert_tokenizer.save_pretrained(__lowerCamelCase     )
      lowerCamelCase__  =     CustomTokenizerFast.from_pretrained(__lowerCamelCase     )

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

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


    '''simple docstring'''

    def    a__				(					self     :     Optional[Any]     )							->							int:

     '''simple docstring'''




     lowerCamelCase__  =     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     :     Optional[Any]     )							->							Any:

     '''simple docstring'''




     lowerCamelCase__  =     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     )							->							Dict:

     '''simple docstring'''




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

    def    a__				(					self     :     Union[str, Any]     )							->							Any:

     '''simple docstring'''




     lowerCamelCase__  =     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     :     Dict     )							->							List[str]:

     '''simple docstring'''




     lowerCamelCase__  =     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     :     Any     )							->							str:

     '''simple docstring'''




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

    def    a__				(					self     :     List[str]     )							->							List[str]:

     '''simple docstring'''




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




    def    a__				(					self     :     Union[str, Any]     )							->							List[str]:

     '''simple docstring'''




     lowerCamelCase__  =     Trie()
     lowerCamelCase__  =     trie.cut_text("ABC"				,     [0, 0, 2, 1, 2, 3]     )
     self.assertEqual(__lowerCamelCase				,     ["AB", "C"]     )









'''simple docstring'''





# Lint as: python3
import sys
from collections.abc import Mapping
from typing import TYPE_CHECKING, Dict, Optional

import numpy as np
import pyarrow as pa

from .. import config
from ..utils.logging import get_logger
from ..utils.py_utils import map_nested
from .formatting import TensorFormatter


if TYPE_CHECKING:
 import jax
 import jaxlib

__A							:							Any  							=		get_logger()

__A							:							Optional[dict]  							=		None
class 				lowercase       (							TensorFormatter[Mapping, "jax.Array", Mapping]      ):


    '''simple docstring'''

    def __init__(					self     :     str				,     __lowerCamelCase     :     Optional[Any]=None				,     __lowerCamelCase     :     Dict=None				,     **__lowerCamelCase     :     str     )							->							List[Any]:

     '''simple docstring'''




     super().__init__(features=__lowerCamelCase     )
     import jax
     from jaxlib.xla_client import Device

     if isinstance(__lowerCamelCase				,     __lowerCamelCase     ):
      raise ValueError(
          f'''Expected {device} to be a `str` not {type(__lowerCamelCase     )}, as `jaxlib.xla_extension.Device` '''
          "is not serializable neither with `pickle` nor with `dill`. Instead you can surround "
          "the device with `str()` to get its string identifier that will be internally mapped "
          "to the actual `jaxlib.xla_extension.Device`."     )
     lowerCamelCase__  =     device if isinstance(__lowerCamelCase				,     __lowerCamelCase     ) else str(jax.devices()[0]     )
     # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
     # with `pickle` nor with `dill`, so we need to use a global variable instead
     global DEVICE_MAPPING
     if DEVICE_MAPPING is None:
      lowerCamelCase__  =     self._map_devices_to_str()
     if self.device not in list(DEVICE_MAPPING.keys()     ):
      logger.warning(
          f'''Device with string identifier {self.device} not listed among the available '''
          f'''devices: {list(DEVICE_MAPPING.keys()     )}, so falling back to the default '''
          f'''device: {str(jax.devices()[0]     )}.'''     )
      lowerCamelCase__  =     str(jax.devices()[0]     )
     lowerCamelCase__  =     jnp_array_kwargs

    @staticmethod
    def    a__				(					)							->							Dict[str, "jaxlib.xla_extension.Device"]:

     '''simple docstring'''




     import jax

     return {str(__lowerCamelCase     ): device for device in jax.devices()}

    def    a__				(					self     :     Optional[Any]				,     __lowerCamelCase     :     Optional[int]     )							->							List[str]:

     '''simple docstring'''




     import jax
     import jax.numpy as jnp

     if isinstance(__lowerCamelCase				,     __lowerCamelCase     ) and column:
      if all(
          isinstance(__lowerCamelCase				,     jax.Array     ) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column     ):
       return jnp.stack(__lowerCamelCase				,     axis=0     )
     return column

    def    a__				(					self     :     List[Any]				,     __lowerCamelCase     :     List[Any]     )							->							Optional[int]:

     '''simple docstring'''




     import jax
     import jax.numpy as jnp

     if isinstance(__lowerCamelCase				,     (str, bytes, type(__lowerCamelCase     ))     ):
      return value
     elif isinstance(__lowerCamelCase				,     (np.character, np.ndarray)     ) and np.issubdtype(value.dtype				,     np.character     ):
      return value.tolist()

     lowerCamelCase__  =     {}

     if isinstance(__lowerCamelCase				,     (np.number, np.ndarray)     ) and np.issubdtype(value.dtype				,     np.integer     ):
      # the default int precision depends on the jax config
      # see https://jax.readthedocs.io/en/latest/notebooks/Common_Gotchas_in_JAX.html#double-64bit-precision
      if jax.config.jax_enable_xaa:
       lowerCamelCase__  =     {"dtype": jnp.intaa}
      else:
       lowerCamelCase__  =     {"dtype": jnp.intaa}
     elif isinstance(__lowerCamelCase				,     (np.number, np.ndarray)     ) and np.issubdtype(value.dtype				,     np.floating     ):
      lowerCamelCase__  =     {"dtype": jnp.floataa}
     elif config.PIL_AVAILABLE and "PIL" in sys.modules:
      import PIL.Image

      if isinstance(__lowerCamelCase				,     PIL.Image.Image     ):
       lowerCamelCase__  =     np.asarray(__lowerCamelCase     )

        # using global variable since `jaxlib.xla_extension.Device` is not serializable neither
        # with `pickle` nor with `dill`, so we need to use a global variable instead
     global DEVICE_MAPPING
     if DEVICE_MAPPING is None:
      lowerCamelCase__  =     self._map_devices_to_str()

     with jax.default_device(DEVICE_MAPPING[self.device]     ):
      # calling jnp.array on a np.ndarray does copy the data
      # see https://github.com/google/jax/issues/4486
      return jnp.array(__lowerCamelCase				,     **{**default_dtype, **self.jnp_array_kwargs}     )

    def    a__				(					self     :     List[str]				,     __lowerCamelCase     :     Optional[Any]     )							->							Dict:

     '''simple docstring'''




     import jax

     # support for torch, tf, jax etc.
     if config.TORCH_AVAILABLE and "torch" in sys.modules:
      import torch

      if isinstance(__lowerCamelCase				,     torch.Tensor     ):
       return self._tensorize(data_struct.detach().cpu().numpy()[()]     )
     if hasattr(__lowerCamelCase				,     "__array__"     ) and not isinstance(__lowerCamelCase				,     jax.Array     ):
      lowerCamelCase__  =     data_struct.__array__()
     # support for nested types like struct of list of struct
     if isinstance(__lowerCamelCase				,     np.ndarray     ):
      if data_struct.dtype == object:  # jax arrays cannot be instantied from an array of objects
       return self._consolidate([self.recursive_tensorize(__lowerCamelCase     ) for substruct in data_struct]     )
     elif isinstance(__lowerCamelCase				,     (list, tuple)     ):
      return self._consolidate([self.recursive_tensorize(__lowerCamelCase     ) for substruct in data_struct]     )
     return self._tensorize(__lowerCamelCase     )

    def    a__				(					self     :     Any				,     __lowerCamelCase     :     dict     )							->							int:

     '''simple docstring'''




     return map_nested(self._recursive_tensorize				,     __lowerCamelCase				,     map_list=__lowerCamelCase     )

    def    a__				(					self     :     Union[str, Any]				,     __lowerCamelCase     :     pa.Table     )							->							Mapping:

     '''simple docstring'''




     lowerCamelCase__  =     self.numpy_arrow_extractor().extract_row(__lowerCamelCase     )
     lowerCamelCase__  =     self.python_features_decoder.decode_row(__lowerCamelCase     )
     return self.recursive_tensorize(__lowerCamelCase     )

    def    a__				(					self     :     List[Any]				,     __lowerCamelCase     :     pa.Table     )							->							"jax.Array":

     '''simple docstring'''




     lowerCamelCase__  =     self.numpy_arrow_extractor().extract_column(__lowerCamelCase     )
     lowerCamelCase__  =     self.python_features_decoder.decode_column(__lowerCamelCase				,     pa_table.column_names[0]     )
     lowerCamelCase__  =     self.recursive_tensorize(__lowerCamelCase     )
     lowerCamelCase__  =     self._consolidate(__lowerCamelCase     )
     return column




    def    a__				(					self     :     List[str]				,     __lowerCamelCase     :     pa.Table     )							->							Mapping:

     '''simple docstring'''




     lowerCamelCase__  =     self.numpy_arrow_extractor().extract_batch(__lowerCamelCase     )
     lowerCamelCase__  =     self.python_features_decoder.decode_batch(__lowerCamelCase     )
     lowerCamelCase__  =     self.recursive_tensorize(__lowerCamelCase     )
     for column_name in batch:
      lowerCamelCase__  =     self._consolidate(batch[column_name]     )
     return batch

