




"""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


_lowerCamelCase  :			List[str]        =       version.parse(version.parse(torch.__version__).base_version) < version.parse('1.11')

def 		lowercase_      (		_UpperCAmelCase  ,      _UpperCAmelCase  ,      _UpperCAmelCase  ,      _UpperCAmelCase  ,      _UpperCAmelCase  ,      _UpperCAmelCase  ,      _UpperCAmelCase  ,      _UpperCAmelCase=False  ,      ):




	"""simple docstring"""






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

@torch.no_grad()
def 		lowercase_      (		_UpperCAmelCase  ,      _UpperCAmelCase  ,      _UpperCAmelCase  ,      _UpperCAmelCase = False	):




	"""simple docstring"""






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

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

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

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

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

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

	A_  :				Dict	   =		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=_UpperCAmelCase  ,      feature_extractor=_UpperCAmelCase  ,      requires_safety_checker=safety_checker is not None  ,      )

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

	del pipeline
	del onnx_pipeline
	A_  :				Any	   =		OnnxStableDiffusionPipeline.from_pretrained(_UpperCAmelCase  ,      provider='''CPUExecutionProvider'''	)
	print('''ONNX pipeline is loadable'''	)


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

		_lowerCamelCase  :			int        =       parser.parse_args()

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











"""simple docstring"""

from typing import Any

import numpy as np

def 		lowercase_      (		_UpperCAmelCase	):




	"""simple docstring"""






	return np.array_equal(_UpperCAmelCase  ,      matrix.conjugate().T	)

def 		lowercase_      (		_UpperCAmelCase  ,      _UpperCAmelCase	):




	"""simple docstring"""






	A_  :				Tuple	   =		v.conjugate().T
	A_  :				Tuple	   =		v_star.dot(_UpperCAmelCase	)
	assert isinstance(_UpperCAmelCase  ,      np.ndarray	)
	return (v_star_dot.dot(_UpperCAmelCase	)) / (v_star.dot(_UpperCAmelCase	))

def 		lowercase_      (		):




	"""simple docstring"""






	A_  :				Optional[Any]	   =		np.array([[2, 2 + 1J, 4], [2 - 1J, 3, 1J], [4, -1J, 1]]	)
	A_  :				str	   =		np.array([[1], [2], [3]]	)
	assert is_hermitian(_UpperCAmelCase	), f"""{a} is not hermitian."""
	print(rayleigh_quotient(_UpperCAmelCase  ,      _UpperCAmelCase	)	)

	A_  :				int	   =		np.array([[1, 2, 4], [2, 3, -1], [4, -1, 1]]	)
	assert is_hermitian(_UpperCAmelCase	), f"""{a} is not hermitian."""
	assert rayleigh_quotient(_UpperCAmelCase  ,      _UpperCAmelCase	) == float(3	)


if __name__ == "__main__":
		import doctest

		doctest.testmod()
		tests()





