
import gc
import math
import unittest

import torch

from diffusers import UNetaDModel
from diffusers.utils import floats_tensor, logging, slow, torch_all_close, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


UpperCAmelCase	      = logging.get_logger(__name__)

enable_full_determinism()







class        A_   (			__lowerCamelCase      ,	__lowerCamelCase      ,	unittest.TestCase   ):




	'''simple docstring'''


	_UpperCamelCase     :     Tuple							 =       UNetaDModel
	_UpperCamelCase     :     Dict							 =       """sample"""





	@property
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase       =    4
		lowercase       =    3
		lowercase       =    (32, 32)

		lowercase       =    floats_tensor((batch_size, num_channels) + sizes       ).to(snake_case       )
		lowercase       =    torch.tensor([10]       ).to(snake_case       )

		return {"sample": noise, "timestep": time_step}





	@property
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		return (3, 32, 32)





	@property
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		return (3, 32, 32)





	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase       =    {
		    'block_out_channels': (32, 64),
		    'down_block_types': ('DownBlock2D', 'AttnDownBlock2D'),
		    'up_block_types': ('AttnUpBlock2D', 'UpBlock2D'),
		    'attention_head_dim': 3,
		    'out_channels': 3,
		    'in_channels': 3,
		    'layers_per_block': 2,
		    'sample_size': 32,
		}
		lowercase       =    self.dummy_input
		return init_dict, inputs_dict







class        A_   (			__lowerCamelCase      ,	__lowerCamelCase      ,	unittest.TestCase   ):




	'''simple docstring'''


	_UpperCamelCase     :     Any							 =       UNetaDModel
	_UpperCamelCase     :     Dict							 =       """sample"""





	@property
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase       =    4
		lowercase       =    4
		lowercase       =    (32, 32)

		lowercase       =    floats_tensor((batch_size, num_channels) + sizes       ).to(snake_case       )
		lowercase       =    torch.tensor([10]       ).to(snake_case       )

		return {"sample": noise, "timestep": time_step}





	@property
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		return (4, 32, 32)





	@property
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		return (4, 32, 32)





	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase       =    {
		    'sample_size': 32,
		    'in_channels': 4,
		    'out_channels': 4,
		    'layers_per_block': 2,
		    'block_out_channels': (32, 64),
		    'attention_head_dim': 32,
		    'down_block_types': ('DownBlock2D', 'DownBlock2D'),
		    'up_block_types': ('UpBlock2D', 'UpBlock2D'),
		}
		lowercase       =    self.dummy_input
		return init_dict, inputs_dict





	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase  ,					lowercase       =    UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'		,    output_loading_info=snake_case       )

		self.assertIsNotNone(snake_case       )
		self.assertEqual(len(loading_info['missing_keys']       )		,    0       )

		model.to(snake_case       )
		lowercase       =    model(**self.dummy_input       ).sample

		assert image is not None, "Make sure output is not None"





	@unittest.skipIf(torch_device != 'cuda'		,    'This test is supposed to run on GPU'       )
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase  ,					lowercase       =    UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'		,    output_loading_info=snake_case       )
		model.to(snake_case       )
		lowercase       =    model(**self.dummy_input       ).sample

		assert image is not None, "Make sure output is not None"





	@unittest.skipIf(torch_device != 'cuda'		,    'This test is supposed to run on GPU'       )
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		# by defautl model loading will use accelerate as `low_cpu_mem_usage=True`
		lowercase  ,					lowercase       =    UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'		,    output_loading_info=snake_case       )
		model_accelerate.to(snake_case       )
		model_accelerate.eval()

		lowercase       =    torch.randn(
		    1		,    model_accelerate.config.in_channels		,    model_accelerate.config.sample_size		,    model_accelerate.config.sample_size		,    generator=torch.manual_seed(0       )		,    )
		lowercase       =    noise.to(snake_case       )
		lowercase       =    torch.tensor([10] * noise.shape[0]       ).to(snake_case       )

		lowercase       =    model_accelerate(snake_case		,    snake_case       )['sample']

		# two models don't need to stay in the device at the same time
		del model_accelerate
		torch.cuda.empty_cache()
		gc.collect()

		lowercase  ,					lowercase       =    UNetaDModel.from_pretrained(
		    'fusing/unet-ldm-dummy-update'		,    output_loading_info=snake_case		,    low_cpu_mem_usage=snake_case       )
		model_normal_load.to(snake_case       )
		model_normal_load.eval()
		lowercase       =    model_normal_load(snake_case		,    snake_case       )['sample']

		assert torch_all_close(snake_case		,    snake_case		,    rtol=1E-3       )





	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase       =    UNetaDModel.from_pretrained('fusing/unet-ldm-dummy-update'       )
		model.eval()
		model.to(snake_case       )

		lowercase       =    torch.randn(
		    1		,    model.config.in_channels		,    model.config.sample_size		,    model.config.sample_size		,    generator=torch.manual_seed(0       )		,    )
		lowercase       =    noise.to(snake_case       )
		lowercase       =    torch.tensor([10] * noise.shape[0]       ).to(snake_case       )

		with torch.no_grad():
			lowercase       =    model(snake_case		,    snake_case       ).sample

		lowercase       =    output[0, -1, -3:, -3:].flatten().cpu()
		# fmt: off
		lowercase       =    torch.tensor([-13.3_258, -20.1_100, -15.9_873, -17.6_617, -23.0_596, -17.9_419, -13.3_675, -16.1_889, -12.3_800]       )
		# fmt: on

		self.assertTrue(torch_all_close(snake_case		,    snake_case		,    rtol=1E-3       )       )







