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


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

_lowercase							:      List[Any]      							={
    '''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''',
    }
}

_lowercase							:      Any      							={
    '''AI-Sweden/gpt-sw3-126m''': 2_0_4_8,
    '''AI-Sweden/gpt-sw3-350m''': 2_0_4_8,
    '''AI-Sweden/gpt-sw3-1.6b''': 2_0_4_8,
    '''AI-Sweden/gpt-sw3-6.7b''': 2_0_4_8,
    '''AI-Sweden/gpt-sw3-20b''': 2_0_4_8,
}



class  SCREAMING_SNAKE_CASE_ (					__a			):
  '''simple docstring'''


  lowercase		:     List[str]            =				VOCAB_FILES_NAMES
  lowercase		:     int            =				PRETRAINED_VOCAB_FILES_MAP
  lowercase		:     Union[str, Any]            =				PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  lowercase		:     Optional[Any]            =				["input_ids", "attention_mask"]







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

         A						:     Tuple         =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						:     str         ="""None"""

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

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

         A						:     Dict         =do_lower_case
         A						:     Any         =remove_space
         A						:     Optional[int]         =keep_accents
         A						:     Optional[int]         =vocab_file

         A						:     Dict         =spm.SentencePieceProcessor(**self.sp_model_kwargs     )
         self.sp_model.Load(a_     )

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

         # Regular expression to remove non-printing characters (e.g. some unicode control chars) in preprocessing
         A						:     Union[str, Any]         =re.compile(
             f'[{"".join(map(a_      ,	list(range(0      ,	9     )     ) + list(range(11      ,	32     )     ) + list(range(1_27      ,	1_60     )     ) + [1_60, 1_73, 82_03]     )     )}]'     )







  def __getstate__(				self    :					Any     )    ->				Any:
         A						:     Tuple         =self.__dict__.copy()
         A						:     int         =None
         return state







  def __setstate__(				self    :					Any      ,	SCREAMING_SNAKE_CASE__    :					Optional[int]     )    ->				Tuple:
         A						:     Optional[Any]         =d

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

         A						:     Union[str, Any]         =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    :					List[str]     )    ->				List[Any]:
         return len(self.sp_model     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Dict      ,	SCREAMING_SNAKE_CASE__    :					str     )    ->				Tuple:
         A						:     Dict         =self.non_printing_characters_re.sub(''      ,	a_     )

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

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







  def 		SCREAMING_SNAKE_CASE_		(				self    :					str      ,	SCREAMING_SNAKE_CASE__    :					str      ,	**SCREAMING_SNAKE_CASE__    :					Union[str, Any]     )    ->				Optional[Any]:
         A						:     Any         =self.preprocess_text(a_     )
         return self.sp_model.encode(a_      ,	out_type=a_     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					List[Any]      ,	SCREAMING_SNAKE_CASE__    :					str     )    ->				Dict:
         return self.sp_model.PieceToId(a_     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Tuple      ,	SCREAMING_SNAKE_CASE__    :					int     )    ->				Dict:
         return self.sp_model.IdToPiece(a_     )







  @staticmethod
  def 		SCREAMING_SNAKE_CASE_		(				SCREAMING_SNAKE_CASE__    :					str     )    ->				Union[str, Any]:
         return out_string







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Tuple      ,	SCREAMING_SNAKE_CASE__    :					List[str]     )    ->				Any:
         A						:     List[str]         =[]
         A						:     Optional[Any]         =""""""
         A						:     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(a_     ) + token
                       A						:     int         =True
                       A						:     int         =[]
                else:
                       current_sub_tokens.append(a_     )
                       A						:     Union[str, Any]         =False
         out_string += self.sp_model.decode(a_     )

         return out_string







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Union[str, Any]     )    ->				str:
         A						:     str         ={self.convert_ids_to_tokens(a_     ): i for i in range(self.vocab_size     )}
         vocab.update(self.added_tokens_encoder     )
         return vocab







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Optional[Any]      ,	SCREAMING_SNAKE_CASE__    :					str      ,	SCREAMING_SNAKE_CASE__    :					Optional[str] = None     )    ->				Optional[Any]:



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

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

         return (out_vocab_file,)







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Any      ,	SCREAMING_SNAKE_CASE__    :					Union[str, List[str]]      ,	SCREAMING_SNAKE_CASE__    :					Union[str, bool] = False     )    ->				Tuple:



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

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

         return token_ids







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Optional[int]      ,	SCREAMING_SNAKE_CASE__    :					Union[int, List[int]]     )    ->				Dict:
         return self.sp_model.decode(a_     )







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



