



"""simple docstring"""






import gc
import random
import unittest

import numpy as np
import torch

from diffusers import DDIMScheduler, KandinskyVaaPipeline, KandinskyVaaPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()






class        __snake_case       (	_lowercase  ,     unittest.TestCase):
  snake_case__				:   Any  			=						KandinskyVaaPipeline
  snake_case__				:   List[str]  			=						[
      "image_embeds",
      "negative_image_embeds",
  ]
  snake_case__				:   Optional[int]  			=						["image_embeds", "negative_image_embeds"]
  snake_case__				:   Any  			=						[
      "generator",
      "height",
      "width",
      "latents",
      "guidance_scale",
      "num_inference_steps",
      "return_dict",
      "guidance_scale",
      "num_images_per_prompt",
      "output_type",
      "return_dict",
  ]
  snake_case__				:   Union[str, Any]  			=						False




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						Dict		):






         """simple docstring"""






         return 3_2




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						Dict		):






         """simple docstring"""






         return 3_2




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						Dict		):






         """simple docstring"""






         return self.time_input_dim




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						Tuple		):






         """simple docstring"""






         return self.time_input_dim * 4




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						Optional[Any]		):






         """simple docstring"""






         return 1_0_0




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						str		):






         """simple docstring"""






         torch.manual_seed(0		)

         _lowerCamelCase						:				Any    			= {
             '''in_channels''': 4,
             # Out channels is double in channels because predicts mean and variance
             '''out_channels''': 8,
             '''addition_embed_type''': '''image''',
             '''down_block_types''': ('''ResnetDownsampleBlock2D''', '''SimpleCrossAttnDownBlock2D'''),
             '''up_block_types''': ('''SimpleCrossAttnUpBlock2D''', '''ResnetUpsampleBlock2D'''),
             '''mid_block_type''': '''UNetMidBlock2DSimpleCrossAttn''',
             '''block_out_channels''': (self.block_out_channels_a, self.block_out_channels_a * 2),
             '''layers_per_block''': 1,
             '''encoder_hid_dim''': self.text_embedder_hidden_size,
             '''encoder_hid_dim_type''': '''image_proj''',
             '''cross_attention_dim''': self.cross_attention_dim,
             '''attention_head_dim''': 4,
             '''resnet_time_scale_shift''': '''scale_shift''',
             '''class_embed_type''': None,
         }

         _lowerCamelCase						:				Tuple    			= UNetaDConditionModel(**__lowerCAmelCase		)
         return model




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						Tuple		):






         """simple docstring"""






         return {
             "block_out_channels": [3_2, 6_4],
             "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
             "in_channels": 3,
             "latent_channels": 4,
             "layers_per_block": 1,
             "norm_num_groups": 8,
             "norm_type": "spatial",
             "num_vq_embeddings": 1_2,
             "out_channels": 3,
             "up_block_types": [
                 "AttnUpDecoderBlock2D",
                 "UpDecoderBlock2D",
             ],
             "vq_embed_dim": 4,
         }




  @property
  def 						SCREAMING_SNAKE_CASE				( self     :						List[Any]		):






         """simple docstring"""






         torch.manual_seed(0		)
         _lowerCamelCase						:				Any    			= VQModel(**self.dummy_movq_kwargs		)
         return model




  def 						SCREAMING_SNAKE_CASE				( self     :						Union[str, Any]		):






         """simple docstring"""






         _lowerCamelCase						:				Optional[Any]    			= self.dummy_unet
         _lowerCamelCase						:				Optional[Any]    			= self.dummy_movq

         _lowerCamelCase						:				Optional[int]    			= DDIMScheduler(
             num_train_timesteps=1_0_0_0						,					beta_schedule='''linear'''						,					beta_start=0.0_00_85						,					beta_end=0.0_12						,					clip_sample=__lowerCAmelCase						,					set_alpha_to_one=__lowerCAmelCase						,					steps_offset=1						,					prediction_type='''epsilon'''						,					thresholding=__lowerCAmelCase						,					)

         _lowerCamelCase						:				Optional[int]    			= {
             '''unet''': unet,
             '''scheduler''': scheduler,
             '''movq''': movq,
         }
         return components




  def 						SCREAMING_SNAKE_CASE				( self     :						Union[str, Any]						,					__lowerCAmelCase     :						Tuple						,					__lowerCAmelCase     :						Optional[Any]=0		):






         """simple docstring"""






         _lowerCamelCase						:				Tuple    			= floats_tensor((1, self.text_embedder_hidden_size)						,					rng=random.Random(__lowerCAmelCase		)		).to(__lowerCAmelCase		)
         _lowerCamelCase						:				Union[str, Any]    			= floats_tensor((1, self.text_embedder_hidden_size)						,					rng=random.Random(seed + 1		)		).to(
             __lowerCAmelCase		)
         if str(__lowerCAmelCase		).startswith('''mps'''		):
                _lowerCamelCase						:				Union[str, Any]    			= torch.manual_seed(__lowerCAmelCase		)
         else:
                _lowerCamelCase						:				Any    			= torch.Generator(device=__lowerCAmelCase		).manual_seed(__lowerCAmelCase		)
         _lowerCamelCase						:				List[str]    			= {
             '''image_embeds''': image_embeds,
             '''negative_image_embeds''': negative_image_embeds,
             '''generator''': generator,
             '''height''': 6_4,
             '''width''': 6_4,
             '''guidance_scale''': 4.0,
             '''num_inference_steps''': 2,
             '''output_type''': '''np''',
         }
         return inputs




  def 						SCREAMING_SNAKE_CASE				( self     :						Any		):






         """simple docstring"""






         _lowerCamelCase						:				Optional[Any]    			= '''cpu'''

         _lowerCamelCase						:				Any    			= self.get_dummy_components()

         _lowerCamelCase						:				Union[str, Any]    			= self.pipeline_class(**__lowerCAmelCase		)
         _lowerCamelCase						:				Tuple    			= pipe.to(__lowerCAmelCase		)

         pipe.set_progress_bar_config(disable=__lowerCAmelCase		)

         _lowerCamelCase						:				Optional[int]    			= pipe(**self.get_dummy_inputs(__lowerCAmelCase		)		)
         _lowerCamelCase						:				List[Any]    			= output.images

         _lowerCamelCase						:				str    			= pipe(
             **self.get_dummy_inputs(__lowerCAmelCase		)						,					return_dict=__lowerCAmelCase						,					)[0]

         _lowerCamelCase						:				Any    			= image[0, -3:, -3:, -1]
         _lowerCamelCase						:				Optional[Any]    			= image_from_tuple[0, -3:, -3:, -1]

         assert image.shape == (1, 6_4, 6_4, 3)

         _lowerCamelCase						:				int    			= np.array(
             [0.6_23_79_76, 1.0, 0.36_44_13_32, 1.0, 0.70_63_96_34, 0.29_87_71_86, 0.85_65_21_25, 0.5_21_68_43, 0.54_45_40_46]		)

         assert (
             np.abs(image_slice.flatten() - expected_slice		).max() < 1E-2
         ), f''' expected_slice {expected_slice}, but got {image_slice.flatten()}'''

         assert (
             np.abs(image_from_tuple_slice.flatten() - expected_slice		).max() < 1E-2
         ), f''' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'''






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




  def 						SCREAMING_SNAKE_CASE				( self     :						List[Any]		):






         """simple docstring"""






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




  def 						SCREAMING_SNAKE_CASE				( self     :						Union[str, Any]		):






         """simple docstring"""






         _lowerCamelCase						:				List[Any]    			= load_numpy(
             '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
             '''/kandinskyv22/kandinskyv22_text2img_cat_fp16.npy'''		)

         _lowerCamelCase						:				Dict    			= KandinskyVaaPriorPipeline.from_pretrained(
             '''kandinsky-community/kandinsky-2-2-prior'''						,					torch_dtype=torch.floataa		)
         pipe_prior.to(__lowerCAmelCase		)

         _lowerCamelCase						:				Any    			= KandinskyVaaPipeline.from_pretrained(
             '''kandinsky-community/kandinsky-2-2-decoder'''						,					torch_dtype=torch.floataa		)
         _lowerCamelCase						:				int    			= pipeline.to(__lowerCAmelCase		)
         pipeline.set_progress_bar_config(disable=__lowerCAmelCase		)

         _lowerCamelCase						:				int    			= '''red cat, 4k photo'''

         _lowerCamelCase						:				Any    			= torch.Generator(device='''cuda'''		).manual_seed(0		)
         _lowerCamelCase			,    _lowerCamelCase						:				int    			= pipe_prior(
             __lowerCAmelCase						,					generator=__lowerCAmelCase						,					num_inference_steps=5						,					negative_prompt=''''''						,					).to_tuple()

         _lowerCamelCase						:				List[Any]    			= torch.Generator(device='''cuda'''		).manual_seed(0		)
         _lowerCamelCase						:				Dict    			= pipeline(
             image_embeds=__lowerCAmelCase						,					negative_image_embeds=__lowerCAmelCase						,					generator=__lowerCAmelCase						,					num_inference_steps=1_0_0						,					output_type='''np'''						,					)

         _lowerCamelCase						:				Optional[int]    			= output.images[0]

         assert image.shape == (5_1_2, 5_1_2, 3)

         assert_mean_pixel_difference(__lowerCAmelCase						,					__lowerCAmelCase		)






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_nllb import NllbTokenizer
else:
    lowerCAmelCase__      		= None


