
"""simple docstring"""


from abc import ABC, abstractmethod
from typing import List, Optional







class 				lowerCAmelCase_				(      snake_case__    ):



						'''simple docstring'''


						def __init__( self     :      int      )     ->   Tuple:
							# test for the above condition
							self.test()


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Tuple      )     ->   Optional[int]:
							A						   =   0
							A						   =   False
							while not completed:
								if counter == 1:
									self.reset()
								A						   =   self.advance()
								if not self.does_advance(lowercase_      ):
									raise Exception(
									    'Custom Constraint is not defined correctly. self.does_advance(self.advance()) must be true.'      )

								A						   =   self.update(lowercase_      )
								counter += 1

								if counter > 1_0000:
									raise Exception('update() does not fulfill the constraint.'      )

							if self.remaining() != 0:
								raise Exception('Custom Constraint is not defined correctly.'      )


						@abstractmethod
						def 						_SCREAMING_SNAKE_CASE      ( self     :      int      )     ->   List[Any]:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'      )


						@abstractmethod
						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[Any]  ,A_     :      str      )     ->   str:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'      )


						@abstractmethod
						def 						_SCREAMING_SNAKE_CASE      ( self     :      Tuple  ,A_     :      Dict      )     ->   Union[str, Any]:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'      )


						@abstractmethod
						def 						_SCREAMING_SNAKE_CASE      ( self     :      Dict      )     ->   str:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'      )


						@abstractmethod
						def 						_SCREAMING_SNAKE_CASE      ( self     :      Tuple      )     ->   Tuple:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'      )




						@abstractmethod
						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[Any]  ,A_     :      Optional[int]=False      )     ->   Tuple:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'      )







class 				lowerCAmelCase_				(      snake_case__    ):



						'''simple docstring'''


						def __init__( self     :      Dict  ,A_     :      Union[str, Any]      )     ->   List[Any]:
							super(lowercase_  ,self      ).__init__()

							if not isinstance(lowercase_  ,lowercase_      ) or len(lowercase_      ) == 0:
								raise ValueError(F'`token_ids` has to be a non-empty list, but is {token_ids}.'      )
							if any((not isinstance(lowercase_  ,lowercase_      ) or token_id < 0) for token_id in token_ids      ):
								raise ValueError(F'Each list in `token_ids` has to be a list of positive integers, but is {token_ids}.'      )

							A						   =   token_ids

							A						   =   len(self.token_ids      )
							A						   =   -1  # the index of the currently fulfilled step
							A						   =   False


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Tuple      )     ->   int:
							if self.completed:
								return None
							return self.token_ids[self.fulfilled_idx + 1]


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[Any]  ,A_     :      List[str]      )     ->   Tuple:
							if not isinstance(lowercase_  ,lowercase_      ):
								raise ValueError(F'`token_id` has to be an `int`, but is {token_id} of type {type(lowercase_      )}'      )

							if self.completed:
								return False

							return token_id == self.token_ids[self.fulfilled_idx + 1]


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Union[str, Any]  ,A_     :      int      )     ->   int:
							if not isinstance(lowercase_  ,lowercase_      ):
								raise ValueError(F'`token_id` has to be an `int`, but is {token_id} of type {type(lowercase_      )}'      )

							A						   =   False
							A						   =   False
							A						   =   False

							if self.does_advance(lowercase_      ):
								self.fulfilled_idx += 1
								A						   =   True
								if self.fulfilled_idx == (self.seqlen - 1):
									A						   =   True
								A						   =   completed
							else:
								# failed to make progress.
								A						   =   True
								self.reset()
							return stepped, completed, reset


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[Any]      )     ->   str:
							A						   =   False
							A						   =   0


						def 						_SCREAMING_SNAKE_CASE      ( self     :      List[Any]      )     ->   List[Any]:
							return self.seqlen - (self.fulfilled_idx + 1)




						def 						_SCREAMING_SNAKE_CASE      ( self     :      int  ,A_     :      Optional[Any]=False      )     ->   List[str]:
							A						   =   PhrasalConstraint(self.token_ids      )

							if stateful:
								A						   =   self.seqlen
								A						   =   self.fulfilled_idx
								A						   =   self.completed

							return new_constraint







