



"""simple docstring"""




import json
import os
from pathlib import Path
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple, Union

import sentencepiece

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


UpperCAmelCase_							:      str      =  logging.get_logger(__name__)

UpperCAmelCase_							:      int      =  """▁"""

UpperCAmelCase_							:      Any      =  {
    """vocab_file""": """vocab.json""",
    """spm_file""": """sentencepiece.bpe.model""",
    """tokenizer_config_file""": """tokenizer_config.json""",
}

UpperCAmelCase_							:      int      =  {
    """vocab_file""": {
        """facebook/m2m100_418M""": """https://huggingface.co/facebook/m2m100_418M/resolve/main/vocab.json""",
        """facebook/m2m100_1.2B""": """https://huggingface.co/facebook/m2m100_1.2B/resolve/main/vocab.json""",
    },
    """spm_file""": {
        """facebook/m2m100_418M""": """https://huggingface.co/facebook/m2m100_418M/resolve/main/sentencepiece.bpe.model""",
        """facebook/m2m100_1.2B""": """https://huggingface.co/facebook/m2m100_1.2B/resolve/main/sentencepiece.bpe.model""",
    },
    """tokenizer_config_file""": {
        """facebook/m2m100_418M""": """https://huggingface.co/facebook/m2m100_418M/resolve/main/tokenizer_config.json""",
        """facebook/m2m100_1.2B""": """https://huggingface.co/facebook/m2m100_1.2B/resolve/main/tokenizer_config.json""",
    },
}

UpperCAmelCase_							:      Tuple      =  {
    """facebook/m2m100_418M""": 1_024,
}

# fmt: off
UpperCAmelCase_							:      Optional[int]      =  {
    """m2m100""": ["""af""", """am""", """ar""", """ast""", """az""", """ba""", """be""", """bg""", """bn""", """br""", """bs""", """ca""", """ceb""", """cs""", """cy""", """da""", """de""", """el""", """en""", """es""", """et""", """fa""", """ff""", """fi""", """fr""", """fy""", """ga""", """gd""", """gl""", """gu""", """ha""", """he""", """hi""", """hr""", """ht""", """hu""", """hy""", """id""", """ig""", """ilo""", """is""", """it""", """ja""", """jv""", """ka""", """kk""", """km""", """kn""", """ko""", """lb""", """lg""", """ln""", """lo""", """lt""", """lv""", """mg""", """mk""", """ml""", """mn""", """mr""", """ms""", """my""", """ne""", """nl""", """no""", """ns""", """oc""", """or""", """pa""", """pl""", """ps""", """pt""", """ro""", """ru""", """sd""", """si""", """sk""", """sl""", """so""", """sq""", """sr""", """ss""", """su""", """sv""", """sw""", """ta""", """th""", """tl""", """tn""", """tr""", """uk""", """ur""", """uz""", """vi""", """wo""", """xh""", """yi""", """yo""", """zh""", """zu"""],
    """wmt21""": ["""en""", """ha""", """is""", """ja""", """cs""", """ru""", """zh""", """de"""]
}






