



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):
							lowerCamelCase__   			=		TextToVideoSDPipeline
							lowerCamelCase__   			=		TEXT_TO_IMAGE_PARAMS
							lowerCamelCase__   			=		TEXT_TO_IMAGE_BATCH_PARAMS
							# No `output_type`.
							lowerCamelCase__   			=		frozenset(
							    [
							        'num_inference_steps',
							        'generator',
							        'latents',
							        'return_dict',
							        'callback',
							        'callback_steps',
							    ])







							def    snake_case__      (   self):

														'''simple docstring'''

														torch.manual_seed(0)
														_lowerCAmelCase     :							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,			)
														_lowerCAmelCase     :							Optional[int]    =    DDIMScheduler(
														    beta_start=0.00_085,			beta_end=0.012,			beta_schedule="scaled_linear",			clip_sample=__a,			set_alpha_to_one=__a,			)
														torch.manual_seed(0)
														_lowerCAmelCase     :							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)
														_lowerCAmelCase     :							List[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,			)
														_lowerCAmelCase     :							Dict    =    CLIPTextModel(__a)
														_lowerCAmelCase     :							Optional[int]    =    CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip")

														_lowerCAmelCase     :							List[Any]    =    {
														    "unet": unet,
														    "scheduler": scheduler,
														    "vae": vae,
														    "text_encoder": text_encoder,
														    "tokenizer": tokenizer,
														}
														return components







							def    snake_case__      (   self,			__a,			__a=0):

														'''simple docstring'''

														if str(__a).startswith("mps"):
																					_lowerCAmelCase     :							str    =    torch.manual_seed(__a)
														else:
																					_lowerCAmelCase     :							Optional[int]    =    torch.Generator(device=__a).manual_seed(__a)
														_lowerCAmelCase     :							Optional[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):

														'''simple docstring'''

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

														_lowerCAmelCase     :							str    =    self.get_dummy_inputs(__a)
														_lowerCAmelCase     :							Union[str, Any]    =    "np"
														_lowerCAmelCase     :							Union[str, Any]    =    sd_pipe(**__a).frames
														_lowerCAmelCase     :							Optional[int]    =    frames[0][-3:, -3:, -1]

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

														'''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    snake_case__      (   self):

														'''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    snake_case__      (   self):

														'''simple docstring'''

														pass







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

														'''simple docstring'''

														pass







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

														'''simple docstring'''

														pass







							def    snake_case__      (   self):

														'''simple docstring'''

														return super().test_progress_bar()



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







							def    snake_case__      (   self):

														'''simple docstring'''

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

														_lowerCAmelCase     :							str    =    TextToVideoSDPipeline.from_pretrained("damo-vilab/text-to-video-ms-1.7b")
														_lowerCAmelCase     :							Tuple    =    DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
														_lowerCAmelCase     :							Union[str, Any]    =    pipe.to("cuda")

														_lowerCAmelCase     :							Any    =    "Spiderman is surfing"
														_lowerCAmelCase     :							Dict    =    torch.Generator(device="cpu").manual_seed(0)

														_lowerCAmelCase     :							Optional[Any]    =    pipe(__a,			generator=__a,			num_inference_steps=25,			output_type="pt").frames
														_lowerCAmelCase     :							List[Any]    =    video_frames.cpu().numpy()

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







							def    snake_case__      (   self):

														'''simple docstring'''

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

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

														_lowerCAmelCase     :							List[Any]    =    "Spiderman is surfing"
														_lowerCAmelCase     :							int    =    torch.Generator(device="cpu").manual_seed(0)

														_lowerCAmelCase     :							int    =    pipe(__a,			generator=__a,			num_inference_steps=2,			output_type="pt").frames
														_lowerCAmelCase     :							str    =    video_frames.cpu().numpy()

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





import argparse
import copy






def 				A				(					_lowerCamelCase  ):





							'''simple docstring'''




							_lowerCAmelCase     :							int    =    {}

							with open(_lowerCamelCase  ) as f:
														for line in f:
																					if line.split()[0] not in dict_of_neighbours:
																												_lowerCAmelCase     :							Tuple    =    []
																												_list.append([line.split()[1], line.split()[2]]  )
																												_lowerCAmelCase     :							Any    =    _list
																					else:
																												dict_of_neighbours[line.split()[0]].append(
																												    [line.split()[1], line.split()[2]]  )
																					if line.split()[1] not in dict_of_neighbours:
																												_lowerCAmelCase     :							str    =    []
																												_list.append([line.split()[0], line.split()[2]]  )
																												_lowerCAmelCase     :							Any    =    _list
																					else:
																												dict_of_neighbours[line.split()[1]].append(
																												    [line.split()[0], line.split()[2]]  )

							return dict_of_neighbours






