





'''simple docstring'''


import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()






@skip_mps
class     lowerCAmelCase__ (				UpperCAmelCase_      , unittest.TestCase    ):

       '''simple docstring'''
       _lowerCamelCase								=VideoToVideoSDPipeline
       _lowerCamelCase								=TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"video"}    ) - {"image", "width", "height"}
       _lowerCamelCase								=TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"video"}    ) - {"image"}
       _lowerCamelCase								=PipelineTesterMixin.required_optional_params - {"latents"}
       _lowerCamelCase								=False

       # No `output_type`.
       _lowerCamelCase								=frozenset(
           [
               "num_inference_steps",
               "generator",
               "latents",
               "return_dict",
               "callback",
               "callback_steps",
           ]    )

       def 			__snake_case      ( self					:				List[str]						):
              torch.manual_seed(0						)
              UpperCAmelCase  =						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       ,     )
              UpperCAmelCase  =						DDIMScheduler(
                  beta_start=0.00_085       ,     beta_end=0.012       ,     beta_schedule='''scaled_linear'''       ,     clip_sample=a__       ,     set_alpha_to_one=a__       ,     )
              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=128       ,     )
              torch.manual_seed(0						)
              UpperCAmelCase  =						CLIPTextConfig(
                  bos_token_id=0       ,     eos_token_id=2       ,     hidden_size=32       ,     intermediate_size=37       ,     layer_norm_eps=1e-0_5       ,     num_attention_heads=4       ,     num_hidden_layers=5       ,     pad_token_id=1       ,     vocab_size=1000       ,     hidden_act='''gelu'''       ,     projection_dim=512       ,     )
              UpperCAmelCase  =						CLIPTextModel(a__						)
              UpperCAmelCase  =						CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''						)

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

       def 			__snake_case      ( self					:				Any       ,     a__					:				int       ,     a__					:				int=0						):
              # 3 frames
              UpperCAmelCase  =						floats_tensor((1, 3, 3, 32, 32)       ,     rng=random.Random(a__						)						).to(a__						)

              if str(a__						).startswith('''mps'''						):
                     UpperCAmelCase  =						torch.manual_seed(a__						)
              else:
                     UpperCAmelCase  =						torch.Generator(device=a__						).manual_seed(a__						)
              UpperCAmelCase  =						{
                  '''prompt''': '''A painting of a squirrel eating a burger''',
                  '''video''': video,
                  '''generator''': generator,
                  '''num_inference_steps''': 2,
                  '''guidance_scale''': 6.0,
                  '''output_type''': '''pt''',
              }
              return inputs

       def 			__snake_case      ( self					:				Dict						):
              UpperCAmelCase  =						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
              UpperCAmelCase  =						self.get_dummy_components()
              UpperCAmelCase  =						VideoToVideoSDPipeline(**a__						)
              UpperCAmelCase  =						sd_pipe.to(a__						)
              sd_pipe.set_progress_bar_config(disable=a__						)

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

              assert frames[0].shape == (32, 32, 3)
              UpperCAmelCase  =						np.array([106, 117, 113, 174, 137, 112, 148, 151, 131]						)

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

       @unittest.skipIf(
           torch_device != '''cuda''' or not is_xformers_available()       ,     reason='''XFormers attention is only available with CUDA and `xformers` installed'''       ,     )
       def 			__snake_case      ( self					:				Tuple						):
              self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=a__       ,     expected_max_diff=5e-3						)

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

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

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



       def 			__snake_case      ( self					:				List[str]						):
              return super().test_progress_bar()






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

       '''simple docstring'''

       def 			__snake_case      ( self					:				List[str]						):
              UpperCAmelCase  =						VideoToVideoSDPipeline.from_pretrained('''cerspense/zeroscope_v2_XL'''       ,     torch_dtype=torch.floataa						)
              pipe.enable_model_cpu_offload()

              # 10 frames
              UpperCAmelCase  =						torch.Generator(device='''cpu'''						).manual_seed(0						)
              UpperCAmelCase  =						torch.randn((1, 10, 3, 1024, 576)       ,     generator=a__						)
              UpperCAmelCase  =						video.to('''cuda'''						)

              UpperCAmelCase  =						'''Spiderman is surfing'''

              UpperCAmelCase  =						pipe(a__       ,     video=a__       ,     generator=a__       ,     num_inference_steps=3       ,     output_type='''pt'''						).frames

              UpperCAmelCase  =						np.array([-1.0_458_984, -1.1_279_297, -0.9_663_086, -0.91_503_906, -0.75_097_656]						)
              assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array						).sum() < 1e-2









"""simple docstring"""





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


__A											=						logging.get_logger(__name__)

__A											=						{"""vocab_file""": """vocab.json""", """merges_file""": """merges.txt""", """tokenizer_file""": """tokenizer.json"""}

__A											=						{
    """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"""
        ),
    },
}

__A											=						{
    """roberta-base""": 512,
    """roberta-large""": 512,
    """roberta-large-mnli""": 512,
    """distilroberta-base""": 512,
    """roberta-base-openai-detector""": 512,
    """roberta-large-openai-detector""": 512,
}






