


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


lowerCAmelCase__    :str       =				version.parse(version.parse(torch.__version__).base_version) < version.parse('''1.11''')

def 			lowerCAmelCase__					(     a__:						Dict     ,    a__:						tuple     ,    a__:						Path     ,    a__:						Optional[int]     ,    a__:						int     ,    a__:						Optional[int]     ,    a__:						Optional[int]     ,    a__:						Tuple=False     ,    )    ->						Optional[Any]:


							'''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 			lowerCAmelCase__					(     a__:						str     ,    a__:						str     ,    a__:						int     ,    a__:						bool = False			)    ->						Dict:


							'''simple docstring'''





							_UpperCAmelCase											=     torch.floataa if fpaa else torch.floataa
							if fpaa and torch.cuda.is_available():
														_UpperCAmelCase											=     'cuda'
							elif fpaa and not torch.cuda.is_available():
														raise ValueError('`float16` model export is only supported on GPUs with CUDA'			)
							else:
														_UpperCAmelCase											=     'cpu'
							_UpperCAmelCase											=     StableDiffusionPipeline.from_pretrained(a__     ,    torch_dtype=a__			).to(a__			)
							_UpperCAmelCase											=     Path(a__			)

							# TEXT ENCODER
							_UpperCAmelCase											=     pipeline.text_encoder.config.max_position_embeddings
							_UpperCAmelCase											=     pipeline.text_encoder.config.hidden_size
							_UpperCAmelCase											=     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
							_UpperCAmelCase											=     pipeline.unet.config.in_channels
							_UpperCAmelCase											=     pipeline.unet.config.sample_size
							_UpperCAmelCase											=     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__     ,    )
							_UpperCAmelCase											=     str(unet_path.absolute().as_posix()			)
							_UpperCAmelCase											=     os.path.dirname(a__			)
							_UpperCAmelCase											=     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
							_UpperCAmelCase											=     pipeline.vae
							_UpperCAmelCase											=     vae_encoder.config.in_channels
							_UpperCAmelCase											=     vae_encoder.config.sample_size
							# need to get the raw tensor output (sample) from the encoder
							_UpperCAmelCase											=     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
							_UpperCAmelCase											=     pipeline.vae
							_UpperCAmelCase											=     vae_decoder.config.latent_channels
							_UpperCAmelCase											=     vae_decoder.config.out_channels
							# forward only through the decoder part
							_UpperCAmelCase											=     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:
														_UpperCAmelCase											=     pipeline.safety_checker
														_UpperCAmelCase											=     safety_checker.config.vision_config.num_channels
														_UpperCAmelCase											=     safety_checker.config.vision_config.image_size
														_UpperCAmelCase											=     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
														_UpperCAmelCase											=     OnnxRuntimeModel.from_pretrained(output_path / 'safety_checker'			)
														_UpperCAmelCase											=     pipeline.feature_extractor
							else:
														_UpperCAmelCase											=     None
														_UpperCAmelCase											=     None

							_UpperCAmelCase											=     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
							_UpperCAmelCase											=     OnnxStableDiffusionPipeline.from_pretrained(a__     ,    provider='CPUExecutionProvider'			)
							print('ONNX pipeline is loadable'			)


if __name__ == "__main__":
		lowerCAmelCase__    :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=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''')

		lowerCAmelCase__    :Union[str, Any]       =				parser.parse_args()

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




from .imports import is_rich_available


if is_rich_available():
		from rich.traceback import install

		install(show_locals=False)

else:
		raise ModuleNotFoundError('''To use the rich extension, install rich with `pip install rich`''')
