


'''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


_lowercase      :      str            =					logging.get_logger(__name__)

_lowercase      :      int            =					"""▁"""

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

_lowercase      :      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""",
    },
}

_lowercase      :      Tuple            =					{
    """facebook/m2m100_418M""": 1024,
}

# fmt: off
_lowercase      :      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__(       lowerCAmelCase	):
				__magic_name__    :				Optional[int]  		=	VOCAB_FILES_NAMES
				__magic_name__    :				List[Any]  		=	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				__magic_name__    :				Tuple  		=	PRETRAINED_VOCAB_FILES_MAP
				__magic_name__    :				Any  		=	["input_ids", "attention_mask"]

				__magic_name__    :				List[int]  		=	[]
				__magic_name__    :				List[int]  		=	[]




				def __init__(		self      :			List[Any]				,   lowerCAmelCase      :			int				,   lowerCAmelCase      :			Dict				,   lowerCAmelCase      :			List[str]=None				,   lowerCAmelCase      :			List[str]=None				,   lowerCAmelCase      :			Any="<s>"				,   lowerCAmelCase      :			Optional[int]="</s>"				,   lowerCAmelCase      :			List[str]="</s>"				,   lowerCAmelCase      :			str="<pad>"				,   lowerCAmelCase      :			Any="<unk>"				,   lowerCAmelCase      :			List[str]="m2m100"				,   lowerCAmelCase      :			Optional[Dict[str, Any]] = None				,   lowerCAmelCase      :			List[Any]=8				,   **lowerCAmelCase      :			Dict				,   )->					None:

										"""simple docstring"""




										UpperCAmelCase					    =     {} if sp_model_kwargs is None else sp_model_kwargs

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

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

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

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

										UpperCAmelCase					    =     len(self.encoder				)

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

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

										UpperCAmelCase					    =     num_madeup_words




				@property
				def 					a__(		self      :			Tuple				)->					int:

										"""simple docstring"""




										return len(self.encoder				) + len(self.lang_token_to_id				)




				@property
				def 					a__(		self      :			int				)->					str:

										"""simple docstring"""




										return self._src_lang




				@src_lang.setter
				def 					a__(		self      :			Any				,   lowerCAmelCase      :			str				)->					None:

										"""simple docstring"""




										UpperCAmelCase					    =     new_src_lang
										self.set_src_lang_special_tokens(self._src_lang				)




				def 					a__(		self      :			Optional[int]				,   lowerCAmelCase      :			str				)->					List[str]:

										"""simple docstring"""




										return self.sp_model.encode(lowerCAmelCase				,   out_type=lowerCAmelCase				)




				def 					a__(		self      :			Optional[Any]				,   lowerCAmelCase      :			int				)->					Dict:

										"""simple docstring"""




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




				def 					a__(		self      :			List[str]				,   lowerCAmelCase      :			int				)->					str:

										"""simple docstring"""




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




				def 					a__(		self      :			Union[str, Any]				,   lowerCAmelCase      :			Union[str, Any]				)->					List[Any]:

										"""simple docstring"""




										UpperCAmelCase					    =     []
										UpperCAmelCase					    =     ''''''
										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(lowerCAmelCase				) + token
																						UpperCAmelCase					    =     []
																else:
																						current_sub_tokens.append(lowerCAmelCase				)
										out_string += self.sp_model.decode(lowerCAmelCase				)
										return out_string.strip()




				def 					a__(		self      :			Dict				,   lowerCAmelCase      :			List[int]				,   lowerCAmelCase      :			Optional[List[int]] = None				,   lowerCAmelCase      :			bool = False				)->					List[int]:

										"""simple docstring"""




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

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




				def 					a__(		self      :			List[Any]				,   lowerCAmelCase      :			List[int]				,   lowerCAmelCase      :			Optional[List[int]] = None				)->					List[int]:

										"""simple docstring"""




										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      :			List[Any]				)->					Dict:

										"""simple docstring"""




										UpperCAmelCase					    =     {self.convert_ids_to_tokens(lowerCAmelCase				): i for i in range(self.vocab_size				)}
										vocab.update(self.added_tokens_encoder				)
										return vocab




				def __getstate__(		self      :			Any				)->					Dict:

										"""simple docstring"""




										UpperCAmelCase					    =     self.__dict__.copy()
										UpperCAmelCase					    =     None
										return state




				def __setstate__(		self      :			Tuple				,   lowerCAmelCase      :			Dict				)->					None:

										"""simple docstring"""




										UpperCAmelCase					    =     d

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

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




				def 					a__(		self      :			Any				,   lowerCAmelCase      :			str				,   lowerCAmelCase      :			Optional[str] = None				)->					Tuple[str]:

										"""simple docstring"""




										UpperCAmelCase					    =     Path(lowerCAmelCase				)
										if not save_dir.is_dir():
																raise OSError(F"""{save_directory} should be a directory"""				)
										UpperCAmelCase					    =     save_dir / (
										    (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''vocab_file''']
										)
										UpperCAmelCase					    =     save_dir / (
										    (filename_prefix + '''-''' if filename_prefix else '''''') + self.vocab_files_names['''spm_file''']
										)

										save_json(self.encoder				,   lowerCAmelCase				)

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

										return (str(lowerCAmelCase				), str(lowerCAmelCase				))




				def 					a__(		self      :			Optional[int]				,   lowerCAmelCase      :			List[str]				,   lowerCAmelCase      :			str = "en"				,   lowerCAmelCase      :			Optional[List[str]] = None				,   lowerCAmelCase      :			str = "ro"				,   **lowerCAmelCase      :			Dict				,   )->					BatchEncoding:

										"""simple docstring"""




										UpperCAmelCase					    =     src_lang
										UpperCAmelCase					    =     tgt_lang
										self.set_src_lang_special_tokens(self.src_lang				)
										return super().prepare_seqaseq_batch(lowerCAmelCase				,   lowerCAmelCase				,   **lowerCAmelCase				)




				def 					a__(		self      :			Union[str, Any]				,   lowerCAmelCase      :			int				,   lowerCAmelCase      :			Optional[str]				,   lowerCAmelCase      :			Optional[str]				,   **lowerCAmelCase      :			Union[str, Any]				)->					int:

										"""simple docstring"""




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




				def 					a__(		self      :			Dict				)->					Dict:

										"""simple docstring"""




										self.set_src_lang_special_tokens(self.src_lang				)




				def 					a__(		self      :			str				)->					Dict:

										"""simple docstring"""




										self.set_tgt_lang_special_tokens(self.tgt_lang				)




				def 					a__(		self      :			Dict				,   lowerCAmelCase      :			str				)->					None:

										"""simple docstring"""




										UpperCAmelCase					    =     self.get_lang_token(lowerCAmelCase				)
										UpperCAmelCase					    =     self.lang_token_to_id[lang_token]
										UpperCAmelCase					    =     [self.cur_lang_id]
										UpperCAmelCase					    =     [self.eos_token_id]




				def 					a__(		self      :			List[str]				,   lowerCAmelCase      :			str				)->					None:

										"""simple docstring"""




										UpperCAmelCase					    =     self.get_lang_token(lowerCAmelCase				)
										UpperCAmelCase					    =     self.lang_token_to_id[lang_token]
										UpperCAmelCase					    =     [self.cur_lang_id]
										UpperCAmelCase					    =     [self.eos_token_id]




				def 					a__(		self      :			Dict				,   lowerCAmelCase      :			str				)->					str:

										"""simple docstring"""




										return self.lang_code_to_token[lang]





				def 					a__(		self      :			List[Any]				,   lowerCAmelCase      :			str				)->					int:

										"""simple docstring"""




										UpperCAmelCase					    =     self.get_lang_token(lowerCAmelCase				)
										return self.lang_token_to_id[lang_token]







