




"""simple docstring"""

import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()







@skip_mps
class    __lowercase     (					_UpperCAmelCase					,    unittest.TestCase):







						"""simple docstring"""







						_A		:							List[str]							       =		TextToVideoSDPipeline
						_A		:							Tuple							       =		TEXT_TO_IMAGE_PARAMS
						_A		:							str							       =		TEXT_TO_IMAGE_BATCH_PARAMS
						# No `output_type`.
						_A		:							List[Any]							       =		frozenset(
						    [
						        """num_inference_steps""",
						        """generator""",
						        """latents""",
						        """return_dict""",
						        """callback""",
						        """callback_steps""",
						    ])







						def 						__UpperCamelCase							(self       ):
											torch.manual_seed(0       )
											snake_case_					:		Optional[Any] 					=					UNetaDConditionModel(
											    block_out_channels=(32, 64, 64, 64)   ,		layers_per_block=2   ,		sample_size=32   ,		in_channels=4   ,		out_channels=4   ,		down_block_types=("""CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """CrossAttnDownBlock3D""", """DownBlock3D""")   ,		up_block_types=("""UpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""", """CrossAttnUpBlock3D""")   ,		cross_attention_dim=32   ,		attention_head_dim=4   ,		)
											snake_case_					:		str 					=					DDIMScheduler(
											    beta_start=0.00085   ,		beta_end=0.012   ,		beta_schedule="""scaled_linear"""   ,		clip_sample=__A   ,		set_alpha_to_one=__A   ,		)
											torch.manual_seed(0       )
											snake_case_					:		List[str] 					=					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       )
											snake_case_					:		Union[str, Any] 					=					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   ,		)
											snake_case_					:		Any 					=					CLIPTextModel(__A       )
											snake_case_					:		Tuple 					=					CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""       )

											snake_case_					:		Optional[int] 					=					{
											    """unet""": unet,
											    """scheduler""": scheduler,
											    """vae""": vae,
											    """text_encoder""": text_encoder,
											    """tokenizer""": tokenizer,
											}
											return components







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__=0       ):
											if str(__A       ).startswith("""mps"""       ):
																snake_case_					:		List[str] 					=					torch.manual_seed(__A       )
											else:
																snake_case_					:		Optional[Any] 					=					torch.Generator(device=__A       ).manual_seed(__A       )
											snake_case_					:		Optional[int] 					=					{
											    """prompt""": """A painting of a squirrel eating a burger""",
											    """generator""": generator,
											    """num_inference_steps""": 2,
											    """guidance_scale""": 6.0,
											    """output_type""": """pt""",
											}
											return inputs







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Tuple 					=					"""cpu"""  # ensure determinism for the device-dependent torch.Generator
											snake_case_					:		Dict 					=					self.get_dummy_components()
											snake_case_					:		List[str] 					=					TextToVideoSDPipeline(**__A       )
											snake_case_					:		Union[str, Any] 					=					sd_pipe.to(__A       )
											sd_pipe.set_progress_bar_config(disable=__A       )

											snake_case_					:		str 					=					self.get_dummy_inputs(__A       )
											snake_case_					:		Any 					=					"""np"""
											snake_case_					:		Any 					=					sd_pipe(**__A       ).frames
											snake_case_					:		Any 					=					frames[0][-3:, -3:, -1]

											assert frames[0].shape == (64, 64, 3)
											snake_case_					:		Tuple 					=					np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0]       )

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







						def 						__UpperCamelCase							(self       ):
											self._test_attention_slicing_forward_pass(test_mean_pixel_difference=__A   ,		expected_max_diff=3e-3       )







						@unittest.skipIf(
						    torch_device != """cuda""" or not is_xformers_available()   ,		reason="""XFormers attention is only available with CUDA and `xformers` installed"""   ,		)
						def 						__UpperCamelCase							(self       ):
											self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=__A   ,		expected_max_diff=1e-2       )







						@unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""       )
						def 						__UpperCamelCase							(self       ):
											pass







						@unittest.skip(reason="""Batching needs to be properly figured out first for this pipeline."""       )
						def 						__UpperCamelCase							(self       ):
											pass







						@unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline."""       )
						def 						__UpperCamelCase							(self       ):
											pass







						def 						__UpperCamelCase							(self       ):
											return super().test_progress_bar()







@slow
@skip_mps
class    __lowercase     (					unittest.TestCase):







						"""simple docstring"""







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		str 					=					load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy"""       )

											snake_case_					:		Optional[Any] 					=					TextToVideoSDPipeline.from_pretrained("""damo-vilab/text-to-video-ms-1.7b"""       )
											snake_case_					:		str 					=					DPMSolverMultistepScheduler.from_config(pipe.scheduler.config       )
											snake_case_					:		Union[str, Any] 					=					pipe.to("""cuda"""       )

											snake_case_					:		Optional[int] 					=					"""Spiderman is surfing"""
											snake_case_					:		List[str] 					=					torch.Generator(device="""cpu"""       ).manual_seed(0       )

											snake_case_					:		Dict 					=					pipe(__A   ,		generator=__A   ,		num_inference_steps=25   ,		output_type="""pt"""       ).frames
											snake_case_					:		int 					=					video_frames.cpu().numpy()

											assert np.abs(expected_video - video       ).mean() < 5e-2







						def 						__UpperCamelCase							(self       ):
											snake_case_					:		Tuple 					=					load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy"""       )

											snake_case_					:		List[Any] 					=					TextToVideoSDPipeline.from_pretrained("""damo-vilab/text-to-video-ms-1.7b"""       )
											snake_case_					:		List[Any] 					=					pipe.to("""cuda"""       )

											snake_case_					:		Optional[int] 					=					"""Spiderman is surfing"""
											snake_case_					:		Optional[Any] 					=					torch.Generator(device="""cpu"""       ).manual_seed(0       )

											snake_case_					:		Dict 					=					pipe(__A   ,		generator=__A   ,		num_inference_steps=2   ,		output_type="""pt"""       ).frames
											snake_case_					:		int 					=					video_frames.cpu().numpy()

											assert np.abs(expected_video - video       ).mean() < 5e-2









