





"""simple docstring"""


class snake_case      :







		def __init__(					self      :       Optional[int]    ,   a__      :       int    ,   a__      :       str=None    ,   a__      :       Dict=None		)       ->						int:





						'''simple docstring'''







						_A					  =			data
						_A					  =			previous
						_A					  =			next_node







		def __str__(					self      :       Any		)       ->						str:





						'''simple docstring'''







						return F"""{self.data}"""







		def  a_ (					self      :       Optional[Any]		)       ->						int:





						'''simple docstring'''







						return self.data







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





						'''simple docstring'''







						return self.next







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





						'''simple docstring'''







						return self.previous





class snake_case      :







		def __init__(					self      :       Tuple    ,   a__      :       Any		)       ->						int:





						'''simple docstring'''







						_A					  =			head







		def __iter__(					self      :       Optional[int]		)       ->						List[str]:





						'''simple docstring'''







						return self







		def  a_ (					self      :       Optional[Any]		)       ->						Tuple:





						'''simple docstring'''







						if not self.current:
										raise StopIteration
						else:
										_A					  =			self.current.get_data()
										_A					  =			self.current.get_next()
										return value





class snake_case      :







		def __init__(					self      :       List[str]		)       ->						Optional[int]:





						'''simple docstring'''







						_A					  =			None  # First node in list
						_A					  =			None  # Last node in list







		def __str__(					self      :       Tuple		)       ->						Optional[int]:





						'''simple docstring'''







						_A					  =			self.head
						_A					  =			[]
						while current is not None:
										nodes.append(current.get_data()		)
										_A					  =			current.get_next()
						return " ".join(str(a__		) for node in nodes		)







		def __contains__(					self      :       int    ,   a__      :       int		)       ->						Any:





						'''simple docstring'''







						_A					  =			self.head
						while current:
										if current.get_data() == value:
														return True
										_A					  =			current.get_next()
						return False







		def __iter__(					self      :       str		)       ->						Tuple:





						'''simple docstring'''







						return LinkedListIterator(self.head		)







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





						'''simple docstring'''







						if self.head:
										return self.head.get_data()
						return None







		def  a_ (					self      :       Optional[Any]		)       ->						int:





						'''simple docstring'''







						if self.tail:
										return self.tail.get_data()
						return None







		def  a_ (					self      :       Any    ,   a__      :       Node		)       ->						None:





						'''simple docstring'''







						if self.head is None:
										_A					  =			node
										_A					  =			node
						else:
										self.insert_before_node(self.head    ,   a__		)







		def  a_ (					self      :       Tuple    ,   a__      :       Node		)       ->						None:





						'''simple docstring'''







						if self.head is None:
										self.set_head(a__		)
						else:
										self.insert_after_node(self.tail    ,   a__		)







		def  a_ (					self      :       Union[str, Any]    ,   a__      :       int		)       ->						None:





						'''simple docstring'''







						_A					  =			Node(a__		)
						if self.head is None:
										self.set_head(a__		)
						else:
										self.set_tail(a__		)







		def  a_ (					self      :       Dict    ,   a__      :       Node    ,   a__      :       Node		)       ->						None:





						'''simple docstring'''







						_A					  =			node
						_A					  =			node.previous

						if node.get_previous() is None:
										_A					  =			node_to_insert
						else:
										_A					  =			node_to_insert

						_A					  =			node_to_insert







		def  a_ (					self      :       Optional[Any]    ,   a__      :       Node    ,   a__      :       Node		)       ->						None:





						'''simple docstring'''







						_A					  =			node
						_A					  =			node.next

						if node.get_next() is None:
										_A					  =			node_to_insert
						else:
										_A					  =			node_to_insert

						_A					  =			node_to_insert







		def  a_ (					self      :       int    ,   a__      :       int    ,   a__      :       int		)       ->						None:





						'''simple docstring'''







						_A					  =			1
						_A					  =			Node(a__		)
						_A					  =			self.head
						while node:
										if current_position == position:
														self.insert_before_node(a__    ,   a__		)
														return
										current_position += 1
										_A					  =			node.next
						self.insert_after_node(self.tail    ,   a__		)







		def  a_ (					self      :       str    ,   a__      :       int		)       ->						Node:





						'''simple docstring'''







						_A					  =			self.head
						while node:
										if node.get_data() == item:
														return node
										_A					  =			node.get_next()
						raise Exception("Node not found"		)







		def  a_ (					self      :       Optional[Any]    ,   a__      :       Union[str, Any]		)       ->						Optional[int]:





						'''simple docstring'''







						if (node := self.get_node(a__		)) is not None:
										if node == self.head:
														_A					  =			self.head.get_next()

										if node == self.tail:
														_A					  =			self.tail.get_previous()

										self.remove_node_pointers(a__		)







		@staticmethod
		def  a_ (					a__      :       Node		)       ->						None:





						'''simple docstring'''







						if node.get_next():
										_A					  =			node.previous

						if node.get_previous():
										_A					  =			node.next

						_A					  =			None
						_A					  =			None







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





						'''simple docstring'''







						return self.head is None






