

import gc
import random
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionInstructPixaPixPipeline,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
from diffusers.utils import floats_tensor, load_image, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()






class      lowerCamelCase_      (	_lowercase						,      _lowercase						,      _lowercase						,      unittest.TestCase	):
	_lowercase		:   List[Any]     					=				StableDiffusionInstructPixaPixPipeline
	_lowercase		:   Tuple     					=				TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {'''height''', '''width''', '''cross_attention_kwargs'''}
	_lowercase		:   Union[str, Any]     					=				TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
	_lowercase		:   Dict     					=				IMAGE_TO_IMAGE_IMAGE_PARAMS
	_lowercase		:   Dict     					=				IMAGE_TO_IMAGE_IMAGE_PARAMS




	def        lowerCAmelCase_       (     self						:     Any							):
		torch.manual_seed(0							)
		__A :			Tuple		    =			UNetaDConditionModel(
		    block_out_channels=(32, 64)					,      layers_per_block=2					,      sample_size=32					,      in_channels=8					,      out_channels=4					,      down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")					,      up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")					,      cross_attention_dim=32					,      )
		__A :			List[Any]		    =			PNDMScheduler(skip_prk_steps=__A							)
		torch.manual_seed(0							)
		__A :			Optional[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					,      )
		torch.manual_seed(0							)
		__A :			List[str]		    =			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					,      )
		__A :			Union[str, Any]		    =			CLIPTextModel(__A							)
		__A :			List[str]		    =			CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""							)

		__A :			Optional[Any]		    =			{
		    """unet""": unet,
		    """scheduler""": scheduler,
		    """vae""": vae,
		    """text_encoder""": text_encoder,
		    """tokenizer""": tokenizer,
		    """safety_checker""": None,
		    """feature_extractor""": None,
		}
		return components




	def        lowerCAmelCase_       (     self						:     str					,      __A						:     Union[str, Any]					,      __A						:     Union[str, Any]=0							):
		__A :			int		    =			floats_tensor((1, 3, 32, 32)					,      rng=random.Random(__A							)							).to(__A							)
		__A :			Dict		    =			image.cpu().permute(0					,      2					,      3					,      1							)[0]
		__A :			Dict		    =			Image.fromarray(np.uinta(__A							)							).convert("""RGB"""							)
		if str(__A							).startswith("""mps"""							):
			__A :			int		    =			torch.manual_seed(__A							)
		else:
			__A :			str		    =			torch.Generator(device=__A							).manual_seed(__A							)
		__A :			Union[str, Any]		    =			{
		    """prompt""": """A painting of a squirrel eating a burger""",
		    """image""": image,
		    """generator""": generator,
		    """num_inference_steps""": 2,
		    """guidance_scale""": 6.0,
		    """image_guidance_scale""": 1,
		    """output_type""": """numpy""",
		}
		return inputs




	def        lowerCAmelCase_       (     self						:     List[str]							):
		__A :			int		    =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
		__A :			str		    =			self.get_dummy_components()
		__A :			List[str]		    =			StableDiffusionInstructPixaPixPipeline(**__A							)
		__A :			List[str]		    =			sd_pipe.to(__A							)
		sd_pipe.set_progress_bar_config(disable=__A							)

		__A :			Tuple		    =			self.get_dummy_inputs(__A							)
		__A :			Union[str, Any]		    =			sd_pipe(**__A							).images
		__A :			List[Any]		    =			image[0, -3:, -3:, -1]
		assert image.shape == (1, 32, 32, 3)
		__A :			List[str]		    =			np.array([0.7_5_2_6, 0.3_7_5_0, 0.4_5_4_7, 0.6_1_1_7, 0.5_8_6_6, 0.5_0_1_6, 0.4_3_2_7, 0.5_6_4_2, 0.4_8_1_5]							)

		assert np.abs(image_slice.flatten() - expected_slice							).max() < 1e-3




	def        lowerCAmelCase_       (     self						:     Optional[Any]							):
		__A :			int		    =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
		__A :			Optional[int]		    =			self.get_dummy_components()
		__A :			Any		    =			StableDiffusionInstructPixaPixPipeline(**__A							)
		__A :			str		    =			sd_pipe.to(__A							)
		sd_pipe.set_progress_bar_config(disable=__A							)

		__A :			Dict		    =			self.get_dummy_inputs(__A							)
		__A :			List[Any]		    =			"""french fries"""
		__A :			List[str]		    =			sd_pipe(**__A					,      negative_prompt=__A							)
		__A :			Union[str, Any]		    =			output.images
		__A :			int		    =			image[0, -3:, -3:, -1]

		assert image.shape == (1, 32, 32, 3)
		__A :			Optional[Any]		    =			np.array([0.7_5_1_1, 0.3_6_4_2, 0.4_5_5_3, 0.6_2_3_6, 0.5_7_9_7, 0.5_0_1_3, 0.4_3_4_3, 0.5_6_1_1, 0.4_8_3_1]							)

		assert np.abs(image_slice.flatten() - expected_slice							).max() < 1e-3




	def        lowerCAmelCase_       (     self						:     Any							):
		__A :			int		    =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
		__A :			Dict		    =			self.get_dummy_components()
		__A :			List[str]		    =			StableDiffusionInstructPixaPixPipeline(**__A							)
		__A :			Tuple		    =			sd_pipe.to(__A							)
		sd_pipe.set_progress_bar_config(disable=__A							)

		__A :			Optional[int]		    =			self.get_dummy_inputs(__A							)
		__A :			int		    =			[inputs["""prompt"""]] * 2

		__A :			str		    =			np.array(inputs["""image"""]							).astype(np.floataa							) / 2_5_5.0
		__A :			Optional[Any]		    =			torch.from_numpy(__A							).unsqueeze(0							).to(__A							)
		__A :			Dict		    =			image / 2 + 0.5
		__A :			List[Any]		    =			image.permute(0					,      3					,      1					,      2							)
		__A :			Optional[int]		    =			image.repeat(2					,      1					,      1					,      1							)

		__A :			List[str]		    =			sd_pipe(**__A							).images
		__A :			List[str]		    =			image[-1, -3:, -3:, -1]

		assert image.shape == (2, 32, 32, 3)
		__A :			Dict		    =			np.array([0.5_8_1_2, 0.5_7_4_8, 0.5_2_2_2, 0.5_9_0_8, 0.5_6_9_5, 0.7_1_7_4, 0.6_8_0_4, 0.5_5_2_3, 0.5_5_7_9]							)

		assert np.abs(image_slice.flatten() - expected_slice							).max() < 1e-3




	def        lowerCAmelCase_       (     self						:     Tuple							):
		__A :			Union[str, Any]		    =			"""cpu"""  # ensure determinism for the device-dependent torch.Generator
		__A :			Dict		    =			self.get_dummy_components()
		__A :			str		    =			EulerAncestralDiscreteScheduler(
		    beta_start=0.0_0_0_8_5					,      beta_end=0.0_1_2					,      beta_schedule="""scaled_linear"""							)
		__A :			List[str]		    =			StableDiffusionInstructPixaPixPipeline(**__A							)
		__A :			Optional[int]		    =			sd_pipe.to(__A							)
		sd_pipe.set_progress_bar_config(disable=__A							)

		__A :			Optional[Any]		    =			self.get_dummy_inputs(__A							)
		__A :			str		    =			sd_pipe(**__A							).images
		__A :			List[str]		    =			image[0, -3:, -3:, -1]

		__A :			List[Any]		    =			[round(__A					,      4							) for x in image_slice.flatten().tolist()]
		print(""",""".join([str(__A							) for x in slice]							)							)

		assert image.shape == (1, 32, 32, 3)
		__A :			Optional[int]		    =			np.array([0.7_4_1_7, 0.3_8_4_2, 0.4_7_3_2, 0.5_7_7_6, 0.5_8_9_1, 0.5_1_3_9, 0.4_0_5_2, 0.5_6_7_3, 0.4_9_8_6]							)

		assert np.abs(image_slice.flatten() - expected_slice							).max() < 1e-3




	def        lowerCAmelCase_       (     self						:     int							):
		super().test_inference_batch_single_identical(expected_max_diff=3e-3							)




	def        lowerCAmelCase_       (     self						:     Optional[int]							):
		__A :			Any		    =			self.get_dummy_components()
		__A :			Any		    =			StableDiffusionInstructPixaPixPipeline(**__A							)
		__A :			Any		    =			VaeImageProcessor(do_resize=__A					,      do_normalize=__A							)
		__A :			Tuple		    =			pipe.to(__A							)
		pipe.set_progress_bar_config(disable=__A							)

		__A :			int		    =			pipe(**self.get_dummy_inputs_by_type(__A					,      input_image_type="""pt"""							)							)[0]

		__A :			List[Any]		    =			components["""vae"""]
		__A :			Union[str, Any]		    =			self.get_dummy_inputs_by_type(__A					,      input_image_type="""pt"""							)

		for image_param in self.image_latents_params:
			if image_param in inputs.keys():
				__A :			Optional[Any]		    =			vae.encode(inputs[image_param]							).latent_dist.mode()

		__A :			Tuple		    =			pipe(**__A							)[0]

		__A :			Tuple		    =			np.abs(out - out_latents_inputs							).max()
		self.assertLess(__A					,      1e-4					,      """passing latents as image input generate different result from passing image"""							)