lowerCAmelCase__      		= logging.get_logger(__name__)


lowerCAmelCase__      		= {'vocab_file': 'sentencepiece.bpe.model', 'tokenizer_file': 'tokenizer.json'}

lowerCAmelCase__      		= {
    'vocab_file': {
        'facebook/nllb-200-distilled-600M': (
            'https://huggingface.co/facebook/nllb-200-distilled-600M/resolve/main/sentencepiece.bpe.model'
        ),
    },
    'tokenizer_file': {
        'facebook/nllb-200-distilled-600M': (
            'https://huggingface.co/facebook/nllb-200-distilled-600M/resolve/main/tokenizer.json'
        ),
    },
}

lowerCAmelCase__      		= {
    'facebook/nllb-large-en-ro': 10_24,
    'facebook/nllb-200-distilled-600M': 10_24,
}

# fmt: off
lowerCAmelCase__      		= ['ace_Arab', 'ace_Latn', 'acm_Arab', 'acq_Arab', 'aeb_Arab', 'afr_Latn', 'ajp_Arab', 'aka_Latn', 'amh_Ethi', 'apc_Arab', 'arb_Arab', 'ars_Arab', 'ary_Arab', 'arz_Arab', 'asm_Beng', 'ast_Latn', 'awa_Deva', 'ayr_Latn', 'azb_Arab', 'azj_Latn', 'bak_Cyrl', 'bam_Latn', 'ban_Latn', 'bel_Cyrl', 'bem_Latn', 'ben_Beng', 'bho_Deva', 'bjn_Arab', 'bjn_Latn', 'bod_Tibt', 'bos_Latn', 'bug_Latn', 'bul_Cyrl', 'cat_Latn', 'ceb_Latn', 'ces_Latn', 'cjk_Latn', 'ckb_Arab', 'crh_Latn', 'cym_Latn', 'dan_Latn', 'deu_Latn', 'dik_Latn', 'dyu_Latn', 'dzo_Tibt', 'ell_Grek', 'eng_Latn', 'epo_Latn', 'est_Latn', 'eus_Latn', 'ewe_Latn', 'fao_Latn', 'pes_Arab', 'fij_Latn', 'fin_Latn', 'fon_Latn', 'fra_Latn', 'fur_Latn', 'fuv_Latn', 'gla_Latn', 'gle_Latn', 'glg_Latn', 'grn_Latn', 'guj_Gujr', 'hat_Latn', 'hau_Latn', 'heb_Hebr', 'hin_Deva', 'hne_Deva', 'hrv_Latn', 'hun_Latn', 'hye_Armn', 'ibo_Latn', 'ilo_Latn', 'ind_Latn', 'isl_Latn', 'ita_Latn', 'jav_Latn', 'jpn_Jpan', 'kab_Latn', 'kac_Latn', 'kam_Latn', 'kan_Knda', 'kas_Arab', 'kas_Deva', 'kat_Geor', 'knc_Arab', 'knc_Latn', 'kaz_Cyrl', 'kbp_Latn', 'kea_Latn', 'khm_Khmr', 'kik_Latn', 'kin_Latn', 'kir_Cyrl', 'kmb_Latn', 'kon_Latn', 'kor_Hang', 'kmr_Latn', 'lao_Laoo', 'lvs_Latn', 'lij_Latn', 'lim_Latn', 'lin_Latn', 'lit_Latn', 'lmo_Latn', 'ltg_Latn', 'ltz_Latn', 'lua_Latn', 'lug_Latn', 'luo_Latn', 'lus_Latn', 'mag_Deva', 'mai_Deva', 'mal_Mlym', 'mar_Deva', 'min_Latn', 'mkd_Cyrl', 'plt_Latn', 'mlt_Latn', 'mni_Beng', 'khk_Cyrl', 'mos_Latn', 'mri_Latn', 'zsm_Latn', 'mya_Mymr', 'nld_Latn', 'nno_Latn', 'nob_Latn', 'npi_Deva', 'nso_Latn', 'nus_Latn', 'nya_Latn', 'oci_Latn', 'gaz_Latn', 'ory_Orya', 'pag_Latn', 'pan_Guru', 'pap_Latn', 'pol_Latn', 'por_Latn', 'prs_Arab', 'pbt_Arab', 'quy_Latn', 'ron_Latn', 'run_Latn', 'rus_Cyrl', 'sag_Latn', 'san_Deva', 'sat_Beng', 'scn_Latn', 'shn_Mymr', 'sin_Sinh', 'slk_Latn', 'slv_Latn', 'smo_Latn', 'sna_Latn', 'snd_Arab', 'som_Latn', 'sot_Latn', 'spa_Latn', 'als_Latn', 'srd_Latn', 'srp_Cyrl', 'ssw_Latn', 'sun_Latn', 'swe_Latn', 'swh_Latn', 'szl_Latn', 'tam_Taml', 'tat_Cyrl', 'tel_Telu', 'tgk_Cyrl', 'tgl_Latn', 'tha_Thai', 'tir_Ethi', 'taq_Latn', 'taq_Tfng', 'tpi_Latn', 'tsn_Latn', 'tso_Latn', 'tuk_Latn', 'tum_Latn', 'tur_Latn', 'twi_Latn', 'tzm_Tfng', 'uig_Arab', 'ukr_Cyrl', 'umb_Latn', 'urd_Arab', 'uzn_Latn', 'vec_Latn', 'vie_Latn', 'war_Latn', 'wol_Latn', 'xho_Latn', 'ydd_Hebr', 'yor_Latn', 'yue_Hant', 'zho_Hans', 'zho_Hant', 'zul_Latn']







