




import tempfile
import unittest

import numpy as np

from diffusers import (
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionPipeline,
    PNDMScheduler,
)
from diffusers.utils.testing_utils import is_onnx_available, nightly, require_onnxruntime, require_torch_gpu

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
	import onnxruntime as ort
class        __snake_case  (_a						,	unittest.TestCase		):
				lowerCAmelCase__				=     "hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"



				def       SCREAMING_SNAKE_CASE       (      self      :		Optional[Any]		,	_UpperCAmelCase      :		Tuple=0      )  ->    List[Any]:





								'''simple docstring'''




								_lowerCAmelCase    :    List[Any]										=      np.random.RandomState(_UpperCAmelCase      )
								_lowerCAmelCase    :    Dict										=      {
								    """prompt""": """A painting of a squirrel eating a burger""",
								    """generator""": generator,
								    """num_inference_steps""": 2,
								    """guidance_scale""": 7.5,
								    """output_type""": """numpy""",
								}
								return inputs



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





								'''simple docstring'''




								_lowerCAmelCase    :    List[Any]										=      OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,	provider="""CPUExecutionProvider"""      )
								pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    List[Any]										=      self.get_dummy_inputs()
								_lowerCAmelCase    :    Tuple										=      pipe(**_UpperCAmelCase      ).images
								_lowerCAmelCase    :    Tuple										=      image[0, -3:, -3:, -1]

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase    :    Optional[Any]										=      np.array([0.65072, 0.58492, 0.48219, 0.55521, 0.53180, 0.55939, 0.50697, 0.39800, 0.46455]      )

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



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





								'''simple docstring'''




								_lowerCAmelCase    :    Optional[Any]										=      OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,	provider="""CPUExecutionProvider"""      )
								_lowerCAmelCase    :    int										=      PNDMScheduler.from_config(pipe.scheduler.config		,	skip_prk_steps=_UpperCAmelCase      )
								pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    Optional[Any]										=      self.get_dummy_inputs()
								_lowerCAmelCase    :    str										=      pipe(**_UpperCAmelCase      ).images
								_lowerCAmelCase    :    Optional[int]										=      image[0, -3:, -3:, -1]

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase    :    List[str]										=      np.array([0.65863, 0.59425, 0.49326, 0.56313, 0.53875, 0.56627, 0.51065, 0.39777, 0.46330]      )

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



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





								'''simple docstring'''




								_lowerCAmelCase    :    Optional[Any]										=      OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,	provider="""CPUExecutionProvider"""      )
								_lowerCAmelCase    :    Dict										=      LMSDiscreteScheduler.from_config(pipe.scheduler.config      )
								pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    Tuple										=      self.get_dummy_inputs()
								_lowerCAmelCase    :    Union[str, Any]										=      pipe(**_UpperCAmelCase      ).images
								_lowerCAmelCase    :    List[str]										=      image[0, -3:, -3:, -1]

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase    :    Union[str, Any]										=      np.array([0.53755, 0.60786, 0.47402, 0.49488, 0.51869, 0.49819, 0.47985, 0.38957, 0.44279]      )

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



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





								'''simple docstring'''




								_lowerCAmelCase    :    Dict										=      OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,	provider="""CPUExecutionProvider"""      )
								_lowerCAmelCase    :    int										=      EulerDiscreteScheduler.from_config(pipe.scheduler.config      )
								pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    str										=      self.get_dummy_inputs()
								_lowerCAmelCase    :    Optional[int]										=      pipe(**_UpperCAmelCase      ).images
								_lowerCAmelCase    :    int										=      image[0, -3:, -3:, -1]

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase    :    Union[str, Any]										=      np.array([0.53755, 0.60786, 0.47402, 0.49488, 0.51869, 0.49819, 0.47985, 0.38957, 0.44279]      )

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



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





								'''simple docstring'''




								_lowerCAmelCase    :    Dict										=      OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,	provider="""CPUExecutionProvider"""      )
								_lowerCAmelCase    :    Optional[int]										=      EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config      )
								pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    List[Any]										=      self.get_dummy_inputs()
								_lowerCAmelCase    :    Any										=      pipe(**_UpperCAmelCase      ).images
								_lowerCAmelCase    :    Any										=      image[0, -3:, -3:, -1]

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase    :    Union[str, Any]										=      np.array([0.53817, 0.60812, 0.47384, 0.49530, 0.51894, 0.49814, 0.47984, 0.38958, 0.44271]      )

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



				def       SCREAMING_SNAKE_CASE       (      self      :		Optional[int]      )  ->    Optional[int]:





								'''simple docstring'''




								_lowerCAmelCase    :    List[str]										=      OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,	provider="""CPUExecutionProvider"""      )
								_lowerCAmelCase    :    str										=      DPMSolverMultistepScheduler.from_config(pipe.scheduler.config      )
								pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    Tuple										=      self.get_dummy_inputs()
								_lowerCAmelCase    :    Optional[Any]										=      pipe(**_UpperCAmelCase      ).images
								_lowerCAmelCase    :    Dict										=      image[0, -3:, -3:, -1]

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase    :    Any										=      np.array([0.53895, 0.60808, 0.47933, 0.49608, 0.51886, 0.49950, 0.48053, 0.38957, 0.44200]      )

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



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





								'''simple docstring'''




								_lowerCAmelCase    :    Tuple										=      OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,	provider="""CPUExecutionProvider"""      )
								pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    Tuple										=      self.get_dummy_inputs()
								_lowerCAmelCase    :    str										=      3 * [inputs["""prompt"""]]

								# forward
								_lowerCAmelCase    :    Tuple										=      pipe(**_UpperCAmelCase      )
								_lowerCAmelCase    :    str										=      output.images[0, -3:, -3:, -1]

								_lowerCAmelCase    :    Optional[int]										=      self.get_dummy_inputs()
								_lowerCAmelCase    :    Optional[Any]										=      3 * [inputs.pop("""prompt"""      )]

								_lowerCAmelCase    :    Union[str, Any]										=      pipe.tokenizer(
								    _UpperCAmelCase		,	padding="""max_length"""		,	max_length=pipe.tokenizer.model_max_length		,	truncation=_UpperCAmelCase		,	return_tensors="""np"""		,	)
								_lowerCAmelCase    :    str										=      text_inputs["""input_ids"""]

								_lowerCAmelCase    :    str										=      pipe.text_encoder(input_ids=text_inputs.astype(np.intaa      )      )[0]

								_lowerCAmelCase    :    Tuple										=      prompt_embeds

								# forward
								_lowerCAmelCase    :    List[Any]										=      pipe(**_UpperCAmelCase      )
								_lowerCAmelCase    :    List[Any]										=      output.images[0, -3:, -3:, -1]

								assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()      ).max() < 1E-4



				def       SCREAMING_SNAKE_CASE       (      self      :		List[Any]      )  ->    Tuple:





								'''simple docstring'''




								_lowerCAmelCase    :    Tuple										=      OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,	provider="""CPUExecutionProvider"""      )
								pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    Any										=      self.get_dummy_inputs()
								_lowerCAmelCase    :    Any										=      3 * ["""this is a negative prompt"""]
								_lowerCAmelCase    :    List[Any]										=      negative_prompt
								_lowerCAmelCase    :    List[str]										=      3 * [inputs["""prompt"""]]

								# forward
								_lowerCAmelCase    :    List[str]										=      pipe(**_UpperCAmelCase      )
								_lowerCAmelCase    :    Tuple										=      output.images[0, -3:, -3:, -1]

								_lowerCAmelCase    :    Any										=      self.get_dummy_inputs()
								_lowerCAmelCase    :    Optional[int]										=      3 * [inputs.pop("""prompt"""      )]

								_lowerCAmelCase    :    List[Any]										=      []
								for p in [prompt, negative_prompt]:
												_lowerCAmelCase    :    Optional[int]										=      pipe.tokenizer(
												    _UpperCAmelCase		,	padding="""max_length"""		,	max_length=pipe.tokenizer.model_max_length		,	truncation=_UpperCAmelCase		,	return_tensors="""np"""		,	)
												_lowerCAmelCase    :    Tuple										=      text_inputs["""input_ids"""]

												embeds.append(pipe.text_encoder(input_ids=text_inputs.astype(np.intaa      )      )[0]      )

								_lowerCAmelCase					,    _lowerCAmelCase    :    Dict										=      embeds

								# forward
								_lowerCAmelCase    :    str										=      pipe(**_UpperCAmelCase      )
								_lowerCAmelCase    :    Any										=      output.images[0, -3:, -3:, -1]

								assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()      ).max() < 1E-4






