


import random
import unittest

import numpy as np

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

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
    import onnxruntime as ort



class 	__lowerCamelCase			(_a				,      unittest.TestCase						):
       _lowercase											=     """hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline"""







       def 		snake_case_	(			self: int,A_: Tuple=0					):






            '''simple docstring'''




            __UpperCamelCase =			floats_tensor((1, 3, 128, 128),rng=random.Random(A_					)					)
            __UpperCamelCase =			np.random.RandomState(A_					)
            __UpperCamelCase =			{
                'prompt': 'A painting of a squirrel eating a burger',
                'image': image,
                'generator': generator,
                'num_inference_steps': 3,
                'strength': 0.7_5,
                'guidance_scale': 7.5,
                'output_type': 'numpy',
            }
            return inputs







       def 		snake_case_	(			self: Union[str, Any]					):






            '''simple docstring'''




            __UpperCamelCase =			OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint,provider='CPUExecutionProvider'					)
            pipe.set_progress_bar_config(disable=A_					)

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

            assert image.shape == (1, 128, 128, 3)
            __UpperCamelCase =			np.array([0.6_9_6_4_3, 0.5_8_4_8_4, 0.5_0_3_1_4, 0.5_8_7_6_0, 0.5_5_3_6_8, 0.5_9_6_4_3, 0.5_1_5_2_9, 0.4_1_2_1_7, 0.4_9_0_8_7]					)
            assert np.abs(image_slice - expected_slice					).max() < 1E-1







       def 		snake_case_	(			self: Tuple					):






            '''simple docstring'''




            __UpperCamelCase =			OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint,provider='CPUExecutionProvider'					)
            __UpperCamelCase =			PNDMScheduler.from_config(pipe.scheduler.config,skip_prk_steps=A_					)
            pipe.set_progress_bar_config(disable=A_					)

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

            assert image.shape == (1, 128, 128, 3)
            __UpperCamelCase =			np.array([0.6_1_7_3_7, 0.5_4_6_4_2, 0.5_3_1_8_3, 0.5_4_4_6_5, 0.5_2_7_4_2, 0.6_0_5_2_5, 0.4_9_9_6_9, 0.4_0_6_5_5, 0.4_8_1_5_4]					)

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







       def 		snake_case_	(			self: List[str]					):






            '''simple docstring'''




            __UpperCamelCase =			OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint,provider='CPUExecutionProvider'					)
            __UpperCamelCase =			LMSDiscreteScheduler.from_config(pipe.scheduler.config					)
            pipe.set_progress_bar_config(disable=A_					)

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

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

            assert image.shape == (1, 128, 128, 3)
            __UpperCamelCase =			np.array([0.5_2_7_6_1, 0.5_9_9_7_7, 0.4_9_0_3_3, 0.4_9_6_1_9, 0.5_4_2_8_2, 0.5_0_3_1_1, 0.4_7_6_0_0, 0.4_0_9_1_8, 0.4_5_2_0_3]					)

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







       def 		snake_case_	(			self: Optional[Any]					):






            '''simple docstring'''




            __UpperCamelCase =			OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint,provider='CPUExecutionProvider'					)
            __UpperCamelCase =			EulerDiscreteScheduler.from_config(pipe.scheduler.config					)
            pipe.set_progress_bar_config(disable=A_					)

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

            assert image.shape == (1, 128, 128, 3)
            __UpperCamelCase =			np.array([0.5_2_9_1_1, 0.6_0_0_0_4, 0.4_9_2_2_9, 0.4_9_8_0_5, 0.5_4_5_0_2, 0.5_0_6_8_0, 0.4_7_7_7_7, 0.4_1_0_2_8, 0.4_5_3_0_4]					)

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







       def 		snake_case_	(			self: List[Any]					):






            '''simple docstring'''




            __UpperCamelCase =			OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint,provider='CPUExecutionProvider'					)
            __UpperCamelCase =			EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config					)
            pipe.set_progress_bar_config(disable=A_					)

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

            assert image.shape == (1, 128, 128, 3)
            __UpperCamelCase =			np.array([0.5_2_9_1_1, 0.6_0_0_0_4, 0.4_9_2_2_9, 0.4_9_8_0_5, 0.5_4_5_0_2, 0.5_0_6_8_0, 0.4_7_7_7_7, 0.4_1_0_2_8, 0.4_5_3_0_4]					)

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







       def 		snake_case_	(			self: str					):






            '''simple docstring'''




            __UpperCamelCase =			OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint,provider='CPUExecutionProvider'					)
            __UpperCamelCase =			DPMSolverMultistepScheduler.from_config(pipe.scheduler.config					)
            pipe.set_progress_bar_config(disable=A_					)

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

            assert image.shape == (1, 128, 128, 3)
            __UpperCamelCase =			np.array([0.6_5_3_3_1, 0.5_8_2_7_7, 0.4_8_2_0_4, 0.5_6_0_5_9, 0.5_3_6_6_5, 0.5_6_2_3_5, 0.5_0_9_6_9, 0.4_0_0_0_9, 0.4_6_5_5_2]					)

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



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







       @property
       def 		snake_case_	(			self: Any					):






            '''simple docstring'''




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







       @property
       def 		snake_case_	(			self: Dict					):






            '''simple docstring'''




            __UpperCamelCase =			ort.SessionOptions()
            __UpperCamelCase =			False
            return options







       def 		snake_case_	(			self: Dict					):






            '''simple docstring'''




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

            __UpperCamelCase =			'A fantasy landscape, trending on artstation'

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

            assert images.shape == (1, 512, 768, 3)
            __UpperCamelCase =			np.array([0.4_9_0_9, 0.5_0_5_9, 0.5_3_7_2, 0.4_6_2_3, 0.4_8_7_6, 0.5_0_4_9, 0.4_8_2_0, 0.4_9_5_6, 0.5_0_1_9]					)
            # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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







       def 		snake_case_	(			self: List[str]					):






            '''simple docstring'''




            __UpperCamelCase =			load_image(
                'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'
                '/img2img/sketch-mountains-input.jpg'					)
            __UpperCamelCase =			init_image.resize((768, 512)					)
            __UpperCamelCase =			LMSDiscreteScheduler.from_pretrained(
                'runwayml/stable-diffusion-v1-5',subfolder='scheduler',revision='onnx'					)
            __UpperCamelCase =			OnnxStableDiffusionImgaImgPipeline.from_pretrained(
                'runwayml/stable-diffusion-v1-5',revision='onnx',scheduler=A_,safety_checker=A_,feature_extractor=A_,provider=self.gpu_provider,sess_options=self.gpu_options,)
            pipe.set_progress_bar_config(disable=A_					)

            __UpperCamelCase =			'A fantasy landscape, trending on artstation'

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

            assert images.shape == (1, 512, 768, 3)
            __UpperCamelCase =			np.array([0.8_0_4_3, 0.9_2_6, 0.9_5_8_1, 0.8_1_1_9, 0.8_9_5_4, 0.9_1_3, 0.7_2_0_9, 0.7_4_6_3, 0.7_4_3_1]					)
            # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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




