


'''simple docstring'''






import concurrent.futures
import json
import os
import shutil
import tempfile
import unittest

from transformers import AutoTokenizer, PreTrainedTokenizerFast
from transformers.testing_utils import require_tokenizers

from ..test_tokenization_common import TokenizerTesterMixin
@require_tokenizers
class   __UpperCamelCase		(  a_  ,     unittest.TestCase							):
							lowercase	:						List[Any]      						=     PreTrainedTokenizerFast
							lowercase	:						Tuple      						=     False
							lowercase	:						Optional[int]      						=     True
							lowercase	:						Union[str, Any]      						=     "tokenizer_file"





							def 						a__      (     self						:int	):
									snake_case_						:							Union[str, Any]					=  False  # because we don't have pretrained_vocab_files_map
									super().setUp()
									snake_case_						:							Union[str, Any]					=  True

									snake_case_						:							List[str]					=  ["""robot-test/dummy-tokenizer-fast""", """robot-test/dummy-tokenizer-wordlevel"""]
									snake_case_						:							int					=  """SaulLu/dummy-tokenizer-bytelevel-bpe"""

									# Inclusion of 2 tokenizers to test different types of models (Unigram and WordLevel for the moment)
									snake_case_						:							int					=  [(PreTrainedTokenizerFast, model_path, {}) for model_path in model_paths]

									snake_case_						:							Optional[Any]					=  PreTrainedTokenizerFast.from_pretrained(model_paths[0]	)
									tokenizer.save_pretrained(self.tmpdirname	)





							def 						a__      (     self						:Optional[int]	):
									pass





							def 						a__      (     self						:Optional[Any]	):
									pass





							def 						a__      (     self						:Tuple	):
									pass





							def 						a__      (     self						:Dict	):
									pass





							def 						a__      (     self						:Optional[int]	):
									snake_case_						:							Dict					=  self.tmpdirname
									# Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
									for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
											with self.subTest(F'''{tokenizer.__class__.__name__} ({pretrained_name})'''	):
													try:
															snake_case_						:							str					=  tempfile.mkdtemp()
															snake_case_						:							Optional[Any]					=  self.rust_tokenizer_class.from_pretrained(_UpperCamelCase     ,**_UpperCamelCase	)

															tokenizer.save_pretrained(self.tmpdirname	)
															super().test_training_new_tokenizer()
													finally:
															# Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
															# is restored
															shutil.rmtree(self.tmpdirname	)
															snake_case_						:							List[str]					=  tmpdirname_orig





							def 						a__      (     self						:Optional[int]	):
									snake_case_						:							str					=  self.tmpdirname
									# Here we want to test the 2 available tokenizers that use 2 different types of models: Unigram and WordLevel.
									for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
											with self.subTest(F'''{tokenizer.__class__.__name__} ({pretrained_name})'''	):
													try:
															snake_case_						:							Tuple					=  tempfile.mkdtemp()
															snake_case_						:							Dict					=  self.rust_tokenizer_class.from_pretrained(_UpperCamelCase     ,**_UpperCamelCase	)

															tokenizer.save_pretrained(self.tmpdirname	)
															super().test_training_new_tokenizer_with_special_tokens_change()
													finally:
															# Even if the test fails, we must be sure that the folder is deleted and that the default tokenizer
															# is restored
															shutil.rmtree(self.tmpdirname	)
															snake_case_						:							Union[str, Any]					=  tmpdirname_orig





							def 						a__      (     self						:str	):
									snake_case_						:							Optional[Any]					=  self.rust_tokenizer_class.from_pretrained(self.bytelevel_bpe_model_name	)

									snake_case_						:							Union[str, Any]					=  ("""a""" for _ in range(1_0_0_0	))
									snake_case_						:							int					=  tokenizer.train_new_from_iterator(text_iterator=_UpperCamelCase     ,length=1_0_0_0     ,vocab_size=5_0	)

									snake_case_						:							str					=  new_tokenizer.encode("""a🤗"""	)
									self.assertEqual(_UpperCamelCase     ,[6_4, 1_7_2, 2_5_3, 9_7, 2_4_5]	)





