



"""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


__snake_case							:   str      =    logging.get_logger(__name__)
__snake_case							:   Union[str, Any]      =    {'vocab_file': 'spiece.model'}

__snake_case							:   int      =    {
    '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',
    }
}

__snake_case							:   Optional[Any]      =    {
    'AI-Sweden/gpt-sw3-126m': 2_048,
    'AI-Sweden/gpt-sw3-350m': 2_048,
    'AI-Sweden/gpt-sw3-1.6b': 2_048,
    'AI-Sweden/gpt-sw3-6.7b': 2_048,
    'AI-Sweden/gpt-sw3-20b': 2_048,
}



class 			A__   (  __SCREAMING_SNAKE_CASE							):




	'''simple docstring'''





	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', 'attention_mask']

	def __init__(     self:		int       ,			_SCREAMING_SNAKE_CASE:		int       ,			_SCREAMING_SNAKE_CASE:		Union[str, Any]=False       ,			_SCREAMING_SNAKE_CASE:		Tuple=False       ,			_SCREAMING_SNAKE_CASE:		Any=False       ,			_SCREAMING_SNAKE_CASE:		List[str]=None       ,			_SCREAMING_SNAKE_CASE:		List[Any]=None       ,			_SCREAMING_SNAKE_CASE:		Dict=None       ,			_SCREAMING_SNAKE_CASE:		Dict=None       ,			_SCREAMING_SNAKE_CASE:		Optional[Dict[str, Any]] = None       ,			**_SCREAMING_SNAKE_CASE:		Any       ,			)				->       None:

				"""simple docstring"""

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

				__lowerCAmelCase      :  Dict												=       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")
							__lowerCAmelCase      :  List[str]												=       "None"

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

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

				__lowerCAmelCase      :  Any												=       do_lower_case
				__lowerCAmelCase      :  str												=       remove_space
				__lowerCAmelCase      :  Tuple												=       keep_accents
				__lowerCAmelCase      :  List[Any]												=       vocab_file

				__lowerCAmelCase      :  Dict												=       spm.SentencePieceProcessor(**self.sp_model_kwargs)
				self.sp_model.Load(_SCREAMING_SNAKE_CASE)

				# Used for whitespace normalization in input texts
				# fmt : off
				__lowerCAmelCase      :  Any												=       {" ", " ", " ", " ", " ", "　", " ", " ", " ", " ", "￼", ""}
				# fmt : on

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

	def __getstate__(     self:		Any)				->       Union[str, Any]:

				"""simple docstring"""

				__lowerCAmelCase      :  Optional[int]												=       self.__dict__.copy()
				__lowerCAmelCase      :  Tuple												=       None
				return state

	def __setstate__(     self:		Optional[Any]       ,			_SCREAMING_SNAKE_CASE:		Union[str, Any])				->       Any:

				"""simple docstring"""

				__lowerCAmelCase      :  Optional[Any]												=       d

				# for backward compatibility
				if not hasattr(self       ,			"sp_model_kwargs"):
							__lowerCAmelCase      :  Union[str, Any]												=       {}

				__lowerCAmelCase      :  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 							_SCREAMING_SNAKE_CASE   (     self:		int)				->       int:

				"""simple docstring"""

				return len(self.sp_model)

	def 							_SCREAMING_SNAKE_CASE   (     self:		Any       ,			_SCREAMING_SNAKE_CASE:		str)				->       str:

				"""simple docstring"""

				__lowerCAmelCase      :  Optional[int]												=       self.non_printing_characters_re.sub(""       ,			_SCREAMING_SNAKE_CASE)

				# Normalize whitespaces
				__lowerCAmelCase      :  List[Any]												=       "".join([char if char not in self.whitespaces else " " for char in text])

				# NFC Unicode normalization
				__lowerCAmelCase      :  Dict												=       unicodedata.normalize("NFC"       ,			_SCREAMING_SNAKE_CASE)
				return text

	def 							_SCREAMING_SNAKE_CASE   (     self:		str       ,			_SCREAMING_SNAKE_CASE:		str       ,			**_SCREAMING_SNAKE_CASE:		Any)				->       List[str]:

				"""simple docstring"""

				__lowerCAmelCase      :  Dict												=       self.preprocess_text(_SCREAMING_SNAKE_CASE)
				return self.sp_model.encode(_SCREAMING_SNAKE_CASE       ,			out_type=_SCREAMING_SNAKE_CASE)

	def 							_SCREAMING_SNAKE_CASE   (     self:		str       ,			_SCREAMING_SNAKE_CASE:		str)				->       int:

				"""simple docstring"""

				return self.sp_model.PieceToId(_SCREAMING_SNAKE_CASE)

	def 							_SCREAMING_SNAKE_CASE   (     self:		Union[str, Any]       ,			_SCREAMING_SNAKE_CASE:		int)				->       str:

				"""simple docstring"""

				return self.sp_model.IdToPiece(_SCREAMING_SNAKE_CASE)

	@staticmethod
	def 							_SCREAMING_SNAKE_CASE   (     _SCREAMING_SNAKE_CASE:		str)				->       str:

				"""simple docstring"""

				return out_string

	def 							_SCREAMING_SNAKE_CASE   (     self:		List[Any]       ,			_SCREAMING_SNAKE_CASE:		List[str])				->       str:

				"""simple docstring"""

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

				return out_string

	def 							_SCREAMING_SNAKE_CASE   (     self:		Union[str, Any])				->       Dict[str, int]:

				"""simple docstring"""

				__lowerCAmelCase      :  List[str]												=       {self.convert_ids_to_tokens(_SCREAMING_SNAKE_CASE): i for i in range(self.vocab_size)}
				vocab.update(self.added_tokens_encoder)
				return vocab

	def 							_SCREAMING_SNAKE_CASE   (     self:		int       ,			_SCREAMING_SNAKE_CASE:		str       ,			_SCREAMING_SNAKE_CASE:		Optional[str] = None)				->       Tuple[str]:

				"""simple docstring"""

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

				if os.path.abspath(self.vocab_file) != os.path.abspath(_SCREAMING_SNAKE_CASE) and os.path.isfile(self.vocab_file):
							copyfile(self.vocab_file       ,			_SCREAMING_SNAKE_CASE)
				elif not os.path.isfile(self.vocab_file):
							with open(_SCREAMING_SNAKE_CASE       ,			"wb") as fi:
										__lowerCAmelCase      :  Tuple												=       self.sp_model.serialized_model_proto()
										fi.write(_SCREAMING_SNAKE_CASE)

				return (out_vocab_file,)

	def 							_SCREAMING_SNAKE_CASE   (     self:		Union[str, Any]       ,			_SCREAMING_SNAKE_CASE:		Union[str, List[str]]       ,			_SCREAMING_SNAKE_CASE:		Union[str, bool] = False)				->       Union[List[int], List[List[int]], "torch.Tensor"]:

				"""simple docstring"""

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

				if return_tensors is True or return_tensors == "pt":
							__lowerCAmelCase      :  Optional[int]												=       torch.tensor(_SCREAMING_SNAKE_CASE)

				return token_ids

	def 							_SCREAMING_SNAKE_CASE   (     self:		Any       ,			_SCREAMING_SNAKE_CASE:		Union[int, List[int]])				->       str:

				"""simple docstring"""

				return self.sp_model.decode(_SCREAMING_SNAKE_CASE)







	def 							_SCREAMING_SNAKE_CASE   (     self:		Any       ,			_SCREAMING_SNAKE_CASE:		"Conversation")				->       List[int]:

				"""simple docstring"""

				__lowerCAmelCase      :  Optional[Any]												=       [F"""User: {text}""" if is_user else F"""Bot: {text}""" for is_user, text in conversation.iter_texts()]
				__lowerCAmelCase      :  int												=       (
				    F"""{self.eos_token}{self.bos_token}""" + F"""{self.bos_token}""".join(_SCREAMING_SNAKE_CASE) + F"""{self.bos_token}Bot:"""
				)
				return self.encode(text=_SCREAMING_SNAKE_CASE)




"""simple docstring"""






import gc
import threading
import time

import psutil
import torch



class 			A__   :




	'''simple docstring'''





	def __init__(     self:		str)				->       Optional[Any]:

				"""simple docstring"""

				__lowerCAmelCase      :  Optional[Any]												=       psutil.Process()
				__lowerCAmelCase      :  str												=       False

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

				"""simple docstring"""

				__lowerCAmelCase      :  Optional[Any]												=       -1

				while True:
							__lowerCAmelCase      :  str												=       max(self.process.memory_info().rss       ,			self.cpu_memory_peak)

							# can't sleep or will not catch the peak right (this comment is here on purpose)
							if not self.peak_monitoring:
										break

	def 							_SCREAMING_SNAKE_CASE   (     self:		Tuple)				->       int:

				"""simple docstring"""

				__lowerCAmelCase      :  List[str]												=       True
				__lowerCAmelCase      :  str												=       threading.Thread(target=self.peak_monitor)
				__lowerCAmelCase      :  Tuple												=       True
				self.thread.start()







	def 							_SCREAMING_SNAKE_CASE   (     self:		Optional[Any])				->       List[str]:

				"""simple docstring"""

				__lowerCAmelCase      :  Tuple												=       False
				self.thread.join()
				return self.cpu_memory_peak






__snake_case							:   Tuple      =    PeakCPUMemory()


def 			_lowercase				(  )						->   str:
			# Time
			__lowerCAmelCase      :  str												=       {"time": time.time()}

			gc.collect()
			torch.cuda.empty_cache()

			# CPU mem
			__lowerCAmelCase      :  Optional[Any]												=       psutil.Process().memory_info().rss
			cpu_peak_tracker.start()

			# GPU mem
			for i in range(torch.cuda.device_count()	):
						__lowerCAmelCase      :  Union[str, Any]												=       torch.cuda.memory_allocated(__snake_case	)
			torch.cuda.reset_peak_memory_stats()

			return measures


def 			_lowercase				(  __snake_case	)						->   Optional[Any]:
			# Time
			__lowerCAmelCase      :  str												=       {"time": time.time() - start_measures["time"]}

			gc.collect()
			torch.cuda.empty_cache()

			# CPU mem
			__lowerCAmelCase      :  str												=       (psutil.Process().memory_info().rss - start_measures["cpu"]) / 2**20
			__lowerCAmelCase      :  List[str]												=       (cpu_peak_tracker.stop() - start_measures["cpu"]) / 2**20

			# GPU mem
			for i in range(torch.cuda.device_count()	):
						__lowerCAmelCase      :  Union[str, Any]												=       (torch.cuda.memory_allocated(__snake_case	) - start_measures[str(__snake_case	)]) / 2**20
						__lowerCAmelCase      :  Any												=       (torch.cuda.max_memory_allocated(__snake_case	) - start_measures[str(__snake_case	)]) / 2**20

			return measures





def 			_lowercase				(  __snake_case					,__snake_case	)						->   Dict:
			print(F"""{description}:"""	)
			print(F"""- Time: {measures['time']:.2f}s"""	)
			for i in range(torch.cuda.device_count()	):
						print(F"""- GPU {i} allocated: {measures[str(__snake_case	)]:.2f}MiB"""	)
						__lowerCAmelCase      :  Optional[Any]												=       measures[F"""{i}-peak"""]
						print(F"""- GPU {i} peak: {peak:.2f}MiB"""	)
			print(F"""- CPU RAM allocated: {measures['cpu']:.2f}MiB"""	)
			print(F"""- CPU RAM peak: {measures['cpu-peak']:.2f}MiB"""	)