




"""simple docstring"""


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







 '''simple docstring'''





 lowercase__			 = StableDiffusionInstructPixaPixPipeline
 lowercase__			 = TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"height", "width", "cross_attention_kwargs"}
 lowercase__			 = TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
 lowercase__			 = IMAGE_TO_IMAGE_IMAGE_PARAMS
 lowercase__			 = IMAGE_TO_IMAGE_IMAGE_PARAMS


 def 			UpperCamelCase_       (    self:		Any		):




   '''simple docstring'''



   torch.manual_seed(0		)
   _snake_case							:		List[Any]								=  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,				)
   _snake_case							:		Optional[Any]								=  PNDMScheduler(skip_prk_steps=a_		)
   torch.manual_seed(0		)
   _snake_case							:		Tuple								=  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		)
   _snake_case							:		Optional[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=1_000,				)
   _snake_case							:		List[Any]								=  CLIPTextModel(a_		)
   _snake_case							:		Union[str, Any]								=  CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""		)

   _snake_case							:		Tuple								=  {
       """unet""": unet,
       """scheduler""": scheduler,
       """vae""": vae,
       """text_encoder""": text_encoder,
       """tokenizer""": tokenizer,
       """safety_checker""": None,
       """feature_extractor""": None,
   }
   return components


 def 			UpperCamelCase_       (    self:		List[str],				a_:		Dict,				a_:		Optional[int]=0		):




   '''simple docstring'''



   _snake_case							:		Any								=  floats_tensor((1, 3, 32, 32),				rng=random.Random(a_		)		).to(a_		)
   _snake_case							:		List[Any]								=  image.cpu().permute(0,				2,				3,				1		)[0]
   _snake_case							:		Optional[Any]								=  Image.fromarray(np.uinta(a_		)		).convert("""RGB"""		)
   if str(a_		).startswith("""mps"""		):
     _snake_case							:		Tuple								=  torch.manual_seed(a_		)
   else:
     _snake_case							:		List[Any]								=  torch.Generator(device=a_		).manual_seed(a_		)
   _snake_case							:		List[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 			UpperCamelCase_       (    self:		Dict		):




   '''simple docstring'''



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

   _snake_case							:		Optional[int]								=  self.get_dummy_inputs(a_		)
   _snake_case							:		List[Any]								=  sd_pipe(**a_		).images
   _snake_case							:		Union[str, Any]								=  image[0, -3:, -3:, -1]
   assert image.shape == (1, 32, 32, 3)
   _snake_case							:		Optional[int]								=  np.array([0.7_526, 0.3_750, 0.4_547, 0.6_117, 0.5_866, 0.5_016, 0.4_327, 0.5_642, 0.4_815]		)

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


 def 			UpperCamelCase_       (    self:		Optional[Any]		):




   '''simple docstring'''



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

   _snake_case							:		Optional[Any]								=  self.get_dummy_inputs(a_		)
   _snake_case							:		Any								=  """french fries"""
   _snake_case							:		Tuple								=  sd_pipe(**a_,				negative_prompt=a_		)
   _snake_case							:		Union[str, Any]								=  output.images
   _snake_case							:		int								=  image[0, -3:, -3:, -1]

   assert image.shape == (1, 32, 32, 3)
   _snake_case							:		int								=  np.array([0.7_511, 0.3_642, 0.4_553, 0.6_236, 0.5_797, 0.5_013, 0.4_343, 0.5_611, 0.4_831]		)

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


 def 			UpperCamelCase_       (    self:		List[Any]		):




   '''simple docstring'''



   _snake_case							:		Dict								=  """cpu"""  # ensure determinism for the device-dependent torch.Generator
   _snake_case							:		Optional[Any]								=  self.get_dummy_components()
   _snake_case							:		Dict								=  StableDiffusionInstructPixaPixPipeline(**a_		)
   _snake_case							:		Dict								=  sd_pipe.to(a_		)
   sd_pipe.set_progress_bar_config(disable=a_		)

   _snake_case							:		Optional[int]								=  self.get_dummy_inputs(a_		)
   _snake_case							:		List[Any]								=  [inputs["""prompt"""]] * 2

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

   _snake_case							:		Tuple								=  sd_pipe(**a_		).images
   _snake_case							:		Dict								=  image[-1, -3:, -3:, -1]

   assert image.shape == (2, 32, 32, 3)
   _snake_case							:		str								=  np.array([0.5_812, 0.5_748, 0.5_222, 0.5_908, 0.5_695, 0.7_174, 0.6_804, 0.5_523, 0.5_579]		)

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


 def 			UpperCamelCase_       (    self:		List[str]		):




   '''simple docstring'''



   _snake_case							:		int								=  """cpu"""  # ensure determinism for the device-dependent torch.Generator
   _snake_case							:		Dict								=  self.get_dummy_components()
   _snake_case							:		int								=  EulerAncestralDiscreteScheduler(
       beta_start=0.00_085,				beta_end=0.012,				beta_schedule="""scaled_linear"""		)
   _snake_case							:		str								=  StableDiffusionInstructPixaPixPipeline(**a_		)
   _snake_case							:		Any								=  sd_pipe.to(a_		)
   sd_pipe.set_progress_bar_config(disable=a_		)

   _snake_case							:		List[Any]								=  self.get_dummy_inputs(a_		)
   _snake_case							:		List[Any]								=  sd_pipe(**a_		).images
   _snake_case							:		Tuple								=  image[0, -3:, -3:, -1]

   _snake_case							:		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)
   _snake_case							:		Tuple								=  np.array([0.7_417, 0.3_842, 0.4_732, 0.5_776, 0.5_891, 0.5_139, 0.4_052, 0.5_673, 0.4_986]		)

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


 def 			UpperCamelCase_       (    self:		str		):




   '''simple docstring'''



   super().test_inference_batch_single_identical(expected_max_diff=3E-3		)


 def 			UpperCamelCase_       (    self:		Optional[int]		):




   '''simple docstring'''



   _snake_case							:		int								=  self.get_dummy_components()
   _snake_case							:		Any								=  StableDiffusionInstructPixaPixPipeline(**a_		)
   _snake_case							:		Union[str, Any]								=  VaeImageProcessor(do_resize=a_,				do_normalize=a_		)
   _snake_case							:		Dict								=  pipe.to(a_		)
   pipe.set_progress_bar_config(disable=a_		)

   _snake_case							:		List[str]								=  pipe(**self.get_dummy_inputs_by_type(a_,				input_image_type="""pt"""		)		)[0]

   _snake_case							:		Tuple								=  components["""vae"""]
   _snake_case							:		List[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():
       _snake_case							:		Optional[int]								=  vae.encode(inputs[image_param]		).latent_dist.mode()

   _snake_case							:		Union[str, Any]								=  pipe(**a_		)[0]

   _snake_case							:		List[str]								=  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  lowercase(					unittest.TestCase			):







 '''simple docstring'''





 def 			UpperCamelCase_       (    self:		Optional[int]		):




   '''simple docstring'''



   super().tearDown()
   gc.collect()
   torch.cuda.empty_cache()


 def 			UpperCamelCase_       (    self:		Dict,				a_:		str=0		):




   '''simple docstring'''



   _snake_case							:		Any								=  torch.manual_seed(a_		)
   _snake_case							:		str								=  load_image(
       """https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/stable_diffusion_pix2pix/example.jpg"""		)
   _snake_case							:		Any								=  {
       """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 			UpperCamelCase_       (    self:		int		):




   '''simple docstring'''



   _snake_case							:		Tuple								=  StableDiffusionInstructPixaPixPipeline.from_pretrained(
       """timbrooks/instruct-pix2pix""",				safety_checker=a_		)
   pipe.to(a_		)
   pipe.set_progress_bar_config(disable=a_		)
   pipe.enable_attention_slicing()

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

   assert image.shape == (1, 512, 512, 3)
   _snake_case							:		int								=  np.array([0.5_902, 0.6_015, 0.6_027, 0.5_983, 0.6_092, 0.6_061, 0.5_765, 0.5_785, 0.5_555]		)

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


 def 			UpperCamelCase_       (    self:		Any		):




   '''simple docstring'''



   _snake_case							:		Tuple								=  StableDiffusionInstructPixaPixPipeline.from_pretrained(
       """timbrooks/instruct-pix2pix""",				safety_checker=a_		)
   _snake_case							:		int								=  LMSDiscreteScheduler.from_config(pipe.scheduler.config		)
   pipe.to(a_		)
   pipe.set_progress_bar_config(disable=a_		)
   pipe.enable_attention_slicing()

   _snake_case							:		str								=  self.get_inputs()
   _snake_case							:		Optional[int]								=  pipe(**a_		).images
   _snake_case							:		Optional[int]								=  image[0, -3:, -3:, -1].flatten()

   assert image.shape == (1, 512, 512, 3)
   _snake_case							:		Optional[Any]								=  np.array([0.6_578, 0.6_817, 0.6_972, 0.6_761, 0.6_856, 0.6_916, 0.6_428, 0.6_516, 0.6_301]		)

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


 def 			UpperCamelCase_       (    self:		int		):




   '''simple docstring'''



   _snake_case							:		Optional[int]								=  StableDiffusionInstructPixaPixPipeline.from_pretrained(
       """timbrooks/instruct-pix2pix""",				safety_checker=a_		)
   _snake_case							:		Tuple								=  DDIMScheduler.from_config(pipe.scheduler.config		)
   pipe.to(a_		)
   pipe.set_progress_bar_config(disable=a_		)
   pipe.enable_attention_slicing()

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

   assert image.shape == (1, 512, 512, 3)
   _snake_case							:		int								=  np.array([0.3_828, 0.3_834, 0.3_818, 0.3_792, 0.3_865, 0.3_752, 0.3_792, 0.3_847, 0.3_753]		)

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


 def 			UpperCamelCase_       (    self:		Union[str, Any]		):




   '''simple docstring'''



   _snake_case							:		Any								=  0

   def callback_fn(a_:		int,				a_:		int,				a_:		torch.FloatTensor		) -> None:
     _snake_case							:		int								=  True
     nonlocal number_of_steps
     number_of_steps += 1
     if step == 1:
       _snake_case							:		Union[str, Any]								=  latents.detach().cpu().numpy()
       assert latents.shape == (1, 4, 64, 64)
       _snake_case							:		Tuple								=  latents[0, -3:, -3:, -1]
       _snake_case							:		Optional[Any]								=  np.array([-0.2_463, -0.4_644, -0.9_756, 1.5_176, 1.4_414, 0.7_866, 0.9_897, 0.8_521, 0.7_983]		)

       assert np.abs(latents_slice.flatten() - expected_slice		).max() < 5E-2
     elif step == 2:
       _snake_case							:		Optional[Any]								=  latents.detach().cpu().numpy()
       assert latents.shape == (1, 4, 64, 64)
       _snake_case							:		Dict								=  latents[0, -3:, -3:, -1]
       _snake_case							:		Union[str, Any]								=  np.array([-0.2_644, -0.4_626, -0.9_653, 1.5_176, 1.4_551, 0.7_686, 0.9_805, 0.8_452, 0.8_115]		)

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

   _snake_case							:		Optional[Any]								=  False

   _snake_case							:		Union[str, Any]								=  StableDiffusionInstructPixaPixPipeline.from_pretrained(
       """timbrooks/instruct-pix2pix""",				safety_checker=a_,				torch_dtype=torch.floataa		)
   _snake_case							:		Any								=  pipe.to(a_		)
   pipe.set_progress_bar_config(disable=a_		)
   pipe.enable_attention_slicing()

   _snake_case							:		List[Any]								=  self.get_inputs()
   pipe(**a_,				callback=a_,				callback_steps=1		)
   assert callback_fn.has_been_called
   assert number_of_steps == 3


 def 			UpperCamelCase_       (    self:		Tuple		):




   '''simple docstring'''



   torch.cuda.empty_cache()
   torch.cuda.reset_max_memory_allocated()
   torch.cuda.reset_peak_memory_stats()

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

   _snake_case							:		int								=  self.get_inputs()
   _snake_case							:		int								=  pipe(**a_		)

   _snake_case							:		str								=  torch.cuda.max_memory_allocated()
   # make sure that less than 2.2 GB is allocated
   assert mem_bytes < 2.2 * 10**9


 def 			UpperCamelCase_       (    self:		Dict		):




   '''simple docstring'''



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

   _snake_case							:		Tuple								=  """timbrooks/instruct-pix2pix"""
   _snake_case							:		Union[str, Any]								=  StableDiffusionInstructPixaPixPipeline.from_pretrained(
       a_,				safety_checker=a_,				)
   pipe.to(a_		)
   pipe.set_progress_bar_config(disable=a_		)
   pipe.enable_attention_slicing()

   _snake_case							:		Optional[Any]								=  pipe(**a_		)
   _snake_case							:		List[Any]								=  output.images[0]

   _snake_case							:		List[str]								=  image[255:258, 383:386, -1]

   assert image.shape == (504, 504, 3)
   _snake_case							:		int								=  np.array([0.2_726, 0.2_529, 0.2_664, 0.2_655, 0.2_641, 0.2_642, 0.2_591, 0.2_649, 0.2_590]		)

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








"""simple docstring"""


from typing import Any






class  lowercase:







 '''simple docstring'''





 def __init__(    self:		Dict,				a_:		Any		):




   '''simple docstring'''



   _snake_case							:		Dict								=  data
   _snake_case							:		Optional[Any]								=  None







class  lowercase:







 '''simple docstring'''





 def __init__(    self:		str		):




   '''simple docstring'''



   _snake_case							:		Any								=  None


 def 			UpperCamelCase_       (    self:		Any		):




   '''simple docstring'''



   _snake_case							:		List[Any]								=  self.head
   while temp is not None:
     print(temp.data,				end=""" """		)
     _snake_case							:		int								=  temp.next
   print()


 def 			UpperCamelCase_       (    self:		Union[str, Any],				a_:		Any		):




   '''simple docstring'''



   _snake_case							:		Optional[Any]								=  Node(a_		)
   _snake_case							:		Union[str, Any]								=  self.head
   _snake_case							:		List[Any]								=  new_node


 def 			UpperCamelCase_       (    self:		Tuple,				a_:		List[str],				a_:		Union[str, Any]		):




   '''simple docstring'''



   if node_data_a == node_data_a:
     return
   else:
     _snake_case							:		int								=  self.head
     while node_a is not None and node_a.data != node_data_a:
       _snake_case							:		List[Any]								=  node_a.next

     _snake_case							:		List[Any]								=  self.head
     while node_a is not None and node_a.data != node_data_a:
       _snake_case							:		List[Any]								=  node_a.next

     if node_a is None or node_a is None:
       return

     _snake_case		, _snake_case							:		int								=  node_a.data, node_a.data


if __name__ == "__main__":
     A_      =    LinkedList()
     for i in range(5, 0, -1):
          ll.push(i)

     ll.print_list()

     ll.swap_nodes(1, 4)
     print('''After swapping''')
     ll.print_list()