class    _lowerCAmelCase  (	a			):
       """simple docstring"""




       __magic_name__	:str						    =   VOCAB_FILES_NAMES
       __magic_name__	:List[Any]						    =   PRETRAINED_VOCAB_FILES_MAP
       __magic_name__	:Optional[Any]						    =   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
       __magic_name__	:str						    =   ["""input_ids""", """attention_mask"""]
       __magic_name__	:Any						    =   RobertaTokenizer






       def __init__(       self       ,					__UpperCAmelCase=None       ,					__UpperCAmelCase=None       ,					__UpperCAmelCase=None       ,					__UpperCAmelCase="replace"       ,					__UpperCAmelCase="<s>"       ,					__UpperCAmelCase="</s>"       ,					__UpperCAmelCase="</s>"       ,					__UpperCAmelCase="<s>"       ,					__UpperCAmelCase="<unk>"       ,					__UpperCAmelCase="<pad>"       ,					__UpperCAmelCase="<mask>"       ,					__UpperCAmelCase=False       ,					__UpperCAmelCase=True       ,					**__UpperCAmelCase       ,					):



         '''simple docstring'''

         super().__init__(
             __UpperCAmelCase       ,					__UpperCAmelCase       ,					tokenizer_file=__UpperCAmelCase       ,					errors=__UpperCAmelCase       ,					bos_token=__UpperCAmelCase       ,					eos_token=__UpperCAmelCase       ,					sep_token=__UpperCAmelCase       ,					cls_token=__UpperCAmelCase       ,					unk_token=__UpperCAmelCase       ,					pad_token=__UpperCAmelCase       ,					mask_token=__UpperCAmelCase       ,					add_prefix_space=__UpperCAmelCase       ,					trim_offsets=__UpperCAmelCase       ,					**__UpperCAmelCase       ,					)

         lowerCAmelCase__     :Dict											=				json.loads(self.backend_tokenizer.pre_tokenizer.__getstate__() )
         if pre_tok_state.get('add_prefix_space'       ,					__UpperCAmelCase ) != add_prefix_space:
           lowerCAmelCase__     :Optional[int]											=				getattr(__UpperCAmelCase       ,					pre_tok_state.pop('type' ) )
           lowerCAmelCase__     :List[Any]											=				add_prefix_space
           lowerCAmelCase__     :str											=				pre_tok_class(**__UpperCAmelCase )

         lowerCAmelCase__     :List[str]											=				add_prefix_space

         lowerCAmelCase__     :str											=				'post_processor'
         lowerCAmelCase__     :Optional[Any]											=				getattr(self.backend_tokenizer       ,					__UpperCAmelCase       ,					__UpperCAmelCase )
         if tokenizer_component_instance:
           lowerCAmelCase__     :Optional[Any]											=				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:
             lowerCAmelCase__     :Any											=				tuple(state['sep'] )
           if "cls" in state:
             lowerCAmelCase__     :int											=				tuple(state['cls'] )

           lowerCAmelCase__     :List[Any]											=				False

           if state.get('add_prefix_space'       ,					__UpperCAmelCase ) != add_prefix_space:
             lowerCAmelCase__     :Union[str, Any]											=				add_prefix_space
             lowerCAmelCase__     :Any											=				True

           if state.get('trim_offsets'       ,					__UpperCAmelCase ) != trim_offsets:
             lowerCAmelCase__     :Union[str, Any]											=				trim_offsets
             lowerCAmelCase__     :Optional[int]											=				True

           if changes_to_apply:
             lowerCAmelCase__     :str											=				getattr(__UpperCAmelCase       ,					state.pop('type' ) )
             lowerCAmelCase__     :Any											=				component_class(**__UpperCAmelCase )
             setattr(self.backend_tokenizer       ,					__UpperCAmelCase       ,					__UpperCAmelCase )






       @property
       def 	snake_case		(       self ):



         '''simple docstring'''

         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 	snake_case		(       self       ,					__UpperCAmelCase ):



         '''simple docstring'''

         lowerCAmelCase__     :List[Any]											=				AddedToken(__UpperCAmelCase       ,					lstrip=__UpperCAmelCase       ,					rstrip=__UpperCAmelCase ) if isinstance(__UpperCAmelCase       ,					__UpperCAmelCase ) else value
         lowerCAmelCase__     :List[str]											=				value






       def 	snake_case		(       self       ,					*__UpperCAmelCase       ,					**__UpperCAmelCase ):



         '''simple docstring'''

         lowerCAmelCase__     :Optional[Any]											=				kwargs.get('is_split_into_words'       ,					__UpperCAmelCase )
         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(*__UpperCAmelCase       ,					**__UpperCAmelCase )






       def 	snake_case		(       self       ,					*__UpperCAmelCase       ,					**__UpperCAmelCase ):



         '''simple docstring'''

         lowerCAmelCase__     :Any											=				kwargs.get('is_split_into_words'       ,					__UpperCAmelCase )

         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(*__UpperCAmelCase       ,					**__UpperCAmelCase )






       def 	snake_case		(       self       ,					__UpperCAmelCase       ,					__UpperCAmelCase = None ):



         '''simple docstring'''

         lowerCAmelCase__     :Union[str, Any]											=				self._tokenizer.model.save(__UpperCAmelCase       ,					name=__UpperCAmelCase )
         return tuple(__UpperCAmelCase )






       def 	snake_case		(       self       ,					__UpperCAmelCase       ,					__UpperCAmelCase=None ):



         '''simple docstring'''

         lowerCAmelCase__     :str											=				[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 	snake_case		(       self       ,					__UpperCAmelCase       ,					__UpperCAmelCase = None ):



         '''simple docstring'''

         lowerCAmelCase__     :List[Any]											=				[self.sep_token_id]
         lowerCAmelCase__     :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]
