





"""simple docstring"""




import io
import os
import unicodedata
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


a		:				Dict  			=    logging.get_logger(__name__)

a		:				int  			=    '''▁'''

a		:				Dict  			=    {'''vocab_file''': '''vocab.txt''', '''sentencepiece_model_ckpt''': '''sentencepiece.bpe.model'''}

a		:				Tuple  			=    {
    '''sentencepiece_model_file''': '''sentencepiece.bpe.model''',
    '''vocab_file''': '''vocab.txt''',
}

a		:				Tuple  			=    {
    '''vocab_file''': {
        '''ernie-m-base''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/vocab.txt''',
        '''ernie-m-large''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/vocab.txt''',
    },
    '''sentencepiece_model_file''': {
        '''ernie-m-base''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/sentencepiece.bpe.model''',
        '''ernie-m-large''': '''https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/sentencepiece.bpe.model''',
    },
}

a		:				Any  			=    {
    '''ernie-m-base''': 514,
    '''ernie-m-large''': 514,
}

a		:				Dict  			=    {
    '''ernie-m-base''': {'''do_lower_case''': False},
    '''ernie-m-large''': {'''do_lower_case''': False},
}


class  __UpperCamelCase       (		a__		):
							lowerCamelCase     :      List[str]         =["input_ids"]

							lowerCamelCase     :      Union[str, Any]         =VOCAB_FILES_NAMES
							lowerCamelCase     :      Dict         =PRETRAINED_INIT_CONFIGURATION
							lowerCamelCase     :      List[Any]         =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							lowerCamelCase     :      Optional[Any]         =PRETRAINED_VOCAB_FILES_MAP
							lowerCamelCase     :      Dict         =RESOURCE_FILES_NAMES






							def __init__(     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__=None   ,      lowerCAmelCase__=False   ,      lowerCAmelCase__="utf8"   ,      lowerCAmelCase__="[UNK]"   ,      lowerCAmelCase__="[SEP]"   ,      lowerCAmelCase__="[PAD]"   ,      lowerCAmelCase__="[CLS]"   ,      lowerCAmelCase__="[MASK]"   ,      lowerCAmelCase__ = None   ,      **lowerCAmelCase__   ,      )			->				None:
								# Mask token behave like a normal word, i.e. include the space before it and
								# is included in the raw text, there should be a match in a non-normalized sentence.

								a   :	Optional[int]			=       {} if sp_model_kwargs is None else sp_model_kwargs
								super().__init__(
								    do_lower_case=lowerCAmelCase__   ,      unk_token=lowerCAmelCase__   ,      sep_token=lowerCAmelCase__   ,      pad_token=lowerCAmelCase__   ,      cls_token=lowerCAmelCase__   ,      mask_token=lowerCAmelCase__   ,      vocab_file=lowerCAmelCase__   ,      encoding=lowerCAmelCase__   ,      sp_model_kwargs=self.sp_model_kwargs   ,      **lowerCAmelCase__   ,      )
								a   :	List[str]			=       do_lower_case
								a   :	Any			=       sentencepiece_model_ckpt
								a   :	Optional[Any]			=       spm.SentencePieceProcessor(**self.sp_model_kwargs   )
								self.sp_model.Load(lowerCAmelCase__   )

								# to mimic paddlenlp.transformers.ernie_m.tokenizer.ErnieMTokenizer functioning
								if vocab_file is not None:
									a   :	Any			=       self.load_vocab(filepath=lowerCAmelCase__   )
								else:
									a   :	Dict			=       {self.sp_model.id_to_piece(lowerCAmelCase__   ): id for id in range(self.sp_model.get_piece_size()   )}
								a   :	Any			=       {v: k for k, v in self.vocab.items()}






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				Any:
								if text is None:
									return None

								a   :	List[Any]			=       self.tokenize(lowerCAmelCase__   )
								a, a   :	Any			=       "", []

								for i, ch in enumerate(lowerCAmelCase__   ):
									if ch in self.SP_CHAR_MAPPING:
										a   :	Any			=       self.SP_CHAR_MAPPING.get(lowerCAmelCase__   )
									else:
										a   :	Optional[Any]			=       unicodedata.normalize("NFKC"   ,      lowerCAmelCase__   )
									if self.is_whitespace(lowerCAmelCase__   ):
										continue
									normalized_text += ch
									char_mapping.extend([i] * len(lowerCAmelCase__   )   )

								a, a, a   :	Union[str, Any]			=       normalized_text, [], 0

								if self.do_lower_case:
									a   :	List[Any]			=       text.lower()

								for token in split_tokens:
									if token[:1] == "▁":
										a   :	Tuple			=       token[1:]
									a   :	Any			=       text[offset:].index(lowerCAmelCase__   ) + offset
									a   :	Union[str, Any]			=       start + len(lowerCAmelCase__   )

									token_mapping.append((char_mapping[start], char_mapping[end - 1] + 1)   )
									a   :	Any			=       end
								return token_mapping






							@property
							def 					__a     (     self   )			->				Dict:
								return len(self.vocab   )






							def 					__a     (     self   )			->				List[Any]:
								return dict(self.vocab   ,      **self.added_tokens_encoder   )






							def __getstate__(     self   )			->				Optional[Any]:
								a   :	Optional[int]			=       self.__dict__.copy()
								a   :	Any			=       None
								return state






							def __setstate__(     self   ,      lowerCAmelCase__   )			->				Optional[int]:
								a   :	str			=       d

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

								a   :	Any			=       spm.SentencePieceProcessor(**self.sp_model_kwargs   )
								self.sp_model.Load(self.sentencepiece_model_ckpt   )






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				List[str]:
								return "".join((self.SP_CHAR_MAPPING.get(lowerCAmelCase__   ,      lowerCAmelCase__   ) for c in text)   )






							def 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__=False   ,      lowerCAmelCase__=64   ,      lowerCAmelCase__=0.1   )			->				List[Any]:

								if self.sp_model_kwargs.get("enable_sampling"   ) is True:
									a   :	int			=       True
								if self.sp_model_kwargs.get("alpha"   ) is not None:
									a   :	Any			=       self.sp_model_kwargs.get("alpha"   )
								if self.sp_model_kwargs.get("nbest_size"   ) is not None:
									a   :	str			=       self.sp_model_kwargs.get("nbest_size"   )

								if not enable_sampling:
									a   :	int			=       self.sp_model.EncodeAsPieces(lowerCAmelCase__   )
								else:
									a   :	List[str]			=       self.sp_model.SampleEncodeAsPieces(lowerCAmelCase__   ,      lowerCAmelCase__   ,      lowerCAmelCase__   )
								a   :	Optional[Any]			=       []
								for pi, piece in enumerate(lowerCAmelCase__   ):
									if piece == SPIECE_UNDERLINE:
										if not pieces[pi + 1].startswith(lowerCAmelCase__   ) and pi != 0:
											new_pieces.append(lowerCAmelCase__   )
											continue
										else:
											continue
									a   :	Union[str, Any]			=       0
									for i, chunk in enumerate(lowerCAmelCase__   ):
										if chunk == SPIECE_UNDERLINE:
											continue
										if self.is_ch_char(lowerCAmelCase__   ) or self.is_punct(lowerCAmelCase__   ):
											if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
												new_pieces.append(piece[lst_i:i]   )
											new_pieces.append(lowerCAmelCase__   )
											a   :	Tuple			=       i + 1
										elif chunk.isdigit() and i > 0 and not piece[i - 1].isdigit():
											if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
												new_pieces.append(piece[lst_i:i]   )
											a   :	Dict			=       i
										elif not chunk.isdigit() and i > 0 and piece[i - 1].isdigit():
											if i > lst_i and piece[lst_i:i] != SPIECE_UNDERLINE:
												new_pieces.append(piece[lst_i:i]   )
											a   :	Optional[Any]			=       i
									if len(lowerCAmelCase__   ) > lst_i:
										new_pieces.append(piece[lst_i:]   )
								return new_pieces






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				Optional[Any]:
								a   :	Any			=       "".join(lowerCAmelCase__   ).replace(lowerCAmelCase__   ,      " "   ).strip()
								return out_string






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				Optional[int]:
								a   :	Optional[Any]			=       self.convert_ids_to_tokens(lowerCAmelCase__   )
								a   :	Optional[Any]			=       "".join(lowerCAmelCase__   ).replace(lowerCAmelCase__   ,      " "   ).strip()
								return out_string






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				str:
								return self.vocab.get(lowerCAmelCase__   ,      self.vocab.get(self.unk_token   )   )






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				str:
								return self.reverse_vocab.get(lowerCAmelCase__   ,      self.unk_token   )






							def 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__=None   )			->				Dict:
								if token_ids_a is None:
									return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
								a   :	Tuple			=       [self.cls_token_id]
								a   :	Any			=       [self.sep_token_id]
								return _cls + token_ids_a + _sep + _sep + token_ids_a + _sep






							def 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__=None   )			->				Any:
								if offset_mapping_a is None:
									return [(0, 0)] + offset_mapping_a + [(0, 0)]

								return [(0, 0)] + offset_mapping_a + [(0, 0), (0, 0)] + offset_mapping_a + [(0, 0)]






							def 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__=None   ,      lowerCAmelCase__=False   )			->				Tuple:

								if already_has_special_tokens:
									if token_ids_a is not None:
										raise ValueError(
										    "You should not supply a second sequence if the provided sequence of "
										    "ids is already formatted with special tokens for the model."   )
									return [1 if x in [self.sep_token_id, self.cls_token_id] else 0 for x in token_ids_a]

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






							def 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__ = None   )			->				List[int]:
								# called when `add_special_tokens` is True, so align with `build_inputs_with_special_tokens` method
								if token_ids_a is None:
									# [CLS] X [SEP]
									return (len(lowerCAmelCase__   ) + 2) * [0]

								# [CLS] A [SEP] [SEP] B [SEP]
								return [0] * (len(lowerCAmelCase__   ) + 1) + [1] * (len(lowerCAmelCase__   ) + 3)






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				List[str]:
								if "\u4e00" <= char <= "\u9fff":
									return True
								return False






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				str:
								if ("a" <= char <= "z") or ("A" <= char <= "Z"):
									return True
								return False






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				Any:
								if char in ",;:.?!~，；：。？！《》【】":
									return True
								return False






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				Dict:
								if char == " " or char == "\t" or char == "\n" or char == "\r":
									return True
								if len(lowerCAmelCase__   ) == 1:
									a   :	List[str]			=       unicodedata.category(lowerCAmelCase__   )
									if cat == "Zs":
										return True
								return False






							def 					__a     (     self   ,      lowerCAmelCase__   )			->				Optional[int]:
								a   :	List[Any]			=       {}
								with io.open(lowerCAmelCase__   ,      "r"   ,      encoding="utf-8"   ) as f:
									for index, line in enumerate(lowerCAmelCase__   ):
										a   :	List[Any]			=       line.rstrip("\n"   )
										a   :	List[Any]			=       int(lowerCAmelCase__   )

								return token_to_idx






							def 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__ = None   )			->				Tuple[str]:
								a   :	List[str]			=       0
								if os.path.isdir(lowerCAmelCase__   ):
									a   :	Dict			=       os.path.join(
									    lowerCAmelCase__   ,      (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]   )
								else:
									a   :	Optional[int]			=       (filename_prefix + "-" if filename_prefix else "") + save_directory
								with open(lowerCAmelCase__   ,      "w"   ,      encoding="utf-8"   ) as writer:
									for token, token_index in sorted(self.vocab.items()   ,      key=lambda lowerCAmelCase__   : kv[1]   ):
										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   :	str			=       token_index
										writer.write(token + "\n"   )
										index += 1

								a   :	Any			=       os.path.join(lowerCAmelCase__   ,      "sentencepiece.bpe.model"   )
								with open(lowerCAmelCase__   ,      "wb"   ) as fi:
									a   :	Union[str, Any]			=       self.sp_model.serialized_model_proto()
									fi.write(lowerCAmelCase__   )

								return (vocab_file,)









