
"""simple docstring"""




import collections
import json
import math
import os
import re
import time
from fnmatch import fnmatch
from typing import Dict

import requests
from slack_sdk import WebClient


lowerCamelCase_  :	List[str] 				=					WebClient(token=os.environ["""CI_SLACK_BOT_TOKEN"""])







def     A__   (  lowerCamelCase  )					->		List[Any]:
				UpperCamelCase_:							Optional[Any]		=  test_results.split(""" """  )

				UpperCamelCase_:							int		=  0
				UpperCamelCase_:							str		=  0

				# When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
				# When it is too long, those signs are not present.
				UpperCamelCase_:							Union[str, Any]		=  expressions[-2] if """=""" in expressions[-1] else expressions[-1]

				for i, expression in enumerate(lowerCamelCase  ):
								if "failed" in expression:
												failed += int(expressions[i - 1]  )
								if "passed" in expression:
												success += int(expressions[i - 1]  )

				return failed, success, time_spent







def     A__   (  lowerCamelCase  )					->		Optional[Any]:
				UpperCamelCase_:							Union[str, Any]		=  {}
				UpperCamelCase_:							List[str]		=  None
				UpperCamelCase_:							Dict		=  False
				for line in failures_short_lines.split("""\n"""  ):
								if re.search(r"""_ \[doctest\]"""       ,      lowerCamelCase  ):
												UpperCamelCase_:							Optional[Any]		=  True
												UpperCamelCase_:							Optional[int]		=  line.split(""" """  )[2]
								elif in_error and not line.split(""" """  )[0].isdigit():
												UpperCamelCase_:							Dict		=  line
												UpperCamelCase_:							int		=  False

				return failures