class      UpperCamelCase    (						_UpperCAmelCase     ):
	lowerCAmelCase	:   Optional[int]				=	VOCAB_FILES_NAMES
	lowerCAmelCase	:   List[Any]				=	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
	lowerCAmelCase	:   Tuple				=	PRETRAINED_VOCAB_FILES_MAP
	lowerCAmelCase	:   Any				=	["input_ids", "attention_mask"]

	lowerCAmelCase	:   List[int]				=	[]
	lowerCAmelCase	:   List[int]				=	[]



	def __init__(       self			,	UpperCAmelCase__			,	UpperCAmelCase__			,	UpperCAmelCase__=None			,	UpperCAmelCase__=None			,	UpperCAmelCase__="<s>"			,	UpperCAmelCase__="</s>"			,	UpperCAmelCase__="</s>"			,	UpperCAmelCase__="<pad>"			,	UpperCAmelCase__="<unk>"			,	UpperCAmelCase__="m2m100"			,	UpperCAmelCase__ = None			,	UpperCAmelCase__=8			,	**UpperCAmelCase__			,	):
							A__							=       {} if sp_model_kwargs is None else sp_model_kwargs

							A__							=       language_codes
							A__							=       FAIRSEQ_LANGUAGE_CODES[language_codes]
							A__							=       {lang_code: F"""__{lang_code}__""" for lang_code in fairseq_language_code}

							A__							=       kwargs.get("additional_special_tokens"			,	[]		)
							kwargs["additional_special_tokens"] += [
							    self.get_lang_token(UpperCAmelCase__		)
							    for lang_code in fairseq_language_code
							    if self.get_lang_token(UpperCAmelCase__		) not in kwargs["additional_special_tokens"]
							]

							super().__init__(
							    src_lang=UpperCAmelCase__			,	tgt_lang=UpperCAmelCase__			,	bos_token=UpperCAmelCase__			,	eos_token=UpperCAmelCase__			,	sep_token=UpperCAmelCase__			,	unk_token=UpperCAmelCase__			,	pad_token=UpperCAmelCase__			,	language_codes=UpperCAmelCase__			,	sp_model_kwargs=self.sp_model_kwargs			,	num_madeup_words=UpperCAmelCase__			,	**UpperCAmelCase__			,	)

							A__							=       vocab_file
							A__							=       load_json(UpperCAmelCase__		)
							A__							=       {v: k for k, v in self.encoder.items()}
							A__							=       spm_file
							A__							=       load_spm(UpperCAmelCase__			,	self.sp_model_kwargs		)

							A__							=       len(self.encoder		)

							A__							=       {
							    self.get_lang_token(UpperCAmelCase__		): self.encoder_size + i for i, lang_code in enumerate(UpperCAmelCase__		)
							}
							A__							=       {lang_code: self.encoder_size + i for i, lang_code in enumerate(UpperCAmelCase__		)}
							A__							=       {v: k for k, v in self.lang_token_to_id.items()}

							A__							=       src_lang if src_lang is not None else "en"
							A__							=       tgt_lang
							A__							=       self.get_lang_id(self._src_lang		)
							self.set_src_lang_special_tokens(self._src_lang		)

							A__							=       num_madeup_words



	@property
	def 			__A							(       self		):
							return len(self.encoder		) + len(self.lang_token_to_id		)



	@property
	def 			__A							(       self		):
							return self._src_lang



	@src_lang.setter
	def 			__A							(       self			,	UpperCAmelCase__		):
							A__							=       new_src_lang
							self.set_src_lang_special_tokens(self._src_lang		)



	def 			__A							(       self			,	UpperCAmelCase__		):
							return self.sp_model.encode(UpperCAmelCase__			,	out_type=UpperCAmelCase__		)



	def 			__A							(       self			,	UpperCAmelCase__		):




							if token in self.lang_token_to_id:
													return self.lang_token_to_id[token]
							return self.encoder.get(UpperCAmelCase__			,	self.encoder[self.unk_token]		)



	def 			__A							(       self			,	UpperCAmelCase__		):




							if index in self.id_to_lang_token:
													return self.id_to_lang_token[index]
							return self.decoder.get(UpperCAmelCase__			,	self.unk_token		)



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



	def 			__A							(       self			,	UpperCAmelCase__			,	UpperCAmelCase__ = None			,	UpperCAmelCase__ = False		):




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

							A__							=       [1] * len(self.prefix_tokens		)
							A__							=       [1] * len(self.suffix_tokens		)
							if token_ids_a is None:
													return prefix_ones + ([0] * len(UpperCAmelCase__		)) + suffix_ones
							return prefix_ones + ([0] * len(UpperCAmelCase__		)) + ([0] * len(UpperCAmelCase__		)) + suffix_ones



	def 			__A							(       self			,	UpperCAmelCase__			,	UpperCAmelCase__ = None		):




							if token_ids_a is None:
													return self.prefix_tokens + token_ids_a + self.suffix_tokens
							# We don't expect to process pairs, but leave the pair logic for API consistency
							return self.prefix_tokens + token_ids_a + token_ids_a + self.suffix_tokens



	def 			__A							(       self		):
							A__							=       {self.convert_ids_to_tokens(UpperCAmelCase__		): 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			,	UpperCAmelCase__		):
							A__							=       d

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

							A__							=       load_spm(self.spm_file			,	self.sp_model_kwargs		)



	def 			__A							(       self			,	UpperCAmelCase__			,	UpperCAmelCase__ = None		):
							A__							=       Path(UpperCAmelCase__		)
							if not save_dir.is_dir():
													raise OSError(F"""{save_directory} should be a directory"""		)
							A__							=       save_dir / (
							    (filename_prefix + "-" if filename_prefix else "") + self.vocab_files_names["vocab_file"]
							)
							A__							=       save_dir / (
							    (filename_prefix + "-" if filename_prefix else "") + self.vocab_files_names["spm_file"]
							)

							save_json(self.encoder			,	UpperCAmelCase__		)

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

							return (str(UpperCAmelCase__		), str(UpperCAmelCase__		))



	def 			__A							(       self			,	UpperCAmelCase__			,	UpperCAmelCase__ = "en"			,	UpperCAmelCase__ = None			,	UpperCAmelCase__ = "ro"			,	**UpperCAmelCase__			,	):
							A__							=       src_lang
							A__							=       tgt_lang
							self.set_src_lang_special_tokens(self.src_lang		)
							return super().prepare_seqaseq_batch(UpperCAmelCase__			,	UpperCAmelCase__			,	**UpperCAmelCase__		)



	def 			__A							(       self			,	UpperCAmelCase__			,	UpperCAmelCase__			,	UpperCAmelCase__			,	**UpperCAmelCase__		):




							if src_lang is None or tgt_lang is None:
													raise ValueError("Translation requires a `src_lang` and a `tgt_lang` for this model"		)
							A__							=       src_lang
							A__							=       self(UpperCAmelCase__			,	add_special_tokens=UpperCAmelCase__			,	**UpperCAmelCase__		)
							A__							=       self.get_lang_id(UpperCAmelCase__		)
							A__							=       tgt_lang_id
							return inputs



	def 			__A							(       self		):
							self.set_src_lang_special_tokens(self.src_lang		)



	def 			__A							(       self		):
							self.set_tgt_lang_special_tokens(self.tgt_lang		)



	def 			__A							(       self			,	UpperCAmelCase__		):
							A__							=       self.get_lang_token(UpperCAmelCase__		)
							A__							=       self.lang_token_to_id[lang_token]
							A__							=       [self.cur_lang_id]
							A__							=       [self.eos_token_id]



	def 			__A							(       self			,	UpperCAmelCase__		):
							A__							=       self.get_lang_token(UpperCAmelCase__		)
							A__							=       self.lang_token_to_id[lang_token]
							A__							=       [self.cur_lang_id]
							A__							=       [self.eos_token_id]



	def 			__A							(       self			,	UpperCAmelCase__		):
							return self.lang_code_to_token[lang]



	def 			__A							(       self			,	UpperCAmelCase__		):
							A__							=       self.get_lang_token(UpperCAmelCase__		)
							return self.lang_token_to_id[lang_token]