def      a__ (				)	->	None:
				pass


if __name__ == "__main__":
		import doctest

		doctest.testmod()



"""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, PNDMScheduler, StableDiffusionInpaintPipeline, UNetaDConditionModel
from diffusers.utils import floats_tensor, load_image, load_numpy, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, slow

from ..pipeline_params import TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS, TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class snake_case      (  _UpperCamelCase     ,				_UpperCamelCase     ,				_UpperCamelCase     ,				unittest.TestCase):
		__UpperCamelCase           =			StableDiffusionInpaintPipeline
		__UpperCamelCase           =			TEXT_GUIDED_IMAGE_INPAINTING_PARAMS
		__UpperCamelCase           =			TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS
		__UpperCamelCase           =			frozenset(
		    [])  # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
		__UpperCamelCase           =			frozenset([])







		def  a_ (					self      :       Dict		)       ->						Optional[int]:





						'''simple docstring'''







						torch.manual_seed(0		)
						_A					  =			UNetaDConditionModel(
						    block_out_channels=(32, 64)    ,   layers_per_block=2    ,   sample_size=32    ,   in_channels=9    ,   out_channels=4    ,   down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")    ,   up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")    ,   cross_attention_dim=32    ,   attention_head_dim=(2, 4)    ,   use_linear_projection=a__    ,   )
						_A					  =			PNDMScheduler(skip_prk_steps=a__		)
						torch.manual_seed(0		)
						_A					  =			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		)
						_A					  =			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=10_00    ,   hidden_act="gelu"    ,   projection_dim=5_12    ,   )
						_A					  =			CLIPTextModel(a__		)
						_A					  =			CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"		)

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







		def  a_ (					self      :       Optional[Any]    ,   a__      :       List[str]    ,   a__      :       Tuple=0		)       ->						int:





						'''simple docstring'''







						_A					  =			floats_tensor((1, 3, 32, 32)    ,   rng=random.Random(a__		)		).to(a__		)
						_A					  =			image.cpu().permute(0    ,   2    ,   3    ,   1		)[0]
						_A					  =			Image.fromarray(np.uinta(a__		)		).convert("RGB"		).resize((64, 64)		)
						_A					  =			Image.fromarray(np.uinta(image + 4		)		).convert("RGB"		).resize((64, 64)		)
						if str(a__		).startswith("mps"		):
										_A					  =			torch.manual_seed(a__		)
						else:
										_A					  =			torch.Generator(device=a__		).manual_seed(a__		)
						_A					  =			{
						    "prompt": "A painting of a squirrel eating a burger",
						    "image": init_image,
						    "mask_image": mask_image,
						    "generator": generator,
						    "num_inference_steps": 2,
						    "guidance_scale": 6.0,
						    "output_type": "numpy",
						}
						return inputs







		def  a_ (					self      :       Optional[Any]		)       ->						Optional[Any]:





						'''simple docstring'''







						_A					  =			"cpu"  # ensure determinism for the device-dependent torch.Generator
						_A					  =			self.get_dummy_components()
						_A					  =			StableDiffusionInpaintPipeline(**a__		)
						_A					  =			sd_pipe.to(a__		)
						sd_pipe.set_progress_bar_config(disable=a__		)

						_A					  =			self.get_dummy_inputs(a__		)
						_A					  =			sd_pipe(**a__		).images
						_A					  =			image[0, -3:, -3:, -1]

						assert image.shape == (1, 64, 64, 3)
						_A					  =			np.array([0.4_7_2_7, 0.5_7_3_5, 0.3_9_4_1, 0.5_4_4_6, 0.5_9_2_6, 0.4_3_9_4, 0.5_0_6_2, 0.4_6_5_4, 0.4_4_7_6]		)

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







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





						'''simple docstring'''







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





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







		def  a_ (					self      :       List[Any]		)       ->						Any:





						'''simple docstring'''







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







		def  a_ (					self      :       Union[str, Any]		)       ->						Tuple:





						'''simple docstring'''







						_A					  =			load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/sd2-inpaint/init_image.png"		)
						_A					  =			load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint/mask.png"		)
						_A					  =			load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint"
						    "/yellow_cat_sitting_on_a_park_bench.npy"		)

						_A					  =			"stabilityai/stable-diffusion-2-inpainting"
						_A					  =			StableDiffusionInpaintPipeline.from_pretrained(a__    ,   safety_checker=a__		)
						pipe.to(a__		)
						pipe.set_progress_bar_config(disable=a__		)
						pipe.enable_attention_slicing()

						_A					  =			"Face of a yellow cat, high resolution, sitting on a park bench"

						_A					  =			torch.manual_seed(0		)
						_A					  =			pipe(
						    prompt=a__    ,   image=a__    ,   mask_image=a__    ,   generator=a__    ,   output_type="np"    ,   )
						_A					  =			output.images[0]

						assert image.shape == (5_12, 5_12, 3)
						assert np.abs(expected_image - image		).max() < 9E-3







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





						'''simple docstring'''







						_A					  =			load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/sd2-inpaint/init_image.png"		)
						_A					  =			load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint/mask.png"		)
						_A					  =			load_numpy(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint"
						    "/yellow_cat_sitting_on_a_park_bench_fp16.npy"		)

						_A					  =			"stabilityai/stable-diffusion-2-inpainting"
						_A					  =			StableDiffusionInpaintPipeline.from_pretrained(
						    a__    ,   torch_dtype=torch.floataa    ,   safety_checker=a__    ,   )
						pipe.to(a__		)
						pipe.set_progress_bar_config(disable=a__		)
						pipe.enable_attention_slicing()

						_A					  =			"Face of a yellow cat, high resolution, sitting on a park bench"

						_A					  =			torch.manual_seed(0		)
						_A					  =			pipe(
						    prompt=a__    ,   image=a__    ,   mask_image=a__    ,   generator=a__    ,   output_type="np"    ,   )
						_A					  =			output.images[0]

						assert image.shape == (5_12, 5_12, 3)
						assert np.abs(expected_image - image		).max() < 5E-1







		def  a_ (					self      :       List[Any]		)       ->						Union[str, Any]:





						'''simple docstring'''







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

						_A					  =			load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
						    "/sd2-inpaint/init_image.png"		)
						_A					  =			load_image(
						    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd2-inpaint/mask.png"		)

						_A					  =			"stabilityai/stable-diffusion-2-inpainting"
						_A					  =			PNDMScheduler.from_pretrained(a__    ,   subfolder="scheduler"		)
						_A					  =			StableDiffusionInpaintPipeline.from_pretrained(
						    a__    ,   safety_checker=a__    ,   scheduler=a__    ,   torch_dtype=torch.floataa    ,   )
						pipe.to(a__		)
						pipe.set_progress_bar_config(disable=a__		)
						pipe.enable_attention_slicing(1		)
						pipe.enable_sequential_cpu_offload()

						_A					  =			"Face of a yellow cat, high resolution, sitting on a park bench"

						_A					  =			torch.manual_seed(0		)
						_A					  =			pipe(
						    prompt=a__    ,   image=a__    ,   mask_image=a__    ,   generator=a__    ,   num_inference_steps=2    ,   output_type="np"    ,   )

						_A					  =			torch.cuda.max_memory_allocated()
						# make sure that less than 2.65 GB is allocated
						assert mem_bytes < 2.6_5 * 10**9