






'''simple docstring'''





import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionModelEditingPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


@skip_mps
class 				snake_case__     (			lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							unittest.TestCase):
					a_       =			StableDiffusionModelEditingPipeline
					a_       =			TEXT_TO_IMAGE_PARAMS
					a_       =			TEXT_TO_IMAGE_BATCH_PARAMS
					a_       =			TEXT_TO_IMAGE_IMAGE_PARAMS
					a_       =			TEXT_TO_IMAGE_IMAGE_PARAMS




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

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




					def 						A     (       self       :						List[str]							,				_A       :						Tuple							,				_A       :						str=0     )  ->    Dict:
								UpperCAmelCase_     :   Optional[Any]       =   torch.manual_seed(_snake_case     )
								UpperCAmelCase_     :   Dict       =   {
								    'prompt': 'A field of roses',
								    'generator': generator,
								    # Setting height and width to None to prevent OOMs on CPU.
								    'height': None,
								    'width': None,
								    'num_inference_steps': 2,
								    'guidance_scale': 6.0,
								    'output_type': 'numpy',
								}
								return inputs




					def 						A     (       self       :						List[str]     )  ->    Tuple:
								UpperCAmelCase_     :   Dict       =   'cpu'  # ensure determinism for the device-dependent torch.Generator
								UpperCAmelCase_     :   Tuple       =   self.get_dummy_components()
								UpperCAmelCase_     :   List[str]       =   StableDiffusionModelEditingPipeline(**_snake_case     )
								UpperCAmelCase_     :   Union[str, Any]       =   sd_pipe.to(_snake_case     )
								sd_pipe.set_progress_bar_config(disable=_snake_case     )

								UpperCAmelCase_     :   Optional[int]       =   self.get_dummy_inputs(_snake_case     )
								UpperCAmelCase_     :   Optional[Any]       =   sd_pipe(**_snake_case     ).images
								UpperCAmelCase_     :   List[Any]       =   image[0, -3:, -3:, -1]
								assert image.shape == (1, 64, 64, 3)

								UpperCAmelCase_     :   Optional[Any]       =   np.array([0.4_755, 0.5_132, 0.4_976, 0.3_904, 0.3_554, 0.4_765, 0.5_139, 0.5_158, 0.4_889]     )

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




					def 						A     (       self       :						Optional[int]     )  ->    int:
								UpperCAmelCase_     :   Tuple       =   'cpu'  # ensure determinism for the device-dependent torch.Generator
								UpperCAmelCase_     :   List[Any]       =   self.get_dummy_components()
								UpperCAmelCase_     :   List[str]       =   StableDiffusionModelEditingPipeline(**_snake_case     )
								UpperCAmelCase_     :   Any       =   sd_pipe.to(_snake_case     )
								sd_pipe.set_progress_bar_config(disable=_snake_case     )

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

								assert image.shape == (1, 64, 64, 3)

								UpperCAmelCase_     :   List[Any]       =   np.array([0.4_992, 0.5_101, 0.5_004, 0.3_949, 0.3_604, 0.4_735, 0.5_216, 0.5_204, 0.4_913]     )

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




					def 						A     (       self       :						Optional[int]     )  ->    Optional[int]:
								UpperCAmelCase_     :   Optional[int]       =   'cpu'  # ensure determinism for the device-dependent torch.Generator
								UpperCAmelCase_     :   Union[str, Any]       =   self.get_dummy_components()
								UpperCAmelCase_     :   List[Any]       =   EulerAncestralDiscreteScheduler(
								    beta_start=0.00_085							,				beta_end=0.012							,				beta_schedule='''scaled_linear'''     )
								UpperCAmelCase_     :   List[Any]       =   StableDiffusionModelEditingPipeline(**_snake_case     )
								UpperCAmelCase_     :   List[str]       =   sd_pipe.to(_snake_case     )
								sd_pipe.set_progress_bar_config(disable=_snake_case     )

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

								assert image.shape == (1, 64, 64, 3)

								UpperCAmelCase_     :   Any       =   np.array([0.4_747, 0.5_372, 0.4_779, 0.4_982, 0.5_543, 0.4_816, 0.5_238, 0.4_904, 0.5_027]     )

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




					def 						A     (       self       :						Any     )  ->    Union[str, Any]:
								UpperCAmelCase_     :   int       =   'cpu'  # ensure determinism for the device-dependent torch.Generator
								UpperCAmelCase_     :   str       =   self.get_dummy_components()
								UpperCAmelCase_     :   Union[str, Any]       =   PNDMScheduler()
								UpperCAmelCase_     :   Union[str, Any]       =   StableDiffusionModelEditingPipeline(**_snake_case     )
								UpperCAmelCase_     :   List[str]       =   sd_pipe.to(_snake_case     )
								sd_pipe.set_progress_bar_config(disable=_snake_case     )

								UpperCAmelCase_     :   Dict       =   self.get_dummy_inputs(_snake_case     )
								# the pipeline does not expect pndm so test if it raises error.
								with self.assertRaises(_snake_case     ):
											UpperCAmelCase_     :   Tuple       =   sd_pipe(**_snake_case     ).images




					def 						A     (       self       :						List[str]     )  ->    Tuple:
								super().test_inference_batch_single_identical(expected_max_diff=5e-3     )





					def 						A     (       self       :						int     )  ->    List[Any]:
								super().test_attention_slicing_forward_pass(expected_max_diff=5e-3     )






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




					def 						A     (       self       :						int     )  ->    Optional[int]:
								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()




					def 						A     (       self       :						Tuple							,				_A       :						int=0     )  ->    Dict:
								UpperCAmelCase_     :   Dict       =   torch.manual_seed(_snake_case     )
								UpperCAmelCase_     :   str       =   {
								    'prompt': 'A field of roses',
								    'generator': generator,
								    'num_inference_steps': 3,
								    'guidance_scale': 7.5,
								    'output_type': 'numpy',
								}
								return inputs




					def 						A     (       self       :						int     )  ->    int:
								UpperCAmelCase_     :   Union[str, Any]       =   'CompVis/stable-diffusion-v1-4'
								UpperCAmelCase_     :   Tuple       =   StableDiffusionModelEditingPipeline.from_pretrained(_snake_case							,				safety_checker=_snake_case     )
								pipe.to(_snake_case     )
								pipe.set_progress_bar_config(disable=_snake_case     )
								pipe.enable_attention_slicing()

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

								assert image.shape == (1, 5_12, 5_12, 3)

								UpperCAmelCase_     :   List[Any]       =   np.array(
								    [0.6_749_496, 0.6_386_453, 0.51_443_267, 0.66_094_905, 0.61_921_215, 0.5_491_332, 0.5_744_417, 0.58_075_106, 0.5_174_658]     )

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

								# make sure image changes after editing
								pipe.edit_model('''A pack of roses'''							,				'''A pack of blue roses'''     )

								UpperCAmelCase_     :   Tuple       =   pipe(**_snake_case     ).images
								UpperCAmelCase_     :   Optional[Any]       =   image[0, -3:, -3:, -1].flatten()

								assert image.shape == (1, 5_12, 5_12, 3)

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





					def 						A     (       self       :						Dict     )  ->    Any:
								torch.cuda.empty_cache()
								torch.cuda.reset_max_memory_allocated()
								torch.cuda.reset_peak_memory_stats()

								UpperCAmelCase_     :   List[Any]       =   'CompVis/stable-diffusion-v1-4'
								UpperCAmelCase_     :   int       =   DDIMScheduler.from_pretrained(_snake_case							,				subfolder='''scheduler'''     )
								UpperCAmelCase_     :   List[Any]       =   StableDiffusionModelEditingPipeline.from_pretrained(
								    _snake_case							,				scheduler=_snake_case							,				safety_checker=_snake_case     )
								UpperCAmelCase_     :   List[Any]       =   pipe.to(_snake_case     )
								pipe.set_progress_bar_config(disable=_snake_case     )
								pipe.enable_attention_slicing(1     )
								pipe.enable_sequential_cpu_offload()

								UpperCAmelCase_     :   Dict       =   self.get_inputs()
								UpperCAmelCase_     :   Any       =   pipe(**_snake_case     )

								UpperCAmelCase_     :   Dict       =   torch.cuda.max_memory_allocated()
								# make sure that less than 4.4 GB is allocated
								assert mem_bytes < 4.4 * 10**9










