



from __future__ import annotations

from sys import maxsize
from typing import Generic, TypeVar

SCREAMING_SNAKE_CASE     				=						TypeVar("T")







def snake_case__  (			__SCREAMING_SNAKE_CASE )		->							int:
		return (position - 1) // 2







def snake_case__  (			__SCREAMING_SNAKE_CASE )		->							int:
		return (2 * position) + 1







def snake_case__  (			__SCREAMING_SNAKE_CASE )		->							int:
		return (2 * position) + 2



class 	lowerCamelCase      (						Generic[T]  ):







					'''simple docstring'''







					def __init__(			self     ):
							UpperCAmelCase_						=				[]
							UpperCAmelCase_						=				{}
							UpperCAmelCase_						=				0



					def __len__(			self     ):
							return self.elements



					def __repr__(			self     ):
							return str(self.heap     )



					def    A__		(			self     ):
							# Check if the priority queue is empty
							return self.elements == 0



					def    A__		(			self    ,	lowerCAmelCase    ,	lowerCAmelCase     ):
							# Add an element with given priority to the queue
							self.heap.append((elem, weight)     )
							UpperCAmelCase_						=				self.elements
							self.elements += 1
							self._bubble_up(lowerCAmelCase     )



					def    A__		(			self     ):
							# Remove and return the element with lowest weight (highest priority)
							if self.elements > 1:
									self._swap_nodes(0    ,	self.elements - 1     )
							UpperCAmelCase_					,						UpperCAmelCase_						=				self.heap.pop()
							del self.position_map[elem]
							self.elements -= 1
							if self.elements > 0:
									UpperCAmelCase_					,						UpperCAmelCase_						=				self.heap[0]
									self._bubble_down(lowerCAmelCase     )
							return elem



					def    A__		(			self    ,	lowerCAmelCase    ,	lowerCAmelCase     ):
							# Update the weight of the given key
							UpperCAmelCase_						=				self.position_map[elem]
							UpperCAmelCase_						=				(elem, weight)
							if position > 0:
									UpperCAmelCase_						=				get_parent_position(lowerCAmelCase     )
									UpperCAmelCase_					,						UpperCAmelCase_						=				self.heap[parent_position]
									if parent_weight > weight:
											self._bubble_up(lowerCAmelCase     )
									else:
											self._bubble_down(lowerCAmelCase     )
							else:
									self._bubble_down(lowerCAmelCase     )



					def    A__		(			self    ,	lowerCAmelCase     ):
							# Place a node at the proper position (upward movement) [to be used internally
							# only]
							UpperCAmelCase_						=				self.position_map[elem]
							if curr_pos == 0:
									return None
							UpperCAmelCase_						=				get_parent_position(lowerCAmelCase     )
							UpperCAmelCase_					,						UpperCAmelCase_						=				self.heap[curr_pos]
							UpperCAmelCase_					,						UpperCAmelCase_						=				self.heap[parent_position]
							if parent_weight > weight:
									self._swap_nodes(lowerCAmelCase    ,	lowerCAmelCase     )
									return self._bubble_up(lowerCAmelCase     )
							return None



					def    A__		(			self    ,	lowerCAmelCase     ):
							# Place a node at the proper position (downward movement) [to be used
							# internally only]
							UpperCAmelCase_						=				self.position_map[elem]
							UpperCAmelCase_					,						UpperCAmelCase_						=				self.heap[curr_pos]
							UpperCAmelCase_						=				get_child_left_position(lowerCAmelCase     )
							UpperCAmelCase_						=				get_child_right_position(lowerCAmelCase     )
							if child_left_position < self.elements and child_right_position < self.elements:
									UpperCAmelCase_					,						UpperCAmelCase_						=				self.heap[child_left_position]
									UpperCAmelCase_					,						UpperCAmelCase_						=				self.heap[child_right_position]
									if child_right_weight < child_left_weight and child_right_weight < weight:
											self._swap_nodes(lowerCAmelCase    ,	lowerCAmelCase     )
											return self._bubble_down(lowerCAmelCase     )
							if child_left_position < self.elements:
									UpperCAmelCase_					,						UpperCAmelCase_						=				self.heap[child_left_position]
									if child_left_weight < weight:
											self._swap_nodes(lowerCAmelCase    ,	lowerCAmelCase     )
											return self._bubble_down(lowerCAmelCase     )
							else:
									return None
							if child_right_position < self.elements:
									UpperCAmelCase_					,						UpperCAmelCase_						=				self.heap[child_right_position]
									if child_right_weight < weight:
											self._swap_nodes(lowerCAmelCase    ,	lowerCAmelCase     )
											return self._bubble_down(lowerCAmelCase     )
							return None



					def    A__		(			self    ,	lowerCAmelCase    ,	lowerCAmelCase     ):
							# Swap the nodes at the given positions
							UpperCAmelCase_						=				self.heap[nodea_pos][0]
							UpperCAmelCase_						=				self.heap[nodea_pos][0]
							UpperCAmelCase_					,						UpperCAmelCase_						=				(
							    self.heap[nodea_pos],
							    self.heap[nodea_pos],
							)
							UpperCAmelCase_						=				nodea_pos
							UpperCAmelCase_						=				nodea_pos





