





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(						__snake_case			, unittest.TestCase	):
     '''simple docstring'''


     __magic_name__   =			'hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'




     def        lowerCAmelCase__		(    self				,     snake_case_=0							):
         _A     =							np.random.RandomState(snake_case_							)
         _A     =							{
             '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							):
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint				,     provider='CPUExecutionProvider'							)
         pipe.set_progress_bar_config(disable=snake_case_							)

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

         assert image.shape == (1, 128, 128, 3)
         _A     =							np.array([0.6_5072, 0.5_8492, 0.4_8219, 0.5_5521, 0.5_3180, 0.5_5939, 0.5_0697, 0.3_9800, 0.4_6455]							)

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




     def        lowerCAmelCase__		(    self							):
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint				,     provider='CPUExecutionProvider'							)
         _A     =							PNDMScheduler.from_config(pipe.scheduler.config				,     skip_prk_steps=snake_case_							)
         pipe.set_progress_bar_config(disable=snake_case_							)

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

         assert image.shape == (1, 128, 128, 3)
         _A     =							np.array([0.6_5863, 0.5_9425, 0.4_9326, 0.5_6313, 0.5_3875, 0.5_6627, 0.5_1065, 0.3_9777, 0.4_6330]							)

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




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

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

         assert image.shape == (1, 128, 128, 3)
         _A     =							np.array([0.5_3755, 0.6_0786, 0.4_7402, 0.4_9488, 0.5_1869, 0.4_9819, 0.4_7985, 0.3_8957, 0.4_4279]							)

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




     def        lowerCAmelCase__		(    self							):
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint				,     provider='CPUExecutionProvider'							)
         _A     =							EulerDiscreteScheduler.from_config(pipe.scheduler.config							)
         pipe.set_progress_bar_config(disable=snake_case_							)

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

         assert image.shape == (1, 128, 128, 3)
         _A     =							np.array([0.5_3755, 0.6_0786, 0.4_7402, 0.4_9488, 0.5_1869, 0.4_9819, 0.4_7985, 0.3_8957, 0.4_4279]							)

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




     def        lowerCAmelCase__		(    self							):
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint				,     provider='CPUExecutionProvider'							)
         _A     =							EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config							)
         pipe.set_progress_bar_config(disable=snake_case_							)

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

         assert image.shape == (1, 128, 128, 3)
         _A     =							np.array([0.5_3817, 0.6_0812, 0.4_7384, 0.4_9530, 0.5_1894, 0.4_9814, 0.4_7984, 0.3_8958, 0.4_4271]							)

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




     def        lowerCAmelCase__		(    self							):
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint				,     provider='CPUExecutionProvider'							)
         _A     =							DPMSolverMultistepScheduler.from_config(pipe.scheduler.config							)
         pipe.set_progress_bar_config(disable=snake_case_							)

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

         assert image.shape == (1, 128, 128, 3)
         _A     =							np.array([0.5_3895, 0.6_0808, 0.4_7933, 0.4_9608, 0.5_1886, 0.4_9950, 0.4_8053, 0.3_8957, 0.4_4200]							)

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




     def        lowerCAmelCase__		(    self							):
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint				,     provider='CPUExecutionProvider'							)
         pipe.set_progress_bar_config(disable=snake_case_							)

         _A     =							self.get_dummy_inputs()
         _A     =							3 * [inputs['prompt']]

         # forward
         _A     =							pipe(**snake_case_							)
         _A     =							output.images[0, -3:, -3:, -1]

         _A     =							self.get_dummy_inputs()
         _A     =							3 * [inputs.pop('prompt'							)]

         _A     =							pipe.tokenizer(
             snake_case_				,     padding='max_length'				,     max_length=pipe.tokenizer.model_max_length				,     truncation=snake_case_				,     return_tensors='np'				,     )
         _A     =							text_inputs['input_ids']

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

         _A     =							prompt_embeds

         # forward
         _A     =							pipe(**snake_case_							)
         _A     =							output.images[0, -3:, -3:, -1]

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




     def        lowerCAmelCase__		(    self							):
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint				,     provider='CPUExecutionProvider'							)
         pipe.set_progress_bar_config(disable=snake_case_							)

         _A     =							self.get_dummy_inputs()
         _A     =							3 * ['this is a negative prompt']
         _A     =							negative_prompt
         _A     =							3 * [inputs['prompt']]

         # forward
         _A     =							pipe(**snake_case_							)
         _A     =							output.images[0, -3:, -3:, -1]

         _A     =							self.get_dummy_inputs()
         _A     =							3 * [inputs.pop('prompt'							)]

         _A     =							[]
         for p in [prompt, negative_prompt]:
             _A     =							pipe.tokenizer(
                 snake_case_				,     padding='max_length'				,     max_length=pipe.tokenizer.model_max_length				,     truncation=snake_case_				,     return_tensors='np'				,     )
             _A     =							text_inputs['input_ids']

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

         _A,			_A     =							embeds

         # forward
         _A     =							pipe(**snake_case_							)
         _A     =							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	):
     '''simple docstring'''


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




     @property
     def        lowerCAmelCase__		(    self							):
         _A     =							ort.SessionOptions()
         _A     =							False
         return options




     def        lowerCAmelCase__		(    self							):
         # using the PNDM scheduler by default
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(
             'CompVis/stable-diffusion-v1-4'				,     revision='onnx'				,     safety_checker=snake_case_				,     feature_extractor=snake_case_				,     provider=self.gpu_provider				,     sess_options=self.gpu_options				,     )
         sd_pipe.set_progress_bar_config(disable=snake_case_							)

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

         _A     =							image[0, -3:, -3:, -1]

         assert image.shape == (1, 512, 512, 3)
         _A     =							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        lowerCAmelCase__		(    self							):
         _A     =							DDIMScheduler.from_pretrained(
             'runwayml/stable-diffusion-v1-5'				,     subfolder='scheduler'				,     revision='onnx'							)
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(
             'runwayml/stable-diffusion-v1-5'				,     revision='onnx'				,     scheduler=snake_case_				,     safety_checker=snake_case_				,     feature_extractor=snake_case_				,     provider=self.gpu_provider				,     sess_options=self.gpu_options				,     )
         sd_pipe.set_progress_bar_config(disable=snake_case_							)

         _A     =							'open neural network exchange'
         _A     =							np.random.RandomState(0							)
         _A     =							sd_pipe([prompt]				,     guidance_scale=7.5				,     num_inference_steps=10				,     generator=snake_case_				,     output_type='np'							)
         _A     =							output.images
         _A     =							image[0, -3:, -3:, -1]

         assert image.shape == (1, 512, 512, 3)
         _A     =							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        lowerCAmelCase__		(    self							):
         _A     =							LMSDiscreteScheduler.from_pretrained(
             'runwayml/stable-diffusion-v1-5'				,     subfolder='scheduler'				,     revision='onnx'							)
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(
             'runwayml/stable-diffusion-v1-5'				,     revision='onnx'				,     scheduler=snake_case_				,     safety_checker=snake_case_				,     feature_extractor=snake_case_				,     provider=self.gpu_provider				,     sess_options=self.gpu_options				,     )
         sd_pipe.set_progress_bar_config(disable=snake_case_							)

         _A     =							'open neural network exchange'
         _A     =							np.random.RandomState(0							)
         _A     =							sd_pipe([prompt]				,     guidance_scale=7.5				,     num_inference_steps=10				,     generator=snake_case_				,     output_type='np'							)
         _A     =							output.images
         _A     =							image[0, -3:, -3:, -1]

         assert image.shape == (1, 512, 512, 3)
         _A     =							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        lowerCAmelCase__		(    self							):
         _A     =							0

         def test_callback_fn(snake_case_				,     snake_case_				,     snake_case_							) -> None:
             _A     =							True
             nonlocal number_of_steps
             number_of_steps += 1
             if step == 0:
                 assert latents.shape == (1, 4, 64, 64)
                 _A     =							latents[0, -3:, -3:, -1]
                 _A     =							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)
                 _A     =							latents[0, -3:, -3:, -1]
                 _A     =							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

         _A     =							False

         _A     =							OnnxStableDiffusionPipeline.from_pretrained(
             'runwayml/stable-diffusion-v1-5'				,     revision='onnx'				,     safety_checker=snake_case_				,     feature_extractor=snake_case_				,     provider=self.gpu_provider				,     sess_options=self.gpu_options				,     )
         pipe.set_progress_bar_config(disable=snake_case_							)

         _A     =							'Andromeda galaxy in a bottle'

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




     def        lowerCAmelCase__		(    self							):
         _A     =							OnnxStableDiffusionPipeline.from_pretrained(
             'runwayml/stable-diffusion-v1-5'				,     revision='onnx'				,     safety_checker=snake_case_				,     feature_extractor=snake_case_				,     provider=self.gpu_provider				,     sess_options=self.gpu_options				,     )
         assert isinstance(snake_case_				,     snake_case_							)
         assert pipe.safety_checker is None

         _A     =							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(snake_case_							)
             _A     =							OnnxStableDiffusionPipeline.from_pretrained(snake_case_							)

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












