



'''simple docstring'''






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

import sentencepiece as spm

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


if is_torch_available():
						import torch


if TYPE_CHECKING:
						from transformers.pipelines.conversational import Conversation


UpperCamelCase__:     List[Any]										=    logging.get_logger(__name__)
UpperCamelCase__:     Dict										=    {"vocab_file": "spiece.model"}

UpperCamelCase__:     List[str]										=    {
    "vocab_file": {
        "AI-Sweden/gpt-sw3-126m": "https://huggingface.co/AI-Sweden/gpt-sw3-126m/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-350m": "https://huggingface.co/AI-Sweden/gpt-sw3-350m/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-1.6b": "https://huggingface.co/AI-Sweden/gpt-sw3-1.6b/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-6.7b": "https://huggingface.co/AI-Sweden/gpt-sw3-6.7b/resolve/main/spiece.model",
        "AI-Sweden/gpt-sw3-20b": "https://huggingface.co/AI-Sweden/gpt-sw3-20b/resolve/main/spiece.model",
    }
}

UpperCamelCase__:     Optional[Any]										=    {
    "AI-Sweden/gpt-sw3-126m": 2048,
    "AI-Sweden/gpt-sw3-350m": 2048,
    "AI-Sweden/gpt-sw3-1.6b": 2048,
    "AI-Sweden/gpt-sw3-6.7b": 2048,
    "AI-Sweden/gpt-sw3-20b": 2048,
}

