
import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


__lowerCamelCase								=					logging.get_logger(__name__)

__lowerCamelCase								=					{
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

__lowerCamelCase								=					{
    'vocab_file': {'openai-gpt': 'https://huggingface.co/openai-gpt/resolve/main/vocab.json'},
    'merges_file': {'openai-gpt': 'https://huggingface.co/openai-gpt/resolve/main/merges.txt'},
}

__lowerCamelCase								=					{
    'openai-gpt': 5_1_2,
}




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE  )   ->      str:

							"""simple docstring"""






							_UpperCAmelCase  :			Optional[Any]	       =		text.strip()
							if not text:
														return []
							_UpperCAmelCase  :			str	       =		text.split()
							return tokens
class    _UpperCamelCase(   SCREAMING_SNAKE_CASE				):

						def __init__(					self		:			str	,						_lowerCamelCase		:			Union[str, Any]=True	,						_lowerCamelCase		:			List[Any]=None	,						_lowerCamelCase		:			List[Any]=True	,						_lowerCamelCase		:			List[str]=None	,						_lowerCamelCase		:			List[str]=True	,						):





													if never_split is None:
																				_UpperCAmelCase  :			Any	       =		[]
													_UpperCAmelCase  :			Optional[Any]	       =		do_lower_case
													_UpperCAmelCase  :			Any	       =		set(_lowerCamelCase			)
													_UpperCAmelCase  :			Union[str, Any]	       =		tokenize_chinese_chars
													_UpperCAmelCase  :			str	       =		strip_accents
													_UpperCAmelCase  :			Optional[int]	       =		do_split_on_punc

						def 						a__ (					self		:			List[str]	,						_lowerCamelCase		:			List[Any]	,						_lowerCamelCase		:			Optional[Any]=None			):
													_UpperCAmelCase  :			List[Any]	       =		self.never_split.union(set(_lowerCamelCase			)			) if never_split else self.never_split
													_UpperCAmelCase  :			List[str]	       =		self._clean_text(_lowerCamelCase			)

													# This was added on November 1st, 2018 for the multilingual and Chinese
													# models. This is also applied to the English models now, but it doesn't
													# matter since the English models were not trained on any Chinese data
													# and generally don't have any Chinese data in them (there are Chinese
													# characters in the vocabulary because Wikipedia does have some Chinese
													# words in the English Wikipedia.).
													if self.tokenize_chinese_chars:
																				_UpperCAmelCase  :			List[str]	       =		self._tokenize_chinese_chars(_lowerCamelCase			)
													# prevents treating the same character with different unicode codepoints as different characters
													_UpperCAmelCase  :			List[Any]	       =		unicodedata.normalize("NFC"	,						_lowerCamelCase			)
													_UpperCAmelCase  :			Any	       =		whitespace_tokenize(_lowerCamelCase			)
													_UpperCAmelCase  :			List[str]	       =		[]
													for token in orig_tokens:
																				if token not in never_split:
																											if self.do_lower_case:
																																		_UpperCAmelCase  :			Dict	       =		token.lower()
																																		if self.strip_accents is not False:
																																									_UpperCAmelCase  :			Dict	       =		self._run_strip_accents(_lowerCamelCase			)
																											elif self.strip_accents:
																																		_UpperCAmelCase  :			Optional[Any]	       =		self._run_strip_accents(_lowerCamelCase			)
																				split_tokens.extend(self._run_split_on_punc(_lowerCamelCase	,						_lowerCamelCase			)			)

													_UpperCAmelCase  :			int	       =		whitespace_tokenize(" ".join(_lowerCamelCase			)			)
													return output_tokens

						def 						a__ (					self		:			List[Any]	,						_lowerCamelCase		:			int			):
													_UpperCAmelCase  :			List[str]	       =		unicodedata.normalize("NFD"	,						_lowerCamelCase			)
													_UpperCAmelCase  :			List[str]	       =		[]
													for char in text:
																				_UpperCAmelCase  :			int	       =		unicodedata.category(_lowerCamelCase			)
																				if cat == "Mn":
																											continue
																				output.append(_lowerCamelCase			)
													return "".join(_lowerCamelCase			)

						def 						a__ (					self		:			Union[str, Any]	,						_lowerCamelCase		:			List[Any]	,						_lowerCamelCase		:			int=None			):





													if not self.do_split_on_punc or (never_split is not None and text in never_split):
																				return [text]
													_UpperCAmelCase  :			List[str]	       =		list(_lowerCamelCase			)
													_UpperCAmelCase  :			Union[str, Any]	       =		0
													_UpperCAmelCase  :			List[str]	       =		True
													_UpperCAmelCase  :			int	       =		[]
													while i < len(_lowerCamelCase			):
																				_UpperCAmelCase  :			List[Any]	       =		chars[i]
																				if _is_punctuation(_lowerCamelCase			):
																											output.append([char]			)
																											_UpperCAmelCase  :			str	       =		True
																				else:
																											if start_new_word:
																																		output.append([]			)
																											_UpperCAmelCase  :			str	       =		False
																											output[-1].append(_lowerCamelCase			)
																				i += 1

													return ["".join(_lowerCamelCase			) for x in output]

						def 						a__ (					self		:			int	,						_lowerCamelCase		:			Tuple			):
													_UpperCAmelCase  :			Any	       =		[]
													for char in text:
																				_UpperCAmelCase  :			List[Any]	       =		ord(_lowerCamelCase			)
																				if self._is_chinese_char(_lowerCamelCase			):
																											output.append(" "			)
																											output.append(_lowerCamelCase			)
																											output.append(" "			)
																				else:
																											output.append(_lowerCamelCase			)
													return "".join(_lowerCamelCase			)

						def 						a__ (					self		:			List[str]	,						_lowerCamelCase		:			str			):





													if (
													    (cp >= 0X4_e_0_0 and cp <= 0X9_f_f_f)
													    or (cp >= 0X3_4_0_0 and cp <= 0X4_d_b_f)  #
													    or (cp >= 0X2_0_0_0_0 and cp <= 0X2_a_6_d_f)  #
													    or (cp >= 0X2_a_7_0_0 and cp <= 0X2_b_7_3_f)  #
													    or (cp >= 0X2_b_7_4_0 and cp <= 0X2_b_8_1_f)  #
													    or (cp >= 0X2_b_8_2_0 and cp <= 0X2_c_e_a_f)  #
													    or (cp >= 0Xf_9_0_0 and cp <= 0Xf_a_f_f)
													    or (cp >= 0X2_f_8_0_0 and cp <= 0X2_f_a_1_f)  #
													):  #
																				return True

													return False






						def 						a__ (					self		:			int	,						_lowerCamelCase		:			Union[str, Any]			):
													_UpperCAmelCase  :			List[Any]	       =		[]
													for char in text:
																				_UpperCAmelCase  :			Dict	       =		ord(_lowerCamelCase			)
																				if cp == 0 or cp == 0Xf_f_f_d or _is_control(_lowerCamelCase			):
																											continue
																				if _is_whitespace(_lowerCamelCase			):
																											output.append(" "			)
																				else:
																											output.append(_lowerCamelCase			)
													return "".join(_lowerCamelCase			)







