





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 		__lowerCAmelCase			(   self				) ->      Union[str, Any]:





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

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




						def 		__lowerCAmelCase			(   self	,		SCREAMING_SNAKE_CASE	,		SCREAMING_SNAKE_CASE=0				) ->      Optional[int]:





												"""simple docstring"""
												if str(SCREAMING_SNAKE_CASE				).startswith("mps"				):
																		UpperCamelCase  							= torch.manual_seed(SCREAMING_SNAKE_CASE				)
												else:
																		UpperCamelCase  							= torch.Generator(device=SCREAMING_SNAKE_CASE				).manual_seed(SCREAMING_SNAKE_CASE				)
												UpperCamelCase  							= {
												    "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				) ->      Optional[Any]:





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

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

												assert frames[0].shape == (64, 64, 3)
												UpperCamelCase  							= 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				) ->      Any:





												"""simple docstring"""
												self._test_attention_slicing_forward_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE	,		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				) ->      str:





												"""simple docstring"""
												self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=SCREAMING_SNAKE_CASE	,		expected_max_diff=1e-2				)




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





												"""simple docstring"""
												pass




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





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


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



						"""simple docstring"""







						def 		__lowerCAmelCase			(   self				) ->      str:





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

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

												UpperCamelCase  							= "Spiderman is surfing"
												UpperCamelCase  							= torch.Generator(device="cpu"				).manual_seed(0				)

												UpperCamelCase  							= pipe(SCREAMING_SNAKE_CASE	,		generator=SCREAMING_SNAKE_CASE	,		num_inference_steps=25	,		output_type="pt"				).frames
												UpperCamelCase  							= video_frames.cpu().numpy()

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




						def 		__lowerCAmelCase			(   self				) ->      Tuple:





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

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

												UpperCamelCase  							= "Spiderman is surfing"
												UpperCamelCase  							= torch.Generator(device="cpu"				).manual_seed(0				)

												UpperCamelCase  							= pipe(SCREAMING_SNAKE_CASE	,		generator=SCREAMING_SNAKE_CASE	,		num_inference_steps=2	,		output_type="pt"				).frames
												UpperCamelCase  							= video_frames.cpu().numpy()

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












import heapq as hq
import math
from collections.abc import Iterator
class  _UpperCAmelCase   :

				"""simple docstring"""




				def __init__(	self		:       str,    lowerCamelCase		:       List[Any]    ):

								'''simple docstring'''
								lowercase__  					=     str(id_    )
								lowercase__  					=     None
								lowercase__  					=     None
								lowercase__  					=     []
								lowercase__  					=     {}  # {vertex:distance}




				def __lt__(	self		:       Dict,    lowerCamelCase		:       Tuple    ):

								'''simple docstring'''
								return self.key < other.key




				def __repr__(	self		:       Dict    ):

								'''simple docstring'''
								return self.id




				def 							lowercase__  (	self		:       str,    lowerCamelCase		:       List[str]    ):

								'''simple docstring'''
								self.neighbors.append(lowerCamelCase    )




				def 							lowercase__  (	self		:       Any,    lowerCamelCase		:       Dict,    lowerCamelCase		:       Optional[Any]    ):

								'''simple docstring'''
								lowercase__  					=     weight






def     a     (		lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_						,  lowerCamelCase_							):






				'''simple docstring'''
				# add the neighbors:
				graph[a - 1].add_neighbor(graph[b - 1]							)
				graph[b - 1].add_neighbor(graph[a - 1]							)
				# add the edges:
				graph[a - 1].add_edge(graph[b - 1]						,  lowerCamelCase_							)
				graph[b - 1].add_edge(graph[a - 1]						,  lowerCamelCase_							)






def     a     (		lowerCamelCase_						,  lowerCamelCase_							):






				'''simple docstring'''
				lowercase__  					=     []
				for u in graph:
								lowercase__  					=     math.inf
								lowercase__  					=     None
				lowercase__  					=     0
				lowercase__  					=     graph[:]
				while q:
								lowercase__  					=     min(lowerCamelCase_							)
								q.remove(lowerCamelCase_							)
								for v in u.neighbors:
												if (v in q) and (u.edges[v.id] < v.key):
																lowercase__  					=     u
																lowercase__  					=     u.edges[v.id]
				for i in range(1						,  len(lowerCamelCase_							)							):
								a.append((int(graph[i].id							) + 1, int(graph[i].pi.id							) + 1)							)
				return a






def     a     (		lowerCamelCase_						,  lowerCamelCase_							):






				'''simple docstring'''
				for u in graph:
								lowercase__  					=     math.inf
								lowercase__  					=     None
				lowercase__  					=     0

				lowercase__  					=     list(lowerCamelCase_							)
				hq.heapify(lowerCamelCase_							)

				while h:
								lowercase__  					=     hq.heappop(lowerCamelCase_							)
								for v in u.neighbors:
												if (v in h) and (u.edges[v.id] < v.key):
																lowercase__  					=     u
																lowercase__  					=     u.edges[v.id]
																hq.heapify(lowerCamelCase_							)

				for i in range(1						,  len(lowerCamelCase_							)							):
								yield (int(graph[i].id							) + 1, int(graph[i].pi.id							) + 1)






def     a     (		):






				'''simple docstring'''


if __name__ == "__main__":
							import doctest

							doctest.testmod()




