






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

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


__magic_name__:					Optional[Any]							=     logging.get_logger(__name__)

__magic_name__:					int							=     "▁"

__magic_name__:					Dict							=     {"vocab_file": "prophetnet.tokenizer"}

__magic_name__:					Optional[int]							=     {
    "vocab_file": {
        "microsoft/xprophetnet-large-wiki100-cased": (
            "https://huggingface.co/microsoft/xprophetnet-large-wiki100-cased/resolve/main/prophetnet.tokenizer"
        ),
    }
}

__magic_name__:					List[str]							=     {
    "microsoft/xprophetnet-large-wiki100-cased": {"do_lower_case": False},
}

__magic_name__:					Optional[Any]							=     {
    "microsoft/xprophetnet-large-wiki100-cased": 512,
}

def    UpperCamelCase						(	_A			):






						"""simple docstring"""







						__magic_name__      :       Optional[Any]       						= collections.OrderedDict()
						with open(_A,   """r""",   encoding="""utf-8"""			) as reader:
												__magic_name__      :       Tuple       						= reader.readlines()
						for index, token in enumerate(_A			):
												__magic_name__      :       Optional[Any]       						= token.rstrip("""\n"""			)
												__magic_name__      :       Any       						= index
						return vocab





class 	snake_case__ (							_lowerCAmelCase				):
	lowercase__						:					Optional[int]											=   VOCAB_FILES_NAMES
	lowercase__						:					Dict											=   PRETRAINED_VOCAB_FILES_MAP
	lowercase__						:					str											=   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	lowercase__						:					Union[str, Any]											=   ['''input_ids''', '''attention_mask''']
	def __init__(    self      ,       lowerCAmelCase__      ,       lowerCAmelCase__="[SEP]"      ,       lowerCAmelCase__="[SEP]"      ,       lowerCAmelCase__="[SEP]"      ,       lowerCAmelCase__="[UNK]"      ,       lowerCAmelCase__="[PAD]"      ,       lowerCAmelCase__="[CLS]"      ,       lowerCAmelCase__="[MASK]"      ,       lowerCAmelCase__ = None      ,       **lowerCAmelCase__      ,       )							->				None:
							__magic_name__      :       List[str]       						= {} if sp_model_kwargs is None else sp_model_kwargs

							super().__init__(
							    bos_token=lowerCAmelCase__      ,       eos_token=lowerCAmelCase__      ,       sep_token=lowerCAmelCase__      ,       unk_token=lowerCAmelCase__      ,       pad_token=lowerCAmelCase__      ,       cls_token=lowerCAmelCase__      ,       mask_token=lowerCAmelCase__      ,       sp_model_kwargs=self.sp_model_kwargs      ,       **lowerCAmelCase__      ,       )

							try:
													import sentencepiece as spm
							except ImportError:
													logger.warning(
													    """You need to install SentencePiece to use XLMRobertaTokenizer: https://github.com/google/sentencepiece"""
													    """ pip install sentencepiece"""			)
													raise

							__magic_name__      :       Tuple       						= spm.SentencePieceProcessor(**self.sp_model_kwargs			)
							self.sp_model.Load(str(lowerCAmelCase__			)			)
							__magic_name__      :       str       						= vocab_file

							# Original fairseq vocab and spm vocab must be "aligned":
							# Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
							# -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
							# fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
							# spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

							# put special tokens and [unused] tokens into the vocab
							__magic_name__      :       Tuple       						= {"""[PAD]""": 0, """[CLS]""": 1, """[SEP]""": 2, """[UNK]""": 3, """[MASK]""": 4}

							for i in range(10			):
													__magic_name__      :       Optional[int]       						= F'[unused{i}]'
													__magic_name__      :       Tuple       						= 5 + i

							# The first "real" token "," has position 15 in the embedding vocab and position 3 in the spm vocab
							__magic_name__      :       Any       						= 12
							__magic_name__      :       Optional[int]       						= {v: k for k, v in self.fairseq_tokens_to_ids.items()}
							for k in self.fairseq_tokens_to_ids.keys():
													self.unique_no_split_tokens.append(lowerCAmelCase__			)
	def __getstate__(    self			)							->				Union[str, Any]:
							__magic_name__      :       Optional[int]       						= self.__dict__.copy()
							__magic_name__      :       Optional[Any]       						= None
							return state
	def __setstate__(    self      ,       lowerCAmelCase__			)							->				Any:
							__magic_name__      :       Optional[int]       						= d
							try:
													import sentencepiece as spm
							except ImportError:
													logger.warning(
													    """You need to install SentencePiece to use XLMRobertaTokenizer: https://github.com/google/sentencepiece"""
													    """ pip install sentencepiece"""			)
													raise

							# for backward compatibility
							if not hasattr(self      ,       """sp_model_kwargs"""			):
													__magic_name__      :       List[str]       						= {}

							__magic_name__      :       Union[str, Any]       						= spm.SentencePieceProcessor(**self.sp_model_kwargs			)
							self.sp_model.Load(self.vocab_file			)
	def       __magic_name__    (    self      ,       lowerCAmelCase__      ,       lowerCAmelCase__ = None      ,       lowerCAmelCase__ = False			)							->				List[int]:

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

							if token_ids_a is None:
													return ([0] * len(lowerCAmelCase__			)) + [1]
							return ([0] * len(lowerCAmelCase__			)) + [1] + ([0] * len(lowerCAmelCase__			)) + [1]
	def       __magic_name__    (    self      ,       lowerCAmelCase__      ,       lowerCAmelCase__ = None			)							->				List[int]:
							__magic_name__      :       int       						= [self.sep_token_id]

							if token_ids_a is None:
													return len(token_ids_a + sep			) * [0]
							return len(token_ids_a + sep + sep + token_ids_a + sep			) * [0]
	@property
	def       __magic_name__    (    self			)							->				List[Any]:
							return len(self.sp_model			) + self.fairseq_offset
	def       __magic_name__    (    self			)							->				Any:
							__magic_name__      :       List[Any]       						= {self.convert_ids_to_tokens(lowerCAmelCase__			): i for i in range(self.vocab_size			)}
							vocab.update(self.added_tokens_encoder			)
							return vocab
	def       __magic_name__    (    self      ,       lowerCAmelCase__			)							->				str:
							return self.sp_model.encode(lowerCAmelCase__      ,       out_type=lowerCAmelCase__			)
	def       __magic_name__    (    self      ,       lowerCAmelCase__			)							->				Tuple:
							if token in self.fairseq_tokens_to_ids:
													return self.fairseq_tokens_to_ids[token]
							__magic_name__      :       Dict       						= self.sp_model.PieceToId(lowerCAmelCase__			)

							# Need to return unknown token if the SP model returned 0
							return spm_id + self.fairseq_offset if spm_id else self.unk_token_id
	def       __magic_name__    (    self      ,       lowerCAmelCase__			)							->				int:
							if index in self.fairseq_ids_to_tokens:
													return self.fairseq_ids_to_tokens[index]
							return self.sp_model.IdToPiece(index - self.fairseq_offset			)
	def       __magic_name__    (    self      ,       lowerCAmelCase__			)							->				str:
							__magic_name__      :       Optional[Any]       						= """""".join(lowerCAmelCase__			).replace(lowerCAmelCase__      ,       """ """			).strip()
							return out_string
	def       __magic_name__    (    self      ,       lowerCAmelCase__      ,       lowerCAmelCase__ = None			)							->				Tuple[str]:
							if not os.path.isdir(lowerCAmelCase__			):
													logger.error(F'Vocabulary path ({save_directory}) should be a directory'			)
													return
							__magic_name__      :       Optional[Any]       						= os.path.join(
							    lowerCAmelCase__      ,       (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]			)

							if os.path.abspath(self.vocab_file			) != os.path.abspath(lowerCAmelCase__			) and os.path.isfile(self.vocab_file			):
													copyfile(self.vocab_file      ,       lowerCAmelCase__			)
							elif not os.path.isfile(self.vocab_file			):
													with open(lowerCAmelCase__      ,       """wb"""			) as fi:
																			__magic_name__      :       List[str]       						= self.sp_model.serialized_model_proto()
																			fi.write(lowerCAmelCase__			)

							return (out_vocab_file,)







	def       __magic_name__    (    self      ,       lowerCAmelCase__      ,       lowerCAmelCase__ = None			)							->				List[int]:

							if token_ids_a is None:
													return token_ids_a + [self.sep_token_id]
							__magic_name__      :       Dict       						= [self.sep_token_id]
							return token_ids_a + sep + token_ids_a + sep








