



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    SCREAMING_SNAKE_CASE__ (							__UpperCAmelCase      ,  unittest.TestCase   ):
    lowercase__      =   VideoToVideoSDPipeline
    lowercase__      =   TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"video"}   ) - {"image", "width", "height"}
    lowercase__      =   TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"video"}   ) - {"image"}
    lowercase__      =   PipelineTesterMixin.required_optional_params - {"latents"}
    lowercase__      =   False

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

    def 					_UpperCAmelCase	(				self			:				Optional[Any]):







      """simple docstring"""







      torch.manual_seed(0)
      lowercase_								= 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 ,  )
      lowercase_								= DDIMScheduler(
          beta_start=0.00_085 ,  beta_end=0.012 ,  beta_schedule="""scaled_linear""" ,  clip_sample=lowerCAmelCase_ ,  set_alpha_to_one=lowerCAmelCase_ ,  )
      torch.manual_seed(0)
      lowercase_								= 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)
      lowercase_								= 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 ,  )
      lowercase_								= CLIPTextModel(lowerCAmelCase_)
      lowercase_								= CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""")

      lowercase_								= {
          """unet""": unet,
          """scheduler""": scheduler,
          """vae""": vae,
          """text_encoder""": text_encoder,
          """tokenizer""": tokenizer,
      }
      return components

    def 					_UpperCAmelCase	(				self			:				Tuple ,  lowerCAmelCase_			:				int ,  lowerCAmelCase_			:				Optional[Any]=0):







      """simple docstring"""







      lowercase_								= floats_tensor((1, 3, 3, 3_2, 3_2) ,  rng=random.Random(lowerCAmelCase_)).to(lowerCAmelCase_)

      if str(lowerCAmelCase_).startswith("""mps"""):
        lowercase_								= torch.manual_seed(lowerCAmelCase_)
      else:
        lowercase_								= torch.Generator(device=lowerCAmelCase_).manual_seed(lowerCAmelCase_)
      lowercase_								= {
          """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 					_UpperCAmelCase	(				self			:				int):







      """simple docstring"""







      lowercase_								= """cpu"""  # ensure determinism for the device-dependent torch.Generator
      lowercase_								= self.get_dummy_components()
      lowercase_								= VideoToVideoSDPipeline(**lowerCAmelCase_)
      lowercase_								= sd_pipe.to(lowerCAmelCase_)
      sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_)

      lowercase_								= self.get_dummy_inputs(lowerCAmelCase_)
      lowercase_								= """np"""
      lowercase_								= sd_pipe(**lowerCAmelCase_).frames
      lowercase_								= frames[0][-3:, -3:, -1]

      assert frames[0].shape == (3_2, 3_2, 3)
      lowercase_								= 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 					_UpperCAmelCase	(				self			:				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 					_UpperCAmelCase	(				self			:				Dict):







      """simple docstring"""







      pass

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







      """simple docstring"""







      pass

    @unittest.skip(reason="""`num_images_per_prompt` argument is not supported for this pipeline.""")
    def 					_UpperCAmelCase	(				self			:				List[Any]):







      """simple docstring"""







      pass




    def 					_UpperCAmelCase	(				self			:				Optional[Any]):







      """simple docstring"""







      return super().test_progress_bar()







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

    def 					_UpperCAmelCase	(				self			:				List[str]):







      """simple docstring"""







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

      # 10 frames
      lowercase_								= torch.Generator(device="""cpu""").manual_seed(0)
      lowercase_								= torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6) ,  generator=lowerCAmelCase_)
      lowercase_								= video.to("""cuda""")

      lowercase_								= """Spiderman is surfing"""

      lowercase_								= pipe(lowerCAmelCase_ ,  video=lowerCAmelCase_ ,  generator=lowerCAmelCase_ ,  num_inference_steps=3 ,  output_type="""pt""").frames

      lowercase_								= 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"""
from typing import TYPE_CHECKING

from ...utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_tf_available,
    is_tokenizers_available,
    is_torch_available,
)


UpperCAmelCase						:		Union[str, Any]			= {
    "configuration_longformer": [
        "LONGFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP",
        "LongformerConfig",
        "LongformerOnnxConfig",
    ],
    "tokenization_longformer": ["LongformerTokenizer"],
}

try:
       if not is_tokenizers_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       UpperCAmelCase						:		List[Any]			= ["LongformerTokenizerFast"]

try:
       if not is_torch_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       UpperCAmelCase						:		Tuple			= [
           "LONGFORMER_PRETRAINED_MODEL_ARCHIVE_LIST",
           "LongformerForMaskedLM",
           "LongformerForMultipleChoice",
           "LongformerForQuestionAnswering",
           "LongformerForSequenceClassification",
           "LongformerForTokenClassification",
           "LongformerModel",
           "LongformerPreTrainedModel",
           "LongformerSelfAttention",
       ]

try:
       if not is_tf_available():
              raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
       pass
else:
       UpperCAmelCase						:		List[Any]			= [
           "TF_LONGFORMER_PRETRAINED_MODEL_ARCHIVE_LIST",
           "TFLongformerForMaskedLM",
           "TFLongformerForMultipleChoice",
           "TFLongformerForQuestionAnswering",
           "TFLongformerForSequenceClassification",
           "TFLongformerForTokenClassification",
           "TFLongformerModel",
           "TFLongformerPreTrainedModel",
           "TFLongformerSelfAttention",
       ]


if TYPE_CHECKING:
       from .configuration_longformer import (
           LONGFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP,
           LongformerConfig,
           LongformerOnnxConfig,
       )
       from .tokenization_longformer import LongformerTokenizer

       try:
              if not is_tokenizers_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .tokenization_longformer_fast import LongformerTokenizerFast

       try:
              if not is_torch_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_longformer import (
                  LONGFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
                  LongformerForMaskedLM,
                  LongformerForMultipleChoice,
                  LongformerForQuestionAnswering,
                  LongformerForSequenceClassification,
                  LongformerForTokenClassification,
                  LongformerModel,
                  LongformerPreTrainedModel,
                  LongformerSelfAttention,
              )

       try:
              if not is_tf_available():
                     raise OptionalDependencyNotAvailable()
       except OptionalDependencyNotAvailable:
              pass
       else:
              from .modeling_tf_longformer import (
                  TF_LONGFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
                  TFLongformerForMaskedLM,
                  TFLongformerForMultipleChoice,
                  TFLongformerForQuestionAnswering,
                  TFLongformerForSequenceClassification,
                  TFLongformerForTokenClassification,
                  TFLongformerModel,
                  TFLongformerPreTrainedModel,
                  TFLongformerSelfAttention,
              )

else:
       import sys

       UpperCAmelCase						:		Optional[Any]			= _LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)