class     _UpperCamelCase			:






			'''simple docstring'''

			def __init__(    self	:    List[str]  ,		snake_case_	:    str  ,		snake_case_	:    Dict ):
							UpperCamelCase_:							Any		=  title

							UpperCamelCase_:							Optional[int]		=  doc_test_results["""time_spent"""].split(""",""" )[0]
							UpperCamelCase_:							Tuple		=  doc_test_results["""success"""]
							UpperCamelCase_:							Tuple		=  doc_test_results["""failures"""]
							UpperCamelCase_:							str		=  self.n_success + self.n_failures

							# Failures and success of the modeling tests
							UpperCamelCase_:							List[Any]		=  doc_test_results

			@property
			def  lowerCAmelCase__		(    self	:    Optional[Any] ):
							UpperCamelCase_:							Optional[int]		=  [self._time_spent]
							UpperCamelCase_:							Optional[Any]		=  0

							for time in time_spent:
											UpperCamelCase_:							List[Any]		=  time.split(""":""" )

											# Time can be formatted as xx:xx:xx, as .xx, or as x.xx if the time spent was less than a minute.
											if len(snake_case_ ) == 1:
															UpperCamelCase_:							str		=  [0, 0, time_parts[0]]

											UpperCamelCase_:							Optional[int]		=  int(time_parts[0] ), int(time_parts[1] ), float(time_parts[2] )
											total_secs += hours * 3600 + minutes * 60 + seconds

							UpperCamelCase_:							List[str]		=  total_secs // 3600, (total_secs % 3600) // 60, total_secs % 60
							return f'''{int(snake_case_ )}h{int(snake_case_ )}m{int(snake_case_ )}s'''

			@property
			def  lowerCAmelCase__		(    self	:    Union[str, Any] ):
							return {"type": "header", "text": {"type": "plain_text", "text": self.title}}

			@property
			def  lowerCAmelCase__		(    self	:    List[Any] ):
							return {
							    "type": "section",
							    "text": {
							        "type": "plain_text",
							        "text": f'''🌞 There were no failures: all {self.n_tests} tests passed. The suite ran in {self.time}.''',
							        "emoji": True,
							    },
							    "accessory": {
							        "type": "button",
							        "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
							        "url": f'''https://github.com/huggingface/transformers/actions/runs/{os.environ["GITHUB_RUN_ID"]}''',
							    },
							}

			@property
			def  lowerCAmelCase__		(    self	:    Any ):
							return {
							    "type": "section",
							    "text": {
							        "type": "plain_text",
							        "text": (
							            f'''There were {self.n_failures} failures, out of {self.n_tests} tests.\nThe suite ran in'''
							            f''' {self.time}.'''
							        ),
							        "emoji": True,
							    },
							    "accessory": {
							        "type": "button",
							        "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
							        "url": f'''https://github.com/huggingface/transformers/actions/runs/{os.environ["GITHUB_RUN_ID"]}''',
							    },
							}

			@property
			def  lowerCAmelCase__		(    self	:    Tuple ):
							UpperCamelCase_:							Dict		=  40
							UpperCamelCase_:							Optional[int]		=  {k: v["""failed"""] for k, v in doc_test_results.items() if isinstance(snake_case_  ,		snake_case_ )}

							UpperCamelCase_:							Any		=  """"""
							for category, failures in category_failures.items():
											if len(snake_case_ ) == 0:
															continue

											if report != "":
															report += "\n\n"

											report += f'''*{category} failures*:'''.ljust(line_length // 2 ).rjust(line_length // 2 ) + "\n"
											report += "`"
											report += "`\n`".join(snake_case_ )
											report += "`"

							return {
							    "type": "section",
							    "text": {
							        "type": "mrkdwn",
							        "text": f'''The following examples had failures:\n\n\n{report}\n''',
							    },
							}

			@property
			def  lowerCAmelCase__		(    self	:    List[str] ):
							UpperCamelCase_:							int		=  [self.header]

							if self.n_failures > 0:
											blocks.append(self.failures )

							if self.n_failures > 0:
											blocks.extend([self.category_failures] )

							if self.n_failures == 0:
											blocks.append(self.no_failures )

							return json.dumps(snake_case_ )

			@staticmethod
			def  lowerCAmelCase__		(    ):
							UpperCamelCase_:							Union[str, Any]		=  [
							    {
							        """type""": """section""",
							        """text""": {
							            """type""": """plain_text""",
							            """text""": """There was an issue running the tests.""",
							        },
							        """accessory""": {
							            """type""": """button""",
							            """text""": {"""type""": """plain_text""", """text""": """Check Action results""", """emoji""": True},
							            """url""": f'''https://github.com/huggingface/transformers/actions/runs/{os.environ["GITHUB_RUN_ID"]}''',
							        },
							    }
							]

							print("""Sending the following payload""" )
							print(json.dumps({"""blocks""": json.loads(snake_case_ )} ) )

							client.chat_postMessage(
							    channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]  ,		text="""There was an issue running the tests."""  ,		blocks=snake_case_  ,		)

			def  lowerCAmelCase__		(    self	:    Tuple ):
							print("""Sending the following payload""" )
							print(json.dumps({"""blocks""": json.loads(self.payload )} ) )

							UpperCamelCase_:							str		=  f'''{self.n_failures} failures out of {self.n_tests} tests,''' if self.n_failures else """All tests passed."""

							UpperCamelCase_:							Optional[Any]		=  client.chat_postMessage(
							    channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]  ,		blocks=self.payload  ,		text=snake_case_  ,		)

			def  lowerCAmelCase__		(    self	:    int  ,		snake_case_	:    List[Any]  ,		snake_case_	:    Dict  ,		snake_case_	:    List[Any]  ,		snake_case_	:    Tuple ):
							UpperCamelCase_:							Tuple		=  """"""
							for key, value in failures.items():
											UpperCamelCase_:							Dict		=  value[:200] + """ [Truncated]""" if len(snake_case_ ) > 250 else value
											failures_text += f'''*{key}*\n_{value}_\n\n'''

							UpperCamelCase_:							Tuple		=  job_name
							UpperCamelCase_:							Any		=  {"""type""": """section""", """text""": {"""type""": """mrkdwn""", """text""": text}}

							if job_link is not None:
											UpperCamelCase_:							Union[str, Any]		=  {
											    """type""": """button""",
											    """text""": {"""type""": """plain_text""", """text""": """GitHub Action job""", """emoji""": True},
											    """url""": job_link,
											}

							return [
							    {"type": "header", "text": {"type": "plain_text", "text": title.upper(), "emoji": True}},
							    content,
							    {"type": "section", "text": {"type": "mrkdwn", "text": failures_text}},
							]




			def  lowerCAmelCase__		(    self	:    int ):
							if self.thread_ts is None:
											raise ValueError("""Can only post reply if a post has been made.""" )

							UpperCamelCase_:							Tuple		=  self.doc_test_results.pop("""job_link""" )
							self.doc_test_results.pop("""failures""" )
							self.doc_test_results.pop("""success""" )
							self.doc_test_results.pop("""time_spent""" )

							UpperCamelCase_:							Optional[int]		=  sorted(self.doc_test_results.items()  ,		key=lambda snake_case_ : t[0] )
							for job, job_result in sorted_dict:
											if len(job_result["""failures"""] ):
															UpperCamelCase_:							Optional[Any]		=  f'''*Num failures* :{len(job_result["failed"] )} \n'''
															UpperCamelCase_:							Optional[int]		=  job_result["""failures"""]
															UpperCamelCase_:							Union[str, Any]		=  self.get_reply_blocks(snake_case_  ,		snake_case_  ,		snake_case_  ,		text=snake_case_ )

															print("""Sending the following reply""" )
															print(json.dumps({"""blocks""": blocks} ) )

															client.chat_postMessage(
															    channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]  ,		text=f'''Results for {job}'''  ,		blocks=snake_case_  ,		thread_ts=self.thread_ts["""ts"""]  ,		)

															time.sleep(1 )