def 				__lowerCAmelCase(       _SCREAMING_SNAKE_CASE							,     _SCREAMING_SNAKE_CASE				)							-> int:




    """simple docstring"""




    print('\nThe shortest path matrix using Floyd Warshall algorithm\n'				)
    for i in range(_SCREAMING_SNAKE_CASE				):
        for j in range(_SCREAMING_SNAKE_CASE				):
            if dist[i][j] != float('inf'				):
                print(int(dist[i][j]				)							,     end='\t'				)
            else:
                print('INF'							,     end='\t'				)
        print()





def 				__lowerCAmelCase(       _SCREAMING_SNAKE_CASE							,     _SCREAMING_SNAKE_CASE				)							-> str:




    """simple docstring"""




    _A     =							[[float('inf'				) for _ in range(_SCREAMING_SNAKE_CASE				)] for _ in range(_SCREAMING_SNAKE_CASE				)]

    for i in range(_SCREAMING_SNAKE_CASE				):
        for j in range(_SCREAMING_SNAKE_CASE				):
            _A     =							graph[i][j]

            # check vertex k against all other vertices (i, j)
    for k in range(_SCREAMING_SNAKE_CASE				):
        # looping through rows of graph array
        for i in range(_SCREAMING_SNAKE_CASE				):
            # looping through columns of graph array
            for j in range(_SCREAMING_SNAKE_CASE				):
                if (
                    dist[i][k] != float('inf'				)
                    and dist[k][j] != float('inf'				)
                    and dist[i][k] + dist[k][j] < dist[i][j]
                ):
                    _A     =							dist[i][k] + dist[k][j]

    _print_dist(_SCREAMING_SNAKE_CASE							,     _SCREAMING_SNAKE_CASE				)
    return dist, v