from typing import Optional

import numpy as np
import torch
from torch import nn
from transformers import GPTaConfig, GPTaLMHeadModel
from transformers.modeling_utils import ModuleUtilsMixin

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin



class  SCREAMING_SNAKE_CASE_ (					lowerCAmelCase_   ,						lowerCAmelCase_   ,						lowerCAmelCase_			):
  '''simple docstring'''


  lowercase		:     Tuple            =				[r"h\.\d+\.attn\.bias", r"h\.\d+\.attn\.masked_bias"]







  @register_to_config
  def __init__(				self    :					Any      ,	SCREAMING_SNAKE_CASE__    :					int      ,	SCREAMING_SNAKE_CASE__    :					int      ,	SCREAMING_SNAKE_CASE__    :					Optional[int] = None      ,	SCREAMING_SNAKE_CASE__    :					int = 5_02_57      ,	SCREAMING_SNAKE_CASE__    :					int = 10_24      ,	SCREAMING_SNAKE_CASE__    :					int = 7_68      ,	SCREAMING_SNAKE_CASE__    :					int = 12      ,	SCREAMING_SNAKE_CASE__    :					int = 12      ,	SCREAMING_SNAKE_CASE__    :					Optional[int] = None      ,	SCREAMING_SNAKE_CASE__    :					str = "gelu_new"      ,	SCREAMING_SNAKE_CASE__    :					float = 0.1      ,	SCREAMING_SNAKE_CASE__    :					float = 0.1      ,	SCREAMING_SNAKE_CASE__    :					float = 0.1      ,	SCREAMING_SNAKE_CASE__    :					float = 1e-5      ,	SCREAMING_SNAKE_CASE__    :					float = 0.0_2      ,	SCREAMING_SNAKE_CASE__    :					bool = True      ,	SCREAMING_SNAKE_CASE__    :					bool = True      ,	SCREAMING_SNAKE_CASE__    :					bool = False      ,	SCREAMING_SNAKE_CASE__    :					bool = False      ,	)    ->				List[str]:
         super().__init__()

         A						:     str         =prefix_length

         if prefix_inner_dim != n_embd and prefix_hidden_dim is None:
                raise ValueError(
                    f'`prefix_hidden_dim` cannot be `None` when `prefix_inner_dim`: {prefix_hidden_dim} and'
                    f' `n_embd`: {n_embd} are not equal.'     )

         A						:     List[Any]         =prefix_inner_dim
         A						:     Dict         =prefix_hidden_dim

         A						:     List[str]         =(
             nn.Linear(self.prefix_inner_dim      ,	self.prefix_hidden_dim     )
             if self.prefix_hidden_dim is not None
             else nn.Identity()
         )
         A						:     Optional[int]         =(
             nn.Linear(self.prefix_hidden_dim      ,	SCREAMING_SNAKE_CASE__     ) if self.prefix_hidden_dim is not None else nn.Identity()
         )

         A						:     Dict         =GPTaConfig(
             vocab_size=SCREAMING_SNAKE_CASE__      ,	n_positions=SCREAMING_SNAKE_CASE__      ,	n_embd=SCREAMING_SNAKE_CASE__      ,	n_layer=SCREAMING_SNAKE_CASE__      ,	n_head=SCREAMING_SNAKE_CASE__      ,	n_inner=SCREAMING_SNAKE_CASE__      ,	activation_function=SCREAMING_SNAKE_CASE__      ,	resid_pdrop=SCREAMING_SNAKE_CASE__      ,	embd_pdrop=SCREAMING_SNAKE_CASE__      ,	attn_pdrop=SCREAMING_SNAKE_CASE__      ,	layer_norm_epsilon=SCREAMING_SNAKE_CASE__      ,	initializer_range=SCREAMING_SNAKE_CASE__      ,	scale_attn_weights=SCREAMING_SNAKE_CASE__      ,	use_cache=SCREAMING_SNAKE_CASE__      ,	scale_attn_by_inverse_layer_idx=SCREAMING_SNAKE_CASE__      ,	reorder_and_upcast_attn=SCREAMING_SNAKE_CASE__      ,	)
         A						:     Dict         =GPTaLMHeadModel(SCREAMING_SNAKE_CASE__     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					List[Any]      ,	SCREAMING_SNAKE_CASE__    :					torch.Tensor      ,	SCREAMING_SNAKE_CASE__    :					torch.Tensor      ,	SCREAMING_SNAKE_CASE__    :					Optional[torch.Tensor] = None      ,	SCREAMING_SNAKE_CASE__    :					Optional[torch.Tensor] = None      ,	)    ->				Optional[Any]:
         A						:     str         =self.transformer.transformer.wte(SCREAMING_SNAKE_CASE__     )
         A						:     Any         =self.encode_prefix(SCREAMING_SNAKE_CASE__     )
         A						:     Optional[Any]         =self.decode_prefix(SCREAMING_SNAKE_CASE__     )
         A						:     Optional[int]         =torch.cat((prefix_embeds, embedding_text)      ,	dim=1     )

         if labels is not None:
                A						:     int         =self.get_dummy_token(input_ids.shape[0]      ,	input_ids.device     )
                A						:     Optional[int]         =torch.cat((dummy_token, input_ids)      ,	dim=1     )
         A						:     Dict         =self.transformer(inputs_embeds=SCREAMING_SNAKE_CASE__      ,	labels=SCREAMING_SNAKE_CASE__      ,	attention_mask=SCREAMING_SNAKE_CASE__     )
         if self.prefix_hidden_dim is not None:
                return out, hidden
         else:
                return out







  def 		SCREAMING_SNAKE_CASE_		(				self    :					List[str]      ,	SCREAMING_SNAKE_CASE__    :					int      ,	SCREAMING_SNAKE_CASE__    :					torch.device     )    ->				torch.Tensor:
         return torch.zeros(SCREAMING_SNAKE_CASE__      ,	self.prefix_length      ,	dtype=torch.intaa      ,	device=SCREAMING_SNAKE_CASE__     )







  def 		SCREAMING_SNAKE_CASE_		(				self    :					Union[str, Any]      ,	SCREAMING_SNAKE_CASE__    :					Optional[int]     )    ->				List[str]:
         return self.encode_prefix(SCREAMING_SNAKE_CASE__     )







  @torch.no_grad()
  def 		SCREAMING_SNAKE_CASE_		(				self    :					List[Any]      ,	SCREAMING_SNAKE_CASE__    :					List[str]      ,	SCREAMING_SNAKE_CASE__    :					Union[str, Any]      ,	SCREAMING_SNAKE_CASE__    :					Any     )    ->				Dict:
         A						:     Dict         =torch.split(SCREAMING_SNAKE_CASE__      ,	1      ,	dim=0     )
         A						:     int         =[]
         A						:     Optional[int]         =[]
         for feature in features:
                A						:     int         =self.decode_prefix(feature.to(SCREAMING_SNAKE_CASE__     )     )  # back to the clip feature
                # Only support beam search for now
                A  ,  A						:     Dict         =self.generate_beam(
                    input_embeds=SCREAMING_SNAKE_CASE__      ,	device=SCREAMING_SNAKE_CASE__      ,	eos_token_id=SCREAMING_SNAKE_CASE__     )
                generated_tokens.append(output_tokens[0]     )
                generated_seq_lengths.append(seq_lengths[0]     )
         A						:     str         =torch.stack(SCREAMING_SNAKE_CASE__     )
         A						:     int         =torch.stack(SCREAMING_SNAKE_CASE__     )
         return generated_tokens, generated_seq_lengths







  @torch.no_grad()
  def 		SCREAMING_SNAKE_CASE_		(				self    :					Optional[int]      ,	SCREAMING_SNAKE_CASE__    :					Union[str, Any]=None      ,	SCREAMING_SNAKE_CASE__    :					List[Any]=None      ,	SCREAMING_SNAKE_CASE__    :					Tuple=None      ,	SCREAMING_SNAKE_CASE__    :					int = 5      ,	SCREAMING_SNAKE_CASE__    :					int = 67      ,	SCREAMING_SNAKE_CASE__    :					float = 1.0      ,	SCREAMING_SNAKE_CASE__    :					Optional[int] = None      ,	)    ->				Dict:
         A						:     Dict         =eos_token_id
         A						:     str         =None
         A						:     List[Any]         =None
         A						:     List[Any]         =torch.ones(SCREAMING_SNAKE_CASE__      ,	device=SCREAMING_SNAKE_CASE__      ,	dtype=torch.int     )
         A						:     str         =torch.zeros(SCREAMING_SNAKE_CASE__      ,	device=SCREAMING_SNAKE_CASE__      ,	dtype=torch.bool     )

         if input_embeds is not None:
                A						:     Any         =input_embeds
         else:
                A						:     List[Any]         =self.transformer.transformer.wte(SCREAMING_SNAKE_CASE__     )

         for i in range(SCREAMING_SNAKE_CASE__     ):
                A						:     Any         =self.transformer(inputs_embeds=SCREAMING_SNAKE_CASE__     )
                A						:     str         =outputs.logits
                A						:     Union[str, Any]         =logits[:, -1, :] / (temperature if temperature > 0 else 1.0)
                A						:     List[str]         =logits.softmax(-1     ).log()

                if scores is None:
                       A  ,  A						:     Any         =logits.topk(SCREAMING_SNAKE_CASE__      ,	-1     )
                       A						:     Any         =generated.expand(SCREAMING_SNAKE_CASE__      ,	*generated.shape[1:]     )
                       A  ,  A						:     Tuple         =next_tokens.permute(1      ,	0     ), scores.squeeze(0     )
                       if tokens is None:
                              A						:     Union[str, Any]         =next_tokens
                       else:
                              A						:     str         =tokens.expand(SCREAMING_SNAKE_CASE__      ,	*tokens.shape[1:]     )
                              A						:     Optional[int]         =torch.cat((tokens, next_tokens)      ,	dim=1     )
                else:
                       A						:     Optional[Any]         =-float(np.inf     )
                       A						:     Tuple         =0
                       A						:     Optional[Any]         =scores[:, None] + logits
                       seq_lengths[~is_stopped] += 1
                       A						:     int         =scores_sum / seq_lengths[:, None]
                       A  ,  A						:     Optional[int]         =scores_sum_average.view(-1     ).topk(SCREAMING_SNAKE_CASE__      ,	-1     )
                       A						:     Dict         =next_tokens // scores_sum.shape[1]
                       A						:     Optional[Any]         =seq_lengths[next_tokens_source]
                       A						:     Tuple         =next_tokens % scores_sum.shape[1]
                       A						:     Optional[Any]         =next_tokens.unsqueeze(1     )
                       A						:     Optional[Any]         =tokens[next_tokens_source]
                       A						:     Any         =torch.cat((tokens, next_tokens)      ,	dim=1     )
                       A						:     List[str]         =generated[next_tokens_source]
                       A						:     List[Any]         =scores_sum_average * seq_lengths
                       A						:     Optional[Any]         =is_stopped[next_tokens_source]

                A						:     Optional[int]         =self.transformer.transformer.wte(next_tokens.squeeze()     ).view(generated.shape[0]      ,	1      ,	-1     )
                A						:     Any         =torch.cat((generated, next_token_embed)      ,	dim=1     )
                A						:     Optional[int]         =is_stopped + next_tokens.eq(SCREAMING_SNAKE_CASE__     ).squeeze()
                if is_stopped.all():
                       break

         A						:     Optional[Any]         =scores / seq_lengths
         A						:     str         =scores.argsort(descending=SCREAMING_SNAKE_CASE__     )
         # tokens tensors are already padded to max_seq_length
         A						:     Optional[Any]         =[tokens[i] for i in order]
         A						:     Any         =torch.stack(SCREAMING_SNAKE_CASE__      ,	dim=0     )
         A						:     str         =torch.tensor([seq_lengths[i] for i in order]      ,	dtype=seq_lengths.dtype     )
         return output_texts, seq_lengths