class 	lowerCamelCase      (						Generic[T]  ):







					'''simple docstring'''







					def __init__(			self     ):
							UpperCAmelCase_						=				{}
							UpperCAmelCase_						=				0



					def __repr__(			self     ):
							return str(self.connections     )



					def __len__(			self     ):
							return self.nodes



					def    A__		(			self    ,	lowerCAmelCase     ):
							# Add a node in the graph if it is not in the graph
							if node not in self.connections:
									UpperCAmelCase_						=				{}
									self.nodes += 1



					def    A__		(			self    ,	lowerCAmelCase    ,	lowerCAmelCase    ,	lowerCAmelCase     ):
							# Add an edge between 2 nodes in the graph
							self.add_node(lowerCAmelCase     )
							self.add_node(lowerCAmelCase     )
							UpperCAmelCase_						=				weight
							UpperCAmelCase_						=				weight







def snake_case__  (			__SCREAMING_SNAKE_CASE			,     )		->							tuple[dict[T, int], dict[T, T | None]]:
		UpperCAmelCase_						=				{node: maxsize for node in graph.connections}
		UpperCAmelCase_						=				{node: None for node in graph.connections}

		UpperCAmelCase_						=				MinPriorityQueue()
		for node, weight in dist.items():
				priority_queue.push(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE )

		if priority_queue.is_empty():
				return dist, parent

		# initialization
		UpperCAmelCase_						=				priority_queue.extract_min()
		UpperCAmelCase_						=				0
		for neighbour in graph.connections[node]:
				if dist[neighbour] > dist[node] + graph.connections[node][neighbour]:
						UpperCAmelCase_						=				dist[node] + graph.connections[node][neighbour]
						priority_queue.update_key(__SCREAMING_SNAKE_CASE			,     dist[neighbour] )
						UpperCAmelCase_						=				node

    # running prim's algorithm
		while not priority_queue.is_empty():
				UpperCAmelCase_						=				priority_queue.extract_min()
				for neighbour in graph.connections[node]:
						if dist[neighbour] > dist[node] + graph.connections[node][neighbour]:
								UpperCAmelCase_						=				dist[node] + graph.connections[node][neighbour]
								priority_queue.update_key(__SCREAMING_SNAKE_CASE			,     dist[neighbour] )
								UpperCAmelCase_						=				node
		return dist, parent






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







					'''simple docstring'''







					lowerCAmelCase_				:    int									=  TextToVideoSDPipeline
					lowerCAmelCase_				:    Dict									=  TEXT_TO_IMAGE_PARAMS
					lowerCAmelCase_				:    Optional[Any]									=  TEXT_TO_IMAGE_BATCH_PARAMS
					# No `output_type`.
					lowerCAmelCase_				:    Optional[Any]									=  frozenset(
					    [
					        'num_inference_steps',
					        'generator',
					        'latents',
					        'return_dict',
					        'callback',
					        'callback_steps',
					    ]  )



					def    A__		(			self     ):
							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.00085    ,	beta_end=0.012    ,	beta_schedule="scaled_linear"    ,	clip_sample=lowerCAmelCase    ,	set_alpha_to_one=lowerCAmelCase    ,	)
							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-0_5    ,	num_attention_heads=4    ,	num_hidden_layers=5    ,	pad_token_id=1    ,	vocab_size=1000    ,	hidden_act="gelu"    ,	projection_dim=512    ,	)
							UpperCAmelCase_						=				CLIPTextModel(lowerCAmelCase     )
							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    A__		(			self    ,	lowerCAmelCase    ,	lowerCAmelCase=0     ):
							if str(lowerCAmelCase     ).startswith("mps"     ):
									UpperCAmelCase_						=				torch.manual_seed(lowerCAmelCase     )
							else:
									UpperCAmelCase_						=				torch.Generator(device=lowerCAmelCase     ).manual_seed(lowerCAmelCase     )
							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    A__		(			self     ):
							UpperCAmelCase_						=				"cpu"  # ensure determinism for the device-dependent torch.Generator
							UpperCAmelCase_						=				self.get_dummy_components()
							UpperCAmelCase_						=				TextToVideoSDPipeline(**lowerCAmelCase     )
							UpperCAmelCase_						=				sd_pipe.to(lowerCAmelCase     )
							sd_pipe.set_progress_bar_config(disable=lowerCAmelCase     )

							UpperCAmelCase_						=				self.get_dummy_inputs(lowerCAmelCase     )
							UpperCAmelCase_						=				"np"
							UpperCAmelCase_						=				sd_pipe(**lowerCAmelCase     ).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    A__		(			self     ):
							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    A__		(			self     ):
							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    A__		(			self     ):
							pass



					@unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."     )
					def    A__		(			self     ):
							pass



					@unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."     )
					def    A__		(			self     ):
							pass



					def    A__		(			self     ):
							return super().test_progress_bar()





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







					'''simple docstring'''







					def    A__		(			self     ):
							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(lowerCAmelCase    ,	generator=lowerCAmelCase    ,	num_inference_steps=25    ,	output_type="pt"     ).frames
							UpperCAmelCase_						=				video_frames.cpu().numpy()

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



					def    A__		(			self     ):
							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(lowerCAmelCase    ,	generator=lowerCAmelCase    ,	num_inference_steps=2    ,	output_type="pt"     ).frames
							UpperCAmelCase_						=				video_frames.cpu().numpy()

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

