
import argparse
import os
import shutil
from pathlib import Path

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

from diffusers import OnnxRuntimeModel, OnnxStableDiffusionPipeline, StableDiffusionPipeline


lowercase__		:Optional[int]   =    version.parse(version.parse(torch.__version__).base_version) < version.parse("1.11")



def        UpperCamelCase    (							lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__=False ,			):
					'''simple docstring'''
					output_path.parent.mkdir(parents=lowerCAmelCase__ ,			exist_ok=lowerCAmelCase__		)
					# 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(
										    lowerCAmelCase__ ,			lowerCAmelCase__ ,			f=output_path.as_posix() ,			input_names=lowerCAmelCase__ ,			output_names=lowerCAmelCase__ ,			dynamic_axes=lowerCAmelCase__ ,			do_constant_folding=lowerCAmelCase__ ,			use_external_data_format=lowerCAmelCase__ ,			enable_onnx_checker=lowerCAmelCase__ ,			opset_version=lowerCAmelCase__ ,			)
					else:
										export(
										    lowerCAmelCase__ ,			lowerCAmelCase__ ,			f=output_path.as_posix() ,			input_names=lowerCAmelCase__ ,			output_names=lowerCAmelCase__ ,			dynamic_axes=lowerCAmelCase__ ,			do_constant_folding=lowerCAmelCase__ ,			opset_version=lowerCAmelCase__ ,			)