@slow
@require_torch_gpu
class      lowerCamelCase_      (	unittest.TestCase	):




	def        lowerCAmelCase_       (     self						:     Dict							):
		super().tearDown()
		gc.collect()
		torch.cuda.empty_cache()




	def        lowerCAmelCase_       (     self						:     List[Any]					,      __A						:     Union[str, Any]=0							):
		__A :			Tuple		    =			torch.manual_seed(__A							)
		__A :			Optional[int]		    =			load_image(
		    """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_pix2pix/example.jpg"""							)
		__A :			Dict		    =			{
		    """prompt""": """turn him into a cyborg""",
		    """image""": image,
		    """generator""": generator,
		    """num_inference_steps""": 3,
		    """guidance_scale""": 7.5,
		    """image_guidance_scale""": 1.0,
		    """output_type""": """numpy""",
		}
		return inputs




	def        lowerCAmelCase_       (     self						:     Any							):
		__A :			List[str]		    =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
		    """timbrooks/instruct-pix2pix"""					,      safety_checker=__A							)
		pipe.to(__A							)
		pipe.set_progress_bar_config(disable=__A							)
		pipe.enable_attention_slicing()

		__A :			Dict		    =			self.get_inputs()
		__A :			List[Any]		    =			pipe(**__A							).images
		__A :			List[Any]		    =			image[0, -3:, -3:, -1].flatten()

		assert image.shape == (1, 512, 512, 3)
		__A :			str		    =			np.array([0.5_9_0_2, 0.6_0_1_5, 0.6_0_2_7, 0.5_9_8_3, 0.6_0_9_2, 0.6_0_6_1, 0.5_7_6_5, 0.5_7_8_5, 0.5_5_5_5]							)

		assert np.abs(expected_slice - image_slice							).max() < 1e-3




	def        lowerCAmelCase_       (     self						:     int							):
		__A :			Tuple		    =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
		    """timbrooks/instruct-pix2pix"""					,      safety_checker=__A							)
		__A :			Dict		    =			LMSDiscreteScheduler.from_config(pipe.scheduler.config							)
		pipe.to(__A							)
		pipe.set_progress_bar_config(disable=__A							)
		pipe.enable_attention_slicing()

		__A :			Union[str, Any]		    =			self.get_inputs()
		__A :			str		    =			pipe(**__A							).images
		__A :			Optional[Any]		    =			image[0, -3:, -3:, -1].flatten()

		assert image.shape == (1, 512, 512, 3)
		__A :			Dict		    =			np.array([0.6_5_7_8, 0.6_8_1_7, 0.6_9_7_2, 0.6_7_6_1, 0.6_8_5_6, 0.6_9_1_6, 0.6_4_2_8, 0.6_5_1_6, 0.6_3_0_1]							)

		assert np.abs(expected_slice - image_slice							).max() < 1e-3




	def        lowerCAmelCase_       (     self						:     List[str]							):
		__A :			Optional[int]		    =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
		    """timbrooks/instruct-pix2pix"""					,      safety_checker=__A							)
		__A :			List[Any]		    =			DDIMScheduler.from_config(pipe.scheduler.config							)
		pipe.to(__A							)
		pipe.set_progress_bar_config(disable=__A							)
		pipe.enable_attention_slicing()

		__A :			Tuple		    =			self.get_inputs()
		__A :			List[Any]		    =			pipe(**__A							).images
		__A :			Optional[int]		    =			image[0, -3:, -3:, -1].flatten()

		assert image.shape == (1, 512, 512, 3)
		__A :			Optional[Any]		    =			np.array([0.3_8_2_8, 0.3_8_3_4, 0.3_8_1_8, 0.3_7_9_2, 0.3_8_6_5, 0.3_7_5_2, 0.3_7_9_2, 0.3_8_4_7, 0.3_7_5_3]							)

		assert np.abs(expected_slice - image_slice							).max() < 1e-3




	def        lowerCAmelCase_       (     self						:     int							):
		__A :			List[Any]		    =			0

		def callback_fn(__A						:     int					,      __A						:     int					,      __A						:     torch.FloatTensor							) -> None:
			__A :			str		    =			True
			nonlocal number_of_steps
			number_of_steps += 1
			if step == 1:
				__A :			Dict		    =			latents.detach().cpu().numpy()
				assert latents.shape == (1, 4, 64, 64)
				__A :			List[str]		    =			latents[0, -3:, -3:, -1]
				__A :			List[Any]		    =			np.array([-0.2_4_6_3, -0.4_6_4_4, -0.9_7_5_6, 1.5_1_7_6, 1.4_4_1_4, 0.7_8_6_6, 0.9_8_9_7, 0.8_5_2_1, 0.7_9_8_3]							)

				assert np.abs(latents_slice.flatten() - expected_slice							).max() < 5e-2
			elif step == 2:
				__A :			List[Any]		    =			latents.detach().cpu().numpy()
				assert latents.shape == (1, 4, 64, 64)
				__A :			Union[str, Any]		    =			latents[0, -3:, -3:, -1]
				__A :			List[Any]		    =			np.array([-0.2_6_4_4, -0.4_6_2_6, -0.9_6_5_3, 1.5_1_7_6, 1.4_5_5_1, 0.7_6_8_6, 0.9_8_0_5, 0.8_4_5_2, 0.8_1_1_5]							)

				assert np.abs(latents_slice.flatten() - expected_slice							).max() < 5e-2

		__A :			Optional[int]		    =			False

		__A :			Dict		    =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
		    """timbrooks/instruct-pix2pix"""					,      safety_checker=__A					,      torch_dtype=torch.floataa							)
		__A :			Optional[Any]		    =			pipe.to(__A							)
		pipe.set_progress_bar_config(disable=__A							)
		pipe.enable_attention_slicing()

		__A :			Dict		    =			self.get_inputs()
		pipe(**__A					,      callback=__A					,      callback_steps=1							)
		assert callback_fn.has_been_called
		assert number_of_steps == 3




	def        lowerCAmelCase_       (     self						:     str							):
		torch.cuda.empty_cache()
		torch.cuda.reset_max_memory_allocated()
		torch.cuda.reset_peak_memory_stats()

		__A :			Optional[Any]		    =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
		    """timbrooks/instruct-pix2pix"""					,      safety_checker=__A					,      torch_dtype=torch.floataa							)
		__A :			Optional[int]		    =			pipe.to(__A							)
		pipe.set_progress_bar_config(disable=__A							)
		pipe.enable_attention_slicing(1							)
		pipe.enable_sequential_cpu_offload()

		__A :			List[Any]		    =			self.get_inputs()
		__A :			str		    =			pipe(**__A							)

		__A :			Union[str, Any]		    =			torch.cuda.max_memory_allocated()
		# make sure that less than 2.2 GB is allocated
		assert mem_bytes < 2.2 * 10**9




	def        lowerCAmelCase_       (     self						:     Optional[int]							):
		__A :			Optional[Any]		    =			self.get_inputs()
		# resize to resolution that is divisible by 8 but not 16 or 32
		__A :			int		    =			inputs["""image"""].resize((504, 504)							)

		__A :			Dict		    =			"""timbrooks/instruct-pix2pix"""
		__A :			str		    =			StableDiffusionInstructPixaPixPipeline.from_pretrained(
		    __A					,      safety_checker=__A					,      )
		pipe.to(__A							)
		pipe.set_progress_bar_config(disable=__A							)
		pipe.enable_attention_slicing()

		__A :			List[str]		    =			pipe(**__A							)
		__A :			Union[str, Any]		    =			output.images[0]

		__A :			Dict		    =			image[255:258, 383:386, -1]

		assert image.shape == (504, 504, 3)
		__A :			Tuple		    =			np.array([0.2_7_2_6, 0.2_5_2_9, 0.2_6_6_4, 0.2_6_5_5, 0.2_6_4_1, 0.2_6_4_2, 0.2_5_9_1, 0.2_6_4_9, 0.2_5_9_0]							)

		assert np.abs(image_slice.flatten() - expected_slice							).max() < 5e-3








