






import json
import os
from functools import lru_cache
from typing import List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


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


__UpperCAmelCase							:   Optional[Any]							=  {"vocab_file": "vocab.json", "merges_file": "merges.txt"}

# See all BART models at https://huggingface.co/models?filter=bart
__UpperCAmelCase							:   Optional[int]							=  {
    "vocab_file": {
        "facebook/bart-base": "https://huggingface.co/facebook/bart-base/resolve/main/vocab.json",
        "facebook/bart-large": "https://huggingface.co/facebook/bart-large/resolve/main/vocab.json",
        "facebook/bart-large-mnli": "https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json",
        "facebook/bart-large-cnn": "https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json",
        "facebook/bart-large-xsum": "https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json",
        "yjernite/bart_eli5": "https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json",
    },
    "merges_file": {
        "facebook/bart-base": "https://huggingface.co/facebook/bart-base/resolve/main/merges.txt",
        "facebook/bart-large": "https://huggingface.co/facebook/bart-large/resolve/main/merges.txt",
        "facebook/bart-large-mnli": "https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt",
        "facebook/bart-large-cnn": "https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt",
        "facebook/bart-large-xsum": "https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt",
        "yjernite/bart_eli5": "https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt",
    },
}

__UpperCAmelCase							:   Tuple							=  {
    "facebook/bart-base": 1_0_2_4,
    "facebook/bart-large": 1_0_2_4,
    "facebook/bart-large-mnli": 1_0_2_4,
    "facebook/bart-large-cnn": 1_0_2_4,
    "facebook/bart-large-xsum": 1_0_2_4,
    "yjernite/bart_eli5": 1_0_2_4,
}





@lru_cache()
def 					lowercase_ (    )     -> int:





				'''simple docstring'''


				snake_case__      :Dict         =   (
				    list(range(ord("!"      )	,	ord("~"      ) + 1      )      ) + list(range(ord("¡"      )	,	ord("¬"      ) + 1      )      ) + list(range(ord("®"      )	,	ord("ÿ"      ) + 1      )      )
				)
				snake_case__      :Union[str, Any]         =   bs[:]
				snake_case__      :Optional[int]         =   0
				for b in range(2**8      ):
								if b not in bs:
												bs.append(__snake_case      )
												cs.append(2**8 + n      )
												n += 1
				snake_case__      :List[str]         =   [chr(__snake_case      ) for n in cs]
				return dict(zip(__snake_case	,	__snake_case      )      )





def 					lowercase_ (    __snake_case     :							int      )     -> Any:





				'''simple docstring'''


				snake_case__      :Optional[Any]         =   set()
				snake_case__      :Any         =   word[0]
				for char in word[1:]:
								pairs.add((prev_char, char)      )
								snake_case__      :Tuple         =   char
				return pairs