@nightly
@require_onnxruntime
@require_torch_gpu
class        __snake_case  (unittest.TestCase		):



				@property
				def       SCREAMING_SNAKE_CASE       (      self      :		List[str]      )  ->    Union[str, Any]:





								'''simple docstring'''




								return (
								    "CUDAExecutionProvider",
								    {
								        "gpu_mem_limit": "15000000000",  # 15GB
								        "arena_extend_strategy": "kSameAsRequested",
								    },
								)



				@property
				def       SCREAMING_SNAKE_CASE       (      self      :		Any      )  ->    Optional[int]:





								'''simple docstring'''




								_lowerCAmelCase    :    str										=      ort.SessionOptions()
								_lowerCAmelCase    :    Optional[Any]										=      False
								return options



				def       SCREAMING_SNAKE_CASE       (      self      :		Tuple      )  ->    str:





								'''simple docstring'''




								_lowerCAmelCase    :    int										=      OnnxStableDiffusionPipeline.from_pretrained(
								    """CompVis/stable-diffusion-v1-4"""		,	revision="""onnx"""		,	safety_checker=_UpperCAmelCase		,	feature_extractor=_UpperCAmelCase		,	provider=self.gpu_provider		,	sess_options=self.gpu_options		,	)
								sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    Dict										=      """A painting of a squirrel eating a burger"""
								np.random.seed(0      )
								_lowerCAmelCase    :    List[str]										=      sd_pipe([prompt]		,	guidance_scale=6.0		,	num_inference_steps=10		,	output_type="""np"""      )
								_lowerCAmelCase    :    int										=      output.images

								_lowerCAmelCase    :    int										=      image[0, -3:, -3:, -1]

								assert image.shape == (1, 512, 512, 3)
								_lowerCAmelCase    :    Optional[Any]										=      np.array([0.0452, 0.0390, 0.0087, 0.0350, 0.0617, 0.0364, 0.0544, 0.0523, 0.0720]      )

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



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





								'''simple docstring'''




								_lowerCAmelCase    :    int										=      DDIMScheduler.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,	subfolder="""scheduler"""		,	revision="""onnx"""      )
								_lowerCAmelCase    :    Optional[int]										=      OnnxStableDiffusionPipeline.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,	revision="""onnx"""		,	scheduler=_UpperCAmelCase		,	safety_checker=_UpperCAmelCase		,	feature_extractor=_UpperCAmelCase		,	provider=self.gpu_provider		,	sess_options=self.gpu_options		,	)
								sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    Optional[int]										=      """open neural network exchange"""
								_lowerCAmelCase    :    str										=      np.random.RandomState(0      )
								_lowerCAmelCase    :    List[Any]										=      sd_pipe([prompt]		,	guidance_scale=7.5		,	num_inference_steps=10		,	generator=_UpperCAmelCase		,	output_type="""np"""      )
								_lowerCAmelCase    :    Optional[Any]										=      output.images
								_lowerCAmelCase    :    Tuple										=      image[0, -3:, -3:, -1]

								assert image.shape == (1, 512, 512, 3)
								_lowerCAmelCase    :    Tuple										=      np.array([0.2867, 0.1974, 0.1481, 0.7294, 0.7251, 0.6667, 0.4194, 0.5642, 0.6486]      )

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



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





								'''simple docstring'''




								_lowerCAmelCase    :    Tuple										=      LMSDiscreteScheduler.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,	subfolder="""scheduler"""		,	revision="""onnx"""      )
								_lowerCAmelCase    :    List[str]										=      OnnxStableDiffusionPipeline.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,	revision="""onnx"""		,	scheduler=_UpperCAmelCase		,	safety_checker=_UpperCAmelCase		,	feature_extractor=_UpperCAmelCase		,	provider=self.gpu_provider		,	sess_options=self.gpu_options		,	)
								sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    Optional[Any]										=      """open neural network exchange"""
								_lowerCAmelCase    :    Union[str, Any]										=      np.random.RandomState(0      )
								_lowerCAmelCase    :    int										=      sd_pipe([prompt]		,	guidance_scale=7.5		,	num_inference_steps=10		,	generator=_UpperCAmelCase		,	output_type="""np"""      )
								_lowerCAmelCase    :    List[str]										=      output.images
								_lowerCAmelCase    :    int										=      image[0, -3:, -3:, -1]

								assert image.shape == (1, 512, 512, 3)
								_lowerCAmelCase    :    List[str]										=      np.array([0.2306, 0.1959, 0.1593, 0.6549, 0.6394, 0.5408, 0.5065, 0.6010, 0.6161]      )

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



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





								'''simple docstring'''




								_lowerCAmelCase    :    Optional[int]										=      0

								def test_callback_fn(_UpperCAmelCase      :		int		,	_UpperCAmelCase      :		int		,	_UpperCAmelCase      :		np.ndarray      ) -> None:
												_lowerCAmelCase    :    Tuple										=      True
												nonlocal number_of_steps
												number_of_steps += 1
												if step == 0:
																assert latents.shape == (1, 4, 64, 64)
																_lowerCAmelCase    :    Tuple										=      latents[0, -3:, -3:, -1]
																_lowerCAmelCase    :    Optional[Any]										=      np.array(
																    [-0.6772, -0.3835, -1.2456, 0.1905, -1.0974, 0.6967, -1.9353, 0.0178, 1.0167]      )

																assert np.abs(latents_slice.flatten() - expected_slice      ).max() < 1E-3
												elif step == 5:
																assert latents.shape == (1, 4, 64, 64)
																_lowerCAmelCase    :    List[Any]										=      latents[0, -3:, -3:, -1]
																_lowerCAmelCase    :    List[Any]										=      np.array(
																    [-0.3351, 0.2241, -0.1837, -0.2325, -0.6577, 0.3393, -0.0241, 0.5899, 1.3875]      )

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

								_lowerCAmelCase    :    Tuple										=      False

								_lowerCAmelCase    :    List[Any]										=      OnnxStableDiffusionPipeline.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,	revision="""onnx"""		,	safety_checker=_UpperCAmelCase		,	feature_extractor=_UpperCAmelCase		,	provider=self.gpu_provider		,	sess_options=self.gpu_options		,	)
								pipe.set_progress_bar_config(disable=_UpperCAmelCase      )

								_lowerCAmelCase    :    str										=      """Andromeda galaxy in a bottle"""

								_lowerCAmelCase    :    int										=      np.random.RandomState(0      )
								pipe(
								    prompt=_UpperCAmelCase		,	num_inference_steps=5		,	guidance_scale=7.5		,	generator=_UpperCAmelCase		,	callback=_UpperCAmelCase		,	callback_steps=1		,	)
								assert test_callback_fn.has_been_called
								assert number_of_steps == 6



				def       SCREAMING_SNAKE_CASE       (      self      :		Any      )  ->    str:





								'''simple docstring'''




								_lowerCAmelCase    :    Optional[Any]										=      OnnxStableDiffusionPipeline.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,	revision="""onnx"""		,	safety_checker=_UpperCAmelCase		,	feature_extractor=_UpperCAmelCase		,	provider=self.gpu_provider		,	sess_options=self.gpu_options		,	)
								assert isinstance(_UpperCAmelCase		,	_UpperCAmelCase      )
								assert pipe.safety_checker is None

								_lowerCAmelCase    :    Optional[Any]										=      pipe("""example prompt"""		,	num_inference_steps=2      ).images[0]
								assert image is not None

								# check that there's no error when saving a pipeline with one of the models being None
								with tempfile.TemporaryDirectory() as tmpdirname:
												pipe.save_pretrained(_UpperCAmelCase      )
												_lowerCAmelCase    :    Union[str, Any]										=      OnnxStableDiffusionPipeline.from_pretrained(_UpperCAmelCase      )

								# sanity check that the pipeline still works
								assert pipe.safety_checker is None
								_lowerCAmelCase    :    List[str]										=      pipe("""example prompt"""		,	num_inference_steps=2      ).images[0]
								assert image is not None