class 				lowerCAmelCase_				:



						'''simple docstring'''


						def __init__( self     :      Any  ,A_     :      int  ,A_     :      Optional[Any]=True      )     ->   List[str]:
							A						   =   max([len(lowercase_      ) for one in nested_token_ids]      )

							A						   =   {}
							for token_ids in nested_token_ids:
								A						   =   root
								for tidx, token_id in enumerate(lowercase_      ):
									if token_id not in level:
										A						   =   {}

									A						   =   level[token_id]

							if no_subsets and self.has_subsets(lowercase_  ,lowercase_      ):
								raise ValueError(
								    'Each list in `nested_token_ids` can\'t be a complete subset of another list, but is'
								    F' {nested_token_ids}.'      )

							A						   =   root


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Dict  ,A_     :      List[Any]      )     ->   Union[str, Any]:
							A						   =   self.trie

							for current_token in current_seq:
								A						   =   start[current_token]

							A						   =   list(start.keys()      )

							return next_tokens


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Any  ,A_     :      Tuple      )     ->   str:
							A						   =   self.next_tokens(lowercase_      )

							return len(lowercase_      ) == 0


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Tuple  ,A_     :      Tuple      )     ->   str:
							A						   =   list(root.values()      )
							if len(lowercase_      ) == 0:
								return 1
							else:
								return sum([self.count_leaves(lowercase_      ) for nn in next_nodes]      )




						def 						_SCREAMING_SNAKE_CASE      ( self     :      str  ,A_     :      int  ,A_     :      Optional[int]      )     ->   Tuple:
							A						   =   self.count_leaves(lowercase_      )
							return len(lowercase_      ) != leaf_count







class 				lowerCAmelCase_				(      snake_case__    ):



						'''simple docstring'''


						def __init__( self     :      Any  ,A_     :      Tuple      )     ->   Optional[int]:
							super(lowercase_  ,self      ).__init__()

							if not isinstance(lowercase_  ,lowercase_      ) or len(lowercase_      ) == 0:
								raise ValueError(F'`nested_token_ids` has to be a non-empty list, but is {nested_token_ids}.'      )
							if any(not isinstance(lowercase_  ,lowercase_      ) for token_ids in nested_token_ids      ):
								raise ValueError(F'`nested_token_ids` has to be a list of lists, but is {nested_token_ids}.'      )
							if any(
							    any((not isinstance(lowercase_  ,lowercase_      ) or token_id < 0) for token_id in token_ids      )
							    for token_ids in nested_token_ids      ):
								raise ValueError(
								    F'Each list in `nested_token_ids` has to be a list of positive integers, but is {nested_token_ids}.'      )

							A						   =   DisjunctiveTrie(lowercase_      )
							A						   =   nested_token_ids

							A						   =   self.trie.max_height
							A						   =   []
							A						   =   False


						def 						_SCREAMING_SNAKE_CASE      ( self     :      int      )     ->   Dict:
							A						   =   self.trie.next_tokens(self.current_seq      )

							if len(lowercase_      ) == 0:
								return None
							else:
								return token_list


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Union[str, Any]  ,A_     :      Optional[int]      )     ->   str:
							if not isinstance(lowercase_  ,lowercase_      ):
								raise ValueError(F'`token_id` is supposed to be type `int`, but is {token_id} of type {type(lowercase_      )}'      )

							A						   =   self.trie.next_tokens(self.current_seq      )

							return token_id in next_tokens


						def 						_SCREAMING_SNAKE_CASE      ( self     :      str  ,A_     :      List[Any]      )     ->   str:
							if not isinstance(lowercase_  ,lowercase_      ):
								raise ValueError(F'`token_id` is supposed to be type `int`, but is {token_id} of type {type(lowercase_      )}'      )

							A						   =   False
							A						   =   False
							A						   =   False

							if self.does_advance(lowercase_      ):
								self.current_seq.append(lowercase_      )
								A						   =   True
							else:
								A						   =   True
								self.reset()

							A						   =   self.trie.reached_leaf(self.current_seq      )
							A						   =   completed

							return stepped, completed, reset


						def 						_SCREAMING_SNAKE_CASE      ( self     :      int      )     ->   Tuple:
							A						   =   False
							A						   =   []


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[Any]      )     ->   Tuple:
							if self.completed:
								# since this can be completed without reaching max height
								return 0
							else:
								return self.seqlen - len(self.current_seq      )




						def 						_SCREAMING_SNAKE_CASE      ( self     :      Dict  ,A_     :      int=False      )     ->   Dict:
							A						   =   DisjunctiveConstraint(self.token_ids      )

							if stateful:
								A						   =   self.seqlen
								A						   =   self.current_seq
								A						   =   self.completed

							return new_constraint







