

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

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







				def  lowerCamelCase_	(       self     :		List[Any]   ):
										'''simple docstring'''
										torch.manual_seed(0   )
										lowercase :							List[str]    		=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		,					)
										lowercase :							List[str]    		=DDIMScheduler(
										    beta_start=0.0_00_85		,					beta_end=0.0_12		,					beta_schedule='''scaled_linear'''		,					clip_sample=__A		,					set_alpha_to_one=__A		,					)
										torch.manual_seed(0   )
										lowercase :							List[str]    		=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   )
										lowercase :							Optional[int]    		=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		,					)
										lowercase :							Optional[Any]    		=CLIPTextModel(__A   )
										lowercase :							Optional[int]    		=CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''   )

										lowercase :							Optional[Any]    		={
										    '''unet''': unet,
										    '''scheduler''': scheduler,
										    '''vae''': vae,
										    '''text_encoder''': text_encoder,
										    '''tokenizer''': tokenizer,
										}
										return components







				def  lowerCamelCase_	(       self     :		Optional[int]		,					UpperCAmelCase__     :		Union[str, Any]		,					UpperCAmelCase__     :		Tuple=0   ):
										'''simple docstring'''
										# 3 frames
										lowercase :							Dict    		=floats_tensor((1, 3, 3, 32, 32)		,					rng=random.Random(__A   )   ).to(__A   )

										if str(__A   ).startswith('''mps'''   ):
																lowercase :							int    		=torch.manual_seed(__A   )
										else:
																lowercase :							Optional[Any]    		=torch.Generator(device=__A   ).manual_seed(__A   )
										lowercase :							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  lowerCamelCase_	(       self     :		Tuple   ):
										'''simple docstring'''
										lowercase :							Union[str, Any]    		='''cpu'''  # ensure determinism for the device-dependent torch.Generator
										lowercase :							Dict    		=self.get_dummy_components()
										lowercase :							int    		=VideoToVideoSDPipeline(**__A   )
										lowercase :							List[Any]    		=sd_pipe.to(__A   )
										sd_pipe.set_progress_bar_config(disable=__A   )

										lowercase :							List[str]    		=self.get_dummy_inputs(__A   )
										lowercase :							Union[str, Any]    		='''np'''
										lowercase :							List[Any]    		=sd_pipe(**__A   ).frames
										lowercase :							List[Any]    		=frames[0][-3:, -3:, -1]

										assert frames[0].shape == (32, 32, 3)
										lowercase :							Tuple    		=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     :		Tuple   ):
										'''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  lowerCamelCase_	(       self     :		Union[str, Any]   ):
										'''simple docstring'''
										pass







				@unittest.skip(reason='''Batching needs to be properly figured out first for this pipeline.'''   )
				def  lowerCamelCase_	(       self     :		Optional[int]   ):
										'''simple docstring'''
										pass







				@unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''   )
				def  lowerCamelCase_	(       self     :		Any   ):
										'''simple docstring'''
										pass







				def  lowerCamelCase_	(       self     :		List[Any]   ):
										'''simple docstring'''
										return super().test_progress_bar()





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







				def  lowerCamelCase_	(       self     :		Tuple   ):
										'''simple docstring'''
										lowercase :							List[Any]    		=VideoToVideoSDPipeline.from_pretrained('''cerspense/zeroscope_v2_XL'''		,					torch_dtype=torch.floataa   )
										pipe.enable_model_cpu_offload()

										# 10 frames
										lowercase :							Any    		=torch.Generator(device='''cpu'''   ).manual_seed(0   )
										lowercase :							str    		=torch.randn((1, 10, 3, 1024, 576)		,					generator=__A   )
										lowercase :							List[Any]    		=video.to('''cuda'''   )

										lowercase :							List[str]    		='''Spiderman is surfing'''

										lowercase :							Optional[int]    		=pipe(__A		,					video=__A		,					generator=__A		,					num_inference_steps=3		,					output_type='''pt'''   ).frames

										lowercase :							Optional[Any]    		=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 importlib
import os

import fsspec
import pytest
from fsspec import register_implementation
from fsspec.registry import _registry as _fsspec_registry

from datasets.filesystems import COMPRESSION_FILESYSTEMS, HfFileSystem, extract_path_from_uri, is_remote_filesystem

from .utils import require_lza, require_zstandard
def 							SCREAMING_SNAKE_CASE_	(					UpperCAmelCase_	:     List[Any]				)			->						Dict:
       assert "mock" in _fsspec_registry
       assert "bz2" in _fsspec_registry
def 							SCREAMING_SNAKE_CASE_	(					)			->						Union[str, Any]:
       assert "mock" not in _fsspec_registry
       assert "bz2" in _fsspec_registry
def 							SCREAMING_SNAKE_CASE_	(					)			->						Dict:
       SCREAMING_SNAKE_CASE_							:				Optional[Any]		='''mock-s3-bucket'''
       SCREAMING_SNAKE_CASE_							:				Any		=f's3://{mock_bucket}'
       SCREAMING_SNAKE_CASE_							:				Dict		=extract_path_from_uri(UpperCAmelCase_				)
       assert dataset_path.startswith('''s3://'''				) is False

       SCREAMING_SNAKE_CASE_							:				Dict		='''./local/path'''
       SCREAMING_SNAKE_CASE_							:				str		=extract_path_from_uri(UpperCAmelCase_				)
       assert dataset_path == new_dataset_path
