





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):
      _lowercase				:					int						=			VideoToVideoSDPipeline
      _lowercase				:					List[Any]						=			TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"""video"""}) - {"""image""", """width""", """height"""}
      _lowercase				:					int						=			TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""video"""}) - {"""image"""}
      _lowercase				:					Any						=			PipelineTesterMixin.required_optional_params - {"""latents"""}
      _lowercase				:					Dict						=			False

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







      def    _lowercase  (    self							)   ->						List[Any]:



        '''simple docstring'''




        torch.manual_seed(0							)
        a__	:					int										=UNetaDConditionModel(
            block_out_channels=(3_2, 6_4, 6_4, 6_4)		,      layers_per_block=2		,      sample_size=3_2		,      in_channels=4		,      out_channels=4		,      down_block_types=("CrossAttnDownBlock3D", "CrossAttnDownBlock3D", "CrossAttnDownBlock3D", "DownBlock3D")		,      up_block_types=("UpBlock3D", "CrossAttnUpBlock3D", "CrossAttnUpBlock3D", "CrossAttnUpBlock3D")		,      cross_attention_dim=3_2		,      attention_head_dim=4		,      )
        a__	:					str										=DDIMScheduler(
            beta_start=0.0_00_85		,      beta_end=0.0_12		,      beta_schedule="scaled_linear"		,      clip_sample=lowerCAmelCase__		,      set_alpha_to_one=lowerCAmelCase__		,      )
        torch.manual_seed(0							)
        a__	:					Optional[Any]										=AutoencoderKL(
            block_out_channels=[3_2, 6_4]		,      in_channels=3		,      out_channels=3		,      down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]		,      up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]		,      latent_channels=4		,      sample_size=1_2_8		,      )
        torch.manual_seed(0							)
        a__	:					List[Any]										=CLIPTextConfig(
            bos_token_id=0		,      eos_token_id=2		,      hidden_size=3_2		,      intermediate_size=3_7		,      layer_norm_eps=1E-05		,      num_attention_heads=4		,      num_hidden_layers=5		,      pad_token_id=1		,      vocab_size=1_0_0_0		,      hidden_act="gelu"		,      projection_dim=5_1_2		,      )
        a__	:					Union[str, Any]										=CLIPTextModel(lowerCAmelCase__							)
        a__	:					Optional[Any]										=CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"							)

        a__	:					Union[str, Any]										={
            "unet": unet,
            "scheduler": scheduler,
            "vae": vae,
            "text_encoder": text_encoder,
            "tokenizer": tokenizer,
        }
        return components







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__=0							)   ->						Optional[int]:



        '''simple docstring'''




        a__	:					Dict										=floats_tensor((1, 3, 3, 3_2, 3_2)		,      rng=random.Random(lowerCAmelCase__							)							).to(lowerCAmelCase__							)

        if str(lowerCAmelCase__							).startswith("mps"							):
          a__	:					List[Any]										=torch.manual_seed(lowerCAmelCase__							)
        else:
          a__	:					List[str]										=torch.Generator(device=lowerCAmelCase__							).manual_seed(lowerCAmelCase__							)
        a__	:					Union[str, Any]										={
            "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    _lowercase  (    self							)   ->						Any:



        '''simple docstring'''




        a__	:					List[str]										="cpu"  # ensure determinism for the device-dependent torch.Generator
        a__	:					Tuple										=self.get_dummy_components()
        a__	:					int										=VideoToVideoSDPipeline(**lowerCAmelCase__							)
        a__	:					Any										=sd_pipe.to(lowerCAmelCase__							)
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__							)

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

        assert frames[0].shape == (3_2, 3_2, 3)
        a__	:					Optional[Any]										=np.array([1_0_6, 1_1_7, 1_1_3, 1_7_4, 1_3_7, 1_1_2, 1_4_8, 1_5_1, 1_3_1]							)

        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    _lowercase  (    self							)   ->						Optional[Any]:



        '''simple docstring'''




        self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCAmelCase__		,      expected_max_diff=5E-3							)







      @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."							)
      def    _lowercase  (    self							)   ->						Optional[int]:



        '''simple docstring'''




        pass







      @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."							)
      def    _lowercase  (    self							)   ->						int:



        '''simple docstring'''




        pass







      @unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."							)
      def    _lowercase  (    self							)   ->						Union[str, Any]:



        '''simple docstring'''




        pass







      def    _lowercase  (    self							)   ->						Any:



        '''simple docstring'''




        return super().test_progress_bar()







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







      def    _lowercase  (    self							)   ->						Any:



        '''simple docstring'''




        a__	:					Dict										=VideoToVideoSDPipeline.from_pretrained("cerspense/zeroscope_v2_XL"		,      torch_dtype=torch.floataa							)
        pipe.enable_model_cpu_offload()

        # 10 frames
        a__	:					Optional[Any]										=torch.Generator(device="cpu"							).manual_seed(0							)
        a__	:					Optional[Any]										=torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6)		,      generator=lowerCAmelCase__							)
        a__	:					List[str]										=video.to("cuda"							)

        a__	:					Tuple										="Spiderman is surfing"

        a__	:					int										=pipe(lowerCAmelCase__		,      video=lowerCAmelCase__		,      generator=lowerCAmelCase__		,      num_inference_steps=3		,      output_type="pt"							).frames

        a__	:					int										=np.array([-1.0_45_89_84, -1.1_27_92_97, -0.9_66_30_86, -0.91_50_39_06, -0.75_09_76_56]							)
        assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array							).sum() < 1E-2













import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_funnel import FunnelTokenizer


UpperCAmelCase	:	int		     =       logging.get_logger(__name__)

UpperCAmelCase	:	Optional[Any]		     =       {"""vocab_file""": """vocab.txt""", """tokenizer_file""": """tokenizer.json"""}

UpperCAmelCase	:	List[Any]		     =       [
    """small""",
    """small-base""",
    """medium""",
    """medium-base""",
    """intermediate""",
    """intermediate-base""",
    """large""",
    """large-base""",
    """xlarge""",
    """xlarge-base""",
]

UpperCAmelCase	:	Optional[int]		     =       {
    """vocab_file""": {
        """funnel-transformer/small""": """https://huggingface.co/funnel-transformer/small/resolve/main/vocab.txt""",
        """funnel-transformer/small-base""": """https://huggingface.co/funnel-transformer/small-base/resolve/main/vocab.txt""",
        """funnel-transformer/medium""": """https://huggingface.co/funnel-transformer/medium/resolve/main/vocab.txt""",
        """funnel-transformer/medium-base""": (
            """https://huggingface.co/funnel-transformer/medium-base/resolve/main/vocab.txt"""
        ),
        """funnel-transformer/intermediate""": (
            """https://huggingface.co/funnel-transformer/intermediate/resolve/main/vocab.txt"""
        ),
        """funnel-transformer/intermediate-base""": (
            """https://huggingface.co/funnel-transformer/intermediate-base/resolve/main/vocab.txt"""
        ),
        """funnel-transformer/large""": """https://huggingface.co/funnel-transformer/large/resolve/main/vocab.txt""",
        """funnel-transformer/large-base""": """https://huggingface.co/funnel-transformer/large-base/resolve/main/vocab.txt""",
        """funnel-transformer/xlarge""": """https://huggingface.co/funnel-transformer/xlarge/resolve/main/vocab.txt""",
        """funnel-transformer/xlarge-base""": (
            """https://huggingface.co/funnel-transformer/xlarge-base/resolve/main/vocab.txt"""
        ),
    },
    """tokenizer_file""": {
        """funnel-transformer/small""": """https://huggingface.co/funnel-transformer/small/resolve/main/tokenizer.json""",
        """funnel-transformer/small-base""": (
            """https://huggingface.co/funnel-transformer/small-base/resolve/main/tokenizer.json"""
        ),
        """funnel-transformer/medium""": """https://huggingface.co/funnel-transformer/medium/resolve/main/tokenizer.json""",
        """funnel-transformer/medium-base""": (
            """https://huggingface.co/funnel-transformer/medium-base/resolve/main/tokenizer.json"""
        ),
        """funnel-transformer/intermediate""": (
            """https://huggingface.co/funnel-transformer/intermediate/resolve/main/tokenizer.json"""
        ),
        """funnel-transformer/intermediate-base""": (
            """https://huggingface.co/funnel-transformer/intermediate-base/resolve/main/tokenizer.json"""
        ),
        """funnel-transformer/large""": """https://huggingface.co/funnel-transformer/large/resolve/main/tokenizer.json""",
        """funnel-transformer/large-base""": (
            """https://huggingface.co/funnel-transformer/large-base/resolve/main/tokenizer.json"""
        ),
        """funnel-transformer/xlarge""": """https://huggingface.co/funnel-transformer/xlarge/resolve/main/tokenizer.json""",
        """funnel-transformer/xlarge-base""": (
            """https://huggingface.co/funnel-transformer/xlarge-base/resolve/main/tokenizer.json"""
        ),
    },
}
UpperCAmelCase	:	Optional[int]		     =       {F"""funnel-transformer/{name}""": 512 for name in _model_names}
UpperCAmelCase	:	Optional[int]		     =       {F"""funnel-transformer/{name}""": {"""do_lower_case""": True} for name in _model_names}