def  UpperCamelCase						(       _A			:				str			,							_A			:				Dict[str, Any]    )->       List[Any]:





						"""simple docstring"""





						A__							=       sentencepiece.SentencePieceProcessor(**_A    )
						spm.Load(str(_A    )    )
						return spm
def  UpperCamelCase						(       _A			:				str    )->       List[Any]:





						"""simple docstring"""





						with open(_A			,							"r"    ) as f:
												return json.load(_A    )
def  UpperCamelCase						(       _A			:				Optional[Any]			,							_A			:				str    )->       str:





						"""simple docstring"""





						with open(_A			,							"w"    ) as f:
												json.dump(_A			,							_A			,							indent=2    )










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


UpperCAmelCase_							:      Dict      =  False






@skip_mps
class      UpperCamelCase    (						_UpperCAmelCase       ,    _UpperCAmelCase       ,    _UpperCAmelCase       ,    unittest.TestCase     ):
	lowerCAmelCase	:   Optional[Any]				=	StableDiffusionAttendAndExcitePipeline
	lowerCAmelCase	:   Optional[Any]				=	False
	lowerCAmelCase	:   Any				=	TEXT_TO_IMAGE_PARAMS
	lowerCAmelCase	:   Dict				=	TEXT_TO_IMAGE_BATCH_PARAMS.union({"""token_indices"""}     )
	lowerCAmelCase	:   List[str]				=	TEXT_TO_IMAGE_IMAGE_PARAMS
	lowerCAmelCase	:   List[Any]				=	TEXT_TO_IMAGE_IMAGE_PARAMS



	@classmethod
	def 			__A							(       cls		):
							super().setUpClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	@classmethod
	def 			__A							(       cls		):
							super().tearDownClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	def 			__A							(       self		):
							torch.manual_seed(0		)
							A__							=       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=UpperCAmelCase__			,	)
							A__							=       DDIMScheduler(
							    beta_start=0.00_085			,	beta_end=0.012			,	beta_schedule="scaled_linear"			,	clip_sample=UpperCAmelCase__			,	set_alpha_to_one=UpperCAmelCase__			,	)
							torch.manual_seed(0		)
							A__							=       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=128			,	)
							torch.manual_seed(0		)
							A__							=       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=1_000			,	hidden_act="gelu"			,	projection_dim=512			,	)
							A__							=       CLIPTextModel(UpperCAmelCase__		)
							A__							=       CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"		)

							A__							=       {
							    "unet": unet,
							    "scheduler": scheduler,
							    "vae": vae,
							    "text_encoder": text_encoder,
							    "tokenizer": tokenizer,
							    "safety_checker": None,
							    "feature_extractor": None,
							}

							return components



	def 			__A							(       self			,	UpperCAmelCase__			,	UpperCAmelCase__=0		):
							if str(UpperCAmelCase__		).startswith("mps"		):
													A__							=       torch.manual_seed(UpperCAmelCase__		)
							else:
													A__							=       torch.Generator(device=UpperCAmelCase__		).manual_seed(UpperCAmelCase__		)
							A__							=       A__							=       {
							    "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 			__A							(       self		):
							A__							=       "cpu"

							A__							=       self.get_dummy_components()
							A__							=       self.pipeline_class(**UpperCAmelCase__		)
							pipe.to(UpperCAmelCase__		)
							pipe.set_progress_bar_config(disable=UpperCAmelCase__		)

							A__							=       self.get_dummy_inputs(UpperCAmelCase__		)
							A__							=       pipe(**UpperCAmelCase__		).images
							A__							=       image[0, -3:, -3:, -1]

							self.assertEqual(image.shape			,	(1, 64, 64, 3)		)
							A__							=       np.array(
							    [0.63_905_364, 0.62_897_307, 0.48_599_017, 0.5_133_624, 0.5_550_048, 0.45_769_516, 0.50_326_973, 0.5_023_139, 0.45_384_496]		)
							A__							=       np.abs(image_slice.flatten() - expected_slice		).max()
							self.assertLessEqual(UpperCAmelCase__			,	1e-3		)



	def 			__A							(       self		):
							super().test_cpu_offload_forward_pass(expected_max_diff=5e-4		)



	def 			__A							(       self		):
							# NOTE: Larger batch sizes cause this test to timeout, only test on smaller batches
							self._test_inference_batch_consistent(batch_sizes=[1, 2]		)



	def 			__A							(       self		):
							self._test_inference_batch_single_identical(batch_size=2			,	expected_max_diff=7e-4		)



	def 			__A							(       self		):
							super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3		)



	def 			__A							(       self		):
							super().test_pt_np_pil_outputs_equivalent(expected_max_diff=5e-4		)



	def 			__A							(       self		):
							super().test_save_load_local(expected_max_difference=5e-4		)



	def 			__A							(       self		):
							super().test_save_load_optional_components(expected_max_difference=4e-4		)






@require_torch_gpu
@slow
class      UpperCamelCase    (						unittest.TestCase     ):



	@classmethod
	def 			__A							(       cls		):
							super().setUpClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	@classmethod
	def 			__A							(       cls		):
							super().tearDownClass()
							torch.use_deterministic_algorithms(UpperCAmelCase__		)



	def 			__A							(       self		):
							super().tearDown()
							gc.collect()
							torch.cuda.empty_cache()



	def 			__A							(       self		):
							A__							=       torch.manual_seed(51		)

							A__							=       StableDiffusionAttendAndExcitePipeline.from_pretrained(
							    "CompVis/stable-diffusion-v1-4"			,	safety_checker=UpperCAmelCase__			,	torch_dtype=torch.floataa		)
							pipe.to("cuda"		)

							A__							=       "a painting of an elephant with glasses"
							A__							=       [5, 7]

							A__							=       pipe(
							    prompt=UpperCAmelCase__			,	token_indices=UpperCAmelCase__			,	guidance_scale=7.5			,	generator=UpperCAmelCase__			,	num_inference_steps=5			,	max_iter_to_alter=5			,	output_type="numpy"			,	).images[0]

							A__							=       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





