



"""simple docstring"""






import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import torch

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


__snake_case							:     Union[str, Any]   =					logging.get_logger(__name__)



class     A__     (						__SCREAMING_SNAKE_CASE							):




	'''simple docstring'''





	SCREAMING_SNAKE_CASE					   =			['input_features', 'is_longer']

	def __init__(  self:	Union[str, Any] ,    _SCREAMING_SNAKE_CASE:	Optional[int]=64 ,    _SCREAMING_SNAKE_CASE:	str=4_8000 ,    _SCREAMING_SNAKE_CASE:	Optional[Any]=480 ,    _SCREAMING_SNAKE_CASE:	Optional[Any]=10 ,    _SCREAMING_SNAKE_CASE:	str=1024 ,    _SCREAMING_SNAKE_CASE:	Optional[Any]=0.0 ,    _SCREAMING_SNAKE_CASE:	Tuple=False ,    _SCREAMING_SNAKE_CASE:	float = 0 ,    _SCREAMING_SNAKE_CASE:	float = 1_4000 ,    _SCREAMING_SNAKE_CASE:	int = None ,    _SCREAMING_SNAKE_CASE:	str = "fusion" ,    _SCREAMING_SNAKE_CASE:	str = "repeatpad" ,    **_SCREAMING_SNAKE_CASE:	List[Any] ,    )       -> Union[str, Any]:

				"""simple docstring"""

				super().__init__(
				    feature_size=_SCREAMING_SNAKE_CASE ,    sampling_rate=_SCREAMING_SNAKE_CASE ,    padding_value=_SCREAMING_SNAKE_CASE ,    return_attention_mask=_SCREAMING_SNAKE_CASE ,    **_SCREAMING_SNAKE_CASE ,    )
				__lowerCAmelCase   :						str				= top_db
				__lowerCAmelCase   :						str				= truncation
				__lowerCAmelCase   :						str				= padding
				__lowerCAmelCase   :						Optional[Any]				= fft_window_size
				__lowerCAmelCase   :						Union[str, Any]				= (fft_window_size >> 1) + 1
				__lowerCAmelCase   :						List[str]				= hop_length
				__lowerCAmelCase   :						Dict				= max_length_s
				__lowerCAmelCase   :						List[Any]				= max_length_s * sampling_rate
				__lowerCAmelCase   :						Optional[Any]				= sampling_rate
				__lowerCAmelCase   :						List[str]				= frequency_min
				__lowerCAmelCase   :						List[Any]				= frequency_max
				__lowerCAmelCase   :						Union[str, Any]				= mel_filter_bank(
				    num_frequency_bins=self.nb_frequency_bins ,    num_mel_filters=_SCREAMING_SNAKE_CASE ,    min_frequency=_SCREAMING_SNAKE_CASE ,    max_frequency=_SCREAMING_SNAKE_CASE ,    sampling_rate=_SCREAMING_SNAKE_CASE ,    norm=_SCREAMING_SNAKE_CASE ,    mel_scale="htk" ,    )
				__lowerCAmelCase   :						Union[str, Any]				= mel_filter_bank(
				    num_frequency_bins=self.nb_frequency_bins ,    num_mel_filters=_SCREAMING_SNAKE_CASE ,    min_frequency=_SCREAMING_SNAKE_CASE ,    max_frequency=_SCREAMING_SNAKE_CASE ,    sampling_rate=_SCREAMING_SNAKE_CASE ,    norm="slaney" ,    mel_scale="slaney" ,    )

	def 							_SCREAMING_SNAKE_CASE     (  self:	List[str])       -> Dict[str, Any]:

				"""simple docstring"""

				__lowerCAmelCase   :						Tuple				= copy.deepcopy(self.__dict__)
				__lowerCAmelCase   :						int				= self.__class__.__name__
				if "mel_filters" in output:
							del output["mel_filters"]
				if "mel_filters_slaney" in output:
							del output["mel_filters_slaney"]
				return output

	def 							_SCREAMING_SNAKE_CASE     (  self:	Union[str, Any] ,    _SCREAMING_SNAKE_CASE:	np.array ,    _SCREAMING_SNAKE_CASE:	Optional[np.array] = None)       -> np.ndarray:

				"""simple docstring"""

				__lowerCAmelCase   :						Union[str, Any]				= spectrogram(
				    _SCREAMING_SNAKE_CASE ,    window_function(self.fft_window_size ,    "hann") ,    frame_length=self.fft_window_size ,    hop_length=self.hop_length ,    power=2.0 ,    mel_filters=_SCREAMING_SNAKE_CASE ,    log_mel="dB" ,    )
				return log_mel_spectrogram.T

	def 							_SCREAMING_SNAKE_CASE     (  self:	int ,    _SCREAMING_SNAKE_CASE:	Optional[int] ,    _SCREAMING_SNAKE_CASE:	Optional[int] ,    _SCREAMING_SNAKE_CASE:	Tuple)       -> Tuple:

				"""simple docstring"""

				__lowerCAmelCase   :						Dict				= np.array_split(list(range(0 ,    total_frames - chunk_frames + 1)) ,    3)
				if len(ranges[1]) == 0:
							# if the audio is too short, we just use the first chunk
							__lowerCAmelCase   :						Optional[Any]				= [0]
				if len(ranges[2]) == 0:
							# if the audio is too short, we just use the first chunk
							__lowerCAmelCase   :						Dict				= [0]
				# randomly choose index for each part
				__lowerCAmelCase   :						Optional[Any]				= np.random.choice(ranges[0])
				__lowerCAmelCase   :						Union[str, Any]				= np.random.choice(ranges[1])
				__lowerCAmelCase   :						Tuple				= np.random.choice(ranges[2])

				__lowerCAmelCase   :						int				= mel[idx_front : idx_front + chunk_frames, :]
				__lowerCAmelCase   :						List[str]				= mel[idx_middle : idx_middle + chunk_frames, :]
				__lowerCAmelCase   :						Optional[Any]				= mel[idx_back : idx_back + chunk_frames, :]

				__lowerCAmelCase   :						int				= torch.tensor(mel[None, None, :])
				__lowerCAmelCase   :						int				= torch.nn.functional.interpolate(
				    _SCREAMING_SNAKE_CASE ,    size=[chunk_frames, 64] ,    mode="bilinear" ,    align_corners=_SCREAMING_SNAKE_CASE)
				__lowerCAmelCase   :						Optional[Any]				= mel_shrink[0][0].numpy()
				__lowerCAmelCase   :						List[str]				= np.stack([mel_shrink, mel_chunk_front, mel_chunk_middle, mel_chunk_back] ,    axis=0)
				return mel_fusion

	def 							_SCREAMING_SNAKE_CASE     (  self:	int ,    _SCREAMING_SNAKE_CASE:	np.array ,    _SCREAMING_SNAKE_CASE:	Tuple ,    _SCREAMING_SNAKE_CASE:	Union[str, Any] ,    _SCREAMING_SNAKE_CASE:	Dict)       -> np.array:

				"""simple docstring"""

				if waveform.shape[0] > max_length:
							if truncation == "rand_trunc":
										__lowerCAmelCase   :						List[Any]				= True
										# random crop to max_length (for compatibility) -> this should be handled by self.pad
										__lowerCAmelCase   :						Optional[int]				= len(_SCREAMING_SNAKE_CASE) - max_length
										__lowerCAmelCase   :						Union[str, Any]				= np.random.randint(0 ,    overflow + 1)
										__lowerCAmelCase   :						Any				= waveform[idx : idx + max_length]
										__lowerCAmelCase   :						str				= self._np_extract_fbank_features(_SCREAMING_SNAKE_CASE ,    self.mel_filters_slaney)[None, :]
							elif truncation == "fusion":
										__lowerCAmelCase   :						Optional[Any]				= self._np_extract_fbank_features(_SCREAMING_SNAKE_CASE ,    self.mel_filters)
										__lowerCAmelCase   :						List[Any]				= max_length // self.hop_length + 1  # the +1 related to how the spectrogram is computed
										__lowerCAmelCase   :						Union[str, Any]				= mel.shape[0]
										if chunk_frames == total_frames:
													# there is a corner case where the audio length is larger than max_length but smaller than max_length+hop_length.
													# In this case, we just use the whole audio.
													__lowerCAmelCase   :						Union[str, Any]				= np.stack([mel, mel, mel, mel] ,    axis=0)
													__lowerCAmelCase   :						Union[str, Any]				= False
										else:
													__lowerCAmelCase   :						List[Any]				= self._random_mel_fusion(_SCREAMING_SNAKE_CASE ,    _SCREAMING_SNAKE_CASE ,    _SCREAMING_SNAKE_CASE)
													__lowerCAmelCase   :						Optional[int]				= True
							else:
										raise NotImplementedError(F"""data_truncating {truncation} not implemented""")

				else:
							__lowerCAmelCase   :						Union[str, Any]				= False
							# only use repeat as a new possible value for padding. you repeat the audio before applying the usual max_length padding
							if waveform.shape[0] < max_length:
										if padding == "repeat":
													__lowerCAmelCase   :						Dict				= int(max_length / len(_SCREAMING_SNAKE_CASE))
													__lowerCAmelCase   :						Dict				= np.stack(np.tile(_SCREAMING_SNAKE_CASE ,    n_repeat + 1))[:max_length]
										if padding == "repeatpad":
													__lowerCAmelCase   :						Optional[Any]				= int(max_length / len(_SCREAMING_SNAKE_CASE))
													__lowerCAmelCase   :						List[str]				= np.stack(np.tile(_SCREAMING_SNAKE_CASE ,    _SCREAMING_SNAKE_CASE))
										__lowerCAmelCase   :						Union[str, Any]				= np.pad(_SCREAMING_SNAKE_CASE ,    (0, max_length - waveform.shape[0]) ,    mode="constant" ,    constant_values=0)

							if truncation == "fusion":
										__lowerCAmelCase   :						Optional[int]				= self._np_extract_fbank_features(_SCREAMING_SNAKE_CASE ,    self.mel_filters)
										__lowerCAmelCase   :						Union[str, Any]				= np.stack([input_mel, input_mel, input_mel, input_mel] ,    axis=0)
							else:
										__lowerCAmelCase   :						str				= self._np_extract_fbank_features(_SCREAMING_SNAKE_CASE ,    self.mel_filters_slaney)[None, :]

				return input_mel, longer







	def __call__(  self:	Dict ,    _SCREAMING_SNAKE_CASE:	Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]] ,    _SCREAMING_SNAKE_CASE:	str = None ,    _SCREAMING_SNAKE_CASE:	Optional[str] = None ,    _SCREAMING_SNAKE_CASE:	Optional[int] = None ,    _SCREAMING_SNAKE_CASE:	Optional[int] = None ,    _SCREAMING_SNAKE_CASE:	Optional[Union[str, TensorType]] = None ,    **_SCREAMING_SNAKE_CASE:	Optional[Any] ,    )       -> BatchFeature:

				"""simple docstring"""

				__lowerCAmelCase   :						Any				= truncation if truncation is not None else self.truncation
				__lowerCAmelCase   :						Any				= padding if padding else self.padding

				if sampling_rate is not None:
							if sampling_rate != self.sampling_rate:
										raise ValueError(
										    F"""The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a"""
										    F""" sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input"""
										    F""" was sampled with {self.sampling_rate} and not {sampling_rate}.""")
				else:
							logger.warning(
							    "It is strongly recommended to pass the `sampling_rate` argument to this function. "
							    "Failing to do so can result in silent errors that might be hard to debug.")

				__lowerCAmelCase   :						Optional[Any]				= isinstance(_SCREAMING_SNAKE_CASE ,    np.ndarray) and len(raw_speech.shape) > 1
				if is_batched_numpy and len(raw_speech.shape) > 2:
							raise ValueError(F"""Only mono-channel audio is supported for input to {self}""")
				__lowerCAmelCase   :						List[str]				= is_batched_numpy or (
				    isinstance(_SCREAMING_SNAKE_CASE ,    (list, tuple)) and (isinstance(raw_speech[0] ,    (np.ndarray, tuple, list)))
				)

				if is_batched:
							__lowerCAmelCase   :						int				= [np.asarray(_SCREAMING_SNAKE_CASE ,    dtype=np.floataa) for speech in raw_speech]
				elif not is_batched and not isinstance(_SCREAMING_SNAKE_CASE ,    np.ndarray):
							__lowerCAmelCase   :						Union[str, Any]				= np.asarray(_SCREAMING_SNAKE_CASE ,    dtype=np.floataa)
				elif isinstance(_SCREAMING_SNAKE_CASE ,    np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
							__lowerCAmelCase   :						Any				= raw_speech.astype(np.floataa)

				# always return batch
				if not is_batched:
							__lowerCAmelCase   :						List[Any]				= [np.asarray(_SCREAMING_SNAKE_CASE)]

				# convert to mel spectrogram, truncate and pad if needed.
				__lowerCAmelCase   :						Tuple				= [
				    self._get_input_mel(_SCREAMING_SNAKE_CASE ,    max_length if max_length else self.nb_max_samples ,    _SCREAMING_SNAKE_CASE ,    _SCREAMING_SNAKE_CASE)
				    for waveform in raw_speech
				]

				__lowerCAmelCase   :						Optional[int]				= []
				__lowerCAmelCase   :						Any				= []
				for mel, longer in padded_inputs:
							input_mel.append(_SCREAMING_SNAKE_CASE)
							is_longer.append(_SCREAMING_SNAKE_CASE)

				if truncation == "fusion" and sum(_SCREAMING_SNAKE_CASE) == 0:
							# if no audio is longer than 10s, then randomly select one audio to be longer
							__lowerCAmelCase   :						List[str]				= np.random.randint(0 ,    len(_SCREAMING_SNAKE_CASE))
							__lowerCAmelCase   :						Optional[Any]				= True

				if isinstance(input_mel[0] ,    _SCREAMING_SNAKE_CASE):
							__lowerCAmelCase   :						int				= [np.asarray(_SCREAMING_SNAKE_CASE ,    dtype=np.floataa) for feature in input_mel]

				# is_longer is a list of bool
				__lowerCAmelCase   :						Union[str, Any]				= [[longer] for longer in is_longer]

				__lowerCAmelCase   :						Optional[Any]				= {"input_features": input_mel, "is_longer": is_longer}
				__lowerCAmelCase   :						Tuple				= BatchFeature(_SCREAMING_SNAKE_CASE)

				if return_tensors is not None:
							__lowerCAmelCase   :						Optional[Any]				= input_features.convert_to_tensors(_SCREAMING_SNAKE_CASE)

				return input_features




"""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     A__     (						__SCREAMING_SNAKE_CASE						,	unittest.TestCase							):




	'''simple docstring'''





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

	def 							_SCREAMING_SNAKE_CASE     (  self:	int ,    _SCREAMING_SNAKE_CASE:	Any=0)       -> Union[str, Any]:

				"""simple docstring"""

				__lowerCAmelCase   :						str				= np.random.RandomState(_SCREAMING_SNAKE_CASE)
				__lowerCAmelCase   :						Optional[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 							_SCREAMING_SNAKE_CASE     (  self:	Optional[int])       -> Dict:

				"""simple docstring"""

				__lowerCAmelCase   :						Union[str, Any]				= OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint ,    provider="CPUExecutionProvider")
				pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE)

				__lowerCAmelCase   :						Tuple				= self.get_dummy_inputs()
				__lowerCAmelCase   :						Optional[Any]				= pipe(**_SCREAMING_SNAKE_CASE).images
				__lowerCAmelCase   :						Union[str, Any]				= image[0, -3:, -3:, -1]

				assert image.shape == (1, 128, 128, 3)
				__lowerCAmelCase   :						List[str]				= 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 							_SCREAMING_SNAKE_CASE     (  self:	Optional[Any])       -> Any:

				"""simple docstring"""

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

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

				assert image.shape == (1, 128, 128, 3)
				__lowerCAmelCase   :						str				= 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 							_SCREAMING_SNAKE_CASE     (  self:	str)       -> Any:

				"""simple docstring"""

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

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

				assert image.shape == (1, 128, 128, 3)
				__lowerCAmelCase   :						Any				= 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 							_SCREAMING_SNAKE_CASE     (  self:	List[Any])       -> Dict:

				"""simple docstring"""

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

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

				assert image.shape == (1, 128, 128, 3)
				__lowerCAmelCase   :						Tuple				= 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 							_SCREAMING_SNAKE_CASE     (  self:	List[Any])       -> List[Any]:

				"""simple docstring"""

				__lowerCAmelCase   :						Dict				= OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint ,    provider="CPUExecutionProvider")
				__lowerCAmelCase   :						int				= EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config)
				pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE)

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

				assert image.shape == (1, 128, 128, 3)
				__lowerCAmelCase   :						List[Any]				= 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 							_SCREAMING_SNAKE_CASE     (  self:	Union[str, Any])       -> Union[str, Any]:

				"""simple docstring"""

				__lowerCAmelCase   :						Optional[int]				= OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint ,    provider="CPUExecutionProvider")
				__lowerCAmelCase   :						Any				= DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
				pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE)

				__lowerCAmelCase   :						List[Any]				= self.get_dummy_inputs()
				__lowerCAmelCase   :						List[Any]				= pipe(**_SCREAMING_SNAKE_CASE).images
				__lowerCAmelCase   :						Dict				= image[0, -3:, -3:, -1]

				assert image.shape == (1, 128, 128, 3)
				__lowerCAmelCase   :						Optional[Any]				= 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 							_SCREAMING_SNAKE_CASE     (  self:	Any)       -> str:

				"""simple docstring"""

				__lowerCAmelCase   :						List[str]				= OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint ,    provider="CPUExecutionProvider")
				pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE)

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

				# forward
				__lowerCAmelCase   :						Optional[Any]				= pipe(**_SCREAMING_SNAKE_CASE)
				__lowerCAmelCase   :						List[Any]				= output.images[0, -3:, -3:, -1]

				__lowerCAmelCase   :						Union[str, Any]				= self.get_dummy_inputs()
				__lowerCAmelCase   :						Union[str, Any]				= 3 * [inputs.pop("prompt")]

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

				__lowerCAmelCase   :						str				= pipe.text_encoder(input_ids=text_inputs.astype(np.intaa))[0]

				__lowerCAmelCase   :						Union[str, Any]				= prompt_embeds

				# forward
				__lowerCAmelCase   :						Tuple				= pipe(**_SCREAMING_SNAKE_CASE)
				__lowerCAmelCase   :						Union[str, Any]				= output.images[0, -3:, -3:, -1]

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







	def 							_SCREAMING_SNAKE_CASE     (  self:	Any)       -> int:

				"""simple docstring"""

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

				__lowerCAmelCase   :						Any				= self.get_dummy_inputs()
				__lowerCAmelCase   :						Optional[int]				= 3 * ["this is a negative prompt"]
				__lowerCAmelCase   :						Union[str, Any]				= negative_prompt
				__lowerCAmelCase   :						Union[str, Any]				= 3 * [inputs["prompt"]]

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

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

				__lowerCAmelCase   :						Dict				= []
				for p in [prompt, negative_prompt]:
							__lowerCAmelCase   :						Optional[Any]				= pipe.tokenizer(
							    _SCREAMING_SNAKE_CASE ,    padding="max_length" ,    max_length=pipe.tokenizer.model_max_length ,    truncation=_SCREAMING_SNAKE_CASE ,    return_tensors="np" ,    )
							__lowerCAmelCase   :						Any				= text_inputs["input_ids"]

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

				__lowerCAmelCase		,			__lowerCAmelCase   :						List[str]				= embeds

				# forward
				__lowerCAmelCase   :						int				= pipe(**_SCREAMING_SNAKE_CASE)
				__lowerCAmelCase   :						Union[str, 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     A__     (						unittest.TestCase							):




	'''simple docstring'''





	@property
	def 							_SCREAMING_SNAKE_CASE     (  self:	List[str])       -> int:

				"""simple docstring"""

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

	@property
	def 							_SCREAMING_SNAKE_CASE     (  self:	Union[str, Any])       -> Union[str, Any]:

				"""simple docstring"""

				__lowerCAmelCase   :						str				= ort.SessionOptions()
				__lowerCAmelCase   :						List[str]				= False
				return options

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

				"""simple docstring"""

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

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

				__lowerCAmelCase   :						Any				= image[0, -3:, -3:, -1]

				assert image.shape == (1, 512, 512, 3)
				__lowerCAmelCase   :						Dict				= 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 							_SCREAMING_SNAKE_CASE     (  self:	str)       -> Union[str, Any]:

				"""simple docstring"""

				__lowerCAmelCase   :						Tuple				= DDIMScheduler.from_pretrained(
				    "runwayml/stable-diffusion-v1-5" ,    subfolder="scheduler" ,    revision="onnx")
				__lowerCAmelCase   :						List[str]				= OnnxStableDiffusionPipeline.from_pretrained(
				    "runwayml/stable-diffusion-v1-5" ,    revision="onnx" ,    scheduler=_SCREAMING_SNAKE_CASE ,    safety_checker=_SCREAMING_SNAKE_CASE ,    feature_extractor=_SCREAMING_SNAKE_CASE ,    provider=self.gpu_provider ,    sess_options=self.gpu_options ,    )
				sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE)

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

				assert image.shape == (1, 512, 512, 3)
				__lowerCAmelCase   :						Optional[Any]				= 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 							_SCREAMING_SNAKE_CASE     (  self:	Optional[int])       -> str:

				"""simple docstring"""

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

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

				assert image.shape == (1, 512, 512, 3)
				__lowerCAmelCase   :						List[Any]				= 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 							_SCREAMING_SNAKE_CASE     (  self:	str)       -> Optional[int]:

				"""simple docstring"""

				__lowerCAmelCase   :						str				= 0

				def test_callback_fn(_SCREAMING_SNAKE_CASE:	int ,    _SCREAMING_SNAKE_CASE:	int ,    _SCREAMING_SNAKE_CASE:	np.ndarray) -> None:
							__lowerCAmelCase   :						Optional[int]				= True
							nonlocal number_of_steps
							number_of_steps += 1
							if step == 0:
										assert latents.shape == (1, 4, 64, 64)
										__lowerCAmelCase   :						Optional[int]				= latents[0, -3:, -3:, -1]
										__lowerCAmelCase   :						List[str]				= 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)
										__lowerCAmelCase   :						Tuple				= latents[0, -3:, -3:, -1]
										__lowerCAmelCase   :						Any				= 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

				__lowerCAmelCase   :						Dict				= False

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

				__lowerCAmelCase   :						Dict				= "Andromeda galaxy in a bottle"

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







	def 							_SCREAMING_SNAKE_CASE     (  self:	str)       -> Tuple:

				"""simple docstring"""

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

				__lowerCAmelCase   :						Optional[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(_SCREAMING_SNAKE_CASE)
							__lowerCAmelCase   :						Tuple				= OnnxStableDiffusionPipeline.from_pretrained(_SCREAMING_SNAKE_CASE)

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