class 				__lowerCAmelCase       (    UpperCamelCase__):
      _lowercase				:					str						=			VOCAB_FILES_NAMES
      _lowercase				:					List[Any]						=			PRETRAINED_VOCAB_FILES_MAP
      _lowercase				:					Dict						=			PRETRAINED_INIT_CONFIGURATION
      _lowercase				:					Union[str, Any]						=			FunnelTokenizer
      _lowercase				:					Any						=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _lowercase				:					int						=			2







      def __init__(    self		,      lowerCAmelCase__=None		,      lowerCAmelCase__=None		,      lowerCAmelCase__=True		,      lowerCAmelCase__="<unk>"		,      lowerCAmelCase__="<sep>"		,      lowerCAmelCase__="<pad>"		,      lowerCAmelCase__="<cls>"		,      lowerCAmelCase__="<mask>"		,      lowerCAmelCase__="<s>"		,      lowerCAmelCase__="</s>"		,      lowerCAmelCase__=True		,      lowerCAmelCase__=True		,      lowerCAmelCase__=None		,      lowerCAmelCase__="##"		,      **lowerCAmelCase__		,      )   ->						Optional[int]:



        '''simple docstring'''




        super().__init__(
            lowerCAmelCase__		,      tokenizer_file=lowerCAmelCase__		,      do_lower_case=lowerCAmelCase__		,      unk_token=lowerCAmelCase__		,      sep_token=lowerCAmelCase__		,      pad_token=lowerCAmelCase__		,      cls_token=lowerCAmelCase__		,      mask_token=lowerCAmelCase__		,      bos_token=lowerCAmelCase__		,      eos_token=lowerCAmelCase__		,      clean_text=lowerCAmelCase__		,      tokenize_chinese_chars=lowerCAmelCase__		,      strip_accents=lowerCAmelCase__		,      wordpieces_prefix=lowerCAmelCase__		,      **lowerCAmelCase__		,      )

        a__	:					Optional[Any]										=json.loads(self.backend_tokenizer.normalizer.__getstate__()							)
        if (
            normalizer_state.get("lowercase"		,      lowerCAmelCase__							) != do_lower_case
            or normalizer_state.get("strip_accents"		,      lowerCAmelCase__							) != strip_accents
            or normalizer_state.get("handle_chinese_chars"		,      lowerCAmelCase__							) != tokenize_chinese_chars
        ):
          a__	:					List[str]										=getattr(lowerCAmelCase__		,      normalizer_state.pop("type"							)							)
          a__	:					Union[str, Any]										=do_lower_case
          a__	:					Any										=strip_accents
          a__	:					Optional[Any]										=tokenize_chinese_chars
          a__	:					Dict										=normalizer_class(**lowerCAmelCase__							)

        a__	:					Any										=do_lower_case







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__=None							)   ->						str:



        '''simple docstring'''




        a__	:					Dict										=[self.cls_token_id] + token_ids_a + [self.sep_token_id]

        if token_ids_a:
          output += token_ids_a + [self.sep_token_id]

        return output







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = None							)   ->						List[int]:



        '''simple docstring'''




        a__	:					Optional[int]										=[self.sep_token_id]
        a__	:					Union[str, Any]										=[self.cls_token_id]
        if token_ids_a is None:
          return len(cls							) * [self.cls_token_type_id] + len(token_ids_a + sep							) * [0]
        return len(cls							) * [self.cls_token_type_id] + len(token_ids_a + sep							) * [0] + len(token_ids_a + sep							) * [1]







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = None							)   ->						Tuple[str]:



        '''simple docstring'''




        a__	:					Tuple										=self._tokenizer.model.save(lowerCAmelCase__		,      name=lowerCAmelCase__							)
        return tuple(lowerCAmelCase__							)






