






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


SCREAMING_SNAKE_CASE__				     =			logging.get_logger(__name__)
SCREAMING_SNAKE_CASE__				     =			{"vocab_file": "spiece.model"}

SCREAMING_SNAKE_CASE__				     =			{
    "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",
    }
}

SCREAMING_SNAKE_CASE__				     =			{
    "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 			_UpperCamelCase(       __UpperCamelCase		):
		__SCREAMING_SNAKE_CASE  :       int							=							VOCAB_FILES_NAMES
		__SCREAMING_SNAKE_CASE  :       Optional[Any]							=							PRETRAINED_VOCAB_FILES_MAP
		__SCREAMING_SNAKE_CASE  :       Union[str, Any]							=							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
		__SCREAMING_SNAKE_CASE  :       List[str]							=							["input_ids", "attention_mask"]

		def __init__(					self		:    List[str]    ,       SCREAMING_SNAKE_CASE__		:    int    ,       SCREAMING_SNAKE_CASE__		:    str=False    ,       SCREAMING_SNAKE_CASE__		:    Optional[int]=False    ,       SCREAMING_SNAKE_CASE__		:    int=False    ,       SCREAMING_SNAKE_CASE__		:    str=None    ,       SCREAMING_SNAKE_CASE__		:    str=None    ,       SCREAMING_SNAKE_CASE__		:    Any=None    ,       SCREAMING_SNAKE_CASE__		:    Optional[Any]=None    ,       SCREAMING_SNAKE_CASE__		:    Optional[Dict[str, Any]] = None    ,       **SCREAMING_SNAKE_CASE__		:    List[Any]    ,       ):



					'''simple docstring'''



					__a						:  int									=     {} if sp_model_kwargs is None else sp_model_kwargs

					__a						:  int									=     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'    )
								__a						:  Tuple									=     'None'

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

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

					__a						:  List[str]									=     do_lower_case
					__a						:  Tuple									=     remove_space
					__a						:  str									=     keep_accents
					__a						:  Dict									=     vocab_file

					__a						:  int									=     spm.SentencePieceProcessor(**self.sp_model_kwargs    )
					self.sp_model.Load(snake_case__    )

					# Used for whitespace normalization in input texts
					# fmt : off
					__a						:  Optional[Any]									=     {' ', ' ', ' ', ' ', ' ', '　', ' ', ' ', ' ', ' ', '￼', ''}
					# fmt : on

					# Regular expression to remove non-printing characters (e.g. some unicode control chars) in preprocessing
					__a						:  Optional[int]									=     re.compile(
					    f'''[{"".join(map(snake_case__    ,       list(range(0    ,       9    )    ) + list(range(1_1    ,       3_2    )    ) + list(range(1_2_7    ,       1_6_0    )    ) + [1_6_0, 1_7_3, 8_2_0_3]    )    )}]'''    )

		def __getstate__(					self		:    Any    ):



					'''simple docstring'''



					__a						:  List[str]									=     self.__dict__.copy()
					__a						:  List[Any]									=     None
					return state

		def __setstate__(					self		:    Union[str, Any]    ,       SCREAMING_SNAKE_CASE__		:    Tuple    ):



					'''simple docstring'''



					__a						:  Optional[int]									=     d

					# for backward compatibility
					if not hasattr(self    ,       'sp_model_kwargs'    ):
								__a						:  Any									=     {}

					__a						:  Optional[int]									=     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    __lowerCAmelCase  (					self		:    int    ):



					'''simple docstring'''



					return len(self.sp_model    )

		def    __lowerCAmelCase  (					self		:    int    ,       SCREAMING_SNAKE_CASE__		:    str    ):



					'''simple docstring'''



					__a						:  Optional[Any]									=     self.non_printing_characters_re.sub(''    ,       snake_case__    )

					# Normalize whitespaces
					__a						:  int									=     ''.join([char if char not in self.whitespaces else ' ' for char in text]    )

					# NFC Unicode normalization
					__a						:  Optional[Any]									=     unicodedata.normalize('NFC'    ,       snake_case__    )
					return text

		def    __lowerCAmelCase  (					self		:    Any    ,       SCREAMING_SNAKE_CASE__		:    str    ,       **SCREAMING_SNAKE_CASE__		:    str    ):



					'''simple docstring'''



					__a						:  List[str]									=     self.preprocess_text(snake_case__    )
					return self.sp_model.encode(snake_case__    ,       out_type=snake_case__    )

		def    __lowerCAmelCase  (					self		:    List[str]    ,       SCREAMING_SNAKE_CASE__		:    str    ):



					'''simple docstring'''



					return self.sp_model.PieceToId(snake_case__    )

		def    __lowerCAmelCase  (					self		:    int    ,       SCREAMING_SNAKE_CASE__		:    int    ):



					'''simple docstring'''



					return self.sp_model.IdToPiece(snake_case__    )

		@staticmethod
		def    __lowerCAmelCase  (					SCREAMING_SNAKE_CASE__		:    str    ):



					'''simple docstring'''



					return out_string

		def    __lowerCAmelCase  (					self		:    int    ,       SCREAMING_SNAKE_CASE__		:    List[str]    ):



					'''simple docstring'''



					__a						:  Any									=     []
					__a						:  Optional[Any]									=     ''
					__a						:  str									=     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(snake_case__    ) + token
											__a						:  Any									=     True
											__a						:  str									=     []
								else:
											current_sub_tokens.append(snake_case__    )
											__a						:  List[Any]									=     False
					out_string += self.sp_model.decode(snake_case__    )

					return out_string

		def    __lowerCAmelCase  (					self		:    Tuple    ):



					'''simple docstring'''



					__a						:  Optional[Any]									=     {self.convert_ids_to_tokens(snake_case__    ): i for i in range(self.vocab_size    )}
					vocab.update(self.added_tokens_encoder    )
					return vocab

		def    __lowerCAmelCase  (					self		:    Optional[Any]    ,       SCREAMING_SNAKE_CASE__		:    str    ,       SCREAMING_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
					__a						:  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:
											__a						:  Optional[int]									=     self.sp_model.serialized_model_proto()
											fi.write(snake_case__    )

					return (out_vocab_file,)

		def    __lowerCAmelCase  (					self		:    int    ,       SCREAMING_SNAKE_CASE__		:    Union[str, List[str]]    ,       SCREAMING_SNAKE_CASE__		:    Union[str, bool] = False    ):



					'''simple docstring'''



					if isinstance(snake_case__    ,       snake_case__    ):
								__a						:  Dict									=     self.preprocess_text(snake_case__    )
								__a						:  int									=     self.sp_model.encode(snake_case__    )
					else:
								__a						:  Tuple									=     [self.preprocess_text(snake_case__    ) for t in text]
								__a						:  Optional[int]									=     self.sp_model.encode(snake_case__    )

					if return_tensors is True or return_tensors == "pt":
								__a						:  Tuple									=     torch.tensor(snake_case__    )

					return token_ids

		def    __lowerCAmelCase  (					self		:    str    ,       SCREAMING_SNAKE_CASE__		:    Union[int, List[int]]    ):



					'''simple docstring'''



					return self.sp_model.decode(snake_case__    )



		def    __lowerCAmelCase  (					self		:    str    ,       SCREAMING_SNAKE_CASE__		:    "Conversation"    ):



					'''simple docstring'''



					__a						:  Tuple									=     [f'''User: {text}''' if is_user else f'''Bot: {text}''' for is_user, text in conversation.iter_texts()]
					__a						:  Union[str, Any]									=     (
					    f'''{self.eos_token}{self.bos_token}''' + f'''{self.bos_token}'''.join(snake_case__    ) + f'''{self.bos_token}Bot:'''
					)
					return self.encode(text=snake_case__    )






"""simple docstring"""




import logging
import os
import threading
import time


try:
			import warnings
except ImportError:
			__lowerCAmelCase     :				Optional[int]    				=    None

try:
			import msvcrt
except ImportError:
			__lowerCAmelCase     :				List[Any]    				=    None

try:
			import fcntl
except ImportError:
			__lowerCAmelCase     :				Optional[int]    				=    None


# Backward compatibility
# ------------------------------------------------
try:
			TimeoutError
except NameError:
			__lowerCAmelCase     :				int    				=    OSError


# Data
# ------------------------------------------------
__lowerCAmelCase     :				Dict    				=    [
    "Timeout",
    "BaseFileLock",
    "WindowsFileLock",
    "UnixFileLock",
    "SoftFileLock",
    "FileLock",
]

__lowerCAmelCase     :				str    				=    "3.0.12"


__lowerCAmelCase     :				Optional[int]    				=    None







def 				_UpperCAmelCase		(							):





		"""simple docstring"""




		global _logger
		lowerCAmelCase__   = _logger or logging.getLogger(__name__  )
		return _logger

class    a_   (  __UpperCamelCase      ):




					def __init__(						self    :  str	,    snake_case__    :  Optional[int]  ):
							lowerCAmelCase__   = lock_file
							return None




					def __str__(						self    :  Any  ):
							lowerCAmelCase__   = F"""The file lock '{self.lock_file}' could not be acquired."""
							return temp

class    a_   :




					def __init__(						self    :  Tuple	,    snake_case__    :  Optional[int]  ):
							lowerCAmelCase__   = lock
							return None




					def __enter__(						self    :  Union[str, Any]  ):
							return self.lock




					def __exit__(						self    :  List[Any]	,    snake_case__    :  Dict	,    snake_case__    :  str	,    snake_case__    :  List[Any]  ):
							self.lock.release()
							return None

class    a_   :




					def __init__(						self    :  Dict	,    snake_case__    :  Tuple	,    snake_case__    :  Dict=-1	,    snake_case__    :  Dict=None  ):
							lowerCAmelCase__   = max_filename_length if max_filename_length is not None else 255
							# Hash the filename if it's too long
							lowerCAmelCase__   = self.hash_filename_if_too_long(snake_case__	,    snake_case__  )
							# The path to the lock file.
							lowerCAmelCase__   = lock_file

							# The file descriptor for the *_lock_file* as it is returned by the
							# os.open() function.
							# This file lock is only NOT None, if the object currently holds the
							# lock.
							lowerCAmelCase__   = None

							# The default timeout value.
							lowerCAmelCase__   = timeout

							# We use this lock primarily for the lock counter.
							lowerCAmelCase__   = threading.Lock()

							# The lock counter is used for implementing the nested locking
							# mechanism. Whenever the lock is acquired, the counter is increased and
							# the lock is only released, when this value is 0 again.
							lowerCAmelCase__   = 0
							return None




					@property
					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple  ):
							return self._lock_file




					@property
					def     _SCREAMING_SNAKE_CASE    (						self    :  Union[str, Any]  ):
							return self._timeout




					@timeout.setter
					def     _SCREAMING_SNAKE_CASE    (						self    :  str	,    snake_case__    :  str  ):
							lowerCAmelCase__   = float(snake_case__  )
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[str]  ):
							raise NotImplementedError()




					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							raise NotImplementedError()




					@property
					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							return self._lock_file_fd is not None




					def     _SCREAMING_SNAKE_CASE    (						self    :  List[Any]	,    snake_case__    :  Union[str, Any]=None	,    snake_case__    :  Optional[int]=0.05  ):
							# Use the default timeout, if no timeout is provided.
							if timeout is None:
									lowerCAmelCase__   = self.timeout

							# Increment the number right at the beginning.
							# We can still undo it, if something fails.
							with self._thread_lock:
									self._lock_counter += 1

							lowerCAmelCase__   = id(self  )
							lowerCAmelCase__   = self._lock_file
							lowerCAmelCase__   = time.time()
							try:
									while True:
											with self._thread_lock:
													if not self.is_locked:
															logger().debug(F"""Attempting to acquire lock {lock_id} on {lock_filename}"""  )
															self._acquire()

											if self.is_locked:
													logger().debug(F"""Lock {lock_id} acquired on {lock_filename}"""  )
													break
											elif timeout >= 0 and time.time() - start_time > timeout:
													logger().debug(F"""Timeout on acquiring lock {lock_id} on {lock_filename}"""  )
													raise Timeout(self._lock_file  )
											else:
													logger().debug(
													    F"""Lock {lock_id} not acquired on {lock_filename}, waiting {poll_intervall} seconds ..."""  )
													time.sleep(snake_case__  )
							except:  # noqa
									# Something did go wrong, so decrement the counter.
									with self._thread_lock:
											lowerCAmelCase__   = max(0	,    self._lock_counter - 1  )

									raise
							return _Acquire_ReturnProxy(lock=self  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]	,    snake_case__    :  Tuple=False  ):
							with self._thread_lock:
									if self.is_locked:
											self._lock_counter -= 1

											if self._lock_counter == 0 or force:
													lowerCAmelCase__   = id(self  )
													lowerCAmelCase__   = self._lock_file

													logger().debug(F"""Attempting to release lock {lock_id} on {lock_filename}"""  )
													self._release()
													lowerCAmelCase__   = 0
													logger().debug(F"""Lock {lock_id} released on {lock_filename}"""  )

							return None




					def __enter__(						self    :  Optional[Any]  ):
							self.acquire()
							return self




					def __exit__(						self    :  Any	,    snake_case__    :  Optional[Any]	,    snake_case__    :  Optional[Any]	,    snake_case__    :  Optional[Any]  ):
							self.release()
							return None




					def __del__(						self    :  Optional[int]  ):
							self.release(force=snake_case__  )
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]	,    snake_case__    :  str	,    snake_case__    :  int  ):
							lowerCAmelCase__   = os.path.basename(snake_case__  )
							if len(snake_case__  ) > max_length and max_length > 0:
									lowerCAmelCase__   = os.path.dirname(snake_case__  )
									lowerCAmelCase__   = str(hash(snake_case__  )  )
									lowerCAmelCase__   = filename[: max_length - len(snake_case__  ) - 8] + """...""" + hashed_filename + """.lock"""
									return os.path.join(snake_case__	,    snake_case__  )
							else:
									return path