class        A_   (			__lowerCamelCase      ,	__lowerCamelCase      ,	unittest.TestCase   ):




	'''simple docstring'''


	_UpperCamelCase     :     Optional[Any]							 =       UNetaDModel
	_UpperCamelCase     :     Dict							 =       """sample"""





	@property
	def 	SCREAMING_SNAKE_CASE__						(       self		,    snake_case=(32, 32)       ):
		lowercase       =    4
		lowercase       =    3

		lowercase       =    floats_tensor((batch_size, num_channels) + sizes       ).to(snake_case       )
		lowercase       =    torch.tensor(batch_size * [10]       ).to(dtype=torch.intaa		,    device=snake_case       )

		return {"sample": noise, "timestep": time_step}





	@property
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		return (3, 32, 32)





	@property
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		return (3, 32, 32)





	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase       =    {
		    'block_out_channels': [32, 64, 64, 64],
		    'in_channels': 3,
		    'layers_per_block': 1,
		    'out_channels': 3,
		    'time_embedding_type': 'fourier',
		    'norm_eps': 1E-6,
		    'mid_block_scale_factor': math.sqrt(2.0       ),
		    'norm_num_groups': None,
		    'down_block_types': [
		        'SkipDownBlock2D',
		        'AttnSkipDownBlock2D',
		        'SkipDownBlock2D',
		        'SkipDownBlock2D',
		    ],
		    'up_block_types': [
		        'SkipUpBlock2D',
		        'SkipUpBlock2D',
		        'AttnSkipUpBlock2D',
		        'SkipUpBlock2D',
		    ],
		}
		lowercase       =    self.dummy_input
		return init_dict, inputs_dict





	@slow
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase  ,					lowercase       =    UNetaDModel.from_pretrained('google/ncsnpp-celebahq-256'		,    output_loading_info=snake_case       )
		self.assertIsNotNone(snake_case       )
		self.assertEqual(len(loading_info['missing_keys']       )		,    0       )

		model.to(snake_case       )
		lowercase       =    self.dummy_input
		lowercase       =    floats_tensor((4, 3) + (256, 256)       ).to(snake_case       )
		lowercase       =    noise
		lowercase       =    model(**snake_case       )

		assert image is not None, "Make sure output is not None"





	@slow
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase       =    UNetaDModel.from_pretrained('google/ncsnpp-celebahq-256'       )
		model.to(snake_case       )

		lowercase       =    4
		lowercase       =    3
		lowercase       =    (256, 256)

		lowercase       =    torch.ones((batch_size, num_channels) + sizes       ).to(snake_case       )
		lowercase       =    torch.tensor(batch_size * [1E-4]       ).to(snake_case       )

		with torch.no_grad():
			lowercase       =    model(snake_case		,    snake_case       ).sample

		lowercase       =    output[0, -3:, -3:, -1].flatten().cpu()
		# fmt: off
		lowercase       =    torch.tensor([-4_842.8_691, -6_499.6_631, -3_800.1_953, -7_978.2_686, -10_980.7_129, -20_028.8_535, 8_148.2_822, 2_342.2_905, 567.7_608]       )
		# fmt: on

		self.assertTrue(torch_all_close(snake_case		,    snake_case		,    rtol=1E-2       )       )





	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase       =    UNetaDModel.from_pretrained('fusing/ncsnpp-ffhq-ve-dummy-update'       )
		model.to(snake_case       )

		lowercase       =    4
		lowercase       =    3
		lowercase       =    (32, 32)

		lowercase       =    torch.ones((batch_size, num_channels) + sizes       ).to(snake_case       )
		lowercase       =    torch.tensor(batch_size * [1E-4]       ).to(snake_case       )

		with torch.no_grad():
			lowercase       =    model(snake_case		,    snake_case       ).sample

		lowercase       =    output[0, -3:, -3:, -1].flatten().cpu()
		# fmt: off
		lowercase       =    torch.tensor([-0.0_325, -0.0_900, -0.0_869, -0.0_332, -0.0_725, -0.0_270, -0.0_101, 0.0_227, 0.0_256]       )
		# fmt: on

		self.assertTrue(torch_all_close(snake_case		,    snake_case		,    rtol=1E-2       )       )





	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		# not required for this model
		pass



