


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 							a						(_lowerCAmelCase	,							_lowerCAmelCase	,							unittest.TestCase ):




							"""simple docstring"""


							__UpperCAmelCase       :				str	    =    AutoencoderKL
							__UpperCAmelCase       :				Optional[Any]	    =    "sample"
							__UpperCAmelCase       :				Optional[int]	    =    1e-2

							@property
							def        __snake_case   (				self			:    Dict							)      ->       Optional[Any]:
														__snake_case						:     Optional[Any]									=  4
														__snake_case						:     Tuple									=  3
														__snake_case						:     List[str]									=  (32, 32)

														__snake_case						:     str									=  floats_tensor((batch_size, num_channels) + sizes							).to(lowerCamelCase							)

														return {"sample": image}

							@property
							def        __snake_case   (				self			:    Union[str, Any]							)      ->       Tuple:
														return (3, 32, 32)

							@property
							def        __snake_case   (				self			:    int							)      ->       int:
														return (3, 32, 32)

							def        __snake_case   (				self			:    Optional[Any]							)      ->       Dict:
														__snake_case						:     Optional[Any]									=  {
														    "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						:     Any									=  self.dummy_input
														return init_dict, inputs_dict

							def        __snake_case   (				self			:    str							)      ->       Dict:
														pass

							def        __snake_case   (				self			:    Tuple							)      ->       List[str]:
														pass

							@unittest.skipIf(torch_device == "mps"       ,      "Gradient checkpointing skipped on MPS"							)
							def        __snake_case   (				self			:    Any							)      ->       Optional[Any]:
														# enable deterministic behavior for gradient checkpointing
														__snake_case					,     __snake_case						:     int									=  self.prepare_init_args_and_inputs_for_common()
														__snake_case						:     str									=  self.model_class(**lowerCamelCase							)
														model.to(lowerCamelCase							)

														assert not model.is_gradient_checkpointing and model.training

														__snake_case						:     str									=  model(**lowerCamelCase							).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						:     Any									=  torch.randn_like(lowerCamelCase							)
														__snake_case						:     str									=  (out - labels).mean()
														loss.backward()

														# re-instantiate the model now enabling gradient checkpointing
														__snake_case						:     Optional[int]									=  self.model_class(**lowerCamelCase							)
														# clone model
														model_a.load_state_dict(model.state_dict()							)
														model_a.to(lowerCamelCase							)
														model_a.enable_gradient_checkpointing()

														assert model_a.is_gradient_checkpointing and model_a.training

														__snake_case						:     int									=  model_a(**lowerCamelCase							).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						:     Union[str, Any]									=  (out_a - labels).mean()
														loss_a.backward()

														# compare the output and parameters gradients
														self.assertTrue((loss - loss_a).abs() < 1E-5							)
														__snake_case						:     Optional[int]									=  dict(model.named_parameters()							)
														__snake_case						:     List[Any]									=  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        __snake_case   (				self			:    List[Any]							)      ->       Optional[int]:
														__snake_case					,     __snake_case						:     Optional[Any]									=  AutoencoderKL.from_pretrained("fusing/autoencoder-kl-dummy"       ,      output_loading_info=lowerCamelCase							)
														self.assertIsNotNone(lowerCamelCase							)
														self.assertEqual(len(loading_info["missing_keys"]							)       ,      0							)

														model.to(lowerCamelCase							)
														__snake_case						:     Optional[Any]									=  model(**self.dummy_input							)

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






							def        __snake_case   (				self			:    Optional[Any]							)      ->       Union[str, Any]:
														__snake_case						:     Tuple									=  AutoencoderKL.from_pretrained("fusing/autoencoder-kl-dummy"							)
														__snake_case						:     Dict									=  model.to(lowerCamelCase							)
														model.eval()

														if torch_device == "mps":
																					__snake_case						:     int									=  torch.manual_seed(0							)
														else:
																					__snake_case						:     str									=  torch.Generator(device=lowerCamelCase							).manual_seed(0							)

														__snake_case						:     List[str]									=  torch.randn(
														    1       ,      model.config.in_channels       ,      model.config.sample_size       ,      model.config.sample_size       ,      generator=torch.manual_seed(0							)       ,      )
														__snake_case						:     Union[str, Any]									=  image.to(lowerCamelCase							)
														with torch.no_grad():
																					__snake_case						:     str									=  model(lowerCamelCase       ,      sample_posterior=lowerCamelCase       ,      generator=lowerCamelCase							).sample

														__snake_case						:     List[Any]									=  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						:     Union[str, Any]									=  torch.tensor(
																					    [
																					        -4.0078E-01,
																					        -3.8323E-04,
																					        -1.2681E-01,
																					        -1.1462E-01,
																					        2.0095E-01,
																					        1.0893E-01,
																					        -8.8247E-02,
																					        -3.0361E-01,
																					        -9.8644E-03,
																					    ]							)
														elif torch_device == "cpu":
																					__snake_case						:     Tuple									=  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						:     List[str]									=  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(lowerCamelCase       ,      lowerCamelCase       ,      rtol=1E-2							)							)