class 				lowerCAmelCase_				:



						'''simple docstring'''


						def __init__( self     :      Dict  ,A_     :      Optional[Any]      )     ->   Union[str, Any]:
							A						   =   constraints

							# max # of steps required to fulfill a given constraint
							A						   =   max([c.seqlen for c in constraints]      )
							A						   =   len(lowercase_      )
							A						   =   False

							self.init_state()


						def 						_SCREAMING_SNAKE_CASE      ( self     :      List[Any]      )     ->   List[str]:
							A						   =   []
							A						   =   None
							A						   =   [constraint.copy(stateful=lowercase_      ) for constraint in self.constraints]


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[Any]      )     ->   List[Any]:
							A						   =   0
							if self.inprogress_constraint:
								# extra points for having a constraint mid-fulfilled
								add += self.max_seqlen - self.inprogress_constraint.remaining()

							return (len(self.complete_constraints      ) * self.max_seqlen) + add


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[int]      )     ->   Optional[int]:
							A						   =   []
							if self.inprogress_constraint is None:
								for constraint in self.pending_constraints:  # "pending" == "unfulfilled yet"
									A						   =   constraint.advance()
									if isinstance(lowercase_  ,lowercase_      ):
										token_list.append(lowercase_      )
									elif isinstance(lowercase_  ,lowercase_      ):
										token_list.extend(lowercase_      )
							else:
								A						   =   self.inprogress_constraint.advance()
								if isinstance(lowercase_  ,lowercase_      ):
									token_list.append(lowercase_      )
								elif isinstance(lowercase_  ,lowercase_      ):
									token_list.extend(lowercase_      )

							if len(lowercase_      ) == 0:
								return None
							else:
								return token_list


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[Any]  ,A_     :      List[Any]      )     ->   Union[str, Any]:
							self.init_state()

							if token_ids is not None:
								for token in token_ids:
									# completes or steps **one** constraint
									A						   =   self.add(lowercase_      )

									# the entire list of constraints are fulfilled
									if self.completed:
										break


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[int]  ,A_     :      int      )     ->   Optional[int]:
							if not isinstance(lowercase_  ,lowercase_      ):
								raise ValueError(F'`token_id` should be an `int`, but is `{token_id}`.'      )

							A						   =   False, False

							if self.completed:
								A						   =   True
								A						   =   False
								return complete, stepped

							if self.inprogress_constraint is not None:
								# In the middle of fulfilling a constraint. If the `token_id` *does* makes an incremental progress to current
								# job, simply update the state

								A						   =   self.inprogress_constraint.update(lowercase_      )
								if reset:
									# 1. If the next token breaks the progress, then we must restart.
									#     e.g. constraint = "I love pies" and sequence so far is "I love" but `token_id` == "books".

									#     But that doesn't mean we self.init_state(), since we only reset the state for this particular
									#     constraint, not the full list of constraints.

									self.pending_constraints.append(self.inprogress_constraint.copy(stateful=lowercase_      )      )
									A						   =   None

								if complete:
									# 2. If the next token completes the constraint, move it to completed list, set
									#     inprogress to None. If there are no pending constraints either, then this full list of constraints
									#     is complete.

									self.complete_constraints.append(self.inprogress_constraint      )
									A						   =   None

									if len(self.pending_constraints      ) == 0:
										# we're done!
										A						   =   True

							else:
								# Not in the middle of fulfilling a constraint. So does this `token_id` helps us step towards any of our list
								# of constraints?

								for cidx, pending_constraint in enumerate(self.pending_constraints      ):
									if pending_constraint.does_advance(lowercase_      ):
										A						   =   pending_constraint.update(lowercase_      )

										if not stepped:
											raise Exception(
											    '`constraint.update(token_id)` is not yielding incremental progress, '
											    'even though `constraint.does_advance(token_id)` is true.'      )

										if complete:
											self.complete_constraints.append(lowercase_      )
											A						   =   None

										if not complete and stepped:
											A						   =   pending_constraint

										if complete or stepped:
											# If we made any progress at all, then it's at least not a "pending constraint".

											A						   =   (
											    self.pending_constraints[:cidx] + self.pending_constraints[cidx + 1 :]
											)

											if len(self.pending_constraints      ) == 0 and self.inprogress_constraint is None:
												# If there's no longer any pending after this and no inprogress either, then we must be
												# complete.

												A						   =   True

											break  # prevent accidentally stepping through multiple constraints with just one token.

							return complete, stepped




						def 						_SCREAMING_SNAKE_CASE      ( self     :      int  ,A_     :      Dict=True      )     ->   Optional[int]:
							A						   =   ConstraintListState(self.constraints      )  # we actually never though self.constraints objects
							# throughout this process. So it's at initialization state.

							if stateful:
								A						   =   [
								    constraint.copy(stateful=lowercase_      ) for constraint in self.complete_constraints
								]
								if self.inprogress_constraint is not None:
									A						   =   self.inprogress_constraint.copy(stateful=lowercase_      )
								A						   =   [constraint.copy() for constraint in self.pending_constraints]

							return new_state






