

import collections
import json
import os
import re
from typing import TYPE_CHECKING, List, Optional, Tuple

import numpy as np

from ...tokenization_utils_fast import PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
	from transformers.pipelines.conversational import Conversation


A_ :				Any						 =    logging.get_logger(__name__)

A_ :				Tuple						 =    {'vocab_file': 'vocab.txt', 'emoji_file': 'emoji.json'}

A_ :				List[str]						 =    {
    'vocab_file': {
        'abeja/gpt-neox-japanese-2.7b': 'https://huggingface.co/abeja/gpt-neox-japanese-2.7b/resolve/main/vocab.txt',
    },
    'emoji_file': {
        'abeja/gpt-neox-japanese-2.7b': 'https://huggingface.co/abeja/gpt-neox-japanese-2.7b/resolve/main/emoji.json',
    },
}

A_ :				Optional[Any]						 =    {
    'abeja/gpt-neox-japanese-2.7b': 2048,
}

def 		UpperCamelCase      (lowercase_:				Tuple					,   lowercase_:				Union[str, Any]      )  ->   str:
	with open(lowerCamelCase__					,   """r"""					,   encoding="""utf-8"""      ) as f:
		A__							:					Tuple       =	json.loads(f.read()      )

	A__							:					Dict       =	collections.OrderedDict()
	A__							:					List[str]       =	collections.OrderedDict()
	A__							:					Union[str, Any]       =	collections.OrderedDict()
	with open(lowerCamelCase__					,   """r"""					,   encoding="""utf-8"""      ) as f:
		A__							:					Union[str, Any]       =	f.readlines()
	A__							:					Union[str, Any]       =	[[t.rstrip("""\n"""      )] if (t == """,""" or """,""" not in t) else t.rstrip("""\n"""      ).split(""","""      ) for t in token]
	for idx, b in enumerate(lowerCamelCase__      ):
		A__							:					int       =	b
		A__							:					Any       =	idx
		for wd in b:
			A__							:					Tuple       =	idx

	return vocab, raw_vocab, ids_to_tokens, emoji



