





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     ):



					'''simple docstring'''

					__snake_case					:			List[str]				=    VideoToVideoSDPipeline
					__snake_case					:			int				=    TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({"video"}     ) - {"image", "width", "height"}
					__snake_case					:			Dict				=    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"video"}     ) - {"image"}
					__snake_case					:			Any				=    PipelineTesterMixin.required_optional_params - {"latents"}
					__snake_case					:			Any				=    False

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






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




											'''simple docstring'''


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

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






					def        SCREAMING_SNAKE_CASE__						(							self		:					Any		,lowerCamelCase__		:					Optional[int]		,lowerCamelCase__		:					List[Any]=0 )  ->	int:




											'''simple docstring'''


											SCREAMING_SNAKE_CASE	=		floats_tensor((1, 3, 3, 32, 32)		,rng=random.Random(__a ) ).to(__a )

											if str(__a ).startswith("""mps""" ):
																	SCREAMING_SNAKE_CASE	=		torch.manual_seed(__a )
											else:
																	SCREAMING_SNAKE_CASE	=		torch.Generator(device=__a ).manual_seed(__a )
											SCREAMING_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        SCREAMING_SNAKE_CASE__						(							self		:					List[str] )  ->	List[Any]:




											'''simple docstring'''


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

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

											assert frames[0].shape == (32, 32, 3)
											SCREAMING_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        SCREAMING_SNAKE_CASE__						(							self		:					int )  ->	Optional[Any]:




											'''simple docstring'''


											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        SCREAMING_SNAKE_CASE__						(							self		:					Tuple )  ->	Optional[int]:




											'''simple docstring'''


											pass






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




											'''simple docstring'''


											pass






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




											'''simple docstring'''


											pass






					def        SCREAMING_SNAKE_CASE__						(							self		:					str )  ->	List[Any]:




											'''simple docstring'''


											return super().test_progress_bar()

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



					'''simple docstring'''

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




											'''simple docstring'''


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

											# 10 frames
											SCREAMING_SNAKE_CASE	=		torch.Generator(device="""cpu""" ).manual_seed(0 )
											SCREAMING_SNAKE_CASE	=		torch.randn((1, 10, 3, 1024, 576)		,generator=__a )
											SCREAMING_SNAKE_CASE	=		video.to("""cuda""" )

											SCREAMING_SNAKE_CASE	=		"Spiderman is surfing"

											SCREAMING_SNAKE_CASE	=		pipe(__a		,video=__a		,generator=__a		,num_inference_steps=3		,output_type="""pt""" ).frames

											SCREAMING_SNAKE_CASE	=		np.array([-1.0458984, -1.1279297, -0.9663086, -0.91503906, -0.75097656] )
											assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array ).sum() < 1e-2









import inspect
import re
from hashlib import shaaaa
from typing import Dict, List

from .arrow import arrow
from .audiofolder import audiofolder
from .csv import csv
from .imagefolder import imagefolder
from .json import json
from .pandas import pandas
from .parquet import parquet
from .sql import sql  # noqa F401
from .text import text



def      __lowercase     (   _SCREAMING_SNAKE_CASE			)				->     str:


      '''simple docstring'''


      SCREAMING_SNAKE_CASE	=		[]
      for line in lines:
            SCREAMING_SNAKE_CASE	=		re.sub(r"""#.*"""				,		""""""				,		_SCREAMING_SNAKE_CASE			)  # remove comments
            if line:
                  filtered_lines.append(_SCREAMING_SNAKE_CASE			)
      SCREAMING_SNAKE_CASE	=		"""\n""".join(_SCREAMING_SNAKE_CASE			)

      # Make a hash from all this code
      SCREAMING_SNAKE_CASE	=		full_str.encode("""utf-8"""			)
      return shaaaa(_SCREAMING_SNAKE_CASE			).hexdigest()


# get importable module names and hash for caching
SCREAMING_SNAKE_CASE_		     = {
    """csv""": (csv.__name__, _hash_python_lines(inspect.getsource(csv).splitlines())),
    """json""": (json.__name__, _hash_python_lines(inspect.getsource(json).splitlines())),
    """pandas""": (pandas.__name__, _hash_python_lines(inspect.getsource(pandas).splitlines())),
    """parquet""": (parquet.__name__, _hash_python_lines(inspect.getsource(parquet).splitlines())),
    """arrow""": (arrow.__name__, _hash_python_lines(inspect.getsource(arrow).splitlines())),
    """text""": (text.__name__, _hash_python_lines(inspect.getsource(text).splitlines())),
    """imagefolder""": (imagefolder.__name__, _hash_python_lines(inspect.getsource(imagefolder).splitlines())),
    """audiofolder""": (audiofolder.__name__, _hash_python_lines(inspect.getsource(audiofolder).splitlines())),
}

# Used to infer the module to use based on the data files extensions
SCREAMING_SNAKE_CASE_		     = {
    """.csv""": ("""csv""", {}),
    """.tsv""": ("""csv""", {"""sep""": """\t"""}),
    """.json""": ("""json""", {}),
    """.jsonl""": ("""json""", {}),
    """.parquet""": ("""parquet""", {}),
    """.arrow""": ("""arrow""", {}),
    """.txt""": ("""text""", {}),
}
_EXTENSION_TO_MODULE.update({ext: ("""imagefolder""", {}) for ext in imagefolder.ImageFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext.upper(): ("""imagefolder""", {}) for ext in imagefolder.ImageFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext: ("""audiofolder""", {}) for ext in audiofolder.AudioFolder.EXTENSIONS})
_EXTENSION_TO_MODULE.update({ext.upper(): ("""audiofolder""", {}) for ext in audiofolder.AudioFolder.EXTENSIONS})
SCREAMING_SNAKE_CASE_		     = {"""imagefolder""", """audiofolder"""}

# Used to filter data files based on extensions given a module name
SCREAMING_SNAKE_CASE_		     = {}
for _ext, (_module, _) in _EXTENSION_TO_MODULE.items():
      _MODULE_TO_EXTENSIONS.setdefault(_module, []).append(_ext)

_MODULE_TO_EXTENSIONS["imagefolder"].append(""".zip""")



_MODULE_TO_EXTENSIONS["audiofolder"].append(""".zip""")