def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE  )   ->      List[Any]:

							"""simple docstring"""






							_UpperCAmelCase  :			Optional[Any]	       =		set()
							_UpperCAmelCase  :			int	       =		word[0]
							for char in word[1:]:
														pairs.add((prev_char, char)  )
														_UpperCAmelCase  :			Union[str, Any]	       =		char
							return pairs




def 		lowerCAmelCase_     (     _SCREAMING_SNAKE_CASE  )   ->      Tuple:

							"""simple docstring"""






							_UpperCAmelCase  :			List[str]	       =		text.replace("—"      ,	"-"  )
							_UpperCAmelCase  :			Optional[int]	       =		text.replace("–"      ,	"-"  )
							_UpperCAmelCase  :			List[str]	       =		text.replace("―"      ,	"-"  )
							_UpperCAmelCase  :			int	       =		text.replace("…"      ,	"..."  )
							_UpperCAmelCase  :			str	       =		text.replace("´"      ,	"'"  )
							_UpperCAmelCase  :			List[Any]	       =		re.sub(R"(-+|~+|!+|\"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)"      ,	R" \1 "      ,	UpperCAmelCase__  )
							_UpperCAmelCase  :			Tuple	       =		re.sub(R"\s*\n\s*"      ,	" \n "      ,	UpperCAmelCase__  )
							_UpperCAmelCase  :			Tuple	       =		re.sub(R"[^\S\n]+"      ,	" "      ,	UpperCAmelCase__  )
							return text.strip()