class    a_   (  __UpperCamelCase      ):




					def __init__(						self    :  List[Any]	,    snake_case__    :  int	,    snake_case__    :  str=-1	,    snake_case__    :  Tuple=None  ):
							from .file_utils import relative_to_absolute_path

							super().__init__(snake_case__	,    timeout=snake_case__	,    max_filename_length=snake_case__  )
							lowerCAmelCase__   = """\\\\?\\""" + relative_to_absolute_path(self.lock_file  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple  ):
							lowerCAmelCase__   = os.O_RDWR | os.O_CREAT | os.O_TRUNC

							try:
									lowerCAmelCase__   = os.open(self._lock_file	,    snake_case__  )
							except OSError:
									pass
							else:
									try:
											msvcrt.locking(snake_case__	,    msvcrt.LK_NBLCK	,    1  )
									except OSError:
											os.close(snake_case__  )
									else:
											lowerCAmelCase__   = fd
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]  ):
							lowerCAmelCase__   = self._lock_file_fd
							lowerCAmelCase__   = None
							msvcrt.locking(snake_case__	,    msvcrt.LK_UNLCK	,    1  )
							os.close(snake_case__  )

							try:
									os.remove(self._lock_file  )
							# Probably another instance of the application
							# that acquired the file lock.
							except OSError:
									pass
							return None