'''simple docstring'''





from __future__ import annotations

from collections import deque
from enum import Enum
from math import atana, degrees
from sys import maxsize


class 				snake_case__     (			UpperCamelCase):
					a_       =			1
					a_       =			2
					a_       =			3




					def __repr__(       self       :						str     )  ->    Optional[int]:
								return F"{self.__class__.__name__}.{self.name}"






def __UpperCAmelCase (			A			:	tuple[int, int]	,   A			:	int	,   A			:	int						)  -> float:
			UpperCAmelCase_  ,     UpperCAmelCase_     :   Any       =   point
			return degrees(atana(y - miny	,   x - minx						)						)
def __UpperCAmelCase (			A			:	tuple[int, int]	,   A			:	tuple[int, int]	,   A			:	tuple[int, int]						)  -> Direction:
			UpperCAmelCase_  ,     UpperCAmelCase_     :   Tuple       =   starting
			UpperCAmelCase_  ,     UpperCAmelCase_     :   Optional[Any]       =   via
			UpperCAmelCase_  ,     UpperCAmelCase_     :   Optional[int]       =   target
			UpperCAmelCase_     :   Optional[int]       =   degrees(atana(ya - ya	,   xa - xa						)						)
			via_angle %= 3_6_0
			UpperCAmelCase_     :   Dict       =   degrees(atana(ya - ya	,   xa - xa						)						)
			target_angle %= 3_6_0
			# t-
			#  \ \
			#   \ v
			#    \|
			#     s
			# via_angle is always lower than target_angle, if direction is left.
			# If they are same, it means they are on a same line of convex hull.
			if target_angle > via_angle:
						return Direction.left
			elif target_angle == via_angle:
						return Direction.straight
			else:
						return Direction.right