class    _UpperCamelCase(   SCREAMING_SNAKE_CASE				):
						__A:						Optional[int]				   =      VOCAB_FILES_NAMES
						__A:						int				   =      PRETRAINED_VOCAB_FILES_MAP
						__A:						Any				   =      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						__A:						int				   =      ["""input_ids""", """attention_mask"""]

						def __init__(					self		:			List[Any]	,						_lowerCamelCase		:			List[str]	,						_lowerCamelCase		:			int	,						_lowerCamelCase		:			List[str]="<unk>"	,						**_lowerCamelCase		:			int			):
													super().__init__(unk_token=_lowerCamelCase	,						**_lowerCamelCase			)

													try:
																				import ftfy
																				from spacy.lang.en import English

																				_UpperCAmelCase  :			Tuple	       =		English()
																				_UpperCAmelCase  :			List[Any]	       =		_nlp.tokenizer
																				_UpperCAmelCase  :			Optional[Any]	       =		ftfy.fix_text
													except ImportError:
																				logger.warning("ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy."			)
																				_UpperCAmelCase  :			List[Any]	       =		BasicTokenizer(do_lower_case=_lowerCamelCase			)
																				_UpperCAmelCase  :			List[str]	       =		None

													with open(_lowerCamelCase	,						encoding="utf-8"			) as vocab_handle:
																				_UpperCAmelCase  :			int	       =		json.load(_lowerCamelCase			)
													_UpperCAmelCase  :			Union[str, Any]	       =		{v: k for k, v in self.encoder.items()}
													with open(_lowerCamelCase	,						encoding="utf-8"			) as merges_handle:
																				_UpperCAmelCase  :			List[str]	       =		merges_handle.read().split("\n"			)[1:-1]
													_UpperCAmelCase  :			Optional[int]	       =		[tuple(merge.split()			) for merge in merges]
													_UpperCAmelCase  :			int	       =		dict(zip(_lowerCamelCase	,						range(len(_lowerCamelCase			)			)			)			)
													_UpperCAmelCase  :			Union[str, Any]	       =		{}

						@property
						def 						a__ (					self		:			List[str]			):
													return True

						@property
						def 						a__ (					self		:			str			):
													return len(self.encoder			)

						def 						a__ (					self		:			List[Any]			):
													return dict(self.encoder	,						**self.added_tokens_encoder			)

						def 						a__ (					self		:			int	,						_lowerCamelCase		:			Optional[int]			):
													_UpperCAmelCase  :			str	       =		tuple(token[:-1]			) + (token[-1] + """</w>""",)
													if token in self.cache:
																				return self.cache[token]
													_UpperCAmelCase  :			Union[str, Any]	       =		get_pairs(_lowerCamelCase			)

													if not pairs:
																				return token + "</w>"

													while True:
																				_UpperCAmelCase  :			Optional[int]	       =		min(_lowerCamelCase	,						key=lambda _lowerCamelCase			: self.bpe_ranks.get(_lowerCamelCase	,						float("inf"			)			)			)
																				if bigram not in self.bpe_ranks:
																											break
																				_UpperCAmelCase  :			int	       =		bigram
																				_UpperCAmelCase  :			Optional[int]	       =		[]
																				_UpperCAmelCase  :			Tuple	       =		0
																				while i < len(_lowerCamelCase			):
																											try:
																																		_UpperCAmelCase  :			Union[str, Any]	       =		word.index(_lowerCamelCase	,						_lowerCamelCase			)
																											except ValueError:
																																		new_word.extend(word[i:]			)
																																		break
																											else:
																																		new_word.extend(word[i:j]			)
																																		_UpperCAmelCase  :			List[str]	       =		j

																											if word[i] == first and i < len(_lowerCamelCase			) - 1 and word[i + 1] == second:
																																		new_word.append(first + second			)
																																		i += 2
																											else:
																																		new_word.append(word[i]			)
																																		i += 1
																				_UpperCAmelCase  :			Dict	       =		tuple(_lowerCamelCase			)
																				_UpperCAmelCase  :			Dict	       =		new_word
																				if len(_lowerCamelCase			) == 1:
																											break
																				else:
																											_UpperCAmelCase  :			Tuple	       =		get_pairs(_lowerCamelCase			)
													_UpperCAmelCase  :			str	       =		""" """.join(_lowerCamelCase			)
													if word == "\n  </w>":
																				_UpperCAmelCase  :			Union[str, Any]	       =		"""\n</w>"""
													_UpperCAmelCase  :			Tuple	       =		word
													return word

						def 						a__ (					self		:			int	,						_lowerCamelCase		:			Optional[int]			):
													_UpperCAmelCase  :			List[str]	       =		[]
													if self.fix_text is None:
																				# Using BERT's BasicTokenizer
																				_UpperCAmelCase  :			Union[str, Any]	       =		self.nlp.tokenize(_lowerCamelCase			)
																				for token in text:
																											split_tokens.extend(list(self.bpe(_lowerCamelCase			).split(" "			)			)			)
													else:
																				# Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
																				_UpperCAmelCase  :			Union[str, Any]	       =		self.nlp(text_standardize(self.fix_text(_lowerCamelCase			)			)			)
																				for token in text:
																											split_tokens.extend(list(self.bpe(token.text.lower()			).split(" "			)			)			)
													return split_tokens

						def 						a__ (					self		:			Union[str, Any]	,						_lowerCamelCase		:			Union[str, Any]			):
													return self.encoder.get(_lowerCamelCase	,						self.encoder.get(self.unk_token			)			)

						def 						a__ (					self		:			Union[str, Any]	,						_lowerCamelCase		:			Any			):
													return self.decoder.get(_lowerCamelCase	,						self.unk_token			)

						def 						a__ (					self		:			Dict	,						_lowerCamelCase		:			List[str]			):
													_UpperCAmelCase  :			Optional[Any]	       =		"""""".join(_lowerCamelCase			).replace("</w>"	,						" "			).strip()
													return out_string






						def 						a__ (					self		:			int	,						_lowerCamelCase		:			str	,						_lowerCamelCase		:			Optional[str] = None			):





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

													with open(_lowerCamelCase	,						"w"	,						encoding="utf-8"			) as f:
																				f.write(json.dumps(self.encoder	,						indent=2	,						sort_keys=_lowerCamelCase	,						ensure_ascii=_lowerCamelCase			) + "\n"			)

													_UpperCAmelCase  :			str	       =		0
													with open(_lowerCamelCase	,						"w"	,						encoding="utf-8"			) as writer:
																				writer.write("#version: 0.2\n"			)
																				for bpe_tokens, token_index in sorted(self.bpe_ranks.items()	,						key=lambda _lowerCamelCase			: kv[1]			):
																											if index != token_index:
																																		logger.warning(
																																		    f"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
																																		    " Please check that the tokenizer is not corrupted!"			)
																																		_UpperCAmelCase  :			Dict	       =		token_index
																											writer.write(" ".join(_lowerCamelCase			) + "\n"			)
																											index += 1

													return vocab_file, merge_file