import math






class      lowerCamelCase_      :




	def __init__(     self						:     Union[str, Any]					,      __A						:     List[str]=0							):  # a graph with Node 0,1,...,N-1
		__A :			List[str]		    =			n
		__A :			List[str]		    =			[
		    [math.inf for j in range(0					,      __A							)] for i in range(0					,      __A							)
		]  # adjacency matrix for weight
		__A :			str		    =			[
		    [math.inf for j in range(0					,      __A							)] for i in range(0					,      __A							)
		]  # dp[i][j] stores minimum distance from i to j




	def        lowerCAmelCase_       (     self						:     str					,      __A						:     Union[str, Any]					,      __A						:     Any					,      __A						:     Optional[int]							):
		__A :			List[Any]		    =			w




	def        lowerCAmelCase_       (     self						:     Union[str, Any]							):
		for k in range(0					,      self.n							):
			for i in range(0					,      self.n							):
				for j in range(0					,      self.n							):
					__A :			List[Any]		    =			min(self.dp[i][j]					,      self.dp[i][k] + self.dp[k][j]							)




	def        lowerCAmelCase_       (     self						:     int					,      __A						:     List[str]					,      __A						:     List[str]							):
		return self.dp[u][v]


if __name__ == "__main__":
			UpperCAmelCase_							:					Tuple		 =   Graph(5)
			graph.add_edge(0, 2, 9)
			graph.add_edge(0, 4, 10)
			graph.add_edge(1, 3, 5)
			graph.add_edge(2, 3, 7)
			graph.add_edge(3, 0, 10)
			graph.add_edge(3, 1, 2)
			graph.add_edge(3, 2, 1)
			graph.add_edge(3, 4, 6)
			graph.add_edge(4, 1, 3)
			graph.add_edge(4, 2, 4)
			graph.add_edge(4, 3, 9)
			graph.floyd_warshall()
			graph.show_min(1, 4)
			graph.show_min(0, 3)





