

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       SCREAMING_SNAKE_CASE__					(__snake_case		,       unittest.TestCase     ):
    __lowerCamelCase						:       Union[str, Any]														=     """hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"""





    def       snake_case_				(      self						,   a=0):
         lowercase__    :   List[Any]      =       np.random.RandomState(a)
         lowercase__    :   int      =       {
             '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       snake_case_				(      self):
         lowercase__    :   List[Any]      =       OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint						,   provider='CPUExecutionProvider')
         pipe.set_progress_bar_config(disable=a)

         lowercase__    :   List[Any]      =       self.get_dummy_inputs()
         lowercase__    :   Tuple      =       pipe(**a).images
         lowercase__    :   Optional[int]      =       image[0, -3:, -3:, -1]

         assert image.shape == (1, 128, 128, 3)
         lowercase__    :   List[str]      =       np.array([0.65_072, 0.58_492, 0.48_219, 0.55_521, 0.53_180, 0.55_939, 0.50_697, 0.39_800, 0.46_455])

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





    def       snake_case_				(      self):
         lowercase__    :   Union[str, Any]      =       OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint						,   provider='CPUExecutionProvider')
         lowercase__    :   str      =       PNDMScheduler.from_config(pipe.scheduler.config						,   skip_prk_steps=a)
         pipe.set_progress_bar_config(disable=a)

         lowercase__    :   Tuple      =       self.get_dummy_inputs()
         lowercase__    :   int      =       pipe(**a).images
         lowercase__    :   Union[str, Any]      =       image[0, -3:, -3:, -1]

         assert image.shape == (1, 128, 128, 3)
         lowercase__    :   Union[str, Any]      =       np.array([0.65_863, 0.59_425, 0.49_326, 0.56_313, 0.53_875, 0.56_627, 0.51_065, 0.39_777, 0.46_330])

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





    def       snake_case_				(      self):
         lowercase__    :   Tuple      =       OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint						,   provider='CPUExecutionProvider')
         lowercase__    :   Tuple      =       LMSDiscreteScheduler.from_config(pipe.scheduler.config)
         pipe.set_progress_bar_config(disable=a)

         lowercase__    :   Tuple      =       self.get_dummy_inputs()
         lowercase__    :   str      =       pipe(**a).images
         lowercase__    :   Dict      =       image[0, -3:, -3:, -1]

         assert image.shape == (1, 128, 128, 3)
         lowercase__    :   int      =       np.array([0.53_755, 0.60_786, 0.47_402, 0.49_488, 0.51_869, 0.49_819, 0.47_985, 0.38_957, 0.44_279])

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





    def       snake_case_				(      self):
         lowercase__    :   Optional[int]      =       OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint						,   provider='CPUExecutionProvider')
         lowercase__    :   Any      =       EulerDiscreteScheduler.from_config(pipe.scheduler.config)
         pipe.set_progress_bar_config(disable=a)

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

         assert image.shape == (1, 128, 128, 3)
         lowercase__    :   Dict      =       np.array([0.53_755, 0.60_786, 0.47_402, 0.49_488, 0.51_869, 0.49_819, 0.47_985, 0.38_957, 0.44_279])

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





    def       snake_case_				(      self):
         lowercase__    :   List[str]      =       OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint						,   provider='CPUExecutionProvider')
         lowercase__    :   str      =       EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config)
         pipe.set_progress_bar_config(disable=a)

         lowercase__    :   Dict      =       self.get_dummy_inputs()
         lowercase__    :   str      =       pipe(**a).images
         lowercase__    :   Tuple      =       image[0, -3:, -3:, -1]

         assert image.shape == (1, 128, 128, 3)
         lowercase__    :   List[str]      =       np.array([0.53_817, 0.60_812, 0.47_384, 0.49_530, 0.51_894, 0.49_814, 0.47_984, 0.38_958, 0.44_271])

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





    def       snake_case_				(      self):
         lowercase__    :   Optional[int]      =       OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint						,   provider='CPUExecutionProvider')
         lowercase__    :   str      =       DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
         pipe.set_progress_bar_config(disable=a)

         lowercase__    :   str      =       self.get_dummy_inputs()
         lowercase__    :   int      =       pipe(**a).images
         lowercase__    :   Union[str, Any]      =       image[0, -3:, -3:, -1]

         assert image.shape == (1, 128, 128, 3)
         lowercase__    :   Dict      =       np.array([0.53_895, 0.60_808, 0.47_933, 0.49_608, 0.51_886, 0.49_950, 0.48_053, 0.38_957, 0.44_200])

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





    def       snake_case_				(      self):
         lowercase__    :   Tuple      =       OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint						,   provider='CPUExecutionProvider')
         pipe.set_progress_bar_config(disable=a)

         lowercase__    :   Optional[int]      =       self.get_dummy_inputs()
         lowercase__    :   Dict      =       3 * [inputs['prompt']]

         # forward
         lowercase__    :   int      =       pipe(**a)
         lowercase__    :   Optional[Any]      =       output.images[0, -3:, -3:, -1]

         lowercase__    :   Optional[int]      =       self.get_dummy_inputs()
         lowercase__    :   List[str]      =       3 * [inputs.pop('prompt')]

         lowercase__    :   List[str]      =       pipe.tokenizer(
             a						,   padding='max_length'						,   max_length=pipe.tokenizer.model_max_length						,   truncation=a						,   return_tensors='np'						,   )
         lowercase__    :   Optional[Any]      =       text_inputs['input_ids']

         lowercase__    :   int      =       pipe.text_encoder(input_ids=text_inputs.astype(np.intaa))[0]

         lowercase__    :   List[Any]      =       prompt_embeds

         # forward
         lowercase__    :   Optional[Any]      =       pipe(**a)
         lowercase__    :   Optional[int]      =       output.images[0, -3:, -3:, -1]

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





    def       snake_case_				(      self):
         lowercase__    :   Optional[Any]      =       OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint						,   provider='CPUExecutionProvider')
         pipe.set_progress_bar_config(disable=a)

         lowercase__    :   List[str]      =       self.get_dummy_inputs()
         lowercase__    :   Optional[Any]      =       3 * ['this is a negative prompt']
         lowercase__    :   Union[str, Any]      =       negative_prompt
         lowercase__    :   Optional[Any]      =       3 * [inputs['prompt']]

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

         lowercase__    :   List[str]      =       self.get_dummy_inputs()
         lowercase__    :   Any      =       3 * [inputs.pop('prompt')]

         lowercase__    :   int      =       []
         for p in [prompt, negative_prompt]:
              lowercase__    :   Any      =       pipe.tokenizer(
                  a						,   padding='max_length'						,   max_length=pipe.tokenizer.model_max_length						,   truncation=a						,   return_tensors='np'						,   )
              lowercase__    :   Optional[int]      =       text_inputs['input_ids']

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

         lowercase__      , lowercase__    :   Tuple      =       embeds

         # forward
         lowercase__    :   Any      =       pipe(**a)
         lowercase__    :   List[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       SCREAMING_SNAKE_CASE__					(unittest.TestCase     ):





    @property
    def       snake_case_				(      self):
         return (
             "CUDAExecutionProvider",
             {
                 "gpu_mem_limit": "15000000000",  # 15GB
                 "arena_extend_strategy": "kSameAsRequested",
             },
         )





    @property
    def       snake_case_				(      self):
         lowercase__    :   Any      =       ort.SessionOptions()
         lowercase__    :   Dict      =       False
         return options





    def       snake_case_				(      self):
         # using the PNDM scheduler by default
         lowercase__    :   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)

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

         lowercase__    :   List[str]      =       image[0, -3:, -3:, -1]

         assert image.shape == (1, 512, 512, 3)
         lowercase__    :   Optional[int]      =       np.array([0.0_452, 0.0_390, 0.0_087, 0.0_350, 0.0_617, 0.0_364, 0.0_544, 0.0_523, 0.0_720])

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





    def       snake_case_				(      self):
         lowercase__    :   Any      =       DDIMScheduler.from_pretrained(
             'runwayml/stable-diffusion-v1-5'						,   subfolder='scheduler'						,   revision='onnx')
         lowercase__    :   int      =       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)

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

         assert image.shape == (1, 512, 512, 3)
         lowercase__    :   Union[str, Any]      =       np.array([0.2_867, 0.1_974, 0.1_481, 0.7_294, 0.7_251, 0.6_667, 0.4_194, 0.5_642, 0.6_486])

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





    def       snake_case_				(      self):
         lowercase__    :   Dict      =       LMSDiscreteScheduler.from_pretrained(
             'runwayml/stable-diffusion-v1-5'						,   subfolder='scheduler'						,   revision='onnx')
         lowercase__    :   List[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)

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

         assert image.shape == (1, 512, 512, 3)
         lowercase__    :   Dict      =       np.array([0.2_306, 0.1_959, 0.1_593, 0.6_549, 0.6_394, 0.5_408, 0.5_065, 0.6_010, 0.6_161])

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





    def       snake_case_				(      self):
         lowercase__    :   List[Any]      =       0

         def test_callback_fn(a						,   a						,   a) -> None:
              lowercase__    :   Tuple      =       True
              nonlocal number_of_steps
              number_of_steps += 1
              if step == 0:
                   assert latents.shape == (1, 4, 64, 64)
                   lowercase__    :   List[str]      =       latents[0, -3:, -3:, -1]
                   lowercase__    :   Any      =       np.array(
                       [-0.6_772, -0.3_835, -1.2_456, 0.1_905, -1.0_974, 0.6_967, -1.9_353, 0.0_178, 1.0_167])

                   assert np.abs(latents_slice.flatten() - expected_slice).max() < 1e-3
              elif step == 5:
                   assert latents.shape == (1, 4, 64, 64)
                   lowercase__    :   Optional[int]      =       latents[0, -3:, -3:, -1]
                   lowercase__    :   Dict      =       np.array(
                       [-0.3_351, 0.2_241, -0.1_837, -0.2_325, -0.6_577, 0.3_393, -0.0_241, 0.5_899, 1.3_875])

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

         lowercase__    :   str      =       False

         lowercase__    :   List[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)

         lowercase__    :   Optional[Any]      =       'Andromeda galaxy in a bottle'

         lowercase__    :   Optional[Any]      =       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       snake_case_				(      self):
         lowercase__    :   int      =       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

         lowercase__    :   List[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(a)
              lowercase__    :   Dict      =       OnnxStableDiffusionPipeline.from_pretrained(a)

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




from __future__ import annotations

def 	snake_case__			(    SCREAMING_SNAKE_CASE_		:    dict	,     SCREAMING_SNAKE_CASE_		:    str			):



     '''simple docstring'''





     lowercase__      , lowercase__    :   List[str]      =       set(SCREAMING_SNAKE_CASE_			), [start]

     while stack:
          lowercase__    :   Union[str, Any]      =       stack.pop()
          explored.add(SCREAMING_SNAKE_CASE_			)
          # Differences from BFS:
          # 1) pop last element instead of first one
          # 2) add adjacent elements to stack without exploring them
          for adj in reversed(graph[v]			):
               if adj not in explored:
                    stack.append(SCREAMING_SNAKE_CASE_			)
     return explored


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

if __name__ == "__main__":
       import doctest

       doctest.testmod()
       print(depth_first_search(G, '''A'''))