@require_tokenizers
class   __UpperCamelCase		(  unittest.TestCase							):





							def 						a__      (     self						:Optional[Any]	):
									snake_case_						:							Tuple					=  AutoTokenizer.from_pretrained("""bert-base-cased"""	)
									snake_case_						:							Optional[Any]					=  json.loads(tokenizer._tokenizer.to_str()	)
									snake_case_						:							Dict					=  len(_UpperCamelCase	)

									with tempfile.TemporaryDirectory() as tmp_dir:
											# Hack to save this in the tokenizer_config.json
											snake_case_						:							Optional[int]					=  ["""tokenizer.4.0.0.json"""]
											tokenizer.save_pretrained(_UpperCamelCase	)
											json.dump(_UpperCamelCase     ,open(os.path.join(_UpperCamelCase     ,"""tokenizer.4.0.0.json"""	)     ,"""w"""	)	)

											# This should pick the new tokenizer file as the version of Transformers is > 4.0.0
											snake_case_						:							Union[str, Any]					=  AutoTokenizer.from_pretrained(_UpperCamelCase	)
											self.assertEqual(len(_UpperCamelCase	)     ,len(_UpperCamelCase	) + 1	)
											snake_case_						:							List[Any]					=  json.loads(new_tokenizer._tokenizer.to_str()	)
											self.assertIn("""huggingface"""     ,json_tokenizer["""model"""]["""vocab"""]	)

											# Will need to be adjusted if we reach v42 and this test is still here.
											# Should pick the old tokenizer file as the version of Transformers is < 4.0.0
											shutil.move(os.path.join(_UpperCamelCase     ,"""tokenizer.4.0.0.json"""	)     ,os.path.join(_UpperCamelCase     ,"""tokenizer.42.0.0.json"""	)	)
											snake_case_						:							Optional[int]					=  ["""tokenizer.42.0.0.json"""]
											tokenizer.save_pretrained(_UpperCamelCase	)
											snake_case_						:							Optional[int]					=  AutoTokenizer.from_pretrained(_UpperCamelCase	)
											self.assertEqual(len(_UpperCamelCase	)     ,len(_UpperCamelCase	)	)
											snake_case_						:							str					=  json.loads(new_tokenizer._tokenizer.to_str()	)
											self.assertNotIn("""huggingface"""     ,json_tokenizer["""model"""]["""vocab"""]	)





							def 						a__      (     self						:Optional[Any]	):
									snake_case_						:							Optional[int]					=  """hf-internal-testing/test-two-tokenizers"""

									# This should pick the new tokenizer file as the version of Transformers is > 4.0.0
									snake_case_						:							List[Any]					=  AutoTokenizer.from_pretrained(_UpperCamelCase	)
									self.assertEqual(len(_UpperCamelCase	)     ,2_8_9_9_7	)
									snake_case_						:							List[Any]					=  json.loads(tokenizer._tokenizer.to_str()	)
									self.assertIn("""huggingface"""     ,json_tokenizer["""model"""]["""vocab"""]	)

									# Testing an older version by monkey-patching the version in the module it's used.
									import transformers as old_transformers

									snake_case_						:							Dict					=  """3.0.0"""
									snake_case_						:							str					=  old_transformers.models.auto.AutoTokenizer.from_pretrained(_UpperCamelCase	)
									self.assertEqual(len(_UpperCamelCase	)     ,2_8_9_9_6	)
									snake_case_						:							Any					=  json.loads(old_tokenizer._tokenizer.to_str()	)
									self.assertNotIn("""huggingface"""     ,json_tokenizer["""model"""]["""vocab"""]	)





@require_tokenizers
class   __UpperCamelCase		(  unittest.TestCase							):





							def 						a__      (     self						:Dict	):
									snake_case_						:							Optional[int]					=  PreTrainedTokenizerFast.from_pretrained("""robot-test/dummy-tokenizer-wordlevel"""	)
									snake_case_						:							List[str]					=  """The Matrix is a 1999 science fiction action film."""

									with concurrent.futures.ThreadPoolExecutor() as executor:
											snake_case_						:							Union[str, Any]					=  [executor.submit(self.fetch     ,_UpperCamelCase     ,_UpperCamelCase	) for i in range(1_0	)]
											snake_case_						:							List[Any]					=  [future.result() for future in futures]
											self.assertEqual(_UpperCamelCase     ,[[1, 1_0, 0, 8, 0, 1_8, 0, 0, 0, 2] for i in range(1_0	)]	)





							def 						a__      (     self						:List[Any]     ,_UpperCamelCase						:List[str]     ,_UpperCamelCase						:List[str]	):
									return tokenizer.encode(_UpperCamelCase     ,truncation="""longest_first"""     ,padding="""longest"""	)


