


'''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 (	__lowercase						,					unittest.TestCase							):
   _SCREAMING_SNAKE_CASE  :  str          =	TextToVideoSDPipeline
   _SCREAMING_SNAKE_CASE  :  Tuple          =	TEXT_TO_IMAGE_PARAMS
   _SCREAMING_SNAKE_CASE  :  List[str]          =	TEXT_TO_IMAGE_BATCH_PARAMS
   # No `output_type`.
   _SCREAMING_SNAKE_CASE  :  List[str]          =	frozenset(
       [
           "num_inference_steps",
           "generator",
           "latents",
           "return_dict",
           "callback",
           "callback_steps",
       ]							)
   def a  (						self       :      Union[str, Any]      ) ->							Dict:
        torch.manual_seed(0      )
        __snake_case           =						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           =						DDIMScheduler(
            beta_start=0.0_0_0_8_5       ,   beta_end=0.0_1_2       ,   beta_schedule='scaled_linear'       ,   clip_sample=SCREAMING_SNAKE_CASE_       ,   set_alpha_to_one=SCREAMING_SNAKE_CASE_       ,   )
        torch.manual_seed(0      )
        __snake_case           =						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      )
        __snake_case           =						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       ,   hidden_act='gelu'       ,   projection_dim=512       ,   )
        __snake_case           =						CLIPTextModel(SCREAMING_SNAKE_CASE_      )
        __snake_case           =						CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'      )

        __snake_case           =						{
            'unet': unet,
            'scheduler': scheduler,
            'vae': vae,
            'text_encoder': text_encoder,
            'tokenizer': tokenizer,
        }
        return components
   def a  (						self       :      Any       ,   SCREAMING_SNAKE_CASE_       :      Optional[int]       ,   SCREAMING_SNAKE_CASE_       :      Union[str, Any]=0      ) ->							Union[str, Any]:
        if str(SCREAMING_SNAKE_CASE_      ).startswith('mps'      ):
             __snake_case           =						torch.manual_seed(SCREAMING_SNAKE_CASE_      )
        else:
             __snake_case           =						torch.Generator(device=SCREAMING_SNAKE_CASE_      ).manual_seed(SCREAMING_SNAKE_CASE_      )
        __snake_case           =						{
            '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 a  (						self       :      Any      ) ->							Union[str, Any]:
        __snake_case           =						'cpu'  # ensure determinism for the device-dependent torch.Generator
        __snake_case           =						self.get_dummy_components()
        __snake_case           =						TextToVideoSDPipeline(**SCREAMING_SNAKE_CASE_      )
        __snake_case           =						sd_pipe.to(SCREAMING_SNAKE_CASE_      )
        sd_pipe.set_progress_bar_config(disable=SCREAMING_SNAKE_CASE_      )

        __snake_case           =						self.get_dummy_inputs(SCREAMING_SNAKE_CASE_      )
        __snake_case           =						'np'
        __snake_case           =						sd_pipe(**SCREAMING_SNAKE_CASE_      ).frames
        __snake_case           =						frames[0][-3:, -3:, -1]

        assert frames[0].shape == (64, 64, 3)
        __snake_case           =						np.array([1_5_8.0, 1_6_0.0, 1_5_3.0, 1_2_5.0, 1_0_0.0, 1_2_1.0, 1_1_1.0, 9_3.0, 1_1_3.0]      )

        assert np.abs(image_slice.flatten() - expected_slice      ).max() < 1e-2
   def a  (						self       :      Optional[int]      ) ->							str:
        self._test_attention_slicing_forward_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE_       ,   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 a  (						self       :      Any      ) ->							int:
        self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE_       ,   expected_max_diff=1e-2      )
   @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'      )
   def a  (						self       :      List[str]      ) ->							Optional[int]:
        pass
   @unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'      )
   def a  (						self       :      Optional[int]      ) ->							Any:
        pass
   @unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'      )
   def a  (						self       :      List[Any]      ) ->							List[str]:
        pass





   def a  (						self       :      Dict      ) ->							Optional[int]:
        return super().test_progress_bar()




@slow
@skip_mps
class 			_lowercase (	unittest.TestCase							):
   def a  (						self       :      List[Any]      ) ->							str:
        __snake_case           =						load_numpy(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'      )

        __snake_case           =						TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'      )
        __snake_case           =						DPMSolverMultistepScheduler.from_config(pipe.scheduler.config      )
        __snake_case           =						pipe.to('cuda'      )

        __snake_case           =						'Spiderman is surfing'
        __snake_case           =						torch.Generator(device='cpu'      ).manual_seed(0      )

        __snake_case           =						pipe(SCREAMING_SNAKE_CASE_       ,   generator=SCREAMING_SNAKE_CASE_       ,   num_inference_steps=25       ,   output_type='pt'      ).frames
        __snake_case           =						video_frames.cpu().numpy()

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





   def a  (						self       :      Dict      ) ->							Any:
        __snake_case           =						load_numpy(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'      )

        __snake_case           =						TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'      )
        __snake_case           =						pipe.to('cuda'      )

        __snake_case           =						'Spiderman is surfing'
        __snake_case           =						torch.Generator(device='cpu'      ).manual_seed(0      )

        __snake_case           =						pipe(SCREAMING_SNAKE_CASE_       ,   generator=SCREAMING_SNAKE_CASE_       ,   num_inference_steps=2       ,   output_type='pt'      ).frames
        __snake_case           =						video_frames.cpu().numpy()

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












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

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


SCREAMING_SNAKE_CASE__   :      str				=		logging.get_logger(__name__)


SCREAMING_SNAKE_CASE__   :      Tuple				=		"""▁"""

SCREAMING_SNAKE_CASE__   :      Union[str, Any]				=		{"""vocab_file""": """spiece.model"""}

SCREAMING_SNAKE_CASE__   :      List[Any]				=		{
    """vocab_file""": {
        """google/reformer-crime-and-punishment""": (
            """https://huggingface.co/google/reformer-crime-and-punishment/resolve/main/spiece.model"""
        )
    }
}

SCREAMING_SNAKE_CASE__   :      Optional[int]				=		{
    """google/reformer-crime-and-punishment""": 52_42_88,
}
class     lowerCamelCase_					(   lowerCamelCase			):
				a__       							=	VOCAB_FILES_NAMES
				a__       							=	PRETRAINED_VOCAB_FILES_MAP
				a__       							=	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				a__       							=	['''input_ids''', '''attention_mask''']







				def __init__(    self							,       __lowerCAmelCase							,       __lowerCAmelCase="</s>"							,       __lowerCAmelCase="<unk>"							,       __lowerCAmelCase=[]							,       __lowerCAmelCase = None							,       **__lowerCAmelCase							,       ):
								"""simple docstring"""
								__magic_name__		:int   =   {} if sp_model_kwargs is None else sp_model_kwargs

								super().__init__(
								    eos_token=__lowerCAmelCase							,       unk_token=__lowerCAmelCase							,       additional_special_tokens=__lowerCAmelCase							,       sp_model_kwargs=self.sp_model_kwargs							,       **__lowerCAmelCase							,       )

								__magic_name__		:Optional[Any]   =   vocab_file
								__magic_name__		:int   =   spm.SentencePieceProcessor(**self.sp_model_kwargs  )
								self.sp_model.Load(__lowerCAmelCase  )







				@property
				def       A						(    self  ):
								"""simple docstring"""
								return self.sp_model.get_piece_size()







				def       A						(    self  ):
								"""simple docstring"""
								__magic_name__		:str   =   {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  ):
								"""simple docstring"""
								__magic_name__		:Optional[Any]   =   self.__dict__.copy()
								__magic_name__		:Optional[Any]   =   None
								return state







				def __setstate__(    self							,       __lowerCAmelCase  ):
								"""simple docstring"""
								__magic_name__		:Any   =   d

								# for backward compatibility
								if not hasattr(self							,       '''sp_model_kwargs'''  ):
												__magic_name__		:Optional[int]   =   {}

								__magic_name__		:Union[str, Any]   =   spm.SentencePieceProcessor(**self.sp_model_kwargs  )
								self.sp_model.Load(self.vocab_file  )







				def       A						(    self							,       __lowerCAmelCase  ):
								"""simple docstring"""
								return self.sp_model.encode(__lowerCAmelCase							,       out_type=__lowerCAmelCase  )







				def       A						(    self							,       __lowerCAmelCase  ):
								"""simple docstring"""
								return self.sp_model.piece_to_id(__lowerCAmelCase  )







				def       A						(    self							,       __lowerCAmelCase  ):
								"""simple docstring"""
								if index < self.sp_model.get_piece_size():
												__magic_name__		:int   =   self.sp_model.IdToPiece(__lowerCAmelCase  )
								return token







				def       A						(    self							,       __lowerCAmelCase  ):
								"""simple docstring"""
								__magic_name__		:Optional[Any]   =   []
								__magic_name__		:Tuple   =   ''''''
								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
																__magic_name__		:Optional[Any]   =   []
												else:
																current_sub_tokens.append(__lowerCAmelCase  )
								out_string += self.sp_model.decode(__lowerCAmelCase  )
								return out_string.strip()







				def       A						(    self							,       __lowerCAmelCase							,       __lowerCAmelCase = None  ):
								"""simple docstring"""
								if not os.path.isdir(__lowerCAmelCase  ):
												logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''  )
												return
								__magic_name__		:Optional[int]   =   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  ) and os.path.isfile(self.vocab_file  ):
												copyfile(self.vocab_file							,       __lowerCAmelCase  )
								elif not os.path.isfile(self.vocab_file  ):
												with open(__lowerCAmelCase							,       '''wb'''  ) as fi:
																__magic_name__		:Dict   =   self.sp_model.serialized_model_proto()
																fi.write(__lowerCAmelCase  )

								return (out_vocab_file,)