class      lowerCAmelCase__       (				a):




     '''simple docstring'''
     __SCREAMING_SNAKE_CASE  			= VOCAB_FILES_NAMES
     __SCREAMING_SNAKE_CASE  			= PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     __SCREAMING_SNAKE_CASE  			= PRETRAINED_VOCAB_FILES_MAP
     __SCREAMING_SNAKE_CASE  			= ["input_ids", "attention_mask"]
     __SCREAMING_SNAKE_CASE  			= NllbTokenizer

     __SCREAMING_SNAKE_CASE  			= []
     __SCREAMING_SNAKE_CASE  			= []


     def __init__(						self ,						__lowerCamelCase=None ,						__lowerCamelCase=None ,						__lowerCamelCase="<s>" ,						__lowerCamelCase="</s>" ,						__lowerCamelCase="</s>" ,						__lowerCamelCase="<s>" ,						__lowerCamelCase="<unk>" ,						__lowerCamelCase="<pad>" ,						__lowerCamelCase="<mask>" ,						__lowerCamelCase=None ,						__lowerCamelCase=None ,						__lowerCamelCase=None ,						__lowerCamelCase=False ,						**__lowerCamelCase ,						)			->				Tuple:
            # Mask token behave like a normal word, i.e. include the space before it
            _A      :       Any				    = AddedToken(__lowerCamelCase ,						lstrip=__lowerCamelCase ,						rstrip=__lowerCamelCase) if isinstance(__lowerCamelCase ,						__lowerCamelCase) else mask_token
            _A      :       Optional[int]				    = legacy_behaviour
            super().__init__(
                vocab_file=__lowerCamelCase ,						tokenizer_file=__lowerCamelCase ,						bos_token=__lowerCamelCase ,						eos_token=__lowerCamelCase ,						sep_token=__lowerCamelCase ,						cls_token=__lowerCamelCase ,						unk_token=__lowerCamelCase ,						pad_token=__lowerCamelCase ,						mask_token=__lowerCamelCase ,						src_lang=__lowerCamelCase ,						tgt_lang=__lowerCamelCase ,						additional_special_tokens=__lowerCamelCase ,						legacy_behaviour=__lowerCamelCase ,						**__lowerCamelCase ,						)

            _A      :       int				    = vocab_file
            _A      :       Optional[Any]				    = False if not self.vocab_file else True

            _A      :       Tuple				    = FAIRSEQ_LANGUAGE_CODES.copy()

            if additional_special_tokens is not None:
                   # Only add those special tokens if they are not already there.
                   _additional_special_tokens.extend(
                       [t for t in additional_special_tokens if t not in _additional_special_tokens])

            self.add_special_tokens({"additional_special_tokens": _additional_special_tokens})
            _A      :       Union[str, Any]				    = {
                lang_code: self.convert_tokens_to_ids(__lowerCamelCase) for lang_code in FAIRSEQ_LANGUAGE_CODES
            }

            _A      :       Optional[int]				    = src_lang if src_lang is not None else "eng_Latn"
            _A      :       Union[str, Any]				    = self.convert_tokens_to_ids(self._src_lang)
            _A      :       List[str]				    = tgt_lang
            self.set_src_lang_special_tokens(self._src_lang)


     @property
     def  _lowerCamelCase (						self)			->				str:
            return self._src_lang


     @src_lang.setter
     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				None:
            _A      :       Tuple				    = new_src_lang
            self.set_src_lang_special_tokens(self._src_lang)


     def  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase = None)			->				List[int]:
            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  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase = None)			->				List[int]:
            _A      :       Tuple				    = [self.sep_token_id]
            _A      :       List[Any]				    = [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 + sep + token_ids_a + sep) * [0]


     def  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						**__lowerCamelCase)			->				Optional[int]:
            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      :       List[Any]				    = src_lang
            _A      :       Optional[int]				    = self(__lowerCamelCase ,						add_special_tokens=__lowerCamelCase ,						return_tensors=__lowerCamelCase ,						**__lowerCamelCase)
            _A      :       Tuple				    = self.convert_tokens_to_ids(__lowerCamelCase)
            _A      :       Tuple				    = tgt_lang_id
            return inputs


     def  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase = "eng_Latn" ,						__lowerCamelCase = None ,						__lowerCamelCase = "fra_Latn" ,						**__lowerCamelCase ,						)			->				BatchEncoding:
            _A      :       Tuple				    = src_lang
            _A      :       int				    = tgt_lang
            return super().prepare_seqaseq_batch(__lowerCamelCase ,						__lowerCamelCase ,						**__lowerCamelCase)


     def  _lowerCamelCase (						self)			->				str:
            return self.set_src_lang_special_tokens(self.src_lang)


     def  _lowerCamelCase (						self)			->				List[str]:
            return self.set_tgt_lang_special_tokens(self.tgt_lang)


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				None:
            _A      :       Dict				    = self.convert_tokens_to_ids(__lowerCamelCase)

            if self.legacy_behaviour:
                   _A      :       List[str]				    = []
                   _A      :       Dict				    = [self.eos_token_id, self.cur_lang_code]
            else:
                   _A      :       Tuple				    = [self.cur_lang_code]
                   _A      :       Optional[Any]				    = [self.eos_token_id]

            _A      :       Optional[int]				    = self.convert_ids_to_tokens(self.prefix_tokens)
            _A      :       int				    = self.convert_ids_to_tokens(self.suffix_tokens)

            _A      :       List[Any]				    = 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  _lowerCamelCase (						self ,						__lowerCamelCase)			->				None:
            _A      :       Optional[Any]				    = self.convert_tokens_to_ids(__lowerCamelCase)
            if self.legacy_behaviour:
                   _A      :       Tuple				    = []
                   _A      :       Any				    = [self.eos_token_id, self.cur_lang_code]
            else:
                   _A      :       Union[str, Any]				    = [self.cur_lang_code]
                   _A      :       str				    = [self.eos_token_id]

            _A      :       Optional[Any]				    = self.convert_ids_to_tokens(self.prefix_tokens)
            _A      :       Dict				    = self.convert_ids_to_tokens(self.suffix_tokens)

            _A      :       Union[str, Any]				    = 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  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase = None)			->				Tuple[str]:
            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(__lowerCamelCase):
                   logger.error(F"Vocabulary path ({save_directory}) should be a directory.")
                   return
            _A      :       Dict				    = 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):
                   copyfile(self.vocab_file ,						__lowerCamelCase)

            return (out_vocab_file,)