"""simple docstring"""



import os

from packaging import version

from .. import __version__
from .accelerate_utils import apply_forward_hook
from .constants import (
    CONFIG_NAME,
    DEPRECATED_REVISION_ARGS,
    DIFFUSERS_CACHE,
    DIFFUSERS_DYNAMIC_MODULE_NAME,
    FLAX_WEIGHTS_NAME,
    HF_MODULES_CACHE,
    HUGGINGFACE_CO_RESOLVE_ENDPOINT,
    ONNX_EXTERNAL_WEIGHTS_NAME,
    ONNX_WEIGHTS_NAME,
    SAFETENSORS_WEIGHTS_NAME,
    TEXT_ENCODER_ATTN_MODULE,
    WEIGHTS_NAME,
)
from .deprecation_utils import deprecate
from .doc_utils import replace_example_docstring
from .dynamic_modules_utils import get_class_from_dynamic_module
from .hub_utils import (
    HF_HUB_OFFLINE,
    _add_variant,
    _get_model_file,
    extract_commit_hash,
    http_user_agent,
)
from .import_utils import (
    BACKENDS_MAPPING,
    ENV_VARS_TRUE_AND_AUTO_VALUES,
    ENV_VARS_TRUE_VALUES,
    USE_JAX,
    USE_TF,
    USE_TORCH,
    DummyObject,
    OptionalDependencyNotAvailable,
    is_accelerate_available,
    is_accelerate_version,
    is_bsa_available,
    is_flax_available,
    is_ftfy_available,
    is_inflect_available,
    is_invisible_watermark_available,
    is_k_diffusion_available,
    is_k_diffusion_version,
    is_librosa_available,
    is_note_seq_available,
    is_omegaconf_available,
    is_onnx_available,
    is_safetensors_available,
    is_scipy_available,
    is_tensorboard_available,
    is_tf_available,
    is_torch_available,
    is_torch_version,
    is_torchsde_available,
    is_transformers_available,
    is_transformers_version,
    is_unidecode_available,
    is_wandb_available,
    is_xformers_available,
    requires_backends,
)
from .logging import get_logger
from .outputs import BaseOutput
from .pil_utils import PIL_INTERPOLATION, numpy_to_pil, pt_to_pil
from .torch_utils import is_compiled_module, randn_tensor


if is_torch_available():
						from .testing_utils import (
						    floats_tensor,
						    load_hf_numpy,
						    load_image,
						    load_numpy,
						    load_pt,
						    nightly,
						    parse_flag_from_env,
						    print_tensor_test,
						    require_torch_a,
						    require_torch_gpu,
						    skip_mps,
						    slow,
						    torch_all_close,
						    torch_device,
						)
						from .torch_utils import maybe_allow_in_graph

from .testing_utils import export_to_gif, export_to_video


A_     =				get_logger(__name__)




def   _lowerCAmelCase		(				UpperCAmelCase__							:       Dict   )		->Union[str, Any]:
						if version.parse(UpperCAmelCase__   ) < version.parse(UpperCAmelCase__   ):
												if "dev" in min_version:
																		A__    :	Optional[int]			=	(
																		    """This example requires a source install from HuggingFace diffusers (see """
																		    """`https://huggingface.co/docs/diffusers/installation#install-from-source`),"""
																		)
												else:
																		A__    :	Union[str, Any]			=	f'This example requires a minimum version of {min_version},'
												error_message += f' but the version found is {__version__}.\n'
												raise ImportError(UpperCAmelCase__   )






