
"""simple docstring"""






import numpy as np
import torch
from torch.utils.data import Dataset, IterableDataset

from ..utils.generic import ModelOutput

class 					__a     ( __a       ):



							'''simple docstring'''







							def __init__(     self  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase       )			->							List[Any]:




										'''simple docstring'''
										__lowercase       =       dataset
										__lowercase       =       process
										__lowercase       =       params
							def __len__(     self       )			->							str:




										'''simple docstring'''
										return len(self.dataset       )





							def __getitem__(     self  ,    _lowerCamelCase       )			->							str:




										'''simple docstring'''
										__lowercase       =       self.dataset[i]
										__lowercase       =       self.process(_lowerCamelCase  ,    **self.params       )
										return processed





class 					__a     ( __a       ):



							'''simple docstring'''







							def __init__(     self  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase=None       )			->							List[str]:




										'''simple docstring'''
										__lowercase       =       loader
										__lowercase       =       infer
										__lowercase       =       params
										if loader_batch_size == 1:
													# Let's spare some time by deactivating altogether
													__lowercase       =       None
										__lowercase       =       loader_batch_size

										# Internal bookkeeping
										__lowercase       =       None
										__lowercase       =       None
							def __len__(     self       )			->							List[Any]:




										'''simple docstring'''
										return len(self.loader       )
							def __iter__(     self       )			->							Optional[Any]:




										'''simple docstring'''
										__lowercase       =       iter(self.loader       )
										return self
							def   SCREAMING_SNAKE_CASE						(     self       )			->							str:




										'''simple docstring'''
										if isinstance(self._loader_batch_data  ,    torch.Tensor       ):
													# Batch data is simple tensor, just fetch the slice
													__lowercase       =       self._loader_batch_data[self._loader_batch_index]
										else:
													# Batch data is assumed to be BaseModelOutput (or dict)
													__lowercase       =       {}
													for k, element in self._loader_batch_data.items():
																if isinstance(_lowerCamelCase  ,    _lowerCamelCase       ):
																			# Convert ModelOutput to tuple first
																			__lowercase       =       element.to_tuple()
																			if isinstance(element[0]  ,    torch.Tensor       ):
																						__lowercase       =       tuple(el[self._loader_batch_index].unsqueeze(0       ) for el in element       )
																			elif isinstance(element[0]  ,    np.ndarray       ):
																						__lowercase       =       tuple(np.expand_dims(el[self._loader_batch_index]  ,    0       ) for el in element       )
																			continue
																if k in {"hidden_states", "past_key_values", "attentions"} and isinstance(_lowerCamelCase  ,    _lowerCamelCase       ):
																			# Those are stored as lists of tensors so need specific unbatching.
																			if isinstance(element[0]  ,    torch.Tensor       ):
																						__lowercase       =       tuple(el[self._loader_batch_index].unsqueeze(0       ) for el in element       )
																			elif isinstance(element[0]  ,    np.ndarray       ):
																						__lowercase       =       tuple(np.expand_dims(el[self._loader_batch_index]  ,    0       ) for el in element       )
																			continue
																if element is None:
																			# This can happen for optional data that get passed around
																			__lowercase       =       None
																elif isinstance(element[self._loader_batch_index]  ,    torch.Tensor       ):
																			# Take correct batch data, but make it looked like batch_size=1
																			# For compatibility with other methods within transformers

																			__lowercase       =       element[self._loader_batch_index].unsqueeze(0       )
																elif isinstance(element[self._loader_batch_index]  ,    np.ndarray       ):
																			# Take correct batch data, but make it looked like batch_size=1
																			# For compatibility with other methods within transformers
																			__lowercase       =       np.expand_dims(element[self._loader_batch_index]  ,    0       )
																else:
																			# This is typically a list, so no need to `unsqueeze`.
																			__lowercase       =       element[self._loader_batch_index]
            # Recreate the element by reusing the original class to make it look
            # batch_size=1
													__lowercase       =       self._loader_batch_data.__class__(_lowerCamelCase       )
										self._loader_batch_index += 1
										return result





							def   SCREAMING_SNAKE_CASE						(     self       )			->							Optional[int]:




										'''simple docstring'''
										if self._loader_batch_index is not None and self._loader_batch_index < self.loader_batch_size:
													# We are currently unrolling a batch so we just need to return
													# the current item within a batch
													return self.loader_batch_item()

										# We're out of items within a batch
										__lowercase       =       next(self.iterator       )
										__lowercase       =       self.infer(_lowerCamelCase  ,    **self.params       )
										# We now have a batch of "inferred things".
										if self.loader_batch_size is not None:
													# Try to infer the size of the batch
													if isinstance(_lowerCamelCase  ,    torch.Tensor       ):
																__lowercase       =       processed
													else:
																__lowercase       =       list(processed.keys()       )[0]
																__lowercase       =       processed[key]
													if isinstance(_lowerCamelCase  ,    _lowerCamelCase       ):
																__lowercase       =       len(_lowerCamelCase       )
													else:
																__lowercase       =       first_tensor.shape[0]
													if 0 < observed_batch_size < self.loader_batch_size:
																# could be last batch so we can't unroll as many
																# elements.
																__lowercase       =       observed_batch_size
													# Setting internal index to unwrap the batch
													__lowercase       =       processed
													__lowercase       =       0
													return self.loader_batch_item()
										else:
													# We're not unrolling batches
													return processed





