




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

      # No `output_type`.
      UpperCAmelCase_						=						frozenset(
          [
              """num_inference_steps""",
              """generator""",
              """latents""",
              """return_dict""",
              """callback""",
              """callback_steps""",
          ]						)
      def 		UpperCAmelCase_  (	self      :Union[str, Any]					)   ->	List[Any]:
       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.0_00_85       ,					beta_end=0.0_12       ,					beta_schedule="scaled_linear"       ,					clip_sample=lowerCAmelCase__       ,					set_alpha_to_one=lowerCAmelCase__       ,					)
       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-05       ,					num_attention_heads=4       ,					num_hidden_layers=5       ,					pad_token_id=1       ,					vocab_size=1000       ,					hidden_act="gelu"       ,					projection_dim=512       ,					)
       UpperCAmelCase__							       =  CLIPTextModel(lowerCAmelCase__					)
       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 		UpperCAmelCase_  (	self      :Optional[int]       ,					lowerCamelCase      :Dict       ,					lowerCamelCase      :Union[str, Any]=0					)   ->	Optional[int]:
       UpperCAmelCase__							       =  floats_tensor((1, 3, 3, 32, 32)       ,					rng=random.Random(lowerCAmelCase__					)					).to(lowerCAmelCase__					)

       if str(lowerCAmelCase__					).startswith("mps"					):
        UpperCAmelCase__							       =  torch.manual_seed(lowerCAmelCase__					)
       else:
        UpperCAmelCase__							       =  torch.Generator(device=lowerCAmelCase__					).manual_seed(lowerCAmelCase__					)
       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 		UpperCAmelCase_  (	self      :Optional[int]					)   ->	Any:
       UpperCAmelCase__							       =  "cpu"  # ensure determinism for the device-dependent torch.Generator
       UpperCAmelCase__							       =  self.get_dummy_components()
       UpperCAmelCase__							       =  VideoToVideoSDPipeline(**lowerCAmelCase__					)
       UpperCAmelCase__							       =  sd_pipe.to(lowerCAmelCase__					)
       sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__					)

       UpperCAmelCase__							       =  self.get_dummy_inputs(lowerCAmelCase__					)
       UpperCAmelCase__							       =  "np"
       UpperCAmelCase__							       =  sd_pipe(**lowerCAmelCase__					).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 		UpperCAmelCase_  (	self      :str					)   ->	Optional[Any]:
       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      :List[Any]					)   ->	Optional[int]:
       pass
      @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."					)
      def 		UpperCAmelCase_  (	self      :List[Any]					)   ->	int:
       pass
      @unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."					)
      def 		UpperCAmelCase_  (	self      :Any					)   ->	Union[str, Any]:
       pass







      def 		UpperCAmelCase_  (	self      :Union[str, Any]					)   ->	Any:
       return super().test_progress_bar()



@slow
@skip_mps
class 	_UpperCamelCase					( unittest.TestCase						):
      def 		UpperCAmelCase_  (	self      :Optional[int]					)   ->	Any:
       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=lowerCAmelCase__					)
       UpperCAmelCase__							       =  video.to("cuda"					)

       UpperCAmelCase__							       =  "Spiderman is surfing"

       UpperCAmelCase__							       =  pipe(lowerCAmelCase__       ,					video=lowerCAmelCase__       ,					generator=lowerCAmelCase__       ,					num_inference_steps=3       ,					output_type="pt"					).frames

       UpperCAmelCase__							       =  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












from typing import Optional

import pyspark

from .. import Features, NamedSplit
from ..download import DownloadMode
from ..packaged_modules.spark.spark import Spark
from .abc import AbstractDatasetReader







class 				__lowerCAmelCase       (    UpperCamelCase__):







      def __init__(    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = None		,      lowerCAmelCase__ = None		,      lowerCAmelCase__ = True		,      lowerCAmelCase__ = None		,      lowerCAmelCase__ = False		,      lowerCAmelCase__ = None		,      lowerCAmelCase__ = True		,      lowerCAmelCase__ = "arrow"		,      **lowerCAmelCase__		,      )   ->						Optional[int]:



        '''simple docstring'''




        super().__init__(
            split=lowerCAmelCase__		,      features=lowerCAmelCase__		,      cache_dir=lowerCAmelCase__		,      keep_in_memory=lowerCAmelCase__		,      streaming=lowerCAmelCase__		,      **lowerCAmelCase__		,      )
        a__	:					int										=load_from_cache_file
        a__	:					Tuple										=file_format
        a__	:					List[Any]										=Spark(
            df=lowerCAmelCase__		,      features=lowerCAmelCase__		,      cache_dir=lowerCAmelCase__		,      working_dir=lowerCAmelCase__		,      **lowerCAmelCase__		,      )







      def    _lowercase  (    self							)   ->						str:



        '''simple docstring'''




        if self.streaming:
          return self.builder.as_streaming_dataset(split=self.split							)
        a__	:					str										=None if self._load_from_cache_file else DownloadMode.FORCE_REDOWNLOAD
        self.builder.download_and_prepare(
            download_mode=lowerCAmelCase__		,      file_format=self._file_format		,      )
        return self.builder.as_dataset(split=self.split							)






