

import gc
import unittest

import torch
from parameterized import parameterized

from diffusers import AutoencoderKL
from diffusers.utils import floats_tensor, load_hf_numpy, require_torch_gpu, slow, torch_all_close, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


enable_full_determinism()







class 			lowercase    (	lowercase_     ,  lowercase_     ,  unittest.TestCase      ):
				__SCREAMING_SNAKE_CASE :			Dict	      =      AutoencoderKL
				__SCREAMING_SNAKE_CASE :			List[str]	      =      '''sample'''
				__SCREAMING_SNAKE_CASE :			Any	      =      1e-2




				@property
				def    a							( self	):
											snake_case_          =   4
											snake_case_          =   3
											snake_case_          =   (32, 32)

											snake_case_          =   floats_tensor((batch_size, num_channels) + sizes	).to(snake_case	)

											return {"sample": image}




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




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




				def    a							( self	):
											snake_case_          =   {
											    'block_out_channels': [32, 64],
											    'in_channels': 3,
											    'out_channels': 3,
											    'down_block_types': ['DownEncoderBlock2D', 'DownEncoderBlock2D'],
											    'up_block_types': ['UpDecoderBlock2D', 'UpDecoderBlock2D'],
											    'latent_channels': 4,
											}
											snake_case_          =   self.dummy_input
											return init_dict, inputs_dict




				def    a							( self	):
											pass




				def    a							( self	):
											pass




				@unittest.skipIf(torch_device == 'mps'				,       'Gradient checkpointing skipped on MPS'	)
				def    a							( self	):
											# enable deterministic behavior for gradient checkpointing
											snake_case_    ,   snake_case_          =   self.prepare_init_args_and_inputs_for_common()
											snake_case_          =   self.model_class(**snake_case	)
											model.to(snake_case	)

											assert not model.is_gradient_checkpointing and model.training

											snake_case_          =   model(**snake_case	).sample
											# run the backwards pass on the model. For backwards pass, for simplicity purpose,
											# we won't calculate the loss and rather backprop on out.sum()
											model.zero_grad()

											snake_case_          =   torch.randn_like(snake_case	)
											snake_case_          =   (out - labels).mean()
											loss.backward()

											# re-instantiate the model now enabling gradient checkpointing
											snake_case_          =   self.model_class(**snake_case	)
											# clone model
											model_a.load_state_dict(model.state_dict()	)
											model_a.to(snake_case	)
											model_a.enable_gradient_checkpointing()

											assert model_a.is_gradient_checkpointing and model_a.training

											snake_case_          =   model_a(**snake_case	).sample
											# run the backwards pass on the model. For backwards pass, for simplicity purpose,
											# we won't calculate the loss and rather backprop on out.sum()
											model_a.zero_grad()
											snake_case_          =   (out_a - labels).mean()
											loss_a.backward()

											# compare the output and parameters gradients
											self.assertTrue((loss - loss_a).abs() < 1e-5	)
											snake_case_          =   dict(model.named_parameters()	)
											snake_case_          =   dict(model_a.named_parameters()	)
											for name, param in named_params.items():
																		self.assertTrue(torch_all_close(param.grad.data				,       named_params_a[name].grad.data				,       atol=5e-5	)	)




				def    a							( self	):
											snake_case_    ,   snake_case_          =   AutoencoderKL.from_pretrained('fusing/autoencoder-kl-dummy'				,       output_loading_info=snake_case	)
											self.assertIsNotNone(snake_case	)
											self.assertEqual(len(loading_info['missing_keys']	)				,       0	)

											model.to(snake_case	)
											snake_case_          =   model(**self.dummy_input	)

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




				def    a							( self	):
											snake_case_          =   AutoencoderKL.from_pretrained('fusing/autoencoder-kl-dummy'	)
											snake_case_          =   model.to(snake_case	)
											model.eval()

											if torch_device == "mps":
																		snake_case_          =   torch.manual_seed(0	)
											else:
																		snake_case_          =   torch.Generator(device=snake_case	).manual_seed(0	)

											snake_case_          =   torch.randn(
											    1				,       model.config.in_channels				,       model.config.sample_size				,       model.config.sample_size				,       generator=torch.manual_seed(0	)				,       )
											snake_case_          =   image.to(snake_case	)
											with torch.no_grad():
																		snake_case_          =   model(snake_case				,       sample_posterior=snake_case				,       generator=snake_case	).sample

											snake_case_          =   output[0, -1, -3:, -3:].flatten().cpu()

											# Since the VAE Gaussian prior's generator is seeded on the appropriate device,
											# the expected output slices are not the same for CPU and GPU.
											if torch_device == "mps":
																		snake_case_          =   torch.tensor(
																		    [
																		        -4.0_0_7_8e-0_1,
																		        -3.8_3_2_3e-0_4,
																		        -1.2_6_8_1e-0_1,
																		        -1.1_4_6_2e-0_1,
																		        2.0_0_9_5e-0_1,
																		        1.0_8_9_3e-0_1,
																		        -8.8_2_4_7e-0_2,
																		        -3.0_3_6_1e-0_1,
																		        -9.8_6_4_4e-0_3,
																		    ]	)
											elif torch_device == "cpu":
																		snake_case_          =   torch.tensor(
																		    [-0.13_52, 0.08_78, 0.04_19, -0.08_18, -0.10_69, 0.06_88, -0.14_58, -0.44_46, -0.00_26]	)
											else:
																		snake_case_          =   torch.tensor(
																		    [-0.24_21, 0.46_42, 0.25_07, -0.04_38, 0.06_82, 0.31_60, -0.20_18, -0.07_27, 0.24_85]	)

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







