





'''simple docstring'''






import os
import re
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


A :			Optional[Any]							   =      logging.get_logger(__name__)
A :			List[str]							   =      {'vocab_file': 'spiece.model'}

A :			Optional[int]							   =      {
    'vocab_file': {
        'google/fnet-base': 'https://huggingface.co/google/fnet-base/resolve/main/spiece.model',
        'google/fnet-large': 'https://huggingface.co/google/fnet-large/resolve/main/spiece.model',
    },
}

A :			List[str]							   =      {
    'google/fnet-base': 512,
    'google/fnet-large': 512,
}

A :			Optional[int]							   =      '▁'







class       lowerCamelCase       (    __UpperCAmelCase	):
					_SCREAMING_SNAKE_CASE				 =     VOCAB_FILES_NAMES
					_SCREAMING_SNAKE_CASE				 =     PRETRAINED_VOCAB_FILES_MAP
					_SCREAMING_SNAKE_CASE				 =     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					_SCREAMING_SNAKE_CASE				 =     ["input_ids", "token_type_ids"]


					def __init__(       self						:	List[str]							,  __snake_case						:	List[str]							,  __snake_case						:	Tuple=False							,  __snake_case						:	int=True							,  __snake_case						:	str=True							,  __snake_case						:	List[str]="<unk>"							,  __snake_case						:	List[Any]="[SEP]"							,  __snake_case						:	Optional[int]="<pad>"							,  __snake_case						:	Dict="[CLS]"							,  __snake_case						:	List[str]="[MASK]"							,  __snake_case						:	Optional[Dict[str, Any]] = None							,  **__snake_case						:	Any							,  ):





											'''simple docstring'''





											_snake_case: int	       =		(
											    AddedToken(__snake_case							,  lstrip=__snake_case							,  rstrip=__snake_case							,  normalized=__snake_case							)
											    if isinstance(__snake_case							,  __snake_case							)
											    else mask_token
											)

											_snake_case: Any	       =		{} if sp_model_kwargs is None else sp_model_kwargs

											super().__init__(
											    do_lower_case=__snake_case							,  remove_space=__snake_case							,  keep_accents=__snake_case							,  unk_token=__snake_case							,  sep_token=__snake_case							,  pad_token=__snake_case							,  cls_token=__snake_case							,  mask_token=__snake_case							,  sp_model_kwargs=self.sp_model_kwargs							,  **__snake_case							,  )

											_snake_case: List[str]	       =		do_lower_case
											_snake_case: str	       =		remove_space
											_snake_case: str	       =		keep_accents
											_snake_case: int	       =		vocab_file

											_snake_case: Dict	       =		spm.SentencePieceProcessor(**self.sp_model_kwargs							)
											self.sp_model.Load(__snake_case							)


					@property
					def  SCREAMING_SNAKE_CASE_		(       self						:	int							):





											'''simple docstring'''





											return len(self.sp_model							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	int							):





											'''simple docstring'''





											_snake_case: Dict	       =		{self.convert_ids_to_tokens(__snake_case							): i for i in range(self.vocab_size							)}
											vocab.update(self.added_tokens_encoder							)
											return vocab


					def __getstate__(       self						:	List[Any]							):





											'''simple docstring'''





											_snake_case: List[Any]	       =		self.__dict__.copy()
											_snake_case: Optional[Any]	       =		None
											return state


					def __setstate__(       self						:	str							,  __snake_case						:	int							):





											'''simple docstring'''





											_snake_case: Optional[int]	       =		d

											# for backward compatibility
											if not hasattr(self							,  'sp_model_kwargs'							):
																	_snake_case: List[Any]	       =		{}

											_snake_case: List[Any]	       =		spm.SentencePieceProcessor(**self.sp_model_kwargs							)
											self.sp_model.Load(self.vocab_file							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	str							,  __snake_case						:	Any							):





											'''simple docstring'''





											if self.remove_space:
																	_snake_case: Any	       =		' '.join(inputs.strip().split()							)
											else:
																	_snake_case: Any	       =		inputs
											_snake_case: str	       =		outputs.replace('``'							,  '"'							).replace('\'\''							,  '"'							)

											if not self.keep_accents:
																	_snake_case: Union[str, Any]	       =		unicodedata.normalize('NFKD'							,  __snake_case							)
																	_snake_case: Union[str, Any]	       =		''.join([c for c in outputs if not unicodedata.combining(__snake_case							)]							)
											if self.do_lower_case:
																	_snake_case: List[Any]	       =		outputs.lower()

											return outputs


					def  SCREAMING_SNAKE_CASE_		(       self						:	Dict							,  __snake_case						:	str							):





											'''simple docstring'''





											_snake_case: Dict	       =		self.preprocess_text(__snake_case							)
											_snake_case: Optional[int]	       =		self.sp_model.encode(__snake_case							,  out_type=__snake_case							)
											_snake_case: str	       =		[]
											for piece in pieces:
																	if len(__snake_case							) > 1 and piece[-1] == str(','							) and piece[-2].isdigit():
																							_snake_case: Union[str, Any]	       =		self.sp_model.EncodeAsPieces(piece[:-1].replace(__snake_case							,  ''							)							)
																							if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
																													if len(cur_pieces[0]							) == 1:
																																			_snake_case: List[Any]	       =		cur_pieces[1:]
																													else:
																																			_snake_case: Dict	       =		cur_pieces[0][1:]
																							cur_pieces.append(piece[-1]							)
																							new_pieces.extend(__snake_case							)
																	else:
																							new_pieces.append(__snake_case							)

											return new_pieces


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							,  __snake_case						:	str							):





											'''simple docstring'''





											return self.sp_model.PieceToId(__snake_case							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[Any]							,  __snake_case						:	Dict							):





											'''simple docstring'''





											return self.sp_model.IdToPiece(__snake_case							)


					def  SCREAMING_SNAKE_CASE_		(       self						:	Dict							,  __snake_case						:	Optional[Any]							):





											'''simple docstring'''





											_snake_case: List[str]	       =		[]
											_snake_case: Any	       =		''
											_snake_case: Union[str, Any]	       =		False
											for token in tokens:
																	# make sure that special tokens are not decoded using sentencepiece model
																	if token in self.all_special_tokens:
																							if not prev_is_special:
																													out_string += " "
																							out_string += self.sp_model.decode(__snake_case							) + token
																							_snake_case: Optional[Any]	       =		True
																							_snake_case: Any	       =		[]
																	else:
																							current_sub_tokens.append(__snake_case							)
																							_snake_case: Optional[int]	       =		False
											out_string += self.sp_model.decode(__snake_case							)
											return out_string.strip()


					def  SCREAMING_SNAKE_CASE_		(       self						:	List[str]							,  __snake_case						:	List[int]							,  __snake_case						:	bool = False							,  __snake_case						:	bool = None							,  __snake_case						:	bool = True							,  **__snake_case						:	Optional[Any]							,  ):





											'''simple docstring'''





											_snake_case: Optional[Any]	       =		kwargs.pop('use_source_tokenizer'							,  __snake_case							)

											_snake_case: Any	       =		self.convert_ids_to_tokens(__snake_case							,  skip_special_tokens=__snake_case							)

											# To avoid mixing byte-level and unicode for byte-level BPT
											# we need to build string separately for added tokens and byte-level tokens
											# cf. https://github.com/huggingface/transformers/issues/1133
											_snake_case: Tuple	       =		[]
											_snake_case: int	       =		[]
											for token in filtered_tokens:
																	if skip_special_tokens and token in self.all_special_ids:
																							continue
																	if token in self.added_tokens_encoder:
																							if current_sub_text:
																													sub_texts.append(self.convert_tokens_to_string(__snake_case							)							)
																													_snake_case: Optional[int]	       =		[]
																							sub_texts.append(__snake_case							)
																	else:
																							current_sub_text.append(__snake_case							)
											if current_sub_text:
																	sub_texts.append(self.convert_tokens_to_string(__snake_case							)							)

											# Mimic the behavior of the Rust tokenizer:
											# No space after <unk>
											if spaces_between_special_tokens:
																	_snake_case: Dict	       =		re.sub(R'(<unk>) '							,  R'\1'							,  ' '.join(__snake_case							)							)
											else:
																	_snake_case: List[str]	       =		''.join(__snake_case							)

											_snake_case: List[Any]	       =		(
											    clean_up_tokenization_spaces
											    if clean_up_tokenization_spaces is not None
											    else self.clean_up_tokenization_spaces
											)
											if clean_up_tokenization_spaces:
																	_snake_case: Tuple	       =		self.clean_up_tokenization(__snake_case							)
																	return clean_text
											else:
																	return text


					def  SCREAMING_SNAKE_CASE_		(       self						:	Dict							,  __snake_case						:	List[int]							,  __snake_case						:	Optional[List[int]] = None							):





											'''simple docstring'''





											_snake_case: Dict	       =		[self.sep_token_id]
											_snake_case: Dict	       =		[self.cls_token_id]
											if token_ids_a is None:
																	return cls + token_ids_a + sep
											return cls + token_ids_a + sep + token_ids_a + sep


					def  SCREAMING_SNAKE_CASE_		(       self						:	Any							,  __snake_case						:	List[int]							,  __snake_case						:	Optional[List[int]] = None							,  __snake_case						:	bool = False							):





											'''simple docstring'''





											if already_has_special_tokens:
																	return super().get_special_tokens_mask(
																	    token_ids_a=__snake_case							,  token_ids_a=__snake_case							,  already_has_special_tokens=__snake_case							)

											if token_ids_a is not None:
																	return [1] + ([0] * len(__snake_case							)) + [1] + ([0] * len(__snake_case							)) + [1]
											return [1] + ([0] * len(__snake_case							)) + [1]


					def  SCREAMING_SNAKE_CASE_		(       self						:	Dict							,  __snake_case						:	List[int]							,  __snake_case						:	Optional[List[int]] = None							):





											'''simple docstring'''





											_snake_case: Any	       =		[self.sep_token_id]
											_snake_case: Dict	       =		[self.cls_token_id]

											if token_ids_a is None:
																	return len(cls + token_ids_a + sep							) * [0]
											return len(cls + token_ids_a + sep							) * [0] + len(token_ids_a + sep							) * [1]




					def  SCREAMING_SNAKE_CASE_		(       self						:	Any							,  __snake_case						:	str							,  __snake_case						:	Optional[str] = None							):





											'''simple docstring'''





											if not os.path.isdir(__snake_case							):
																	logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''							)
																	return
											_snake_case: List[str]	       =		os.path.join(
											    __snake_case							,  (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file']							)

											if os.path.abspath(self.vocab_file							) != os.path.abspath(__snake_case							) and os.path.isfile(self.vocab_file							):
																	copyfile(self.vocab_file							,  __snake_case							)
											elif not os.path.isfile(self.vocab_file							):
																	with open(__snake_case							,  'wb'							) as fi:
																							_snake_case: List[str]	       =		self.sp_model.serialized_model_proto()
																							fi.write(__snake_case							)

											return (out_vocab_file,)












'''simple docstring'''






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    lowercase_   (		lowercase__					,	lowercase__    ) ->List[str]:
						assert isinstance(lowercase__					,	lowercase__    )
						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    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->str:
						_snake_case: Optional[int]	       =		tmp_path / 'cache'
						_snake_case: 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():
												_snake_case: Any	       =		CsvDatasetReader(lowercase__					,	cache_dir=lowercase__					,	keep_in_memory=lowercase__    ).read()
						_check_csv_dataset(lowercase__					,	lowercase__    )





@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    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->List[Any]:
						_snake_case: str	       =		tmp_path / 'cache'
						# CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
						_snake_case: Union[str, Any]	       =		{'col_1': 'int64', 'col_2': 'int64', 'col_3': 'float64'}
						_snake_case: List[str]	       =		features.copy() if features else default_expected_features
						_snake_case: str	       =		(
						    Features({feature: Value(lowercase__    ) for feature, dtype in features.items()}    ) if features is not None else None
						)
						_snake_case: Optional[int]	       =		CsvDatasetReader(lowercase__					,	features=lowercase__					,	cache_dir=lowercase__    ).read()
						_check_csv_dataset(lowercase__					,	lowercase__    )





@pytest.mark.parametrize('split'					,	[None, NamedSplit('train'    ), 'train', 'test']    )
def    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->List[str]:
						_snake_case: Dict	       =		tmp_path / 'cache'
						_snake_case: int	       =		{'col_1': 'int64', 'col_2': 'int64', 'col_3': 'float64'}
						_snake_case: str	       =		CsvDatasetReader(lowercase__					,	cache_dir=lowercase__					,	split=lowercase__    ).read()
						_check_csv_dataset(lowercase__					,	lowercase__    )
						assert dataset.split == split if split else "train"





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





def    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__=("train",)    ) ->str:
						assert isinstance(lowercase__					,	lowercase__    )
						for split in splits:
												_snake_case: Optional[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    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->Optional[int]:
						_snake_case: Optional[Any]	       =		tmp_path / 'cache'
						_snake_case: 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():
												_snake_case: Dict	       =		CsvDatasetReader({'train': csv_path}					,	cache_dir=lowercase__					,	keep_in_memory=lowercase__    ).read()
						_check_csv_datasetdict(lowercase__					,	lowercase__    )





@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    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->Optional[int]:
						_snake_case: List[str]	       =		tmp_path / 'cache'
						# CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
						_snake_case: str	       =		{'col_1': 'int64', 'col_2': 'int64', 'col_3': 'float64'}
						_snake_case: str	       =		features.copy() if features else default_expected_features
						_snake_case: int	       =		(
						    Features({feature: Value(lowercase__    ) for feature, dtype in features.items()}    ) if features is not None else None
						)
						_snake_case: Tuple	       =		CsvDatasetReader({'train': csv_path}					,	features=lowercase__					,	cache_dir=lowercase__    ).read()
						_check_csv_datasetdict(lowercase__					,	lowercase__    )





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





def    lowercase_   (		lowercase__    ) ->Union[str, Any]:
						with open(lowercase__					,	encoding='utf-8'    ) as csvfile:
												yield from csv.reader(lowercase__    )





def    lowercase_   (		lowercase__					,	lowercase__    ) ->str:
						_snake_case: List[Any]	       =		tmp_path / 'cache'
						_snake_case: Dict	       =		os.path.join(lowercase__					,	'tmp.csv'    )
						_snake_case: Optional[Any]	       =		CsvDatasetReader({'train': csv_path}					,	cache_dir=lowercase__    ).read()
						CsvDatasetWriter(dataset['train']					,	lowercase__					,	num_proc=1    ).write()

						_snake_case: Optional[int]	       =		iter_csv_file(lowercase__    )
						_snake_case: List[Any]	       =		iter_csv_file(lowercase__    )

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





def    lowercase_   (		lowercase__					,	lowercase__    ) ->str:
						_snake_case: List[str]	       =		tmp_path / 'cache'
						_snake_case: List[str]	       =		os.path.join(lowercase__					,	'tmp.csv'    )
						_snake_case: List[str]	       =		CsvDatasetReader({'train': csv_path}					,	cache_dir=lowercase__    ).read()
						CsvDatasetWriter(dataset['train']					,	lowercase__					,	num_proc=2    ).write()

						_snake_case: Optional[Any]	       =		iter_csv_file(lowercase__    )
						_snake_case: Tuple	       =		iter_csv_file(lowercase__    )

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





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





