





'''simple docstring'''




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


lowercase__       : Optional[int]         =       logging.get_logger(__name__)

lowercase__       : int         =       {'''vocab_file''': '''vocab.txt''', '''emoji_file''': '''emoji.json'''}

lowercase__       : Tuple         =       {
    '''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''',
    },
}

lowercase__       : str         =       {
    '''abeja/gpt-neox-japanese-2.7b''': 20_48,
}






def 			_lowerCAmelCase     (    __snake_case    :					Dict					,				__snake_case    :					Dict     )	->	Any:
     with open(__snake_case					,				'r'					,				encoding='utf-8'     ) as f:
          __A		:					Optional[int]						=			json.loads(f.read()     )

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

     return vocab, raw_vocab, ids_to_tokens, emoji


class 					SCREAMING_SNAKE_CASE       (a__				):
     lowerCAmelCase						     =			VOCAB_FILES_NAMES
     lowerCAmelCase						     =			PRETRAINED_VOCAB_FILES_MAP
     lowerCAmelCase						     =			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     lowerCAmelCase						     =			['''input_ids''', '''attention_mask''']


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase="<|endoftext|>"		,      _UpperCAmelCase="<|endoftext|>"		,      _UpperCAmelCase="<|startoftext|>"		,      _UpperCAmelCase="<|endoftext|>"		,      _UpperCAmelCase=False		,      **_UpperCAmelCase		,      ):

          '''simple docstring'''






          super().__init__(
              unk_token=_UpperCAmelCase		,      pad_token=_UpperCAmelCase		,      bos_token=_UpperCAmelCase		,      eos_token=_UpperCAmelCase		,      do_clean_text=_UpperCAmelCase		,      **_UpperCAmelCase		,      )
          if not os.path.isfile(_UpperCAmelCase):
               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(_UpperCAmelCase):
               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		:					Tuple						=			do_clean_text
          __A					,__A					,__A					,__A		:					Dict						=			load_vocab_and_emoji(_UpperCAmelCase		,      _UpperCAmelCase)
          __A		:					int						=			SubWordJapaneseTokenizer(
              vocab=self.vocab		,      ids_to_tokens=self.ids_to_tokens		,      emoji=self.emoji)


     @property
     def 				SCREAMING_SNAKE_CASE    (		self):

          '''simple docstring'''






          return len(self.raw_vocab)


     def 				SCREAMING_SNAKE_CASE    (		self):

          '''simple docstring'''






          return dict(self.raw_vocab		,      **self.added_tokens_encoder)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          return self.subword_tokenizer.tokenize(_UpperCAmelCase		,      clean=self.do_clean_text)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          return self.vocab.get(_UpperCAmelCase		,      self.vocab.get(self.unk_token))


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          return self.subword_tokenizer.convert_id_to_token(_UpperCAmelCase)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					int						=			''.join(_UpperCAmelCase).strip()
          return out_string


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






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

          if len(_UpperCAmelCase) > self.model_max_length:
               __A		:					Optional[int]						=			input_ids[-self.model_max_length :]
          return input_ids


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase = None):

          '''simple docstring'''






          __A		:					List[str]						=			0
          if os.path.isdir(_UpperCAmelCase):
               __A		:					Dict						=			os.path.join(
                   _UpperCAmelCase		,      (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file'])
               __A		:					int						=			os.path.join(
                   _UpperCAmelCase		,      (filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['emoji_file'])
          else:
               __A		:					Tuple						=			(
                   (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(_UpperCAmelCase		,      '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(_UpperCAmelCase) + '\n')
                    index += 1
          with open(_UpperCAmelCase		,      'w'		,      encoding='utf-8') as writer:
               json.dump(self.emoji		,      _UpperCAmelCase)
          return vocab_file, emoji_file




class 					SCREAMING_SNAKE_CASE       (a__				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					int						=			vocab  # same as swe
          __A		:					Union[str, Any]						=			ids_to_tokens  # same as bpe
          __A		:					Optional[int]						=			emoji
          __A		:					List[str]						=			np.max([len(_UpperCAmelCase) for w in self.vocab.keys()])
          __A		:					Optional[int]						=			re.compile(R'(https?|ftp)(:\/\/[-_\.!~*\'()a-zA-Z0-9;\/?:\@&=\+$,%#]+)')
          __A		:					Optional[Any]						=			re.compile(R'[A-Za-z0-9\._+]*@[\-_0-9A-Za-z]+(\.[A-Za-z]+)*')
          __A		:					str						=			re.compile(R'[\(]{0,1}[0-9]{2,4}[\)\-\(]{0,1}[0-9]{2,4}[\)\-]{0,1}[0-9]{3,4}')
          __A		:					Any						=			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		:					List[str]						=			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		:					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		:					str						=			'─━│┃┄┅┆┇┈┉┊┋┌┍┎┏┐┑┒┓└┕┖┗┘┙┚┛├┝┞┟┠┡┢┣┤┥┦┧┨┩┪┫┬┭┮┯┰┱┲┳┴┵┶┷┸┹┺┻┼┽┾┿╀╁╂╃╄╅╆╇╈╉╊╋╌╍╎╏═║╒╓╔╕╖╗╘╙╚╛╜╝╞╟╠╡╢╣╤╥╦╧╨╩╪╫╬╭╮╯╰╱╲╳╴╵╶╷╸╹╺╻╼╽╾╿'
          __A		:					Dict						=			'▀▁▂▃▄▅▆▇█▉▊▋▌▍▎▏▐░▒▓▔▕▖▗▘▙▚▛▜▝▞▟'
          __A		:					int						=			str.maketrans({k: '<BLOCK>' for k in keisen + blocks})


     def __len__(		self):

          '''simple docstring'''






          return len(self.ids_to_tokens)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






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


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase=False):

          '''simple docstring'''






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

          def check_simbol(_UpperCAmelCase):
               __A		:					str						=			x.encode()
               if len(_UpperCAmelCase) == 1 and len(_UpperCAmelCase) == 2:
                    __A		:					Dict						=			(int(e[0]) << 8) + int(e[1])
                    if (
                        (c >= 0Xc_2_a_1 and c <= 0Xc_2_b_f)
                        or (c >= 0Xc_7_8_0 and c <= 0Xc_7_8_3)
                        or (c >= 0Xc_a_b_9 and c <= 0Xc_b_b_f)
                        or (c >= 0Xc_c_8_0 and c <= 0Xc_d_a_2)
                    ):
                         return True
               return False

          def checkuae(_UpperCAmelCase):
               __A		:					Optional[int]						=			x.encode()
               if len(_UpperCAmelCase) == 1 and len(_UpperCAmelCase) == 3:
                    __A		:					Dict						=			(int(e[0]) << 16) + (int(e[1]) << 8) + int(e[2])
                    if c >= 0Xe_2_8_0_8_0 and c <= 0Xe_2_b_0_7_f:
                         return True
               return False

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


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase="\n"):

          '''simple docstring'''






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






'''simple docstring'''




from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import get_down_block, get_mid_block, get_out_block, get_up_block


@dataclass
class 					SCREAMING_SNAKE_CASE       (a__				):
     lowerCAmelCase						     =			42




class 					SCREAMING_SNAKE_CASE       (a__   ,						a__				):


     @register_to_config
     def __init__(		self		,      _UpperCAmelCase = 6_5536		,      _UpperCAmelCase = None		,      _UpperCAmelCase = 2		,      _UpperCAmelCase = 2		,      _UpperCAmelCase = 0		,      _UpperCAmelCase = "fourier"		,      _UpperCAmelCase = True		,      _UpperCAmelCase = False		,      _UpperCAmelCase = 0.0		,      _UpperCAmelCase = ("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")		,      _UpperCAmelCase = ("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")		,      _UpperCAmelCase = "UNetMidBlock1D"		,      _UpperCAmelCase = None		,      _UpperCAmelCase = (32, 32, 64)		,      _UpperCAmelCase = None		,      _UpperCAmelCase = 8		,      _UpperCAmelCase = 1		,      _UpperCAmelCase = False		,      ):

          '''simple docstring'''






          super().__init__()
          __A		:					Dict						=			sample_size

          # time
          if time_embedding_type == "fourier":
               __A		:					int						=			GaussianFourierProjection(
                   embedding_size=8		,      set_W_to_weight=_UpperCAmelCase		,      log=_UpperCAmelCase		,      flip_sin_to_cos=_UpperCAmelCase)
               __A		:					Any						=			2 * block_out_channels[0]
          elif time_embedding_type == "positional":
               __A		:					List[str]						=			Timesteps(
                   block_out_channels[0]		,      flip_sin_to_cos=_UpperCAmelCase		,      downscale_freq_shift=_UpperCAmelCase)
               __A		:					List[str]						=			block_out_channels[0]

          if use_timestep_embedding:
               __A		:					Optional[Any]						=			block_out_channels[0] * 4
               __A		:					Optional[int]						=			TimestepEmbedding(
                   in_channels=_UpperCAmelCase		,      time_embed_dim=_UpperCAmelCase		,      act_fn=_UpperCAmelCase		,      out_dim=block_out_channels[0]		,      )

          __A		:					Dict						=			nn.ModuleList([])
          __A		:					Dict						=			None
          __A		:					Tuple						=			nn.ModuleList([])
          __A		:					Tuple						=			None

          # down
          __A		:					Any						=			in_channels
          for i, down_block_type in enumerate(_UpperCAmelCase):
               __A		:					Tuple						=			output_channel
               __A		:					Optional[Any]						=			block_out_channels[i]

               if i == 0:
                    input_channel += extra_in_channels

               __A		:					List[str]						=			i == len(_UpperCAmelCase) - 1

               __A		:					int						=			get_down_block(
                   _UpperCAmelCase		,      num_layers=_UpperCAmelCase		,      in_channels=_UpperCAmelCase		,      out_channels=_UpperCAmelCase		,      temb_channels=block_out_channels[0]		,      add_downsample=not is_final_block or downsample_each_block		,      )
               self.down_blocks.append(_UpperCAmelCase)

          # mid
          __A		:					str						=			get_mid_block(
              _UpperCAmelCase		,      in_channels=block_out_channels[-1]		,      mid_channels=block_out_channels[-1]		,      out_channels=block_out_channels[-1]		,      embed_dim=block_out_channels[0]		,      num_layers=_UpperCAmelCase		,      add_downsample=_UpperCAmelCase		,      )

          # up
          __A		:					Optional[int]						=			list(reversed(_UpperCAmelCase))
          __A		:					Optional[int]						=			reversed_block_out_channels[0]
          if out_block_type is None:
               __A		:					str						=			out_channels
          else:
               __A		:					List[Any]						=			block_out_channels[0]

          for i, up_block_type in enumerate(_UpperCAmelCase):
               __A		:					Optional[Any]						=			output_channel
               __A		:					Optional[Any]						=			(
                   reversed_block_out_channels[i + 1] if i < len(_UpperCAmelCase) - 1 else final_upsample_channels
               )

               __A		:					Dict						=			i == len(_UpperCAmelCase) - 1

               __A		:					str						=			get_up_block(
                   _UpperCAmelCase		,      num_layers=_UpperCAmelCase		,      in_channels=_UpperCAmelCase		,      out_channels=_UpperCAmelCase		,      temb_channels=block_out_channels[0]		,      add_upsample=not is_final_block		,      )
               self.up_blocks.append(_UpperCAmelCase)
               __A		:					Optional[int]						=			output_channel

          # out
          __A		:					str						=			norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4		,      32)
          __A		:					Optional[Any]						=			get_out_block(
              out_block_type=_UpperCAmelCase		,      num_groups_out=_UpperCAmelCase		,      embed_dim=block_out_channels[0]		,      out_channels=_UpperCAmelCase		,      act_fn=_UpperCAmelCase		,      fc_dim=block_out_channels[-1] // 4		,      )


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase = True		,      ):

          '''simple docstring'''






          __A		:					Any						=			timestep
          if not torch.is_tensor(_UpperCAmelCase):
               __A		:					Any						=			torch.tensor([timesteps]		,      dtype=torch.long		,      device=sample.device)
          elif torch.is_tensor(_UpperCAmelCase) and len(timesteps.shape) == 0:
               __A		:					Any						=			timesteps[None].to(sample.device)

          __A		:					List[Any]						=			self.time_proj(_UpperCAmelCase)
          if self.config.use_timestep_embedding:
               __A		:					Dict						=			self.time_mlp(_UpperCAmelCase)
          else:
               __A		:					Dict						=			timestep_embed[..., None]
               __A		:					Tuple						=			timestep_embed.repeat([1, 1, sample.shape[2]]).to(sample.dtype)
               __A		:					List[Any]						=			timestep_embed.broadcast_to((sample.shape[:1] + timestep_embed.shape[1:]))

          # 2. down
          __A		:					int						=			()
          for downsample_block in self.down_blocks:
               __A					,__A		:					int						=			downsample_block(hidden_states=_UpperCAmelCase		,      temb=_UpperCAmelCase)
               down_block_res_samples += res_samples

          # 3. mid
          if self.mid_block:
               __A		:					Optional[int]						=			self.mid_block(_UpperCAmelCase		,      _UpperCAmelCase)

          # 4. up
          for i, upsample_block in enumerate(self.up_blocks):
               __A		:					Any						=			down_block_res_samples[-1:]
               __A		:					Optional[int]						=			down_block_res_samples[:-1]
               __A		:					Any						=			upsample_block(_UpperCAmelCase		,      res_hidden_states_tuple=_UpperCAmelCase		,      temb=_UpperCAmelCase)

          # 5. post-process
          if self.out_block:
               __A		:					Dict						=			self.out_block(_UpperCAmelCase		,      _UpperCAmelCase)

          if not return_dict:
               return (sample,)

          return UNetaDOutput(sample=_UpperCAmelCase)