class 					__a     ( __a       ):



							'''simple docstring'''







							def __init__(     self  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase=None       )			->							Optional[Any]:




										'''simple docstring'''
										super().__init__(_lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase       )
							def __iter__(     self       )			->							Optional[int]:




										'''simple docstring'''
										__lowercase       =       iter(self.loader       )
										__lowercase       =       None
										return self





							def   SCREAMING_SNAKE_CASE						(     self       )			->							int:




										'''simple docstring'''
										if self.subiterator is None:
													__lowercase       =       self.infer(next(self.iterator       )  ,    **self.params       )
										try:
													# Try to return next item
													__lowercase       =       next(self.subiterator       )
										except StopIteration:
													# When a preprocess iterator ends, we can start lookig at the next item
													# ChunkIterator will keep feeding until ALL elements of iterator
													# all have created their subiterator and have been iterating against.
													#
													# Another way to look at it, is we're basically flattening lists of lists
													# into a single list, but with generators
													__lowercase       =       self.infer(next(self.iterator       )  ,    **self.params       )
													__lowercase       =       next(self.subiterator       )
										return processed





class 					__a     ( __a       ):



							'''simple docstring'''







							def __iter__(     self       )			->							Any:




										'''simple docstring'''
										__lowercase       =       iter(self.loader       )
										return self





							def   SCREAMING_SNAKE_CASE						(     self       )			->							Any:




										'''simple docstring'''
										# Extremely similar to PipelineIterator in its unpacking mechanism
										# BUT, we have an extra required item which is the presence of `is_last`
										# That is because everything is flattened by `PipelineChunkIterator` we
										# need to keep track of how to regroup here in the original `process`
										# boundaries so that `process` and `postprocess` see the same data.

										# This iterator accumulates items (possibly while unbatching) until it
										# its a `is_last` and then just passes it on to the caller.
										__lowercase       =       False
										__lowercase       =       []
										if self._loader_batch_index is not None and self._loader_batch_index < self.loader_batch_size:
													while self._loader_batch_index < self.loader_batch_size:
																__lowercase       =       self.loader_batch_item()
																__lowercase       =       item.pop("is_last"       )
																accumulator.append(_lowerCamelCase       )
																if is_last:
																			return accumulator

										while not is_last:
													__lowercase       =       self.infer(next(self.iterator       )  ,    **self.params       )
													if self.loader_batch_size is not None:
																if isinstance(_lowerCamelCase  ,    torch.Tensor       ):
																			__lowercase       =       processed
																else:
																			__lowercase       =       list(processed.keys()       )[0]
																			__lowercase       =       processed[key]
																if isinstance(_lowerCamelCase  ,    _lowerCamelCase       ):
																			__lowercase       =       len(_lowerCamelCase       )
																else:
																			__lowercase       =       first_tensor.shape[0]
																if 0 < observed_batch_size < self.loader_batch_size:
																			# could be last batch so we can't unroll as many
																			# elements.
																			__lowercase       =       observed_batch_size
																__lowercase       =       processed
																__lowercase       =       0
																while self._loader_batch_index < self.loader_batch_size:
																			__lowercase       =       self.loader_batch_item()
																			__lowercase       =       item.pop("is_last"       )
																			accumulator.append(_lowerCamelCase       )
																			if is_last:
																						return accumulator
													else:
																__lowercase       =       processed
																__lowercase       =       item.pop("is_last"       )
																accumulator.append(_lowerCamelCase       )
										return accumulator





class 					__a     ( __a       ):



							'''simple docstring'''







							def __init__(     self  ,    _lowerCamelCase  ,    _lowerCamelCase       )			->							List[str]:




										'''simple docstring'''
										__lowercase       =       dataset
										__lowercase       =       key
							def __len__(     self       )			->							List[Any]:




										'''simple docstring'''
										return len(self.dataset       )





							def __getitem__(     self  ,    _lowerCamelCase       )			->							List[str]:




										'''simple docstring'''
										return self.dataset[i][self.key]