def 				A				(					_lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							with open(_lowerCamelCase  ) as f:
														_lowerCAmelCase     :							str    =    f.read(1  )
							_lowerCAmelCase     :							str    =    start_node

							_lowerCAmelCase     :							List[str]    =    []

							_lowerCAmelCase     :							Any    =    start_node

							_lowerCAmelCase     :							str    =    0
							while visiting not in first_solution:
														_lowerCAmelCase     :							Dict    =    10_000
														for k in dict_of_neighbours[visiting]:
																					if int(k[1]  ) < int(_lowerCamelCase  ) and k[0] not in first_solution:
																												_lowerCAmelCase     :							List[str]    =    k[1]
																												_lowerCAmelCase     :							List[Any]    =    k[0]

														first_solution.append(_lowerCamelCase  )
														_lowerCAmelCase     :							Optional[int]    =    distance_of_first_solution + int(_lowerCamelCase  )
														_lowerCAmelCase     :							str    =    best_node

							first_solution.append(_lowerCamelCase  )

							_lowerCAmelCase     :							Union[str, Any]    =    0
							for k in dict_of_neighbours[first_solution[-2]]:
														if k[0] == start_node:
																					break
														position += 1

							_lowerCAmelCase     :							Tuple    =    (
							    distance_of_first_solution
							    + int(dict_of_neighbours[first_solution[-2]][position][1]  )
							    - 10_000
							)
							return first_solution, distance_of_first_solution






def 				A				(					_lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							_lowerCAmelCase     :							Tuple    =    []

							for n in solution[1:-1]:
														_lowerCAmelCase     :							Dict    =    solution.index(_lowerCamelCase  )
														for kn in solution[1:-1]:
																					_lowerCAmelCase     :							Dict    =    solution.index(_lowerCamelCase  )
																					if n == kn:
																												continue

																					_lowerCAmelCase     :							Optional[int]    =    copy.deepcopy(_lowerCamelCase  )
																					_lowerCAmelCase     :							int    =    kn
																					_lowerCAmelCase     :							Dict    =    n

																					_lowerCAmelCase     :							Optional[int]    =    0

																					for k in _tmp[:-1]:
																												_lowerCAmelCase     :							str    =    _tmp[_tmp.index(_lowerCamelCase  ) + 1]
																												for i in dict_of_neighbours[k]:
																																			if i[0] == next_node:
																																										_lowerCAmelCase     :							Optional[Any]    =    distance + int(i[1]  )
																					_tmp.append(_lowerCamelCase  )

																					if _tmp not in neighborhood_of_solution:
																												neighborhood_of_solution.append(_tmp  )

							_lowerCAmelCase     :							List[Any]    =    len(neighborhood_of_solution[0]  ) - 1

							neighborhood_of_solution.sort(key=lambda _lowerCamelCase  : x[index_of_last_item_in_the_list]  )
							return neighborhood_of_solution






def 				A				(					_lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase  ):





							'''simple docstring'''




							_lowerCAmelCase     :							List[str]    =    1
							_lowerCAmelCase     :							int    =    first_solution
							_lowerCAmelCase     :							Tuple    =    []
							_lowerCAmelCase     :							Tuple    =    distance_of_first_solution
							_lowerCAmelCase     :							Optional[int]    =    solution

							while count <= iters:
														_lowerCAmelCase     :							int    =    find_neighborhood(_lowerCamelCase				, _lowerCamelCase  )
														_lowerCAmelCase     :							Tuple    =    0
														_lowerCAmelCase     :							Dict    =    neighborhood[index_of_best_solution]
														_lowerCAmelCase     :							int    =    len(_lowerCamelCase  ) - 1

														_lowerCAmelCase     :							Union[str, Any]    =    False
														while not found:
																					_lowerCAmelCase     :							Tuple    =    0
																					while i < len(_lowerCamelCase  ):
																												if best_solution[i] != solution[i]:
																																			_lowerCAmelCase     :							str    =    best_solution[i]
																																			_lowerCAmelCase     :							Tuple    =    solution[i]
																																			break
																												_lowerCAmelCase     :							int    =    i + 1

																					if [first_exchange_node, second_exchange_node] not in tabu_list and [
																					    second_exchange_node,
																					    first_exchange_node,
																					] not in tabu_list:
																												tabu_list.append([first_exchange_node, second_exchange_node]  )
																												_lowerCAmelCase     :							Optional[int]    =    True
																												_lowerCAmelCase     :							Optional[Any]    =    best_solution[:-1]
																												_lowerCAmelCase     :							Tuple    =    neighborhood[index_of_best_solution][best_cost_index]
																												if cost < best_cost:
																																			_lowerCAmelCase     :							Union[str, Any]    =    cost
																																			_lowerCAmelCase     :							List[Any]    =    solution
																					else:
																												_lowerCAmelCase     :							Optional[Any]    =    index_of_best_solution + 1
																												_lowerCAmelCase     :							Optional[Any]    =    neighborhood[index_of_best_solution]

														if len(_lowerCamelCase  ) >= size:
																					tabu_list.pop(0  )

														_lowerCAmelCase     :							int    =    count + 1

							return best_solution_ever, best_cost






def 				A				(					_lowerCamelCase=None  ):





							'''simple docstring'''




							_lowerCAmelCase     :							int    =    generate_neighbours(args.File  )

							_lowerCAmelCase   ,    _lowerCAmelCase     :							List[str]    =    generate_first_solution(
							    args.File				, _lowerCamelCase  )

							_lowerCAmelCase   ,    _lowerCAmelCase     :							Any    =    tabu_search(
							    _lowerCamelCase				, _lowerCamelCase				, _lowerCamelCase				, args.Iterations				, args.Size				, )

							print(F"Best solution: {best_sol}, with total distance: {best_cost}."  )


if __name__ == "__main__":
						_snake_case		=  argparse.ArgumentParser(description="Tabu Search")
						parser.add_argument(
						    "-f",
						    "--File",
						    type=str,
						    help="Path to the file containing the data",
						    required=True,
						)
						parser.add_argument(
						    "-i",
						    "--Iterations",
						    type=int,
						    help="How many iterations the algorithm should perform",
						    required=True,
						)
						parser.add_argument(
						    "-s", "--Size", type=int, help="Size of the tabu list", required=True
						)

						# Pass the arguments to main method
						main(parser.parse_args())
