






import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    TextToVideoSDPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import is_xformers_available, load_numpy, skip_mps, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
@skip_mps
class  _UpperCAmelCase   (			A__						,unittest.TestCase      ):

				"""simple docstring"""




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




				def 							lowercase__  (	self		:       Any    ):

								'''simple docstring'''
								torch.manual_seed(0    )
								lowercase__  					=     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__  					=     DDIMScheduler(
								    beta_start=0.00085,    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=[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__  					=     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=1_000,    hidden_act='''gelu''',    projection_dim=512,    )
								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 							lowercase__  (	self		:       int,    lowerCamelCase		:       Any,    lowerCamelCase		:       Optional[int]=0    ):

								'''simple docstring'''
								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''',
								    '''generator''': generator,
								    '''num_inference_steps''': 2,
								    '''guidance_scale''': 6.0,
								    '''output_type''': '''pt''',
								}
								return inputs




				def 							lowercase__  (	self		:       List[str]    ):

								'''simple docstring'''
								lowercase__  					=     '''cpu'''  # ensure determinism for the device-dependent torch.Generator
								lowercase__  					=     self.get_dummy_components()
								lowercase__  					=     TextToVideoSDPipeline(**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 == (64, 64, 3)
								lowercase__  					=     np.array([158.0, 160.0, 153.0, 125.0, 100.0, 121.0, 111.0, 93.0, 113.0]    )

								assert np.abs(image_slice.flatten() - expected_slice    ).max() < 1E-2




				def 							lowercase__  (	self		:       List[str]    ):

								'''simple docstring'''
								self._test_attention_slicing_forward_pass(test_mean_pixel_difference=lowerCamelCase,    expected_max_diff=3E-3    )




				@unittest.skipIf(
				    torch_device != '''cuda''' or not is_xformers_available(),    reason='''XFormers attention is only available with CUDA and `xformers` installed''',    )
				def 							lowercase__  (	self		:       Optional[Any]    ):

								'''simple docstring'''
								self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=lowerCamelCase,    expected_max_diff=1E-2    )




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

								'''simple docstring'''
								pass




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

								'''simple docstring'''
								pass




				@unittest.skip(reason='''`num_images_per_prompt` argument is not supported for this pipeline.'''    )
				def 							lowercase__  (	self		:       List[str]    ):

								'''simple docstring'''
								pass




				def 							lowercase__  (	self		:       Dict    ):

								'''simple docstring'''
								return super().test_progress_bar()



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

				"""simple docstring"""




				def 							lowercase__  (	self		:       Optional[int]    ):

								'''simple docstring'''
								lowercase__  					=     load_numpy(
								    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'''    )

								lowercase__  					=     TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''    )
								lowercase__  					=     DPMSolverMultistepScheduler.from_config(pipe.scheduler.config    )
								lowercase__  					=     pipe.to('''cuda'''    )

								lowercase__  					=     '''Spiderman is surfing'''
								lowercase__  					=     torch.Generator(device='''cpu'''    ).manual_seed(0    )

								lowercase__  					=     pipe(lowerCamelCase,    generator=lowerCamelCase,    num_inference_steps=25,    output_type='''pt'''    ).frames
								lowercase__  					=     video_frames.cpu().numpy()

								assert np.abs(expected_video - video    ).mean() < 5E-2




				def 							lowercase__  (	self		:       int    ):

								'''simple docstring'''
								lowercase__  					=     load_numpy(
								    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'''    )

								lowercase__  					=     TextToVideoSDPipeline.from_pretrained('''damo-vilab/text-to-video-ms-1.7b'''    )
								lowercase__  					=     pipe.to('''cuda'''    )

								lowercase__  					=     '''Spiderman is surfing'''
								lowercase__  					=     torch.Generator(device='''cpu'''    ).manual_seed(0    )

								lowercase__  					=     pipe(lowerCamelCase,    generator=lowerCamelCase,    num_inference_steps=2,    output_type='''pt'''    ).frames
								lowercase__  					=     video_frames.cpu().numpy()

								assert np.abs(expected_video - video    ).mean() < 5E-2












import importlib
import shutil
import threading
import warnings
from typing import List

import fsspec
import fsspec.asyn

from . import compression
from .hffilesystem import HfFileSystem


A__      :     int       =					importlib.util.find_spec('s3fs') is not None

if _has_safs:
							from .safilesystem import SaFileSystem  # noqa: F401

A__      :     List[compression.BaseCompressedFileFileSystem]       =					[
    compression.BzaFileSystem,
    compression.GzipFileSystem,
    compression.LzaFileSystem,
    compression.XzFileSystem,
    compression.ZstdFileSystem,
]

# Register custom filesystems
for fs_class in COMPRESSION_FILESYSTEMS + [HfFileSystem]:
							if fs_class.protocol in fsspec.registry and fsspec.registry[fs_class.protocol] is not fs_class:
														warnings.warn(F"A filesystem protocol was already set for {fs_class.protocol} and will be overwritten.")
							fsspec.register_implementation(fs_class.protocol, fs_class, clobber=True)






def     a     (		lowerCamelCase_							):






				'''simple docstring'''
				if "://" in dataset_path:
								lowercase__  					=     dataset_path.split('''://'''							)[1]
				return dataset_path






def     a     (		lowerCamelCase_							):






				'''simple docstring'''
				if fs is not None and fs.protocol != "file":
								return True
				else:
								return False






def     a     (		lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_							):






				'''simple docstring'''
				lowercase__  					=     not is_remote_filesystem(lowerCamelCase_							)
				if is_local:
								# LocalFileSystem.mv does copy + rm, it is more efficient to simply move a local directory
								shutil.move(fs._strip_protocol(lowerCamelCase_							)						,  fs._strip_protocol(lowerCamelCase_							)							)
				else:
								fs.mv(lowerCamelCase_						,  lowerCamelCase_						,  recursive=lowerCamelCase_							)






def     a     (		):






				'''simple docstring'''
				if hasattr(fsspec.asyn						,  '''reset_lock'''							):
								# for future fsspec>2022.05.0
								fsspec.asyn.reset_lock()
				else:
								lowercase__  					=     None
								lowercase__  					=     None
								lowercase__  					=     threading.Lock()




