
'''simple docstring'''



from typing import List

import numpy as np


def A (__lowerCamelCase :dict						):
						_lowerCAmelCase    		=					{key: len(__lowerCamelCase						) for key, value in gen_kwargs.items() if isinstance(__lowerCamelCase					,      __lowerCamelCase						)}
						if len(set(lists_lengths.values()						)						) > 1:
												raise RuntimeError(
												    (
												        """Sharding is ambiguous for this dataset: """
												        + """we found several data sources lists of different lengths, and we don't know over which list we should parallelize:\n"""
												        + """\n""".join(f'\t- key {key} has length {length}' for key, length in lists_lengths.items()						)
												        + """\nTo fix this, check the 'gen_kwargs' and make sure to use lists only for data sources, """
												        + """and use tuples otherwise. In the end there should only be one single list, or several lists with the same length."""
												    )						)
						_lowerCAmelCase    		=					max(lists_lengths.values()					,      default=0						)
						return max(1					,      __lowerCamelCase						)


def A (__lowerCamelCase :int					,      __lowerCamelCase :int						):
						_lowerCAmelCase    		=					[]
						for group_idx in range(__lowerCamelCase						):
												_lowerCAmelCase    		=					num_shards // max_num_jobs + (group_idx < (num_shards % max_num_jobs))
												if num_shards_to_add == 0:
																		break
												_lowerCAmelCase    		=					shards_indices_per_group[-1].stop if shards_indices_per_group else 0
												_lowerCAmelCase    		=					range(__lowerCamelCase					,      start + num_shards_to_add						)
												shards_indices_per_group.append(__lowerCamelCase						)
						return shards_indices_per_group


def A (__lowerCamelCase :dict					,      __lowerCamelCase :int						):
						_lowerCAmelCase    		=					_number_of_shards_in_gen_kwargs(__lowerCamelCase						)
						if num_shards == 1:
												return [dict(__lowerCamelCase						)]
						else:
												_lowerCAmelCase    		=					_distribute_shards(num_shards=__lowerCamelCase					,      max_num_jobs=__lowerCamelCase						)
												return [
												    {
												        key: [value[shard_idx] for shard_idx in shard_indices_per_group[group_idx]]
												        if isinstance(__lowerCamelCase					,      __lowerCamelCase						)
												        else value
												        for key, value in gen_kwargs.items()
												    }
												    for group_idx in range(len(__lowerCamelCase						)						)
												]


def A (__lowerCamelCase :List[dict]						):
						return {
						    key: [value for gen_kwargs in gen_kwargs_list for value in gen_kwargs[key]]
						    if isinstance(gen_kwargs_list[0][key]					,      __lowerCamelCase						)
						    else gen_kwargs_list[0][key]
						    for key in gen_kwargs_list[0]
						}





def A (__lowerCamelCase :np.random.Generator					,      __lowerCamelCase :dict						):
						_lowerCAmelCase    		=					{len(__lowerCamelCase						) for value in gen_kwargs.values() if isinstance(__lowerCamelCase					,      __lowerCamelCase						)}
						_lowerCAmelCase    		=					{}
						for size in list_sizes:
												_lowerCAmelCase    		=					list(range(__lowerCamelCase						)						)
												rng.shuffle(indices_per_size[size]						)
						# Now let's copy the gen_kwargs and shuffle the lists based on their sizes
						_lowerCAmelCase    		=					dict(__lowerCamelCase						)
						for key, value in shuffled_kwargs.items():
												if isinstance(__lowerCamelCase					,      __lowerCamelCase						):
																		_lowerCAmelCase    		=					[value[i] for i in indices_per_size[len(__lowerCamelCase						)]]
						return shuffled_kwargs




import tempfile
import unittest

import numpy as np

from diffusers import (
    DDIMScheduler,
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionPipeline,
    PNDMScheduler,
)
from diffusers.utils.testing_utils import is_onnx_available, nightly, require_onnxruntime, require_torch_gpu

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
				import onnxruntime as ort




