
"""simple docstring"""




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

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


SCREAMING_SNAKE_CASE							:str										=    logging.get_logger(__name__)

SCREAMING_SNAKE_CASE							:List[Any]										=    '▁'

SCREAMING_SNAKE_CASE							:List[Any]										=    {'vocab_file': 'sentencepiece.bpe.model'}

SCREAMING_SNAKE_CASE							:str										=    {
    'vocab_file': {
        'facebook/xglm-564M': 'https://huggingface.co/facebook/xglm-564M/resolve/main/sentencepiece.bpe.model',
    }
}

SCREAMING_SNAKE_CASE							:Optional[Any]										=    {
    'facebook/xglm-564M': 2048,
}





class 							UpperCAmelCase				(  __SCREAMING_SNAKE_CASE  ):




	'''simple docstring'''





	snake_case_    =					VOCAB_FILES_NAMES
	snake_case_    =					PRETRAINED_VOCAB_FILES_MAP
	snake_case_    =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	snake_case_    =					["input_ids", "attention_mask"]






	def __init__(	self :			Optional[int]							,A :			List[Any]							,A :			Tuple="<s>"							,A :			Any="</s>"							,A :			Any="</s>"							,A :			Dict="<s>"							,A :			Tuple="<unk>"							,A :			str="<pad>"							,A :			Optional[Dict[str, Any]] = None							,**A :			List[Any]							,):
					__A             =			{} if sp_model_kwargs is None else sp_model_kwargs

					# Compatibility with the original tokenizer
					__A             =			7
					__A             =			[f'''<madeupword{i}>''' for i in range(self.num_madeup_words			)]

					__A             =			kwargs.get("additional_special_tokens"							,[]			)
					kwargs["additional_special_tokens"] += [
					    word for word in madeup_words if word not in kwargs["additional_special_tokens"]
					]

					super().__init__(
					    bos_token=A							,eos_token=A							,unk_token=A							,sep_token=A							,cls_token=A							,pad_token=A							,sp_model_kwargs=self.sp_model_kwargs							,**A							,)

					__A             =			spm.SentencePieceProcessor(**self.sp_model_kwargs			)
					self.sp_model.Load(str(A			)			)
					__A             =			vocab_file

					# Original fairseq vocab and spm vocab must be "aligned":
					# Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
					# -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
					# fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
					# spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

					# The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
					__A             =			1

					# Mimic fairseq token-to-id alignment for the first 4 token
					__A             =			{"<s>": 0, "<pad>": 1, "</s>": 2, "<unk>": 3}

					__A             =			len(self.sp_model			)
					__A             =			{f'''<madeupword{i}>''': sp_size + i + self.fairseq_offset for i in range(self.num_madeup_words			)}
					self.fairseq_tokens_to_ids.update(A			)

					__A             =			{v: k for k, v in self.fairseq_tokens_to_ids.items()}






	def __getstate__(	self :			int			):
					__A             =			self.__dict__.copy()
					__A             =			None
					__A             =			self.sp_model.serialized_model_proto()
					return state






	def __setstate__(	self :			Optional[int]							,A :			int			):
					__A             =			d

					# for backward compatibility
					if not hasattr(self							,"sp_model_kwargs"			):
									__A             =			{}

					__A             =			spm.SentencePieceProcessor(**self.sp_model_kwargs			)
					self.sp_model.LoadFromSerializedProto(self.sp_model_proto			)






	def     UpperCamelCase_  (	self :			Optional[Any]							,A :			List[int]							,A :			Optional[List[int]] = None			):

					if token_ids_a is None:
									return [self.sep_token_id] + token_ids_a
					__A             =			[self.sep_token_id]
					return sep + token_ids_a + sep + sep + token_ids_a






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

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

					if token_ids_a is None:
									return [1] + ([0] * len(A			))
					return [1] + ([0] * len(A			)) + [1, 1] + ([0] * len(A			))






	def     UpperCamelCase_  (	self :			Union[str, Any]							,A :			List[int]							,A :			Optional[List[int]] = None			):
					__A             =			[self.sep_token_id]

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






	@property
	def     UpperCamelCase_  (	self :			Tuple			):
					return len(self.sp_model			) + self.fairseq_offset + self.num_madeup_words






	def     UpperCamelCase_  (	self :			List[Any]			):
					__A             =			{self.convert_ids_to_tokens(A			): i for i in range(self.vocab_size			)}
					vocab.update(self.added_tokens_encoder			)
					return vocab






	def     UpperCamelCase_  (	self :			str							,A :			str			):
					return self.sp_model.encode(A							,out_type=A			)






	def     UpperCamelCase_  (	self :			Tuple							,A :			Optional[int]			):
					if token in self.fairseq_tokens_to_ids:
									return self.fairseq_tokens_to_ids[token]
					__A             =			self.sp_model.PieceToId(A			)

					# Need to return unknown token if the SP model returned 0
					return spm_id + self.fairseq_offset if spm_id else self.unk_token_id






	def     UpperCamelCase_  (	self :			List[Any]							,A :			Dict			):
					if index in self.fairseq_ids_to_tokens:
									return self.fairseq_ids_to_tokens[index]
					return self.sp_model.IdToPiece(index - self.fairseq_offset			)






	def     UpperCamelCase_  (	self :			List[Any]							,A :			List[Any]			):
					__A             =			"".join(A			).replace(A							," "			).strip()
					return out_string







	def     UpperCamelCase_  (	self :			str							,A :			str							,A :			Optional[str] = None			):
					if not os.path.isdir(A			):
									logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''			)
									return
					__A             =			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             =			self.sp_model.serialized_model_proto()
													fi.write(A			)

					return (out_vocab_file,)


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 							UpperCAmelCase				(  __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE      ,      __SCREAMING_SNAKE_CASE  ):




	'''simple docstring'''





	snake_case_    =					[R"h\.\d+\.attn\.bias", R"h\.\d+\.attn\.masked_bias"]






	@register_to_config
	def __init__(	self :			Tuple							,A :			int							,A :			int							,A :			Optional[int] = None							,A :			int = 5_02_57							,A :			int = 10_24							,A :			int = 7_68							,A :			int = 12							,A :			int = 12							,A :			Optional[int] = None							,A :			str = "gelu_new"							,A :			float = 0.1							,A :			float = 0.1							,A :			float = 0.1							,A :			float = 1E-5							,A :			float = 0.02							,A :			bool = True							,A :			bool = True							,A :			bool = False							,A :			bool = False							,):
					super().__init__()

					__A             =			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             =			prefix_inner_dim
					__A             =			prefix_hidden_dim

					__A             =			(
					    nn.Linear(self.prefix_inner_dim							,self.prefix_hidden_dim			)
					    if self.prefix_hidden_dim is not None
					    else nn.Identity()
					)
					__A             =			(
					    nn.Linear(self.prefix_hidden_dim							,A			) if self.prefix_hidden_dim is not None else nn.Identity()
					)

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






	def     UpperCamelCase_  (	self :			Union[str, Any]							,A :			torch.Tensor							,A :			torch.Tensor							,A :			Optional[torch.Tensor] = None							,A :			Optional[torch.Tensor] = None							,):
					__A             =			self.transformer.transformer.wte(A			)
					__A             =			self.encode_prefix(A			)
					__A             =			self.decode_prefix(A			)
					__A             =			torch.cat((prefix_embeds, embedding_text)							,dim=1			)

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






	def     UpperCamelCase_  (	self :			int							,A :			int							,A :			torch.device			):
					return torch.zeros(A							,self.prefix_length							,dtype=torch.intaa							,device=A			)






	def     UpperCamelCase_  (	self :			Optional[Any]							,A :			List[Any]			):
					return self.encode_prefix(A			)






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







	@torch.no_grad()
	def     UpperCamelCase_  (	self :			Tuple							,A :			List[str]=None							,A :			Dict=None							,A :			Optional[int]=None							,A :			int = 5							,A :			int = 67							,A :			float = 1.0							,A :			Optional[int] = None							,):
					__A             =			eos_token_id
					__A             =			None
					__A             =			None
					__A             =			torch.ones(A							,device=A							,dtype=torch.int			)
					__A             =			torch.zeros(A							,device=A							,dtype=torch.bool			)

					if input_embeds is not None:
									__A             =			input_embeds
					else:
									__A             =			self.transformer.transformer.wte(A			)

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

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

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

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