import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionAttendAndExcitePipeline,
    UNetaDConditionModel,
)
from diffusers.utils import load_numpy, skip_mps, slow
from diffusers.utils.testing_utils import require_torch_gpu

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


__magic_name__:					Union[str, Any]							=     False





@skip_mps
class 	snake_case__ (							_lowerCAmelCase						,  _lowerCAmelCase						,  _lowerCAmelCase						,  unittest.TestCase				):
	lowercase__						:					Optional[int]											=   StableDiffusionAttendAndExcitePipeline
	lowercase__						:					Tuple											=   False
	lowercase__						:					List[str]											=   TEXT_TO_IMAGE_PARAMS
	lowercase__						:					Dict											=   TEXT_TO_IMAGE_BATCH_PARAMS.union({'''token_indices'''}				)
	lowercase__						:					Union[str, Any]											=   TEXT_TO_IMAGE_IMAGE_PARAMS
	lowercase__						:					List[Any]											=   TEXT_TO_IMAGE_IMAGE_PARAMS
	@classmethod
	def       __magic_name__    (    cls			)							->				Tuple:
							super().setUpClass()
							torch.use_deterministic_algorithms(lowerCAmelCase__			)
	@classmethod
	def       __magic_name__    (    cls			)							->				Optional[Any]:
							super().tearDownClass()
							torch.use_deterministic_algorithms(lowerCAmelCase__			)
	def       __magic_name__    (    self			)							->				str:
							torch.manual_seed(0			)
							__magic_name__      :       Union[str, Any]       						= UNetaDConditionModel(
							    block_out_channels=(32, 64)      ,       layers_per_block=1      ,       sample_size=32      ,       in_channels=4      ,       out_channels=4      ,       down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")      ,       up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")      ,       cross_attention_dim=32      ,       attention_head_dim=(2, 4)      ,       use_linear_projection=lowerCAmelCase__      ,       )
							__magic_name__      :       List[Any]       						= DDIMScheduler(
							    beta_start=0.0_0_0_8_5      ,       beta_end=0.0_1_2      ,       beta_schedule="""scaled_linear"""      ,       clip_sample=lowerCAmelCase__      ,       set_alpha_to_one=lowerCAmelCase__      ,       )
							torch.manual_seed(0			)
							__magic_name__      :       List[Any]       						= AutoencoderKL(
							    block_out_channels=[32, 64]      ,       in_channels=3      ,       out_channels=3      ,       down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]      ,       up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]      ,       latent_channels=4      ,       sample_size=1_28      ,       )
							torch.manual_seed(0			)
							__magic_name__      :       List[str]       						= CLIPTextConfig(
							    bos_token_id=0      ,       eos_token_id=2      ,       hidden_size=32      ,       intermediate_size=37      ,       layer_norm_eps=1e-0_5      ,       num_attention_heads=4      ,       num_hidden_layers=5      ,       pad_token_id=1      ,       vocab_size=10_00      ,       hidden_act="""gelu"""      ,       projection_dim=5_12      ,       )
							__magic_name__      :       Any       						= CLIPTextModel(lowerCAmelCase__			)
							__magic_name__      :       Optional[int]       						= CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""			)

							__magic_name__      :       Optional[int]       						= {
							    """unet""": unet,
							    """scheduler""": scheduler,
							    """vae""": vae,
							    """text_encoder""": text_encoder,
							    """tokenizer""": tokenizer,
							    """safety_checker""": None,
							    """feature_extractor""": None,
							}

							return components
	def       __magic_name__    (    self      ,       lowerCAmelCase__      ,       lowerCAmelCase__=0			)							->				Optional[Any]:
							if str(lowerCAmelCase__			).startswith("""mps"""			):
													__magic_name__      :       int       						= torch.manual_seed(lowerCAmelCase__			)
							else:
													__magic_name__      :       int       						= torch.Generator(device=lowerCAmelCase__			).manual_seed(lowerCAmelCase__			)
							__magic_name__      :       Optional[Any]       						= {
							    """prompt""": """a cat and a frog""",
							    """token_indices""": [2, 5],
							    """generator""": generator,
							    """num_inference_steps""": 1,
							    """guidance_scale""": 6.0,
							    """output_type""": """numpy""",
							    """max_iter_to_alter""": 2,
							    """thresholds""": {0: 0.7},
							}
							return inputs
	def       __magic_name__    (    self			)							->				Dict:
							__magic_name__      :       int       						= """cpu"""

							__magic_name__      :       Union[str, Any]       						= self.get_dummy_components()
							__magic_name__      :       int       						= self.pipeline_class(**lowerCAmelCase__			)
							pipe.to(lowerCAmelCase__			)
							pipe.set_progress_bar_config(disable=lowerCAmelCase__			)

							__magic_name__      :       Union[str, Any]       						= self.get_dummy_inputs(lowerCAmelCase__			)
							__magic_name__      :       str       						= pipe(**lowerCAmelCase__			).images
							__magic_name__      :       Any       						= image[0, -3:, -3:, -1]

							self.assertEqual(image.shape      ,       (1, 64, 64, 3)			)
							__magic_name__      :       Dict       						= np.array(
							    [0.6_3_9_0_5_3_6_4, 0.6_2_8_9_7_3_0_7, 0.4_8_5_9_9_0_1_7, 0.5_1_3_3_6_2_4, 0.5_5_5_0_0_4_8, 0.4_5_7_6_9_5_1_6, 0.5_0_3_2_6_9_7_3, 0.5_0_2_3_1_3_9, 0.4_5_3_8_4_4_9_6]			)
							__magic_name__      :       Tuple       						= np.abs(image_slice.flatten() - expected_slice			).max()
							self.assertLessEqual(lowerCAmelCase__      ,       1e-3			)
	def       __magic_name__    (    self			)							->				List[Any]:
							super().test_cpu_offload_forward_pass(expected_max_diff=5e-4			)
	def       __magic_name__    (    self			)							->				Union[str, Any]:
							# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
							self._test_inference_batch_consistent(batch_sizes=[1, 2]			)
	def       __magic_name__    (    self			)							->				int:
							self._test_inference_batch_single_identical(batch_size=2      ,       expected_max_diff=7e-4			)
	def       __magic_name__    (    self			)							->				List[str]:
							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3			)
	def       __magic_name__    (    self			)							->				Any:
							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4			)
	def       __magic_name__    (    self			)							->				Dict:
							super().test_save_load_local(expected_max_difference=5e-4			)







	def       __magic_name__    (    self			)							->				List[Any]:
							super().test_save_load_optional_components(expected_max_difference=4e-4			)