"""simple docstring"""


import random
import unittest

import numpy as np

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionImgaImgPipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
		import onnxruntime as ort







class 				lowerCAmelCase_				(      _lowercase		,					unittest.TestCase    ):



						'''simple docstring'''


						_lowerCamelCase:    Tuple      =    '''hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'''


						def 						_SCREAMING_SNAKE_CASE      ( self     :      List[Any]  ,A_     :      List[str]=0      )     ->   Dict:
							A						   =   floats_tensor((1, 3, 128, 128)  ,rng=random.Random(A_      )      )
							A						   =   np.random.RandomState(A_      )
							A						   =   {
							    'prompt': 'A painting of a squirrel eating a burger',
							    'image': image,
							    'generator': generator,
							    'num_inference_steps': 3,
							    'strength': 0.75,
							    'guidance_scale': 7.5,
							    'output_type': 'numpy',
							}
							return inputs


						def 						_SCREAMING_SNAKE_CASE      ( self     :      str      )     ->   Optional[int]:
							A						   =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,provider='CPUExecutionProvider'      )
							pipe.set_progress_bar_config(disable=A_      )

							A						   =   self.get_dummy_inputs()
							A						   =   pipe(**A_      ).images
							A						   =   image[0, -3:, -3:, -1].flatten()

							assert image.shape == (1, 128, 128, 3)
							A						   =   np.array([0.6_96_43, 0.5_84_84, 0.5_03_14, 0.5_87_60, 0.5_53_68, 0.5_96_43, 0.5_15_29, 0.4_12_17, 0.4_90_87]      )
							assert np.abs(image_slice - expected_slice      ).max() < 1e-1


						def 						_SCREAMING_SNAKE_CASE      ( self     :      str      )     ->   Union[str, Any]:
							A						   =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,provider='CPUExecutionProvider'      )
							A						   =   PNDMScheduler.from_config(pipe.scheduler.config  ,skip_prk_steps=A_      )
							pipe.set_progress_bar_config(disable=A_      )

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

							assert image.shape == (1, 128, 128, 3)
							A						   =   np.array([0.6_17_37, 0.5_46_42, 0.5_31_83, 0.5_44_65, 0.5_27_42, 0.6_05_25, 0.4_99_69, 0.4_06_55, 0.4_81_54]      )

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


						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[Any]      )     ->   Dict:
							A						   =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,provider='CPUExecutionProvider'      )
							A						   =   LMSDiscreteScheduler.from_config(pipe.scheduler.config      )
							pipe.set_progress_bar_config(disable=A_      )

							# warmup pass to apply optimizations
							A						   =   pipe(**self.get_dummy_inputs()      )

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

							assert image.shape == (1, 128, 128, 3)
							A						   =   np.array([0.5_27_61, 0.5_99_77, 0.4_90_33, 0.4_96_19, 0.5_42_82, 0.5_03_11, 0.4_76_00, 0.4_09_18, 0.4_52_03]      )

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


						def 						_SCREAMING_SNAKE_CASE      ( self     :      int      )     ->   Optional[int]:
							A						   =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,provider='CPUExecutionProvider'      )
							A						   =   EulerDiscreteScheduler.from_config(pipe.scheduler.config      )
							pipe.set_progress_bar_config(disable=A_      )

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

							assert image.shape == (1, 128, 128, 3)
							A						   =   np.array([0.5_29_11, 0.6_00_04, 0.4_92_29, 0.4_98_05, 0.5_45_02, 0.5_06_80, 0.4_77_77, 0.4_10_28, 0.4_53_04]      )

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


						def 						_SCREAMING_SNAKE_CASE      ( self     :      List[str]      )     ->   Optional[int]:
							A						   =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,provider='CPUExecutionProvider'      )
							A						   =   EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config      )
							pipe.set_progress_bar_config(disable=A_      )

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

							assert image.shape == (1, 128, 128, 3)
							A						   =   np.array([0.5_29_11, 0.6_00_04, 0.4_92_29, 0.4_98_05, 0.5_45_02, 0.5_06_80, 0.4_77_77, 0.4_10_28, 0.4_53_04]      )

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




						def 						_SCREAMING_SNAKE_CASE      ( self     :      str      )     ->   Optional[Any]:
							A						   =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,provider='CPUExecutionProvider'      )
							A						   =   DPMSolverMultistepScheduler.from_config(pipe.scheduler.config      )
							pipe.set_progress_bar_config(disable=A_      )

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

							assert image.shape == (1, 128, 128, 3)
							A						   =   np.array([0.6_53_31, 0.5_82_77, 0.4_82_04, 0.5_60_59, 0.5_36_65, 0.5_62_35, 0.5_09_69, 0.4_00_09, 0.4_65_52]      )

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







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



						'''simple docstring'''


						@property
						def 						_SCREAMING_SNAKE_CASE      ( self     :      Any      )     ->   Union[str, Any]:
							return (
							    "CUDAExecutionProvider",
							    {
							        "gpu_mem_limit": "15000000000",  # 15GB
							        "arena_extend_strategy": "kSameAsRequested",
							    },
							)


						@property
						def 						_SCREAMING_SNAKE_CASE      ( self     :      Dict      )     ->   str:
							A						   =   ort.SessionOptions()
							A						   =   False
							return options


						def 						_SCREAMING_SNAKE_CASE      ( self     :      str      )     ->   Tuple:
							A						   =   load_image(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
							    '/img2img/sketch-mountains-input.jpg'      )
							A						   =   init_image.resize((768, 512)      )
							# using the PNDM scheduler by default
							A						   =   OnnxStableDiffusionImgaImgPipeline.from_pretrained(
							    'CompVis/stable-diffusion-v1-4'  ,revision='onnx'  ,safety_checker=A_  ,feature_extractor=A_  ,provider=self.gpu_provider  ,sess_options=self.gpu_options  ,)
							pipe.set_progress_bar_config(disable=A_      )

							A						   =   'A fantasy landscape, trending on artstation'

							A						   =   np.random.RandomState(0      )
							A						   =   pipe(
							    prompt=A_  ,image=A_  ,strength=0.75  ,guidance_scale=7.5  ,num_inference_steps=10  ,generator=A_  ,output_type='np'  ,)
							A						   =   output.images
							A						   =   images[0, 255:258, 383:386, -1]

							assert images.shape == (1, 512, 768, 3)
							A						   =   np.array([0.49_09, 0.50_59, 0.53_72, 0.46_23, 0.48_76, 0.50_49, 0.48_20, 0.49_56, 0.50_19]      )
							# TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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




						def 						_SCREAMING_SNAKE_CASE      ( self     :      Optional[Any]      )     ->   Any:
							A						   =   load_image(
							    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
							    '/img2img/sketch-mountains-input.jpg'      )
							A						   =   init_image.resize((768, 512)      )
							A						   =   LMSDiscreteScheduler.from_pretrained(
							    'runwayml/stable-diffusion-v1-5'  ,subfolder='scheduler'  ,revision='onnx'      )
							A						   =   OnnxStableDiffusionImgaImgPipeline.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  ,)
							pipe.set_progress_bar_config(disable=A_      )

							A						   =   'A fantasy landscape, trending on artstation'

							A						   =   np.random.RandomState(0      )
							A						   =   pipe(
							    prompt=A_  ,image=A_  ,strength=0.75  ,guidance_scale=7.5  ,num_inference_steps=20  ,generator=A_  ,output_type='np'  ,)
							A						   =   output.images
							A						   =   images[0, 255:258, 383:386, -1]

							assert images.shape == (1, 512, 768, 3)
							A						   =   np.array([0.80_43, 0.9_26, 0.95_81, 0.81_19, 0.89_54, 0.9_13, 0.72_09, 0.74_63, 0.74_31]      )
							# TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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