class 					_a		(__magic_name__       ):


					'''simple docstring'''






					UpperCAmelCase__:     Optional[int]													= VOCAB_FILES_NAMES
					UpperCAmelCase__:     Any													= PRETRAINED_VOCAB_FILES_MAP
					UpperCAmelCase__:     Optional[Any]													= PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
					UpperCAmelCase__:     str													= ["""input_ids""", """attention_mask"""]






					def __init__(				self					, A__					, A__					, A__="<|endoftext|>"					, A__="<|endoftext|>"					, A__="<|startoftext|>"					, A__="<|endoftext|>"					, A__=False					, **A__					, ):
						super().__init__(
						    unk_token=A__					, pad_token=A__					, bos_token=A__					, eos_token=A__					, do_clean_text=A__					, **A__					, )
						if not os.path.isfile(A__    ):
							raise ValueError(
							    F"""Can't find a vocabulary file at path '{vocab_file}'. To load the vocabulary from a Google pretrained"""
							    """ model use `tokenizer = GPTNeoXJapaneseokenizer.from_pretrained(PRETRAINED_MODEL_NAME)`"""    )
						if not os.path.isfile(A__    ):
							raise ValueError(
							    F"""Can't find a emoji file at path '{emoji_file}'. To load the emoji information from a Google"""
							    """ pretrained model use `tokenizer = GPTNeoXJapaneseokenizer.from_pretrained(PRETRAINED_MODEL_NAME)`"""    )
						A__							:					int       =	do_clean_text
						A__     ,     A__     ,     A__     ,     A__							:					Union[str, Any]       =	load_vocab_and_emoji(A__					, A__    )
						A__							:					Optional[Any]       =	SubWordJapaneseTokenizer(
						    vocab=self.vocab					, ids_to_tokens=self.ids_to_tokens					, emoji=self.emoji    )






					@property
					def 						__A    (				self    ):
						return len(self.raw_vocab    )






					def 						__A    (				self    ):
						return dict(self.raw_vocab					, **self.added_tokens_encoder    )






					def 						__A    (				self					, A__    ):
						return self.subword_tokenizer.tokenize(A__					, clean=self.do_clean_text    )






					def 						__A    (				self					, A__    ):
						return self.vocab.get(A__					, self.vocab.get(self.unk_token    )    )






					def 						__A    (				self					, A__    ):
						return self.subword_tokenizer.convert_id_to_token(A__    )






					def 						__A    (				self					, A__    ):
						A__							:					Union[str, Any]       =	"""""".join(A__    ).strip()
						return out_string






					def 						__A    (				self					, A__    ):
						A__							:					Tuple       =	[]
						for is_user, text in conversation.iter_texts():
							input_ids.extend(self.encode(A__					, add_special_tokens=A__    ) + [self.eos_token_id]    )

						if len(A__    ) > self.model_max_length:
							A__							:					Union[str, Any]       =	input_ids[-self.model_max_length :]
						return input_ids






					def 						__A    (				self					, A__					, A__ = None    ):
						A__							:					Union[str, Any]       =	0
						if os.path.isdir(A__    ):
							A__							:					Union[str, Any]       =	os.path.join(
							    A__					, (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]    )
							A__							:					Tuple       =	os.path.join(
							    A__					, (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""emoji_file"""]    )
						else:
							A__							:					int       =	(
							    (filename_prefix + """-""" if filename_prefix else """""") + save_directory + VOCAB_FILES_NAMES["""vocab_file"""]
							)
							A__							:					int       =	(
							    (filename_prefix + """-""" if filename_prefix else """""") + save_directory + VOCAB_FILES_NAMES["""emoji_file"""]
							)
						with open(A__					, """w"""					, encoding="""utf-8"""    ) as writer:
							for token_index, token in self.ids_to_tokens.items():
								if index != token_index:
									logger.warning(
									    F"""Saving vocabulary to {vocab_file}: vocabulary indices are not consecutive."""
									    """ Please check that the vocabulary is not corrupted!"""    )
									A__							:					Dict       =	token_index
								writer.write(""",""".join(A__    ) + """\n"""    )
								index += 1
						with open(A__					, """w"""					, encoding="""utf-8"""    ) as writer:
							json.dump(self.emoji					, A__    )
						return vocab_file, emoji_file



class 					_a		(__magic_name__       ):


					'''simple docstring'''






					def __init__(				self					, A__					, A__					, A__    ):
						A__							:					str       =	vocab  # same as swe
						A__							:					Tuple       =	ids_to_tokens  # same as bpe
						A__							:					Any       =	emoji
						A__							:					str       =	np.max([len(A__    ) for w in self.vocab.keys()]    )
						A__							:					Dict       =	re.compile(r"""(https?|ftp)(:\/\/[-_\.!~*\'()a-zA-Z0-9;\/?:\@&=\+$,%#]+)"""    )
						A__							:					Any       =	re.compile(r"""[A-Za-z0-9\._+]*@[\-_0-9A-Za-z]+(\.[A-Za-z]+)*"""    )
						A__							:					Optional[Any]       =	re.compile(r"""[\(]{0,1}[0-9]{2,4}[\)\-\(]{0,1}[0-9]{2,4}[\)\-]{0,1}[0-9]{3,4}"""    )
						A__							:					Dict       =	re.compile(
						    r"""([12]\d{3}[/\-年])*(0?[1-9]|1[0-2])[/\-月]((0?[1-9]|[12][0-9]|3[01])日?)*(\d{1,2}|:|\d{1,2}時|\d{1,2}分|\(日\)|\(月\)|\(火\)|\(水\)|\(木\)|\(金\)|\(土\)|㈰|㈪|㈫|㈬|㈭|㈮|㈯)*"""    )
						A__							:					Union[str, Any]       =	re.compile(
						    r"""(明治|大正|昭和|平成|令和|㍾|㍽|㍼|㍻|\u32ff)\d{1,2}年(0?[1-9]|1[0-2])月(0?[1-9]|[12][0-9]|3[01])日(\d{1,2}|:|\d{1,2}時|\d{1,2}分|\(日\)|\(月\)|\(火\)|\(水\)|\(木\)|\(金\)|\(土\)|㈰|㈪|㈫|㈬|㈭|㈮|㈯)*"""    )
						A__							:					Union[str, Any]       =	re.compile(
						    r"""((0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*億)*((0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*万)*((0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*千)*(0|[1-9]\d*|[1-9]\d{0,2}(,\d{3})+)*(千円|万円|千万円|円|千ドル|万ドル|千万ドル|ドル|千ユーロ|万ユーロ|千万ユーロ|ユーロ)+(\(税込\)|\(税抜\)|\+tax)*"""    )
						A__							:					Dict       =	"""─━│┃┄┅┆┇┈┉┊┋┌┍┎┏┐┑┒┓└┕┖┗┘┙┚┛├┝┞┟┠┡┢┣┤┥┦┧┨┩┪┫┬┭┮┯┰┱┲┳┴┵┶┷┸┹┺┻┼┽┾┿╀╁╂╃╄╅╆╇╈╉╊╋╌╍╎╏═║╒╓╔╕╖╗╘╙╚╛╜╝╞╟╠╡╢╣╤╥╦╧╨╩╪╫╬╭╮╯╰╱╲╳╴╵╶╷╸╹╺╻╼╽╾╿"""
						A__							:					List[Any]       =	"""▀▁▂▃▄▅▆▇█▉▊▋▌▍▎▏▐░▒▓▔▕▖▗▘▙▚▛▜▝▞▟"""
						A__							:					Union[str, Any]       =	str.maketrans({k: """<BLOCK>""" for k in keisen + blocks}    )






					def __len__(				self    ):
						return len(self.ids_to_tokens    )






					def 						__A    (				self					, A__    ):
						A__							:					Optional[Any]       =	self.content_repattera.sub("""<URL>"""					, A__    )
						A__							:					Optional[int]       =	self.content_repattera.sub("""<EMAIL>"""					, A__    )
						A__							:					Optional[Any]       =	self.content_repattera.sub("""<TEL>"""					, A__    )
						A__							:					Union[str, Any]       =	self.content_repattera.sub("""<DATE>"""					, A__    )
						A__							:					Union[str, Any]       =	self.content_repattera.sub("""<DATE>"""					, A__    )
						A__							:					Optional[Any]       =	self.content_repattera.sub("""<PRICE>"""					, A__    )
						A__							:					Tuple       =	content.translate(self.content_transa    )
						while "<BLOCK><BLOCK>" in content:
							A__							:					Union[str, Any]       =	content.replace("""<BLOCK><BLOCK>"""					, """<BLOCK>"""    )
						return content






					def 						__A    (				self					, A__					, A__=False    ):
						A__							:					str       =	text.replace(""" """					, """<SP>"""    )
						A__							:					Dict       =	text.replace("""　"""					, """<SP>"""    )
						A__							:					str       =	text.replace("""\r\n"""					, """<BR>"""    )
						A__							:					Dict       =	text.replace("""\n"""					, """<BR>"""    )
						A__							:					str       =	text.replace("""\r"""					, """<BR>"""    )
						A__							:					Optional[int]       =	text.replace("""\t"""					, """<TAB>"""    )
						A__							:					List[str]       =	text.replace("""—"""					, """ー"""    )
						A__							:					str       =	text.replace("""−"""					, """ー"""    )
						for k, v in self.emoji["emoji"].items():
							if k in text:
								A__							:					Any       =	text.replace(A__					, A__    )
						if clean:
							A__							:					Optional[Any]       =	self.clean_text(A__    )

						def check_simbol(A__    ):
							A__							:					Dict       =	x.encode()
							if len(A__    ) == 1 and len(A__    ) == 2:
								A__							:					Optional[Any]       =	(int(e[0]    ) << 8) + int(e[1]    )
								if (
								    (c >= 0xC_2A1 and c <= 0xC_2BF)
								    or (c >= 0xC_780 and c <= 0xC_783)
								    or (c >= 0xC_AB9 and c <= 0xC_BBF)
								    or (c >= 0xC_C80 and c <= 0xC_DA2)
								):
									return True
							return False

						def checkuae(A__    ):
							A__							:					Tuple       =	x.encode()
							if len(A__    ) == 1 and len(A__    ) == 3:
								A__							:					Union[str, Any]       =	(int(e[0]    ) << 16) + (int(e[1]    ) << 8) + int(e[2]    )
								if c >= 0xE28_080 and c <= 0xE2B_07F:
									return True
							return False

						A__							:					List[str]       =	0
						A__							:					Dict       =	[]
						while pos < len(A__    ):
							A__							:					Dict       =	min(len(A__    )					, pos + self.maxlen + 1    ) if text[pos] == """<""" else pos + 3
							A__							:					str       =	[]  # (token_id, token, pos)
							for e in range(A__					, A__					, -1    ):
								A__							:					Union[str, Any]       =	text[pos:e]
								if wd in self.vocab:
									if wd[0] == "<" and len(A__    ) > 2:
										A__							:					List[Any]       =	[(self.vocab[wd], wd, e)]
										break
									else:
										candidates.append((self.vocab[wd], wd, e)    )
							if len(A__    ) > 0:
								# the smallest token_id is adopted
								A__     ,     A__     ,     A__							:					Any       =	sorted(A__					, key=lambda A__    : x[0]    )[0]
								result.append(A__    )
								A__							:					Dict       =	e
							else:
								A__							:					Optional[Any]       =	pos + 1
								A__							:					str       =	text[pos:end]
								if check_simbol(A__    ):
									result.append("""<KIGOU>"""    )
								elif checkuae(A__    ):
									result.append("""<U2000U2BFF>"""    )
								else:
									for i in wd.encode("""utf-8"""    ):
										result.append("""<|byte%d|>""" % i    )
								A__							:					int       =	end
						return result






					def 						__A    (				self					, A__					, A__="\n"    ):
						A__							:					List[str]       =	[]
						A__							:					Any       =	[]
						A__							:					str       =	self.ids_to_tokens[index][0]
						if word[:6] == "<|byte" and word[-2:] == "|>":
							byte_tokens.append(int(word[6:-2]    )    )
						else:
							if len(A__    ) > 0:
								words.append(bytearray(A__    ).decode("""utf-8"""					, errors="""replace"""    )    )
								A__							:					List[Any]       =	[]
							if word[:7] == "<|emoji" and word[-2:] == "|>":
								words.append(self.emoji["""emoji_inv"""][word]    )
							elif word == "<SP>":
								words.append(""" """    )
							elif word == "<BR>":
								words.append(A__    )
							elif word == "<TAB>":
								words.append("""\t"""    )
							elif word == "<BLOCK>":
								words.append("""▀"""    )
							elif word == "<KIGOU>":
								words.append("""ǀ"""    )
							elif word == "<U2000U2BFF>":
								words.append("""‖"""    )
							else:
								words.append(A__    )
						if len(A__    ) > 0:
							words.append(bytearray(A__    ).decode("""utf-8"""					, errors="""replace"""    )    )
						A__							:					Optional[Any]       =	"""""".join(A__    )
						return text






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 					_a		(__magic_name__	,     __magic_name__	,     __magic_name__       ):


					'''simple docstring'''






					UpperCAmelCase__:     str													= [r'''h\.\d+\.attn\.bias''', r'''h\.\d+\.attn\.masked_bias''']






					@register_to_config
					def __init__(				self					, A__					, A__					, A__ = None					, A__ = 5_0257					, A__ = 1024					, A__ = 768					, A__ = 12					, A__ = 12					, A__ = None					, A__ = "gelu_new"					, A__ = 0.1					, A__ = 0.1					, A__ = 0.1					, A__ = 1e-5					, A__ = 0.0_2					, A__ = True					, A__ = True					, A__ = False					, A__ = False					, ):
						super().__init__()

						A__							:					Union[str, Any]       =	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__							:					str       =	prefix_inner_dim
						A__							:					Optional[Any]       =	prefix_hidden_dim

						A__							:					Tuple       =	(
						    nn.Linear(self.prefix_inner_dim					, self.prefix_hidden_dim    )
						    if self.prefix_hidden_dim is not None
						    else nn.Identity()
						)
						A__							:					int       =	(
						    nn.Linear(self.prefix_hidden_dim					, A__    ) if self.prefix_hidden_dim is not None else nn.Identity()
						)

						A__							:					Tuple       =	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__							:					int       =	GPTaLMHeadModel(A__    )






					def 						__A    (				self					, A__					, A__					, A__ = None					, A__ = None					, ):
						A__							:					List[str]       =	self.transformer.transformer.wte(A__    )
						A__							:					int       =	self.encode_prefix(A__    )
						A__							:					int       =	self.decode_prefix(A__    )
						A__							:					Optional[Any]       =	torch.cat((prefix_embeds, embedding_text)					, dim=1    )

						if labels is not None:
							A__							:					Any       =	self.get_dummy_token(input_ids.shape[0]					, input_ids.device    )
							A__							:					List[Any]       =	torch.cat((dummy_token, input_ids)					, dim=1    )
						A__							:					List[str]       =	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 						__A    (				self					, A__					, A__    ):
						return torch.zeros(A__					, self.prefix_length					, dtype=torch.intaa					, device=A__    )






					def 						__A    (				self					, A__    ):
						return self.encode_prefix(A__    )






					@torch.no_grad()
					def 						__A    (				self					, A__					, A__					, A__    ):
						A__							:					List[Any]       =	torch.split(A__					, 1					, dim=0    )
						A__							:					Optional[int]       =	[]
						A__							:					str       =	[]
						for feature in features:
							A__							:					Dict       =	self.decode_prefix(feature.to(A__    )    )  # back to the clip feature
							# Only support beam search for now
							A__     ,     A__							:					Union[str, Any]       =	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__							:					int       =	torch.stack(A__    )
						A__							:					List[Any]       =	torch.stack(A__    )
						return generated_tokens, generated_seq_lengths






					@torch.no_grad()
					def 						__A    (				self					, A__=None					, A__=None					, A__=None					, A__ = 5					, A__ = 67					, A__ = 1.0					, A__ = None					, ):
						A__							:					Any       =	eos_token_id
						A__							:					Any       =	None
						A__							:					Optional[int]       =	None
						A__							:					Optional[Any]       =	torch.ones(A__					, device=A__					, dtype=torch.int    )
						A__							:					Any       =	torch.zeros(A__					, device=A__					, dtype=torch.bool    )

						if input_embeds is not None:
							A__							:					Dict       =	input_embeds
						else:
							A__							:					str       =	self.transformer.transformer.wte(A__    )

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

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

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

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