class        SCREAMING_SNAKE_CASE(							lowercase_    ):







	"""simple docstring"""






	lowerCamelCase__    				= VOCAB_FILES_NAMES
	lowerCamelCase__    				= PRETRAINED_VOCAB_FILES_MAP
	lowerCamelCase__    				= PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	lowerCamelCase__    				= ['''input_ids''', '''attention_mask''']







	def __init__(		self    :  Optional[int]	,    __snake_case    :  List[Any]	,    __snake_case    :  int=False	,    __snake_case    :  Tuple=False	,    __snake_case    :  List[Any]=False	,    __snake_case    :  Union[str, Any]=None	,    __snake_case    :  Any=None	,    __snake_case    :  List[Any]=None	,    __snake_case    :  str=None	,    __snake_case    :  Optional[Dict[str, Any]] = None	,    **__snake_case    :  List[str]	,    )    ->	List[Any]:
								UpperCAmelCase				: str											=	{} if sp_model_kwargs is None else sp_model_kwargs

								UpperCAmelCase				: Any											=	kwargs.get('''name_or_path'''	)
								if name_or_path is None:
															logger.warning(
															    '''name_or_path not provided, will work for all GPTSw3 models except gpt-sw3-7b,'''
															    ''' you are testing the model, this can safely be ignored'''	)
															UpperCAmelCase				: Union[str, Any]											=	'''None'''

								# Default definitions for our 2 tokenizer versions, with None-checks to enable proper testing
								UpperCAmelCase				: int											=	'''<|endoftext|>''' if eos_token is None else eos_token
								UpperCAmelCase				: List[Any]											=	'''<unk>''' if unk_token is None else unk_token
								if "gpt-sw3-7b" in name_or_path:
															UpperCAmelCase				: Any											=	unk_token if pad_token is None else pad_token
															UpperCAmelCase				: Any											=	eos_token if bos_token is None else bos_token
								else:
															UpperCAmelCase				: Tuple											=	'''<pad>''' if pad_token is None else pad_token
															UpperCAmelCase				: Any											=	'''<s>''' if bos_token is None else bos_token

								super().__init__(
								    do_lower_case=UpperCamelCase__	,    remove_space=UpperCamelCase__	,    keep_accents=UpperCamelCase__	,    bos_token=UpperCamelCase__	,    eos_token=UpperCamelCase__	,    unk_token=UpperCamelCase__	,    pad_token=UpperCamelCase__	,    sp_model_kwargs=self.sp_model_kwargs	,    **UpperCamelCase__	,    )

								UpperCAmelCase				: List[Any]											=	do_lower_case
								UpperCAmelCase				: int											=	remove_space
								UpperCAmelCase				: Union[str, Any]											=	keep_accents
								UpperCAmelCase				: Tuple											=	vocab_file

								UpperCAmelCase				: int											=	spm.SentencePieceProcessor(**self.sp_model_kwargs	)
								self.sp_model.Load(UpperCamelCase__	)

								# Used for whitespace normalization in input texts
								# fmt : off
								UpperCAmelCase				: int											=	{''' ''', ''' ''', ''' ''', ''' ''', ''' ''', '''　''', ''' ''', ''' ''', ''' ''', ''' ''', '''￼''', ''''''}
								# fmt : on

								# Regular expression to remove non-printing characters (e.g. some unicode control chars) in preprocessing
								UpperCAmelCase				: List[Any]											=	re.compile(
								    F"""[{"".join(map(UpperCamelCase__	,    list(range(0	,    9	)	) + list(range(11	,    32	)	) + list(range(127	,    160	)	) + [160, 173, 8203]	)	)}]"""	)







	def __getstate__(		self    :  Tuple	)    ->	int:
								UpperCAmelCase				: Optional[Any]											=	self.__dict__.copy()
								UpperCAmelCase				: Optional[Any]											=	None
								return state







	def __setstate__(		self    :  Dict	,    __snake_case    :  Optional[Any]	)    ->	str:
								UpperCAmelCase				: Union[str, Any]											=	d

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

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







	@property
	# Copied from transformers.models.albert.tokenization_albert.AlbertTokenizer.vocab_size
	def   A			(		self    :  int	)    ->	Union[str, Any]:
								return len(self.sp_model	)







	def   A			(		self    :  List[str]	,    __snake_case    :  str	)    ->	Optional[int]:
								UpperCAmelCase				: str											=	self.non_printing_characters_re.sub(''''''	,    UpperCamelCase__	)

								# Normalize whitespaces
								UpperCAmelCase				: Union[str, Any]											=	''''''.join([char if char not in self.whitespaces else ''' ''' for char in text]	)

								# NFC Unicode normalization
								UpperCAmelCase				: Union[str, Any]											=	unicodedata.normalize('''NFC'''	,    UpperCamelCase__	)
								return text







	def   A			(		self    :  Any	,    __snake_case    :  str	,    **__snake_case    :  List[str]	)    ->	Union[str, Any]:
								UpperCAmelCase				: List[Any]											=	self.preprocess_text(UpperCamelCase__	)
								return self.sp_model.encode(UpperCamelCase__	,    out_type=UpperCamelCase__	)







	def   A			(		self    :  Union[str, Any]	,    __snake_case    :  str	)    ->	Tuple:
								return self.sp_model.PieceToId(UpperCamelCase__	)







	def   A			(		self    :  Tuple	,    __snake_case    :  int	)    ->	Union[str, Any]:
								return self.sp_model.IdToPiece(UpperCamelCase__	)







	@staticmethod
	def   A			(		__snake_case    :  str	)    ->	List[Any]:
								return out_string







	def   A			(		self    :  Union[str, Any]	,    __snake_case    :  List[str]	)    ->	Tuple:
								UpperCAmelCase				: Any											=	[]
								UpperCAmelCase				: Optional[Any]											=	''''''
								UpperCAmelCase				: Optional[int]											=	False
								for token in tokens:
															# make sure that special tokens are not decoded using sentencepiece model
															if token in self.all_special_tokens:
																						# TODO: Check if this is needed, as it ensures that decode(encode(doc)) != doc by adding extra whitespace in the decoded document
																						if not prev_is_special:
																													out_string += " "

																						out_string += self.sp_model.decode(UpperCamelCase__	) + token
																						UpperCAmelCase				: Union[str, Any]											=	True
																						UpperCAmelCase				: Tuple											=	[]
															else:
																						current_sub_tokens.append(UpperCamelCase__	)
																						UpperCAmelCase				: Optional[Any]											=	False
								out_string += self.sp_model.decode(UpperCamelCase__	)

								return out_string







	def   A			(		self    :  Any	)    ->	Dict:
								UpperCAmelCase				: List[str]											=	{self.convert_ids_to_tokens(UpperCamelCase__	): i for i in range(self.vocab_size	)}
								vocab.update(self.added_tokens_encoder	)
								return vocab







	def   A			(		self    :  Optional[Any]	,    __snake_case    :  str	,    __snake_case    :  Optional[str] = None	)    ->	Tuple:







								if not os.path.isdir(UpperCamelCase__	):
															logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""	)
															return
								UpperCAmelCase				: int											=	os.path.join(
								    UpperCamelCase__	,    (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']	)

								if os.path.abspath(self.vocab_file	) != os.path.abspath(UpperCamelCase__	) and os.path.isfile(self.vocab_file	):
															copyfile(self.vocab_file	,    UpperCamelCase__	)
								elif not os.path.isfile(self.vocab_file	):
															with open(UpperCamelCase__	,    '''wb'''	) as fi:
																						UpperCAmelCase				: Optional[int]											=	self.sp_model.serialized_model_proto()
																						fi.write(UpperCamelCase__	)

								return (out_vocab_file,)







	def   A			(		self    :  str	,    __snake_case    :  Union[str, List[str]]	,    __snake_case    :  Union[str, bool] = False	)    ->	Union[str, Any]:







								if isinstance(UpperCamelCase__	,    UpperCamelCase__	):
															UpperCAmelCase				: Any											=	self.preprocess_text(UpperCamelCase__	)
															UpperCAmelCase				: str											=	self.sp_model.encode(UpperCamelCase__	)
								else:
															UpperCAmelCase				: Optional[int]											=	[self.preprocess_text(UpperCamelCase__	) for t in text]
															UpperCAmelCase				: Optional[int]											=	self.sp_model.encode(UpperCamelCase__	)

								if return_tensors is True or return_tensors == "pt":
															UpperCAmelCase				: List[str]											=	torch.tensor(UpperCamelCase__	)

								return token_ids







	def   A			(		self    :  str	,    __snake_case    :  Union[int, List[int]]	)    ->	Optional[Any]:
								return self.sp_model.decode(UpperCamelCase__	)







	def   A			(		self    :  Union[str, Any]	,    __snake_case    :  "Conversation"	)    ->	Tuple:
								UpperCAmelCase				: List[str]											=	[F"""User: {text}""" if is_user else F"""Bot: {text}""" for is_user, text in conversation.iter_texts()]
								UpperCAmelCase				: Tuple											=	(
								    F"""{self.eos_token}{self.bos_token}""" + F"""{self.bos_token}""".join(UpperCamelCase__	) + F"""{self.bos_token}Bot:"""
								)
								return self.encode(text=UpperCamelCase__	)










'''simple docstring'''






import contextlib
import os
import sqlitea

import pytest

from datasets import Dataset, Features, Value
from datasets.io.sql import SqlDatasetReader, SqlDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases, require_sqlalchemy


def 				snake_case_  (      _lowerCAmelCase						:		int     ,   _lowerCAmelCase						:		Tuple						)   ->     str:
							assert isinstance(_lowerCAmelCase     ,   _lowerCAmelCase						)
							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


@require_sqlalchemy
@pytest.mark.parametrize('''keep_in_memory'''     ,   [False, True]						)
def 				snake_case_  (      _lowerCAmelCase						:		Union[str, Any]     ,   _lowerCAmelCase						:		List[str]     ,   _lowerCAmelCase						:		Any     ,   _lowerCAmelCase						:		Optional[Any]						)   ->     List[str]:
							UpperCAmelCase				: int											=	tmp_path / '''cache'''
							UpperCAmelCase				: Dict											=	{'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
							with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
														UpperCAmelCase				: Any											=	SqlDatasetReader(
														    '''dataset'''     ,   '''sqlite:///''' + sqlite_path     ,   cache_dir=_lowerCAmelCase     ,   keep_in_memory=_lowerCAmelCase						).read()
							_check_sql_dataset(_lowerCAmelCase     ,   _lowerCAmelCase						)


@require_sqlalchemy
@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_  (      _lowerCAmelCase						:		int     ,   _lowerCAmelCase						:		Any     ,   _lowerCAmelCase						:		Optional[int]     ,   _lowerCAmelCase						:		str						)   ->     Optional[int]:
							UpperCAmelCase				: Union[str, Any]											=	tmp_path / '''cache'''
							UpperCAmelCase				: Union[str, Any]											=	{'''col_1''': '''string''', '''col_2''': '''int64''', '''col_3''': '''float64'''}
							UpperCAmelCase				: str											=	features.copy() if features else default_expected_features
							UpperCAmelCase				: Optional[Any]											=	(
							    Features({feature: Value(_lowerCAmelCase						) for feature, dtype in features.items()}						) if features is not None else None
							)
							UpperCAmelCase				: Tuple											=	SqlDatasetReader('''dataset'''     ,   '''sqlite:///''' + sqlite_path     ,   features=_lowerCAmelCase     ,   cache_dir=_lowerCAmelCase						).read()
							_check_sql_dataset(_lowerCAmelCase     ,   _lowerCAmelCase						)


def 				snake_case_  (      _lowerCAmelCase						:		List[Any]						)   ->     Dict:
							with contextlib.closing(sqlitea.connect(_lowerCAmelCase						)						) as con:
														UpperCAmelCase				: Any											=	con.cursor()
														cur.execute('''SELECT * FROM dataset'''						)
														for row in cur:
																					yield row


@require_sqlalchemy
def 				snake_case_  (      _lowerCAmelCase						:		Union[str, Any]     ,   _lowerCAmelCase						:		str     ,   _lowerCAmelCase						:		Any						)   ->     Optional[int]:
							UpperCAmelCase				: Optional[int]											=	tmp_path / '''cache'''
							UpperCAmelCase				: int											=	os.path.join(_lowerCAmelCase     ,   '''tmp.sql'''						)
							UpperCAmelCase				: Any											=	SqlDatasetReader('''dataset'''     ,   '''sqlite:///''' + sqlite_path     ,   cache_dir=_lowerCAmelCase						).read()
							SqlDatasetWriter(_lowerCAmelCase     ,   '''dataset'''     ,   '''sqlite:///''' + output_sqlite_path     ,   num_proc=1						).write()

							UpperCAmelCase				: List[Any]											=	iter_sql_file(_lowerCAmelCase						)
							UpperCAmelCase				: List[Any]											=	iter_sql_file(_lowerCAmelCase						)

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


@require_sqlalchemy
def 				snake_case_  (      _lowerCAmelCase						:		int     ,   _lowerCAmelCase						:		Optional[Any]     ,   _lowerCAmelCase						:		Union[str, Any]						)   ->     int:
							UpperCAmelCase				: Optional[Any]											=	tmp_path / '''cache'''
							UpperCAmelCase				: Any											=	os.path.join(_lowerCAmelCase     ,   '''tmp.sql'''						)
							UpperCAmelCase				: List[str]											=	SqlDatasetReader('''dataset'''     ,   '''sqlite:///''' + sqlite_path     ,   cache_dir=_lowerCAmelCase						).read()
							SqlDatasetWriter(_lowerCAmelCase     ,   '''dataset'''     ,   '''sqlite:///''' + output_sqlite_path     ,   num_proc=2						).write()

							UpperCAmelCase				: List[str]											=	iter_sql_file(_lowerCAmelCase						)
							UpperCAmelCase				: Any											=	iter_sql_file(_lowerCAmelCase						)

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




@require_sqlalchemy
def 				snake_case_  (      _lowerCAmelCase						:		Union[str, Any]     ,   _lowerCAmelCase						:		Any     ,   _lowerCAmelCase						:		Optional[Any]						)   ->     List[Any]:
							UpperCAmelCase				: Union[str, Any]											=	tmp_path / '''cache'''
							UpperCAmelCase				: Tuple											=	os.path.join(_lowerCAmelCase     ,   '''tmp.sql'''						)
							UpperCAmelCase				: Optional[int]											=	SqlDatasetReader('''dataset'''     ,   '''sqlite:///''' + sqlite_path     ,   cache_dir=_lowerCAmelCase						).read()
							with pytest.raises(_lowerCAmelCase						):
														SqlDatasetWriter(_lowerCAmelCase     ,   '''dataset'''     ,   '''sqlite:///''' + output_sqlite_path     ,   num_proc=0						).write()





