



import os
import re
import unicodedata
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


UpperCAmelCase							=       logging.get_logger(__name__)
UpperCAmelCase							=       {"""vocab_file""": """spiece.model"""}

UpperCAmelCase							=       {
    """vocab_file""": {
        """google/fnet-base""": """https://huggingface.co/google/fnet-base/resolve/main/spiece.model""",
        """google/fnet-large""": """https://huggingface.co/google/fnet-large/resolve/main/spiece.model""",
    },
}

UpperCAmelCase							=       {
    """google/fnet-base""": 512,
    """google/fnet-large""": 512,
}

UpperCAmelCase							=       """▁"""

class 			__snake_case   ( SCREAMING_SNAKE_CASE):







							'''simple docstring'''






							UpperCamelCase__					:   Tuple       =       VOCAB_FILES_NAMES
							UpperCamelCase__					:   List[str]       =       PRETRAINED_VOCAB_FILES_MAP
							UpperCamelCase__					:   List[Any]       =       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							UpperCamelCase__					:   List[str]       =       ["""input_ids""", """token_type_ids"""]


							def __init__(				self	,	a_	,	a_=False	,	a_=True	,	a_=True	,	a_="<unk>"	,	a_="[SEP]"	,	a_="<pad>"	,	a_="[CLS]"	,	a_="[MASK]"	,	a_ = None	,	**a_	,	):
											# 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__					   =		(
											    AddedToken(a_	,	lstrip=a_	,	rstrip=a_	,	normalized=a_		)
											    if isinstance(a_	,	a_		)
											    else mask_token
											)

											a__					   =		{} if sp_model_kwargs is None else sp_model_kwargs

											super().__init__(
											    do_lower_case=a_	,	remove_space=a_	,	keep_accents=a_	,	unk_token=a_	,	sep_token=a_	,	pad_token=a_	,	cls_token=a_	,	mask_token=a_	,	sp_model_kwargs=self.sp_model_kwargs	,	**a_	,	)

											a__					   =		do_lower_case
											a__					   =		remove_space
											a__					   =		keep_accents
											a__					   =		vocab_file

											a__					   =		spm.SentencePieceProcessor(**self.sp_model_kwargs		)
											self.sp_model.Load(a_		)


							@property
							def _a   (				self		):
											return len(self.sp_model		)


							def _a   (				self		):
											a__					   =		{self.convert_ids_to_tokens(a_		): i for i in range(self.vocab_size		)}
											vocab.update(self.added_tokens_encoder		)
											return vocab


							def __getstate__(				self		):
											a__					   =		self.__dict__.copy()
											a__					   =		None
											return state


							def __setstate__(				self	,	a_		):
											a__					   =		d

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

											a__					   =		spm.SentencePieceProcessor(**self.sp_model_kwargs		)
											self.sp_model.Load(self.vocab_file		)


							def _a   (				self	,	a_		):
											if self.remove_space:
															a__					   =		""" """.join(inputs.strip().split()		)
											else:
															a__					   =		inputs
											a__					   =		outputs.replace("""``"""	,	"""\""""		).replace("""''"""	,	"""\""""		)

											if not self.keep_accents:
															a__					   =		unicodedata.normalize("""NFKD"""	,	a_		)
															a__					   =		"""""".join([c for c in outputs if not unicodedata.combining(a_		)]		)
											if self.do_lower_case:
															a__					   =		outputs.lower()

											return outputs


							def _a   (				self	,	a_		):
											a__					   =		self.preprocess_text(a_		)
											a__					   =		self.sp_model.encode(a_	,	out_type=a_		)
											a__					   =		[]
											for piece in pieces:
															if len(a_		) > 1 and piece[-1] == str(""","""		) and piece[-2].isdigit():
																			a__					   =		self.sp_model.EncodeAsPieces(piece[:-1].replace(a_	,	""""""		)		)
																			if piece[0] != SPIECE_UNDERLINE and cur_pieces[0][0] == SPIECE_UNDERLINE:
																							if len(cur_pieces[0]		) == 1:
																											a__					   =		cur_pieces[1:]
																							else:
																											a__					   =		cur_pieces[0][1:]
																			cur_pieces.append(piece[-1]		)
																			new_pieces.extend(a_		)
															else:
																			new_pieces.append(a_		)

											return new_pieces


							def _a   (				self	,	a_		):
											return self.sp_model.PieceToId(a_		)


							def _a   (				self	,	a_		):
											return self.sp_model.IdToPiece(a_		)


							def _a   (				self	,	a_		):
											a__					   =		[]
											a__					   =		""""""
											a__					   =		False
											for token in tokens:
															# make sure that special tokens are not decoded using sentencepiece model
															if token in self.all_special_tokens:
																			if not prev_is_special:
																							out_string += " "
																			out_string += self.sp_model.decode(a_		) + token
																			a__					   =		True
																			a__					   =		[]
															else:
																			current_sub_tokens.append(a_		)
																			a__					   =		False
											out_string += self.sp_model.decode(a_		)
											return out_string.strip()


							def _a   (				self	,	a_	,	a_ = False	,	a_ = None	,	a_ = True	,	**a_	,	):
											a__					   =		kwargs.pop("""use_source_tokenizer"""	,	a_		)

											a__					   =		self.convert_ids_to_tokens(a_	,	skip_special_tokens=a_		)

											# To avoid mixing byte-level and unicode for byte-level BPT
											# we need to build string separately for added tokens and byte-level tokens
											# cf. https://github.com/huggingface/transformers/issues/1133
											a__					   =		[]
											a__					   =		[]
											for token in filtered_tokens:
															if skip_special_tokens and token in self.all_special_ids:
																			continue
															if token in self.added_tokens_encoder:
																			if current_sub_text:
																							sub_texts.append(self.convert_tokens_to_string(a_		)		)
																							a__					   =		[]
																			sub_texts.append(a_		)
															else:
																			current_sub_text.append(a_		)
											if current_sub_text:
															sub_texts.append(self.convert_tokens_to_string(a_		)		)

											# Mimic the behavior of the Rust tokenizer:
											# No space after <unk>
											if spaces_between_special_tokens:
															a__					   =		re.sub(r"""(<unk>) """	,	r"""\1"""	,	""" """.join(a_		)		)
											else:
															a__					   =		"""""".join(a_		)

											a__					   =		(
											    clean_up_tokenization_spaces
											    if clean_up_tokenization_spaces is not None
											    else self.clean_up_tokenization_spaces
											)
											if clean_up_tokenization_spaces:
															a__					   =		self.clean_up_tokenization(a_		)
															return clean_text
											else:
															return text


							def _a   (				self	,	a_	,	a_ = None		):
											a__					   =		[self.sep_token_id]
											a__					   =		[self.cls_token_id]
											if token_ids_a is None:
															return cls + token_ids_a + sep
											return cls + token_ids_a + sep + token_ids_a + sep


							def _a   (				self	,	a_	,	a_ = None	,	a_ = False		):

											if already_has_special_tokens:
															return super().get_special_tokens_mask(
															    token_ids_a=a_	,	token_ids_a=a_	,	already_has_special_tokens=a_		)

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


							def _a   (				self	,	a_	,	a_ = None		):
											a__					   =		[self.sep_token_id]
											a__					   =		[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		) * [0] + len(token_ids_a + sep		) * [1]







							def _a   (				self	,	a_	,	a_ = None		):
											if not os.path.isdir(a_		):
															logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''		)
															return
											a__					   =		os.path.join(
											    a_	,	(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]		)

											if os.path.abspath(self.vocab_file		) != os.path.abspath(a_		) and os.path.isfile(self.vocab_file		):
															copyfile(self.vocab_file	,	a_		)
											elif not os.path.isfile(self.vocab_file		):
															with open(a_	,	"""wb"""		) as fi:
																			a__					   =		self.sp_model.serialized_model_proto()
																			fi.write(a_		)

											return (out_vocab_file,)





