



import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()







@skip_mps
class       __lowerCAmelCase   (      __snake_case    ,						unittest.TestCase						):
			UpperCamelCase        =					VideoToVideoSDPipeline
			UpperCamelCase        =					TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'''video'''}						) - {'image', 'width', 'height'}
			UpperCamelCase        =					TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''video'''}						) - {'image'}
			UpperCamelCase        =					PipelineTesterMixin.required_optional_params - {'latents'}
			UpperCamelCase        =					False

			# No `output_type`.
			UpperCamelCase        =					frozenset(
			    [
			        '''num_inference_steps''',
			        '''generator''',
			        '''latents''',
			        '''return_dict''',
			        '''callback''',
			        '''callback_steps''',
			    ]						)



			def    _lowerCamelCase	(						self       :  Tuple)       ->   Dict:



						"""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.0_0_0_8_5			,		beta_end=0.0_1_2			,		beta_schedule='scaled_linear'			,		clip_sample=A			,		set_alpha_to_one=A			,		)
						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=1_28			,		)
						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=10_00			,		hidden_act='gelu'			,		projection_dim=5_12			,		)
						_UpperCAmelCase			=       CLIPTextModel(A)
						_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       :  Any			,		A       :  str			,		A       :  Union[str, Any]=0)       ->   List[Any]:



						"""simple docstring"""






						_UpperCAmelCase			=       floats_tensor((1, 3, 3, 32, 32)			,		rng=random.Random(A)).to(A)

						if str(A).startswith('mps'):
									_UpperCAmelCase			=       torch.manual_seed(A)
						else:
									_UpperCAmelCase			=       torch.Generator(device=A).manual_seed(A)
						_UpperCAmelCase			=       {
						    'prompt': 'A painting of a squirrel eating a burger',
						    'video': video,
						    'generator': generator,
						    'num_inference_steps': 2,
						    'guidance_scale': 6.0,
						    'output_type': 'pt',
						}
						return inputs



			def    _lowerCamelCase	(						self       :  Any)       ->   int:



						"""simple docstring"""






						_UpperCAmelCase			=       'cpu'  # ensure determinism for the device-dependent torch.Generator
						_UpperCAmelCase			=       self.get_dummy_components()
						_UpperCAmelCase			=       VideoToVideoSDPipeline(**A)
						_UpperCAmelCase			=       sd_pipe.to(A)
						sd_pipe.set_progress_bar_config(disable=A)

						_UpperCAmelCase			=       self.get_dummy_inputs(A)
						_UpperCAmelCase			=       'np'
						_UpperCAmelCase			=       sd_pipe(**A).frames
						_UpperCAmelCase			=       frames[0][-3:, -3:, -1]

						assert frames[0].shape == (32, 32, 3)
						_UpperCAmelCase			=       np.array([1_06, 1_17, 1_13, 1_74, 1_37, 1_12, 1_48, 1_51, 1_31])

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



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



						"""simple docstring"""






						self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=A			,		expected_max_diff=5E-3)



			@unittest.skip(reason='Batching needs to be properly figured out first for this pipeline.')
			def    _lowerCamelCase	(						self       :  List[str])       ->   List[str]:



						"""simple docstring"""






						pass



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



						"""simple docstring"""






						pass



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



						"""simple docstring"""






						pass



			def    _lowerCamelCase	(						self       :  Optional[Any])       ->   List[Any]:



						"""simple docstring"""






						return super().test_progress_bar()







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



			def    _lowerCamelCase	(						self       :  List[str])       ->   str:



						"""simple docstring"""






						_UpperCAmelCase			=       VideoToVideoSDPipeline.from_pretrained('cerspense/zeroscope_v2_XL'			,		torch_dtype=torch.floataa)
						pipe.enable_model_cpu_offload()

						# 10 frames
						_UpperCAmelCase			=       torch.Generator(device='cpu').manual_seed(0)
						_UpperCAmelCase			=       torch.randn((1, 10, 3, 10_24, 5_76)			,		generator=A)
						_UpperCAmelCase			=       video.to('cuda')

						_UpperCAmelCase			=       'Spiderman is surfing'

						_UpperCAmelCase			=       pipe(A			,		video=A			,		generator=A			,		num_inference_steps=3			,		output_type='pt').frames

						_UpperCAmelCase			=       np.array([-1.0_4_5_8_9_8_4, -1.1_2_7_9_2_9_7, -0.9_6_6_3_0_8_6, -0.9_1_5_0_3_9_0_6, -0.7_5_0_9_7_6_5_6])
						assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array).sum() < 1E-2





UpperCAmelCase__       =	{
    "A": ["B", "C", "E"],
    "B": ["A", "D", "E"],
    "C": ["A", "F", "G"],
    "D": ["B"],
    "E": ["A", "B", "D"],
    "F": ["C"],
    "G": ["C"],
}





def 							A     (					_UpperCAmelCase    :       dict							,       _UpperCAmelCase    :       Optional[int]							,       _UpperCAmelCase    :       List[Any]   )    ->			list[str]:







			'''simple docstring'''
			_UpperCAmelCase			=       set()
			# keep track of all the paths to be checked
			_UpperCAmelCase			=       [[start]]

			# return path if start is goal
			if start == goal:
						return [start]

			# keeps looping until all possible paths have been checked
			while queue:
						# pop the first path from the queue
						_UpperCAmelCase			=       queue.pop(0   )
						# get the last node from the path
						_UpperCAmelCase			=       path[-1]
						if node not in explored:
									_UpperCAmelCase			=       graph[node]
									# go through all neighbour nodes, construct a new path and
									# push it into the queue
									for neighbour in neighbours:
												_UpperCAmelCase			=       list(_UpperCAmelCase   )
												new_path.append(_UpperCAmelCase   )
												queue.append(_UpperCAmelCase   )
												# return path if neighbour is goal
												if neighbour == goal:
															return new_path

            # mark node as explored
									explored.add(_UpperCAmelCase   )

    # in case there's no path between the 2 nodes
			return []





def 							A     (					_UpperCAmelCase    :       dict							,       _UpperCAmelCase    :       Any							,       _UpperCAmelCase    :       List[str]   )    ->			int:







			'''simple docstring'''
			if not graph or start not in graph or target not in graph:
						return -1
			if start == target:
						return 0
			_UpperCAmelCase			=       [start]
			_UpperCAmelCase			=       set(_UpperCAmelCase   )
			# Keep tab on distances from `start` node.
			_UpperCAmelCase			=       {start: 0, target: -1}
			while queue:
						_UpperCAmelCase			=       queue.pop(0   )
						if node == target:
									_UpperCAmelCase			=       (
									    dist[node] if dist[target] == -1 else min(dist[target]							,       dist[node]   )
									)
						for adjacent in graph[node]:
									if adjacent not in visited:
												visited.add(_UpperCAmelCase   )
												queue.append(_UpperCAmelCase   )
												_UpperCAmelCase			=       dist[node] + 1
			return dist[target]


if __name__ == "__main__":
							print(bfs_shortest_path(demo_graph, "G", "D"))  # returns ['G', 'C', 'A', 'B', 'D']
							print(bfs_shortest_path_distance(demo_graph, "G", "D"))  # returns 4