import csv
import os

import pytest

from datasets import Dataset, DatasetDict, Features, NamedSplit, Value
from datasets.io.csv import CsvDatasetReader, CsvDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases





def 	snake_case_      (__A     :						Any					,   __A     :						Union[str, Any]     )			->	int:
   assert isinstance(__A					,   __A     )
   assert dataset.num_rows == 4
   assert dataset.num_columns == 3
   assert dataset.column_names == ["col_1", "col_2", "col_3"]
   for feature, expected_dtype in expected_features.items():
      assert dataset.features[feature].dtype == expected_dtype





@pytest.mark.parametrize("""keep_in_memory"""					,   [False, True]     )
def 	snake_case_      (__A     :						int					,   __A     :						int					,   __A     :						List[str]     )			->	Tuple:
   __lowerCAmelCase  :			Dict					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			int					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
      __lowerCAmelCase  :			List[str]					    =						CsvDatasetReader(__A					,   cache_dir=__A					,   keep_in_memory=__A     ).read()
   _check_csv_dataset(__A					,   __A     )





@pytest.mark.parametrize(
    """features"""					,   [
        None,
        {"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""},
        {"""col_1""": """string""", """col_2""": """string""", """col_3""": """string"""},
        {"""col_1""": """int32""", """col_2""": """int32""", """col_3""": """int32"""},
        {"""col_1""": """float32""", """col_2""": """float32""", """col_3""": """float32"""},
    ]					,   )
def 	snake_case_      (__A     :						Dict					,   __A     :						List[Any]					,   __A     :						List[str]     )			->	Union[str, Any]:
   __lowerCAmelCase  :			str					    =						tmp_path / """cache"""
   # CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
   __lowerCAmelCase  :			Tuple					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			int					    =						features.copy() if features else default_expected_features
   __lowerCAmelCase  :			List[Any]					    =						(
       Features({feature: Value(__A     ) for feature, dtype in features.items()}     ) if features is not None else None
   )
   __lowerCAmelCase  :			Tuple					    =						CsvDatasetReader(__A					,   features=__A					,   cache_dir=__A     ).read()
   _check_csv_dataset(__A					,   __A     )





@pytest.mark.parametrize("""split"""					,   [None, NamedSplit("""train"""     ), """train""", """test"""]     )
def 	snake_case_      (__A     :						str					,   __A     :						Optional[int]					,   __A     :						int     )			->	Union[str, Any]:
   __lowerCAmelCase  :			str					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Tuple					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			Optional[Any]					    =						CsvDatasetReader(__A					,   cache_dir=__A					,   split=__A     ).read()
   _check_csv_dataset(__A					,   __A     )
   assert dataset.split == split if split else "train"





@pytest.mark.parametrize("""path_type"""					,   [str, list]     )
def 	snake_case_      (__A     :						Union[str, Any]					,   __A     :						Any					,   __A     :						List[str]     )			->	Optional[int]:
   if issubclass(__A					,   __A     ):
      __lowerCAmelCase  :			Dict					    =						csv_path
   elif issubclass(__A					,   __A     ):
      __lowerCAmelCase  :			Tuple					    =						[csv_path]
   __lowerCAmelCase  :			Dict					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Dict					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			List[str]					    =						CsvDatasetReader(__A					,   cache_dir=__A     ).read()
   _check_csv_dataset(__A					,   __A     )





def 	snake_case_      (__A     :						int					,   __A     :						Optional[int]					,   __A     :						Any=("train",)     )			->	Optional[int]:
   assert isinstance(__A					,   __A     )
   for split in splits:
      __lowerCAmelCase  :			int					    =						dataset_dict[split]
      assert dataset.num_rows == 4
      assert dataset.num_columns == 3
      assert dataset.column_names == ["col_1", "col_2", "col_3"]
      for feature, expected_dtype in expected_features.items():
         assert dataset.features[feature].dtype == expected_dtype





@pytest.mark.parametrize("""keep_in_memory"""					,   [False, True]     )
def 	snake_case_      (__A     :						Optional[Any]					,   __A     :						List[str]					,   __A     :						Any     )			->	List[str]:
   __lowerCAmelCase  :			int					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Optional[Any]					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
      __lowerCAmelCase  :			Tuple					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A					,   keep_in_memory=__A     ).read()
   _check_csv_datasetdict(__A					,   __A     )