import tempfile
import unittest

from diffusers import (
    DDIMScheduler,
    DDPMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    PNDMScheduler,
    logging,
)
from diffusers.configuration_utils import ConfigMixin, register_to_config
from diffusers.utils.testing_utils import CaptureLogger







class 					snake_case__(				UpperCAmelCase__				):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__   :	Union[str, Any]       =				"""config.json"""




						@register_to_config
						def __init__(					self ,    __lowercase=2 ,    __lowercase=5 ,    __lowercase=(2, 5) ,    __lowercase="for diffusion" ,    __lowercase=[1, 3] ,    )       ->     int:
										pass







class 					snake_case__(				UpperCAmelCase__				):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__   :	int       =				"""config.json"""




						@register_to_config
						def __init__(					self ,    __lowercase=2 ,    __lowercase=5 ,    __lowercase=(2, 5) ,    __lowercase="for diffusion" ,    __lowercase=[1, 3] ,    )       ->     Optional[Any]:
										pass







class 					snake_case__(				UpperCAmelCase__				):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__   :	int       =				"""config.json"""




						@register_to_config
						def __init__(					self ,    __lowercase=2 ,    __lowercase=5 ,    __lowercase=(2, 5) ,    __lowercase="for diffusion" ,    __lowercase=[1, 3] ,    __lowercase=[1, 3] ,    )       ->     List[str]:
										pass