def 							SCREAMING_SNAKE_CASE_	(					UpperCAmelCase_	:     Union[str, Any]				)			->						List[str]:
       SCREAMING_SNAKE_CASE_							:				Union[str, Any]		=is_remote_filesystem(UpperCAmelCase_				)
       assert is_remote is True

       SCREAMING_SNAKE_CASE_							:				Dict		=fsspec.filesystem('''file'''				)

       SCREAMING_SNAKE_CASE_							:				Optional[int]		=is_remote_filesystem(UpperCAmelCase_				)
       assert is_remote is False
@pytest.mark.parametrize('''compression_fs_class'''			,							UpperCAmelCase_				)
def 							SCREAMING_SNAKE_CASE_	(					UpperCAmelCase_	:     Union[str, Any]			,							UpperCAmelCase_	:     Any			,							UpperCAmelCase_	:     Optional[Any]			,							UpperCAmelCase_	:     Optional[Any]			,							UpperCAmelCase_	:     int			,							UpperCAmelCase_	:     str			,							UpperCAmelCase_	:     Any				)			->						List[Any]:
       SCREAMING_SNAKE_CASE_							:				str		={'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_file, '''bz2''': bza_file, '''lz4''': lza_file}
       SCREAMING_SNAKE_CASE_							:				List[Any]		=input_paths[compression_fs_class.protocol]
       if input_path is None:
              SCREAMING_SNAKE_CASE_							:				List[str]		=f'for \'{compression_fs_class.protocol}\' compression protocol, '
              if compression_fs_class.protocol == "lz4":
                     reason += require_lza.kwargs["reason"]
              elif compression_fs_class.protocol == "zstd":
                     reason += require_zstandard.kwargs["reason"]
              pytest.skip(UpperCAmelCase_				)
       SCREAMING_SNAKE_CASE_							:				Union[str, Any]		=fsspec.filesystem(compression_fs_class.protocol			,							fo=UpperCAmelCase_				)
       assert isinstance(UpperCAmelCase_			,							UpperCAmelCase_				)
       SCREAMING_SNAKE_CASE_							:				str		=os.path.basename(UpperCAmelCase_				)
       SCREAMING_SNAKE_CASE_							:				Optional[Any]		=expected_filename[: expected_filename.rindex('''.'''				)]
       assert fs.glob('''*'''				) == [expected_filename]
       with fs.open(UpperCAmelCase_			,							'''r'''			,							encoding='''utf-8'''				) as f, open(UpperCAmelCase_			,							encoding='''utf-8'''				) as expected_file:
              assert f.read() == expected_file.read()
@pytest.mark.parametrize('''protocol'''			,							['''zip''', '''gzip''']				)
def 							SCREAMING_SNAKE_CASE_	(					UpperCAmelCase_	:     Any			,							UpperCAmelCase_	:     Union[str, Any]			,							UpperCAmelCase_	:     str				)			->						Any:
       SCREAMING_SNAKE_CASE_							:				Tuple		={'''zip''': zip_jsonl_path, '''gzip''': jsonl_gz_path}
       SCREAMING_SNAKE_CASE_							:				List[Any]		=compressed_file_paths[protocol]
       SCREAMING_SNAKE_CASE_							:				str		='''dataset.jsonl'''
       SCREAMING_SNAKE_CASE_							:				Dict		=f'{protocol}://{member_file_path}::{compressed_file_path}'
       SCREAMING_SNAKE_CASE_	,     *SCREAMING_SNAKE_CASE_							:				Dict		=fsspec.get_fs_token_paths(UpperCAmelCase_				)
       assert fs.isfile(UpperCAmelCase_				)
       assert not fs.isfile('''non_existing_''' + member_file_path				)
@pytest.mark.integration
def 							SCREAMING_SNAKE_CASE_	(					UpperCAmelCase_	:     List[str]			,							UpperCAmelCase_	:     Optional[int]			,							UpperCAmelCase_	:     Optional[int]			,							UpperCAmelCase_	:     int				)			->						int:
       SCREAMING_SNAKE_CASE_							:				Dict		=hf_api.dataset_info(UpperCAmelCase_			,							token=UpperCAmelCase_				)
       SCREAMING_SNAKE_CASE_							:				Optional[int]		=HfFileSystem(repo_info=UpperCAmelCase_			,							token=UpperCAmelCase_				)
       assert sorted(hffs.glob('''*'''				)				) == [".gitattributes", "data"]
       assert hffs.isdir('''data'''				)
       assert hffs.isfile('''.gitattributes'''				) and hffs.isfile('''data/text_data.txt'''				)
       with open(UpperCAmelCase_				) as f:
              assert hffs.open('''data/text_data.txt'''			,							'''r'''				).read() == f.read()

def 							SCREAMING_SNAKE_CASE_	(					)			->						Union[str, Any]:
       SCREAMING_SNAKE_CASE_							:				Tuple		='''bz2'''

       # Import module
       import datasets.filesystems

       # Overwrite protocol and reload
       register_implementation(UpperCAmelCase_			,							UpperCAmelCase_			,							clobber=UpperCAmelCase_				)
       with pytest.warns(UpperCAmelCase_				) as warning_info:
              importlib.reload(datasets.filesystems				)

       assert len(UpperCAmelCase_				) == 1
       assert (
           str(warning_info[0].message				)
           == f'A filesystem protocol was already set for {protocol} and will be overwritten.'
       )