"""simple docstring"""




import os
import shutil
from pathlib import Path
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ..utils import ONNX_EXTERNAL_WEIGHTS_NAME, ONNX_WEIGHTS_NAME, is_onnx_available, logging


if is_onnx_available():
		import onnxruntime as ort


a		:				Any  			=    logging.get_logger(__name__)

a		:				Tuple  			=    {
    '''tensor(bool)''': np.bool_,
    '''tensor(int8)''': np.inta,
    '''tensor(uint8)''': np.uinta,
    '''tensor(int16)''': np.intaa,
    '''tensor(uint16)''': np.uintaa,
    '''tensor(int32)''': np.intaa,
    '''tensor(uint32)''': np.uintaa,
    '''tensor(int64)''': np.intaa,
    '''tensor(uint64)''': np.uintaa,
    '''tensor(float16)''': np.floataa,
    '''tensor(float)''': np.floataa,
    '''tensor(double)''': np.floataa,
}


class  __UpperCamelCase       :






							def __init__(     self   ,      lowerCAmelCase__=None   ,      **lowerCAmelCase__   )			->				str:
								logger.info("`diffusers.OnnxRuntimeModel` is experimental and might change in the future."   )
								a   :	Optional[int]			=       model
								a   :	int			=       kwargs.get("model_save_dir"   ,      lowerCAmelCase__   )
								a   :	Tuple			=       kwargs.get("latest_model_name"   ,      lowerCAmelCase__   )






							def __call__(     self   ,      **lowerCAmelCase__   )			->				Dict:
								a   :	List[str]			=       {k: np.array(lowerCAmelCase__   ) for k, v in kwargs.items()}
								return self.model.run(lowerCAmelCase__   ,      lowerCAmelCase__   )






							@staticmethod
							def 					__a     (     lowerCAmelCase__   ,      lowerCAmelCase__=None   ,      lowerCAmelCase__=None   )			->				Union[str, Any]:
								if provider is None:
									logger.info("No onnxruntime provider specified, using CPUExecutionProvider"   )
									a   :	List[str]			=       "CPUExecutionProvider"

								return ort.InferenceSession(lowerCAmelCase__   ,      providers=[provider]   ,      sess_options=lowerCAmelCase__   )






							def 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__ = None   ,      **lowerCAmelCase__   )			->				int:
								a   :	List[str]			=       file_name if file_name is not None else ONNX_WEIGHTS_NAME

								a   :	Optional[int]			=       self.model_save_dir.joinpath(self.latest_model_name   )
								a   :	List[str]			=       Path(lowerCAmelCase__   ).joinpath(lowerCAmelCase__   )
								try:
									shutil.copyfile(lowerCAmelCase__   ,      lowerCAmelCase__   )
								except shutil.SameFileError:
									pass

								# copy external weights (for models >2GB)
								a   :	str			=       self.model_save_dir.joinpath(lowerCAmelCase__   )
								if src_path.exists():
									a   :	Any			=       Path(lowerCAmelCase__   ).joinpath(lowerCAmelCase__   )
									try:
										shutil.copyfile(lowerCAmelCase__   ,      lowerCAmelCase__   )
									except shutil.SameFileError:
										pass






							def 					__a     (     self   ,      lowerCAmelCase__   ,      **lowerCAmelCase__   ,      )			->				str:
								if os.path.isfile(lowerCAmelCase__   ):
									logger.error(f"""Provided path ({save_directory}) should be a directory, not a file"""   )
									return

								os.makedirs(lowerCAmelCase__   ,      exist_ok=lowerCAmelCase__   )

								# saving model weights/files
								self._save_pretrained(lowerCAmelCase__   ,      **lowerCAmelCase__   )






							@classmethod
							def 					__a     (     cls   ,      lowerCAmelCase__   ,      lowerCAmelCase__ = None   ,      lowerCAmelCase__ = None   ,      lowerCAmelCase__ = False   ,      lowerCAmelCase__ = None   ,      lowerCAmelCase__ = None   ,      lowerCAmelCase__ = None   ,      lowerCAmelCase__ = None   ,      **lowerCAmelCase__   ,      )			->				Optional[int]:
								a   :	Tuple			=       file_name if file_name is not None else ONNX_WEIGHTS_NAME
								# load model from local directory
								if os.path.isdir(lowerCAmelCase__   ):
									a   :	Tuple			=       OnnxRuntimeModel.load_model(
									    os.path.join(lowerCAmelCase__   ,      lowerCAmelCase__   )   ,      provider=lowerCAmelCase__   ,      sess_options=lowerCAmelCase__   )
									a   :	Tuple			=       Path(lowerCAmelCase__   )
								# load model from hub
								else:
									# download model
									a   :	Optional[Any]			=       hf_hub_download(
									    repo_id=lowerCAmelCase__   ,      filename=lowerCAmelCase__   ,      use_auth_token=lowerCAmelCase__   ,      revision=lowerCAmelCase__   ,      cache_dir=lowerCAmelCase__   ,      force_download=lowerCAmelCase__   ,      )
									a   :	Optional[int]			=       Path(lowerCAmelCase__   ).parent
									a   :	List[Any]			=       Path(lowerCAmelCase__   ).name
									a   :	int			=       OnnxRuntimeModel.load_model(lowerCAmelCase__   ,      provider=lowerCAmelCase__   ,      sess_options=lowerCAmelCase__   )
								return cls(model=lowerCAmelCase__   ,      **lowerCAmelCase__   )






							@classmethod
							def 					__a     (     cls   ,      lowerCAmelCase__   ,      lowerCAmelCase__ = True   ,      lowerCAmelCase__ = None   ,      lowerCAmelCase__ = None   ,      **lowerCAmelCase__   ,      )			->				List[str]:
								a   :	Any			=       None
								if len(str(lowerCAmelCase__   ).split("@"   )   ) == 2:
									a, a   :	Tuple			=       model_id.split("@"   )

								return cls._from_pretrained(
								    model_id=lowerCAmelCase__   ,      revision=lowerCAmelCase__   ,      cache_dir=lowerCAmelCase__   ,      force_download=lowerCAmelCase__   ,      use_auth_token=lowerCAmelCase__   ,      **lowerCAmelCase__   ,      )


