

import os
import pickle
import tempfile
import unittest
from typing import Callable, Optional

import numpy as np

from transformers import (
    BatchEncoding,
    BertTokenizer,
    BertTokenizerFast,
    PreTrainedTokenizer,
    PreTrainedTokenizerFast,
    TensorType,
    TokenSpan,
    is_tokenizers_available,
)
from transformers.models.gpta.tokenization_gpta import GPTaTokenizer
from transformers.testing_utils import CaptureStderr, require_flax, require_tf, require_tokenizers, require_torch, slow


if is_tokenizers_available():
 from tokenizers import Tokenizer
 from tokenizers.models import WordPiece



class     lowercase_       (    unittest.TestCase	):



      def 		UpperCamelCase_						(     self :							Optional[Any]     ,   A__ :							List[str]						)       ->	Union[str, Any]:
        _snake_case		=				list(tokenizer_class.max_model_input_sizes.keys()						)
        for model_name in sa_models[:1]:
          _snake_case		=				tokenizer_class.from_pretrained(A__						)
          self.assertIsNotNone(A__						)
          self.assertIsInstance(A__     ,   A__						)
          self.assertIsInstance(A__     ,   A__						)

          for special_tok in tokenizer.all_special_tokens:
            self.assertIsInstance(A__     ,   A__						)
            _snake_case		=				tokenizer.convert_tokens_to_ids(A__						)
            self.assertIsInstance(A__     ,   A__						)



      def 		UpperCamelCase_						(     self :							Optional[int]     ,   A__ :							BatchEncoding     ,   A__ :							Optional[Callable] = None						)       ->	Union[str, Any]:
        _snake_case		=				pickle.dumps(A__						)
        self.assertIsNotNone(A__						)

        _snake_case		=				pickle.loads(A__						)

        # Ensure is_fast is correctly restored
        self.assertEqual(be_restored.is_fast     ,   be_original.is_fast						)

        # Ensure encodings are potentially correctly restored
        if be_original.is_fast:
          self.assertIsNotNone(be_restored.encodings						)
        else:
          self.assertIsNone(be_restored.encodings						)

        # Ensure the keys are the same
        for original_v, restored_v in zip(be_original.values()     ,   be_restored.values()						):
          if equal_op:
            self.assertTrue(equal_op(A__     ,   A__						)						)
          else:
            self.assertEqual(A__     ,   A__						)



      @slow
      def 		UpperCamelCase_						(     self :							Tuple						)       ->	Any:
        self.check_tokenizer_from_pretrained(A__						)



      def 		UpperCamelCase_						(     self :							List[str]						)       ->	Optional[Any]:
        self.assertEqual(TensorType('''tf'''						)     ,   TensorType.TENSORFLOW						)
        self.assertEqual(TensorType('''pt'''						)     ,   TensorType.PYTORCH						)
        self.assertEqual(TensorType('''np'''						)     ,   TensorType.NUMPY						)



      @require_tokenizers
      def 		UpperCamelCase_						(     self :							Optional[int]						)       ->	List[Any]:
        import numpy as np

        _snake_case		=				BertTokenizer.from_pretrained('''bert-base-cased'''						)
        _snake_case		=				BertTokenizerFast.from_pretrained('''bert-base-cased'''						)

        # Python no tensor
        with self.subTest('''BatchEncoding (Python, return_tensors=None)'''						):
          self.assert_dump_and_restore(tokenizer_p('''Small example to encode'''						)						)

        with self.subTest('''BatchEncoding (Python, return_tensors=NUMPY)'''						):
          self.assert_dump_and_restore(
              tokenizer_p('''Small example to encode'''     ,   return_tensors=TensorType.NUMPY						)     ,   np.array_equal						)

        with self.subTest('''BatchEncoding (Rust, return_tensors=None)'''						):
          self.assert_dump_and_restore(tokenizer_r('''Small example to encode'''						)						)

        with self.subTest('''BatchEncoding (Rust, return_tensors=NUMPY)'''						):
          self.assert_dump_and_restore(
              tokenizer_r('''Small example to encode'''     ,   return_tensors=TensorType.NUMPY						)     ,   np.array_equal						)



      @require_tf
      @require_tokenizers
      def 		UpperCamelCase_						(     self :							Dict						)       ->	Tuple:
        import tensorflow as tf

        def tf_array_equals(A__ :							int     ,   A__ :							Dict						):
          return tf.reduce_all(tf.equal(A__     ,   A__						)						)

        _snake_case		=				BertTokenizer.from_pretrained('''bert-base-cased'''						)
        _snake_case		=				BertTokenizerFast.from_pretrained('''bert-base-cased'''						)

        with self.subTest('''BatchEncoding (Python, return_tensors=TENSORFLOW)'''						):
          self.assert_dump_and_restore(
              tokenizer_p('''Small example to encode'''     ,   return_tensors=TensorType.TENSORFLOW						)     ,   A__						)

        with self.subTest('''BatchEncoding (Rust, return_tensors=TENSORFLOW)'''						):
          self.assert_dump_and_restore(
              tokenizer_r('''Small example to encode'''     ,   return_tensors=TensorType.TENSORFLOW						)     ,   A__						)



      @require_torch
      @require_tokenizers
      def 		UpperCamelCase_						(     self :							Optional[int]						)       ->	int:
        import torch

        _snake_case		=				BertTokenizer.from_pretrained('''bert-base-cased'''						)
        _snake_case		=				BertTokenizerFast.from_pretrained('''bert-base-cased'''						)

        with self.subTest('''BatchEncoding (Python, return_tensors=PYTORCH)'''						):
          self.assert_dump_and_restore(
              tokenizer_p('''Small example to encode'''     ,   return_tensors=TensorType.PYTORCH						)     ,   torch.equal						)

        with self.subTest('''BatchEncoding (Rust, return_tensors=PYTORCH)'''						):
          self.assert_dump_and_restore(
              tokenizer_r('''Small example to encode'''     ,   return_tensors=TensorType.PYTORCH						)     ,   torch.equal						)



      @require_tokenizers
      def 		UpperCamelCase_						(     self :							Dict						)       ->	List[Any]:
        _snake_case		=				BertTokenizer.from_pretrained('''bert-base-cased'''						)
        _snake_case		=				BertTokenizerFast.from_pretrained('''bert-base-cased'''						)

        with self.subTest('''Python Tokenizer'''						):
          self.assertFalse(tokenizer_p('''Small example to_encode'''						).is_fast						)

        with self.subTest('''Rust Tokenizer'''						):
          self.assertTrue(tokenizer_r('''Small example to_encode'''						).is_fast						)



      @require_tokenizers
      def 		UpperCamelCase_						(     self :							Tuple						)       ->	Union[str, Any]:
        _snake_case		=				BertTokenizerFast.from_pretrained('''bert-base-cased'''						)
        _snake_case		=				tokenizer_r(['''Test''', '''\xad''', '''test''']     ,   is_split_into_words=A__						)

        self.assertEqual(encoded.word_to_tokens(0						)     ,   TokenSpan(start=1     ,   end=2						)						)
        self.assertEqual(encoded.word_to_tokens(1						)     ,   A__						)
        self.assertEqual(encoded.word_to_tokens(2						)     ,   TokenSpan(start=2     ,   end=3						)						)



      def 		UpperCamelCase_						(     self :							Tuple						)       ->	Dict:
        _snake_case		=				BatchEncoding({'''inputs''': [[1, 2, 3], [4, 5, 6]], '''labels''': [0, 1]}						)
        _snake_case		=				batch.convert_to_tensors(tensor_type='''np'''						)
        self.assertEqual(tensor_batch['''inputs'''].shape     ,   (2, 3)						)
        self.assertEqual(tensor_batch['''labels'''].shape     ,   (2,)						)
        # test converting the converted
        with CaptureStderr() as cs:
          _snake_case		=				batch.convert_to_tensors(tensor_type='''np'''						)
        self.assertFalse(len(cs.err						)     ,   msg=f"""should have no warning, but got {cs.err}"""						)

        _snake_case		=				BatchEncoding({'''inputs''': [1, 2, 3], '''labels''': 0}						)
        _snake_case		=				batch.convert_to_tensors(tensor_type='''np'''     ,   prepend_batch_axis=A__						)
        self.assertEqual(tensor_batch['''inputs'''].shape     ,   (1, 3)						)
        self.assertEqual(tensor_batch['''labels'''].shape     ,   (1,)						)



      @require_torch
      def 		UpperCamelCase_						(     self :							Union[str, Any]						)       ->	Dict:
        _snake_case		=				BatchEncoding({'''inputs''': [[1, 2, 3], [4, 5, 6]], '''labels''': [0, 1]}						)
        _snake_case		=				batch.convert_to_tensors(tensor_type='''pt'''						)
        self.assertEqual(tensor_batch['''inputs'''].shape     ,   (2, 3)						)
        self.assertEqual(tensor_batch['''labels'''].shape     ,   (2,)						)
        # test converting the converted
        with CaptureStderr() as cs:
          _snake_case		=				batch.convert_to_tensors(tensor_type='''pt'''						)
        self.assertFalse(len(cs.err						)     ,   msg=f"""should have no warning, but got {cs.err}"""						)

        _snake_case		=				BatchEncoding({'''inputs''': [1, 2, 3], '''labels''': 0}						)
        _snake_case		=				batch.convert_to_tensors(tensor_type='''pt'''     ,   prepend_batch_axis=A__						)
        self.assertEqual(tensor_batch['''inputs'''].shape     ,   (1, 3)						)
        self.assertEqual(tensor_batch['''labels'''].shape     ,   (1,)						)



      @require_tf
      def 		UpperCamelCase_						(     self :							int						)       ->	Any:
        _snake_case		=				BatchEncoding({'''inputs''': [[1, 2, 3], [4, 5, 6]], '''labels''': [0, 1]}						)
        _snake_case		=				batch.convert_to_tensors(tensor_type='''tf'''						)
        self.assertEqual(tensor_batch['''inputs'''].shape     ,   (2, 3)						)
        self.assertEqual(tensor_batch['''labels'''].shape     ,   (2,)						)
        # test converting the converted
        with CaptureStderr() as cs:
          _snake_case		=				batch.convert_to_tensors(tensor_type='''tf'''						)
        self.assertFalse(len(cs.err						)     ,   msg=f"""should have no warning, but got {cs.err}"""						)

        _snake_case		=				BatchEncoding({'''inputs''': [1, 2, 3], '''labels''': 0}						)
        _snake_case		=				batch.convert_to_tensors(tensor_type='''tf'''     ,   prepend_batch_axis=A__						)
        self.assertEqual(tensor_batch['''inputs'''].shape     ,   (1, 3)						)
        self.assertEqual(tensor_batch['''labels'''].shape     ,   (1,)						)



      @require_flax
      def 		UpperCamelCase_						(     self :							Dict						)       ->	Optional[int]:
        _snake_case		=				BatchEncoding({'''inputs''': [[1, 2, 3], [4, 5, 6]], '''labels''': [0, 1]}						)
        _snake_case		=				batch.convert_to_tensors(tensor_type='''jax'''						)
        self.assertEqual(tensor_batch['''inputs'''].shape     ,   (2, 3)						)
        self.assertEqual(tensor_batch['''labels'''].shape     ,   (2,)						)
        # test converting the converted
        with CaptureStderr() as cs:
          _snake_case		=				batch.convert_to_tensors(tensor_type='''jax'''						)
        self.assertFalse(len(cs.err						)     ,   msg=f"""should have no warning, but got {cs.err}"""						)

        _snake_case		=				BatchEncoding({'''inputs''': [1, 2, 3], '''labels''': 0}						)
        _snake_case		=				batch.convert_to_tensors(tensor_type='''jax'''     ,   prepend_batch_axis=A__						)
        self.assertEqual(tensor_batch['''inputs'''].shape     ,   (1, 3)						)
        self.assertEqual(tensor_batch['''labels'''].shape     ,   (1,)						)



      def 		UpperCamelCase_						(     self :							int						)       ->	List[Any]:
        _snake_case		=				[{'''input_ids''': np.array([0, 1, 2]						)}, {'''input_ids''': np.array([0, 1, 2, 3]						)}]
        _snake_case		=				BertTokenizer.from_pretrained('''bert-base-cased'''						)

        _snake_case		=				tokenizer.pad(A__     ,   padding=A__						)
        self.assertTrue(isinstance(batch['''input_ids''']     ,   np.ndarray						)						)
        self.assertEqual(batch['''input_ids'''].tolist()     ,   [[0, 1, 2, tokenizer.pad_token_id], [0, 1, 2, 3]]						)
        _snake_case		=				tokenizer.pad(A__     ,   padding=A__     ,   return_tensors='''np'''						)
        self.assertTrue(isinstance(batch['''input_ids''']     ,   np.ndarray						)						)
        self.assertEqual(batch['''input_ids'''].tolist()     ,   [[0, 1, 2, tokenizer.pad_token_id], [0, 1, 2, 3]]						)



      @require_torch
      def 		UpperCamelCase_						(     self :							int						)       ->	List[Any]:
        import torch

        _snake_case		=				[{'''input_ids''': torch.tensor([0, 1, 2]						)}, {'''input_ids''': torch.tensor([0, 1, 2, 3]						)}]
        _snake_case		=				BertTokenizer.from_pretrained('''bert-base-cased'''						)

        _snake_case		=				tokenizer.pad(A__     ,   padding=A__						)
        self.assertTrue(isinstance(batch['''input_ids''']     ,   torch.Tensor						)						)
        self.assertEqual(batch['''input_ids'''].tolist()     ,   [[0, 1, 2, tokenizer.pad_token_id], [0, 1, 2, 3]]						)
        _snake_case		=				tokenizer.pad(A__     ,   padding=A__     ,   return_tensors='''pt'''						)
        self.assertTrue(isinstance(batch['''input_ids''']     ,   torch.Tensor						)						)
        self.assertEqual(batch['''input_ids'''].tolist()     ,   [[0, 1, 2, tokenizer.pad_token_id], [0, 1, 2, 3]]						)



      @require_tf
      def 		UpperCamelCase_						(     self :							str						)       ->	List[Any]:
        import tensorflow as tf

        _snake_case		=				[{'''input_ids''': tf.constant([0, 1, 2]						)}, {'''input_ids''': tf.constant([0, 1, 2, 3]						)}]
        _snake_case		=				BertTokenizer.from_pretrained('''bert-base-cased'''						)

        _snake_case		=				tokenizer.pad(A__     ,   padding=A__						)
        self.assertTrue(isinstance(batch['''input_ids''']     ,   tf.Tensor						)						)
        self.assertEqual(batch['''input_ids'''].numpy().tolist()     ,   [[0, 1, 2, tokenizer.pad_token_id], [0, 1, 2, 3]]						)
        _snake_case		=				tokenizer.pad(A__     ,   padding=A__     ,   return_tensors='''tf'''						)
        self.assertTrue(isinstance(batch['''input_ids''']     ,   tf.Tensor						)						)
        self.assertEqual(batch['''input_ids'''].numpy().tolist()     ,   [[0, 1, 2, tokenizer.pad_token_id], [0, 1, 2, 3]]						)



      @require_tokenizers
      def 		UpperCamelCase_						(     self :							Union[str, Any]						)       ->	Any:
        _snake_case		=				Tokenizer(WordPiece(unk_token='''[UNK]'''						)						)
        PreTrainedTokenizerFast(tokenizer_object=A__						)







      @require_tokenizers
      def 		UpperCamelCase_						(     self :							Optional[Any]						)       ->	int:
        _snake_case		=				Tokenizer(WordPiece(unk_token='''[UNK]'''						)						)
        with tempfile.TemporaryDirectory() as tmpdirname:
          bert_tokenizer.save(os.path.join(A__     ,   '''tokenizer.json'''						)						)
          PreTrainedTokenizerFast(tokenizer_file=os.path.join(A__     ,   '''tokenizer.json'''						)						)




