
"""simple docstring"""

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


_A						=			version.parse(version.parse(torch.__version__).base_version) < version.parse('1.11')
def 		UpperCAmelCase (		a_,						a_,						a_,						a_,						a_,						a_,						a_,						a_=False,						):





		'''simple docstring'''


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





		'''simple docstring'''


		lowerCamelCase					:      List[Any]          =   torch.floataa if fpaa else torch.floataa
		if fpaa and torch.cuda.is_available():
				lowerCamelCase					:      Union[str, Any]          =   'cuda'
		elif fpaa and not torch.cuda.is_available():
				raise ValueError('`float16` model export is only supported on GPUs with CUDA'      )
		else:
				lowerCamelCase					:      Union[str, Any]          =   'cpu'
		lowerCamelCase					:      Optional[Any]          =   StableDiffusionPipeline.from_pretrained(a_,						torch_dtype=a_      ).to(a_      )
		lowerCamelCase					:      str          =   Path(a_      )

		# TEXT ENCODER
		lowerCamelCase					:      int          =   pipeline.text_encoder.config.max_position_embeddings
		lowerCamelCase					:      str          =   pipeline.text_encoder.config.hidden_size
		lowerCamelCase					:      Union[str, Any]          =   pipeline.tokenizer(
		    'A sample prompt',						padding='max_length',						max_length=pipeline.tokenizer.model_max_length,						truncation=a_,						return_tensors='pt',						)
		onnx_export(
		    pipeline.text_encoder,						model_args=(text_input.input_ids.to(device=a_,						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=a_,						)
		del pipeline.text_encoder

		# UNET
		lowerCamelCase					:      Any          =   pipeline.unet.config.in_channels
		lowerCamelCase					:      int          =   pipeline.unet.config.sample_size
		lowerCamelCase					:      Optional[int]          =   output_path / 'unet' / 'model.onnx'
		onnx_export(
		    pipeline.unet,						model_args=(
		        torch.randn(2,						a_,						a_,						a_      ).to(device=a_,						dtype=a_      ),
		        torch.randn(2      ).to(device=a_,						dtype=a_      ),
		        torch.randn(2,						a_,						a_      ).to(device=a_,						dtype=a_      ),
		        False,
		    ),						output_path=a_,						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=a_,						use_external_data_format=a_,						)
		lowerCamelCase					:      List[str]          =   str(unet_path.absolute().as_posix()      )
		lowerCamelCase					:      Any          =   os.path.dirname(a_      )
		lowerCamelCase					:      List[str]          =   onnx.load(a_      )
		# clean up existing tensor files
		shutil.rmtree(a_      )
		os.mkdir(a_      )
		# collate external tensor files into one
		onnx.save_model(
		    a_,						a_,						save_as_external_data=a_,						all_tensors_to_one_file=a_,						location='weights.pb',						convert_attribute=a_,						)
		del pipeline.unet

		# VAE ENCODER
		lowerCamelCase					:      Optional[Any]          =   pipeline.vae
		lowerCamelCase					:      List[str]          =   vae_encoder.config.in_channels
		lowerCamelCase					:      Optional[int]          =   vae_encoder.config.sample_size
		# need to get the raw tensor output (sample) from the encoder
		lowerCamelCase					:      str          =   lambda a_,						a_      : vae_encoder.encode(a_,						a_      )[0].sample()
		onnx_export(
		    a_,						model_args=(
		        torch.randn(1,						a_,						a_,						a_      ).to(device=a_,						dtype=a_      ),
		        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=a_,						)

		# VAE DECODER
		lowerCamelCase					:      str          =   pipeline.vae
		lowerCamelCase					:      Tuple          =   vae_decoder.config.latent_channels
		lowerCamelCase					:      Any          =   vae_decoder.config.out_channels
		# forward only through the decoder part
		lowerCamelCase					:      List[str]          =   vae_encoder.decode
		onnx_export(
		    a_,						model_args=(
		        torch.randn(1,						a_,						a_,						a_      ).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 pipeline.vae

		# SAFETY CHECKER
		if pipeline.safety_checker is not None:
				lowerCamelCase					:      Tuple          =   pipeline.safety_checker
				lowerCamelCase					:      int          =   safety_checker.config.vision_config.num_channels
				lowerCamelCase					:      Any          =   safety_checker.config.vision_config.image_size
				lowerCamelCase					:      Tuple          =   safety_checker.forward_onnx
				onnx_export(
				    pipeline.safety_checker,						model_args=(
				        torch.randn(
				            1,						a_,						a_,						a_,						).to(device=a_,						dtype=a_      ),
				        torch.randn(1,						a_,						a_,						a_      ).to(device=a_,						dtype=a_      ),
				    ),						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=a_,						)
				del pipeline.safety_checker
				lowerCamelCase					:      Any          =   OnnxRuntimeModel.from_pretrained(output_path / 'safety_checker'      )
				lowerCamelCase					:      Any          =   pipeline.feature_extractor
		else:
				lowerCamelCase					:      str          =   None
				lowerCamelCase					:      Tuple          =   None

		lowerCamelCase					:      str          =   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=a_,						feature_extractor=a_,						requires_safety_checker=safety_checker is not None,						)

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

		del pipeline
		del onnx_pipeline
		lowerCamelCase					:      Dict          =   OnnxStableDiffusionPipeline.from_pretrained(a_,						provider='CPUExecutionProvider'      )
		print('ONNX pipeline is loadable'      )


if __name__ == "__main__":
					_A						=			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=1_4,
					    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')

					_A						=			parser.parse_args()

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








"""simple docstring"""

import os
import sys
from contextlib import contextmanager


# Windows only
if os.name == "nt":
					import ctypes
					import msvcrt  # noqa






					class 	_lowercase       (    ctypes.Structure		):
						# _fields is a specific attr expected by ctypes
						lowercase_				=     [('size', ctypes.c_int), ('visible', ctypes.c_byte)]
def 		UpperCAmelCase (		):





		'''simple docstring'''


		if os.name == "nt":
				lowerCamelCase					:      Optional[int]          =   CursorInfo()
				lowerCamelCase					:      Union[str, Any]          =   ctypes.windll.kernelaa.GetStdHandle(-11      )
				ctypes.windll.kernelaa.GetConsoleCursorInfo(a_,						ctypes.byref(a_      )      )
				lowerCamelCase					:      Dict          =   False
				ctypes.windll.kernelaa.SetConsoleCursorInfo(a_,						ctypes.byref(a_      )      )
		elif os.name == "posix":
				sys.stdout.write('\033[?25l'      )
				sys.stdout.flush()
def 		UpperCAmelCase (		):





		'''simple docstring'''


		if os.name == "nt":
				lowerCamelCase					:      List[str]          =   CursorInfo()
				lowerCamelCase					:      List[Any]          =   ctypes.windll.kernelaa.GetStdHandle(-11      )
				ctypes.windll.kernelaa.GetConsoleCursorInfo(a_,						ctypes.byref(a_      )      )
				lowerCamelCase					:      Optional[Any]          =   True
				ctypes.windll.kernelaa.SetConsoleCursorInfo(a_,						ctypes.byref(a_      )      )
		elif os.name == "posix":
				sys.stdout.write('\033[?25h'      )
				sys.stdout.flush()





@contextmanager
def 		UpperCAmelCase (		):





		'''simple docstring'''


		try:
				hide_cursor()
				yield
		finally:
				show_cursor()






