






"""simple docstring"""







import os
from shutil import copyfile
from typing import List, Optional, Tuple

from tokenizers import processors

from ...tokenization_utils import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import is_sentencepiece_available, logging


if is_sentencepiece_available():
 from .tokenization_mbartaa import MBartaaTokenizer
else:
 lowerCAmelCase_		:			int				   =					None


lowerCAmelCase_		:			Tuple				   =					logging.get_logger(__name__)

lowerCAmelCase_		:			List[Any]				   =					{'''vocab_file''': '''sentencepiece.bpe.model''', '''tokenizer_file''': '''tokenizer.json'''}

lowerCAmelCase_		:			List[Any]				   =					{
    '''vocab_file''': {
        '''facebook/mbart-large-50-one-to-many-mmt''': (
            '''https://huggingface.co/facebook/mbart-large-50-one-to-many-mmt/resolve/main/sentencepiece.bpe.model'''
        ),
    },
    '''tokenizer_file''': {
        '''facebook/mbart-large-50-one-to-many-mmt''': (
            '''https://huggingface.co/facebook/mbart-large-50-one-to-many-mmt/resolve/main/tokenizer.json'''
        ),
    },
}

lowerCAmelCase_		:			int				   =					{
    '''facebook/mbart-large-50-one-to-many-mmt''': 1_0_2_4,
}

# fmt: off
lowerCAmelCase_		:			List[Any]				   =					['''ar_AR''', '''cs_CZ''', '''de_DE''', '''en_XX''', '''es_XX''', '''et_EE''', '''fi_FI''', '''fr_XX''', '''gu_IN''', '''hi_IN''', '''it_IT''', '''ja_XX''', '''kk_KZ''', '''ko_KR''', '''lt_LT''', '''lv_LV''', '''my_MM''', '''ne_NP''', '''nl_XX''', '''ro_RO''', '''ru_RU''', '''si_LK''', '''tr_TR''', '''vi_VN''', '''zh_CN''', '''af_ZA''', '''az_AZ''', '''bn_IN''', '''fa_IR''', '''he_IL''', '''hr_HR''', '''id_ID''', '''ka_GE''', '''km_KH''', '''mk_MK''', '''ml_IN''', '''mn_MN''', '''mr_IN''', '''pl_PL''', '''ps_AF''', '''pt_XX''', '''sv_SE''', '''sw_KE''', '''ta_IN''', '''te_IN''', '''th_TH''', '''tl_XX''', '''uk_UA''', '''ur_PK''', '''xh_ZA''', '''gl_ES''', '''sl_SI''']