import sys
from collections.abc import Mapping

import numpy as np
import pyarrow as pa

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



class     lowercase_       (    TensorFormatter[Mapping, np.ndarray, Mapping]	):



      def __init__(     self :							Union[str, Any]     ,   A__ :							Dict=None     ,   **A__ :							str						)       ->	Any:
        super().__init__(features=A__						)
        _snake_case		=				np_array_kwargs



      def 		UpperCamelCase_						(     self :							str     ,   A__ :							Dict						)       ->	Dict:
        if isinstance(A__     ,   A__						):
          if column and all(
              isinstance(A__     ,   np.ndarray						) and x.shape == column[0].shape and x.dtype == column[0].dtype for x in column						):
            return np.stack(A__						)
          else:
            # don't use np.array(column, dtype=object)
            # since it fails in certain cases
            # see https://stackoverflow.com/q/51005699
            _snake_case		=				np.empty(len(A__						)     ,   dtype=A__						)
            _snake_case		=				column
            return out
        return column



      def 		UpperCamelCase_						(     self :							Tuple     ,   A__ :							Optional[int]						)       ->	Optional[Any]:
        if isinstance(A__     ,   (str, bytes, type(A__						))						):
          return value
        elif isinstance(A__     ,   (np.character, np.ndarray)						) and np.issubdtype(value.dtype     ,   np.character						):
          return value
        elif isinstance(A__     ,   np.number						):
          return value

        _snake_case		=				{}

        if isinstance(A__     ,   np.ndarray						) and np.issubdtype(value.dtype     ,   np.integer						):
          _snake_case		=				{'''dtype''': np.intaa}
        elif isinstance(A__     ,   np.ndarray						) and np.issubdtype(value.dtype     ,   np.floating						):
          _snake_case		=				{'''dtype''': np.floataa}
        elif config.PIL_AVAILABLE and "PIL" in sys.modules:
          import PIL.Image

          if isinstance(A__     ,   PIL.Image.Image						):
            return np.asarray(A__     ,   **self.np_array_kwargs						)

        return np.asarray(A__     ,   **{**default_dtype, **self.np_array_kwargs}						)



      def 		UpperCamelCase_						(     self :							Union[str, Any]     ,   A__ :							Union[str, Any]						)       ->	int:
        # support for torch, tf, jax etc.
        if config.TORCH_AVAILABLE and "torch" in sys.modules:
          import torch

          if isinstance(A__     ,   torch.Tensor						):
            return self._tensorize(data_struct.detach().cpu().numpy()[()]						)
        if hasattr(A__     ,   '''__array__'''						) and not isinstance(A__     ,   (np.ndarray, np.character, np.number)						):
          _snake_case		=				data_struct.__array__()
        # support for nested types like struct of list of struct
        if isinstance(A__     ,   np.ndarray						):
          if data_struct.dtype == object:
            return self._consolidate([self.recursive_tensorize(A__						) for substruct in data_struct]						)
        if isinstance(A__     ,   (list, tuple)						):
          return self._consolidate([self.recursive_tensorize(A__						) for substruct in data_struct]						)
        return self._tensorize(A__						)



      def 		UpperCamelCase_						(     self :							List[Any]     ,   A__ :							dict						)       ->	Tuple:
        return map_nested(self._recursive_tensorize     ,   A__     ,   map_list=A__						)



      def 		UpperCamelCase_						(     self :							str     ,   A__ :							pa.Table						)       ->	Mapping:
        _snake_case		=				self.numpy_arrow_extractor().extract_row(A__						)
        _snake_case		=				self.python_features_decoder.decode_row(A__						)
        return self.recursive_tensorize(A__						)



      def 		UpperCamelCase_						(     self :							Optional[Any]     ,   A__ :							pa.Table						)       ->	np.ndarray:
        _snake_case		=				self.numpy_arrow_extractor().extract_column(A__						)
        _snake_case		=				self.python_features_decoder.decode_column(A__     ,   pa_table.column_names[0]						)
        _snake_case		=				self.recursive_tensorize(A__						)
        _snake_case		=				self._consolidate(A__						)
        return column







      def 		UpperCamelCase_						(     self :							Dict     ,   A__ :							pa.Table						)       ->	Mapping:
        _snake_case		=				self.numpy_arrow_extractor().extract_batch(A__						)
        _snake_case		=				self.python_features_decoder.decode_batch(A__						)
        _snake_case		=				self.recursive_tensorize(A__						)
        for column_name in batch:
          _snake_case		=				self._consolidate(batch[column_name]						)
        return batch