import unittest
from pathlib import Path
from tempfile import NamedTemporaryFile, TemporaryDirectory

from transformers import BertConfig, BertTokenizerFast, FeatureExtractionPipeline
from transformers.convert_graph_to_onnx import (
    convert,
    ensure_valid_input,
    generate_identified_filename,
    infer_shapes,
    quantize,
)
from transformers.testing_utils import require_tf, require_tokenizers, require_torch, slow







class        A_   :




	'''simple docstring'''


	def 	SCREAMING_SNAKE_CASE__						(       self		,    snake_case		,    snake_case		,    snake_case       ):
		return None







class        A_   :




	'''simple docstring'''


	def 	SCREAMING_SNAKE_CASE__						(       self		,    snake_case		,    snake_case		,    snake_case		,    snake_case       ):
		return None







class        A_   (			unittest.TestCase   ):




	'''simple docstring'''


	_UpperCamelCase     :     Tuple							 =       [
	    # (model_name, model_kwargs)
	    ("""bert-base-cased""", {}),
	    ("""gpt2""", {"""use_cache""": False}),  # We don't support exporting GPT2 past keys anymore
	]





	@require_tf
	@slow
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
			self._test_export(snake_case		,    'tf'		,    12		,    **snake_case       )





	@require_torch
	@slow
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
			self._test_export(snake_case		,    'pt'		,    12		,    **snake_case       )





	@require_torch
	@slow
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		from transformers import BertModel

		lowercase       =    ['[UNK]', '[SEP]', '[CLS]', '[PAD]', '[MASK]', 'some', 'other', 'words']
		with NamedTemporaryFile(mode='w+t'       ) as vocab_file:
			vocab_file.write('\n'.join(snake_case       )       )
			vocab_file.flush()
			lowercase       =    BertTokenizerFast(vocab_file.name       )

		with TemporaryDirectory() as bert_save_dir:
			lowercase       =    BertModel(BertConfig(vocab_size=len(snake_case       )       )       )
			model.save_pretrained(snake_case       )
			self._test_export(snake_case		,    'pt'		,    12		,    snake_case       )





	@require_tf
	@slow
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
			lowercase       =    self._test_export(snake_case		,    'tf'		,    12		,    **snake_case       )
			lowercase       =    quantize(Path(snake_case       )       )

			# Ensure the actual quantized model is not bigger than the original one
			if quantized_path.stat().st_size >= Path(snake_case       ).stat().st_size:
				self.fail('Quantized model is bigger than initial ONNX model'       )





	@require_torch
	@slow
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		for model, model_kwargs in OnnxExportTestCase.MODEL_TO_TEST:
			lowercase       =    self._test_export(snake_case		,    'pt'		,    12		,    **snake_case       )
			lowercase       =    quantize(snake_case       )

			# Ensure the actual quantized model is not bigger than the original one
			if quantized_path.stat().st_size >= Path(snake_case       ).stat().st_size:
				self.fail('Quantized model is bigger than initial ONNX model'       )





	def 	SCREAMING_SNAKE_CASE__						(       self		,    snake_case		,    snake_case		,    snake_case		,    snake_case=None		,    **snake_case       ):
		try:
			# Compute path
			with TemporaryDirectory() as tempdir:
				lowercase       =    Path(snake_case       ).joinpath('model.onnx'       )

			# Remove folder if exists
			if path.parent.exists():
				path.parent.rmdir()

			# Export
			convert(snake_case		,    snake_case		,    snake_case		,    snake_case		,    snake_case		,    **snake_case       )

			return path
		except Exception as e:
			self.fail(snake_case       )





	@require_torch
	@require_tokenizers
	@slow
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		from transformers import BertModel

		lowercase       =    BertModel(BertConfig.from_pretrained('lysandre/tiny-bert-random'       )       )
		lowercase       =    BertTokenizerFast.from_pretrained('lysandre/tiny-bert-random'       )
		self._test_infer_dynamic_axis(snake_case		,    snake_case		,    'pt'       )





	@require_tf
	@require_tokenizers
	@slow
	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		from transformers import TFBertModel

		lowercase       =    TFBertModel(BertConfig.from_pretrained('lysandre/tiny-bert-random'       )       )
		lowercase       =    BertTokenizerFast.from_pretrained('lysandre/tiny-bert-random'       )
		self._test_infer_dynamic_axis(snake_case		,    snake_case		,    'tf'       )





	def 	SCREAMING_SNAKE_CASE__						(       self		,    snake_case		,    snake_case		,    snake_case       ):
		lowercase       =    FeatureExtractionPipeline(snake_case		,    snake_case       )

		lowercase       =    ['input_ids', 'token_type_ids', 'attention_mask', 'output_0', 'output_1']
		lowercase  ,					lowercase  ,					lowercase  ,					lowercase       =    infer_shapes(snake_case		,    snake_case       )

		# Assert all variables are present
		self.assertEqual(len(snake_case       )		,    len(snake_case       )       )
		self.assertTrue(all(var_name in shapes for var_name in variable_names       )       )
		self.assertSequenceEqual(variable_names[:3]		,    snake_case       )
		self.assertSequenceEqual(variable_names[3:]		,    snake_case       )

		# Assert inputs are {0: batch, 1: sequence}
		for var_name in ["input_ids", "token_type_ids", "attention_mask"]:
			self.assertDictEqual(shapes[var_name]		,    {0: 'batch', 1: 'sequence'}       )

		# Assert outputs are {0: batch, 1: sequence} and {0: batch}
		self.assertDictEqual(shapes['output_0']		,    {0: 'batch', 1: 'sequence'}       )
		self.assertDictEqual(shapes['output_1']		,    {0: 'batch'}       )





	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase       =    ['input_ids', 'attention_mask', 'token_type_ids']
		lowercase       =    {'input_ids': [1, 2, 3, 4], 'attention_mask': [0, 0, 0, 0], 'token_type_ids': [1, 1, 1, 1]}
		lowercase  ,					lowercase       =    ensure_valid_input(FuncContiguousArgs()		,    snake_case		,    snake_case       )

		# Should have exactly the same number of args (all are valid)
		self.assertEqual(len(snake_case       )		,    3       )

		# Should have exactly the same input names
		self.assertEqual(set(snake_case       )		,    set(snake_case       )       )

		# Parameter should be reordered according to their respective place in the function:
		# (input_ids, token_type_ids, attention_mask)
		self.assertEqual(snake_case		,    (tokens['input_ids'], tokens['token_type_ids'], tokens['attention_mask'])       )

		# Generated args are interleaved with another args (for instance parameter "past" in GPT2)
		lowercase  ,					lowercase       =    ensure_valid_input(FuncNonContiguousArgs()		,    snake_case		,    snake_case       )

		# Should have exactly the one arg (all before the one not provided "some_other_args")
		self.assertEqual(len(snake_case       )		,    1       )
		self.assertEqual(len(snake_case       )		,    1       )

		# Should have only "input_ids"
		self.assertEqual(inputs_args[0]		,    tokens['input_ids']       )
		self.assertEqual(ordered_input_names[0]		,    'input_ids'       )





	def 	SCREAMING_SNAKE_CASE__						(       self       ):
		lowercase       =    generate_identified_filename(Path('/home/something/my_fake_model.onnx'       )		,    '-test'       )
		self.assertEqual('/home/something/my_fake_model-test.onnx'		,    generated.as_posix()       )