def     A__   (  )					->		str:
				UpperCamelCase_:							Optional[Any]		=  os.environ["""GITHUB_RUN_ID"""]
				UpperCamelCase_:							Dict		=  F'''https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100'''
				UpperCamelCase_:							Optional[int]		=  requests.get(lowerCamelCase  ).json()
				UpperCamelCase_:							Optional[int]		=  {}

				try:
								jobs.update({job["""name"""]: job["""html_url"""] for job in result["""jobs"""]}  )
								UpperCamelCase_:							List[Any]		=  math.ceil((result["""total_count"""] - 1_00) / 1_00  )

								for i in range(lowerCamelCase  ):
												UpperCamelCase_:							Any		=  requests.get(url + F'''&page={i + 2}'''  ).json()
												jobs.update({job["""name"""]: job["""html_url"""] for job in result["""jobs"""]}  )

								return jobs
				except Exception as e:
								print("""Unknown error, could not fetch links."""       ,      lowerCamelCase  )

				return {}







def     A__   (  lowerCamelCase  )					->		Tuple:
				UpperCamelCase_:							List[Any]		=  {}

				if os.path.exists(lowerCamelCase  ):
								UpperCamelCase_:							Optional[int]		=  os.listdir(lowerCamelCase  )
								for file in files:
												try:
																with open(os.path.join(lowerCamelCase       ,      lowerCamelCase  )       ,      encoding="""utf-8"""  ) as f:
																				UpperCamelCase_:							Tuple		=  f.read()
												except UnicodeDecodeError as e:
																raise ValueError(F'''Could not open {os.path.join(lowerCamelCase       ,      lowerCamelCase  )}.'''  ) from e

				return _artifact







def     A__   (  )					->		Union[str, Any]:

				class     _UpperCamelCase			:






							'''simple docstring'''

							def __init__(    self	:    Tuple  ,		snake_case_	:    str ):
											UpperCamelCase_:							Union[str, Any]		=  name
											UpperCamelCase_:							Optional[Any]		=  []

							def __str__(    self	:    Optional[int] ):
											return self.name




							def  lowerCAmelCase__		(    self	:    List[Any]  ,		snake_case_	:    str ):
											self.paths.append({"""name""": self.name, """path""": path} )




				UpperCamelCase_:							Dict[str, Artifact]		=  {}

				UpperCamelCase_:							Dict		=  filter(os.path.isdir       ,      os.listdir()  )
				for directory in directories:
								UpperCamelCase_:							Union[str, Any]		=  directory
								if artifact_name not in _available_artifacts:
												UpperCamelCase_:							Optional[Any]		=  Artifact(lowerCamelCase  )

												_available_artifacts[artifact_name].add_path(lowerCamelCase  )

				return _available_artifacts