"""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_mbart import MBartTokenizer
else:
			a_			   =					None


a_			   =					logging.get_logger(__name__)


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

a_			   =					{
    '''vocab_file''': {
        '''facebook/mbart-large-en-ro''': (
            '''https://huggingface.co/facebook/mbart-large-en-ro/resolve/main/sentencepiece.bpe.model'''
        ),
        '''facebook/mbart-large-cc25''': (
            '''https://huggingface.co/facebook/mbart-large-cc25/resolve/main/sentencepiece.bpe.model'''
        ),
    },
    '''tokenizer_file''': {
        '''facebook/mbart-large-en-ro''': '''https://huggingface.co/facebook/mbart-large-en-ro/resolve/main/tokenizer.json''',
        '''facebook/mbart-large-cc25''': '''https://huggingface.co/facebook/mbart-large-cc25/resolve/main/tokenizer.json''',
    },
}

a_			   =					{
    '''facebook/mbart-large-en-ro''': 1024,
    '''facebook/mbart-large-cc25''': 1024,
}

# fmt: off
a_			   =					['''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''']







class    __lowercase     (					_UpperCAmelCase):







						"""simple docstring"""







						_A		:							Dict							       =		VOCAB_FILES_NAMES
						_A		:							Union[str, Any]							       =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
						_A		:							Optional[int]							       =		PRETRAINED_VOCAB_FILES_MAP
						_A		:							str							       =		["""input_ids""", """attention_mask"""]
						_A		:							Tuple							       =		MBartTokenizer

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







						def __init__(self   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__="<s>"   ,		lowercase__="</s>"   ,		lowercase__="</s>"   ,		lowercase__="<s>"   ,		lowercase__="<unk>"   ,		lowercase__="<pad>"   ,		lowercase__="<mask>"   ,		lowercase__=None   ,		lowercase__=None   ,		lowercase__=None   ,		**lowercase__   ,		):
											# Mask token behave like a normal word, i.e. include the space before it
											snake_case_					:		int 					=					AddedToken(lowercase__   ,		lstrip=lowercase__   ,		rstrip=lowercase__       ) if isinstance(lowercase__   ,		lowercase__       ) else mask_token

											super().__init__(
											    vocab_file=lowercase__   ,		tokenizer_file=lowercase__   ,		bos_token=lowercase__   ,		eos_token=lowercase__   ,		sep_token=lowercase__   ,		cls_token=lowercase__   ,		unk_token=lowercase__   ,		pad_token=lowercase__   ,		mask_token=lowercase__   ,		src_lang=lowercase__   ,		tgt_lang=lowercase__   ,		additional_special_tokens=lowercase__   ,		**lowercase__   ,		)

											snake_case_					:		Dict 					=					vocab_file
											snake_case_					:		Optional[int] 					=					False if not self.vocab_file else True

											snake_case_					:		Optional[Any] 					=					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}       )
											snake_case_					:		Any 					=					{
											    lang_code: self.convert_tokens_to_ids(lowercase__       ) for lang_code in FAIRSEQ_LANGUAGE_CODES
											}

											snake_case_					:		Tuple 					=					src_lang if src_lang is not None else """en_XX"""
											snake_case_					:		Tuple 					=					self.convert_tokens_to_ids(self._src_lang       )
											snake_case_					:		Tuple 					=					tgt_lang
											self.set_src_lang_special_tokens(self._src_lang       )







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







						@src_lang.setter
						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		Tuple 					=					new_src_lang
											self.set_src_lang_special_tokens(self._src_lang       )







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__ = 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 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__ = None       ):
											snake_case_					:		List[Any] 					=					[self.sep_token_id]
											snake_case_					:		Optional[int] 					=					[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 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__   ,		lowercase__   ,		lowercase__   ,		**lowercase__       ):
											if src_lang is None or tgt_lang is None:
																raise ValueError("""Translation requires a `src_lang` and a `tgt_lang` for this model"""       )
											snake_case_					:		int 					=					src_lang
											snake_case_					:		List[str] 					=					self(lowercase__   ,		add_special_tokens=lowercase__   ,		return_tensors=lowercase__   ,		**lowercase__       )
											snake_case_					:		List[str] 					=					self.convert_tokens_to_ids(lowercase__       )
											snake_case_					:		Union[str, Any] 					=					tgt_lang_id
											return inputs







						def 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__ = "en_XX"   ,		lowercase__ = None   ,		lowercase__ = "ro_RO"   ,		**lowercase__   ,		):
											snake_case_					:		List[str] 					=					src_lang
											snake_case_					:		int 					=					tgt_lang
											return super().prepare_seqaseq_batch(lowercase__   ,		lowercase__   ,		**lowercase__       )







						def 						__UpperCamelCase							(self       ):
											return self.set_src_lang_special_tokens(self.src_lang       )







						def 						__UpperCamelCase							(self       ):
											return self.set_tgt_lang_special_tokens(self.tgt_lang       )







						def 						__UpperCamelCase							(self   ,		lowercase__       ):
											snake_case_					:		int 					=					self.convert_tokens_to_ids(lowercase__       )
											snake_case_					:		Tuple 					=					[]
											snake_case_					:		List[Any] 					=					[self.eos_token_id, self.cur_lang_code]

											snake_case_					:		List[Any] 					=					self.convert_ids_to_tokens(self.prefix_tokens       )
											snake_case_					:		Union[str, Any] 					=					self.convert_ids_to_tokens(self.suffix_tokens       )

											snake_case_					:		Optional[int] 					=					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   ,		lowercase__       ):
											snake_case_					:		Tuple 					=					self.convert_tokens_to_ids(lowercase__       )
											snake_case_					:		Optional[int] 					=					[]
											snake_case_					:		Optional[Any] 					=					[self.eos_token_id, self.cur_lang_code]

											snake_case_					:		Union[str, Any] 					=					self.convert_ids_to_tokens(self.prefix_tokens       )
											snake_case_					:		int 					=					self.convert_ids_to_tokens(self.suffix_tokens       )

											snake_case_					:		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 						__UpperCamelCase							(self   ,		lowercase__   ,		lowercase__ = None       ):
											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(lowercase__       ):
																logger.error(f'Vocabulary path ({save_directory}) should be a directory.'       )
																return
											snake_case_					:		List[str] 					=					os.path.join(
											    lowercase__   ,		(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]       )

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

											return (out_vocab_file,)