import cva
import numpy as np
class        __snake_case  :



				def __init__(      self      :		List[str]		,	_UpperCAmelCase      :		float		,	_UpperCAmelCase      :		int      )  ->    int:





								'''simple docstring'''




								if k in (0.04, 0.06):
												_lowerCAmelCase    :    str										=      k
												_lowerCAmelCase    :    Optional[Any]										=      window_size
								else:
												raise ValueError("""invalid k value"""      )



				def __str__(      self      :		int      )  ->    str:





								'''simple docstring'''




								return str(self.k      )



				def       SCREAMING_SNAKE_CASE       (      self      :		Optional[int]		,	_UpperCAmelCase      :		str      )  ->    tuple[cva.Mat, list[list[int]]]:





								'''simple docstring'''




								_lowerCAmelCase    :    Tuple										=      cva.imread(_UpperCAmelCase		,	0      )
								_lowerCAmelCase					,    _lowerCAmelCase    :    Tuple										=      img.shape
								_lowerCAmelCase    :    list[list[int]]										=      []
								_lowerCAmelCase    :    int										=      img.copy()
								_lowerCAmelCase    :    str										=      cva.cvtColor(_UpperCAmelCase		,	cva.COLOR_GRAY2RGB      )
								_lowerCAmelCase					,    _lowerCAmelCase    :    int										=      np.gradient(_UpperCAmelCase      )
								_lowerCAmelCase    :    Any										=      dx**2
								_lowerCAmelCase    :    Optional[int]										=      dy**2
								_lowerCAmelCase    :    Optional[Any]										=      dx * dy
								_lowerCAmelCase    :    Dict										=      0.04
								_lowerCAmelCase    :    Tuple										=      self.window_size // 2
								for y in range(_UpperCAmelCase		,	h - offset      ):
												for x in range(_UpperCAmelCase		,	w - offset      ):
																_lowerCAmelCase    :    Optional[int]										=      ixx[
																    y - offset : y + offset + 1, x - offset : x + offset + 1
																].sum()
																_lowerCAmelCase    :    int										=      iyy[
																    y - offset : y + offset + 1, x - offset : x + offset + 1
																].sum()
																_lowerCAmelCase    :    Any										=      ixy[
																    y - offset : y + offset + 1, x - offset : x + offset + 1
																].sum()

																_lowerCAmelCase    :    Dict										=      (wxx * wyy) - (wxy**2)
																_lowerCAmelCase    :    Union[str, Any]										=      wxx + wyy
																_lowerCAmelCase    :    int										=      det - k * (trace**2)
																# Can change the value
																if r > 0.5:
																				corner_list.append([x, y, r]      )
																				color_img.itemset((y, x, 0)		,	0      )
																				color_img.itemset((y, x, 1)		,	0      )
																				color_img.itemset((y, x, 2)		,	255      )
								return color_img, corner_list


if __name__ == "__main__":
	_lowerCamelCase      :  int      						=						HarrisCorner(0.0_4, 3)
	_lowerCamelCase						, _lowerCamelCase      :  Any      						=						edge_detect.detect("path_to_image")
	cva.imwrite("detect.png", color_img)