if __name__ == "__main__":
							lowerCamelCase_  :	int 				=					get_job_links()
							lowerCamelCase_  :	int 				=					retrieve_available_artifacts()

							lowerCamelCase_  :	Optional[Any] 				=					collections.OrderedDict(
							    [
							        ("""*.py""", """API Examples"""),
							        ("""*.md""", """MD Examples"""),
							    ]
							)

							# This dict will contain all the information relative to each doc test category:
							# - failed: list of failed tests
							# - failures: dict in the format 'test': 'error_message'
							lowerCamelCase_  :	Optional[Any] 				=					{
							    v: {
							        """failed""": [],
							        """failures""": {},
							    }
							    for v in docs.values()
							}

							# Link to the GitHub Action job
							lowerCamelCase_  :	str 				=					github_actions_job_links.get("""run_doctests""")

							lowerCamelCase_  :	Any 				=					available_artifacts["""doc_tests_gpu_test_reports"""].paths[0]
							lowerCamelCase_  :	List[str] 				=					retrieve_artifact(artifact_path["""name"""])
							if "stats" in artifact:
														lowerCamelCase_  :	Dict 				=					handle_test_results(artifact["""stats"""])
														lowerCamelCase_  :	int 				=					failed
														lowerCamelCase_  :	int 				=					success
														lowerCamelCase_  :	Tuple 				=					time_spent[1:-1] + """, """

														lowerCamelCase_  :	Union[str, Any] 				=					extract_first_line_failure(artifact["""failures_short"""])
														for line in artifact["summary_short"].split("""\n"""):
																					if re.search("""FAILED""", line):
																												lowerCamelCase_  :	Optional[Any] 				=					line.replace("""FAILED """, """""")
																												lowerCamelCase_  :	int 				=					line.split()[0].replace("""\n""", """""")

																												if "::" in line:
																																			lowerCamelCase_  :	Tuple 				=					line.split("""::""")
																												else:
																																			lowerCamelCase_  :	List[Any] 				=					line, line

																												for file_regex in docs.keys():
																																			if fnmatch(file_path, file_regex):
																																										lowerCamelCase_  :	Dict 				=					docs[file_regex]
																																										doc_test_results[category]["failed"].append(test)

																																										lowerCamelCase_  :	Dict 				=					all_failures[test] if test in all_failures else """N/A"""
																																										lowerCamelCase_  :	List[str] 				=					failure
																																										break

							lowerCamelCase_  :	List[str] 				=					Message("""🤗 Results of the doc tests.""", doc_test_results)
							message.post()
							message.post_reply()








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     _UpperCamelCase			(   _A       ,    unittest.TestCase					):






			'''simple docstring'''

			__UpperCamelCase :  Optional[Any]    =	"""hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"""

			def  lowerCAmelCase__		(    self	:    List[str]  ,		snake_case_	:    Optional[int]=0 ):
							UpperCamelCase_:							str		=  floats_tensor((1, 3, 128, 128)  ,		rng=random.Random(snake_case_ ) )
							UpperCamelCase_:							Optional[int]		=  np.random.RandomState(snake_case_ )
							UpperCamelCase_:							Dict		=  {
							    """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  lowerCAmelCase__		(    self	:    Any ):
							UpperCamelCase_:							Any		=  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,		provider="""CPUExecutionProvider""" )
							pipe.set_progress_bar_config(disable=snake_case_ )

							UpperCamelCase_:							str		=  self.get_dummy_inputs()
							UpperCamelCase_:							Any		=  pipe(**snake_case_ ).images
							UpperCamelCase_:							Any		=  image[0, -3:, -3:, -1].flatten()

							assert image.shape == (1, 128, 128, 3)
							UpperCamelCase_:							str		=  np.array([0.6_9643, 0.5_8484, 0.5_0314, 0.5_8760, 0.5_5368, 0.5_9643, 0.5_1529, 0.4_1217, 0.4_9087] )
							assert np.abs(image_slice - expected_slice ).max() < 1e-1

			def  lowerCAmelCase__		(    self	:    List[str] ):
							UpperCamelCase_:							int		=  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,		provider="""CPUExecutionProvider""" )
							UpperCamelCase_:							Tuple		=  PNDMScheduler.from_config(pipe.scheduler.config  ,		skip_prk_steps=snake_case_ )
							pipe.set_progress_bar_config(disable=snake_case_ )

							UpperCamelCase_:							str		=  self.get_dummy_inputs()
							UpperCamelCase_:							Dict		=  pipe(**snake_case_ ).images
							UpperCamelCase_:							Optional[Any]		=  image[0, -3:, -3:, -1]

							assert image.shape == (1, 128, 128, 3)
							UpperCamelCase_:							Any		=  np.array([0.6_1737, 0.5_4642, 0.5_3183, 0.5_4465, 0.5_2742, 0.6_0525, 0.4_9969, 0.4_0655, 0.4_8154] )

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

			def  lowerCAmelCase__		(    self	:    str ):
							UpperCamelCase_:							str		=  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,		provider="""CPUExecutionProvider""" )
							UpperCamelCase_:							List[Any]		=  LMSDiscreteScheduler.from_config(pipe.scheduler.config )
							pipe.set_progress_bar_config(disable=snake_case_ )

							# warmup pass to apply optimizations
							UpperCamelCase_:							Union[str, Any]		=  pipe(**self.get_dummy_inputs() )

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

							assert image.shape == (1, 128, 128, 3)
							UpperCamelCase_:							Optional[int]		=  np.array([0.5_2761, 0.5_9977, 0.4_9033, 0.4_9619, 0.5_4282, 0.5_0311, 0.4_7600, 0.4_0918, 0.4_5203] )

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

			def  lowerCAmelCase__		(    self	:    Union[str, Any] ):
							UpperCamelCase_:							List[str]		=  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,		provider="""CPUExecutionProvider""" )
							UpperCamelCase_:							Optional[int]		=  EulerDiscreteScheduler.from_config(pipe.scheduler.config )
							pipe.set_progress_bar_config(disable=snake_case_ )

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

							assert image.shape == (1, 128, 128, 3)
							UpperCamelCase_:							List[str]		=  np.array([0.5_2911, 0.6_0004, 0.4_9229, 0.4_9805, 0.5_4502, 0.5_0680, 0.4_7777, 0.4_1028, 0.4_5304] )

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

			def  lowerCAmelCase__		(    self	:    Optional[int] ):
							UpperCamelCase_:							List[Any]		=  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,		provider="""CPUExecutionProvider""" )
							UpperCamelCase_:							List[Any]		=  EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config )
							pipe.set_progress_bar_config(disable=snake_case_ )

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

							assert image.shape == (1, 128, 128, 3)
							UpperCamelCase_:							Tuple		=  np.array([0.5_2911, 0.6_0004, 0.4_9229, 0.4_9805, 0.5_4502, 0.5_0680, 0.4_7777, 0.4_1028, 0.4_5304] )

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




			def  lowerCAmelCase__		(    self	:    Union[str, Any] ):
							UpperCamelCase_:							Optional[Any]		=  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint  ,		provider="""CPUExecutionProvider""" )
							UpperCamelCase_:							str		=  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config )
							pipe.set_progress_bar_config(disable=snake_case_ )

							UpperCamelCase_:							List[Any]		=  self.get_dummy_inputs()
							UpperCamelCase_:							Optional[int]		=  pipe(**snake_case_ ).images
							UpperCamelCase_:							Dict		=  image[0, -3:, -3:, -1]

							assert image.shape == (1, 128, 128, 3)
							UpperCamelCase_:							Tuple		=  np.array([0.6_5331, 0.5_8277, 0.4_8204, 0.5_6059, 0.5_3665, 0.5_6235, 0.5_0969, 0.4_0009, 0.4_6552] )

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




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






			'''simple docstring'''

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

			@property
			def  lowerCAmelCase__		(    self	:    List[Any] ):
							UpperCamelCase_:							Dict		=  ort.SessionOptions()
							UpperCamelCase_:							Optional[int]		=  False
							return options

			def  lowerCAmelCase__		(    self	:    Tuple ):
							UpperCamelCase_:							Optional[int]		=  load_image(
							    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
							    """/img2img/sketch-mountains-input.jpg""" )
							UpperCamelCase_:							Union[str, Any]		=  init_image.resize((768, 512) )
							# using the PNDM scheduler by default
							UpperCamelCase_:							int		=  OnnxStableDiffusionImgaImgPipeline.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  ,		)
							pipe.set_progress_bar_config(disable=snake_case_ )

							UpperCamelCase_:							Dict		=  """A fantasy landscape, trending on artstation"""

							UpperCamelCase_:							List[str]		=  np.random.RandomState(0 )
							UpperCamelCase_:							List[str]		=  pipe(
							    prompt=snake_case_  ,		image=snake_case_  ,		strength=0.75  ,		guidance_scale=7.5  ,		num_inference_steps=10  ,		generator=snake_case_  ,		output_type="""np"""  ,		)
							UpperCamelCase_:							List[str]		=  output.images
							UpperCamelCase_:							Optional[Any]		=  images[0, 255:258, 383:386, -1]

							assert images.shape == (1, 512, 768, 3)
							UpperCamelCase_:							Tuple		=  np.array([0.4909, 0.5059, 0.5372, 0.4623, 0.4876, 0.5049, 0.4820, 0.4956, 0.5019] )
							# TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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




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

							UpperCamelCase_:							Any		=  """A fantasy landscape, trending on artstation"""

							UpperCamelCase_:							Dict		=  np.random.RandomState(0 )
							UpperCamelCase_:							Union[str, Any]		=  pipe(
							    prompt=snake_case_  ,		image=snake_case_  ,		strength=0.75  ,		guidance_scale=7.5  ,		num_inference_steps=20  ,		generator=snake_case_  ,		output_type="""np"""  ,		)
							UpperCamelCase_:							Optional[int]		=  output.images
							UpperCamelCase_:							Dict		=  images[0, 255:258, 383:386, -1]

							assert images.shape == (1, 512, 768, 3)
							UpperCamelCase_:							Optional[int]		=  np.array([0.8043, 0.926, 0.9581, 0.8119, 0.8954, 0.913, 0.7209, 0.7463, 0.7431] )
							# TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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