class 					snake_case__(				UpperCAmelCase__				):







						'''simple docstring'''




						SCREAMING_SNAKE_CASE__   :	Optional[Any]       =				"""config.json"""




						@register_to_config
						def __init__(					self ,    __lowercase=2 ,    __lowercase=5 ,    __lowercase=(2, 5) ,    __lowercase="for diffusion" ,    __lowercase=[1, 5] ,    __lowercase=[5, 4] ,    )       ->     List[str]:
										pass







class 					snake_case__(				unittest.TestCase				):







						'''simple docstring'''




						def     lowercase_			(					self				)       ->     str:
										with self.assertRaises(__lowercase				):
														ConfigMixin.load_config('''dummy_path'''				)




						def     lowercase_			(					self				)       ->     Optional[int]:
										lowerCAmelCase_			:   List[str]     					=					SampleObject()
										lowerCAmelCase_			:   Dict     					=					obj.config
										assert config["a"] == 2
										assert config["b"] == 5
										assert config["c"] == (2, 5)
										assert config["d"] == "for diffusion"
										assert config["e"] == [1, 3]

										# init ignore private arguments
										lowerCAmelCase_			:   List[str]     					=					SampleObject(_name_or_path='''lalala'''				)
										lowerCAmelCase_			:   Dict     					=					obj.config
										assert config["a"] == 2
										assert config["b"] == 5
										assert config["c"] == (2, 5)
										assert config["d"] == "for diffusion"
										assert config["e"] == [1, 3]

										# can override default
										lowerCAmelCase_			:   List[Any]     					=					SampleObject(c=6				)
										lowerCAmelCase_			:   Optional[int]     					=					obj.config
										assert config["a"] == 2
										assert config["b"] == 5
										assert config["c"] == 6
										assert config["d"] == "for diffusion"
										assert config["e"] == [1, 3]

										# can use positional arguments.
										lowerCAmelCase_			:   Optional[Any]     					=					SampleObject(1 ,    c=6				)
										lowerCAmelCase_			:   Dict     					=					obj.config
										assert config["a"] == 1
										assert config["b"] == 5
										assert config["c"] == 6
										assert config["d"] == "for diffusion"
										assert config["e"] == [1, 3]




						def     lowercase_			(					self				)       ->     int:
										lowerCAmelCase_			:   List[Any]     					=					SampleObject()
										lowerCAmelCase_			:   str     					=					obj.config

										assert config["a"] == 2
										assert config["b"] == 5
										assert config["c"] == (2, 5)
										assert config["d"] == "for diffusion"
										assert config["e"] == [1, 3]

										with tempfile.TemporaryDirectory() as tmpdirname:
														obj.save_config(__lowercase				)
														lowerCAmelCase_			:   int     					=					SampleObject.from_config(SampleObject.load_config(__lowercase				)				)
														lowerCAmelCase_			:   Optional[Any]     					=					new_obj.config

										# unfreeze configs
										lowerCAmelCase_			:   str     					=					dict(__lowercase				)
										lowerCAmelCase_			:   Any     					=					dict(__lowercase				)

										assert config.pop('''c'''				) == (2, 5)  # instantiated as tuple
										assert new_config.pop('''c'''				) == [2, 5]  # saved & loaded as list because of json
										config.pop('''_use_default_values'''				)
										assert config == new_config




						def     lowercase_			(					self				)       ->     List[str]:
										lowerCAmelCase_			:   Optional[Any]     					=					logging.get_logger('''diffusers.configuration_utils'''				)
										# 30 for warning
										logger.setLevel(3_0				)

										with CaptureLogger(__lowercase				) as cap_logger:
														lowerCAmelCase_			:   int     					=					DDIMScheduler.from_pretrained(
														    '''hf-internal-testing/tiny-stable-diffusion-torch''' ,    subfolder='''scheduler'''				)

										assert ddim.__class__ == DDIMScheduler
										# no warning should be thrown
										assert cap_logger.out == ""




						def     lowercase_			(					self				)       ->     Optional[Any]:
										lowerCAmelCase_			:   List[Any]     					=					logging.get_logger('''diffusers.configuration_utils'''				)
										# 30 for warning
										logger.setLevel(3_0				)

										with CaptureLogger(__lowercase				) as cap_logger:
														lowerCAmelCase_			:   Any     					=					EulerDiscreteScheduler.from_pretrained(
														    '''hf-internal-testing/tiny-stable-diffusion-torch''' ,    subfolder='''scheduler'''				)

										assert euler.__class__ == EulerDiscreteScheduler
										# no warning should be thrown
										assert cap_logger.out == ""




						def     lowercase_			(					self				)       ->     Any:
										lowerCAmelCase_			:   Tuple     					=					logging.get_logger('''diffusers.configuration_utils'''				)
										# 30 for warning
										logger.setLevel(3_0				)

										with CaptureLogger(__lowercase				) as cap_logger:
														lowerCAmelCase_			:   Optional[int]     					=					EulerAncestralDiscreteScheduler.from_pretrained(
														    '''hf-internal-testing/tiny-stable-diffusion-torch''' ,    subfolder='''scheduler'''				)

										assert euler.__class__ == EulerAncestralDiscreteScheduler
										# no warning should be thrown
										assert cap_logger.out == ""




						def     lowercase_			(					self				)       ->     List[Any]:
										lowerCAmelCase_			:   Any     					=					logging.get_logger('''diffusers.configuration_utils'''				)
										# 30 for warning
										logger.setLevel(3_0				)

										with CaptureLogger(__lowercase				) as cap_logger:
														lowerCAmelCase_			:   Dict     					=					PNDMScheduler.from_pretrained(
														    '''hf-internal-testing/tiny-stable-diffusion-torch''' ,    subfolder='''scheduler'''				)

										assert pndm.__class__ == PNDMScheduler
										# no warning should be thrown
										assert cap_logger.out == ""




						def     lowercase_			(					self				)       ->     Tuple:
										lowerCAmelCase_			:   Optional[int]     					=					logging.get_logger('''diffusers.configuration_utils'''				)
										# 30 for warning
										logger.setLevel(3_0				)

										with CaptureLogger(__lowercase				) as cap_logger:
														lowerCAmelCase_			:   int     					=					DDPMScheduler.from_pretrained(
														    '''hf-internal-testing/tiny-stable-diffusion-torch''' ,    subfolder='''scheduler''' ,    prediction_type='''sample''' ,    beta_end=8 ,    )

										with CaptureLogger(__lowercase				) as cap_logger_a:
														lowerCAmelCase_			:   Tuple     					=					DDPMScheduler.from_pretrained('''google/ddpm-celebahq-256''' ,    beta_start=8_8				)

										assert ddpm.__class__ == DDPMScheduler
										assert ddpm.config.prediction_type == "sample"
										assert ddpm.config.beta_end == 8
										assert ddpm_a.config.beta_start == 8_8

										# no warning should be thrown
										assert cap_logger.out == ""
										assert cap_logger_a.out == ""




						def     lowercase_			(					self				)       ->     int:
										lowerCAmelCase_			:   Tuple     					=					logging.get_logger('''diffusers.configuration_utils'''				)
										# 30 for warning
										logger.setLevel(3_0				)

										with CaptureLogger(__lowercase				) as cap_logger:
														lowerCAmelCase_			:   List[str]     					=					DPMSolverMultistepScheduler.from_pretrained(
														    '''hf-internal-testing/tiny-stable-diffusion-torch''' ,    subfolder='''scheduler'''				)

										assert dpm.__class__ == DPMSolverMultistepScheduler
										# no warning should be thrown
										assert cap_logger.out == ""




						def     lowercase_			(					self				)       ->     Optional[Any]:
										# let's first save a config that should be in the form
										#    a=2,
										#    b=5,
										#    c=(2, 5),
										#    d="for diffusion",
										#    e=[1, 3],

										lowerCAmelCase_			:   Dict     					=					SampleObject()

										lowerCAmelCase_			:   Optional[int]     					=					{k: v for k, v in config.config.items() if not k.startswith('''_'''				)}

										# make sure that default config has all keys in `_use_default_values`
										assert set(config_dict.keys()				) == set(config.config._use_default_values				)

										with tempfile.TemporaryDirectory() as tmpdirname:
														config.save_config(__lowercase				)

														# now loading it with SampleObject2 should put f into `_use_default_values`
														lowerCAmelCase_			:   List[Any]     					=					SampleObjecta.from_config(__lowercase				)

														assert "f" in config._use_default_values
														assert config.f == [1, 3]

										# now loading the config, should **NOT** use [1, 3] for `f`, but the default [1, 4] value
										# **BECAUSE** it is part of `config._use_default_values`
										lowerCAmelCase_			:   List[Any]     					=					SampleObjecta.from_config(config.config				)
										assert new_config.f == [5, 4]

										config.config._use_default_values.pop()
										lowerCAmelCase_			:   int     					=					SampleObjecta.from_config(config.config				)
										assert new_config_a.f == [1, 3]

										# Nevertheless "e" should still be correctly loaded to [1, 3] from SampleObject2 instead of defaulting to [1, 5]
										assert new_config_a.e == [1, 3]