






'''simple docstring'''

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 							__magic_name__     (	UpperCAmelCase_,						unittest.TestCase				):







	"""simple docstring"""



	SCREAMING_SNAKE_CASE_					:      Union[str, Any]        =  TextToVideoSDPipeline
	SCREAMING_SNAKE_CASE_					:      str        =  TEXT_TO_IMAGE_PARAMS
	SCREAMING_SNAKE_CASE_					:      str        =  TEXT_TO_IMAGE_BATCH_PARAMS
	# No `output_type`.
	SCREAMING_SNAKE_CASE_					:      Tuple        =  frozenset(
	    [
	        'num_inference_steps',
	        'generator',
	        'latents',
	        'return_dict',
	        'callback',
	        'callback_steps',
	    ]				)
	def        snake_case__		(							self       )						->  List[str]:
							torch.manual_seed(0       )
							lowercase_      :   List[Any]        =			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_      :   Union[str, Any]        =			DDIMScheduler(
							    beta_start=0.0_00_85  ,      beta_end=0.0_12  ,      beta_schedule='scaled_linear'  ,      clip_sample=_lowercase  ,      set_alpha_to_one=_lowercase  ,      )
							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_      :   Dict        =			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  ,      )
							lowercase_      :   Dict        =			CLIPTextModel(_lowercase       )
							lowercase_      :   str        =			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        snake_case__		(							self  ,      _lowercase  ,      _lowercase=0       )						->  int:
							if str(_lowercase       ).startswith('mps'       ):
													lowercase_      :   Tuple        =			torch.manual_seed(_lowercase       )
							else:
													lowercase_      :   List[str]        =			torch.Generator(device=_lowercase       ).manual_seed(_lowercase       )
							lowercase_      :   Union[str, Any]        =			{
							    '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        snake_case__		(							self       )						->  Optional[Any]:
							lowercase_      :   Dict        =			'cpu'  # ensure determinism for the device-dependent torch.Generator
							lowercase_      :   List[str]        =			self.get_dummy_components()
							lowercase_      :   int        =			TextToVideoSDPipeline(**_lowercase       )
							lowercase_      :   Dict        =			sd_pipe.to(_lowercase       )
							sd_pipe.set_progress_bar_config(disable=_lowercase       )

							lowercase_      :   Optional[Any]        =			self.get_dummy_inputs(_lowercase       )
							lowercase_      :   Any        =			'np'
							lowercase_      :   Union[str, Any]        =			sd_pipe(**_lowercase       ).frames
							lowercase_      :   int        =			frames[0][-3:, -3:, -1]

							assert frames[0].shape == (64, 64, 3)
							lowercase_      :   List[str]        =			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        snake_case__		(							self       )						->  Optional[int]:
							self._test_attention_slicing_forward_pass(test_mean_pixel_difference=_lowercase  ,      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        snake_case__		(							self       )						->  Any:
							self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=_lowercase  ,      expected_max_diff=1E-2       )
	@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'       )
	def        snake_case__		(							self       )						->  Union[str, Any]:
							pass
	@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.'       )
	def        snake_case__		(							self       )						->  Optional[Any]:
							pass
	@unittest.skip(reason='`num_images_per_prompt` argument is not supported for this pipeline.'       )
	def        snake_case__		(							self       )						->  Optional[int]:
							pass







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






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







	"""simple docstring"""



	def        snake_case__		(							self       )						->  int:
							lowercase_      :   str        =			load_numpy(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy'       )

							lowercase_      :   List[str]        =			TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'       )
							lowercase_      :   Any        =			DPMSolverMultistepScheduler.from_config(pipe.scheduler.config       )
							lowercase_      :   int        =			pipe.to('cuda'       )

							lowercase_      :   int        =			'Spiderman is surfing'
							lowercase_      :   Tuple        =			torch.Generator(device='cpu'       ).manual_seed(0       )

							lowercase_      :   int        =			pipe(_lowercase  ,      generator=_lowercase  ,      num_inference_steps=25  ,      output_type='pt'       ).frames
							lowercase_      :   Tuple        =			video_frames.cpu().numpy()

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







	def        snake_case__		(							self       )						->  int:
							lowercase_      :   List[Any]        =			load_numpy(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy'       )

							lowercase_      :   Union[str, Any]        =			TextToVideoSDPipeline.from_pretrained('damo-vilab/text-to-video-ms-1.7b'       )
							lowercase_      :   List[str]        =			pipe.to('cuda'       )

							lowercase_      :   str        =			'Spiderman is surfing'
							lowercase_      :   Any        =			torch.Generator(device='cpu'       ).manual_seed(0       )

							lowercase_      :   List[str]        =			pipe(_lowercase  ,      generator=_lowercase  ,      num_inference_steps=2  ,      output_type='pt'       ).frames
							lowercase_      :   Any        =			video_frames.cpu().numpy()

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








'''simple docstring'''

from pathlib import PurePosixPath
from typing import Optional

import fsspec
from fsspec import AbstractFileSystem
from huggingface_hub.hf_api import DatasetInfo

from ..utils.file_utils import get_authentication_headers_for_url
from ..utils.hub import hf_hub_url






class 							__magic_name__     (	UpperCAmelCase_				):







	"""simple docstring"""



	SCREAMING_SNAKE_CASE_					:      Union[str, Any]        =  ''
	SCREAMING_SNAKE_CASE_					:      Union[str, Any]        =  'hf-legacy'  # "hf://"" is reserved for hffs
	def __init__(							self  ,      _lowercase = None  ,      _lowercase = None  ,      **_lowercase  ,      )						->  Optional[Any]:
							super().__init__(self  ,      **_lowercase       )
							lowercase_      :   int        =			repo_info
							lowercase_      :   List[Any]        =			token
							lowercase_      :   Union[str, Any]        =			None
	def        lowerCamelCase__		(							self       )						->  Optional[Any]:
							if self.dir_cache is None:
													lowercase_      :   Optional[Any]        =			{}
													for hf_file in self.repo_info.siblings:
																			# TODO(QL): add sizes
																			lowercase_      :   str        =			{
																			    'name': hf_file.rfilename,
																			    'size': None,
																			    'type': 'file',
																			}
																			self.dir_cache.update(
																			    {
																			        str(_lowercase       ): {'name': str(_lowercase       ), 'size': None, 'type': 'directory'}
																			        for d in list(PurePosixPath(hf_file.rfilename       ).parents       )[:-1]
																			    }       )
	def        lowerCamelCase__		(							self  ,      _lowercase  ,      _lowercase = "rb"  ,      **_lowercase  ,      )						->  Dict:
							if not isinstance(self.repo_info  ,      _lowercase       ):
													raise NotImplementedError(f"Open is only implemented for dataset repositories, but got {self.repo_info}"       )
							lowercase_      :   Optional[int]        =			hf_hub_url(self.repo_info.id  ,      _lowercase  ,      revision=self.repo_info.sha       )
							return fsspec.open(
							    _lowercase  ,      mode=_lowercase  ,      headers=get_authentication_headers_for_url(_lowercase  ,      use_auth_token=self.token       )  ,      client_kwargs={'trust_env': True}  ,      ).open()
	def        lowerCamelCase__		(							self  ,      _lowercase  ,      **_lowercase       )						->  Tuple:
							self._get_dirs()
							lowercase_      :   str        =			self._strip_protocol(_lowercase       )
							if path in self.dir_cache:
													return self.dir_cache[path]
							else:
													raise FileNotFoundError(_lowercase       )







	def        lowerCamelCase__		(							self  ,      _lowercase  ,      _lowercase=False  ,      **_lowercase       )						->  List[str]:
							self._get_dirs()
							lowercase_      :   List[str]        =			PurePosixPath(path.strip('/'       )       )
							lowercase_      :   List[str]        =			{}
							for p, f in self.dir_cache.items():
													lowercase_      :   Tuple        =			PurePosixPath(p.strip('/'       )       )
													lowercase_      :   Optional[int]        =			p.parent
													if root == path:
																			lowercase_      :   List[str]        =			f
							lowercase_      :   List[str]        =			list(paths.values()       )
							if detail:
													return out
							else:
													return sorted(f['name'] for f in out       )