if __name__ == "__main__":
 __A							:			Dict         =			int(input("Enter number of vertices: "))
 __A							:			Union[str, Any]         =			int(input("Enter number of edges: "))

 __A							:			List[str]         =			[[float("inf") for i in range(v)] for j in range(v)]

 for i in range(v):
  __A							:			List[Any]         =			0.0

  # src and dst are indices that must be within the array size graph[e][v]
  # failure to follow this will result in an error
 for i in range(e):
  print("\nEdge ", i + 1)
  __A							:			Union[str, Any]         =			int(input("Enter source:"))
  __A							:			List[str]         =			int(input("Enter destination:"))
  __A							:			Union[str, Any]         =			float(input("Enter weight:"))
  __A							:			Any         =			weight

 floyd_warshall(graph, v)

 # Example Input
 # Enter number of vertices: 3
 # Enter number of edges: 2

 # # generated graph from vertex and edge inputs
 # [[inf, inf, inf], [inf, inf, inf], [inf, inf, inf]]
 # [[0.0, inf, inf], [inf, 0.0, inf], [inf, inf, 0.0]]

 # specify source, destination and weight for edge #1
 # Edge  1
 # Enter source:1
 # Enter destination:2
 # Enter weight:2

 # specify source, destination and weight for edge #2
 # Edge  2
 # Enter source:2
 # Enter destination:1
 # Enter weight:1

 # # Expected Output from the vertice, edge and src, dst, weight inputs!!
 # 0		INF	INF
 # INF	0	2
 # INF	1	0