class 					_snake_case						(			_A ):
	_A          =   VOCAB_FILES_NAMES
	_A          =   PRETRAINED_VOCAB_FILES_MAP
	_A          =   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	_A          =   ['input_ids', 'attention_mask']






	def __init__(		self					,UpperCamelCase					,UpperCamelCase					,UpperCamelCase="replace"					,UpperCamelCase="<s>"					,UpperCamelCase="</s>"					,UpperCamelCase="</s>"					,UpperCamelCase="<s>"					,UpperCamelCase="<unk>"					,UpperCamelCase="<pad>"					,UpperCamelCase="<mask>"					,UpperCamelCase=False					,**UpperCamelCase					,)						->    Optional[int]:
					snake_case__      :str         =   AddedToken(UpperCamelCase					,lstrip=UpperCamelCase					,rstrip=UpperCamelCase       ) if isinstance(UpperCamelCase					,UpperCamelCase       ) else bos_token
					snake_case__      :int         =   AddedToken(UpperCamelCase					,lstrip=UpperCamelCase					,rstrip=UpperCamelCase       ) if isinstance(UpperCamelCase					,UpperCamelCase       ) else eos_token
					snake_case__      :Any         =   AddedToken(UpperCamelCase					,lstrip=UpperCamelCase					,rstrip=UpperCamelCase       ) if isinstance(UpperCamelCase					,UpperCamelCase       ) else sep_token
					snake_case__      :List[str]         =   AddedToken(UpperCamelCase					,lstrip=UpperCamelCase					,rstrip=UpperCamelCase       ) if isinstance(UpperCamelCase					,UpperCamelCase       ) else cls_token
					snake_case__      :Tuple         =   AddedToken(UpperCamelCase					,lstrip=UpperCamelCase					,rstrip=UpperCamelCase       ) if isinstance(UpperCamelCase					,UpperCamelCase       ) else unk_token
					snake_case__      :Dict         =   AddedToken(UpperCamelCase					,lstrip=UpperCamelCase					,rstrip=UpperCamelCase       ) if isinstance(UpperCamelCase					,UpperCamelCase       ) else pad_token

					# Mask token behave like a normal word, i.e. include the space before it
					snake_case__      :List[Any]         =   AddedToken(UpperCamelCase					,lstrip=UpperCamelCase					,rstrip=UpperCamelCase       ) if isinstance(UpperCamelCase					,UpperCamelCase       ) else mask_token

					super().__init__(
					    errors=UpperCamelCase					,bos_token=UpperCamelCase					,eos_token=UpperCamelCase					,unk_token=UpperCamelCase					,sep_token=UpperCamelCase					,cls_token=UpperCamelCase					,pad_token=UpperCamelCase					,mask_token=UpperCamelCase					,add_prefix_space=UpperCamelCase					,**UpperCamelCase					,)

					with open(UpperCamelCase					,encoding="utf-8"       ) as vocab_handle:
									snake_case__      :Optional[int]         =   json.load(UpperCamelCase       )
					snake_case__      :Optional[Any]         =   {v: k for k, v in self.encoder.items()}
					snake_case__      :Dict         =   errors  # how to handle errors in decoding
					snake_case__      :int         =   bytes_to_unicode()
					snake_case__      :List[Any]         =   {v: k for k, v in self.byte_encoder.items()}
					with open(UpperCamelCase					,encoding="utf-8"       ) as merges_handle:
									snake_case__      :Union[str, Any]         =   merges_handle.read().split("\n"       )[1:-1]
					snake_case__      :List[str]         =   [tuple(merge.split()       ) for merge in bpe_merges]
					snake_case__      :Union[str, Any]         =   dict(zip(UpperCamelCase					,range(len(UpperCamelCase       )       )       )       )
					snake_case__      :Dict         =   {}
					snake_case__      :int         =   add_prefix_space

					# Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
					snake_case__      :str         =   re.compile(R"'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"       )






	@property
	def    lowerCAmelCase_							(		self       )						->    Any:
					return len(self.encoder       )






	def    lowerCAmelCase_							(		self       )						->    Optional[int]:
					return dict(self.encoder					,**self.added_tokens_encoder       )






	def    lowerCAmelCase_							(		self					,UpperCamelCase       )						->    int:
					if token in self.cache:
									return self.cache[token]
					snake_case__      :Dict         =   tuple(UpperCamelCase       )
					snake_case__      :List[Any]         =   get_pairs(UpperCamelCase       )

					if not pairs:
									return token

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

													if word[i] == first and i < len(UpperCamelCase       ) - 1 and word[i + 1] == second:
																	new_word.append(first + second       )
																	i += 2
													else:
																	new_word.append(word[i]       )
																	i += 1
									snake_case__      :Any         =   tuple(UpperCamelCase       )
									snake_case__      :Any         =   new_word
									if len(UpperCamelCase       ) == 1:
													break
									else:
													snake_case__      :Union[str, Any]         =   get_pairs(UpperCamelCase       )
					snake_case__      :Tuple         =   " ".join(UpperCamelCase       )
					snake_case__      :Optional[Any]         =   word
					return word






	def    lowerCAmelCase_							(		self					,UpperCamelCase       )						->    int:
					snake_case__      :str         =   []
					for token in re.findall(self.pat					,UpperCamelCase       ):
									snake_case__      :Any         =   "".join(
									    self.byte_encoder[b] for b in token.encode("utf-8"       )       )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
									bpe_tokens.extend(bpe_token for bpe_token in self.bpe(UpperCamelCase       ).split(" "       )       )
					return bpe_tokens






	def    lowerCAmelCase_							(		self					,UpperCamelCase       )						->    int:
					return self.encoder.get(UpperCamelCase					,self.encoder.get(self.unk_token       )       )






	def    lowerCAmelCase_							(		self					,UpperCamelCase       )						->    List[str]:
					return self.decoder.get(UpperCamelCase       )






	def    lowerCAmelCase_							(		self					,UpperCamelCase       )						->    int:
					snake_case__      :Tuple         =   "".join(UpperCamelCase       )
					snake_case__      :Optional[int]         =   bytearray([self.byte_decoder[c] for c in text]       ).decode("utf-8"					,errors=self.errors       )
					return text






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = None       )						->    Tuple[str]:
					if not os.path.isdir(UpperCamelCase       ):
									logger.error(f'Vocabulary path ({save_directory}) should be a directory'       )
									return
					snake_case__      :List[Any]         =   os.path.join(
					    UpperCamelCase					,(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]       )
					snake_case__      :str         =   os.path.join(
					    UpperCamelCase					,(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]       )

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

					snake_case__      :int         =   0
					with open(UpperCamelCase					,"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 UpperCamelCase       : 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!"       )
																	snake_case__      :Tuple         =   token_index
													writer.write(" ".join(UpperCamelCase       ) + "\n"       )
													index += 1

					return vocab_file, merge_file






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = None       )						->    List[int]:
					if token_ids_a is None:
									return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
					snake_case__      :Any         =   [self.cls_token_id]
					snake_case__      :Union[str, Any]         =   [self.sep_token_id]
					return cls + token_ids_a + sep + sep + token_ids_a + sep






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = None					,UpperCamelCase = False       )						->    List[int]:
					if already_has_special_tokens:
									return super().get_special_tokens_mask(
									    token_ids_a=UpperCamelCase					,token_ids_a=UpperCamelCase					,already_has_special_tokens=UpperCamelCase       )

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






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase = None       )						->    List[int]:
					snake_case__      :Union[str, Any]         =   [self.sep_token_id]
					snake_case__      :List[Any]         =   [self.cls_token_id]

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






	def    lowerCAmelCase_							(		self					,UpperCamelCase					,UpperCamelCase=False					,**UpperCamelCase       )						->    Optional[Any]:
					snake_case__      :Tuple         =   kwargs.pop("add_prefix_space"					,self.add_prefix_space       )
					if (is_split_into_words or add_prefix_space) and (len(UpperCamelCase       ) > 0 and not text[0].isspace()):
									snake_case__      :Any         =   " " + text
					return (text, kwargs)







