





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      lowerCamelCase(						__snake_case			, unittest.TestCase	):
     '''simple docstring'''


     __magic_name__   =			TextToVideoSDPipeline
     __magic_name__   =			TEXT_TO_IMAGE_PARAMS
     __magic_name__   =			TEXT_TO_IMAGE_BATCH_PARAMS
     # No `output_type`.
     __magic_name__   =			frozenset(
         [
             'num_inference_steps',
             'generator',
             'latents',
             'return_dict',
             'callback',
             'callback_steps',
         ]	)




     def        lowerCAmelCase__		(    self							):
         torch.manual_seed(0							)
         _A     =							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				,     )
         _A     =							DDIMScheduler(
             beta_start=0.0_0085				,     beta_end=0.012				,     beta_schedule='scaled_linear'				,     clip_sample=snake_case_				,     set_alpha_to_one=snake_case_				,     )
         torch.manual_seed(0							)
         _A     =							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							)
         _A     =							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				,     )
         _A     =							CLIPTextModel(snake_case_							)
         _A     =							CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'							)

         _A     =							{
             'unet': unet,
             'scheduler': scheduler,
             'vae': vae,
             'text_encoder': text_encoder,
             'tokenizer': tokenizer,
         }
         return components




     def        lowerCAmelCase__		(    self				,     snake_case_				,     snake_case_=0							):
         if str(snake_case_							).startswith('mps'							):
             _A     =							torch.manual_seed(snake_case_							)
         else:
             _A     =							torch.Generator(device=snake_case_							).manual_seed(snake_case_							)
         _A     =							{
             '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        lowerCAmelCase__		(    self							):
         _A     =							'cpu'  # ensure determinism for the device-dependent torch.Generator
         _A     =							self.get_dummy_components()
         _A     =							TextToVideoSDPipeline(**snake_case_							)
         _A     =							sd_pipe.to(snake_case_							)
         sd_pipe.set_progress_bar_config(disable=snake_case_							)

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

         assert frames[0].shape == (64, 64, 3)
         _A     =							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        lowerCAmelCase__		(    self							):
         self._test_attention_slicing_forward_pass(test_mean_pixel_difference=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        lowerCAmelCase__		(    self							):
         self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=snake_case_				,     expected_max_diff=1E-2							)




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




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




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




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







@slow
@skip_mps
class      lowerCamelCase(						unittest.TestCase	):
     '''simple docstring'''


     def        lowerCAmelCase__		(    self							):
         _A     =							load_numpy(
             'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'							)

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

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

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

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




     def        lowerCAmelCase__		(    self							):
         _A     =							load_numpy(
             'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'							)

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

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

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

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







import json
from typing import List, Optional, Tuple

from tokenizers import pre_tokenizers, processors

from ...tokenization_utils_base import AddedToken, BatchEncoding
from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_roberta import RobertaTokenizer


__magic_name__  							=	logging.get_logger(__name__)

__magic_name__  							=	{'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt''', '''tokenizer_file''': '''tokenizer.json'''}

__magic_name__  							=	{
    '''vocab_file''': {
        '''roberta-base''': '''https://huggingface.co/roberta-base/resolve/main/vocab.json''',
        '''roberta-large''': '''https://huggingface.co/roberta-large/resolve/main/vocab.json''',
        '''roberta-large-mnli''': '''https://huggingface.co/roberta-large-mnli/resolve/main/vocab.json''',
        '''distilroberta-base''': '''https://huggingface.co/distilroberta-base/resolve/main/vocab.json''',
        '''roberta-base-openai-detector''': '''https://huggingface.co/roberta-base-openai-detector/resolve/main/vocab.json''',
        '''roberta-large-openai-detector''': (
            '''https://huggingface.co/roberta-large-openai-detector/resolve/main/vocab.json'''
        ),
    },
    '''merges_file''': {
        '''roberta-base''': '''https://huggingface.co/roberta-base/resolve/main/merges.txt''',
        '''roberta-large''': '''https://huggingface.co/roberta-large/resolve/main/merges.txt''',
        '''roberta-large-mnli''': '''https://huggingface.co/roberta-large-mnli/resolve/main/merges.txt''',
        '''distilroberta-base''': '''https://huggingface.co/distilroberta-base/resolve/main/merges.txt''',
        '''roberta-base-openai-detector''': '''https://huggingface.co/roberta-base-openai-detector/resolve/main/merges.txt''',
        '''roberta-large-openai-detector''': (
            '''https://huggingface.co/roberta-large-openai-detector/resolve/main/merges.txt'''
        ),
    },
    '''tokenizer_file''': {
        '''roberta-base''': '''https://huggingface.co/roberta-base/resolve/main/tokenizer.json''',
        '''roberta-large''': '''https://huggingface.co/roberta-large/resolve/main/tokenizer.json''',
        '''roberta-large-mnli''': '''https://huggingface.co/roberta-large-mnli/resolve/main/tokenizer.json''',
        '''distilroberta-base''': '''https://huggingface.co/distilroberta-base/resolve/main/tokenizer.json''',
        '''roberta-base-openai-detector''': (
            '''https://huggingface.co/roberta-base-openai-detector/resolve/main/tokenizer.json'''
        ),
        '''roberta-large-openai-detector''': (
            '''https://huggingface.co/roberta-large-openai-detector/resolve/main/tokenizer.json'''
        ),
    },
}

__magic_name__  							=	{
    '''roberta-base''': 512,
    '''roberta-large''': 512,
    '''roberta-large-mnli''': 512,
    '''distilroberta-base''': 512,
    '''roberta-base-openai-detector''': 512,
    '''roberta-large-openai-detector''': 512,
}




class        _SCREAMING_SNAKE_CASE	(   __UpperCamelCase			):
      _A    :		Dict     						=  VOCAB_FILES_NAMES
      _A    :		List[str]     						=  PRETRAINED_VOCAB_FILES_MAP
      _A    :		Any     						=  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _A    :		str     						=  ['input_ids', 'attention_mask']
      _A    :		Dict     						=  RobertaTokenizer
      def __init__(      self		,						lowerCamelCase=None		,						lowerCamelCase=None		,						lowerCamelCase=None		,						lowerCamelCase="replace"		,						lowerCamelCase="<s>"		,						lowerCamelCase="</s>"		,						lowerCamelCase="</s>"		,						lowerCamelCase="<s>"		,						lowerCamelCase="<unk>"		,						lowerCamelCase="<pad>"		,						lowerCamelCase="<mask>"		,						lowerCamelCase=False		,						lowerCamelCase=True		,						**lowerCamelCase		,						):
          super().__init__(
              lowerCamelCase		,						lowerCamelCase		,						tokenizer_file=lowerCamelCase		,						errors=lowerCamelCase		,						bos_token=lowerCamelCase		,						eos_token=lowerCamelCase		,						sep_token=lowerCamelCase		,						cls_token=lowerCamelCase		,						unk_token=lowerCamelCase		,						pad_token=lowerCamelCase		,						mask_token=lowerCamelCase		,						add_prefix_space=lowerCamelCase		,						trim_offsets=lowerCamelCase		,						**lowerCamelCase		,						)

          snake_case__    =					json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__()			)
          if pre_tok_state.get("add_prefix_space"		,						lowerCamelCase			) != add_prefix_space:
              snake_case__    =					getattr(lowerCamelCase		,						pre_tok_state.pop("type"			)			)
              snake_case__    =					add_prefix_space
              snake_case__    =					pre_tok_class(**lowerCamelCase			)

          snake_case__    =					add_prefix_space

          snake_case__    =					"post_processor"
          snake_case__    =					getattr(self.backend_tokenizer		,						lowerCamelCase		,						lowerCamelCase			)
          if tokenizer_component_instance:
              snake_case__    =					json.loads(tokenizer_component_instance.__getstate__()			)

              # The lists 'sep' and 'cls' must be cased in tuples for the object `post_processor_class`
              if "sep" in state:
                  snake_case__    =					tuple(state["sep"]			)
              if "cls" in state:
                  snake_case__    =					tuple(state["cls"]			)

              snake_case__    =					False

              if state.get("add_prefix_space"		,						lowerCamelCase			) != add_prefix_space:
                  snake_case__    =					add_prefix_space
                  snake_case__    =					True

              if state.get("trim_offsets"		,						lowerCamelCase			) != trim_offsets:
                  snake_case__    =					trim_offsets
                  snake_case__    =					True

              if changes_to_apply:
                  snake_case__    =					getattr(lowerCamelCase		,						state.pop("type"			)			)
                  snake_case__    =					component_class(**lowerCamelCase			)
                  setattr(self.backend_tokenizer		,						lowerCamelCase		,						lowerCamelCase			)
      @property
      def A_      (      self			):
          if self._mask_token is None:
              if self.verbose:
                  logger.error("Using mask_token, but it is not set yet."			)
              return None
          return str(self._mask_token			)
      @mask_token.setter
      def A_      (      self		,						lowerCamelCase			):
          snake_case__    =					AddedToken(lowerCamelCase		,						lstrip=lowerCamelCase		,						rstrip=lowerCamelCase			) if isinstance(lowerCamelCase		,						lowerCamelCase			) else value
          snake_case__    =					value
      def A_      (      self		,						*lowerCamelCase		,						**lowerCamelCase			):
          snake_case__    =					kwargs.get("is_split_into_words"		,						lowerCamelCase			)
          assert self.add_prefix_space or not is_split_into_words, (
              F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
              "to use it with pretokenized inputs."
          )

          return super()._batch_encode_plus(*lowerCamelCase		,						**lowerCamelCase			)
      def A_      (      self		,						*lowerCamelCase		,						**lowerCamelCase			):
          snake_case__    =					kwargs.get("is_split_into_words"		,						lowerCamelCase			)

          assert self.add_prefix_space or not is_split_into_words, (
              F"""You need to instantiate {self.__class__.__name__} with add_prefix_space=True """
              "to use it with pretokenized inputs."
          )

          return super()._encode_plus(*lowerCamelCase		,						**lowerCamelCase			)
      def A_      (      self		,						lowerCamelCase		,						lowerCamelCase = None			):
          snake_case__    =					self._tokenizer.model.save(lowerCamelCase		,						name=lowerCamelCase			)
          return tuple(lowerCamelCase			)
      def A_      (      self		,						lowerCamelCase		,						lowerCamelCase=None			):
          snake_case__    =					[self.bos_token_id] + token_ids_a + [self.eos_token_id]
          if token_ids_a is None:
              return output

          return output + [self.eos_token_id] + token_ids_a + [self.eos_token_id]



      def A_      (      self		,						lowerCamelCase		,						lowerCamelCase = None			):
          snake_case__    =					[self.sep_token_id]
          snake_case__    =					[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]