@require_torch_gpu
@slow
class 	snake_case__ (							unittest.TestCase				):
	@classmethod
	def       __magic_name__    (    cls			)							->				Optional[int]:
							super().setUpClass()
							torch.use_deterministic_algorithms(lowerCAmelCase__			)
	@classmethod
	def       __magic_name__    (    cls			)							->				List[Any]:
							super().tearDownClass()
							torch.use_deterministic_algorithms(lowerCAmelCase__			)
	def       __magic_name__    (    self			)							->				Tuple:
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()







	def       __magic_name__    (    self			)							->				Optional[Any]:
							__magic_name__      :       str       						= torch.manual_seed(51			)

							__magic_name__      :       Tuple       						= StableDiffusionAttendAndExcitePipeline.from_pretrained(
							    """CompVis/stable-diffusion-v1-4"""      ,       safety_checker=lowerCAmelCase__      ,       torch_dtype=torch.floataa			)
							pipe.to("""cuda"""			)

							__magic_name__      :       List[str]       						= """a painting of an elephant with glasses"""
							__magic_name__      :       Any       						= [5, 7]

							__magic_name__      :       List[Any]       						= pipe(
							    prompt=lowerCAmelCase__      ,       token_indices=lowerCAmelCase__      ,       guidance_scale=7.5      ,       generator=lowerCAmelCase__      ,       num_inference_steps=5      ,       max_iter_to_alter=5      ,       output_type="""numpy"""      ,       ).images[0]

							__magic_name__      :       Optional[int]       						= load_numpy(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/attend-and-excite/elephant_glasses.npy"""			)
							assert np.abs((expected_image - image).max()			) < 5e-1
