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


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


def 				A	( lowercase   ,	lowercase   ,	lowercase   ,	lowercase   ,	lowercase   ,	lowercase   ,	lowercase   ,	lowercase=False   ,	)       ->  Optional[Any]:
		'''simple docstring'''

		output_path.parent.mkdir(parents=lowercase   ,	exist_ok=lowercase			)
		# 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(
				    lowercase   ,	lowercase   ,	f=output_path.as_posix()   ,	input_names=lowercase   ,	output_names=lowercase   ,	dynamic_axes=lowercase   ,	do_constant_folding=lowercase   ,	use_external_data_format=lowercase   ,	enable_onnx_checker=lowercase   ,	opset_version=lowercase   ,	)
		else:
				export(
				    lowercase   ,	lowercase   ,	f=output_path.as_posix()   ,	input_names=lowercase   ,	output_names=lowercase   ,	dynamic_axes=lowercase   ,	do_constant_folding=lowercase   ,	opset_version=lowercase   ,	)


@torch.no_grad()
def 				A	( lowercase   ,	lowercase   ,	lowercase   ,	lowercase = False			)       ->  Any:
		'''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(lowercase   ,	torch_dtype=lowercase			).to(lowercase			)
		UpperCamelCase          =					Path(lowercase			)

		# 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=lowercase   ,	return_tensors='pt'   ,	)
		onnx_export(
		    pipeline.text_encoder   ,	model_args=(text_input.input_ids.to(device=lowercase   ,	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=lowercase   ,	)
		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   ,	lowercase   ,	lowercase   ,	lowercase			).to(device=lowercase   ,	dtype=lowercase			),
		        torch.randn(2			).to(device=lowercase   ,	dtype=lowercase			),
		        torch.randn(2   ,	lowercase   ,	lowercase			).to(device=lowercase   ,	dtype=lowercase			),
		        False,
		    )   ,	output_path=lowercase   ,	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=lowercase   ,	use_external_data_format=lowercase   ,	)
		UpperCamelCase          =					str(unet_path.absolute().as_posix()			)
		UpperCamelCase          =					os.path.dirname(lowercase			)
		UpperCamelCase          =					onnx.load(lowercase			)
		# clean up existing tensor files
		shutil.rmtree(lowercase			)
		os.mkdir(lowercase			)
		# collate external tensor files into one
		onnx.save_model(
		    lowercase   ,	lowercase   ,	save_as_external_data=lowercase   ,	all_tensors_to_one_file=lowercase   ,	location='weights.pb'   ,	convert_attribute=lowercase   ,	)
		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 lowercase   ,	lowercase			: vae_encoder.encode(lowercase   ,	lowercase			)[0].sample()
		onnx_export(
		    lowercase   ,	model_args=(
		        torch.randn(1   ,	lowercase   ,	lowercase   ,	lowercase			).to(device=lowercase   ,	dtype=lowercase			),
		        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=lowercase   ,	)

		# 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(
		    lowercase   ,	model_args=(
		        torch.randn(1   ,	lowercase   ,	lowercase   ,	lowercase			).to(device=lowercase   ,	dtype=lowercase			),
		        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=lowercase   ,	)
		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   ,	lowercase   ,	lowercase   ,	lowercase   ,	).to(device=lowercase   ,	dtype=lowercase			),
				        torch.randn(1   ,	lowercase   ,	lowercase   ,	lowercase			).to(device=lowercase   ,	dtype=lowercase			),
				    )   ,	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=lowercase   ,	)
				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=lowercase   ,	feature_extractor=lowercase   ,	requires_safety_checker=safety_checker is not None   ,	)

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

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


if __name__ == "__main__":
					_UpperCAmelCase :			Any              =	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")

					_UpperCAmelCase :			List[str]              =	parser.parse_args()

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


from __future__ import annotations


def 				A	( lowercase   ,	lowercase			)       ->  tuple[int, int]:
		'''simple docstring'''

		if b == 0:
				return (1, 0)
		((UpperCamelCase)     ,  (UpperCamelCase))          =					extended_euclid(lowercase   ,	a % b			)
		UpperCamelCase          =					a // b
		return (y, x - k * y)


def 				A	( lowercase   ,	lowercase   ,	lowercase   ,	lowercase			)       ->  int:
		'''simple docstring'''

		((UpperCamelCase)     ,  (UpperCamelCase))          =					extended_euclid(lowercase   ,	lowercase			)
		UpperCamelCase          =					na * na
		UpperCamelCase          =					ra * x * na + ra * y * na
		return (n % m + m) % m


def 				A	( lowercase   ,	lowercase			)       ->  int:
		'''simple docstring'''

		((UpperCamelCase)     ,  (UpperCamelCase))          =					extended_euclid(lowercase   ,	lowercase			)
		if b < 0:
				UpperCamelCase          =					(b % n + n) % n
		return b


def 				A	( lowercase   ,	lowercase   ,	lowercase   ,	lowercase			)       ->  int:
		'''simple docstring'''

		UpperCamelCase     ,  UpperCamelCase          =					invert_modulo(lowercase   ,	lowercase			), invert_modulo(lowercase   ,	lowercase			)
		UpperCamelCase          =					na * na
		UpperCamelCase          =					ra * x * na + ra * y * na
		return (n % m + m) % m


if __name__ == "__main__":
					from doctest import testmod

					testmod(name="chinese_remainder_theorem", verbose=True)
					testmod(name="chinese_remainder_theorem2", verbose=True)
					testmod(name="invert_modulo", verbose=True)
					testmod(name="extended_euclid", verbose=True)
