




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 							__snake_case	(_UpperCamelCase    ,						unittest.TestCase		):
				lowerCAmelCase__      						=  '''hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'''



				def    SCREAMING_SNAKE_CASE							(   self   :		Optional[int]		,						_UpperCAmelCase   :		List[Any]=0   )	->     Optional[Any]:





								'''simple docstring'''




								_lowerCAmelCase     :     str              =   np.random.RandomState(lowercase__   )
								_lowerCAmelCase     :     Dict              =   {
								    '''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]   )	->     int:





								'''simple docstring'''




								_lowerCAmelCase     :     Optional[Any]              =   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,						provider="""CPUExecutionProvider"""   )
								pipe.set_progress_bar_config(disable=lowercase__   )

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

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase     :     Optional[Any]              =   np.array([0.65072, 0.58492, 0.48219, 0.55521, 0.53180, 0.55939, 0.50697, 0.39800, 0.46455]   )

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



				def    SCREAMING_SNAKE_CASE							(   self   :		Optional[int]   )	->     List[str]:





								'''simple docstring'''




								_lowerCAmelCase     :     Optional[int]              =   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,						provider="""CPUExecutionProvider"""   )
								_lowerCAmelCase     :     Optional[int]              =   PNDMScheduler.from_config(pipe.scheduler.config		,						skip_prk_steps=lowercase__   )
								pipe.set_progress_bar_config(disable=lowercase__   )

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

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase     :     List[str]              =   np.array([0.65863, 0.59425, 0.49326, 0.56313, 0.53875, 0.56627, 0.51065, 0.39777, 0.46330]   )

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



				def    SCREAMING_SNAKE_CASE							(   self   :		Any   )	->     List[Any]:





								'''simple docstring'''




								_lowerCAmelCase     :     Tuple              =   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,						provider="""CPUExecutionProvider"""   )
								_lowerCAmelCase     :     List[Any]              =   LMSDiscreteScheduler.from_config(pipe.scheduler.config   )
								pipe.set_progress_bar_config(disable=lowercase__   )

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

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase     :     Union[str, Any]              =   np.array([0.53755, 0.60786, 0.47402, 0.49488, 0.51869, 0.49819, 0.47985, 0.38957, 0.44279]   )

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



				def    SCREAMING_SNAKE_CASE							(   self   :		Optional[int]   )	->     Dict:





								'''simple docstring'''




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

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

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase     :     int              =   np.array([0.53755, 0.60786, 0.47402, 0.49488, 0.51869, 0.49819, 0.47985, 0.38957, 0.44279]   )

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



				def    SCREAMING_SNAKE_CASE							(   self   :		List[Any]   )	->     Optional[Any]:





								'''simple docstring'''




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

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

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase     :     Dict              =   np.array([0.53817, 0.60812, 0.47384, 0.49530, 0.51894, 0.49814, 0.47984, 0.38958, 0.44271]   )

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



				def    SCREAMING_SNAKE_CASE							(   self   :		List[Any]   )	->     Optional[int]:





								'''simple docstring'''




								_lowerCAmelCase     :     Union[str, Any]              =   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,						provider="""CPUExecutionProvider"""   )
								_lowerCAmelCase     :     List[Any]              =   DPMSolverMultistepScheduler.from_config(pipe.scheduler.config   )
								pipe.set_progress_bar_config(disable=lowercase__   )

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

								assert image.shape == (1, 128, 128, 3)
								_lowerCAmelCase     :     Dict              =   np.array([0.53895, 0.60808, 0.47933, 0.49608, 0.51886, 0.49950, 0.48053, 0.38957, 0.44200]   )

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



				def    SCREAMING_SNAKE_CASE							(   self   :		Tuple   )	->     List[Any]:





								'''simple docstring'''




								_lowerCAmelCase     :     Optional[int]              =   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,						provider="""CPUExecutionProvider"""   )
								pipe.set_progress_bar_config(disable=lowercase__   )

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

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

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

								_lowerCAmelCase     :     Tuple              =   pipe.tokenizer(
								    lowercase__		,						padding="""max_length"""		,						max_length=pipe.tokenizer.model_max_length		,						truncation=lowercase__		,						return_tensors="""np"""		,						)
								_lowerCAmelCase     :     Optional[Any]              =   text_inputs['''input_ids''']

								_lowerCAmelCase     :     Union[str, Any]              =   pipe.text_encoder(input_ids=text_inputs.astype(np.intaa   )   )[0]

								_lowerCAmelCase     :     Union[str, Any]              =   prompt_embeds

								# forward
								_lowerCAmelCase     :     Union[str, Any]              =   pipe(**lowercase__   )
								_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   :		List[str]   )	->     List[str]:





								'''simple docstring'''




								_lowerCAmelCase     :     List[Any]              =   OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint		,						provider="""CPUExecutionProvider"""   )
								pipe.set_progress_bar_config(disable=lowercase__   )

								_lowerCAmelCase     :     int              =   self.get_dummy_inputs()
								_lowerCAmelCase     :     Any              =   3 * ['''this is a negative prompt''']
								_lowerCAmelCase     :     Optional[int]              =   negative_prompt
								_lowerCAmelCase     :     List[Any]              =   3 * [inputs['''prompt''']]

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

								_lowerCAmelCase     :     Optional[int]              =   self.get_dummy_inputs()
								_lowerCAmelCase     :     Any              =   3 * [inputs.pop("""prompt"""   )]

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

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

								_lowerCAmelCase     :     Optional[int]              =   embeds

								# forward
								_lowerCAmelCase     :     Union[str, Any]              =   pipe(**lowercase__   )
								_lowerCAmelCase     :     str              =   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 							__snake_case	(unittest.TestCase		):



				@property
				def    SCREAMING_SNAKE_CASE							(   self   :		int   )	->     str:





								'''simple docstring'''




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



				@property
				def    SCREAMING_SNAKE_CASE							(   self   :		Optional[int]   )	->     List[Any]:





								'''simple docstring'''




								_lowerCAmelCase     :     Optional[Any]              =   ort.SessionOptions()
								_lowerCAmelCase     :     List[Any]              =   False
								return options



				def    SCREAMING_SNAKE_CASE							(   self   :		List[str]   )	->     int:





								'''simple docstring'''




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

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

								_lowerCAmelCase     :     str              =   image[0, -3:, -3:, -1]

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





								'''simple docstring'''




								_lowerCAmelCase     :     Optional[Any]              =   DDIMScheduler.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,						subfolder="""scheduler"""		,						revision="""onnx"""   )
								_lowerCAmelCase     :     Dict              =   OnnxStableDiffusionPipeline.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,						revision="""onnx"""		,						scheduler=lowercase__		,						safety_checker=lowercase__		,						feature_extractor=lowercase__		,						provider=self.gpu_provider		,						sess_options=self.gpu_options		,						)
								sd_pipe.set_progress_bar_config(disable=lowercase__   )

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

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





								'''simple docstring'''




								_lowerCAmelCase     :     Union[str, Any]              =   LMSDiscreteScheduler.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,						subfolder="""scheduler"""		,						revision="""onnx"""   )
								_lowerCAmelCase     :     Union[str, Any]              =   OnnxStableDiffusionPipeline.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,						revision="""onnx"""		,						scheduler=lowercase__		,						safety_checker=lowercase__		,						feature_extractor=lowercase__		,						provider=self.gpu_provider		,						sess_options=self.gpu_options		,						)
								sd_pipe.set_progress_bar_config(disable=lowercase__   )

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

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





								'''simple docstring'''




								_lowerCAmelCase     :     Tuple              =   0

								def test_callback_fn(_UpperCAmelCase   :		Dict		,						_UpperCAmelCase   :		Union[str, Any]		,						_UpperCAmelCase   :		Union[str, Any]   ) -> None:
												_lowerCAmelCase     :     Optional[Any]              =   True
												nonlocal number_of_steps
												number_of_steps += 1
												if step == 0:
																assert latents.shape == (1, 4, 64, 64)
																_lowerCAmelCase     :     Union[str, Any]              =   latents[0, -3:, -3:, -1]
																_lowerCAmelCase     :     Optional[Any]              =   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     :     Union[str, Any]              =   latents[0, -3:, -3:, -1]
																_lowerCAmelCase     :     Dict              =   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     :     str              =   False

								_lowerCAmelCase     :     List[str]              =   OnnxStableDiffusionPipeline.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,						revision="""onnx"""		,						safety_checker=lowercase__		,						feature_extractor=lowercase__		,						provider=self.gpu_provider		,						sess_options=self.gpu_options		,						)
								pipe.set_progress_bar_config(disable=lowercase__   )

								_lowerCAmelCase     :     Union[str, Any]              =   '''Andromeda galaxy in a bottle'''

								_lowerCAmelCase     :     List[Any]              =   np.random.RandomState(0   )
								pipe(
								    prompt=lowercase__		,						num_inference_steps=5		,						guidance_scale=7.5		,						generator=lowercase__		,						callback=lowercase__		,						callback_steps=1		,						)
								assert test_callback_fn.has_been_called
								assert number_of_steps == 6



				def    SCREAMING_SNAKE_CASE							(   self   :		int   )	->     int:





								'''simple docstring'''




								_lowerCAmelCase     :     Union[str, Any]              =   OnnxStableDiffusionPipeline.from_pretrained(
								    """runwayml/stable-diffusion-v1-5"""		,						revision="""onnx"""		,						safety_checker=lowercase__		,						feature_extractor=lowercase__		,						provider=self.gpu_provider		,						sess_options=self.gpu_options		,						)
								assert isinstance(lowercase__		,						lowercase__   )
								assert pipe.safety_checker is None

								_lowerCAmelCase     :     Union[str, 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(lowercase__   )
												_lowerCAmelCase     :     List[Any]              =   OnnxStableDiffusionPipeline.from_pretrained(lowercase__   )

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





from __future__ import annotations

import requests

lowerCAmelCase       		=    set(
    """approved_at_utc approved_by author_flair_background_color
author_flair_css_class author_flair_richtext author_flair_template_id author_fullname
author_premium can_mod_post category clicked content_categories created_utc downs
edited gilded gildings hidden hide_score is_created_from_ads_ui is_meta
is_original_content is_reddit_media_domain is_video link_flair_css_class
link_flair_richtext link_flair_text link_flair_text_color media_embed mod_reason_title
name permalink pwls quarantine saved score secure_media secure_media_embed selftext
subreddit subreddit_name_prefixed subreddit_type thumbnail title top_awarded_type
total_awards_received ups upvote_ratio url user_reports""".split()
)




def     __SCREAMING_SNAKE_CASE							(       lowercase_      ,       lowercase_ = 1      ,       lowercase_ = "new"      ,       lowercase_ = None      )		->     dict:




						'''simple docstring'''







						__UpperCAmelCase			:							Tuple	    =	wanted_data or []
						if invalid_search_terms := ", ".join(sorted(set(lowercase_      ) - valid_terms      )      ):
												__UpperCAmelCase			:							List[Any]	    =	f"Invalid search term: {invalid_search_terms}"
												raise ValueError(lowercase_      )
						__UpperCAmelCase			:							Optional[int]	    =	requests.get(
						    f"https://reddit.com/r/{subreddit}/{age}.json?limit={limit}"      ,       headers={'''User-agent''': '''A random string'''}      ,       )
						if response.status_code == 429:
												raise requests.HTTPError

						__UpperCAmelCase			:							List[str]	    =	response.json()
						if not wanted_data:
												return {id_: data["data"]["children"][id_] for id_ in range(lowercase_      )}

						__UpperCAmelCase			:							List[Any]	    =	{}
						for id_ in range(lowercase_      ):
												__UpperCAmelCase			:							str	    =	{
												    item: data['''data''']['''children'''][id_]['''data'''][item] for item in wanted_data
												}
						return data_dict


if __name__ == "__main__":
				# If you get Error 429, that means you are rate limited.Try after some time
				print(get_subreddit_data("""learnpython""", wanted_data=["""title""", """url""", """selftext"""]))