class 					__a     ( __a       ):



							'''simple docstring'''







							def __init__(     self  ,    _lowerCamelCase  ,    _lowerCamelCase  ,    _lowerCamelCase       )			->							str:




										'''simple docstring'''
										__lowercase       =       dataset
										__lowercase       =       keya
										__lowercase       =       keya
							def __len__(     self       )			->							Optional[int]:




										'''simple docstring'''
										return len(self.dataset       )





							def __getitem__(     self  ,    _lowerCamelCase       )			->							int:




										'''simple docstring'''
										return {"text": self.dataset[i][self.keya], "text_pair": self.dataset[i][self.keya]}







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



							'''simple docstring'''







							def   SCREAMING_SNAKE_CASE						(     self       )			->							int:




										'''simple docstring'''
										# A mock response for an HTTP head request to emulate server down
										__lowercase       =       mock.Mock()
										__lowercase       =       500
										__lowercase       =       {}
										__lowercase       =       HTTPError
										__lowercase       =       {}

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




										'''simple docstring'''
										# A mock response for an HTTP head request to emulate server down
										__lowercase       =       mock.Mock()
										__lowercase       =       500
										__lowercase       =       {}
										__lowercase       =       HTTPError
										__lowercase       =       {}

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




										'''simple docstring'''
										# This test is for deprecated behavior and can be removed in v5
										try:
													__lowercase       =       tempfile.mktemp()
													with open(_lowerCamelCase  ,    "wb"       ) as f:
																http_get("https://huggingface.co/albert-base-v1/resolve/main/spiece.model"  ,    _lowerCamelCase       )

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

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





							def   SCREAMING_SNAKE_CASE						(     self       )			->							int:




										'''simple docstring'''
										# This test is for deprecated behavior and can be removed in v5
										__lowercase       =       AlbertTokenizer.from_pretrained("https://huggingface.co/albert-base-v1/resolve/main/spiece.model"       )





@is_staging_test
class 					__a     ( unittest.TestCase       ):



							'''simple docstring'''







							_lowerCamelCase					:			str          =   ["""[UNK]""", """[CLS]""", """[SEP]""", """[PAD]""", """[MASK]""", """bla""", """blou"""]
							@classmethod
							def   SCREAMING_SNAKE_CASE						(     cls       )			->							List[Any]:




										'''simple docstring'''
										__lowercase       =       TOKEN
										HfFolder.save_token(_lowerCamelCase       )
							@classmethod
							def   SCREAMING_SNAKE_CASE						(     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   SCREAMING_SNAKE_CASE						(     self       )			->							Any:




										'''simple docstring'''
										with tempfile.TemporaryDirectory() as tmp_dir:
													__lowercase       =       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]       )       )
													__lowercase       =       BertTokenizer(_lowerCamelCase       )

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

										__lowercase       =       BertTokenizer.from_pretrained(f'''{USER}/test-tokenizer'''       )
										self.assertDictEqual(new_tokenizer.vocab  ,    tokenizer.vocab       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''
										with tempfile.TemporaryDirectory() as tmp_dir:
													__lowercase       =       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]       )       )
													__lowercase       =       BertTokenizer(_lowerCamelCase       )

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

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





							@require_tokenizers
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Optional[int]:




										'''simple docstring'''
										CustomTokenizer.register_for_auto_class()
										with tempfile.TemporaryDirectory() as tmp_dir:
													__lowercase       =       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]       )       )
													__lowercase       =       CustomTokenizer(_lowerCamelCase       )

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

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

													__lowercase       =       BertTokenizerFast.from_pretrained(_lowerCamelCase       )
													bert_tokenizer.save_pretrained(_lowerCamelCase       )
													__lowercase       =       CustomTokenizerFast.from_pretrained(_lowerCamelCase       )

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

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



							'''simple docstring'''







							def   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''
										__lowercase       =       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   SCREAMING_SNAKE_CASE						(     self       )			->							Tuple:




										'''simple docstring'''
										__lowercase       =       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   SCREAMING_SNAKE_CASE						(     self       )			->							Dict:




										'''simple docstring'''
										__lowercase       =       Trie()
										trie.add("A"       )
										self.assertEqual(trie.split("ABC"       )  ,    ["A", "BC"]       )
										self.assertEqual(trie.split("BCA"       )  ,    ["BC", "A"]       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							str:




										'''simple docstring'''
										__lowercase       =       Trie()
										trie.add("TOKEN]"       )
										trie.add("[SPECIAL_TOKEN]"       )
										self.assertEqual(trie.split("This is something [SPECIAL_TOKEN]"       )  ,    ["This is something ", "[SPECIAL_TOKEN]"]       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Union[str, Any]:




										'''simple docstring'''
										__lowercase       =       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   SCREAMING_SNAKE_CASE						(     self       )			->							Any:




										'''simple docstring'''
										__lowercase       =       Trie()
										trie.add("AB"       )
										trie.add("B"       )
										trie.add("C"       )
										self.assertEqual(trie.split("ABC"       )  ,    ["AB", "C"]       )
							def   SCREAMING_SNAKE_CASE						(     self       )			->							Tuple:




										'''simple docstring'''
										__lowercase       =       Trie()
										trie.add("ABC"       )
										trie.add("B"       )
										trie.add("CD"       )
										self.assertEqual(trie.split("ABCD"       )  ,    ["ABC", "D"]       )





							def   SCREAMING_SNAKE_CASE						(     self       )			->							Tuple:




										'''simple docstring'''
										# Even if the offsets are wrong, we necessarily output correct string
										# parts.
										__lowercase       =       Trie()
										__lowercase       =       trie.cut_text("ABC"  ,    [0, 0, 2, 1, 2, 3]       )
										self.assertEqual(_lowerCamelCase  ,    ["AB", "C"]       )