def __UpperCAmelCase (			A			:	list[tuple[int, int]]						)  -> list[tuple[int, int]]:

			if len(A						) <= 2:
						# There is no convex hull
						raise ValueError('''graham_scan: argument must contain more than 3 points.'''						)
			if len(A						) == 3:
						return points
			# find the lowest and the most left point
			UpperCAmelCase_     :   List[str]       =   0
			UpperCAmelCase_  ,     UpperCAmelCase_     :   List[Any]       =   maxsize, maxsize
			for i, point in enumerate(A						):
						UpperCAmelCase_     :   str       =   point[0]
						UpperCAmelCase_     :   List[Any]       =   point[1]
						if y < miny:
									UpperCAmelCase_     :   int       =   y
									UpperCAmelCase_     :   Dict       =   x
									UpperCAmelCase_     :   Any       =   i
						if y == miny and x < minx:
									UpperCAmelCase_     :   str       =   x
									UpperCAmelCase_     :   Optional[int]       =   i

    # remove the lowest and the most left point from points for preparing for sort
			points.pop(A						)

			UpperCAmelCase_     :   Optional[Any]       =   sorted(A	,   key=lambda A						: angle_comparer(A	,   A	,   A						)						)
			# This insert actually costs complexity,
			# and you should instead add (minx, miny) into stack later.
			# I'm using insert just for easy understanding.
			sorted_points.insert(0	,   (minx, miny)						)

			UpperCAmelCase_     :   deque[tuple[int, int]]       =   deque()
			stack.append(sorted_points[0]						)
			stack.append(sorted_points[1]						)
			stack.append(sorted_points[2]						)
			# In any ways, the first 3 points line are towards left.
			# Because we sort them the angle from minx, miny.
			UpperCAmelCase_     :   Any       =   Direction.left

			for i in range(3	,   len(A						)						):
						while True:
									UpperCAmelCase_     :   Any       =   stack[-2]
									UpperCAmelCase_     :   List[Any]       =   stack[-1]
									UpperCAmelCase_     :   Any       =   sorted_points[i]
									UpperCAmelCase_     :   Tuple       =   check_direction(A	,   A	,   A						)

									if next_direction == Direction.left:
												UpperCAmelCase_     :   Optional[Any]       =   Direction.left
												break
									if next_direction == Direction.straight:
												if current_direction == Direction.left:
															# We keep current_direction as left.
															# Because if the straight line keeps as straight,
															# we want to know if this straight line is towards left.
															break
												elif current_direction == Direction.right:
															# If the straight line is towards right,
															# every previous points on those straigh line is not convex hull.
															stack.pop()
									if next_direction == Direction.right:
												stack.pop()
						stack.append(sorted_points[i]						)
			return list(A						)