def 				lowerCamelCase__				(	A       :      str				,       A       :      Dict[str, Any]					):




						'''simple docstring'''






						UpperCAmelCase					    =     sentencepiece.SentencePieceProcessor(**A					)
						spm.Load(str(A					)					)
						return spm


def 				lowerCamelCase__				(	A       :      str					):




						'''simple docstring'''






						with open(A				,       '''r'''					) as f:
												return json.load(A					)




def 				lowerCamelCase__				(	A       :      Optional[Any]				,       A       :      str					):




						'''simple docstring'''






						with open(A				,       '''w'''					) as f:
												json.dump(A				,       A				,       indent=2					)









'''simple docstring'''



import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    StableDiffusionSAGPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, 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 PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class 	UpperCamelCase__(       lowerCAmelCase						,    lowerCAmelCase						,    unittest.TestCase	):
				__magic_name__    :				List[str]  		=	StableDiffusionSAGPipeline
				__magic_name__    :				str  		=	TEXT_TO_IMAGE_PARAMS
				__magic_name__    :				Any  		=	TEXT_TO_IMAGE_BATCH_PARAMS
				__magic_name__    :				Optional[int]  		=	TEXT_TO_IMAGE_IMAGE_PARAMS
				__magic_name__    :				List[Any]  		=	TEXT_TO_IMAGE_IMAGE_PARAMS
				__magic_name__    :				str  		=	False




				def 					a__(		self      :			Union[str, Any]				)->					Tuple:

										"""simple docstring"""




										torch.manual_seed(0				)
										UpperCAmelCase					    =     UNetaDConditionModel(
										    block_out_channels=(32, 64)				,   layers_per_block=2				,   sample_size=32				,   in_channels=4				,   out_channels=4				,   down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')				,   up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')				,   cross_attention_dim=32				,   )
										UpperCAmelCase					    =     DDIMScheduler(
										    beta_start=0.00085				,   beta_end=0.012				,   beta_schedule='''scaled_linear'''				,   clip_sample=lowerCAmelCase				,   set_alpha_to_one=lowerCAmelCase				,   )
										torch.manual_seed(0				)
										UpperCAmelCase					    =     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				,   )
										torch.manual_seed(0				)
										UpperCAmelCase					    =     CLIPTextConfig(
										    bos_token_id=0				,   eos_token_id=2				,   hidden_size=32				,   intermediate_size=37				,   layer_norm_eps=1E-05				,   num_attention_heads=4				,   num_hidden_layers=5				,   pad_token_id=1				,   vocab_size=1000				,   )
										UpperCAmelCase					    =     CLIPTextModel(lowerCAmelCase				)
										UpperCAmelCase					    =     CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''				)

										UpperCAmelCase					    =     {
										    '''unet''': unet,
										    '''scheduler''': scheduler,
										    '''vae''': vae,
										    '''text_encoder''': text_encoder,
										    '''tokenizer''': tokenizer,
										    '''safety_checker''': None,
										    '''feature_extractor''': None,
										}
										return components




				def 					a__(		self      :			Optional[Any]				,   lowerCAmelCase      :			str				,   lowerCAmelCase      :			Tuple=0				)->					str:

										"""simple docstring"""




										if str(lowerCAmelCase				).startswith('''mps'''				):
																UpperCAmelCase					    =     torch.manual_seed(lowerCAmelCase				)
										else:
																UpperCAmelCase					    =     torch.Generator(device=lowerCAmelCase				).manual_seed(lowerCAmelCase				)
										UpperCAmelCase					    =     {
										    '''prompt''': '''.''',
										    '''generator''': generator,
										    '''num_inference_steps''': 2,
										    '''guidance_scale''': 1.0,
										    '''sag_scale''': 1.0,
										    '''output_type''': '''numpy''',
										}
										return inputs





				def 					a__(		self      :			Any				)->					List[str]:

										"""simple docstring"""




										super().test_inference_batch_single_identical(expected_max_diff=3E-3				)







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




				def 					a__(		self      :			Union[str, Any]				)->					Any:

										"""simple docstring"""




										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()




				def 					a__(		self      :			Union[str, Any]				)->					Tuple:

										"""simple docstring"""




										UpperCAmelCase					    =     StableDiffusionSAGPipeline.from_pretrained('''CompVis/stable-diffusion-v1-4'''				)
										UpperCAmelCase					    =     sag_pipe.to(lowerCAmelCase				)
										sag_pipe.set_progress_bar_config(disable=lowerCAmelCase				)

										UpperCAmelCase					    =     '''.'''
										UpperCAmelCase					    =     torch.manual_seed(0				)
										UpperCAmelCase					    =     sag_pipe(
										    [prompt]				,   generator=lowerCAmelCase				,   guidance_scale=7.5				,   sag_scale=1.0				,   num_inference_steps=20				,   output_type='''np'''				)

										UpperCAmelCase					    =     output.images

										UpperCAmelCase					    =     image[0, -3:, -3:, -1]

										assert image.shape == (1, 512, 512, 3)
										UpperCAmelCase					    =     np.array([0.1568, 0.1738, 0.1695, 0.1693, 0.1507, 0.1705, 0.1547, 0.1751, 0.1949]				)

										assert np.abs(image_slice.flatten() - expected_slice				).max() < 5E-2




				def 					a__(		self      :			int				)->					Union[str, Any]:

										"""simple docstring"""




										UpperCAmelCase					    =     StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''				)
										UpperCAmelCase					    =     sag_pipe.to(lowerCAmelCase				)
										sag_pipe.set_progress_bar_config(disable=lowerCAmelCase				)

										UpperCAmelCase					    =     '''.'''
										UpperCAmelCase					    =     torch.manual_seed(0				)
										UpperCAmelCase					    =     sag_pipe(
										    [prompt]				,   generator=lowerCAmelCase				,   guidance_scale=7.5				,   sag_scale=1.0				,   num_inference_steps=20				,   output_type='''np'''				)

										UpperCAmelCase					    =     output.images

										UpperCAmelCase					    =     image[0, -3:, -3:, -1]

										assert image.shape == (1, 512, 512, 3)
										UpperCAmelCase					    =     np.array([0.3459, 0.2876, 0.2537, 0.3002, 0.2671, 0.2160, 0.3026, 0.2262, 0.2371]				)

										assert np.abs(image_slice.flatten() - expected_slice				).max() < 5E-2





				def 					a__(		self      :			Optional[Any]				)->					Union[str, Any]:

										"""simple docstring"""




										UpperCAmelCase					    =     StableDiffusionSAGPipeline.from_pretrained('''stabilityai/stable-diffusion-2-1-base'''				)
										UpperCAmelCase					    =     sag_pipe.to(lowerCAmelCase				)
										sag_pipe.set_progress_bar_config(disable=lowerCAmelCase				)

										UpperCAmelCase					    =     '''.'''
										UpperCAmelCase					    =     torch.manual_seed(0				)
										UpperCAmelCase					    =     sag_pipe(
										    [prompt]				,   width=768				,   height=512				,   generator=lowerCAmelCase				,   guidance_scale=7.5				,   sag_scale=1.0				,   num_inference_steps=20				,   output_type='''np'''				,   )

										UpperCAmelCase					    =     output.images

										assert image.shape == (1, 512, 768, 3)