@pytest.mark.parametrize(
    """features"""					,   [
        None,
        {"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""},
        {"""col_1""": """string""", """col_2""": """string""", """col_3""": """string"""},
        {"""col_1""": """int32""", """col_2""": """int32""", """col_3""": """int32"""},
        {"""col_1""": """float32""", """col_2""": """float32""", """col_3""": """float32"""},
    ]					,   )
def 	snake_case_      (__A     :						List[str]					,   __A     :						List[Any]					,   __A     :						Optional[int]     )			->	Optional[int]:
   __lowerCAmelCase  :			Optional[int]					    =						tmp_path / """cache"""
   # CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
   __lowerCAmelCase  :			Union[str, Any]					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			Optional[int]					    =						features.copy() if features else default_expected_features
   __lowerCAmelCase  :			Optional[Any]					    =						(
       Features({feature: Value(__A     ) for feature, dtype in features.items()}     ) if features is not None else None
   )
   __lowerCAmelCase  :			Union[str, Any]					    =						CsvDatasetReader({"""train""": csv_path}					,   features=__A					,   cache_dir=__A     ).read()
   _check_csv_datasetdict(__A					,   __A     )





@pytest.mark.parametrize("""split"""					,   [None, NamedSplit("""train"""     ), """train""", """test"""]     )
def 	snake_case_      (__A     :						Optional[Any]					,   __A     :						Optional[Any]					,   __A     :						Union[str, Any]     )			->	Optional[int]:
   if split:
      __lowerCAmelCase  :			Any					    =						{split: csv_path}
   else:
      __lowerCAmelCase  :			Dict					    =						{"""train""": csv_path, """test""": csv_path}
   __lowerCAmelCase  :			List[str]					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			int					    =						{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
   __lowerCAmelCase  :			int					    =						CsvDatasetReader(__A					,   cache_dir=__A     ).read()
   _check_csv_datasetdict(__A					,   __A					,   splits=list(path.keys()     )     )
   assert all(dataset[split].split == split for split in path.keys()     )





def 	snake_case_      (__A     :						List[str]     )			->	Optional[Any]:
   with open(__A					,   encoding="""utf-8"""     ) as csvfile:
      yield from csv.reader(__A     )





def 	snake_case_      (__A     :						Optional[int]					,   __A     :						Optional[int]     )			->	Union[str, Any]:
   __lowerCAmelCase  :			Optional[Any]					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			Tuple					    =						os.path.join(__A					,   """tmp.csv"""     )
   __lowerCAmelCase  :			Tuple					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A     ).read()
   CsvDatasetWriter(dataset["""train"""]					,   __A					,   num_proc=1     ).write()

   __lowerCAmelCase  :			Dict					    =						iter_csv_file(__A     )
   __lowerCAmelCase  :			Union[str, Any]					    =						iter_csv_file(__A     )

   for rowa, rowa in zip(__A					,   __A     ):
      assert rowa == rowa





def 	snake_case_      (__A     :						int					,   __A     :						List[str]     )			->	str:
   __lowerCAmelCase  :			Any					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			List[str]					    =						os.path.join(__A					,   """tmp.csv"""     )
   __lowerCAmelCase  :			Dict					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A     ).read()
   CsvDatasetWriter(dataset["""train"""]					,   __A					,   num_proc=2     ).write()

   __lowerCAmelCase  :			Union[str, Any]					    =						iter_csv_file(__A     )
   __lowerCAmelCase  :			Dict					    =						iter_csv_file(__A     )

   for rowa, rowa in zip(__A					,   __A     ):
      assert rowa == rowa





def 	snake_case_      (__A     :						int					,   __A     :						Optional[Any]     )			->	List[str]:
   __lowerCAmelCase  :			Any					    =						tmp_path / """cache"""
   __lowerCAmelCase  :			List[Any]					    =						os.path.join(__A					,   """tmp.csv"""     )
   __lowerCAmelCase  :			List[str]					    =						CsvDatasetReader({"""train""": csv_path}					,   cache_dir=__A     ).read()
   with pytest.raises(__A     ):
      CsvDatasetWriter(dataset["""train"""]					,   __A					,   num_proc=0     )


