





'''simple docstring'''


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 	lowerCAmelCase__			(			UpperCamelCase__    ,       unittest.TestCase	):
			lowerCAmelCase :	List[Any]          =     "hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"




			def 							lowerCAmelCase__    (		self  :     Any	,  lowerCamelCase__  :     Union[str, Any]=0 )   ->List[str]:



					'''simple docstring'''







					_UpperCAmelCase		:       Dict								=      np.random.RandomState(_a )
					_UpperCAmelCase		:       Any								=      {
					    """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 							lowerCAmelCase__    (		self  :     Union[str, Any] )   ->Optional[int]:



					'''simple docstring'''







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

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

					assert image.shape == (1, 1_28, 1_28, 3)
					_UpperCAmelCase		:       Any								=      np.array([0.6_5_0_7_2, 0.5_8_4_9_2, 0.4_8_2_1_9, 0.5_5_5_2_1, 0.5_3_1_8_0, 0.5_5_9_3_9, 0.5_0_6_9_7, 0.3_9_8_0_0, 0.4_6_4_5_5] )

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




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



					'''simple docstring'''







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

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

					assert image.shape == (1, 1_28, 1_28, 3)
					_UpperCAmelCase		:       List[Any]								=      np.array([0.6_5_8_6_3, 0.5_9_4_2_5, 0.4_9_3_2_6, 0.5_6_3_1_3, 0.5_3_8_7_5, 0.5_6_6_2_7, 0.5_1_0_6_5, 0.3_9_7_7_7, 0.4_6_3_3_0] )

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




			def 							lowerCAmelCase__    (		self  :     List[str] )   ->List[str]:



					'''simple docstring'''







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

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

					assert image.shape == (1, 1_28, 1_28, 3)
					_UpperCAmelCase		:       Optional[int]								=      np.array([0.5_3_7_5_5, 0.6_0_7_8_6, 0.4_7_4_0_2, 0.4_9_4_8_8, 0.5_1_8_6_9, 0.4_9_8_1_9, 0.4_7_9_8_5, 0.3_8_9_5_7, 0.4_4_2_7_9] )

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




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



					'''simple docstring'''







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

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

					assert image.shape == (1, 1_28, 1_28, 3)
					_UpperCAmelCase		:       Any								=      np.array([0.5_3_7_5_5, 0.6_0_7_8_6, 0.4_7_4_0_2, 0.4_9_4_8_8, 0.5_1_8_6_9, 0.4_9_8_1_9, 0.4_7_9_8_5, 0.3_8_9_5_7, 0.4_4_2_7_9] )

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




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



					'''simple docstring'''







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

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

					assert image.shape == (1, 1_28, 1_28, 3)
					_UpperCAmelCase		:       Tuple								=      np.array([0.5_3_8_1_7, 0.6_0_8_1_2, 0.4_7_3_8_4, 0.4_9_5_3_0, 0.5_1_8_9_4, 0.4_9_8_1_4, 0.4_7_9_8_4, 0.3_8_9_5_8, 0.4_4_2_7_1] )

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




			def 							lowerCAmelCase__    (		self  :     Any )   ->str:



					'''simple docstring'''







					_UpperCAmelCase		:       int								=      OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint	,  provider="CPUExecutionProvider" )
					_UpperCAmelCase		:       Any								=      DPMSolverMultistepScheduler.from_config(pipe.scheduler.config )
					pipe.set_progress_bar_config(disable=_a )

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

					assert image.shape == (1, 1_28, 1_28, 3)
					_UpperCAmelCase		:       Optional[int]								=      np.array([0.5_3_8_9_5, 0.6_0_8_0_8, 0.4_7_9_3_3, 0.4_9_6_0_8, 0.5_1_8_8_6, 0.4_9_9_5_0, 0.4_8_0_5_3, 0.3_8_9_5_7, 0.4_4_2_0_0] )

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




			def 							lowerCAmelCase__    (		self  :     Tuple )   ->Dict:



					'''simple docstring'''







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

					_UpperCAmelCase		:       str								=      self.get_dummy_inputs()
					_UpperCAmelCase		:       List[str]								=      3 * [inputs["""prompt"""]]

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

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

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

					_UpperCAmelCase		:       Union[str, Any]								=      pipe.text_encoder(input_ids=text_inputs.astype(np.intaa ) )[0]

					_UpperCAmelCase		:       int								=      prompt_embeds

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

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






			def 							lowerCAmelCase__    (		self  :     List[str] )   ->List[str]:



					'''simple docstring'''







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

					_UpperCAmelCase		:       Optional[int]								=      self.get_dummy_inputs()
					_UpperCAmelCase		:       Dict								=      3 * ["""this is a negative prompt"""]
					_UpperCAmelCase		:       List[Any]								=      negative_prompt
					_UpperCAmelCase		:       int								=      3 * [inputs["""prompt"""]]

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

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

					_UpperCAmelCase		:       Tuple								=      []
					for p in [prompt, negative_prompt]:
							_UpperCAmelCase		:       str								=      pipe.tokenizer(
							    _a	,  padding="max_length"	,  max_length=pipe.tokenizer.model_max_length	,  truncation=_a	,  return_tensors="np"	,  )
							_UpperCAmelCase		:       Union[str, Any]								=      text_inputs["""input_ids"""]

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

					_UpperCAmelCase		:       List[Any]								=      embeds

					# forward
					_UpperCAmelCase		:       Tuple								=      pipe(**_a )
					_UpperCAmelCase		:       Optional[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 	lowerCAmelCase__			(			unittest.TestCase	):




			@property
			def 							lowerCAmelCase__    (		self  :     Dict )   ->List[str]:



					'''simple docstring'''







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




			@property
			def 							lowerCAmelCase__    (		self  :     Optional[Any] )   ->Dict:



					'''simple docstring'''







					_UpperCAmelCase		:       Any								=      ort.SessionOptions()
					_UpperCAmelCase		:       int								=      False
					return options




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



					'''simple docstring'''







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

					_UpperCAmelCase		:       Optional[Any]								=      """A painting of a squirrel eating a burger"""
					np.random.seed(0 )
					_UpperCAmelCase		:       Union[str, Any]								=      sd_pipe([prompt]	,  guidance_scale=6.0	,  num_inference_steps=10	,  output_type="np" )
					_UpperCAmelCase		:       Union[str, Any]								=      output.images

					_UpperCAmelCase		:       Union[str, Any]								=      image[0, -3:, -3:, -1]

					assert image.shape == (1, 5_12, 5_12, 3)
					_UpperCAmelCase		:       int								=      np.array([0.0_4_5_2, 0.0_3_9_0, 0.0_0_8_7, 0.0_3_5_0, 0.0_6_1_7, 0.0_3_6_4, 0.0_5_4_4, 0.0_5_2_3, 0.0_7_2_0] )

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




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



					'''simple docstring'''







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

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

					assert image.shape == (1, 5_12, 5_12, 3)
					_UpperCAmelCase		:       Union[str, Any]								=      np.array([0.2_8_6_7, 0.1_9_7_4, 0.1_4_8_1, 0.7_2_9_4, 0.7_2_5_1, 0.6_6_6_7, 0.4_1_9_4, 0.5_6_4_2, 0.6_4_8_6] )

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




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



					'''simple docstring'''







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

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

					assert image.shape == (1, 5_12, 5_12, 3)
					_UpperCAmelCase		:       Union[str, Any]								=      np.array([0.2_3_0_6, 0.1_9_5_9, 0.1_5_9_3, 0.6_5_4_9, 0.6_3_9_4, 0.5_4_0_8, 0.5_0_6_5, 0.6_0_1_0, 0.6_1_6_1] )

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




			def 							lowerCAmelCase__    (		self  :     str )   ->int:



					'''simple docstring'''







					_UpperCAmelCase		:       List[Any]								=      0

					def test_callback_fn(lowerCamelCase__  :     Dict	,  lowerCamelCase__  :     Union[str, Any]	,  lowerCamelCase__  :     Dict ) -> None:
							_UpperCAmelCase		:       List[str]								=      True
							nonlocal number_of_steps
							number_of_steps += 1
							if step == 0:
									assert latents.shape == (1, 4, 64, 64)
									_UpperCAmelCase		:       Optional[int]								=      latents[0, -3:, -3:, -1]
									_UpperCAmelCase		:       str								=      np.array(
									    [-0.6_7_7_2, -0.3_8_3_5, -1.2_4_5_6, 0.1_9_0_5, -1.0_9_7_4, 0.6_9_6_7, -1.9_3_5_3, 0.0_1_7_8, 1.0_1_6_7] )

									assert np.abs(latents_slice.flatten() - expected_slice ).max() < 1E-3
							elif step == 5:
									assert latents.shape == (1, 4, 64, 64)
									_UpperCAmelCase		:       Union[str, Any]								=      latents[0, -3:, -3:, -1]
									_UpperCAmelCase		:       Tuple								=      np.array(
									    [-0.3_3_5_1, 0.2_2_4_1, -0.1_8_3_7, -0.2_3_2_5, -0.6_5_7_7, 0.3_3_9_3, -0.0_2_4_1, 0.5_8_9_9, 1.3_8_7_5] )

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

					_UpperCAmelCase		:       List[Any]								=      False

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

					_UpperCAmelCase		:       List[Any]								=      """Andromeda galaxy in a bottle"""

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






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



					'''simple docstring'''







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

					_UpperCAmelCase		:       Dict								=      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(_a )
							_UpperCAmelCase		:       List[Any]								=      OnnxStableDiffusionPipeline.from_pretrained(_a )

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







'''simple docstring'''


from __future__ import annotations

lowerCamelCase__    	=     {
    'A': ['B', 'C', 'E'],
    'B': ['A', 'D', 'E'],
    'C': ['A', 'F', 'G'],
    'D': ['B'],
    'E': ['A', 'B', 'D'],
    'F': ['C'],
    'G': ['C'],
}






class 	lowerCAmelCase__			:




			def __init__(		self  :     int	,  lowerCamelCase__  :     dict[str, list[str]]	,  lowerCamelCase__  :     str )   ->None:



					'''simple docstring'''







					_UpperCAmelCase		:       Dict								=      graph
					# mapping node to its parent in resulting breadth first tree
					_UpperCAmelCase		:       dict[str, str | None]								=      {}
					_UpperCAmelCase		:       List[Any]								=      source_vertex




			def 							lowerCAmelCase__    (		self  :     Optional[int] )   ->None:



					'''simple docstring'''







					_UpperCAmelCase		:       List[Any]								=      {self.source_vertex}
					_UpperCAmelCase		:       List[Any]								=      None
					_UpperCAmelCase		:       List[str]								=      [self.source_vertex]  # first in first out queue

					while queue:
							_UpperCAmelCase		:       int								=      queue.pop(0 )
							for adjacent_vertex in self.graph[vertex]:
									if adjacent_vertex not in visited:
											visited.add(lowerCamelCase__ )
											_UpperCAmelCase		:       List[Any]								=      vertex
											queue.append(lowerCamelCase__ )






			def 							lowerCAmelCase__    (		self  :     Tuple	,  lowerCamelCase__  :     str )   ->str:



					'''simple docstring'''







					if target_vertex == self.source_vertex:
							return self.source_vertex

					_UpperCAmelCase		:       int								=      self.parent.get(lowerCamelCase__ )
					if target_vertex_parent is None:
							_UpperCAmelCase		:       Tuple								=      (
							    F"""No path from vertex: {self.source_vertex} to vertex: {target_vertex}"""
							)
							raise ValueError(lowerCamelCase__ )

					return self.shortest_path(lowerCamelCase__ ) + F"""->{target_vertex}"""


if __name__ == "__main__":
						lowerCamelCase__    	=     Graph(graph, 'G')
						g.breath_first_search()
						print(g.shortest_path('D'))
						print(g.shortest_path('G'))
						print(g.shortest_path('Foo'))



