




'''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 (			__lowerCAmelCase					, unittest.TestCase):
       __lowercase	:   Dict										=      VideoToVideoSDPipeline
       __lowercase	:   Dict										=      TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'''video'''}) - {'''image''', '''width''', '''height'''}
       __lowercase	:   int										=      TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''video'''}) - {'''image'''}
       __lowercase	:   Optional[int]										=      PipelineTesterMixin.required_optional_params - {'''latents'''}
       __lowercase	:   Any										=      False

       # No `output_type`.
       __lowercase	:   Union[str, Any]										=      frozenset(
           [
               '''num_inference_steps''',
               '''generator''',
               '''latents''',
               '''return_dict''',
               '''callback''',
               '''callback_steps''',
           ])


       def    lowerCAmelCase   ( self	)   ->	List[str]:



            '''simple docstring'''




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

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


       def    lowerCAmelCase   ( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE=0	)   ->	Union[str, Any]:



            '''simple docstring'''




            __snake_case            =	floats_tensor((1, 3, 3, 32, 32)     ,	rng=random.Random(__SCREAMING_SNAKE_CASE	)	).to(__SCREAMING_SNAKE_CASE	)

            if str(__SCREAMING_SNAKE_CASE	).startswith('''mps'''	):
                 __snake_case            =	torch.manual_seed(__SCREAMING_SNAKE_CASE	)
            else:
                 __snake_case            =	torch.Generator(device=__SCREAMING_SNAKE_CASE	).manual_seed(__SCREAMING_SNAKE_CASE	)
            __snake_case            =	{
                '''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    lowerCAmelCase   ( self	)   ->	List[str]:



            '''simple docstring'''




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

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

            assert frames[0].shape == (32, 32, 3)
            __snake_case            =	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    lowerCAmelCase   ( self	)   ->	List[str]:



            '''simple docstring'''




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


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



            '''simple docstring'''




            pass


       @unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''	)
       def    lowerCAmelCase   ( self	)   ->	Union[str, Any]:



            '''simple docstring'''




            pass


       @unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''	)
       def    lowerCAmelCase   ( self	)   ->	int:



            '''simple docstring'''




            pass




       def    lowerCAmelCase   ( self	)   ->	Optional[int]:



            '''simple docstring'''




            return super().test_progress_bar()





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


       def    lowerCAmelCase   ( self	)   ->	str:



            '''simple docstring'''




            __snake_case            =	VideoToVideoSDPipeline.from_pretrained('''cerspense/zeroscope_v2_XL'''     ,	torch_dtype=torch.floataa	)
            pipe.enable_model_cpu_offload()

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

            __snake_case            =	'''Spiderman is surfing'''

            __snake_case            =	pipe(__SCREAMING_SNAKE_CASE     ,	video=__SCREAMING_SNAKE_CASE     ,	generator=__SCREAMING_SNAKE_CASE     ,	num_inference_steps=3     ,	output_type='''pt'''	).frames

            __snake_case            =	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 Optional

import pyspark

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





class      lowerCAmelCase (			__lowerCAmelCase):


       def __init__( self     ,	__SCREAMING_SNAKE_CASE     ,	__SCREAMING_SNAKE_CASE = None     ,	__SCREAMING_SNAKE_CASE = None     ,	__SCREAMING_SNAKE_CASE = True     ,	__SCREAMING_SNAKE_CASE = None     ,	__SCREAMING_SNAKE_CASE = False     ,	__SCREAMING_SNAKE_CASE = None     ,	__SCREAMING_SNAKE_CASE = True     ,	__SCREAMING_SNAKE_CASE = "arrow"     ,	**__SCREAMING_SNAKE_CASE     ,	)   ->	Union[str, Any]:



            '''simple docstring'''




            super().__init__(
                split=__SCREAMING_SNAKE_CASE     ,	features=__SCREAMING_SNAKE_CASE     ,	cache_dir=__SCREAMING_SNAKE_CASE     ,	keep_in_memory=__SCREAMING_SNAKE_CASE     ,	streaming=__SCREAMING_SNAKE_CASE     ,	**__SCREAMING_SNAKE_CASE     ,	)
            __snake_case            =	load_from_cache_file
            __snake_case            =	file_format
            __snake_case            =	Spark(
                df=__SCREAMING_SNAKE_CASE     ,	features=__SCREAMING_SNAKE_CASE     ,	cache_dir=__SCREAMING_SNAKE_CASE     ,	working_dir=__SCREAMING_SNAKE_CASE     ,	**__SCREAMING_SNAKE_CASE     ,	)




       def    lowerCAmelCase   ( self	)   ->	str:



            '''simple docstring'''




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