@slow
class 			lowercase    (	unittest.TestCase      ):




				def    a							( self				,       snake_case				,       snake_case	):
											return F'''gaussian_noise_s={seed}_shape={"_".join([str(snake_case	) for s in shape]	)}.npy'''




				def    a							( self	):
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()
											torch.cuda.empty_cache()




				def    a							( self				,       snake_case=0				,       snake_case=(4, 3, 512, 512)				,       snake_case=False	):
											snake_case_          =   torch.floataa if fpaa else torch.floataa
											snake_case_          =   torch.from_numpy(load_hf_numpy(self.get_file_format(snake_case				,       snake_case	)	)	).to(snake_case	).to(snake_case	)
											return image




				def    a							( self				,       snake_case="CompVis/stable-diffusion-v1-4"				,       snake_case=False	):
											snake_case_          =   'fp16' if fpaa else None
											snake_case_          =   torch.floataa if fpaa else torch.floataa

											snake_case_          =   AutoencoderKL.from_pretrained(
											    snake_case				,       subfolder='vae'				,       torch_dtype=snake_case				,       revision=snake_case				,       )
											model.to(snake_case	).eval()

											return model




				def    a							( self				,       snake_case=0	):
											if torch_device == "mps":
																		return torch.manual_seed(snake_case	)
											return torch.Generator(device=snake_case	).manual_seed(snake_case	)




				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.16_03, 0.98_78, -0.04_95, -0.07_90, -0.27_09, 0.83_75, -0.20_60, -0.08_24], [-0.23_95, 0.00_98, 0.01_02, -0.07_09, -0.28_40, -0.02_74, -0.07_18, -0.18_24]],
				        [47, [-0.23_76, 0.11_68, 0.13_32, -0.48_40, -0.25_08, -0.07_91, -0.04_93, -0.40_89], [0.03_50, 0.08_47, 0.04_67, 0.03_44, -0.08_42, -0.05_47, -0.06_33, -0.11_31]],
				        # fmt: on
				    ]	)
				def    a							( self				,       snake_case				,       snake_case				,       snake_case	):
											snake_case_          =   self.get_sd_vae_model()
											snake_case_          =   self.get_sd_image(snake_case	)
											snake_case_          =   self.get_generator(snake_case	)

											with torch.no_grad():
																		snake_case_          =   model(snake_case				,       generator=snake_case				,       sample_posterior=snake_case	).sample

											assert sample.shape == image.shape

											snake_case_          =   sample[-1, -2:, -2:, :2].flatten().float().cpu()
											snake_case_          =   torch.tensor(expected_slice_mps if torch_device == 'mps' else expected_slice	)

											assert torch_all_close(snake_case				,       snake_case				,       atol=3e-3	)




				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.05_13, 0.02_89, 1.37_99, 0.21_66, -0.25_73, -0.08_71, 0.51_03, -0.09_99]],
				        [47, [-0.41_28, -0.13_20, -0.37_04, 0.19_65, -0.41_16, -0.23_32, -0.33_40, 0.22_47]],
				        # fmt: on
				    ]	)
				@require_torch_gpu
				def    a							( self				,       snake_case				,       snake_case	):
											snake_case_          =   self.get_sd_vae_model(fpaa=snake_case	)
											snake_case_          =   self.get_sd_image(snake_case				,       fpaa=snake_case	)
											snake_case_          =   self.get_generator(snake_case	)

											with torch.no_grad():
																		snake_case_          =   model(snake_case				,       generator=snake_case				,       sample_posterior=snake_case	).sample

											assert sample.shape == image.shape

											snake_case_          =   sample[-1, -2:, :2, -2:].flatten().float().cpu()
											snake_case_          =   torch.tensor(snake_case	)

											assert torch_all_close(snake_case				,       snake_case				,       atol=1e-2	)




				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.16_09, 0.98_66, -0.04_87, -0.07_77, -0.27_16, 0.83_68, -0.20_55, -0.08_14], [-0.23_95, 0.00_98, 0.01_02, -0.07_09, -0.28_40, -0.02_74, -0.07_18, -0.18_24]],
				        [47, [-0.23_77, 0.11_47, 0.13_33, -0.48_41, -0.25_06, -0.08_05, -0.04_91, -0.40_85], [0.03_50, 0.08_47, 0.04_67, 0.03_44, -0.08_42, -0.05_47, -0.06_33, -0.11_31]],
				        # fmt: on
				    ]	)
				def    a							( self				,       snake_case				,       snake_case				,       snake_case	):
											snake_case_          =   self.get_sd_vae_model()
											snake_case_          =   self.get_sd_image(snake_case	)

											with torch.no_grad():
																		snake_case_          =   model(snake_case	).sample

											assert sample.shape == image.shape

											snake_case_          =   sample[-1, -2:, -2:, :2].flatten().float().cpu()
											snake_case_          =   torch.tensor(expected_slice_mps if torch_device == 'mps' else expected_slice	)

											assert torch_all_close(snake_case				,       snake_case				,       atol=3e-3	)




				@parameterized.expand(
				    [
				        # fmt: off
				        [13, [-0.20_51, -0.18_03, -0.23_11, -0.21_14, -0.32_92, -0.35_74, -0.29_53, -0.33_23]],
				        [37, [-0.26_32, -0.26_25, -0.21_99, -0.27_41, -0.45_39, -0.49_90, -0.37_20, -0.49_25]],
				        # fmt: on
				    ]	)
				@require_torch_gpu
				def    a							( self				,       snake_case				,       snake_case	):
											snake_case_          =   self.get_sd_vae_model()
											snake_case_          =   self.get_sd_image(snake_case				,       shape=(3, 4, 64, 64)	)

											with torch.no_grad():
																		snake_case_          =   model.decode(snake_case	).sample

											assert list(sample.shape	) == [3, 3, 512, 512]

											snake_case_          =   sample[-1, -2:, :2, -2:].flatten().cpu()
											snake_case_          =   torch.tensor(snake_case	)

											assert torch_all_close(snake_case				,       snake_case				,       atol=1e-3	)




				@parameterized.expand(
				    [
				        # fmt: off
				        [27, [-0.03_69, 0.02_07, -0.07_76, -0.06_82, -0.17_47, -0.19_30, -0.14_65, -0.20_39]],
				        [16, [-0.16_28, -0.21_34, -0.27_47, -0.26_42, -0.37_74, -0.44_04, -0.36_87, -0.42_77]],
				        # fmt: on
				    ]	)
				@require_torch_gpu
				def    a							( self				,       snake_case				,       snake_case	):
											snake_case_          =   self.get_sd_vae_model(fpaa=snake_case	)
											snake_case_          =   self.get_sd_image(snake_case				,       shape=(3, 4, 64, 64)				,       fpaa=snake_case	)

											with torch.no_grad():
																		snake_case_          =   model.decode(snake_case	).sample

											assert list(sample.shape	) == [3, 3, 512, 512]

											snake_case_          =   sample[-1, -2:, :2, -2:].flatten().float().cpu()
											snake_case_          =   torch.tensor(snake_case	)

											assert torch_all_close(snake_case				,       snake_case				,       atol=5e-3	)




				@parameterized.expand([(13,), (16,), (27,)]	)
				@require_torch_gpu
				@unittest.skipIf(not is_xformers_available()				,       reason='xformers is not required when using PyTorch 2.0.'	)
				def    a							( self				,       snake_case	):
											snake_case_          =   self.get_sd_vae_model(fpaa=snake_case	)
											snake_case_          =   self.get_sd_image(snake_case				,       shape=(3, 4, 64, 64)				,       fpaa=snake_case	)

											with torch.no_grad():
																		snake_case_          =   model.decode(snake_case	).sample

											model.enable_xformers_memory_efficient_attention()
											with torch.no_grad():
																		snake_case_          =   model.decode(snake_case	).sample

											assert list(sample.shape	) == [3, 3, 512, 512]

											assert torch_all_close(snake_case				,       snake_case				,       atol=1e-1	)




				@parameterized.expand([(13,), (16,), (37,)]	)
				@require_torch_gpu
				@unittest.skipIf(not is_xformers_available()				,       reason='xformers is not required when using PyTorch 2.0.'	)
				def    a							( self				,       snake_case	):
											snake_case_          =   self.get_sd_vae_model()
											snake_case_          =   self.get_sd_image(snake_case				,       shape=(3, 4, 64, 64)	)

											with torch.no_grad():
																		snake_case_          =   model.decode(snake_case	).sample

											model.enable_xformers_memory_efficient_attention()
											with torch.no_grad():
																		snake_case_          =   model.decode(snake_case	).sample

											assert list(sample.shape	) == [3, 3, 512, 512]

											assert torch_all_close(snake_case				,       snake_case				,       atol=1e-2	)




				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.30_01, 0.09_18, -2.69_84, -3.97_20, -3.20_99, -5.03_53, 1.73_38, -0.20_65, 3.42_67]],
				        [47, [-1.50_30, -4.38_71, -6.03_55, -9.11_57, -1.66_61, -2.78_53, 2.16_07, -5.08_23, 2.56_33]],
				        # fmt: on
				    ]	)
				def    a							( self				,       snake_case				,       snake_case	):
											snake_case_          =   self.get_sd_vae_model()
											snake_case_          =   self.get_sd_image(snake_case	)
											snake_case_          =   self.get_generator(snake_case	)

											with torch.no_grad():
																		snake_case_          =   model.encode(snake_case	).latent_dist
																		snake_case_          =   dist.sample(generator=snake_case	)

											assert list(sample.shape	) == [image.shape[0], 4] + [i // 8 for i in image.shape[2:]]

											snake_case_          =   sample[0, -1, -3:, -3:].flatten().cpu()
											snake_case_          =   torch.tensor(snake_case	)

											snake_case_          =   3e-3 if torch_device != 'mps' else 1e-2
											assert torch_all_close(snake_case				,       snake_case				,       atol=snake_case	)







def  __lowerCamelCase     (			UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__				):







							'''simple docstring'''






							global f  # a global dp table for knapsack
							if f[i][j] < 0:
														if j < wt[i - 1]:
																					snake_case_          =   mf_knapsack(i - 1       ,							UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__				)
														else:
																					snake_case_          =   max(
																					    mf_knapsack(i - 1       ,							UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__				)       ,							mf_knapsack(i - 1       ,							UpperCamelCase__       ,							UpperCamelCase__       ,							j - wt[i - 1]				) + val[i - 1]       ,							)
														snake_case_          =   val
							return f[i][j]





def  __lowerCamelCase     (			UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__				):







							'''simple docstring'''






							snake_case_          =   [[0] * (w + 1) for _ in range(n + 1				)]

							for i in range(1       ,							n + 1				):
														for w_ in range(1       ,							w + 1				):
																					if wt[i - 1] <= w_:
																												snake_case_          =   max(val[i - 1] + dp[i - 1][w_ - wt[i - 1]]       ,							dp[i - 1][w_]				)
																					else:
																												snake_case_          =   dp[i - 1][w_]

							return dp[n][w_], dp





def  __lowerCamelCase     (			UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__				):







							'''simple docstring'''






							if not (isinstance(UpperCamelCase__       ,							(list, tuple)				) and isinstance(UpperCamelCase__       ,							(list, tuple)				)):
														raise ValueError(
														    'Both the weights and values vectors must be either lists or tuples'				)

							snake_case_          =   len(UpperCamelCase__				)
							if num_items != len(UpperCamelCase__				):
														snake_case_          =   (
														    'The number of weights must be the same as the number of values.\n'
														    F'''But got {num_items} weights and {len(UpperCamelCase__				)} values'''
														)
														raise ValueError(UpperCamelCase__				)
							for i in range(UpperCamelCase__				):
														if not isinstance(wt[i]       ,							UpperCamelCase__				):
																					snake_case_          =   (
																					    'All weights must be integers but got weight of '
																					    F'''type {type(wt[i]				)} at index {i}'''
																					)
																					raise TypeError(UpperCamelCase__				)

							snake_case_    ,   snake_case_          =   knapsack(UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__				)
							snake_case_          =   set()
							_construct_solution(UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__				)

							return optimal_val, example_optional_set





def  __lowerCamelCase     (			UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__       ,							UpperCamelCase__				):







							'''simple docstring'''






							if i > 0 and j > 0:
														if dp[i - 1][j] == dp[i][j]:
																					_construct_solution(UpperCamelCase__       ,							UpperCamelCase__       ,							i - 1       ,							UpperCamelCase__       ,							UpperCamelCase__				)
														else:
																					optimal_set.add(UpperCamelCase__				)
																					_construct_solution(UpperCamelCase__       ,							UpperCamelCase__       ,							i - 1       ,							j - wt[i - 1]       ,							UpperCamelCase__				)


if __name__ == "__main__":
					_UpperCAmelCase		:       int						    =	[3, 2, 4, 4]
					_UpperCAmelCase		:       Tuple						    =	[4, 3, 2, 3]
					_UpperCAmelCase		:       Dict						    =	4
					_UpperCAmelCase		:       int						    =	6
					_UpperCAmelCase		:       Union[str, Any]						    =	[[0] * (w + 1)] + [[0] + [-1] * (w + 1) for _ in range(n + 1)]
					_UpperCAmelCase    ,						_UpperCAmelCase		:       Optional[int]						    =	knapsack(w, wt, val, n)
					print(optimal_solution)
					print(mf_knapsack(n, wt, val, w))  # switched the n and w

					# testing the dynamic programming problem with example
					# the optimal subset for the above example are items 3 and 4
					_UpperCAmelCase    ,						_UpperCAmelCase		:       Optional[int]						    =	knapsack_with_example_solution(w, wt, val)
					assert optimal_solution == 8
					assert optimal_subset == {3, 4}
					print("""optimal_value = """, optimal_solution)
					print("""An optimal subset corresponding to the optimal value""", optimal_subset)