@torch.no_grad()
def        UpperCamelCase    (							lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__ = False		):
					'''simple docstring'''
					lowercase							=						torch.floataa if fpaa else torch.floataa
					if fpaa and torch.cuda.is_available():
										lowercase							=						'''cuda'''
					elif fpaa and not torch.cuda.is_available():
										raise ValueError('''`float16` model export is only supported on GPUs with CUDA'''		)
					else:
										lowercase							=						'''cpu'''
					lowercase							=						StableDiffusionPipeline.from_pretrained(lowerCAmelCase__ ,			torch_dtype=lowerCAmelCase__		).to(lowerCAmelCase__		)
					lowercase							=						Path(lowerCAmelCase__		)

					# TEXT ENCODER
					lowercase							=						pipeline.text_encoder.config.max_position_embeddings
					lowercase							=						pipeline.text_encoder.config.hidden_size
					lowercase							=						pipeline.tokenizer(
					    '''A sample prompt''' ,			padding='''max_length''' ,			max_length=pipeline.tokenizer.model_max_length ,			truncation=lowerCAmelCase__ ,			return_tensors='''pt''' ,			)
					onnx_export(
					    pipeline.text_encoder ,			model_args=(text_input.input_ids.to(device=lowerCAmelCase__ ,			dtype=torch.intaa		)) ,			output_path=output_path / '''text_encoder''' / '''model.onnx''' ,			ordered_input_names=['''input_ids'''] ,			output_names=['''last_hidden_state''', '''pooler_output'''] ,			dynamic_axes={
					        '''input_ids''': {0: '''batch''', 1: '''sequence'''},
					    } ,			opset=lowerCAmelCase__ ,			)
					del pipeline.text_encoder

					# UNET
					lowercase							=						pipeline.unet.config.in_channels
					lowercase							=						pipeline.unet.config.sample_size
					lowercase							=						output_path / '''unet''' / '''model.onnx'''
					onnx_export(
					    pipeline.unet ,			model_args=(
					        torch.randn(2 ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__		).to(device=lowerCAmelCase__ ,			dtype=lowerCAmelCase__		),
					        torch.randn(2		).to(device=lowerCAmelCase__ ,			dtype=lowerCAmelCase__		),
					        torch.randn(2 ,			lowerCAmelCase__ ,			lowerCAmelCase__		).to(device=lowerCAmelCase__ ,			dtype=lowerCAmelCase__		),
					        False,
					    ) ,			output_path=lowerCAmelCase__ ,			ordered_input_names=['''sample''', '''timestep''', '''encoder_hidden_states''', '''return_dict'''] ,			output_names=['''out_sample'''] ,			dynamic_axes={
					        '''sample''': {0: '''batch''', 1: '''channels''', 2: '''height''', 3: '''width'''},
					        '''timestep''': {0: '''batch'''},
					        '''encoder_hidden_states''': {0: '''batch''', 1: '''sequence'''},
					    } ,			opset=lowerCAmelCase__ ,			use_external_data_format=lowerCAmelCase__ ,			)
					lowercase							=						str(unet_path.absolute().as_posix()		)
					lowercase							=						os.path.dirname(lowerCAmelCase__		)
					lowercase							=						onnx.load(lowerCAmelCase__		)
					# clean up existing tensor files
					shutil.rmtree(lowerCAmelCase__		)
					os.mkdir(lowerCAmelCase__		)
					# collate external tensor files into one
					onnx.save_model(
					    lowerCAmelCase__ ,			lowerCAmelCase__ ,			save_as_external_data=lowerCAmelCase__ ,			all_tensors_to_one_file=lowerCAmelCase__ ,			location='''weights.pb''' ,			convert_attribute=lowerCAmelCase__ ,			)
					del pipeline.unet

					# VAE ENCODER
					lowercase							=						pipeline.vae
					lowercase							=						vae_encoder.config.in_channels
					lowercase							=						vae_encoder.config.sample_size
					# need to get the raw tensor output (sample) from the encoder
					lowercase							=						lambda lowerCAmelCase__ ,			lowerCAmelCase__		: vae_encoder.encode(lowerCAmelCase__ ,			lowerCAmelCase__		)[0].sample()
					onnx_export(
					    lowerCAmelCase__ ,			model_args=(
					        torch.randn(1 ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__		).to(device=lowerCAmelCase__ ,			dtype=lowerCAmelCase__		),
					        False,
					    ) ,			output_path=output_path / '''vae_encoder''' / '''model.onnx''' ,			ordered_input_names=['''sample''', '''return_dict'''] ,			output_names=['''latent_sample'''] ,			dynamic_axes={
					        '''sample''': {0: '''batch''', 1: '''channels''', 2: '''height''', 3: '''width'''},
					    } ,			opset=lowerCAmelCase__ ,			)

					# VAE DECODER
					lowercase							=						pipeline.vae
					lowercase							=						vae_decoder.config.latent_channels
					lowercase							=						vae_decoder.config.out_channels
					# forward only through the decoder part
					lowercase							=						vae_encoder.decode
					onnx_export(
					    lowerCAmelCase__ ,			model_args=(
					        torch.randn(1 ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__		).to(device=lowerCAmelCase__ ,			dtype=lowerCAmelCase__		),
					        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=lowerCAmelCase__ ,			)
					del pipeline.vae

					# SAFETY CHECKER
					if pipeline.safety_checker is not None:
										lowercase							=						pipeline.safety_checker
										lowercase							=						safety_checker.config.vision_config.num_channels
										lowercase							=						safety_checker.config.vision_config.image_size
										lowercase							=						safety_checker.forward_onnx
										onnx_export(
										    pipeline.safety_checker ,			model_args=(
										        torch.randn(
										            1 ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			).to(device=lowerCAmelCase__ ,			dtype=lowerCAmelCase__		),
										        torch.randn(1 ,			lowerCAmelCase__ ,			lowerCAmelCase__ ,			lowerCAmelCase__		).to(device=lowerCAmelCase__ ,			dtype=lowerCAmelCase__		),
										    ) ,			output_path=output_path / '''safety_checker''' / '''model.onnx''' ,			ordered_input_names=['''clip_input''', '''images'''] ,			output_names=['''out_images''', '''has_nsfw_concepts'''] ,			dynamic_axes={
										        '''clip_input''': {0: '''batch''', 1: '''channels''', 2: '''height''', 3: '''width'''},
										        '''images''': {0: '''batch''', 1: '''height''', 2: '''width''', 3: '''channels'''},
										    } ,			opset=lowerCAmelCase__ ,			)
										del pipeline.safety_checker
										lowercase							=						OnnxRuntimeModel.from_pretrained(output_path / '''safety_checker'''		)
										lowercase							=						pipeline.feature_extractor
					else:
										lowercase							=						None
										lowercase							=						None

					lowercase							=						OnnxStableDiffusionPipeline(
					    vae_encoder=OnnxRuntimeModel.from_pretrained(output_path / '''vae_encoder'''		) ,			vae_decoder=OnnxRuntimeModel.from_pretrained(output_path / '''vae_decoder'''		) ,			text_encoder=OnnxRuntimeModel.from_pretrained(output_path / '''text_encoder'''		) ,			tokenizer=pipeline.tokenizer ,			unet=OnnxRuntimeModel.from_pretrained(output_path / '''unet'''		) ,			scheduler=pipeline.scheduler ,			safety_checker=lowerCAmelCase__ ,			feature_extractor=lowerCAmelCase__ ,			requires_safety_checker=safety_checker is not None ,			)

					onnx_pipeline.save_pretrained(lowerCAmelCase__		)
					print('''ONNX pipeline saved to''' ,			lowerCAmelCase__		)

					del pipeline
					del onnx_pipeline
					lowercase							=						OnnxStableDiffusionPipeline.from_pretrained(lowerCAmelCase__ ,			provider='''CPUExecutionProvider'''		)
					print('''ONNX pipeline is loadable'''		)


if __name__ == "__main__":
							lowercase__		:Optional[int]   =    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__		:List[Any]   =    parser.parse_args()

							convert_models(args.model_path, args.output_path, args.opset, args.fpaa)








snake_case				:    str       				=     '''
# Transformers installation
! pip install transformers datasets
# To install from source instead of the last release, comment the command above and uncomment the following one.
# ! pip install git+https://github.com/huggingface/transformers.git
'''

snake_case				:    List[Any]       				=     [{'''type''': '''code''', '''content''': INSTALL_CONTENT}]
snake_case				:    int       				=     {
    '''{processor_class}''': '''FakeProcessorClass''',
    '''{model_class}''': '''FakeModelClass''',
    '''{object_class}''': '''FakeObjectClass''',
}