class    a_   (  __UpperCamelCase      ):




					def __init__(						self    :  Union[str, Any]	,    snake_case__    :  List[str]	,    snake_case__    :  str=-1	,    snake_case__    :  List[str]=None  ):
							lowerCAmelCase__   = os.statvfs(os.path.dirname(snake_case__  )  ).f_namemax
							super().__init__(snake_case__	,    timeout=snake_case__	,    max_filename_length=snake_case__  )




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[Any]  ):
							lowerCAmelCase__   = os.O_RDWR | os.O_CREAT | os.O_TRUNC
							lowerCAmelCase__   = os.open(self._lock_file	,    snake_case__  )

							try:
									fcntl.flock(snake_case__	,    fcntl.LOCK_EX | fcntl.LOCK_NB  )
							except OSError:
									os.close(snake_case__  )
							else:
									lowerCAmelCase__   = fd
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  str  ):
							# Do not remove the lockfile:
							#
							#   https://github.com/benediktschmitt/py-filelock/issues/31
							#   https://stackoverflow.com/questions/17708885/flock-removing-locked-file-without-race-condition
							lowerCAmelCase__   = self._lock_file_fd
							lowerCAmelCase__   = None
							fcntl.flock(snake_case__	,    fcntl.LOCK_UN  )
							os.close(snake_case__  )
							return None

class    a_   (  __UpperCamelCase      ):




					def     _SCREAMING_SNAKE_CASE    (						self    :  Tuple  ):
							lowerCAmelCase__   = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
							try:
									lowerCAmelCase__   = os.open(self._lock_file	,    snake_case__  )
							except OSError:
									pass
							else:
									lowerCAmelCase__   = fd
							return None




					def     _SCREAMING_SNAKE_CASE    (						self    :  Optional[int]  ):
							os.close(self._lock_file_fd  )
							lowerCAmelCase__   = None

							try:
									os.remove(self._lock_file  )
							# The file is already deleted and that's what we want.
							except OSError:
									pass
							return None
__lowerCAmelCase     :				Optional[int]    				=    None

if msvcrt:
			__lowerCAmelCase     :				Tuple    				=    WindowsFileLock
elif fcntl:
			__lowerCAmelCase     :				Optional[int]    				=    UnixFileLock
else:
			__lowerCAmelCase     :				Tuple    				=    SoftFileLock

			if warnings is not None:
						warnings.warn("only soft file lock is available")