import os
import sys
import unittest


__UpperCAmelCase							:   str							=  os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, "utils"))

import check_dummies  # noqa: E402
from check_dummies import create_dummy_files, create_dummy_object, find_backend, read_init  # noqa: E402


# Align TRANSFORMERS_PATH in check_dummies with the current path
__UpperCAmelCase							:   Tuple							=  os.path.join(git_repo_path, "src", "diffusers")


class 					_snake_case						(			unittest.TestCase ):






	def    lowerCAmelCase_							(		self       )						->    Union[str, Any]:
					snake_case__      :Tuple         =   find_backend("    if not is_torch_available():"       )
					self.assertEqual(UpperCamelCase					,"torch"       )

					# backend_with_underscore = find_backend("    if not is_tensorflow_text_available():")
					# self.assertEqual(backend_with_underscore, "tensorflow_text")

					snake_case__      :Tuple         =   find_backend("    if not (is_torch_available() and is_transformers_available()):"       )
					self.assertEqual(UpperCamelCase					,"torch_and_transformers"       )

					# double_backend_with_underscore = find_backend(
					#    "    if not (is_sentencepiece_available() and is_tensorflow_text_available()):"
					# )
					# self.assertEqual(double_backend_with_underscore, "sentencepiece_and_tensorflow_text")

					snake_case__      :str         =   find_backend(
					    "    if not (is_torch_available() and is_transformers_available() and is_onnx_available()):"       )
					self.assertEqual(UpperCamelCase					,"torch_and_transformers_and_onnx"       )






	def    lowerCAmelCase_							(		self       )						->    str:
					snake_case__      :int         =   read_init()
					# We don't assert on the exact list of keys to allow for smooth grow of backend-specific objects
					self.assertIn("torch"					,UpperCamelCase       )
					self.assertIn("torch_and_transformers"					,UpperCamelCase       )
					self.assertIn("flax_and_transformers"					,UpperCamelCase       )
					self.assertIn("torch_and_transformers_and_onnx"					,UpperCamelCase       )

					# Likewise, we can't assert on the exact content of a key
					self.assertIn("UNet2DModel"					,objects["torch"]       )
					self.assertIn("FlaxUNet2DConditionModel"					,objects["flax"]       )
					self.assertIn("StableDiffusionPipeline"					,objects["torch_and_transformers"]       )
					self.assertIn("FlaxStableDiffusionPipeline"					,objects["flax_and_transformers"]       )
					self.assertIn("LMSDiscreteScheduler"					,objects["torch_and_scipy"]       )
					self.assertIn("OnnxStableDiffusionPipeline"					,objects["torch_and_transformers_and_onnx"]       )






	def    lowerCAmelCase_							(		self       )						->    Any:
					snake_case__      :Union[str, Any]         =   create_dummy_object("CONSTANT"					,"'torch'"       )
					self.assertEqual(UpperCamelCase					,"\nCONSTANT = None\n"       )

					snake_case__      :Optional[Any]         =   create_dummy_object("function"					,"'torch'"       )
					self.assertEqual(
					    UpperCamelCase					,"\ndef function(*args, **kwargs):\n    requires_backends(function, 'torch')\n"       )

					snake_case__      :str         =   "\nclass FakeClass(metaclass=DummyObject):\n    _backends = 'torch'\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, 'torch')\n\n    @classmethod\n    def from_config(cls, *args, **kwargs):\n        requires_backends(cls, 'torch')\n\n    @classmethod\n    def from_pretrained(cls, *args, **kwargs):\n        requires_backends(cls, 'torch')\n"
					snake_case__      :List[str]         =   create_dummy_object("FakeClass"					,"'torch'"       )
					self.assertEqual(UpperCamelCase					,UpperCamelCase       )






	def    lowerCAmelCase_							(		self       )						->    List[Any]:
					snake_case__      :Tuple         =   "# This file is autogenerated by the command `make fix-copies`, do not edit.\nfrom ..utils import DummyObject, requires_backends\n\n\nCONSTANT = None\n\n\ndef function(*args, **kwargs):\n    requires_backends(function, [\"torch\"])\n\n\nclass FakeClass(metaclass=DummyObject):\n    _backends = [\"torch\"]\n\n    def __init__(self, *args, **kwargs):\n        requires_backends(self, [\"torch\"])\n\n    @classmethod\n    def from_config(cls, *args, **kwargs):\n        requires_backends(cls, [\"torch\"])\n\n    @classmethod\n    def from_pretrained(cls, *args, **kwargs):\n        requires_backends(cls, [\"torch\"])\n"
					snake_case__      :int         =   create_dummy_files({"torch": ["CONSTANT", "function", "FakeClass"]}       )
					self.assertEqual(dummy_files["torch"]					,UpperCamelCase       )