class      UpperCamelCase_       (				a_     ):
  _A				:    Union[str, Any] =		VOCAB_FILES_NAMES
  _A				:    Optional[int] =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  _A				:    Tuple =		PRETRAINED_VOCAB_FILES_MAP
  _A				:    Dict =		['input_ids', 'attention_mask']
  _A				:    Any =		MBartaaTokenizer

  _A				:    List[int] =		[]
  _A				:    List[int] =		[]



  def __init__(					self       ,						snake_case__=None       ,						snake_case__=None       ,						snake_case__=None       ,						snake_case__=None       ,						snake_case__="</s>"       ,						snake_case__="</s>"       ,						snake_case__="<s>"       ,						snake_case__="<unk>"       ,						snake_case__="<pad>"       ,						snake_case__="<mask>"       ,						**snake_case__       ,						)	->	Optional[Any]:



         """simple docstring"""





         UpperCAmelCase					=	AddedToken(snake_case__       ,						lstrip=snake_case__       ,						rstrip=snake_case__ ) if isinstance(snake_case__       ,						snake_case__ ) else mask_token

         UpperCAmelCase					=	kwargs.get("""additional_special_tokens"""       ,						[] )
         kwargs["additional_special_tokens"] += [
             code for code in FAIRSEQ_LANGUAGE_CODES if code not in kwargs["additional_special_tokens"]
         ]

         super().__init__(
             snake_case__       ,						src_lang=snake_case__       ,						tgt_lang=snake_case__       ,						tokenizer_file=snake_case__       ,						eos_token=snake_case__       ,						sep_token=snake_case__       ,						cls_token=snake_case__       ,						unk_token=snake_case__       ,						pad_token=snake_case__       ,						mask_token=snake_case__       ,						**snake_case__       ,						)

         UpperCAmelCase					=	vocab_file
         UpperCAmelCase					=	False if not self.vocab_file else True

         UpperCAmelCase					=	{
             lang_code: self.convert_tokens_to_ids(snake_case__ ) for lang_code in FAIRSEQ_LANGUAGE_CODES
         }

         UpperCAmelCase					=	src_lang if src_lang is not None else """en_XX"""
         UpperCAmelCase					=	tgt_lang
         UpperCAmelCase					=	self.lang_code_to_id[self._src_lang]
         self.set_src_lang_special_tokens(self._src_lang )



  @property
  def 						UpperCamelCase_ (					self )	->	str:



         """simple docstring"""





         return self._src_lang



  @src_lang.setter
  def 						UpperCamelCase_ (					self       ,						snake_case__ )	->	None:



         """simple docstring"""





         UpperCAmelCase					=	new_src_lang
         self.set_src_lang_special_tokens(self._src_lang )



  def 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__ = 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 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__ = "en_XX"       ,						snake_case__ = None       ,						snake_case__ = "ro_RO"       ,						**snake_case__       ,						)	->	BatchEncoding:



         """simple docstring"""





         UpperCAmelCase					=	src_lang
         UpperCAmelCase					=	tgt_lang
         return super().prepare_seqaseq_batch(snake_case__       ,						snake_case__       ,						**snake_case__ )



  def 						UpperCamelCase_ (					self )	->	int:



         """simple docstring"""





         return self.set_src_lang_special_tokens(self.src_lang )



  def 						UpperCamelCase_ (					self )	->	Optional[int]:



         """simple docstring"""





         return self.set_tgt_lang_special_tokens(self.tgt_lang )



  def 						UpperCamelCase_ (					self       ,						snake_case__ )	->	None:



         """simple docstring"""





         UpperCAmelCase					=	self.convert_tokens_to_ids(snake_case__ )
         UpperCAmelCase					=	[self.cur_lang_code_id]
         UpperCAmelCase					=	[self.eos_token_id]

         UpperCAmelCase					=	self.convert_ids_to_tokens(self.prefix_tokens )
         UpperCAmelCase					=	self.convert_ids_to_tokens(self.suffix_tokens )

         UpperCAmelCase					=	processors.TemplateProcessing(
             single=prefix_tokens_str + ["""$A"""] + suffix_tokens_str       ,						pair=prefix_tokens_str + ["""$A""", """$B"""] + suffix_tokens_str       ,						special_tokens=list(zip(prefix_tokens_str + suffix_tokens_str       ,						self.prefix_tokens + self.suffix_tokens ) )       ,						)



  def 						UpperCamelCase_ (					self       ,						snake_case__ )	->	None:



         """simple docstring"""





         UpperCAmelCase					=	self.convert_tokens_to_ids(snake_case__ )
         UpperCAmelCase					=	[self.cur_lang_code_id]
         UpperCAmelCase					=	[self.eos_token_id]

         UpperCAmelCase					=	self.convert_ids_to_tokens(self.prefix_tokens )
         UpperCAmelCase					=	self.convert_ids_to_tokens(self.suffix_tokens )

         UpperCAmelCase					=	processors.TemplateProcessing(
             single=prefix_tokens_str + ["""$A"""] + suffix_tokens_str       ,						pair=prefix_tokens_str + ["""$A""", """$B"""] + suffix_tokens_str       ,						special_tokens=list(zip(prefix_tokens_str + suffix_tokens_str       ,						self.prefix_tokens + self.suffix_tokens ) )       ,						)



  def 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						snake_case__       ,						**snake_case__ )	->	Dict:



         """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(snake_case__       ,						add_special_tokens=snake_case__       ,						return_tensors=snake_case__       ,						**snake_case__ )
         UpperCAmelCase					=	self.convert_tokens_to_ids(snake_case__ )
         UpperCAmelCase					=	tgt_lang_id
         return inputs



  def 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__ = None )	->	Tuple[str]:



         """simple docstring"""





         if not self.can_save_slow_tokenizer:
                raise ValueError(
                    """Your fast tokenizer does not have the necessary information to save the vocabulary for a slow """
                    """tokenizer.""" )

         if not os.path.isdir(snake_case__ ):
                logger.error(f'''Vocabulary path ({save_directory}) should be a directory''' )
                return
         UpperCAmelCase					=	os.path.join(
             snake_case__       ,						(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""] )

         if os.path.abspath(self.vocab_file ) != os.path.abspath(snake_case__ ):
                copyfile(self.vocab_file       ,						snake_case__ )

         return (out_vocab_file,)








"""simple docstring"""







import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    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_       (				a_      ,    a_      ,    unittest.TestCase     ):
  _A				:    str =		StableDiffusionParadigmsPipeline
  _A				:    Any =		TEXT_TO_IMAGE_PARAMS
  _A				:    List[Any] =		TEXT_TO_IMAGE_BATCH_PARAMS
  _A				:    str =		TEXT_TO_IMAGE_IMAGE_PARAMS
  _A				:    Tuple =		TEXT_TO_IMAGE_IMAGE_PARAMS



  def 						UpperCamelCase_ (					self )	->	Dict:



         """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       ,						attention_head_dim=(2, 4)       ,						use_linear_projection=snake_case__       ,						)
         UpperCAmelCase					=	DDIMParallelScheduler(
             beta_start=0.00_085       ,						beta_end=0.012       ,						beta_schedule="""scaled_linear"""       ,						clip_sample=snake_case__       ,						set_alpha_to_one=snake_case__       ,						)
         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       ,						sample_size=1_28       ,						)
         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=10_00       ,						hidden_act="""gelu"""       ,						projection_dim=5_12       ,						)
         UpperCAmelCase					=	CLIPTextModel(snake_case__ )
         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 						UpperCamelCase_ (					self       ,						snake_case__       ,						snake_case__=0 )	->	Any:



         """simple docstring"""





         if str(snake_case__ ).startswith("""mps""" ):
                UpperCAmelCase					=	torch.manual_seed(snake_case__ )
         else:
                UpperCAmelCase					=	torch.Generator(device=snake_case__ ).manual_seed(snake_case__ )
         UpperCAmelCase					=	{
             """prompt""": """a photograph of an astronaut riding a horse""",
             """generator""": generator,
             """num_inference_steps""": 10,
             """guidance_scale""": 6.0,
             """output_type""": """numpy""",
             """parallel""": 3,
             """debug""": True,
         }
         return inputs



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



         """simple docstring"""





         UpperCAmelCase					=	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         UpperCAmelCase					=	self.get_dummy_components()
         UpperCAmelCase					=	StableDiffusionParadigmsPipeline(**snake_case__ )
         UpperCAmelCase					=	sd_pipe.to(snake_case__ )
         sd_pipe.set_progress_bar_config(disable=snake_case__ )

         UpperCAmelCase					=	self.get_dummy_inputs(snake_case__ )
         UpperCAmelCase					=	sd_pipe(**snake_case__ ).images
         UpperCAmelCase					=	image[0, -3:, -3:, -1]
         assert image.shape == (1, 64, 64, 3)

         UpperCAmelCase					=	np.array([0.4_773, 0.5_417, 0.4_723, 0.4_925, 0.5_631, 0.4_752, 0.5_240, 0.4_935, 0.5_023] )

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1e-2



  def 						UpperCamelCase_ (					self )	->	List[Any]:



         """simple docstring"""





         UpperCAmelCase					=	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         UpperCAmelCase					=	self.get_dummy_components()
         torch.manual_seed(0 )
         UpperCAmelCase					=	DDPMParallelScheduler()
         torch.manual_seed(0 )
         UpperCAmelCase					=	StableDiffusionParadigmsPipeline(**snake_case__ )
         UpperCAmelCase					=	sd_pipe.to(snake_case__ )
         sd_pipe.set_progress_bar_config(disable=snake_case__ )

         UpperCAmelCase					=	self.get_dummy_inputs(snake_case__ )
         UpperCAmelCase					=	sd_pipe(**snake_case__ ).images
         UpperCAmelCase					=	image[0, -3:, -3:, -1]
         assert image.shape == (1, 64, 64, 3)

         UpperCAmelCase					=	np.array([0.3_573, 0.4_420, 0.4_960, 0.4_799, 0.3_796, 0.3_879, 0.4_819, 0.4_365, 0.4_468] )

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1e-2



  def 						UpperCamelCase_ (					self )	->	Tuple:



         """simple docstring"""





         super().test_inference_batch_consistent(batch_sizes=[1, 2] )



  def 						UpperCamelCase_ (					self )	->	Optional[int]:



         """simple docstring"""





         super().test_inference_batch_single_identical(batch_size=2       ,						expected_max_diff=3e-3 )



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



         """simple docstring"""





         UpperCAmelCase					=	"""cpu"""  # ensure determinism for the device-dependent torch.Generator
         UpperCAmelCase					=	self.get_dummy_components()
         UpperCAmelCase					=	StableDiffusionParadigmsPipeline(**snake_case__ )
         UpperCAmelCase					=	sd_pipe.to(snake_case__ )
         sd_pipe.set_progress_bar_config(disable=snake_case__ )

         UpperCAmelCase					=	self.get_dummy_inputs(snake_case__ )
         UpperCAmelCase					=	"""french fries"""
         UpperCAmelCase					=	sd_pipe(**snake_case__       ,						negative_prompt=snake_case__ )
         UpperCAmelCase					=	output.images
         UpperCAmelCase					=	image[0, -3:, -3:, -1]

         assert image.shape == (1, 64, 64, 3)

         UpperCAmelCase					=	np.array([0.4_771, 0.5_420, 0.4_683, 0.4_918, 0.5_636, 0.4_725, 0.5_230, 0.4_923, 0.5_015] )

         assert np.abs(image_slice.flatten() - expected_slice ).max() < 1e-2





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



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



         """simple docstring"""





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



  def 						UpperCamelCase_ (					self       ,						snake_case__=0 )	->	int:



         """simple docstring"""





         UpperCAmelCase					=	torch.Generator(device=snake_case__ ).manual_seed(snake_case__ )
         UpperCAmelCase					=	{
             """prompt""": """a photograph of an astronaut riding a horse""",
             """generator""": generator,
             """num_inference_steps""": 10,
             """guidance_scale""": 7.5,
             """output_type""": """numpy""",
             """parallel""": 3,
             """debug""": True,
         }
         return inputs



  def 						UpperCamelCase_ (					self )	->	Tuple:



         """simple docstring"""





         UpperCAmelCase					=	"""stabilityai/stable-diffusion-2-base"""
         UpperCAmelCase					=	DDIMParallelScheduler.from_pretrained(snake_case__       ,						subfolder="""scheduler""" )
         UpperCAmelCase					=	StableDiffusionParadigmsPipeline.from_pretrained(snake_case__       ,						scheduler=snake_case__       ,						safety_checker=snake_case__ )
         pipe.to(snake_case__ )
         pipe.set_progress_bar_config(disable=snake_case__ )
         pipe.enable_attention_slicing()

         UpperCAmelCase					=	self.get_inputs()
         UpperCAmelCase					=	pipe(**snake_case__ ).images
         UpperCAmelCase					=	image[0, -3:, -3:, -1].flatten()

         assert image.shape == (1, 5_12, 5_12, 3)

         UpperCAmelCase					=	np.array([0.9_622, 0.9_602, 0.9_748, 0.9_591, 0.9_630, 0.9_691, 0.9_661, 0.9_631, 0.9_741] )

         assert np.abs(expected_slice - image_slice ).max() < 1e-2