'''simple docstring'''
import argparse
from pathlib import Path

import torch
from packaging import version
from torch.onnx import export

from diffusers import AutoencoderKL


lowercase__											=version.parse(version.parse(torch.__version__).base_version) < version.parse('1.11')






def      UpperCamelCase_				(      A__  ,      A__  ,      A__  ,      A__  ,      A__  ,      A__  ,      A__  ,      A__=False  ,      ):
	output_path.parent.mkdir(parents=A__  ,      exist_ok=A__						)
	# PyTorch deprecated the `enable_onnx_checker` and `use_external_data_format` arguments in v1.11,
	# so we check the torch version for backwards compatibility
	if is_torch_less_than_1_11:
		export(
		    A__  ,      A__  ,      f=output_path.as_posix()  ,      input_names=A__  ,      output_names=A__  ,      dynamic_axes=A__  ,      do_constant_folding=A__  ,      use_external_data_format=A__  ,      enable_onnx_checker=A__  ,      opset_version=A__  ,      )
	else:
		export(
		    A__  ,      A__  ,      f=output_path.as_posix()  ,      input_names=A__  ,      output_names=A__  ,      dynamic_axes=A__  ,      do_constant_folding=A__  ,      opset_version=A__  ,      )






@torch.no_grad()
def      UpperCamelCase_				(      A__  ,      A__  ,      A__  ,      A__ = False						):
	a_      				=						torch.floataa if fpaa else torch.floataa
	if fpaa and torch.cuda.is_available():
		a_      				=						"""cuda"""
	elif fpaa and not torch.cuda.is_available():
		raise ValueError("""`float16` model export is only supported on GPUs with CUDA"""						)
	else:
		a_      				=						"""cpu"""
	a_      				=						Path(A__						)

	# VAE DECODER
	a_      				=						AutoencoderKL.from_pretrained(model_path + """/vae"""						)
	a_      				=						vae_decoder.config.latent_channels
	# forward only through the decoder part
	a_      				=						vae_decoder.decode
	onnx_export(
	    A__  ,      model_args=(
	        torch.randn(1  ,      A__  ,      25  ,      25						).to(device=A__  ,      dtype=A__						),
	        False,
	    )  ,      output_path=output_path / """vae_decoder""" / """model.onnx"""  ,      ordered_input_names=["""latent_sample""", """return_dict"""]  ,      output_names=["""sample"""]  ,      dynamic_axes={
	        """latent_sample""": {0: """batch""", 1: """channels""", 2: """height""", 3: """width"""},
	    }  ,      opset=A__  ,      )
	del vae_decoder


if __name__ == "__main__":
		lowercase__											=argparse.ArgumentParser()

		parser.add_argument(
		    '--model_path',
		    type=str,
		    required=True,
		    help='Path to the `diffusers` checkpoint to convert (either a local directory or on the Hub).',
		)

		parser.add_argument('--output_path', type=str, required=True, help='Path to the output model.')
		parser.add_argument(
		    '--opset',
		    default=14,
		    type=int,
		    help='The version of the ONNX operator set to use.',
		)
		parser.add_argument('--fp16', action='store_true', default=False, help='Export the models in `float16` mode')

		lowercase__											=parser.parse_args()
		print(args.output_path)
		convert_models(args.model_path, args.output_path, args.opset, args.fpaa)
		print('SD: Done: ONNX')
