


"""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 _lowerCAmelCase       (   snake_case_					,							unittest.TestCase			):
			__UpperCAmelCase				: Any													=						TextToVideoSDPipeline
			__UpperCAmelCase				: Any													=						TEXT_TO_IMAGE_PARAMS
			__UpperCAmelCase				: Optional[Any]													=						TEXT_TO_IMAGE_BATCH_PARAMS
			# No `output_type`.
			__UpperCAmelCase				: Optional[int]													=						frozenset(
			    [
			        '''num_inference_steps''',
			        '''generator''',
			        '''latents''',
			        '''return_dict''',
			        '''callback''',
			        '''callback_steps''',
			    ]			)







			def 				lowerCamelCase	(			self		)			->	Tuple:

							'''simple docstring'''




							torch.manual_seed(0		)
							snake_case					:   Optional[int]         =					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					:   Optional[int]         =					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		)
							snake_case					:   Union[str, Any]         =					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					:   Union[str, Any]         =					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					:   Union[str, Any]         =					CLIPTextModel(A__		)
							snake_case					:   Tuple         =					CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"		)

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







			def 				lowerCamelCase	(			self		,			UpperCamelCase__		,			UpperCamelCase__=0		)			->	List[Any]:

							'''simple docstring'''




							if str(A__		).startswith("mps"		):
											snake_case					:   Tuple         =					torch.manual_seed(A__		)
							else:
											snake_case					:   List[str]         =					torch.Generator(device=A__		).manual_seed(A__		)
							snake_case					:   List[str]         =					{
							    """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 				lowerCamelCase	(			self		)			->	Union[str, Any]:

							'''simple docstring'''




							snake_case					:   List[str]         =					"""cpu"""  # ensure determinism for the device-dependent torch.Generator
							snake_case					:   Union[str, Any]         =					self.get_dummy_components()
							snake_case					:   Union[str, Any]         =					TextToVideoSDPipeline(**A__		)
							snake_case					:   int         =					sd_pipe.to(A__		)
							sd_pipe.set_progress_bar_config(disable=A__		)

							snake_case					:   int         =					self.get_dummy_inputs(A__		)
							snake_case					:   int         =					"""np"""
							snake_case					:   Any         =					sd_pipe(**A__		).frames
							snake_case					:   Dict         =					frames[0][-3:, -3:, -1]

							assert frames[0].shape == (64, 64, 3)
							snake_case					:   Optional[Any]         =					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 				lowerCamelCase	(			self		)			->	str:

							'''simple docstring'''




							self._test_attention_slicing_forward_pass(test_mean_pixel_difference=A__		,			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 				lowerCamelCase	(			self		)			->	Tuple:

							'''simple docstring'''




							self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=A__		,			expected_max_diff=1e-2		)







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

							'''simple docstring'''




							pass







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

							'''simple docstring'''




							pass







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

							'''simple docstring'''




							pass







			def 				lowerCamelCase	(			self		)			->	int:

							'''simple docstring'''




							return super().test_progress_bar()







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







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

							'''simple docstring'''




							snake_case					:   Union[str, Any]         =					load_numpy(
							    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video.npy"		)

							snake_case					:   Tuple         =					TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b"		)
							snake_case					:   Any         =					DPMSolverMultistepScheduler.from_config(pipe.scheduler.config		)
							snake_case					:   int         =					pipe.to("cuda"		)

							snake_case					:   Optional[Any]         =					"""Spiderman is surfing"""
							snake_case					:   List[str]         =					torch.Generator(device="cpu"		).manual_seed(0		)

							snake_case					:   Optional[Any]         =					pipe(A__		,			generator=A__		,			num_inference_steps=25		,			output_type="pt"		).frames
							snake_case					:   Dict         =					video_frames.cpu().numpy()

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







			def 				lowerCamelCase	(			self		)			->	Union[str, Any]:

							'''simple docstring'''




							snake_case					:   List[Any]         =					load_numpy(
							    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/text_to_video/video_2step.npy"		)

							snake_case					:   Optional[int]         =					TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b"		)
							snake_case					:   List[str]         =					pipe.to("cuda"		)

							snake_case					:   Dict         =					"""Spiderman is surfing"""
							snake_case					:   Union[str, Any]         =					torch.Generator(device="cpu"		).manual_seed(0		)

							snake_case					:   Optional[int]         =					pipe(A__		,			generator=A__		,			num_inference_steps=2		,			output_type="pt"		).frames
							snake_case					:   Optional[int]         =					video_frames.cpu().numpy()

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









"""simple docstring"""



# this script reports modified .py files under the desired list of top-level sub-dirs passed as a list of arguments, e.g.:
#   python ./utils/get_modified_files.py utils src tests examples
#
# it uses git to find the forking point and which files were modified - i.e. files not under git won't be considered
# since the output of this script is fed into Makefile commands it doesn't print a newline after the results

import re
import subprocess
import sys


__snake_case         =					subprocess.check_output("""git merge-base main HEAD""".split()).decode("""utf-8""")
__snake_case         =					(
    subprocess.check_output(F'''git diff --diff-filter=d --name-only {fork_point_sha}'''.split()).decode("""utf-8""").split()
)

__snake_case         =					"""|""".join(sys.argv[1:])
__snake_case         =					re.compile(RF'''^({joined_dirs}).*?\.py$''')

__snake_case         =					[x for x in modified_files if regex.match(x)]
print(""" """.join(relevant_modified_files), end="""""")