class 							lowerCamelCase   (  _UpperCamelCase							,						unittest.TestCase ):
	_lowerCAmelCase     :		int										=						'''hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'''


	def  A(    self				,     lowercase__=0):
							__UpperCAmelCase			:							str	    =	np.random.RandomState(lowercase__)
							__UpperCAmelCase			:							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  A(    self):
							__UpperCAmelCase			:							Optional[Any]	    =	OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''')
							pipe.set_progress_bar_config(disable=lowercase__)

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

							assert image.shape == (1, 1_2_8, 1_2_8, 3)
							__UpperCAmelCase			:							Optional[Any]	    =	np.array([0.6_5_0_7_2, 0.5_8_4_9_2, 0.4_8_2_1_9, 0.5_5_5_2_1, 0.5_3_1_8_0, 0.5_5_9_3_9, 0.5_0_6_9_7, 0.3_9_8_0_0, 0.4_6_4_5_5])

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


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

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

							assert image.shape == (1, 1_2_8, 1_2_8, 3)
							__UpperCAmelCase			:							List[str]	    =	np.array([0.6_5_8_6_3, 0.5_9_4_2_5, 0.4_9_3_2_6, 0.5_6_3_1_3, 0.5_3_8_7_5, 0.5_6_6_2_7, 0.5_1_0_6_5, 0.3_9_7_7_7, 0.4_6_3_3_0])

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


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

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

							assert image.shape == (1, 1_2_8, 1_2_8, 3)
							__UpperCAmelCase			:							Union[str, Any]	    =	np.array([0.5_3_7_5_5, 0.6_0_7_8_6, 0.4_7_4_0_2, 0.4_9_4_8_8, 0.5_1_8_6_9, 0.4_9_8_1_9, 0.4_7_9_8_5, 0.3_8_9_5_7, 0.4_4_2_7_9])

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


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

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

							assert image.shape == (1, 1_2_8, 1_2_8, 3)
							__UpperCAmelCase			:							int	    =	np.array([0.5_3_7_5_5, 0.6_0_7_8_6, 0.4_7_4_0_2, 0.4_9_4_8_8, 0.5_1_8_6_9, 0.4_9_8_1_9, 0.4_7_9_8_5, 0.3_8_9_5_7, 0.4_4_2_7_9])

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


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

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

							assert image.shape == (1, 1_2_8, 1_2_8, 3)
							__UpperCAmelCase			:							Dict	    =	np.array([0.5_3_8_1_7, 0.6_0_8_1_2, 0.4_7_3_8_4, 0.4_9_5_3_0, 0.5_1_8_9_4, 0.4_9_8_1_4, 0.4_7_9_8_4, 0.3_8_9_5_8, 0.4_4_2_7_1])

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


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

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

							assert image.shape == (1, 1_2_8, 1_2_8, 3)
							__UpperCAmelCase			:							Dict	    =	np.array([0.5_3_8_9_5, 0.6_0_8_0_8, 0.4_7_9_3_3, 0.4_9_6_0_8, 0.5_1_8_8_6, 0.4_9_9_5_0, 0.4_8_0_5_3, 0.3_8_9_5_7, 0.4_4_2_0_0])

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


	def  A(    self):
							__UpperCAmelCase			:							Optional[int]	    =	OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''')
							pipe.set_progress_bar_config(disable=lowercase__)

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

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

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

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

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

							__UpperCAmelCase			:							Union[str, Any]	    =	prompt_embeds

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

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


	def  A(    self):
							__UpperCAmelCase			:							List[Any]	    =	OnnxStableDiffusionPipeline.from_pretrained(self.hub_checkpoint				,     provider='''CPUExecutionProvider''')
							pipe.set_progress_bar_config(disable=lowercase__)

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

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

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

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

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

							__UpperCAmelCase    ,      __UpperCAmelCase			:							Optional[int]	    =	embeds

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


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


	@property
	def  A(    self):
							__UpperCAmelCase			:							Optional[Any]	    =	ort.SessionOptions()
							__UpperCAmelCase			:							List[Any]	    =	False
							return options


	def  A(    self):
							# using the PNDM scheduler by default
							__UpperCAmelCase			:							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__)

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

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

							assert image.shape == (1, 5_1_2, 5_1_2, 3)
							__UpperCAmelCase			:							List[str]	    =	np.array([0.0_4_5_2, 0.0_3_9_0, 0.0_0_8_7, 0.0_3_5_0, 0.0_6_1_7, 0.0_3_6_4, 0.0_5_4_4, 0.0_5_2_3, 0.0_7_2_0])

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


	def  A(    self):
							__UpperCAmelCase			:							Optional[Any]	    =	DDIMScheduler.from_pretrained(
							    '''runwayml/stable-diffusion-v1-5'''				,     subfolder='''scheduler'''				,     revision='''onnx''')
							__UpperCAmelCase			:							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__)

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

							assert image.shape == (1, 5_1_2, 5_1_2, 3)
							__UpperCAmelCase			:							int	    =	np.array([0.2_8_6_7, 0.1_9_7_4, 0.1_4_8_1, 0.7_2_9_4, 0.7_2_5_1, 0.6_6_6_7, 0.4_1_9_4, 0.5_6_4_2, 0.6_4_8_6])

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


	def  A(    self):
							__UpperCAmelCase			:							Union[str, Any]	    =	LMSDiscreteScheduler.from_pretrained(
							    '''runwayml/stable-diffusion-v1-5'''				,     subfolder='''scheduler'''				,     revision='''onnx''')
							__UpperCAmelCase			:							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__)

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

							assert image.shape == (1, 5_1_2, 5_1_2, 3)
							__UpperCAmelCase			:							Optional[int]	    =	np.array([0.2_3_0_6, 0.1_9_5_9, 0.1_5_9_3, 0.6_5_4_9, 0.6_3_9_4, 0.5_4_0_8, 0.5_0_6_5, 0.6_0_1_0, 0.6_1_6_1])

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


	def  A(    self):
							__UpperCAmelCase			:							Tuple	    =	0

							def test_callback_fn(lowercase__				,     lowercase__				,     lowercase__) -> None:
													__UpperCAmelCase			:							Optional[Any]	    =	True
													nonlocal number_of_steps
													number_of_steps += 1
													if step == 0:
																			assert latents.shape == (1, 4, 6_4, 6_4)
																			__UpperCAmelCase			:							Union[str, Any]	    =	latents[0, -3:, -3:, -1]
																			__UpperCAmelCase			:							Optional[Any]	    =	np.array(
																			    [-0.6_7_7_2, -0.3_8_3_5, -1.2_4_5_6, 0.1_9_0_5, -1.0_9_7_4, 0.6_9_6_7, -1.9_3_5_3, 0.0_1_7_8, 1.0_1_6_7])

																			assert np.abs(latents_slice.flatten() - expected_slice).max() < 1e-3
													elif step == 5:
																			assert latents.shape == (1, 4, 6_4, 6_4)
																			__UpperCAmelCase			:							Union[str, Any]	    =	latents[0, -3:, -3:, -1]
																			__UpperCAmelCase			:							Dict	    =	np.array(
																			    [-0.3_3_5_1, 0.2_2_4_1, -0.1_8_3_7, -0.2_3_2_5, -0.6_5_7_7, 0.3_3_9_3, -0.0_2_4_1, 0.5_8_9_9, 1.3_8_7_5])

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

							__UpperCAmelCase			:							str	    =	False

							__UpperCAmelCase			:							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__)

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

							__UpperCAmelCase			:							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  A(    self):
							__UpperCAmelCase			:							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

							__UpperCAmelCase			:							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__)
													__UpperCAmelCase			:							List[Any]	    =	OnnxStableDiffusionPipeline.from_pretrained(lowercase__)

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