@slow
class 							a						(unittest.TestCase ):




							"""simple docstring"""


							def        __snake_case   (				self			:    int       ,      lowerCamelCase			:    Dict       ,      lowerCamelCase			:    Optional[Any]							)      ->       List[str]:
														return F'gaussian_noise_s={seed}_shape={"_".join([str(lowerCamelCase							) for s in shape]							)}.npy'

							def        __snake_case   (				self			:    List[Any]							)      ->       int:
														# clean up the VRAM after each test
														super().tearDown()
														gc.collect()
														torch.cuda.empty_cache()

							def        __snake_case   (				self			:    Tuple       ,      lowerCamelCase			:    List[Any]=0       ,      lowerCamelCase			:    Tuple=(4, 3, 512, 512)       ,      lowerCamelCase			:    Optional[int]=False							)      ->       str:
														__snake_case						:     List[Any]									=  torch.floataa if fpaa else torch.floataa
														__snake_case						:     Tuple									=  torch.from_numpy(load_hf_numpy(self.get_file_format(lowerCamelCase       ,      lowerCamelCase							)							)							).to(lowerCamelCase							).to(lowerCamelCase							)
														return image

							def        __snake_case   (				self			:    Optional[Any]       ,      lowerCamelCase			:    int="CompVis/stable-diffusion-v1-4"       ,      lowerCamelCase			:    int=False							)      ->       int:
														__snake_case						:     str									=  "fp16" if fpaa else None
														__snake_case						:     int									=  torch.floataa if fpaa else torch.floataa

														__snake_case						:     int									=  AutoencoderKL.from_pretrained(
														    lowerCamelCase       ,      subfolder="vae"       ,      torch_dtype=lowerCamelCase       ,      revision=lowerCamelCase       ,      )
														model.to(lowerCamelCase							).eval()

														return model

							def        __snake_case   (				self			:    str       ,      lowerCamelCase			:    int=0							)      ->       Optional[Any]:
														if torch_device == "mps":
																					return torch.manual_seed(lowerCamelCase							)
														return torch.Generator(device=lowerCamelCase							).manual_seed(lowerCamelCase							)

							@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        __snake_case   (				self			:    List[str]       ,      lowerCamelCase			:    Optional[Any]       ,      lowerCamelCase			:    Optional[Any]       ,      lowerCamelCase			:    Optional[Any]							)      ->       List[Any]:
														__snake_case						:     Optional[Any]									=  self.get_sd_vae_model()
														__snake_case						:     List[Any]									=  self.get_sd_image(lowerCamelCase							)
														__snake_case						:     Tuple									=  self.get_generator(lowerCamelCase							)

														with torch.no_grad():
																					__snake_case						:     Optional[Any]									=  model(lowerCamelCase       ,      generator=lowerCamelCase       ,      sample_posterior=lowerCamelCase							).sample

														assert sample.shape == image.shape

														__snake_case						:     List[Any]									=  sample[-1, -2:, -2:, :2].flatten().float().cpu()
														__snake_case						:     int									=  torch.tensor(expected_slice_mps if torch_device == "mps" else expected_slice							)

														assert torch_all_close(lowerCamelCase       ,      lowerCamelCase       ,      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        __snake_case   (				self			:    Any       ,      lowerCamelCase			:    List[str]       ,      lowerCamelCase			:    List[str]							)      ->       Tuple:
														__snake_case						:     Any									=  self.get_sd_vae_model(fpaa=lowerCamelCase							)
														__snake_case						:     List[Any]									=  self.get_sd_image(lowerCamelCase       ,      fpaa=lowerCamelCase							)
														__snake_case						:     List[Any]									=  self.get_generator(lowerCamelCase							)

														with torch.no_grad():
																					__snake_case						:     str									=  model(lowerCamelCase       ,      generator=lowerCamelCase       ,      sample_posterior=lowerCamelCase							).sample

														assert sample.shape == image.shape

														__snake_case						:     Optional[Any]									=  sample[-1, -2:, :2, -2:].flatten().float().cpu()
														__snake_case						:     Any									=  torch.tensor(lowerCamelCase							)

														assert torch_all_close(lowerCamelCase       ,      lowerCamelCase       ,      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        __snake_case   (				self			:    List[Any]       ,      lowerCamelCase			:    List[Any]       ,      lowerCamelCase			:    Any       ,      lowerCamelCase			:    Dict							)      ->       int:
														__snake_case						:     int									=  self.get_sd_vae_model()
														__snake_case						:     List[Any]									=  self.get_sd_image(lowerCamelCase							)

														with torch.no_grad():
																					__snake_case						:     int									=  model(lowerCamelCase							).sample

														assert sample.shape == image.shape

														__snake_case						:     Union[str, Any]									=  sample[-1, -2:, -2:, :2].flatten().float().cpu()
														__snake_case						:     List[str]									=  torch.tensor(expected_slice_mps if torch_device == "mps" else expected_slice							)

														assert torch_all_close(lowerCamelCase       ,      lowerCamelCase       ,      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        __snake_case   (				self			:    List[str]       ,      lowerCamelCase			:    Tuple       ,      lowerCamelCase			:    Any							)      ->       Optional[Any]:
														__snake_case						:     List[str]									=  self.get_sd_vae_model()
														__snake_case						:     List[Any]									=  self.get_sd_image(lowerCamelCase       ,      shape=(3, 4, 64, 64)							)

														with torch.no_grad():
																					__snake_case						:     str									=  model.decode(lowerCamelCase							).sample

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

														__snake_case						:     str									=  sample[-1, -2:, :2, -2:].flatten().cpu()
														__snake_case						:     Optional[int]									=  torch.tensor(lowerCamelCase							)

														assert torch_all_close(lowerCamelCase       ,      lowerCamelCase       ,      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        __snake_case   (				self			:    str       ,      lowerCamelCase			:    Optional[int]       ,      lowerCamelCase			:    Dict							)      ->       int:
														__snake_case						:     int									=  self.get_sd_vae_model(fpaa=lowerCamelCase							)
														__snake_case						:     List[str]									=  self.get_sd_image(lowerCamelCase       ,      shape=(3, 4, 64, 64)       ,      fpaa=lowerCamelCase							)

														with torch.no_grad():
																					__snake_case						:     Union[str, Any]									=  model.decode(lowerCamelCase							).sample

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

														__snake_case						:     Optional[Any]									=  sample[-1, -2:, :2, -2:].flatten().float().cpu()
														__snake_case						:     Optional[Any]									=  torch.tensor(lowerCamelCase							)

														assert torch_all_close(lowerCamelCase       ,      lowerCamelCase       ,      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        __snake_case   (				self			:    Tuple       ,      lowerCamelCase			:    List[Any]							)      ->       Tuple:
														__snake_case						:     Dict									=  self.get_sd_vae_model(fpaa=lowerCamelCase							)
														__snake_case						:     Any									=  self.get_sd_image(lowerCamelCase       ,      shape=(3, 4, 64, 64)       ,      fpaa=lowerCamelCase							)

														with torch.no_grad():
																					__snake_case						:     str									=  model.decode(lowerCamelCase							).sample

														model.enable_xformers_memory_efficient_attention()
														with torch.no_grad():
																					__snake_case						:     Any									=  model.decode(lowerCamelCase							).sample

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

														assert torch_all_close(lowerCamelCase       ,      lowerCamelCase       ,      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        __snake_case   (				self			:    List[Any]       ,      lowerCamelCase			:    Any							)      ->       Optional[int]:
														__snake_case						:     str									=  self.get_sd_vae_model()
														__snake_case						:     Union[str, Any]									=  self.get_sd_image(lowerCamelCase       ,      shape=(3, 4, 64, 64)							)

														with torch.no_grad():
																					__snake_case						:     List[Any]									=  model.decode(lowerCamelCase							).sample

														model.enable_xformers_memory_efficient_attention()
														with torch.no_grad():
																					__snake_case						:     Dict									=  model.decode(lowerCamelCase							).sample

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

														assert torch_all_close(lowerCamelCase       ,      lowerCamelCase       ,      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        __snake_case   (				self			:    List[Any]       ,      lowerCamelCase			:    List[Any]       ,      lowerCamelCase			:    Dict							)      ->       Optional[int]:
														__snake_case						:     str									=  self.get_sd_vae_model()
														__snake_case						:     int									=  self.get_sd_image(lowerCamelCase							)
														__snake_case						:     int									=  self.get_generator(lowerCamelCase							)

														with torch.no_grad():
																					__snake_case						:     Optional[Any]									=  model.encode(lowerCamelCase							).latent_dist
																					__snake_case						:     Dict									=  dist.sample(generator=lowerCamelCase							)

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

														__snake_case						:     List[str]									=  sample[0, -1, -3:, -3:].flatten().cpu()
														__snake_case						:     Dict									=  torch.tensor(lowerCamelCase							)

														__snake_case						:     Dict									=  3E-3 if torch_device != "mps" else 1E-2
														assert torch_all_close(lowerCamelCase       ,      lowerCamelCase       ,      atol=lowerCamelCase							)






from typing import Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
     from PIL import Image

     from ..image_utils import load_image

if is_torch_available():
     from ..models.auto.modeling_auto import MODEL_FOR_VISUAL_QUESTION_ANSWERING_MAPPING

a_      :      str    =     logging.get_logger(__name__)
@add_end_docstrings(_lowercase				)
class    __UpperCamelCase			(       _lowercase				):

 """simple docstring"""



 def __init__(	self		,				*SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      ) ->			int:
  super().__init__(*SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      )
  self.check_model_type(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE=None		,				SCREAMING_SNAKE_CASE=None		,				SCREAMING_SNAKE_CASE=None		,				**SCREAMING_SNAKE_CASE      ) ->			Union[str, Any]:
  a__							,			a__												= {}, {}
  if padding is not None:
   a__												= padding
  if truncation is not None:
   a__												= truncation
  if top_k is not None:
   a__												= top_k
  return preprocess_params, {}, postprocess_params

 def __call__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = None		,				**SCREAMING_SNAKE_CASE      ) ->			Optional[int]:
  if isinstance(SCREAMING_SNAKE_CASE		,				(Image.Image, str)      ) and isinstance(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
   a__												= {'''image''': image, '''question''': question}
  else:
   a__												= image
  a__												= super().__call__(SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      )
  return results

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=False		,				SCREAMING_SNAKE_CASE=False      ) ->			Dict:
  a__												= load_image(inputs['''image''']      )
  a__												= self.tokenizer(
      inputs['''question''']		,				return_tensors=self.framework		,				padding=SCREAMING_SNAKE_CASE		,				truncation=SCREAMING_SNAKE_CASE      )
  a__												= self.image_processor(images=SCREAMING_SNAKE_CASE		,				return_tensors=self.framework      )
  model_inputs.update(SCREAMING_SNAKE_CASE      )
  return model_inputs

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			Union[str, Any]:
  a__												= self.model(**SCREAMING_SNAKE_CASE      )
  return model_outputs



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=5      ) ->			Optional[Any]:
  if top_k > self.model.config.num_labels:
   a__												= self.model.config.num_labels

  if self.framework == "pt":
   a__												= model_outputs.logits.sigmoid()[0]
   a__							,			a__												= probs.topk(SCREAMING_SNAKE_CASE      )
  else:
   raise ValueError(f"Unsupported framework: {self.framework}"      )

  a__												= scores.tolist()
  a__												= ids.tolist()
  return [{"score": score, "answer": self.model.config.idalabel[_id]} for score, _id in zip(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )]





